from __future__ import annotations import hashlib import json from datetime import UTC, datetime from pathlib import Path from typing import Any import numpy as np import pandas as pd from pydantic import BaseModel from .analysis.historical_weekly_projection import ( POSITIONS, build_player_week_examples, evaluation_seasons, feature_columns, load_weekly_stats, ) from .evaluation_trace import EvaluationGateResult, EvaluationMetric, EvaluationRunRecord from .model_improvement import apply_promotion from .model_platform import ( ActiveModelChampion, ChampionDecision, EvaluationSummary, FeatureDefinition, FeatureSnapshot, ModelComponentSpec, ModelConfiguration, ModelExperimentRegistration, SignalCandidate, ) from .models import utc_now from .paths import RESEARCH_ROOT from .sandbox import sha256_file from .weather import GameWeatherFeatureSnapshot, WeatherForecastSnapshot ANALYSIS_VERSION = "1.1.0" STUDY_ID = "weather-model-challenger" SIGNAL_ID = "cutoff-safe-pregame-weather" FORECAST_HORIZON_HOURS = 6 MINIMUM_GAMES = 1_000 MINIMUM_CONDITION_GAMES = 100 MINIMUM_IMPROVEMENT_PCT = 0.10 MINIMUM_WINNING_WINDOWS = 4 MAXIMUM_POSITION_REGRESSION_PCT = 1.50 INTERVAL_COVERAGE_MIN = 0.70 INTERVAL_COVERAGE_MAX = 0.90 WEATHER_RIDGE_PENALTY = 80.0 WIND_THRESHOLD_MPS = 6.7056 # 15 mph BOOTSTRAP_REPETITIONS = 4_000 WEATHER_FEATURES = ( "weather_air_temp_c", "weather_cold_degree_c", "weather_wind_speed_mps", "weather_high_wind_mps", "weather_precip_probability", "weather_wind_precip_interaction", ) WEEKLY_CHAMPION_STATE_PATH = RESEARCH_ROOT / "state" / "weekly-model-champion.json" ACTIVE_WEEKLY_DECISION_PATH = RESEARCH_ROOT / "state" / "active-weekly-champion-decision.json" ACTIVE_WEATHER_PARAMETERS_PATH = RESEARCH_ROOT / "state" / "active-weather-adjustment.json" PUBLIC_DECISION_HREF = "/model-artifacts/weather-model-v1/champion-decision.json" def _canonical(value: Any) -> Any: if isinstance(value, BaseModel): return _canonical(value.model_dump(mode="json")) if isinstance(value, dict): return {key: _canonical(child) for key, child in sorted(value.items())} if isinstance(value, (list, tuple)): return [_canonical(child) for child in value] if isinstance(value, (float, np.floating)): return round(float(value), 10) if isinstance(value, np.integer): return int(value) return value def _hash(value: Any) -> str: encoded = json.dumps(_canonical(value), sort_keys=True, separators=(",", ":"), allow_nan=False) return hashlib.sha256(encoded.encode()).hexdigest() def _write_json(path: Path, value: Any) -> None: if hasattr(value, "model_dump"): value = value.model_dump(mode="json") path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(_canonical(value), indent=2, sort_keys=True, allow_nan=False) + "\n", encoding="utf-8", ) def baseline_weekly_champion() -> ActiveModelChampion: return ActiveModelChampion( configuration=ModelConfiguration( model_id="weekly-ppr-ridge", version="0.2.0", feature_families=[ "prior-fantasy-output", "passing-volume", "rushing-volume", "receiving-volume", ], position_penalties={position: 35 for position in POSITIONS}, source_signal_ids=[], ), activated_at="2026-09-02T00:00:00Z", activation_basis="baseline-registration", ) def load_weekly_champion(path: Path = WEEKLY_CHAMPION_STATE_PATH) -> ActiveModelChampion: if not path.is_file(): return baseline_weekly_champion() return ActiveModelChampion.model_validate_json(path.read_text(encoding="utf-8")) def _registration( active: ActiveModelChampion, *, test_seasons: list[int], locked_at: str ) -> ModelExperimentRegistration: champion = active.configuration major, minor, patch = (int(value) for value in champion.version.split(".")) challenger = champion.model_copy( update={ "version": f"{major}.{minor}.{patch + 1}", "source_signal_ids": sorted({*champion.source_signal_ids, SIGNAL_ID}), } ) return ModelExperimentRegistration( experiment_id=f"model-exp-weather-team-environment-through-{test_seasons[-1]}", proposal_sha256=_hash( { "analysis_version": ANALYSIS_VERSION, "study_id": STUDY_ID, "forecast_horizon_hours": FORECAST_HORIZON_HOURS, "weather_features": WEATHER_FEATURES, "weather_ridge_penalty": WEATHER_RIDGE_PENALTY, } ), champion=champion, challenger=challenger, test_seasons=test_seasons, minimum_improvement_pct=MINIMUM_IMPROVEMENT_PCT, minimum_winning_windows=MINIMUM_WINNING_WINDOWS, maximum_position_regression_pct=MAXIMUM_POSITION_REGRESSION_PCT, required_interval_coverage_min=INTERVAL_COVERAGE_MIN, required_interval_coverage_max=INTERVAL_COVERAGE_MAX, locked_at=locked_at, ) def _feature_definitions() -> list[FeatureDefinition]: common = { "signal_id": SIGNAL_ID, "version": ANALYSIS_VERSION, "grain": "game", "observed_at_rule": ( "Use only a prediction-eligible forecast whose source availability and issue " "timestamps precede the registered six-hour decision cutoff." ), "missingness_rule": ( "Exclude the game from both champion and challenger denominators when a required " "pregame forecast field is missing; never replace missing weather with zero." ), "leakage_risk": "high", "source_keys": ["noaa-hrrr-archive"], "eligible_positions": list(POSITIONS), } return [ FeatureDefinition( feature_id="pregame-air-temperature", name="Pregame air temperature", definition="Six-hour-cutoff HRRR game-window air temperature at the stadium grid point.", units="degrees Celsius", valid_min=-80, valid_max=65, **common, ), FeatureDefinition( feature_id="pregame-wind", name="Pregame wind", definition="Six-hour-cutoff HRRR game-window sustained wind and the excess above 15 mph.", units="meters per second", valid_min=0, valid_max=100, **common, ), FeatureDefinition( feature_id="pregame-precipitation", name="Pregame precipitation", definition="Six-hour-cutoff HRRR precipitation probability and its interaction with wind.", units="probability and interaction", valid_min=0, **common, ), ] def _signal(status: str = "registered", decision_href: str | None = None) -> SignalCandidate: return SignalCandidate( signal_id=SIGNAL_ID, research_run_id="weather-forecast-skill", hypothesis=( "Cutoff-safe stadium weather forecasts add held-out weekly fantasy projection " "accuracy beyond the identical trailing-opportunity champion." ), target_components=["team-environment", "calibration"], source_keys=["noaa-hrrr-archive", "nflverse-weekly-player-stats"], rights_status="approved", prediction_time_available=True, missingness_rule=( "A missing or ineligible forecast removes that game from both compared models; " "current rankings receive no weather adjustment for an unsafe game." ), expected_direction="Weather should reduce held-out weekly PPR MAE without harming a position.", evaluation_plan=( "Use six rolling future-season windows from 2020 through 2025, identical player-week " "denominators, every fantasy position, calibrated intervals, and exact repeat execution." ), status=status, # type: ignore[arg-type] champion_decision_href=decision_href, created_at="2026-09-02T00:00:00Z", ) def forecast_readiness(hrrr_dir: Path) -> dict[str, Any]: evaluation_path = hrrr_dir / "evaluation-run.json" scorecard_path = hrrr_dir / "scorecard.json" snapshots_path = hrrr_dir / "snapshots.json" if not all(path.is_file() for path in (evaluation_path, scorecard_path, snapshots_path)): return { "ready": False, "reason": "The archived forecast backtest has not produced its required evidence.", "completed_forecasts": 0, "coverage": 0.0, "six_hour_games": 0, "condition_games": 0, "latest_forecast_season": None, "source_sha256": {}, } evaluation = EvaluationRunRecord.model_validate_json( evaluation_path.read_text(encoding="utf-8") ) scorecard = json.loads(scorecard_path.read_text(encoding="utf-8")) snapshots = _load_six_hour_snapshots(snapshots_path) condition_games = sum( (snapshot.wind_speed_mps or 0) >= WIND_THRESHOLD_MPS or (snapshot.precip_probability or 0) >= 0.5 for snapshot in snapshots ) ready = ( evaluation.status == "passed" and int(scorecard["completed_forecasts"]) >= MINIMUM_GAMES and float(scorecard["coverage"]) >= 0.85 and len(snapshots) >= MINIMUM_GAMES and condition_games >= MINIMUM_CONDITION_GAMES ) return { "ready": ready, "reason": ( "Every registered forecast-readiness threshold passed." if ready else "The weather challenger remains blocked on forecast sample, coverage, version, or condition support." ), "forecast_backtest_status": evaluation.status, "completed_forecasts": int(scorecard["completed_forecasts"]), "coverage": float(scorecard["coverage"]), "six_hour_games": len(snapshots), "condition_games": condition_games, "latest_forecast_season": max( (int(snapshot.game_id[:4]) for snapshot in snapshots), default=None ), "source_sha256": { "hrrr-evaluation": sha256_file(evaluation_path), "hrrr-scorecard": sha256_file(scorecard_path), "hrrr-snapshots": sha256_file(snapshots_path), }, } def _load_six_hour_snapshots(path: Path) -> list[WeatherForecastSnapshot]: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, list): raise TypeError("Archived weather snapshots must be a JSON list") selected = [] for item in value: snapshot = WeatherForecastSnapshot.model_validate(item) cutoff = datetime.fromisoformat(snapshot.cutoff_at).astimezone(UTC) valid = datetime.fromisoformat(snapshot.valid_at).astimezone(UTC) horizon = round((valid - cutoff).total_seconds() / 3600) if horizon == FORECAST_HORIZON_HOURS: selected.append(snapshot) if len({snapshot.game_id for snapshot in selected}) != len(selected): raise ValueError("Six-hour weather snapshots repeat a game") return sorted(selected, key=lambda item: item.game_id) def attach_weather_features( examples: pd.DataFrame, snapshots: list[WeatherForecastSnapshot] ) -> pd.DataFrame: rows = [] for snapshot in snapshots: _, week, away, home = snapshot.game_id.split("_", 3) values = { "game_id": snapshot.game_id, "season": int(snapshot.game_id[:4]), "week": int(week), "weather_air_temp_c": snapshot.air_temp_c, "weather_wind_speed_mps": snapshot.wind_speed_mps, "weather_precip_probability": snapshot.precip_probability, "cutoff_at": snapshot.cutoff_at, "valid_at": snapshot.valid_at, "source_available_at": snapshot.source_available_at, } for team in (away, home): rows.append(values | {"team": team}) weather = pd.DataFrame(rows) if weather.empty: raise ValueError("No six-hour weather forecasts are available") if weather.duplicated(["season", "week", "team"]).any(): raise ValueError("Weather forecasts repeat a season-week-team key") joined = examples.merge( weather, on=["season", "week", "team"], how="inner", validate="many_to_one", ) required = [ "weather_air_temp_c", "weather_wind_speed_mps", "weather_precip_probability", ] joined = joined.dropna(subset=required).copy() joined["weather_cold_degree_c"] = np.maximum(0, 4 - joined["weather_air_temp_c"]) joined["weather_high_wind_mps"] = np.maximum( 0, joined["weather_wind_speed_mps"] - WIND_THRESHOLD_MPS ) joined["weather_wind_precip_interaction"] = ( joined["weather_wind_speed_mps"] * joined["weather_precip_probability"] ) if joined.empty: raise ValueError("Forecast missingness removed every player-week") return joined def _fit_ridge(frame: pd.DataFrame, columns: list[str], target: str, penalty: float) -> dict: x = frame[columns].to_numpy(float) y = frame[target].to_numpy(float) mean = x.mean(axis=0) scale = x.std(axis=0) scale[scale == 0] = 1.0 z = (x - mean) / scale beta = np.linalg.solve(z.T @ z + penalty * np.eye(len(columns)), z.T @ (y - y.mean())) return { "features": columns, "mean": mean, "scale": scale, "beta": beta, "intercept": float(y.mean()), "training_rows": len(frame), "penalty": penalty, } def _predict(frame: pd.DataFrame, parameters: dict) -> np.ndarray: x = frame[list(parameters["features"])].to_numpy(float) z = (x - parameters["mean"]) / parameters["scale"] return float(parameters["intercept"]) + z @ parameters["beta"] def walk_forward_weather_challenger( examples: pd.DataFrame, ) -> tuple[pd.DataFrame, dict[str, Any]]: predictions = [] parameters: dict[str, Any] = {} seasons = evaluation_seasons(examples) for test_season in seasons: train = examples[examples["season"] < test_season] test = examples[examples["season"].eq(test_season)] if train.empty or test.empty: raise ValueError(f"Missing weather training or test examples for {test_season}") parameters[str(test_season)] = {} for position in POSITIONS: position_train = train[train["position"].eq(position)].copy() position_test = test[test["position"].eq(position)].copy() if len(position_train) < 20 or position_test.empty: raise ValueError(f"Insufficient {position} weather rows for {test_season}") base_columns = feature_columns(position) base = _fit_ridge(position_train, base_columns, "actual_points", 35.0) train_champion = np.clip(_predict(position_train, base), 0, None) test_champion = np.clip(_predict(position_test, base), 0, None) position_train["weather_residual"] = ( position_train["actual_points"].to_numpy(float) - train_champion ) adjustment = _fit_ridge( position_train, list(WEATHER_FEATURES), "weather_residual", WEATHER_RIDGE_PENALTY, ) train_challenger = np.clip( train_champion + _predict(position_train, adjustment), 0, None ) test_challenger = np.clip(test_champion + _predict(position_test, adjustment), 0, None) champion_residual = position_train["actual_points"].to_numpy(float) - train_champion challenger_residual = position_train["actual_points"].to_numpy(float) - train_challenger position_test["champion_points"] = test_champion position_test["challenger_points"] = test_challenger position_test["champion_low"] = np.clip( test_champion + np.quantile(champion_residual, 0.10), 0, None ) position_test["champion_high"] = np.clip( test_champion + np.quantile(champion_residual, 0.90), 0, None ) position_test["challenger_low"] = np.clip( test_challenger + np.quantile(challenger_residual, 0.10), 0, None ) position_test["challenger_high"] = np.clip( test_challenger + np.quantile(challenger_residual, 0.90), 0, None ) predictions.append(position_test) parameters[str(test_season)][position] = _public_parameters(adjustment) full_parameters = {} for position in POSITIONS: position_rows = examples[examples["position"].eq(position)].copy() base = _fit_ridge(position_rows, feature_columns(position), "actual_points", 35.0) champion = np.clip(_predict(position_rows, base), 0, None) position_rows["weather_residual"] = ( position_rows["actual_points"].to_numpy(float) - champion ) full_parameters[position] = _public_parameters( _fit_ridge( position_rows, list(WEATHER_FEATURES), "weather_residual", WEATHER_RIDGE_PENALTY, ) ) return pd.concat(predictions, ignore_index=True), { "folds": parameters, "active_fit": full_parameters, } def _public_parameters(parameters: dict) -> dict[str, Any]: return { "features": list(parameters["features"]), "mean": np.asarray(parameters["mean"]).astype(float).tolist(), "scale": np.asarray(parameters["scale"]).astype(float).tolist(), "beta": np.asarray(parameters["beta"]).astype(float).tolist(), "intercept": float(parameters["intercept"]), "training_rows": int(parameters["training_rows"]), "ridge_penalty": float(parameters["penalty"]), } def apply_active_weather_adjustment( frame: pd.DataFrame, *, season: int, week: int, weekly_champion_state_path: Path = WEEKLY_CHAMPION_STATE_PATH, active_parameters_path: Path = ACTIVE_WEATHER_PARAMETERS_PATH, feature_snapshots_path: Path = ( RESEARCH_ROOT.parent / "artifacts" / "weather-current-snapshots" / "weather-current-v1" / "feature-snapshots.json" ), ) -> tuple[pd.DataFrame, dict[str, Any]]: """Apply an earned weather adjustment; unsafe or absent games stay unchanged.""" active = load_weekly_champion(weekly_champion_state_path) if SIGNAL_ID not in active.configuration.source_signal_ids: return frame.copy(), { "status": "not-promoted", "model_version": active.configuration.version, "adjusted_players": 0, "eligible_games": 0, } if not active_parameters_path.is_file(): raise ValueError("The promoted weather champion is missing active parameters") payload = json.loads(active_parameters_path.read_text(encoding="utf-8")) if ( payload.get("model_version") != active.configuration.version or payload.get("signal_id") != SIGNAL_ID or set(payload.get("positions", {})) != set(POSITIONS) ): raise ValueError("Active weather parameters disagree with the weekly champion") if not feature_snapshots_path.is_file(): return frame.copy(), { "status": "no-current-snapshots", "model_version": active.configuration.version, "adjusted_players": 0, "eligible_games": 0, "source_sha256": { "weekly-weather-champion": sha256_file(weekly_champion_state_path), "weather-adjustment-parameters": sha256_file(active_parameters_path), }, } value = json.loads(feature_snapshots_path.read_text(encoding="utf-8")) if not isinstance(value, list): raise TypeError("Current weather feature snapshots must be a JSON list") selected: dict[str, tuple[int, GameWeatherFeatureSnapshot]] = {} prefix = f"{season}_{week:02d}_" for item in value: snapshot = GameWeatherFeatureSnapshot.model_validate(item) if not snapshot.game_id.startswith(prefix) or not snapshot.prediction_eligible: continue horizon = int(snapshot.snapshot_id.rsplit("-", 1)[-1].removesuffix("h")) current = selected.get(snapshot.game_id) if current is None or horizon < current[0]: selected[snapshot.game_id] = (horizon, snapshot) team_weather: dict[str, GameWeatherFeatureSnapshot] = {} for _, snapshot in selected.values(): _, _, away, home = snapshot.game_id.split("_", 3) team_weather[away] = snapshot team_weather[home] = snapshot adjusted = frame.copy() adjusted_count = 0 for index, row in adjusted.iterrows(): snapshot = team_weather.get(str(row["team"])) if snapshot is None: continue required = ( snapshot.air_temp_c, snapshot.wind_speed_mps, snapshot.precip_probability, ) if any(value is None for value in required): continue air_temp = float(snapshot.air_temp_c) wind = float(snapshot.wind_speed_mps) precip = float(snapshot.precip_probability) feature_values = np.asarray( [ air_temp, max(0, 4 - air_temp), wind, max(0, wind - WIND_THRESHOLD_MPS), precip, wind * precip, ] ) parameters = payload["positions"][str(row["position"])] if parameters.get("features") != list(WEATHER_FEATURES): raise ValueError("Active weather parameter features changed order") mean = np.asarray(parameters["mean"], dtype=float) scale = np.asarray(parameters["scale"], dtype=float) beta = np.asarray(parameters["beta"], dtype=float) if not (len(mean) == len(scale) == len(beta) == len(feature_values)) or (scale <= 0).any(): raise ValueError("Active weather parameters have invalid dimensions") delta = float(parameters["intercept"] + ((feature_values - mean) / scale) @ beta) for column in ("expected_points", "conditional_points", "low_points", "high_points"): adjusted.loc[index, column] = max(0.0, float(row[column]) + delta) adjusted_count += 1 return adjusted, { "status": "applied" if adjusted_count else "no-eligible-current-games", "model_version": active.configuration.version, "adjusted_players": adjusted_count, "eligible_games": len(selected), "forecast_horizon_hours": min((horizon for horizon, _ in selected.values()), default=None), "as_of": max((snapshot.as_of for _, snapshot in selected.values()), default=None), "decision_href": active.champion_decision_href, "source_sha256": { "weekly-weather-champion": sha256_file(weekly_champion_state_path), "weather-adjustment-parameters": sha256_file(active_parameters_path), "current-weather-features": sha256_file(feature_snapshots_path), }, } def _metric(frame: pd.DataFrame, prefix: str) -> dict[str, Any]: actual = frame["actual_points"].to_numpy(float) predicted = frame[f"{prefix}_points"].to_numpy(float) correlations = [] for _, week in frame.groupby(["season", "week", "position"]): correlation = week[f"{prefix}_points"].rank().corr(week["actual_points"].rank()) if pd.notna(correlation): correlations.append(float(correlation)) return { "rows": len(frame), "model_mae": float(np.abs(actual - predicted).mean()), "interval_80_coverage": float( ( (frame["actual_points"] >= frame[f"{prefix}_low"]) & (frame["actual_points"] <= frame[f"{prefix}_high"]) ).mean() ), "mean_weekly_spearman": float(np.mean(correlations)) if correlations else 0.0, } def build_weather_scorecard(predictions: pd.DataFrame) -> dict[str, Any]: return { "overall": { "champion": _metric(predictions, "champion"), "challenger": _metric(predictions, "challenger"), }, "by_season": { str(int(season)): { "champion": _metric(frame, "champion"), "challenger": _metric(frame, "challenger"), } for season, frame in predictions.groupby("season") }, "by_position": { str(position): { "champion": _metric(frame, "champion"), "challenger": _metric(frame, "challenger"), } for position, frame in predictions.groupby("position") }, } def _cluster_bootstrap_error_difference(predictions: pd.DataFrame) -> dict[str, Any]: """Quantify paired model error while preserving within-game dependence.""" errors = predictions.assign( champion_absolute_error=( predictions["actual_points"] - predictions["champion_points"] ).abs(), challenger_absolute_error=( predictions["actual_points"] - predictions["challenger_points"] ).abs(), ) clusters = ( errors.groupby("game_id", sort=True) .agg( champion_error_sum=("champion_absolute_error", "sum"), challenger_error_sum=("challenger_absolute_error", "sum"), player_weeks=("player_id", "size"), ) .reset_index() ) if clusters.empty: raise ValueError("Weather model uncertainty requires at least one game cluster") champion_sums = clusters["champion_error_sum"].to_numpy(dtype=float) challenger_sums = clusters["challenger_error_sum"].to_numpy(dtype=float) counts = clusters["player_weeks"].to_numpy(dtype=float) seed = int.from_bytes( hashlib.sha256(f"{STUDY_ID}:{ANALYSIS_VERSION}:game-bootstrap".encode()).digest()[:8], "big", ) rng = np.random.default_rng(seed) differences: list[np.ndarray] = [] improvements: list[np.ndarray] = [] for offset in range(0, BOOTSTRAP_REPETITIONS, 200): draws = min(200, BOOTSTRAP_REPETITIONS - offset) sampled = rng.integers(0, len(clusters), size=(draws, len(clusters))) denominator = counts[sampled].sum(axis=1) champion_mae = champion_sums[sampled].sum(axis=1) / denominator challenger_mae = challenger_sums[sampled].sum(axis=1) / denominator if not np.all(champion_mae > 0): raise ValueError("Weather model uncertainty requires positive champion error") differences.append(challenger_mae - champion_mae) improvements.append((champion_mae - challenger_mae) / champion_mae * 100) difference_interval = np.quantile(np.concatenate(differences), [0.025, 0.975]) improvement_interval = np.quantile(np.concatenate(improvements), [0.025, 0.975]) return { "method": ( "4,000-draw deterministic cluster bootstrap resampling NFL games; all player-weeks " "from a sampled game move together" ), "cluster_unit": "game_id", "clusters": len(clusters), "repetitions": BOOTSTRAP_REPETITIONS, "challenger_minus_champion_mae_ci_95": [ float(difference_interval[0]), float(difference_interval[1]), ], "overall_improvement_pct_ci_95": [ float(improvement_interval[0]), float(improvement_interval[1]), ], } def _summary(value: dict[str, Any]) -> EvaluationSummary: return EvaluationSummary( rows=int(value["rows"]), model_mae=float(value["model_mae"]), interval_80_coverage=float(value["interval_80_coverage"]), ) def _build_decision( *, predictions: pd.DataFrame, registration: ModelExperimentRegistration, reproducibility_sha256: list[str], created_at: str, ) -> tuple[ChampionDecision, dict[str, Any]]: scorecard = build_weather_scorecard(predictions) champion_overall = scorecard["overall"]["champion"] challenger_overall = scorecard["overall"]["challenger"] improvement = ( (champion_overall["model_mae"] - challenger_overall["model_mae"]) / champion_overall["model_mae"] * 100 ) wins = sum( value["challenger"]["model_mae"] < value["champion"]["model_mae"] for value in scorecard["by_season"].values() ) regressions = { position: ( (value["challenger"]["model_mae"] - value["champion"]["model_mae"]) / value["champion"]["model_mae"] * 100 ) for position, value in scorecard["by_position"].items() } maximum_regression = max(regressions.values()) leakage_safe = bool( (predictions["last_history_week"] < predictions["week"]).all() and ( pd.to_datetime(predictions["source_available_at"], utc=True) <= pd.to_datetime(predictions["cutoff_at"], utc=True) ).all() and ( pd.to_datetime(predictions["cutoff_at"], utc=True) < pd.to_datetime(predictions["valid_at"], utc=True) ).all() ) complete = not predictions[[*WEATHER_FEATURES, "actual_points"]].isna().any().any() coverage = float(challenger_overall["interval_80_coverage"]) gates = { "leakage": leakage_safe, "missingness": complete, "primary-metric": improvement >= registration.minimum_improvement_pct, "multiple-windows": wins >= registration.minimum_winning_windows, "subgroup-stability": maximum_regression <= registration.maximum_position_regression_pct, "calibration": ( registration.required_interval_coverage_min <= coverage <= registration.required_interval_coverage_max ), "reproducibility": len(set(reproducibility_sha256)) == 1, "cost": len(WEATHER_FEATURES) <= 20, } promoted = all(gates.values()) failed = [name for name, passed in gates.items() if not passed] decision = ChampionDecision( decision_id=("champion-decision-" + registration.experiment_id.removeprefix("model-exp-")), registration=registration, champion_overall=_summary(champion_overall), challenger_overall=_summary(challenger_overall), champion_by_season={ season: _summary(value["champion"]) for season, value in scorecard["by_season"].items() }, challenger_by_season={ season: _summary(value["challenger"]) for season, value in scorecard["by_season"].items() }, champion_by_position={ position: _summary(value["champion"]) for position, value in scorecard["by_position"].items() }, challenger_by_position={ position: _summary(value["challenger"]) for position, value in scorecard["by_position"].items() }, overall_improvement_pct=improvement, winning_windows=wins, maximum_position_regression_pct=maximum_regression, gates=gates, reproducibility_sha256=reproducibility_sha256, decision="promoted" if promoted else "rejected", reason=( "The cutoff-safe weather challenger passed every locked weekly projection gate." if promoted else "The weather challenger was rejected and the weekly champion is unchanged. Failed gates: " + ", ".join(failed) + "." ), active_configuration=registration.challenger if promoted else registration.champion, created_at=created_at, ) return decision, scorecard | { "position_regression_pct": regressions, "uncertainty": _cluster_bootstrap_error_difference(predictions), } def _write_static_contracts( *, output_dir: Path, active: ActiveModelChampion, test_seasons: list[int], locked_at: str, ) -> ModelExperimentRegistration: path = output_dir / "experiment-registration.json" existing = ( ModelExperimentRegistration.model_validate_json(path.read_text(encoding="utf-8")) if path.is_file() else None ) preserved_lock = ( existing.locked_at if existing is not None and existing.test_seasons == test_seasons and existing.champion == active.configuration else locked_at ) registration = _registration(active, test_seasons=test_seasons, locked_at=preserved_lock) _write_json(path, registration) _write_json(output_dir / "feature-definitions.json", _feature_definitions()) _write_json(output_dir / "signal-candidate.json", _signal()) _write_json( output_dir / "model-component.json", ModelComponentSpec( component_id="weekly-weather-team-environment", name="Weekly weather team environment", kind="team-environment", version=ANALYSIS_VERSION, model_family="ridge-regularized additive residual adjustment", input_feature_ids=[ "pregame-air-temperature", "pregame-wind", "pregame-precipitation", ], output_fields=["weekly-ppr-adjustment"], training_cutoff="Rolling; strictly before each held-out season", supported_positions=list(POSITIONS), status="shadow", evaluation_href=PUBLIC_DECISION_HREF, ), ) return registration def _model_eligible_counts( joined: pd.DataFrame, readiness: dict[str, Any] ) -> tuple[int, int]: """Reconcile the forecast-ready sample with the weekly model's eligible rows. The weekly champion requires two completed player games, so Weeks 1–2 cannot enter either model even when their archived forecasts are cutoff-safe. The 1,000-game acquisition gate therefore belongs to forecast readiness; this downstream check preserves the registered condition floor and verifies that the join did not create records absent from the accepted forecast sample. """ game_count = int(joined["game_id"].nunique()) condition_count = int( joined.loc[ joined["weather_high_wind_mps"].gt(0) | joined["weather_precip_probability"].ge(0.5), "game_id", ].nunique() ) if game_count <= 0: raise ValueError("No forecast-ready games are eligible for the weekly champion") if game_count > int(readiness["six_hour_games"]): raise ValueError("The model join created games outside forecast readiness") if condition_count > int(readiness["condition_games"]): raise ValueError("The model join created weather conditions outside forecast readiness") if condition_count < MINIMUM_CONDITION_GAMES: raise ValueError("Too few model-eligible material-weather games") return game_count, condition_count def run_weather_projection_challenger( *, hrrr_dir: Path, output_dir: Path, artifact_root: str, code_sha: str, run_id: str, weekly_stats_paths: list[Path] | None = None, weekly_champion_state_path: Path = WEEKLY_CHAMPION_STATE_PATH, active_decision_path: Path = ACTIVE_WEEKLY_DECISION_PATH, active_parameters_path: Path = ACTIVE_WEATHER_PARAMETERS_PATH, completed_at: str | None = None, ) -> EvaluationRunRecord: started = completed_at or utc_now() active = load_weekly_champion(weekly_champion_state_path) output_dir.mkdir(parents=True, exist_ok=True) readiness = forecast_readiness(hrrr_dir) latest_forecast_season = readiness["latest_forecast_season"] or 2024 test_seasons = list(range(int(latest_forecast_season) - 5, int(latest_forecast_season) + 1)) registration = _write_static_contracts( output_dir=output_dir, active=active, test_seasons=test_seasons, locked_at=started, ) _write_json(output_dir / "readiness.json", readiness) evaluation_run_path = output_dir / "evaluation-run.json" evaluation_history_path = output_dir / "evaluation-runs" / f"{run_id}.json" if not readiness["ready"]: gates = [ EvaluationGateResult( gate_id="leakage", status="passed", details="Archived forecast records enforce issue and source availability before cutoff.", evidence_hrefs=[f"{artifact_root}/readiness.json"], ), EvaluationGateResult( gate_id="missingness", status="failed", details="The registered six-hour game coverage floor has not been reached.", evidence_hrefs=[f"{artifact_root}/readiness.json"], ), *[ EvaluationGateResult( gate_id=gate_id, status="not-applicable", details="This locked model gate cannot run until forecast readiness passes.", evidence_hrefs=[f"{artifact_root}/readiness.json"], ) for gate_id in ( "primary-metric", "multiple-windows", "subgroup-stability", "calibration", ) ], EvaluationGateResult( gate_id="reproducibility", status="passed", details="The upstream archived forecast scorecard reproduced exactly.", evidence_hrefs=[f"{artifact_root}/readiness.json"], ), EvaluationGateResult( gate_id="cost", status="passed", details="Readiness checks use no LLM calls and stay inside the registered feature bound.", evidence_hrefs=[f"{artifact_root}/experiment-registration.json"], ), ] evaluation = EvaluationRunRecord( run_id=run_id, workload_id=STUDY_ID, run_kind="model-challenger", status="withheld", started_at=started, completed_at=started, code_sha=code_sha, source_sha256=readiness["source_sha256"], registration_href=f"{artifact_root}/experiment-registration.json", artifact_hrefs=[ f"{artifact_root}/readiness.json", f"{artifact_root}/feature-definitions.json", f"{artifact_root}/signal-candidate.json", ], metrics=[ EvaluationMetric( metric_id="six-hour-games", label="Cutoff-safe six-hour games", value=readiness["six_hour_games"], units="games", population="Registered outdoor NFL games", ), EvaluationMetric( metric_id="condition-games", label="Material weather games", value=readiness["condition_games"], units="games", population="Six-hour forecasts with wind or precipitation", ), ], gates=gates, total_cost_usd=0, champion_version=registration.champion.version, challenger_version=registration.challenger.version, ) _write_json(evaluation_run_path, evaluation) _write_json(evaluation_history_path, evaluation) return evaluation if not weekly_stats_paths: raise ValueError("Forecast-ready weather evaluation requires weekly player statistics") weekly = load_weekly_stats(weekly_stats_paths) examples = build_player_week_examples(weekly) snapshots = _load_six_hour_snapshots(hrrr_dir / "snapshots.json") joined = attach_weather_features(examples, snapshots) game_count, condition_count = _model_eligible_counts(joined, readiness) readiness.update( { "model_eligible_games": game_count, "model_eligible_condition_games": condition_count, "baseline_ineligible_forecast_games": int(readiness["six_hour_games"]) - game_count, } ) _write_json(output_dir / "readiness.json", readiness) first, parameters = walk_forward_weather_challenger(joined) second, second_parameters = walk_forward_weather_challenger(joined) evaluated_seasons = sorted(int(value) for value in first["season"].unique()) if evaluated_seasons != registration.test_seasons: raise ValueError("Weather evaluation windows changed after preregistration") stable_columns = [ "game_id", "player_id", "season", "week", "position", "champion_points", "challenger_points", "champion_low", "champion_high", "challenger_low", "challenger_high", "actual_points", ] first_hash = _hash(first[stable_columns].to_dict(orient="records")) second_hash = _hash(second[stable_columns].to_dict(orient="records")) if parameters != second_parameters: second_hash = _hash({"predictions": second_hash, "parameters": second_parameters}) first_hash = _hash({"predictions": first_hash, "parameters": parameters}) decision, scorecard = _build_decision( predictions=first, registration=registration, reproducibility_sha256=[first_hash, second_hash], created_at=started, ) _write_json(output_dir / "scorecard.json", scorecard) _write_json(output_dir / "model-parameters.json", parameters) _write_json(output_dir / "champion-decision.json", decision) latest = joined.sort_values(["season", "week"]).iloc[-1] feature_snapshot = FeatureSnapshot( snapshot_id=f"weather-features-{int(latest.season)}-week-{int(latest.week)}", as_of=str(latest.cutoff_at), season=int(latest.season), week=int(latest.week), feature_versions={ definition.feature_id: definition.version for definition in _feature_definitions() }, source_sha256={ "noaa-hrrr-archive": readiness["source_sha256"]["hrrr-snapshots"], "nflverse-weekly-player-stats": _hash( {path.name: sha256_file(path) for path in weekly_stats_paths} ), }, row_count=len(joined), excluded_row_count=max(0, len(examples) - len(joined)), data_sha256=_hash( joined[["player_id", "season", "week", *WEATHER_FEATURES]].to_dict(orient="records") ), ) _write_json(output_dir / "feature-snapshot.json", feature_snapshot) terminal_signal = _signal( status="promoted" if decision.decision == "promoted" else "rejected", decision_href=PUBLIC_DECISION_HREF, ) _write_json(output_dir / "signal-candidate.json", terminal_signal) if decision.decision == "promoted": _write_json( active_parameters_path, { "schema_version": "1.0", "model_version": decision.active_configuration.version, "signal_id": SIGNAL_ID, "forecast_horizon_hours": FORECAST_HORIZON_HOURS, "positions": parameters["active_fit"], "decision_href": PUBLIC_DECISION_HREF, }, ) apply_promotion( decision, champion_state_path=weekly_champion_state_path, active_decision_path=active_decision_path, public_decision_href=PUBLIC_DECISION_HREF, ) gates = [ EvaluationGateResult( gate_id=name, status="passed" if passed else "failed", details=( f"Locked ChampionDecision gate {name} passed." if passed else f"Locked ChampionDecision gate {name} failed; the weekly champion is unchanged." ), evidence_hrefs=[f"{artifact_root}/champion-decision.json"], ) for name, passed in decision.gates.items() ] evaluation = EvaluationRunRecord( run_id=run_id, workload_id=STUDY_ID, run_kind="model-challenger", status="passed" if decision.decision == "promoted" else "rejected", started_at=started, completed_at=started, code_sha=code_sha, source_sha256={ **readiness["source_sha256"], "weekly-player-stats": _hash( {path.name: sha256_file(path) for path in weekly_stats_paths} ), }, registration_href=f"{artifact_root}/experiment-registration.json", artifact_hrefs=[ f"{artifact_root}/{name}" for name in ( "readiness.json", "feature-definitions.json", "feature-snapshot.json", "signal-candidate.json", "scorecard.json", "model-parameters.json", "champion-decision.json", ) ], metrics=[ EvaluationMetric( metric_id="eligible-games", label="Model-eligible weather games", value=game_count, units="games", population="Identical champion and challenger denominator", ), EvaluationMetric( metric_id="mae-improvement", label="Weather challenger MAE improvement", value=decision.overall_improvement_pct, units="percent", population="Held-out player-weeks", ), EvaluationMetric( metric_id="winning-windows", label="Winning future-season windows", value=decision.winning_windows, units="seasons", population=( f"Locked {registration.test_seasons[0]}-{registration.test_seasons[-1]} " "evaluation windows" ), ), ], gates=gates, total_cost_usd=0, champion_version=registration.champion.version, challenger_version=registration.challenger.version, ) _write_json(evaluation_run_path, evaluation) _write_json(evaluation_history_path, evaluation) return evaluation