Jyotisha/scripts/research/holdout_v5_build.py

#!/usr/bin/env python3
"""Build minute_rectification_holdout_v5 from the v4 public AA set plus the reviewed v5 roster.

Inputs (all under references/real_case_calibration/):
  - minute_rectification_holdout_v4.json      the 20 inherited cases, copied verbatim
  - holdout_v5_roster.json                    new public Rodden-AA birth records + checks
  - minute_rectification_holdout_v5_events.json  reviewed, quoted events per new case

Output: minute_rectification_holdout_v5.json (deterministic; two builds are byte-identical).
Public biographies only. No private user data. No scoring code is touched.
"""

from __future__ import annotations

import argparse
import json
import sys
from collections import Counter
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[2]
CAL = ROOT / "references" / "real_case_calibration"
V4_PATH = CAL / "minute_rectification_holdout_v4.json"
ROSTER_PATH = CAL / "holdout_v5_roster.json"
EVENTS_PATH = CAL / "minute_rectification_holdout_v5_events.json"
V5_PATH = CAL / "minute_rectification_holdout_v5.json"

RADII = (10, 30, 60)
FALSE_MINUTE_OFFSETS = [-10, -5, -2, -1, 1, 2, 5, 10]
MIN_EVENTS = 7
MIN_DOMAINS = 4
MIN_CASES = 60
DOMAINS = {"education", "relocation", "relationship", "career", "finance", "health_pressure", "family", "appearance", "occupation"}
PRECISION_FORMAT = {"year": 4, "month": 7, "day": 10}


