#!/usr/bin/env python3 """ NFL Win Probability Metrics (nfl_data_py) Computes win probability metrics and live line edge indicators. """ import nfl_data_py as nfl import psycopg2 import numpy as np from datetime import datetime, timezone import sys def load_db_url(): with open('/var/www/html/eventheodds/.env', 'r') as f: for line in f: if line.startswith('SPORTS_DATABASE_URL='): return line.split('=', 1)[1].strip().split('?')[0] return '' def ingest_win_prob_metrics(season=2025): db_url = load_db_url() if not db_url: raise RuntimeError('SPORTS_DATABASE_URL not set') conn = psycopg2.connect(db_url) cur = conn.cursor() stats_added = 0 game_key = f'{season}_NFL_WIN_PROB' game_date = datetime(season, 9, 1, tzinfo=timezone.utc) sql_team = ''' INSERT INTO "TeamGameMetric" (league, season, "gameKey", "gameDate", team, "statKey", value) VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (league, season, "gameKey", team, "statKey") DO UPDATE SET value = EXCLUDED.value ''' print(f'Fetching NFL PBP for {season}...') pbp = nfl.import_pbp_data([season]) print(f' Plays: {len(pbp)}') # Get unique games with outcomes games = pbp.groupby('game_id').agg({ 'home_team': 'first', 'away_team': 'first', 'total_home_score': 'max', 'total_away_score': 'max', 'spread_line': 'first' }).reset_index() games['home_won'] = games['total_home_score'] > games['total_away_score'] games['away_won'] = games['total_away_score'] > games['total_home_score'] # ========== Halftime Win Probability Analysis ========== # Get WP at halftime (end of Q2) halftime = pbp[(pbp['qtr'] == 2)].copy() if len(halftime) > 0: halftime_last = halftime.groupby('game_id').agg({ 'home_wp': 'last', 'away_wp': 'last', 'home_team': 'first', 'away_team': 'first' }).reset_index() halftime_last = halftime_last.merge(games[['game_id', 'home_won', 'away_won']], on='game_id') # Close games at halftime (WP between 40-60%) close_games_home = halftime_last[(halftime_last['home_wp'] >= 0.4) & (halftime_last['home_wp'] <= 0.6)] # Comeback analysis: trailing at halftime but won home_trailing = halftime_last[halftime_last['home_wp'] < 0.5] away_trailing = halftime_last[halftime_last['away_wp'] < 0.5] # Team-level aggregation for team in games['home_team'].unique(): # Home games home_games = halftime_last[halftime_last['home_team'] == team] home_close = close_games_home[close_games_home['home_team'] == team] home_trailing_games = home_trailing[home_trailing['home_team'] == team] # Away games away_games = halftime_last[halftime_last['away_team'] == team] away_close = close_games_home[close_games_home['away_team'] == team] away_trailing_games = away_trailing[away_trailing['away_team'] == team] # Close game win rate close_wins = len(home_close[home_close['home_won']]) + len(away_close[away_close['away_won']]) close_total = len(home_close) + len(away_close) if close_total >= 3: close_win_rate = close_wins / close_total cur.execute(sql_team, ('nfl', season, game_key, game_date, team, 'nfl_close_game_win_rate', float(close_win_rate))) stats_added += 1 # Comeback rate (win when trailing at half) comeback_wins = len(home_trailing_games[home_trailing_games['home_won']]) + len(away_trailing_games[away_trailing_games['away_won']]) trailing_total = len(home_trailing_games) + len(away_trailing_games) if trailing_total >= 2: comeback_rate = comeback_wins / trailing_total cur.execute(sql_team, ('nfl', season, game_key, game_date, team, 'nfl_comeback_rate', float(comeback_rate))) stats_added += 1 # ========== WP vs Vegas WP Differential ========== if 'vegas_wp' in pbp.columns: # Average differential between model WP and Vegas WP valid_wp = pbp[(pbp['wp'].notna()) & (pbp['vegas_wp'].notna())].copy() valid_wp['wp_diff'] = valid_wp['wp'] - valid_wp['vegas_wp'] # When team is on offense (posteam), measure their WP vs Vegas wp_by_team = valid_wp.groupby('posteam')['wp_diff'].mean() for team, diff in wp_by_team.items(): cur.execute(sql_team, ('nfl', season, game_key, game_date, team, 'nfl_wp_vs_vegas_edge', float(diff))) stats_added += 1 # ========== WPA Volatility (swinginess) ========== if 'wpa' in pbp.columns: wpa_std = pbp[pbp['wpa'].notna()].groupby('posteam')['wpa'].std() for team, std in wpa_std.items(): cur.execute(sql_team, ('nfl', season, game_key, game_date, team, 'nfl_wpa_volatility', float(std))) stats_added += 1 # ========== ATS Performance ========== if 'spread_line' in games.columns: games['margin'] = games['total_home_score'] - games['total_away_score'] games['home_covered'] = games['margin'] > -games['spread_line'] games['away_covered'] = games['margin'] < -games['spread_line'] for team in games['home_team'].unique(): home_covers = games[(games['home_team'] == team) & games['home_covered']] away_covers = games[(games['away_team'] == team) & games['away_covered']] home_total = len(games[games['home_team'] == team]) away_total = len(games[games['away_team'] == team]) total_covers = len(home_covers) + len(away_covers) total_games = home_total + away_total if total_games >= 3: ats_rate = total_covers / total_games cur.execute(sql_team, ('nfl', season, game_key, game_date, team, 'nfl_ats_cover_rate', float(ats_rate))) stats_added += 1 conn.commit() cur.close() conn.close() print(f'✅ NFL win probability metrics ingested: {stats_added}') return {'success': True, 'stats_added': stats_added} if __name__ == '__main__': season = int(sys.argv[1]) if len(sys.argv) > 1 else 2025 ingest_win_prob_metrics(season)