Spaces:

peggy44
/

savant

Running

savant / scripts /run_benchmark.py

Pegah

Claude Opus 4.7 (1M context)

add Track-B empirical benchmark (9 methods × 4 tasks on StatsBomb Open + 360)

4b35d65 4 months ago

14.7 kB

"""SAVANT Track-B benchmark orchestrator.
Wires:  config → data load → match-level split → leakage audit → baseline
(and later method) dispatch → per-task predictions → leaderboard.
For BM-4 the CLI runs the 3 baselines end-to-end on a configurable subset
of training competitions and writes their T3 team-match-value predictions
to `data/outputs/benchmark/`. Metric evaluation itself (Brier, Pearson ρ,
etc.) lands in BM-10. Model wrappers (xT, VAEP, DAxT, ...) land in BM-6+.
Usage
-----
    PYTHONPATH=src python scripts/run_benchmark.py                       # all baselines, all train comps
    PYTHONPATH=src python scripts/run_benchmark.py --max-matches 30      # smoke run
    PYTHONPATH=src python scripts/run_benchmark.py --models xg_per_shot  # single baseline
Every run writes:
    data/outputs/benchmark/
      ├── baselines/{baseline_id}__t3_team_match_values.parquet
      ├── run_manifest.json      # timings, config snapshot, git sha
      └── (later) leaderboard.csv, per_task/*.csv
"""

from __future__ import annotations

import argparse
import json
import logging
import subprocess
import sys
import time
from datetime import datetime, timezone
from pathlib import Path

import pandas as pd

REPO = Path(__file__).resolve().parents[1]
if str(REPO / "src") not in sys.path:
    sys.path.insert(0, str(REPO / "src"))

from savant.benchmarking import (
    BASELINES_BY_ID, METHODS_BY_ID, SPADLLoader, StatsBombOpenLoader,
    holdout_competitions, leakage_audit, load_config, split_matches,
    split_spec, train_competitions,
)

logger = logging.getLogger("run_benchmark")


def _git_sha() -> str:
    try:
        r = subprocess.run(
            ["git", "rev-parse", "--short", "HEAD"],
            cwd=REPO, capture_output=True, text=True, timeout=3,
        )
        return r.stdout.strip() or "unknown"
    except Exception:
        return "unknown"


def _step(msg: str, elapsed: float | None = None) -> None:
    prefix = f"[{elapsed:6.2f}s] " if elapsed is not None else " " * 10
    print(f"{prefix}{msg}")


