Something went wrong. Try again.
Stats and whatnot for CFB and NFL
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135#!/usr/bin/env python3"""Cuts a frozen test fixture from nflverse's `schedules`/`pbp` datasets(architecture plan §10, §11 Phase 1 M1).
Fetches one season via the real `NflverseAdapter`, picks a handful of fullgames from one week, and writes two small Parquet files -- one with thegames' pbp rows, one with their schedule rows -- keeping only the columns`sources.nflverse.pbp.normalize_pbp` / `.schedules.normalize_schedules`actually read. Needs network access; the output fixtures are committed sotests never need it."""
import argparseimport sysfrom pathlib import Path
import polars as pl
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "pipeline" / "src"))
from matchupland.contracts.sources import FetchRequest # noqa: E402from matchupland.core.ids import LeagueId # noqa: E402from matchupland.sources.nflverse.adapter import NflverseAdapter # noqa: E402
# Exactly the raw columns `normalize_pbp`/`normalize_schedules` read --# see sources/nflverse/pbp.py and sources/nflverse/schedules.py._PBP_COLUMNS = [ "game_id", "season", "season_type", "week", "home_team", "away_team", "posteam", "defteam", "play_id", "drive", "qtr", "quarter_seconds_remaining", "half_seconds_remaining", "down", "ydstogo", "yardline_100", "score_differential", "play_type", "sack", "rush", "qb_scramble", "pass", "qb_dropback", "yards_gained", "air_yards", "ep", "epa", "wp", "wpa", "qb_hit", "interception", "fumble_lost", "tackled_for_loss", "pass_defense_1_player_id", "forced_fumble_player_1_player_id", "passer_player_id", "passer_player_name", "rusher_player_id", "rusher_player_name", "receiver_player_id", "receiver_player_name", "drive_end_transition", "drive_game_clock_start", "drive_game_clock_end",]
_SCHEDULE_COLUMNS = [ "season", "game_type", "week", "home_team", "away_team", "game_id", "home_score", "away_score", "gameday", "gametime", "stadium", "location", "spread_line", "total_line", "espn",]
def main(argv: list[str]) -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--league", default="nfl") parser.add_argument("--season", type=int, default=2024) parser.add_argument("--week", type=int, default=1) parser.add_argument("--games", type=int, default=3) parser.add_argument( "--out-dir", type=Path, default=Path(__file__).resolve().parents[1] / "fixtures" / "nfl" ) args = parser.parse_args(argv[1:])
league = LeagueId(args.league) adapter = NflverseAdapter()
schedules = adapter.fetch(FetchRequest(league=league, dataset="schedules", season=args.season)) assert schedules.frame is not None week_games = schedules.frame.filter(pl.col("week") == args.week).sort("game_id") chosen = week_games.head(args.games) game_ids = chosen["game_id"].to_list() if len(game_ids) < args.games: raise SystemExit(f"only found {len(game_ids)} games in week {args.week}")
pbp = adapter.fetch(FetchRequest(league=league, dataset="pbp", season=args.season)) assert pbp.frame is not None pbp_fixture = pbp.frame.filter(pl.col("game_id").is_in(game_ids)).select(_PBP_COLUMNS) schedule_fixture = chosen.select(_SCHEDULE_COLUMNS)
args.out_dir.mkdir(parents=True, exist_ok=True) pbp_path = args.out_dir / f"{args.season}_wk{args.week:02d}_{args.games}games.parquet" schedule_path = args.out_dir / f"{args.season}_wk{args.week:02d}_schedule.parquet" pbp_fixture.write_parquet(pbp_path) schedule_fixture.write_parquet(schedule_path)
total_bytes = pbp_path.stat().st_size + schedule_path.stat().st_size print(f"wrote {pbp_path} ({pbp_fixture.height} plays)") print(f"wrote {schedule_path} ({schedule_fixture.height} games)") print(f"total size: {total_bytes / 1024:.1f} KiB") return 0
if __name__ == "__main__": sys.exit(main(sys.argv))