research(rectification): holdout v5 baseline runner and schema tests

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017eEAG8HD3mm8gsKXgk8uU8

scripts/research/holdout_v5_baseline.py

@@ -0,0 +1,158 @@
+#!/usr/bin/env python3
+"""Baseline scorecard for a holdout dataset (v4 subset check / v5 full set).
+
+Runs the production scoring (variant "baseline" only) through the three existing
+research harnesses without modifying them:
+
+  * minute_resolution_sweep      five metrics + six-question replay (head hit)
+  * cluster_width_probe          six-question replay under the live unionStillValidRange
+  * futile_collect_stop_replay   truth / opposite six-cell coverage and width
+
+The harnesses hard-code the v4 dataset and the 2026-09-14 report paths as module
+constants; this runner points them at the requested dataset and at its own output
+files, so the published 09-14 / 09-26 documents are never rewritten.
+
+Usage:
+  python scripts/research/holdout_v5_baseline.py --dataset v5            # full v5
+  python scripts/research/holdout_v5_baseline.py --dataset v5 --case-filter inherited   # v4 subset inside v5
+  python scripts/research/holdout_v5_baseline.py --dataset v4 --out-dir /tmp/x   # published-number check
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+import time
+import traceback
+from pathlib import Path
+from typing import Any
+
+ROOT = Path(__file__).resolve().parents[2]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts.research import cluster_width_probe as cwp  # noqa: E402
+from scripts.research import futile_collect_stop_replay as fcs  # noqa: E402
+from scripts.research import guided_collect_holdout_replay as gch  # noqa: E402
+from scripts.research import minute_resolution_sweep as mrs  # noqa: E402
+
+CAL = ROOT / "references" / "real_case_calibration"
+DATASETS = {
+    "v4": CAL / "minute_rectification_holdout_v4.json",
+    "v5": CAL / "minute_rectification_holdout_v5.json",
+}
+DEFAULT_OUT = ROOT / "docs" / "research"
+RADII = (10, 30, 60)
+
+
+def filtered_dataset(path: Path, case_filter: str | None, out_dir: Path) -> Path:
+    """Write a filtered copy of the dataset so every harness sees the same case list."""
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    cases = list(payload.get("cases") or [])
+    if case_filter == "inherited":
+        cases = [c for c in cases if c.get("inherited_from_v4")]
+    elif case_filter == "new":
+        cases = [c for c in cases if not c.get("inherited_from_v4")]
+    payload["cases"] = cases
+    # The harnesses record HOLDOUT relative to the repository root, so the working
+    # copy must live inside the repository; it is removed once the run finishes.
+    target = CAL / f"_holdout_v5_baseline_tmp_{path.stem}_{case_filter or 'all'}.json"
+    target.write_text(json.dumps(payload, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
+    return target
+
+
+def baseline_only(module: Any) -> None:
+    module.all_variants = lambda: [module.Variant("baseline")]
+
+
+def run_sweep(dataset: Path, out_dir: Path, tag: str, radii: tuple[int, ...]) -> dict[str, Any]:
+    mrs.HOLDOUT = dataset
+    mrs.REPORT_MD = out_dir / f"holdout_v5_baseline_sweep_{tag}.md"
+    mrs.REPORT_JSON = out_dir / f"holdout_v5_baseline_sweep_{tag}.json"
+    baseline_only(mrs)
+    payload = mrs.run(argparse.Namespace(limit=0, radii=[str(r) for r in radii], quick=False))
+    return {"baseline": payload["baseline"], "case_count": payload["case_count"], "errors": payload["errors"]}
+
+
+def run_cluster_width(dataset: Path, out_dir: Path, tag: str, radii: tuple[int, ...]) -> dict[str, Any]:
+    cwp.HOLDOUT = dataset
+    cwp.REPORT_MD = out_dir / f"holdout_v5_baseline_cluster_width_{tag}.md"
+    cwp.REPORT_JSON = out_dir / f"holdout_v5_baseline_cluster_width_{tag}.json"
+    baseline_only(cwp)
+    payload = cwp.run(argparse.Namespace(limit=0, radii=[str(r) for r in radii]))
+    replay = {
+        str(radius): payload["by_radius"][str(radius)]["baseline"].get("replay")
+        for radius in radii
+        if str(radius) in payload["by_radius"]
+    }
+    return {"replay": replay, "case_count": payload["case_count"], "errors": payload.get("errors", [])}
+
+
+def run_futile(dataset: Path, radii: tuple[int, ...]) -> dict[str, Any]:
+    gch.HOLDOUT = dataset
+    cases = gch.load_cases()
+    rows: list[dict[str, Any]] = []
+    for case in cases:
+        for radius in radii:
+            for direction in ("truth", "opposite"):
+                try:
+                    rows.append(fcs.evaluate_case(case, radius, direction))
+                except Exception as exc:  # noqa: BLE001
+                    rows.append({
+                        "case_id": case.get("case_id"), "radius": radius, "direction": direction,
+                        "error": f"{type(exc).__name__}: {exc}", "trace": traceback.format_exc(limit=6),
+                    })
+    summaries = [fcs.summarize(rows, radius, direction) for radius in radii for direction in ("truth", "opposite")]
+    return {
+        "summaries": summaries,
+        "errors": [row for row in rows if row.get("error")],
+        "case_count": len(cases),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--dataset", choices=sorted(DATASETS), default="v5")
+    parser.add_argument("--case-filter", choices=["inherited", "new"], default=None)
+    parser.add_argument("--out-dir", default=str(DEFAULT_OUT))
+    parser.add_argument("--radii", default=",".join(str(r) for r in RADII))
+    parser.add_argument("--skip", default="", help="comma list of harnesses to skip: sweep,cluster,futile")
+    parser.add_argument("--json-out", default=None)
+    args = parser.parse_args()
+
+    out_dir = Path(args.out_dir)
+    out_dir.mkdir(parents=True, exist_ok=True)
+    radii = tuple(int(item) for item in args.radii.split(",") if item.strip())
+    skip = {item.strip() for item in args.skip.split(",") if item.strip()}
+    tag = f"{args.dataset}_{args.case_filter or 'all'}"
+    dataset = filtered_dataset(DATASETS[args.dataset], args.case_filter, out_dir)
+
+    started = time.perf_counter()
+    result: dict[str, Any] = {
+        "dataset": str(DATASETS[args.dataset].relative_to(ROOT)).replace("\\", "/"),
+        "case_filter": args.case_filter,
+        "radii": list(radii),
+        "variant": "baseline (production scoring, no research variants)",
+        "ayanamsa": "raman",
+        "node_mode": "mean",
+    }
+    if "sweep" not in skip:
+        result["sweep"] = run_sweep(dataset, out_dir, tag, radii)
+        print("sweep done", json.dumps(result["sweep"]["baseline"]), flush=True)
+    if "cluster" not in skip:
+        result["cluster_width"] = run_cluster_width(dataset, out_dir, tag, radii)
+        print("cluster_width done", json.dumps(result["cluster_width"]["replay"]), flush=True)
+    if "futile" not in skip:
+        result["futile_collect"] = run_futile(dataset, radii)
+        print("futile done", json.dumps(result["futile_collect"]["summaries"]), flush=True)
+    result["elapsed_s"] = round(time.perf_counter() - started, 1)
+    json_out = Path(args.json_out) if args.json_out else out_dir / f"holdout_v5_baseline_{tag}_2026_09_29.json"
+    json_out.write_text(json.dumps(result, ensure_ascii=True, indent=2) + "\n", encoding="utf-8")
+    dataset.unlink(missing_ok=True)
+    print(f"wrote {json_out} elapsed={result['elapsed_s']}s")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_holdout_v5_schema.py

@@ -0,0 +1,147 @@
+"""Schema and provenance checks for the holdout v5 open evaluation set.
+
+Public Rodden-AA biographies only. The dataset is research data; these tests do not
+touch scoring and never read private material.
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from collections import Counter
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+CAL = ROOT / "references" / "real_case_calibration"
+V4 = CAL / "minute_rectification_holdout_v4.json"
+V5 = CAL / "minute_rectification_holdout_v5.json"
+ROSTER = CAL / "holdout_v5_roster.json"
+EVENTS = CAL / "minute_rectification_holdout_v5_events.json"
+
+DOMAINS = {"education", "relocation", "relationship", "career", "finance", "health_pressure", "family", "appearance", "occupation"}
+DATE_RE = {"year": re.compile(r"^\d{4}$"), "month": re.compile(r"^\d{4}-\d{2}$"), "day": re.compile(r"^\d{4}-\d{2}-\d{2}$")}
+
+
+@pytest.fixture(scope="module")
+def v5() -> dict:
+    return json.loads(V5.read_text(encoding="utf-8"))
+
+
+@pytest.fixture(scope="module")
+def v4() -> dict:
+    return json.loads(V4.read_text(encoding="utf-8"))
+
+
+def test_gate_counts(v5: dict) -> None:
+    gate = v5["minimum_gate"]
+    assert gate == {"public_aa_cases": 60, "events_per_case": 7, "domains_per_case": 4}
+    assert len(v5["cases"]) >= gate["public_aa_cases"]
+    for case in v5["cases"]:
+        events = case["events"]
+        domains = {event["domain"] for event in events}
+        assert len(events) >= gate["events_per_case"], case["case_id"]
+        assert len(domains) >= gate["domains_per_case"], case["case_id"]
+
+
+def test_calculation_profile_matches_v4(v5: dict, v4: dict) -> None:
+    assert v5["ayanamsa"] == v4["ayanamsa"] == "raman"
+    assert v5["node_mode"] == v4["node_mode"] == "mean"
+    assert v5["candidate_radii_minutes"] == [10, 30, 60]
+    assert v5["truth_hidden_from_ranker"] is False
+    assert v5["minute_step"] == 2
+
+
+def test_v4_cases_are_inherited_verbatim(v5: dict, v4: dict) -> None:
+    by_id = {case["case_id"]: case for case in v5["cases"]}
+    for original in v4["cases"]:
+        inherited = by_id[original["case_id"]]
+        assert inherited["inherited_from_v4"] is True
+        assert inherited["birth"] == original["birth"]
+        assert inherited["events"] == original["events"]
+        assert inherited["true_minute"] == original["true_minute"]
+        assert inherited["false_minute_offsets"] == original["false_minute_offsets"]
+    assert sum(1 for case in v5["cases"] if case["inherited_from_v4"]) == len(v4["cases"])
+
+
+def test_case_ids_unique_and_birth_records_complete(v5: dict) -> None:
+    ids = [case["case_id"] for case in v5["cases"]]
+    assert len(ids) == len(set(ids))
+    for case in v5["cases"]:
+        birth = case["birth"]
+        assert re.match(r"^\d{4}-\d{2}-\d{2}$", birth["date"]), case["case_id"]
+        assert re.match(r"^\d{2}:\d{2}$", birth["time"]), case["case_id"]
+        assert isinstance(birth["latitude"], (int, float)) and -90 <= birth["latitude"] <= 90
+        assert isinstance(birth["longitude"], (int, float)) and -180 <= birth["longitude"] <= 180
+        assert isinstance(birth["timezone_offset"], (int, float))
+        assert birth["source"]["rodden_rating"] == "AA", case["case_id"]
+        # v4 rows cite Astro-Databank's adbvip pages; v5 rows cite the wiki record page.
+        assert birth["source"]["url"].startswith("https://www.astro.com/"), case["case_id"]
+        assert birth["source"]["publisher"] == "Astro-Databank", case["case_id"]
+        assert case["true_minute"] == birth["time"]
+
+
+def test_new_cases_carry_checks_and_quoted_events(v5: dict) -> None:
+    new_cases = [case for case in v5["cases"] if not case["inherited_from_v4"]]
+    assert new_cases, "v5 must add cases beyond v4"
+    for case in new_cases:
+        checks = case["checks"]
+        assert checks["timezone_agrees"] is True, case["case_id"]
+        assert checks["timezone_rule"] in {"zoneinfo_match", "standard_time_kept_dst_disagreement", "lmt_from_birthplace_longitude"}
+        assert checks["place_distance_km"] is not None and checks["place_distance_km"] <= 60, case["case_id"]
+        assert case["birth"]["source"]["verified_via"]["url"].startswith("https://"), case["case_id"]
+        for event in case["events"]:
+            source = event["source"]
+            assert source["independent_of_birth_source"] is True
+            assert source["url"].startswith("https://en.wikipedia.org/wiki/")
+            assert len(source["quote"]) >= 20, event["id"]
+
+
+def test_events_domains_dates_and_precision(v5: dict) -> None:
+    for case in v5["cases"]:
+        seen: set[str] = set()
+        for event in case["events"]:
+            assert event["domain"] in DOMAINS, event["id"]
+            assert event["precision"] in DATE_RE, event["id"]
+            assert DATE_RE[event["precision"]].match(str(event["date"])), (event["id"], event["date"])
+            assert event["id"] not in seen
+            seen.add(event["id"])
+            assert int(str(event["date"])[:4]) >= int(case["birth"]["date"][:4]), event["id"]
+
+
+def test_strata_and_summary_agree(v5: dict) -> None:
+    cases = v5["cases"]
+    summary = v5["summary"]
+    assert summary["cases"] == len(cases)
+    assert summary["events_total"] == sum(len(case["events"]) for case in cases)
+    assert summary["southern_hemisphere"] == sum(1 for case in cases if case["birth"]["latitude"] < 0)
+    assert summary["night_birth_22_to_02"] == sum(1 for case in cases if case["strata"]["night_birth_22_to_02"])
+    assert summary["utc_plus_8"] == sum(1 for case in cases if abs(case["birth"]["timezone_offset"] - 8.0) < 0.01)
+    assert len(summary["decades"]) >= 3
+    assert set(summary["latitude_bands"]) == {"lt25", "25to45", "gt45"}
+    assert summary["southern_hemisphere"] >= 8
+    assert summary["night_birth_22_to_02"] >= 6
+    precision = Counter(event["precision"] for case in cases for event in case["events"])
+    assert precision == Counter(summary["events_by_precision"])
+
+
+def test_roster_and_events_files_match_dataset(v5: dict) -> None:
+    roster = json.loads(ROSTER.read_text(encoding="utf-8"))
+    events = json.loads(EVENTS.read_text(encoding="utf-8"))
+    new_ids = {case["case_id"] for case in v5["cases"] if not case["inherited_from_v4"]}
+    assert {row["case_id"] for row in roster["cases"]} == new_ids
+    assert set(events) == new_ids
+    for case in v5["cases"]:
+        if case["inherited_from_v4"]:
+            continue
+        assert sorted(events[case["case_id"]]["events"], key=lambda e: (e["date"], e["id"])) == case["events"]
+
+
+def test_build_is_deterministic(tmp_path: Path) -> None:
+    from scripts.research import holdout_v5_build as build
+
+    payload, problems = build.build(build.load(V4), build.load(ROSTER), build.load(EVENTS))
+    assert problems == []
+    rendered = json.dumps(payload, ensure_ascii=False, indent=1) + "\n"
+    assert rendered == V5.read_text(encoding="utf-8")