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():
    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)
    grouped = sweep.score_window(request, 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 grouped == expected
    old_matrix = sweep.build_event_contribution_matrix(request, static_contexts=contexts)
    old_rows = sweep.score_from_matrix(request, old_matrix)
    assert old_rows != expected


def test_recorded_specification_and_all_prespecified_cells():
    report = json.loads((sweep.ROOT / "docs/research/reported_offset_2026_09_20.json").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"] == "candidate_date_grouped_v2"
    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