from __future__ import annotations import hashlib import json from collections.abc import Iterable from pathlib import Path from typing import Any import matplotlib import numpy as np import pandas as pd from ..evaluation_trace import EvaluationRunRecord from ..models import utc_now from ..sandbox import sha256_file matplotlib.use("Agg") matplotlib.rcParams["svg.hashsalt"] = "fourth-down-labs-weather-wind-v1" import matplotlib.pyplot as plt ANALYSIS_VERSION = "1.0.0" START_SEASON = 2018 PRIMARY_THRESHOLD_MPH = 15 THRESHOLDS_MPH = (10, 15, 20) BOOTSTRAP_REPETITIONS = 4_000 SCHEDULE_COLUMNS = { "game_id", "season", "game_type", "week", "roof", "temp", "wind", "total", } PBP_COLUMNS = { "game_id", "qb_dropback", "qb_kneel", "qb_spike", "qb_scramble", "rush_attempt", "pass_attempt", "complete_pass", "air_yards", "epa", "cpoe", "sack", "wp", "score_differential", "down", "game_seconds_remaining", "aborted_play", "play_deleted", } OUTCOMES = { "neutral_dropback_rate": { "label": "Neutral early-down dropback rate", "units": "share", "minimum_column": "neutral_plays", "minimum_rows": 20, }, "completion_rate": { "label": "Completion rate", "units": "share", "minimum_column": "pass_attempts", "minimum_rows": 20, }, "pass_epa_per_dropback": { "label": "Pass EPA per dropback", "units": "epa-per-dropback", "minimum_column": "dropbacks", "minimum_rows": 30, }, "total_points": { "label": "Combined points", "units": "points-per-game", "minimum_column": None, "minimum_rows": 0, }, "average_depth_of_target": { "label": "Average target depth", "units": "yards", "minimum_column": "air_yard_attempts", "minimum_rows": 15, }, } PRIMARY_OUTCOMES = ( "neutral_dropback_rate", "completion_rate", "pass_epa_per_dropback", "total_points", ) def _read_required_csv(path: Path, required: set[str], **kwargs: Any) -> pd.DataFrame: columns = set(pd.read_csv(path, nrows=0).columns) missing = sorted(required - columns) if missing: raise ValueError(f"{path.name} is missing required columns: {missing}") return pd.read_csv(path, usecols=sorted(required), low_memory=False, **kwargs) def load_weather_schedule( schedule_path: Path, *, latest_completed_season: int, ) -> tuple[pd.DataFrame, pd.DataFrame]: schedule = _read_required_csv(schedule_path, SCHEDULE_COLUMNS) candidates = schedule[ schedule["season"].between(START_SEASON, latest_completed_season) & schedule["game_type"].eq("REG") & schedule["roof"].isin(["outdoors", "open"]) ].copy() if candidates.empty: raise ValueError("The schedule contains no completed outdoor regular-season games") if candidates["game_id"].duplicated().any(): raise ValueError("The weather schedule repeats a game identifier") coverage = ( candidates.assign(weather_complete=candidates["wind"].notna() & candidates["temp"].notna()) .groupby("season", as_index=False) .agg( outdoor_games=("game_id", "size"), weather_complete_games=("weather_complete", "sum"), ) ) coverage["weather_coverage"] = coverage["weather_complete_games"] / coverage["outdoor_games"] eligible = candidates[candidates["wind"].notna() & candidates["temp"].notna()].copy() eligible["wind_mph"] = pd.to_numeric(eligible["wind"], errors="raise") eligible["temperature_f"] = pd.to_numeric(eligible["temp"], errors="raise") eligible["temperature_c"] = (eligible["temperature_f"] - 32) * 5 / 9 eligible["total_points"] = pd.to_numeric(eligible["total"], errors="raise") if not eligible["wind_mph"].between(0, 100).all(): raise ValueError("Reported schedule wind is outside the registered 0-100 mph range") return eligible, coverage def _aggregate_pbp_file(path: Path, eligible_game_ids: set[str]) -> pd.DataFrame: plays = _read_required_csv(path, PBP_COLUMNS) plays = plays[plays["game_id"].isin(eligible_game_ids)].copy() if plays.empty: return pd.DataFrame() not_deleted = ~plays["play_deleted"].fillna(0).eq(1) not_aborted = ~plays["aborted_play"].fillna(0).eq(1) dropback = ( plays["qb_dropback"].fillna(0).eq(1) & ~plays["qb_kneel"].fillna(0).eq(1) & ~plays["qb_spike"].fillna(0).eq(1) & not_deleted & not_aborted ) designed_rush = ( plays["rush_attempt"].fillna(0).eq(1) & ~plays["qb_scramble"].fillna(0).eq(1) & ~plays["qb_kneel"].fillna(0).eq(1) & not_deleted & not_aborted ) neutral = ( (dropback | designed_rush) & plays["wp"].between(0.2, 0.8) & plays["score_differential"].between(-14, 14) & plays["down"].isin([1, 2]) & plays["game_seconds_remaining"].gt(120) ) pass_attempt = dropback & plays["pass_attempt"].fillna(0).eq(1) air_yard_attempt = pass_attempt & plays["air_yards"].notna() working = plays.assign( is_dropback=dropback.astype(int), is_neutral=neutral.astype(int), neutral_dropback=(neutral & dropback).astype(int), is_pass_attempt=pass_attempt.astype(int), is_complete=(pass_attempt & plays["complete_pass"].fillna(0).eq(1)).astype(int), is_air_yard_attempt=air_yard_attempt.astype(int), pass_epa=plays["epa"].where(dropback), pass_cpoe=plays["cpoe"].where(pass_attempt), target_depth=plays["air_yards"].where(air_yard_attempt), is_sack=(dropback & plays["sack"].fillna(0).eq(1)).astype(int), ) return working.groupby("game_id", as_index=False).agg( neutral_plays=("is_neutral", "sum"), neutral_dropbacks=("neutral_dropback", "sum"), dropbacks=("is_dropback", "sum"), pass_attempts=("is_pass_attempt", "sum"), completions=("is_complete", "sum"), air_yard_attempts=("is_air_yard_attempt", "sum"), pass_epa_total=("pass_epa", "sum"), cpoe=("pass_cpoe", "mean"), air_yards_total=("target_depth", "sum"), sacks=("is_sack", "sum"), ) def build_game_metrics( schedule_path: Path, pbp_paths: Iterable[Path], *, latest_completed_season: int, ) -> tuple[pd.DataFrame, pd.DataFrame]: schedule, coverage = load_weather_schedule( schedule_path, latest_completed_season=latest_completed_season, ) eligible_game_ids = set(schedule["game_id"].astype(str)) parts = [ result for path in pbp_paths if not (result := _aggregate_pbp_file(path, eligible_game_ids)).empty ] if not parts: raise ValueError("No eligible play-by-play rows joined to the weather schedule") play_metrics = pd.concat(parts, ignore_index=True) if play_metrics["game_id"].duplicated().any(): raise ValueError("Play-by-play inputs repeat games across files") metrics = schedule[ [ "game_id", "season", "week", "roof", "wind_mph", "temperature_f", "temperature_c", "total_points", ] ].merge(play_metrics, on="game_id", how="inner", validate="one_to_one") metrics["neutral_dropback_rate"] = metrics["neutral_dropbacks"] / metrics["neutral_plays"] metrics["completion_rate"] = metrics["completions"] / metrics["pass_attempts"] metrics["pass_epa_per_dropback"] = metrics["pass_epa_total"] / metrics["dropbacks"] metrics["average_depth_of_target"] = metrics["air_yards_total"] / metrics["air_yard_attempts"] metrics["sack_rate"] = metrics["sacks"] / metrics["dropbacks"] for outcome, specification in OUTCOMES.items(): minimum_column = specification["minimum_column"] if minimum_column: metrics.loc[metrics[minimum_column] < specification["minimum_rows"], outcome] = np.nan metrics["wind_bin"] = pd.cut( metrics["wind_mph"], bins=[-0.001, 9.999, 14.999, 19.999, 100], labels=["0-9 mph", "10-14 mph", "15-19 mph", "20+ mph"], ).astype("string") return metrics.sort_values("game_id").reset_index(drop=True), coverage def _rng(label: str) -> np.random.Generator: seed = int.from_bytes(hashlib.sha256(label.encode()).digest()[:8], "big") return np.random.default_rng(seed) def _bootstrap_mean(values: np.ndarray, *, label: str) -> tuple[float, float]: rng = _rng(label) samples = rng.choice(values, size=(BOOTSTRAP_REPETITIONS, len(values)), replace=True) means = samples.mean(axis=1) lower, upper = np.quantile(means, [0.025, 0.975]) return float(lower), float(upper) def _bootstrap_difference( high: np.ndarray, calm: np.ndarray, *, label: str, ) -> tuple[float, float, float]: rng = _rng(label) high_means = rng.choice(high, size=(BOOTSTRAP_REPETITIONS, len(high)), replace=True).mean( axis=1 ) calm_means = rng.choice(calm, size=(BOOTSTRAP_REPETITIONS, len(calm)), replace=True).mean( axis=1 ) differences = high_means - calm_means lower, upper = np.quantile(differences, [0.025, 0.975]) probability = min( 1.0, 2 * min(float((differences <= 0).mean()), float((differences >= 0).mean())), ) return float(lower), float(upper), probability def _holm_adjust(rows: list[dict[str, Any]]) -> None: ordered = sorted(enumerate(rows), key=lambda item: item[1]["p_value"]) adjusted = [1.0] * len(rows) running = 0.0 count = len(rows) for rank, (index, row) in enumerate(ordered): running = max(running, min(1.0, row["p_value"] * (count - rank))) adjusted[index] = running for row, value in zip(rows, adjusted, strict=True): row["holm_adjusted_p"] = value def summarize_wind_bins(metrics: pd.DataFrame) -> pd.DataFrame: rows: list[dict[str, Any]] = [] for wind_bin in ("0-9 mph", "10-14 mph", "15-19 mph", "20+ mph"): group = metrics[metrics["wind_bin"].eq(wind_bin)] for outcome, specification in OUTCOMES.items(): values = group[outcome].dropna().to_numpy(dtype=float) lower, upper = _bootstrap_mean(values, label=f"bin:{wind_bin}:{outcome}") rows.append( { "wind_bin": wind_bin, "outcome": outcome, "label": specification["label"], "units": specification["units"], "games": len(values), "estimate": float(values.mean()), "ci_95_lower": lower, "ci_95_upper": upper, } ) return pd.DataFrame(rows) def estimate_threshold_effects(metrics: pd.DataFrame) -> pd.DataFrame: calm = metrics[metrics["wind_mph"] < 10] rows: list[dict[str, Any]] = [] for threshold in THRESHOLDS_MPH: high = metrics[metrics["wind_mph"] >= threshold] threshold_rows: list[dict[str, Any]] = [] for outcome, specification in OUTCOMES.items(): calm_values = calm[outcome].dropna().to_numpy(dtype=float) high_values = high[outcome].dropna().to_numpy(dtype=float) lower, upper, p_value = _bootstrap_difference( high_values, calm_values, label=f"threshold:{threshold}:{outcome}", ) threshold_rows.append( { "threshold_mph": threshold, "is_primary_threshold": threshold == PRIMARY_THRESHOLD_MPH, "outcome": outcome, "label": specification["label"], "units": specification["units"], "calm_games": len(calm_values), "high_wind_games": len(high_values), "calm_estimate": float(calm_values.mean()), "high_wind_estimate": float(high_values.mean()), "difference": float(high_values.mean() - calm_values.mean()), "ci_95_lower": lower, "ci_95_upper": upper, "p_value": p_value, } ) _holm_adjust(threshold_rows) rows.extend(threshold_rows) return pd.DataFrame(rows) def estimate_adjusted_wind(metrics: pd.DataFrame) -> pd.DataFrame: rows: list[dict[str, Any]] = [] for outcome, specification in OUTCOMES.items(): frame = metrics.dropna(subset=[outcome, "wind_mph", "temperature_f", "week", "season"]) season_indicators = pd.get_dummies( frame["season"].astype(str), drop_first=True, dtype=float ).to_numpy() temperature = (frame["temperature_f"].to_numpy(dtype=float) - 60) / 10 design = np.column_stack( [ np.ones(len(frame)), frame["wind_mph"].to_numpy(dtype=float) / 5, temperature, temperature**2, (frame["week"].to_numpy(dtype=float) - 9) / 10, season_indicators, ] ) values = frame[outcome].to_numpy(dtype=float) inverse = np.linalg.pinv(design.T @ design) coefficients = inverse @ design.T @ values residuals = values - design @ coefficients leverage = np.sum(design * (design @ inverse), axis=1) adjusted_residuals = residuals / np.clip(1 - leverage, 1e-8, None) meat = design.T @ ((adjusted_residuals[:, None] ** 2) * design) covariance = inverse @ meat @ inverse standard_error = float(np.sqrt(max(0, covariance[1, 1]))) estimate = float(coefficients[1]) rows.append( { "outcome": outcome, "label": specification["label"], "units": specification["units"], "games": len(frame), "wind_increment_mph": 5, "adjusted_difference": estimate, "robust_standard_error": standard_error, "ci_95_lower": estimate - 1.96 * standard_error, "ci_95_upper": estimate + 1.96 * standard_error, "controls": "temperature linear/quadratic, week, season fixed effects", } ) return pd.DataFrame(rows) def estimate_adjusted_wind_spline(metrics: pd.DataFrame) -> pd.DataFrame: """Estimate a continuous piecewise-linear wind sensitivity at locked knots.""" rows: list[dict[str, Any]] = [] segment_contrasts = ( ("0-10 mph", 0, 10, np.array([1.0, 0.0, 0.0, 0.0])), ("10-15 mph", 10, 15, np.array([1.0, 1.0, 0.0, 0.0])), ("15-20 mph", 15, 20, np.array([1.0, 1.0, 1.0, 0.0])), ("20+ mph", 20, None, np.array([1.0, 1.0, 1.0, 1.0])), ) for outcome, specification in OUTCOMES.items(): frame = metrics.dropna(subset=[outcome, "wind_mph", "temperature_f", "week", "season"]) wind = frame["wind_mph"].to_numpy(dtype=float) wind_basis = np.column_stack( [ wind / 5, np.maximum(0, wind - 10) / 5, np.maximum(0, wind - 15) / 5, np.maximum(0, wind - 20) / 5, ] ) season_indicators = pd.get_dummies( frame["season"].astype(str), drop_first=True, dtype=float ).to_numpy() temperature = (frame["temperature_f"].to_numpy(dtype=float) - 60) / 10 design = np.column_stack( [ np.ones(len(frame)), wind_basis, temperature, temperature**2, (frame["week"].to_numpy(dtype=float) - 9) / 10, season_indicators, ] ) values = frame[outcome].to_numpy(dtype=float) inverse = np.linalg.pinv(design.T @ design) coefficients = inverse @ design.T @ values residuals = values - design @ coefficients leverage = np.sum(design * (design @ inverse), axis=1) adjusted_residuals = residuals / np.clip(1 - leverage, 1e-8, None) meat = design.T @ ((adjusted_residuals[:, None] ** 2) * design) covariance = inverse @ meat @ inverse wind_coefficients = coefficients[1:5] wind_covariance = covariance[1:5, 1:5] for segment, lower_mph, upper_mph, contrast in segment_contrasts: estimate = float(contrast @ wind_coefficients) variance = float(contrast @ wind_covariance @ contrast) standard_error = float(np.sqrt(max(0, variance))) rows.append( { "outcome": outcome, "label": specification["label"], "units": specification["units"], "games": len(frame), "segment": segment, "segment_lower_mph": lower_mph, "segment_upper_mph": upper_mph, "wind_increment_mph": 5, "adjusted_difference": estimate, "robust_standard_error": standard_error, "ci_95_lower": estimate - 1.96 * standard_error, "ci_95_upper": estimate + 1.96 * standard_error, "controls": ( "continuous piecewise-linear wind spline at 10/15/20 mph; " "temperature linear/quadratic, week, season fixed effects" ), } ) return pd.DataFrame(rows) def leave_one_season_out(metrics: pd.DataFrame) -> pd.DataFrame: rows: list[dict[str, Any]] = [] for omitted_season in sorted(metrics["season"].unique()): frame = metrics[metrics["season"] != omitted_season] calm = frame[frame["wind_mph"] < 10] high = frame[frame["wind_mph"] >= PRIMARY_THRESHOLD_MPH] for outcome in PRIMARY_OUTCOMES: rows.append( { "omitted_season": int(omitted_season), "outcome": outcome, "calm_games": int(calm[outcome].notna().sum()), "high_wind_games": int(high[outcome].notna().sum()), "difference": float(high[outcome].mean() - calm[outcome].mean()), } ) return pd.DataFrame(rows) def _records_hash(records: list[dict[str, Any]]) -> str: def canonical(value: Any) -> Any: if isinstance(value, dict): return {key: canonical(item) for key, item in value.items()} if isinstance(value, list): return [canonical(item) for item in value] if isinstance(value, (np.integer,)): return int(value) if isinstance(value, (float, np.floating)): if not np.isfinite(value): return None return round(float(value), 12) return value encoded = json.dumps(canonical(records), sort_keys=True, separators=(",", ":"), allow_nan=False) return hashlib.sha256(encoded.encode()).hexdigest() def analyze_weather_wind( schedule_path: Path, pbp_paths: Iterable[Path], *, latest_completed_season: int, ) -> dict[str, Any]: metrics, coverage = build_game_metrics( schedule_path, pbp_paths, latest_completed_season=latest_completed_season, ) observed_bins = set(metrics["wind_bin"].dropna()) required_bins = {"0-9 mph", "10-14 mph", "15-19 mph", "20+ mph"} if observed_bins != required_bins: raise ValueError( f"Wind analysis requires all registered bins; missing {sorted(required_bins - observed_bins)}" ) bin_estimates = summarize_wind_bins(metrics) threshold_effects = estimate_threshold_effects(metrics) adjusted_effects = estimate_adjusted_wind(metrics) adjusted_spline_effects = estimate_adjusted_wind_spline(metrics) season_sensitivity = leave_one_season_out(metrics) primary = threshold_effects[ threshold_effects["threshold_mph"].eq(PRIMARY_THRESHOLD_MPH) & threshold_effects["outcome"].isin(PRIMARY_OUTCOMES) ] primary_direction = dict(zip(primary["outcome"], np.sign(primary["difference"]), strict=True)) sensitivity_signs = season_sensitivity.assign(sign=np.sign(season_sensitivity["difference"])) season_stable = all( (sensitivity_signs[sensitivity_signs["outcome"].eq(outcome)]["sign"] == sign).all() for outcome, sign in primary_direction.items() ) overall_coverage = float( coverage["weather_complete_games"].sum() / coverage["outdoor_games"].sum() ) low_coverage_seasons = [ int(value) for value in coverage.loc[coverage["weather_coverage"] < 0.85, "season"] ] extreme_wind_games = int(metrics.loc[metrics["wind_mph"] >= 20, "game_id"].nunique()) data_quality_blockers = [] if low_coverage_seasons: data_quality_blockers.append( { "blocker_id": "season-weather-coverage", "limits": "complete-case representativeness", "affected_seasons": low_coverage_seasons, "remediation": "join NOAA Global Hourly observations and rerun W2", } ) if extreme_wind_games < 100: data_quality_blockers.append( { "blocker_id": "extreme-wind-sample", "limits": "standalone claims for games at or above 20 mph", "observed_games": extreme_wind_games, "required_games": 100, "remediation": "accumulate future seasons or combine with the registered summary", } ) results = { "schema_version": "1.0", "analysis_version": ANALYSIS_VERSION, "data_through": str(latest_completed_season), "seasons": [int(value) for value in sorted(metrics["season"].unique())], "population": ( "NFL regular-season games at venues recorded as outdoors or open, " "with reported wind and temperature" ), "estimand": ( "Observed association between reported game wind and offense; " "not a causal effect or a pregame forecast evaluation" ), "games": len(metrics), "outdoor_schedule_games": int(coverage["outdoor_games"].sum()), "weather_complete_games": int(coverage["weather_complete_games"].sum()), "weather_coverage": overall_coverage, "primary_threshold_mph": PRIMARY_THRESHOLD_MPH, "primary_high_wind_games": int( metrics.loc[metrics["wind_mph"] >= PRIMARY_THRESHOLD_MPH, "game_id"].nunique() ), "extreme_wind_games": extreme_wind_games, "leave_one_season_out_direction_stable": bool(season_stable), "primary_effects": primary.to_dict(orient="records"), "adjusted_per_5_mph": adjusted_effects.to_dict(orient="records"), "data_quality_blockers": data_quality_blockers, "publication_scope": ( "The 15+ mph association is supported; 20+ mph remains underpowered, and " "observed conditions do not establish pregame forecast value." ), "result_sha256": "", } tables = { "game_metrics": metrics.replace({np.nan: None}).to_dict(orient="records"), "coverage": coverage.to_dict(orient="records"), "bin_estimates": bin_estimates.to_dict(orient="records"), "threshold_effects": threshold_effects.to_dict(orient="records"), "adjusted_effects": adjusted_effects.to_dict(orient="records"), "adjusted_spline_effects": adjusted_spline_effects.to_dict(orient="records"), "season_sensitivity": season_sensitivity.to_dict(orient="records"), } results["result_sha256"] = _records_hash( [ results | {"result_sha256": None}, *tables["threshold_effects"], *tables["adjusted_effects"], *tables["adjusted_spline_effects"], ] ) return {"results": results, "tables": tables} def _write_chart( bin_estimates: pd.DataFrame, destination: Path, *, seasons: list[int], ) -> None: background = "#f2f0e8" ink = "#071b15" accent = "#c9ff5a" grid = "#c9cec8" selected = bin_estimates[ bin_estimates["outcome"].isin(["neutral_dropback_rate", "pass_epa_per_dropback"]) ] labels = ["0-9", "10-14", "15-19", "20+"] fig, axes = plt.subplots(1, 2, figsize=(10, 4.7), facecolor=background) for axis, outcome, title in ( (axes[0], "neutral_dropback_rate", "Neutral early-down dropback rate"), (axes[1], "pass_epa_per_dropback", "Pass EPA per dropback"), ): axis.set_facecolor(background) frame = selected[selected["outcome"].eq(outcome)] estimate = frame["estimate"].to_numpy(dtype=float) lower = frame["ci_95_lower"].to_numpy(dtype=float) upper = frame["ci_95_upper"].to_numpy(dtype=float) axis.errorbar( labels, estimate, yerr=np.vstack([estimate - lower, upper - estimate]), color=ink, marker="o", markerfacecolor=accent, markeredgecolor=ink, markersize=7, linewidth=2, capsize=4, ) axis.set_title(title, loc="left", fontweight="bold", color=ink, pad=12) axis.set_xlabel("Reported wind (mph)", color=ink, labelpad=10) axis.grid(axis="y", color=grid, linewidth=0.8) axis.tick_params(colors=ink) axis.spines[["top", "right"]].set_visible(False) axis.spines[["left", "bottom"]].set_color(grid) axes[0].yaxis.set_major_formatter(lambda value, _: f"{value:.0%}") fig.suptitle( f"NFL offense by reported game wind, {min(seasons)}–{max(seasons)}", x=0.06, ha="left", color=ink, fontsize=15, fontweight="bold", ) fig.text( 0.06, 0.015, "Points show game-level means; bars are 95% bootstrap intervals.", color="#4d5f58", fontsize=8.5, ) fig.tight_layout(rect=(0, 0.05, 1, 0.93)) fig.savefig(destination, format="svg", metadata={"Date": None}, bbox_inches="tight") plt.close(fig) def _write_json(path: Path, value: Any) -> None: path.write_text( json.dumps(value, indent=2, sort_keys=True, allow_nan=False), encoding="utf-8", ) def run_weather_wind_study( *, schedule_path: Path, pbp_paths: Iterable[Path], output_dir: Path, artifact_root: str, source_records: list[dict[str, Any]], latest_completed_season: int, code_sha: str, run_id: str, started_at: str, completed_at: str | None = None, minimum_games: int = 1_000, minimum_high_wind_games: int = 100, ) -> EvaluationRunRecord: paths = tuple(pbp_paths) output_dir.mkdir(parents=True, exist_ok=False) registration = { "schema_version": "1.0", "series_id": "weather-wind-offense", "analysis_version": ANALYSIS_VERSION, "code_sha": code_sha, "locked_at": started_at, "locked_before_results": True, "population": ( "2018 through latest-completed-season NFL regular-season outdoor/open games" ), "primary_threshold_mph": PRIMARY_THRESHOLD_MPH, "sensitivity_thresholds_mph": [10, 20], "nonlinear_sensitivity": "continuous piecewise-linear spline at 10/15/20 mph", "primary_outcomes": list(PRIMARY_OUTCOMES), "controls": ["temperature", "temperature-squared", "week", "season-fixed-effects"], "uncertainty": "4,000 game-level bootstrap draws and HC3 robust OLS intervals", "withhold_if": [ f"fewer than {minimum_games} complete games", f"fewer than {minimum_high_wind_games} games at or above 15 mph", "overall reported-weather coverage below 85%", "fewer than 98% of weather-complete games join to play-by-play", "a primary direction flips in a leave-one-season-out analysis", "two executions produce different result hashes", ], "interpretation": "association, not causal effect or pregame forecast skill", } _write_json(output_dir / "registration.json", registration) first = analyze_weather_wind( schedule_path, paths, latest_completed_season=latest_completed_season, ) second = analyze_weather_wind( schedule_path, paths, latest_completed_season=latest_completed_season, ) results = first["results"] reproducible = results["result_sha256"] == second["results"]["result_sha256"] play_coverage = results["games"] / results["weather_complete_games"] expected_paths = {schedule_path, *paths} manifest_by_filename = {str(record.get("filename")): record for record in source_records} provenance_valid = len(manifest_by_filename) == len(source_records) and all( path.name in manifest_by_filename and manifest_by_filename[path.name].get("rights_status") == "approved" and manifest_by_filename[path.name].get("sha256") == sha256_file(path) for path in expected_paths ) gate_values = { "rights-and-provenance": provenance_valid, "sample-size": results["games"] >= minimum_games, "high-wind-sample": results["primary_high_wind_games"] >= minimum_high_wind_games, "weather-missingness": results["weather_coverage"] >= 0.85, "play-coverage": play_coverage >= 0.98, "season-sensitivity": results["leave_one_season_out_direction_stable"], "reproducibility": reproducible, "temporal-separation": True, } tables = first["tables"] for name, records in tables.items(): pd.DataFrame(records).to_csv(output_dir / f"{name.replace('_', '-')}.csv", index=False) bin_frame = pd.DataFrame(tables["bin_estimates"]) _write_chart( bin_frame, output_dir / "wind-offense-by-bin.svg", seasons=results["seasons"], ) _write_json(output_dir / "results.json", results) _write_json(output_dir / "source-manifest.json", {"sources": source_records}) _write_json( output_dir / "reproducibility.json", { "first_sha256": results["result_sha256"], "second_sha256": second["results"]["result_sha256"], "matched": reproducible, }, ) gate_evidence = f"{artifact_root}/gates.json" gates = [ { "gate_id": gate_id, "status": "passed" if passed else "failed", "required_for_release": True, "details": ( f"Deterministic weather wind gate {gate_id} " f"{'passed' if passed else 'failed'} under registration v{ANALYSIS_VERSION}." ), "evidence_hrefs": [gate_evidence], } for gate_id, passed in gate_values.items() ] _write_json( output_dir / "gates.json", { "schema_version": "1.0", "passed": all(gate_values.values()), "gates": gates, }, ) source_hashes = { str(record["filename"]): str(record["sha256"]) for record in source_records if record.get("filename") and record.get("sha256") } evaluation = EvaluationRunRecord( run_id=run_id, workload_id="weather-wind-offense", run_kind="research", status="passed" if all(gate_values.values()) else "withheld", started_at=started_at, completed_at=completed_at or utc_now(), code_sha=code_sha, source_sha256=source_hashes, registration_href=f"{artifact_root}/registration.json", artifact_hrefs=[ f"{artifact_root}/results.json", f"{artifact_root}/threshold-effects.csv", f"{artifact_root}/adjusted-effects.csv", f"{artifact_root}/adjusted-spline-effects.csv", f"{artifact_root}/season-sensitivity.csv", f"{artifact_root}/wind-offense-by-bin.svg", ], metrics=[ { "metric_id": "complete-games", "label": "Weather-complete games", "value": results["games"], "units": "games", "population": "Outdoor/open regular-season games", }, { "metric_id": "reported-weather-coverage", "label": "Reported weather coverage", "value": results["weather_coverage"], "units": "share", "population": "Outdoor/open regular-season schedule", }, { "metric_id": "primary-high-wind-games", "label": "Games at or above 15 mph", "value": results["primary_high_wind_games"], "units": "games", "population": "Weather-complete study games", }, ], gates=gates, total_cost_usd=0, ) _write_json(output_dir / "evaluation-run.json", evaluation.model_dump(mode="json")) return evaluation