Jyotisha/tests/test_reported_offset_research.py

from __future__ import annotations

import json
from datetime import datetime, timedelta

import pytest

from scripts.research import reported_offset_sweep as sweep
from scripts.research.sealed_holdout_rerun import DATASET, opaque_order


def public_case():
    return json.loads(DATASET.read_text(encoding="utf-8"))["cases"][0]


def test_zero_offset_centres_on_truth_and_has_complete_grid():
    case = public_case()
    truth = datetime.fromisoformat(f"{case['birth']['date']}T{case['birth']['time']}:00")
    request, candidates = sweep.shifted_window(case, 0, 15)
    assert candidates[15] == truth
    assert len(candidates) == 31
    assert request["start_time"] == candidates[0].strftime("%H:%M")
    assert request["end_time"] == candidates[-1].strftime("%H:%M")
    assert truth in candidates
    assert not any("truth" in key or "offset" in key for key in request)


@pytest.mark.parametrize("offset", [-30, -20, 20, 30])
def test_offset_beyond_radius_has_no_truth_candidate(offset):
    case = public_case()
    truth = datetime.fromisoformat(f"{case['birth']['date']}T{case['birth']['time']}:00")
    _, candidates = sweep.shifted_window(case, offset, 15)
    assert truth not in candidates
    rows = [{"time": moment.strftime("%H:%M"), "score": 1} for moment in candidates]
    metrics = sweep.reveal_metrics(rows, candidates, candidates, truth, "test", "test")
    assert metrics["truth_in_window"] is False
    assert metrics["true_rank"] is None
    assert metrics["delivery_covers_truth"] is False


def test_opaque_tie_break_is_independent_of_truth_and_input_order():
    # Explicitly synthetic unit inputs, not a fabricated engine-contract fixture.
    moments = [datetime(2000, 1, 1, 12) + timedelta(minutes=i) for i in range(3)]
    rows = [{"time": moment.strftime("%H:%M"), "score": 1} for moment in moments]
    ordered = opaque_order("test", "synthetic", rows)
    assert ordered == opaque_order("test", "synthetic", list(reversed(rows)))
    winner = next(moment for moment in moments if moment.strftime("%H:%M") == ordered[0]["time"])
    for truth in moments:
        metrics = sweep.reveal_metrics(rows, moments, moments, truth, "test", "synthetic")
        assert metrics["top_1_hit"] is (truth == winner)
        assert metrics["true_rank"] == 1 + next(i for i, row in enumerate(ordered) if row["time"] == truth.strftime("%H:%M"))


@pytest.mark.parametrize("clock,offset", [("00:02", -5), ("23:58", 5)])
def test_shifted_window_preserves_dates_across_midnight(clock, offset):
    case = public_case()
    # Synthetic boundary mutation of the public case; no real user's birth data.
    case["birth"] = {**case["birth"], "time": clock}
    request, candidates = sweep.shifted_window(case, offset, 15)
    truth = datetime.fromisoformat(f"{case['birth']['date']}T{clock}:00")
    assert candidates[15] == truth + timedelta(minutes=offset)
    assert request["birth_date"] == candidates[0].date().isoformat()
    rows = [{"time": moment.strftime("%H:%M"), "score": int(moment == truth)} for moment in candidates]
    metrics = sweep.reveal_metrics(rows, candidates, candidates, truth, "test", "synthetic")
    assert metrics["top_1_minute_error"] == 0
    assert metrics["truth_in_window"] is True
    assert metrics["delivery_covers_truth"] is True
    assert metrics["delivery_width_minutes"] == 31


def test_cross_midnight_real_engine_scores_match_candidate_date_replay(monkeypatch):
    case = json.loads(DATASET.read_text(encoding="utf-8"))["cases"][5]
    request, candidates = sweep.shifted_window(case, 0, 60)
    assert len({candidate.date() for candidate in candidates}) == 2
    contexts = sweep.compute_candidate_static_contexts(request, candidates=candidates)
    calls = []
    build = sweep.build_event_contribution_matrix
    def traced(request, **kwargs):
        calls.append(len(kwargs["static_contexts"]))
        return build(request, **kwargs)
    with monkeypatch.context() as patch:
        patch.setattr(sweep, "build_event_contribution_matrix", traced)
        native = sweep.score_window(request, contexts)
    assert calls == [len(contexts)]
    expected = []
    for context in contexts:
        dated = {**request, "birth_date": context["candidate_at"].date().isoformat()}
        built = sweep.build_event_contribution_matrix(dated, static_contexts=[context])
        expected.extend(sweep.score_from_matrix(dated, built))
    assert native == expected
    native_matrix = sweep.build_event_contribution_matrix(request, static_contexts=contexts)
    native_rows = sweep.score_from_matrix(request, native_matrix)
    assert native_rows == expected


