"""The streaming evaluation protocol.
Simulates a lender scoring an arriving stream of applications whose outcomes
are revealed only after a maturation delay. See :mod:`pamir.evaluate_iid` for
the conventional train/test split, included for comparison with other tabular
benchmarks.
Both protocols call a user-supplied ``predict_fn(X_train, y_train, X_test)``
and account for every call that raises — see :mod:`pamir.failures`.
"""
import warnings
from typing import Callable, Dict, Optional, Sequence
import numpy as np
import pandas as pd
from sklearn.metrics import roc_auc_score
from pamir.catalog import list_datasets
from pamir.failures import FailureLog
from pamir.loader import load_dataset
from pamir.summary import fleet_summary, format_fleet_summary
# Columns of the per-dataset frame returned by `evaluate`. `refit_points` is
# deliberately absent: it is a list per row, which blocks `to_csv` and every
# other tabular export. Read it from `evaluate_one` when you want the curve.
RESULT_COLUMNS = ["dataset", "n_rows", "n_defaults", "DR", "n_refits",
"n_calls", "n_failures", "auc_final"]
# Minimum resolved rows and defaults before a cumulative AUC is meaningful.
_MIN_SCORED_ROWS = 10
_MIN_SCORED_DEFAULTS = 3
[docs]
def evaluate_one(
predict_fn: Callable,
dataset_id: str,
lag: int = 1000,
k_refit: int = 10,
max_n: Optional[int] = 20000,
min_defaults: int = 3,
on_error: str = "warn",
) -> Dict:
"""Run the streaming protocol on one dataset.
Parameters
----------
predict_fn : callable
``predict_fn(X_train, y_train, X_test) -> np.ndarray`` of scores
(higher = more likely to default). Called once per refit point.
``X_train`` contains only rows whose outcome is resolved; ``X_test`` is
the remaining unscored stream, and the returned array must have one
score per row of it.
dataset_id : str
PaMIR dataset id.
lag : int
Label-maturation delay. Outcome of application at position *t* is
revealed only at position *t + lag*.
k_refit : int
Refit every *k_refit* newly resolved defaults. This changes the score
as well as the cost — see the protocol documentation — so report it.
max_n : int or None
Truncate the stream to at most this many rows.
min_defaults : int
Minimum resolved defaults before the first refit.
on_error : {"warn", "raise", "ignore"}
What to do when ``predict_fn`` raises. See :mod:`pamir.failures`.
Returns
-------
dict with keys: dataset, n_rows, n_defaults, DR, n_refits, n_calls,
n_failures, errors, auc_final, refit_points (list of dicts with pos,
resolved_n, cum_defaults, cum_auc).
"""
X, y, _meta = load_dataset(dataset_id, max_rows=max_n)
n = len(y)
preds = np.full(n, np.nan, dtype=np.float64)
log = FailureLog(on_error, context=dataset_id)
cum_defaults = 0
defaults_since_refit = 0
refit_points = []
def score_and_commit(X_ctx, y_ctx, X_rem, start):
"""Call the model and commit its scores, validating the contract."""
scores = np.asarray(predict_fn(X_ctx, y_ctx, X_rem), dtype=np.float64)
if scores.shape != (len(X_rem),):
raise ValueError(
f"predict_fn returned scores of shape {scores.shape} for "
f"{len(X_rem)} rows; expected ({len(X_rem)},)."
)
preds[start:] = scores
for t in range(n):
resolved_end = max(0, t + 1 - lag)
if t >= lag and y[t - lag] == 1:
cum_defaults += 1
defaults_since_refit += 1
need_refit = (cum_defaults >= min_defaults
and defaults_since_refit >= k_refit)
if cum_defaults == min_defaults and not refit_points:
need_refit = True
if not (need_refit and resolved_end > 0 and t + 1 < n):
continue
if resolved_end - cum_defaults < min_defaults:
continue
log.call(score_and_commit, X.iloc[:resolved_end], y[:resolved_end],
X.iloc[resolved_end:], resolved_end)
defaults_since_refit = 0
refit_points.append(_refit_point(t, resolved_end, cum_defaults,
y, preds))
log.warn_if_failed()
return {
"dataset": dataset_id,
"n_rows": n,
"n_defaults": int(y.sum()),
"DR": float(y.mean()),
"n_refits": len(refit_points),
**log.as_dict(),
"auc_final": _safe_auc(y, preds),
"refit_points": refit_points,
}
def _refit_point(t: int, resolved_end: int, cum_defaults: int,
y: np.ndarray, preds: np.ndarray) -> Dict:
"""Record cumulative AUC over everything resolved so far."""
point = {"pos": t + 1, "resolved_n": resolved_end,
"cum_defaults": cum_defaults}
auc = _safe_auc(y[:resolved_end], preds[:resolved_end])
if auc is not None:
point["cum_auc"] = auc
return point
def _safe_auc(y: np.ndarray, preds: np.ndarray) -> Optional[float]:
"""ROC AUC over scored rows, or None when too few to be meaningful."""
valid = ~np.isnan(preds)
if valid.sum() <= _MIN_SCORED_ROWS or y[valid].sum() < _MIN_SCORED_DEFAULTS:
return None
return float(roc_auc_score(y[valid], preds[valid]))
[docs]
def evaluate(
predict_fn: Callable,
datasets: Optional[Sequence[str]] = None,
lag: int = 1000,
k_refit: int = 10,
max_n: Optional[int] = 20000,
min_defaults: int = 3,
verbose: bool = True,
on_error: str = "warn",
) -> pd.DataFrame:
"""Run the streaming protocol across the full PaMIR fleet.
Parameters
----------
predict_fn : callable
Same signature as in :func:`evaluate_one`.
datasets : list of str, optional
Subset of dataset ids. Default: all 19.
lag, k_refit, max_n, min_defaults : int
Protocol parameters (see :func:`evaluate_one`).
verbose : bool
Print per-dataset progress and the fleet summary.
on_error : {"warn", "raise", "ignore"}
What to do when ``predict_fn`` raises.
Returns
-------
DataFrame with one row per dataset and the columns in ``RESULT_COLUMNS``.
Pass it to :func:`pamir.fleet_summary` for the headline numbers — a mean
taken by hand over this frame silently excludes the datasets the model
failed on.
"""
ds_ids = list(datasets) if datasets else list_datasets()
rows = []
for ds in ds_ids:
if verbose:
print(f" {ds}...", end=" ", flush=True)
result = evaluate_one(predict_fn, ds, lag=lag, k_refit=k_refit,
max_n=max_n, min_defaults=min_defaults,
on_error=on_error)
if verbose:
print(_format_dataset_line(result))
rows.append(result)
df = pd.DataFrame(rows)[RESULT_COLUMNS]
summary = fleet_summary(df)
if verbose and len(rows) > 1:
print(format_fleet_summary(summary))
if not summary["complete"] and on_error == "warn":
warnings.warn(
f"the model did not score {summary['n_datasets'] - summary['n_scored']} "
f"of {summary['n_datasets']} datasets "
f"({', '.join(summary['failed_datasets'])}). There is no fleet mean "
"for a partial run — see pamir.fleet_summary.",
UserWarning,
stacklevel=2,
)
return df
def _format_dataset_line(result: Dict) -> str:
auc = result["auc_final"]
head = f"AUC={auc:.4f}" if auc is not None else "NO AUC"
tail = f"({result['n_refits']} refits"
if result["n_failures"]:
tail += f", {result['n_failures']}/{result['n_calls']} calls FAILED"
return f"{head} {tail})"