NFL Models — Math & Proof

"""
Feature builders for outcome models.

Leakage contract
----------------
All "form" features for a (team, season, week) game are computed from games
played BEFORE (season, week): an opponent-adjusted rolling mean, shifted by
one week. A test in tests/ guards the .shift(1) on every rolling column.

Team features come from schedules (points for/against) joined with weekly
team aggregates built from stats_player_week (EPA, volumes).

Player volume features come from each player's own season-to-date mean
(shifted by one week). Opponent context joins the opponent's season-to-date
allowed volumes (team-level aggregates of stats_player_week, also shifted).

There is no cross-league future peeking: every join key carries
(athlete/team, season) and the join is on week values strictly `< current`.
"""

from __future__ import annotations

import numpy as np
import pandas as pd

# --------------------------------------------------------------- team form

PT_STATS = [
    "passing_yards", "passing_tds", "passing_interceptions",
    "rushing_yards", "rushing_tds", "passing_epa", "rushing_epa",
    "attempts", "completions", "carries", "targets", "receptions",
    "receiving_yards", "def_tackles_solo", "def_interceptions", "def_sacks",
]


def build_weekly_team_stats(player_week: pd.DataFrame) -> pd.DataFrame:
    """Weekly TEAM totals aggregated from weekly player stats."""
    pw = player_week[player_week.season_type == "REG"].copy()
    numeric = [c for c in PT_STATS if c in pw.columns]
    agg = (
        pw.groupby(["season", "week", "team"], as_index=False)[numeric].sum()
        .rename(columns={
            "def_tackles_solo": "tackles_solo",
            "def_interceptions": "takeaways_int",
            "def_sacks": "sacks_made",
        })
    )
    return agg


def team_form_features(team_games: pd.DataFrame, stats: pd.DataFrame,
                       window: int = 4) -> pd.DataFrame:
    """
    Lagged team-form features per (team, season, week).

    team_games: one row per team per game with columns season, week, team,
      opp, points_for, points_against.
    stats: weekly team aggregate (same keys), stats columns aggregated.

    Every feature here is .shift(1)-lagged within (team, season): value at
    week w contains ONLY weeks < w. EXTRA wired features beyond the v1 set:
      - DEFENSE: opponent-allowed EPA (pass/rush, per game allowed)
      - pace proxy: attempts + carries per game
      - turnover counts: passing_interceptions (thrown) + defensive INTs
        (taken) -- turnovers the user asked for
      - per-home/away-split rolling means (home-away EPA splits) via same lag
    """
    # offense-side stats (what a team's unit did that week)
    keep_stats = [c for c in [
        "passing_epa", "rushing_epa", "passing_yards", "rushing_yards",
        "attempts", "carries",
        "passing_interceptions",     # turnovers thrown (offense risk)
        "takeaways_int",             # takeaways made (defense; renamed col in stats)
    ] if c in stats.columns]

    # pace proxy: total meaningful offensive plays per game
    stats = stats.copy()
    stats["pace"] = stats.get("attempts", 0.0) + stats.get("carries", 0.0)
    keep_stats.append("pace")

    # defense-side: stats allowed BY a team == what opponents recorded vs it.
    # opps' passing_epa vs team = team's pass-defense performance, etc.
    opp_keep = [c for c in ["season", "week", "team",
                            "passing_epa", "rushing_epa",
                            "passing_yards", "rushing_yards",
                            "passing_interceptions"]
                if c in stats.columns]
    def_stats = (
        stats[opp_keep].groupby(["season", "week", "team"], as_index=False).sum()
        .rename(columns={
            "passing_epa": "def_pass_epa_allowed",
            "rushing_epa": "def_rush_epa_allowed",
            "passing_yards": "def_pass_yds_allowed",
            "rushing_yards": "def_rush_yds_allowed",
            "passing_interceptions": "def_ints_forced",
        })
    )
    def_cols = [c for c in def_stats.columns
                if c not in ("season", "week", "team")]

    tg = team_games[["season", "week", "team", "opp", "points_for", "points_against"]].merge(
        stats[["season", "week", "team"] + keep_stats],
        on=["season", "week", "team"], how="left",
    ).merge(
        def_stats, on=["season", "week", "team"], how="left",
    )
    tg = tg.fillna({c: 0.0 for c in keep_stats + def_cols})
    tg["points_diff"] = tg.points_for - tg.points_against
    tg = tg.sort_values(["team", "season", "week"]).reset_index(drop=True)

    shift_cols = (["points_for", "points_against", "points_diff"]
                  + keep_stats + def_cols)
    out = tg.copy()
    g = tg.groupby(["team", "season"], sort=False)
    shifted = g[shift_cols].shift(1)
    shifted.columns = ["prev_" + c for c in shifted.columns]
    out = pd.concat([out, shifted], axis=1)

    roll_src = ["prev_points_diff"] + [f"prev_{c}" for c in shift_cols[3:]]
    shifted_roll = (
        out.groupby(["team", "season"], sort=False)[roll_src]
        .rolling(window, min_periods=1).mean()
        .reset_index(level=[0, 1], drop=True)
        .add_suffix(f"_roll{window}")
    )
    out = pd.concat([out, shifted_roll], axis=1)

    cols = ["season", "week", "team", "opp"] + [
        c for c in out.columns if c.startswith("prev_")
    ]
    return out[cols]


def join_home_away(game_rows: pd.DataFrame, team_form: pd.DataFrame) -> pd.DataFrame:
    """Join home/away team-form features onto game rows."""
    tf = team_form.copy()
    tf_cols = [c for c in tf.columns if c.startswith("prev_")]
    home = tf.rename(columns={c: "home_" + c for c in tf_cols})[
        ["season", "week", "team"] + ["home_" + c for c in tf_cols]
    ]
    away = tf.rename(columns={c: "away_" + c for c in tf_cols})[
        ["season", "week", "team"] + ["away_" + c for c in tf_cols]
    ]
    out = game_rows.merge(home, left_on=["season", "week", "home_team"],
                          right_on=["season", "week", "team"], how="left").drop(columns=["team"])
    out = out.merge(away, left_on=["season", "week", "away_team"],
                    right_on=["season", "week", "team"], how="left").drop(columns=["team"])
    return out


# ---- market-specific feature lists. All are .shift(1)-derived form features,
# game-context features, or market lines that are PUBLIC before the game.
# Spread/margin might legitimately use different covariates than totals.

