API: tools

offset

Logit-offset (central tendency) adjustment with audit trail.

Theory — including the King–Zeng / Elkan / Tasche equivalences and the uniqueness of the mode-B root: docs/concepts/offset.md.

References

King & Zeng (2001); Elkan (2001); Tasche (2013) — full records in the documentation.

AuditReport dataclass

AuditReport(delta: float, pre_mean: float, post_mean: float, timestamp: str, guardrails_before: GuardrailReport, guardrails_after: GuardrailReport)

Pre/post record of a logit-offset application, for validators.

ATTRIBUTE	DESCRIPTION
delta	Applied log-odds shift. TYPE: float
pre_mean, post_mean	Portfolio mean probability before and after the shift. TYPE: float
timestamp	ISO-8601 UTC time at which the offset was fitted. TYPE: str
guardrails_before, guardrails_after	The three-flag calibration health summary on the input and output probabilities. TYPE: GuardrailReport

LogitOffset

LogitOffset(delta: float | None = None, target_mean: float | None = None)

Bases: JsonIO

Uniform log-odds shift: p' = sigma(logit(p) + delta).

Mode A takes delta explicitly; mode B takes target_mean and solves mean(p') = target_mean for delta by bisection — the portfolio mean is strictly increasing in delta, so the root is unique (stated in the offset chapter and unit-tested). Exactly one of the two arguments must be given.

The offset is deliberately not folded into any calibrator's parameters: CalibratedModel.offset_to appends it as a separate, inspectable pipeline stage.

PARAMETER	DESCRIPTION
delta	Mode A: the log-odds shift to apply directly. Mutually exclusive with target_mean — :meth:fit requires exactly one of the two. TYPE: float or None DEFAULT: None
target_mean	Mode B: the desired post-shift portfolio mean probability in (0, 1); delta is solved by bisection. Mutually exclusive with delta. TYPE: float or None DEFAULT: None

ATTRIBUTE	DESCRIPTION
delta_	Fitted (or given) shift in log-odds. TYPE: float
pre_mean_, post_mean_	Portfolio mean before and after, recorded at fit time. TYPE: float
timestamp_	ISO-8601 UTC fit time — part of the audit trail. TYPE: str

Source code in src/probcal/offset.py

def __init__(self, delta: float | None = None, target_mean: float | None = None) -> None:
    self.delta = delta
    self.target_mean = target_mean

affine_logit_coeffs_ property

affine_logit_coeffs_: tuple[float, float]

(1, delta): the offset is affine on the logit scale.

fit

fit(p: object, *, y: object = None, sample_weight: object = None) -> Self

Fix delta (mode A) or solve it against the target mean (mode B).

PARAMETER	DESCRIPTION
p	Current calibrated probabilities of the portfolio. TYPE: array_like
y	Ignored; accepted for compatibility with the chain fit protocol. TYPE: (array_like or None, keyword - only) DEFAULT: None
sample_weight	Weights for the portfolio mean. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Self	The fitted offset.

RAISES	DESCRIPTION
ValueError	If not exactly one of delta/target_mean is set, or target_mean is outside (0, 1).
UnattainableTargetError	If target_mean cannot be reached by any shift in [-40, 40]; the message names the attainable range.

Source code in src/probcal/offset.py

def fit(self, p: object, *, y: object = None, sample_weight: object = None) -> Self:
    """Fix ``delta`` (mode A) or solve it against the target mean (mode B).

    Parameters
    ----------
    p : array_like
        Current calibrated probabilities of the portfolio.
    y : array_like or None, keyword-only
        Ignored; accepted for compatibility with the chain fit protocol.
    sample_weight : array_like or None, keyword-only
        Weights for the portfolio mean.

    Returns
    -------
    Self
        The fitted offset.

    Raises
    ------
    ValueError
        If not exactly one of ``delta``/``target_mean`` is set, or
        ``target_mean`` is outside ``(0, 1)``.
    UnattainableTargetError
        If ``target_mean`` cannot be reached by any shift in
        ``[-40, 40]``; the message names the attainable range.
    """
    if (self.delta is None) == (self.target_mean is None):
        raise ValueError("LogitOffset: give exactly one of delta or target_mean")
    p_arr = validate_scores(p, name="p")
    w = validate_weights(sample_weight, len(p_arr))
    z = logit(p_arr)
    self.pre_mean_ = float(np.average(p_arr, weights=w))
    if self.delta is not None:
        self.delta_ = float(self.delta)
    else:
        target = float(self.target_mean)  # type: ignore[arg-type]
        if not 0.0 < target < 1.0:
            raise ValueError("target_mean must lie in (0, 1)")
        self.delta_ = _solve_delta(z, w, target)
    self.post_mean_ = float(np.average(expit(z + self.delta_), weights=w))
    self.timestamp_ = datetime.now(UTC).isoformat(timespec="seconds")
    self.fit_meta_ = {
        "n_obs": int(p_arr.shape[0]),
        "weight_sum": float(w.sum()),
        "fitted_at_utc": self.timestamp_,
        "data_fingerprint": data_fingerprint(p_arr, w),
    }
    self.fitted_ = True
    return self

transform

transform(p: object) -> ndarray

Apply the fitted shift to probabilities.

Source code in src/probcal/offset.py

def transform(self, p: object) -> np.ndarray:
    """Apply the fitted shift to probabilities."""
    self._check_fitted()
    return expit(logit(validate_scores(p, name="p")) + self.delta_)

__sklearn_is_fitted__

__sklearn_is_fitted__() -> bool

Fitted state for sklearn >= 1.6 (delta_ fixed or solved).

Source code in src/probcal/offset.py

def __sklearn_is_fitted__(self) -> bool:
    """Fitted state for sklearn >= 1.6 (``delta_`` fixed or solved)."""
    return bool(self.fitted_)

interpret

interpret() -> Interpretation

Read delta in log-odds, odds-factor, and central-tendency terms.

Source code in src/probcal/offset.py

def interpret(self) -> Interpretation:
    """Read delta in log-odds, odds-factor, and central-tendency terms."""
    self._check_fitted()
    return Interpretation(
        method=type(self).__name__,
        param_names=("delta",),
        param_values=(self.delta_,),
        messages=(
            f"delta = {self.delta_:+.4f} log-odds: every observation's odds are "
            f"multiplied by exp(delta) = {np.exp(self.delta_):.4f} uniformly",
            f"portfolio mean re-anchored from {self.pre_mean_:.5f} to "
            f"{self.post_mean_:.5f} (credit-risk central tendency adjustment)",
            "equivalent to King-Zeng prior correction and to Elkan's base-rate "
            "adjustment (see the offset chapter for the derivations)",
            "ranking is untouched: the shift is strictly increasing",
        ),
    )

interval_inverse

interval_inverse(lo: float, hi: float, *, space: str = 'probability', buffer_logit: float = 0.0) -> tuple[float, float]

Closed-form preimage: subtract delta on the logit scale.

Same protocol as BaseCalibrator.interval_inverse; the offset's output range is the full unit interval, so only a crossed buffer can make a target unattainable.

PARAMETER	DESCRIPTION
lo	Calibrated-probability bounds; lo=0/hi=1 map to the full raw range (−inf/+inf on the logit scale). TYPE: float
hi	Calibrated-probability bounds; lo=0/hi=1 map to the full raw range (−inf/+inf on the logit scale). TYPE: float
space	Scale of the returned raw bounds. TYPE: (probability, logit) DEFAULT: "probability"
buffer_logit	Shrink the calibrated interval by this margin in logit space before inverting (must be finite and >= 0). TYPE: (float, keyword - only) DEFAULT: 0.0

RETURNS	DESCRIPTION
tuple of float	(raw_lo, raw_hi) preimage bounds, on the scale requested by space.

RAISES	DESCRIPTION
UnattainableTargetError	If a crossed buffer_logit empties the calibrated interval, or the (buffered) interval does not intersect the offset map's representable output range [sigma(delta - logit(1 - 1e-12)), sigma(delta + logit(1 - 1e-12))] — bounds beyond that range collapse to the full-range sentinels (0/1, ±inf) instead of raw values below the clip that transform could not round-trip.
ValueError	If lo, hi are not ordered in [0, 1], space is unknown, or buffer_logit is negative.

Source code in src/probcal/offset.py

def interval_inverse(
    self,
    lo: float,
    hi: float,
    *,
    space: str = "probability",
    buffer_logit: float = 0.0,
) -> tuple[float, float]:
    """Closed-form preimage: subtract delta on the logit scale.

    Same protocol as ``BaseCalibrator.interval_inverse``; the offset's
    output range is the full unit interval, so only a crossed buffer can
    make a target unattainable.

    Parameters
    ----------
    lo, hi : float
        Calibrated-probability bounds; ``lo=0``/``hi=1`` map to the full
        raw range (−inf/+inf on the logit scale).
    space : {"probability", "logit"}, keyword-only
        Scale of the returned raw bounds.
    buffer_logit : float, keyword-only
        Shrink the calibrated interval by this margin in logit space
        before inverting (must be finite and ``>= 0``).

    Returns
    -------
    tuple of float
        ``(raw_lo, raw_hi)`` preimage bounds, on the scale requested by
        ``space``.

    Raises
    ------
    UnattainableTargetError
        If a crossed ``buffer_logit`` empties the calibrated interval,
        or the (buffered) interval does not intersect the offset map's
        representable output range ``[sigma(delta - logit(1 - 1e-12)),
        sigma(delta + logit(1 - 1e-12))]`` — bounds beyond that range
        collapse to the full-range sentinels (0/1, ±inf) instead of raw
        values below the clip that ``transform`` could not round-trip.
    ValueError
        If ``lo``, ``hi`` are not ordered in ``[0, 1]``, ``space`` is
        unknown, or ``buffer_logit`` is negative.
    """
    self._check_fitted()
    validate_interval(lo, hi)
    validate_space(space)
    lo_b, hi_b = shrink_interval(lo, hi, buffer_logit)
    # Representable output range: raw scores are clipped to
    # [1e-12, 1 - 1e-12] by every forward entry point, so the shifted map
    # attains only [sigma(delta - _LOGIT_CLIP), sigma(delta + _LOGIT_CLIP)].
    # Bounds beyond it collapse to the full-range sentinels (0.0/1.0,
    # -inf/+inf) exactly as in BaseCalibrator.interval_inverse; a raw
    # bound below the clip (e.g. 4.5e-14) could not round-trip through
    # transform — the silent break the no-silent-clamp doctrine forbids.
    gmin = expit1(self.delta_ - _LOGIT_CLIP)
    gmax = expit1(self.delta_ + _LOGIT_CLIP)
    if lo_b > gmax or hi_b < gmin:
        raise UnattainableTargetError(
            f"calibrated target [{lo_b:.6g}, {hi_b:.6g}] does not intersect the "
            f"offset map's representable output range [{gmin:.6g}, {gmax:.6g}]"
        )
    lo_z = -np.inf if lo_b <= gmin else logit1(lo_b) - self.delta_
    hi_z = np.inf if hi_b >= gmax else logit1(hi_b) - self.delta_
    if space == "logit":
        return lo_z, hi_z
    raw_lo = 0.0 if np.isneginf(lo_z) else expit1(lo_z)
    raw_hi = 1.0 if np.isposinf(hi_z) else expit1(hi_z)
    return raw_lo, raw_hi

point_inverse

point_inverse(p: object, *, space: str = 'probability') -> ndarray

Raw scores whose shifted probabilities equal p (exact preimage).

Closed form: subtract delta on the logit scale. Same protocol and boundary doctrine as :meth:BaseCalibrator.point_inverse (shared validation helpers).

PARAMETER	DESCRIPTION
p	Shifted probabilities strictly inside (0, 1); boundary and out-of-range values raise UnattainableTargetError (all-or-nothing, no silent clamp). TYPE: array_like
space	Scale of the returned raw values. TYPE: (probability, logit) DEFAULT: "probability"

RETURNS	DESCRIPTION
ndarray	Raw scores (or logits, if space="logit") whose shifted probability equals p.

RAISES	DESCRIPTION
RuntimeError	If not yet fitted.
ValueError	If space is not "probability" or "logit".
UnattainableTargetError	If any element of p lies outside the open interval (0, 1); or if space="probability" and the raw logit of any result exceeds logit(1 - 1e-12) in magnitude — the probability representation would round to 0.0/1.0 and silently fail to round-trip; space="logit" is exact there.

Source code in src/probcal/offset.py

def point_inverse(self, p: object, *, space: str = "probability") -> np.ndarray:
    """Raw scores whose shifted probabilities equal ``p`` (exact preimage).

    Closed form: subtract ``delta`` on the logit scale. Same protocol
    and boundary doctrine as :meth:`BaseCalibrator.point_inverse`
    (shared validation helpers).

    Parameters
    ----------
    p : array_like
        Shifted probabilities strictly inside ``(0, 1)``; boundary and
        out-of-range values raise ``UnattainableTargetError``
        (all-or-nothing, no silent clamp).
    space : {"probability", "logit"}, keyword-only
        Scale of the returned raw values.

    Returns
    -------
    numpy.ndarray
        Raw scores (or logits, if ``space="logit"``) whose shifted
        probability equals ``p``.

    Raises
    ------
    RuntimeError
        If not yet fitted.
    ValueError
        If ``space`` is not ``"probability"`` or ``"logit"``.
    UnattainableTargetError
        If any element of ``p`` lies outside the open interval
        ``(0, 1)``; or if ``space="probability"`` and the raw logit of
        any result exceeds ``logit(1 - 1e-12)`` in magnitude — the
        probability representation would round to 0.0/1.0 and silently
        fail to round-trip; ``space="logit"`` is exact there.
    """
    self._check_fitted()
    validate_space(space)
    arr = _validate_point_targets(p)
    z = logit(arr) - self.delta_
    _check_representable(z, space)
    return z if space == "logit" else expit(z)

audit_report

audit_report(y: object, p: object, *, sample_weight: object = None) -> AuditReport

Pre/post guardrail comparison for the validator's one-table view.

Source code in src/probcal/offset.py

def audit_report(self, y: object, p: object, *, sample_weight: object = None) -> AuditReport:
    """Pre/post guardrail comparison for the validator's one-table view."""
    self._check_fitted()
    before = calibration_guardrails(y, p, sample_weight=sample_weight)
    after = calibration_guardrails(y, self.transform(p), sample_weight=sample_weight)
    return AuditReport(
        delta=self.delta_,
        pre_mean=self.pre_mean_,
        post_mean=self.post_mean_,
        timestamp=self.timestamp_,
        guardrails_before=before,
        guardrails_after=after,
    )

to_dict

to_dict() -> dict[str, object]

Versioned JSON-native snapshot (see BaseCalibrator.to_dict).

fit_meta records n_obs, weight_sum, fitted_at_utc, and the data_fingerprint of the (p, w) pair — no n_events because the offset is fitted on probabilities alone. fingerprint is blind to the audit timestamp_: identical fits fingerprint identically.

Source code in src/probcal/offset.py

def to_dict(self) -> dict[str, object]:
    """Versioned JSON-native snapshot (see ``BaseCalibrator.to_dict``).

    ``fit_meta`` records ``n_obs``, ``weight_sum``, ``fitted_at_utc``,
    and the ``data_fingerprint`` of the ``(p, w)`` pair — no ``n_events``
    because the offset is fitted on probabilities alone. ``fingerprint``
    is blind to the audit ``timestamp_``: identical fits fingerprint
    identically.
    """
    self._check_fitted()
    return envelope(
        self,
        params=encode_value({"delta": self.delta, "target_mean": self.target_mean}),
        state=encode_value(
            {
                "delta_": self.delta_,
                "pre_mean_": self.pre_mean_,
                "post_mean_": self.post_mean_,
                "timestamp_": self.timestamp_,
            }
        ),
        fit_meta=encode_value(self.fit_meta_),
    )

from_dict classmethod

from_dict(d: dict) -> LogitOffset

Rebuild a fitted offset from :meth:to_dict output.

RAISES	DESCRIPTION
ValueError	If the schema version is unknown or the payload class differs.

Source code in src/probcal/offset.py

@classmethod
def from_dict(cls, d: dict) -> "LogitOffset":
    """Rebuild a fitted offset from :meth:`to_dict` output.

    Raises
    ------
    ValueError
        If the schema version is unknown or the payload class differs.
    """
    check_payload(cls, d)
    obj = cls(**decode_value(d["params"], arrays=False))  # type: ignore[arg-type]
    for key, value in d["state"].items():
        setattr(obj, key, decode_value(value, arrays=False))
    obj.fit_meta_ = dict(decode_value(d["fit_meta"], arrays=False))  # type: ignore[call-overload]
    obj.fitted_ = True
    return obj

estimate_offset

estimate_offset(y: object, p: object, *, sample_weight: object = None) -> OffsetEstimate

Offset-only logistic MLE of delta given p, with its standard error.

Fits the single-parameter model y ~ Bernoulli(sigma(logit(p) + delta)) by maximum likelihood. The score equation sum(w * (y - sigma(logit(p) + delta))) = 0 is exactly the mean-matching condition solved by LogitOffset(target_mean=mean_w(y)), so delta is found by the same bisection root-finder (:func:_offset_mle, shared with probcal.monitor._processes.plug_in_delta). Weights are relative frequency weights (:mod:probcal.metrics._common): delta and se do not change when every weight is multiplied by a constant. With q = sigma(logit(p) + delta), I = sum(w * q * (1 - q)) and J = sum(w**2 * q * (1 - q)) (the exact variance of the weighted score when each row is one Bernoulli draw), the standard error is sqrt(J) / I; for unit weights that is the Fisher 1 / sqrt(I).

PARAMETER	DESCRIPTION
y	Binary outcomes in {0, 1}. Both classes must be present. TYPE: array_like
p	Predicted probabilities in [0, 1]. TYPE: array_like
sample_weight	Optional non-negative weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
OffsetEstimate	The fitted delta, its standard error, and the fit's n, events, and weight_sum.

RAISES	DESCRIPTION
ValueError	If y or p fail validation (shape, range, finiteness) or if y contains only one class — the offset MLE does not exist then.

Examples:

>>> import numpy as np
>>> from probcal.offset import estimate_offset
>>> from probcal._math import expit
>>> rng = np.random.default_rng(0)
>>> z = rng.normal(0.0, 1.0, 2000)
>>> p = expit(z)
>>> y = (rng.random(2000) < expit(z + 0.5)).astype(float)
>>> est = estimate_offset(y, p)
>>> est.n
2000
>>> abs(est.delta - 0.5) < 3 * est.se
True

Source code in src/probcal/offset.py

def estimate_offset(y: object, p: object, *, sample_weight: object = None) -> OffsetEstimate:
    """Offset-only logistic MLE of ``delta`` given ``p``, with its standard error.

    Fits the single-parameter model ``y ~ Bernoulli(sigma(logit(p) + delta))``
    by maximum likelihood. The score equation
    ``sum(w * (y - sigma(logit(p) + delta))) = 0`` is exactly the mean-matching
    condition solved by ``LogitOffset(target_mean=mean_w(y))``, so ``delta`` is
    found by the same bisection root-finder (:func:`_offset_mle`, shared with
    ``probcal.monitor._processes.plug_in_delta``). Weights are relative
    frequency weights (:mod:`probcal.metrics._common`): ``delta`` and ``se``
    do not change when every weight is multiplied by a constant. With
    ``q = sigma(logit(p) + delta)``, ``I = sum(w * q * (1 - q))`` and
    ``J = sum(w**2 * q * (1 - q))`` (the exact variance of the weighted score
    when each row is one Bernoulli draw), the standard error is
    ``sqrt(J) / I``; for unit weights that is the Fisher ``1 / sqrt(I)``.

    Parameters
    ----------
    y : array_like
        Binary outcomes in ``{0, 1}``. Both classes must be present.
    p : array_like
        Predicted probabilities in ``[0, 1]``.
    sample_weight : array_like or None, keyword-only
        Optional non-negative weights, same length as ``y``.

    Returns
    -------
    OffsetEstimate
        The fitted ``delta``, its standard error, and the fit's ``n``,
        ``events``, and ``weight_sum``.

    Raises
    ------
    ValueError
        If ``y`` or ``p`` fail validation (shape, range, finiteness) or if
        ``y`` contains only one class — the offset MLE does not exist then.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.offset import estimate_offset
    >>> from probcal._math import expit
    >>> rng = np.random.default_rng(0)
    >>> z = rng.normal(0.0, 1.0, 2000)
    >>> p = expit(z)
    >>> y = (rng.random(2000) < expit(z + 0.5)).astype(float)
    >>> est = estimate_offset(y, p)
    >>> est.n
    2000
    >>> abs(est.delta - 0.5) < 3 * est.se
    True
    """
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    z = logit(p_arr)
    delta = _offset_mle(z, y_arr, w)
    q = expit(z + delta)
    v = w * q * (1.0 - q)
    info = float(np.sum(v))
    # sqrt(J / I) / sqrt(I) == sqrt(J) / I, written so that unit weights
    # (w * v == v, J / I == 1.0 exactly) give the Fisher 1 / sqrt(I) bit for bit.
    se = math.sqrt(float(np.sum(w * v)) / info) / math.sqrt(info)
    return OffsetEstimate(
        delta=delta,
        se=se,
        n=int(len(y_arr)),
        events=float(np.sum(w * y_arr)),
        weight_sum=float(w.sum()),
    )

offset_from_estimate

offset_from_estimate(est: OffsetEstimate, p: object) -> LogitOffset

Build a fitted :class:LogitOffset (mode A) from an :class:OffsetEstimate.

Equivalent to LogitOffset(delta=est.delta).fit(p) — a convenience for turning the audited MLE into the same offset object used elsewhere in the package (transform, interpret, to_dict, ...).

PARAMETER	DESCRIPTION
est	Result of :func:estimate_offset. TYPE: OffsetEstimate
p	Probabilities to fit the offset's audit trail (pre_mean_, post_mean_) against. TYPE: array_like

RETURNS	DESCRIPTION
LogitOffset	Fitted with delta = est.delta.

Examples:

>>> import numpy as np
>>> from probcal.offset import estimate_offset, offset_from_estimate
>>> rng = np.random.default_rng(0)
>>> p = rng.uniform(0.01, 0.5, 500)
>>> y = (rng.random(500) < p).astype(float)
>>> est = estimate_offset(y, p)
>>> off = offset_from_estimate(est, p)
>>> off.delta_ == est.delta
True

Source code in src/probcal/offset.py

def offset_from_estimate(est: OffsetEstimate, p: object) -> LogitOffset:
    """Build a fitted :class:`LogitOffset` (mode A) from an :class:`OffsetEstimate`.

    Equivalent to ``LogitOffset(delta=est.delta).fit(p)`` — a convenience for
    turning the audited MLE into the same offset object used elsewhere in
    the package (``transform``, ``interpret``, ``to_dict``, ...).

    Parameters
    ----------
    est : OffsetEstimate
        Result of :func:`estimate_offset`.
    p : array_like
        Probabilities to fit the offset's audit trail (``pre_mean_``,
        ``post_mean_``) against.

    Returns
    -------
    LogitOffset
        Fitted with ``delta = est.delta``.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.offset import estimate_offset, offset_from_estimate
    >>> rng = np.random.default_rng(0)
    >>> p = rng.uniform(0.01, 0.5, 500)
    >>> y = (rng.random(500) < p).astype(float)
    >>> est = estimate_offset(y, p)
    >>> off = offset_from_estimate(est, p)
    >>> off.delta_ == est.delta
    True
    """
    return LogitOffset(delta=est.delta).fit(p)

OffsetEstimate dataclass

OffsetEstimate(delta: float, se: float, n: int, events: float, weight_sum: float)

Bases: _ResultBase

Offset-only logistic MLE of delta given p, with its standard error.

ATTRIBUTE	DESCRIPTION
delta	MLE of the logit-offset shift: the mean-matching root of sum(w * (y - sigma(logit(p) + delta))) = 0 (the offset-only logistic score equation), found by bisection. TYPE: float
se	Asymptotic standard error of delta, sqrt(sum(w**2 * q * (1 - q))) / sum(w * q * (1 - q)) at q = sigma(logit(p) + delta): invariant to the scale of the weights, and the Fisher 1 / sqrt(sum(q * (1 - q))) for unit weights. TYPE: float
n	Number of observations. TYPE: int
events	Weighted event count, sum(w * y). TYPE: float
weight_sum	Sum of weights (equals n for unit weights). TYPE: float

wrapper

CalibratedModel: model-level wrapper with prefit and cross-validation flows.

Theory of the flows (why prefit is the credit-risk canon, why the pooled cv variant is the recommended default): docs/concepts/data-splitting.md.

CalibratedModel

CalibratedModel(model: Any, calibrator: BaseCalibrator, flow: str = 'prefit', cv: int = 5, ensemble: bool = False, random_state: int = 42, *, model_id: str | None = None)

Bases: CalibratorProtocolMixin, JsonIO

Wrap any scoring model with a probcal calibrator (and optional offsets).

PARAMETER	DESCRIPTION
model	Duck-typed model with predict_proba(X) or decision_function(X). For flow="cv" it must also have fit(X, y) and be clonable. X is always passed to the model as given (never converted), so DataFrames, column transformers, and string features work. TYPE: object
calibrator	Unfitted calibrator instance (its parameters are cloned per fold in the cv flow via get_params). TYPE: BaseCalibrator
flow	"prefit": the model is already trained; fit(X_cal, y_cal) scores the calibration set and fits the calibrator — the canonical credit-risk flow. "cv": the model is cloned and retrained per fold; every observation is scored by a model that did not train on it. TYPE: (prefit, cv) DEFAULT: "prefit"
cv	Fold count for the cv flow (stratified, seeded); an integer >= 2. TYPE: int DEFAULT: 5
ensemble	False (recommended default): one calibrator on pooled out-of-fold scores, final model refit on all data — a single auditable mapping. True: keep the per-fold (model, calibrator) pairs and average their predictions. TYPE: bool DEFAULT: False
random_state	Seed for the fold assignment. TYPE: int DEFAULT: 42

ATTRIBUTE	DESCRIPTION
model_	The deployed model (the input model for prefit; the full-data refit for pooled cv). TYPE: object
calibrator_	The fitted calibrator (pooled/prefit flows). TYPE: BaseCalibrator
ensemble_	The fold pairs (ensemble flow only). TYPE: list[tuple[model, BaseCalibrator]]
offsets_	Appended offset stages, each separately inspectable. TYPE: list[LogitOffset]

Notes

interval_inverse and point_inverse invert the full pipeline (calibrator, then every offset) through :attr:chain_ — see :meth:Chain.interval_inverse and :meth:Chain.point_inverse; the bounds are on the model's probability output (or its logit). The ensemble flow (K distinct maps) has no single preimage, so both raise NotImplementedError there.

Source code in src/probcal/wrapper.py

def __init__(
    self,
    model: Any,
    calibrator: BaseCalibrator,
    flow: str = "prefit",
    cv: int = 5,
    ensemble: bool = False,
    random_state: int = 42,
    *,
    model_id: str | None = None,
) -> None:
    self.model = model
    self.calibrator = calibrator
    self.flow = flow
    self.cv = cv
    self.ensemble = ensemble
    self.random_state = random_state
    self.model_id = model_id

is_monotone_ property

is_monotone_: bool

Monotone iff every calibrator stage is (offsets always are).

affine_logit_coeffs_ property

affine_logit_coeffs_: tuple[float, float] | None

Composed (a, b + sum(deltas)) when the calibrator stage is affine.

chain_ property

chain_: Chain

The equivalent model-free :class:probcal.Chain (calibrator + offsets).

Hand this to a recourse engine when the base model stays behind: the chain calibrates on the model probability, so its space="logit" bounds are bounds on the raw margin.

The returned chain aliases this wrapper's own fitted calibrator and offsets rather than copying them, so calling fit on the chain refits this :class:CalibratedModel's calibrator and offsets in place.

RAISES	DESCRIPTION
NotImplementedError	For the ensemble flow.

fit

fit(X: object, y: object, sample_weight: object = None) -> Self

Fit the calibration stage per the configured flow.

PARAMETER	DESCRIPTION
X	Calibration-set inputs, passed untouched to the model (flow="cv": row subsets via .iloc or numpy indexing) or scored directly by the already-trained model (flow="prefit"). TYPE: array_like, DataFrame, or any model input
y	Binary outcomes in {0, 1}; both classes must be present. TYPE: array_like
sample_weight	Positive observation weights. Always used by the calibrator; in the cv flow also forwarded to model.fit when its signature names sample_weight (or, with sklearn metadata routing on, the model is a router); otherwise a UserWarning says the model is trained unweighted. Also kept for :meth:offset_to without X. TYPE: array_like or None DEFAULT: None

RETURNS	DESCRIPTION
Self	The fitted wrapper.

RAISES	DESCRIPTION
ValueError	If flow is not "prefit" or "cv", cv is not an integer >= 2 (or a class has fewer than 2 members; with ensemble=True, fewer than cv members, since every fold's calibrator needs both classes), X and y differ in length, or the model exposes classes_ other than {0, 1}.
TypeError	If flow="cv" and the model has no fit(X, y) method.

Source code in src/probcal/wrapper.py

def fit(self, X: object, y: object, sample_weight: object = None) -> Self:
    """Fit the calibration stage per the configured flow.

    Parameters
    ----------
    X : array_like, DataFrame, or any model input
        Calibration-set inputs, passed untouched to the model
        (``flow="cv"``: row subsets via ``.iloc`` or numpy indexing) or
        scored directly by the already-trained model (``flow="prefit"``).
    y : array_like
        Binary outcomes in ``{0, 1}``; both classes must be present.
    sample_weight : array_like or None
        Positive observation weights. Always used by the calibrator; in
        the cv flow also forwarded to ``model.fit`` when its signature
        names ``sample_weight`` (or, with sklearn metadata routing on,
        the model is a router); otherwise a ``UserWarning`` says the
        model is trained unweighted. Also kept for
        :meth:`offset_to` without ``X``.

    Returns
    -------
    Self
        The fitted wrapper.

    Raises
    ------
    ValueError
        If ``flow`` is not ``"prefit"`` or ``"cv"``, ``cv`` is not an
        integer ``>= 2`` (or a class has fewer than 2 members; with
        ``ensemble=True``, fewer than ``cv`` members, since every fold's
        calibrator needs both classes), ``X`` and ``y`` differ in
        length, or the model exposes ``classes_`` other than ``{0, 1}``.
    TypeError
        If ``flow="cv"`` and the model has no ``fit(X, y)`` method.
    """
    if self.flow not in ("prefit", "cv"):
        raise ValueError(f"flow must be 'prefit' or 'cv', got {self.flow!r}")
    y_arr = validate_binary_y(y)
    if _n_rows(X) != len(y_arr):
        raise ValueError(f"X has {_n_rows(X)} rows but y has {len(y_arr)}")
    w_arr = validate_weights(sample_weight, len(y_arr))
    weighted = sample_weight is not None
    self.offsets_: list[LogitOffset] = []
    self.ensemble_: list[tuple[Any, BaseCalibrator]] = []
    if self.flow == "prefit":
        self.model_ = self.model
        s = _model_scores(self.model_, X)
        self.calibrator_ = self._fresh_calibrator().fit(s, y_arr, sample_weight=w_arr)
    else:
        cv = validate_cv(self.cv, y_arr, held_out_both_classes=self.ensemble)
        s = self._fit_cv(X, y_arr, w_arr, cv, weighted=weighted)
    self._cal_scores = s
    self._cal_weights = w_arr if weighted else None
    self.fit_meta_ = {
        "n_obs": int(len(y_arr)),
        "n_events": float(np.sum(w_arr * y_arr)),
        "weight_sum": float(w_arr.sum()),
        "fitted_at_utc": datetime.now(UTC).isoformat(timespec="seconds"),
        "data_fingerprint": data_fingerprint(s, y_arr, w_arr),
    }
    self.fitted_ = True
    return self

__sklearn_is_fitted__

__sklearn_is_fitted__() -> bool

Fitted state for sklearn >= 1.6 (model and calibrator both fitted).

Source code in src/probcal/wrapper.py

def __sklearn_is_fitted__(self) -> bool:
    """Fitted state for sklearn >= 1.6 (model and calibrator both fitted)."""
    return bool(getattr(self, "fitted_", False))

predict_proba

predict_proba(X: object) -> ndarray

Calibrated (and offset) probabilities P(y=1) for new inputs.

PARAMETER	DESCRIPTION
X	New inputs, passed untouched to the deployed model (or every ensemble fold's model, averaged). TYPE: array_like, DataFrame, or any model input

RETURNS	DESCRIPTION
numpy.ndarray of shape (n,)	Calibrated probabilities, after any appended offset stages.

Source code in src/probcal/wrapper.py

def predict_proba(self, X: object) -> np.ndarray:
    """Calibrated (and offset) probabilities ``P(y=1)`` for new inputs.

    Parameters
    ----------
    X : array_like, DataFrame, or any model input
        New inputs, passed untouched to the deployed model (or every
        ensemble fold's model, averaged).

    Returns
    -------
    numpy.ndarray of shape (n,)
        Calibrated probabilities, after any appended offset stages.
    """
    self._check_fitted()
    return self._pipeline(lambda model: _model_scores(model, X))

offset_to

offset_to(*, target_mean: float | None = None, delta: float | None = None, X: object = None, sample_weight: object = None) -> Self

Append an inspectable :class:LogitOffset stage.

Mode B (target_mean) anchors the portfolio mean of the current pipeline output — computed on X when given (weighted by sample_weight), else on the stored calibration scores, weighted by the sample_weight given to :meth:fit. The offset is never folded into the calibrator's parameters.

PARAMETER	DESCRIPTION
target_mean	Mode B: desired post-shift portfolio mean; mutually exclusive with delta (enforced by :class:LogitOffset). TYPE: (float or None, keyword - only) DEFAULT: None
delta	Mode A: the log-odds shift to apply directly; mutually exclusive with target_mean. TYPE: (float or None, keyword - only) DEFAULT: None
X	Inputs to compute the current pipeline output on; None uses the stored calibration scores (and calibration weights) instead. TYPE: (array_like or None, keyword - only) DEFAULT: None
sample_weight	Positive weights for the portfolio mean on X; only valid together with X. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Self	The wrapper, with the new offset appended to offsets_.

Source code in src/probcal/wrapper.py

def offset_to(
    self,
    *,
    target_mean: float | None = None,
    delta: float | None = None,
    X: object = None,
    sample_weight: object = None,
) -> Self:
    """Append an inspectable :class:`LogitOffset` stage.

    Mode B (``target_mean``) anchors the portfolio mean of the current
    pipeline output — computed on ``X`` when given (weighted by
    ``sample_weight``), else on the stored calibration scores, weighted
    by the ``sample_weight`` given to :meth:`fit`. The offset is never
    folded into the calibrator's parameters.

    Parameters
    ----------
    target_mean : float or None, keyword-only
        Mode B: desired post-shift portfolio mean; mutually exclusive
        with ``delta`` (enforced by :class:`LogitOffset`).
    delta : float or None, keyword-only
        Mode A: the log-odds shift to apply directly; mutually exclusive
        with ``target_mean``.
    X : array_like or None, keyword-only
        Inputs to compute the current pipeline output on; ``None`` uses
        the stored calibration scores (and calibration weights) instead.
    sample_weight : array_like or None, keyword-only
        Positive weights for the portfolio mean on ``X``; only valid
        together with ``X``.

    Returns
    -------
    Self
        The wrapper, with the new offset appended to ``offsets_``.
    """
    self._check_fitted()
    if X is not None:
        p_now = self.predict_proba(X)
        weights = sample_weight
    else:
        if sample_weight is not None:
            raise ValueError(
                "sample_weight is only accepted together with X; without X the "
                "calibration weights given to fit() are used"
            )
        cal_scores = self._cal_scores
        if cal_scores is None:
            raise RuntimeError(
                "calibration scores are not serialized; after from_dict, pass X so "
                "the offset stage can record its audit means on current output"
            )
        p_now = self._pipeline(lambda _: cal_scores)
        weights = self._cal_weights
    off = LogitOffset(delta=delta, target_mean=target_mean)
    off.fit(p_now, sample_weight=weights)
    self.offsets_.append(off)
    return self

with_offset

with_offset(offset: LogitOffset) -> CalibratedModel

A deep copy of this wrapper with an already-fitted offset appended.

Needs no data: unlike :meth:offset_to, nothing is fitted — the given offset (e.g. from :func:probcal.offset_from_estimate or a monitor recommendation) is copied and appended as the last stage. self is left untouched.

PARAMETER	DESCRIPTION
offset	A fitted offset. TYPE: LogitOffset

RETURNS	DESCRIPTION
CalibratedModel	The new wrapper (offsets_ == [*self.offsets_, offset], copied).

RAISES	DESCRIPTION
TypeError	If offset is not a :class:LogitOffset.
RuntimeError	If this wrapper or the offset is not fitted.

Source code in src/probcal/wrapper.py

def with_offset(self, offset: LogitOffset) -> "CalibratedModel":
    """A deep copy of this wrapper with an already-fitted offset appended.

    Needs no data: unlike :meth:`offset_to`, nothing is fitted — the
    given offset (e.g. from :func:`probcal.offset_from_estimate` or a
    monitor recommendation) is copied and appended as the last stage.
    ``self`` is left untouched.

    Parameters
    ----------
    offset : LogitOffset
        A fitted offset.

    Returns
    -------
    CalibratedModel
        The new wrapper (``offsets_ == [*self.offsets_, offset]``, copied).

    Raises
    ------
    TypeError
        If ``offset`` is not a :class:`LogitOffset`.
    RuntimeError
        If this wrapper or the offset is not fitted.
    """
    self._check_fitted()
    if not isinstance(offset, LogitOffset):
        raise TypeError(f"offset must be a LogitOffset, got {type(offset).__name__}")
    offset._check_fitted()
    new = copy.deepcopy(self)
    new.offsets_.append(copy.deepcopy(offset))
    return new

interpret

interpret() -> Interpretation

Concatenated interpretation of the calibrator and every offset stage.

RETURNS	DESCRIPTION
Interpretation	Parameters and messages concatenated across the calibrator stage(s) and every appended offset, in application order.

Source code in src/probcal/wrapper.py

def interpret(self) -> Interpretation:
    """Concatenated interpretation of the calibrator and every offset stage.

    Returns
    -------
    Interpretation
        Parameters and messages concatenated across the calibrator
        stage(s) and every appended offset, in application order.
    """
    stages: list[BaseCalibrator | LogitOffset] = [*self._calibrators(), *self.offsets_]
    parts = [stage.interpret() for stage in stages]
    return _concat_interpretations("CalibratedModel", parts, qualify=False)

to_dict

to_dict() -> dict[str, object]

Versioned snapshot: nested calibrator, offsets, and a model reference.

The base model is never serialized — only a reference (class name, the user-supplied model_id, and get_params() when available and JSON-encodable); reattach it on load via CalibratedModel.from_dict(d, model=...). A wrapper loaded without a model writes back the reference it was loaded with, so its fingerprint is unchanged. The stored calibration scores are not serialized either: after a reload, offset_to needs an explicit X.

RAISES	DESCRIPTION
RuntimeError	If not yet fitted.
NotImplementedError	For the ensemble flow: K fold models cannot be referenced.

Source code in src/probcal/wrapper.py

def to_dict(self) -> dict[str, object]:
    """Versioned snapshot: nested calibrator, offsets, and a model *reference*.

    The base model is never serialized — only a reference (class name,
    the user-supplied ``model_id``, and ``get_params()`` when available
    and JSON-encodable); reattach it on load via
    ``CalibratedModel.from_dict(d, model=...)``. A wrapper loaded without
    a model writes back the reference it was loaded with, so its
    fingerprint is unchanged. The stored calibration
    scores are not serialized either: after a reload,
    ``offset_to`` needs an explicit ``X``.

    Raises
    ------
    RuntimeError
        If not yet fitted.
    NotImplementedError
        For the ensemble flow: K fold models cannot be referenced.
    """
    self._check_fitted()
    if self.ensemble_:
        raise NotImplementedError(
            "the ensemble flow holds K fold models that cannot be referenced; "
            "serialize a pooled (ensemble=False) or prefit wrapper instead"
        )
    ref_model = self.model_ if self.model_ is not None else self.model
    if ref_model is None and self._model_ref is not None:
        # Loaded without a model: keep the reference it was saved with.
        model_ref = self._model_ref
    else:
        model_ref = self._model_ref_of(ref_model)
    return envelope(
        self,
        params={
            "flow": self.flow,
            "cv": self.cv,
            "ensemble": self.ensemble,
            "random_state": self.random_state,
            "model_id": self.model_id,
        },
        state={
            "calibrator": self.calibrator_.to_dict(),
            "offsets": [off.to_dict() for off in self.offsets_],
            "model_ref": model_ref,
        },
        fit_meta=encode_value(self.fit_meta_),
    )

from_dict classmethod

from_dict(d: dict, model: Any = None) -> CalibratedModel

Rebuild a fitted wrapper, reattaching the base model.

PARAMETER	DESCRIPTION
d	Output of :meth:to_dict. TYPE: dict
model	The base model to reattach (matched against the stored reference is the caller's responsibility). With None, the loaded wrapper can serialize and introspect but predict_proba raises until a model is assigned to model_. TYPE: object or None DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the schema version is unknown or the payload class differs.

Source code in src/probcal/wrapper.py

@classmethod
def from_dict(cls, d: dict, model: Any = None) -> "CalibratedModel":
    """Rebuild a fitted wrapper, reattaching the base model.

    Parameters
    ----------
    d : dict
        Output of :meth:`to_dict`.
    model : object or None
        The base model to reattach (matched against the stored reference
        is the caller's responsibility). With ``None``, the loaded
        wrapper can serialize and introspect but ``predict_proba``
        raises until a model is assigned to ``model_``.

    Raises
    ------
    ValueError
        If the schema version is unknown or the payload class differs.
    """
    check_payload(cls, d)
    state = d["state"]
    calibrator = load(state["calibrator"])
    obj = cls(model, calibrator, **d["params"])  # type: ignore[arg-type]
    obj.calibrator_ = calibrator  # type: ignore[assignment]
    obj.offsets_ = [LogitOffset.from_dict(o) for o in state["offsets"]]
    obj.ensemble_ = []
    obj.model_ = model
    obj._model_ref = dict(state["model_ref"])
    obj.fit_meta_ = dict(d["fit_meta"])
    obj.fitted_ = True
    return obj

from_json classmethod

from_json(path_or_str: object, model: Any = None) -> Self

Load from a JSON string or a filesystem path (see :meth:from_dict).

Source code in src/probcal/wrapper.py

@classmethod
def from_json(cls, path_or_str: object, model: Any = None) -> Self:  # type: ignore[override]
    """Load from a JSON string or a filesystem path (see :meth:`from_dict`)."""
    return super().from_json(path_or_str, model=model)

interval_inverse

interval_inverse(lo: float, hi: float, *, space: str = 'probability', buffer_logit: float = 0.0) -> tuple[float, float]

Preimage of a calibrated interval on the calibrator's input scale (delegated).

Source code in src/probcal/_protocol.py

def interval_inverse(
    self, lo: float, hi: float, *, space: str = "probability", buffer_logit: float = 0.0
) -> tuple[float, float]:
    """Preimage of a calibrated interval on the calibrator's input scale (delegated)."""
    return self._fitted_calibrator().interval_inverse(
        lo, hi, space=space, buffer_logit=buffer_logit
    )

point_inverse

point_inverse(p: object, *, space: str = 'probability') -> ndarray

Exact preimage of calibrated probabilities (delegated).

Source code in src/probcal/_protocol.py

def point_inverse(self, p: object, *, space: str = "probability") -> np.ndarray:
    """Exact preimage of calibrated probabilities (delegated)."""
    return self._fitted_calibrator().point_inverse(p, space=space)

selection

CalibratorSelector: automatic method selection under nested validation.

The selector's scoring path only ever receives out-of-fold predictions — selection on fitting data is an unrepresentable state, not a documented misuse. Protocol, criteria, and report reading: docs/concepts/auto-selection.md.

CalibratorSelector

CalibratorSelector(candidates: dict[str, BaseCalibrator] | None = None, scoring: str = 'log_loss', cv: int = 5, random_state: int = 42)

Bases: CalibratorProtocolMixin, BaseCalibrator

Choose a calibrator by inner cross-validation on the calibration data.

Custom candidates declare their tie-break position by overriding complexity_rank (lower = simpler; default 100.0 ranks last).

Once fitted, the selector is its winner for every protocol question: predict_proba, interpret, is_monotone_, affine_logit_coeffs_, complexity_rank, point_inverse and interval_inverse all delegate to best_calibrator_. fit raises ValueError if scoring is not one of the accepted criteria (plain ECE and Hosmer–Lemeshow are refused as selection criteria), cv is not an integer >= 2, or a class has fewer than cv members (every held-out fold is scored on its own and needs both classes; lower cv for low-default data).

PARAMETER	DESCRIPTION
candidates	Candidate instances (cloned per fold via get_params). None uses the default menu: platt, temperature, beta_abm, isotonic, cir, histogram_mass, scaling_binning, ivap. The heavier methods (spline, BBQ, ENIR, CVAP) join by explicit opt-in. TYPE: dict[str, BaseCalibrator] or None DEFAULT: None
scoring	Out-of-fold selection criterion, lower is better. Plain ECE and Hosmer–Lemeshow are refused — see the metrics chapter's table. TYPE: (log_loss, brier, ici, smooth_ece, ece_sweep) DEFAULT: "log_loss"
cv	Inner stratified fold count; an integer >= 2. TYPE: int DEFAULT: 5
random_state	Seed for the fold assignment. TYPE: int DEFAULT: 42

ATTRIBUTE	DESCRIPTION
best_name_	Winning candidate's name. TYPE: str
best_calibrator_	The winner refitted on the full calibration set. TYPE: BaseCalibrator
report_	Ranked table: mean ± sd of the criterion, guardrail flags, chosen marker. TYPE: SelectionReport

Source code in src/probcal/selection.py

def __init__(
    self,
    candidates: dict[str, BaseCalibrator] | None = None,
    scoring: str = "log_loss",
    cv: int = 5,
    random_state: int = 42,
) -> None:
    self.candidates = candidates
    self.scoring = scoring
    self.cv = cv
    self.random_state = random_state

is_monotone_ property

is_monotone_: bool

The winner's is_monotone_ (True before fitting).

complexity_rank property

complexity_rank: float

The winner's parsimony rank once fitted; 100.0 before.

affine_logit_coeffs_ property

affine_logit_coeffs_: tuple[float, float] | None

Affine-logit coefficients of the calibration map, if any (delegated).

interpret

interpret() -> Interpretation

The fitted calibrator's plain-language reading (delegated).

Source code in src/probcal/_protocol.py

def interpret(self) -> "Interpretation":
    """The fitted calibrator's plain-language reading (delegated)."""
    return self._fitted_calibrator().interpret()

interval_inverse

interval_inverse(lo: float, hi: float, *, space: str = 'probability', buffer_logit: float = 0.0) -> tuple[float, float]

Preimage of a calibrated interval on the calibrator's input scale (delegated).

Source code in src/probcal/_protocol.py

def interval_inverse(
    self, lo: float, hi: float, *, space: str = "probability", buffer_logit: float = 0.0
) -> tuple[float, float]:
    """Preimage of a calibrated interval on the calibrator's input scale (delegated)."""
    return self._fitted_calibrator().interval_inverse(
        lo, hi, space=space, buffer_logit=buffer_logit
    )

point_inverse

point_inverse(p: object, *, space: str = 'probability') -> ndarray

Exact preimage of calibrated probabilities (delegated).

Source code in src/probcal/_protocol.py

def point_inverse(self, p: object, *, space: str = "probability") -> np.ndarray:
    """Exact preimage of calibrated probabilities (delegated)."""
    return self._fitted_calibrator().point_inverse(p, space=space)

curves

Reliability-curve builders and the GiViTI-style calibration belt.

Numpy-only; every result is a frozen dataclass carrying both probability- and logit-scale coordinates, plotting-backend-agnostic (rendering lives in probcal.plots). Theory: docs/concepts/visualization.md.

References

Austin & Steyerberg (2014); Nattino, Finazzi & Bertolini (2014); Nattino, Lemeshow, Phillips, Finazzi & Bertolini (2017) — full records in the documentation. The belt is reimplemented from the papers; no GPL code is used.

EcceCurve dataclass

EcceCurve(frac: ndarray, cumdev: ndarray, sd_null: ndarray, stat_max: float, argmax_frac: float)

Bases: _ResultBase

Cumulative-deviation walk over predictions sorted ascending (ECCE).

Compares by value (arrays element-wise); unhashable.

ATTRIBUTE	DESCRIPTION
frac	Cumulative weight fraction at the end of each block of tied predictions (1/n .. 1 for distinct scores and unit weights). TYPE: ndarray
cumdev	Cumulative-deviation walk value at each frac. TYPE: ndarray
sd_null	Pointwise standard deviation of the walk under calibration. TYPE: ndarray
stat_max	Maximum absolute value of cumdev (agrees with metrics.ecce's stat_max). TYPE: float
argmax_frac	frac at which the maximum is attained. TYPE: float

reliability_binned

reliability_binned(y: object, p: object, *, n_bins: int = 10, strategy: str = 'mass', sample_weight: object = None) -> ReliabilityCurve

Binned reliability curve with Wilson confidence intervals.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
n_bins	Requested bin count. TYPE: int DEFAULT: 10
strategy	Equal-count (default) or equal-width bins. TYPE: (mass, width) DEFAULT: "mass"
sample_weight	Weights for the bin means; Wilson CIs use raw counts. TYPE: array_like or None DEFAULT: None

RETURNS	DESCRIPTION
ReliabilityCurve	Per-bin mean prediction, event rate, count, Wilson CI, and the logit-scale coordinates.

Source code in src/probcal/curves.py

def reliability_binned(
    y: object,
    p: object,
    *,
    n_bins: int = 10,
    strategy: str = "mass",
    sample_weight: object = None,
) -> ReliabilityCurve:
    """Binned reliability curve with Wilson confidence intervals.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    n_bins : int
        Requested bin count.
    strategy : {"mass", "width"}
        Equal-count (default) or equal-width bins.
    sample_weight : array_like or None
        Weights for the bin means; Wilson CIs use raw counts.

    Returns
    -------
    ReliabilityCurve
        Per-bin mean prediction, event rate, count, Wilson CI, and the
        logit-scale coordinates.
    """
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    bs = bin_stats(y_arr, p_arr, w, n_bins, strategy)
    counts = bs.n_obs
    pred_mean = bs.p_mean
    event_rate = bs.rate
    ci_low, ci_high = _wilson(event_rate, counts.astype(np.float64))
    # The Wilson interval contains the point estimate analytically; enforce it
    # against floating-point noise at 0/1-rate bins (negative yerr otherwise).
    ci_low = np.minimum(np.clip(ci_low, 0.0, 1.0), event_rate)
    ci_high = np.maximum(np.clip(ci_high, 0.0, 1.0), event_rate)
    return ReliabilityCurve(
        pred_mean=pred_mean,
        event_rate=event_rate,
        count=counts.astype(np.int64),
        ci_low=ci_low,
        ci_high=ci_high,
        pred_mean_logit=logit(pred_mean),
    )

reliability_loess

reliability_loess(y: object, p: object, *, frac: float = 0.75, grid_size: int = 100, sample_weight: object = None) -> SmoothReliabilityCurve

LOESS-smoothed reliability curve on a grid (Austin & Steyerberg, 2014).

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
frac	LOESS smoothing fraction. TYPE: (float, keyword - only) DEFAULT: 0.75
grid_size	Number of evaluation points, spanning the 0.5th to 99.5th percentile of p. TYPE: (int, keyword - only) DEFAULT: 100
sample_weight	Optional positive weights. Non-uniform weights multiply the tricube kernel weights of each local-linear fit (same nearest-neighbour windows); None or all-equal weights give the unweighted LOESS curve unchanged. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
SmoothReliabilityCurve	Grid coordinates (probability and logit scale) and the smoothed event rate at each point.

Source code in src/probcal/curves.py

def reliability_loess(
    y: object,
    p: object,
    *,
    frac: float = 0.75,
    grid_size: int = 100,
    sample_weight: object = None,
) -> SmoothReliabilityCurve:
    """LOESS-smoothed reliability curve on a grid (Austin & Steyerberg, 2014).

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    frac : float, keyword-only
        LOESS smoothing fraction.
    grid_size : int, keyword-only
        Number of evaluation points, spanning the 0.5th to 99.5th percentile
        of ``p``.
    sample_weight : array_like or None, keyword-only
        Optional positive weights. Non-uniform weights multiply the tricube
        kernel weights of each local-linear fit (same nearest-neighbour
        windows); ``None`` or all-equal weights give the unweighted LOESS
        curve unchanged.

    Returns
    -------
    SmoothReliabilityCurve
        Grid coordinates (probability and logit scale) and the smoothed
        event rate at each point.
    """
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    grid = _grid(p_arr, grid_size)
    if is_uniform(w):
        fitted = loess(p_arr, y_arr, frac=frac, xeval=grid)
    else:
        fitted = _weighted_local_linear(p_arr, y_arr, w, grid, frac)
    rate = np.clip(fitted, 0.0, 1.0)
    return SmoothReliabilityCurve(grid_p=grid, grid_logit=logit(grid), event_rate=rate)

reliability_spline

reliability_spline(y: object, p: object, *, grid_size: int = 100, sample_weight: object = None) -> SmoothReliabilityCurve

Spline-smoothed reliability curve on a grid.

Penalized natural cubic spline of the outcome on the logit prediction.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
grid_size	Number of evaluation points, spanning the 0.5th to 99.5th percentile of p. TYPE: (int, keyword - only) DEFAULT: 100
sample_weight	Optional positive weights passed to the spline fit. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
SmoothReliabilityCurve	Grid coordinates (probability and logit scale) and the smoothed event rate at each point.

Source code in src/probcal/curves.py

def reliability_spline(
    y: object,
    p: object,
    *,
    grid_size: int = 100,
    sample_weight: object = None,
) -> SmoothReliabilityCurve:
    """Spline-smoothed reliability curve on a grid.

    Penalized natural cubic spline of the outcome on the logit prediction.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    grid_size : int, keyword-only
        Number of evaluation points, spanning the 0.5th to 99.5th percentile
        of ``p``.
    sample_weight : array_like or None, keyword-only
        Optional positive weights passed to the spline fit.

    Returns
    -------
    SmoothReliabilityCurve
        Grid coordinates (probability and logit scale) and the smoothed
        event rate at each point.
    """
    from .spline import SplineCalibrator

    y_arr, p_arr, w = _prep(y, p, sample_weight)
    cal = SplineCalibrator()
    cal.fit(p_arr, y_arr, sample_weight=w)
    grid = _grid(p_arr, grid_size)
    return SmoothReliabilityCurve(
        grid_p=grid, grid_logit=logit(grid), event_rate=cal.predict_proba(grid)
    )

reliability_smooth

reliability_smooth(y: object, p: object, *, sample_weight: object = None, grid_size: int = 200, n_boot: int = 100, confidence: float = 0.9, random_state: int = 42, bins: int | None = 8192) -> KernelReliabilityCurve

smECE-consistent kernel reliability curve (Blasiok-Nakkiran).

Shares its bandwidth and lattice with metrics.smooth_ece: both solve the same fixed point sigma_star on the same equal-width logit lattice (metrics.smooth._smece_solve), so curve.smooth_ece reproduces metrics.smooth_ece(y, p, bins=bins) exactly instead of merely agreeing with it. The event rate and prediction density are then Nadaraya-Watson kernel estimates at that one fixed sigma_star — rate = K*bincount(w*y) / K*bincount(w) on the lattice, interpolated onto grid_logit — using the same truncated Gaussian kernel smooth_ece used to reach sigma_star (metrics.smooth._lattice_kernel_smooth). When smooth_ece's path selection falls back to its exact (non-lattice) computation — degenerate logit range, bins=None, or an infeasible/under-resolved refinement — the curve falls back the same way, to direct O(n * grid_size) Gaussian smoothing on logit(p) at sigma_star.

The confidence ribbon bootstraps (y, p, sample_weight) triples (numpy.random.default_rng(random_state), resampling with replacement) and recomputes the rate at the point estimate's fixed sigma_star — the ribbon conditions on the bandwidth, it does not reflect uncertainty in choosing it. The ribbon is clamped to contain the point estimate (ci_low <= event_rate <= ci_high), so a bootstrap quantile falling on the wrong side of it is pulled back to it. n_boot=0 disables the ribbon (ci_low and ci_high both equal event_rate).

Grid points with no observation within the kernel's +-5 sigma_star reach (data gaps, on the logit scale) have no estimate: event_rate, ci_low and ci_high are nan there. Bootstrap resamples that leave a grid point uncovered are skipped in that point's quantiles.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
sample_weight	Optional positive weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None
grid_size	Number of evaluation points, spanning the 0.5th to 99.5th percentile of p (curves._grid). TYPE: (int, keyword - only) DEFAULT: 200
n_boot	Number of bootstrap resamples for the confidence ribbon; 0 disables it. TYPE: (int, keyword - only) DEFAULT: 100
confidence	Nominal coverage level of the ribbon; must satisfy 0 < confidence < 1. TYPE: (float, keyword - only) DEFAULT: 0.9
random_state	Seed for numpy.random.default_rng, used by the bootstrap. TYPE: (int, keyword - only) DEFAULT: 42
bins	Lattice bin count passed through to the shared smECE solve; see metrics.smooth_ece. None forces the exact path. TYPE: (int or None, keyword - only) DEFAULT: 8192

RETURNS	DESCRIPTION
KernelReliabilityCurve	Grid coordinates, kernel-smoothed event rate and density, the bootstrap ribbon, and the shared sigma_star / smooth_ece.

RAISES	DESCRIPTION
ValueError	If confidence is not in (0, 1).

Examples:

>>> import numpy as np
>>> from probcal import make_pd_portfolio
>>> from probcal.curves import reliability_smooth
>>> d = make_pd_portfolio(n=2000, random_state=0)
>>> curve = reliability_smooth(d.y, d.scores, n_boot=0)
>>> len(curve.grid_p) == 200
True
>>> abs(float(curve.density.sum()) - 1.0) < 1e-10
True

Source code in src/probcal/curves.py

def reliability_smooth(
    y: object,
    p: object,
    *,
    sample_weight: object = None,
    grid_size: int = 200,
    n_boot: int = 100,
    confidence: float = 0.9,
    random_state: int = 42,
    bins: int | None = 8192,
) -> KernelReliabilityCurve:
    """smECE-consistent kernel reliability curve (Blasiok-Nakkiran).

    Shares its bandwidth and lattice with ``metrics.smooth_ece``: both solve
    the same fixed point ``sigma_star`` on the same equal-width logit
    lattice (``metrics.smooth._smece_solve``), so
    ``curve.smooth_ece`` reproduces ``metrics.smooth_ece(y, p, bins=bins)``
    exactly instead of merely agreeing with it. The event rate and
    prediction density are then Nadaraya-Watson kernel estimates at that one
    fixed ``sigma_star`` — ``rate = K*bincount(w*y) / K*bincount(w)`` on the
    lattice, interpolated onto ``grid_logit`` — using the same truncated
    Gaussian kernel ``smooth_ece`` used to reach ``sigma_star``
    (``metrics.smooth._lattice_kernel_smooth``). When ``smooth_ece``'s path
    selection falls back to its exact (non-lattice) computation — degenerate
    logit range, ``bins=None``, or an infeasible/under-resolved refinement —
    the curve falls back the same way, to direct O(n * grid_size) Gaussian
    smoothing on ``logit(p)`` at ``sigma_star``.

    The confidence ribbon bootstraps ``(y, p, sample_weight)`` triples
    (``numpy.random.default_rng(random_state)``, resampling with
    replacement) and recomputes the rate at the point estimate's *fixed*
    ``sigma_star`` — the ribbon conditions on the bandwidth, it does not
    reflect uncertainty in choosing it. The ribbon is clamped to contain the
    point estimate (``ci_low <= event_rate <= ci_high``), so a bootstrap
    quantile falling on the wrong side of it is pulled back to it.
    ``n_boot=0`` disables the ribbon (``ci_low`` and ``ci_high`` both equal
    ``event_rate``).

    Grid points with no observation within the kernel's ``+-5 sigma_star``
    reach (data gaps, on the logit scale) have no estimate: ``event_rate``,
    ``ci_low`` and ``ci_high`` are ``nan`` there. Bootstrap resamples that leave a grid point
    uncovered are skipped in that point's quantiles.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    sample_weight : array_like or None, keyword-only
        Optional positive weights, same length as ``y``.
    grid_size : int, keyword-only
        Number of evaluation points, spanning the 0.5th to 99.5th percentile
        of ``p`` (``curves._grid``).
    n_boot : int, keyword-only
        Number of bootstrap resamples for the confidence ribbon; ``0``
        disables it.
    confidence : float, keyword-only
        Nominal coverage level of the ribbon; must satisfy
        ``0 < confidence < 1``.
    random_state : int, keyword-only
        Seed for ``numpy.random.default_rng``, used by the bootstrap.
    bins : int or None, keyword-only
        Lattice bin count passed through to the shared smECE solve; see
        ``metrics.smooth_ece``. ``None`` forces the exact path.

    Returns
    -------
    KernelReliabilityCurve
        Grid coordinates, kernel-smoothed event rate and density, the
        bootstrap ribbon, and the shared ``sigma_star`` / ``smooth_ece``.

    Raises
    ------
    ValueError
        If ``confidence`` is not in ``(0, 1)``.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal import make_pd_portfolio
    >>> from probcal.curves import reliability_smooth
    >>> d = make_pd_portfolio(n=2000, random_state=0)
    >>> curve = reliability_smooth(d.y, d.scores, n_boot=0)
    >>> len(curve.grid_p) == 200
    True
    >>> abs(float(curve.density.sum()) - 1.0) < 1e-10
    True
    """
    conf = validate_level(confidence, "confidence")
    n_boot = validate_nonneg_int(n_boot, "n_boot")
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    grid_p = _grid(p_arr, grid_size)
    grid_logit = logit(grid_p)
    t = logit(p_arr)
    mass = (w / w.sum()) * (y_arr - p_arr)
    smooth_ece_value, sigma_star, lattice = _smece_solve(t, mass, bins)

    event_rate, density = _kernel_rate_density(t, y_arr, w, sigma_star, grid_logit, lattice)

    if n_boot > 0:
        rng = np.random.default_rng(random_state)
        n = len(y_arr)
        boot_rate = np.empty((n_boot, grid_size))
        for i in range(n_boot):
            idx_b = rng.integers(0, n, n)
            # `lattice` is the point estimate's fixed (width, t_lo, b) — never
            # rederived here. The ribbon is meant to reflect uncertainty in the
            # rate given sigma_star, not uncertainty in sigma_star or its
            # lattice; re-solving the smECE fixed point per resample would also
            # make each resample's rate estimate use a different bandwidth and
            # bin grid, so resamples would stop being comparable pointwise.
            boot_rate[i], _ = _kernel_rate_density(
                t[idx_b], y_arr[idx_b], w[idx_b], sigma_star, grid_logit, lattice
            )
        a = (1.0 - conf) / 2.0
        with warnings.catch_warnings():
            warnings.simplefilter("ignore", RuntimeWarning)  # all-NaN columns
            q_lo = np.nanquantile(boot_rate, a, axis=0)
            q_hi = np.nanquantile(boot_rate, 1.0 - a, axis=0)
        gap = np.isnan(event_rate)
        ci_low = np.where(gap, np.nan, np.fmin(q_lo, event_rate))
        ci_high = np.where(gap, np.nan, np.fmax(q_hi, event_rate))
    else:
        ci_low = event_rate.copy()
        ci_high = event_rate.copy()

    return KernelReliabilityCurve(
        grid_p=grid_p,
        grid_logit=grid_logit,
        event_rate=event_rate,
        density=density,
        ci_low=ci_low,
        ci_high=ci_high,
        sigma_star=sigma_star,
        smooth_ece=smooth_ece_value,
    )

corp_reliability

corp_reliability(y: object, p: object, *, sample_weight: object = None, bands: str | None = 'consistency', confidence: float = 0.9, n_resamples: int = 200, random_state: int = 42) -> CorpResult

CORP reliability diagram with the Brier/log-loss MCB-DSC-UNC decomposition.

Fits the isotonic (PAV) recalibration map of y on p — the unique "consistent, optimally binned, reproducible" reliability diagram of Dimitriadis, Gneiting & Jordan (2021) — and decomposes both the Brier score and log loss into miscalibration (MCB), discrimination (DSC), and uncertainty (UNC) terms, with score == mcb - dsc + unc holding exactly. Log loss clips PAV levels and predictions to [1e-12, 1 - 1e-12] before taking logarithms, so degenerate blocks (exact 0 or 1 event rate) stay finite.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
sample_weight	Optional positive weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None
bands	Band type to compute around the PAV fit. "consistency" resamples y ~ Bernoulli(p) under the null that p is calibrated; "confidence" bootstraps (y, p, sample_weight) triples. Both give pointwise, not simultaneous, bands (see Notes). TYPE: (consistency, confidence, None) DEFAULT: "consistency"
confidence	Nominal coverage level of the bands; must satisfy 0 < confidence < 1 (stored as CorpResult.level). TYPE: (float, keyword - only) DEFAULT: 0.9
n_resamples	Number of resamples used to build the bands; 0 skips them (as bands=None). TYPE: (int, keyword - only) DEFAULT: 200
random_state	Seed for numpy.random.default_rng, used by the band resampling. TYPE: (int, keyword - only) DEFAULT: 42

RETURNS	DESCRIPTION
CorpResult	PAV block structure, the pointwise fit, the Brier/log-loss decomposition, and the (possibly empty) bands.

RAISES	DESCRIPTION
ValueError	If bands is not one of "consistency", "confidence", or None, or if confidence is not in (0, 1).

Notes

Bands are pointwise: at each grid point, confidence of resamples fall inside, not that the whole curve does so simultaneously (the docs/scripts/corp_sim.py coverage simulation reports the gap between pointwise and uniform coverage). corp_reliability with n=10_000, n_resamples=200 takes about 3.5 s (measured once on the development machine) — the PAV step is a Python loop over unique scores (_math.pava), and the bands refit PAV n_resamples times.

Examples:

>>> import numpy as np
>>> from probcal import corp_reliability
>>> rng = np.random.default_rng(0)
>>> p = rng.uniform(0.1, 0.9, 200)
>>> y = (rng.random(200) < p).astype(float)
>>> r = corp_reliability(y, p, bands=None)
>>> abs(r.brier - (r.brier_mcb - r.brier_dsc + r.brier_unc)) < 1e-12
True

Source code in src/probcal/curves.py

def corp_reliability(
    y: object,
    p: object,
    *,
    sample_weight: object = None,
    bands: str | None = "consistency",
    confidence: float = 0.9,
    n_resamples: int = 200,
    random_state: int = 42,
) -> CorpResult:
    """CORP reliability diagram with the Brier/log-loss MCB-DSC-UNC decomposition.

    Fits the isotonic (PAV) recalibration map of ``y`` on ``p`` — the unique
    "consistent, optimally binned, reproducible" reliability diagram of
    Dimitriadis, Gneiting & Jordan (2021) — and decomposes both the Brier
    score and log loss into miscalibration (MCB), discrimination (DSC), and
    uncertainty (UNC) terms, with ``score == mcb - dsc + unc`` holding
    exactly. Log loss clips PAV levels and predictions to
    ``[1e-12, 1 - 1e-12]`` before taking logarithms, so degenerate blocks
    (exact 0 or 1 event rate) stay finite.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    sample_weight : array_like or None, keyword-only
        Optional positive weights, same length as ``y``.
    bands : {"consistency", "confidence", None}, keyword-only
        Band type to compute around the PAV fit. ``"consistency"`` resamples
        ``y ~ Bernoulli(p)`` under the null that ``p`` is calibrated;
        ``"confidence"`` bootstraps ``(y, p, sample_weight)`` triples. Both
        give pointwise, not simultaneous, bands (see Notes).
    confidence : float, keyword-only
        Nominal coverage level of the bands; must satisfy
        ``0 < confidence < 1`` (stored as ``CorpResult.level``).
    n_resamples : int, keyword-only
        Number of resamples used to build the bands; ``0`` skips them (as
        ``bands=None``).
    random_state : int, keyword-only
        Seed for ``numpy.random.default_rng``, used by the band resampling.

    Returns
    -------
    CorpResult
        PAV block structure, the pointwise fit, the Brier/log-loss
        decomposition, and the (possibly empty) bands.

    Raises
    ------
    ValueError
        If ``bands`` is not one of ``"consistency"``, ``"confidence"``, or
        ``None``, or if ``confidence`` is not in ``(0, 1)``.

    Notes
    -----
    Bands are pointwise: at each grid point, ``confidence`` of resamples fall
    inside, not that the whole curve does so simultaneously (the
    ``docs/scripts/corp_sim.py`` coverage simulation reports the gap between
    pointwise and uniform coverage). ``corp_reliability`` with
    ``n=10_000, n_resamples=200`` takes about 3.5 s (measured once on the
    development machine) — the PAV step is a Python loop over unique scores
    (``_math.pava``), and the bands refit PAV ``n_resamples`` times.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal import corp_reliability
    >>> rng = np.random.default_rng(0)
    >>> p = rng.uniform(0.1, 0.9, 200)
    >>> y = (rng.random(200) < p).astype(float)
    >>> r = corp_reliability(y, p, bands=None)
    >>> abs(r.brier - (r.brier_mcb - r.brier_dsc + r.brier_unc)) < 1e-12
    True
    """
    if bands not in _BANDS:
        raise ValueError('bands must be "consistency", "confidence", or None')
    conf = validate_level(confidence, "confidence")
    n_resamples = validate_nonneg_int(n_resamples, "n_resamples")
    if n_resamples == 0:
        bands = None
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    lo, hi, level_b, w_b, pav = corp_fit(y_arr, p_arr, w)
    b = decompose(y_arr, p_arr, pav, w, "brier")
    ll = decompose(y_arr, p_arr, pav, w, "log_loss")
    grid, low, high = corp_bands(y_arr, p_arr, w, bands, conf, n_resamples, random_state)
    return CorpResult(
        block_lo=lo,
        block_hi=hi,
        block_level=level_b,
        block_weight=w_b,
        pav=pav,
        brier=b[0],
        brier_mcb=b[1],
        brier_dsc=b[2],
        brier_unc=b[3],
        log_loss=ll[0],
        log_loss_mcb=ll[1],
        log_loss_dsc=ll[2],
        log_loss_unc=ll[3],
        bands=bands,
        level=conf,
        band_grid=grid,
        band_low=low,
        band_high=high,
        n=int(len(y_arr)),
        events=int(y_arr.sum()),
    )

ecce_curve

ecce_curve(y: object, p: object, *, sample_weight: object = None) -> EcceCurve

Cumulative-deviation walk for the ECCE plot (Arrieta-Ibarra et al., 2022).

Sorts by prediction and accumulates weighted residuals with the same helper as metrics.ecce, so stat_max agrees with the metric. The walk is reported only at the end of each block of tied predictions (the order of tied rows is arbitrary, so intermediate points carry no information); frac is the cumulative weight fraction. sd_null is the pointwise standard deviation of the walk under calibration — an envelope for reading, not a simultaneous band — the exact null SD sqrt(cumsum(w**2 p (1 - p))) / sum(w) of the weighted walk (sqrt(cumsum(p (1 - p))) / n for unit weights; invariant to rescaling the weights).

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
sample_weight	Optional positive weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
EcceCurve	Cumulative walk, null-envelope SD, and the max-deviation summary.

Source code in src/probcal/curves.py

def ecce_curve(y: object, p: object, *, sample_weight: object = None) -> EcceCurve:
    """Cumulative-deviation walk for the ECCE plot (Arrieta-Ibarra et al., 2022).

    Sorts by prediction and accumulates weighted residuals with the same
    helper as ``metrics.ecce``, so ``stat_max`` agrees with the metric. The
    walk is reported only at the end of each block of tied predictions (the
    order of tied rows is arbitrary, so intermediate points carry no
    information); ``frac`` is the cumulative weight fraction. ``sd_null`` is
    the pointwise standard deviation of the walk under calibration — an
    envelope for reading, not a simultaneous band — the exact null SD
    ``sqrt(cumsum(w**2 p (1 - p))) / sum(w)`` of the weighted walk
    (``sqrt(cumsum(p (1 - p))) / n`` for unit weights; invariant to
    rescaling the weights).

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    sample_weight : array_like or None, keyword-only
        Optional positive weights, same length as ``y``.

    Returns
    -------
    EcceCurve
        Cumulative walk, null-envelope SD, and the max-deviation summary.
    """
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    walk = _ecce_walk(y_arr, p_arr, w, sd=True)
    assert walk.sd_null is not None
    k = int(np.argmax(np.abs(walk.cumdev)))
    return EcceCurve(
        frac=walk.frac,
        cumdev=walk.cumdev,
        sd_null=walk.sd_null,
        stat_max=float(np.abs(walk.cumdev[k])),
        argmax_frac=float(walk.frac[k]),
    )

calibration_belt

calibration_belt(y: object, p: object, *, confidence: tuple[float, float] = (0.8, 0.95), grid_size: int = 100, sample_weight: object = None) -> BeltResult

GiViTI-style calibration belt (Nattino et al., 2014, 2017).

Fits a polynomial logistic recalibration of the outcome on logit(p), selecting the degree by forward likelihood-ratio testing (p < 0.05 to add a term, capped at degree 4). The polynomial is fitted on the centred and scaled logit (same model, better-conditioned design).

Both outputs are approximations of the published construction:

• Belt. At each confidence level the band is the image of the Wald (information-matrix) ellipsoid with radius chi2_ppf(level, degree + 1) — a quadratic approximation of the likelihood-ratio confidence region the papers invert. Using the degree + 1-df radius makes it a simultaneous (Scheffé-type) band over the grid for the selected polynomial, not a pointwise one; it conditions on the selected degree.
• p-value. The likelihood-ratio test of the selected polynomial against the identity, referred to chi2(degree + 1); it ignores the forward selection of the degree, whose exact null distribution (Nattino et al., 2014) puts more mass in the upper tail, so the p-value is anti-conservative.

Where the band excludes the diagonal, the data reject calibration in that region. With non-uniform weights (package convention, :mod:probcal.metrics._common) the fits maximise the weighted likelihood, the band uses the sandwich covariance I**-1 J I**-1 with J = X' diag(w**2 mu (1 - mu)) X (the exact variance of the weighted score under the fitted model), the forward-selection steps divide each 1-df LR statistic by the added coefficient's design effect (sandwich over model-based variance), and the p-value is second-order Rao-Scott corrected (:func:probcal.metrics._common.rao_scott, its reference distribution chi2 with the corrected, possibly fractional, degrees of freedom). All of it is invariant to rescaling the weights; unit weights give the unweighted belt unchanged.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
confidence	The two (low, high) confidence levels for the bands, e.g. (0.8, 0.95). TYPE: tuple of float, keyword-only DEFAULT: (0.8, 0.95)
grid_size	Number of evaluation points, spanning the 0.5th to 99.5th percentile of p. TYPE: (int, keyword - only) DEFAULT: 100
sample_weight	Optional positive weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
BeltResult	Grid coordinates, the band bands[level] = (lower, upper) for each of the two levels, the selected polynomial degree, and the associated calibration-test p-value.

Source code in src/probcal/curves.py

def calibration_belt(
    y: object,
    p: object,
    *,
    confidence: tuple[float, float] = (0.8, 0.95),
    grid_size: int = 100,
    sample_weight: object = None,
) -> BeltResult:
    """GiViTI-style calibration belt (Nattino et al., 2014, 2017).

    Fits a polynomial logistic recalibration of the outcome on
    ``logit(p)``, selecting the degree by forward likelihood-ratio testing
    (p < 0.05 to add a term, capped at degree 4). The polynomial is fitted on
    the centred and scaled logit (same model, better-conditioned design).

    Both outputs are approximations of the published construction:

    * **Belt.** At each confidence level the band is the image of the Wald
      (information-matrix) ellipsoid with radius ``chi2_ppf(level, degree +
      1)`` — a quadratic approximation of the likelihood-ratio confidence
      region the papers invert. Using the ``degree + 1``-df radius makes it
      a *simultaneous* (Scheffé-type) band over the grid for the selected
      polynomial, not a pointwise one; it conditions on the selected degree.
    * **p-value.** The likelihood-ratio test of the selected polynomial
      against the identity, referred to ``chi2(degree + 1)``; it ignores the
      forward selection of the degree, whose exact null distribution
      (Nattino et al., 2014) puts more mass in the upper tail, so the
      p-value is anti-conservative.

    Where the band excludes the diagonal, the data reject calibration in
    that region. With non-uniform weights (package convention,
    :mod:`probcal.metrics._common`) the fits maximise the weighted
    likelihood, the band uses the sandwich covariance ``I**-1 J I**-1``
    with ``J = X' diag(w**2 mu (1 - mu)) X`` (the exact variance of the
    weighted score under the fitted model), the forward-selection steps
    divide each 1-df LR statistic by the added coefficient's design effect
    (sandwich over model-based variance), and the p-value is second-order
    Rao-Scott corrected (:func:`probcal.metrics._common.rao_scott`, its
    reference distribution ``chi2`` with the corrected, possibly fractional,
    degrees of freedom). All of it is invariant to rescaling the weights;
    unit weights give the unweighted belt unchanged.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    confidence : tuple of float, keyword-only
        The two (low, high) confidence levels for the bands, e.g. ``(0.8,
        0.95)``.
    grid_size : int, keyword-only
        Number of evaluation points, spanning the 0.5th to 99.5th percentile
        of ``p``.
    sample_weight : array_like or None, keyword-only
        Optional positive weights, same length as ``y``.

    Returns
    -------
    BeltResult
        Grid coordinates, the band ``bands[level] = (lower, upper)`` for
        each of the two ``levels``, the selected polynomial degree, and the
        associated calibration-test p-value.
    """
    levels = tuple(float(c) for c in confidence)
    if len(levels) != 2 or not all(0.0 < c < 1.0 for c in levels):
        raise ValueError("confidence must be two levels in (0, 1)")
    grid_size = validate_positive_int(grid_size, "grid_size")
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    # Mean-one weights (unit weights unchanged) keep the ridge and the fit
    # tolerances on the same scale whatever the weights' scale.
    w = w / float(np.mean(w))
    uniform = is_uniform(w)
    z = logit(p_arr)
    centre = float(np.average(z, weights=w))
    spread = float(np.sqrt(np.average((z - centre) ** 2, weights=w)))
    if not spread > 0.0:
        spread = 1.0

    def design(deg: int, zz: np.ndarray) -> np.ndarray:
        t = (zz - centre) / spread
        return np.column_stack([t**k for k in range(deg + 1)])

    def information_inv(X: np.ndarray, mu: np.ndarray) -> np.ndarray:
        """Inverse of the (ridged) model-based information."""
        return np.linalg.inv(gram(X, w * mu * (1.0 - mu)) + 1e-10 * np.eye(X.shape[1]))

    def covariance(X: np.ndarray, mu: np.ndarray) -> np.ndarray:
        """Coefficient covariance: ``I**-1`` scaled to unit weights, or the sandwich."""
        info_inv = information_inv(X, mu)
        if uniform:
            return info_inv * float(w[0])
        return info_inv @ gram(X, w * w * mu * (1.0 - mu)) @ info_inv

    def step_stat(lr: float, X: np.ndarray, mu: np.ndarray) -> float:
        """1-df LR statistic for the last column, divided by its design effect."""
        if uniform:
            return lr / float(w[0])
        deff = covariance(X, mu)[-1, -1] / information_inv(X, mu)[-1, -1]
        return lr / deff if deff > 0.0 else lr

    # Forward LR selection of the polynomial degree. A separated fit's
    # coefficients come from the ridge fallback: usable as a terminal fit,
    # never a basis for extension.
    degree = 1
    fit = irls_logistic(design(1, z), y_arr, w=w)
    ll = bern_loglik(y_arr, expit(design(1, z) @ fit.beta), w)
    while degree < 4 and not fit.separation:
        X_cand = design(degree + 1, z)
        cand = irls_logistic(X_cand, y_arr, w=w)
        if cand.separation:
            break
        mu_cand = expit(X_cand @ cand.beta)
        ll_cand = bern_loglik(y_arr, mu_cand, w)
        if chi2_sf(step_stat(max(2.0 * (ll_cand - ll), 0.0), X_cand, mu_cand), 1.0) >= 0.05:
            break
        degree += 1
        fit, ll = cand, ll_cand

    # Associated calibration test: fitted polynomial vs the identity map.
    df = degree + 1
    X = design(degree, z)
    stat, df_rs = rao_scott(max(2.0 * (ll - bern_loglik(y_arr, p_arr, w)), 0.0), X, p_arr, w)
    p_value = chi2_sf(stat, df_rs)

    # Bands from the (sandwich) covariance ellipsoid.
    mu = expit(np.clip(X @ fit.beta, -30.0, 30.0))
    info_inv = covariance(X, mu)
    grid_p = _grid(p_arr, grid_size)
    grid_z = logit(grid_p)
    Xg = design(degree, grid_z)
    eta = Xg @ fit.beta
    se_sq = np.einsum("ij,jk,ik->i", Xg, info_inv, Xg)
    bands: dict[float, tuple[np.ndarray, np.ndarray]] = {}
    for conf in levels:
        radius = np.sqrt(chi2_ppf(conf, float(df)) * se_sq)
        bands[conf] = (expit(eta - radius), expit(eta + radius))
    return BeltResult(
        grid_p=grid_p,
        grid_logit=grid_z,
        levels=(levels[0], levels[1]),
        bands=bands,
        degree=degree,
        p_value=p_value,
    )

plots

Matplotlib plotting helpers (requires the [viz] extra; import-guarded).

All computation lives in probcal.curves and probcal.metrics; this module only renders. The logit-scale views are the flagship for low-PD portfolios: axis ticks sit at logit positions but are labeled in probabilities, so the low-probability region stays readable. Styling is applied per call via rc_context — global rcParams are never touched — and the axes-level part of it (spines, grid) is applied explicitly as well, so a caller-supplied ax is styled the same as one created here. Figures this module creates are made through pyplot (so plt.show() and notebook display work) and are therefore registered in pyplot's figure manager; close them with plt.close(fig) when rendering many in a loop. A call that raises closes any figure it created before failing. scale is "probability" (the default) or "logit"; anything else raises ValueError. Theory: docs/concepts/visualization.md.

plot_attributes

plot_attributes(y: object, p: object, *, method: str = 'binned', n_bins: int = 10, scale: str = 'probability', sample_weight: object = None, ax: Any = None) -> Any

Attributes diagram: reliability curve against climatology and no-skill references.

Draws the classic Hsu & Murphy (1986) attributes diagram: the identity (perfect calibration), the horizontal and vertical climatology references at the weighted base rate :math:\bar y, the no-skill line :math:y = (x + \bar y) / 2 (equidistant between the climatology level and the identity), and a light shading of the region where a point beats climatology, :math:(y - x)^2 \le (x - \bar y)^2 — i.e. where the point sits closer to the identity than the horizontal no-resolution line, the geometric criterion for positive Brier skill. The reliability curve is drawn on top: method="binned" overlays :func:probcal.curves.reliability_binned as markers sized by bin count; method="corp" overlays the PAV step fit from :func:probcal.curves.corp_reliability (bands=None), the same convention as :func:probcal.plots.plot_corp. scale="logit" transforms every drawn quantity (clipped to [1e-12, 1 - 1e-12]) through :func:probcal._math.logit and relabels the axes in probabilities.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
method	Reliability construction to overlay. TYPE: (binned, corp) DEFAULT: "binned"
n_bins	Bin count passed to :func:probcal.curves.reliability_binned (method="binned" only). TYPE: (int, keyword - only) DEFAULT: 10
scale	Axis scale; "logit" stretches the low-probability region. TYPE: (probability, logit) DEFAULT: "probability"
sample_weight	Optional non-negative weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the diagram was drawn on.

RAISES	DESCRIPTION
ValueError	If method is not "binned" or "corp".

Notes

The identity, climatology, no-skill, and shading layers are evaluated on a fixed 400-point grid over [0, 1] — a plotting-fidelity choice (dense enough to render as smooth curves at the figure's default figsize=(6.5, 6)), not a statistical one; it does not depend on n_bins or the data.

Examples:

>>> import numpy as np
>>> from probcal.plots import plot_attributes
>>> rng = np.random.default_rng(0)
>>> p = rng.uniform(0.05, 0.5, 300)
>>> y = (rng.random(300) < p).astype(float)
>>> ax = plot_attributes(y, p, method="corp")

Source code in src/probcal/_plots_diag.py

def plot_attributes(
    y: object,
    p: object,
    *,
    method: str = "binned",
    n_bins: int = 10,
    scale: str = "probability",
    sample_weight: object = None,
    ax: Any = None,
) -> Any:
    """Attributes diagram: reliability curve against climatology and no-skill references.

    Draws the classic Hsu & Murphy (1986) attributes diagram: the identity
    (perfect calibration), the horizontal and vertical climatology
    references at the weighted base rate :math:`\\bar y`, the no-skill line
    :math:`y = (x + \\bar y) / 2` (equidistant between the climatology
    level and the identity), and a light shading of the region where a
    point beats climatology, :math:`(y - x)^2 \\le (x - \\bar y)^2` — i.e.
    where the point sits closer to the identity than the horizontal
    no-resolution line, the geometric criterion for positive Brier skill.
    The reliability curve is drawn on top: ``method="binned"`` overlays
    :func:`probcal.curves.reliability_binned` as markers sized by bin
    count; ``method="corp"`` overlays the PAV step fit from
    :func:`probcal.curves.corp_reliability` (``bands=None``), the same
    convention as :func:`probcal.plots.plot_corp`. ``scale="logit"``
    transforms every drawn quantity (clipped to ``[1e-12, 1 - 1e-12]``)
    through :func:`probcal._math.logit` and relabels the axes in
    probabilities.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    method : {"binned", "corp"}, keyword-only
        Reliability construction to overlay.
    n_bins : int, keyword-only
        Bin count passed to :func:`probcal.curves.reliability_binned`
        (``method="binned"`` only).
    scale : {"probability", "logit"}, keyword-only
        Axis scale; ``"logit"`` stretches the low-probability region.
    sample_weight : array_like or None, keyword-only
        Optional non-negative weights, same length as ``y``.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the diagram was drawn on.

    Raises
    ------
    ValueError
        If ``method`` is not ``"binned"`` or ``"corp"``.

    Notes
    -----
    The identity, climatology, no-skill, and shading layers are evaluated on
    a fixed 400-point grid over ``[0, 1]`` — a plotting-fidelity choice
    (dense enough to render as smooth curves at the figure's default
    ``figsize=(6.5, 6)``), not a statistical one; it does not depend on
    ``n_bins`` or the data.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.plots import plot_attributes
    >>> rng = np.random.default_rng(0)
    >>> p = rng.uniform(0.05, 0.5, 300)
    >>> y = (rng.random(300) < p).astype(float)
    >>> ax = plot_attributes(y, p, method="corp")  # doctest: +SKIP
    """
    _require_mpl()
    _check_scale(scale)
    if method not in ("binned", "corp"):
        raise ValueError('method must be "binned" or "corp"')
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    ybar = float(np.average(y_arr, weights=w))

    def _t(x: Any) -> np.ndarray:
        return _tr(x, scale)

    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))

        grid = np.linspace(0.0, 1.0, 400)
        ybar_t = float(_t([ybar])[0])
        ax.plot(_t(grid), _t(grid), ls="--", color=_GREY, lw=1, label="identity")
        ax.axhline(ybar_t, color=_GREY, ls=":", lw=1, label="climatology / no resolution")
        ax.axvline(ybar_t, color=_GREY, ls=":", lw=1, label="climatology")
        no_skill = (grid + ybar) / 2.0
        ax.plot(_t(grid), _t(no_skill), color=_GREY, ls="-.", lw=1, label="no skill")

        half_width = np.abs(grid - ybar)
        y_upper = np.clip(grid + half_width, 0.0, 1.0)
        y_lower = np.clip(grid - half_width, 0.0, 1.0)
        ax.fill_between(_t(grid), _t(y_lower), _t(y_upper), color=_GREEN, alpha=0.08)

        if method == "binned":
            curve = reliability_binned(y_arr, p_arr, n_bins=n_bins, sample_weight=w)
            sizes = 15.0 + 200.0 * curve.count / curve.count.max()
            ax.scatter(
                _t(curve.pred_mean),
                _t(curve.event_rate),
                s=sizes,
                color=_BLUE,
                zorder=5,
                label="binned",
            )
        else:
            result = corp_reliability(y_arr, p_arr, sample_weight=w, bands=None)
            x_edges, y_levels = _pav_step_xy(result.block_lo, result.block_hi, result.block_level)
            ax.step(_t(x_edges), _t(y_levels), where="post", color=_BLUE, lw=2, label="PAV fit")

        _scale_axis_labels(ax, scale, "predicted probability", "observed event rate")

        ax.set_title("attributes diagram")
        ax.legend(loc="lower right")
        return ax

plot_corp

plot_corp(result: CorpResult, *, scale: str = 'probability', show_decomposition: bool = True, ax: Any = None) -> Any

CORP reliability diagram: PAV step fit, resampled bands, and the Brier decomposition.

Draws the PAV-recalibrated step function (each block's [block_lo, block_hi] at its block_level, joined vertically between consecutive blocks) against the identity, with the resampled bands from :func:probcal.curves.corp_reliability shaded around it. Grey tick marks along the x-axis show each PAV block's centre, scaled to its weight share of the portfolio. scale="logit" clips edges to [1e-12, 1 - 1e-12] before the logit transform and stretches the low-probability region — the recommended view for PD portfolios.

PARAMETER	DESCRIPTION
result	Result of :func:probcal.curves.corp_reliability. TYPE: CorpResult
scale	Axis scale; "logit" stretches the low-probability region. TYPE: (probability, logit) DEFAULT: "probability"
show_decomposition	If True (default), draw the Brier/MCB/DSC/UNC decomposition box. TYPE: (bool, keyword - only) DEFAULT: True
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the diagram was drawn on.

Examples:

>>> import numpy as np
>>> from probcal.curves import corp_reliability
>>> from probcal.plots import plot_corp
>>> rng = np.random.default_rng(0)
>>> p = rng.uniform(0.05, 0.5, 300)
>>> y = (rng.random(300) < p).astype(float)
>>> ax = plot_corp(corp_reliability(y, p, n_resamples=20))

Source code in src/probcal/_plots_diag.py

def plot_corp(
    result: CorpResult,
    *,
    scale: str = "probability",
    show_decomposition: bool = True,
    ax: Any = None,
) -> Any:
    """CORP reliability diagram: PAV step fit, resampled bands, and the Brier decomposition.

    Draws the PAV-recalibrated step function (each block's ``[block_lo,
    block_hi]`` at its ``block_level``, joined vertically between
    consecutive blocks) against the identity, with the resampled bands from
    :func:`probcal.curves.corp_reliability` shaded around it. Grey tick
    marks along the x-axis show each PAV block's centre, scaled to its
    weight share of the portfolio. ``scale="logit"`` clips edges to
    ``[1e-12, 1 - 1e-12]`` before the logit transform and stretches the
    low-probability region — the recommended view for PD portfolios.

    Parameters
    ----------
    result : CorpResult
        Result of :func:`probcal.curves.corp_reliability`.
    scale : {"probability", "logit"}, keyword-only
        Axis scale; ``"logit"`` stretches the low-probability region.
    show_decomposition : bool, keyword-only
        If ``True`` (default), draw the Brier/MCB/DSC/UNC decomposition box.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the diagram was drawn on.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.curves import corp_reliability
    >>> from probcal.plots import plot_corp
    >>> rng = np.random.default_rng(0)
    >>> p = rng.uniform(0.05, 0.5, 300)
    >>> y = (rng.random(300) < p).astype(float)
    >>> ax = plot_corp(corp_reliability(y, p, n_resamples=20))  # doctest: +SKIP
    """
    _require_mpl()
    _check_scale(scale)
    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))
        lo, hi, level, weight = (
            result.block_lo,
            result.block_hi,
            result.block_level,
            result.block_weight,
        )
        domain = _tr([lo[0], hi[-1]], scale)
        ax.plot(domain, domain, ls="--", c=_GREY, lw=1, label="identity")

        if len(result.band_grid) > 0:
            ax.fill_between(
                _tr(result.band_grid, scale),
                _tr(result.band_low, scale),
                _tr(result.band_high, scale),
                color=_BLUE,
                alpha=0.15,
                label=f"{result.level:.0%} {result.bands} band",
            )

        # Each block contributes [lo, hi] at its level; steps-post joins
        # consecutive blocks with a vertical segment at the right edge.
        x_edges, y_levels = _pav_step_xy(lo, hi, level)
        ax.step(
            _tr(x_edges, scale), _tr(y_levels, scale), where="post", color=_BLUE, lw=2,
            label="PAV fit",
        )  # fmt: skip

        centres = _tr((lo + hi) / 2.0, scale)
        heights = 0.08 * weight / weight.max()
        ax.vlines(centres, 0.0, heights, color=_GREY, alpha=0.6, transform=ax.get_xaxis_transform())

        _scale_axis_labels(ax, scale, "predicted probability", "PAV-recalibrated probability")

        if show_decomposition:
            txt = (
                f"Brier {result.brier:.4f}\n"
                f"MCB {result.brier_mcb:.4f}\n"
                f"DSC {result.brier_dsc:.4f}\n"
                f"UNC {result.brier_unc:.4f}"
            )
            _text_box(ax, txt)

        ax.set_title("CORP reliability diagram")
        ax.legend(loc="lower right")
        return ax

plot_mcb_dsc

plot_mcb_dsc(candidates: Mapping[str, tuple[Any, Any]] | SelectionReport, *, score: str = 'brier', ax: Any = None) -> Any

MCB-DSC plane: CORP miscalibration vs. discrimination, one point per candidate.

Each candidate is a point at (DSC, MCB) from its CORP decomposition (:func:probcal.curves.corp_reliability). Dashed grey iso-score diagonals trace MCB = DSC + (S̄ - UNC) for five values of the mean score S̄ spaced between the candidates' min and max — candidates on the same diagonal tie on score despite different miscalibration/ discrimination splits, so the plane separates "worse calibrated" from "less discriminating" for two methods that score the same. Lower-right is better: more discrimination (DSC) for no more miscalibration (MCB).

PARAMETER	DESCRIPTION
candidates	Either a {name: (y, p)} mapping — each entry's CORP decomposition is computed fresh via corp_reliability(y, p, bands=None), and every entry's y must share the same weighted mean to 1e-12 (UNC, and therefore the iso-score diagonals, is only shared across candidates when it is) — or a fitted :class:probcal.selection.CalibratorSelector's report_, whose mcb/dsc/unc columns are plotted directly; score is then informational only, since the report already fixed Brier vs. log loss at selection time. TYPE: mapping of str to (y, p), or SelectionReport
score	Which CORP decomposition to plot for a mapping input. TYPE: (brier, log_loss) DEFAULT: "brier"
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the plane was drawn on.

RAISES	DESCRIPTION
ValueError	If score is not "brier" or "log_loss", or if a mapping's candidates do not share the same weighted mean y.

Examples:

>>> import numpy as np
>>> from probcal.plots import plot_mcb_dsc
>>> rng = np.random.default_rng(0)
>>> p_a = rng.uniform(0.05, 0.5, 300)
>>> y = (rng.random(300) < p_a).astype(float)
>>> p_b = np.clip(p_a * 0.9, 1e-6, 1 - 1e-6)
>>> ax = plot_mcb_dsc({"a": (y, p_a), "b": (y, p_b)})

Source code in src/probcal/_plots_diag.py

def plot_mcb_dsc(
    candidates: "Mapping[str, tuple[Any, Any]] | SelectionReport",
    *,
    score: str = "brier",
    ax: Any = None,
) -> Any:
    """MCB-DSC plane: CORP miscalibration vs. discrimination, one point per candidate.

    Each candidate is a point at ``(DSC, MCB)`` from its CORP decomposition
    (:func:`probcal.curves.corp_reliability`). Dashed grey iso-score
    diagonals trace ``MCB = DSC + (S̄ - UNC)`` for five values of the mean
    score S̄ spaced between the candidates' min and max — candidates on the
    same diagonal tie on ``score`` despite different miscalibration/
    discrimination splits, so the plane separates "worse calibrated" from
    "less discriminating" for two methods that score the same. Lower-right
    is better: more discrimination (DSC) for no more miscalibration (MCB).

    Parameters
    ----------
    candidates : mapping of str to (y, p), or SelectionReport
        Either a ``{name: (y, p)}`` mapping — each entry's CORP
        decomposition is computed fresh via ``corp_reliability(y, p,
        bands=None)``, and every entry's ``y`` must share the same weighted
        mean to ``1e-12`` (UNC, and therefore the iso-score diagonals, is
        only shared across candidates when it is) — or a fitted
        :class:`probcal.selection.CalibratorSelector`'s ``report_``, whose
        ``mcb``/``dsc``/``unc`` columns are plotted directly; ``score`` is
        then informational only, since the report already fixed Brier vs.
        log loss at selection time.
    score : {"brier", "log_loss"}, keyword-only
        Which CORP decomposition to plot for a mapping input.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the plane was drawn on.

    Raises
    ------
    ValueError
        If ``score`` is not ``"brier"`` or ``"log_loss"``, or if a
        mapping's candidates do not share the same weighted mean ``y``.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.plots import plot_mcb_dsc
    >>> rng = np.random.default_rng(0)
    >>> p_a = rng.uniform(0.05, 0.5, 300)
    >>> y = (rng.random(300) < p_a).astype(float)
    >>> p_b = np.clip(p_a * 0.9, 1e-6, 1 - 1e-6)
    >>> ax = plot_mcb_dsc({"a": (y, p_a), "b": (y, p_b)})  # doctest: +SKIP
    """
    _require_mpl()
    if score not in ("brier", "log_loss"):
        raise ValueError('score must be "brier" or "log_loss"')

    if isinstance(candidates, SelectionReport):
        names = list(candidates.methods)
        mcb = np.asarray(candidates.mcb, dtype=np.float64)
        dsc = np.asarray(candidates.dsc, dtype=np.float64)
        unc = float(candidates.unc)
    else:
        names = list(candidates)
        mcb = np.empty(len(names))
        dsc = np.empty(len(names))
        unc = 0.0
        ybar0: float | None = None
        for i, name in enumerate(names):
            y_i, p_i = candidates[name]
            y_arr, p_arr, w_arr = _prep(y_i, p_i, None)
            ybar = float(np.average(y_arr, weights=w_arr))
            if ybar0 is None:
                ybar0 = ybar
            elif abs(ybar - ybar0) > 1e-12:
                raise ValueError(
                    "candidates must share the same weighted mean y "
                    "(UNC would differ across candidates)"
                )
            r = corp_reliability(y_arr, p_arr, sample_weight=w_arr, bands=None)
            mcb[i] = getattr(r, f"{score}_mcb")
            dsc[i] = getattr(r, f"{score}_dsc")
            unc = getattr(r, f"{score}_unc")

    mean_score = mcb - dsc + unc

    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))

        spread = float(dsc.max() - dsc.min())
        pad = spread * 0.15 if spread > 0 else max(0.05, 0.1 * float(dsc.max()))
        x0, x1 = max(0.0, float(dsc.min()) - pad), float(dsc.max()) + pad

        for s_bar in np.linspace(mean_score.min(), mean_score.max(), 5):
            y0, y1 = x0 + (s_bar - unc), x1 + (s_bar - unc)
            ax.plot([x0, x1], [y0, y1], color=_GREY, lw=0.8, ls=":", zorder=1)
            ax.annotate(
                f"S̄={s_bar:.4g}",
                xy=(x1, y1),
                fontsize=8,
                color=_GREY,
                ha="right",
                va="bottom",
            )

        ax.scatter(dsc, mcb, color=_BLUE, zorder=3)
        for name, d, m in zip(names, dsc, mcb, strict=True):
            ax.annotate(name, (d, m), textcoords="offset points", xytext=(5, 5), fontsize=9)

        ax.set_xlim(x0, x1)
        ax.set_xlabel("DSC (discrimination)")
        ax.set_ylabel("MCB (miscalibration)")
        ax.set_title("MCB-DSC plane")
        return ax

plot_murphy

plot_murphy(curves: MurphyCurve | Mapping[str, MurphyCurve] | Mapping[str, tuple[Any, Any]], *, diff: bool = False, n_boot: int = 200, random_state: int = 42, ax: Any = None) -> Any

Murphy diagram: mean elementary score across a threshold grid, or a paired difference.

diff=False (default) draws one line per curve: a single :class:probcal.metrics.MurphyCurve, or a {name: MurphyCurve} mapping with a legend. diff=True instead requires a mapping of exactly two {name: (y, p)} raw-data entries — a MurphyCurve does not retain y/p, so the pointwise difference and its bootstrap band are recomputed from the paired data — and draws S_theta(A) - S_theta(B) (on A's default 513-point threshold grid) with a seeded pointwise bootstrap band (paired-index resampling, 5th/95th percentile) and a zero reference line.

PARAMETER	DESCRIPTION
curves	See above; the last form only when diff=True. TYPE: MurphyCurve, mapping of str to MurphyCurve, or mapping of str to (y, p)
diff	If True, draw the paired difference of the two named curves' elementary scores instead of the curves themselves. TYPE: (bool, keyword - only) DEFAULT: False
n_boot	Bootstrap resamples for the difference band (diff=True only). TYPE: (int, keyword - only) DEFAULT: 200
random_state	Seed for numpy.random.default_rng, used by the bootstrap band. TYPE: (int, keyword - only) DEFAULT: 42
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the diagram was drawn on.

RAISES	DESCRIPTION
ValueError	If diff=True and curves is not a mapping of exactly two raw (y, p) pairs of equal length.

Examples:

>>> import numpy as np
>>> from probcal.metrics import murphy_curve
>>> from probcal.plots import plot_murphy
>>> rng = np.random.default_rng(0)
>>> p = rng.uniform(0.05, 0.5, 300)
>>> y = (rng.random(300) < p).astype(float)
>>> ax = plot_murphy({"model": murphy_curve(y, p)})

Source code in src/probcal/_plots_diag.py

def plot_murphy(
    curves: "MurphyCurve | Mapping[str, MurphyCurve] | Mapping[str, tuple[Any, Any]]",
    *,
    diff: bool = False,
    n_boot: int = 200,
    random_state: int = 42,
    ax: Any = None,
) -> Any:
    """Murphy diagram: mean elementary score across a threshold grid, or a paired difference.

    ``diff=False`` (default) draws one line per curve: a single
    :class:`probcal.metrics.MurphyCurve`, or a ``{name: MurphyCurve}``
    mapping with a legend. ``diff=True`` instead requires a mapping of
    exactly two ``{name: (y, p)}`` raw-data entries — a ``MurphyCurve``
    does not retain ``y``/``p``, so the pointwise difference and its
    bootstrap band are recomputed from the paired data — and draws
    ``S_theta(A) - S_theta(B)`` (on ``A``'s default 513-point threshold
    grid) with a seeded pointwise bootstrap band (paired-index resampling,
    5th/95th percentile) and a zero reference line.

    Parameters
    ----------
    curves : MurphyCurve, mapping of str to MurphyCurve, or mapping of str to (y, p)
        See above; the last form only when ``diff=True``.
    diff : bool, keyword-only
        If ``True``, draw the paired difference of the two named curves'
        elementary scores instead of the curves themselves.
    n_boot : int, keyword-only
        Bootstrap resamples for the difference band (``diff=True`` only).
    random_state : int, keyword-only
        Seed for ``numpy.random.default_rng``, used by the bootstrap band.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the diagram was drawn on.

    Raises
    ------
    ValueError
        If ``diff=True`` and ``curves`` is not a mapping of exactly two
        raw ``(y, p)`` pairs of equal length.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.metrics import murphy_curve
    >>> from probcal.plots import plot_murphy
    >>> rng = np.random.default_rng(0)
    >>> p = rng.uniform(0.05, 0.5, 300)
    >>> y = (rng.random(300) < p).astype(float)
    >>> ax = plot_murphy({"model": murphy_curve(y, p)})  # doctest: +SKIP
    """
    _require_mpl()
    msg = "diff=True needs a mapping of exactly two {name: (y, p)} raw-data pairs"
    # Validate (and compute) everything before a figure exists, so a bad call
    # leaves no orphan figure registered in pyplot.
    if diff:
        if not isinstance(curves, Mapping) or len(curves) != 2:
            raise ValueError(msg)
        items: list[tuple[str, Any]] = list(curves.items())
        (name_a, pair_a), (name_b, pair_b) = items
        if not all(isinstance(pr, tuple) and len(pr) == 2 for pr in (pair_a, pair_b)):
            raise ValueError(msg)
        y_a, p_a, _w_a = _prep(pair_a[0], pair_a[1], None)
        y_b, p_b, _w_b = _prep(pair_b[0], pair_b[1], None)
        if len(y_a) != len(y_b):
            raise ValueError("diff=True needs paired (y, p) of equal length")

        curve_a = murphy_curve(y_a, p_a)
        theta = curve_a.thresholds
        delta = curve_a.score - murphy_curve(y_b, p_b, thresholds=theta).score
        rng = np.random.default_rng(random_state)
        n = len(y_a)
        boot = np.empty((n_boot, len(theta)))
        for b in range(n_boot):
            idx = rng.integers(0, n, n)
            ca = murphy_curve(y_a[idx], p_a[idx], thresholds=theta)
            cb = murphy_curve(y_b[idx], p_b[idx], thresholds=theta)
            boot[b] = ca.score - cb.score
        band_lo = np.percentile(boot, 5, axis=0)
        band_hi = np.percentile(boot, 95, axis=0)
    elif not isinstance(curves, MurphyCurve):
        for curve in curves.values():
            if not isinstance(curve, MurphyCurve):
                raise ValueError(
                    "plot_murphy: mapping values must be MurphyCurve objects; "
                    "pass diff=True with (y, p) pairs for a difference plot"
                )

    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))
        if diff:
            ax.fill_between(
                theta, band_lo, band_hi, color=_BLUE, alpha=0.15, label="90% bootstrap band"
            )
            ax.plot(theta, delta, color=_BLUE, lw=2, label=f"{name_a} - {name_b}")
            ax.axhline(0.0, color=_GREY, lw=1, ls="--", label="zero")
            ax.legend(loc="best")
        elif isinstance(curves, MurphyCurve):
            ax.plot(curves.thresholds, curves.score, color=_BLUE, lw=2)
        else:
            curve_map: Mapping[str, Any] = curves
            for name, curve in curve_map.items():
                ax.plot(curve.thresholds, curve.score, lw=2, label=name)
            ax.legend(loc="best")

        ax.set_xlabel("threshold θ")
        ax.set_ylabel("mean elementary score")
        ax.set_title("Murphy diagram")
        return ax

