Skip to content

Monitoring a deployed calibration

Data these snippets assume

Blocks on this page use a fixed set of names for held-out data instead of re-deriving it each time. To run them, define the names first:

# docs: no-run — the definitions the snippets on this page assume
import numpy as np
from probcal import BetaCalibrator, Masterscale, make_pd_portfolio
from probcal.monitor import CalibrationMonitor

cal_set, new_set = make_pd_portfolio(n=3000, random_state=0), make_pd_portfolio(n=1000, random_state=1)
s_cal, y_cal = cal_set.scores, cal_set.y          # held-out calibration scores and outcomes
w_cal = np.ones_like(y_cal)                       # uniform sample weights
s_new = new_set.scores                            # scores of new obligors
ms = Masterscale.from_edges([0.01, 0.05], names=["G1", "G2", "G3"])   # the masterscale
p_cal = BetaCalibrator().fit(s_cal, y_cal).predict_proba(s_cal)         # calibrated PDs
grades = ms.assign(p_cal)                                              # rating labels
segments = np.array(["seg-a", "seg-b", "seg-c"])[np.arange(len(s_cal)) % 3]  # segment labels
model = ...   # any object with predict_proba(X); the docs use a stub that reads X[:, 0]
mon = CalibrationMonitor(alpha=0.05)              # with a few batches of a calibrated forecast applied

Every block also names, in its first comment line, which of these it uses.

How-to; the statistics (and why fixed-sample tests are invalid under optional stopping) live in the Monitoring concepts chapter.

Assumptions and scope

The guarantee is: under the null that the deployed forecast is calibrated, the alarm fires with probability at most alpha, at any look and however long monitoring runs. It holds when:

  • Batches arrive in order and are never back-filled. Every plug-in for batch k is computed from batches 1..k-1 only, which is what makes the running product a supermartingale. Persist the monitor state between batches rather than recomputing from raw data.
  • Outcomes are the matured outcomes of the forecast being watched, the calibrated probability, not the raw score, and not a forecast refitted since. After re-calibrating, start a new monitor.
  • The null is conditional calibration, E[y | past, p] = p. A covariate shift that leaves the forecast calibrated does not trip the alarm; a level or slope drift does, whatever caused it.

Outside the guarantee: the recommendation (re-offset versus re-fit) and the onset estimate are diagnostics with no error control, and every component process is reported so they can be second-guessed. Derivations and the simulation evidence are in Monitoring.

# s_cal, y_cal, ms: held-out calibration scores, outcomes, the masterscale
from probcal import BetaCalibrator
from probcal.monitor import CalibrationMonitor

deployed = BetaCalibrator().fit(s_cal, y_cal)   # the map actually in production
mon = CalibrationMonitor(alpha=0.05, grades=ms.names)   # fixed grade universe

# Each time a cohort's outcomes mature (arrival order — never reordered).
# The monitor watches the *calibrated* forecast, never the raw score:
y_batch, p_batch = y_cal[:200], deployed.predict_proba(s_cal[:200])
step = mon.update(y_batch, p_batch, grade=ms, label="2026Q3")   # grades assigned by the scale
step.e_global      # the alarm statistic (alarm when it ever reaches 1/alpha)
step.delta_ci      # anytime-valid CI for the current offset: (-3.0, 0.55) here,
                   # i.e. one 200-row batch buys almost no precision; the
                   # sequence tightens as cohorts accumulate

# Persist between batches; resuming reproduces the trajectory bit-for-bit:
mon.to_json("monitor-state.json")
mon = CalibrationMonitor.from_json("monitor-state.json")

rep = mon.report()
rep.alarm_at        # first crossing label, or None
rep.recommendation  # "none" | "re-offset" | "re-fit" (diagnostic, not a test)
rep.reasoning       # the plain-language trail behind it

from probcal.plots import plot_e_process   # probcal[viz]
plot_e_process(rep)

plot_e_process(rep, grades_panel=True) adds the per-grade confidence sequences below the wealth curves; the monitoring chapter shows the rendered figure on a twelve-cohort drift scenario and reads it line by line.

Passing a Masterscale as grade assigns the labels from the calibrated probabilities and records the scale's fingerprint on first use; a different scale later raises, since it would change the grade universe mid-stream. Declare that universe at construction (grades=ms.names): only declared grades enter the alarm statistic, from the first batch on (a grade with no data yet contributes e = 1), and a label outside the declaration raises. Without grades=, per-grade e-values and confidence sequences are still reported as diagnostics, but they stay out of e_global and the monitor warns once. The reason is anytime validity: a mixture whose components change mid-stream is no longer an e-process (Monitoring).

Three operational rules. Persist the state instead of recomputing from raw data; predictability is what makes the guarantee hold. A portfolio-wide macro shock should trip the alarm: that is the monitor working, not a false positive. After re-calibrating, start a new monitor on the new forecasts. The delta_ci half-width is also the principled buffer_logit for recourse certificates; see the treecf guide.

Re-offsetting inside a pipeline

When the deployed calibrator lives inside an sklearn pipeline (see the scikit-learn guide), acting on Closing the loop's recommendation means swapping the offset step and refitting it alone, leaving the calibrator untouched:

# s_cal, y_cal: held-out calibration scores and outcomes
from sklearn.pipeline import Pipeline

from probcal.sklearn import SklearnCalibrator, SklearnOffset

pipe = Pipeline(
    [("cal", SklearnCalibrator()), ("off", SklearnOffset(target_mean=0.03))]
).fit(s_cal.reshape(-1, 1), y_cal)
fp_before = pipe.named_steps["cal"].calibrator_.fingerprint()

# apply_recommendation() returned a new delta; swap only the offset step:
new_delta = 0.05
pipe.set_params(off=SklearnOffset(delta=new_delta))
p_cal = pipe.named_steps["cal"].transform(s_cal.reshape(-1, 1))
pipe.named_steps["off"].fit(p_cal)
assert pipe.named_steps["cal"].calibrator_.fingerprint() == fp_before

The calibrator's fingerprint survives the swap because nothing there called its fit again. A single-step Chain([BetaCalibrator(), LogitOffset(...)]) inside SklearnCalibrator is the other option when one auditable object, rather than an independently swappable offset, is what governance wants.