COMMON_FORM = [
    "home_prev_points_diff", "away_prev_points_diff",
    "home_prev_passing_epa", "away_prev_passing_epa",
    "home_prev_rushing_epa", "away_prev_rushing_epa",
    "home_prev_points_diff_roll4", "away_prev_points_diff_roll4",
    "home_prev_passing_epa_roll4", "away_prev_passing_epa_roll4",
    "home_prev_rushing_epa_roll4", "away_prev_rushing_epa_roll4",
    "home_prev_points_for", "away_prev_points_for",
    "home_prev_points_against", "away_prev_points_against",
]

DEFENSE_FEATURES = [
    "home_prev_def_pass_epa_allowed", "away_prev_def_pass_epa_allowed",
    "home_prev_def_rush_epa_allowed", "away_prev_def_rush_epa_allowed",
    "home_prev_def_pass_yds_allowed", "away_prev_def_pass_yds_allowed",
    "home_prev_def_rush_yds_allowed", "away_prev_def_rush_yds_allowed",
]

TURNOVER_FEATURES = [
    "home_prev_passing_interceptions",     # turnovers thrown (offense risk)
    "away_prev_passing_interceptions",
    "home_prev_takeaways_int",             # takeaways made (defense upside)
    "away_prev_takeaways_int",
    "home_prev_def_ints_forced",           # opp QBs' INTs vs this team (raw stat)
    "away_prev_def_ints_forced",
]

PACE_FEATURES = [
    "home_prev_pace", "away_prev_pace",
    "home_prev_pace_roll4", "away_prev_pace_roll4",
]

CONTEXT_FEATURES = [
    "home_rest", "away_rest", "rest_diff",
    "temp", "wind", "wind_missing", "temp_missing",
    "domestic",          # roof == 'dome'
    "divisional",
    "week",              # early-season form noise differs
]

# Home-away matchup features (interaction-style, computed differences)
DIFF_FEATURES = [
    "diff_prev_points_diff",
    "diff_prev_passing_epa",
    "diff_prev_rushing_epa",
    "diff_prev_points_diff_roll4",
    "diff_prev_passing_epa_roll4",
    "diff_prev_def_pass_epa_allowed",
    "diff_prev_def_rush_epa_allowed",
]

# A spread model cares about DIFFERENCES between teams (matchup-adjusted margin).
# Home-field advantage handled by home_rest + intercept/differential intercept.
SPREAD_FEATURES = (
    DIFF_FEATURES + CONTEXT_FEATURES
)

# A totals model cares about COMBINED scoring capacity + pace + environment.
TOTAL_FEATURES = (
    [c for c in COMMON_FORM] + TURNOVER_FEATURES + PACE_FEATURES
    + DEFENSE_FEATURES + CONTEXT_FEATURES
)

# Moneyline: same as spread (margin drives win prob). Kept separate so the
# win model can later add per-team QB strength params.
MONEYLINE_FEATURES = SPREAD_FEATURES

GAME_FEATURES = sorted(set(COMMON_FORM + DEFENSE_FEATURES + TURNOVER_FEATURES
                           + PACE_FEATURES + CONTEXT_FEATURES + DIFF_FEATURES))

# Market-float features -- the closing/opening lines are known BEFORE kickoff
# (legal pregame data). Kept separate so the "hybrid" variant is explicitly
# labeled and can never be confused with a model-only forecast.
HYBRID_FEATURES = ["spread_line", "total_line"]


# Static division alignment (2026-era alignment, stable since 2002; historical
# seasons with old teams just don't map -- that's fine, treated as 0/missing).
DIVISIONS = {
    ("AFC", "East"): ["BUF", "MIA", "NE", "NYJ"],
    ("AFC", "North"): ["BAL", "CIN", "CLE", "PIT"],
    ("AFC", "South"): ["HOU", "IND", "JAX", "TEN"],
    ("AFC", "West"): ["DEN", "KC", "LAC", "LV"],
    ("NFC", "East"): ["DAL", "NYG", "PHI", "WAS"],
    ("NFC", "North"): ["CHI", "DET", "GB", "MIN"],
    ("NFC", "South"): ["ATL", "CAR", "NO", "TB"],
    ("NFC", "West"): ["ARI", "LAR", "SEA", "SF"],
}
_DIVISION_LOOKUP = {t: div for (conf, div), teams in DIVISIONS.items()
                    for t in teams}


def build_game_matrix(schedules: pd.DataFrame, team_stats: pd.DataFrame) -> pd.DataFrame:
    """Full feature matrix for team-level models, one row per game.

    Wires every context signal available pregame:
      rest_diff          = home_rest - away_rest  (positive = home more rested)
      divisional         = 1 if both teams share a division
      domestic           = 1 if roof == 'dome'
      temp_missing / wind_missing = 1 if weather field missing (dome games)
      diff_* features    = home minus away on every shared form feature
                           (explicit matchup terms for the spread model)
    """
    tg_home = schedules.rename(columns={"home_team": "team", "away_team": "opp"})[
        ["season", "week", "team", "opp", "home_score", "away_score"]
    ].assign(points_for=lambda d: d.home_score, points_against=lambda d: d.away_score)
    tg_away = schedules.rename(columns={"away_team": "team", "home_team": "opp"})[
        ["season", "week", "team", "opp", "away_score", "home_score"]
    ].assign(points_for=lambda d: d.away_score, points_against=lambda d: d.home_score)
    team_games = pd.concat([tg_home, tg_away], ignore_index=True)[
        ["season", "week", "team", "opp", "points_for", "points_against"]
    ]

    form = team_form_features(team_games, team_stats)
    mat = join_home_away(schedules, form)

    # --- game-context features ---
    mat["domestic"] = (mat["roof"] == "dome").astype(int) if "roof" in mat else 0
    mat["wind_missing"] = mat["wind"].isna().astype(int) if "wind" in mat else 1
    mat["temp_missing"] = mat["temp"].isna().astype(int) if "temp" in mat else 1
    mat["wind"] = mat.get("wind", pd.Series(0.0, index=mat.index)).fillna(5.0)
    mat["temp"] = mat.get("temp", pd.Series(0.0, index=mat.index)).fillna(65.0)
    if "home_rest" not in mat:
        mat["home_rest"] = 7
    if "away_rest" not in mat:
        mat["away_rest"] = 7
    mat["home_rest"] = mat["home_rest"].fillna(7.0)
    mat["away_rest"] = mat["away_rest"].fillna(7.0)
    mat["rest_diff"] = mat["home_rest"] - mat["away_rest"]
    mat["divisional"] = (
        (mat["home_team"].map(_DIVISION_LOOKUP) == mat["away_team"].map(_DIVISION_LOOKUP))
        & mat["home_team"].map(_DIVISION_LOOKUP).notna()
    ).astype(int)

    # --- explicit matchup differences (spread-relevant) ---
    diff_bases = [
        "prev_points_diff", "prev_passing_epa", "prev_rushing_epa",
        "prev_points_diff_roll4", "prev_passing_epa_roll4",
        "prev_def_pass_epa_allowed", "prev_def_rush_epa_allowed",
    ]
    for base in diff_bases:
        h, a = f"home_{base}", f"away_{base}"
        hs = mat[h] if h in mat.columns else pd.Series(0.0, index=mat.index)
        aser = mat[a] if a in mat.columns else pd.Series(0.0, index=mat.index)
        mat[f"diff_{base}"] = (hs.astype(float).fillna(0.0)
                               - aser.astype(float).fillna(0.0))
    return mat