plot_reliability

plot_reliability(curve: ReliabilityCurve, *, smooth: SmoothReliabilityCurve | KernelReliabilityCurve | None = None, scale: str = 'probability', y: object = None, p: object = None, annotate: bool = True, rug: bool = True, counts: bool = False, ax: Any = None, stats: bool | MetricReport | GroupedMetricReport = False, risk_dist: str | None = 'rug', by: object = None) -> Any

Annotated reliability diagram.

Binned points with Wilson CIs, optional smooth overlay, stats box, and event/non-event risk distribution.

scale="logit" stretches the low-probability region — the recommended view for PD portfolios. Bins whose event rate is exactly 0 or 1 have no finite logit and are omitted from the logit-scale point layer; they remain visible in the risk distribution (or the counts=True margin). Confidence bounds at exactly 0 or 1 are clipped to [1e-6, 1 - 1e-6] before the logit transform.

Passing a :class:probcal.curves.KernelReliabilityCurve (from :func:probcal.curves.reliability_smooth) as smooth renders the density-weighted variable-width curve instead of a plain line: a LineCollection whose width tracks the local prediction density (one width per segment, density[:-1] — the density at the left endpoint of each [grid[i], grid[i+1]] segment, since a LineCollection of len(grid) - 1 segments needs exactly that many widths), the shaded miscalibration area between the curve and the identity, the bootstrap ribbon, and an smECE = ... readout in the bottom-right corner.

