#!/usr/bin/env python3 """ NBA Team Stats Ingestion using nba_api Fetches team standings, advanced stats, and ratings from NBA.com Ingests into SportsDB TeamGameMetric table """ import json import os import subprocess from datetime import datetime # nba_api imports from nba_api.stats.endpoints import leaguestandings, teamestimatedmetrics, leaguedashteamstats from nba_api.stats.static import teams def sql_value(val): if val is None or val == '': return 'NULL' if isinstance(val, (int, float)): return str(val) s = str(val).replace("'", "''") return f"'{s}'" def load_env_db_url(): db_url = os.environ.get('SPORTS_DATABASE_URL', '') if db_url: return db_url env_path = '/var/www/html/eventheodds/.env' if os.path.exists(env_path): with open(env_path, 'r', encoding='utf-8') as f: for line in f: if line.startswith('SPORTS_DATABASE_URL='): return line.split('=', 1)[1].strip() return '' def run_sql(sql): db_url = load_env_db_url() db_url = db_url.split('?schema=')[0] if '?schema=' in db_url else db_url if not db_url: raise RuntimeError('SPORTS_DATABASE_URL not set') tmp_path = '/tmp/nba_team_stats.sql' with open(tmp_path, 'w', encoding='utf-8') as f: f.write(sql) result = subprocess.run( ['psql', db_url, '-f', tmp_path], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, timeout=300, ) if result.returncode != 0: err = (result.stderr or '').strip() raise RuntimeError(f'psql failed: {err[:500]}') def batch_insert(rows): if not rows: return 0 columns = ['league', 'season', 'gameKey', 'gameDate', 'team', 'opponent', 'statKey', 'value'] values_sql = ',\n'.join( '(' + ', '.join(sql_value(row.get(col)) for col in columns) + ')' for row in rows ) cols_sql = ', '.join(f'"{c}"' for c in columns) sql = f'''INSERT INTO "TeamGameMetric" ({cols_sql}) VALUES {values_sql} ON CONFLICT (league, season, "gameKey", team, "statKey") DO UPDATE SET value=EXCLUDED.value;''' run_sql(sql) return len(rows) def fetch_and_ingest_standings(season='2025-26'): """Fetch NBA standings and team stats""" print(f'[nba_api] Fetching standings for {season}...') try: # Get league standings standings = leaguestandings.LeagueStandings(season=season) standings_df = standings.get_data_frames()[0] rows = [] today = datetime.now().strftime('%Y-%m-%d') season_year = int(season.split('-')[0]) for _, team in standings_df.iterrows(): team_abbr = team.get('TeamSlug', '').upper() or team.get('TeamCity', '')[:3].upper() game_key = f'standings:{season}:{team_abbr}' # Extract all available stats stats_to_save = { 'wins': team.get('WINS'), 'losses': team.get('LOSSES'), 'win_pct': team.get('WinPCT'), 'home_record_wins': team.get('HOME').split('-')[0] if team.get('HOME') else None, 'road_record_wins': team.get('ROAD').split('-')[0] if team.get('ROAD') else None, 'l10_wins': team.get('L10').split('-')[0] if team.get('L10') else None, 'streak': team.get('CurrentStreak'), 'conf_rank': team.get('ConferenceRank'), 'div_rank': team.get('DivisionRank'), 'ppg': team.get('PointsPG'), 'opp_ppg': team.get('OppPointsPG'), 'diff_ppg': team.get('DiffPointsPG'), } for stat_key, value in stats_to_save.items(): if value is not None: try: val = float(value) if value else None if val is not None: rows.append({ 'league': 'nba', 'season': season_year, 'gameKey': game_key, 'gameDate': today, 'team': team_abbr, 'opponent': None, 'statKey': f'nba_{stat_key}', 'value': val, }) except (ValueError, TypeError): pass inserted = batch_insert(rows) print(f'[nba_api] Inserted {inserted} standing stats') return inserted except Exception as e: print(f'[nba_api] Error fetching standings: {e}') return 0 def fetch_and_ingest_advanced_stats(season='2025-26'): """Fetch advanced team stats (ratings, pace, etc.)""" print(f'[nba_api] Fetching advanced stats for {season}...') try: # Get team estimated metrics (net rating, etc.) metrics = teamestimatedmetrics.TeamEstimatedMetrics(season=season) metrics_df = metrics.get_data_frames()[0] rows = [] today = datetime.now().strftime('%Y-%m-%d') season_year = int(season.split('-')[0]) for _, team in metrics_df.iterrows(): team_id = team.get('TEAM_ID') team_name = team.get('TEAM_NAME', '') # Get team abbreviation from static data nba_teams = teams.get_teams() team_info = next((t for t in nba_teams if t['id'] == team_id), None) team_abbr = team_info['abbreviation'] if team_info else team_name[:3].upper() game_key = f'advanced:{season}:{team_abbr}' # Extract advanced stats stats_to_save = { 'gp': team.get('GP'), 'w': team.get('W'), 'l': team.get('L'), 'w_pct': team.get('W_PCT'), 'min': team.get('MIN'), 'e_off_rating': team.get('E_OFF_RATING'), # Estimated Offensive Rating 'e_def_rating': team.get('E_DEF_RATING'), # Estimated Defensive Rating 'e_net_rating': team.get('E_NET_RATING'), # Estimated Net Rating 'e_pace': team.get('E_PACE'), # Estimated Pace 'e_ast_ratio': team.get('E_AST_RATIO'), 'e_oreb_pct': team.get('E_OREB_PCT'), 'e_dreb_pct': team.get('E_DREB_PCT'), 'e_reb_pct': team.get('E_REB_PCT'), 'e_tm_tov_pct': team.get('E_TM_TOV_PCT'), } for stat_key, value in stats_to_save.items(): if value is not None: try: val = float(value) rows.append({ 'league': 'nba', 'season': season_year, 'gameKey': game_key, 'gameDate': today, 'team': team_abbr, 'opponent': None, 'statKey': f'nba_{stat_key}', 'value': val, }) except (ValueError, TypeError): pass inserted = batch_insert(rows) print(f'[nba_api] Inserted {inserted} advanced stats') return inserted except Exception as e: print(f'[nba_api] Error fetching advanced stats: {e}') return 0 def fetch_and_ingest_team_stats(season='2025-26'): """Fetch detailed team stats""" print(f'[nba_api] Fetching team stats for {season}...') try: # Get league dash team stats team_stats = leaguedashteamstats.LeagueDashTeamStats(season=season) stats_df = team_stats.get_data_frames()[0] rows = [] today = datetime.now().strftime('%Y-%m-%d') season_year = int(season.split('-')[0]) for _, team in stats_df.iterrows(): team_abbr = team.get('TEAM_ABBREVIATION', '') game_key = f'teamstats:{season}:{team_abbr}' # Extract team stats stats_to_save = { 'gp': team.get('GP'), 'w': team.get('W'), 'l': team.get('L'), 'w_pct': team.get('W_PCT'), 'min': team.get('MIN'), 'fgm': team.get('FGM'), 'fga': team.get('FGA'), 'fg_pct': team.get('FG_PCT'), 'fg3m': team.get('FG3M'), 'fg3a': team.get('FG3A'), 'fg3_pct': team.get('FG3_PCT'), 'ftm': team.get('FTM'), 'fta': team.get('FTA'), 'ft_pct': team.get('FT_PCT'), 'oreb': team.get('OREB'), 'dreb': team.get('DREB'), 'reb': team.get('REB'), 'ast': team.get('AST'), 'tov': team.get('TOV'), 'stl': team.get('STL'), 'blk': team.get('BLK'), 'blka': team.get('BLKA'), 'pf': team.get('PF'), 'pfd': team.get('PFD'), 'pts': team.get('PTS'), 'plus_minus': team.get('PLUS_MINUS'), } for stat_key, value in stats_to_save.items(): if value is not None: try: val = float(value) rows.append({ 'league': 'nba', 'season': season_year, 'gameKey': game_key, 'gameDate': today, 'team': team_abbr, 'opponent': None, 'statKey': f'nba_team_{stat_key}', 'value': val, }) except (ValueError, TypeError): pass inserted = batch_insert(rows) print(f'[nba_api] Inserted {inserted} team stats') return inserted except Exception as e: print(f'[nba_api] Error fetching team stats: {e}') return 0 def main(): print('[nba_api] Starting NBA team stats ingestion...') total = 0 total += fetch_and_ingest_standings() total += fetch_and_ingest_advanced_stats() total += fetch_and_ingest_team_stats() print(f'[nba_api] Complete! Total stats inserted: {total}') return total if __name__ == '__main__': main()