import argparse import hashlib import json import math import pathlib import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np import pandas as pd WINDOWS = [1, 2, 3, 4, 6, 8] OUT_LEN = 4 SEED = 271828 B = 1000 FIRST, LAST = 2018, 2025 EVAL_SEASONS = list(range(2019, 2026)) FINAL_SEASONS = [2024, 2025] K_GRID = [1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0] TEAM_ALIAS = {"OAK": "LV", "SD": "LAC", "STL": "LA", "LAR": "LA", "JAX": "JAC"} EXPECTED_INPUT_HASHES = { "games.csv": "c9d3896dafb1ca89b0bd9196d7e5948b16d56cb258a145bd71ef771069c3a605", "stats_player_week_2018.csv": "c53021bcfc9f89d30edd6f1fe145a1d7d6ce469efeec75974fe9f0adac568642", "stats_player_week_2019.csv": "a05558126b1209b80fe11a8f18938cb35816370bf9990975a87de49e9eed7376", "stats_player_week_2020.csv": "05b992676faccc8940efbf6242cb76358069372d31d7c2e02c1a4acdcd3cbe18", "stats_player_week_2021.csv": "41915fb49238902ad1f129ebf0405b11a1e710454ae0fe8f7b3e4f9145875f48", "stats_player_week_2022.csv": "ad426c3fe5bf1cc30c3f137fdfe96d054e19d400879ee4413129da49fa7b54be", "stats_player_week_2023.csv": "f19cb71a5de0dce7fd09376026237c9ee9d5a93fe13815a2ea3ec2d37204cb17", "stats_player_week_2024.csv": "3ddc45a84f759aa348ce465ae001752c530575455717657cdfe1f8abfcdb4759", "stats_player_week_2025.csv": "e5e0615b3d96a3eaebfaee91e55afb4a4e7fe0caf057454177bcd7d6ad4bcfc2", } EXPECTED_COMPLETED_GAMES = { 2018: 256, 2019: 256, 2020: 256, 2021: 272, 2022: 271, 2023: 272, 2024: 272, 2025: 272, } SCHEDULE_FIELDS = {"game_id", "season", "game_type", "week", "gameday", "away_team", "away_score", "home_team", "home_score"} KEY_FIELDS = ["player_id", "season", "week", "game_id", "team", "position", "targets"] STATS_FIELDS = set(KEY_FIELDS + ["season_type"]) CODE_VERSION = "wr-target-share-stability-analysis-3.0" FIG_FILE = "figures/wr-target-share-error-curve-figure.png" TABLE_FILE = "tables/wr-target-share-main-results-table.csv" ASSET_FILE = "publication-assets.json" SCOPE = ("Findings describe historical forecast reliability for the 2019-2025 NFL seasons and do " "not constitute recommendations for 2026 players; bootstrap intervals describe " "uncertainty in average error, not individual-player prediction intervals.") OBS_COLUMNS = ["player_id", "season", "team", "window", "player_targets", "team_targets", "share", "outcome", "team_changed", "obs_last_date", "out_first_date", "obs_games", "out_games"] COMBINED_COMPARATOR_LABEL = "Prior-season share / earlier-season WR-mean fallback" BAND_LABEL = "Observed-share MAE: 95% pointwise player-cluster CI" TOLERANCE_LABEL = "Registered editorial MAE tolerance (6 pp)" FIGURE_TITLE = ("2019-2025 NFL wide receivers: next-four-team-game target-share forecast MAE\n" "First N completed regular-season team games -> immediately following four team games") FIGURE_X_LABEL = "First N completed regular-season team games" SAMPLE_NOTE = "n = eligible player-windows." TARGET_SHARE_NOTE = ("Target share = player summed targets / all reported player targets for that team over the same games " "(RB and TE included;\nuntargeted pass attempts excluded; targets summed before division).") UNCERTAINTY_NOTE = ("Band: observed-share mean MAE 95% pointwise player-cluster bootstrap CI " "(1,000 replicates; seed 271828).\nNot an individual-player prediction interval; " "intervals are not simultaneous across N; other-line intervals are not displayed.\n" "Player-only clustering does not separately account for dependence among receivers " "sharing team-game context.") COHORT_NOTE = ("Each N changes the forecast origin and immediately following outcome block; eligible cohorts vary. " "Connected pooled estimates\nare not a within-player causal trajectory or the isolated effect of adding games.") FIGURE_SIZE_INCHES = (14.0, 9.0) FIGURE_DPI = 120 FIGURE_WIDTH_PX = 1680 FIGURE_HEIGHT_PX = 1080 MINIMUM_DISPLAY_WIDTH_PX = 840 MINIMUM_EFFECTIVE_TEXT_PX = 8.0 MIN_ESSENTIAL_FONT_POINTS = 10.0 NOTE_FONT_SIZE = 10.0 LEGEND_FONT_SIZE = 10.0 SAMPLE_FONT_SIZE = 10.0 TABLE_CSV_COLUMNS = [ "first_n_completed_regular_season_team_games", "n_players", "n_player_windows", "mae_observed_share_pp", "mae_observed_share_ci_low_pp", "mae_observed_share_ci_high_pp", "mae_training_only_shrinkage_pp", "mae_prior_share_or_earlier_season_wr_mean_fallback_pp", "mae_combined_prior_fallback_ci_low_pp", "mae_combined_prior_fallback_ci_high_pp", "mae_earlier_season_wr_mean_baseline_pp", "earlier_season_wr_mean_fallback_player_windows_n", "improvement_combined_comparator_mae_minus_observed_share_mae_pp_positive_favors_observed_share", "improvement_combined_comparator_mae_minus_observed_share_mae_ci_low_pp", "improvement_combined_comparator_mae_minus_observed_share_mae_ci_high_pp", ] TABLE_READER_COLUMNS = [ "First N completed regular-season team games", "Unique NFL wide receivers", "Eligible player-windows (n)", "Observed-share error (pp)", "Observed-share 95% pointwise interval (pp)", "Training-only shrinkage error (pp)", "Prior-season share / earlier-season WR-mean fallback error (pp)", "Combined comparator 95% pointwise interval (pp)", "Earlier-season WR mean baseline error (pp)", "Earlier-season WR-mean fallback player-windows (n)", "Improvement: combined-comparator MAE - observed-share MAE (pp; positive favors observed share)", "Improvement 95% pointwise interval (pp; same subtraction direction)", ] class InputValidationError(ValueError): pass def sha256_file(path): h = hashlib.sha256() with path.open("rb") as fh: while True: block = fh.read(1024 * 1024) if not block: break h.update(block) return h.hexdigest() def clear_publication_artifacts(output_dir): output_dir = pathlib.Path(output_dir) for relative in (ASSET_FILE, FIG_FILE, TABLE_FILE): path = output_dir / relative if path.is_symlink() or path.is_file(): path.replace(path.with_name(path.name + ".stale")) elif path.exists(): raise InputValidationError("registered publication artifact path is not a file: %s" % path) def to_native(obj): if isinstance(obj, dict): return {str(k): to_native(v) for k, v in obj.items()} if isinstance(obj, (list, tuple)): return [to_native(v) for v in obj] if isinstance(obj, np.integer): return int(obj) if isinstance(obj, np.bool_): return bool(obj) if isinstance(obj, np.floating): obj = float(obj) if isinstance(obj, float) and math.isnan(obj): return None return obj def finite(value): return value is not None and isinstance(value, (int, float)) and math.isfinite(value) def normalize_text(series, upper=False): out = series.copy().astype(object) mask = out.notna() vals = out.loc[mask].astype(str).str.strip() if upper: vals = vals.str.upper() out.loc[mask] = vals return out def normalize_team(series): return normalize_text(series, upper=True).replace(TEAM_ALIAS) def integer_values(series, label): numeric = pd.to_numeric(series, errors="coerce") arr = numeric.to_numpy(dtype=float) bad = numeric.isna().to_numpy() | ~np.isfinite(arr) | (arr != np.floor(arr)) if bool(np.any(bad)): raise InputValidationError("%s contains missing or non-integer values (%d rows)" % (label, int(np.sum(bad)))) return numeric.astype(np.int64) def require_columns(frame, fields, label): missing = sorted(set(fields) - set(frame.columns)) if missing: raise InputValidationError("%s is missing required columns: %s" % (label, ", ".join(missing))) def require_semantic_phrases(text, phrases, label): normalized = " ".join(str(text).lower().split()) missing = [phrase for phrase in phrases if phrase.lower() not in normalized] if missing: raise InputValidationError("%s lacks required standalone semantics: %s" % (label, ", ".join(missing))) def load_inputs(ind, fixture_mode=False, return_manifest=False): ind = pathlib.Path(ind) required = ["games.csv"] + ["stats_player_week_%d.csv" % year for year in range(FIRST, LAST + 1)] missing = [name for name in required if not (ind / name).is_file()] if not fixture_mode and missing: raise InputValidationError("missing required production inputs: %s" % ", ".join(missing)) if fixture_mode and not (ind / "games.csv").is_file(): raise InputValidationError("fixture mode still requires games.csv") names = required if not fixture_mode else [name for name in required if (ind / name).is_file()] if not any(name.startswith("stats_player_week_") for name in names): raise InputValidationError("no weekly statistics files were supplied") entries = [] loaded = {} for name in names: path = ind / name digest = sha256_file(path) expected = EXPECTED_INPUT_HASHES.get(name) if not fixture_mode and digest != expected: raise InputValidationError("approved SHA-256 mismatch for %s" % name) frame = pd.read_csv(path, low_memory=False) require_columns(frame, SCHEDULE_FIELDS if name == "games.csv" else STATS_FIELDS, name) entries.append({"file": name, "sha256": digest, "bytes": int(path.stat().st_size), "rows": int(len(frame)), "approved_hash": bool(not fixture_mode and digest == expected)}) loaded[name] = frame sched = loaded["games.csv"] stats_frames = [] for year in range(FIRST, LAST + 1): name = "stats_player_week_%d.csv" % year if name not in loaded: continue frame = loaded[name].copy() numeric_year = pd.to_numeric(frame["season"], errors="coerce") wrong_file = numeric_year.notna() & (numeric_year != year) if bool(wrong_file.any()): raise InputValidationError("%s contains rows assigned to another season" % name) frame["_source_file"] = name frame["_source_row"] = np.arange(len(frame), dtype=np.int64) stats_frames.append(frame) stats = pd.concat(stats_frames, ignore_index=True, sort=False) manifest = { "code_version": CODE_VERSION, "mode": "fixture" if fixture_mode else "production", "required_files_complete": bool(not missing), "hashes_verified": bool(not fixture_mode and all(item["approved_hash"] for item in entries)), "approved": bool(not fixture_mode and not missing and all(item["approved_hash"] for item in entries)), "files": entries, } if return_manifest: return sched, stats, manifest return sched, stats def prepare(sched, stats): require_columns(sched, SCHEDULE_FIELDS, "schedule") require_columns(stats, STATS_FIELDS, "weekly statistics") sched = sched.copy().reset_index(drop=True) stats = stats.copy().reset_index(drop=True) stats["_row_id"] = np.arange(len(stats), dtype=np.int64) raw_stats_rows = int(len(stats)) raw_missing = {field: int(stats[field].isna().sum()) for field in KEY_FIELDS} sched["season"] = integer_values(sched["season"], "schedule.season") sched["week"] = integer_values(sched["week"], "schedule.week") sched["game_id"] = normalize_text(sched["game_id"]) sched["game_type"] = normalize_text(sched["game_type"], upper=True) sched["home_team"] = normalize_team(sched["home_team"]) sched["away_team"] = normalize_team(sched["away_team"]) if bool(sched[["game_id", "game_type", "home_team", "away_team"]].isna().any().any()): raise InputValidationError("schedule has missing canonical key values") if bool((sched["game_id"] == "").any()): raise InputValidationError("schedule has blank game_id values") duplicate_schedule = sched["game_id"].duplicated(keep=False) if bool(duplicate_schedule.any()): raise InputValidationError("schedule game_id is not unique (%d rows)" % int(duplicate_schedule.sum())) if bool((sched["home_team"] == sched["away_team"]).any()): raise InputValidationError("schedule contains a game with identical home and away teams") parsed_dates = pd.to_datetime(sched["gameday"], errors="coerce") bad_date = sched["gameday"].notna() & parsed_dates.isna() if bool(bad_date.any()): raise InputValidationError("schedule contains invalid gameday values") sched["_parsed_gameday"] = parsed_dates stats["season"] = integer_values(stats["season"], "statistics.season") stats["week"] = integer_values(stats["week"], "statistics.week") stats["season_type"] = normalize_text(stats["season_type"], upper=True) stats["game_id"] = normalize_text(stats["game_id"]) stats["team"] = normalize_team(stats["team"]) stats["player_id"] = normalize_text(stats["player_id"]) stats["position"] = normalize_text(stats["position"], upper=True) targets = pd.to_numeric(stats["targets"], errors="coerce") target_arr = targets.to_numpy(dtype=float) target_bad = targets.isna().to_numpy() | ~np.isfinite(target_arr) | (target_arr < 0) if bool(np.any(target_bad)): raise InputValidationError("statistics.targets contains missing, negative, or nonnumeric values") stats["targets"] = targets.astype(float) in_range = stats["season"].between(FIRST, LAST) excluded_out = ~in_range excluded_nonreg = in_range & (stats["season_type"] != "REG") candidate = stats[in_range & (stats["season_type"] == "REG")].copy() if bool(candidate[["game_id", "team", "season_type"]].isna().any().any()): raise InputValidationError("regular-season statistics have missing game or team keys") if bool((candidate["game_id"] == "").any() | (candidate["team"] == "").any()): raise InputValidationError("regular-season statistics have blank game or team keys") schedule_in_range = sched["season"].between(FIRST, LAST) schedule_regular = schedule_in_range & (sched["game_type"] == "REG") complete = (schedule_regular & sched["home_score"].notna() & sched["away_score"].notna() & sched["_parsed_gameday"].notna()) sc = sched[complete].copy() sc["gameday"] = sc["_parsed_gameday"].dt.strftime("%Y-%m-%d") sc = sc.sort_values(["gameday", "game_id"], kind="mergesort").reset_index(drop=True) canonical = sc.set_index("game_id", drop=False) game_map = canonical.to_dict("index") defect = {"unmatched_game": 0, "season_mismatch": 0, "week_mismatch": 0, "team_mismatch": 0} for row in candidate[["game_id", "season", "week", "team"]].itertuples(index=False): game = game_map.get(row.game_id) if game is None: defect["unmatched_game"] += 1 continue if int(row.season) != int(game["season"]): defect["season_mismatch"] += 1 if int(row.week) != int(game["week"]): defect["week_mismatch"] += 1 if row.team not in (game["home_team"], game["away_team"]): defect["team_mismatch"] += 1 if any(defect.values()): raise InputValidationError("canonical schedule reconciliation failed: " + json.dumps(defect, sort_keys=True)) keyed = candidate[candidate["player_id"].notna()].copy() duplicate_mask = keyed.duplicated(["game_id", "team", "player_id"], keep=False) duplicate_rows = int(duplicate_mask.sum()) position_conflicts = 0 if duplicate_rows: grouped = keyed[duplicate_mask].groupby(["game_id", "team", "player_id"], dropna=False) position_conflicts = int(sum(1 for _, group in grouped if group["position"].dropna().nunique() > 1)) multi_team = int(sum(1 for _, group in keyed.groupby(["game_id", "player_id"], dropna=False) if group["team"].nunique() > 1)) if duplicate_rows or multi_team: raise InputValidationError( "player-game key reconciliation failed: duplicate_rows=%d, position_conflicts=%d, " "multi_team_player_games=%d" % (duplicate_rows, position_conflicts, multi_team)) st = candidate.reset_index(drop=True) team_games = {} gameday = {} completed_counts = {} for row in sc.itertuples(index=False): season = int(row.season) completed_counts[season] = completed_counts.get(season, 0) + 1 gameday[row.game_id] = str(row.gameday) for team in (row.home_team, row.away_team): team_games.setdefault((season, team), []).append(row.game_id) stp = st[st["player_id"].notna()].copy() team_t = st.groupby(["game_id", "team"], sort=True)["targets"].sum().to_dict() pt = stp.groupby(["game_id", "team", "player_id"], sort=True)["targets"].sum().to_dict() rows_by_game = {} wr_rows = set() for row in stp[["game_id", "player_id", "position", "team", "targets"]].itertuples(index=False): rows_by_game.setdefault(row.game_id, []).append( (row.player_id, row.position, row.team, float(row.targets))) if row.position == "WR": wr_rows.add((row.game_id, row.team, row.player_id)) player_games = {} last_team = {} for row in stp[["game_id", "player_id", "team", "season"]].itertuples(index=False): day = gameday.get(row.game_id) if day is None: continue player_games.setdefault(row.player_id, []).append((day, row.team)) key = (row.player_id, int(row.season)) value = (day, str(row.game_id), str(row.team)) if key not in last_team or value > last_team[key]: last_team[key] = value team_season_targets = st.groupby(["season", "team"], sort=True)["targets"].sum().to_dict() player_season_team = stp.groupby(["player_id", "season", "team"], sort=True).agg( targets=("targets", "sum"), is_wr=("position", lambda values: bool((values == "WR").any())), ).reset_index() share_ps = {} wr_keys = set() for row in player_season_team.itertuples(index=False): total = team_season_targets.get((row.season, row.team)) share = float(row.targets) / float(total) if total and total > 0 else float("nan") key = (row.player_id, int(row.season), row.team) share_ps[key] = share if row.is_wr: wr_keys.add(key) mu = {} mu_src = {} for season in range(FIRST, LAST + 1): keys = [key for key in sorted(wr_keys) if key[1] < season and finite(share_ps.get(key))] mu[season] = float(np.mean([share_ps[key] for key in keys])) if keys else None mu_src[season] = int(max(key[1] for key in keys)) if keys else None missing_team_sides = set() missing_games = set() for row in sc.itertuples(index=False): for team in (row.home_team, row.away_team): if (row.game_id, team) not in team_t: missing_team_sides.add((row.game_id, team)) missing_games.add(row.game_id) schedule_reconciliation = { "schedule_rows": int(len(sched)), "postseason_rows_excluded": int((schedule_in_range & (sched["game_type"] != "REG")).sum()), "out_of_range_season_rows_excluded": int((~schedule_in_range).sum()), "in_range_unplayed_rows_excluded": int(schedule_regular.sum() - complete.sum()), "completed_regular_season_games_used": int(len(sc)), "completed_games_with_no_team_stats": int(len(missing_games)), "completed_team_games_with_no_team_stats": int(len(missing_team_sides)), "stats_games_not_in_completed_schedule": 0, "stats_rows_with_unmatched_team": 0, "stats_rows_with_season_mismatch": 0, "stats_rows_with_week_mismatch": 0, "duplicate_player_game_team_rows": 0, "position_conflict_keys": 0, } source_dispositions = { "raw_statistics_rows": raw_stats_rows, "excluded_out_of_range_season_rows": int(excluded_out.sum()), "excluded_non_regular_season_rows": int(excluded_nonreg.sum()), "excluded_canonical_join_or_key_failure_rows": 0, "accepted_canonical_regular_season_rows": int(len(st)), } disposition_sum = (source_dispositions["excluded_out_of_range_season_rows"] + source_dispositions["excluded_non_regular_season_rows"] + source_dispositions["excluded_canonical_join_or_key_failure_rows"] + source_dispositions["accepted_canonical_regular_season_rows"]) if disposition_sum != raw_stats_rows: raise AssertionError("source row dispositions do not reconcile") return { "team_games": team_games, "gameday": gameday, "team_t": team_t, "pt": pt, "rows_by_game": rows_by_game, "wr_rows": wr_rows, "player_games": player_games, "share_ps": share_ps, "mu": mu, "mu_src": mu_src, "st": st, "last_team": last_team, "wr_keys": wr_keys, "raw_missing_by_field": raw_missing, "source_dispositions": source_dispositions, "schedule_reconciliation": schedule_reconciliation, "completed_counts": completed_counts, "primary_team_games": set(), "non_disjoint_player_windows": 0, } def validate_production_completeness(pr, manifest): if (not manifest.get("approved") or not manifest.get("required_files_complete") or not manifest.get("hashes_verified")): raise InputValidationError("production inputs are not the complete approved manifest") actual = {season: int(pr["completed_counts"].get(season, 0)) for season in range(FIRST, LAST + 1)} if actual != EXPECTED_COMPLETED_GAMES: raise InputValidationError("completed regular-season schedule is incomplete: expected %s, got %s" % (json.dumps(EXPECTED_COMPLETED_GAMES, sort_keys=True), json.dumps(actual, sort_keys=True))) def build_obs(pr): records = [] kept = 0 primary_team_games = set() non_disjoint_players = 0 drops = {"incomplete_season_team_windows": 0, "zero_denominator_team_windows": 0, "unresolved_team_game_joins": 0, "non_disjoint_team_windows": 0} for (season, team), games in sorted(pr["team_games"].items()): for window in WINDOWS: if len(games) < window + OUT_LEN: drops["incomplete_season_team_windows"] += 1 continue observed_games = games[:window] outcome_games = games[window:window + OUT_LEN] all_games = observed_games + outcome_games if any((game, team) not in pr["team_t"] for game in all_games): drops["unresolved_team_game_joins"] += 1 continue observed_total = sum(pr["team_t"][(game, team)] for game in observed_games) outcome_total = sum(pr["team_t"][(game, team)] for game in outcome_games) if observed_total <= 0 or outcome_total <= 0: drops["zero_denominator_team_windows"] += 1 continue eligible = {player for game in observed_games for player, position, row_team, _ in pr["rows_by_game"].get(game, []) if position == "WR" and row_team == team} observed_last = pr["gameday"][observed_games[-1]] outcome_first = pr["gameday"][outcome_games[0]] if not observed_last < outcome_first: drops["non_disjoint_team_windows"] += 1 non_disjoint_players += len(eligible) continue kept += 1 primary_team_games.update((game, team) for game in all_games) outcome_last = pr["gameday"][outcome_games[-1]] for player in sorted(eligible): player_observed = sum(pr["pt"].get((game, team, player), 0.0) for game in observed_games) player_outcome = sum(pr["pt"].get((game, team, player), 0.0) for game in outcome_games) changed = any(outcome_first <= day <= outcome_last and other_team != team for day, other_team in pr["player_games"].get(player, [])) records.append({ "player_id": player, "season": int(season), "team": team, "window": int(window), "player_targets": float(player_observed), "team_targets": float(observed_total), "share": float(player_observed / observed_total), "outcome": float(player_outcome / outcome_total), "team_changed": bool(changed), "obs_last_date": observed_last, "out_first_date": outcome_first, "obs_games": list(observed_games), "out_games": list(outcome_games), }) pr["primary_team_games"] = primary_team_games pr["non_disjoint_player_windows"] = int(non_disjoint_players) return pd.DataFrame(records, columns=OBS_COLUMNS), drops, kept def attach_priors(frame, share_ps, mu, last_team): prior = [] fallback = [] wr_mean = [] source = [] for row in frame.itertuples(index=False): mean = mu.get(row.season) mean_value = float(mean) if mean is not None else float("nan") wr_mean.append(mean_value) previous = last_team.get((row.player_id, row.season - 1)) value = share_ps.get((row.player_id, row.season - 1, previous[2])) if previous else None if previous is None: prior.append(mean_value) fallback.append(True) source.append("wr_mean_no_prior_history") elif not finite(value): prior.append(mean_value) fallback.append(True) source.append("wr_mean_prior_zero_denominator") else: prior.append(float(value)) fallback.append(False) source.append("prior_same_team" if previous[2] == row.team else "prior_other_team") out = frame.copy() out["prior_pred"] = prior out["prior_fallback"] = fallback out["prior_source"] = source out["wr_mean_pred"] = wr_mean return out def fit_k(frame, mu): values = {} training_max = {} for season in EVAL_SEASONS: mean = mu.get(season) training = frame[(frame["season"] < season) & frame["outcome"].notna()] for window in WINDOWS: sample = training[training["window"] == window] training_max[(season, window)] = int(sample["season"].max()) if len(sample) else None if mean is None or len(sample) == 0: values[(season, window)] = 8.0 continue best_error = None best_k = 8.0 for k in K_GRID: prediction = ((sample["player_targets"] + k * mean) / (sample["team_targets"] + k)) error = float((prediction - sample["outcome"]).abs().mean()) if best_error is None or error < best_error: best_error = error best_k = k values[(season, window)] = float(best_k) return values, training_max def mae_pair(frame): if len(frame) == 0: return {"n": 0, "mae_observed_pp": None, "improvement_over_prior_pp": None} outcome = frame["outcome"].to_numpy(dtype=float) observed_error = np.abs(frame["share"].to_numpy(dtype=float) - outcome) prior_error = np.abs(frame["prior_pred"].to_numpy(dtype=float) - outcome) return {"n": int(len(frame)), "mae_observed_pp": float(observed_error.mean() * 100.0), "improvement_over_prior_pp": float((prior_error.mean() - observed_error.mean()) * 100.0)} def boot_ci(observed_error, prior_error, players): unique = np.unique(players) indices = {player: np.where(players == player)[0] for player in unique} player_list = np.array(sorted(indices)) rng = np.random.default_rng(SEED) observed_means = [] prior_means = [] improvements = [] for _ in range(B): sampled = rng.choice(player_list, size=len(player_list), replace=True) selected = np.concatenate([indices[player] for player in sampled]) observed = float(observed_error[selected].mean()) prior = float(prior_error[selected].mean()) observed_means.append(observed) prior_means.append(prior) improvements.append(prior - observed) def interval(values): arr = np.asarray(values, dtype=float) return [float(np.percentile(arr, 2.5) * 100.0), float(np.percentile(arr, 97.5) * 100.0)] return interval(observed_means), interval(prior_means), interval(improvements) def window_results(evaluation, k_values): results = {} for window in WINDOWS: sample = evaluation[evaluation["window"] == window] if len(sample) == 0: results[window] = None continue shrinkage = [] for row in sample.itertuples(index=False): k = k_values[(row.season, window)] shrinkage.append((row.player_targets + k * row.wr_mean_pred) / (row.team_targets + k)) outcome = sample["outcome"].to_numpy(dtype=float) observed_error = np.abs(sample["share"].to_numpy(dtype=float) - outcome) shrinkage_error = np.abs(np.asarray(shrinkage) - outcome) prior_error = np.abs(sample["prior_pred"].to_numpy(dtype=float) - outcome) mean_error = np.abs(sample["wr_mean_pred"].to_numpy(dtype=float) - outcome) observed_ci, prior_ci, improvement_ci = boot_ci( observed_error, prior_error, sample["player_id"].to_numpy()) results[window] = { "n": int(len(sample)), "n_players": int(sample["player_id"].nunique()), "mae_observed_pp": float(observed_error.mean() * 100.0), "mae_observed_ci_pp": observed_ci, "mae_shrinkage_pp": float(shrinkage_error.mean() * 100.0), "mae_prior_baseline_pp": float(prior_error.mean() * 100.0), "mae_prior_baseline_ci_pp": prior_ci, "mae_wr_mean_baseline_pp": float(mean_error.mean() * 100.0), "improvement_over_prior_pp": float((prior_error.mean() - observed_error.mean()) * 100.0), "improvement_over_prior_ci_pp": improvement_ci, "prior_fallback_n": int(sample["prior_fallback"].sum()), "prior_source_counts": {str(key): int(value) for key, value in sample["prior_source"].value_counts().sort_index().items()}, } return results def sensitivities(evaluation, results): common = set() for _, season_frame in evaluation.groupby("season"): intersection = None for window in WINDOWS: sample = season_frame[season_frame["window"] == window] keys = set(zip(sample["player_id"], sample["season"], sample["team"])) intersection = keys if intersection is None else intersection & keys if intersection: common |= intersection output = {"common_cohort": {}, "team_change": {}, "low_team_targets": {}, "precision_target": {}, "per_season": {}} for window in WINDOWS: sample = evaluation[evaluation["window"] == window] common_mask = np.asarray([(player, season, team) in common for player, season, team in zip(sample["player_id"], sample["season"], sample["team"])], dtype=bool) output["common_cohort"][str(window)] = mae_pair(sample[common_mask]) changed = sample[sample["team_changed"]] unchanged = sample[~sample["team_changed"]] output["team_change"][str(window)] = dict( mae_pair(unchanged), n_team_changed=int(len(changed))) output["low_team_targets"][str(window)] = mae_pair( sample[sample["team_targets"] >= 20.0 * window]) result = results.get(window) if result is None: output["precision_target"][str(window)] = None else: output["precision_target"][str(window)] = { "mae_le_4pp": bool(result["mae_observed_pp"] <= 4.0), "mae_le_6pp": bool(result["mae_observed_pp"] <= 6.0), "mae_le_8pp": bool(result["mae_observed_pp"] <= 8.0), "improvement_ge_0_5pp": bool(result["improvement_over_prior_pp"] >= 0.5), "meets_registered_target": bool(result["mae_observed_pp"] <= 6.0 and result["improvement_over_prior_pp"] >= 0.5), } for season in FINAL_SEASONS: output["per_season"][str(season)] = { str(window): mae_pair(evaluation[(evaluation["season"] == season) & (evaluation["window"] == window)]) for window in WINDOWS} return output def reconcile_row_use(pr, frame): st = pr["st"] accepted_ids = set(int(value) for value in st["_row_id"]) primary_ids = { int(row_id) for row_id, game_id, team in st[["_row_id", "game_id", "team"]].itertuples( index=False, name=None) if (game_id, team) in pr["primary_team_games"] } mean_pairs = set() wr_rows = st[(st["player_id"].notna()) & (st["position"] == "WR") & (st["season"] < LAST)] mean_pairs.update((int(season), team) for season, team in zip(wr_rows["season"], wr_rows["team"])) prior_pairs = set() prior_players = set() evaluated = frame[frame["season"].isin(EVAL_SEASONS)] for row in evaluated.itertuples(index=False): prior_players.add((row.player_id, row.season - 1)) previous = pr["last_team"].get((row.player_id, row.season - 1)) if previous is not None: prior_pairs.add((row.season - 1, previous[2])) baseline_ids = set() for row_id, season, team, player_id in st[ ["_row_id", "season", "team", "player_id"]].itertuples(index=False, name=None): pair = (int(season), team) player_pair = (player_id, int(season)) if pd.notna(player_id) else None if pair in mean_pairs or pair in prior_pairs or player_pair in prior_players: baseline_ids.add(int(row_id)) both = primary_ids & baseline_ids primary_only = primary_ids - baseline_ids baseline_only = baseline_ids - primary_ids wholly_unused = accepted_ids - primary_ids - baseline_ids roles = { "primary_window_only_rows": len(primary_only), "prior_or_training_only_rows": len(baseline_only), "primary_and_prior_or_training_rows": len(both), "wholly_unused_accepted_rows": len(wholly_unused), } if sum(roles.values()) != len(accepted_ids): raise AssertionError("accepted-row analytical roles do not reconcile") source = dict(pr["source_dispositions"]) used_rows = len(primary_ids | baseline_ids) excluded_rows = source["raw_statistics_rows"] - used_rows reconciled = bool( source["raw_statistics_rows"] == source["excluded_out_of_range_season_rows"] + source["excluded_non_regular_season_rows"] + source["excluded_canonical_join_or_key_failure_rows"] + source["accepted_canonical_regular_season_rows"] and sum(roles.values()) == source["accepted_canonical_regular_season_rows"] and used_rows + excluded_rows == source["raw_statistics_rows"]) if not reconciled: raise AssertionError("raw-input and analytical-use ledgers do not reconcile") return { "source_dispositions": source, "analytic_roles_within_accepted_rows": roles, "analytically_used_rows": int(used_rows), "excluded_from_all_estimators_rows": int(excluded_rows), "accepted_missing_player_id_rows": int(st["player_id"].isna().sum()), "accepted_missing_position_rows": int(st["position"].isna().sum()), "reconciled": True, } def validate_figure_artists(fig, axis, windows): require_semantic_phrases(axis.get_title(), [ "2019-2025 nfl wide receivers", "next-four-team-game target-share forecast mae", "first n completed regular-season team games", "immediately following four team games", ], "figure title") note_text = "\n".join(item.get_text() for item in fig.texts) require_semantic_phrases(note_text, [ "n = eligible player-windows", "all reported player targets", "rb and te included", "untargeted pass attempts excluded", "not an individual-player prediction interval", "not simultaneous across n", "other-line intervals are not displayed", "does not separately account for dependence among receivers sharing team-game context", "eligible cohorts vary", "not a within-player causal trajectory", ], "in-image figure notes") expected_labels = ["n=%d" % int(window_n) for window_n in windows] actual_labels = [item.get_text() for item in axis.texts] if actual_labels != expected_labels: raise InputValidationError("figure sample-size labels do not match eligible player-windows") legend = axis.get_legend() if legend is None: raise InputValidationError("figure legend is missing") legend_labels = [item.get_text() for item in legend.get_texts()] required_legend = [BAND_LABEL, "Observed-window share forecast", "Training-only shrinkage estimate", COMBINED_COMPARATOR_LABEL, "Earlier-season WR mean baseline", TOLERANCE_LABEL] if legend_labels != required_legend: raise InputValidationError("figure legend does not identify all registered estimators") essential_artists = list(fig.texts) + list(axis.texts) + list(legend.get_texts()) if not essential_artists: raise InputValidationError("figure has no essential explanatory text") minimum_font = min(float(item.get_fontsize()) for item in essential_artists) if minimum_font < MIN_ESSENTIAL_FONT_POINTS: raise InputValidationError("essential figure text is below the %.1f-point minimum" % MIN_ESSENTIAL_FONT_POINTS) width_inches, height_inches = fig.get_size_inches() native_width = int(round(float(width_inches) * FIGURE_DPI)) native_height = int(round(float(height_inches) * FIGURE_DPI)) if native_width != FIGURE_WIDTH_PX or native_height != FIGURE_HEIGHT_PX: raise InputValidationError("figure canvas dimensions violate the publication contract") display_scale = float(MINIMUM_DISPLAY_WIDTH_PX) / float(native_width) effective_px = minimum_font * float(FIGURE_DPI) / 72.0 * display_scale if effective_px < MINIMUM_EFFECTIVE_TEXT_PX: raise InputValidationError("essential figure text is illegible at minimum publication width") return { "x_positions": [float(value) for value in windows], "title": axis.get_title(), "x_label": axis.get_xlabel(), "tolerance_y_pp": 6.0, "tolerance_label": TOLERANCE_LABEL, "uncertainty_band_label": BAND_LABEL, "uncertainty_note": UNCERTAINTY_NOTE, "target_share_note": TARGET_SHARE_NOTE, "cohort_and_origin_note": COHORT_NOTE, "sample_size_definition": SAMPLE_NOTE, "sample_size_labels": ["N=%d: n=%d eligible player-windows" % (window, window_n) for window, window_n in zip(WINDOWS, windows)], "legend_labels": legend_labels, "minimum_essential_font_points": minimum_font, "minimum_display_width_px": MINIMUM_DISPLAY_WIDTH_PX, "minimum_effective_text_px": float(effective_px), "native_width_px": native_width, "native_height_px": native_height, } def validate_figure_raster(output_path, spec): image = plt.imread(output_path) if image.ndim not in (2, 3): raise InputValidationError("generated figure raster has an unsupported shape") height, width = int(image.shape[0]), int(image.shape[1]) if width != spec["native_width_px"] or height != spec["native_height_px"]: raise InputValidationError("generated figure raster dimensions do not match metadata") if width < 2 * spec["minimum_display_width_px"]: raise InputValidationError("generated figure lacks the tested publication downscaling margin") def make_figure(results, output_path): windows = [window for window in WINDOWS if results.get(window) is not None] x_values = np.asarray(windows, dtype=float) fig, axis = plt.subplots(figsize=FIGURE_SIZE_INCHES) try: if windows: observed = [results[window]["mae_observed_pp"] for window in windows] low = [results[window]["mae_observed_ci_pp"][0] for window in windows] high = [results[window]["mae_observed_ci_pp"][1] for window in windows] axis.fill_between(x_values, low, high, alpha=0.2, color="tab:blue", label=BAND_LABEL) axis.plot(x_values, observed, "o-", color="tab:blue", label="Observed-window share forecast") axis.plot(x_values, [results[window]["mae_shrinkage_pp"] for window in windows], "s--", color="tab:green", label="Training-only shrinkage estimate") axis.plot(x_values, [results[window]["mae_prior_baseline_pp"] for window in windows], "^:", color="tab:orange", label=COMBINED_COMPARATOR_LABEL) axis.plot(x_values, [results[window]["mae_wr_mean_baseline_pp"] for window in windows], "v-.", color="tab:red", label="Earlier-season WR mean baseline") for value, window in zip(x_values, windows): axis.text(value, 0.985, "n=%d" % results[window]["n"], transform=axis.get_xaxis_transform(), ha="center", va="top", fontsize=SAMPLE_FONT_SIZE) axis.axhline(6.0, color="gray", linewidth=0.8, linestyle="--", label=TOLERANCE_LABEL) axis.set_xticks(x_values) axis.set_xticklabels([str(window) for window in windows]) axis.set_xlabel(FIGURE_X_LABEL, fontsize=11) axis.set_ylabel("Held-out mean absolute error (percentage points)", fontsize=11) axis.set_title(FIGURE_TITLE, fontsize=14) axis.tick_params(axis="both", labelsize=10) axis.margins(y=0.15) axis.legend(fontsize=LEGEND_FONT_SIZE, ncol=2, loc="best") fig.text(0.01, 0.285, SAMPLE_NOTE, fontsize=NOTE_FONT_SIZE, ha="left", va="bottom", fontweight="bold") fig.text(0.01, 0.205, TARGET_SHARE_NOTE, fontsize=NOTE_FONT_SIZE, ha="left", va="bottom") fig.text(0.01, 0.095, UNCERTAINTY_NOTE, fontsize=NOTE_FONT_SIZE, ha="left", va="bottom") fig.text(0.01, 0.018, COHORT_NOTE, fontsize=NOTE_FONT_SIZE, ha="left", va="bottom") fig.tight_layout(rect=(0, 0.31, 1, 1)) sample_counts = [results[window]["n"] for window in windows] spec = validate_figure_artists(fig, axis, sample_counts) spec["x_positions"] = [float(value) for value in x_values] spec["sample_size_labels"] = [ "N=%d: n=%d eligible player-windows" % (window, results[window]["n"]) for window in windows] fig.savefig(output_path, dpi=FIGURE_DPI, metadata={"Software": CODE_VERSION}) validate_figure_raster(output_path, spec) return spec finally: plt.close(fig) def validate_table_semantics(caption, reader_columns, csv_columns, reader_rows): require_semantic_phrases(caption, [ "2019-2025 nfl wide receivers", "next-four-team-game target-share forecast", "first n completed regular-season team games", "immediately following four team games", "eligible player-windows", "improvement equals combined-comparator mae minus observed-share mae", "positive values favor observed share", "not individual-player prediction intervals", "does not separately account for dependence among receivers sharing team-game context", "eligible cohorts vary", ], "table caption") reader_text = " | ".join(reader_columns) require_semantic_phrases(reader_text, [ "eligible player-windows (n)", "wr-mean fallback player-windows (n)", "combined-comparator mae - observed-share mae", "positive favors observed share", ], "table reader headers") csv_text = " | ".join(csv_columns).lower() for required in ("fallback_player_windows_n", "comparator_mae_minus_observed_share_mae", "positive_favors_observed_share"): if required not in csv_text: raise InputValidationError("table CSV headers lack required standalone semantics: %s" % required) if len(reader_columns) != 12: raise InputValidationError("publication table must contain exactly 12 reader columns") if len(reader_rows) != len(WINDOWS): raise InputValidationError("publication table must contain all six registered windows") if any(len(row) != len(reader_columns) for row in reader_rows): raise InputValidationError("publication table rows do not match reader columns") def check(name, passed, details): return {"name": name, "passed": bool(passed), "details": details} def classify_registered_hypothesis(results): classification = {} for window in WINDOWS: result = results.get(window) if result is None: classification[str(window)] = None continue qualifies_numerically = bool( finite(result.get("mae_observed_pp")) and finite(result.get("improvement_over_prior_pp")) and result["mae_observed_pp"] <= 6.0 and result["improvement_over_prior_pp"] >= 0.5) classification[str(window)] = bool(window >= 4 and qualifies_numerically) return classification def registered_hypothesis_predicate_ok(hypothesis, results): expected = classify_registered_hypothesis(results) if not isinstance(hypothesis, dict) or set(hypothesis) != set(expected): return False for key, expected_value in expected.items(): actual = hypothesis[key] if expected_value is None: if actual is not None: return False elif (not isinstance(actual, (bool, np.bool_)) or bool(actual) != expected_value): return False return True def publication_reasons(manifest, row_reconciliation, results, observation_drops, temporal_ok, per_season_ok, audit, hypothesis=None): reasons = [] if not registered_hypothesis_predicate_ok(hypothesis, results): reasons.append("registered_hypothesis_predicate_failed") if not manifest.get("approved"): reasons.append("unapproved_fixture_inputs") counts = {str(window): (results[window]["n"] if results.get(window) else 0) for window in WINDOWS} low = [key for key, value in counts.items() if value < 100] if low: reasons.append("fewer_than_100_observations:" + ",".join(low)) if any(results.get(window) is None for window in WINDOWS): reasons.append("full_error_curve_unavailable") if observation_drops["non_disjoint_player_windows"] > 0: reasons.append("non_disjoint_observations_dropped") if not row_reconciliation.get("reconciled"): reasons.append("row_reconciliation_failed") if not temporal_ok: reasons.append("temporal_validation_failed") if not per_season_ok: reasons.append("final_season_sensitivity_unavailable") audit_checks = [audit["temporal_validation"], audit["uncertainty"]] for section in ("baseline_checks", "sensitivity_checks", "leakage_checks", "withholding_checks"): audit_checks.extend(audit[section]) for item in audit_checks: if not item["passed"]: reasons.append("research_audit_failed:" + str(item["name"])) if not audit["missingness"]["reconciled"]: reasons.append("research_audit_failed:Missingness reconciliation") return reasons def publication_copy(results): first = results[WINDOWS[0]] last = results[WINDOWS[-1]] summary = ("At N=1, observed-share MAE is %.2f pp versus %.2f pp for the combined " "prior/fallback comparator; at N=8 the values are %.2f and %.2f pp." % (first["mae_observed_pp"], first["mae_prior_baseline_pp"], last["mae_observed_pp"], last["mae_prior_baseline_pp"])) definition = ("Target share is the player's summed targets divided by all reported player " "targets for that team over the same games, including RB and TE targets and " "excluding untargeted pass attempts; targets are summed before division.") fallback = ("The combined comparator uses the player's prior-season share when usable and " "otherwise a training-only earlier-season WR mean fallback.") limitations = ("Each N has a different forecast origin and immediately following four-game " "outcome block, and eligible cohorts vary, so connected pooled estimates are " "not a within-player causal trajectory or the isolated effect of adding games.") uncertainty = ("The shaded band applies only to observed-share mean MAE and is a 95% pointwise " "player-cluster bootstrap interval (1,000 replicates; seed 271828), not an " "individual-player prediction interval and not a simultaneous interval across N; " "other-line intervals are not displayed. Player-cluster intervals do not " "separately account for dependence among receivers sharing team-game context.") alt = ("Chart of held-out next-four-team-game target-share forecast mean absolute error for " "2019-2025 NFL wide receivers. Forecasts use the first N completed regular-season team " "games and are scored on the immediately following four team games. In-image n labels " "are eligible player-windows. " + summary + " Lines show the observed-window share " "forecast, training-only shrinkage estimate, prior-season share / earlier-season WR-mean " "fallback comparator, and separate earlier-season WR mean baseline. " + uncertainty + " " + definition + " " + limitations) figure_caption = ("2019-2025 NFL wide receivers: held-out next-four-team-game target-share " "forecast mean absolute error in percentage points after each team's first N " "completed regular-season games, scored on the immediately following four " "team games. In-image n labels are eligible player-windows. " + definition + " " + fallback + " " + uncertainty + " " + limitations + " The dashed line is the registered 6-percentage-point editorial tolerance.") table_caption = ("2019-2025 NFL wide receivers: held-out next-four-team-game target-share " "forecast MAE and paired improvement after the first N completed regular-season " "team games, with each forecast scored on the immediately following four team " "games. Rows report eligible player-windows. Improvement equals combined-comparator " "MAE minus observed-share MAE, so positive values favor observed share. " + definition + " " + fallback + " The earlier-season WR mean is also reported " "as a separate baseline. Intervals are 95% pointwise player-cluster bootstrap " "intervals for mean error or paired mean improvement, not individual-player " "prediction intervals or simultaneous intervals across N. The player-cluster " "method does not separately account for dependence among receivers sharing " "team-game context. " + limitations) return alt, figure_caption, table_caption def validate_publication_outputs(output_dir, figure_spec): output_dir = pathlib.Path(output_dir) asset_path = output_dir / ASSET_FILE figure_path = output_dir / FIG_FILE table_path = output_dir / TABLE_FILE if not asset_path.is_file() or not figure_path.is_file() or not table_path.is_file(): raise InputValidationError("registered publication outputs are incomplete") assets = json.loads(asset_path.read_text(encoding="utf-8")) if len(assets.get("figures", [])) != 1 or len(assets.get("tables", [])) != 1: raise InputValidationError("publication asset inventory is not the registered one-figure one-table set") figure = assets["figures"][0] table = assets["tables"][0] if figure.get("id") != "wr-target-share-error-curve-figure" or figure.get("file") != FIG_FILE: raise InputValidationError("publication figure identity or path changed") if table.get("id") != "wr-target-share-main-results-table": raise InputValidationError("publication table identity changed") if figure.get("sha256") != sha256_file(figure_path): raise InputValidationError("publication figure hash does not match the generated raster") if figure.get("width") != figure_spec["native_width_px"] or \ figure.get("height") != figure_spec["native_height_px"]: raise InputValidationError("publication figure metadata dimensions are incorrect") require_semantic_phrases(figure.get("alt", "") + " " + figure.get("caption", ""), [ "2019-2025 nfl wide receivers", "next-four-team-game target-share forecast", "n labels are eligible player-windows", "not an individual-player prediction interval", "dependence among receivers sharing team-game context", "eligible cohorts vary", ], "figure accessibility metadata") csv_frame = pd.read_csv(table_path) if list(csv_frame.columns) != TABLE_CSV_COLUMNS: raise InputValidationError("downloadable table columns do not match the semantic contract") validate_table_semantics(table.get("caption", ""), table.get("columns", []), list(csv_frame.columns), table.get("rows", [])) if len(csv_frame) != len(WINDOWS): raise InputValidationError("downloadable table does not contain all six windows") validate_figure_raster(figure_path, figure_spec) def emit_publication(results, output_dir): output_dir = pathlib.Path(output_dir) clear_publication_artifacts(output_dir) (output_dir / "figures").mkdir(parents=True, exist_ok=True) (output_dir / "tables").mkdir(parents=True, exist_ok=True) try: figure_spec = make_figure(results, output_dir / FIG_FILE) rows = [] for window in WINDOWS: result = results.get(window) if result is None: continue rows.append([ window, result["n_players"], result["n"], round(result["mae_observed_pp"], 2), round(result["mae_observed_ci_pp"][0], 2), round(result["mae_observed_ci_pp"][1], 2), round(result["mae_shrinkage_pp"], 2), round(result["mae_prior_baseline_pp"], 2), round(result["mae_prior_baseline_ci_pp"][0], 2), round(result["mae_prior_baseline_ci_pp"][1], 2), round(result["mae_wr_mean_baseline_pp"], 2), result["prior_fallback_n"], round(result["improvement_over_prior_pp"], 2), round(result["improvement_over_prior_ci_pp"][0], 2), round(result["improvement_over_prior_ci_pp"][1], 2), ]) pd.DataFrame(rows, columns=TABLE_CSV_COLUMNS).to_csv(output_dir / TABLE_FILE, index=False) reader_rows = [ [row[0], row[1], row[2], row[3], "%.2f to %.2f" % (row[4], row[5]), row[6], row[7], "%.2f to %.2f" % (row[8], row[9]), row[10], row[11], row[12], "%.2f to %.2f" % (row[13], row[14])] for row in rows ] alt, figure_caption, table_caption = publication_copy(results) validate_table_semantics(table_caption, TABLE_READER_COLUMNS, TABLE_CSV_COLUMNS, reader_rows) figure_hash = sha256_file(output_dir / FIG_FILE) assets = { "figures": [{ "id": "wr-target-share-error-curve-figure", "file": FIG_FILE, "alt": alt, "caption": figure_caption, "width": FIGURE_WIDTH_PX, "height": FIGURE_HEIGHT_PX, "sha256": figure_hash, }], "tables": [{ "id": "wr-target-share-main-results-table", "caption": table_caption, "columns": TABLE_READER_COLUMNS, "rows": to_native(reader_rows), }], } (output_dir / ASSET_FILE).write_text( json.dumps(to_native(assets), indent=2, allow_nan=False), encoding="utf-8") validate_publication_outputs(output_dir, figure_spec) except Exception: clear_publication_artifacts(output_dir) raise def finalize_publication(results, output_dir, reasons): if reasons: clear_publication_artifacts(output_dir) else: emit_publication(results, output_dir) def main(argv=None): parser = argparse.ArgumentParser(description="WR target-share stability analysis") parser.add_argument("--input", required=True) parser.add_argument("--output", required=True) parser.add_argument("--fixture-mode", action="store_true", help="Allow partial synthetic inputs; publication remains withheld") args = parser.parse_args(argv) input_dir = pathlib.Path(args.input) output_dir = pathlib.Path(args.output) clear_publication_artifacts(output_dir) sched, stats, manifest = load_inputs(input_dir, fixture_mode=args.fixture_mode, return_manifest=True) pr = prepare(sched, stats) if not args.fixture_mode: validate_production_completeness(pr, manifest) output_dir.mkdir(parents=True, exist_ok=True) (output_dir / "figures").mkdir(parents=True, exist_ok=True) (output_dir / "tables").mkdir(parents=True, exist_ok=True) observations, drops, kept = build_obs(pr) frame = attach_priors(observations, pr["share_ps"], pr["mu"], pr["last_team"]) k_values, training_max = fit_k(frame, pr["mu"]) evaluation = frame[frame["season"].isin(EVAL_SEASONS) & frame["prior_pred"].notna() & frame["wr_mean_pred"].notna() & frame["outcome"].notna()].copy() results = window_results(evaluation, k_values) sensitivity = sensitivities(evaluation, results) row_reconciliation = reconcile_row_use(pr, frame) if row_reconciliation["analytically_used_rows"] < 1: raise InputValidationError("no source rows were used by any registered estimator") disjoint = bool((frame["obs_last_date"] < frame["out_first_date"]).all()) if len(frame) else True eligibility_ok = all(any((game, row.team, row.player_id) in pr["wr_rows"] for game in row.obs_games) for row in frame.itertuples(index=False)) if len(frame) else True training_ok = (all(value is None or value < season for (season, _), value in training_max.items()) and all(pr["mu_src"][season] is None or pr["mu_src"][season] < season for season in EVAL_SEASONS)) temporal_ok = bool(training_ok and disjoint and (len(evaluation) == 0 or int(evaluation["season"].min()) > FIRST)) all_windows = all(results.get(window) is not None for window in WINDOWS) counts = {str(window): (results[window]["n"] if results.get(window) else 0) for window in WINDOWS} hypothesis = classify_registered_hypothesis(results) string_windows = [str(window) for window in WINDOWS] common = sensitivity["common_cohort"] team_change = sensitivity["team_change"] low_targets = sensitivity["low_team_targets"] precision = sensitivity["precision_target"] per_season = sensitivity["per_season"] common_ok = (len({common[key]["n"] for key in string_windows}) == 1 and all(0 < common[key]["n"] <= counts[key] for key in string_windows)) team_change_ok = all(team_change[key]["n"] + team_change[key]["n_team_changed"] == counts[key] for key in string_windows) low_targets_ok = all(low_targets[key]["n"] <= counts[key] and (low_targets[key]["n"] == 0 or finite(low_targets[key]["mae_observed_pp"])) for key in string_windows) precision_ok = all(precision[key] is not None and (not precision[key]["mae_le_4pp"] or precision[key]["mae_le_6pp"]) and (not precision[key]["mae_le_6pp"] or precision[key]["mae_le_8pp"]) for key in string_windows) per_season_ok = True for season in FINAL_SEASONS: for window in WINDOWS: key = str(window) expected = int(((evaluation["season"] == season) & (evaluation["window"] == window)).sum()) reported = per_season[str(season)][key]["n"] if expected <= 0 or reported != expected or reported > counts[key]: per_season_ok = False if per_season_ok: per_season_ok = all( sum(per_season[str(season)][key]["n"] for season in FINAL_SEASONS) <= counts[key] for key in string_windows) team_window_total = len(pr["team_games"]) * len(WINDOWS) drops_ok = kept + sum(drops.values()) == team_window_total defaulted = ["%d_%d" % (season, window) for (season, window), value in sorted(training_max.items()) if value is None] source_counts = ({str(key): int(value) for key, value in evaluation["prior_source"].value_counts().sort_index().items()} if len(evaluation) else {}) fallback_ok = bool((evaluation["prior_fallback"].to_numpy() == evaluation["prior_source"].str.startswith("wr_mean").to_numpy()).all()) \ if len(evaluation) else True prior_ok = bool(all_windows and fallback_ok and np.isfinite(evaluation["prior_pred"].to_numpy(dtype=float)).all() and all(results[window]["prior_fallback_n"] <= results[window]["n"] for window in WINDOWS)) mean_ok = bool(all_windows and all(finite(results[window]["mae_wr_mean_baseline_pp"]) for window in WINDOWS)) observed_ok = bool(all_windows and all(finite(results[window]["mae_observed_pp"]) for window in WINDOWS)) uncertainty_ok = bool(all_windows and B == 1000 and all( len(results[window][field]) == 2 and finite(results[window][field][0]) and finite(results[window][field][1]) and results[window][field][0] <= results[window][field][1] for window in WINDOWS for field in ("mae_observed_ci_pp", "mae_prior_baseline_ci_pp", "improvement_over_prior_ci_pp"))) scope_ok = bool("2026" in SCOPE and "not" in SCOPE and all(value is not None for value in hypothesis.values())) observation_drops = { "non_disjoint_player_windows": int(pr["non_disjoint_player_windows"])} missingness = { "input_rows": int(pr["source_dispositions"]["raw_statistics_rows"]), "analyzed_rows": int(row_reconciliation["analytically_used_rows"]), "excluded_rows": int(row_reconciliation["excluded_from_all_estimators_rows"]), "reconciled": bool(row_reconciliation["reconciled"]), "missing_by_field": pr["raw_missing_by_field"], } audit = { "temporal_validation": check( "Rolling temporal validation", temporal_ok, "Verified earlier-season fitting sources, strictly ordered feature and outcome dates, " "and exclusion of the training-only 2018 season from evaluation."), "baseline_checks": [ check("Prior-season target-share baseline with training-only fallback", prior_ok, "Prior shares use the last dated prior-season team and its own denominator; " "fallbacks are earlier-season WR means. Sources: " + json.dumps(source_counts)), check("Earlier-season WR mean baseline", mean_ok, "The earlier-season WR mean is evaluated separately for every registered window."), check("Observed-window share without shrinkage", observed_ok, "Raw observed target share is evaluated with finite MAE for every window."), ], "uncertainty": check( "Player-cluster bootstrap intervals", uncertainty_ok, "1000 player-cluster bootstrap replicates use seed 271828; intervals are pointwise " "mean-error and paired-improvement intervals, not player prediction intervals. Player " "clustering does not separately account for shared team-game dependence."), "missingness": missingness, "sensitivity_checks": [ check("Common-cohort sensitivity", common_ok, "Player-season-team cohorts eligible in every window; counts: " + json.dumps({key: common[key]["n"] for key in string_windows})), check("Team-change sensitivity", team_change_ok, "Unchanged plus changed observations reconcile to each window; changed counts: " + json.dumps({key: team_change[key]["n_team_changed"] for key in string_windows})), check("Low-team-target sensitivity", low_targets_ok, "The threshold is 20 team targets per observed game; counts: " + json.dumps({key: low_targets[key]["n"] for key in string_windows})), check("Precision-target sensitivity", precision_ok, "MAE tolerances at 4, 6, and 8 points and improvement of 0.5 points are evaluated " "as editorial thresholds and checked for logical monotonicity."), check("Per-season sensitivity", per_season_ok, "Positive observations are required for every window in both 2024 and 2025; " "reported season-window counts are checked against pooled evaluation records."), ], "leakage_checks": [ check("Feature-outcome window disjointness", disjoint and observation_drops["non_disjoint_player_windows"] == 0, "Retained observations have feature-window end dates before outcome-window start dates; " "invalid observations are dropped and counted before evaluation."), check("Training-only parameter fitting", training_ok, "Shrinkage fitting frames and WR-mean sources have maximum seasons below each " "evaluation season; defaulted season-windows are reported."), check("Window-only WR eligibility", eligibility_ok, "Every retained player has a WR-position row for his team in the observation window."), ], "withholding_checks": [ check("Minimum 100 observations per window", all(value >= 100 for value in counts.values()), "Per-window eligible observation counts: " + json.dumps(counts)), check("Incomplete-window drops reported", drops_ok, "Kept and mutually exclusive team-window drops reconcile to %d candidate team-windows: %s" % (team_window_total, json.dumps(dict(drops, analyzed_team_windows=kept)))), check("Full error curve for all six windows", all_windows, "The required 1, 2, 3, 4, 6, and 8 game windows are all required before publication."), check("Scope limitation on game-count claims", scope_ok, "The scope is historical 2019-2025 reliability and explicitly excludes 2026 recommendations."), ], } reasons = publication_reasons(manifest, row_reconciliation, results, observation_drops, temporal_ok, per_season_ok, audit, hypothesis) payload = { "series_id": "wr-target-share-stability", "estimand": "Held-out mean absolute error (percentage points) of next-four-team-game " "target share predictions by observation window length (1, 2, 3, 4, 6, 8 team " "games), with paired improvement over the prior-season baseline.", "population": "NFL wide receivers appearing in regular-season weekly player statistics " "2018-2025; WR eligibility established only within the observation window.", "outcome_definition": "Player summed targets divided by all reported player targets for " "his team summed over the same four team games (RB and TE targets " "included; untargeted pass attempts excluded); targets summed before " "dividing; full-season share never used as the outcome.", "evaluation_seasons": EVAL_SEASONS, "input_manifest": manifest, "input_validation": { "code_version": CODE_VERSION, "approved": bool(manifest["approved"]), "required_files_complete": bool(manifest["required_files_complete"]), "hashes_verified": bool(manifest["hashes_verified"]), "canonical_schedule_and_keys_reconciled": True, }, "row_reconciliation": row_reconciliation, "windows": {str(window): results[window] for window in WINDOWS}, "sensitivities": sensitivity, "hypothesis_evaluation": hypothesis, "drops": drops, "observation_drops": observation_drops, "analyzed_team_windows": int(kept), "schedule_reconciliation": pr["schedule_reconciliation"], "shrinkage_k": {"%d_%d" % (season, window): k_values[(season, window)] for season in EVAL_SEASONS for window in WINDOWS}, "shrinkage_k_defaulted": defaulted, "prior_baseline_rule": "Prior-season share of the player's last dated prior-season team " "(latest gameday, then game_id, then team), divided by that team's " "own prior-season target total; earlier-season WR mean only when no " "usable prior-season history exists.", "prior_source_counts": source_counts, "scope_statement": SCOPE, "publication_status": {"status": "withheld" if reasons else "published", "reasons": reasons}, "research_audit": audit, } finalize_publication(results, output_dir, reasons) (output_dir / "analysis.json").write_text( json.dumps(to_native(payload), indent=2, allow_nan=False), encoding="utf-8") if __name__ == "__main__": main()