scripts/recalibration_evaluation.py

"""Post-hoc temperature-scaling recalibration for the shipped Dixon-Coles Elo
model.

The walk-forward folds from ``backtest_walk_forward.py`` (refit per test
year at ``HALF_LIFE_DAYS=180.0``, the current shipped half-life -- see
``models/upgrade_evaluation.json``) are reused to collect out-of-fold
forecasts for every match 2018-2026. A single temperature ``T`` (see
``forecasting/calibration.py``) is fit on the pooled SELECTION_YEARS
(2018-2025) out-of-fold forecasts and evaluated against the ``T=1``
(no-op) baseline using the same three-way selection/confirmation discipline
as ``upgrade_evaluation.py``:

  - SELECTION_YEARS (2018-2025): the fitted T must beat T=1 on mean log loss
    here.
  - CONFIRMATION_YEAR (2026): held out from fitting. The fitted T must ALSO
    beat T=1 on this fold, both overall and on the neutral-venue subset.

Writes models/recalibration_evaluation.json. Does not modify
src/underdog_lab/world_cup/forecasting.py or models/backtest_report.json --
adopting the fitted temperature is a separate, human-reviewed step.

Usage:
  python scripts/recalibration_evaluation.py
"""

from __future__ import annotations

import json
import random
from datetime import date

from underdog_lab.config import MODEL_DIR
from underdog_lab.forecasting.calibration import apply_temperature, fit_temperature

from backtest_common import (
    fit_dixon_coles,
    load_matches_with_self_elo,
    observed_outcome,
    score_candidate,
)

REPORT_PATH = MODEL_DIR / "recalibration_evaluation.json"

HALF_LIFE_DAYS = 180.0
SELECTION_YEARS = list(range(2018, 2026))
CONFIRMATION_YEAR = 2026
ALL_YEARS = SELECTION_YEARS + [CONFIRMATION_YEAR]

EMPTY_TOTALS = {"log_loss": 0.0, "brier": 0.0, "rps": 0.0}
BOOTSTRAP_ITERATIONS = 2000


def fold_matches(all_matches: list[dict], test_year: int):
    train_cutoff = date(test_year - 1, 12, 31)
    train_matches = [m for m in all_matches if m["date"] <= train_cutoff]
    test_matches = [m for m in all_matches if m["date"].year == test_year]
    return train_cutoff, train_matches, test_matches


def collect_out_of_fold(all_matches: list[dict], years: list[int]) -> dict[int, list[tuple]]:
    """For each year, fit a model on data strictly before it and forecast
    every match played during it. Returns {year: [(forecast, outcome, is_neutral), ...]}."""
    per_year: dict[int, list[tuple]] = {}
    for year in years:
        train_cutoff, train_matches, test_matches = fold_matches(all_matches, year)
        if not test_matches:
            continue
        model = fit_dixon_coles(train_matches, train_cutoff, HALF_LIFE_DAYS)
        rows = []
        for match in test_matches:
            forecast = model.forecast(match["home_elo"], match["away_elo"], neutral_venue=match["neutral"])
            outcome = observed_outcome(match["home_goals"], match["away_goals"])
            rows.append((forecast, outcome, match["neutral"]))
        per_year[year] = rows
    return per_year


def _aggregate(rows: list[tuple], temperature: float) -> tuple[dict, dict, int, int]:
    """rows: list of (forecast, outcome, is_neutral). Returns (mean_all, mean_neutral, n_all, n_neutral)."""
    totals_all = dict(EMPTY_TOTALS)
    totals_neutral = dict(EMPTY_TOTALS)
    n_all = 0
    n_neutral = 0
    for forecast, outcome, is_neutral in rows:
        scores = score_candidate(apply_temperature(forecast, temperature), outcome)
        for key, value in scores.items():
            totals_all[key] += value
        n_all += 1
        if is_neutral:
            for key, value in scores.items():
                totals_neutral[key] += value
            n_neutral += 1
    mean_all = {k: v / n_all for k, v in totals_all.items()} if n_all else {}
    mean_neutral = {k: v / n_neutral for k, v in totals_neutral.items()} if n_neutral else {}
    return mean_all, mean_neutral, n_all, n_neutral


