Source code for pamir.summary

"""Fleet-level summary of a results frame, with coverage as a first-class field.

The headline number of a benchmark run is only defined when the model scored
every dataset it was given.  ``fleet_summary`` therefore reports ``auc_mean``
only for a complete run, and puts the partial average in a separately named
field so it cannot be mistaken for one.
"""

from typing import Dict, List, Optional

import pandas as pd

# The AUC column is named by the protocol that produced the frame.
AUC_COLUMNS = ("auc_final", "auc_mean")


def _auc_column(results: pd.DataFrame) -> str:
    for col in AUC_COLUMNS:
        if col in results.columns:
            return col
    raise ValueError(
        f"no auc column found: expected one of {AUC_COLUMNS}, "
        f"got {list(results.columns)}. Pass a frame returned by "
        "pamir.evaluate or pamir.evaluate_iid."
    )




def fleet_summary(results: pd.DataFrame) -> Dict[str, Optional[float]]:
    """Summarize a fleet run, withholding the mean when coverage is incomplete.

    Parameters
    ----------
    results : DataFrame
        As returned by :func:`pamir.evaluate` or :func:`pamir.evaluate_iid`.

    Returns
    -------
    dict with keys:

    ``n_datasets``, ``n_scored``, ``coverage``, ``complete``
        How much of the fleet the model actually scored.
    ``auc_mean``, ``gini_mean``
        The headline numbers — ``None`` unless every dataset was scored.
    ``auc_mean_scored_only``
        The average over the datasets that did score.  Diagnostic only: it is
        an average over a subset the model chose by failing, so it is not
        comparable across models.
    ``n_failures``, ``failed_datasets``
        What went wrong and where.
    """
    auc_col = _auc_column(results)
    scored = results[auc_col].notna()

    n_datasets = len(results)
    n_scored = int(scored.sum())
    complete = n_datasets > 0 and n_scored == n_datasets
    partial_mean = float(results.loc[scored, auc_col].mean()) if n_scored else None

    failed: List[str] = (
        results.loc[~scored, "dataset"].tolist()
        if "dataset" in results.columns else []
    )

    return {
        "n_datasets": n_datasets,
        "n_scored": n_scored,
        "coverage": n_scored / n_datasets if n_datasets else 0.0,
        "complete": complete,
        "auc_mean": partial_mean if complete else None,
        "gini_mean": 2 * partial_mean - 1 if complete and partial_mean else None,
        "auc_mean_scored_only": partial_mean,
        "n_failures": (int(results["n_failures"].sum())
                       if "n_failures" in results.columns else None),
        "failed_datasets": failed,
    }



def format_fleet_summary(summary: Dict) -> str:
    """One or two lines of human-readable summary, for ``verbose=True``."""
    if summary["complete"]:
        return (f"\n  Fleet mean AUC: {summary['auc_mean']:.4f}  "
                f"Gini: {summary['gini_mean']:.3f}  "
                f"({summary['n_datasets']} datasets, all scored)")

    failed = ", ".join(summary["failed_datasets"]) or "—"
    partial = summary["auc_mean_scored_only"]
    partial_text = f"{partial:.4f}" if partial is not None else "n/a"
    return (
        f"\n  NO FLEET MEAN: {summary['n_scored']} of {summary['n_datasets']} "
        f"datasets scored (failed: {failed})."
        f"\n  Mean over the scored subset only: {partial_text} — not comparable "
        "with a model that scored the whole fleet."
    )