from __future__ import annotations from collections import defaultdict from dataclasses import dataclass from datetime import datetime, timezone import math from statistics import mean from app.models import FinishedMatch @dataclass(frozen=True) class TeamStats: games: int venue_games: int effective_games: float venue_effective_games: float points_rate: float venue_points_rate: float gf: float ga: float venue_gf: float venue_ga: float last_date: datetime | None @dataclass(frozen=True) class LeagueSummary: home_goals: float away_goals: float draw_rate: float sample_size: int rho: float def _weighted_average(values: list[tuple[float, float]], default: float = 0.0) -> float: if not values: return default weight_sum = sum(weight for _, weight in values) return sum(value * weight for value, weight in values) / weight_sum if weight_sum else default def _age_weight(match_date: datetime, as_of: datetime, half_life_days: float = 75.0) -> float: age_days = max(0.0, (as_of - match_date).total_seconds() / 86400.0) return 0.5 ** (age_days / half_life_days) def team_stats( team_key: str, matches: list[FinishedMatch], venue: str, as_of: datetime, ) -> TeamStats: relevant = [ m for m in matches if m.utc_date < as_of and (m.home_key == team_key or m.away_key == team_key) ] relevant = sorted(relevant, key=lambda x: x.utc_date, reverse=True)[:30] points_values: list[tuple[float, float]] = [] gf_values: list[tuple[float, float]] = [] ga_values: list[tuple[float, float]] = [] venue_points: list[tuple[float, float]] = [] venue_gf: list[tuple[float, float]] = [] venue_ga: list[tuple[float, float]] = [] last_date = relevant[0].utc_date if relevant else None for m in relevant: is_home = m.home_key == team_key gf = m.home_goals if is_home else m.away_goals ga = m.away_goals if is_home else m.home_goals pts_rate = 1.0 if gf > ga else (1.0 / 3.0 if gf == ga else 0.0) weight = _age_weight(m.utc_date, as_of) points_values.append((pts_rate, weight)) gf_values.append((float(gf), weight)) ga_values.append((float(ga), weight)) correct_venue = (venue == "home" and is_home) or (venue == "away" and not is_home) if correct_venue: venue_points.append((pts_rate, weight)) venue_gf.append((float(gf), weight)) venue_ga.append((float(ga), weight)) generic_points = _weighted_average(points_values, 0.44) generic_gf = _weighted_average(gf_values, 1.30) generic_ga = _weighted_average(ga_values, 1.30) return TeamStats( games=len(relevant), venue_games=len(venue_points), effective_games=sum(w for _, w in points_values), venue_effective_games=sum(w for _, w in venue_points), points_rate=generic_points, venue_points_rate=_weighted_average(venue_points, generic_points), gf=generic_gf, ga=generic_ga, venue_gf=_weighted_average(venue_gf, generic_gf), venue_ga=_weighted_average(venue_ga, generic_ga), last_date=last_date, ) def build_elo( matches: list[FinishedMatch], k: float = 22.0, home_advantage: float = 55.0, as_of: datetime | None = None, ) -> dict[str, float]: ratings: dict[str, float] = defaultdict(lambda: 1500.0) for m in sorted(matches, key=lambda x: x.utc_date): if as_of is not None and m.utc_date >= as_of: continue rh, ra = ratings[m.home_key], ratings[m.away_key] exp_h = 1.0 / (1.0 + 10 ** ((ra - (rh + home_advantage)) / 400.0)) if m.home_goals > m.away_goals: actual = 1.0 elif m.home_goals == m.away_goals: actual = 0.5 else: actual = 0.0 margin = abs(m.home_goals - m.away_goals) margin_multiplier = min(1.75, 1.0 + 0.12 * margin) delta = k * margin_multiplier * (actual - exp_h) ratings[m.home_key] = rh + delta ratings[m.away_key] = ra - delta return dict(ratings) def poisson_1x2(lambda_home: float, lambda_away: float, max_goals: int = 9) -> tuple[float, float, float]: return dixon_coles_1x2(lambda_home, lambda_away, rho=0.0, max_goals=max_goals) def _dc_tau(home_goals: int, away_goals: int, lh: float, la: float, rho: float) -> float: if home_goals == 0 and away_goals == 0: return max(0.01, 1.0 - lh * la * rho) if home_goals == 0 and away_goals == 1: return max(0.01, 1.0 + lh * rho) if home_goals == 1 and away_goals == 0: return max(0.01, 1.0 + la * rho) if home_goals == 1 and away_goals == 1: return max(0.01, 1.0 - rho) return 1.0 def dixon_coles_1x2( lambda_home: float, lambda_away: float, rho: float, max_goals: int = 9, ) -> tuple[float, float, float]: def pois(k: int, lam: float) -> float: return math.exp(-lam) * (lam ** k) / math.factorial(k) ph = pd = pa = total = 0.0 for h in range(max_goals + 1): for a in range(max_goals + 1): p = pois(h, lambda_home) * pois(a, lambda_away) p *= _dc_tau(h, a, lambda_home, lambda_away, rho) total += p if h > a: ph += p elif h == a: pd += p else: pa += p if total <= 0: return 1 / 3, 1 / 3, 1 / 3 return ph / total, pd / total, pa / total def _estimate_rho(home_goals: float, away_goals: float, draw_rate: float, sample_size: int) -> float: if sample_size < 60: return 0.0 best_rho = 0.0 best_error = float("inf") for step in range(-15, 11): rho = step / 100.0 _, predicted_draw, _ = dixon_coles_1x2(home_goals, away_goals, rho) error = abs(predicted_draw - draw_rate) if error < best_error: best_error = error best_rho = rho return best_rho def league_summary( matches: list[FinishedMatch], competition: str, as_of: datetime, ) -> LeagueSummary: sample = [m for m in matches if m.competition == competition and m.utc_date < as_of] sample = sorted(sample, key=lambda m: m.utc_date, reverse=True)[:350] if not sample: return LeagueSummary(1.45, 1.15, 0.27, 0, 0.0) hg = mean(m.home_goals for m in sample) ag = mean(m.away_goals for m in sample) draw_rate = sum(m.home_goals == m.away_goals for m in sample) / len(sample) rho = _estimate_rho(hg, ag, draw_rate, len(sample)) return LeagueSummary( home_goals=max(0.70, min(2.20, hg)), away_goals=max(0.60, min(1.90, ag)), draw_rate=draw_rate, sample_size=len(sample), rho=rho, ) def predictive_models( home_key: str, away_key: str, matches: list[FinishedMatch], elo: dict[str, float], competition: str | None = None, as_of: datetime | None = None, ensemble_weights: tuple[float, float, float] | None = None, ) -> dict[str, tuple[float, float, float] | float]: as_of = as_of or datetime.now(timezone.utc) comp = competition or (matches[-1].competition if matches else "UNKNOWN") comp_matches = [m for m in matches if m.competition == comp and m.utc_date < as_of] hs = team_stats(home_key, comp_matches, "home", as_of) aw = team_stats(away_key, comp_matches, "away", as_of) league = league_summary(comp_matches, comp, as_of) def shrink(rate: float, effective_n: float, prior: float, prior_strength: float = 5.5) -> float: n = max(0.0, effective_n) return (rate * n + prior * prior_strength) / (n + prior_strength) # Venue information is valuable but noisy. Blend venue rates with overall rates, # then shrink both towards competition scoring baselines. home_attack_venue = shrink(hs.venue_gf, hs.venue_effective_games, league.home_goals) home_attack_all = shrink(hs.gf, hs.effective_games, (league.home_goals + league.away_goals) / 2) home_attack = 0.68 * home_attack_venue + 0.32 * home_attack_all away_def_venue = shrink(aw.venue_ga, aw.venue_effective_games, league.home_goals) away_def_all = shrink(aw.ga, aw.effective_games, (league.home_goals + league.away_goals) / 2) away_def = 0.68 * away_def_venue + 0.32 * away_def_all away_attack_venue = shrink(aw.venue_gf, aw.venue_effective_games, league.away_goals) away_attack_all = shrink(aw.gf, aw.effective_games, (league.home_goals + league.away_goals) / 2) away_attack = 0.68 * away_attack_venue + 0.32 * away_attack_all home_def_venue = shrink(hs.venue_ga, hs.venue_effective_games, league.away_goals) home_def_all = shrink(hs.ga, hs.effective_games, (league.home_goals + league.away_goals) / 2) home_def = 0.68 * home_def_venue + 0.32 * home_def_all # Geometric combination is deliberately less explosive than multiplying # attack/defence strengths directly. lam_h = math.sqrt(max(0.08, home_attack) * max(0.08, away_def)) lam_a = math.sqrt(max(0.08, away_attack) * max(0.08, home_def)) lam_h = min(3.50, max(0.30, lam_h)) lam_a = min(3.20, max(0.22, lam_a)) poisson = dixon_coles_1x2(lam_h, lam_a, league.rho) draw_anchor = poisson[1] rh = elo.get(home_key, 1500.0) ra = elo.get(away_key, 1500.0) q_home = 1.0 / (1.0 + 10 ** ((ra - (rh + 55.0)) / 400.0)) elo_p = ( (1 - draw_anchor) * q_home, draw_anchor, (1 - draw_anchor) * (1 - q_home), ) form_delta = ( 0.60 * hs.venue_points_rate + 0.40 * hs.points_rate - 0.60 * aw.venue_points_rate - 0.40 * aw.points_rate ) q_form = 1.0 / (1.0 + math.exp(-2.15 * form_delta)) form_p = ( (1 - draw_anchor) * q_form, draw_anchor, (1 - draw_anchor) * (1 - q_form), ) general_q = min(1.0, min(hs.effective_games, aw.effective_games) / 10.0) venue_q = min(1.0, min(hs.venue_effective_games, aw.venue_effective_games) / 4.0) league_q = min(1.0, league.sample_size / 160.0) last_dates = [d for d in (hs.last_date, aw.last_date) if d] if len(last_dates) == 2: days = max((as_of - d).days for d in last_dates) recency = 1.0 if days <= 14 else 0.92 if days <= 30 else 0.75 if days <= 60 else 0.45 else: recency = 0.20 quality = ( 0.36 * general_q + 0.28 * venue_q + 0.22 * league_q + 0.14 * recency ) if ensemble_weights is None: # Default prior weights. When samples are shallow, trust the slow-moving # Elo component slightly more. poisson_w = 0.42 + 0.08 * quality elo_w = 0.38 - 0.05 * quality form_w = 1.0 - poisson_w - elo_w else: pw, ew, fw = ensemble_weights total_w = max(1e-9, pw + ew + fw) poisson_w, elo_w, form_w = pw / total_w, ew / total_w, fw / total_w # Walk-forward tuning is competition-level; event-level low sample still # receives a small stability shift from form toward Elo. low_sample_shift = max(0.0, 0.55 - quality) * 0.12 shifted = min(form_w * 0.45, low_sample_shift) form_w -= shifted elo_w += shifted ensemble = tuple( poisson_w * poisson[i] + elo_w * elo_p[i] + form_w * form_p[i] for i in range(3) ) total = sum(ensemble) ensemble = tuple(p / total for p in ensemble) return { "poisson": poisson, "elo": elo_p, "form": form_p, "ensemble": ensemble, "quality": quality, "lambda_home": lam_h, "lambda_away": lam_a, "rho": league.rho, "league_draw_rate": league.draw_rate, "league_sample": float(league.sample_size), "home_games": float(hs.games), "away_games": float(aw.games), "home_venue_games": float(hs.venue_games), "away_venue_games": float(aw.venue_games), "weight_poisson": float(poisson_w), "weight_elo": float(elo_w), "weight_form": float(form_w), } def tune_ensemble_weights( matches: list[FinishedMatch], competition: str, *, evaluation_matches: int = 56, minimum_training_matches: int = 70, ) -> dict[str, float | tuple[float, float, float]]: """ Time-aware competition-level weight tuning. Every evaluation match is predicted using only matches that happened before it. The selected weights minimize multiclass Brier score on the older part of the walk-forward slice, are shrunk toward a conservative prior, and are accepted only when they hold up on the newer validation part. """ ordered = sorted( [m for m in matches if m.competition == competition], key=lambda m: m.utc_date, ) if len(ordered) < minimum_training_matches + 20: return { "weights": (0.46, 0.34, 0.20), "samples": 0.0, "validation_samples": 0.0, "brier": 0.0, "default_brier": 0.0, "climatology_brier": 0.0, "brier_skill": 0.0, "gain": 0.0, } start = max(minimum_training_matches, len(ordered) - evaluation_matches) rows: list[tuple[ tuple[float, float, float], tuple[float, float, float], tuple[float, float, float], tuple[float, float, float], tuple[float, float, float], ]] = [] for idx in range(start, len(ordered)): target = ordered[idx] train = ordered[:idx] # Need a minimally informative history for both teams. home_count = sum(target.home_key in (m.home_key, m.away_key) for m in train) away_count = sum(target.away_key in (m.home_key, m.away_key) for m in train) if min(home_count, away_count) < 5: continue elo = build_elo(train, as_of=target.utc_date) model = predictive_models( target.home_key, target.away_key, train, elo, competition=competition, as_of=target.utc_date, ensemble_weights=None, ) y = ( (1.0, 0.0, 0.0) if target.home_goals > target.away_goals else (0.0, 1.0, 0.0) if target.home_goals == target.away_goals else (0.0, 0.0, 1.0) ) # Time-safe climatology: computed only from matches available before # the target. It gives us a genuine walk-forward skill baseline instead # of judging the model merely by whether tuned weights beat default weights. baseline_sample = train[-220:] n_base = max(1, len(baseline_sample)) climatology = ( sum(m.home_goals > m.away_goals for m in baseline_sample) / n_base, sum(m.home_goals == m.away_goals for m in baseline_sample) / n_base, sum(m.home_goals < m.away_goals for m in baseline_sample) / n_base, ) rows.append(( tuple(float(x) for x in model["poisson"]), tuple(float(x) for x in model["elo"]), tuple(float(x) for x in model["form"]), climatology, y, )) if len(rows) < 18: return { "weights": (0.46, 0.34, 0.20), "samples": float(len(rows)), "validation_samples": 0.0, "brier": 0.0, "default_brier": 0.0, "climatology_brier": 0.0, "brier_skill": 0.0, "gain": 0.0, } # Tune weights on the older part of the walk-forward predictions and report # skill only on the newer holdout. The base predictions are time-safe, but # selecting and scoring weights on the same slice would still be optimistic. split_at = max(12, int(len(rows) * 0.70)) split_at = min(split_at, len(rows) - 6) tuning_rows = rows[:split_at] validation_rows = rows[split_at:] def brier(weights: tuple[float, float, float], sample=validation_rows) -> float: pw, ew, fw = weights total = 0.0 for pp, ep, fp, _clim, y in sample: pred = tuple(pw * pp[i] + ew * ep[i] + fw * fp[i] for i in range(3)) total += sum((pred[i] - y[i]) ** 2 for i in range(3)) / 3.0 return total / len(sample) default = (0.46, 0.34, 0.20) default_brier = brier(default) climatology_brier = sum( sum((clim[i] - y[i]) ** 2 for i in range(3)) / 3.0 for _pp, _ep, _fp, clim, y in validation_rows ) / len(validation_rows) candidates: list[tuple[float, float, float]] = [default] # Coarse grid is deliberate; a fine grid would overfit the short walk-forward # sample and create fake precision. for pi in range(2, 8): pw = pi / 10.0 for ei in range(2, 8): ew = ei / 10.0 fw = 1.0 - pw - ew if 0.10 <= fw <= 0.40: candidates.append((pw, ew, fw)) best = min(candidates, key=lambda weights: brier(weights, tuning_rows)) # Empirical-Bayes style shrinkage toward prior weights. trust = min(0.70, len(tuning_rows) / (len(tuning_rows) + 45.0)) shrunk = tuple(default[i] * (1.0 - trust) + best[i] * trust for i in range(3)) total_w = sum(shrunk) shrunk = tuple(w / total_w for w in shrunk) shrunk_brier = brier(shrunk) if shrunk_brier > default_brier: shrunk = default shrunk_brier = default_brier # Multiclass Brier Skill Score against a time-safe competition climatology. # Positive = internal model beat the baseline out of sample. Negative = it did not. brier_skill = ( 1.0 - shrunk_brier / climatology_brier if climatology_brier > 1e-12 else 0.0 ) return { "weights": shrunk, "samples": float(len(rows)), "validation_samples": float(len(validation_rows)), "brier": float(shrunk_brier), "default_brier": float(default_brier), "climatology_brier": float(climatology_brier), "brier_skill": float(brier_skill), "gain": float(max(0.0, default_brier - shrunk_brier)), }