def load(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def latitude_band(lat: float) -> str:
    a = abs(lat)
    if a < 25:
        return "lt25"
    if a <= 45:
        return "25to45"
    return "gt45"


def night_birth(time: str) -> bool:
    hour = int(time.split(":")[0])
    return hour >= 22 or hour < 2


def strata_for(birth: dict[str, Any]) -> dict[str, Any]:
    year = int(birth["date"][:4])
    return {
        "decade": f"{year // 10 * 10}s",
        "latitude_band": latitude_band(float(birth["latitude"])),
        "hemisphere": "south" if float(birth["latitude"]) < 0 else "north",
        "night_birth_22_to_02": night_birth(birth["time"]),
        "utc_plus_8": abs(float(birth["timezone_offset"]) - 8.0) < 0.01,
    }


def validate_event(case_id: str, event: dict[str, Any], problems: list[str]) -> None:
    if event.get("domain") not in DOMAINS:
        problems.append(f"{case_id}: bad domain {event.get('domain')}")
    precision = event.get("precision")
    date = str(event.get("date", ""))
    if precision not in PRECISION_FORMAT or len(date) != PRECISION_FORMAT[precision]:
        problems.append(f"{case_id}: date {date!r} does not match precision {precision!r}")
    source = event.get("source") or {}
    if not source.get("url") or not source.get("independent_of_birth_source"):
        problems.append(f"{case_id}: event {event.get('id')} lacks an independent source url")


def build_new_case(roster_row: dict[str, Any], events: list[dict[str, Any]], problems: list[str]) -> dict[str, Any]:
    case_id = roster_row["case_id"]
    birth = roster_row["birth"]
    if (birth.get("source") or {}).get("rodden_rating") != "AA":
        problems.append(f"{case_id}: rodden_rating is not AA")
    checks = roster_row.get("checks") or {}
    if not checks.get("timezone_agrees"):
        problems.append(f"{case_id}: timezone check did not agree")
    if checks.get("place_distance_km") is None or float(checks["place_distance_km"]) > 60:
        problems.append(f"{case_id}: place/coordinate check missing or too far")
    for event in events:
        validate_event(case_id, event, problems)
    domains = sorted({e["domain"] for e in events})
    if len(events) < MIN_EVENTS or len(domains) < MIN_DOMAINS:
        problems.append(f"{case_id}: {len(events)} events / {len(domains)} domains below gate")
    return {
        "case_id": case_id,
        "subject_label": roster_row["subject_label"],
        "holdout_partition": "open_scale_evaluation",
        "excluded_from_tuning": True,
        "inherited_from_v4": False,
        "candidate_radius_minutes": RADII[0],
        "false_minute_offsets": list(FALSE_MINUTE_OFFSETS),
        "birth": birth,
        "checks": checks,
        "strata": strata_for(birth),
        "events": sorted(events, key=lambda e: (e["date"], e["id"])),
        "candidate_radii_minutes": list(RADII),
        "true_minute": birth["time"],
        "true_cluster_note": "Cluster membership depends on the scoring variant and radius; recorded in the sweep results.",
    }


def inherit_v4_case(case: dict[str, Any]) -> dict[str, Any]:
    out = dict(case)
    out["inherited_from_v4"] = True
    out["strata"] = strata_for(case["birth"])
    return out


def summarize(cases: list[dict[str, Any]]) -> dict[str, Any]:
    strata = [c["strata"] for c in cases]
    return {
        "cases": len(cases),
        "inherited_from_v4": sum(1 for c in cases if c.get("inherited_from_v4")),
        "events_total": sum(len(c["events"]) for c in cases),
        "events_by_precision": dict(sorted(Counter(e["precision"] for c in cases for e in c["events"]).items())),
        "events_by_domain": dict(sorted(Counter(e["domain"] for c in cases for e in c["events"]).items())),
        "decades": dict(sorted(Counter(s["decade"] for s in strata).items())),
        "latitude_bands": dict(sorted(Counter(s["latitude_band"] for s in strata).items())),
        "southern_hemisphere": sum(1 for s in strata if s["hemisphere"] == "south"),
        "night_birth_22_to_02": sum(1 for s in strata if s["night_birth_22_to_02"]),
        "utc_plus_8": sum(1 for s in strata if s["utc_plus_8"]),
        "countries": dict(sorted(Counter(c["birth"]["place"].split(",")[-1].strip() for c in cases).items())),
    }


def build(v4: dict[str, Any], roster: dict[str, Any], events_by_case: dict[str, Any]) -> tuple[dict[str, Any], list[str]]:
    problems: list[str] = []
    cases = [inherit_v4_case(c) for c in v4["cases"]]
    known = {c["case_id"] for c in cases}
    for row in roster["cases"]:
        case_id = row["case_id"]
        if case_id in known:
            problems.append(f"duplicate case_id {case_id}")
            continue
        events = (events_by_case.get(case_id) or {}).get("events") or []
        if not events:
            problems.append(f"{case_id}: no events file entry")
        cases.append(build_new_case(row, events, problems))
        known.add(case_id)
    cases.sort(key=lambda c: (not c.get("inherited_from_v4"), c["case_id"]))
    if len(cases) < MIN_CASES:
        problems.append(f"only {len(cases)} cases (< {MIN_CASES})")
    payload = {
        "schema_version": "minute-rectification-holdout-v5",
        "benchmark_id": "minute_rectification_scale_holdout_v5",
        "description": (
            "Public Rodden-AA minute set: the 20 v4 cases verbatim plus reviewed public biographies, "
            "each with at least 7 dated events and 4 domains, scored at ±10 / ±30 / ±60. "
            "Open evaluation for scoring-scale research, not a sealed blind release gate."
        ),
        "truth_hidden_from_ranker": False,
        "frozen_before_replay": False,
        "source_audit_status": "public_biography_expansion_2026_09_29",
        "ayanamsa": v4["ayanamsa"],
        "node_mode": v4["node_mode"],
        "minute_step": 2,
        "candidate_radii_minutes": list(RADII),
        "minimum_gate": {"public_aa_cases": MIN_CASES, "events_per_case": MIN_EVENTS, "domains_per_case": MIN_DOMAINS},
        "parent_dataset": str(V4_PATH.relative_to(ROOT)).replace("\\", "/"),
        "roster": str(ROSTER_PATH.relative_to(ROOT)).replace("\\", "/"),
        "events_source": str(EVENTS_PATH.relative_to(ROOT)).replace("\\", "/"),
        "birth_record_protocol": roster.get("protocol"),
        "summary": summarize(cases),
        "cases": cases,
        "boundary": (
            "Not a sealed blind holdout. Events are public biographies quoted from their source pages. "
            "Do not mix these numbers with upstream true-node runs."
        ),
    }
    return payload, problems


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--check", action="store_true", help="validate only; do not write")
    args = parser.parse_args()
    payload, problems = build(load(V4_PATH), load(ROSTER_PATH), load(EVENTS_PATH))
    for problem in problems:
        print("PROBLEM:", problem)
    if problems:
        return 1
    if not args.check:
        V5_PATH.write_text(json.dumps(payload, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
        print(f"wrote {V5_PATH.relative_to(ROOT)}")
    print(json.dumps(payload["summary"], ensure_ascii=False, indent=1))
    return 0


if __name__ == "__main__":
    sys.exit(main())