feat: sync blind timing holdout gate

scripts/day_level_holdout_template.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python3
+"""Create a blank independent day-level timing holdout annotation template."""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+
+TEMPLATE = {
+    "case_id": "",
+    "subject": {
+        "name": "",
+        "public_profile_url": "",
+        "birth_time_rating": "AA/A only preferred",
+    },
+    "domain": "career|marriage|wealth|health",
+    "label": "target_event|no_target_event",
+    "start": "YYYY-MM-DD",
+    "end": "YYYY-MM-DD",
+    "event_description": "",
+    "event_absent_assertion": "",
+    "source_url": "https://",
+    "source_quote_or_summary": "",
+    "adjudicator": "",
+    "time_uncertainty_days": 0,
+    "independent_human_reviewed": True,
+    "frozen_before_scoring": True,
+    "source_path": "",
+    "notes": "",
+}
+
+
+def build_template() -> dict:
+    return {
+        "template_type": "day_level_holdout_annotation_v3",
+        "instructions": [
+            "Use target_event for known dated events.",
+            "Use no_target_event only when a public source supports that the target event did not occur in the interval.",
+            "Do not use old control dates or rows observed before preregistration for tuning.",
+            "Freeze labels before running timing_ranker_blind_eval.py.",
+        ],
+        "annotation": TEMPLATE,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--output", type=Path)
+    args = parser.parse_args()
+    text = json.dumps(build_template(), ensure_ascii=False, indent=2) + "\n"
+    if args.output:
+        args.output.write_text(text, encoding="utf-8")
+    else:
+        print(text, end="")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/day_level_holdout_validator.py

@@ -11,11 +11,14 @@ REQUIRED={"case_id","domain","label","start","end","source_url","adjudicator","t
 def validate(path: Path) -> dict:
     data=json.loads(path.read_text(encoding="utf-8")); rows=data.get("annotations") or []; errors=[]
     mode=data.get("validation_mode", "independent")
+    prohibited=set(data.get("prohibited_tuning_data") or [])
     allowed_labels={"target_event", "no_target_event"} if mode == "independent" else {"target_event", "observational_non_target_date"}
     for i,row in enumerate(rows):
         for key in sorted(REQUIRED-set(row)): errors.append({"row":i,"field":key,"error":"missing"})
         if row.get("label") not in allowed_labels: errors.append({"row":i,"field":"label","error":"invalid"})
         if not str(row.get("source_url") or "").startswith(("https://","http://")): errors.append({"row":i,"field":"source_url","error":"not_public_url"})
+        if mode == "independent" and row.get("independent_human_reviewed") is not True: errors.append({"row":i,"field":"independent_human_reviewed","error":"not_independently_human_reviewed"})
+        if row.get("source_path") in prohibited: errors.append({"row":i,"field":"source_path","error":"prohibited_tuning_source"})
     positives=sum(r.get("label")=="target_event" for r in rows); negatives=sum(r.get("label") in {"no_target_event", "observational_non_target_date"} for r in rows)
     gate=data.get("frozen_gate") or {}; ready=not errors and positives>=gate.get("minimum_independent_cases",20) and negatives>=gate.get("minimum_independent_negative_intervals",80)
     status=("observational_ready_not_independent" if ready else "awaiting_observational_labels") if mode == "observational" else ("ready_for_blind_replay" if ready else "awaiting_independent_labels")

scripts/timing_ranker_blind_eval.py

@@ -0,0 +1,71 @@
+#!/usr/bin/env python3
+"""Evaluate frozen day/month timing candidates against independent holdout labels."""
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+
+from scripts.day_level_holdout_validator import validate
+
+
+def _load(path: Path) -> dict:
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def _key(row: dict) -> tuple[str, str, str]:
+    return (str(row.get("case_id") or ""), str(row.get("start") or ""), str(row.get("end") or ""))
+
+
+def evaluate(manifest_path: Path, candidates_path: Path) -> dict:
+    validation = validate(manifest_path)
+    manifest = _load(manifest_path)
+    candidates = sorted(
+        (_load(candidates_path).get("candidate_windows") or []),
+        key=lambda row: float(row.get("score") or 0),
+        reverse=True,
+    )
+    labels = {_key(row): row.get("label") for row in manifest.get("annotations") or []}
+    ranked = [{**row, "rank": index + 1, "label": labels.get(_key(row))} for index, row in enumerate(candidates)]
+    positives = [row for row in ranked if row.get("label") == "target_event"]
+    negatives = [row for row in ranked if row.get("label") == "no_target_event"]
+    top_3_positive = sum(1 for row in positives if row["rank"] <= 3)
+    positive_top_3_rate = top_3_positive / len(positives) if positives else 0.0
+    min_positive_score = min((float(row.get("score") or 0) for row in positives), default=0.0)
+    false_positive_negatives = sum(1 for row in negatives if float(row.get("score") or 0) >= min_positive_score)
+    specificity = 1 - (false_positive_negatives / len(negatives)) if negatives else 0.0
+    gate = manifest.get("frozen_gate") or {}
+    blockers = []
+    if validation["status"] != "ready_for_blind_replay":
+        blockers.append("holdout_not_ready")
+    if positive_top_3_rate < gate.get("minimum_positive_top_3_rate", 1):
+        blockers.append("positive_top_3_rate_below_gate")
+    if specificity < gate.get("minimum_specificity", 1):
+        blockers.append("specificity_below_gate")
+    passed = not blockers
+    return {
+        "scope": "timing_ranker_blind_eval",
+        "status": "pass" if passed else "blocked",
+        "claim_status": "calibrated_day_level" if passed else "exploratory_unvalidated",
+        "production_tuning_allowed": bool(passed),
+        "positive_count": len(positives),
+        "negative_count": len(negatives),
+        "positive_top_3_rate": positive_top_3_rate,
+        "specificity": specificity,
+        "blockers": blockers,
+        "validation": validation,
+        "ranked_windows": ranked,
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("manifest", type=Path)
+    parser.add_argument("candidates", type=Path)
+    args = parser.parse_args()
+    print(json.dumps(evaluate(args.manifest, args.candidates), ensure_ascii=False, indent=2, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())