Passing the raw y/p enables the stats box and the risk distribution; both are silently skipped when y/p are absent. annotate=True (default) draws the classic stats box, computed by :func:probcal.metrics.reliability_summary. stats=True replaces it with a box reporting n, events, intercept, slope, ICI, smECE, Brier instead (annotate is then ignored); stats=<MetricReport> instead reports name = value [ci_low, ci_high] for whichever of {"intercept", "slope", "ici", "smooth_ece", "brier"} the report carries, plus n/events computed from y.

risk_dist selects the density layer: "rug" (default) draws event/non-event tick marks along the top/bottom edges, deterministically thinned to at most 1000 marks per class; "split" replaces it with a 30-equal-mass-bin spike histogram of p (events up, non-events down, from a y=0.12 baseline in axis-fraction coordinates, heights scaled so the taller class reaches the full 0.12 — axis coordinates cannot go below 0, so both classes share the one baseline); None draws no density layer. rug=False disables the density layer regardless of risk_dist (equivalent to risk_dist=None). counts=True adds the twin-axis count-bar margin, independent of risk_dist.

Passing by switches to a faceted grid: one panel per sorted, stringified group in by (grouped exactly as by :func:probcal.metrics.evaluate's by=, which raises when two distinct labels such as 1 and "1" display alike) plus a leading "pooled" panel, each a fresh :func:probcal.curves.reliability_binned panel built from that group's slice of y/p — the given curve is ignored for the panels (it would otherwise be ambiguous which group it represents). y and p are required in this mode. Every panel honours scale, annotate, stats, rug, risk_dist and counts exactly as the single-panel diagram does (pass annotate=False, rug=False for light panels), with one exception for stats: stats=True computes each panel's box from that panel's data, and a pooled stats=<MetricReport> raises (it cannot describe each group) — pass the :class:probcal.metrics.GroupedMetricReport from evaluate(..., by=...) instead, whose pooled and per-group reports fill the matching panels; sharing x/y limits across the grid; the function then returns the Figure, not an Axes (unlike the by=None default, matching :func:plot_comparison). ax and smooth are not supported with by and raise ValueError. "pooled" is a reserved panel title: a group of your own by that name is indistinguishable from the pooled panel. Group-conditional statistical testing is out of scope here — see docs/guide/groups.md.

PARAMETER	DESCRIPTION
curve	Binned curve, e.g. from :func:probcal.curves.reliability_binned. Ignored when by is given. TYPE: ReliabilityCurve
smooth	Optional smooth overlay, e.g. from :func:probcal.curves.reliability_loess or :func:probcal.curves.reliability_smooth. TYPE: SmoothReliabilityCurve, KernelReliabilityCurve, or None, keyword-only DEFAULT: None
scale	Axis scale; "logit" stretches the low-probability region. TYPE: (probability, logit) DEFAULT: "probability"
y	Raw outcomes and predictions; must be given together (or not at all). Enables the stats box and risk distribution; required when by is given. TYPE: (array_like or None, keyword - only) DEFAULT: None
p	Raw outcomes and predictions; must be given together (or not at all). Enables the stats box and risk distribution; required when by is given. TYPE: (array_like or None, keyword - only) DEFAULT: None
annotate	If True (default) and y/p are given, draw the classic stats box; ignored when stats is truthy. TYPE: (bool, keyword - only) DEFAULT: True
rug	If True (default) and y/p are given, draw the density layer selected by risk_dist. TYPE: (bool, keyword - only) DEFAULT: True
counts	If True, add a twin-axis bar strip of per-bin counts. TYPE: (bool, keyword - only) DEFAULT: False
ax	Axes to draw on; a new figure and axes are created if None. Must be None when by is given (a new figure of panels is always created). TYPE: (Axes or None, keyword - only) DEFAULT: None
stats	If truthy and y/p are given, draw the n, events, intercept, slope, ICI, smECE, Brier stats box (True) or a MetricReport-driven box, replacing annotate's box. With by, a GroupedMetricReport over the same groups (a plain MetricReport raises); without by, not a GroupedMetricReport. TYPE: (bool, MetricReport or GroupedMetricReport, keyword - only) DEFAULT: False
risk_dist	Density-layer style; see above. Anything else raises ValueError. TYPE: (rug, split) DEFAULT: "rug"
by	Optional group labels, one per observation (same length as y); see above. None (default) is the single-panel diagram above, unchanged. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes or Figure	The axes the diagram was drawn on (by=None, the default), or the figure of faceted panels (by given).

RAISES	DESCRIPTION
ValueError	If scale is not "probability" or "logit", y/p are not given together, or risk_dist is not one of "rug", "split", None; if by is given without both y and p, with a length that does not match y, with labels that display alike, together with ax or smooth, or with a stats report that does not match its groups (see above).

Examples:

>>> import numpy as np
>>> from probcal.curves import reliability_binned
>>> from probcal.plots import plot_reliability
>>> rng = np.random.default_rng(0)
>>> p = rng.uniform(0.05, 0.5, 300)
>>> y = (rng.random(300) < p).astype(float)
>>> curve = reliability_binned(y, p, n_bins=10)
>>> ax = plot_reliability(curve, scale="logit", y=y, p=p)
>>> segment = np.where(p < 0.2, "low", "high")
>>> fig = plot_reliability(curve, y=y, p=p, by=segment)

Source code in src/probcal/plots.py

def plot_reliability(
    curve: ReliabilityCurve,
    *,
    smooth: SmoothReliabilityCurve | KernelReliabilityCurve | None = None,
    scale: str = "probability",
    y: object = None,
    p: object = None,
    annotate: bool = True,
    rug: bool = True,
    counts: bool = False,
    ax: Any = None,
    stats: bool | MetricReport | GroupedMetricReport = False,
    risk_dist: str | None = "rug",
    by: object = None,
) -> Any:
    """Annotated reliability diagram.

    Binned points with Wilson CIs, optional smooth overlay, stats box, and
    event/non-event risk distribution.

    ``scale="logit"`` stretches the low-probability region — the recommended
    view for PD portfolios. Bins whose event rate is exactly 0 or 1 have no
    finite logit and are omitted from the logit-scale point layer; they remain
    visible in the risk distribution (or the ``counts=True`` margin).
    Confidence bounds at exactly 0 or 1 are clipped to ``[1e-6, 1 - 1e-6]``
    before the logit transform.

    Passing a :class:`probcal.curves.KernelReliabilityCurve` (from
    :func:`probcal.curves.reliability_smooth`) as ``smooth`` renders the
    density-weighted variable-width curve instead of a plain line: a
    ``LineCollection`` whose width tracks the local prediction density (one
    width per segment, ``density[:-1]`` — the density at the *left* endpoint
    of each ``[grid[i], grid[i+1]]`` segment, since a ``LineCollection`` of
    ``len(grid) - 1`` segments needs exactly that many widths), the shaded
    miscalibration area between the curve and the identity, the bootstrap
    ribbon, and an ``smECE = ...`` readout in the bottom-right corner.

    Passing the raw ``y``/``p`` enables the stats box and the risk
    distribution; both are silently skipped when ``y``/``p`` are absent.
    ``annotate=True`` (default) draws the classic stats box, computed by
    :func:`probcal.metrics.reliability_summary`. ``stats=True`` replaces it
    with a box reporting ``n, events, intercept, slope, ICI, smECE, Brier``
    instead (``annotate`` is then ignored); ``stats=<MetricReport>`` instead
    reports ``name = value [ci_low, ci_high]`` for whichever of
    ``{"intercept", "slope", "ici", "smooth_ece", "brier"}`` the report
    carries, plus ``n``/``events`` computed from ``y``.

    ``risk_dist`` selects the density layer: ``"rug"`` (default) draws
    event/non-event tick marks along the top/bottom edges,
    deterministically thinned to at most 1000 marks per class; ``"split"``
    replaces it with a 30-equal-mass-bin spike histogram of ``p`` (events
    up, non-events down, from a ``y=0.12`` baseline in axis-fraction
    coordinates, heights scaled so the taller class reaches the full 0.12 —
    axis coordinates cannot go below 0, so both classes share the one
    baseline); ``None`` draws no density layer. ``rug=False`` disables the
    density layer regardless of ``risk_dist`` (equivalent to
    ``risk_dist=None``). ``counts=True`` adds the twin-axis count-bar
    margin, independent of ``risk_dist``.

    Passing ``by`` switches to a faceted grid: one panel per sorted,
    stringified group in ``by`` (grouped exactly as by
    :func:`probcal.metrics.evaluate`'s ``by=``, which raises when two distinct
    labels such as ``1`` and ``"1"`` display alike) plus a leading "pooled" panel, each a fresh
    :func:`probcal.curves.reliability_binned` panel built from that group's
    slice of ``y``/``p`` — the given ``curve`` is ignored for the panels
    (it would otherwise be ambiguous which group it represents). ``y`` and
    ``p`` are required in this mode. Every panel honours ``scale``,
    ``annotate``, ``stats``, ``rug``, ``risk_dist`` and ``counts`` exactly as
    the single-panel diagram does (pass ``annotate=False, rug=False`` for
    light panels), with one exception for ``stats``: ``stats=True`` computes
    each panel's box from that panel's data, and a pooled
    ``stats=<MetricReport>`` raises (it cannot describe each group) — pass the
    :class:`probcal.metrics.GroupedMetricReport` from
    ``evaluate(..., by=...)`` instead, whose pooled and per-group reports fill
    the matching panels; sharing x/y limits across the grid; the function then
    returns the **Figure**, not an ``Axes`` (unlike the ``by=None`` default,
    matching :func:`plot_comparison`). ``ax`` and ``smooth`` are not
    supported with ``by`` and raise ``ValueError``.
    ``"pooled"`` is a reserved panel title: a group of your own by that name
    is indistinguishable from the pooled panel. Group-conditional
    statistical *testing* is out of scope here — see
    ``docs/guide/groups.md``.

    Parameters
    ----------
    curve : ReliabilityCurve
        Binned curve, e.g. from :func:`probcal.curves.reliability_binned`.
        Ignored when ``by`` is given.
    smooth : SmoothReliabilityCurve, KernelReliabilityCurve, or None, keyword-only
        Optional smooth overlay, e.g. from
        :func:`probcal.curves.reliability_loess` or
        :func:`probcal.curves.reliability_smooth`.
    scale : {"probability", "logit"}, keyword-only
        Axis scale; ``"logit"`` stretches the low-probability region.
    y, p : array_like or None, keyword-only
        Raw outcomes and predictions; must be given together (or not at all).
        Enables the stats box and risk distribution; required when ``by``
        is given.
    annotate : bool, keyword-only
        If ``True`` (default) and ``y``/``p`` are given, draw the classic
        stats box; ignored when ``stats`` is truthy.
    rug : bool, keyword-only
        If ``True`` (default) and ``y``/``p`` are given, draw the density
        layer selected by ``risk_dist``.
    counts : bool, keyword-only
        If ``True``, add a twin-axis bar strip of per-bin counts.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.
        Must be ``None`` when ``by`` is given (a new figure of panels is
        always created).
    stats : bool, MetricReport or GroupedMetricReport, keyword-only
        If truthy and ``y``/``p`` are given, draw the ``n, events,
        intercept, slope, ICI, smECE, Brier`` stats box (``True``) or a
        ``MetricReport``-driven box, replacing ``annotate``'s box. With
        ``by``, a ``GroupedMetricReport`` over the same groups (a plain
        ``MetricReport`` raises); without ``by``, not a ``GroupedMetricReport``.
    risk_dist : {"rug", "split"} or None, keyword-only
        Density-layer style; see above. Anything else raises ``ValueError``.
    by : array_like or None, keyword-only
        Optional group labels, one per observation (same length as ``y``);
        see above. ``None`` (default) is the single-panel diagram above,
        unchanged.

    Returns
    -------
    matplotlib.axes.Axes or matplotlib.figure.Figure
        The axes the diagram was drawn on (``by=None``, the default), or
        the figure of faceted panels (``by`` given).

    Raises
    ------
    ValueError
        If ``scale`` is not ``"probability"`` or ``"logit"``, ``y``/``p``
        are not given together, or ``risk_dist`` is not one of ``"rug"``,
        ``"split"``, ``None``; if ``by`` is given without both ``y`` and
        ``p``, with a length that does not match ``y``, with labels that
        display alike, together with ``ax`` or ``smooth``, or with a
        ``stats`` report that does not match its groups (see above).

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.curves import reliability_binned
    >>> from probcal.plots import plot_reliability
    >>> rng = np.random.default_rng(0)
    >>> p = rng.uniform(0.05, 0.5, 300)
    >>> y = (rng.random(300) < p).astype(float)
    >>> curve = reliability_binned(y, p, n_bins=10)
    >>> ax = plot_reliability(curve, scale="logit", y=y, p=p)  # doctest: +SKIP
    >>> segment = np.where(p < 0.2, "low", "high")
    >>> fig = plot_reliability(curve, y=y, p=p, by=segment)  # doctest: +SKIP
    """
    _require_mpl()
    _check_scale(scale)
    if (y is None) != (p is None):
        raise ValueError("y and p must be given together")
    if risk_dist not in ("rug", "split", None):
        raise ValueError('risk_dist must be one of "rug", "split", None')
    if by is not None:
        if y is None or p is None:
            raise ValueError("by requires y and p")
        if ax is not None:
            raise ValueError(
                "ax is not supported with by= (a new figure of panels is always created); "
                "draw each group on your own axes with by=None instead"
            )
        if smooth is not None:
            raise ValueError(
                "smooth is not supported with by= (one smooth curve cannot describe every "
                "group); draw each group with by=None and its own smooth curve instead"
            )
        return _plot_reliability_faceted(
            y,
            p,
            by,
            scale=scale,
            annotate=annotate,
            rug=rug,
            counts=counts,
            stats=stats,
            risk_dist=risk_dist,
        )
    if isinstance(stats, GroupedMetricReport):
        raise ValueError(
            "stats=<GroupedMetricReport> needs by= (one panel per group); for a single "
            "panel pass its .pooled report"
        )
    y_arr = None if y is None else np.asarray(y, dtype=np.float64)
    p_arr = None if p is None else np.asarray(p, dtype=np.float64)
    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))
        _draw_reliability(
            ax,
            curve,
            smooth=smooth,
            scale=scale,
            y_arr=y_arr,
            p_arr=p_arr,
            annotate=annotate,
            rug=rug,
            counts=counts,
            stats=stats,
            risk_dist=risk_dist,
        )
        return ax