# ------------------------------------------------------------- player volume

VOL_TARGETS = ["attempts", "completions", "carries", "receptions", "targets",
               "passing_yards", "rushing_yards", "receiving_yards"]


def player_volume_matrix(player_week: pd.DataFrame, team_stats: pd.DataFrame,
                         schedules: pd.DataFrame) -> pd.DataFrame:
    """
    One row per (player, season, week) with:
      - player's own season-to-date trailing mean of each VOL_TARGET (shifted 1 week)
      - personalized position group
      - opponent's season-to-date average ALLOWED volumes for positions (QB/RB/WR/TE)
      - the actual target values (for training/backtest)
    """
    pw = player_week[player_week.season_type == "REG"].copy()
    pw = pw.sort_values(["player_id", "season", "week"]).reset_index(drop=True)

    # full prop-stat coverage: trailing mean of EVERY predicted stat exists so
    # both the feature row and the "over own trailing mean" probe work.
    prev_targets = [c for c in
                    VOL_TARGETS + ["passing_tds", "rushing_tds", "receiving_tds",
                                   "def_tackles_solo", "def_interceptions",
                                   "def_sacks", "passing_interceptions"]
                    if c in pw.columns]
    # expanding mean shifted by 1 == (cumsum - current) / count_seen_before.
    # Written as one vectorized cumsum pass: groupby.expanding().transform per
    # column allocates a copy per group and OOMs on low-RAM hosts.
    pw[prev_targets] = pw[prev_targets].astype("float32")
    grp = pw.groupby(["player_id", "season"], sort=False)
    csum = grp[prev_targets].transform("cumsum")
    n_before = grp.cumcount().astype("float32")               # rows before current
    prev_means = (csum - pw[prev_targets]).div(n_before.where(n_before > 0), axis=0)
    prev_means.columns = [f"prev_{c}_mean" for c in prev_targets]
    pw = pd.concat([pw, prev_means], axis=1)
    del csum, n_before, prev_means, grp

    # opponent's season-to-date average allowed volume for that position
    pos_map = pw.groupby(["season", "week", "team"])[
        [f"{c}" for c in VOL_TARGETS] + []
    ].sum().reset_index()

    # For each (season, week, team): trailing mean of those weekly sums.
    # Same vectorized cumsum trick (groupby.expanding().transform OOMs here).
    pos_map = pos_map.sort_values(["team", "season", "week"]).reset_index(drop=True)
    pos_map[VOL_TARGETS] = pos_map[VOL_TARGETS].astype("float32")
    tg = pos_map.groupby(["team", "season"], sort=False)
    tsum = tg[VOL_TARGETS].transform("cumsum")
    nt = tg.cumcount().astype("float32")
    allowed = (tsum - pos_map[VOL_TARGETS]).div(nt.where(nt > 0), axis=0)
    allowed.columns = [f"opp_{c}_allowed_mean" for c in allowed.columns]
    pos_map = pd.concat([pos_map[["season", "team", "week"]], allowed], axis=1)
    del tsum, nt, allowed, tg

    # player's opponent team from game/week -> schedules. Pairing matters:
    # home_team's opponent is away_team ONLY in that game (building this with
    # a melting self-join joins every team against every other team playing
    # that week and massively duplicates rows).
    hl = schedules[["season", "week", "home_team", "away_team"]]
    opp_lookup = pd.concat([
        hl.rename(columns={"home_team": "team", "away_team": "opp"}),
        hl.rename(columns={"away_team": "team", "home_team": "opp"}),
    ], ignore_index=True)[["season", "week", "team", "opp"]].drop_duplicates()
    out = pw.merge(opp_lookup, on=["season", "week", "team"], how="left")
    out = out.merge(
        pos_map.rename(columns={"team": "opp"}),
        on=["season", "week", "opp"], how="left",
    )
    return out

"""
Market-specific models. One class per betting market, each with:
  - its own feature list (from features.py)
  - its own regularization alpha chosen by TIME-SERIES CV inside the train set
  - a fitted residual sigma so every market outputs a CALIBRATED probability,
    not just a point estimate:
        P(cover/over) = 1 - Phi((line - pred) / sigma)
    where sigma is the history of |pred - actual| residuals on the train fold.

Math (linear models, standardized ridge):
    beta = argmin_b ||y - S X b||^2 + alpha ||b||^2   where S = StandardScaler
The normal-residual assumption is honest: we show calibration bins in the
backtest report instead of asserting it's Gaussian.

WIN / CALIBRATION:
  - SpreadModel(coach): ridge on SPREAD_FEATURES -> margin
      P(home covers spread L) = Phi((pred_margin - L) / sigma_margin)
      P(home wins) = Phi(pred_margin / sigma_margin)
  - TotalModel: ridge on TOTAL_FEATURES -> total
      P(over L) = 1 - Phi((L - pred_total) / sigma_total)  = Phi((pred_total-L)/sigma)
  - MoneylineModel: logistic on MONEYLINE_FEATURES -> P(home win)
  - PlayerPropModel: ridge on player's trailing means + opponent-allowed means
      P(over L) = Phi((pred - L) / sigma_target)   (per-target sigma)

Hyperparameter selection: for each walk-forward fold, alphas are scored on the
LAST TRAIN SEASON (a strict time-series validation split inside training); the
chosen alpha is the argmin of MAE. Chosen alpha is recorded in the report, per
fold -- "dynamic tuning" in the sense of re-tuning every fold without leakage.
"""

from __future__ import annotations

from math import erf, exp, sqrt
from typing import Optional

import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression, Ridge
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