def test_recorded_specification_and_all_prespecified_cells():
    report = json.loads(sweep.REPORT.read_text(encoding="utf-8"))
    spec = report["specification"]
    assert spec["ayanamsa"] == "raman"
    assert spec["node_mode"] == "mean"
    assert spec["radii_minutes"] == list(sweep.RADII)
    assert spec["offsets_minutes"] == list(sweep.OFFSETS)
    assert spec["minute_step"] == 1
    assert spec["implementation_sha256_prefix"] == spec["implementation_sha256"][:16]
    assert spec["dataset_sha256"] == sweep.file_sha256(DATASET)
    assert spec["evaluator_sha256"] == sweep.file_sha256(sweep.ROOT / "scripts/research/reported_offset_sweep.py")
    assert spec["production_scoring_sha256"] == sweep.implementation_sha256(spec["production_scoring_files"])
    assert spec["research_implementation_sha256"] == sweep.implementation_sha256(spec["research_files"])
    assert spec["replay_revision"] == "native_candidate_date_v3"
    frozen = json.loads(sweep.FREEZE.read_text(encoding="utf-8"))
    assert report["frozen_record"] == spec == frozen
    sweep.verify_frozen_record(frozen, sweep.freeze_record())
    assert report["implementation_hash_matches_at_replay"] is True
    assert report["dataset_hash_matches_at_replay"] is True
    assert frozen["frozen_at_utc"] <= report["replay_started_at_utc"] <= report["replay_finished_at_utc"]
    assert spec["official_valid_independent_blind"] is False
    assert spec["official_blind_trial_count"] == 0
    assert spec["results_previously_seen"] is True
    assert spec["must_not_use_for_tuning"] is True
    assert spec["truth_hidden_from_ranker"] is True
    assert spec["is_blind_evaluation"] is False
    assert report["trial_count"] == 20 * len(sweep.RADII) * len(sweep.OFFSETS)
    assert len(report["summary"]) == len(sweep.RADII) * len(sweep.OFFSETS)
    assert report["summary"] == sweep.summarize(report["trials"], sweep.RADII, sweep.OFFSETS)
    assert {(row["case_ordinal"], row["radius_minutes"], row["offset_minutes"]) for row in report["trials"]} == {
        (case, radius, offset) for case in range(1, 21) for radius in sweep.RADII for offset in sweep.OFFSETS
    }
    assert report["widening_geometry"]["radius_120_scored"] is False
    for trial in report["trials"]:
        assert not ({"predicted_time", "birth", "latitude", "longitude"} & trial.keys())
    for row in report["summary"]:
        expected = float(abs(row["offset_minutes"]) <= row["radius_minutes"])
        assert row["truth_in_window_rate"] == expected
        assert row["delivery_coverage_rate"] <= expected
        assert row["top_1_rate"] <= expected
    assert report["historical_comparison"] == sweep.historical_comparison(
        sweep.LEGACY_REPORT, report["trials"], ("case_ordinal", "radius_minutes", "offset_minutes"),
    )


@pytest.mark.parametrize("key", ["dataset_sha256", "production_scoring_sha256", "research_implementation_sha256", "evaluator_sha256"])
def test_sweep_changed_frozen_identity_fails_before_scoring(tmp_path, monkeypatch, key):
    frozen = sweep.freeze_record()
    frozen[key] = "0" * 64
    path = tmp_path / "bad-freeze.json"
    path.write_text(json.dumps(frozen), encoding="utf-8")
    def unexpected(*args, **kwargs):
        raise AssertionError("must reject identity before scoring")
    monkeypatch.setattr(sweep, "compute_candidate_static_contexts", unexpected)
    with pytest.raises(ValueError, match=f"frozen_record_mismatch:{key}"):
        sweep.run(freeze_path=path)