plot_belt

plot_belt(belt: BeltResult, *, scale: str = 'probability', ax: Any = None) -> Any

GiViTI-style calibration belt with 80/95% bands and the test p-value.

PARAMETER	DESCRIPTION
belt	Result of :func:probcal.curves.calibration_belt. TYPE: BeltResult
scale	Axis scale; "logit" stretches the low-probability region. TYPE: (probability, logit) DEFAULT: "probability"
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the belt was drawn on.

Source code in src/probcal/plots.py

def plot_belt(belt: BeltResult, *, scale: str = "probability", ax: Any = None) -> Any:
    """GiViTI-style calibration belt with 80/95% bands and the test p-value.

    Parameters
    ----------
    belt : BeltResult
        Result of :func:`probcal.curves.calibration_belt`.
    scale : {"probability", "logit"}, keyword-only
        Axis scale; ``"logit"`` stretches the low-probability region.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the belt was drawn on.
    """
    _require_mpl()
    _check_scale(scale)

    def _band(bound: np.ndarray) -> np.ndarray:
        return _tr(bound, scale, _CI_CLIP_EPS)

    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))
        x = belt.grid_logit if scale == "logit" else belt.grid_p
        ax.plot(x, x, ls="--", c=_GREY, lw=1)
        # Widest level first so the narrower band sits on top of it.
        for level, alpha in zip(sorted(belt.levels, reverse=True), (0.2, 0.35), strict=True):
            lo, hi = belt.bands[level]
            ax.fill_between(x, _band(lo), _band(hi), color=_BLUE, alpha=alpha, label=f"{level:.0%}")
        ax.set_title(f"calibration belt (degree {belt.degree}, p = {belt.p_value:.3g})")
        _scale_axis_labels(ax, scale, "predicted probability", "event rate")
        ax.legend(loc="upper left")
        return ax

plot_comparison

plot_comparison(before: ReliabilityCurve, after: ReliabilityCurve, *, scale: str = 'probability', labels: tuple[str, str] = ('before', 'after')) -> Any

Side-by-side reliability diagrams (pre/post calibration or offset).

PARAMETER	DESCRIPTION
before	Binned curves to compare, e.g. raw vs calibrated. TYPE: ReliabilityCurve
after	Binned curves to compare, e.g. raw vs calibrated. TYPE: ReliabilityCurve
scale	Axis scale; "logit" stretches the low-probability region. TYPE: (probability, logit) DEFAULT: "probability"
labels	Panel titles for (before, after). TYPE: tuple of str, keyword-only DEFAULT: ('before', 'after')

RETURNS	DESCRIPTION
Figure	The figure containing both panels.

Source code in src/probcal/plots.py

def plot_comparison(
    before: ReliabilityCurve,
    after: ReliabilityCurve,
    *,
    scale: str = "probability",
    labels: tuple[str, str] = ("before", "after"),
) -> Any:
    """Side-by-side reliability diagrams (pre/post calibration or offset).

    Parameters
    ----------
    before, after : ReliabilityCurve
        Binned curves to compare, e.g. raw vs calibrated.
    scale : {"probability", "logit"}, keyword-only
        Axis scale; ``"logit"`` stretches the low-probability region.
    labels : tuple of str, keyword-only
        Panel titles for ``(before, after)``.

    Returns
    -------
    matplotlib.figure.Figure
        The figure containing both panels.
    """
    _require_mpl()
    _check_scale(scale)
    with _plot_context():
        fig, axes = _plt.subplots(1, 2, figsize=(12, 5.5), sharey=True)
        panel_colors = (_RED, _GREEN)
        for ax, curve, label, color in zip(
            axes, (before, after), labels, panel_colors, strict=True
        ):
            # The binned series takes the panel's before/after color at draw
            # time, so the legend handle matches the points.
            _style_axes(ax)
            _draw_reliability(ax, curve, scale=scale, color=color)
            ax.set_title(label)
        return fig

plot_interval

plot_interval(intervals: ndarray, s: ndarray, *, ax: Any = None) -> Any

Venn–Abers interval widths against the score: where is calibration uncertain?

PARAMETER	DESCRIPTION
intervals	(p0, p1) Venn–Abers interval bounds per score, e.g. from :meth:probcal.vennabers.CrossVennAbersCalibrator.predict_interval. TYPE: numpy.ndarray of shape (n, 2)
s	Scores the intervals are plotted against; need not be sorted (the pairs are drawn in ascending order of s). TYPE: numpy.ndarray of shape (n,)
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the intervals were drawn on.

Source code in src/probcal/plots.py