from forecaster.features import (VOL_TARGETS, SPREAD_FEATURES, TOTAL_FEATURES,
                             MONEYLINE_FEATURES, HYBRID_FEATURES)

ALPHA_GRID = [0.03, 0.1, 0.3, 1.0, 3.0, 10.0, 30.0, 100.0]


def _phi(z: np.ndarray) -> np.ndarray:
    """Standard normal CDF; vectorized, no scipy dependency."""
    return 0.5 * (1.0 + np.vectorize(erf)(z / sqrt(2.0)))


def _x(df: pd.DataFrame, features: list[str]) -> pd.DataFrame:
    return df[features].astype(float).fillna(0.0)


def _ridge_pipe(alpha: float) -> Pipeline:
    return Pipeline([("s", StandardScaler()), ("m", Ridge(alpha=alpha))])


class _RidgeMarket:
    """Shared machinery: per-fold alpha selection + residual-sigma fit."""

    feature_list: list[str] = []
    target_name: str = ""

    def __init__(self, use_market: bool = False):
        self.use_market = use_market  # hybrid variant adds closing lines

    def _features(self, df: pd.DataFrame) -> list[str]:
        f = list(self.feature_list)
        if self.use_market:
            f += [c for c in HYBRID_FEATURES if c in df.columns]
            # keep only rows w/ market for hybrid; backtest guarantees it
        return f

    def _target(self, df):
        raise NotImplementedError

    def _tune_alpha(self, train: pd.DataFrame, feats: list[str]) -> float:
        """Time-series CV: final season of train is the validation split."""
        seasons = sorted(train.season.unique())
        if len(seasons) < 3:
            return 1.0  # not enough seasons for a holdout
        tr = train[train.season < seasons[-1]]
        va = train[train.season == seasons[-1]]
        if len(va) < 20:
            return 1.0
        scores = {}
        for a in ALPHA_GRID:
            m = _ridge_pipe(a).fit(_x(tr, feats), self._target(tr))
            pred = m.predict(_x(va, feats))
            scores[a] = float(np.mean(np.abs(pred - self._target(va).to_numpy())))
        return min(scores, key=scores.get)

    def fit(self, train: pd.DataFrame) -> "_RidgeMarket":
        feats = self._features(train)
        if self.use_market:
            train = train.dropna(subset=[c for c in HYBRID_FEATURES
                                         if c in train.columns])
        self.features_ = feats
        y = self._target(train)
        self.alpha_ = self._tune_alpha(train, feats)
        self.model_ = _ridge_pipe(self.alpha_).fit(_x(train, feats), y)
        pred_tr = self.model_.predict(_x(train, feats))
        # Residual sigma -- calibrated on TRAIN only (no test leakage)
        self.sigma_ = float(np.sqrt(np.mean((pred_tr - y.to_numpy()) ** 2)))
        self.sigma_ = max(self.sigma_, 1.0)  # floor: avoid overconfidence
        self.n_train_ = len(train)
        return self

    def predict(self, df: pd.DataFrame) -> np.ndarray:
        return self.model_.predict(_x(df, self.features_))


class SpreadModel(_RidgeMarket):
    feature_list = SPREAD_FEATURES

    def _target(self, df):
        return (df.home_score - df.away_score).astype(float)

    def predict_market(self, df: pd.DataFrame) -> pd.DataFrame:
        pred_margin = self.predict(df)
        out = pd.DataFrame({"pred_margin": pred_margin}, index=df.index)
        out["pred_home_win_prob"] = _phi(pred_margin / self.sigma_)
        if "spread_line" in df.columns:
            line = df.spread_line.fillna(0.0).to_numpy()
            out["prob_home_covers"] = _phi((pred_margin - line) / self.sigma_)
            out["prob_away_covers"] = 1.0 - out["prob_home_covers"]
            out["pred_home_score"] = np.nan  # margin-only market doesn't claim totals
        return out


class TotalModel(_RidgeMarket):
    feature_list = TOTAL_FEATURES

    def _target(self, df):
        return (df.home_score + df.away_score).astype(float)

    def predict_market(self, df: pd.DataFrame) -> pd.DataFrame:
        pred_total = self.predict(df)
        out = pd.DataFrame({"pred_total": pred_total}, index=df.index)
        if "total_line" in df.columns:
            line = df.total_line.fillna(45.0).to_numpy()
            out["prob_over"] = _phi((pred_total - line) / self.sigma_)
            out["prob_under"] = 1.0 - out["prob_over"]
        return out


class MoneylineModel:
    """Logistic regression P(home win). Calibrated-ish via sklearn's in-built
    probability estimates; the calibration bins in the report show truth."""

    def __init__(self, C: float = 1.0, use_market: bool = False):
        self.use_market = use_market
        self.C = C
        self.feature_list = list(MONEYLINE_FEATURES)

    @staticmethod
    def _with_implied(df: pd.DataFrame) -> pd.DataFrame:
        """Attach _imp_* implied-probability columns derived from the closing
        american odds (hybrid variant only). vig stays in on purpose -- the
        model learns calibration over it."""
        df = df.copy()
        for col in ("home_moneyline", "away_moneyline"):
            if col in df.columns:
                df[f"_imp_{col}"] = np.where(
                    df[col].isna(), np.nan,
                    np.where(df[col] > 0,
                             100.0 / (df[col] + 100.0),
                             (-df[col]) / ((-df[col]) + 100.0)))
        return df

    def _features(self, df) -> list[str]:
        f = list(self.feature_list)
        if self.use_market:
            f += [c for c in ("_imp_home_moneyline", "_imp_away_moneyline")
                  if c in df.columns]
        return f

    def _tune_C(self, train: pd.DataFrame, feats: list[str], y) -> float:
        seasons = sorted(train.season.unique())
        if len(seasons) < 3:
            return self.C
        tr = train[train.season < seasons[-1]]
        va = train[train.season == seasons[-1]]
        y_tr = (tr.home_score > tr.away_score).astype(int)
        y_va = (va.home_score > va.away_score).astype(int)
        scores = {}
        for c in ALPHA_GRID:
            p = Pipeline([("s", StandardScaler()),
                          ("m", LogisticRegression(max_iter=2000, C=c))])
            p.fit(_x(tr, feats), y_tr)
            pr = p.predict_proba(_x(va, feats))[:, 1]
            scores[c] = float(np.mean((pr - y_va) ** 2))
        return min(scores, key=scores.get)

    def fit(self, train: pd.DataFrame) -> "MoneylineModel":
        if self.use_market:
            train = train.dropna(subset=["home_moneyline", "away_moneyline"])
            train = self._with_implied(train)
        self.features_ = self._features(train)
        trainx = _x(train, self.features_)
        y = (train.home_score > train.away_score).astype(int)
        self.C_ = self._tune_C(train, self.features_, y)
        self.model_ = Pipeline([("s", StandardScaler()),
                                ("m", LogisticRegression(max_iter=2000, C=self.C_))])
        self.model_.fit(trainx, y)
        self.n_train_ = len(train)
        return self

    def predict_market(self, df: pd.DataFrame) -> pd.DataFrame:
        if self.use_market:
            df = df.dropna(subset=["home_moneyline", "away_moneyline"])
            df = self._with_implied(df)
        feats = self._features(df)
        p = self.model_.predict_proba(_x(df, feats))[:, 1]
        return pd.DataFrame({"pred_home_win_prob": p}, index=df.index)


