pippen.cli

The command-line entry point. Run pippen --help for usage.

pippen.cli

Command-line interface for pippen.

version

version()

Print the installed package version.

Source code in src/pippen/cli.py

@app.command()
def version() -> None:
    """Print the installed package version."""
    console.print(__version__)

paths

paths()

Show where pippen reads and writes data on this machine.

Source code in src/pippen/cli.py

@app.command()
def paths() -> None:
    """Show where pippen reads and writes data on this machine."""
    table = Table(title="pippen data layout")
    table.add_column("Stage", style="bold")
    table.add_column("Path")
    table.add_column("Exists", justify="center")

    table.add_row("root", str(data_root()), "yes" if data_root().is_dir() else "no")
    for stage in ("raw", "interim", "processed", "sources"):
        path = stage_dir(stage)
        table.add_row(stage, str(path), "yes" if path.is_dir() else "no")

    console.print(table)
    console.print(f"\nOverride the root by setting [bold]{ENV_VAR}[/bold].")

fetch

fetch(seasons, dataset='play_by_play', force=False)

Download one hoopR table for a range of seasons.

Files already present and readable are skipped unless --force is given, so re-running after an interruption costs only the seasons that are missing.

Source code in src/pippen/cli.py

@app.command()
def fetch(
    seasons: Annotated[
        str,
        typer.Option(help="A season, 2024, or an inclusive range, 2015-2024."),
    ],
    dataset: Annotated[
        str,
        typer.Option(help="Which hoopR table to download."),
    ] = "play_by_play",
    force: Annotated[
        bool,
        typer.Option(help="Re-download even when a valid file is already present."),
    ] = False,
) -> None:
    """Download one hoopR table for a range of seasons.

    Files already present and readable are skipped unless --force is given, so
    re-running after an interruption costs only the seasons that are missing.
    """
    wanted = _parse_seasons(seasons)
    if dataset not in hoopr.DATASETS:
        known = ", ".join(sorted(hoopr.DATASETS))
        console.print(f"[red]unknown dataset[/red] {dataset!r}. Known datasets: {known}")
        raise typer.Exit(code=2)

    if hoopr.is_master_dataset(dataset):
        # Published as one file covering every season, so the requested range
        # selects nothing. Downloading it once per season would store the same
        # file repeatedly.
        console.print(f"{dataset} is published as one file covering every season")
        results = [hoopr.download_master(dataset, force=force)]
        console.print(f"{results[0].status} {dataset}")
    else:
        results = hoopr.download_seasons(dataset, wanted, force=force, console=console)

    failed = [r for r in results if r.status == "failed"]
    downloaded = sum(1 for r in results if r.status == "downloaded")
    skipped = sum(1 for r in results if r.status == "skipped")
    console.print(
        f"\n{len(results)} season(s): {downloaded} downloaded, "
        f"{skipped} already present, {len(failed)} failed"
    )
    if failed:
        raise typer.Exit(code=1)

rapm

rapm(seasons, window=3, top=20)

Compute regularized adjusted plus-minus over a rolling multi-season window.

Windows are pooled rather than fitted per season because a single season is too few possessions to separate teammates who rarely sit apart. A window of 1 is allowed and is what the sensitivity check uses, not what gets published.

Source code in src/pippen/cli.py

