Spaces:
Paused
Paused
AIBRUH
Claude Opus 5.5
Add IB 2.0 pass-rush matchup score, calibrated from 2023-2025 play by play
6fe7db5 | """ | |
| IB 2.0 deep dive -- what actually rattles a QB, 2023-2025, play by play. | |
| Joins nflverse play_by_play + pbp_participation (was_pressure, pass rushers, | |
| time_to_throw, coverage) + FTN charting (blitzers, interception-worthy throws, | |
| QB out of pocket). Raw files live in lake/bronze/nflverse/ (gitignored). | |
| Sections | |
| A play level: which conditions produce interceptions / INT-worthy throws | |
| B rattle carryover: does being hit earlier make a QB worse later, with the | |
| current play's pressure held fixed | |
| C defense level: pre-game IB 1.0 (sacks/INTs/points, the original formula) | |
| vs IB 2.0 (true pressure, 4-man pressure, blitz, INT-worthy forced) -- | |
| which one predicts the opponent's passing game, trained 2023-24, | |
| scored on 2025 untouched | |
| D in-game: first-half pressure as a live signal for the second half | |
| E receivers: where the targets go when the pocket collapses | |
| Run from the repo root: | |
| backend/.venv/bin/python scripts/ib_deep_dive.py | |
| """ | |
| from __future__ import annotations | |
| import json | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| from sklearn.linear_model import LinearRegression, LogisticRegression | |
| from sklearn.metrics import roc_auc_score | |
| REPO = Path(__file__).resolve().parent.parent | |
| BRONZE = REPO / "lake" / "bronze" / "nflverse" | |
| GOLD = REPO / "lake" / "gold" / "nfl" | |
| SEASONS = (2023, 2024, 2025) | |
| TRAIN, TEST = (2023, 2024), 2025 | |
| PRIOR_DB = 150 # dropbacks of prior-season weight before a defense's own games take over | |
| PBP_COLS = [ | |
| "game_id", "play_id", "season", "week", "season_type", "posteam", "defteam", "home_team", | |
| "passer_player_id", "passer_player_name", "receiver_player_id", "receiver_player_name", | |
| "qb_dropback", "pass_attempt", "sack", "qb_hit", "qb_scramble", "qb_spike", "qb_kneel", "interception", | |
| "complete_pass", "passing_yards", "receiving_yards", "air_yards", "down", "ydstogo", "yardline_100", "qtr", | |
| "game_half", "score_differential", "half_seconds_remaining", "wp", "shotgun", "no_huddle", | |
| "epa", "cpoe", "two_point_attempt", "spread_line", "total_line", "fumble_lost", "pass_touchdown", | |
| ] | |
| # ── load ───────────────────────────────────────────────────────────────────── | |
| PART_COLS = ["was_pressure", "number_of_pass_rushers", "time_to_throw", "defense_coverage_type", | |
| "defense_man_zone_type", "defenders_in_box"] | |
| FTN_COLS = ["n_blitzers", "ftn_rushers", "is_interception_worthy", "is_qb_out_of_pocket", "is_play_action", | |
| "is_screen_pass", "is_throw_away", "is_qb_fault_sack"] | |
| def load_dropbacks(seasons=SEASONS) -> pd.DataFrame: | |
| """Participation / FTN files are optional per season (in-season years may only have play-by-play).""" | |
| frames = [] | |
| for s in seasons: | |
| pbp = pd.read_csv(BRONZE / f"play_by_play_{s}.csv.gz", usecols=PBP_COLS, low_memory=False) | |
| pbp = pbp[(pbp.season_type == "REG") & (pbp.qb_dropback == 1) & (pbp.two_point_attempt.fillna(0) == 0) | |
| & (pbp.qb_spike.fillna(0) == 0) & (pbp.qb_kneel.fillna(0) == 0)] | |
| df = pbp | |
| if (BRONZE / f"pbp_participation_{s}.csv").exists(): | |
| part = pd.read_csv(BRONZE / f"pbp_participation_{s}.csv", low_memory=False, | |
| usecols=["nflverse_game_id", "play_id"] + PART_COLS) | |
| df = df.merge(part.rename(columns={"nflverse_game_id": "game_id"}), on=["game_id", "play_id"], how="left") | |
| if (BRONZE / f"ftn_charting_{s}.csv").exists(): | |
| ftn = pd.read_csv(BRONZE / f"ftn_charting_{s}.csv", low_memory=False, | |
| usecols=["nflverse_game_id", "nflverse_play_id", "n_blitzers", "n_pass_rushers", | |
| "is_interception_worthy", "is_qb_out_of_pocket", "is_play_action", | |
| "is_screen_pass", "is_throw_away", "is_qb_fault_sack"]) | |
| ftn = ftn.rename(columns={"nflverse_game_id": "game_id", "nflverse_play_id": "play_id", | |
| "n_pass_rushers": "ftn_rushers"}) | |
| df = df.merge(ftn, on=["game_id", "play_id"], how="left") | |
| frames.append(df.reindex(columns=list(df.columns) + [c for c in PART_COLS + FTN_COLS if c not in df.columns])) | |
| df = pd.concat(frames, ignore_index=True) | |
| for c in ["was_pressure", "is_interception_worthy", "is_qb_out_of_pocket", "is_play_action", | |
| "is_screen_pass", "is_throw_away", "is_qb_fault_sack"]: | |
| df[c] = df[c].map({True: 1, False: 0, "True": 1, "False": 0, 1: 1, 0: 0}).astype("float") | |
| df["rushers"] = df["number_of_pass_rushers"].fillna(df["ftn_rushers"]) | |
| df["blitz"] = (df["n_blitzers"].fillna(0) > 0).astype(int) | |
| df["pressure"] = df["was_pressure"].fillna(((df.sack == 1) | (df.qb_hit == 1)).astype(float)) | |
| df["pressure4"] = ((df.pressure == 1) & (df.rushers <= 4)).astype(int) # pressure without sending extra men | |
| df["hit_or_sack"] = ((df.sack == 1) | (df.qb_hit == 1)).astype(int) | |
| df["int_worthy"] = df["is_interception_worthy"].fillna(df["interception"]) | |
| df["pass_yds"] = df["passing_yards"].fillna(0) | |
| return df | |
| # ── A: conditions behind interceptions ─────────────────────────────────────── | |
| def rate_table(df: pd.DataFrame, col: str, label: str) -> list[dict]: | |
| g = df.groupby(col, observed=True).agg(dropbacks=("interception", "size"), ints=("interception", "sum"), | |
| int_worthy=("int_worthy", "sum"), sacks=("sack", "sum"), | |
| epa=("epa", "mean")) | |
| base = df.interception.mean() | |
| out = [] | |
| for k, r in g.iterrows(): | |
| if r.dropbacks < 300: | |
| continue | |
| out.append({"factor": label, "bucket": str(k), "dropbacks": int(r.dropbacks), | |
| "int_rate": round(r.ints / r.dropbacks, 4), "int_worthy_rate": round(r.int_worthy / r.dropbacks, 4), | |
| "sack_rate": round(r.sacks / r.dropbacks, 4), "epa_per_db": round(r.epa, 3), | |
| "int_lift": round((r.ints / r.dropbacks) / base, 2)}) | |
| return out | |
| def section_a(df: pd.DataFrame) -> dict: | |
| d = df.copy() | |
| d["pressure_b"] = d.pressure.map({1.0: "PRESSURED", 0.0: "CLEAN"}) | |
| d["blitzers_b"] = pd.cut(d.n_blitzers, [-1, 0, 1, 2, 20], labels=["0", "1", "2", "3+"]) | |
| d["ttt_b"] = pd.cut(d.time_to_throw, [0, 2.0, 2.5, 3.0, 3.5, 20], labels=["<2.0s", "2.0-2.5s", "2.5-3.0s", "3.0-3.5s", "3.5s+"]) | |
| d["down_dist"] = np.select( | |
| [d.down == 3, d.down == 4, d.ydstogo >= 10], | |
| [np.where(d.ydstogo >= 7, "3rd & 7+", "3rd & <7"), "4th down", "1st/2nd & 10+"], "1st/2nd & <10") | |
| d["game_state"] = np.select( | |
| [(d.qtr >= 4) & (d.score_differential <= -8), (d.qtr >= 4) & (d.score_differential < 0), | |
| d.score_differential >= 8, d.score_differential <= -8], | |
| ["4Q trailing 9+", "4Q trailing 1-8", "leading 8+", "trailing 8+ (Q1-3)"], "within one score") | |
| d["coverage"] = d.defense_coverage_type.where(d.defense_coverage_type.isin( | |
| ["COVER_0", "COVER_1", "COVER_2", "COVER_3", "COVER_4", "COVER_6", "2_MAN", "COVER_9"])) | |
| d["pocket"] = d.is_qb_out_of_pocket.map({1.0: "OUT OF POCKET", 0.0: "IN POCKET"}) | |
| d["pa"] = d.is_play_action.map({1.0: "PLAY ACTION", 0.0: "STRAIGHT DROPBACK"}) | |
| table = [] | |
| for col, label in [("pressure_b", "pressure"), ("blitzers_b", "blitzers"), ("ttt_b", "time_to_throw"), | |
| ("down_dist", "down_distance"), ("game_state", "game_state"), ("coverage", "coverage"), | |
| ("pocket", "pocket"), ("pa", "play_action")]: | |
| table += rate_table(d, col, label) | |
| # Pressure x game state: the "desperate and hunted" combination | |
| combo = d.assign(combo=d.pressure_b + " · " + d.game_state) | |
| table += rate_table(combo, "combo", "pressure_x_state") | |
| # Logistic model on INT-worthy throws (less noisy than INTs: drops and lucky picks wash out) | |
| att = d[d.pass_attempt == 1].copy() | |
| feats = pd.DataFrame({ | |
| "pressure": att.pressure.fillna(0), | |
| "blitzers": att.n_blitzers.fillna(0).clip(0, 4), | |
| "rushers5plus": (att.rushers >= 5).astype(int), | |
| "out_of_pocket": att.is_qb_out_of_pocket.fillna(0), | |
| "play_action": att.is_play_action.fillna(0), | |
| "third_long": ((att.down == 3) & (att.ydstogo >= 7)).astype(int), | |
| "trailing_4q": ((att.qtr >= 4) & (att.score_differential < 0)).astype(int), | |
| "wp_low": (att.wp < 0.2).astype(int), | |
| "air_yards_15plus": (att.air_yards >= 15).astype(int), | |
| "cover_0_1": att.defense_coverage_type.isin(["COVER_0", "COVER_1"]).astype(int), | |
| "cover_2_4_6": att.defense_coverage_type.isin(["COVER_2", "COVER_4", "COVER_6", "2_MAN"]).astype(int), | |
| }) | |
| y = att.int_worthy.fillna(0).astype(int) | |
| tr, te = att.season.isin(TRAIN), att.season == TEST | |
| m = LogisticRegression(max_iter=2000).fit(feats[tr], y[tr]) | |
| p = m.predict_proba(feats[te])[:, 1] | |
| q = pd.qcut(p, 10, labels=False, duplicates="drop") | |
| top, bot = y[te].values[q == q.max()].mean(), y[te].values[q == q.min()].mean() | |
| return { | |
| "table": table, | |
| "odds_ratios": {f: round(float(np.exp(c)), 2) for f, c in zip(feats.columns, m.coef_[0])}, | |
| "auc_2025": round(float(roc_auc_score(y[te], p)), 3), | |
| "top_decile_rate": round(float(top), 4), "bottom_decile_rate": round(float(bot), 4), | |
| "base_rate": round(float(y[te].mean()), 4), | |
| } | |
| # ── B: rattle carryover ────────────────────────────────────────────────────── | |
| def section_b(df: pd.DataFrame) -> dict: | |
| d = df.sort_values(["game_id", "play_id"]).copy() | |
| grp = d.groupby(["game_id", "passer_player_id"], sort=False) | |
| d["prior_hits"] = grp.hit_or_sack.cumsum() - d.hit_or_sack | |
| d["prior_db"] = grp.cumcount() | |
| d["hit_last_db"] = grp.hit_or_sack.shift(1).fillna(0) | |
| d["prior_hit_rate"] = d.prior_hits / d.prior_db.replace(0, np.nan) | |
| att = d[(d.pass_attempt == 1) & (d.prior_db >= 8)].copy() | |
| att["rattled"] = pd.cut(att.prior_hit_rate, [-0.01, 0.08, 0.16, 0.25, 1.0], | |
| labels=["0-8% hit so far", "8-16%", "16-25%", "25%+ hit so far"]) | |
| def summarize(frame): | |
| return {"attempts": int(len(frame)), "int_worthy_rate": round(float(frame.int_worthy.mean()), 4), | |
| "int_rate": round(float(frame.interception.mean()), 4), | |
| "ttt": round(float(frame.time_to_throw.mean()), 2), "epa": round(float(frame.epa.mean()), 3), | |
| "cpoe": round(float(frame.cpoe.mean()), 2)} | |
| # Only CLEAN pockets: if the QB still misfires with nobody on him, that is the rattle | |
| clean = att[att.pressure == 0] | |
| by_rattle = {str(k): summarize(g) for k, g in clean.groupby("rattled", observed=True)} | |
| after_hit = {"clean pocket, hit on previous dropback": summarize(clean[clean.hit_last_db == 1]), | |
| "clean pocket, not hit on previous dropback": summarize(clean[clean.hit_last_db == 0])} | |
| X = pd.DataFrame({"prior_hit_rate": att.prior_hit_rate.fillna(0), "pressure_now": att.pressure.fillna(0), | |
| "third_long": ((att.down == 3) & (att.ydstogo >= 7)).astype(int), | |
| "trailing": (att.score_differential < 0).astype(int), "qtr": att.qtr}) | |
| m = LogisticRegression(max_iter=2000).fit(X, att.int_worthy.fillna(0).astype(int)) | |
| # Within-game test: same QB, same game, same defense -- clean-pocket throws right after a hit vs the rest. | |
| # Demeaning by QB-game removes defense quality, weather, game plan and QB talent. | |
| c = clean.assign(qg=clean.game_id + clean.passer_player_id) | |
| c = c[c.groupby("qg").hit_last_db.transform("nunique") == 2] | |
| fe = {} | |
| for col in ["int_worthy", "epa", "cpoe"]: | |
| y = c[col] - c.groupby("qg")[col].transform("mean") | |
| x = c.hit_last_db - c.groupby("qg").hit_last_db.transform("mean") | |
| ok = y.notna() | |
| beta = float((x[ok] * y[ok]).sum() / (x[ok] ** 2).sum()) | |
| resid = y[ok] - beta * x[ok] | |
| se = float(np.sqrt((resid ** 2).sum() / (ok.sum() - 1) / (x[ok] ** 2).sum())) | |
| fe[col] = {"effect_after_hit": round(beta, 4), "se": round(se, 4), "t": round(beta / se, 2)} | |
| return {"within_game_after_hit_clean_pocket": fe, "within_game_n": int(len(c)), | |
| "clean_pocket_by_prior_hit_rate": by_rattle, "clean_pocket_after_hit": after_hit, | |
| "odds_ratio_per_+10pct_prior_hit_rate": round(float(np.exp(m.coef_[0][0] * 0.10)), 3), | |
| "odds_ratio_pressure_now": round(float(np.exp(m.coef_[0][1])), 2)} | |
| # ── C: defense IB 1.0 vs IB 2.0, pre-game, predicting the opponent's passing game ── | |
| DEF_STATS = { # name -> (numerator column, denominator column) | |
| "pressure_rate": ("pressure", "db"), | |
| "pressure4_rate": ("pressure4", "rush4_db"), | |
| "blitz_rate": ("blitz", "db"), | |
| "sack_rate": ("sack", "db"), | |
| "hit_rate": ("hit_or_sack", "db"), | |
| "int_worthy_rate": ("int_worthy", "att"), | |
| "int_rate": ("interception", "att"), | |
| "epa_per_db": ("epa", "db"), | |
| "pass_yds_per_game": ("pass_yds", "games"), | |
| "points_allowed": ("points", "games"), | |
| } | |
| def defense_games(df: pd.DataFrame) -> pd.DataFrame: | |
| d = df.assign(att=df.pass_attempt, db=1, rush4_db=(df.rushers <= 4).astype(int)) | |
| g = d.groupby(["season", "week", "game_id", "defteam", "posteam"]).agg( | |
| db=("db", "sum"), att=("att", "sum"), rush4_db=("rush4_db", "sum"), pressure=("pressure", "sum"), | |
| pressure4=("pressure4", "sum"), blitz=("blitz", "sum"), sack=("sack", "sum"), hit_or_sack=("hit_or_sack", "sum"), | |
| int_worthy=("int_worthy", "sum"), interception=("interception", "sum"), epa=("epa", "sum"), | |
| pass_yds=("pass_yds", "sum"), pass_td=("pass_touchdown", "sum"), spread_line=("spread_line", "first"), | |
| total_line=("total_line", "first"), home_team=("home_team", "first"), | |
| ).reset_index() | |
| g["games"] = 1 | |
| # points allowed from team_game_stats (already in the lake) | |
| tgs = pd.read_csv(GOLD / "team_game_stats.csv")[["game_id", "team", "points_against"]] | |
| g = g.merge(tgs.rename(columns={"team": "defteam", "points_against": "points"}), on=["game_id", "defteam"], how="left") | |
| g["points"] = g.points.fillna(g.points.mean()) | |
| return g.sort_values(["defteam", "season", "week"]).reset_index(drop=True) | |
| def pregame(g: pd.DataFrame, key: str, stats: dict) -> pd.DataFrame: | |
| """Each row gets the team's rate from games BEFORE this one, shrunk toward last season (or league mean).""" | |
| out = g.copy() | |
| cols = {c for pair in stats.values() for c in pair} | |
| season_tot = g.groupby([key, "season"])[list(cols)].sum() | |
| league = {name: g[n].sum() / g[dn].sum() for name, (n, dn) in stats.items()} | |
| grp = out.groupby([key, "season"], sort=False) | |
| for name, (n, dn) in stats.items(): | |
| cum_n = grp[n].cumsum() - out[n] | |
| cum_d = grp[dn].cumsum() - out[dn] | |
| prev = [season_tot.loc[(t, s - 1)] if (t, s - 1) in season_tot.index else None | |
| for t, s in zip(out[key], out.season)] | |
| prior = np.array([p[n] / p[dn] if p is not None and p[dn] else league[name] for p in prev]) | |
| prior = 0.67 * prior + 0.33 * league[name] # regress last season a third of the way to the mean | |
| w = PRIOR_DB if dn in ("db", "att", "rush4_db") else 4 # 4 games of prior weight for per-game stats | |
| out[f"pre_{name}"] = (cum_n + prior * w) / (cum_d + w) | |
| return out | |
| def zscore(s: pd.Series, by: pd.Series) -> pd.Series: | |
| return s.groupby(by).transform(lambda x: (x - x.mean()) / (x.std() or 1)) | |
| def section_c(df: pd.DataFrame) -> tuple[dict, pd.DataFrame]: | |
| g = pregame(defense_games(df), "defteam", DEF_STATS) | |
| # offense side: this offense's own pre-game passing level and pressure allowed | |
| off = g.rename(columns={"defteam": "_d", "posteam": "offteam"}) | |
| off = pregame(off, "offteam", {"off_pass_yds": ("pass_yds", "games"), "off_pressure_allowed": ("pressure", "db"), | |
| "off_int_worthy": ("int_worthy", "att"), "off_sack_allowed": ("sack", "db")}) | |
| g = g.merge(off[["game_id", "offteam", "pre_off_pass_yds", "pre_off_pressure_allowed", "pre_off_int_worthy", | |
| "pre_off_sack_allowed"]].rename(columns={"offteam": "posteam"}), on=["game_id", "posteam"]) | |
| wk = g.season.astype(str) + "-" + g.week.astype(str) | |
| # IB 1.0 -- the original lake formula's ingredients: sacks, interceptions, points allowed | |
| g["ib1_raw"] = zscore(g.pre_sack_rate, wk) + zscore(g.pre_int_rate, wk) - zscore(g.pre_points_allowed, wk) | |
| # IB 2.0 -- true pressure, pressure with four, INT-worthy forced, EPA allowed; blitz is NOT rewarded on its own | |
| g["ib2_raw"] = (1.0 * zscore(g.pre_pressure_rate, wk) + 0.8 * zscore(g.pre_pressure4_rate, wk) | |
| + 0.6 * zscore(g.pre_int_worthy_rate, wk) - 0.8 * zscore(g.pre_epa_per_db, wk)) | |
| # MATCHUP -- the defense's pressure against this offense's protection | |
| g["exp_pressure"] = g.pre_pressure_rate * g.pre_off_pressure_allowed / (g.pressure.sum() / g.db.sum()) | |
| g["ib2_matchup_raw"] = g.ib2_raw + 1.2 * zscore(g.pre_off_pressure_allowed, wk) + 0.6 * zscore(g.pre_off_int_worthy, wk) | |
| for c in ["ib1", "ib2", "ib2_matchup"]: | |
| g[c] = pd.qcut(g[f"{c}_raw"].rank(method="first"), 5, labels=[1, 2, 3, 4, 5]).astype(int) | |
| g["team_spread"] = np.where(g.posteam == g.home_team, g.spread_line, -g.spread_line) # + = offense favored | |
| g["pass_under"] = (g.pass_yds < g.pre_off_pass_yds).astype(int) | |
| g["actual_pressure_rate"] = g.pressure / g.db | |
| test = g[g.season == TEST] | |
| buckets = {} | |
| for c in ["ib1", "ib2", "ib2_matchup"]: | |
| t = test.groupby(c).agg(games=("pass_yds", "size"), pass_yds=("pass_yds", "mean"), | |
| vs_own_avg=("pass_yds", lambda x: 0), under_rate=("pass_under", "mean"), | |
| ints=("interception", "mean"), sacks=("sack", "mean"), | |
| pressure=("actual_pressure_rate", "mean")) | |
| t["vs_own_avg"] = test.groupby(c).apply(lambda x: (x.pass_yds - x.pre_off_pass_yds).mean(), include_groups=False) | |
| buckets[c] = [{"score": int(k), **{kk: round(float(v), 3) for kk, v in r.items()}} for k, r in t.iterrows()] | |
| # Incremental value beyond the offense's own level and the Vegas lines | |
| base = ["pre_off_pass_yds", "team_spread", "total_line"] | |
| tr, te = g.season.isin(TRAIN), g.season == TEST | |
| def r2(cols, target): | |
| mdl = LinearRegression().fit(g.loc[tr, cols].fillna(0), g.loc[tr, target]) | |
| pred = mdl.predict(g.loc[te, cols].fillna(0)) | |
| resid = g.loc[te, target] - pred | |
| return round(float(1 - (resid ** 2).sum() / ((g.loc[te, target] - g.loc[te, target].mean()) ** 2).sum()), 4), mdl | |
| incr = {} | |
| for target in ["pass_yds", "interception", "sack", "int_worthy"]: | |
| b, _ = r2(base, target) | |
| v1, _ = r2(base + ["ib1_raw"], target) | |
| v2, _ = r2(base + ["ib2_raw"], target) | |
| vm, m = r2(base + ["ib2_matchup_raw"], target) | |
| incr[target] = {"lines_only": b, "plus_ib1": v1, "plus_ib2": v2, "plus_ib2_matchup": vm, | |
| "yds_per_1sd_ib2_matchup": round(float(m.coef_[-1]), 2) if target == "pass_yds" else None} | |
| # Does the pre-game rate predict the in-game pressure? (stability of each ingredient) | |
| stab = {n: round(float(test[f"pre_{n}"].corr(test[a] / test[b])), 3) | |
| for n, (a, b) in DEF_STATS.items() if b in ("db", "att", "rush4_db")} | |
| stab["exp_pressure(matchup)"] = round(float(test.exp_pressure.corr(test.actual_pressure_rate)), 3) | |
| # Blitz-heavy vs pressure-with-four defenses: two different animals | |
| test = test.assign(style=np.select([(test.pre_blitz_rate > test.pre_blitz_rate.quantile(0.67)), | |
| (test.pre_pressure4_rate > test.pre_pressure4_rate.quantile(0.67))], | |
| ["BLITZ-HEAVY", "PRESSURE WITH FOUR"], "OTHER")) | |
| style = test.groupby("style").agg(games=("pass_yds", "size"), pass_yds_vs_own=("pass_yds", "mean"), | |
| ints=("interception", "mean"), pass_td=("pass_td", "mean"), | |
| epa=("epa", "sum")).reset_index() | |
| style["pass_yds_vs_own"] = test.groupby("style").apply(lambda x: (x.pass_yds - x.pre_off_pass_yds).mean(), | |
| include_groups=False).values | |
| style["epa"] = test.groupby("style").apply(lambda x: x.epa.sum() / x.db.sum(), include_groups=False).values | |
| # Is it already in the lines? Residual after the lines-only model, by IB 2.0 matchup quintile. | |
| _, lm = r2(base, "pass_yds") | |
| # Residuals are centered within each season so a league-wide scoring shift doesn't read as an IB effect. | |
| g["pass_yds_vs_lines"] = g.pass_yds - lm.predict(g[base].fillna(0)) | |
| g["pass_yds_vs_lines"] -= g.groupby("season").pass_yds_vs_lines.transform("median") | |
| _, sm = r2(base, "sack") | |
| g["sacks_vs_lines"] = g.sack - sm.predict(g[base].fillna(0)) | |
| g["sacks_vs_lines"] -= g.groupby("season").sacks_vs_lines.transform("mean") | |
| vs_lines = {} | |
| for season_set, label in [((TEST,), "2025_holdout"), (SEASONS, "2023_2025_all")]: | |
| sub = g[g.season.isin(season_set)] | |
| vs_lines[label] = sub.groupby("ib2_matchup").agg( | |
| games=("pass_yds", "size"), pass_yds_vs_lines=("pass_yds_vs_lines", "mean"), | |
| under_lines_rate=("pass_yds_vs_lines", lambda x: (x < 0).mean()), | |
| sacks_vs_lines=("sacks_vs_lines", "mean"), sacks=("sack", "mean"), ints=("interception", "mean"), | |
| dropbacks=("db", "mean")).round(3).reset_index().to_dict("records") | |
| return {"buckets_2025": buckets, "incremental_r2_2025": incr, "pregame_to_ingame_corr_2025": stab, | |
| "style_2025": style.round(3).to_dict("records"), "vs_lines": vs_lines}, g | |
| # ── D: first-half pressure as a live second-half signal ────────────────────── | |
| def section_d(df: pd.DataFrame) -> dict: | |
| d = df[df.qtr <= 4] | |
| h = d.groupby(["game_id", "posteam", "game_half"]).agg(db=("pressure", "size"), score=("score_differential", "last"), | |
| pressure=("pressure", "mean"), | |
| hits=("hit_or_sack", "mean"), iw=("int_worthy", "sum"), | |
| ints=("interception", "sum"), yds=("pass_yds", "sum"), | |
| epa=("epa", "mean")).reset_index() | |
| h1 = h[h.game_half == "Half1"].set_index(["game_id", "posteam"]) | |
| h2 = h[h.game_half == "Half2"].set_index(["game_id", "posteam"]) | |
| j = h1.join(h2, lsuffix="_1h", rsuffix="_2h", how="inner") | |
| j = j[(j.db_1h >= 12) & (j.db_2h >= 8)] | |
| j["bucket"] = pd.qcut(j.hits_1h, 4, labels=["low 1H hits", "mid-low", "mid-high", "high 1H hits"]) | |
| t = j.groupby("bucket", observed=True).agg(games=("db_2h", "size"), hit_rate_1h=("hits_1h", "mean"), | |
| pressure_2h=("pressure_2h", "mean"), iw_per_100_2h=("iw_2h", "sum"), | |
| ints_2h=("ints_2h", "mean"), epa_2h=("epa_2h", "mean")) | |
| t["iw_per_100_2h"] = j.groupby("bucket", observed=True).apply(lambda x: 100 * x.iw_2h.sum() / x.db_2h.sum()).values | |
| t["pass_yds_2h"] = j.groupby("bucket", observed=True).yds_2h.mean().values | |
| t["dropbacks_2h"] = j.groupby("bucket", observed=True).db_2h.mean().values | |
| t["halftime_score_diff"] = j.groupby("bucket", observed=True).score_1h.mean().values | |
| # Same comparison among games within one score at the half -- strips out "trailing teams throw more" | |
| close = j[j.score_1h.abs() <= 8] | |
| tc = close.groupby("bucket", observed=True).agg(games=("db_2h", "size"), epa_2h=("epa_2h", "mean"), | |
| pass_yds_2h=("yds_2h", "mean"), ints_2h=("ints_2h", "mean")) | |
| return {"by_first_half_hits": [{"bucket": str(k), **{kk: round(float(v), 3) for kk, v in r.items()}} | |
| for k, r in t.iterrows()], | |
| "close_at_half": [{"bucket": str(k), **{kk: round(float(v), 3) for kk, v in r.items()}} | |
| for k, r in tc.iterrows()], | |
| "corr_1h_hits_2h_epa": round(float(j.hits_1h.corr(j.epa_2h)), 3), | |
| "corr_1h_pressure_2h_pressure": round(float(j.pressure_1h.corr(j.pressure_2h)), 3)} | |
| # ── E: receivers under a collapsing pocket ─────────────────────────────────── | |
| def section_e(df: pd.DataFrame, games: pd.DataFrame) -> dict: | |
| att = df[(df.pass_attempt == 1) & df.receiver_player_id.notna()].copy() | |
| att = att.merge(games[["game_id", "posteam", "ib2_matchup"]], on=["game_id", "posteam"], how="left") | |
| att["depth"] = pd.cut(att.air_yards, [-99, 0, 9, 19, 99], labels=["behind LOS", "short 1-9", "intermediate 10-19", "deep 20+"]) | |
| mix = att.groupby(["ib2_matchup", "depth"], observed=True).size().unstack(fill_value=0) | |
| mix = mix.div(mix.sum(axis=1), axis=0).round(3) | |
| adot = att.groupby("ib2_matchup").air_yards.mean().round(2) | |
| # Receiver game lines vs each receiver's own pre-game average (prior games that season) | |
| rg = att.groupby(["season", "week", "game_id", "posteam", "receiver_player_id"]).agg( | |
| tgt=("pass_attempt", "sum"), rec=("complete_pass", "sum"), yds=("receiving_yards", "sum"), | |
| adot=("air_yards", "mean"), ib=("ib2_matchup", "first")).reset_index() | |
| rg["yds"] = rg.yds.fillna(0) | |
| rg = rg.sort_values(["receiver_player_id", "season", "week"]) | |
| grp = rg.groupby(["receiver_player_id", "season"]) | |
| for c in ["rec", "yds", "tgt", "adot"]: | |
| rg[f"pre_{c}"] = (grp[c].cumsum() - rg[c]) / grp.cumcount().replace(0, np.nan) | |
| rg["n_prior"] = grp.cumcount() | |
| rg = rg[rg.n_prior >= 3] | |
| primary = rg[rg.pre_yds >= 50] # WR1/TE1-type volume receivers | |
| checkdown = rg[(rg.pre_adot < 4) & (rg.pre_tgt >= 3)] # RBs/TEs living behind or near the line | |
| def by_ib(frame, stat): | |
| t = frame.groupby("ib").apply(lambda x: pd.Series({ | |
| "players": len(x), "avg_vs_own": (x[stat] - x[f"pre_{stat}"]).mean(), | |
| "under_own_rate": (x[stat] < x[f"pre_{stat}"]).mean()}), include_groups=False) | |
| return [{"ib2_matchup": int(k), **{kk: round(float(v), 3) for kk, v in r.items()}} for k, r in t.iterrows()] | |
| return {"target_depth_mix_by_ib2_matchup": {int(k): v for k, v in mix.to_dict("index").items()}, | |
| "adot_by_ib2_matchup": {int(k): float(v) for k, v in adot.items()}, | |
| "primary_receiver_yds": by_ib(primary, "yds"), | |
| "checkdown_receptions": by_ib(checkdown, "rec")} | |
| # ── F: QB pressure profile -- is the QB's reaction to pressure a stable trait? ── | |
| def section_f(df: pd.DataFrame) -> tuple[dict, pd.DataFrame]: | |
| d = df[df.passer_player_id.notna()] | |
| rows = [] | |
| for (pid, s), x in d.groupby(["passer_player_id", "season"]): | |
| if len(x) < 200: | |
| continue | |
| pr, cl = x[x.pressure == 1], x[x.pressure == 0] | |
| rows.append({"passer_player_id": pid, "season": s, "name": x.passer_player_name.mode().iat[0], | |
| "team": x.posteam.mode().iat[0], "dropbacks": len(x), | |
| "pressure_rate_faced": x.pressure.mean(), | |
| "pressure_to_sack": pr.sack.mean(), | |
| "epa_clean": cl.epa.mean(), "epa_pressured": pr.epa.mean(), | |
| "pressure_epa_drop": cl.epa.mean() - pr.epa.mean(), | |
| "int_worthy_clean": cl.int_worthy.mean(), "int_worthy_pressured": pr.int_worthy.mean(), | |
| "ttt": x.time_to_throw.mean(), "hold_3s_rate": (x.time_to_throw >= 3).mean(), | |
| "out_of_pocket_rate": x.is_qb_out_of_pocket.mean(), "int_worthy_rate": x.int_worthy.mean()}) | |
| q = pd.DataFrame(rows) | |
| nxt = q.merge(q.assign(season=q.season - 1), on=["passer_player_id", "season"], suffixes=("", "_next")) | |
| stab = {c: round(float(nxt[c].corr(nxt[f"{c}_next"])), 3) | |
| for c in ["pressure_rate_faced", "pressure_to_sack", "pressure_epa_drop", "epa_clean", "epa_pressured", | |
| "int_worthy_clean", "int_worthy_pressured", "ttt", "hold_3s_rate", "int_worthy_rate"]} | |
| return {"year_over_year_corr": stab, "qb_pairs": int(len(nxt))}, q | |
| def main() -> None: | |
| df = load_dropbacks() | |
| print(f"dropbacks {len(df):,} pressure coverage {df.was_pressure.notna().mean():.1%} " | |
| f"FTN coverage {df.is_interception_worthy.notna().mean():.1%}") | |
| a = section_a(df) | |
| b = section_b(df) | |
| c, games = section_c(df) | |
| d = section_d(df) | |
| e = section_e(df, games) | |
| f, qb = section_f(df) | |
| qb.round(4).to_csv(GOLD / "ib2_qb_pressure_profile.csv", index=False) | |
| # 2026 proxy check: can pbp-only (sack/hit/scramble) stand in for true pressure at the team level? | |
| proxy = df.assign(proxy=((df.sack == 1) | (df.qb_hit == 1) | (df.qb_scramble == 1)).astype(int)) \ | |
| .groupby(["season", "defteam"]).agg(p=("pressure", "mean"), x=("proxy", "mean")) | |
| c["proxy_team_corr"] = round(float(proxy.p.corr(proxy.x)), 3) | |
| summary = {"A_int_conditions": a, "B_rattle": b, "C_defense": c, "D_in_game": d, "E_receivers": e, "F_qb": f, | |
| "n_dropbacks": int(len(df)), "n_int": int(df.interception.sum()), | |
| "n_int_worthy": int(df.int_worthy.sum())} | |
| GOLD.mkdir(parents=True, exist_ok=True) | |
| pd.DataFrame(a["table"]).to_csv(GOLD / "ib2_int_conditions.csv", index=False) | |
| keep = ["season", "week", "game_id", "defteam", "posteam", "db", "pressure", "sack", "interception", "int_worthy", | |
| "pass_yds", "pre_off_pass_yds", "pre_pressure_rate", "pre_pressure4_rate", "pre_blitz_rate", | |
| "pre_int_worthy_rate", "pre_epa_per_db", "pre_off_pressure_allowed", "exp_pressure", | |
| "ib1", "ib2", "ib2_matchup", "ib2_raw", "ib2_matchup_raw"] | |
| games[keep].round(4).to_csv(GOLD / "ib2_defense_game_backtest.csv", index=False) | |
| (REPO / "lake" / "gold" / "nfl" / "ib2_deep_dive_summary.json").write_text(json.dumps(summary, indent=1, default=str)) | |
| print(json.dumps({k: v for k, v in summary.items() if k != "A_int_conditions"}, indent=1, default=str)[:12000]) | |
| print(json.dumps({k: v for k, v in a.items() if k != "table"}, indent=1)) | |
| if __name__ == "__main__": | |
| main() | |