class PlayerPropModel:
    """Multi-target ridge over player's trailing means + opponent allowance,
    with per-target residual sigma so P(over L) = Phi((pred - L) / sigma)."""

    def __init__(self, targets: Optional[list[str]] = None):
        self.targets = targets or VOL_TARGETS + ["def_tackles_solo",
                                                 "def_interceptions", "def_sacks",
                                                 "passing_interceptions",
                                                 "passing_tds", "rushing_tds",
                                                 "receiving_tds"]
        self.models_: dict[str, Pipeline] = {}
        self.sigma_: dict[str, float] = {}
        self.alpha_: dict[str, float] = {}

    def _features(self, df: pd.DataFrame) -> list[str]:
        # every prop target's own trailing mean (incl. TDs + defensive stats so
        # tackle/sack/INT models use the defender's history) + opponent-allowed
        # context for the volume stats.
        prev_any = VOL_TARGETS + ["passing_tds", "rushing_tds", "receiving_tds",
                                  "def_tackles_solo", "def_interceptions",
                                  "def_sacks", "passing_interceptions"]
        feats = ([f"prev_{c}_mean" for c in prev_any]
                 + [f"opp_{c}_allowed_mean" for c in VOL_TARGETS])
        return [f for f in feats if f in df.columns]

    def _tune_alpha(self, x, y) -> float:
        # single split: last 20% of the sorted training data
        n = max(int(len(x) * 0.2), 30)
        if len(x) < 2 * n:
            return 1.0
        xtr, xva = x.iloc[:-n], x.iloc[-n:]
        ytr, yva = y.iloc[:-n], y.iloc[-n:]
        scores = {}
        for a in ALPHA_GRID:
            m = _ridge_pipe(a).fit(xtr, ytr)
            scores[a] = float(np.mean(np.abs(m.predict(xva) - yva)))
        return min(scores, key=scores.get)

    def fit(self, train: pd.DataFrame) -> "PlayerPropModel":
        self.features_ = self._features(train)
        x = train[self.features_].astype(float).fillna(0.0)
        for target in self.targets:
            if target not in train.columns:
                continue
            ok = train[target].notna() & x.notna().all(axis=1)
            if int(ok.sum()) < 80:
                continue
            xx = x[ok]
            yy = train[target][ok].astype(float)
            a = self._tune_alpha(xx, yy)
            m = _ridge_pipe(a).fit(xx, yy)
            self.models_[target] = m
            self.alpha_[target] = a
            sig = float(np.sqrt(np.mean((m.predict(xx) - yy) ** 2)))
            self.sigma_[target] = max(sig, 0.5)
        return self

    def predict(self, df: pd.DataFrame) -> pd.DataFrame:
        x = df[self.features_].astype(float).fillna(0.0)
        out = pd.DataFrame(index=df.index)
        for t, m in self.models_.items():
            pred = m.predict(x)
            out[f"pred_{t}"] = pred
            out[f"sigma_{t}"] = self.sigma_[t]
            # per-row prob over: needs a line; if df has it, emit probability
            if f"line_{t}" in df.columns:
                out[f"prob_over_{t}"] = _phi((pred - df[f"line_{t}"].to_numpy())
                                             / self.sigma_[t])
        return out

"""
Walk-forward backtester with a hard no-leakage contract.

def walk_forward(games_df, build_features_fn, fit_predict_fn, test_seasons):
  for each season t in test_seasons:
    train = games_df[season < t]          (strict)
    test  = games_df[season == t]
    feats_train = build_features_fn(train)  # features computed from PAST only
    feats_test  = build_features_fn(train + test, truncate to season==t)
    model.fit(feats_train)
    preds = model.predict(feats_test)

Because features are .shift(1) lagged means strictly, only data from weeks
< current week feed predictions. Users further verify with
tests/test_no_leakage.py.

Metrics per season: MAE/RMSE for margin + total, logistic Brier + accuracy;
comparison vs the closing market (spread_line, total_line, moneyline).
"""

from __future__ import annotations

from typing import Callable

import numpy as np
import pandas as pd


def brier_score(p: np.ndarray, y: np.ndarray) -> float:
    return float(np.mean((p - y) ** 2))