@app.command()
def rapm(
    seasons: Annotated[str, typer.Option(help="Season range, for example 2016-2024.")],
    window: Annotated[int, typer.Option(help="Number of seasons per RAPM window.")] = 3,
    top: Annotated[int, typer.Option(help="How many players to print.")] = 20,
) -> None:
    """Compute regularized adjusted plus-minus over a rolling multi-season window.

    Windows are pooled rather than fitted per season because a single season is
    too few possessions to separate teammates who rarely sit apart. A window of
    1 is allowed and is what the sensitivity check uses, not what gets
    published.
    """
    wanted = _parse_seasons(seasons)
    if window < 1:
        console.print("[red]window must be at least 1[/red]")
        raise typer.Exit(code=2)
    if window > len(wanted):
        console.print(f"[red]window {window} is longer than the {len(wanted)} seasons given[/red]")
        raise typer.Exit(code=2)

    frames = []
    for season in wanted:
        try:
            frames.append(rapm_possessions.read_season_stints(season))
        except FileNotFoundError as exc:
            console.print(f"[red]{exc}[/red]")
            raise typer.Exit(code=2) from exc

    for start in range(0, len(wanted) - window + 1):
        span = wanted[start : start + window]
        pooled = pd.concat(frames[start : start + window], ignore_index=True)
        design = rapm_design.build_design(pooled)
        console.print(f"\n[bold]{_compact(span)}[/bold]  {design.describe()}")
        if design.dropped_rows:
            console.print(f"  dropped {design.dropped_rows:,} stints with malformed lineups")

        fit, table = rapm_ridge.solve(design)
        best = table["weighted_mse"].min()
        console.print(f"  alpha {fit.alpha:,.0f}  intercept {fit.intercept:.2f}  cv mse {best:.1f}")

        ratings = fit.ratings().head(top)
        rendered = Table(title=f"top {top} by impact per 100 possessions, {_compact(span)}")
        rendered.add_column("player_id", justify="right")
        for name in ("offensive", "defensive", "total"):
            rendered.add_column(name, justify="right")
        for row in ratings.itertuples(index=False):
            rendered.add_row(
                str(row.player_id),
                f"{row.offensive:+.2f}",
                f"{row.defensive:+.2f}",
                f"{row.total:+.2f}",
            )
        console.print(rendered)

train

train(seasons, factors=2, top=20, write=True)

Fit the reliability-weighted measurement model for one window.

Seasons are NBA labels, so 2016 means the 2016-17 season. The reliability bounds come from pippen reliability, which must have run first.

Source code in src/pippen/cli.py

@app.command()
def train(
    seasons: Annotated[
        str,
        typer.Option(help="Season range to pool, NBA labels, for example 2016-2018."),
    ],
    factors: Annotated[int, typer.Option(help="Factors in the fusion model.")] = 2,
    top: Annotated[int, typer.Option(help="How many players to print.")] = 20,
    write: Annotated[bool, typer.Option(help="Write ratings to the processed stage.")] = True,
) -> None:
    """Fit the reliability-weighted measurement model for one window.

    Seasons are NBA labels, so 2016 means the 2016-17 season. The reliability
    bounds come from `pippen reliability`, which must have run first.
    """
    wanted = _parse_seasons(seasons)
    try:
        bounds = study.load_measured_reliability()
        dataset, crosswalks = rapm_dataset.build_fusion_dataset(tuple(wanted))
    except FileNotFoundError as exc:
        console.print(f"[red]{exc}[/red]")
        raise typer.Exit(code=2) from exc

    console.print(dataset.describe())
    for crosswalk in crosswalks:
        if len(crosswalk.unmatched) or len(crosswalk.ambiguous):
            console.print(f"  [yellow]{crosswalk.describe()}[/yellow]")

    fit = fusion.fit_fusion(dataset, bounds, n_factors=factors)
    console.print(fit.describe())
    if not fit.converged:
        console.print("[yellow]the published quantities did not converge[/yellow]")

    loadings = Table(title="loadings, how much of each metric is the latent quantity")
    for column in ("metric", "loading", "bound", "share of bound"):
        loadings.add_column(column, justify="right" if column != "metric" else "left")
    for row in fit.loadings.itertuples(index=False):
        loadings.add_row(
            str(row.metric), f"{row.loading:+.3f}", f"{row.bound:.3f}", f"{row.share_of_bound:.2f}"
        )
    console.print(loadings)

    ratings = Table(title=f"top {top} by fused impact")
    for column in ("nba_id", "impact", "sd", "lower", "upper"):
        ratings.add_column(column, justify="right")
    for row in fit.ratings.head(top).itertuples(index=False):
        ratings.add_row(
            str(row.nba_id),
            f"{row.impact:+.2f}",
            f"{row.sd:.2f}",
            f"{row.lower:+.2f}",
            f"{row.upper:+.2f}",
        )
    console.print(ratings)
    console.print(
        "Intervals are a floor on the true uncertainty, not a faithful estimate. "
        "See docs/methodology/calibration.md."
    )

    if write:
        target = stage_dir("processed", create=True) / f"ratings_{_compact(wanted)}.parquet"
        fit.ratings.to_parquet(target, index=False)
        console.print(f"wrote {target}")

