Spaces:
Running
Running
File size: 3,638 Bytes
b05ddd5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 | """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
|