def walk_forward(games_df: pd.DataFrame,
                 build_features_fn: Callable[[pd.DataFrame], pd.DataFrame],
                 fit_predict_fn: Callable[[pd.DataFrame, pd.DataFrame], pd.DataFrame],
                 test_seasons: range | list[int],
                 min_train_games: int = 512) -> tuple[pd.DataFrame, pd.DataFrame]:
    """
    Returns (predictions_one_row_per_test_game, metrics_per_season).
    fit_predict_fn(train_with_features, test_with_features) -> DataFrame with
    pred_* columns aligned to test rows.
    """
    preds_all: list[pd.DataFrame] = []
    metrics: list[dict] = []
    seasons = sorted(games_df.season.unique())
    for t in test_seasons:
        train = games_df[games_df.season < t]
        test = games_df[games_df.season == t]
        if len(train) < min_train_games or len(test) == 0:
            continue
        # features computed on train-only for train rows, and on all games for
        # test rows but with strictly-lagged stats so no current-week facts leak.
        feats_all = build_features_fn(pd.concat([train, test], ignore_index=True))
        tr = feats_all[feats_all.season < t].copy()
        te = feats_all[feats_all.season == t].copy()
        tr = tr.dropna(subset=["home_score", "away_score"])
        ytest_present = te.dropna(subset=["home_score", "away_score"])
        if len(ytest_present) == 0:
            continue
        pred = fit_predict_fn(tr, ytest_present)
        pred.index = ytest_present.index
        joined = ytest_present.join(pred)
        preds_all.append(joined)

        actual_margin = joined.home_score - joined.away_score
        actual_total = joined.home_score + joined.away_score
        actual_win = (joined.home_score > joined.away_score).astype(float)

        row = {
            "season": int(t),
            "n_games": int(len(joined)),
            "mae_margin": float(np.mean(np.abs(joined.pred_margin - actual_margin))),
            "mae_total": float(np.mean(np.abs(joined.pred_total - actual_total))),
            "rmse_margin": float(np.sqrt(np.mean((joined.pred_margin - actual_margin) ** 2))),
            "rmse_total": float(np.sqrt(np.mean((joined.pred_total - actual_total) ** 2))),
            "brier_win": brier_score(joined.pred_home_prob.to_numpy(), actual_win.to_numpy()),
            "accuracy_win": float(np.mean((joined.pred_home_prob > 0.5).astype(float) == actual_win)),
        }

        # market comparisons (only where closing lines exist)
        if "spread_line" in joined.columns and joined.spread_line.notna().any():
            mkt = joined[joined.spread_line.notna()]
            row["n_market"] = int(len(mkt))
            # nflverse convention: spread_line > 0 means home favored by that
            # many (MARKET expected home margin). Verified vs 2024 results:
            # corr(spread_line, home_margin)=+0.50.
            market_margin = mkt.spread_line
            row["mae_margin_market"] = float(
                np.mean(np.abs(market_margin - (mkt.home_score - mkt.away_score)))
            )
        if "total_line" in joined.columns and joined.total_line.notna().any():
            mkt = joined[joined.total_line.notna()]
            row["mae_total_market"] = float(
                np.mean(np.abs(mkt.total_line - (mkt.home_score + mkt.away_score)))
            )
        metrics.append(row)
    return pd.concat(preds_all, ignore_index=True), pd.DataFrame(metrics)

