From 57118b2697461bef6919310756db7f70137231bc Mon Sep 17 00:00:00 2001 From: Jesse_Chen Date: Tue, 29 Sep 2026 19:14:08 +0800 Subject: [PATCH] research(rectification): holdout v5 baseline runner and schema tests Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_017eEAG8HD3mm8gsKXgk8uU8 --- scripts/research/holdout_v5_baseline.py | 158 ++++++++++++++++++++++++ tests/test_holdout_v5_schema.py | 147 ++++++++++++++++++++++ 2 files changed, 305 insertions(+) create mode 100644 scripts/research/holdout_v5_baseline.py create mode 100644 tests/test_holdout_v5_schema.py diff --git a/scripts/research/holdout_v5_baseline.py b/scripts/research/holdout_v5_baseline.py new file mode 100644 index 00000000..502ef84c --- /dev/null +++ b/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()) diff --git a/tests/test_holdout_v5_schema.py b/tests/test_holdout_v5_schema.py new file mode 100644 index 00000000..7725d93c --- /dev/null +++ b/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")