#!/usr/bin/env python3 """ SportsGameOdds API - Historical Data Backfill Script Backfills historical games, odds, scores, and player stats from SportsGameOdds API. Supports date ranges for 2024, 2025, and 2026 seasons. Usage: python scripts/backfill_sportsGameOdds.py [league] [--start YYYY-MM-DD] [--end YYYY-MM-DD] python scripts/backfill_sportsGameOdds.py --all-leagues --year 2025 python scripts/backfill_sportsGameOdds.py nba --season 2024-25 Examples: python scripts/backfill_sportsGameOdds.py nba --start 2025-01-01 --end 2025-12-31 python scripts/backfill_sportsGameOdds.py --all-leagues --year 2025 python scripts/backfill_sportsGameOdds.py nfl --season 2025-26 """ import os import sys import json import time import hashlib import argparse import psycopg2 from datetime import datetime, timezone, timedelta from pathlib import Path from typing import Dict, List, Any, Optional, Tuple import requests sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from team_names import normalize_to_full # Load environment ENV_FILE = Path(__file__).parent.parent / ".env" if ENV_FILE.exists(): with open(ENV_FILE) as f: for line in f: if line.strip() and not line.startswith("#") and "=" in line: key, val = line.strip().split("=", 1) os.environ.setdefault(key, val.strip('"').strip("'")) # Configuration SGO_BASE_URL = os.environ.get("SPORTSGAMEODDS_BASE_URL", "https://api.sportsgameodds.com/v2") SGO_API_KEY = os.environ.get("SPORTSGAMEODDS_API_KEY") SPORTS_DB_URL = os.environ.get("SPORTS_DATABASE_URL", "") if not SGO_API_KEY: print("ERROR: SPORTSGAMEODDS_API_KEY not set in environment") sys.exit(1) # League mappings with season info LEAGUES = { # US Major Sports "nba": {"id": "NBA", "season_start": 10, "season_end": 6}, # Oct-Jun "nfl": {"id": "NFL", "season_start": 9, "season_end": 2}, # Sep-Feb "mlb": {"id": "MLB", "season_start": 3, "season_end": 10}, # Mar-Oct "nhl": {"id": "NHL", "season_start": 10, "season_end": 6}, # Oct-Jun "ncaab": {"id": "NCAAB", "season_start": 11, "season_end": 4}, # Nov-Apr "ncaaf": {"id": "NCAAF", "season_start": 8, "season_end": 1}, # Aug-Jan "wnba": {"id": "WNBA", "season_start": 5, "season_end": 10}, # May-Oct # MMA "mma": {"id": "UFC", "season_start": 1, "season_end": 12}, # Year-round "ufc": {"id": "UFC", "season_start": 1, "season_end": 12}, # European Soccer "epl": {"id": "EPL", "season_start": 8, "season_end": 5}, # Aug-May "laliga": {"id": "LA_LIGA", "season_start": 8, "season_end": 5}, "seriea": {"id": "IT_SERIE_A", "season_start": 8, "season_end": 5}, "bundesliga": {"id": "BUNDESLIGA", "season_start": 8, "season_end": 5}, "ligue1": {"id": "FR_LIGUE_1", "season_start": 8, "season_end": 5}, "ucl": {"id": "UEFA_CHAMPIONS_LEAGUE", "season_start": 9, "season_end": 6}, "mls": {"id": "MLS", "season_start": 2, "season_end": 12}, } # All leagues for --all-leagues flag ALL_LEAGUES = ["nba", "nfl", "mlb", "nhl", "ncaab", "ncaaf", "mma", "epl", "laliga", "seriea", "bundesliga", "ligue1"] # Reverse mapping: SGO API league ID → canonical lowercase DB name SGO_TO_DB_LEAGUE = {v["id"]: k for k, v in LEAGUES.items()} SGO_TO_DB_LEAGUE["UFC"] = "mma" # Fix duplicate class SportsGameOddsBackfiller: """Backfill historical data from SportsGameOdds API""" def __init__(self, api_key: str, base_url: str = SGO_BASE_URL): self.api_key = api_key self.base_url = base_url self.session = requests.Session() self.session.headers.update({ "x-api-key": api_key, "Content-Type": "application/json", }) self.requests_made = 0 self.events_fetched = 0 self.players_fetched = 0 def _request(self, endpoint: str, params: Dict = None) -> Dict: """Make API request with retry logic""" url = f"{self.base_url}/{endpoint}" max_retries = 3 for attempt in range(max_retries): try: resp = self.session.get(url, params=params, timeout=60) self.requests_made += 1 if resp.status_code == 429: wait = min(60, (attempt + 1) * 10) print(f" Rate limited, waiting {wait}s...") time.sleep(wait) continue if resp.status_code == 401: return {"success": False, "error": "Invalid API key"} resp.raise_for_status() return resp.json() except requests.exceptions.RequestException as e: if attempt < max_retries - 1: time.sleep(5) continue return {"success": False, "error": str(e)} return {"success": False, "error": "Max retries exceeded"} def fetch_events_in_range(self, league_id: str, start_date: str, end_date: str, include_completed: bool = True) -> List[Dict]: """Fetch all events in a date range with pagination""" all_events = [] cursor = None batch = 0 params = { "leagueID": league_id, "startsAfter": f"{start_date}T00:00:00Z", "startsBefore": f"{end_date}T23:59:59Z", "limit": 100, } # Don't filter by oddsAvailable for backfill - we want all games if not include_completed: params["oddsAvailable"] = "true" while True: batch += 1 if cursor: params["cursor"] = cursor result = self._request("events", params) if not result.get("success", True) or "error" in result: print(f" Error: {result.get('error', 'Unknown error')}") break events = result.get("data", []) if not events: break all_events.extend(events) self.events_fetched += len(events) print(f" Batch {batch}: {len(events)} events (total: {len(all_events)})") cursor = result.get("nextCursor") if not cursor: break # Rate limiting time.sleep(0.3) return all_events def fetch_players(self, league_id: str) -> List[Dict]: """Fetch all players for a league""" all_players = [] cursor = None while True: params = {"leagueID": league_id, "limit": 500} if cursor: params["cursor"] = cursor result = self._request("players", params) if not result.get("success", True): break players = result.get("data", []) if not players: break all_players.extend(players) self.players_fetched += len(players) cursor = result.get("nextCursor") if not cursor: break time.sleep(0.2) return all_players def parse_odds_value(value) -> Optional[int]: """Parse odds string to integer""" if value is None: return None if isinstance(value, (int, float)): return int(value) if isinstance(value, str): try: return int(value.replace("+", "")) except ValueError: return None return None def parse_line_value(value) -> Optional[float]: """Parse line value to float""" if value is None: return None if isinstance(value, (int, float)): return float(value) if isinstance(value, str): try: return float(value) except ValueError: return None return None def extract_odds(event: Dict) -> Dict: """Extract odds data from event""" odds_data = event.get("odds", {}) # Moneylines ml_home = odds_data.get("points-home-game-ml-home", {}) ml_away = odds_data.get("points-away-game-ml-away", {}) home_ml = parse_odds_value(ml_home.get("bookOdds") or ml_home.get("fairOdds")) away_ml = parse_odds_value(ml_away.get("bookOdds") or ml_away.get("fairOdds")) # Spreads sp_home = odds_data.get("points-home-game-sp-home", {}) spread_home = parse_line_value(sp_home.get("bookSpread") or sp_home.get("fairSpread")) # Totals ou_over = odds_data.get("points-all-game-ou-over", {}) total = parse_line_value(ou_over.get("bookOverUnder") or ou_over.get("fairOverUnder")) # Opening lines if available open_home_ml = parse_odds_value(ml_home.get("openBookOdds") or ml_home.get("openFairOdds")) open_spread = parse_line_value(sp_home.get("openBookSpread") or sp_home.get("openFairSpread")) open_total = parse_line_value(ou_over.get("openBookOverUnder") or ou_over.get("openFairOverUnder")) return { "moneylineHome": home_ml, "moneylineAway": away_ml, "spreadHome": spread_home, "spreadAway": -spread_home if spread_home else None, "total": total, "openMoneylineHome": open_home_ml, "openSpreadHome": open_spread, "openTotal": open_total, } def extract_scores(event: Dict) -> Tuple[Optional[int], Optional[int]]: """Extract final scores from event""" results = event.get("results", {}) game_results = results.get("game", {}) home_score = game_results.get("home", {}).get("points") away_score = game_results.get("away", {}).get("points") # Also check teams object if home_score is None: teams = event.get("teams", {}) home_score = teams.get("home", {}).get("score") away_score = teams.get("away", {}).get("score") return home_score, away_score def extract_player_stats(event: Dict) -> List[Dict]: """Extract player game stats from event results""" stats = [] event_id = event.get("eventID") sgo_league = event.get("leagueID") league = SGO_TO_DB_LEAGUE.get(sgo_league, sgo_league.lower() if sgo_league else "unknown") status = event.get("status", {}) start_time = status.get("startsAt") if start_time: game_date = datetime.fromisoformat(start_time.replace("Z", "+00:00")) season = game_date.year if game_date.month >= 7 else game_date.year - 1 else: return stats results = event.get("results", {}) players_data = event.get("players", {}) # Get team mappings teams = event.get("teams", {}) home_team = teams.get("home", {}).get("names", {}).get("short", "UNK") away_team = teams.get("away", {}).get("names", {}).get("short", "UNK") # Check for player stats in results game_results = results.get("game", {}) for team_key in ["home", "away"]: team_results = game_results.get(team_key, {}) team_name = home_team if team_key == "home" else away_team opponent = away_team if team_key == "home" else home_team # Look for player-level data in results players_in_team = team_results.get("players", {}) for player_id, player_stats in players_in_team.items(): if not isinstance(player_stats, dict): continue player_info = players_data.get(player_id, {}) player_name = player_info.get("name", player_id) position = player_info.get("position") # Extract relevant stats stat_mappings = { "points": player_stats.get("points"), "rebounds": player_stats.get("rebounds"), "assists": player_stats.get("assists"), "steals": player_stats.get("steals"), "blocks": player_stats.get("blocks"), "turnovers": player_stats.get("turnovers"), "minutes": player_stats.get("minutes"), "fieldGoalsMade": player_stats.get("fieldGoalsMade"), "fieldGoalsAttempted": player_stats.get("fieldGoalsAttempted"), "threePointersMade": player_stats.get("threePointersMade"), "threePointersAttempted": player_stats.get("threePointersAttempted"), "freeThrowsMade": player_stats.get("freeThrowsMade"), "freeThrowsAttempted": player_stats.get("freeThrowsAttempted"), } for stat_key, value in stat_mappings.items(): if value is not None: stats.append({ "league": league, "season": season, "gameKey": f"g:{event_id}", "gameDate": game_date, "playerExternalId": player_id, "playerName": player_name, "position": position, "team": team_name, "opponent": opponent, "statKey": stat_key, "value": float(value), }) return stats def get_db_connection(): """Get PostgreSQL connection""" if not SPORTS_DB_URL: return None try: url = SPORTS_DB_URL.replace("postgresql://", "") if "?" in url: url = url.split("?")[0] user_pass, host_db = url.split("@") user, password = user_pass.split(":") host_port, database = host_db.split("/") if ":" in host_port: host, port = host_port.split(":") else: host, port = host_port, "5432" conn = psycopg2.connect( host=host, port=int(port), database=database, user=user, password=password, ) return conn except Exception as e: print(f"Database connection error: {e}") return None def upsert_games(conn, events: List[Dict], league: str) -> Tuple[int, int]: """Upsert games to SportsGame table""" if not conn or not events: return 0, 0 cursor = conn.cursor() inserted = 0 updated = 0 for event in events: try: event_id = event.get("eventID") teams = event.get("teams", {}) home_team = normalize_to_full(teams.get("home", {}).get("names", {}).get("long", "") or teams.get("home", {}).get("names", {}).get("short", "UNK"), league) away_team = normalize_to_full(teams.get("away", {}).get("names", {}).get("long", "") or teams.get("away", {}).get("names", {}).get("short", "UNK"), league) status_data = event.get("status", {}) start_time = status_data.get("startsAt") if status_data.get("completed"): status = "final" elif status_data.get("live"): status = "live" elif status_data.get("cancelled"): status = "cancelled" else: status = "scheduled" home_score, away_score = extract_scores(event) odds = extract_odds(event) if start_time: game_date = datetime.fromisoformat(start_time.replace("Z", "+00:00")) season = game_date.year if game_date.month >= 7 else game_date.year - 1 else: continue # Check if exists cursor.execute(''' SELECT id FROM "SportsGame" WHERE "externalGameId" = %s AND league = %s ''', (event_id, league.lower())) existing = cursor.fetchone() if existing: cursor.execute(''' UPDATE "SportsGame" SET "homeScore" = COALESCE(%s, "homeScore"), "awayScore" = COALESCE(%s, "awayScore"), status = %s, "moneylineHome" = COALESCE(%s, "moneylineHome"), "moneylineAway" = COALESCE(%s, "moneylineAway"), "spreadHome" = COALESCE(%s, "spreadHome"), "spreadAway" = COALESCE(%s, "spreadAway"), total = COALESCE(%s, total), "oddsSource" = %s, "oddsUpdatedAt" = NOW(), "updatedAt" = NOW() WHERE "externalGameId" = %s AND league = %s ''', ( home_score, away_score, status, odds["moneylineHome"], odds["moneylineAway"], odds["spreadHome"], odds["spreadAway"], odds["total"], "sportsgameodds-backfill", event_id, league.lower() )) updated += 1 else: cursor.execute(''' INSERT INTO "SportsGame" ( league, season, "gameDate", "homeTeam", "awayTeam", "externalGameId", "homeScore", "awayScore", status, "moneylineHome", "moneylineAway", "spreadHome", "spreadAway", total, "oddsSource", "oddsUpdatedAt", "createdAt", "updatedAt" ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW(), NOW() ) ON CONFLICT (league, season, "gameDate", "homeTeam", "awayTeam") DO UPDATE SET "homeScore" = COALESCE(EXCLUDED."homeScore", "SportsGame"."homeScore"), "awayScore" = COALESCE(EXCLUDED."awayScore", "SportsGame"."awayScore"), status = EXCLUDED.status, "moneylineHome" = COALESCE(EXCLUDED."moneylineHome", "SportsGame"."moneylineHome"), "moneylineAway" = COALESCE(EXCLUDED."moneylineAway", "SportsGame"."moneylineAway"), "spreadHome" = COALESCE(EXCLUDED."spreadHome", "SportsGame"."spreadHome"), "spreadAway" = COALESCE(EXCLUDED."spreadAway", "SportsGame"."spreadAway"), total = COALESCE(EXCLUDED.total, "SportsGame".total), "oddsSource" = EXCLUDED."oddsSource", "oddsUpdatedAt" = NOW(), "updatedAt" = NOW() ''', ( league.lower(), season, game_date, home_team, away_team, event_id, home_score, away_score, status, odds["moneylineHome"], odds["moneylineAway"], odds["spreadHome"], odds["spreadAway"], odds["total"], "sportsgameodds-backfill" )) inserted += 1 except Exception as e: continue conn.commit() cursor.close() return inserted, updated def upsert_player_stats(conn, stats: List[Dict]) -> int: """Upsert player game metrics""" if not conn or not stats: return 0 cursor = conn.cursor() inserted = 0 for stat in stats: try: cursor.execute(''' INSERT INTO "PlayerGameMetric" ( league, season, "gameKey", "gameDate", "playerExternalId", "playerName", position, team, opponent, "statKey", value, "createdAt" ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) ON CONFLICT (league, season, "gameKey", "playerExternalId", "statKey") DO UPDATE SET value = EXCLUDED.value ''', ( stat["league"], stat["season"], stat["gameKey"], stat["gameDate"], stat["playerExternalId"], stat.get("playerName"), stat.get("position"), stat.get("team"), stat.get("opponent"), stat["statKey"], stat["value"] )) if cursor.rowcount > 0: inserted += 1 except Exception as e: continue conn.commit() cursor.close() return inserted def upsert_players(conn, players: List[Dict], league: str) -> int: """Upsert player records""" if not conn or not players: return 0 cursor = conn.cursor() inserted = 0 for player in players: try: cursor.execute(''' INSERT INTO "Player" ( league, "externalPlayerId", name, position, team, raw, "createdAt", "updatedAt" ) VALUES (%s, %s, %s, %s, %s, %s, NOW(), NOW()) ON CONFLICT (league, "externalPlayerId") DO UPDATE SET name = EXCLUDED.name, position = COALESCE(EXCLUDED.position, "Player".position), team = COALESCE(EXCLUDED.team, "Player".team), "updatedAt" = NOW() ''', ( league.lower(), player.get("playerID"), player.get("name"), player.get("position"), player.get("teamID"), json.dumps(player) )) if cursor.rowcount > 0: inserted += 1 except Exception as e: continue conn.commit() cursor.close() return inserted def save_external_feed(conn, events: List[Dict], league: str) -> int: """Save raw events to ExternalFeedRecord""" if not conn or not events: return 0 cursor = conn.cursor() inserted = 0 for event in events: try: raw_json = json.dumps(event, default=str) content_hash = hashlib.md5(raw_json.encode()).hexdigest() cursor.execute(''' INSERT INTO "ExternalFeedRecord" ( source, kind, league, "contentHash", raw, "fetchedAt", "createdAt" ) VALUES (%s, %s, %s, %s, %s, NOW(), NOW()) ON CONFLICT (source, "contentHash") DO NOTHING ''', ("sportsgameodds-backfill", "historical", league.lower(), content_hash, raw_json)) if cursor.rowcount > 0: inserted += 1 except Exception: continue conn.commit() cursor.close() return inserted def get_season_dates(league: str, season: str) -> Tuple[str, str]: """Get start and end dates for a season like '2024-25' or '2025'""" league_info = LEAGUES.get(league, {}) season_start_month = league_info.get("season_start", 1) season_end_month = league_info.get("season_end", 12) if "-" in season: # Format: 2024-25 start_year = int(season.split("-")[0]) end_year = start_year + 1 else: # Format: 2025 year = int(season) if season_start_month > season_end_month: # Cross-year season (e.g., NBA Oct-Jun) start_year = year - 1 end_year = year else: start_year = year end_year = year start_date = f"{start_year}-{season_start_month:02d}-01" # Calculate end date if season_end_month == 12: end_date = f"{end_year}-12-31" else: # Get last day of end month if season_end_month in [1, 3, 5, 7, 8, 10]: last_day = 31 elif season_end_month == 2: last_day = 28 else: last_day = 30 end_date = f"{end_year}-{season_end_month:02d}-{last_day}" return start_date, end_date def main(): parser = argparse.ArgumentParser(description="Backfill historical data from SportsGameOdds API") parser.add_argument("league", nargs="?", default="nba", help="League to backfill") parser.add_argument("--start", type=str, help="Start date (YYYY-MM-DD)") parser.add_argument("--end", type=str, help="End date (YYYY-MM-DD)") parser.add_argument("--year", type=int, help="Year to backfill (e.g., 2025)") parser.add_argument("--season", type=str, help="Season to backfill (e.g., 2024-25)") parser.add_argument("--all-leagues", action="store_true", help="Backfill all major leagues") parser.add_argument("--include-players", action="store_true", help="Also fetch and store player data") parser.add_argument("--dry-run", action="store_true", help="Don't save to database") args = parser.parse_args() print("=" * 70) print("SPORTSGAMEODDS HISTORICAL BACKFILL") print("=" * 70) backfiller = SportsGameOddsBackfiller(SGO_API_KEY) # Check API usage usage = backfiller._request("account/usage") if usage.get("data"): rate_limits = usage["data"].get("rateLimits", {}) print(f"API Tier: {usage['data'].get('tier', 'unknown')}") print(f"Daily requests: {rate_limits.get('per-day', {}).get('current-requests', 0)} / {rate_limits.get('per-day', {}).get('max-requests', 'unlimited')}") print(f"Daily entities: {rate_limits.get('per-day', {}).get('current-entities', 0)} / {rate_limits.get('per-day', {}).get('max-entities', 'unlimited')}") # Determine leagues to backfill if args.all_leagues: leagues = ALL_LEAGUES else: leagues = [args.league.lower()] # Determine date range if args.start and args.end: start_date = args.start end_date = args.end elif args.year: start_date = f"{args.year}-01-01" end_date = f"{args.year}-12-31" elif args.season: # Will be calculated per-league start_date = None end_date = None else: # Default: last 90 days end_date = datetime.now().strftime("%Y-%m-%d") start_date = (datetime.now() - timedelta(days=90)).strftime("%Y-%m-%d") # Connect to database conn = None if args.dry_run else get_db_connection() if conn: print("Connected to PostgreSQL database") elif not args.dry_run: print("WARNING: Could not connect to database") total_games = 0 total_inserted = 0 total_updated = 0 total_player_stats = 0 total_players = 0 for league in leagues: league_info = LEAGUES.get(league) if not league_info: print(f"\nUnknown league: {league}") continue league_id = league_info["id"] # Calculate dates for season if specified if args.season: start_date, end_date = get_season_dates(league, args.season) print(f"\n{'='*60}") print(f"BACKFILLING: {league_id}") print(f"Date range: {start_date} to {end_date}") print(f"{'='*60}") # Fetch events events = backfiller.fetch_events_in_range(league_id, start_date, end_date) if not events: print(f" No events found for {league_id}") continue # Count completed games completed = sum(1 for e in events if e.get("status", {}).get("completed")) with_odds = sum(1 for e in events if e.get("odds")) with_scores = sum(1 for e in events if extract_scores(e)[0] is not None) print(f" Total events: {len(events)}") print(f" Completed games: {completed}") print(f" With odds: {with_odds}") print(f" With scores: {with_scores}") total_games += len(events) # Save to database if conn: inserted, updated = upsert_games(conn, events, league) print(f" Database: {inserted} inserted, {updated} updated") total_inserted += inserted total_updated += updated # Save external feed records feed_count = save_external_feed(conn, events, league) print(f" External feed records: {feed_count}") # Extract and save player stats from completed games all_stats = [] for event in events: if event.get("status", {}).get("completed"): stats = extract_player_stats(event) all_stats.extend(stats) if all_stats: stats_inserted = upsert_player_stats(conn, all_stats) print(f" Player stats: {stats_inserted}") total_player_stats += stats_inserted # Fetch players if requested if args.include_players: print(f" Fetching players...") players = backfiller.fetch_players(league_id) if players and conn: players_inserted = upsert_players(conn, players, league) print(f" Players: {players_inserted}") total_players += players_inserted # Sample output if events: print("\n Sample events:") for event in events[:3]: teams = event.get("teams", {}) home = teams.get("home", {}).get("names", {}).get("medium", "?") away = teams.get("away", {}).get("names", {}).get("medium", "?") start = event.get("status", {}).get("startsAt", "?")[:10] home_score, away_score = extract_scores(event) status = "Final" if event.get("status", {}).get("completed") else "Scheduled" score_str = f"{away_score}-{home_score}" if home_score else "TBD" print(f" {start}: {away} @ {home} ({score_str}) [{status}]") # Summary print("\n" + "=" * 70) print("BACKFILL SUMMARY") print("=" * 70) print(f"Total events fetched: {total_games}") print(f"Database inserts: {total_inserted}") print(f"Database updates: {total_updated}") print(f"Player stats saved: {total_player_stats}") print(f"Players saved: {total_players}") print(f"API requests made: {backfiller.requests_made}") # Final usage check final_usage = backfiller._request("account/usage") if final_usage.get("data"): rate_limits = final_usage["data"].get("rateLimits", {}) print(f"\nRemaining daily requests: {rate_limits.get('per-day', {}).get('max-requests', 'unlimited')} - {rate_limits.get('per-day', {}).get('current-requests', 0)}") if conn: conn.close() if __name__ == "__main__": main()