"""
Per-market walk-forward backtest. Shows the work per market:

  SPREAD    -- ridge margin; MAE vs closing spread; ATS pick accuracy;
               P(cover) calibration (binned, model vs empirical)
  TOTAL     -- ridge total; MAE vs closing total; O/U pick accuracy;
               P(over) calibration
  MONEYLINE -- logistic win prob; Brier; accuracy; calibration
  PROPS     -- per-stat ridge on player trailing means + opp allowance;
               MAE, corr, P(over own-mean) calibration per position

Each market runs BOTH a pure-model variant and a MARKET-HYBRID variant that
includes closing lines as features (legal pregame inputs, explicitly labeled),
so we can measure our marginal value-add OVER the market vs just matching it.

Run:  .venv\\Scripts\\python scripts\\backtest_markets.py
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import numpy as np
import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))

from forecaster.data_loader import load_schedules, load_player_week
from forecaster.features import (build_game_matrix, build_weekly_team_stats,
                             player_volume_matrix, VOL_TARGETS)
from forecaster.market_models import (SpreadModel, TotalModel, MoneylineModel,
                                  PlayerPropModel, _phi)

TEST_SEASONS = list(range(2004, 2026))       # full modern era: spreads/totals exist 1999+; moneylines 2006+
PROP_TEST_SEASONS = list(range(2004, 2026))  # props evaluated vs own trailing mean (no historical prop lines exist)
OUTDIR = Path(__file__).resolve().parent.parent / "results"


def calibration_bins(prob: pd.Series, hit: pd.Series, bins=None) -> pd.DataFrame:
    """Empirical hit rate inside probability decile bins = calibration curve."""
    bins = bins or [0.0, .3, .4, .45, .5, .55, .6, .7, 1.0]
    df = pd.DataFrame({"p": prob, "hit": hit}).dropna()
    df["bin"] = pd.cut(df.p, bins, include_lowest=True)
    out = df.groupby("bin", observed=True).agg(
        n=("hit", "size"), avg_pred=("p", "mean"), empirical=("hit", "mean"))
    return out.reset_index().astype({"n": int, "avg_pred": float, "empirical": float})


_MAT_CACHE: dict = {}  # full feature matrix is built ONCE (leakage-safe:
# every form feature is intra-(team,season) .shift(1)-lagged, so a season-t
# row can never be made from season->=t information)


def _full_mat(schedules, team_stats) -> pd.DataFrame:
    if "mat" not in _MAT_CACHE:
        _MAT_CACHE["mat"] = build_game_matrix(schedules, team_stats)
    return _MAT_CACHE["mat"]


def walk_forward_market(schedules, team_stats, model_cls, use_market,
                        feature_hybrid_note, min_train=200):
    rows, chosen_params = [], []
    mat = _full_mat(schedules, team_stats)
    for t in TEST_SEASONS:
        tr = mat[mat.season < t].dropna(subset=["home_score", "away_score"])
        te = mat[(mat.season == t) & mat.home_score.notna()]
        if len(tr) < min_train or len(te) == 0:
            continue
        if use_market and model_cls is MoneylineModel:
            # hybrid moneyline needs moneylines in TRAIN; pre-2006 seasons have none
            tr = tr.dropna(subset=["home_moneyline", "away_moneyline"])
            if len(tr) < min_train:
                continue
        m = model_cls(use_market=use_market).fit(tr)
        pred = m.predict_market(te)
        te = te.join(pred)
        # hybrid moneyline drops rows without moneylines; guard empty folds
        first_pred = next((c for c in te.columns if c.startswith("pred_")), None)
        if first_pred is None:
            continue
        te = te.dropna(subset=[first_pred])
        if len(te) == 0:
            continue
        actual_margin = te.home_score - te.away_score
        actual_total = te.home_score + te.away_score
        row = {"season": int(t), "n": int(len(te)),
               "alpha": getattr(m, "alpha_", getattr(m, "C_", None))}
        chosen_params.append(row)

        # ---- spread metrics ----
        if isinstance(m, SpreadModel):
            row["mae_margin"] = float(np.mean(np.abs(te.pred_margin - actual_margin)))
            have_line = te.spread_line.notna()
            te_l = te[have_line]
            if len(te_l) > 0:
                act_m = te_l.home_score - te_l.away_score
                push = np.isclose(act_m, te_l.spread_line)
                home_cover = (act_m > te_l.spread_line) & ~push
                pick_home = te_l.prob_home_covers > 0.5
                wins = (pick_home == home_cover) & ~push
                row["n_atspicks"] = int((~push).sum())
                row["ats_pick_rate"] = float(wins.mean())
                row["hold_vs_spread"] = float(
                    np.mean(np.abs(te_l.pred_margin - act_m))
                    - np.mean(np.abs(te_l.spread_line - act_m)))
                row["_calib"] = calibration_bins(te_l.prob_home_covers, home_cover)
        # ---- totals metrics ----
        if isinstance(m, TotalModel):
            row["mae_total"] = float(np.mean(np.abs(te.pred_total - actual_total)))
            have_line = te.total_line.notna()
            te_l = te[have_line]
            if len(te_l) > 0:
                act_t = te_l.home_score + te_l.away_score
                push = np.isclose(act_t, te_l.total_line)
                over = (act_t > te_l.total_line) & ~push
                pick_over = te_l.prob_over > 0.5
                wins = (pick_over == over) & ~push
                row["n_oupicks"] = int((~push).sum())
                row["ou_pick_rate"] = float(wins.mean())
                row["_calib"] = calibration_bins(te_l.prob_over, over)
        # ---- moneyline metrics ----
        if isinstance(m, MoneylineModel):
            y = (te.home_score > te.away_score).astype(float)
            row["brier"] = float(np.mean((te.pred_home_win_prob - y) ** 2))
            row["pick_rate"] = float(((te.pred_home_win_prob > .5).astype(float) == y).mean())
            row["_calib"] = calibration_bins(te.pred_home_win_prob, y)
        rows.append(row)
    return pd.DataFrame(rows), chosen_params


def props_backtest(schedules, team_stats, player_week):
    """Player props walk-forward over PROP_TEST_SEASONS.

    ONE feature matrix built once (leakage-safe: every prev_*_mean is an
    expanding mean shifted 1 week within (player, season)), then per test
    season t the model fits on seasons < t only and is evaluated on season t.
    Replaces the old per-fold reload (O(n^2) parquet reads) with a slice.
    """
    mat = player_volume_matrix(player_week, team_stats, schedules)
    out_rows = []
    for t in PROP_TEST_SEASONS:
        tr = mat[mat.season < t]
        te = mat[mat.season == t]
        if len(tr) < 200 or len(te) < 20:
            continue
        m = PlayerPropModel().fit(tr)
        pr = m.predict(te)
        joined = te.join(pr)

        for pos_label, pos_groups, pos_targets in (
            ("QB", ["QB"], ["attempts", "completions", "passing_yards",
                            "passing_tds", "passing_interceptions"]),
            ("RB", ["RB"], ["carries", "rushing_yards", "rushing_tds", "receptions"]),
            ("WR", ["WR"], ["receptions", "targets", "receiving_yards", "receiving_tds"]),
            ("TE", ["TE"], ["receptions", "targets", "receiving_yards"]),
            ("DEF (LB/DB/DL)", ["LB", "DB", "DL"],
             ["def_tackles_solo", "def_interceptions", "def_sacks"]),
        ):
            df = joined[joined.position_group.isin(pos_groups)] if "position_group" in joined else joined
            for target in pos_targets:
                col = f"pred_{target}"
                if col not in joined.columns or target not in df.columns:
                    continue
                sub = df.dropna(subset=[target, col]).dropna(subset=[f"prev_{target}_mean"])
                if len(sub) < 30:
                    continue
                pos = pos_label
                mae = float(np.mean(np.abs(sub[col] - sub[target])))
                sig = m.sigma_.get(target, 1.0)
                # realistic outcome: did the player go OVER their own trailing
                # mean? (proxy for a posted prop line of "season-to-date mean")
                over = (sub[target] > sub[f"prev_{target}_mean"]).astype(float)
                prob_over = _phi((sub[col] - sub[f"prev_{target}_mean"]) / sig)
                out_rows.append({
                    "season": int(t), "position": pos, "stat": target,
                    "n": int(len(sub)), "mae": mae,
                    "corr": float(np.corrcoef(sub[col], sub[target])[0, 1]),
                    "sigma": sig,
                    "alpha": m.alpha_.get(target),
                    "avg_prob_over": float(prob_over.mean()),
                    "empirical_over": float(over.mean()),
                })
    return pd.DataFrame(out_rows)


# output filenames: datetime-stamped + descriptive, one folder per run so a
# file on disk always says WHAT it is and WHEN it was produced. No guessing.
MARKET_FILENAMES = {
    "spread":            "NFL_SPREAD_MODEL_PURE_NO_MARKET_INPUTS_WALKFORWARD_METRICS_2004_2025.csv",
    "spread_hybrid":     "NFL_SPREAD_MODEL_HYBRID_WITH_CLOSING_SPREAD_LINE_WALKFORWARD_METRICS_2004_2025.csv",
    "total":             "NFL_TOTAL_MODEL_PURE_NO_MARKET_INPUTS_WALKFORWARD_METRICS_2004_2025.csv",
    "total_hybrid":      "NFL_TOTAL_MODEL_HYBRID_WITH_CLOSING_TOTAL_LINE_WALKFORWARD_METRICS_2004_2025.csv",
    "moneyline":         "NFL_MONEYLINE_MODEL_PURE_LOGISTIC_WALKFORWARD_METRICS_2004_2025.csv",
    "moneyline_hybrid":  "NFL_MONEYLINE_MODEL_HYBRID_WITH_CLOSING_MONEYLINES_WALKFORWARD_METRICS_2004_2025.csv",
}


def main():
    from datetime import datetime
    run_stamp = datetime.now().strftime("%Y%m%d_%H%M")
    outdir = OUTDIR / f"{run_stamp}_NFL_OUTCOME_MARKETS_WALKFORWARD_2004_2025_RUN"
    print("loading data...", flush=True)
    schedules = load_schedules()
    player_week = load_player_week(list(range(1999, 2027)))
    print(f"  schedules={len(schedules)}, player_week={len(player_week)}", flush=True)
    team_stats = build_weekly_team_stats(player_week)

    outdir.mkdir(parents=True, exist_ok=True)
    report = {}

    for market, cls, use_market, note in (
        ("spread", SpreadModel, False, "pure model (no market inputs)"),
        ("spread_hybrid", SpreadModel, True, "adds closing spread_line"),
        ("total", TotalModel, False, "pure model"),
        ("total_hybrid", TotalModel, True, "adds closing total_line"),
        ("moneyline", MoneylineModel, False, "logistic win prob"),
        ("moneyline_hybrid", MoneylineModel, True, "adds closing moneylines"),
    ):
        print(f"\n=== {market} ({note}) ===", flush=True)
        df, params = walk_forward_market(schedules, team_stats, cls, use_market, note)
        df.to_csv(outdir / f"{run_stamp}_{MARKET_FILENAMES[market]}", index=False)
        show = df[[c for c in df.columns if not c.startswith("_calib")]]
        print(show.to_string(index=False))
        report[market] = df.to_dict("records")

    print("\n=== player props (2004-25) ===", flush=True)
    import gc
    _MAT_CACHE.clear()       # team matrices no longer needed (RAM)
    props = props_backtest(schedules, team_stats, player_week)
    del player_week
    gc.collect()
    if len(props):
        props.to_csv(
            outdir / f"{run_stamp}_NFL_PLAYER_PROPS_RIDGE_BY_POSITION_AND_STAT_WALKFORWARD_METRICS_2004_2025.csv",
            index=False)
        print(props.round(3).to_string(index=False))
        report["props"] = props.to_dict("records")

    (outdir / f"{run_stamp}_NFL_ALL_MARKETS_WALKFORWARD_FULL_JSON_REPORT.json").write_text(
        json.dumps(report, indent=2, default=str))
    print(f"\nartifacts -> {outdir}")


if __name__ == "__main__":
    main()

"""
Headless math-check harness (GATE 2) -- the same live checks the local math
page ran, so the cloud build proves the published math is the math that runs:

  A: leakage contract  - team_form_features week-N prev contains NO week-N data
  B: Phi numerics      - Phi(0)=0.5, Phi(1)+Phi(-1)=1, Phi(0.2)=0.5792597
  C: cumsum identity   - vectorized (cumsum-x)/count == expanding().mean().shift(1)
  D: spread sign       - corr(spread_line, home margin) > 0 on real schedules
                         (SKIP w/ notice if data not fetched -- CI fetches first)

