""" NBA ML Prediction System - Live Data Collector =============================================== Real-time data collection from NBA Live API endpoints. """ import logging from datetime import datetime, timezone from typing import Dict, List, Optional import time from nba_api.live.nba.endpoints import scoreboard, boxscore from src.config import API_CONFIG, NBA_TEAMS logger = logging.getLogger(__name__) class LiveDataCollector: """ Collects live game data from NBA API. Uses nba_api.live endpoints: - scoreboard.ScoreBoard() for today's games with live scores - boxscore.BoxScore(game_id) for detailed game box scores """ GAME_STATUS_MAP = { 1: "NOT_STARTED", 2: "IN_PROGRESS", 3: "FINAL" } def __init__(self): self._last_scoreboard_fetch = None self._cached_scoreboard = None self._cache_ttl = 10 # Seconds to cache scoreboard def get_live_scoreboard(self, force_refresh: bool = False) -> List[Dict]: """ Get today's games with live scores. Returns list of games with: - game_id, game_code - home_team, away_team (tricodes) - home_score, away_score - status (NOT_STARTED, IN_PROGRESS, FINAL) - period, clock - home_record, away_record """ # Use cache if available and fresh if not force_refresh and self._cached_scoreboard: if self._last_scoreboard_fetch: elapsed = (datetime.now() - self._last_scoreboard_fetch).total_seconds() if elapsed < self._cache_ttl: return self._cached_scoreboard try: sb = scoreboard.ScoreBoard() games_data = sb.games.get_dict() games_list = [] for game in games_data: game_status = game.get("gameStatus", 1) home_team = game.get("homeTeam", {}) away_team = game.get("awayTeam", {}) # Parse periods for quarter scores home_periods = [p.get("score", 0) for p in home_team.get("periods", [])] away_periods = [p.get("score", 0) for p in away_team.get("periods", [])] games_list.append({ "game_id": game.get("gameId", ""), "game_code": game.get("gameCode", ""), "game_date": game.get("gameEt", "")[:10] if game.get("gameEt") else "", "game_time_utc": game.get("gameTimeUTC", ""), "game_time_et": game.get("gameEt", ""), # Teams "home_team": home_team.get("teamTricode", ""), "away_team": away_team.get("teamTricode", ""), "home_team_id": home_team.get("teamId", 0), "away_team_id": away_team.get("teamId", 0), "home_team_name": home_team.get("teamName", ""), "away_team_name": away_team.get("teamName", ""), # Scores "home_score": home_team.get("score", 0), "away_score": away_team.get("score", 0), "home_periods": home_periods, "away_periods": away_periods, # Status "status": self.GAME_STATUS_MAP.get(game_status, "UNKNOWN"), "status_text": game.get("gameStatusText", ""), "period": game.get("period", 0), "clock": game.get("gameClock", ""), # Records "home_wins": home_team.get("wins", 0), "home_losses": home_team.get("losses", 0), "away_wins": away_team.get("wins", 0), "away_losses": away_team.get("losses", 0), "home_record": f"{home_team.get('wins', 0)}-{home_team.get('losses', 0)}", "away_record": f"{away_team.get('wins', 0)}-{away_team.get('losses', 0)}", # Leaders (for in-progress/final games) "home_leader": game.get("gameLeaders", {}).get("homeLeaders", {}), "away_leader": game.get("gameLeaders", {}).get("awayLeaders", {}), }) # Update cache self._cached_scoreboard = games_list self._last_scoreboard_fetch = datetime.now() logger.info(f"Fetched {len(games_list)} games from Live Scoreboard") return games_list except Exception as e: logger.error(f"Failed to fetch live scoreboard: {e}") return self._cached_scoreboard or [] def get_game_boxscore(self, game_id: str) -> Optional[Dict]: """ Get detailed box score for a specific game. Returns: Dict with game details, team stats, player stats """ try: box = boxscore.BoxScore(game_id) game_data = box.game.get_dict() return { "game_id": game_data.get("gameId", game_id), "game_status": game_data.get("gameStatus", 1), "game_status_text": game_data.get("gameStatusText", ""), "period": game_data.get("period", 0), "clock": game_data.get("gameClock", ""), "home_team": game_data.get("homeTeam", {}), "away_team": game_data.get("awayTeam", {}), "arena": game_data.get("arena", {}), } except Exception as e: logger.error(f"Failed to fetch boxscore for {game_id}: {e}") return None def get_game_status(self, game_id: str) -> str: """ Get current status for a specific game. Returns: 'NOT_STARTED', 'IN_PROGRESS', or 'FINAL' """ games = self.get_live_scoreboard() for game in games: if game["game_id"] == game_id: return game["status"] return "UNKNOWN" def get_winner(self, game_id: str) -> Optional[str]: """ Get the winner of a completed game. Returns: Team tricode of winner, or None if game not finished """ games = self.get_live_scoreboard() for game in games: if game["game_id"] == game_id: if game["status"] == "FINAL": if game["home_score"] > game["away_score"]: return game["home_team"] else: return game["away_team"] return None def get_games_by_status(self, status: str) -> List[Dict]: """ Filter games by status. Args: status: 'NOT_STARTED', 'IN_PROGRESS', or 'FINAL' """ games = self.get_live_scoreboard() return [g for g in games if g["status"] == status] def get_live_games(self) -> List[Dict]: """Get all currently in-progress games.""" return self.get_games_by_status("IN_PROGRESS") def get_final_games(self) -> List[Dict]: """Get all completed games from today.""" return self.get_games_by_status("FINAL") def get_upcoming_games(self) -> List[Dict]: """Get all not-yet-started games from today.""" return self.get_games_by_status("NOT_STARTED") def format_game_summary(self, game: Dict) -> str: """Format a game into a readable summary string.""" status = game["status"] away = game["away_team"] home = game["home_team"] if status == "NOT_STARTED": return f"{away} @ {home} - {game['status_text']}" elif status == "IN_PROGRESS": return f"{away} {game['away_score']} @ {home} {game['home_score']} - {game['status_text']}" else: # FINAL return f"{away} {game['away_score']} @ {home} {game['home_score']} - FINAL" # ============================================================================= # CLI INTERFACE # ============================================================================= if __name__ == "__main__": logging.basicConfig(level=logging.INFO) collector = LiveDataCollector() print("\n=== Today's NBA Games ===\n") games = collector.get_live_scoreboard() if not games: print("No games scheduled for today") else: for game in games: print(collector.format_game_summary(game)) if game["status"] == "IN_PROGRESS": print(f" Q{game['period']} {game['clock']}") print()