MLB Statcast Data Explorer

from processors import DailyGamesProcessor

# Process all MLB games for a specific date
processor = DailyGamesProcessor(date="2024-07-15")
processor.run()

# Or process a date range with monthly chunking
from processors import RangeGamesProcessor
processor = RangeGamesProcessor(
    start_date="2024-04-01",
    end_date="2024-09-30",
    chunk_size="monthly"
)
processor.run()

class GameProcessorBase(ABC):
    """Abstract base class for all processors"""

    @abstractmethod
    def run(self) -> None:
        pass

class DailyGamesProcessor(GameProcessorBase):
    """Batch-optimized: fetches Statcast ONCE per day"""

    def run(self):
        # 1. Get all games for date
        games = statsapi.schedule(date=self.date)

        # 2. Fetch Statcast data ONCE (not per-game)
        pitch_df = pybaseball.statcast(self.date)

        # 3. Process each game with shared data
        for game in games:
            processor = MLBGameProcessor(game, pitch_df)
            processor.process()

def compute_pitcher_snapshot(pitcher_id: int, as_of_date: date):
    """Compute cumulative stats BEFORE game date (no data leakage)"""

    return db.execute("""
        SELECT
            COUNT(DISTINCT game_id) as games_started,
            AVG(release_speed) as avg_velocity,
            SUM(CASE WHEN events = 'strikeout' THEN 1 ELSE 0 END) /
                NULLIF(COUNT(DISTINCT pa_id), 0) as k_rate,
            -- All stats use data BEFORE as_of_date
        FROM pitches p
        JOIN games g ON p.game_id = g.game_id
        WHERE p.pitcher_id = :pitcher_id
          AND g.game_date < :as_of_date  -- Critical: no leakage
    """, {"pitcher_id": pitcher_id, "as_of_date": as_of_date})