evaluate

evaluate(seasons, window=3, trials=40, factors=2)

Compare PIPPEN against every input metric on next-season team net rating.

This is the claim the project committed to answering before the model was built. It runs every candidate on its own, then two supervised baselines over all of them at once, then the same baselines with RAPM and the fused rating removed, then SHAP attributions, then paired tests. The answer is printed whichever way it comes out.

Slow: one fusion fit per window plus nested hyperparameter selection.

Source code in src/pippen/cli.py

@app.command()
def evaluate(
    seasons: Annotated[
        str,
        typer.Option(help="Seasons available, NBA labels, for example 2016-2024."),
    ],
    window: Annotated[int, typer.Option(help="Seasons pooled into one rating.")] = 3,
    trials: Annotated[int, typer.Option(help="Optuna trials per search, per baseline.")] = 40,
    factors: Annotated[int, typer.Option(help="Factors in the fusion model.")] = 2,
) -> None:
    """Compare PIPPEN against every input metric on next-season team net rating.

    This is the claim the project committed to answering before the model was
    built. It runs every candidate on its own, then two supervised baselines
    over all of them at once, then the same baselines with RAPM and the fused
    rating removed, then SHAP attributions, then paired tests. The answer is
    printed whichever way it comes out.

    Slow: one fusion fit per window plus nested hyperparameter selection.
    """
    wanted = _parse_seasons(seasons)
    try:
        panel, _fits = study.build_team_panel(
            wanted,
            window=window,
            n_factors=factors,
            on_progress=lambda record: console.print(
                f"  {_compact(list(record.nba_seasons))} -> {record.target_season}: "
                f"r-hat {record.fit.diagnostics['max_r_hat']:.4f}"
            ),
        )
    except (FileNotFoundError, ValueError) as exc:
        console.print(f"[red]{exc}[/red]")
        raise typer.Exit(code=2) from exc

    console.print(f"\n{panel.describe()}")
    report = study.evaluate_claim(panel, trials=trials)

    single = Table(title="every candidate on its own, leave-one-season-out")
    for column in ("candidate", "rmse", "vs best"):
        single.add_column(column, justify="right" if column != "candidate" else "left")
    for row in report.single.table.itertuples(index=False):
        single.add_row(str(row.candidate), f"{row.rmse:.3f}", f"{row.vs_best:+.3f}")
    console.print(single)

    for title, results in (
        ("baselines over every candidate", report.baselines),
        ("baselines over the box score alone", report.box_only_baselines),
    ):
        rendered = Table(title=title)
        rendered.add_column("model")
        rendered.add_column("rmse", justify="right")
        for result in results:
            rendered.add_row(result.name, f"{result.rmse:.3f}")
        console.print(rendered)

    if report.attribution is not None:
        attribution = Table(title=f"SHAP importance, {report.attributed_to}")
        attribution.add_column("metric")
        attribution.add_column("mean |shap|", justify="right")
        for row in report.attribution.importance.head(10).itertuples(index=False):
            attribution.add_row(str(row.metric), f"{row.mean_abs_shap:.3f}")
        console.print(attribution)

    console.print(f"\n[bold]{report.single.describe()}[/bold]")
    for comparison in report.comparisons:
        console.print(f"  {comparison.describe()}")
    if report.box_only_comparison is not None:
        console.print(f"  box score combined: {report.box_only_comparison.describe()}")