Exit code 0 only if every executed check PASSes. Prints PASS/FAIL lines.
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from forecaster import features as F          # noqa: E402
from forecaster import market_models as MM    # noqa: E402

results: list[tuple[str, bool]] = []


def record(name: str, ok: bool, detail: str) -> None:
    results.append((name, ok))
    print(f"{'PASS' if ok else 'FAIL'} -- {name} -- {detail}", flush=True)


def check_a_leakage() -> None:
    games = pd.DataFrame({
        "season": [2022, 2022, 2022],
        "week": [1, 2, 3],
        "team": ["AAA"] * 3, "opp": ["BBB", "CCC", "DDD"],
        "points_for": [30.0, 10.0, 20.0],
        "points_against": [10.0, 40.0, 20.0],
    })
    stats = pd.DataFrame({
        "season": [2022] * 3, "week": [1, 2, 3], "team": ["AAA"] * 3,
        "passing_epa": [5.0, -2.0, 1.0], "rushing_epa": [1.0, 0.0, 0.0],
        "passing_yards": [300.0, 150.0, 200.0],
        "rushing_yards": [100.0, 50.0, 90.0],
        "attempts": [40.0, 25.0, 30.0], "carries": [25.0, 15.0, 20.0],
        "passing_interceptions": [1.0, 2.0, 0.0], "takeaways_int": [0.0, 1.0, 1.0],
    })
    form = F.team_form_features(games, stats)
    aaa = form[form.team == "AAA"].sort_values("week")
    ok = (pd.isna(aaa.iloc[0].prev_points_diff)
          and float(aaa.iloc[1].prev_points_diff) == 20.0
          and float(aaa.iloc[2].prev_points_diff) == -30.0)
    record("A. leakage contract (shift(1))", bool(ok),
           "week1 NaN; week2=+20 (30-10); week3=-30 (10-40)")


def check_b_phi() -> None:
    phi0 = float(MM._phi(np.array([0.0]))[0])
    z1 = float(MM._phi(np.array([1.0]))[0])
    z2 = float(MM._phi(np.array([-1.0]))[0])
    p = float(MM._phi(np.array([(3.0 - 1.0) / 10.0]))[0])
    ok = abs(phi0 - 0.5) < 1e-12 and abs(z1 + z2 - 1.0) < 1e-12 and abs(p - 0.5792597) < 1e-6
    record("B. Phi numerics", ok,
           f"Phi(0)={phi0:.12f}; Phi(1)+Phi(-1)={z1 + z2:.12f}; Phi(0.2)={p:.7f}")


def check_c_cumsum_identity() -> None:
    rng = np.random.default_rng(7)
    pw = pd.DataFrame({
        "player_id": ["p1"] * 6 + ["p2"] * 4,
        "season": [2024] * 10,
        "week": [1, 2, 3, 4, 5, 6, 1, 2, 3, 4],
        "attempts": rng.integers(10, 50, 10).astype(float),
    }).sort_values(["player_id", "week"]).reset_index(drop=True)
    ref = (pw.groupby(["player_id", "season"], sort=False)["attempts"]
             .transform(lambda s: s.expanding().mean().shift(1)))
    grp = pw.groupby(["player_id", "season"], sort=False)
    csum = grp["attempts"].transform("cumsum")
    n_before = grp.cumcount()
    got = (csum - pw["attempts"]).div(n_before.where(n_before > 0), axis=0)
    maxdiff = float(np.nanmax(np.abs(ref.to_numpy(float) - got.to_numpy(float))))
    record("C. cumsum identity == expanding().mean().shift(1)",
           bool(np.allclose(ref, got, equal_nan=True)),
           f"max abs diff {maxdiff:.2e}")


def check_d_sign() -> None:
    from forecaster.data_loader import DATA_DIR, SCHEDULES_GLOB
    path = DATA_DIR / SCHEDULES_GLOB
    if not path.exists():
        print("SKIP -- D. spread sign -- schedules parquet not fetched yet", flush=True)
        return
    from forecaster.data_loader import load_schedules
    s = load_schedules().dropna(subset=["spread_line", "home_score", "away_score"])
    margin = s.home_score - s.away_score
    corr = float(np.corrcoef(s.spread_line.astype(float), margin)[0, 1])
    mae = float(np.mean(np.abs(s.spread_line - margin)))
    record("D. spread_line sign on real data", corr > 0,
           f"corr={corr:+.3f}; MAE={mae:.2f}")


def main() -> int:
    check_a_leakage()
    check_b_phi()
    check_c_cumsum_identity()
    check_d_sign()
    failed = [n for n, ok in results if not ok]
    print(f"\nmath_checks: {len(results) - len(failed)}/{len(results)} passed", flush=True)
    if failed:
        print("FAILED: " + "; ".join(failed), flush=True)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())