scripts/backtest_common.py

"""Shared walk-forward backtest helpers.

Used by both ``backtest_walk_forward.py`` (the official ship-gated backtest
for the currently shipped MODEL) and ``upgrade_evaluation.py`` (the
half-life / ensemble experiments). Keeping the fold-fitting and scoring
logic in one place means an experiment and the official backtest can never
silently diverge in how a fold is built or scored.
"""

from __future__ import annotations

from datetime import date

from underdog_lab.domain import Outcome
from underdog_lab.forecasting.dixon_coles import DixonColesEloModel
from underdog_lab.forecasting.scoring import brier_score, log_loss, rank_probability_score
from underdog_lab.forecasting.self_elo import compute_self_elo

from fit_elo_dixon_coles import DEFAULT_BOUNDS, DEFAULT_X0, fit_params, load_matches, time_decay_weights


def load_matches_with_self_elo(cutoff: date) -> list[dict]:
    """Load matches and attach pre-match self-computed Elo ratings.

    ``self_home_elo``/``self_away_elo`` are independent of the eloratings.net
    ``home_elo``/``away_elo`` columns -- see ``forecasting/self_elo.py``.
    """
    matches = load_matches(cutoff)
    for match, (self_home, self_away) in zip(matches, compute_self_elo(matches)):
        match["self_home_elo"] = self_home
        match["self_away_elo"] = self_away
    return matches


def observed_outcome(home_goals: int, away_goals: int) -> Outcome:
    if home_goals > away_goals:
        return "home"
    if home_goals < away_goals:
        return "away"
    return "draw"


def score_candidate(forecast, outcome: str) -> dict[str, float]:
    return {
        "log_loss": log_loss(forecast, outcome),
        "brier": brier_score(forecast, outcome),
        "rps": rank_probability_score(forecast, outcome),
    }


def fit_dixon_coles(
    train_matches: list[dict],
    train_cutoff: date,
    half_life_days: float,
    elo_keys: tuple[str, str] = ("home_elo", "away_elo"),
) -> DixonColesEloModel:
    """Fit a DixonColesEloModel on ``train_matches`` using the given Elo
    source columns and time-decay half-life. Same MLE procedure as
    ``fit_elo_dixon_coles.py``."""
    weights = time_decay_weights(train_matches, train_cutoff, half_life_days)
    if elo_keys != ("home_elo", "away_elo"):
        train_matches = [
            {**m, "home_elo": m[elo_keys[0]], "away_elo": m[elo_keys[1]]}
            for m in train_matches
        ]
    result = fit_params(train_matches, weights, DEFAULT_X0, DEFAULT_BOUNDS)
    intercept, elo_scale, home_adv_logshift, rho = result.x
    return DixonColesEloModel(
        intercept=float(intercept),
        elo_scale=float(elo_scale),
        home_advantage_elo=float(home_adv_logshift / elo_scale),
        rho=float(rho),
    )


def calibration_table(rows: list[tuple[float, bool]]) -> list[dict]:
    """Bucket predicted home-win probability into deciles and compare to
    the realized home-win frequency in each bucket (basic calibration)."""
    buckets: list[list[tuple[float, bool]]] = [[] for _ in range(10)]
    for p_home, was_home in rows:
        index = min(9, int(p_home * 10))
        buckets[index].append((p_home, was_home))

    table = []
    for index, bucket in enumerate(buckets):
        if not bucket:
            continue
        table.append(
            {
                "predicted_range": [index / 10, (index + 1) / 10],
                "n": len(bucket),
                "predicted_mean": sum(row[0] for row in bucket) / len(bucket),
                "observed_home_win_rate": (
                    sum(row[1] for row in bucket) / len(bucket)
                ),
            }
        )
    return table