def run(
    baseline_ids: list[str] | None = None,
    method_ids: list[str] | None = None,
    max_matches: int | None = None,
    out_dir: Path | None = None,
    competition_index: int | None = None,
    holdout: bool = False,
) -> dict:
    """Programmatic entrypoint. Returns a manifest dict.
    Set `competition_index` to N (0-indexed) to restrict the run to a
    single training competition — used by the by-competition runner so
    that 360-heavy competitions don't OOM alongside the others.
    """
    t_total = time.perf_counter()
    print("\n=== SAVANT Track-B benchmark ===\n")

    baseline_ids = baseline_ids if baseline_ids is not None else list(BASELINES_BY_ID.keys())
    method_ids = method_ids if method_ids is not None else list(METHODS_BY_ID.keys())
    out_dir = out_dir or (REPO / "data" / "outputs" / "benchmark")
    (out_dir / "baselines").mkdir(parents=True, exist_ok=True)
    (out_dir / "methods").mkdir(parents=True, exist_ok=True)
    (out_dir / "methods_per_action").mkdir(parents=True, exist_ok=True)

    cfg = load_config()
    spec = split_spec(cfg)
    # `holdout=True` reroutes the pool from the training corpus to the
    # T4 evaluation set (WSL 2020/21 + AFCON 2023 per configs/benchmark_tasks.yaml).
    train_comps = (holdout_competitions(cfg) if holdout else train_competitions(cfg))
    corpus_label = "holdout" if holdout else "train"
    if competition_index is not None:
        if not (0 <= competition_index < len(train_comps)):
            raise IndexError(
                f"--competition-index {competition_index} out of range "
                f"[0, {len(train_comps)})",
            )
        train_comps = [train_comps[competition_index]]
        _step(f"config loaded: 1 {corpus_label} competition (index {competition_index}: "
              f"{train_comps[0].name} {train_comps[0].season}), "
              f"{len(baseline_ids)} baselines + {len(method_ids)} methods to run")
    else:
        _step(f"config loaded: {len(train_comps)} {corpus_label} competitions, "
              f"{len(baseline_ids)} baselines + {len(method_ids)} methods to run")

    # --- Data load -----------------------------------------------------
    spadl_loader = SPADLLoader()
    sb_loader = StatsBombOpenLoader()   # for raw events (xG baseline uses these)

    all_spadl: list[pd.DataFrame] = []
    all_events: list[pd.DataFrame] = []
    all_frames: list[pd.DataFrame] = []
    total_matches = 0
    remaining = max_matches
    need_360 = any(
        getattr(METHODS_BY_ID.get(mid, None), "requires_360", False)
        for mid in method_ids
    )

    for comp_spec in train_comps:
        t0 = time.perf_counter()
        sc = spadl_loader.resolve(comp_spec.name, comp_spec.season)
        sb_comp = sb_loader.resolve(comp_spec.name, comp_spec.season)
        matches = spadl_loader.list_matches(sc)
        game_ids = matches["game_id"].tolist()
        if remaining is not None:
            game_ids = game_ids[:remaining]
            remaining = max(0, remaining - len(game_ids))
        if not game_ids:
            continue

        spadl_df = spadl_loader.load_actions_batch(sc, game_ids)
        events_df = sb_loader.load_events_batch(sb_comp, game_ids)
        # Normalise the match-id column across the two loaders for the
        # baselines to share a common `game_id`.
        if "match_id" in events_df.columns and "game_id" not in events_df.columns:
            events_df = events_df.rename(columns={"match_id": "game_id"})

        all_spadl.append(spadl_df)
        all_events.append(events_df)

        # Only fetch 360 if a selected method actually needs it (avoids the
        # extra round-trip on runs that don't include OBSO_FF).
        if need_360 and sb_comp.has_360:
            frames_df = sb_loader.load_360_batch(sb_comp, game_ids)
            if frames_df is not None and len(frames_df):
                if "match_id" in frames_df.columns and "game_id" not in frames_df.columns:
                    frames_df = frames_df.rename(columns={"match_id": "game_id"})
                all_frames.append(frames_df)

        total_matches += len(game_ids)
        _step(f"loaded {comp_spec.name} {comp_spec.season}: "
              f"{len(game_ids)} matches → {len(spadl_df):,} SPADL actions",
              time.perf_counter() - t0)
        if remaining is not None and remaining == 0:
            break

    if not all_spadl:
        raise RuntimeError("No matches loaded — check config or --max-matches")

    actions = pd.concat(all_spadl, ignore_index=True)
    events = pd.concat(all_events, ignore_index=True)
    frames = pd.concat(all_frames, ignore_index=True) if all_frames else None
    matches_meta = pd.DataFrame({"game_id": sorted(actions["game_id"].unique())})
    combined_msg = (f"combined: {total_matches} matches, {len(actions):,} SPADL actions, "
                    f"{len(events):,} raw events")
    if frames is not None:
        combined_msg += f", {len(frames):,} 360 frame rows"
    _step(combined_msg)

    # --- Split + leakage audit ----------------------------------------
    t0 = time.perf_counter()
    split = split_matches(matches_meta["game_id"].tolist(), spec=spec)
    audit = leakage_audit(actions, split, match_id_column="game_id")
    _step(f"split (seed={spec.seed}): "
          f"{audit['n_matches_train']}/{audit['n_matches_val']}/{audit['n_matches_test']} matches, "
          f"{audit['n_events_train']:,}/{audit['n_events_val']:,}/{audit['n_events_test']:,} actions "
          "(leakage audit PASSED)",
          time.perf_counter() - t0)

    # --- Baseline dispatch --------------------------------------------
    baseline_results: dict[str, dict] = {}
    for bid in baseline_ids:
        if bid not in BASELINES_BY_ID:
            print(f"  WARNING: unknown baseline {bid!r}, skipping")
            continue
        t0 = time.perf_counter()
        model_cls = BASELINES_BY_ID[bid]
        model = model_cls()
        model.fit(actions, matches_meta)

        # T3: team-match values. xG baseline needs raw events, the other
        # two use SPADL — the model's method signature is the same either way.
        input_frame = events if bid == "xg_per_shot" else actions
        preds = model.predict_team_match_values(input_frame, matches_meta)

        out_path = out_dir / "baselines" / f"{bid}__t3_team_match_values.parquet"
        preds.to_parquet(out_path, index=False)
        try:
            out_path_rel = str(out_path.relative_to(REPO))
        except ValueError:
            out_path_rel = str(out_path)   # out_dir outside repo (e.g. tests)
        baseline_results[bid] = {
            "n_predictions": int(len(preds)),
            "n_distinct_games": int(preds["game_id"].nunique()),
            "n_distinct_teams": int(preds["team_id"].nunique()),
            "value_summary": {
                "mean": float(preds["value"].mean()),
                "std": float(preds["value"].std()),
                "min": float(preds["value"].min()),
                "max": float(preds["value"].max()),
            },
            "output_path": out_path_rel,
            "duration_s": round(time.perf_counter() - t0, 3),
        }
        _step(f"baseline {bid}: {len(preds)} team-match predictions "
              f"(mean={preds['value'].mean():.2f}, max={preds['value'].max():.2f})",
              time.perf_counter() - t0)

    # --- Method dispatch (fit on train split; predict on all matches) ---
    train_actions = actions[actions["game_id"].isin(split.train)]
    train_events = events[events["game_id"].isin(split.train)]
    method_results: dict[str, dict] = {}
    for mid in method_ids:
        if mid not in METHODS_BY_ID:
            print(f"  WARNING: unknown method {mid!r}, skipping")
            continue
        t0 = time.perf_counter()
        model_cls = METHODS_BY_ID[mid]
        model = model_cls()
        needs_frames = getattr(model_cls, "requires_360", False)
        # xT-Action's fit accepts optional raw_events for real xG labelling.
        if mid == "xt_action_sumpter":
            model.fit(train_actions, matches_meta, raw_events=train_events)
        elif needs_frames:
            train_frames = (frames[frames["game_id"].isin(split.train)]
                            if frames is not None else None)
            model.fit(train_actions, matches_meta, frames=train_frames)
        else:
            model.fit(train_actions, matches_meta)
        if needs_frames:
            preds = model.predict_team_match_values(actions, matches_meta, frames=frames)
            per_action_df = model.predict_per_action(actions, frames=frames)
        else:
            preds = model.predict_team_match_values(actions, matches_meta)
            per_action_df = model.predict_per_action(actions)

        out_path = out_dir / "methods" / f"{mid}__t3_team_match_values.parquet"
        preds.to_parquet(out_path, index=False)

        # BM-12: per-action values for T1 evaluation. Include NaNs so
        # DAxT / OBSO_FF partial-coverage methods stay diagnosable downstream.
        pa_path = out_dir / "methods_per_action" / f"{mid}__per_action_values.parquet"
        per_action_df.to_parquet(pa_path, index=False)
        try:
            out_path_rel = str(out_path.relative_to(REPO))
        except ValueError:
            out_path_rel = str(out_path)
        method_results[mid] = {
            "n_predictions": int(len(preds)),
            "n_distinct_games": int(preds["game_id"].nunique()),
            "n_distinct_teams": int(preds["team_id"].nunique()),
            "value_summary": {
                "mean": float(preds["value"].mean()),
                "std":  float(preds["value"].std()),
                "min":  float(preds["value"].min()),
                "max":  float(preds["value"].max()),
            },
            "output_path": out_path_rel,
            "duration_s": round(time.perf_counter() - t0, 3),
        }
        _step(f"method {mid}: {len(preds)} team-match predictions "
              f"(mean={preds['value'].mean():.3f}, max={preds['value'].max():.3f})",
              time.perf_counter() - t0)

    # --- Manifest -----------------------------------------------------
    manifest = {
        "run_at": datetime.now(timezone.utc).isoformat(),
        "duration_s": round(time.perf_counter() - t_total, 3),
        "git_sha": _git_sha(),
        "config_path": "configs/benchmark_tasks.yaml",
        "n_train_competitions": len(train_comps),
        "n_matches_total": int(total_matches),
        "split_seed": spec.seed,
        "audit": audit,
        "baselines": baseline_results,
        "methods": method_results,
    }
    manifest_path = out_dir / "run_manifest.json"
    manifest_path.write_text(json.dumps(manifest, indent=2, default=str) + "\n")
    try:
        manifest_pretty = manifest_path.relative_to(REPO)
    except ValueError:
        manifest_pretty = manifest_path
    _step(f"manifest saved → {manifest_pretty}")

    print(f"\n=== done in {manifest['duration_s']:.1f}s ===\n")
    return manifest


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument(
        "--baselines", nargs="*", default=None,
        help="baseline ids to run. Default: all baselines. Empty string = none.",
    )
    parser.add_argument(
        "--methods", nargs="*", default=None,
        help="method ids to run. Default: all methods. Empty string = none.",
    )
    parser.add_argument(
        "--max-matches", type=int, default=None,
        help="Cap total matches across all competitions (for smoke runs).",
    )
    parser.add_argument(
        "--out-dir", type=Path, default=None,
        help="Override output directory.",
    )
    parser.add_argument(
        "--competition-index", type=int, default=None,
        help="If set, only process the competition at this 0-indexed position "
             "in the selected corpus (train or holdout). Used by "
             "scripts/run_benchmark_by_competition.py to spread memory "
             "pressure over sequential runs.",
    )
    parser.add_argument(
        "--holdout", action="store_true",
        help="Run on the T4 holdout competitions (WSL 2020/21 + AFCON 2023) "
             "instead of the training corpus.",
    )
    args = parser.parse_args()
    run(
        baseline_ids=args.baselines,
        method_ids=args.methods,
        max_matches=args.max_matches,
        out_dir=args.out_dir,
        competition_index=args.competition_index,
        holdout=args.holdout,
    )