def plot_interval(intervals: np.ndarray, s: np.ndarray, *, ax: Any = None) -> Any:
    """Venn–Abers interval widths against the score: where is calibration uncertain?

    Parameters
    ----------
    intervals : numpy.ndarray of shape (n, 2)
        ``(p0, p1)`` Venn–Abers interval bounds per score, e.g. from
        :meth:`probcal.vennabers.CrossVennAbersCalibrator.predict_interval`.
    s : numpy.ndarray of shape (n,)
        Scores the intervals are plotted against; need not be sorted (the
        pairs are drawn in ascending order of ``s``).
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the intervals were drawn on.
    """
    _require_mpl()
    with _plot_context():
        ax = _get_axes(ax, (6.5, 4.5))
        s_arr = np.asarray(s, dtype=np.float64)
        order = np.argsort(s_arr, kind="stable")
        s_arr = s_arr[order]
        p0 = np.asarray(intervals, dtype=np.float64)[order, 0]
        p1 = np.asarray(intervals, dtype=np.float64)[order, 1]
        ax.fill_between(s_arr, p0, p1, color=_BLUE, alpha=0.3, label="Venn–Abers interval")
        ax.plot(s_arr, p1 / (1.0 - p0 + p1), lw=1.2, c=_ORANGE, label="scalarized")
        ax.set_xlabel("score")
        ax.set_ylabel("calibrated probability")
        ax.legend(loc="upper left")
        return ax

plot_selection

plot_selection(report: SelectionReport, *, ax: Any = None) -> Any

SelectionReport as a ranked dot plot with fold-spread whiskers.

PARAMETER	DESCRIPTION
report	Result of :meth:probcal.selection.CalibratorSelector.fit, read from its report_ attribute. TYPE: SelectionReport
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the dot plot was drawn on.

Source code in src/probcal/plots.py

def plot_selection(report: SelectionReport, *, ax: Any = None) -> Any:
    """SelectionReport as a ranked dot plot with fold-spread whiskers.

    Parameters
    ----------
    report : SelectionReport
        Result of :meth:`probcal.selection.CalibratorSelector.fit`, read
        from its ``report_`` attribute.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the dot plot was drawn on.
    """
    _require_mpl()
    with _plot_context():
        ax = _get_axes(ax, (6.5, 0.6 * len(report.methods) + 1.5))
        order = np.argsort(report.score_mean)
        ys = np.arange(len(order))
        for rank, i in enumerate(order):
            ok = report.guardrails_ok[i]
            marker = "o" if ok else "x"
            color = _GREEN if report.chosen[i] else (_BLUE if ok else _RED)
            ax.errorbar(
                report.score_mean[i],
                rank,
                xerr=report.score_sd[i],
                fmt=marker,
                color=color,
                capsize=3,
            )
        ax.set_yticks(ys)
        ax.set_yticklabels([report.methods[i] for i in order])
        ax.set_xlabel(report.criterion)
        ax.set_title("calibrator selection (chosen in green; x = guardrail flag)")
        return ax

plot_ecce

plot_ecce(curves: Any, *, labels: Any = None, show_band: bool = True, ax: Any = None) -> Any

ECCE cumulative-drift walk(s) from :func:probcal.curves.ecce_curve.

Accepts a single EcceCurve or a sequence (e.g. raw vs calibrated). The grey envelope (show_band=True, from the first curve) is ±2 pointwise standard deviations under calibration — an aid for reading the walk, NOT a simultaneous confidence band; the formal max-statistic test of Arrieta-Ibarra et al. (2022) is not implemented.

PARAMETER	DESCRIPTION
curves	One or more cumulative-drift walks to overlay. TYPE: EcceCurve or sequence of EcceCurve
labels	Legend labels, aligned with curves; None uses "curve 1", "curve 2", .... TYPE: sequence of str or None, keyword-only DEFAULT: None
show_band	If True (default), draw the ±2 SD envelope from the first curve. TYPE: (bool, keyword - only) DEFAULT: True
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the walk(s) were drawn on.

Source code in src/probcal/plots.py

def plot_ecce(
    curves: Any,
    *,
    labels: Any = None,
    show_band: bool = True,
    ax: Any = None,
) -> Any:
    """ECCE cumulative-drift walk(s) from :func:`probcal.curves.ecce_curve`.

    Accepts a single ``EcceCurve`` or a sequence (e.g. raw vs calibrated).
    The grey envelope (``show_band=True``, from the first curve) is ±2
    *pointwise* standard deviations under calibration — an aid for reading
    the walk, NOT a simultaneous confidence band; the formal max-statistic
    test of Arrieta-Ibarra et al. (2022) is not implemented.

    Parameters
    ----------
    curves : EcceCurve or sequence of EcceCurve
        One or more cumulative-drift walks to overlay.
    labels : sequence of str or None, keyword-only
        Legend labels, aligned with ``curves``; ``None`` uses
        ``"curve 1", "curve 2", ...``.
    show_band : bool, keyword-only
        If ``True`` (default), draw the ±2 SD envelope from the first curve.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the walk(s) were drawn on.
    """
    _require_mpl()
    if isinstance(curves, EcceCurve):
        curves = [curves]
    curves = list(curves)
    if labels is None:
        labels = [f"curve {i + 1}" for i in range(len(curves))]
    palette = [_RED, _GREEN, _BLUE, _ORANGE]
    with _plot_context():
        ax = _get_axes(ax, (7.5, 4.8))
        if show_band:
            c0 = curves[0]
            ax.fill_between(
                c0.frac,
                -2.0 * c0.sd_null,
                2.0 * c0.sd_null,
                color=_GREY,
                alpha=0.3,
                label="±2 SD under calibration (pointwise)",
            )
        ax.axhline(0.0, ls="--", c=_GREY, lw=1)
        for i, (c, label) in enumerate(zip(curves, labels, strict=True)):
            color = palette[i % len(palette)]
            ax.plot(
                c.frac, c.cumdev, lw=1.6, c=color, label=f"{label} (max drift {c.stat_max:.4f})"
            )
            ax.axvline(c.argmax_frac, ls=":", c=color, lw=1, alpha=0.7)
        ax.set_xlabel("cumulative share of portfolio (sorted by prediction)")
        ax.set_ylabel("cumulative deviation")
        ax.legend(loc="best")
        return ax

plot_grade_backtest

plot_grade_backtest(result: Any, *, log_scale: bool = True, ax: Any = None) -> Any

Per-grade traffic-light backtest chart (Jeffreys or exact binomial).

Observed default rates as circles colored by the grade's traffic light, grey 90% display intervals (ci_low/ci_high), and the assigned PDs as wide blue dashes. The intervals are display companions only — the verdict is carried by the lights from the unchanged one-sided tests, so no p-values are printed on the canvas. log_scale=True is the right default for PD grades spanning orders of magnitude.

PARAMETER	DESCRIPTION
result	Per-grade backtest result, from :func:probcal.metrics.binomial_grade_test or :func:probcal.metrics.jeffreys_grade_test. TYPE: BinomialGradeResult or JeffreysGradeResult
log_scale	If True (default), use a log-scale y-axis — the right default for PD grades spanning orders of magnitude. TYPE: (bool, keyword - only) DEFAULT: True
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the backtest chart was drawn on.

Source code in src/probcal/plots.py

def plot_grade_backtest(result: Any, *, log_scale: bool = True, ax: Any = None) -> Any:
    """Per-grade traffic-light backtest chart (Jeffreys or exact binomial).

    Observed default rates as circles colored by the grade's traffic light,
    grey 90% display intervals (``ci_low``/``ci_high``), and the assigned PDs
    as wide blue dashes. The intervals are display companions only — the
    verdict is carried by the lights from the unchanged one-sided tests, so
    no p-values are printed on the canvas. ``log_scale=True`` is the right
    default for PD grades spanning orders of magnitude.

    Parameters
    ----------
    result : BinomialGradeResult or JeffreysGradeResult
        Per-grade backtest result, from
        :func:`probcal.metrics.binomial_grade_test` or
        :func:`probcal.metrics.jeffreys_grade_test`.
    log_scale : bool, keyword-only
        If ``True`` (default), use a log-scale y-axis — the right default
        for PD grades spanning orders of magnitude.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the backtest chart was drawn on.
    """
    _require_mpl()
    light_color = {"green": _GREEN, "yellow": _AMBER, "amber": _AMBER, "red": _RED}
    name = "Jeffreys" if isinstance(result, JeffreysGradeResult) else "exact binomial"
    x = np.arange(len(result.grades))
    rate = result.k / result.n
    colors = [light_color.get(li, _GREY) for li in result.light]
    with _plot_context():
        ax = _get_axes(ax, (1.1 * len(x) + 3.0, 4.8))
        ax.scatter(x, result.pd, marker="_", s=500, c=_BLUE, zorder=2, label="assigned PD")
        ax.errorbar(
            x,
            rate,
            yerr=[np.maximum(rate - result.ci_low, 0.0), np.maximum(result.ci_high - rate, 0.0)],
            fmt="none",
            ecolor=_GREY,
            capsize=4,
            zorder=2,
        )
        ax.scatter(x, rate, s=90, c=colors, edgecolors="white", zorder=3, label="observed rate")
        for i in range(len(x)):
            ax.annotate(
                f"n={int(result.n[i]):,}\nk={int(result.k[i])}",
                xy=(float(x[i]), float(result.ci_high[i])),
                xytext=(0, 5),
                textcoords="offset points",
                ha="center",
                fontsize=8.5,
                color="#666666",
                clip_on=True,
            )
        if log_scale:
            ax.set_yscale("log")
        # Headroom so the n/k labels never collide with the title.
        lo, hi = ax.get_ylim()
        if log_scale:
            ax.set_ylim(lo, hi * (hi / lo) ** 0.12)
        else:
            ax.set_ylim(lo, hi + 0.12 * (hi - lo))
        ax.set_xticks(x)
        ax.set_xticklabels(result.grades)
        ax.set_xlabel("grade")
        ax.set_ylabel("default rate")
        ax.set_title(f"per-grade backtest ({name}, 90% display intervals)")
        ax.legend(loc="upper left")
        return ax

plot_offset_audit

plot_offset_audit(offset: Any, *, ax: Any = None) -> Any

Audit chart for a fitted :class:probcal.offset.LogitOffset stage.

Draws the offset map t -> t + delta on the logit scale against the identity, marks the pre- and post-adjustment central tendencies, and prints the audit numbers read directly from the fitted attributes. This chart audits the stage, not the outcomes — for the before/after guardrail comparison use LogitOffset.audit_report(y, p).

PARAMETER	DESCRIPTION
offset	A fitted :class:probcal.offset.LogitOffset instance. TYPE: LogitOffset
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The axes the audit chart was drawn on.

Source code in src/probcal/plots.py

def plot_offset_audit(offset: Any, *, ax: Any = None) -> Any:
    """Audit chart for a fitted :class:`probcal.offset.LogitOffset` stage.

    Draws the offset map ``t -> t + delta`` on the logit scale against the
    identity, marks the pre- and post-adjustment central tendencies, and
    prints the audit numbers read directly from the fitted attributes. This
    chart audits the *stage*, not the outcomes — for the before/after
    guardrail comparison use ``LogitOffset.audit_report(y, p)``.

    Parameters
    ----------
    offset : LogitOffset
        A fitted :class:`probcal.offset.LogitOffset` instance.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The axes the audit chart was drawn on.
    """
    _require_mpl()
    if not getattr(offset, "fitted_", False):
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    t = np.linspace(*_DEFAULT_LOGIT_SPAN, 200)
    lp = logit1(offset.pre_mean_)
    lq = logit1(offset.post_mean_)
    with _plot_context():
        ax = _get_axes(ax, (6.5, 6))
        ax.plot(t, t, ls="--", c=_GREY, lw=1, label="identity")
        ax.plot(t, t + offset.delta_, c=_BLUE, lw=1.6, label="offset map")
        if offset.target_mean is not None:
            ax.axhline(logit1(offset.target_mean), c=_GREY, lw=0.8, alpha=0.7)
        pre_xy = (lp, lp)
        post_xy = (lq - offset.delta_, lq)
        ax.scatter(*pre_xy, c=_RED, s=60, zorder=3, label="pre mean")
        ax.scatter(*post_xy, c=_GREEN, s=60, zorder=3, label="post mean")
        ax.annotate(
            "", xy=post_xy, xytext=pre_xy, arrowprops={"arrowstyle": "->", "color": "#555555"}
        )
        ax.annotate(
            f"δ = {offset.delta_:+.3f}",
            xy=((pre_xy[0] + post_xy[0]) / 2.0, (pre_xy[1] + post_xy[1]) / 2.0),
            xytext=(8, 0),
            textcoords="offset points",
            fontsize=9,
            color="#555555",
        )
        txt = (
            f"delta = {offset.delta_:+.4f} log-odds\n"
            f"odds factor = {math.exp(offset.delta_):.3f}\n"
            f"pre mean = {offset.pre_mean_:.4%}\n"
            f"post mean = {offset.post_mean_:.4%}\n"
            f"fitted {offset.timestamp_}"
        )
        _text_box(ax, txt)
        _scale_axis_labels(ax, "logit", "input probability", "shifted probability")
        ax.set_title("logit offset audit")
        ax.legend(loc="lower right")
        return ax

plot_e_process

plot_e_process(report: Any, *, grades_panel: bool = False, ax: Any = None) -> Any

Monitoring wealth per component on a log scale, with the 1/alpha line.

