#!/usr/bin/env python3 """Reported-centre sensitivity on exposed public cases, without oracle answers.""" from __future__ import annotations import argparse import json import sys from datetime import datetime, timedelta, timezone 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.active_rectification_event_engine import ( AYANAMSA, NODE_MODE, compute_candidate_static_contexts, ) from scripts.minute_rectification_blind_eval import _opaque_winner, implementation_sha256 from scripts.minute_rectification_holdout_validator import validate from scripts.rectification.candidate_contrast import select_signature_representatives from scripts.rectification.scoring_service import build_event_contribution_matrix, score_from_matrix from scripts.research.cluster_width_lib import SEPARATION_LEAD, still_valid_public from scripts.research.probe_supply_after_six import request_from_case from scripts.research.sealed_holdout_rerun import ( DATASET, file_sha256, historical_comparison, implementation_identity, opaque_order, verify_frozen_record, ) OFFSETS = (-30, -20, -15, -10, -8, -5, -3, 0, 3, 5, 8, 10, 15, 20, 30) RADII = (15, 30, 60) MINUTE_STEP = 1 FREEZE = ROOT / "docs/research/reported_offset_cross_midnight_2026_09_20.final.freeze.json" REPORT = ROOT / "docs/research/reported_offset_cross_midnight_2026_09_20.json" LEGACY_REPORT = ROOT / "docs/research/reported_offset_2026_09_20.json" def shifted_window(case: dict[str, Any], offset: int, radius: int) -> tuple[dict[str, Any], list[datetime]]: if radius not in (*RADII, 120): raise ValueError("unsupported_radius") true_at = datetime.fromisoformat(f"{case['birth']['date']}T{case['birth']['time']}:00") reported_at = true_at + timedelta(minutes=offset) start = reported_at - timedelta(minutes=radius) end = reported_at + timedelta(minutes=radius) candidates = [start + timedelta(minutes=i) for i in range(0, radius * 2 + 1, MINUTE_STEP)] # Reuse only the established event normalization, then replace the window. # The ranker sees neither the truth label nor the simulated offset. request = request_from_case(case) request.update({ "birth_date": start.date().isoformat(), "start_time": start.strftime("%H:%M"), "end_time": end.strftime("%H:%M"), "minute_step": MINUTE_STEP, "ayanamsa": AYANAMSA, "node_mode": NODE_MODE, }) return request, candidates def score_window(request: dict[str, Any], contexts: list[dict[str, Any]]) -> list[dict[str, Any]]: """Use the repaired native single matrix, with candidate dates preserved. The historical grouped adapter is archived, not a second active scoring path. This is a native scoring replay, still not a production Q&A replay. """ built = build_event_contribution_matrix(request, static_contexts=contexts) return score_from_matrix(request, built) def delivery_moments(public: list[dict[str, Any]], candidates: list[datetime]) -> list[datetime]: """Initial delivery envelope in date-aware order; no truth-based narrowing.""" valid = still_valid_public(public, {}, lead=SEPARATION_LEAD) clocks = {str(value)[:5] for row in valid for value in (row.get("cluster_times") or [row["time"]])} return [candidate for candidate in candidates if candidate.strftime("%H:%M") in clocks] def reveal_metrics( rows: list[dict[str, Any]], candidates: list[datetime], delivery: list[datetime], truth: datetime, benchmark_id: str, case_id: str, ) -> dict[str, Any]: ordered = opaque_order(benchmark_id, case_id, rows) predicted_clock = _opaque_winner(benchmark_id, case_id, rows) by_clock = {candidate.strftime("%H:%M"): candidate for candidate in candidates} if len(by_clock) != len(candidates): raise ValueError("ambiguous_candidate_clock") predicted = by_clock[predicted_clock] within = min(candidates) <= truth <= max(candidates) in_candidates = truth in candidates rank = next((i for i, row in enumerate(ordered, 1) if by_clock[row["time"]] == truth), None) covered = bool(delivery) and min(delivery) <= truth <= max(delivery) return { "truth_in_window": within, "truth_in_candidates": in_candidates, "true_rank": rank, "top_1_hit": predicted == truth, "top_1_minute_error": abs((predicted - truth).total_seconds()) / 60, "delivery_covers_truth": covered, "candidate_count": len(candidates), "delivery_width_minutes": int((max(delivery) - min(delivery)).total_seconds() / 60) + 1 if delivery else 0, } def summarize(trials: list[dict[str, Any]], radii: tuple[int, ...], offsets: tuple[int, ...]) -> list[dict[str, Any]]: result = [] for radius in radii: for offset in offsets: group = [row for row in trials if row["radius_minutes"] == radius and row["offset_minutes"] == offset] count = len(group) result.append({ "radius_minutes": radius, "offset_minutes": offset, "trial_count": count, **{key: round(sum(row[source] for row in group) / count, 4) if count else None for key, source in ( ("truth_in_window_rate", "truth_in_window"), ("top_1_rate", "top_1_hit"), ("delivery_coverage_rate", "delivery_covers_truth"), ("mean_absolute_minute_error", "top_1_minute_error"), )}, }) return result def freeze_record(dataset: Path = DATASET, radii: tuple[int, ...] = RADII, offsets: tuple[int, ...] = OFFSETS) -> dict[str, Any]: manifest = json.loads(dataset.read_text(encoding="utf-8")) legacy_hash = implementation_sha256(manifest["frozen_scoring"]["files"]) return { "record_version": "reported-offset-native-candidate-date-v3", "frozen_at_utc": datetime.now(timezone.utc).isoformat(), "ayanamsa": AYANAMSA, "node_mode": NODE_MODE, "radii_minutes": list(radii), "offsets_minutes": list(offsets), "minute_step": MINUTE_STEP, "dataset": dataset.relative_to(ROOT).as_posix(), "dataset_sha256": file_sha256(dataset), "dataset_benchmark_id": manifest["benchmark_id"], "implementation_sha256": legacy_hash, "implementation_sha256_prefix": legacy_hash[:16], **implementation_identity(dataset), "scorer": "native_event_contribution_matrix_not_shadow_fact_ranker", "delivery": "initial_signature_clusters_peak_gap_lt_8_envelope_no_answers_no_elimination", "rank": "score_desc_then_sha256(benchmark_id:case_id:candidate_time)", "grid": "every_minute_inclusive_not_production_two_minute_sampling", "cross_midnight": "date_aware_candidates_distance_and_native_single_matrix", "evaluator_sha256": file_sha256(Path(__file__)), "replay_revision": "native_candidate_date_v3", "supersedes": "candidate_date_grouped_v2_identity_and_path_not_assumed_numerically_wrong", "is_blind_evaluation": False, "truth_hidden_from_ranker": True, "official_valid_independent_blind": False, "official_blind_trial_count": 0, "results_previously_seen": True, "must_not_use_for_tuning": True, } def run(dataset: Path = DATASET, radii: tuple[int, ...] = RADII, offsets: tuple[int, ...] = OFFSETS, freeze_path: Path = FREEZE) -> dict[str, Any]: frozen = json.loads(freeze_path.read_text(encoding="utf-8")) verify_frozen_record(frozen, freeze_record(dataset, radii, offsets)) replay_started_at = datetime.now(timezone.utc).isoformat() manifest = json.loads(dataset.read_text(encoding="utf-8")) validation = validate(dataset) invalid = validation["invalid_cases"] trials = [] for index, case in enumerate(manifest["cases"], 1): if case["case_id"] in invalid: continue # Reuse static chart calculations, not window-dependent scores or ranks. contexts_by_moment: dict[datetime, dict[str, Any]] = {} for radius in radii: for offset in offsets: request, candidates = shifted_window(case, offset, radius) missing = [moment for moment in candidates if moment not in contexts_by_moment] if missing: contexts_by_moment.update(zip(missing, compute_candidate_static_contexts(request, candidates=missing), strict=True)) contexts = [contexts_by_moment[moment] for moment in candidates] rows = score_window(request, contexts) public = select_signature_representatives(rows, contexts) delivery = delivery_moments(public, candidates) # All scores and the delivery envelope are locked before reveal. truth = datetime.fromisoformat(f"{case['birth']['date']}T{case['birth']['time']}:00") trials.append({ "case_ordinal": index, "offset_minutes": offset, "radius_minutes": radius, **reveal_metrics(rows, candidates, delivery, truth, manifest["benchmark_id"], case["case_id"]), }) verify_frozen_record(frozen, freeze_record(dataset, radii, offsets)) return { "scope": "reported_offset_sensitivity_not_product_accuracy", "replay_started_at_utc": replay_started_at, "replay_finished_at_utc": datetime.now(timezone.utc).isoformat(), "specification": frozen, "frozen_record": frozen, "freeze_record_path": freeze_path.relative_to(ROOT).as_posix(), "implementation_hash_matches_at_replay": True, "dataset_hash_matches_at_replay": True, "historical_comparison": historical_comparison(LEGACY_REPORT, trials, ("case_ordinal", "radius_minutes", "offset_minutes")), "excluded_cases": invalid, "case_count": validation["valid_public_aa_cases"], "trial_count": len(trials), "trials": trials, "summary": summarize(trials, radii, offsets), "widening_geometry": { "scored_radii": list(radii), "radius_120_scored": 120 in radii, "truth_in_window_condition": "abs(reported_offset_minutes) <= radius_minutes", "radius_15_first_integer_minute_outside": 16, "all_tested_offsets_within_radius_30_60_120": max(map(abs, offsets)) <= 30, "boundary": "Geometry rescues candidate inclusion only, not ranking or delivery coverage; no population frequency without a reported-offset distribution.", }, } if __name__ == "__main__": parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--json", action="store_true") parser.add_argument("--freeze", action="store_true", help="Exclusively create the record before replay") parser.add_argument("--freeze-path", type=Path, default=FREEZE) args = parser.parse_args() if args.freeze: with args.freeze_path.open("x", encoding="utf-8", newline="\n") as handle: json.dump(freeze_record(), handle, ensure_ascii=False, indent=2) handle.write("\n") print(args.freeze_path.relative_to(ROOT).as_posix()) else: print(json.dumps(run(freeze_path=args.freeze_path), ensure_ascii=False, indent=2))