From 4d9364a156c0355046bcb3ce8befa1e1f978c0c4 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 13:30:11 +0000 Subject: [PATCH 1/2] slice-8: season chaining and gp season Chains rounds r+1..R within each trial, carrying the car random walk forward; sprints via simulate_weekend; drivers' and constructors' tables from the same trials. gp season writes drivers_championship.csv and constructors_championship.csv; the remaining calendar comes from FastF1 through a stubbable loader and the command fails if it cannot be read. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QRJHJJN7CD3ym1wMdmWnsM --- docs/autopilot/decisions.md | 12 ++ docs/autopilot/queue.md | 2 +- src/cli.py | 155 +++++++++++++++++++++++++- src/sim/race.py | 17 ++- src/sim/season.py | 196 +++++++++++++++++++++++++++++++++ tests/test_season.py | 212 ++++++++++++++++++++++++++++++++++++ 6 files changed, 586 insertions(+), 8 deletions(-) create mode 100644 src/sim/season.py create mode 100644 tests/test_season.py diff --git a/docs/autopilot/decisions.md b/docs/autopilot/decisions.md index 0e38cf3..e3be7fb 100644 --- a/docs/autopilot/decisions.md +++ b/docs/autopilot/decisions.md @@ -160,3 +160,15 @@ Question: How should `collect_v2` flag a quali row whose time came from the s Chosen: `q_segment = "laps"` on the row plus a WARNING log naming the drivers. Per row: only a row with a `Position` and no Q1/Q2/Q3 time falls back, to that driver's fastest lap with `Deleted` laps excluded. Pole is still the official Q3→Q2→Q1 minimum wherever any official time exists; lap times set pole only when the session has none at all. Rejected: A new `time_source` column. Clearer, but it changes the `driver_rounds.csv` schema (COLUMNS, the validator, every existing file) for one round in nine seasons; `q_segment` has no readers in `src/` or `tests/`, so overloading it is free. Also rejected: triggering only when the whole session lacks times. Simpler, but the criterion is per row, and a partial upstream gap would then still drop drivers silently. Blast radius: `_fastest_session_laps` and the fallback block in `collect_v2._build_quali_rows`. Fully-timed sessions are byte-for-byte unchanged (the laps are never read; a test asserts it). Lap-derived times can order near-ties differently from the official classification (NOR/PIA/ANT in Miami sit within 0.002s); `finish_position` still comes from the results, so the grid is unaffected. + +## 2026-10-21 — slice-8 +Question: §3.4 says carry the team random walk forward. It does not say whether the other posterior terms (driver skill, noise scales, both DNF hazards) are redrawn per round or held for the season. +Chosen: One posterior draw per trial, held for every remaining round. Only `car` steps between rounds, by the pace model's own `SIGMA_DEV` (doubled for a reset year's first rounds, as in `_step_sigma_scale`). The incident hazard's grid slot is the season-start posterior-mean quali ranking (`cli._hazard_grid`, pre-weekend rule). +Rejected: Redrawing the parameters each round. That treats parameter uncertainty as round-to-round noise, so it averages out over a season and makes title odds too confident. +Blast radius: `sim.season.simulate_season`, `cli.forecast_season`. Nothing in `gp race` or the backtest changes. + +## 2026-10-21 — slice-8 +Question: Which line-up races the remaining rounds, and how are ties on points broken? +Chosen: The last completed round's entrants. Drivers who scored earlier but are no longer entered keep their points and score nothing more. Team current points come from the recorded points on R/S rows, so they follow whoever drove the car. Ties go to more Grand Prix wins, then column order. +Rejected: An explicit entry list (nothing in the repo maintains one) and the full FIA countback (second places onward). The countback adds a per-position tally for ties that fractional expected points almost never hit. +Blast radius: `cli.forecast_season`, `sim.season.final_positions`. diff --git a/docs/autopilot/queue.md b/docs/autopilot/queue.md index 857686c..e9f3752 100644 --- a/docs/autopilot/queue.md +++ b/docs/autopilot/queue.md @@ -379,7 +379,7 @@ Acceptance: ## slice-8: season chaining and `gp season` -Status: todo +Status: in-progress Spec: `docs/grid-prophet-v2-spec.md` §3.4, §6 (`gp season`, the two championship CSVs), §8 Phase 3. diff --git a/src/cli.py b/src/cli.py index 65ecc30..5324b74 100644 --- a/src/cli.py +++ b/src/cli.py @@ -6,7 +6,7 @@ Implemented: `fit`, `race --mode pre`, `race --mode post-quali` (Phase 2, spec sec 6 conditioning), sprint weekends (sec 3.4), and `backtest` (the -sec 7 evaluation). `season` is Phase 3. +sec 7 evaluation), and `season` (sec 3.4 chaining, Phase 3). """ from __future__ import annotations @@ -469,6 +469,152 @@ def cmd_race(args): print(f"Saved -> {sprint_path}") +# --------------------------------------------------------------------------- +# gp season +# --------------------------------------------------------------------------- + +def _season_so_far(driver_rounds: pd.DataFrame, season: int, through_round: int): + """Points and Grand Prix wins already scored in `season` up to and + including `through_round`, as ({driver: pts}, {team: pts}, {driver: wins}, + {team: wins}). Read from the recorded `points` of R and S rows, so team + totals follow whoever actually drove the car that weekend.""" + run = driver_rounds[(driver_rounds["season"] == season) & (driver_rounds["round"] <= through_round) + & (driver_rounds["session_type"].isin(["R", "S"]))] + pts = run.assign(points=run["points"].fillna(0.0)) + won = run[(run["session_type"] == "R") & (run["finish_position"] == 1)] + return (pts.groupby("abbreviation")["points"].sum().to_dict(), + pts.groupby("team")["points"].sum().to_dict(), + won.groupby("abbreviation").size().to_dict(), + won.groupby("team").size().to_dict()) + + +def forecast_season(idata, data, reliability_data, driver_rounds: pd.DataFrame, circuits: pd.DataFrame, + n_trials: int = 10_000, seed: int = 0, schedule_loader=None): + """Chain the rest of the fitted season (spec sec 3.4). + Returns `(drivers_championship, constructors_championship)`. + + The season and the last completed round are the fit's own cutoff, so the + chain starts exactly where the posterior ends. Entrants for the remaining + rounds are the last completed round's line-up.""" + from model.overtaking import build_overtaking_data, circuit_difficulty + from model.pace import CIRCUIT_TYPES + from model.reliability import sample_incident_dnf_prob, sample_mechanical_dnf_prob + from sim.race import load_points_table + from sim.season import ( + SeasonRound, SeasonTrialInputs, constructors_championship, drivers_championship, + remaining_calendar, simulate_season, step_sigma_for_round, + ) + + season, through = max(data.round_to_idx, key=data.round_to_idx.get) + calendar = remaining_calendar(season, through, loader=schedule_loader) + rng = np.random.default_rng(seed) + + entrants = _entrants_for_round(driver_rounds, season, through) + driver_ids, teams = entrants["abbreviation"].tolist(), entrants["team"].tolist() + drv_pts, team_pts, drv_wins, team_wins = _season_so_far(driver_rounds, season, through) + team_names = sorted(set(teams) | set(team_pts)) + + post = idata.posterior + n_posterior = post.sizes["chain"] * post.sizes["draw"] + draw_idx = rng.integers(0, n_posterior, size=n_trials) + + def flat(name): + arr = np.asarray(post[name]) + return arr.reshape(n_posterior, *arr.shape[2:])[draw_idx] + + car_all = np.asarray(post["car"].isel(round=data.round_to_idx[(season, through)]) + .transpose("chain", "draw", "team")).reshape(n_posterior, -1)[draw_idx] + car0_mean = flat("car0").mean(axis=1) + fitted_teams = list(post["car"].coords["team"].values) + car = np.column_stack([car_all[:, fitted_teams.index(t)] if t in fitted_teams else car0_mean + for t in team_names]) + + fitted_drivers = list(post["driver_skill"].coords["driver"].values) + skill_all, offset_all, track_all = flat("driver_skill"), flat("quali_offset"), flat("driver_track") + n_drivers = len(driver_ids) + skill, offset = np.zeros((n_trials, n_drivers)), np.zeros((n_trials, n_drivers)) + track = np.zeros((n_trials, n_drivers, len(CIRCUIT_TYPES))) + for i, d in enumerate(driver_ids): + if d in fitted_drivers: # an unseen driver keeps the field-average zeros + j = fitted_drivers.index(d) + skill[:, i], offset[:, i], track[:, i] = skill_all[:, j], offset_all[:, j], track_all[:, j] + + # Hazards are posterior parameters: one draw per trial, held for the season. + team_mech = {t: sample_mechanical_dnf_prob(reliability_data, t, season, n_trials, rng) + for t in sorted(set(teams))} + mech = np.column_stack([team_mech[t] for t in teams]) + quali_pace = car[:, [team_names.index(t) for t in teams]] + skill + offset + hazard_grid = _hazard_grid(None, quali_pace) + incident = np.column_stack([sample_incident_dnf_prob(reliability_data, d, hazard_grid[i], n_trials, rng) + for i, d in enumerate(driver_ids)]) + + overtaking = build_overtaking_data(driver_rounds, circuits, before_season=season, before_round=through + 1) + rounds = [] + for row in calendar.itertuples(index=False): + match = circuits[circuits["event_name"] == row.event_name] + ctype = match.iloc[0]["circuit_type"] if not match.empty else "mixed" + hand = float(match.iloc[0]["overtaking_difficulty"]) if not match.empty else None + rounds.append(SeasonRound( + round=int(row.round), has_sprint=bool(row.has_sprint), + circuit_type_idx=CIRCUIT_TYPES.index(ctype) if ctype in CIRCUIT_TYPES else 0, + overtaking_difficulty=circuit_difficulty(overtaking, row.event_name, hand), + step_sigma=step_sigma_for_round(season, int(row.round)))) + + inputs = SeasonTrialInputs( + driver_ids=driver_ids, teams=teams, team_names=team_names, car=car, + driver_skill=skill, driver_track=track, quali_offset=offset, + race_noise_nu=flat("race_nu"), race_noise_sigma=flat("race_sigma"), + quali_noise_sigma=flat("quali_sigma"), + mechanical_dnf_prob=mech, incident_dnf_prob=incident) + outcome = simulate_season(inputs, rounds, load_points_table("grand_prix", season), + load_points_table("sprint", season), rng) + + # Drivers who scored this season but are not on the current line-up keep + # their points and score nothing more. + all_drivers = driver_ids + sorted(set(drv_pts) - set(driver_ids)) + last_team = (driver_rounds[driver_rounds["season"] == season] + .sort_values("round").groupby("abbreviation")["team"].last()) + all_teams = teams + [last_team[d] for d in all_drivers[n_drivers:]] + future = np.zeros((n_trials, len(all_drivers))) + future_wins = np.zeros_like(future) + future[:, :n_drivers], future_wins[:, :n_drivers] = outcome.driver_points, outcome.driver_wins + cur_d = np.array([drv_pts.get(d, 0.0) for d in all_drivers]) + cur_dw = np.array([drv_wins.get(d, 0) for d in all_drivers]) + cur_t = np.array([team_pts.get(t, 0.0) for t in team_names]) + cur_tw = np.array([team_wins.get(t, 0) for t in team_names]) + drivers = drivers_championship(cur_d + future, cur_dw + future_wins, all_drivers, all_teams, cur_d) + constructors = constructors_championship(cur_t + outcome.team_points, cur_tw + outcome.team_wins, + team_names, cur_t) + return drivers, constructors + + +def cmd_season(args): + from sim.season import ScheduleUnavailable + + fit_dir = args.fit_dir + if fit_dir is None: + if not os.path.exists(LATEST_POINTER): + print("No fit found. Run `gp fit --through ...` first, or pass --fit-dir.", file=sys.stderr) + sys.exit(1) + fit_dir = open(LATEST_POINTER).read().strip() + idata, data, reliability_data, _ = load_fit(fit_dir) + try: + drivers, constructors = forecast_season(idata, data, reliability_data, _load_driver_rounds(), + _load_circuits(), n_trials=args.n_trials, seed=args.seed) + except ScheduleUnavailable as exc: + print(str(exc), file=sys.stderr) + sys.exit(1) + + out_dir = args.out_dir or fit_dir + os.makedirs(out_dir, exist_ok=True) + for name, table in (("drivers_championship.csv", drivers), + ("constructors_championship.csv", constructors)): + path = os.path.join(out_dir, name) + table.to_csv(path, index=False) + print(table.to_string(index=False)) + print(f"Saved -> {path}\n") + + # --------------------------------------------------------------------------- # gp backtest # --------------------------------------------------------------------------- @@ -520,6 +666,13 @@ def build_parser() -> argparse.ArgumentParser: "pit-lane starts; a sprint's own grid is unaffected)") p_race.set_defaults(func=cmd_race) + p_season = sub.add_parser("season", help="Chain the rest of the fitted season: both championships") + p_season.add_argument("--fit-dir", default=None) + p_season.add_argument("--out-dir", default=None) + p_season.add_argument("--n-trials", type=int, default=10_000) + p_season.add_argument("--seed", type=int, default=0) + p_season.set_defaults(func=cmd_season) + p_bt = sub.add_parser("backtest", help="Rolling backtest (spec sec 7)") p_bt.add_argument("--seasons", default="2022-2025") p_bt.add_argument("--rounds", default="coarse", choices=["coarse", "all"]) diff --git a/src/sim/race.py b/src/sim/race.py index cffc06c..a744abd 100644 --- a/src/sim/race.py +++ b/src/sim/race.py @@ -252,16 +252,21 @@ def simulate_weekend(inputs: RaceTrialInputs, rng: np.random.Generator, sprint_positions=sprint_positions, sprint_dnf=sprint_dnf) -def summarize_trials(positions: np.ndarray, dnf: np.ndarray, driver_ids: list, teams: list, - points_table: dict) -> pd.DataFrame: - """Turn (n_trials, n_drivers) position/dnf arrays into race_forecast.csv rows.""" - n_trials, n_drivers = positions.shape +def points_for_trials(positions: np.ndarray, dnf: np.ndarray, points_table: dict) -> np.ndarray: + """(n_trials, n_drivers) points scored, zero for a DNF.""" + n_drivers = positions.shape[1] points_by_pos = np.zeros(n_drivers + 1) for pos, pts in points_table.items(): if 1 <= pos <= n_drivers: points_by_pos[pos] = pts - trial_points = points_by_pos[positions] - trial_points = np.where(dnf, 0.0, trial_points) + return np.where(dnf, 0.0, points_by_pos[positions]) + + +def summarize_trials(positions: np.ndarray, dnf: np.ndarray, driver_ids: list, teams: list, + points_table: dict) -> pd.DataFrame: + """Turn (n_trials, n_drivers) position/dnf arrays into race_forecast.csv rows.""" + n_trials, n_drivers = positions.shape + trial_points = points_for_trials(positions, dnf, points_table) rows = [] for i, driver in enumerate(driver_ids): diff --git a/src/sim/season.py b/src/sim/season.py new file mode 100644 index 0000000..b356502 --- /dev/null +++ b/src/sim/season.py @@ -0,0 +1,196 @@ +"""Season chaining (spec sec 3.4): both championships from one set of trials. + +Each trial starts from one posterior draw of `car[t]` at the last fitted round +and, before each future round, steps every team's car by the pace model's +random walk (sec 3.1). The step lands on the running value, not the fitted +one, so the wander compounds over the rest of the year. Each round runs +through `sim.race.simulate_weekend` (a sprint shares its Grand Prix's pace +draw). Team points are the sum of that team's drivers' points in the same +trial, so the two championships are consistent by construction. + +Everything else that is a posterior parameter (driver skill, track +interaction, quali offset, noise scales, both DNF hazards) is drawn once per +trial and held for the season: only `car` has a random walk in sec 3.1. + +The remaining calendar comes from FastF1 through `remaining_calendar`'s +`loader` seam. If it cannot be read the command fails; it never invents one. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import numpy as np +import pandas as pd + +from sim.race import RaceTrialInputs, points_for_trials, simulate_weekend + + +class ScheduleUnavailable(RuntimeError): + """The season's event schedule could not be read. `gp season` stops.""" + + +def _fastf1_schedule(season: int) -> pd.DataFrame: + import fastf1 + + return fastf1.get_event_schedule(season, include_testing=False) + + +def remaining_calendar(season: int, after_round: int, loader=None) -> pd.DataFrame: + """Rounds after `after_round` as `round, event_name, has_sprint`, from a + FastF1-shaped `loader(season)`. Sprints are read off `EventFormat` as in + `collect_v2`. An unreadable or empty schedule raises; an empty result + only ever means the season is over.""" + loader = loader or _fastf1_schedule + try: + schedule = loader(season) + except Exception as exc: + raise ScheduleUnavailable( + f"Could not read the {season} event schedule from FastF1 ({exc}). " + "gp season will not guess the remaining calendar.") from exc + needed = {"RoundNumber", "EventName", "EventFormat"} + if schedule is None or len(schedule) == 0 or not needed.issubset(schedule.columns): + raise ScheduleUnavailable( + f"The {season} event schedule came back empty or without {sorted(needed)}. " + "gp season will not guess the remaining calendar.") + round_numbers = schedule["RoundNumber"].astype(int) + rounds = schedule[round_numbers > max(after_round, 0)].sort_values("RoundNumber") + return pd.DataFrame({ + "round": rounds["RoundNumber"].astype(int).to_numpy(), + "event_name": rounds["EventName"].astype(str).to_numpy(), + "has_sprint": rounds["EventFormat"].astype(str).str.lower().str.contains("sprint").to_numpy(), + }) + + +def step_sigma_for_round(season: int, round_: int) -> float: + """Random-walk step into `round_`, as `model.pace._step_sigma_scale` within a season.""" + from model.pace import ( + RULE_CHANGE_EARLY_ROUNDS, RULE_CHANGE_EARLY_SCALE, RULE_CHANGE_SEASONS, SIGMA_DEV, + ) + + early = season in RULE_CHANGE_SEASONS and round_ - 1 < RULE_CHANGE_EARLY_ROUNDS + return SIGMA_DEV * (RULE_CHANGE_EARLY_SCALE if early else 1.0) + + +@dataclass +class SeasonRound: + round: int + has_sprint: bool + circuit_type_idx: int # column of `SeasonTrialInputs.driver_track` + overtaking_difficulty: float + step_sigma: float # sd of the car random-walk step into this round + + +@dataclass +class SeasonTrialInputs: + """One posterior draw per trial, held for the whole season except `car`.""" + driver_ids: list + teams: list # per driver, len n_drivers + team_names: list # columns of `car` + car: np.ndarray # (n_trials, n_teams) at the last fitted round + driver_skill: np.ndarray # (n_trials, n_drivers) + driver_track: np.ndarray # (n_trials, n_drivers, n_circuit_types) + quali_offset: np.ndarray # (n_trials, n_drivers) + race_noise_nu: np.ndarray # (n_trials,) + race_noise_sigma: np.ndarray # (n_trials,) + quali_noise_sigma: np.ndarray # (n_trials,) + mechanical_dnf_prob: np.ndarray # (n_trials, n_drivers) + incident_dnf_prob: np.ndarray # (n_trials, n_drivers) + + +@dataclass +class SeasonOutcome: + """Simulated rounds only; `cli` adds the points already scored.""" + driver_points: np.ndarray # (n_trials, n_drivers) + driver_wins: np.ndarray # (n_trials, n_drivers), Grand Prix wins + team_points: np.ndarray # (n_trials, n_teams) + team_wins: np.ndarray # (n_trials, n_teams) + car_path: np.ndarray # (n_trials, n_rounds, n_teams), car at each simulated round + + +def simulate_season(inputs: SeasonTrialInputs, rounds: list, gp_points: dict, sprint_points: dict, + rng: np.random.Generator) -> SeasonOutcome: + """Chain `rounds` in order within every trial (sec 3.4).""" + from model.reliability import combined_dnf_prob + + n_trials, n_drivers = inputs.driver_skill.shape + team_col = np.array([inputs.team_names.index(t) for t in inputs.teams]) + dnf_prob = combined_dnf_prob(inputs.mechanical_dnf_prob, inputs.incident_dnf_prob) + + car = np.array(inputs.car, dtype=float, copy=True) + car_path = np.zeros((n_trials, len(rounds), car.shape[1])) + points = np.zeros((n_trials, n_drivers)) + wins = np.zeros((n_trials, n_drivers)) + for k, rnd in enumerate(rounds): + # The step lands on the running value: this is the carry-forward. + car = car + rng.normal(0.0, rnd.step_sigma, size=car.shape) + car_path[:, k] = car + car_d = car[:, team_col] + race_inputs = RaceTrialInputs( + driver_ids=inputs.driver_ids, + race_pace=car_d + inputs.driver_skill + inputs.driver_track[:, :, rnd.circuit_type_idx], + race_noise_nu=inputs.race_noise_nu, race_noise_sigma=inputs.race_noise_sigma, + quali_pace=car_d + inputs.driver_skill + inputs.quali_offset, + quali_noise_sigma=inputs.quali_noise_sigma, + dnf_prob=dnf_prob, + mechanical_dnf_prob=inputs.mechanical_dnf_prob, + incident_dnf_prob=inputs.incident_dnf_prob, + overtaking_difficulty=rnd.overtaking_difficulty, + ) + outcome = simulate_weekend(race_inputs, rng, has_sprint=rnd.has_sprint) + points += points_for_trials(outcome.positions, outcome.dnf, gp_points) + wins += (outcome.positions == 1) & ~outcome.dnf + if outcome.has_sprint: + points += points_for_trials(outcome.sprint_positions, outcome.sprint_dnf, sprint_points) + + membership = np.zeros((n_drivers, len(inputs.team_names))) + membership[np.arange(n_drivers), team_col] = 1.0 + return SeasonOutcome(driver_points=points, driver_wins=wins, + team_points=points @ membership, team_wins=wins @ membership, + car_path=car_path) + + +def final_positions(points: np.ndarray, wins: np.ndarray) -> np.ndarray: + """1-indexed championship position per trial; points ties go to more GP wins.""" + # Points come in steps of >= 0.5, wins are < 50 a season, so this key + # orders by points first and wins second without a per-trial lexsort. + key = np.asarray(points, dtype=float) * 100.0 + np.asarray(wins, dtype=float) + order = np.argsort(-key, axis=1, kind="stable") + return np.argsort(order, axis=1, kind="stable") + 1 + + +def _position_columns(pos: np.ndarray) -> dict: + return {"p10_final_position": float(np.percentile(pos, 10)), + "p50_final_position": float(np.percentile(pos, 50)), + "p90_final_position": float(np.percentile(pos, 90))} + + +def drivers_championship(total_points: np.ndarray, total_wins: np.ndarray, driver_ids: list, + teams: list, current_points) -> pd.DataFrame: + """sec 6 `drivers_championship.csv`, plus `team` and `current_points`.""" + pos = final_positions(total_points, total_wins) + rows = [] + for i, driver in enumerate(driver_ids): + rows.append({"driver": driver, "team": teams[i], "current_points": float(current_points[i]), + "p_title": float(np.mean(pos[:, i] == 1)), + "exp_final_points": float(np.mean(total_points[:, i])), + **_position_columns(pos[:, i]), + "p_top3": float(np.mean(pos[:, i] <= 3))}) + return pd.DataFrame(rows).sort_values("exp_final_points", ascending=False).reset_index(drop=True) + + +def constructors_championship(total_points: np.ndarray, total_wins: np.ndarray, team_names: list, + current_points) -> pd.DataFrame: + """sec 6 `constructors_championship.csv`, plus `current_points`. + `exp_share` is the per-trial share of all constructors' points, averaged.""" + pos = final_positions(total_points, total_wins) + totals = total_points.sum(axis=1, keepdims=True) + share = np.divide(total_points, totals, out=np.zeros_like(total_points), where=totals > 0) + rows = [] + for j, team in enumerate(team_names): + rows.append({"team": team, "current_points": float(current_points[j]), + "p_title": float(np.mean(pos[:, j] == 1)), + "exp_final_points": float(np.mean(total_points[:, j])), + "exp_share": float(np.mean(share[:, j])), + **_position_columns(pos[:, j])}) + return pd.DataFrame(rows).sort_values("exp_final_points", ascending=False).reset_index(drop=True) diff --git a/tests/test_season.py b/tests/test_season.py new file mode 100644 index 0000000..2fb2318 --- /dev/null +++ b/tests/test_season.py @@ -0,0 +1,212 @@ +"""Season chaining and `gp season` (spec sec 3.4, sec 6; autopilot slice-8).""" + +import numpy as np +import pandas as pd +import pytest + +import sim.season +from sim.season import ( + ScheduleUnavailable, SeasonRound, SeasonTrialInputs, final_positions, remaining_calendar, + simulate_season, +) + +GP = {1: 25, 2: 18, 3: 15, 4: 12, 5: 10, 6: 8, 7: 6, 8: 4, 9: 2, 10: 1} +SPRINT = {1: 8, 2: 7, 3: 6, 4: 5, 5: 4, 6: 3, 7: 2, 8: 1} + + +def _inputs(n_trials=2000, n_teams=5, dnf=0.0, seed=0): + """Synthetic posterior: two drivers per team, every trial's car identical + at the start, so any spread in `car_path` is the random walk's own.""" + rng = np.random.default_rng(seed) + n_drivers = 2 * n_teams + team_names = [f"T{j}" for j in range(n_teams)] + shape = (n_trials, n_drivers) + return SeasonTrialInputs( + driver_ids=[f"D{i}" for i in range(n_drivers)], + teams=[team_names[i // 2] for i in range(n_drivers)], + team_names=team_names, + car=np.tile(np.linspace(0.0, 1.0, n_teams), (n_trials, 1)), + driver_skill=rng.normal(0.0, 0.1, size=shape), + driver_track=np.zeros((*shape, 4)), + quali_offset=np.zeros(shape), + race_noise_nu=np.full(n_trials, 5.0), race_noise_sigma=np.full(n_trials, 0.3), + quali_noise_sigma=np.full(n_trials, 0.2), + mechanical_dnf_prob=np.full(shape, dnf), incident_dnf_prob=np.full(shape, dnf), + ) + + +def _rounds(n, sprint_rounds=(), step=0.08): + return [SeasonRound(round=k, has_sprint=k in sprint_rounds, circuit_type_idx=0, + overtaking_difficulty=3.0, step_sigma=step) for k in range(1, n + 1)] + + +class TestRandomWalkCarriesForward: + def test_late_round_car_draw_is_wider_than_early(self): + out = simulate_season(_inputs(), _rounds(12), GP, SPRINT, np.random.default_rng(1)) + early = out.car_path[:, 0].std(axis=0) + late = out.car_path[:, -1].std(axis=0) + # Compounding drift: sd after k steps is step * sqrt(k). Re-drawing each + # round's step from the fitted value would leave every round at one step. + assert np.all(late > 2.5 * early) + np.testing.assert_allclose(late, 0.08 * np.sqrt(12), rtol=0.1) + + def test_rounds_draw_pace_from_the_wandered_car(self): + # A huge step makes the order random by the last round even though T0's + # fitted car starts a full second clear: proof the resolver reads the + # running car, not `inputs.car`. + inputs = _inputs(n_trials=500) + calm = simulate_season(inputs, _rounds(3, step=0.0), GP, SPRINT, np.random.default_rng(2)) + wild = simulate_season(inputs, _rounds(3, step=5.0), GP, SPRINT, np.random.default_rng(2)) + assert calm.team_points[:, 0].mean() > wild.team_points[:, 0].mean() + 20 + + +class TestPoints: + def test_sprint_rounds_score_their_own_table(self): + out = simulate_season(_inputs(n_trials=50), _rounds(3, sprint_rounds=(2,)), GP, SPRINT, + np.random.default_rng(3)) + # No DNFs: every trial hands out three GP tables and one sprint table. + np.testing.assert_allclose(out.driver_points.sum(axis=1), 3 * 101 + 36) + + def test_sprint_goes_through_simulate_weekend(self, monkeypatch): + calls = [] + real = sim.season.simulate_weekend + + def spy(inputs, rng, has_sprint=False, **kw): + calls.append(has_sprint) + return real(inputs, rng, has_sprint=has_sprint, **kw) + + monkeypatch.setattr(sim.season, "simulate_weekend", spy) + simulate_season(_inputs(n_trials=10), _rounds(3, sprint_rounds=(3,)), GP, SPRINT, + np.random.default_rng(0)) + assert calls == [False, False, True] + + def test_constructor_points_are_the_sum_of_their_drivers_per_trial(self): + inputs = _inputs(n_trials=300, dnf=0.1) + out = simulate_season(inputs, _rounds(4, sprint_rounds=(2,)), GP, SPRINT, + np.random.default_rng(4)) + for j, team in enumerate(inputs.team_names): + cols = [i for i, t in enumerate(inputs.teams) if t == team] + np.testing.assert_array_equal(out.team_points[:, j], out.driver_points[:, cols].sum(axis=1)) + np.testing.assert_array_equal(out.team_wins[:, j], out.driver_wins[:, cols].sum(axis=1)) + + +def test_final_positions_break_points_ties_on_wins(): + points = np.array([[100.0, 100.0, 50.0]]) + wins = np.array([[1, 3, 0]]) + np.testing.assert_array_equal(final_positions(points, wins), [[2, 1, 3]]) + + +def _schedule(rounds=6, sprint=(3,)): + return pd.DataFrame({ + "RoundNumber": list(range(0, rounds + 1)), + "EventName": ["Pre-Season Testing"] + [f"Event {r}" for r in range(1, rounds + 1)], + "EventFormat": ["testing"] + ["sprint_qualifying" if r in sprint else "conventional" + for r in range(1, rounds + 1)], + }) + + +class TestRemainingCalendar: + def test_rounds_after_cutoff_with_sprint_flags(self): + cal = remaining_calendar(2026, 2, loader=lambda s: _schedule()) + assert cal["round"].tolist() == [3, 4, 5, 6] + assert cal["has_sprint"].tolist() == [True, False, False, False] + + def test_unreadable_schedule_fails_rather_than_inventing_one(self): + def broken(season): + raise ConnectionError("no network") + + with pytest.raises(ScheduleUnavailable, match="will not guess"): + remaining_calendar(2026, 2, loader=broken) + + +# --------------------------------------------------------------------------- +# gp season end to end, on test_cli's fake posterior +# --------------------------------------------------------------------------- + +def _fit_inputs(): + from model.reliability import build_reliability_data + from test_cli import _fake_fit, _sprint_weekend_rounds + + driver_rounds = _sprint_weekend_rounds() + driver_rounds["points"] = 0.0 + is_race = driver_rounds["session_type"] == "R" + driver_rounds.loc[is_race, "points"] = driver_rounds.loc[is_race, "finish_position"].map(GP).fillna(0.0) + idata, data = _fake_fit(driver_rounds) + reliability = build_reliability_data(driver_rounds, through_season=2024, through_round=2) + circuits = pd.DataFrame({"event_name": ["Bahrain Grand Prix"], "circuit_type": ["mixed"], + "overtaking_difficulty": [4.5]}) + return idata, data, reliability, driver_rounds, circuits + + +def test_forecast_season_tables(monkeypatch): + from cli import forecast_season + + idata, data, reliability, driver_rounds, circuits = _fit_inputs() + drivers, constructors = forecast_season(idata, data, reliability, driver_rounds, circuits, + n_trials=400, seed=0, + schedule_loader=lambda s: _schedule(rounds=5)) + for col in ["driver", "p_title", "exp_final_points", "p10_final_position", + "p50_final_position", "p90_final_position", "p_top3"]: + assert col in drivers.columns + for col in ["team", "p_title", "exp_final_points", "exp_share", "p10_final_position", + "p50_final_position", "p90_final_position"]: + assert col in constructors.columns + + # current points: d1 won both rounds already run + current = drivers.set_index("driver")["current_points"] + assert current["d1"] == 50.0 and current["d2"] == 36.0 + assert (drivers["exp_final_points"] >= drivers["current_points"]).all() + assert drivers["p_title"].sum() == pytest.approx(1.0) + assert constructors["exp_share"].sum() == pytest.approx(1.0) + # same trials: constructor expectations are their drivers' expectations summed + by_team = drivers.groupby("team")["exp_final_points"].sum() + for team, exp in constructors.set_index("team")["exp_final_points"].items(): + assert exp == pytest.approx(by_team[team]) + + +def test_season_finished_means_standings_are_final(): + from cli import forecast_season + + idata, data, reliability, driver_rounds, circuits = _fit_inputs() + drivers, _ = forecast_season(idata, data, reliability, driver_rounds, circuits, n_trials=50, + schedule_loader=lambda s: _schedule(rounds=2)) + leader = drivers.iloc[0] + assert leader["driver"] == "d1" and leader["p_title"] == 1.0 + np.testing.assert_allclose(drivers["exp_final_points"], drivers["current_points"]) + + +class TestCmdSeason: + def _args(self, tmp_path): + import argparse + + return argparse.Namespace(fit_dir=str(tmp_path), out_dir=str(tmp_path), n_trials=50, seed=0) + + def _patch(self, monkeypatch, schedule): + import cli + + idata, data, reliability, driver_rounds, circuits = _fit_inputs() + monkeypatch.setattr(cli, "load_fit", lambda d: (idata, data, reliability, None)) + monkeypatch.setattr(cli, "_load_driver_rounds", lambda: driver_rounds) + monkeypatch.setattr(cli, "_load_circuits", lambda: circuits) + monkeypatch.setattr(sim.season, "_fastf1_schedule", schedule) + + def test_writes_both_championship_csvs(self, monkeypatch, tmp_path): + import cli + + self._patch(monkeypatch, lambda s: _schedule(rounds=4)) + cli.cmd_season(self._args(tmp_path)) + assert pd.read_csv(tmp_path / "drivers_championship.csv").shape[0] == 4 + assert pd.read_csv(tmp_path / "constructors_championship.csv").shape[0] == 2 + + def test_unreadable_schedule_exits_with_a_message(self, monkeypatch, tmp_path, capsys): + import cli + + def broken(season): + raise ConnectionError("rate limited") + + self._patch(monkeypatch, broken) + with pytest.raises(SystemExit): + cli.cmd_season(self._args(tmp_path)) + assert "rate limited" in capsys.readouterr().err + assert not (tmp_path / "drivers_championship.csv").exists() + From 503aded6f361ffe31dc169574fec7f87e1d6cdc2 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 21 Oct 2026 13:30:56 +0000 Subject: [PATCH 2/2] queue: slice-8 in-review (PR #21) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QRJHJJN7CD3ym1wMdmWnsM --- docs/autopilot/queue.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/autopilot/queue.md b/docs/autopilot/queue.md index e9f3752..bd00ecc 100644 --- a/docs/autopilot/queue.md +++ b/docs/autopilot/queue.md @@ -379,7 +379,7 @@ Acceptance: ## slice-8: season chaining and `gp season` -Status: in-progress +Status: in-review — PR #21 Spec: `docs/grid-prophet-v2-spec.md` §3.4, §6 (`gp season`, the two championship CSVs), §8 Phase 3.