PARAMETER	DESCRIPTION
report	Result of :meth:probcal.monitor.CalibrationMonitor.report. TYPE: MonitorReport
grades_panel	Add a second, shorter axes below the main plot showing each grade's offset confidence-sequence band (MonitorStep.grade_delta_ci) across steps. The panel is carved out of the main axes' own area (mpl_toolkits.axes_grid1 divider, shared x-axis), so it stays inside the figure and inside a caller-supplied ax's slot; the step labels then move to the panel. When this function creates the figure, it is made taller (7.5 x 6.5 in) to fit the panel. Additive: the default (False) output is pinned by tests/test_plots_regression.py. TYPE: bool DEFAULT: False
ax	Axes to draw on; a new figure and axes are created if None. TYPE: (Axes or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Axes	The main axes the e-processes were drawn on (unchanged even when grades_panel=True adds a second axes to the same figure).

Source code in src/probcal/plots.py

def plot_e_process(report: Any, *, grades_panel: bool = False, ax: Any = None) -> Any:
    """Monitoring wealth per component on a log scale, with the 1/alpha line.

    Parameters
    ----------
    report : MonitorReport
        Result of :meth:`probcal.monitor.CalibrationMonitor.report`.
    grades_panel : bool, default False
        Add a second, shorter axes below the main plot showing each grade's
        offset confidence-sequence band (``MonitorStep.grade_delta_ci``)
        across steps. The panel is carved out of the main axes' own area
        (``mpl_toolkits.axes_grid1`` divider, shared x-axis), so it stays
        inside the figure and inside a caller-supplied ``ax``'s slot; the
        step labels then move to the panel. When this function creates the
        figure, it is made taller (7.5 x 6.5 in) to fit the panel. Additive:
        the default (``False``) output is pinned by
        ``tests/test_plots_regression.py``.
    ax : matplotlib.axes.Axes or None, keyword-only
        Axes to draw on; a new figure and axes are created if ``None``.

    Returns
    -------
    matplotlib.axes.Axes
        The main axes the e-processes were drawn on (unchanged even when
        ``grades_panel=True`` adds a second axes to the same figure).
    """
    _require_mpl()
    with _plot_context():
        ax = _get_axes(ax, (7.5, 6.5) if grades_panel else (7.5, 4.2))
        steps = report.steps
        x = np.arange(1, len(steps) + 1)
        series = [
            ("global", [s.e_global for s in steps], "black", 2.0),
            ("offset", [s.e_offset for s in steps], _BLUE, 1.4),
            ("shape", [s.e_shape for s in steps], _ORANGE, 1.4),
        ]
        grades = sorted({g for s in steps for g in s.e_grades})
        for g in grades:
            series.append((f"grade {g}", [s.e_grades.get(g, np.nan) for s in steps], _GREY, 1.0))
        for name, values, color, lw in series:
            vals = np.asarray(values, dtype=np.float64)
            if np.all(np.isnan(vals)):
                continue
            ax.plot(x, vals, label=name, color=color, linewidth=lw, marker=".")
        ax.set_yscale("log")
        ax.axhline(1.0 / report.alpha, color=_RED, linestyle="--", linewidth=1.2, label="1/alpha")
        alarm_x = next((i + 1 for i, s in enumerate(steps) if s.alarm), None)
        if alarm_x is not None:
            ax.axvline(alarm_x, color=_RED, linestyle=":", linewidth=1.0)
            ax.annotate(
                f"alarm: {report.alarm_at}",
                xy=(alarm_x, 1.0),
                xytext=(4, 6),
                textcoords="offset points",
                color=_RED,
                fontsize=9,
            )
        ax.axhline(1.0, color=_GREY, linewidth=0.8)
        ax.set_xticks(x)
        ax.set_xticklabels([s.label for s in steps], rotation=45, ha="right", fontsize=8)
        ax.set_ylabel("e-process wealth (log scale)")
        ax.set_title("anytime-valid calibration monitoring")
        ax.legend(loc="upper left", fontsize=9)
        if grades_panel:
            _draw_grades_panel(ax, steps, x)
        return ax

report

Self-contained HTML/markdown validation report.

:func:validation_report assembles a single document — one HTML file with base64-embedded PNG figures, or a markdown file plus a sibling directory of PNGs — out of the diagnostics already computed elsewhere in the package: the reliability diagrams (curves/plots), the metric catalog (metrics.evaluate), the CORP score decomposition, the per-grade Jeffreys/Pluto-Tasche backtests, grouped evaluation, and monitor trajectories. Nothing here computes new statistics; every number and figure is produced by the existing public API and merely rendered into one document for handoff (a model-risk file, an audit trail, a stakeholder readout).

Import cost: this module is stdlib + numpy + probcal at import time — matplotlib is only ever imported lazily, inside the figure-rendering path, so import probcal.report never pulls in the [viz] extra even when it is installed. Calling :func:validation_report does require it (every section renders at least one figure from y/p alone); the ImportError it raises without the extra names probcal[viz].

Determinism: every resampling site (metrics.evaluate, curves.corp_reliability, curves.reliability_smooth) is driven by the single seed keyword, and n_boot sizes all of them at once — the report is bit-reproducible given the same inputs and seed, apart from the one Generated ... UTC timestamp line.

validation_report

validation_report(y: object, p: object, *, calibrator: Any = None, monitor: Any = None, grades: object = None, by: object = None, title: str | None = None, path: str | PathLike[str] | None = None, format: str = 'html', n_boot: int = 200, seed: int = 42) -> str

Self-contained validation report: reliability, metrics, grades, monitoring.

One document — HTML with base64-embedded PNG figures, or markdown with a sibling directory of PNGs — built entirely from the existing public API (curves, metrics, plots, monitor): nothing here computes a new statistic. Sections are omitted, not left blank, when their input is absent: reliability, the metric report, and the CORP decomposition always render (they need only y/p); the rating-grade backtests render only when grades is given, the grouped-evaluation panel only when by is given, the monitoring trajectory only when monitor is given, and the calibrator appendix only when calibrator is given.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
calibrator	A fitted calibrator; adds its fingerprint to the header and an appendix with its serialized state (:meth:BaseCalibrator.to_json) and :meth:BaseCalibrator.interpret messages. TYPE: (BaseCalibrator or None, keyword - only) DEFAULT: None
monitor	Adds its fingerprint to the header and a monitoring section with the e-process trajectory (:func:probcal.plots.plot_e_process) and its report's recommendation/reasoning/onset_label. TYPE: (CalibrationMonitor or None, keyword - only) DEFAULT: None
grades	Rating grade label per observation, or a :class:probcal.Masterscale. Adds a rating-grades section: the Jeffreys per-grade backtest and chart, the Pluto-Tasche most-prudent PD table, and the Jeffreys upper masterscale bands, all ordered best to worst (by mean predicted probability for a label array, by the scale for a masterscale, stated in the section). With a masterscale the section opens with its grade table and the scale's fingerprint joins the header. TYPE: array_like, Masterscale, or None, keyword-only DEFAULT: None
by	Group labels, one per observation. Adds a grouped-evaluation section: the faceted reliability panel (:func:probcal.plots.plot_reliability with by=) plus the pooled-and-per-group metric table. TYPE: (array_like or None, keyword - only) DEFAULT: None
title	Document title; None uses "probcal validation report". TYPE: (str or None, keyword - only) DEFAULT: None
path	When given, the rendered document is written here in addition to being returned. Required when format="markdown" (figures are written to <path stem>_figures/ next to it). TYPE: (path - like or None, keyword - only) DEFAULT: None
format	Output format. "html" embeds every figure as a base64 PNG data URI; "markdown" writes GFM tables and references PNG files written alongside path. TYPE: (html, markdown) DEFAULT: "html"
n_boot	Bootstrap/resample count shared by every resampling site in the report (metrics.evaluate, curves.corp_reliability, curves.reliability_smooth) — one knob to keep the whole report fast (small n_boot) or tighter (large n_boot). TYPE: (int, keyword - only) DEFAULT: 200
seed	RNG seed shared by the same resampling sites; the report is bit-reproducible given the same inputs and seed apart from its one Generated ... UTC timestamp line. TYPE: (int, keyword - only) DEFAULT: 42

RETURNS	DESCRIPTION
str	The rendered document text (also written to path when given).

RAISES	DESCRIPTION
ValueError	If format is not "html" or "markdown", if format="markdown" is given without path, or if y/p fail validation (non-binary y, p outside [0, 1], unequal lengths) — checked once, before anything is rendered.
ImportError	If matplotlib is not installed (every section renders at least one figure); names the probcal[viz] extra.

Examples:

>>> from probcal import make_pd_portfolio
>>> from probcal.report import validation_report
>>> d = make_pd_portfolio(n=500, random_state=0)
>>> html = validation_report(d.y, d.scores, n_boot=20)
>>> "Generated" in html
True

Source code in src/probcal/report.py

def validation_report(
    y: object,
    p: object,
    *,
    calibrator: Any = None,
    monitor: Any = None,
    grades: object = None,
    by: object = None,
    title: "str | None" = None,
    path: "str | os.PathLike[str] | None" = None,
    format: str = "html",
    n_boot: int = 200,
    seed: int = 42,
) -> str:
    """Self-contained validation report: reliability, metrics, grades, monitoring.

    One document — HTML with base64-embedded PNG figures, or markdown with a
    sibling directory of PNGs — built entirely from the existing public API
    (``curves``, ``metrics``, ``plots``, ``monitor``): nothing here computes
    a new statistic. Sections are omitted, not left blank, when their input
    is absent: reliability, the metric report, and the CORP decomposition
    always render (they need only ``y``/``p``); the rating-grade backtests
    render only when ``grades`` is given, the grouped-evaluation panel only
    when ``by`` is given, the monitoring trajectory only when ``monitor`` is
    given, and the calibrator appendix only when ``calibrator`` is given.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    calibrator : BaseCalibrator or None, keyword-only
        A fitted calibrator; adds its fingerprint to the header and an
        appendix with its serialized state (:meth:`BaseCalibrator.to_json`)
        and :meth:`BaseCalibrator.interpret` messages.
    monitor : CalibrationMonitor or None, keyword-only
        Adds its fingerprint to the header and a monitoring section with the
        e-process trajectory (:func:`probcal.plots.plot_e_process`) and its
        report's ``recommendation``/``reasoning``/``onset_label``.
    grades : array_like, Masterscale, or None, keyword-only
        Rating grade label per observation, or a :class:`probcal.Masterscale`.
        Adds a rating-grades section: the Jeffreys per-grade backtest and
        chart, the Pluto-Tasche most-prudent PD table, and the Jeffreys upper
        masterscale bands, all ordered best to worst (by mean predicted
        probability for a label array, by the scale for a masterscale, stated
        in the section). With a masterscale the section opens with its grade
        table and the scale's fingerprint joins the header.
    by : array_like or None, keyword-only
        Group labels, one per observation. Adds a grouped-evaluation
        section: the faceted reliability panel
        (:func:`probcal.plots.plot_reliability` with ``by=``) plus the
        pooled-and-per-group metric table.
    title : str or None, keyword-only
        Document title; ``None`` uses ``"probcal validation report"``.
    path : path-like or None, keyword-only
        When given, the rendered document is written here in addition to
        being returned. Required when ``format="markdown"`` (figures are
        written to ``<path stem>_figures/`` next to it).
    format : {"html", "markdown"}, keyword-only
        Output format. ``"html"`` embeds every figure as a base64 PNG data
        URI; ``"markdown"`` writes GFM tables and references PNG files
        written alongside ``path``.
    n_boot : int, keyword-only
        Bootstrap/resample count shared by every resampling site in the
        report (``metrics.evaluate``, ``curves.corp_reliability``,
        ``curves.reliability_smooth``) — one knob to keep the whole report
        fast (small ``n_boot``) or tighter (large ``n_boot``).
    seed : int, keyword-only
        RNG seed shared by the same resampling sites; the report is
        bit-reproducible given the same inputs and ``seed`` apart from its
        one ``Generated ... UTC`` timestamp line.

    Returns
    -------
    str
        The rendered document text (also written to ``path`` when given).

    Raises
    ------
    ValueError
        If ``format`` is not ``"html"`` or ``"markdown"``, if
        ``format="markdown"`` is given without ``path``, or if ``y``/``p``
        fail validation (non-binary ``y``, ``p`` outside ``[0, 1]``,
        unequal lengths) — checked once, before anything is rendered.
    ImportError
        If matplotlib is not installed (every section renders at least one
        figure); names the ``probcal[viz]`` extra.

    Examples
    --------
    >>> from probcal import make_pd_portfolio
    >>> from probcal.report import validation_report
    >>> d = make_pd_portfolio(n=500, random_state=0)
    >>> html = validation_report(d.y, d.scores, n_boot=20)  # doctest: +SKIP
    >>> "Generated" in html  # doctest: +SKIP
    True
    """
    if format not in ("html", "markdown"):
        raise ValueError(f'format must be "html" or "markdown", got {format!r}')
    if format == "markdown" and path is None:
        raise ValueError('format="markdown" requires path (figures are written next to it)')

    y_arr = validate_binary_y(y)
    p_arr = validate_scores(p, name="p")
    if len(y_arr) != len(p_arr):
        raise ValueError(f"y and p must have equal length, got {len(y_arr)} and {len(p_arr)}")
    sink = _FigureSink(format, path)

    page_title = title if title is not None else "probcal validation report"
    if format == "html":
        page_title = _escape_html(page_title)
    ts = datetime.now(UTC).strftime("%Y-%m-%d %H:%M:%S")
    timestamp = f"Generated {ts} UTC"

    fp_pairs: list[tuple[str, object]] = [("data", data_fingerprint(y_arr, p_arr))]
    if calibrator is not None:
        fp_pairs.append(("calibrator", calibrator.fingerprint()))
    if monitor is not None:
        fp_pairs.append(("monitor", monitor.fingerprint()))
    if isinstance(grades, Masterscale):
        fp_pairs.append(("masterscale", grades.fingerprint()))
    fingerprints = _kv(format, fp_pairs)

    corp = corp_reliability(y_arr, p_arr, n_resamples=n_boot, random_state=seed)
    sections = [
        _section_header(format, y_arr, p_arr),
        _section_reliability(format, y_arr, p_arr, corp, n_boot=n_boot, seed=seed, sink=sink),
        _section_evaluate(format, y_arr, p_arr, n_boot=n_boot, seed=seed),
        _section_corp_decomposition(format, corp),
    ]
    if grades is not None:
        sections.append(_section_grades(format, y_arr, p_arr, grades, sink=sink))
    if by is not None:
        sections.append(
            _section_groups(format, y_arr, p_arr, by, n_boot=n_boot, seed=seed, sink=sink)
        )
    if monitor is not None:
        sections.append(_section_monitor(format, monitor, sink=sink))
    if calibrator is not None:
        sections.append(_section_appendix(format, calibrator))

    template = _HTML_TEMPLATE if format == "html" else _MD_TEMPLATE
    text = template.substitute(
        title=page_title,
        version=__version__,
        timestamp=timestamp,
        fingerprints=fingerprints,
        sections="".join(sections),
    )

    if path is not None:
        with open(path, "w", encoding="utf-8") as fh:
            fh.write(text)
    return text

attribution

SHAP / additive-attribution adjustment to calibrated outputs.

Post-hoc calibration breaks SHAP local accuracy: base + sum(phi) reconstructs the raw score, not the calibrated probability. This module restores additivity on the calibrated scale — exactly for calibrators affine on the logit scale, and by the Aumann–Shapley secant rescaling in general. Theory, identifiability limits, and invariance properties: docs/concepts/shap-calibration.md.

No shap import: plain arrays are accepted, and shap.Explanation objects are duck-typed via their .values / .base_values attributes.

References

Lundberg & Lee (2017); Lundberg et al. (2020); Sundararajan, Taly & Yan (2017); Aumann & Shapley (1974) — full records in the documentation.

AdjustedAttribution dataclass

AdjustedAttribution(phi_adj: ndarray, base_adj: ndarray, target: ndarray, method_used: str, max_reconstruction_error: float)

Bases: _ResultBase

Attributions rescaled to the calibrated output scale.

ATTRIBUTE	DESCRIPTION
phi_adj	Adjusted per-feature attributions. TYPE: numpy.ndarray of shape (n, d)
base_adj	Adjusted base values. TYPE: numpy.ndarray of shape (n,)
target	The calibrated output each row reconstructs (base_adj + phi_adj.sum(axis=1)). TYPE: numpy.ndarray of shape (n,)
method_used	"affine-exact" (exact Shapley values by linearity) or "aumann-shapley" (exact additivity; nonlinearity distributed proportionally to phi). TYPE: str
max_reconstruction_error	max |base_adj + sum(phi_adj) - target| over rows. TYPE: float

adjust_attributions

adjust_attributions(phi: object, base_value: object, calibrator: Any, *, scale: str = 'logit', method: str = 'auto') -> AdjustedAttribution

Rescale additive attributions so they sum to the calibrated output.

PARAMETER	DESCRIPTION
phi	Raw attributions on the model's score scale (log-odds margins for scale="logit", probabilities for scale="probability"). Objects exposing .values and .base_values are duck-typed; base_value is then ignored. TYPE: array_like of shape (n, d) or shap.Explanation-like
base_value	SHAP base value(s) on the same scale as phi. TYPE: float or array_like of shape (n,)
calibrator	Any object with predict_proba on model probabilities; affine_logit_coeffs_ (when not None) enables the exact affine path. A :class:~probcal.CalibratedModel is read through its model-free :attr:~probcal.CalibratedModel.chain_ (its own predict_proba takes features, not scores); a CalibratedScorecard and the sklearn adapters (probcal.sklearn.SklearnCalibrator and CalibratedClassifier) through their fitted probcal calibrator_, which maps the model's probability (for the adapters, the probability for classes_[1]; for method="decision_function", the expit of its margin) — so phi must explain that score. TYPE: fitted calibrator, Chain, or wrapper
scale	Working scale of the attributions. Affine-exactness exists only on the logit scale. TYPE: (logit, probability) DEFAULT: "logit"
method	"auto" uses affine-exact when available, else Aumann–Shapley. TYPE: (auto, affine, aumann - shapley) DEFAULT: "auto"

RETURNS	DESCRIPTION
AdjustedAttribution

RAISES	DESCRIPTION
ValueError	If method="affine" is forced for a calibrator that is not affine on the logit scale (or on the probability scale, where no calibrator is affine).

Source code in src/probcal/attribution.py

def adjust_attributions(
    phi: object,
    base_value: object,
    calibrator: Any,
    *,
    scale: str = "logit",
    method: str = "auto",
) -> AdjustedAttribution:
    """Rescale additive attributions so they sum to the calibrated output.

    Parameters
    ----------
    phi : array_like of shape (n, d) or shap.Explanation-like
        Raw attributions on the model's score scale (log-odds margins for
        ``scale="logit"``, probabilities for ``scale="probability"``).
        Objects exposing ``.values`` and ``.base_values`` are duck-typed;
        ``base_value`` is then ignored.
    base_value : float or array_like of shape (n,)
        SHAP base value(s) on the same scale as ``phi``.
    calibrator : fitted calibrator, Chain, or wrapper
        Any object with ``predict_proba`` on model probabilities;
        ``affine_logit_coeffs_`` (when not None) enables the exact affine
        path. A :class:`~probcal.CalibratedModel` is read through its
        model-free :attr:`~probcal.CalibratedModel.chain_` (its own
        ``predict_proba`` takes features, not scores); a
        ``CalibratedScorecard`` and the sklearn adapters
        (``probcal.sklearn.SklearnCalibrator`` and ``CalibratedClassifier``)
        through their fitted probcal ``calibrator_``, which maps the model's
        probability (for the adapters, the probability for ``classes_[1]``;
        for ``method="decision_function"``, the ``expit`` of its margin) —
        so ``phi`` must explain that score.
    scale : {"logit", "probability"}
        Working scale of the attributions. Affine-exactness exists only on
        the logit scale.
    method : {"auto", "affine", "aumann-shapley"}
        ``"auto"`` uses affine-exact when available, else Aumann–Shapley.

    Returns
    -------
    AdjustedAttribution

    Raises
    ------
    ValueError
        If ``method="affine"`` is forced for a calibrator that is not affine
        on the logit scale (or on the probability scale, where no calibrator
        is affine).
    """
    if scale not in ("logit", "probability"):
        raise ValueError(f"scale must be 'logit' or 'probability', got {scale!r}")
    if method not in ("auto", "affine", "aumann-shapley"):
        raise ValueError(f"unknown method {method!r}")
    if isinstance(calibrator, CalibratorProtocolMixin):
        # Wrappers whose predict_proba takes features, not scores: read the
        # fitted score-level map instead.
        calibrator = calibrator._fitted_calibrator()
    phi_arr, base_arr = _extract(phi, base_value)
    s = base_arr + phi_arr.sum(axis=1)

    coeffs = getattr(calibrator, "affine_logit_coeffs_", None)
    affine_available = scale == "logit" and coeffs is not None
    if method == "affine" and not affine_available:
        raise ValueError(
            "method='affine' requires a calibrator affine on the logit scale "
            "(affine_logit_coeffs_ is None, or scale='probability' was requested)"
        )
    use_affine = affine_available and method in ("auto", "affine")

    if scale == "logit":

        def g_work(t: np.ndarray) -> np.ndarray:
            return logit(calibrator.predict_proba(expit(t)))

    else:

        def g_work(t: np.ndarray) -> np.ndarray:
            return calibrator.predict_proba(np.clip(t, 1e-12, 1.0 - 1e-12))

    if use_affine:
        assert coeffs is not None
        a, b = coeffs
        # Exact on the logit scale: no round trip through predict_proba,
        # whose probability clip would truncate |a * s + b| > logit(1 - 1e-12).
        target = a * s + b
        phi_adj = a * phi_arr
        base_adj = a * base_arr + b
        method_used = "affine-exact"
    else:
        target = g_work(s)
        g_s0 = g_work(base_arr)
        diff = s - base_arr
        multiplier = np.empty(len(s))
        regular = np.abs(diff) >= _DEGENERATE_EPS
        multiplier[regular] = (target[regular] - g_s0[regular]) / diff[regular]
        if np.any(~regular):
            b0 = base_arr[~regular]
            h = _CENTRAL_DIFF_H
            multiplier[~regular] = (g_work(b0 + h) - g_work(b0 - h)) / (2.0 * h)
        phi_adj = phi_arr * multiplier[:, None]
        base_adj = target - phi_adj.sum(axis=1)
        # base_adj equals g(s0) exactly on regular rows (telescoping); the
        # assignment above additionally zeroes reconstruction error on
        # degenerate rows where the local-slope multiplier is approximate.
        method_used = "aumann-shapley"

    recon_err = float(np.max(np.abs(base_adj + phi_adj.sum(axis=1) - target), initial=0.0))
    return AdjustedAttribution(
        phi_adj=phi_adj,
        base_adj=base_adj,
        target=target,
        method_used=method_used,
        max_reconstruction_error=recon_err,
    )

thresholds

Calibrated-to-raw interval and masterscale-band mapping.

Thin functional wrappers over the calibrators' interval_inverse protocol: numpy-only, arrays and floats, no knowledge of any consumer. The canonical rating-grade workflow — masterscale bands defined on calibrated PD, translated once per recalibration into raw-score intervals — is calibrated_bands_to_raw; its output plugs directly into band-style raw targets of a counterfactual engine. build_masterscale runs the other way: it designs the bands from data.

calibrated_interval_to_raw

calibrated_interval_to_raw(calibrator: object, lo: float, hi: float, *, space: str = 'probability', buffer_logit: float = 0.0) -> tuple[float, float]

Translate one calibrated-probability interval into raw-score bounds.

PARAMETER	DESCRIPTION
calibrator	Any object implementing the duck-typed protocol interval_inverse(lo, hi, *, space, buffer_logit) with is_monotone_. TYPE: fitted calibrator
lo	Calibrated bounds; lo=0 / hi=1 map to the full raw range. TYPE: float
hi	Calibrated bounds; lo=0 / hi=1 map to the full raw range. TYPE: float
space	Scale of the returned bounds. TYPE: (probability, logit) DEFAULT: "probability"
buffer_logit	Robustness margin applied in logit space before inversion. TYPE: float DEFAULT: 0.0

RETURNS	DESCRIPTION
tuple of float	(raw_lo, raw_hi) bounds, on the scale requested by space.

Source code in src/probcal/thresholds.py

def calibrated_interval_to_raw(
    calibrator: object,
    lo: float,
    hi: float,
    *,
    space: str = "probability",
    buffer_logit: float = 0.0,
) -> tuple[float, float]:
    """Translate one calibrated-probability interval into raw-score bounds.

    Parameters
    ----------
    calibrator : fitted calibrator
        Any object implementing the duck-typed protocol
        ``interval_inverse(lo, hi, *, space, buffer_logit)`` with
        ``is_monotone_``.
    lo, hi : float
        Calibrated bounds; ``lo=0`` / ``hi=1`` map to the full raw range.
    space : {"probability", "logit"}
        Scale of the returned bounds.
    buffer_logit : float
        Robustness margin applied in logit space before inversion.

    Returns
    -------
    tuple of float
        ``(raw_lo, raw_hi)`` bounds, on the scale requested by ``space``.
    """
    return calibrator.interval_inverse(lo, hi, space=space, buffer_logit=buffer_logit)  # type: ignore[attr-defined]

calibrated_bands_to_raw

calibrated_bands_to_raw(calibrator: object, bands: object, *, space: str = 'probability', buffer_logit: float = 0.0) -> dict

Translate a masterscale {grade: (lo, hi)} on calibrated PD to raw intervals.

Grade edges are policy artifacts that outlive model versions; this translation is what changes when the calibrator is refitted.

The raw intervals follow the masterscale's half-open convention (lo <= p < hi, top band closed): for a non-decreasing fitted map g, band [lo, hi) becomes

[raw_lo, raw_hi) = [inf{s : g(s) >= lo}, inf{s : g(s) >= hi}),

and the top band (the one with the largest hi) becomes the closed [inf{s : g(s) >= lo}, sup{s : g(s) <= hi}]. Adjacent bands therefore share their raw edge and the intervals partition the raw axis: a raw score s lies in the interval of exactly the grade that Masterscale.assign(g(s)) gives it — also for step calibrators (isotonic, histogram binning), where a calibrated edge sitting on a plateau carries positive raw-score mass, which half-open inversion assigns to exactly one grade. For a single interval with closed-bound semantics use :func:calibrated_interval_to_raw.

PARAMETER	DESCRIPTION
calibrator	Any object implementing the duck-typed protocol interval_inverse(lo, hi, *, space, buffer_logit) with is_monotone_. TYPE: fitted calibrator
bands	Mapping of grade label to (lo, hi) calibrated-probability bounds, or a :class:probcal.Masterscale (its bands are read). TYPE: dict or Masterscale
space	Scale of the returned bounds. TYPE: (probability, logit) DEFAULT: "probability"
buffer_logit	Robustness margin applied in logit space before inversion: each band shrinks to [lo', hi') first, so buffered bands do not touch. TYPE: (float, keyword - only) DEFAULT: 0.0

RETURNS	DESCRIPTION
dict	Mapping of grade label to (raw_lo, raw_hi) bounds, on the scale requested by space: raw_lo <= s < raw_hi (<= raw_hi for the top band).

RAISES	DESCRIPTION
UnattainableTargetError	If a band does not intersect the calibrator's output range.

Source code in src/probcal/thresholds.py

def calibrated_bands_to_raw(
    calibrator: object,
    bands: object,
    *,
    space: str = "probability",
    buffer_logit: float = 0.0,
) -> dict:
    """Translate a masterscale ``{grade: (lo, hi)}`` on calibrated PD to raw intervals.

    Grade edges are policy artifacts that outlive model versions; this
    translation is what changes when the calibrator is refitted.

    The raw intervals follow the masterscale's half-open convention
    (``lo <= p < hi``, top band closed): for a non-decreasing fitted map
    ``g``, band ``[lo, hi)`` becomes

    ``[raw_lo, raw_hi) = [inf{s : g(s) >= lo}, inf{s : g(s) >= hi})``,

    and the top band (the one with the largest ``hi``) becomes the closed
    ``[inf{s : g(s) >= lo}, sup{s : g(s) <= hi}]``. Adjacent bands therefore
    share their raw edge and the intervals partition the raw axis: a raw
    score ``s`` lies in the interval of exactly the grade that
    ``Masterscale.assign(g(s))`` gives it — also for step calibrators
    (isotonic, histogram binning), where a calibrated edge sitting on a
    plateau carries positive raw-score mass, which half-open inversion
    assigns to exactly one grade. For a single interval with closed-bound semantics use
    :func:`calibrated_interval_to_raw`.

    Parameters
    ----------
    calibrator : fitted calibrator
        Any object implementing the duck-typed protocol
        ``interval_inverse(lo, hi, *, space, buffer_logit)`` with
        ``is_monotone_``.
    bands : dict or Masterscale
        Mapping of grade label to ``(lo, hi)`` calibrated-probability bounds,
        or a :class:`probcal.Masterscale` (its ``bands`` are read).
    space : {"probability", "logit"}, keyword-only
        Scale of the returned bounds.
    buffer_logit : float, keyword-only
        Robustness margin applied in logit space before inversion: each band
        shrinks to ``[lo', hi')`` first, so buffered bands do not touch.

    Returns
    -------
    dict
        Mapping of grade label to ``(raw_lo, raw_hi)`` bounds, on the scale
        requested by ``space``: ``raw_lo <= s < raw_hi`` (``<= raw_hi`` for
        the top band).

    Raises
    ------
    UnattainableTargetError
        If a band does not intersect the calibrator's output range.
    """
    if isinstance(bands, Masterscale):
        bands = bands.bands
    items = list(bands.items())  # type: ignore[attr-defined]
    top_hi = max(float(hi) for _, (_, hi) in items) if items else 1.0
    inv = calibrator.interval_inverse  # type: ignore[attr-defined]
    out: dict = {}
    for grade, (lo, hi) in items:
        raw_lo, raw_hi = inv(lo, hi, space=space, buffer_logit=buffer_logit)
        if float(hi) < top_hi:
            # Exclusive upper bound: the left inverse of the (buffered) hi.
            hi_b = shrink_interval(lo, hi, buffer_logit)[1]
            try:
                raw_hi = inv(hi_b, 1.0, space=space)[0]
            except UnattainableTargetError:
                pass  # hi above the output range: no score reaches it; keep sup
        out[grade] = (raw_lo, raw_hi)
    return out

build_masterscale

build_masterscale(y: object, p: object, *, n_grades: int, min_count: float = 0, min_events: float = 0, objective: str = 'likelihood', target_shares: object = None, prebins: int = 512, sample_weight: object = None, names: object = None) -> Masterscale

Design a masterscale from data by exact dynamic programming.

p is sorted and cut into prebins equal-mass pre-bins whose boundaries are observed values (ties merge pre-bins). Grades are runs of consecutive pre-bins, so the returned :class:Masterscale carries exact edges and its half-open assign reproduces the partition the optimizer scored. Complexity O(n_grades * prebins^2) after an O(n log n) sort; at the default 512 pre-bins that is well under a second.

PARAMETER	DESCRIPTION
y	Outcomes and calibrated probabilities. TYPE: array_like
p	Outcomes and calibrated probabilities. TYPE: array_like
n_grades	Number of grades (a positive integer). TYPE: (int, keyword - only)
min_count	Floors per grade on the (weighted) observation and event counts. TYPE: (float, keyword - only) DEFAULT: 0
min_events	Floors per grade on the (weighted) observation and event counts. TYPE: (float, keyword - only) DEFAULT: 0
objective	"likelihood" maximizes the grade-level binomial log-likelihood (equivalently, minimizes within-grade PD heterogeneity); "target_shares" minimizes the squared deviation of each grade's share of the sample from target_shares. TYPE: (likelihood, target_shares) DEFAULT: "likelihood"
target_shares	Required with objective="target_shares": one share per grade, best to worst, summing to 1. TYPE: sequence of float or None, keyword-only DEFAULT: None
prebins	Pre-bin count (a positive integer; an upper bound on the number of candidate edges). TYPE: (int, keyword - only) DEFAULT: 512
sample_weight	Weights; counts become weighted sums. TYPE: (array_like or None, keyword - only) DEFAULT: None
names	Grade names best to worst; None gives "G1", "G2", ... TYPE: sequence of str or None, keyword-only DEFAULT: None

RETURNS	DESCRIPTION
Masterscale	With provenance recording the objective, its value, the pre-bin counts, the floors, and which floors bind at the optimum.

RAISES	DESCRIPTION
ValueError	If no partition satisfies the floors (the message names the binding floor), or on an invalid objective, target_shares, names, n_grades, or prebins.

Source code in src/probcal/thresholds.py

def build_masterscale(
    y: object,
    p: object,
    *,
    n_grades: int,
    min_count: float = 0,
    min_events: float = 0,
    objective: str = "likelihood",
    target_shares: object = None,
    prebins: int = 512,
    sample_weight: object = None,
    names: object = None,
) -> Masterscale:
    """Design a masterscale from data by exact dynamic programming.

    ``p`` is sorted and cut into ``prebins`` equal-mass pre-bins whose
    boundaries are observed values (ties merge pre-bins). Grades are runs of
    consecutive pre-bins, so the returned :class:`Masterscale` carries exact
    edges and its half-open ``assign`` reproduces the partition the optimizer
    scored. Complexity ``O(n_grades * prebins^2)`` after an ``O(n log n)``
    sort; at the default 512 pre-bins that is well under a second.

    Parameters
    ----------
    y, p : array_like
        Outcomes and calibrated probabilities.
    n_grades : int, keyword-only
        Number of grades (a positive integer).
    min_count, min_events : float, keyword-only
        Floors per grade on the (weighted) observation and event counts.
    objective : {"likelihood", "target_shares"}, keyword-only
        ``"likelihood"`` maximizes the grade-level binomial log-likelihood
        (equivalently, minimizes within-grade PD heterogeneity);
        ``"target_shares"`` minimizes the squared deviation of each grade's
        share of the sample from ``target_shares``.
    target_shares : sequence of float or None, keyword-only
        Required with ``objective="target_shares"``: one share per grade,
        best to worst, summing to 1.
    prebins : int, keyword-only
        Pre-bin count (a positive integer; an upper bound on the number of
        candidate edges).
    sample_weight : array_like or None, keyword-only
        Weights; counts become weighted sums.
    names : sequence of str or None, keyword-only
        Grade names best to worst; ``None`` gives ``"G1"``, ``"G2"``, ...

    Returns
    -------
    Masterscale
        With ``provenance`` recording the objective, its value, the pre-bin
        counts, the floors, and which floors bind at the optimum.

    Raises
    ------
    ValueError
        If no partition satisfies the floors (the message names the binding
        floor), or on an invalid ``objective``, ``target_shares``, ``names``,
        ``n_grades``, or ``prebins``.
    """
    y_arr = validate_binary_y(y)
    p_arr = _validate_probabilities(p, name="p")
    if len(y_arr) != len(p_arr):
        raise ValueError("y and p must have equal length")
    w_arr = validate_weights(sample_weight, len(p_arr))
    n_grades = validate_positive_int(n_grades, "n_grades")
    prebins = validate_positive_int(prebins, "prebins")
    if objective not in ("likelihood", "target_shares"):
        raise ValueError(f'objective must be "likelihood" or "target_shares", got {objective!r}')
    target: np.ndarray | None = None
    if objective == "target_shares":
        if target_shares is None:
            raise ValueError('objective="target_shares" requires target_shares')
        target = np.asarray(target_shares, dtype=np.float64).reshape(-1)
        if len(target) != n_grades or np.any(target < 0):
            raise ValueError(f"target_shares must hold {n_grades} non-negative shares")
        if abs(float(target.sum()) - 1.0) > 1e-9:
            raise ValueError("target_shares must sum to 1")
    labels = None if names is None else [str(x) for x in list(names)]  # type: ignore[call-overload]
    if labels is not None and len(labels) != n_grades:
        raise ValueError(f"names must have {n_grades} entries, got {len(labels)}")

    W, E, cuts = _prebin(p_arr, y_arr, w_arr, prebins)
    mc, me = float(min_count), float(min_events)
    solved = _solve(W, E, n_grades, objective, target, mc, me)
    if solved is None:
        binding = []
        if _solve(W, E, n_grades, objective, target, mc, 0.0) is not None:
            binding.append("min_events")
        if _solve(W, E, n_grades, objective, target, 0.0, me) is not None:
            binding.append("min_count")
        if not binding:
            binding = (
                ["min_count", "min_events"]
                if (mc or me)
                else ["n_grades (more grades than distinct pre-bins)"]
            )
        raise ValueError(
            f"no {n_grades}-grade partition satisfies the floors; binding: {', '.join(binding)} "
            f"(min_count={min_count}, min_events={min_events}, {len(W)} pre-bins)"
        )
    value, cut_pos = solved
    edges = [float(cuts[j - 1]) for j in cut_pos]
    bounds = list(zip([0, *cut_pos], [*cut_pos, len(W)], strict=True))
    seg_n = np.array([W[a:b].sum() for a, b in bounds])
    seg_e = np.array([E[a:b].sum() for a, b in bounds])
    binding_at_opt = []
    if mc and np.any(np.isclose(seg_n, mc)):
        binding_at_opt.append("min_count")
    if me and np.any(np.isclose(seg_e, me)):
        binding_at_opt.append("min_events")
    provenance = {
        "objective": objective,
        "objective_value": value,
        "prebins": int(prebins),
        "prebins_effective": int(len(W)),
        "n_grades": int(n_grades),
        "min_count": mc,
        "min_events": me,
        "target_shares": None if target is None else [float(x) for x in target],
        "binding": binding_at_opt,
    }
    return Masterscale.from_edges(edges, names=labels, provenance=provenance)

datasets

Synthetic dataset generators (make_pd_portfolio).

PdPortfolio dataclass

PdPortfolio(scores: ndarray, y: ndarray, p_true: ndarray)

Bases: _ResultBase

Synthetic PD portfolio: model scores, outcomes, and the true probabilities.

ATTRIBUTE	DESCRIPTION
scores	The model's reported PDs — miscalibrated unless generated with slope=1, asymmetry=0, intercept=0. TYPE: ndarray
y	Bernoulli outcomes drawn from p_true. TYPE: ndarray
p_true	True conditional probabilities (mean anchored at event_rate when intercept=0). TYPE: ndarray

make_pd_portfolio

make_pd_portfolio(n: int = 5000, *, event_rate: float = 0.03, slope: float = 0.7, intercept: float = 0.0, asymmetry: float = 0.4, score_location: float = -3.2, score_scale: float = 1.1, random_state: int = 42) -> PdPortfolio

Generate a synthetic, controllably miscalibrated PD portfolio.

The model's scores are drawn as s = sigma(N(score_location, score_scale)); the true probability follows the beta-calibration family

logit p_true = a_lo * ln(s) - a_hi * ln(1 - s) + c

with a_lo = slope * (1 + asymmetry) (low-PD tail) and a_hi = slope (high tail), so asymmetry != 0 produces exactly the one-sided tail distortion low-event-rate portfolios exhibit, and BetaCalibrator can recover the generative exponents. c = c_anchor + intercept: c_anchor is solved so that the undistorted-by-intercept portfolio has mean(p_true) == event_rate (unique by monotonicity, via bisection), and intercept is then added on top — a deliberate calibration-in-the-large error of the scores (the true mean moves away from event_rate; the scores do not).

The identity distortion (slope=1, asymmetry=0) anchors the scores instead: their logits are shifted so that mean(scores) == event_rate, and p_true = sigma(logit(scores) + intercept). With intercept=0 too, the scores are exactly calibrated and the portfolio mean is event_rate.

PARAMETER	DESCRIPTION
n	Portfolio size. TYPE: int DEFAULT: 5000
event_rate	Target mean of p_true (the central tendency), ~3% by default. TYPE: float DEFAULT: 0.03
slope	Base exponent of the distortion; < 1 means the model's scores are too spread out (overconfident). TYPE: float DEFAULT: 0.7
intercept	Log-odds shift of p_true applied after the mean anchor: intercept > 0 means the scores underestimate risk (calibration-in-the-large error). It is applied after the anchor because the anchor would otherwise absorb it. TYPE: float DEFAULT: 0.0
asymmetry	Relative extra distortion of the low-PD tail (a_lo/a_hi - 1). TYPE: float DEFAULT: 0.4
score_location	Parameters of the normal generating the score logits. TYPE: float DEFAULT: -3.2
score_scale	Parameters of the normal generating the score logits. TYPE: float DEFAULT: -3.2
random_state	Seed. TYPE: int DEFAULT: 42

RETURNS	DESCRIPTION
PdPortfolio

Source code in src/probcal/datasets.py

def make_pd_portfolio(
    n: int = 5000,
    *,
    event_rate: float = 0.03,
    slope: float = 0.7,
    intercept: float = 0.0,
    asymmetry: float = 0.4,
    score_location: float = -3.2,
    score_scale: float = 1.1,
    random_state: int = 42,
) -> PdPortfolio:
    """Generate a synthetic, controllably miscalibrated PD portfolio.

    The model's scores are drawn as ``s = sigma(N(score_location,
    score_scale))``; the true probability follows the beta-calibration family

    ``logit p_true = a_lo * ln(s) - a_hi * ln(1 - s) + c``

    with ``a_lo = slope * (1 + asymmetry)`` (low-PD tail) and ``a_hi = slope``
    (high tail), so ``asymmetry != 0`` produces exactly the one-sided tail
    distortion low-event-rate portfolios exhibit, and `BetaCalibrator` can
    recover the generative exponents. ``c = c_anchor + intercept``:
    ``c_anchor`` is solved so that the undistorted-by-intercept portfolio
    has ``mean(p_true) == event_rate`` (unique by monotonicity, via
    bisection), and ``intercept`` is then added on top — a deliberate
    calibration-in-the-large error of the scores (the true mean moves away
    from ``event_rate``; the scores do not).

    The identity distortion (``slope=1, asymmetry=0``) anchors the *scores*
    instead: their logits are shifted so that ``mean(scores) ==
    event_rate``, and ``p_true = sigma(logit(scores) + intercept)``. With
    ``intercept=0`` too, the scores are exactly calibrated and the
    portfolio mean is ``event_rate``.

    Parameters
    ----------
    n : int
        Portfolio size.
    event_rate : float
        Target mean of ``p_true`` (the central tendency), ~3% by default.
    slope : float
        Base exponent of the distortion; ``< 1`` means the model's scores are
        too spread out (overconfident).
    intercept : float
        Log-odds shift of ``p_true`` applied *after* the mean anchor:
        ``intercept > 0`` means the scores underestimate risk
        (calibration-in-the-large error). It is applied after the anchor
        because the anchor would otherwise absorb it.
    asymmetry : float
        Relative extra distortion of the low-PD tail (``a_lo/a_hi - 1``).
    score_location, score_scale : float
        Parameters of the normal generating the score logits.
    random_state : int
        Seed.

    Returns
    -------
    PdPortfolio
    """
    if not 0.0 < event_rate < 1.0:
        raise ValueError("event_rate must lie in (0, 1)")
    rng = np.random.default_rng(random_state)
    s = expit(rng.normal(score_location, score_scale, n))
    a_lo = slope * (1.0 + asymmetry)
    a_hi = slope
    identity_case = slope == 1.0 and asymmetry == 0.0
    core = logit(s) if identity_case else a_lo * np.log(s) - a_hi * np.log1p(-s)

    def gap(c: float) -> float:
        return float(np.mean(expit(core + c))) - event_rate

    anchored = core + bisect(gap, -60.0, 60.0, tol=1e-14)
    if identity_case:
        s = expit(anchored)
    p_true = expit(anchored + intercept)
    y = (rng.random(n) < p_true).astype(np.float64)
    return PdPortfolio(scores=s, y=y, p_true=p_true)

chain

Chain: model-free composition of a calibrator with logit-offset stages.

The object a recourse engine inverts after a macro re-offset: recourse must run through offset ∘ calibrator exactly, and every stage stays separately inspectable. CalibratedModel.chain_ builds the equivalent chain for users who fitted through the wrapper and want to hand it on without the model.

Chain

Chain(stages: Sequence[object])

Bases: JsonIO

A calibrator followed by zero or more LogitOffset stages.

Exposes the full calibrator protocol — forward map, exact inverse maps, monotonicity, affine coefficients, interpretation, serialization — for the composed map sigma(logit(g(s)) + delta_1 + ... + delta_m).

Stages may be given fitted (the chain is then immediately usable) or unfitted (the chain must be fitted with :meth:fit before any reading method is called). fit always refits every stage in place, sequentially: the calibrator on (s, y, sample_weight), then each offset in turn on the running calibrated probabilities.

PARAMETER	DESCRIPTION
stages	A :class:~probcal.base.BaseCalibrator first, then zero or more :class:~probcal.offset.LogitOffset stages, in application order. Stored verbatim (as self.stages) when a list is given. TYPE: Sequence

ATTRIBUTE	DESCRIPTION
stages	The stages, in application order, stored verbatim. TYPE: list[object]
calibrator_	Read-only view of stages[0]. TYPE: BaseCalibrator
offsets_	Read-only view of stages[1:]. TYPE: tuple[LogitOffset, ...]
fitted_	True iff every stage is fitted. TYPE: bool
is_monotone_	True iff every stage is monotone (offsets always are). TYPE: bool

Source code in src/probcal/chain.py

def __init__(self, stages: "Sequence[object]") -> None:
    listed = stages if isinstance(stages, list) else list(stages)
    if not listed:
        raise ValueError("Chain needs at least a calibrator stage")
    head, tail = listed[0], listed[1:]
    if not isinstance(head, BaseCalibrator):
        raise ValueError(f"the first stage must be a calibrator, got {type(head).__name__}")
    for off in tail:
        if not isinstance(off, LogitOffset):
            raise ValueError(
                f"every stage after the first must be a LogitOffset, got {type(off).__name__}"
            )
    # Stored verbatim when a list is given: sklearn's clone() constructs
    # with the params it just cloned and then checks identity via get_params.
    self.stages: list[object] = listed
    self.fitted_: bool = all(getattr(st, "fitted_", False) for st in listed)

calibrator_ property

calibrator_: BaseCalibrator

Read-only view of the first stage.

offsets_ property

offsets_: tuple[LogitOffset, ...]

Read-only view of the offset stages, in application order.

delta_ property

delta_: float

Total log-odds shift of the offset stages.

is_monotone_ property

is_monotone_: bool

True iff the calibrator stage is monotone (offsets always are).

affine_logit_coeffs_ property

affine_logit_coeffs_: tuple[float, float] | None

(a, b + sum(delta)) when the calibrator is affine on the logit scale.

fit

fit(s: object, y: object, sample_weight: object = None) -> Chain

Fit every stage sequentially on the same calibration data.

The head calibrator is fitted on (s, y, sample_weight); each offset is then fitted on the running calibrated probabilities, so the offset anchors the calibrator's in-sample output — exactly what CalibratedModel.offset_to does. There is no cross-fitting inside a chain and no automatic MLE offset (estimate_offset remains an explicit choice). fit always refits every stage, including stages that were already fitted at construction; to keep a stage frozen, compose fitted objects and skip fit.

Source code in src/probcal/chain.py

def fit(self, s: object, y: object, sample_weight: object = None) -> "Chain":
    """Fit every stage sequentially on the same calibration data.

    The head calibrator is fitted on ``(s, y, sample_weight)``; each
    offset is then fitted on the running calibrated probabilities, so
    the offset anchors the calibrator's in-sample output — exactly what
    ``CalibratedModel.offset_to`` does. There is no cross-fitting inside
    a chain and no automatic MLE offset (``estimate_offset`` remains an
    explicit choice). ``fit`` always refits every stage, including
    stages that were already fitted at construction; to keep a stage
    frozen, compose fitted objects and skip ``fit``.
    """
    self.calibrator_.fit(s, y, sample_weight=sample_weight)
    p = self.calibrator_.predict_proba(s)
    for off in self.offsets_:
        off.fit(p, y=y, sample_weight=sample_weight)
        p = off.transform(p)
    self.fitted_ = True
    return self

get_params

get_params(deep: bool = True) -> dict[str, object]

Constructor parameters, with stages__i[__param] nesting when deep.

Source code in src/probcal/chain.py

def get_params(self, deep: bool = True) -> dict[str, object]:
    """Constructor parameters, with ``stages__i[__param]`` nesting when ``deep``."""
    params: dict[str, object] = {"stages": self.stages}
    if deep:
        for i, stage in enumerate(self.stages):
            params[f"stages__{i}"] = stage
            for key, value in stage.get_params(deep=True).items():  # type: ignore[attr-defined]
                params[f"stages__{i}__{key}"] = value
    return params

set_params

set_params(**params: object) -> Chain

Set stages wholesale, one stage (stages__i), or a nested stage param.

Stage replacements (stages__i) validate the whole candidate list before anything on the chain changes, so a rejected replacement leaves the chain as it was; a wholesale stages= key is applied first and independently of the indexed keys in the same call. Like the stages' own set_params, setting a nested parameter (stages__i__param) does not clear fitted_; call fit again for the new value to take effect.

Source code in src/probcal/chain.py

def set_params(self, **params: object) -> "Chain":
    """Set ``stages`` wholesale, one stage (``stages__i``), or a nested stage param.

    Stage replacements (``stages__i``) validate the whole candidate
    list before anything on the chain changes, so a rejected
    replacement leaves the chain as it was; a wholesale ``stages=``
    key is applied first and independently of the indexed keys in the
    same call. Like the stages' own ``set_params``, setting a nested
    parameter (``stages__i__param``) does not clear ``fitted_``; call
    ``fit`` again for the new value to take effect.
    """
    if "stages" in params:
        self.__init__(params.pop("stages"))  # type: ignore[misc]
    nested: dict[int, dict[str, object]] = {}
    replacements: dict[int, object] = {}
    for key, value in params.items():
        prefix, _, rest = key.partition("__")
        idx_text, _, sub = rest.partition("__")
        if prefix != "stages" or not idx_text.isdigit():
            raise ValueError(f"invalid parameter {key!r} for Chain")
        idx = int(idx_text)
        if idx >= len(self.stages):
            raise ValueError(
                f"invalid parameter {key!r} for Chain: stage index {idx} is out "
                f"of range for {len(self.stages)} stages"
            )
        if not sub:
            replacements[idx] = value
        else:
            nested.setdefault(idx, {})[sub] = value
    if replacements:
        candidate = list(self.stages)
        for idx, stage in replacements.items():
            candidate[idx] = stage
        self.__init__(candidate)  # type: ignore[misc] # validates before mutating
    for idx, sub_params in nested.items():
        self.stages[idx].set_params(**sub_params)  # type: ignore[attr-defined]
    return self

predict_proba

predict_proba(s: object) -> ndarray

The composed calibrated probability, applied stage by stage.

Source code in src/probcal/chain.py

def predict_proba(self, s: object) -> np.ndarray:
    """The composed calibrated probability, applied stage by stage."""
    self._check_fitted()
    p = self.calibrator_.predict_proba(s)
    for off in self.offsets_:
        p = off.transform(p)
    return p

__sklearn_is_fitted__

__sklearn_is_fitted__() -> bool

Fitted state for sklearn >= 1.6 (True iff every stage is fitted).

Source code in src/probcal/chain.py

def __sklearn_is_fitted__(self) -> bool:
    """Fitted state for sklearn >= 1.6 (``True`` iff every stage is fitted)."""
    return bool(self.fitted_)

interval_inverse

interval_inverse(lo: float, hi: float, *, space: str = 'probability', buffer_logit: float = 0.0) -> tuple[float, float]

Preimage of a calibrated interval through every stage.

The buffer applies to the final calibrated scale, then the bounds travel back through the offsets on the logit scale, then the calibrator's own generalized inverse finishes the job — every refusal (empty buffered interval, unattainable target) is raised by the same doctrine as the underlying stages.

PARAMETER	DESCRIPTION
lo	Calibrated-probability bounds on the chain's output scale. TYPE: float
hi	Calibrated-probability bounds on the chain's output scale. TYPE: float
space	Scale of the returned raw bounds. TYPE: (probability, logit) DEFAULT: "probability"
buffer_logit	Logit-space shrinkage applied before inverting. TYPE: float DEFAULT: 0.0

RETURNS	DESCRIPTION
tuple of float	(raw_lo, raw_hi) on the requested scale.

RAISES	DESCRIPTION
UnattainableTargetError	If the buffered interval is empty or does not intersect the chain's output range.

Source code in src/probcal/chain.py

def interval_inverse(
    self,
    lo: float,
    hi: float,
    *,
    space: str = "probability",
    buffer_logit: float = 0.0,
) -> tuple[float, float]:
    """Preimage of a calibrated interval through every stage.

    The buffer applies to the *final* calibrated scale, then the bounds
    travel back through the offsets on the logit scale, then the
    calibrator's own generalized inverse finishes the job — every
    refusal (empty buffered interval, unattainable target) is raised by
    the same doctrine as the underlying stages.

    Parameters
    ----------
    lo, hi : float
        Calibrated-probability bounds on the chain's output scale.
    space : {"probability", "logit"}
        Scale of the returned raw bounds.
    buffer_logit : float
        Logit-space shrinkage applied before inverting.

    Returns
    -------
    tuple of float
        ``(raw_lo, raw_hi)`` on the requested scale.

    Raises
    ------
    UnattainableTargetError
        If the buffered interval is empty or does not intersect the
        chain's output range.
    """
    self._check_fitted()
    validate_interval(lo, hi)
    validate_space(space)
    lo_b, hi_b = shrink_interval(lo, hi, buffer_logit)
    lo_c = self._shift_bound(lo_b, is_lower=True)
    hi_c = self._shift_bound(hi_b, is_lower=False)
    return self.calibrator_.interval_inverse(lo_c, hi_c, space=space, buffer_logit=0.0)

point_inverse

point_inverse(p: object, *, space: str = 'probability') -> ndarray

Exact preimage of composed calibrated probabilities.

Shifts the targets back through the offsets on the logit scale, then the calibrator's own exact point inverse finishes; the boundary doctrine (strict (0, 1) targets, representable probability-space results) is inherited from the stages.

RAISES	DESCRIPTION
UnattainableTargetError	If a target lies outside (0, 1), is unattainable for the calibrator, or the probability-space result is not representable.

Source code in src/probcal/chain.py

def point_inverse(self, p: object, *, space: str = "probability") -> np.ndarray:
    """Exact preimage of composed calibrated probabilities.

    Shifts the targets back through the offsets on the logit scale, then
    the calibrator's own exact point inverse finishes; the boundary
    doctrine (strict ``(0, 1)`` targets, representable probability-space
    results) is inherited from the stages.

    Raises
    ------
    UnattainableTargetError
        If a target lies outside ``(0, 1)``, is unattainable for the
        calibrator, or the probability-space result is not
        representable.
    """
    self._check_fitted()
    validate_space(space)
    arr = _validate_point_targets(p)
    shifted_z = logit(arr) - self.delta_
    _check_representable(shifted_z, "probability")  # the intermediate must round-trip
    z = self.calibrator_.point_inverse(expit(shifted_z), space="logit")
    _check_representable(np.asarray(z), space)
    return np.asarray(z) if space == "logit" else expit(np.asarray(z))

interpret

interpret() -> Interpretation

Concatenated interpretation of every stage.

Source code in src/probcal/chain.py

def interpret(self) -> Interpretation:
    """Concatenated interpretation of every stage."""
    self._check_fitted()
    parts = [self.calibrator_.interpret()] + [off.interpret() for off in self.offsets_]
    return _concat_interpretations("Chain", parts, qualify=True)

to_dict

to_dict() -> dict[str, object]

Versioned snapshot: the stages' own envelopes, in order.

Source code in src/probcal/chain.py

def to_dict(self) -> dict[str, object]:
    """Versioned snapshot: the stages' own envelopes, in order."""
    self._check_fitted()
    return envelope(
        self,
        params={},
        state={
            "stages": [self.calibrator_.to_dict()] + [off.to_dict() for off in self.offsets_],
        },
        fit_meta={},
    )

from_dict classmethod

from_dict(d: dict) -> Chain

Rebuild the chain by loading every stage through the registry.

Source code in src/probcal/chain.py

@classmethod
def from_dict(cls, d: dict) -> "Chain":
    """Rebuild the chain by loading every stage through the registry."""
    check_payload(cls, d)
    stages = [load(sd) for sd in d["state"]["stages"]]
    return cls(stages)

monitor

Anytime-valid calibration monitoring by e-processes.

Theory, validity conditions, and the simulation verification: docs/concepts/monitoring.md. numpy + stdlib only, like the core.

AppliedAction dataclass

AppliedAction(kind: str, offset: LogitOffset | None, composed: object | None, monitor: CalibrationMonitor | None, window: tuple[str, ...], audit: dict)

Bases: JsonIO

The result of :meth:CalibrationMonitor.apply_recommendation.

ATTRIBUTE	DESCRIPTION
kind	The recommendation :meth:CalibrationMonitor.report produced. TYPE: {re - offset, re - fit, none}
offset	The fitted correction; only for kind="re-offset". TYPE: LogitOffset or None
composed	offset applied to the caller's target (a :class:~probcal.chain.Chain or a :class:~probcal.wrapper.CalibratedModel) -- None when no target was given or kind != "re-offset". TYPE: object or None
monitor	A fresh monitor with the same constructor parameters, ready to watch the corrected pipeline; only for kind="re-offset" (see the "why fresh" note on apply_recommendation). TYPE: CalibrationMonitor or None
window	Batch labels the offset (or the suggested re-fit window) was estimated from; empty when kind="none". TYPE: tuple[str, ...]
audit	Provenance: alarm_at, onset_label, fingerprints of the old and new monitor/offset/target (None where not applicable), and the estimated delta/se (None unless kind="re-offset"). TYPE: dict

Examples:

>>> import numpy as np
>>> from probcal._math import expit, logit
>>> from probcal.datasets import make_pd_portfolio
>>> from probcal.monitor import CalibrationMonitor
>>> mon = CalibrationMonitor(alpha=0.05)
>>> for k in range(6):
...     d = make_pd_portfolio(n=1000, random_state=k)
...     rng = np.random.default_rng(k + 1000)
...     y = (rng.random(1000) < expit(logit(d.scores) + 0.8)).astype(float)
...     _ = mon.update(y, d.scores, label=f"m{k}")
>>> action = mon.apply_recommendation()
>>> action.kind
're-offset'
>>> action.offset.delta_ > 0
True

to_dict

to_dict() -> dict[str, object]

Versioned snapshot; offset/composed/monitor are nested envelopes.

Each nested field is stored via its own to_dict (None stays None): a CalibratedModel composed target stores only a model reference, reattached on load via AppliedAction.from_dict(d, model=...) -- see CalibratedModel.to_dict.

Source code in src/probcal/monitor/_actions.py

def to_dict(self) -> dict[str, object]:
    """Versioned snapshot; ``offset``/``composed``/``monitor`` are nested envelopes.

    Each nested field is stored via its own ``to_dict`` (``None`` stays
    ``None``): a ``CalibratedModel`` composed target stores only a
    model *reference*, reattached on load via
    ``AppliedAction.from_dict(d, model=...)`` -- see
    ``CalibratedModel.to_dict``.
    """
    return envelope(
        self,
        params={},
        state={
            "kind": self.kind,
            "offset": self.offset.to_dict() if self.offset is not None else None,
            "composed": (
                self.composed.to_dict()  # type: ignore[attr-defined]
                if self.composed is not None
                else None
            ),
            "monitor": self.monitor.to_dict() if self.monitor is not None else None,
            "window": list(self.window),
            "audit": dict(self.audit),
        },
        fit_meta={},
    )

from_dict classmethod

from_dict(d: dict, *, model: object = None) -> AppliedAction

Rebuild from :meth:to_dict output (from_json forwards model=).

PARAMETER	DESCRIPTION
d	Output of :meth:to_dict. TYPE: dict
model	Passed through to CalibratedModel.from_dict when composed was a CalibratedModel (only a reference to the base model is serialized, never the model itself). TYPE: (object or None, keyword - only) DEFAULT: None

RAISES	DESCRIPTION
ValueError	If the schema version is unknown or the payload class differs.

Source code in src/probcal/monitor/_actions.py

@classmethod
def from_dict(cls, d: dict, *, model: object = None) -> "AppliedAction":
    """Rebuild from :meth:`to_dict` output (``from_json`` forwards ``model=``).

    Parameters
    ----------
    d : dict
        Output of :meth:`to_dict`.
    model : object or None, keyword-only
        Passed through to ``CalibratedModel.from_dict`` when
        ``composed`` was a ``CalibratedModel`` (only a reference to the
        base model is serialized, never the model itself).

    Raises
    ------
    ValueError
        If the schema version is unknown or the payload class differs.
    """
    check_payload(cls, d)
    st = d["state"]
    offset = LogitOffset.from_dict(st["offset"]) if st["offset"] is not None else None
    composed: object | None = None
    if st["composed"] is not None:
        if st["composed"].get("class") == "CalibratedModel":
            from ..wrapper import CalibratedModel

            composed = CalibratedModel.from_dict(st["composed"], model=model)
        else:
            composed = load(st["composed"])
    monitor = CalibrationMonitor.from_dict(st["monitor"]) if st["monitor"] is not None else None
    return cls(
        kind=st["kind"],
        offset=offset,
        composed=composed,
        monitor=monitor,
        window=tuple(st["window"]),
        audit=dict(st["audit"]),
    )

CalibrationMonitor

CalibrationMonitor(alpha: float = 0.05, components: tuple[str, ...] = ('offset', 'shape'), grades: tuple | None = None, mixture_grid: tuple[float, ...] = (0.1, 0.25, 0.5, 1.0), delta_ci_grid: tuple[float, float, int] = (-3.0, 3.0, 241), min_history: int = 1, plug_in_window: int | None = None, *, recommendation_window: str = 'since_onset')

Bases: JsonIO

Anytime-valid calibration monitoring by e-processes.

Feed matured outcome batches in arrival order; the alarm rule "E >= 1/alpha" has type-I error at most alpha at every stopping time (Ville's inequality), however long monitoring runs. Persist the state between batches with :meth:to_json — never re-run or reorder past batches. Theory: docs/concepts/monitoring.md.

PARAMETER	DESCRIPTION
alpha	Alarm level in (0, 1). TYPE: float DEFAULT: 0.05
components	Which portfolio-level processes drive the global alarm. TYPE: tuple of {"offset", "shape"} DEFAULT: ('offset', 'shape')
grades	The grade universe, fixed for the monitor's lifetime. When given, the per-grade component (the equal-weight average of one offset e-process per declared grade) is part of the global mixture from the first batch on — a grade without data yet contributes e = 1 — and a grade array carrying a label outside this universe raises ValueError. When None, per-grade e-processes and confidence sequences are still tracked and reported for every label seen, but they are not part of the global mixture (a warning says so the first time a grade array is passed). Labels are compared through one normalization: 1, 1.0 and np.int64(1) are the same grade "1". TYPE: tuple or None DEFAULT: None
mixture_grid	Non-empty tuple of positive, finite shifts for the offset mixture (symmetrized to ±). TYPE: tuple of float DEFAULT: (0.1, 0.25, 0.5, 1.0)
delta_ci_grid	Grid of offset nulls for the confidence sequence: linspace(lo, hi, count) with finite lo <= 0 <= hi, lo < hi and an integer count >= 2. TYPE: tuple(lo, hi, count) DEFAULT: (-3.0, 3.0, 241)
min_history	Number of past batches (>= 0) required before the plug-ins engage (before that they are the identity and their factors equal 1). TYPE: int DEFAULT: 1
plug_in_window	Trailing number of past batches (>= 1) used by the plug-ins, and by the recommendation rule when recommendation_window="trailing"; None uses all past batches. Every batch refits the plug-ins on the window, so with None the per-batch cost grows linearly with the history and the total cost of k batches quadratically — set a window for long-running monitors. TYPE: int or None DEFAULT: None
recommendation_window	Which batches feed report()'s trailing-window diagnostics (delta_now, the Cox slope CI, the residual-shape LR) once an alarm has fired. "since_onset" (the default) uses batches from the estimated drift onset (:func:~probcal.monitor._onset. estimate_onset on MonitorStep.log_e_increment) onward — the rationale is that a window starting where the evidence trail actually turns is more informative than one anchored to plug_in_window, which predates any alarm. When plug_in_window is also set, the window starts at the LATER of the two starts (max(onset_idx, n_batches - plug_in_window)), so a short plug_in_window still bounds how far back the since-onset window can reach. "trailing" ignores the onset estimate for those diagnostic INPUTS and uses plug_in_window (or all past batches) instead, unconditionally. onset_label and the onset sentence in reasoning are populated under both modes — only the diagnostic window differs. TYPE: (since_onset, trailing) DEFAULT: "since_onset"

ATTRIBUTE	DESCRIPTION
steps_	The processed batches, in arrival order. TYPE: list[MonitorStep]
masterscale_fingerprint_	Fingerprint of the Masterscale passed to :meth:update, once one has been; serialized with the state. TYPE: str or None

Notes

Anytime validity needs the global mixture's components and weights to be fixed in advance: a grade first seen mid-stream that joined the average with e = 1 would re-weight the existing components, and the global value could then jump toward 1 without any evidence. Declaring grades fixes the universe up front; without it the per-grade processes are diagnostics only.

Source code in src/probcal/monitor/_monitor.py

def __init__(
    self,
    alpha: float = 0.05,
    components: tuple[str, ...] = ("offset", "shape"),
    grades: tuple | None = None,
    mixture_grid: tuple[float, ...] = (0.1, 0.25, 0.5, 1.0),
    delta_ci_grid: tuple[float, float, int] = (-3.0, 3.0, 241),
    min_history: int = 1,
    plug_in_window: int | None = None,
    *,
    recommendation_window: str = "since_onset",
) -> None:
    validate_level(alpha, "alpha")
    unknown = [c for c in components if c not in _COMPONENTS]
    if unknown or not components:
        raise ValueError(
            f"components must be a non-empty subset of {_COMPONENTS}, got {components!r}"
        )
    if recommendation_window not in _RECOMMENDATION_WINDOWS:
        raise ValueError(
            "recommendation_window must be one of "
            f"{_RECOMMENDATION_WINDOWS}, got {recommendation_window!r}"
        )
    validate_nonneg_int(min_history, "min_history")
    if plug_in_window is not None:
        validate_positive_int(plug_in_window, "plug_in_window")
    mixture_grid = tuple(mixture_grid)
    if not mixture_grid or not all(
        isinstance(d, (int, float, np.integer, np.floating))
        and not isinstance(d, (bool, np.bool_))
        and np.isfinite(d)
        and d > 0
        for d in mixture_grid
    ):
        raise ValueError(
            "mixture_grid must be a non-empty tuple of positive finite shifts, "
            f"got {mixture_grid!r}"
        )
    _validate_delta_ci_grid(delta_ci_grid)
    if grades is not None:
        keys = [grade_key(g) for g in grades]
        if not keys:
            raise ValueError("grades must be None or a non-empty tuple of labels")
        if len(set(keys)) != len(keys):
            raise ValueError(
                f"grades contains duplicate labels after normalization: {grades!r}"
            )
    self.alpha = alpha
    self.components = tuple(components)
    self.grades = grades
    self.mixture_grid = mixture_grid
    self.delta_ci_grid = tuple(delta_ci_grid)
    self.min_history = min_history
    self.plug_in_window = plug_in_window
    self.recommendation_window = recommendation_window
    self._init_state()

update

update(y: object, p: object, sample_weight: object = None, grade: object = None, label: str | None = None) -> MonitorStep

Process one matured batch (arrival order is the process order).

PARAMETER	DESCRIPTION
y	Matured binary outcomes in {0, 1} (a one-class batch is legal — quiet months happen). TYPE: array_like
p	The probabilities the deployed forecast assigned to this batch. TYPE: array_like
sample_weight	Positive weights, fixed before the outcomes are seen. In every batch factor they enter as likelihood-ratio exponents min(1, w / mean(w)) (over the observations that factor covers), which keeps each factor an e-value: the anytime-valid guarantee holds exactly for any weights. Rescaling all weights by a constant changes nothing, and constant weights are equivalent to none; skewed weights reduce power (an observation never counts as more than one observation's evidence). The plug-ins are fitted with the same exponents. TYPE: array_like or None DEFAULT: None
grade	Optional per-observation grade labels, or a :class:probcal.Masterscale that assigns them from p; either advances the per-grade offset processes (part of the global alarm only when grades was declared). The first masterscale seen is recorded as masterscale_fingerprint_ and serialized once the batch is accepted; a different one later raises. TYPE: array_like, Masterscale, or None DEFAULT: None
label	Batch label for reporting; defaults to batch-<k>. TYPE: str or None DEFAULT: None

RETURNS	DESCRIPTION
MonitorStep	The running record after this batch.

RAISES	DESCRIPTION
ValueError	On an empty batch, mismatched lengths, invalid values, or a grade label outside the declared grades.

Source code in src/probcal/monitor/_monitor.py

def update(
    self,
    y: object,
    p: object,
    sample_weight: object = None,
    grade: object = None,
    label: str | None = None,
) -> MonitorStep:
    """Process one matured batch (arrival order is the process order).

    Parameters
    ----------
    y : array_like
        Matured binary outcomes in ``{0, 1}`` (a one-class batch is
        legal — quiet months happen).
    p : array_like
        The probabilities the deployed forecast assigned to this batch.
    sample_weight : array_like or None
        Positive weights, fixed before the outcomes are seen. In every
        batch factor they enter as likelihood-ratio exponents
        ``min(1, w / mean(w))`` (over the observations that factor
        covers), which keeps each factor an e-value: the anytime-valid
        guarantee holds exactly for any weights. Rescaling all weights
        by a constant changes nothing, and constant weights are
        equivalent to none; skewed weights reduce power (an
        observation never counts as more than one observation's
        evidence). The plug-ins are fitted with the same exponents.
    grade : array_like, Masterscale, or None
        Optional per-observation grade labels, or a
        :class:`probcal.Masterscale` that assigns them from ``p``;
        either advances the per-grade offset processes (part of the
        global alarm only when ``grades`` was declared). The first
        masterscale seen is recorded as ``masterscale_fingerprint_`` and
        serialized once the batch is accepted; a different one later
        raises.
    label : str or None
        Batch label for reporting; defaults to ``batch-<k>``.

    Returns
    -------
    MonitorStep
        The running record after this batch.

    Raises
    ------
    ValueError
        On an empty batch, mismatched lengths, invalid values, or a
        grade label outside the declared ``grades``.
    """
    # One-class batches are legal: a quiet month can mature with zero events.
    y_arr = validate_binary_y(y, require_both_classes=False)
    p_arr = validate_scores(p, name="p")
    if len(y_arr) != len(p_arr):
        raise ValueError("y and p must have equal length")
    w_arr = validate_weights(sample_weight, len(p_arr))
    self._pending_fingerprint = None
    g_arr = self._grade_array(grade, p_arr) if grade is not None else None
    z = logit(p_arr)

    # Predictable plug-ins: strictly past data only.
    history_ready = len(self._z) >= self.min_history
    if history_ready:
        pz, py, pw = self._past()
        delta_hat = plug_in_delta(pz, py, pw)
        c_hat, a_hat = plug_in_shape(pz, py, pw)
    else:
        delta_hat, (c_hat, a_hat) = 0.0, (0.0, 1.0)

    offset_inc = self._offset.update(z, p_arr, y_arr, w_arr, delta_hat)
    shape_inc = 0.0
    if (c_hat, a_hat) != (0.0, 1.0):  # identity plug-in: factor exactly 1
        shape_inc = bern_log_lr(y_arr, p_arr, expit(c_hat + a_hat * z), exponent_weights(w_arr))
        self._log_shape += shape_inc
    # The onset series sums the plug-in increments of the processes that
    # drive the alarm, and nothing else.
    log_e_increment = 0.0
    if "offset" in self.components:
        log_e_increment += offset_inc
    if "shape" in self.components:
        log_e_increment += shape_inc

    if g_arr is not None:
        grade_inc = 0.0
        for g in (str(x) for x in np.unique(g_arr)):
            if g not in self._grade_procs:
                self._grade_procs[g] = OffsetProcess(self._sym_grid())
            if g not in self._grade_cs:
                self._grade_cs[g] = ConfidenceSequence(self._cs.grid)
            mask = g_arr == g
            if history_ready:
                d_g = plug_in_delta(*self._past(grade=g))
            else:
                d_g = 0.0
            zg, pg, yg, wg = z[mask], p_arr[mask], y_arr[mask], w_arr[mask]
            grade_inc += self._grade_procs[g].update(zg, pg, yg, wg, d_g)
            # Per-grade CS: the global construction on this grade's own
            # slice, with the grade's own plug-in as the alternative.
            self._grade_cs[g].update(zg, pg, yg, wg, d_g)
        if self.grades is not None:
            log_e_increment += grade_inc

    self._cs.update(z, p_arr, y_arr, w_arr, delta_hat)

    # Store the batch AFTER the plug-ins consumed only the past.
    self._z.append(z)
    self._y.append(y_arr)
    self._w.append(w_arr)
    self._g.append(g_arr)

    step = self._make_step(label, y_arr, w_arr, delta_hat, a_hat, log_e_increment)
    self.steps_.append(step)
    if self._pending_fingerprint is not None:
        self.masterscale_fingerprint_ = self._pending_fingerprint
        self._pending_fingerprint = None
    return step

report

report() -> MonitorReport

Trajectory plus the diagnostic re-offset/re-fit recommendation.

Source code in src/probcal/monitor/_monitor.py

def report(self) -> MonitorReport:
    """Trajectory plus the diagnostic re-offset/re-fit recommendation."""
    alarm_at = next((s.label for s in self.steps_ if s.alarm), None)
    grade_table = dict(self.steps_[-1].e_grades) if self.steps_ else {}
    onset_label = None
    recommendation = "none"
    reasoning: tuple[str, ...] = ("no alarm: the global e-process never reached 1/alpha",)
    if alarm_at is not None:
        onset_idx = self._onset_index()
        onset_label = self.steps_[onset_idx].label
        # Shared with apply_recommendation(), so the diagnostic window here
        # and the action window there never disagree.
        pz, py, pw = self._since(self._recommendation_window_start(onset_idx))
        recommendation, reasoning = recommend(
            pz,
            py,
            pw,
            alarm_at=alarm_at,
            e_shape=self.steps_[-1].e_shape,
            alpha=self.alpha,
            onset_label=onset_label,
        )
    return MonitorReport(
        steps=tuple(self.steps_),
        alarm_at=alarm_at,
        recommendation=recommendation,
        reasoning=reasoning,
        alpha=self.alpha,
        grade_table=grade_table,
        onset_label=onset_label,
    )

apply_recommendation

apply_recommendation(target: object = None) -> Any

Apply :meth:report's recommendation once, closing the re-offset loop.

"re-offset": estimates the log-odds shift by maximum likelihood (:func:~probcal.offset.estimate_offset) on the batches of the recommendation window (the same window :meth:report uses for its trailing-window diagnostics), composes the fitted offset onto target (see below), and returns a fresh monitor with the same constructor parameters (:meth:_ctor_params) to watch the corrected pipeline. The monitor is fresh, not continued: its e-process is a martingale under the null "the CURRENTLY DEPLOYED forecast is calibrated"; once target changes, the accumulated evidence describes a forecast that no longer exists.

"re-fit"/"none": no offset, composed target, or fresh monitor is produced. Automatic re-fitting is deliberately out of scope: a slope drift needs a human to choose and validate a new calibrator, not a mechanical action this method could take safely.

Composing the fitted offset onto target:

• None (default) -- composed is None; only the offset (and the fresh monitor) come back.
• :class:~probcal.chain.Chain -- a new Chain([target.calibrator_, *target.offsets_, offset]); target itself is untouched.
• :class:~probcal.wrapper.CalibratedModel -- target.with_offset(offset): a copy with the fitted offset appended (works for a model loaded from JSON too, since no calibration data is needed); target itself is untouched.

PARAMETER	DESCRIPTION
target	The currently deployed pipeline to correct. None (default) returns the fitted offset alone. TYPE: Chain, CalibratedModel, or None DEFAULT: None

RETURNS	DESCRIPTION
AppliedAction	kind, the fitted offset (None unless kind="re-offset"), the composed pipeline (None unless kind="re-offset" and a target was given), a fresh monitor (None unless kind="re-offset"), the window of batch labels the estimate used, and an audit trail of fingerprints and the estimated delta/se.

RAISES	DESCRIPTION
TypeError	If target is not None, a Chain, or a CalibratedModel.

Notes

self is never mutated: the returned monitor is a brand-new object.

Examples:

>>> import numpy as np
>>> from probcal._math import expit, logit
>>> from probcal.datasets import make_pd_portfolio
>>> from probcal.monitor import CalibrationMonitor
>>> mon = CalibrationMonitor(alpha=0.05)
>>> for k in range(6):
...     d = make_pd_portfolio(n=1000, random_state=k)
...     rng = np.random.default_rng(k + 1000)
...     y = (rng.random(1000) < expit(logit(d.scores) + 0.8)).astype(float)
...     _ = mon.update(y, d.scores, label=f"m{k}")
>>> action = mon.apply_recommendation()
>>> action.kind
're-offset'
>>> action.monitor is not mon
True

Source code in src/probcal/monitor/_monitor.py

def apply_recommendation(self, target: object = None) -> Any:
    """Apply :meth:`report`'s recommendation once, closing the re-offset loop.

    ``"re-offset"``: estimates the log-odds shift by maximum likelihood
    (:func:`~probcal.offset.estimate_offset`) on the batches of the
    recommendation window (the same window :meth:`report` uses for its
    trailing-window diagnostics), composes the fitted offset onto ``target`` (see
    below), and returns a **fresh** monitor with the same constructor
    parameters (:meth:`_ctor_params`) to watch the corrected pipeline.
    The monitor is fresh, not continued: its e-process is a martingale
    under the null "the CURRENTLY DEPLOYED forecast is calibrated";
    once ``target`` changes, the accumulated evidence describes a
    forecast that no longer exists.

    ``"re-fit"``/``"none"``: no offset, composed target, or fresh
    monitor is produced. Automatic re-fitting is deliberately out of
    scope: a slope drift needs a human to choose and validate a new
    calibrator, not a mechanical action this method could take safely.

    Composing the fitted offset onto ``target``:

    - ``None`` (default) -- ``composed`` is ``None``; only the offset
      (and the fresh monitor) come back.
    - :class:`~probcal.chain.Chain` -- a new
      ``Chain([target.calibrator_, *target.offsets_, offset])``;
      ``target`` itself is untouched.
    - :class:`~probcal.wrapper.CalibratedModel` --
      ``target.with_offset(offset)``: a copy with the fitted offset
      appended (works for a model loaded from JSON too, since no
      calibration data is needed); ``target`` itself is untouched.

    Parameters
    ----------
    target : Chain, CalibratedModel, or None
        The currently deployed pipeline to correct. ``None`` (default)
        returns the fitted offset alone.

    Returns
    -------
    AppliedAction
        ``kind``, the fitted ``offset`` (``None`` unless
        ``kind="re-offset"``), the ``composed`` pipeline (``None``
        unless ``kind="re-offset"`` and a ``target`` was given), a
        fresh ``monitor`` (``None`` unless ``kind="re-offset"``), the
        ``window`` of batch labels the estimate used, and an ``audit``
        trail of fingerprints and the estimated ``delta``/``se``.

    Raises
    ------
    TypeError
        If ``target`` is not ``None``, a ``Chain``, or a
        ``CalibratedModel``.

    Notes
    -----
    ``self`` is never mutated: the returned monitor is a brand-new
    object.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal._math import expit, logit
    >>> from probcal.datasets import make_pd_portfolio
    >>> from probcal.monitor import CalibrationMonitor
    >>> mon = CalibrationMonitor(alpha=0.05)
    >>> for k in range(6):
    ...     d = make_pd_portfolio(n=1000, random_state=k)
    ...     rng = np.random.default_rng(k + 1000)
    ...     y = (rng.random(1000) < expit(logit(d.scores) + 0.8)).astype(float)
    ...     _ = mon.update(y, d.scores, label=f"m{k}")
    >>> action = mon.apply_recommendation()
    >>> action.kind
    're-offset'
    >>> action.monitor is not mon
    True
    """
    from ._actions import apply_recommendation

    return apply_recommendation(self, target)

to_dict

to_dict() -> dict[str, object]

Versioned snapshot; the state includes every past batch — that is what makes each decision reproducible (spec invariant).

Source code in src/probcal/monitor/_monitor.py

def to_dict(self) -> dict[str, object]:
    """Versioned snapshot; the state includes every past batch — that is
    what makes each decision reproducible (spec invariant)."""
    params = self._ctor_params()
    for k in _SEQUENCE_PARAMS:
        if params[k] is not None:
            params[k] = list(params[k])
    return envelope(
        self,
        params=params,
        state=encode_value(self._state_dict()),
        fit_meta={
            "n_batches": len(self._z),
            "n_obs": int(sum(len(a) for a in self._z)),
        },
    )

from_dict classmethod

from_dict(d: dict) -> CalibrationMonitor

Rebuild a monitor mid-stream; the trajectory continues bit-for-bit.

RAISES	DESCRIPTION
ValueError	If the schema version is unknown or the payload class differs.

Source code in src/probcal/monitor/_monitor.py

@classmethod
def from_dict(cls, d: dict) -> "CalibrationMonitor":
    """Rebuild a monitor mid-stream; the trajectory continues bit-for-bit.

    Raises
    ------
    ValueError
        If the schema version is unknown or the payload class differs.
    """
    check_payload(cls, d)
    params = dict(d["params"])
    for k in _SEQUENCE_PARAMS:
        if params[k] is not None:
            params[k] = tuple(params[k])
    mon = cls(**params)
    st: dict[str, Any] = decode_value(d["state"], arrays=False)  # type: ignore[assignment]
    mon._z = [np.asarray(a, dtype=np.float64) for a in st["z"]]
    mon._y = [np.asarray(a, dtype=np.float64) for a in st["y"]]
    mon._w = [np.asarray(a, dtype=np.float64) for a in st["w"]]
    mon._g = [np.asarray(a, dtype=str) if a is not None else None for a in st["g"]]
    mon._offset.set_state(st["offset"])
    mon._log_shape = float(st["log_shape"])
    mon._grade_procs = {}
    for g, ps in st["grade_procs"].items():
        proc = OffsetProcess(mon._sym_grid())
        proc.set_state(ps)
        mon._grade_procs[g] = proc
    mon._cs.set_state(st["cs_log"], st["cs_max"])
    mon._grade_cs = {}
    for g, log in st["grade_cs_log"].items():
        cs = ConfidenceSequence(mon._cs.grid)
        cs.set_state(log, st["grade_cs_max"][g])
        mon._grade_cs[g] = cs
    mon._max_log_global = float(st["max_log_global"])
    mon._alarmed = bool(st["alarmed"])
    mon._warned_grades = bool(st.get("warned_grades", False))
    mon.masterscale_fingerprint_ = st.get("masterscale_fingerprint")
    mon.steps_ = []
    for sd in st["steps"]:
        sd = dict(sd)
        sd["delta_ci"] = tuple(sd["delta_ci"]) if sd["delta_ci"] is not None else None
        sd["grade_delta_ci"] = {
            g: (tuple(v) if v is not None else None) for g, v in sd["grade_delta_ci"].items()
        }
        mon.steps_.append(MonitorStep(**sd))
    return mon

MonitorReport dataclass

MonitorReport(steps: tuple[MonitorStep, ...], alarm_at: str | None, recommendation: str, reasoning: tuple[str, ...], alpha: float = 0.05, grade_table: dict[str, float] = dict(), onset_label: str | None = None)

Full monitoring trajectory with the diagnostic recommendation.

ATTRIBUTE	DESCRIPTION
steps	Every processed batch, in arrival order. TYPE: tuple[MonitorStep, ...]
alarm_at	Label of the first batch at which the alarm fired. TYPE: str or None
recommendation	Diagnostic (no error guarantee — the component e-values are the evidence; see the monitoring chapter). TYPE: {none, re - offset, re - fit}
reasoning	Plain-language trail behind the recommendation. TYPE: tuple[str, ...]
alpha	The monitor's alarm level (drawn as the 1/alpha line by probcal.plots.plot_e_process). TYPE: float
grade_table	Latest per-grade e-values. TYPE: dict[str, float]
onset_label	Label of the batch :func:~probcal.monitor._onset.estimate_onset points to as the drift onset (backward-CUSUM argmax of MonitorStep.log_e_increment); None unless alarm_at is also set. An estimate, not a test. TYPE: str or None

to_frame

to_frame() -> object

Steps as a list of dicts, or a pandas DataFrame when pandas is importable.

Source code in src/probcal/monitor/_monitor.py

def to_frame(self) -> object:
    """Steps as a list of dicts, or a pandas DataFrame when pandas is importable."""
    return _to_frame([asdict(s) for s in self.steps])

MonitorStep dataclass

MonitorStep(label: str, n: int, n_events: float, e_offset: float, e_shape: float, e_grades: dict[str, float], e_global: float, p_anytime: float, alarm: bool, delta_ci: tuple[float, float] | None, delta_hat: float, slope_hat: float, grade_delta_ci: dict[str, tuple[float, float] | None], log_e_increment: float)

One matured batch's monitoring record (all e-values are running values).

ATTRIBUTE	DESCRIPTION
label	Caller-supplied batch label (opaque; arrival order is what counts). TYPE: str
n	Batch size. TYPE: int
n_events	Weighted event count of the batch. TYPE: float
e_offset, e_shape	Running component e-values after this batch (nan for components not in components). TYPE: float
e_grades	Running per-grade offset e-values (empty when no grades were given). TYPE: dict[str, float]
e_global	Running mean of the active components — the alarm statistic. TYPE: float
p_anytime	min(1, 1 / max_k E_k) — a p-value valid at every stopping time. TYPE: float
alarm	Whether E has ever reached 1/alpha (sticky). TYPE: bool
delta_ci	Time-uniform confidence sequence for the current offset (grid endpoints still surviving); None if every grid null is rejected. TYPE: tuple[float, float] or None
delta_hat, slope_hat	The predictable plug-ins used for this batch (from past batches only) — recorded for auditability. TYPE: float
grade_delta_ci	Per-grade time-uniform confidence sequence for that grade's own offset, same construction and grid as delta_ci (empty when no grades were given). TYPE: dict[str, tuple[float, float] | None]
log_e_increment	This batch's additive plug-in log-LR increment: the sum of the plug-in bern_log_lr contributions of the components in the alarm — offset and shape when monitored (0.0 when a plug-in is the identity), plus the declared grades' plug-ins. Unlike e_global — a logsumexp mixture, not additive across batches — this is the purely additive series :func:~probcal.monitor._onset.estimate_onset localizes drift onset from. TYPE: float

moc_offset

moc_offset(monitor_or_report: CalibrationMonitor | MonitorReport, *, level: float | None = None) -> LogitOffset

Margin-of-conservatism offset from a monitor's confidence sequence.

CalibrationMonitor maintains, at every batch, a time-uniform confidence sequence (CS) for the current offset: the set of shifts delta such that sigma(z + delta) -- applying that shift to the monitored logits -- would itself be calibrated is covered with probability >= 1 - alpha simultaneously at every stopping time (MonitorStep.delta_ci, the surviving grid nulls' hull; None if every grid null has been rejected). Its upper end, hi, is a margin-of-conservatism offset: applying delta=hi shifts the portfolio at least as far as the CS says drift plausibly runs, so (loosely) it corrects for the drift with high confidence rather than only for its point estimate.

Two ways to get hi:

• level=None (default): take hi from steps[-1].delta_ci as-is, at the monitor's own alpha.
• level given: recompute the surviving grid nulls at that confidence level directly from the monitor's own running confidence-sequence state (the nulls whose running maximum log-e stays below -log(1 - level)) and take their max. This needs the live monitor object, not a frozen :class:~probcal.monitor.MonitorReport snapshot, so it raises TypeError for a report.

The returned :class:~probcal.offset.LogitOffset is fit on the last monitored batch's probabilities (expit(mon._z[-1])), which fixes its pre_mean_/post_mean_ audit fields and its data fingerprint to that batch. A :class:~probcal.monitor.MonitorReport retains no batch data at all, so in that case the offset is fit on the placeholder np.array([0.5]) instead -- delta_ is exact either way, but pre_mean_/post_mean_ and the fingerprint are then placeholders, not a real portfolio's summary.

PARAMETER	DESCRIPTION
monitor_or_report	The monitor (or its report) to read the confidence sequence from. TYPE: CalibrationMonitor or MonitorReport
level	Confidence level in (0, 1) to recompute the surviving grid nulls at; None (default) uses the last step's delta_ci at the monitor's own alpha. Requires a CalibrationMonitor. TYPE: (float or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
LogitOffset	Fitted offset with delta_ equal to the confidence sequence's upper end.

RAISES	DESCRIPTION
ValueError	If no batches have been processed yet, or the surviving grid-null set is empty (every null rejected) -- widen delta_ci_grid.
TypeError	If level is given but monitor_or_report is a MonitorReport rather than a live CalibrationMonitor.

Examples:

>>> import numpy as np
>>> from probcal.datasets import make_pd_portfolio
>>> from probcal.monitor import CalibrationMonitor, moc_offset
>>> mon = CalibrationMonitor(alpha=0.05)
>>> from probcal import expit, logit
>>> for seed in range(3):
...     d = make_pd_portfolio(n=500, random_state=seed)
...     rng = np.random.default_rng(seed)
...     true_pd = expit(logit(d.scores) + 0.3)  # +0.3 log-odds drift injected
...     y = (rng.random(500) < true_pd).astype(float)
...     _ = mon.update(y, d.scores, label=f"b{seed}")
>>> off = moc_offset(mon)
>>> off.delta_ >= mon.steps_[-1].delta_hat
True

Source code in src/probcal/monitor/_actions.py

def moc_offset(
    monitor_or_report: "CalibrationMonitor | MonitorReport", *, level: float | None = None
) -> LogitOffset:
    """Margin-of-conservatism offset from a monitor's confidence sequence.

    ``CalibrationMonitor`` maintains, at every batch, a time-uniform
    confidence sequence (CS) for the current offset: the set of shifts
    ``delta`` such that ``sigma(z + delta)`` -- applying that shift to the
    monitored logits -- would itself be calibrated is covered with
    probability ``>= 1 - alpha`` *simultaneously at every stopping time*
    (``MonitorStep.delta_ci``, the surviving grid nulls' hull; ``None`` if
    every grid null has been rejected). Its upper end, ``hi``, is a
    margin-of-conservatism offset: applying ``delta=hi`` shifts the
    portfolio at least as far as the CS says drift plausibly runs, so
    (loosely) it corrects for the drift with high confidence rather than
    only for its point estimate.

    Two ways to get ``hi``:

    - ``level=None`` (default): take ``hi`` from ``steps[-1].delta_ci``
      as-is, at the monitor's own ``alpha``.
    - ``level`` given: recompute the surviving grid nulls at that
      confidence level directly from the monitor's own running
      confidence-sequence state (the nulls whose running maximum log-e
      stays below ``-log(1 - level)``) and take their max. This needs the
      live monitor object, not a frozen
      :class:`~probcal.monitor.MonitorReport` snapshot, so it raises
      ``TypeError`` for a report.

    The returned :class:`~probcal.offset.LogitOffset` is fit on the last
    monitored batch's probabilities (``expit(mon._z[-1])``), which fixes
    its ``pre_mean_``/``post_mean_`` audit fields and its data fingerprint
    to that batch. A :class:`~probcal.monitor.MonitorReport` retains no
    batch data at all, so in that case the offset is fit on the
    placeholder ``np.array([0.5])`` instead -- ``delta_`` is exact either
    way, but ``pre_mean_``/``post_mean_`` and the fingerprint are then
    placeholders, not a real portfolio's summary.

    Parameters
    ----------
    monitor_or_report : CalibrationMonitor or MonitorReport
        The monitor (or its report) to read the confidence sequence from.
    level : float or None, keyword-only
        Confidence level in ``(0, 1)`` to recompute the surviving grid
        nulls at; ``None`` (default) uses the last step's ``delta_ci`` at
        the monitor's own ``alpha``. Requires a ``CalibrationMonitor``.

    Returns
    -------
    LogitOffset
        Fitted offset with ``delta_`` equal to the confidence sequence's
        upper end.

    Raises
    ------
    ValueError
        If no batches have been processed yet, or the surviving grid-null
        set is empty (every null rejected) -- widen ``delta_ci_grid``.
    TypeError
        If ``level`` is given but ``monitor_or_report`` is a
        ``MonitorReport`` rather than a live ``CalibrationMonitor``.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.datasets import make_pd_portfolio
    >>> from probcal.monitor import CalibrationMonitor, moc_offset
    >>> mon = CalibrationMonitor(alpha=0.05)
    >>> from probcal import expit, logit
    >>> for seed in range(3):
    ...     d = make_pd_portfolio(n=500, random_state=seed)
    ...     rng = np.random.default_rng(seed)
    ...     true_pd = expit(logit(d.scores) + 0.3)  # +0.3 log-odds drift injected
    ...     y = (rng.random(500) < true_pd).astype(float)
    ...     _ = mon.update(y, d.scores, label=f"b{seed}")
    >>> off = moc_offset(mon)
    >>> off.delta_ >= mon.steps_[-1].delta_hat
    True
    """
    if isinstance(monitor_or_report, CalibrationMonitor):
        mon, steps = monitor_or_report, monitor_or_report.steps_
    elif isinstance(monitor_or_report, MonitorReport):
        mon, steps = None, list(monitor_or_report.steps)
    else:
        raise TypeError(
            "moc_offset requires a CalibrationMonitor or a MonitorReport, got "
            f"{type(monitor_or_report).__name__}"
        )
    if not steps:
        raise ValueError("moc_offset: no batches have been processed yet")

    if level is None:
        delta_ci = steps[-1].delta_ci
        if delta_ci is None:
            raise ValueError(
                "moc_offset: every grid null in delta_ci is rejected (delta_ci is "
                "None); widen delta_ci_grid to include the true offset"
            )
        hi = delta_ci[1]
    else:
        if mon is None:
            raise TypeError(
                "moc_offset: level requires a live CalibrationMonitor -- recomputing "
                "the surviving grid nulls at a new confidence level reads the "
                "monitor's running confidence-sequence state, which a MonitorReport "
                "(a frozen snapshot) does not retain; pass the monitor itself, or "
                "omit level to use its last step's delta_ci as-is"
            )
        validate_level(level, "level")
        surviving = mon._cs.surviving(-np.log(1.0 - level))
        if surviving.size == 0:
            raise ValueError("moc_offset: no grid nulls survive at this level; widen delta_ci_grid")
        hi = float(surviving.max())

    batch_p = expit(mon._z[-1]) if mon is not None else np.array([0.5])
    return LogitOffset(delta=hi).fit(batch_p)

moc_offset_from_counts

moc_offset_from_counts(y: object, p: object, *, level: float = 0.9, sample_weight: object = None) -> LogitOffset

Margin-of-conservatism offset from raw event counts (mode B, no monitor).

The Jeffreys posterior upper bound on the observed event rate, q = beta_ppf(level, k + 0.5, n - k + 0.5) with the Kish effective counts n = sum(w)^2 / sum(w^2) and k = n * sum(w * y) / sum(w) (unit weights give the raw counts; rescaling every weight by a constant leaves q unchanged) -- the same one-sided Jeffreys quantile metrics.jeffreys_grade_test/metrics.jeffreys_upper_bands use -- becomes the offset's target mean: LogitOffset(target_mean=q) (mode B) solves for the log-odds shift that re-anchors p's mean at q, a conservative re-anchoring against the observed outcomes rather than a shift read off a monitor's confidence sequence.

PARAMETER	DESCRIPTION
y	Binary outcomes in {0, 1}. TYPE: array_like
p	Predicted probabilities in [0, 1] to be re-anchored. TYPE: array_like
level	Confidence level in (0, 1) for the Jeffreys upper quantile. TYPE: (float, keyword - only) DEFAULT: 0.9
sample_weight	Optional non-negative weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
LogitOffset	Fitted offset with post_mean_ equal to the Jeffreys upper quantile q.

RAISES	DESCRIPTION
ValueError	If y/p are invalid, or level is not in (0, 1).

Examples:

>>> import numpy as np
>>> from probcal.monitor import moc_offset_from_counts
>>> y = np.array([0.0] * 970 + [1.0] * 30)
>>> p = np.full(1000, 0.02)
>>> off = moc_offset_from_counts(y, p, level=0.9)
>>> off.post_mean_ > 0.03
True

Source code in src/probcal/monitor/_actions.py

def moc_offset_from_counts(
    y: object,
    p: object,
    *,
    level: float = 0.9,
    sample_weight: object = None,
) -> LogitOffset:
    """Margin-of-conservatism offset from raw event counts (mode B, no monitor).

    The Jeffreys posterior upper bound on the observed event rate,
    ``q = beta_ppf(level, k + 0.5, n - k + 0.5)`` with the Kish effective
    counts ``n = sum(w)^2 / sum(w^2)`` and ``k = n * sum(w * y) / sum(w)``
    (unit weights give the raw counts; rescaling every weight by a constant
    leaves ``q`` unchanged) -- the same one-sided Jeffreys quantile
    ``metrics.jeffreys_grade_test``/``metrics.jeffreys_upper_bands`` use --
    becomes the offset's target mean: ``LogitOffset(target_mean=q)`` (mode
    B) solves for the log-odds shift that re-anchors ``p``'s mean at
    ``q``, a conservative re-anchoring against the observed outcomes
    rather than a shift read off a monitor's confidence sequence.

    Parameters
    ----------
    y : array_like
        Binary outcomes in ``{0, 1}``.
    p : array_like
        Predicted probabilities in ``[0, 1]`` to be re-anchored.
    level : float, keyword-only
        Confidence level in ``(0, 1)`` for the Jeffreys upper quantile.
    sample_weight : array_like or None, keyword-only
        Optional non-negative weights, same length as ``y``.

    Returns
    -------
    LogitOffset
        Fitted offset with ``post_mean_`` equal to the Jeffreys upper
        quantile ``q``.

    Raises
    ------
    ValueError
        If ``y``/``p`` are invalid, or ``level`` is not in ``(0, 1)``.

    Examples
    --------
    >>> import numpy as np
    >>> from probcal.monitor import moc_offset_from_counts
    >>> y = np.array([0.0] * 970 + [1.0] * 30)
    >>> p = np.full(1000, 0.02)
    >>> off = moc_offset_from_counts(y, p, level=0.9)
    >>> off.post_mean_ > 0.03
    True
    """
    y_arr, p_arr, w_arr = _prep(y, p, sample_weight)
    validate_level(level, "level")
    # Kish effective counts: events and trials scaled by the same factor
    # (unit weights come back bit-identical: the factor is exactly 1.0).
    w_eff = w_arr * (float(w_arr.sum()) / float(np.dot(w_arr, w_arr)))
    k = float(np.sum(w_eff * y_arr))
    n = float(np.sum(w_eff))
    q = beta_ppf(level, k + 0.5, n - k + 0.5)
    return LogitOffset(target_mean=q).fit(p_arr, sample_weight=w_arr)

Numerical core: PAVA, IRLS logistic regression, special functions, LOESS, spline basis.

Pure numpy + stdlib. Special functions are hand-rolled (continued fractions, series, math.erfc/math.lgamma) and verified against scipy in tests/test_math_reference.py.

expit

expit(z: object) -> ndarray

Logistic sigmoid 1 / (1 + exp(-z)), overflow-safe.

PARAMETER	DESCRIPTION
z	Logits; any real values, including large magnitudes. TYPE: array_like

RETURNS	DESCRIPTION
ndarray	Probabilities in [0, 1].

Source code in src/probcal/_math.py

def expit(z: object) -> np.ndarray:
    """Logistic sigmoid ``1 / (1 + exp(-z))``, overflow-safe.

    Parameters
    ----------
    z : array_like
        Logits; any real values, including large magnitudes.

    Returns
    -------
    numpy.ndarray
        Probabilities in ``[0, 1]``.
    """
    arr = np.asarray(z, dtype=np.float64)
    out = np.empty_like(arr)
    pos = arr >= 0
    out[pos] = 1.0 / (1.0 + np.exp(-arr[pos]))
    ez = np.exp(arr[~pos])
    out[~pos] = ez / (1.0 + ez)
    return out

logit

logit(p: object) -> ndarray

Log-odds of p, clipped to keep the output finite.

PARAMETER	DESCRIPTION
p	Probabilities; values outside [1e-12, 1 - 1e-12] are clipped. TYPE: array_like

RETURNS	DESCRIPTION
ndarray	log(p / (1 - p)) elementwise.

Source code in src/probcal/_math.py

def logit(p: object) -> np.ndarray:
    """Log-odds of ``p``, clipped to keep the output finite.

    Parameters
    ----------
    p : array_like
        Probabilities; values outside ``[1e-12, 1 - 1e-12]`` are clipped.

    Returns
    -------
    numpy.ndarray
        ``log(p / (1 - p))`` elementwise.
    """
    arr = np.clip(np.asarray(p, dtype=np.float64), EPS, 1.0 - EPS)
    return np.log(arr) - np.log1p(-arr)

masterscale

A masterscale as one value object: the grade ladder on the calibrated scale.

A masterscale is the fixed ladder of PD bands that defines rating grades. probcal consumes it in two shapes -- a {name: (lo, hi)} dict for translation and a per-observation label array for testing and monitoring -- and :class:Masterscale is the single place both shapes come from.

Boundary convention, stated once: :meth:Masterscale.assign is half-open, lo <= p < hi, with the top band closed at its upper edge, so every p in [edges[0], edges[-1]] belongs to exactly one grade. Band inversion (:func:probcal.thresholds.calibrated_bands_to_raw) follows the same rule on the raw axis: band [lo, hi) maps to [inf{s: g(s) >= lo}, inf{s: g(s) >= hi}), top band closed, so the raw intervals partition the raw axis and counting raw scores per interval reproduces :meth:Masterscale.assign counts exactly — also for step calibrators (isotonic, histogram binning), whose plateaus give a shared edge positive raw-score mass.

GradeTable dataclass

GradeTable(grades: tuple[str, ...], lo: ndarray, hi: ndarray, n: ndarray, events: ndarray, mean_pd: ndarray, observed_rate: ndarray)

Bases: _ResultBase

Per-grade counts against a masterscale: the standard grade table.

ATTRIBUTE	DESCRIPTION
grades	Every grade of the masterscale, best to worst (empty grades included). TYPE: tuple of str
lo, hi	Band bounds per grade. TYPE: ndarray
n	Observation count per grade (weighted sum when sample_weight is given). TYPE: ndarray
events	Event count per grade (weighted). TYPE: ndarray
mean_pd	Mean assigned probability per grade; nan for an empty grade. TYPE: ndarray
observed_rate	events / n; nan for an empty grade. TYPE: ndarray

Masterscale

Masterscale(bands: dict[str, tuple[float, float]], *, provenance: dict | None = None)

Bases: JsonIO

Frozen grade ladder on the calibrated-probability scale.

PARAMETER	DESCRIPTION
bands	Grade name to (lo, hi) calibrated-probability bounds. Bands may be given in any order; they are stored sorted by lo. Every bound must lie in [0, 1], each band needs lo < hi, consecutive bands must share their edge exactly (gaps and overlaps both raise), and names must be unique. TYPE: dict[str, tuple[float, float]]
provenance	Optional record of how the scale was built (set by :func:probcal.thresholds.build_masterscale); None for a hand-built scale. Deep-frozen on construction (later changes to the dict passed in do not reach the scale), serialized with the object, and shown by :meth:interpret. Values must be JSON-encodable. TYPE: (dict or None, keyword - only) DEFAULT: None

ATTRIBUTE	DESCRIPTION
names	Grade names, best (lowest PD) to worst. TYPE: tuple of str
edges	The n_grades + 1 band edges, ascending. TYPE: ndarray
n_grades	Number of grades. TYPE: int

Notes

Assignment is half-open, lo <= p < hi, with the top band closed at its upper edge. Values outside [edges[0], edges[-1]] raise; a masterscale is expected to cover [0, 1] (pass lo and hi to :meth:from_edges explicitly if yours does not, or accept the error). Band inversion through :func:probcal.thresholds.calibrated_bands_to_raw uses the same half-open rule on the raw axis, so raw-interval counts and :meth:assign counts agree exactly.

Equality and hashing cover the bands and the provenance, so a == b exactly when a.fingerprint() == b.fingerprint(); compare a.bands == b.bands for "same ladder, however built". Instances are immutable and support :func:copy.deepcopy and pickling.

Examples:

>>> ms = Masterscale.from_edges([0.01, 0.05], names=["A", "B", "C"])
>>> ms.assign([0.005, 0.01, 0.05, 1.0]).tolist()
['A', 'B', 'C', 'C']

Source code in src/probcal/masterscale.py

def __init__(
    self, bands: dict[str, tuple[float, float]], *, provenance: dict | None = None
) -> None:
    if not bands:
        raise ValueError("a masterscale needs at least one band")
    items: list[tuple[float, float, str]] = []
    for name, bounds in bands.items():
        if len(bounds) != 2:
            raise ValueError(f"band {name!r}: expected (lo, hi), got {bounds!r}")
        # ``+ 0.0`` folds -0.0 into 0.0, so equal ladders share one fingerprint.
        lo, hi = float(bounds[0]) + 0.0, float(bounds[1]) + 0.0
        in_range = np.isfinite(lo) and np.isfinite(hi) and 0.0 <= lo <= 1.0 and 0.0 <= hi <= 1.0
        if not in_range:
            raise ValueError(f"band {name!r}: bounds must lie in [0, 1], got ({lo}, {hi})")
        if not lo < hi:
            raise ValueError(f"band {name!r}: lo must be < hi, got ({lo}, {hi})")
        items.append((lo, hi, str(name)))
    items.sort(key=lambda t: t[0])
    names = tuple(t[2] for t in items)
    if len(set(names)) != len(names):
        raise ValueError(f"grade names must be unique, got {names}")
    for (_lo0, hi0, n0), (lo1, _hi1, n1) in zip(items, items[1:], strict=False):
        if hi0 < lo1:
            raise ValueError(f"gap between band {n0!r} (hi={hi0}) and band {n1!r} (lo={lo1})")
        if hi0 > lo1:
            raise ValueError(f"band {n1!r} (lo={lo1}) overlaps band {n0!r} (hi={hi0})")
    object.__setattr__(self, "_bands", tuple((n, (lo, hi)) for lo, hi, n in items))
    object.__setattr__(self, "_edges", np.array([t[0] for t in items] + [items[-1][1]]))
    object.__setattr__(self, "_names", names)
    frozen = None if provenance is None else _freeze(dict(provenance))
    object.__setattr__(self, "_provenance", frozen)

bands property

bands: dict[str, tuple[float, float]]

{name: (lo, hi)} best to worst: the shape every band consumer accepts.

edges property

edges: ndarray

The n_grades + 1 band edges, ascending (a copy).

names property

names: tuple[str, ...]

Grade names, best to worst.

n_grades property

n_grades: int

Number of grades.

provenance property

provenance: dict | None

How the scale was built (None for a hand-built scale); a fresh copy.

from_edges classmethod

from_edges(edges: object, names: object = None, lo: float = 0.0, hi: float = 1.0, *, provenance: dict | None = None) -> Masterscale

Build from interior edges: len(edges) + 1 grades on [lo, hi].

PARAMETER	DESCRIPTION
edges	Interior band edges (any order; sorted here). TYPE: array_like
names	One name per grade, best to worst; None gives "G1", "G2", ... TYPE: sequence of str or None DEFAULT: None
lo	Outer bounds, 0.0 and 1.0 by default. TYPE: float DEFAULT: 0.0
hi	Outer bounds, 0.0 and 1.0 by default. TYPE: float DEFAULT: 0.0
provenance	Passed to the constructor (see :class:Masterscale). TYPE: (dict or None, keyword - only) DEFAULT: None

Source code in src/probcal/masterscale.py

@classmethod
def from_edges(
    cls,
    edges: object,
    names: object = None,
    lo: float = 0.0,
    hi: float = 1.0,
    *,
    provenance: dict | None = None,
) -> "Masterscale":
    """Build from interior edges: ``len(edges) + 1`` grades on ``[lo, hi]``.

    Parameters
    ----------
    edges : array_like
        Interior band edges (any order; sorted here).
    names : sequence of str or None
        One name per grade, best to worst; ``None`` gives ``"G1"``, ``"G2"``, ...
    lo, hi : float
        Outer bounds, ``0.0`` and ``1.0`` by default.
    provenance : dict or None, keyword-only
        Passed to the constructor (see :class:`Masterscale`).
    """
    inner = sorted(float(e) for e in np.asarray(edges, dtype=np.float64).reshape(-1))
    full = [float(lo), *inner, float(hi)]
    n = len(full) - 1
    if names is None:
        labels = [f"G{i + 1}" for i in range(n)]
    else:
        labels = [str(x) for x in list(names)]  # type: ignore[call-overload]
    if len(labels) != n:
        raise ValueError(f"{n} grades from {len(inner)} edges, but {len(labels)} names given")
    if len(set(labels)) != n:
        raise ValueError(f"grade names must be unique, got {tuple(labels)}")
    return cls({labels[i]: (full[i], full[i + 1]) for i in range(n)}, provenance=provenance)

index

index(p: object) -> ndarray

Zero-based grade index per observation, lo <= p < hi, top band closed.

p is validated like any probability input (a two-column predict_proba matrix is accepted) but not clipped away from 0 and 1, so an edge closer than 1e-12 to either end still decides.

RAISES	DESCRIPTION
ValueError	If any p lies outside [edges[0], edges[-1]].

Source code in src/probcal/masterscale.py

def index(self, p: object) -> np.ndarray:
    """Zero-based grade index per observation, ``lo <= p < hi``, top band closed.

    ``p`` is validated like any probability input (a two-column
    ``predict_proba`` matrix is accepted) but not clipped away from 0
    and 1, so an edge closer than ``1e-12`` to either end still decides.

    Raises
    ------
    ValueError
        If any ``p`` lies outside ``[edges[0], edges[-1]]``.
    """
    p_arr = _validate_probabilities(p, name="p")
    lo, hi = self._edges[0], self._edges[-1]
    bad = (p_arr < lo) | (p_arr > hi)
    if np.any(bad):
        raise ValueError(
            f"{int(bad.sum())} value(s) of p lie outside the masterscale range "
            f"[{lo}, {hi}] (first offender {p_arr[bad][0]!r}); a masterscale is expected "
            "to cover [0, 1]"
        )
    return np.searchsorted(self._edges[1:-1], p_arr, side="right").astype(np.int64)

assign

assign(p: object) -> ndarray

Grade name per observation (see :meth:index for the rule).

Source code in src/probcal/masterscale.py

def assign(self, p: object) -> np.ndarray:
    """Grade name per observation (see :meth:`index` for the rule)."""
    return np.asarray(self._names, dtype=object)[self.index(p)].astype(str)

table

table(y: object, p: object, sample_weight: object = None) -> GradeTable

Per-grade counts of y against p assigned through this scale.

Every grade is listed, including empty ones (n == 0, rates nan); weights, when given, turn counts into weighted sums.

Source code in src/probcal/masterscale.py

def table(self, y: object, p: object, sample_weight: object = None) -> GradeTable:
    """Per-grade counts of ``y`` against ``p`` assigned through this scale.

    Every grade is listed, including empty ones (``n == 0``, rates ``nan``);
    weights, when given, turn counts into weighted sums.
    """
    y_arr = validate_binary_y(y)
    p_arr = _validate_probabilities(p, name="p")
    if len(y_arr) != len(p_arr):
        raise ValueError("y and p must have equal length")
    w_arr = validate_weights(sample_weight, len(p_arr))
    idx = self.index(p_arr)
    k = self.n_grades
    n = np.bincount(idx, weights=w_arr, minlength=k).astype(np.float64)
    events = np.bincount(idx, weights=w_arr * y_arr, minlength=k).astype(np.float64)
    wp = np.bincount(idx, weights=w_arr * p_arr, minlength=k).astype(np.float64)
    with np.errstate(divide="ignore", invalid="ignore"):
        mean_pd = np.where(n > 0, wp / n, np.nan)
        observed = np.where(n > 0, events / n, np.nan)
    return GradeTable(
        grades=self._names,
        lo=self._edges[:-1].copy(),
        hi=self._edges[1:].copy(),
        n=n,
        events=events,
        mean_pd=mean_pd,
        observed_rate=observed,
    )

interpret

interpret() -> Interpretation

The band table in words, with the boundary convention and provenance.

param_names/param_values list each grade's band as two entries, "<grade>.lo" and "<grade>.hi".

Source code in src/probcal/masterscale.py

def interpret(self) -> Interpretation:
    """The band table in words, with the boundary convention and provenance.

    ``param_names``/``param_values`` list each grade's band as two
    entries, ``"<grade>.lo"`` and ``"<grade>.hi"``.
    """
    last = self.n_grades - 1
    messages = [
        f"{n}: {lo:.6g} <= p < {hi:.6g}" + (" (top band, closed at hi)" if i == last else "")
        for i, (n, (lo, hi)) in enumerate(self._bands)
    ]
    messages.append(
        "assignment is half-open, lo <= p < hi, with the top band closed at its upper "
        "edge; band inversion (calibrated_bands_to_raw) returns the same half-open "
        "intervals on the raw axis"
    )
    prov = self.provenance
    if prov is not None:
        messages.append(
            "built by build_masterscale: " + ", ".join(f"{k}={v}" for k, v in prov.items())
        )
    return Interpretation(
        method="Masterscale",
        param_names=tuple(f"{n}.{side}" for n in self._names for side in ("lo", "hi")),
        param_values=tuple(float(b) for _, (lo, hi) in self._bands for b in (lo, hi)),
        messages=tuple(messages),
    )

to_dict

to_dict() -> dict[str, object]

Versioned JSON-native snapshot (schema 1, the envelope every class uses).

Source code in src/probcal/masterscale.py

def to_dict(self) -> dict[str, object]:
    """Versioned JSON-native snapshot (schema 1, the envelope every class uses)."""
    return envelope(
        self,
        params={"bands": {n: [lo, hi] for n, (lo, hi) in self._bands}},
        state={"provenance": encode_value(self.provenance)},
        fit_meta={},
    )

from_dict classmethod

from_dict(d: dict) -> Masterscale

Rebuild from :meth:to_dict output.

RAISES	DESCRIPTION
ValueError	If the schema version is unknown or the payload class differs.

Source code in src/probcal/masterscale.py

@classmethod
def from_dict(cls, d: dict) -> "Masterscale":
    """Rebuild from :meth:`to_dict` output.

    Raises
    ------
    ValueError
        If the schema version is unknown or the payload class differs.
    """
    check_payload(cls, d)
    bands = {str(n): (float(b[0]), float(b[1])) for n, b in d["params"]["bands"].items()}
    prov = decode_value(d["state"]["provenance"], arrays=False)
    return cls(bands, provenance=prov)  # type: ignore[arg-type]