#!/usr/bin/env python3 """ Multi-Sport Data Fetcher Fetches real game data from SportsData.io API for all supported sports Supported Sports: - NBA, NFL, MLB, NHL (Pro leagues) - CFB, CBB (College) - Soccer, MMA/UFC, Golf, NASCAR, Tennis """ import os import json import requests from datetime import datetime, timedelta from typing import Dict, List, Any, Optional import hashlib import time class SportsDataFetcher: """ Fetches real sports data from SportsData.io API. Includes caching to avoid redundant API calls. """ # API base URLs for different sports API_BASES = { 'nfl': 'https://api.sportsdata.io/v3/nfl', 'nba': 'https://api.sportsdata.io/v3/nba', 'mlb': 'https://api.sportsdata.io/v3/mlb', 'nhl': 'https://api.sportsdata.io/v3/nhl', 'cfb': 'https://api.sportsdata.io/v3/cfb', 'cbb': 'https://api.sportsdata.io/v3/cbb', 'soccer': 'https://api.sportsdata.io/v3/soccer', 'mma': 'https://api.sportsdata.io/v3/mma', 'golf': 'https://api.sportsdata.io/v3/golf', 'nascar': 'https://api.sportsdata.io/v3/nascar', 'tennis': 'https://api.sportsdata.io/v3/tennis', } # Cache duration in seconds (30 days for historical data) CACHE_DURATION = 30 * 24 * 60 * 60 def __init__(self, api_key: Optional[str] = None, cache_dir: str = './data/cache'): """ Initialize the fetcher with API key. Args: api_key: SportsData.io API key (or set SPORTSDATA_API_KEY env var) cache_dir: Directory to store cached responses """ self.api_key = api_key or os.environ.get('SPORTSDATA_API_KEY') self.cache_dir = cache_dir # Create cache directory if it doesn't exist os.makedirs(cache_dir, exist_ok=True) def _get_cache_key(self, sport: str, endpoint: str, params: Dict) -> str: """Generate a unique cache key for the request""" param_str = json.dumps(params, sort_keys=True) key_str = f"{sport}:{endpoint}:{param_str}" return hashlib.md5(key_str.encode()).hexdigest() def _get_from_cache(self, cache_key: str) -> Optional[Dict]: """Get cached response if valid""" cache_file = os.path.join(self.cache_dir, f"{cache_key}.json") if os.path.exists(cache_file): try: with open(cache_file, 'r') as f: cached = json.load(f) # Check if cache is still valid cached_at = cached.get('cached_at', 0) if time.time() - cached_at < self.CACHE_DURATION: return cached.get('data') except (json.JSONDecodeError, IOError): pass return None def _save_to_cache(self, cache_key: str, data: Any) -> None: """Save response to cache""" cache_file = os.path.join(self.cache_dir, f"{cache_key}.json") try: with open(cache_file, 'w') as f: json.dump({ 'cached_at': time.time(), 'data': data }, f) except IOError: pass def _fetch_api(self, sport: str, endpoint: str, params: Dict = None) -> Any: """ Fetch data from the SportsData.io API. Args: sport: Sport code (nfl, nba, mlb, etc.) endpoint: API endpoint path params: Query parameters Returns: API response data """ params = params or {} # Check cache first cache_key = self._get_cache_key(sport, endpoint, params) cached = self._get_from_cache(cache_key) if cached is not None: return cached # Build URL base_url = self.API_BASES.get(sport) if not base_url: raise ValueError(f"Unsupported sport: {sport}") url = f"{base_url}/{endpoint}" # Add API key to params headers = { 'Ocp-Apim-Subscription-Key': self.api_key } if self.api_key else {} try: response = requests.get(url, params=params, headers=headers, timeout=30) response.raise_for_status() data = response.json() # Cache the response self._save_to_cache(cache_key, data) return data except requests.RequestException as e: print(f"API request failed for {sport}/{endpoint}: {e}") return None def _normalize_game(self, game: Dict, sport: str) -> Dict: """ Normalize game data to a common format across all sports. Returns: Normalized game dictionary """ # Common field mapping for different sports if sport in ('nfl', 'cfb'): return { 'game_id': str(game.get('GameID') or game.get('GameKey')), 'game_date': game.get('DateTime') or game.get('Day'), 'sport': sport, 'season': game.get('Season'), 'week': game.get('Week'), 'home_team': game.get('HomeTeam'), 'away_team': game.get('AwayTeam'), 'home_score': game.get('HomeScore'), 'away_score': game.get('AwayScore'), 'home_win': (game.get('HomeScore') or 0) > (game.get('AwayScore') or 0) if game.get('HomeScore') is not None else None, 'point_margin': (game.get('HomeScore') or 0) - (game.get('AwayScore') or 0) if game.get('HomeScore') is not None else None, 'total_points': (game.get('HomeScore') or 0) + (game.get('AwayScore') or 0) if game.get('HomeScore') is not None else None, 'status': game.get('Status'), 'point_spread': game.get('PointSpread'), 'over_under': game.get('OverUnder'), 'stadium': game.get('StadiumDetails', {}).get('Name') if isinstance(game.get('StadiumDetails'), dict) else game.get('Stadium'), } elif sport in ('nba', 'cbb'): return { 'game_id': str(game.get('GameID')), 'game_date': game.get('DateTime') or game.get('Day'), 'sport': sport, 'season': game.get('Season'), 'home_team': game.get('HomeTeam'), 'away_team': game.get('AwayTeam'), 'home_score': game.get('HomeTeamScore'), 'away_score': game.get('AwayTeamScore'), 'home_win': (game.get('HomeTeamScore') or 0) > (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'point_margin': (game.get('HomeTeamScore') or 0) - (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'total_points': (game.get('HomeTeamScore') or 0) + (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'status': game.get('Status'), 'point_spread': game.get('PointSpread'), 'over_under': game.get('OverUnder'), 'quarter': game.get('Quarter') or game.get('Period'), } elif sport == 'mlb': return { 'game_id': str(game.get('GameID')), 'game_date': game.get('DateTime') or game.get('Day'), 'sport': sport, 'season': game.get('Season'), 'home_team': game.get('HomeTeam'), 'away_team': game.get('AwayTeam'), 'home_score': game.get('HomeTeamRuns'), 'away_score': game.get('AwayTeamRuns'), 'home_win': (game.get('HomeTeamRuns') or 0) > (game.get('AwayTeamRuns') or 0) if game.get('HomeTeamRuns') is not None else None, 'point_margin': (game.get('HomeTeamRuns') or 0) - (game.get('AwayTeamRuns') or 0) if game.get('HomeTeamRuns') is not None else None, 'total_points': (game.get('HomeTeamRuns') or 0) + (game.get('AwayTeamRuns') or 0) if game.get('HomeTeamRuns') is not None else None, 'status': game.get('Status'), 'inning': game.get('Inning'), 'innings': game.get('Innings'), } elif sport == 'nhl': return { 'game_id': str(game.get('GameID')), 'game_date': game.get('DateTime') or game.get('Day'), 'sport': sport, 'season': game.get('Season'), 'home_team': game.get('HomeTeam'), 'away_team': game.get('AwayTeam'), 'home_score': game.get('HomeTeamScore'), 'away_score': game.get('AwayTeamScore'), 'home_win': (game.get('HomeTeamScore') or 0) > (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'point_margin': (game.get('HomeTeamScore') or 0) - (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'total_points': (game.get('HomeTeamScore') or 0) + (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'status': game.get('Status'), 'period': game.get('Period'), 'overtime': game.get('OverTime'), } elif sport == 'soccer': return { 'game_id': str(game.get('GameId')), 'game_date': game.get('DateTime'), 'sport': sport, 'competition': game.get('Competition', {}).get('Name') if isinstance(game.get('Competition'), dict) else None, 'home_team': game.get('HomeTeamName'), 'away_team': game.get('AwayTeamName'), 'home_score': game.get('HomeTeamScore'), 'away_score': game.get('AwayTeamScore'), 'home_win': (game.get('HomeTeamScore') or 0) > (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'draw': (game.get('HomeTeamScore') or 0) == (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'point_margin': (game.get('HomeTeamScore') or 0) - (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'total_points': (game.get('HomeTeamScore') or 0) + (game.get('AwayTeamScore') or 0) if game.get('HomeTeamScore') is not None else None, 'status': game.get('Status'), } else: # Generic format for other sports return { 'game_id': str(game.get('GameID') or game.get('GameId') or game.get('EventId')), 'game_date': game.get('DateTime') or game.get('Day') or game.get('Date'), 'sport': sport, 'status': game.get('Status'), 'raw_data': game, # Include raw data for sport-specific processing } def get_games(self, sport: str, season: int = None, date: str = None, week: int = None) -> List[Dict]: """ Get games for a sport. Args: sport: Sport code (nfl, nba, mlb, etc.) season: Season year (e.g., 2024) date: Specific date (YYYY-MM-DD format) week: Week number (for NFL/CFB) Returns: List of normalized game dictionaries """ sport = sport.lower() # Determine default season if not specified if season is None: today = datetime.now() # Most seasons span two years if today.month >= 9: season = today.year else: season = today.year - 1 if sport in ('nfl', 'cfb', 'cbb', 'nba', 'nhl') else today.year games = [] try: if sport == 'nfl': if week: data = self._fetch_api(sport, f'scores/json/ScoresByWeek/{season}/{week}') else: # Get all weeks all_games = [] for w in range(1, 19): # Regular season + playoffs week_data = self._fetch_api(sport, f'scores/json/ScoresByWeek/{season}/{w}') if week_data: all_games.extend(week_data) data = all_games elif sport == 'nba': if date: data = self._fetch_api(sport, f'scores/json/GamesByDate/{date}') else: data = self._fetch_api(sport, f'scores/json/Games/{season}') elif sport == 'mlb': if date: data = self._fetch_api(sport, f'scores/json/GamesByDate/{date}') else: data = self._fetch_api(sport, f'scores/json/Games/{season}') elif sport == 'nhl': if date: data = self._fetch_api(sport, f'scores/json/GamesByDate/{date}') else: data = self._fetch_api(sport, f'scores/json/Games/{season}') elif sport == 'cfb': if week: data = self._fetch_api(sport, f'scores/json/GamesByWeek/{season}/{week}') else: # Get all weeks all_games = [] for w in range(1, 16): week_data = self._fetch_api(sport, f'scores/json/GamesByWeek/{season}/{w}') if week_data: all_games.extend(week_data) data = all_games elif sport == 'cbb': if date: data = self._fetch_api(sport, f'scores/json/GamesByDate/{date}') else: data = self._fetch_api(sport, f'scores/json/Games/{season}') elif sport == 'soccer': # Soccer requires competition parameter data = self._fetch_api(sport, f'scores/json/Games/{season}') else: # Generic endpoint data = self._fetch_api(sport, f'scores/json/Games/{season}') if data: for game in data: normalized = self._normalize_game(game, sport) # Only include games with final scores if normalized.get('home_score') is not None: games.append(normalized) except Exception as e: print(f"Error fetching {sport} games: {e}") # Sort by date games.sort(key=lambda x: x.get('game_date') or '') return games def get_teams(self, sport: str) -> List[Dict]: """Get teams for a sport""" sport = sport.lower() try: data = self._fetch_api(sport, 'scores/json/Teams') return data or [] except Exception as e: print(f"Error fetching {sport} teams: {e}") return [] def get_players(self, sport: str, team: str = None) -> List[Dict]: """Get players for a sport""" sport = sport.lower() try: if team: data = self._fetch_api(sport, f'scores/json/Players/{team}') else: data = self._fetch_api(sport, 'scores/json/Players') return data or [] except Exception as e: print(f"Error fetching {sport} players: {e}") return [] def get_supported_sports(self) -> List[str]: """Get list of supported sports""" return list(self.API_BASES.keys()) # Fallback to NBA API when SportsData.io key not available class NBAApiFallback: """Fallback to free nba_api when SportsData.io not configured""" def __init__(self): try: from nba_api.stats.endpoints import leaguegamefinder import pandas as pd self.leaguegamefinder = leaguegamefinder self.pd = pd self.available = True except ImportError: self.available = False def get_games(self, season: str = '2023-24') -> List[Dict]: """Get NBA games using free nba_api""" if not self.available: return [] try: gamefinder = self.leaguegamefinder.LeagueGameFinder( season_nullable=season, season_type_nullable='Regular Season' ) df = gamefinder.get_data_frames()[0] # Group by game ID to pair home/away game_pairs = {} for _, row in df.iterrows(): game_id = str(row['GAME_ID']) matchup = row['MATCHUP'] is_home = ' vs. ' in matchup if game_id not in game_pairs: game_pairs[game_id] = { 'game_id': game_id, 'game_date': str(row['GAME_DATE']), 'sport': 'nba', 'season': season, } score = int(row['PTS']) if self.pd.notna(row['PTS']) else None win_loss = row.get('WL', '') if is_home: game_pairs[game_id]['home_team'] = row['TEAM_NAME'] game_pairs[game_id]['home_score'] = score game_pairs[game_id]['home_win'] = win_loss == 'W' else: game_pairs[game_id]['away_team'] = row['TEAM_NAME'] game_pairs[game_id]['away_score'] = score # Process and filter games = [] for game in game_pairs.values(): if game.get('home_score') is not None and game.get('away_score') is not None: game['point_margin'] = game['home_score'] - game['away_score'] game['total_points'] = game['home_score'] + game['away_score'] game['status'] = 'Final' games.append(game) games.sort(key=lambda x: x.get('game_date', '')) return games except Exception as e: print(f"NBA API fallback error: {e}") return [] def get_fetcher() -> SportsDataFetcher: """Get the appropriate data fetcher based on available credentials""" api_key = os.environ.get('SPORTSDATA_API_KEY') if api_key: return SportsDataFetcher(api_key) else: # Return fetcher that will use fallbacks print("Warning: SPORTSDATA_API_KEY not set. Some sports may not be available.") return SportsDataFetcher(None) # Test the fetcher if __name__ == '__main__': print("Testing SportsDataFetcher...") # Test with NBA API fallback fallback = NBAApiFallback() if fallback.available: games = fallback.get_games('2023-24') print(f"NBA API Fallback: {len(games)} games") if games: print(f"Sample game: {games[0]}") else: print("nba_api not available") # Test SportsData.io fetcher fetcher = get_fetcher() print(f"Supported sports: {fetcher.get_supported_sports()}") print("\n✅ SportsDataFetcher tests completed!")