#!/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 keep_published_reports_untouched() -> None: """cluster_width_probe.run rewrites the errata table inside the 09-14 sweep report; never do that here.""" cwp.append_previous_errata = lambda *args, **kwargs: None 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) keep_published_reports_untouched() 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())