research(rectification): score D12 parent probes, sample underpowered (BUG-1308)
This commit is contained in:
jesse-ux
@@ -0,0 +1,595 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Pre-registered D12 parent-probe scoring. Research only.
|
||||
|
||||
The reading rules live in
|
||||
``docs/research/d12_parent_probes_preregistration_2026_10_12.json``.
|
||||
This module does not compute an ephemeris. Callers pass longitudes from
|
||||
``domain_calculation_service.compute_chart`` and signs from ``varga.calc_varga``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import sys
|
||||
from collections.abc import Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
from random import Random
|
||||
from typing import Any
|
||||
|
||||
_SCRIPTS = str(Path(__file__).resolve().parents[1])
|
||||
if _SCRIPTS not in sys.path:
|
||||
sys.path.insert(0, _SCRIPTS)
|
||||
|
||||
import varga
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[2]
|
||||
PREREG_PATH = ROOT / "docs" / "research" / "d12_parent_probes_preregistration_2026_10_12.json"
|
||||
HOLDOUT_PATH = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v5.json"
|
||||
FACTS_PATH = ROOT / "references" / "real_case_calibration" / "parent_facts_v1.json"
|
||||
RESULTS_PATH = ROOT / "docs" / "research" / "d12_parent_probes_results_2026_10_12.json"
|
||||
|
||||
PROBES: tuple[str, ...] = (
|
||||
"mother_more_daily_care",
|
||||
"father_financial_lead",
|
||||
"father_guided_study",
|
||||
"father_work_institutional",
|
||||
"father_work_commerce",
|
||||
"parents_long_apart",
|
||||
)
|
||||
KENDRA = frozenset({1, 4, 7, 10})
|
||||
SEPARATED_DELTAS = frozenset({1, 5, 7, 11})
|
||||
ANSWERS = frozenset({"yes", "no", "unclear"})
|
||||
PREDICTIONS = frozenset({"yes", "no", "abstain"})
|
||||
GRADES = frozenset({"biography", "interview", "encyclopedia"})
|
||||
ALPHA = Decimal("0.0125")
|
||||
ROBUST_MAX = Decimal("0.2")
|
||||
CONTROL_P_MAX = Decimal("0.01")
|
||||
MIN_DECISIVE = 4
|
||||
MIN_INCLUDED = 60
|
||||
MIN_DIFFERING = 20
|
||||
PERMUTATIONS = 999
|
||||
PERM_SEED = 20261012
|
||||
CONTROL_SEED = 20261013
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class Placement:
|
||||
d12_lagna: int
|
||||
d1_lagna: int
|
||||
d12_signs: Mapping[str, int]
|
||||
d1_signs: Mapping[str, int]
|
||||
|
||||
|
||||
def canonical_bytes(payload: Any) -> bytes:
|
||||
return (
|
||||
json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
|
||||
).encode("utf-8")
|
||||
|
||||
|
||||
def case_id_digest(case_ids: Sequence[str]) -> str:
|
||||
text = "\n".join(sorted(case_ids))
|
||||
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def load_prereg(path: Path = PREREG_PATH) -> dict[str, Any]:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def assert_prereg_locked(prereg: Mapping[str, Any]) -> None:
|
||||
if prereg.get("id") != "d12_parent_probes_preregistration_2026_10_12":
|
||||
raise SystemExit("preregistration id mismatch")
|
||||
if list(prereg.get("probes") or []) != list(PROBES):
|
||||
raise SystemExit("preregistration probe list mismatch")
|
||||
if int(prereg["permutation"]["n"]) != PERMUTATIONS:
|
||||
raise SystemExit("preregistration permutation count mismatch")
|
||||
if int(prereg["permutation"]["seed"]) != PERM_SEED:
|
||||
raise SystemExit("preregistration permutation seed mismatch")
|
||||
if int(prereg["positive_control"]["seed"]) != CONTROL_SEED:
|
||||
raise SystemExit("preregistration control seed mismatch")
|
||||
if int(prereg["positive_control"]["minimum_differing_cases"]) != MIN_DIFFERING:
|
||||
raise SystemExit("preregistration control minimum mismatch")
|
||||
if int(prereg["birth_records"]["expected_case_count"]) != 77:
|
||||
raise SystemExit("preregistration case count mismatch")
|
||||
if "at least 60" not in str(prereg["gates"]["sample"]):
|
||||
raise SystemExit("preregistration sample gate mismatch")
|
||||
if format(prereg["family"]["bonferroni_alpha"], ".4f") != "0.0125":
|
||||
raise SystemExit("preregistration alpha mismatch")
|
||||
if format(prereg["robustness"]["gate_max_change_rate"], ".1f") != "0.2":
|
||||
raise SystemExit("preregistration robustness line mismatch")
|
||||
if format(prereg["positive_control"]["permutation_p_max"], ".2f") != "0.01":
|
||||
raise SystemExit("preregistration control p mismatch")
|
||||
|
||||
|
||||
def house_of(body_sign: int, lagna: int) -> int:
|
||||
return (int(body_sign) - int(lagna)) % 12 + 1
|
||||
|
||||
|
||||
def lord_of(lagna: int, house: int) -> str:
|
||||
sign_index = (int(lagna) + int(house) - 1) % 12
|
||||
return varga.SIGN_LORDS[varga.SIGNS[sign_index]]
|
||||
|
||||
|
||||
def career_house(father_house: int) -> int:
|
||||
return ((int(father_house) - 1 + 9) % 12) + 1
|
||||
|
||||
|
||||
def separated(sign_a: int, sign_b: int) -> bool:
|
||||
return (int(sign_b) - int(sign_a)) % 12 in SEPARATED_DELTAS
|
||||
|
||||
|
||||
def _unique(planets: Sequence[str]) -> tuple[str, ...]:
|
||||
return tuple(dict.fromkeys(planets))
|
||||
|
||||
|
||||
def _in_kendra(signs: Mapping[str, int], lagna: int, planet: str) -> bool:
|
||||
return house_of(signs[planet], lagna) in KENDRA
|
||||
|
||||
|
||||
def read_probes(
|
||||
signs: Mapping[str, int],
|
||||
lagna: int,
|
||||
father_house: int,
|
||||
) -> dict[str, str]:
|
||||
"""Return yes / no / abstain for the six pre-registered probes."""
|
||||
mother = _unique(("Moon", lord_of(lagna, 4)))
|
||||
father = _unique(("Sun", lord_of(lagna, father_house)))
|
||||
mother_kendra = sum(_in_kendra(signs, lagna, planet) for planet in mother)
|
||||
father_kendra = sum(_in_kendra(signs, lagna, planet) for planet in father)
|
||||
if mother_kendra > father_kendra:
|
||||
care = "yes"
|
||||
elif father_kendra > mother_kendra:
|
||||
care = "no"
|
||||
elif _in_kendra(signs, lagna, "Moon") and not _in_kendra(signs, lagna, "Sun"):
|
||||
care = "yes"
|
||||
elif _in_kendra(signs, lagna, "Sun") and not _in_kendra(signs, lagna, "Moon"):
|
||||
care = "no"
|
||||
else:
|
||||
care = "abstain"
|
||||
|
||||
lord2 = lord_of(lagna, 2)
|
||||
financial = (
|
||||
any(house_of(signs[planet], lagna) in {2, 11} for planet in father)
|
||||
or house_of(signs[lord2], lagna) == father_house
|
||||
or any(signs[lord2] == signs[planet] for planet in father)
|
||||
)
|
||||
lord5 = lord_of(lagna, 5)
|
||||
study = (
|
||||
any(house_of(signs[planet], lagna) == 5 for planet in father)
|
||||
or house_of(signs[lord5], lagna) == father_house
|
||||
or house_of(signs["Jupiter"], lagna) == father_house
|
||||
or any(signs["Jupiter"] == signs[planet] for planet in father)
|
||||
)
|
||||
career = career_house(father_house)
|
||||
career_lord = lord_of(lagna, career)
|
||||
institutional = (
|
||||
house_of(signs["Saturn"], lagna) == career
|
||||
or house_of(signs["Sun"], lagna) == career
|
||||
or career_lord in {"Saturn", "Sun"}
|
||||
)
|
||||
commerce = (
|
||||
house_of(signs["Mercury"], lagna) == career
|
||||
or house_of(signs["Venus"], lagna) == career
|
||||
or career_lord in {"Mercury", "Venus"}
|
||||
)
|
||||
lord4 = lord_of(lagna, 4)
|
||||
father_lord = lord_of(lagna, father_house)
|
||||
lords_apart = lord4 != father_lord and separated(signs[lord4], signs[father_lord])
|
||||
rahu_house = house_of(signs["Rahu"], lagna) if "Rahu" in signs else None
|
||||
apart = (
|
||||
separated(signs["Sun"], signs["Moon"])
|
||||
or lords_apart
|
||||
or rahu_house in {4, father_house}
|
||||
)
|
||||
return {
|
||||
"mother_more_daily_care": care,
|
||||
"father_financial_lead": "yes" if financial else "no",
|
||||
"father_guided_study": "yes" if study else "no",
|
||||
"father_work_institutional": "yes" if institutional else "no",
|
||||
"father_work_commerce": "yes" if commerce else "no",
|
||||
"parents_long_apart": "yes" if apart else "no",
|
||||
}
|
||||
|
||||
|
||||
def placement_from_longitudes(
|
||||
ascendant_longitude: float,
|
||||
planet_longitudes: Mapping[str, float],
|
||||
planet_d1_signs: Mapping[str, int],
|
||||
) -> Placement:
|
||||
d12_signs = {
|
||||
name: int(varga.calc_varga(float(longitude), 12)["sign_idx"])
|
||||
for name, longitude in planet_longitudes.items()
|
||||
}
|
||||
d1_signs = {name: int(sign) for name, sign in planet_d1_signs.items()}
|
||||
ascendant = float(ascendant_longitude)
|
||||
return Placement(
|
||||
d12_lagna=int(varga.calc_varga(ascendant, 12)["sign_idx"]),
|
||||
d1_lagna=int(ascendant // 30) % 12,
|
||||
d12_signs=d12_signs,
|
||||
d1_signs=d1_signs,
|
||||
)
|
||||
|
||||
|
||||
def predict(placement: Placement, chart: str, father_house: int) -> dict[str, str]:
|
||||
if chart == "D12":
|
||||
return read_probes(placement.d12_signs, placement.d12_lagna, father_house)
|
||||
if chart == "D1":
|
||||
return read_probes(placement.d1_signs, placement.d1_lagna, father_house)
|
||||
raise ValueError(f"unknown chart {chart}")
|
||||
|
||||
|
||||
def cuts(sign_by_offset: Mapping[int, int]) -> list[tuple[int, int]]:
|
||||
offsets = sorted(int(item) for item in sign_by_offset)
|
||||
found: list[tuple[int, int]] = []
|
||||
for left, right in zip(offsets, offsets[1:]):
|
||||
if right == left + 1 and sign_by_offset[left] != sign_by_offset[right]:
|
||||
found.append((left, right))
|
||||
return found
|
||||
|
||||
|
||||
def nearest_cut(pairs: Sequence[tuple[int, int]]) -> tuple[int, int] | None:
|
||||
if not pairs:
|
||||
return None
|
||||
return min(pairs, key=lambda pair: (min(abs(pair[0]), abs(pair[1])), abs(pair[0]), pair[0]))
|
||||
|
||||
|
||||
def true_side(
|
||||
sign_by_offset: Mapping[int, int],
|
||||
pair: tuple[int, int],
|
||||
) -> tuple[int, int]:
|
||||
truth = sign_by_offset[0]
|
||||
left, right = pair
|
||||
left_true = sign_by_offset[left] == truth
|
||||
right_true = sign_by_offset[right] == truth
|
||||
if left_true == right_true:
|
||||
raise RuntimeError("boundary does not separate the recorded sign")
|
||||
if left_true:
|
||||
return left, right
|
||||
return right, left
|
||||
|
||||
|
||||
def segments(sign_by_offset: Mapping[int, int]) -> list[list[int]]:
|
||||
offsets = sorted(int(item) for item in sign_by_offset)
|
||||
if not offsets:
|
||||
return []
|
||||
runs: list[list[int]] = [[offsets[0]]]
|
||||
for offset in offsets[1:]:
|
||||
if offset == runs[-1][-1] + 1 and sign_by_offset[offset] == sign_by_offset[runs[-1][-1]]:
|
||||
runs[-1].append(offset)
|
||||
else:
|
||||
runs.append([offset])
|
||||
return runs
|
||||
|
||||
|
||||
def segment_representative(run: Sequence[int]) -> int:
|
||||
return min(run, key=lambda offset: (abs(offset), offset))
|
||||
|
||||
|
||||
def side_score(prediction: Mapping[str, str], answers: Mapping[str, str]) -> int:
|
||||
score = 0
|
||||
for probe in PROBES:
|
||||
answer = answers[probe]
|
||||
label = prediction[probe]
|
||||
if answer not in {"yes", "no"} or label not in {"yes", "no"}:
|
||||
continue
|
||||
score += 1 if label == answer else -1
|
||||
return score
|
||||
|
||||
|
||||
def fisher_yates(items: Sequence[str], rng: Random) -> list[str]:
|
||||
out = list(items)
|
||||
for index in range(len(out) - 1, 0, -1):
|
||||
swap = rng.randrange(index + 1)
|
||||
out[index], out[swap] = out[swap], out[index]
|
||||
return out
|
||||
|
||||
|
||||
def permutation_maps(case_ids: Sequence[str], seed: int, count: int = PERMUTATIONS) -> list[list[str]]:
|
||||
rng = Random(seed)
|
||||
ordered = sorted(case_ids)
|
||||
return [fisher_yates(ordered, rng) for _ in range(count)]
|
||||
|
||||
|
||||
def assign_answers(
|
||||
destinations: Sequence[str],
|
||||
sources: Sequence[str],
|
||||
answers: Mapping[str, Mapping[str, str]],
|
||||
) -> dict[str, Mapping[str, str]]:
|
||||
if len(destinations) != len(sources):
|
||||
raise RuntimeError("permutation length mismatch")
|
||||
return {dest: answers[source] for dest, source in zip(destinations, sources)}
|
||||
|
||||
|
||||
def selection_count(
|
||||
case_ids: Sequence[str],
|
||||
answers: Mapping[str, Mapping[str, str]],
|
||||
true_predictions: Mapping[str, Mapping[str, str]],
|
||||
other_predictions: Mapping[str, Mapping[str, str]],
|
||||
) -> int:
|
||||
selected = 0
|
||||
for case_id in case_ids:
|
||||
true_score = side_score(true_predictions[case_id], answers[case_id])
|
||||
other_score = side_score(other_predictions[case_id], answers[case_id])
|
||||
if true_score > other_score:
|
||||
selected += 1
|
||||
return selected
|
||||
|
||||
|
||||
def ratio_below(numer: int, denom: int, threshold: Decimal) -> bool:
|
||||
if denom <= 0:
|
||||
return False
|
||||
return Decimal(numer) / Decimal(denom) < threshold
|
||||
|
||||
|
||||
def display_ratio(numer: int, denom: int) -> str:
|
||||
if denom <= 0:
|
||||
return "0.000000"
|
||||
return format(numer / denom, ".6f")
|
||||
|
||||
|
||||
def permutation_summary(observed: int, denom: int, perm_counts: Sequence[int]) -> dict[str, Any]:
|
||||
extreme = sum(1 for count in perm_counts if count >= observed)
|
||||
below = sum(1 for count in perm_counts if count < observed)
|
||||
null_sum = sum(perm_counts)
|
||||
n_perm = len(perm_counts)
|
||||
greater = observed * n_perm > null_sum if denom > 0 else False
|
||||
return {
|
||||
"observed_numer": observed,
|
||||
"observed_denom": denom,
|
||||
"observed_display": display_ratio(observed, denom),
|
||||
"extreme": extreme,
|
||||
"p_numer": extreme + 1,
|
||||
"p_denom": n_perm + 1,
|
||||
"p_display": display_ratio(extreme + 1, n_perm + 1),
|
||||
"null_sum": null_sum,
|
||||
"null_mean_display": display_ratio(null_sum, n_perm * denom) if denom else "0.000000",
|
||||
"above_null_mean": greater,
|
||||
"percentile_numer": below,
|
||||
"percentile_denom": n_perm,
|
||||
"percentile_display": display_ratio(below, n_perm),
|
||||
}
|
||||
|
||||
|
||||
def cell_is_significant(summary: Mapping[str, Any]) -> bool:
|
||||
return bool(summary["above_null_mean"]) and ratio_below(
|
||||
int(summary["p_numer"]),
|
||||
int(summary["p_denom"]),
|
||||
ALPHA,
|
||||
)
|
||||
|
||||
|
||||
def vectors_differ(
|
||||
true_prediction: Mapping[str, str],
|
||||
other_prediction: Mapping[str, str],
|
||||
) -> bool:
|
||||
return any(
|
||||
true_prediction[probe] in {"yes", "no"} and true_prediction[probe] != other_prediction[probe]
|
||||
for probe in PROBES
|
||||
)
|
||||
|
||||
|
||||
def plant_answers(true_prediction: Mapping[str, str]) -> dict[str, str]:
|
||||
planted: dict[str, str] = {}
|
||||
for probe in PROBES:
|
||||
label = true_prediction[probe]
|
||||
planted[probe] = label if label in {"yes", "no"} else "unclear"
|
||||
return planted
|
||||
|
||||
|
||||
def change_count(
|
||||
baseline: Mapping[str, Mapping[int, Mapping[str, str]]],
|
||||
alternate: Mapping[str, Mapping[int, Mapping[str, str]]],
|
||||
sides: Mapping[str, tuple[int, int]],
|
||||
) -> int:
|
||||
changes = 0
|
||||
for case_id, pair in sides.items():
|
||||
for offset in pair:
|
||||
for probe in PROBES:
|
||||
if baseline[case_id][offset][probe] != alternate[case_id][offset][probe]:
|
||||
changes += 1
|
||||
return changes
|
||||
|
||||
|
||||
def decisive_count(answers: Mapping[str, str]) -> int:
|
||||
return sum(answer in {"yes", "no"} for answer in answers.values())
|
||||
|
||||
|
||||
def validate_fact_case(case: Mapping[str, Any]) -> dict[str, str]:
|
||||
probes = case.get("probes")
|
||||
if not isinstance(probes, dict) or set(probes) != set(PROBES):
|
||||
raise SystemExit(f"fact probes incomplete for {case.get('case_id')}")
|
||||
answers: dict[str, str] = {}
|
||||
for probe in PROBES:
|
||||
row = probes[probe]
|
||||
if not isinstance(row, dict):
|
||||
raise SystemExit(f"fact row missing for {case.get('case_id')} {probe}")
|
||||
answer = str(row.get("answer") or "")
|
||||
grade = str(row.get("source_grade") or "")
|
||||
url = str(row.get("url") or "")
|
||||
excerpt = str(row.get("excerpt") or "").strip()
|
||||
if answer not in ANSWERS or grade not in GRADES or not url.startswith("https://") or not excerpt:
|
||||
raise SystemExit(f"fact source incomplete for {case.get('case_id')} {probe}")
|
||||
answers[probe] = answer
|
||||
return answers
|
||||
|
||||
|
||||
def answer_tally(cases: Sequence[Mapping[str, str]]) -> dict[str, dict[str, int]]:
|
||||
tally = {probe: {"yes": 0, "no": 0, "unclear": 0} for probe in PROBES}
|
||||
for answers in cases:
|
||||
for probe in PROBES:
|
||||
tally[probe][answers[probe]] += 1
|
||||
return tally
|
||||
|
||||
|
||||
def window_offsets(radius: int) -> list[int]:
|
||||
return list(range(-int(radius), int(radius) + 1))
|
||||
|
||||
|
||||
def _slice_map(values: Mapping[int, Any], offsets: Sequence[int]) -> dict[int, Any]:
|
||||
return {int(offset): values[int(offset)] for offset in offsets}
|
||||
|
||||
|
||||
def case_window(
|
||||
sign_by_offset: Mapping[int, int],
|
||||
prediction_by_offset: Mapping[int, Mapping[str, str]],
|
||||
answers: Mapping[str, str],
|
||||
radius: int,
|
||||
) -> dict[str, Any]:
|
||||
signs = _slice_map(sign_by_offset, window_offsets(radius))
|
||||
pair = nearest_cut(cuts(signs))
|
||||
if pair is None:
|
||||
return {
|
||||
"boundary": False,
|
||||
"cut": None,
|
||||
"true_offset": None,
|
||||
"other_offset": None,
|
||||
"true_score": None,
|
||||
"other_score": None,
|
||||
"selected": False,
|
||||
"true_predictions": None,
|
||||
"other_predictions": None,
|
||||
"segments": 1 if signs else 0,
|
||||
"head_hit": None,
|
||||
"retained": None,
|
||||
}
|
||||
true_offset, other_offset = true_side(signs, pair)
|
||||
true_prediction = dict(prediction_by_offset[true_offset])
|
||||
other_prediction = dict(prediction_by_offset[other_offset])
|
||||
true_score = side_score(true_prediction, answers)
|
||||
other_score = side_score(other_prediction, answers)
|
||||
runs = segments(signs)
|
||||
scored: list[tuple[list[int], int]] = []
|
||||
for run in runs:
|
||||
representative = segment_representative(run)
|
||||
scored.append((run, side_score(prediction_by_offset[representative], answers)))
|
||||
true_segment = next(score for run, score in scored if 0 in run)
|
||||
other_scores = [score for run, score in scored if 0 not in run]
|
||||
return {
|
||||
"boundary": True,
|
||||
"cut": [pair[0], pair[1]],
|
||||
"true_offset": true_offset,
|
||||
"other_offset": other_offset,
|
||||
"true_score": true_score,
|
||||
"other_score": other_score,
|
||||
"selected": true_score > other_score,
|
||||
"true_predictions": true_prediction,
|
||||
"other_predictions": other_prediction,
|
||||
"segments": len(runs),
|
||||
"head_hit": true_segment > max(other_scores),
|
||||
"retained": true_segment >= max(other_scores),
|
||||
"differ": vectors_differ(true_prediction, other_prediction),
|
||||
}
|
||||
|
||||
|
||||
def evaluate_cell(
|
||||
case_ids: Sequence[str],
|
||||
signs: Mapping[str, Mapping[int, int]],
|
||||
predictions: Mapping[str, Mapping[int, Mapping[str, str]]],
|
||||
answers: Mapping[str, Mapping[str, str]],
|
||||
radius: int,
|
||||
perm_maps: Sequence[Sequence[str]],
|
||||
) -> dict[str, Any]:
|
||||
ordered = sorted(case_ids)
|
||||
rows = {
|
||||
case_id: case_window(signs[case_id], predictions[case_id], answers[case_id], radius)
|
||||
for case_id in ordered
|
||||
}
|
||||
boundary_ids = [case_id for case_id in ordered if rows[case_id]["boundary"]]
|
||||
true_predictions = {case_id: rows[case_id]["true_predictions"] for case_id in boundary_ids}
|
||||
other_predictions = {case_id: rows[case_id]["other_predictions"] for case_id in boundary_ids}
|
||||
observed = selection_count(boundary_ids, answers, true_predictions, other_predictions)
|
||||
perm_counts = []
|
||||
for sources in perm_maps:
|
||||
assigned = assign_answers(ordered, sources, answers)
|
||||
perm_counts.append(
|
||||
selection_count(boundary_ids, assigned, true_predictions, other_predictions)
|
||||
)
|
||||
summary = permutation_summary(observed, len(boundary_ids), perm_counts)
|
||||
head_hits = sum(1 for case_id in boundary_ids if rows[case_id]["head_hit"])
|
||||
retained = sum(1 for case_id in boundary_ids if rows[case_id]["retained"])
|
||||
return {
|
||||
"summary": summary,
|
||||
"significant": cell_is_significant(summary),
|
||||
"head_hit_numer": head_hits,
|
||||
"head_hit_denom": len(boundary_ids),
|
||||
"head_hit_display": display_ratio(head_hits, len(boundary_ids)),
|
||||
"retained_numer": retained,
|
||||
"retained_denom": len(boundary_ids),
|
||||
"retained_display": display_ratio(retained, len(boundary_ids)),
|
||||
"no_boundary": len(ordered) - len(boundary_ids),
|
||||
"cases": rows,
|
||||
"not_a_gate_segment_rates": True,
|
||||
}
|
||||
|
||||
|
||||
def positive_control_cell(
|
||||
case_ids: Sequence[str],
|
||||
signs: Mapping[str, Mapping[int, int]],
|
||||
predictions: Mapping[str, Mapping[int, Mapping[str, str]]],
|
||||
radius: int,
|
||||
) -> dict[str, Any]:
|
||||
"""Plant the true side's own labels. Permute only cases that have a cut.
|
||||
|
||||
Those are the cases with a planted vector. The stream starts at the
|
||||
pre-registered control seed for every cell.
|
||||
"""
|
||||
blank = {probe: "unclear" for probe in PROBES}
|
||||
boundary_ids: list[str] = []
|
||||
true_predictions: dict[str, Mapping[str, str]] = {}
|
||||
other_predictions: dict[str, Mapping[str, str]] = {}
|
||||
planted: dict[str, dict[str, str]] = {}
|
||||
missed: list[str] = []
|
||||
differing = 0
|
||||
for case_id in sorted(case_ids):
|
||||
row = case_window(signs[case_id], predictions[case_id], blank, radius)
|
||||
if not row["boundary"]:
|
||||
continue
|
||||
boundary_ids.append(case_id)
|
||||
true_predictions[case_id] = row["true_predictions"]
|
||||
other_predictions[case_id] = row["other_predictions"]
|
||||
planted[case_id] = plant_answers(row["true_predictions"])
|
||||
if row["differ"]:
|
||||
differing += 1
|
||||
checked = case_window(signs[case_id], predictions[case_id], planted[case_id], radius)
|
||||
if not checked["selected"]:
|
||||
missed.append(case_id)
|
||||
maps = permutation_maps(boundary_ids, CONTROL_SEED, PERMUTATIONS)
|
||||
observed = selection_count(boundary_ids, planted, true_predictions, other_predictions)
|
||||
perm_counts = [
|
||||
selection_count(
|
||||
boundary_ids,
|
||||
assign_answers(boundary_ids, sources, planted),
|
||||
true_predictions,
|
||||
other_predictions,
|
||||
)
|
||||
for sources in maps
|
||||
]
|
||||
summary = permutation_summary(observed, len(boundary_ids), perm_counts)
|
||||
return {
|
||||
"summary": summary,
|
||||
"differing": differing,
|
||||
"pipeline_exact": not missed,
|
||||
"missed": missed,
|
||||
"p_below": ratio_below(int(summary["p_numer"]), int(summary["p_denom"]), CONTROL_P_MAX),
|
||||
}
|
||||
|
||||
|
||||
def decide_status(
|
||||
*,
|
||||
included: int,
|
||||
control_exact: bool,
|
||||
differing: int,
|
||||
control_p_below: bool,
|
||||
any_robust_significant: bool,
|
||||
) -> str:
|
||||
if not control_exact:
|
||||
return "control_invalid"
|
||||
if differing < MIN_DIFFERING or not control_p_below:
|
||||
return "control_underpowered"
|
||||
if included < MIN_INCLUDED:
|
||||
return "underpowered_sample"
|
||||
if any_robust_significant:
|
||||
return "pass"
|
||||
return "fail_closed"
|
||||
@@ -0,0 +1,338 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Score the pre-registered D12 parent probes. Research only.
|
||||
|
||||
Charts come from domain_calculation_service.compute_chart. D12 signs come
|
||||
from varga.calc_varga. The script writes one JSON file and no timestamps.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import sys
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
_SCRIPTS = str(Path(__file__).resolve().parents[1])
|
||||
if _SCRIPTS not in sys.path:
|
||||
sys.path.insert(0, _SCRIPTS)
|
||||
|
||||
import domain_calculation_service
|
||||
import swisseph as swe
|
||||
|
||||
from scripts.research import d12_parent_probes_lib as lib
|
||||
|
||||
PROFILES = {
|
||||
"father_9": ("D12", 9),
|
||||
"father_10": ("D12", 10),
|
||||
"d1_father_9": ("D1", 9),
|
||||
}
|
||||
FAMILY = (("father_9", 10), ("father_9", 30), ("father_10", 10), ("father_10", 30))
|
||||
SETTINGS = {
|
||||
"baseline": ("raman", "mean"),
|
||||
"lahiri_mean": ("lahiri", "mean"),
|
||||
"raman_true": ("raman", "true"),
|
||||
}
|
||||
OFFSETS = list(range(-30, 31))
|
||||
NEEDED = ("Sun", "Moon", "Mercury", "Venus", "Mars", "Jupiter", "Saturn", "Rahu")
|
||||
|
||||
|
||||
def file_sha256(path: Path) -> str:
|
||||
return hashlib.sha256(path.read_bytes()).hexdigest()
|
||||
|
||||
|
||||
def birth_moment(birth: dict[str, Any]) -> datetime:
|
||||
clock = str(birth["time"])[:5]
|
||||
return datetime.strptime(f"{birth['date']} {clock}", "%Y-%m-%d %H:%M")
|
||||
|
||||
|
||||
def placement_at(moment: datetime, birth: dict[str, Any], ayanamsa: str, node_mode: str) -> lib.Placement:
|
||||
chart = domain_calculation_service.compute_chart(
|
||||
{
|
||||
"year": moment.year,
|
||||
"month": moment.month,
|
||||
"day": moment.day,
|
||||
"hour": moment.hour,
|
||||
"minute": moment.minute,
|
||||
"second": 0,
|
||||
"lat": float(birth["latitude"]),
|
||||
"lon": float(birth["longitude"]),
|
||||
"tz": float(birth["timezone_offset"]),
|
||||
"ayanamsa": ayanamsa,
|
||||
"node_mode": node_mode,
|
||||
}
|
||||
)
|
||||
planets = chart["planets"]
|
||||
longitudes: dict[str, float] = {}
|
||||
d1_signs: dict[str, int] = {}
|
||||
for name in NEEDED:
|
||||
row = planets.get(name)
|
||||
if not isinstance(row, dict) or "lon" not in row:
|
||||
raise RuntimeError(f"chart is missing {name}")
|
||||
longitudes[name] = float(row["lon"])
|
||||
if "sign_idx" in row:
|
||||
d1_signs[name] = int(row["sign_idx"])
|
||||
else:
|
||||
d1_signs[name] = int(float(row["lon"]) // 30) % 12
|
||||
ascendant = chart.get("ascendant") or {}
|
||||
if "lon" not in ascendant:
|
||||
raise RuntimeError("chart is missing the ascendant longitude")
|
||||
return lib.placement_from_longitudes(float(ascendant["lon"]), longitudes, d1_signs)
|
||||
|
||||
|
||||
def sky_grid(cases: list[dict[str, Any]], ayanamsa: str, node_mode: str) -> dict[str, dict[int, lib.Placement]]:
|
||||
grid: dict[str, dict[int, lib.Placement]] = {}
|
||||
for index, case in enumerate(cases, start=1):
|
||||
moment = birth_moment(case["birth"])
|
||||
rows: dict[int, lib.Placement] = {}
|
||||
for offset in OFFSETS:
|
||||
rows[offset] = placement_at(moment + timedelta(minutes=offset), case["birth"], ayanamsa, node_mode)
|
||||
grid[str(case["case_id"])] = rows
|
||||
print(f"{ayanamsa}/{node_mode} {index}/{len(cases)}", file=sys.stderr)
|
||||
return grid
|
||||
|
||||
|
||||
def signs_of(grid: dict[str, dict[int, lib.Placement]]) -> dict[str, dict[int, int]]:
|
||||
return {
|
||||
case_id: {offset: placement.d12_lagna for offset, placement in rows.items()}
|
||||
for case_id, rows in grid.items()
|
||||
}
|
||||
|
||||
|
||||
def predictions_of(
|
||||
grid: dict[str, dict[int, lib.Placement]],
|
||||
chart: str,
|
||||
father_house: int,
|
||||
) -> dict[str, dict[int, dict[str, str]]]:
|
||||
return {
|
||||
case_id: {
|
||||
offset: lib.predict(placement, chart, father_house)
|
||||
for offset, placement in rows.items()
|
||||
}
|
||||
for case_id, rows in grid.items()
|
||||
}
|
||||
|
||||
|
||||
def load_births(prereg: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
payload = json_load(lib.HOLDOUT_PATH)
|
||||
cases = list(payload["cases"])
|
||||
digest = lib.case_id_digest([str(case["case_id"]) for case in cases])
|
||||
if digest != prereg["birth_records"]["case_id_sha256"]:
|
||||
raise SystemExit("birth case-id digest does not match the preregistration")
|
||||
if len(cases) != 77:
|
||||
raise SystemExit("birth file does not contain 77 cases")
|
||||
return cases
|
||||
|
||||
|
||||
def json_load(path: Path) -> dict[str, Any]:
|
||||
import json
|
||||
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def load_answers(birth_ids: set[str]) -> tuple[dict[str, dict[str, str]], list[dict[str, Any]], list[str]]:
|
||||
if not lib.FACTS_PATH.is_file():
|
||||
raise SystemExit(f"missing facts: {lib.FACTS_PATH}")
|
||||
payload = json_load(lib.FACTS_PATH)
|
||||
coded: dict[str, dict[str, str]] = {}
|
||||
excluded: list[dict[str, Any]] = []
|
||||
for case in payload.get("cases") or []:
|
||||
case_id = str(case.get("case_id") or "")
|
||||
if case_id not in birth_ids:
|
||||
raise SystemExit(f"fact case is not in the frozen birth list: {case_id}")
|
||||
answers = lib.validate_fact_case(case)
|
||||
decisive = lib.decisive_count(answers)
|
||||
if decisive >= lib.MIN_DECISIVE:
|
||||
coded[case_id] = answers
|
||||
else:
|
||||
excluded.append({"case_id": case_id, "decisive": decisive})
|
||||
not_coded = sorted(birth_ids - set(coded) - {row["case_id"] for row in excluded})
|
||||
return coded, excluded, not_coded
|
||||
|
||||
|
||||
def robustness_block(
|
||||
rows: dict[str, dict[str, Any]],
|
||||
baseline: dict[str, dict[int, dict[str, str]]],
|
||||
alternate: dict[str, dict[int, dict[str, str]]],
|
||||
) -> dict[str, Any]:
|
||||
sides = {
|
||||
case_id: (int(row["true_offset"]), int(row["other_offset"]))
|
||||
for case_id, row in rows.items()
|
||||
if row["boundary"]
|
||||
}
|
||||
changes = lib.change_count(baseline, alternate, sides)
|
||||
denom = len(sides) * 2 * len(lib.PROBES)
|
||||
return {
|
||||
"changes": changes,
|
||||
"denom": denom,
|
||||
"display": lib.display_ratio(changes, denom),
|
||||
"below_line": lib.ratio_below(changes, denom, lib.ROBUST_MAX),
|
||||
}
|
||||
|
||||
|
||||
def public_cases(rows: dict[str, dict[str, Any]]) -> list[dict[str, Any]]:
|
||||
published = []
|
||||
for case_id in sorted(rows):
|
||||
row = dict(rows[case_id])
|
||||
row["case_id"] = case_id
|
||||
row.pop("differ", None)
|
||||
published.append(row)
|
||||
return published
|
||||
|
||||
|
||||
def ephemeris_version() -> str:
|
||||
version = getattr(swe, "version", None)
|
||||
return str(version) if version else "unknown"
|
||||
|
||||
|
||||
def run() -> dict[str, Any]:
|
||||
prereg = lib.load_prereg()
|
||||
lib.assert_prereg_locked(prereg)
|
||||
births = load_births(prereg)
|
||||
birth_ids = {str(case["case_id"]) for case in births}
|
||||
answers, excluded, not_coded = load_answers(birth_ids)
|
||||
included_ids = sorted(answers)
|
||||
included_births = [case for case in births if str(case["case_id"]) in answers]
|
||||
included_births.sort(key=lambda case: str(case["case_id"]))
|
||||
skies = {
|
||||
name: sky_grid(included_births, ayanamsa, node_mode)
|
||||
for name, (ayanamsa, node_mode) in SETTINGS.items()
|
||||
}
|
||||
sign_grid = signs_of(skies["baseline"])
|
||||
predicted = {
|
||||
profile: predictions_of(skies["baseline"], chart, father_house)
|
||||
for profile, (chart, father_house) in PROFILES.items()
|
||||
}
|
||||
alternate = {
|
||||
setting: {
|
||||
profile: predictions_of(skies[setting], "D12", father_house)
|
||||
for profile, father_house in (("father_9", 9), ("father_10", 10))
|
||||
}
|
||||
for setting in ("lahiri_mean", "raman_true")
|
||||
}
|
||||
maps = lib.permutation_maps(included_ids, lib.PERM_SEED, lib.PERMUTATIONS)
|
||||
cells: dict[str, Any] = {}
|
||||
controls: dict[str, Any] = {}
|
||||
any_robust_significant = False
|
||||
for profile, radius in FAMILY:
|
||||
key = f"{profile}_r{radius}"
|
||||
evaluated = lib.evaluate_cell(
|
||||
included_ids,
|
||||
sign_grid,
|
||||
predicted[profile],
|
||||
answers,
|
||||
radius,
|
||||
maps,
|
||||
)
|
||||
ayanamsa_shift = robustness_block(
|
||||
evaluated["cases"],
|
||||
predicted[profile],
|
||||
alternate["lahiri_mean"][profile],
|
||||
)
|
||||
node_shift = robustness_block(
|
||||
evaluated["cases"],
|
||||
predicted[profile],
|
||||
alternate["raman_true"][profile],
|
||||
)
|
||||
robust = bool(ayanamsa_shift["below_line"] and node_shift["below_line"])
|
||||
if evaluated["significant"] and robust:
|
||||
any_robust_significant = True
|
||||
control = lib.positive_control_cell(included_ids, sign_grid, predicted[profile], radius)
|
||||
cells[key] = {
|
||||
"profile": profile,
|
||||
"radius_minutes": radius,
|
||||
"summary": evaluated["summary"],
|
||||
"significant": evaluated["significant"],
|
||||
"head_hit_numer": evaluated["head_hit_numer"],
|
||||
"head_hit_denom": evaluated["head_hit_denom"],
|
||||
"head_hit_display": evaluated["head_hit_display"],
|
||||
"retained_numer": evaluated["retained_numer"],
|
||||
"retained_denom": evaluated["retained_denom"],
|
||||
"retained_display": evaluated["retained_display"],
|
||||
"no_boundary": evaluated["no_boundary"],
|
||||
"robustness_ayanamsa": ayanamsa_shift,
|
||||
"robustness_node": node_shift,
|
||||
"robust": robust,
|
||||
"not_a_gate_segment_rates": True,
|
||||
"cases": public_cases(evaluated["cases"]),
|
||||
}
|
||||
controls[key] = {
|
||||
"summary": control["summary"],
|
||||
"differing": control["differing"],
|
||||
"pipeline_exact": control["pipeline_exact"],
|
||||
"missed": control["missed"],
|
||||
"p_below_0_01": control["p_below"],
|
||||
}
|
||||
d1_cells = {}
|
||||
for radius in (10, 30):
|
||||
evaluated = lib.evaluate_cell(
|
||||
included_ids,
|
||||
sign_grid,
|
||||
predicted["d1_father_9"],
|
||||
answers,
|
||||
radius,
|
||||
maps,
|
||||
)
|
||||
d1_cells[f"d1_father_9_r{radius}"] = {
|
||||
"summary": evaluated["summary"],
|
||||
"significant_if_it_were_in_the_family": evaluated["significant"],
|
||||
"head_hit_display": evaluated["head_hit_display"],
|
||||
"retained_display": evaluated["retained_display"],
|
||||
"no_boundary": evaluated["no_boundary"],
|
||||
"not_a_gate": True,
|
||||
}
|
||||
control_exact = all(row["pipeline_exact"] for row in controls.values())
|
||||
min_differing = min((row["differing"] for row in controls.values()), default=0)
|
||||
control_p_below = all(row["p_below_0_01"] for row in controls.values())
|
||||
status = lib.decide_status(
|
||||
included=len(included_ids),
|
||||
control_exact=control_exact,
|
||||
differing=min_differing,
|
||||
control_p_below=control_p_below,
|
||||
any_robust_significant=any_robust_significant,
|
||||
)
|
||||
coded_rows = []
|
||||
facts_payload = json_load(lib.FACTS_PATH)
|
||||
for case in facts_payload.get("cases") or []:
|
||||
coded_rows.append(lib.validate_fact_case(case))
|
||||
return {
|
||||
"id": "d12_parent_probes_results_2026_10_12",
|
||||
"preregistration": "docs/research/d12_parent_probes_preregistration_2026_10_12.json",
|
||||
"preregistration_sha256": file_sha256(lib.PREREG_PATH),
|
||||
"facts_sha256": file_sha256(lib.FACTS_PATH),
|
||||
"ephemeris": ephemeris_version(),
|
||||
"python": f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}",
|
||||
"settings": {
|
||||
"baseline": ["raman", "mean"],
|
||||
"ayanamsa_alt": ["lahiri", "mean"],
|
||||
"node_alt": ["raman", "true"],
|
||||
},
|
||||
"coded_cases": len(coded_rows),
|
||||
"included_cases": len(included_ids),
|
||||
"included_case_ids": included_ids,
|
||||
"excluded_few_decisive": excluded,
|
||||
"not_coded": not_coded,
|
||||
"unclear_rate_coded": lib.answer_tally(coded_rows),
|
||||
"unclear_rate_included": lib.answer_tally([answers[case_id] for case_id in included_ids]),
|
||||
"cells": cells,
|
||||
"positive_control": controls,
|
||||
"d1_control": d1_cells,
|
||||
"decision": status,
|
||||
"decision_note": {
|
||||
"pass": "过门",
|
||||
"fail_closed": "不上线",
|
||||
"underpowered_sample": "欠力",
|
||||
"control_underpowered": "欠力",
|
||||
"control_invalid": "结论无效",
|
||||
}[status],
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
payload = run()
|
||||
lib.RESULTS_PATH.write_bytes(lib.canonical_bytes(payload))
|
||||
print(f"wrote {lib.RESULTS_PATH} decision={payload['decision']}", file=sys.stderr)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user