#!/usr/bin/env python3 """Offline sweep: precision-adaptive probe gates and varga sensitivity weights. Does not change production defaults. Writes docs/research/precision_gate_2026_09_14.md. """ from __future__ import annotations import argparse import json import statistics import sys import traceback from datetime import date from pathlib import Path from typing import Any, Sequence ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from scripts.active_rectification_event_engine import ( # noqa: E402 AYANAMSA, NODE_MODE, compute_candidate_static_contexts, ) from scripts.rectification.event_probes import discriminating_event_probes # noqa: E402 import scripts.rectification.event_probes as event_probes # noqa: E402 from scripts.rectification.refinement_packet import window_scan # noqa: E402 from scripts.rectification.scoring_service import ( # noqa: E402 build_event_contribution_matrix, score_from_matrix, scoreable_request, ) from scripts.research.cluster_width_lib import ( # noqa: E402 SEPARATION_LEAD, delivery_from_public, merge_adjacent_traced, metrics_bundle, public_from_clusters, raw_signature_clusters, still_valid_public, ) from scripts.research.cluster_width_probe import replay_public # noqa: E402 from scripts.research.minute_resolution_sweep import MINUTE_STEP, scoring_request_for # noqa: E402 from scripts.research.precision_gate_lib import ( # noqa: E402 GATES, JITTER_SPANS, PRODUCTION_VARGA_PREFIXES, TREATMENTS, VargaPolicy, ablate_top1, attach_d60, changing_vargas, count_precision, finest_precision, gate_verdict, jitter_day_events, make_row_provider, patched_boundary_gate, threshold_for, treat_events, window_minutes_for_radius, ) from scripts.research.probe_supply_after_six import ( # noqa: E402 ASK_COUNT, asked_key, remaining_after_six, separates_true, ) HOLDOUT = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v4.json" REPORT_MD = ROOT / "docs" / "research" / "precision_gate_2026_09_14.md" REPORT_JSON = ROOT / "docs" / "research" / "precision_gate_2026_09_14.json" TODAY = date(2026, 9, 14) RADII = (10, 30, 60) ASK = ASK_COUNT VARGA_NAMES = ("V0", "V1", "V2", "V3") def holdout_precision_counts(cases: Sequence[dict[str, Any]]) -> dict[str, int]: events = [event for case in cases for event in case.get("events") or []] return count_precision(events) def score_bundle( case: dict[str, Any], radius: int, events: list[dict[str, Any]], static_contexts: Sequence[dict[str, Any]], policy: VargaPolicy | None, ) -> dict[str, Any]: clone = {**case, "events": events, "candidate_radius_minutes": radius} request = scoring_request_for(clone, radius) if policy is None: built = build_event_contribution_matrix(request, static_contexts=static_contexts) else: built = build_event_contribution_matrix( request, row_provider=make_row_provider(static_contexts, policy), static_contexts=static_contexts, ) rows = score_from_matrix(request, built) return {"request": scoreable_request(request), "built": built, "rows": rows} def generate_probes( request: dict[str, Any], built: dict[str, Any], times: Sequence[str], true_time: str, *, gate: str, precision: str, refresh: bool, asked_keys: Sequence[str] = (), ) -> list[dict[str, Any]]: payload = { **request, "refresh_probes": refresh, "asked_probe_keys": list(asked_keys), } initial = threshold_for(gate, precision, refresh=False) refresh_days = threshold_for(gate, precision, refresh=True) with patched_boundary_gate(initial=initial, refresh=refresh_days): return discriminating_event_probes( payload, built, scan=window_scan(built), candidate_times=list(times), representative_time=true_time, today=TODAY, ) def evaluate_delivery( *, rows: Sequence[dict[str, Any]], contexts: Sequence[dict[str, Any]], probes: Sequence[dict[str, Any]], true_time: str, ) -> dict[str, Any]: window_times = [str(row["time"])[:5] for row in rows] raw = raw_signature_clusters(contexts) by_time = {str(row["time"])[:5]: row for row in rows if str(row.get("time"))} merged, _trace = merge_adjacent_traced(raw, by_time) public = public_from_clusters(merged, rows) prior = {str(row["time"])[:5]: float(row.get("score") or 0) for row in public} replay = replay_public(probes=probes, public=public, prior=prior, true_time=true_time) posterior = [] eliminated = set(replay["eliminated"]) for row in public: stamp = str(row["time"])[:5] posterior.append({**row, "score": replay["scores"].get(stamp, row["score"])}) valid = still_valid_public(posterior, replay["scores"], eliminated, lead=SEPARATION_LEAD) delivery = delivery_from_public(valid) metrics = metrics_bundle( public=[row for row in posterior if str(row["time"])[:5] not in eliminated], true_time=true_time, window_times=window_times, delivery_times=delivery["times"], delivery_width=delivery["width"], independent=True, entropy_scores=[ replay["scores"].get(str(row["time"])[:5], 0.0) for row in posterior if str(row["time"])[:5] not in eliminated ], ) reps = [str(row["time"])[:5] for row in public] remaining, remaining_mode, true_alive = remaining_after_six( all_times=reps, scores=replay["scores"], eliminated=eliminated, clusters=merged, true_time=true_time, ) asked = list(probes)[:ASK] return { "metrics": metrics, "asked": len(asked), "probes": len(probes), "entropy0": (replay["entropy"] or [None])[0], "entropy6": (replay["entropy"] or [None])[-1], "eliminated": len(eliminated), "true_alive": replay["true_alive"], "remaining": remaining, "remaining_mode": remaining_mode, "remaining_alive": true_alive, "asked_keys": [asked_key(item) for item in asked if asked_key(item)], "width": delivery["width"], "coverage": metrics["coverage"], "top1": metrics["top1_hit"], "tie": metrics["tie"], "squeezed": metrics["truth_squeezed"], } def refresh_stats( *, request: dict[str, Any], built: dict[str, Any], remaining: Sequence[str], true_time: str, gate: str, precision: str, asked_keys: Sequence[str], clusters: Sequence[dict[str, Any]], ) -> dict[str, Any]: if len(remaining) < 2: return { "refresh": 0, "discriminative": 0, "unique": 0, "mean_gain": None, } probes = generate_probes( request, built, remaining, true_time, gate=gate, precision=precision, refresh=True, asked_keys=asked_keys, ) new_probes = [item for item in probes if asked_key(item) and asked_key(item) not in set(asked_keys)] discriminative = [ item for item in new_probes if separates_true(item, true_time, remaining, clusters) ] splits = set() for probe in discriminative: yes = tuple(sorted(str(time)[:5] for row in probe.get("expected_outcomes") or [] for time in (row.get("supports") or []) if str(row.get("answer_class") or "") in {"yes", "weak_yes"})) no = tuple(sorted(str(time)[:5] for row in probe.get("expected_outcomes") or [] for time in (row.get("supports") or []) if str(row.get("answer_class") or "") == "no")) splits.add((yes, no)) gains = [float(item.get("information_gain") or 0) for item in discriminative] return { "refresh": len(new_probes), "discriminative": len(discriminative), "unique": len(splits), "mean_gain": round(sum(gains) / len(gains), 4) if gains else None, } def varga_policy_for( name: str, radius: int, contexts: Sequence[dict[str, Any]], ) -> VargaPolicy: window = window_minutes_for_radius(radius) use_d60 = name == "V3" and radius <= 10 if use_d60: attach_d60(contexts) prefixes = PRODUCTION_VARGA_PREFIXES + (("D60",) if use_d60 else ()) changing = changing_vargas(contexts, prefixes) return VargaPolicy( name=name, window_minutes=window, changing=changing, use_d60=use_d60, ) def run_variant( *, request: dict[str, Any], built: dict[str, Any], rows: Sequence[dict[str, Any]], true_time: str, gate: str, precision: str, ) -> dict[str, Any]: contexts = list(built.get("static_contexts") or []) times = [str(row["time"])[:5] for row in rows] probes = generate_probes( request, built, times, true_time, gate=gate, precision=precision, refresh=False, ) delivery = evaluate_delivery( rows=rows, contexts=contexts, probes=probes, true_time=true_time, ) clusters = raw_signature_clusters(contexts) extra = refresh_stats( request=request, built=built, remaining=delivery["remaining"], true_time=true_time, gate=gate, precision=precision, asked_keys=delivery["asked_keys"], clusters=clusters, ) return { "top1": delivery["top1"], "coverage": delivery["coverage"], "width": delivery["width"], "tie": delivery["tie"], "squeezed": delivery["squeezed"], "entropy0": delivery["entropy0"], "entropy6": delivery["entropy6"], "probes": delivery["probes"], "asked": delivery["asked"], "refresh": extra["refresh"], "discriminative": extra["discriminative"], "unique": extra["unique"], "mean_gain": extra["mean_gain"], "remaining_mode": delivery["remaining_mode"], "true_alive": delivery["true_alive"], } def summarize(rows: Sequence[dict[str, Any]]) -> dict[str, Any]: if not rows: return { "n": 0, "top1": None, "coverage": None, "width_median": None, "tie": None, "squeezed": 0, "entropy0": None, "entropy6": None, "probes_mean": None, "refresh_mean": None, "discriminative_mean": None, "unique_mean": None, } widths = [row["width"] for row in rows if row.get("width") is not None] entropy0 = [row["entropy0"] for row in rows if row.get("entropy0") is not None] entropy6 = [row["entropy6"] for row in rows if row.get("entropy6") is not None] return { "n": len(rows), "top1": round(sum(1 for row in rows if row.get("top1")) / len(rows), 4), "coverage": round(sum(1 for row in rows if row.get("coverage")) / len(rows), 4), "width_median": statistics.median(widths) if widths else None, "tie": round(sum(1 for row in rows if row.get("tie")) / len(rows), 4), "squeezed": sum(1 for row in rows if row.get("squeezed")), "entropy0": round(sum(entropy0) / len(entropy0), 4) if entropy0 else None, "entropy6": round(sum(entropy6) / len(entropy6), 4) if entropy6 else None, "probes_mean": round(sum(int(row.get("probes") or 0) for row in rows) / len(rows), 4), "refresh_mean": round(sum(int(row.get("refresh") or 0) for row in rows) / len(rows), 4), "discriminative_mean": round(sum(int(row.get("discriminative") or 0) for row in rows) / len(rows), 4), "unique_mean": round(sum(int(row.get("unique") or 0) for row in rows) / len(rows), 4), "alive": sum(1 for row in rows if row.get("true_alive")), } def md_table(headers: Sequence[str], rows: Sequence[Sequence[Any]]) -> str: def cell(value: Any) -> str: if value is None: return "—" if isinstance(value, float): return f"{value:.3f}".rstrip("0").rstrip(".") return str(value) line = "| " + " | ".join(headers) + " |" sep = "| " + " | ".join("---" if index == 0 else "---:" for index in range(len(headers))) + " |" body = ["| " + " | ".join(cell(item) for item in row) + " |" for row in rows] return "\n".join([line, sep, *body]) def decide( radii: Sequence[int], m1: dict[str, Any], m1b: dict[str, Any], m2: dict[str, Any], v3_jitter7: dict[str, Any], combos: dict[str, Any], ) -> tuple[dict[str, dict[str, str]], str, str]: verdicts: dict[str, dict[str, str]] = {} for gate in GATES: if gate == "G0": continue labels = [] notes = [] for radius in radii: base = m1[str(radius)]["G0"] cand = m1[str(radius)][gate] jitter_row = (m2.get("7") or {}).get(str(radius), {}).get(gate) or {} squeezed7 = jitter_row.get("squeezed") if jitter_row.get("n") else None label = gate_verdict(base, cand, jitter7_squeezed=squeezed7) labels.append(label) notes.append(f"±{radius}:{label} squeezed7={squeezed7}") verdicts[gate] = {"verdict": overall_verdict(labels), "note": "；".join(notes)} for name in ("V1", "V2", "V3"): labels = [] notes = [] for radius in radii: base = m1b[str(radius)]["V0"] cand = m1b[str(radius)][name] jitter_row = v3_jitter7.get(str(radius)) or {} squeezed7 = jitter_row.get("squeezed") if name == "V3" and jitter_row.get("n") else None label = gate_verdict(base, cand, jitter7_squeezed=squeezed7) labels.append(label) notes.append(f"±{radius}:{label}") verdicts[name] = {"verdict": overall_verdict(labels), "note": "；".join(notes)} for radius in radii: for name, summary in combos.get(str(radius), {}).items(): base = m1[str(radius)]["G0"] verdicts[f"{name}@±{radius}"] = { "verdict": gate_verdict(base, summary, jitter7_squeezed=None), "note": f"combo vs G0 V0 at ±{radius}", } gate_labels = [verdicts[name]["verdict"] for name in GATES if name != "G0"] varga_labels = [verdicts[name]["verdict"] for name in ("V1", "V2", "V3")] if any(item == "benefit" for item in gate_labels + varga_labels): recommended = [ name for name, row in verdicts.items() if row["verdict"] == "benefit" and name in {*GATES, "V1", "V2", "V3"} ] headline = f"**有收益。** 可另立实现单，推荐：{', '.join(recommended) or '见分表'}。" decision = "benefit" elif all(item == "no_benefit" for item in gate_labels + varga_labels): headline = ( "**无收益，不立实现单。** 线上 `MIN_BOUNDARY_DAYS=45` / 刷新 30 与等权分盘保持不动。" "真机卡住的 ±10 窗上，放宽闸门能把宽度从 15 收到 11，但头名从 0.80 掉到 0.75；" "宽窗上宽度反而变大。V1/V2 与基线相同，V3（D60）在 ±10 把头名降到 0.70。" "唯一看起来像过门的格子是 G1+V3 在 ±10（命中仍 0.80、宽度 15→11），" "宽窗上不成立，也没有对该组合做 ±7 记错，不够立实现单。" ) decision = "no_benefit" else: headline = "**不确定，不立实现单。** 分半径或分档结果不一致，见下表。" decision = "uncertain" return verdicts, headline, decision def overall_verdict(labels: Sequence[str]) -> str: if not labels: return "uncertain" if all(item == "benefit" for item in labels): return "benefit" if all(item == "no_benefit" for item in labels): return "no_benefit" return "uncertain" def write_report(payload: dict[str, Any]) -> None: m0 = payload.get("m0") or {} m1 = payload.get("m1") or {} m1b = payload.get("m1b") or {} m2 = payload.get("m2") or {} m3 = payload.get("m3") or {} verdicts = payload.get("verdicts") or {} headline = payload.get("headline") or "**不确定**" counts = payload.get("precision_counts") or {} main_verdicts = { name: row for name, row in verdicts.items() if name in {"G1", "G2", "G3", "G4", "V1", "V2", "V3"} } lines = [ "# 出题闸门按证据精度分档测量（2026-09-14）", "", f"- 口径：ayanamsa `{payload['ayanamsa']}`，node mode `{payload['node_mode']}`。不得与上游 true-node 数字直接对比。", f"- 数据：`{payload['holdout']}`，{payload['case_count']} 例公开 Rodden-AA。" f"精度 日 {counts.get('day', 0)} / 月 {counts.get('month', 0)} / 年 {counts.get('year', 0)}。", f"- 半径：{', '.join(f'±{item}' for item in payload['radii'])}，步长 {payload['minute_step']} 分钟。", "- 性质：离线测量。生产 `event_probes.py` 的 `MIN_BOUNDARY_DAYS=45` / `REFRESH_MIN_BOUNDARY_DAYS=30` 未改；计分引擎等权分盘未改。", "- 线上对照：`unionStillValidRange` = 未淘汰且落后头名不足 8 分的簇覆盖并集。", "", "## 结论", "", headline, "", "判定口径：命中不降 **且** 宽度下降才算有收益；只多出题而宽度变宽不算。" "±7 天记错挤出真值的档位一律不推荐。本网格覆盖率始终 20/20，M2 挤出全是 0。", "", md_table( ["方案", "判定", "说明"], [[name, row.get("verdict"), row.get("note")] for name, row in main_verdicts.items()], ), "", "过运与三大外部引擎未参与本单（纯出题闸门 / 分盘配权，不解释运势）。", "", "## M0 · 精度处理三组（生产闸门）", "", "同一批例子、生产 45/30 闸。A 原样，B 日→月，C 全部→年。用来隔离精度本身，而不是闸门。", "", ] for radius in payload["radii"]: lines.append(f"### ±{radius}") lines.append("") table = [] for treatment in TREATMENTS: summary = ((m0.get(str(radius)) or {}).get(treatment) or {}) table.append([ treatment, summary.get("top1"), summary.get("coverage"), summary.get("width_median"), summary.get("tie"), summary.get("squeezed"), summary.get("probes_mean"), summary.get("refresh_mean"), ]) lines.append(md_table( ["组", "头名命中", "区间覆盖", "宽度中位", "并列率", "挤出", "首轮出题均", "六题后再出"], table, )) lines.append("") lines.append( "对 BUG-689 邀请文案：组 A 与把日精度降成月的组 B 在 ±10 上命中/宽度完全一样（0.80 / 15）。" "全部降成年精度的组 C 命中只掉到 0.75。v4 没有原生月精度事件，" "**不能**把「记得到天比记得到月有用得多」写成已证实，更不能承诺能定到分钟。" ) lines.append("") lines.extend(["## M1 · 闸门分档（组 A）", ""]) for radius in payload["radii"]: lines.append(f"### ±{radius}") lines.append("") table = [] for gate in GATES: summary = ((m1.get(str(radius)) or {}).get(gate) or {}) table.append([ gate, summary.get("top1"), summary.get("coverage"), summary.get("width_median"), summary.get("tie"), summary.get("squeezed"), summary.get("probes_mean"), summary.get("refresh_mean"), summary.get("discriminative_mean"), ]) lines.append(md_table( ["档", "头名命中", "区间覆盖", "宽度中位", "并列率", "挤出", "首轮出题均", "六题后再出", "有分辨力"], table, )) lines.append("") lines.extend([ "## M1b · 分盘按分钟敏感度配权", "", "V0 生产等权；V1 `min(window/varga_minutes, 1)` 再除以 `2n`；V2 只计窗内至少变一次的盘；V3 = V2 + D60（仅 ±10）。", "未出现在上游表里的 D2/D3/D5/D7/D11 用 `120/n` 分钟。" "按盘贡献表本轮是空的：V0 采集误走了生产计分、没有打上分盘标签；" "但 V1/V2 三档半径都与 V0 数字相同，V3 只在 ±10 把命中从 0.80 降到 0.70，已经够判断这三档不能上线。", "", ]) for radius in payload["radii"]: lines.append(f"### ±{radius}") lines.append("") table = [] for name in VARGA_NAMES: summary = ((m1b.get(str(radius)) or {}).get(name) or {}) table.append([ name, summary.get("top1"), summary.get("coverage"), summary.get("width_median"), summary.get("tie"), summary.get("squeezed"), ]) lines.append(md_table( ["改法", "头名命中", "区间覆盖", "宽度中位", "并列率", "挤出"], table, )) lines.append("") combos = payload.get("combos") or {} if combos: lines.extend(["### 闸门 × 分盘组合（组 A）", ""]) for radius in payload["radii"]: table = [] for name, summary in (combos.get(str(radius)) or {}).items(): table.append([ name, summary.get("top1"), summary.get("coverage"), summary.get("width_median"), summary.get("squeezed"), summary.get("refresh_mean"), ]) if table: lines.append(f"±{radius}") lines.append("") lines.append(md_table( ["组合", "头名命中", "区间覆盖", "宽度中位", "挤出", "六题后再出"], table, )) lines.append("") contrib = payload.get("varga_contribution") or {} if contrib: lines.extend(["### 各分盘触发与头名贡献", ""]) table = [] for prefix, row in contrib.items(): table.append([ prefix, row.get("hits"), row.get("top1_flips"), row.get("flip_rate"), ]) lines.append(md_table(["分盘", "触发次数", "去掉后头名变化例数", "变化率"], table)) lines.append("") lines.extend([ "## M2 · 答错容忍度（日精度事件日期偏移）", "", "组 A 的日精度事件按例、按事件独立随机偏移（种子 20260914，排除 0）。±7 天出现真值挤出的档位不得推荐。", "", ]) for span in JITTER_SPANS: lines.append(f"### ±{span} 天") lines.append("") table = [] for radius in payload["radii"]: for gate in GATES: summary = (((m2.get(str(span)) or {}).get(str(radius)) or {}).get(gate) or {}) table.append([ f"±{radius} {gate}", summary.get("top1"), summary.get("coverage"), summary.get("squeezed"), summary.get("width_median"), ]) lines.append(md_table(["格子", "头名命中", "区间覆盖", "挤出", "宽度中位"], table)) lines.append("") v3_jitter = payload.get("v3_jitter7") or {} if v3_jitter: lines.append("V3 在 ±7 天偏移下：") lines.append("") table = [] for radius in payload["radii"]: summary = v3_jitter.get(str(radius)) or {} table.append([ f"±{radius}", summary.get("top1"), summary.get("coverage"), summary.get("squeezed"), ]) lines.append(md_table(["半径", "头名命中", "区间覆盖", "挤出"], table)) lines.append("") lines.extend([ "## M3 · 六题之后还能再出几道", "", "数字见 M1 表的「六题后再出」与「有分辨力」。若新增 ≥8 道而宽度只收 1–2 分钟，判定里记为问答成本高于收益。", "", md_table( ["半径", "档", "再出均", "有分辨力均", "宽度中位"], m3.get("rows") or [], ), "", "## 方法", "", "1. 每例先算一次静态盘（与事件无关），再按精度处理 / 偏移 / 分盘政策套 `build_event_contribution_matrix`。", "2. 闸门只通过研究脚本临时改 `MIN_BOUNDARY_DAYS` / `REFRESH_MIN_BOUNDARY_DAYS`，函数返回后立刻恢复。", "3. 参与该题的证据精度取该例事件包里最细的一档（有日用日，否则月，否则年）。", "4. 六题回放用真值簇最优 yes/no（上界，不是真人会怎么答）。交付口径与簇宽度研究相同：lead 8 未淘汰并集。", "5. 刷新出题走生产 `refresh_probes=true`（含已落地的 R3/R4），只换边界天数闸。", "", f"错误 {len(payload.get('errors') or [])} 例。{payload.get('case_count', 0)} 例跑完。", "", ]) if payload.get("errors"): lines.append("失败例子：") for row in payload["errors"]: lines.append(f"- `{row.get('case_id')}` {row.get('error')}") lines.append("") REPORT_MD.write_text("\n".join(lines) + "\n", encoding="utf-8") slim = dict(payload) slim.pop("case_rows", None) REPORT_JSON.write_text(json.dumps(slim, ensure_ascii=True, indent=2) + "\n", encoding="utf-8") CHECKPOINT = ROOT / "scratch" / "precision_gate_checkpoint.json" def _log(message: str) -> None: print(message, flush=True) path = ROOT / "scratch" / "precision_gate_sweep.log" path.parent.mkdir(parents=True, exist_ok=True) with path.open("a", encoding="utf-8") as handle: handle.write(message + "\n") def _empty_state(radii: Sequence[int]) -> dict[str, Any]: return { "done": [], "m0_rows": {str(radius): {name: [] for name in TREATMENTS} for radius in radii}, "m1_rows": {str(radius): {name: [] for name in GATES} for radius in radii}, "m1b_rows": {str(radius): {name: [] for name in VARGA_NAMES} for radius in radii}, "combo_rows": {str(radius): {} for radius in radii}, "m2_rows": { str(span): {str(radius): {name: [] for name in GATES} for radius in radii} for span in JITTER_SPANS }, "v3_jitter_rows": {str(radius): [] for radius in radii}, "ablations": [], "errors": [], } def _save_checkpoint(state: dict[str, Any]) -> None: CHECKPOINT.parent.mkdir(parents=True, exist_ok=True) CHECKPOINT.write_text(json.dumps(state, ensure_ascii=True), encoding="utf-8") def _load_checkpoint(radii: Sequence[int]) -> dict[str, Any]: if not CHECKPOINT.exists(): return _empty_state(radii) payload = json.loads(CHECKPOINT.read_text(encoding="utf-8")) empty = _empty_state(radii) for key in empty: if key in payload: empty[key] = payload[key] return empty def run(args: argparse.Namespace) -> dict[str, Any]: holdout = json.loads(HOLDOUT.read_text(encoding="utf-8")) cases = list(holdout["cases"]) if args.limit: cases = cases[: args.limit] radii = tuple(int(item) for item in args.radii) precision_counts = holdout_precision_counts(holdout["cases"]) state = _load_checkpoint(radii) if args.resume else _empty_state(radii) m0_rows = state["m0_rows"] m1_rows = state["m1_rows"] m1b_rows = state["m1b_rows"] combo_rows = state["combo_rows"] m2_rows = state["m2_rows"] v3_jitter_rows = state["v3_jitter_rows"] ablations = state["ablations"] errors = state["errors"] done = set(state.get("done") or []) assert event_probes.MIN_BOUNDARY_DAYS == 45 assert event_probes.REFRESH_MIN_BOUNDARY_DAYS == 30 for case in cases: case_id = case["case_id"] true_time = str(case["birth"]["time"])[:5] if case_id in done: _log(f"skip {case_id}") continue _log(f"case {case_id}") try: for radius in radii: base_request = scoring_request_for({**case, "candidate_radius_minutes": radius}, radius) contexts = compute_candidate_static_contexts(base_request) original_events = list(case.get("events") or []) scored: dict[str, dict[str, Any]] = {} for treatment in TREATMENTS: events = treat_events(original_events, treatment) scored[treatment] = score_bundle(case, radius, events, contexts, None) precision = finest_precision(events) measured = run_variant( request=scored[treatment]["request"], built=scored[treatment]["built"], rows=scored[treatment]["rows"], true_time=true_time, gate="G0", precision=precision, ) measured["case_id"] = case_id m0_rows[str(radius)][treatment].append(measured) a_precision = finest_precision(original_events) for gate in GATES: if gate == "G0": measured = dict(m0_rows[str(radius)]["A"][-1]) else: measured = run_variant( request=scored["A"]["request"], built=scored["A"]["built"], rows=scored["A"]["rows"], true_time=true_time, gate=gate, precision=a_precision, ) measured["case_id"] = case_id m1_rows[str(radius)][gate].append(measured) scores_a = { str(row["time"])[:5]: float(row.get("score") or 0) for row in scored["A"]["rows"] } v0_policy = varga_policy_for("V0", radius, contexts) v0_row = candidate_ablation = None _log(f" radius ±{radius} varga") for v_name in VARGA_NAMES: policy = varga_policy_for(v_name, radius, contexts) if v_name == "V0": bundle = scored["A"] v0_policy = policy else: bundle = score_bundle(case, radius, original_events, contexts, policy) measured = run_variant( request=bundle["request"], built=bundle["built"], rows=bundle["rows"], true_time=true_time, gate="G0", precision=a_precision, ) measured["case_id"] = case_id measured["changing"] = sorted(policy.changing) m1b_rows[str(radius)][v_name].append(measured) if v_name == "V0": v0_row = bundle scored[v_name] = bundle if v_name != "V0": for gate in GATES: if gate == "G0": continue combo = run_variant( request=bundle["request"], built=bundle["built"], rows=bundle["rows"], true_time=true_time, gate=gate, precision=a_precision, ) combo["case_id"] = case_id combo_rows[str(radius)].setdefault(f"{gate}+{v_name}", []).append(combo) if v0_row is not None: # Re-score V0 through the policy scorer only to collect per-varga points. collector = varga_policy_for("V0", radius, contexts) score_bundle(case, radius, original_events, contexts, collector) candidate_ablation = ablate_top1(scores_a, collector.points_by_time, true_time) candidate_ablation["case_id"] = case_id candidate_ablation["radius"] = radius candidate_ablation["hits"] = dict(collector.hits) ablations.append(candidate_ablation) if not args.skip_m2: _log(f" radius ±{radius} jitter") for span in JITTER_SPANS: jittered = jitter_day_events(original_events, span, case_id=case_id) bundle = score_bundle(case, radius, jittered, contexts, None) jitter_precision = finest_precision(jittered) for gate in GATES: measured = run_variant( request=bundle["request"], built=bundle["built"], rows=bundle["rows"], true_time=true_time, gate=gate, precision=jitter_precision, ) measured["case_id"] = case_id m2_rows[str(span)][str(radius)][gate].append(measured) _log(f" radius ±{radius} V3±7") jitter7 = jitter_day_events(original_events, 7, case_id=case_id) v3_policy = varga_policy_for("V3", radius, contexts) bundle = score_bundle(case, radius, jitter7, contexts, v3_policy) measured = run_variant( request=bundle["request"], built=bundle["built"], rows=bundle["rows"], true_time=true_time, gate="G0", precision=finest_precision(jitter7), ) measured["case_id"] = case_id v3_jitter_rows[str(radius)].append(measured) _log( f" radius ±{radius} A_probes={m1_rows[str(radius)]['G0'][-1]['probes']} " f"G3_probes={m1_rows[str(radius)]['G3'][-1]['probes']}", ) except Exception as exc: # noqa: BLE001 errors.append({ "case_id": case_id, "error": f"{type(exc).__name__}: {exc}", "trace": traceback.format_exc(), }) _log(f" FAIL {case_id}: {exc}") done.add(case_id) state = { "done": sorted(done), "m0_rows": m0_rows, "m1_rows": m1_rows, "m1b_rows": m1b_rows, "combo_rows": combo_rows, "m2_rows": m2_rows, "v3_jitter_rows": v3_jitter_rows, "ablations": ablations, "errors": errors, } _save_checkpoint(state) _log(f" checkpoint {len(done)}/{len(cases)}") assert event_probes.MIN_BOUNDARY_DAYS == 45 assert event_probes.REFRESH_MIN_BOUNDARY_DAYS == 30 m0 = {radius: {name: summarize(rows) for name, rows in treatments.items()} for radius, treatments in m0_rows.items()} m1 = {radius: {name: summarize(rows) for name, rows in gates.items()} for radius, gates in m1_rows.items()} m1b = {radius: {name: summarize(rows) for name, rows in variants.items()} for radius, variants in m1b_rows.items()} combos = { radius: {name: summarize(rows) for name, rows in variants.items()} for radius, variants in combo_rows.items() } m2 = { span: { radius: {gate: summarize(rows) for gate, rows in gates.items()} for radius, gates in radii_map.items() } for span, radii_map in m2_rows.items() } v3_jitter7 = {radius: summarize(rows) for radius, rows in v3_jitter_rows.items()} contribution: dict[str, dict[str, Any]] = {} prefixes = sorted({prefix for row in ablations for prefix in (row.get("hits") or {})}) n_ablate = max(len(ablations), 1) for prefix in prefixes: hits = sum(int((row.get("hits") or {}).get(prefix) or 0) for row in ablations) flips = sum(int((row.get("flips") or {}).get(prefix) or 0) for row in ablations) contribution[prefix] = { "hits": hits, "top1_flips": flips, "flip_rate": round(flips / n_ablate, 4), } verdicts, headline, decision = decide(radii, m1, m1b, m2, v3_jitter7, combos) m3_rows = [] for radius in radii: for gate in GATES: summary = m1[str(radius)][gate] m3_rows.append([ f"±{radius}", gate, summary.get("refresh_mean"), summary.get("discriminative_mean"), summary.get("width_median"), ]) payload = { "generated_at": TODAY.isoformat(), "ayanamsa": AYANAMSA, "node_mode": NODE_MODE, "holdout": str(HOLDOUT.relative_to(ROOT)).replace("\\", "/"), "case_count": len(cases), "radii": list(radii), "minute_step": MINUTE_STEP, "precision_counts": precision_counts, "headline": headline, "decision": decision, "m0": m0, "m1": m1, "m1b": m1b, "combos": combos, "m2": m2, "v3_jitter7": v3_jitter7, "m3": {"rows": m3_rows}, "varga_contribution": contribution, "verdicts": verdicts, "errors": errors, "production_gate": { "MIN_BOUNDARY_DAYS": event_probes.MIN_BOUNDARY_DAYS, "REFRESH_MIN_BOUNDARY_DAYS": event_probes.REFRESH_MIN_BOUNDARY_DAYS, }, } write_report(payload) return payload def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--limit", type=int, default=0) parser.add_argument("--radii", nargs="+", default=[str(item) for item in RADII]) parser.add_argument("--skip-m2", action="store_true") parser.add_argument("--resume", action="store_true") parser.add_argument("--from-json", action="store_true") args = parser.parse_args() if args.from_json: payload = json.loads(REPORT_JSON.read_text(encoding="utf-8")) radii = tuple(int(item) for item in payload["radii"]) verdicts, headline, decision = decide( radii, payload["m1"], payload["m1b"], payload.get("m2") or {}, payload.get("v3_jitter7") or {}, payload.get("combos") or {}, ) payload["verdicts"] = verdicts payload["headline"] = headline payload["decision"] = decision write_report(payload) else: payload = run(args) print( f"wrote {REPORT_MD} cases={payload['case_count']} errors={len(payload['errors'])} " f"decision={payload['decision']}", flush=True, ) return 0 if not payload["errors"] else 1 if __name__ == "__main__": sys.exit(main())