def paired_log_loss_interval(
    rows: list[tuple],
    temperature: float,
    *,
    seed: int = 2026,
) -> list[float]:
    """Bootstrap fitted-minus-baseline log-loss differences."""
    differences = [
        score_candidate(apply_temperature(forecast, temperature), outcome)[
            "log_loss"
        ]
        - score_candidate(forecast, outcome)["log_loss"]
        for forecast, outcome, _ in rows
    ]
    rng = random.Random(seed)
    samples = [
        sum(rng.choice(differences) for _ in differences) / len(differences)
        for _ in range(BOOTSTRAP_ITERATIONS)
    ]
    samples.sort()
    return [
        samples[int(0.025 * BOOTSTRAP_ITERATIONS)],
        samples[int(0.975 * BOOTSTRAP_ITERATIONS)],
    ]


def main() -> None:
    all_matches = load_matches_with_self_elo(date(2026, 6, 12))
    per_year = collect_out_of_fold(all_matches, ALL_YEARS)

    selection_rows = [row for year in SELECTION_YEARS for row in per_year.get(year, [])]
    confirmation_rows = per_year.get(CONFIRMATION_YEAR, [])
    confirmation_neutral_rows = [
        row for row in confirmation_rows if row[2]
    ]

    temperature = fit_temperature([(forecast, outcome) for forecast, outcome, _ in selection_rows])

    selection_baseline, _, n_selection, _ = _aggregate(selection_rows, 1.0)
    selection_fitted, _, _, _ = _aggregate(selection_rows, temperature)

    confirm_baseline_all, confirm_baseline_neutral, n_confirm, n_confirm_neutral = _aggregate(
        confirmation_rows, 1.0
    )
    confirm_fitted_all, confirm_fitted_neutral, _, _ = _aggregate(confirmation_rows, temperature)

    beats_on_selection = selection_fitted["log_loss"] < selection_baseline["log_loss"]
    beats_on_confirmation = confirm_fitted_all["log_loss"] < confirm_baseline_all["log_loss"]
    beats_on_confirmation_neutral = (
        confirm_fitted_neutral["log_loss"] < confirm_baseline_neutral["log_loss"]
    )
    intervals = {
        "selection": paired_log_loss_interval(selection_rows, temperature),
        "confirmation": paired_log_loss_interval(
            confirmation_rows,
            temperature,
        ),
        "confirmation_neutral": paired_log_loss_interval(
            confirmation_neutral_rows,
            temperature,
        ),
    }
    intervals_below_zero = all(interval[1] < 0.0 for interval in intervals.values())
    adopted = (
        temperature != 1.0
        and beats_on_selection
        and beats_on_confirmation
        and beats_on_confirmation_neutral
        and intervals_below_zero
    )

    report = {
        "half_life_days": HALF_LIFE_DAYS,
        "selection_years": SELECTION_YEARS,
        "confirmation_year": CONFIRMATION_YEAR,
        "n_selection": n_selection,
        "n_confirmation": n_confirm,
        "n_confirmation_neutral": n_confirm_neutral,
        "fitted_temperature": temperature,
        "selection": {
            "baseline_t1": selection_baseline,
            "fitted_t": selection_fitted,
        },
        "confirmation": {
            "baseline_t1": confirm_baseline_all,
            "fitted_t": confirm_fitted_all,
        },
        "confirmation_neutral": {
            "baseline_t1": confirm_baseline_neutral,
            "fitted_t": confirm_fitted_neutral,
        },
        "paired_log_loss_difference_bootstrap_95": intervals,
        "gate": {
            "beats_baseline_on_selection": beats_on_selection,
            "beats_baseline_on_confirmation": beats_on_confirmation,
            "beats_baseline_on_confirmation_neutral": beats_on_confirmation_neutral,
            "all_paired_bootstrap_intervals_below_zero": intervals_below_zero,
        },
        "adopted": adopted,
        "criterion": (
            "The fitted temperature must differ from 1.0 and produce a lower "
            "mean log loss than the T=1 (no-op) baseline on the pooled "
            "2018-2025 selection folds AND on the held-out 2026 confirmation "
            "fold, both overall and on its neutral-venue subset. Paired 95% "
            "bootstrap intervals for all three reported slices must remain "
            "below zero. The neutral subset overlaps the overall 2026 fold "
            "and is not an independent confirmation sample."
        ),
    }

    MODEL_DIR.mkdir(parents=True, exist_ok=True)
    REPORT_PATH.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8")
    print(f"Wrote {REPORT_PATH}")
    print("fitted temperature:", temperature, "(adopted:", adopted, ")")
    print(json.dumps(report["selection"], indent=2))
    print(json.dumps(report["confirmation"], indent=2))
    print(json.dumps(report["confirmation_neutral"], indent=2))
    print(json.dumps(report["gate"], indent=2))


if __name__ == "__main__":
    main()