if __name__ == "__main__":
    main()

	"""SAVANT Track-B benchmark orchestrator.

	Wires: config → data load → match-level split → leakage audit → baseline
	(and later method) dispatch → per-task predictions → leaderboard.

	For BM-4 the CLI runs the 3 baselines end-to-end on a configurable subset
	of training competitions and writes their T3 team-match-value predictions
	to `data/outputs/benchmark/`. Metric evaluation itself (Brier, Pearson ρ,
	etc.) lands in BM-10. Model wrappers (xT, VAEP, DAxT, ...) land in BM-6+.

	Usage
	-----
	PYTHONPATH=src python scripts/run_benchmark.py # all baselines, all train comps
	PYTHONPATH=src python scripts/run_benchmark.py --max-matches 30 # smoke run
	PYTHONPATH=src python scripts/run_benchmark.py --models xg_per_shot # single baseline

	Every run writes:
	data/outputs/benchmark/
	├── baselines/{baseline_id}__t3_team_match_values.parquet
	├── run_manifest.json # timings, config snapshot, git sha
	└── (later) leaderboard.csv, per_task/*.csv
	"""

	from __future__ import annotations

	import argparse
	import json
	import logging
	import subprocess
	import sys
	import time
	from datetime import datetime, timezone
	from pathlib import Path

	import pandas as pd

	REPO = Path(__file__).resolve().parents[1]
	if str(REPO / "src") not in sys.path:
	sys.path.insert(0, str(REPO / "src"))

	from savant.benchmarking import (
	BASELINES_BY_ID, METHODS_BY_ID, SPADLLoader, StatsBombOpenLoader,
	holdout_competitions, leakage_audit, load_config, split_matches,
	split_spec, train_competitions,
	)

	logger = logging.getLogger("run_benchmark")


	def _git_sha() -> str:
	try:
	r = subprocess.run(
	["git", "rev-parse", "--short", "HEAD"],
	cwd=REPO, capture_output=True, text=True, timeout=3,
	)
	return r.stdout.strip() or "unknown"
	except Exception:
	return "unknown"


	def _step(msg: str, elapsed: float \| None = None) -> None:
	prefix = f"[{elapsed:6.2f}s] " if elapsed is not None else " " * 10
	print(f"{prefix}{msg}")


	def run(
	baseline_ids: list[str] \| None = None,
	method_ids: list[str] \| None = None,
	max_matches: int \| None = None,
	out_dir: Path \| None = None,
	competition_index: int \| None = None,
	holdout: bool = False,
	) -> dict:
	"""Programmatic entrypoint. Returns a manifest dict.

	Set `competition_index` to N (0-indexed) to restrict the run to a
	single training competition — used by the by-competition runner so
	that 360-heavy competitions don't OOM alongside the others.
	"""
	t_total = time.perf_counter()
	print("\n=== SAVANT Track-B benchmark ===\n")

	baseline_ids = baseline_ids if baseline_ids is not None else list(BASELINES_BY_ID.keys())
	method_ids = method_ids if method_ids is not None else list(METHODS_BY_ID.keys())
	out_dir = out_dir or (REPO / "data" / "outputs" / "benchmark")
	(out_dir / "baselines").mkdir(parents=True, exist_ok=True)
	(out_dir / "methods").mkdir(parents=True, exist_ok=True)
	(out_dir / "methods_per_action").mkdir(parents=True, exist_ok=True)

	cfg = load_config()
	spec = split_spec(cfg)
	# `holdout=True` reroutes the pool from the training corpus to the
	# T4 evaluation set (WSL 2020/21 + AFCON 2023 per configs/benchmark_tasks.yaml).
	train_comps = (holdout_competitions(cfg) if holdout else train_competitions(cfg))
	corpus_label = "holdout" if holdout else "train"
	if competition_index is not None:
	if not (0 <= competition_index < len(train_comps)):
	raise IndexError(
	f"--competition-index {competition_index} out of range "
	f"[0, {len(train_comps)})",
	)
	train_comps = [train_comps[competition_index]]
	_step(f"config loaded: 1 {corpus_label} competition (index {competition_index}: "
	f"{train_comps[0].name} {train_comps[0].season}), "
	f"{len(baseline_ids)} baselines + {len(method_ids)} methods to run")
	else:
	_step(f"config loaded: {len(train_comps)} {corpus_label} competitions, "
	f"{len(baseline_ids)} baselines + {len(method_ids)} methods to run")

	# --- Data load -----------------------------------------------------
	spadl_loader = SPADLLoader()
	sb_loader = StatsBombOpenLoader() # for raw events (xG baseline uses these)

	all_spadl: list[pd.DataFrame] = []
	all_events: list[pd.DataFrame] = []
	all_frames: list[pd.DataFrame] = []
	total_matches = 0
	remaining = max_matches
	need_360 = any(
	getattr(METHODS_BY_ID.get(mid, None), "requires_360", False)
	for mid in method_ids
	)

	for comp_spec in train_comps:
	t0 = time.perf_counter()
	sc = spadl_loader.resolve(comp_spec.name, comp_spec.season)
	sb_comp = sb_loader.resolve(comp_spec.name, comp_spec.season)
	matches = spadl_loader.list_matches(sc)
	game_ids = matches["game_id"].tolist()
	if remaining is not None:
	game_ids = game_ids[:remaining]
	remaining = max(0, remaining - len(game_ids))
	if not game_ids:
	continue

	spadl_df = spadl_loader.load_actions_batch(sc, game_ids)
	events_df = sb_loader.load_events_batch(sb_comp, game_ids)
	# Normalise the match-id column across the two loaders for the
	# baselines to share a common `game_id`.
	if "match_id" in events_df.columns and "game_id" not in events_df.columns:
	events_df = events_df.rename(columns={"match_id": "game_id"})

	all_spadl.append(spadl_df)
	all_events.append(events_df)

	# Only fetch 360 if a selected method actually needs it (avoids the
	# extra round-trip on runs that don't include OBSO_FF).
	if need_360 and sb_comp.has_360:
	frames_df = sb_loader.load_360_batch(sb_comp, game_ids)
	if frames_df is not None and len(frames_df):
	if "match_id" in frames_df.columns and "game_id" not in frames_df.columns:
	frames_df = frames_df.rename(columns={"match_id": "game_id"})
	all_frames.append(frames_df)

	total_matches += len(game_ids)
	_step(f"loaded {comp_spec.name} {comp_spec.season}: "
	f"{len(game_ids)} matches → {len(spadl_df):,} SPADL actions",
	time.perf_counter() - t0)
	if remaining is not None and remaining == 0:
	break

	if not all_spadl:
	raise RuntimeError("No matches loaded — check config or --max-matches")

	actions = pd.concat(all_spadl, ignore_index=True)
	events = pd.concat(all_events, ignore_index=True)
	frames = pd.concat(all_frames, ignore_index=True) if all_frames else None
	matches_meta = pd.DataFrame({"game_id": sorted(actions["game_id"].unique())})
	combined_msg = (f"combined: {total_matches} matches, {len(actions):,} SPADL actions, "
	f"{len(events):,} raw events")
	if frames is not None:
	combined_msg += f", {len(frames):,} 360 frame rows"
	_step(combined_msg)

	# --- Split + leakage audit ----------------------------------------
	t0 = time.perf_counter()
	split = split_matches(matches_meta["game_id"].tolist(), spec=spec)
	audit = leakage_audit(actions, split, match_id_column="game_id")
	_step(f"split (seed={spec.seed}): "
	f"{audit['n_matches_train']}/{audit['n_matches_val']}/{audit['n_matches_test']} matches, "
	f"{audit['n_events_train']:,}/{audit['n_events_val']:,}/{audit['n_events_test']:,} actions "
	"(leakage audit PASSED)",
	time.perf_counter() - t0)

	# --- Baseline dispatch --------------------------------------------
	baseline_results: dict[str, dict] = {}
	for bid in baseline_ids:
	if bid not in BASELINES_BY_ID:
	print(f" WARNING: unknown baseline {bid!r}, skipping")
	continue
	t0 = time.perf_counter()
	model_cls = BASELINES_BY_ID[bid]
	model = model_cls()
	model.fit(actions, matches_meta)

	# T3: team-match values. xG baseline needs raw events, the other
	# two use SPADL — the model's method signature is the same either way.
	input_frame = events if bid == "xg_per_shot" else actions
	preds = model.predict_team_match_values(input_frame, matches_meta)

	out_path = out_dir / "baselines" / f"{bid}__t3_team_match_values.parquet"
	preds.to_parquet(out_path, index=False)
	try:
	out_path_rel = str(out_path.relative_to(REPO))
	except ValueError:
	out_path_rel = str(out_path) # out_dir outside repo (e.g. tests)
	baseline_results[bid] = {
	"n_predictions": int(len(preds)),
	"n_distinct_games": int(preds["game_id"].nunique()),
	"n_distinct_teams": int(preds["team_id"].nunique()),
	"value_summary": {
	"mean": float(preds["value"].mean()),
	"std": float(preds["value"].std()),
	"min": float(preds["value"].min()),
	"max": float(preds["value"].max()),
	},
	"output_path": out_path_rel,
	"duration_s": round(time.perf_counter() - t0, 3),
	}
	_step(f"baseline {bid}: {len(preds)} team-match predictions "
	f"(mean={preds['value'].mean():.2f}, max={preds['value'].max():.2f})",
	time.perf_counter() - t0)

	# --- Method dispatch (fit on train split; predict on all matches) ---
	train_actions = actions[actions["game_id"].isin(split.train)]
	train_events = events[events["game_id"].isin(split.train)]
	method_results: dict[str, dict] = {}
	for mid in method_ids:
	if mid not in METHODS_BY_ID:
	print(f" WARNING: unknown method {mid!r}, skipping")
	continue
	t0 = time.perf_counter()
	model_cls = METHODS_BY_ID[mid]
	model = model_cls()
	needs_frames = getattr(model_cls, "requires_360", False)
	# xT-Action's fit accepts optional raw_events for real xG labelling.
	if mid == "xt_action_sumpter":
	model.fit(train_actions, matches_meta, raw_events=train_events)
	elif needs_frames:
	train_frames = (frames[frames["game_id"].isin(split.train)]
	if frames is not None else None)
	model.fit(train_actions, matches_meta, frames=train_frames)
	else:
	model.fit(train_actions, matches_meta)
	if needs_frames:
	preds = model.predict_team_match_values(actions, matches_meta, frames=frames)
	per_action_df = model.predict_per_action(actions, frames=frames)
	else:
	preds = model.predict_team_match_values(actions, matches_meta)
	per_action_df = model.predict_per_action(actions)

	out_path = out_dir / "methods" / f"{mid}__t3_team_match_values.parquet"
	preds.to_parquet(out_path, index=False)

	# BM-12: per-action values for T1 evaluation. Include NaNs so
	# DAxT / OBSO_FF partial-coverage methods stay diagnosable downstream.
	pa_path = out_dir / "methods_per_action" / f"{mid}__per_action_values.parquet"
	per_action_df.to_parquet(pa_path, index=False)
	try:
	out_path_rel = str(out_path.relative_to(REPO))
	except ValueError:
	out_path_rel = str(out_path)
	method_results[mid] = {
	"n_predictions": int(len(preds)),
	"n_distinct_games": int(preds["game_id"].nunique()),
	"n_distinct_teams": int(preds["team_id"].nunique()),
	"value_summary": {
	"mean": float(preds["value"].mean()),
	"std": float(preds["value"].std()),
	"min": float(preds["value"].min()),
	"max": float(preds["value"].max()),
	},
	"output_path": out_path_rel,
	"duration_s": round(time.perf_counter() - t0, 3),
	}
	_step(f"method {mid}: {len(preds)} team-match predictions "
	f"(mean={preds['value'].mean():.3f}, max={preds['value'].max():.3f})",
	time.perf_counter() - t0)

	# --- Manifest -----------------------------------------------------
	manifest = {
	"run_at": datetime.now(timezone.utc).isoformat(),
	"duration_s": round(time.perf_counter() - t_total, 3),
	"git_sha": _git_sha(),
	"config_path": "configs/benchmark_tasks.yaml",
	"n_train_competitions": len(train_comps),
	"n_matches_total": int(total_matches),
	"split_seed": spec.seed,
	"audit": audit,
	"baselines": baseline_results,
	"methods": method_results,
	}
	manifest_path = out_dir / "run_manifest.json"
	manifest_path.write_text(json.dumps(manifest, indent=2, default=str) + "\n")
	try:
	manifest_pretty = manifest_path.relative_to(REPO)
	except ValueError:
	manifest_pretty = manifest_path
	_step(f"manifest saved → {manifest_pretty}")

	print(f"\n=== done in {manifest['duration_s']:.1f}s ===\n")
	return manifest


	def main() -> None:
	parser = argparse.ArgumentParser(description=__doc__,
	formatter_class=argparse.RawDescriptionHelpFormatter)
	parser.add_argument(
	"--baselines", nargs="*", default=None,
	help="baseline ids to run. Default: all baselines. Empty string = none.",
	)
	parser.add_argument(
	"--methods", nargs="*", default=None,
	help="method ids to run. Default: all methods. Empty string = none.",
	)
	parser.add_argument(
	"--max-matches", type=int, default=None,
	help="Cap total matches across all competitions (for smoke runs).",
	)
	parser.add_argument(
	"--out-dir", type=Path, default=None,
	help="Override output directory.",
	)
	parser.add_argument(
	"--competition-index", type=int, default=None,
	help="If set, only process the competition at this 0-indexed position "
	"in the selected corpus (train or holdout). Used by "
	"scripts/run_benchmark_by_competition.py to spread memory "
	"pressure over sequential runs.",
	)
	parser.add_argument(
	"--holdout", action="store_true",
	help="Run on the T4 holdout competitions (WSL 2020/21 + AFCON 2023) "
	"instead of the training corpus.",
	)
	args = parser.parse_args()
	run(
	baseline_ids=args.baselines,
	method_ids=args.methods,
	max_matches=args.max_matches,
	out_dir=args.out_dir,
	competition_index=args.competition_index,
	holdout=args.holdout,
	)


	if __name__ == "__main__":
	main()