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)

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, sample_weight: object = None, *, y: 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
sample_weight	Weights for the portfolio mean. TYPE: array_like or None DEFAULT: None
y	Ignored; accepted for compatibility with the chain fit protocol. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
Self	The fitted offset.

Source code in src/probcal/offset.py

def fit(self, p: object, sample_weight: object = None, *, y: 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.
    sample_weight : array_like or None
        Weights for the portfolio mean.
    y : array_like or None, keyword-only
        Ignored; accepted for compatibility with the chain fit protocol.

    Returns
    -------
    Self
        The fitted offset.
    """
    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)")

        def gap(d: float) -> float:
            return float(np.average(expit(z + d), weights=w)) - target

        self.delta_ = bisect(gap, -_DELTA_BRACKET, _DELTA_BRACKET, tol=1e-14)
    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."""
    if not self.fitted_:
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    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_)

get_params

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

Constructor parameters as a dict (manual sklearn-compatible clone info).

Source code in src/probcal/offset.py

def get_params(self, deep: bool = True) -> dict[str, object]:
    """Constructor parameters as a dict (manual sklearn-compatible clone info)."""
    sig = inspect.signature(type(self).__init__)
    return {
        name: getattr(self, name)
        for name in sig.parameters
        if name not in ("self", "args", "kwargs")
    }

set_params

set_params(**params: object) -> Self

Set constructor parameters; unknown names raise ValueError.

Source code in src/probcal/offset.py

def set_params(self, **params: object) -> Self:
    """Set constructor parameters; unknown names raise ``ValueError``."""
    valid = self.get_params()
    for key, value in params.items():
        if key not in valid:
            raise ValueError(
                f"unknown parameter {key!r} for {type(self).__name__}; "
                f"valid: {sorted(valid)}"
            )
        setattr(self, key, value)
    return self

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."""
    if not self.fitted_:
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    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. 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].

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.

    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]``.
    """
    if not self.fitted_:
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    if not 0.0 <= lo <= hi <= 1.0:
        raise ValueError(f"need 0 <= lo <= hi <= 1, got lo={lo}, hi={hi}")
    from .base import UnattainableTargetError

    lo_b, hi_b = float(lo), float(hi)
    if buffer_logit > 0.0:
        if lo > 0.0:
            lo_b = float(expit(np.array([logit(np.array([lo]))[0] + buffer_logit]))[0])
        if hi < 1.0:
            hi_b = float(expit(np.array([logit(np.array([hi]))[0] - buffer_logit]))[0])
        if lo_b > hi_b:
            raise UnattainableTargetError(
                f"buffer_logit={buffer_logit} empties the calibrated interval [{lo}, {hi}]"
            )
    # 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 = float(expit(np.array([self.delta_ - _LOGIT_CLIP]))[0])
    gmax = float(expit(np.array([self.delta_ + _LOGIT_CLIP]))[0])
    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 float(logit(np.array([lo_b]))[0]) - self.delta_
    hi_z = np.inf if hi_b >= gmax else float(logit(np.array([hi_b]))[0]) - self.delta_
    if space == "logit":
        return lo_z, hi_z
    raw_lo = 0.0 if np.isneginf(lo_z) else float(expit(np.array([lo_z]))[0])
    raw_hi = 1.0 if np.isposinf(hi_z) else float(expit(np.array([hi_z]))[0])
    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 as :meth:BaseCalibrator.point_inverse — LogitOffset is not a BaseCalibrator subclass, so the fit-guard and validation are duplicated here rather than shared (the existing offset.py precedent, e.g. :meth:interval_inverse).

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 as
    :meth:`BaseCalibrator.point_inverse` — ``LogitOffset`` is not a
    ``BaseCalibrator`` subclass, so the fit-guard and validation are
    duplicated here rather than shared (the existing ``offset.py``
    precedent, e.g. :meth:`interval_inverse`).

    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.
    """
    if not self.fitted_:
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    if space not in ("probability", "logit"):
        raise ValueError(f"space must be 'probability' or 'logit', got {space!r}")
    from .base import _check_representable, _validate_point_targets

    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."""
    if not self.fitted_:
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    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.

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.
    """
    if not self.fitted_:
        raise RuntimeError("LogitOffset is not fitted; call fit() first")
    from . import __version__

    return {
        "probcal_schema": SCHEMA_VERSION,
        "probcal_version": __version__,
        "class": type(self).__name__,
        "params": {"delta": self.delta, "target_mean": self.target_mean},
        "state": {
            "delta_": self.delta_,
            "pre_mean_": self.pre_mean_,
            "post_mean_": self.post_mean_,
            "timestamp_": self.timestamp_,
        },
        "fit_meta": dict(getattr(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_schema(d)
    if d.get("class") != cls.__name__:
        raise ValueError(f"payload was written by {d.get('class')!r}, not {cls.__name__}")
    params = d.get("params", {})
    obj = cls(delta=params.get("delta"), target_mean=params.get("target_mean"))
    for key, value in d.get("state", {}).items():
        setattr(obj, key, value)
    obj.fit_meta_ = dict(d.get("fit_meta", {}))
    obj.fitted_ = True
    return obj

to_json

to_json(path: str | PathLike[str] | None = None, *, indent: int = 2) -> str | None

Serialize to JSON text, or to path when given (returns None then).

Source code in src/probcal/offset.py

def to_json(
    self, path: "str | os.PathLike[str] | None" = None, *, indent: int = 2
) -> str | None:
    """Serialize to JSON text, or to ``path`` when given (returns None then)."""
    text = json.dumps(self.to_dict(), indent=indent)
    if path is None:
        return text
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return None

from_json classmethod

from_json(path_or_str: object) -> LogitOffset

Load from a JSON string or a filesystem path.

Source code in src/probcal/offset.py

@classmethod
def from_json(cls, path_or_str: object) -> "LogitOffset":
    """Load from a JSON string or a filesystem path."""
    text = str(path_or_str)
    if not text.lstrip().startswith("{"):
        with open(text, encoding="utf-8") as fh:
            text = fh.read()
    return cls.from_dict(json.loads(text))

fingerprint

fingerprint() -> str

SHA-256 of the canonical serialized form, blind to versions and to the audit timestamp_ — identical fits fingerprint identically.

Source code in src/probcal/offset.py

def fingerprint(self) -> str:
    """SHA-256 of the canonical serialized form, blind to versions and
    to the audit ``timestamp_`` — identical fits fingerprint identically."""
    return fingerprint_of_dict(self.to_dict())

estimate_offset

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

Offset-only logistic MLE of delta given p, with a Fisher 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). The Fisher information for this one-parameter model is sum(w * q * (1 - q)) at q = sigma(logit(p) + delta), so the standard error is its inverse square root. That reading of the weights is the frequency one — w counts observations — so the SE is only valid for frequency weights; importance (or otherwise non-count) weights inflate the information and understate the SE.

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 (metrics.scores._prep performs this check).

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 a Fisher 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``). The Fisher information for
    this one-parameter model is ``sum(w * q * (1 - q))`` at
    ``q = sigma(logit(p) + delta)``, so the standard error is its inverse
    square root. That reading of the weights is the frequency one — ``w``
    counts observations — so the SE is only valid for frequency weights;
    importance (or otherwise non-count) weights inflate the information and
    understate the SE.

    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
        (``metrics.scores._prep`` performs this check).

    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)
    se = 1.0 / np.sqrt(float(np.sum(w * q * (1.0 - q))))
    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)

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)

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. 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). 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]

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 the 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_: object

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.

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 to the model (flow="cv") or scored directly by the already-trained model (flow="prefit"). TYPE: array_like
y	Binary outcomes in {0, 1}; both classes must be present. TYPE: array_like
sample_weight	Positive observation weights. TYPE: array_like or None DEFAULT: None

RETURNS	DESCRIPTION
Self	The fitted wrapper.

RAISES	DESCRIPTION
ValueError	If flow is not "prefit" or "cv".
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
        Calibration-set inputs, passed to the model (``flow="cv"``) 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.

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

    Raises
    ------
    ValueError
        If ``flow`` is not ``"prefit"`` or ``"cv"``.
    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}")
    X_arr = np.asarray(X, dtype=np.float64)
    y_arr = validate_binary_y(y)
    w_arr = validate_weights(sample_weight, len(y_arr))
    self.offsets_: list[LogitOffset] = []
    self.ensemble_: list[tuple[Any, BaseCalibrator]] = []
    if self.flow == "prefit":
        self.model_ = self.model
        s = _model_scores(self.model_, X_arr)
        self.calibrator_ = self._fresh_calibrator().fit(s, y_arr, sample_weight=w_arr)
        self._cal_scores = s
    else:
        self._fit_cv(X_arr, y_arr, w_arr)
    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(self._cal_scores, 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 to the deployed model (or every ensemble fold's model, averaged). TYPE: array_like

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
        New inputs, passed 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()
    p = self._base_predict(np.asarray(X, dtype=np.float64))
    for off in self.offsets_:
        p = off.transform(p)
    return p

predict_proba_2d

predict_proba_2d(X: object) -> ndarray

Sklearn-style (n, 2) probability matrix.

Source code in src/probcal/wrapper.py

def predict_proba_2d(self, X: object) -> np.ndarray:
    """Sklearn-style ``(n, 2)`` probability matrix."""
    p = self.predict_proba(X)
    return np.column_stack([1.0 - p, p])

offset_to

offset_to(target_mean: float | None = None, delta: float | None = None, X: 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, else on the stored calibration scores. 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 DEFAULT: None
delta	Mode A: the log-odds shift to apply directly; mutually exclusive with target_mean. TYPE: float or None DEFAULT: None
X	Inputs to compute the current pipeline output on; None uses the stored calibration scores instead. TYPE: array_like or None 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,
) -> 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, else on the stored
    calibration scores. The offset is never folded into
    the calibrator's parameters.

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

    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)
    else:
        if self._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._base_predict_from_scores(self._cal_scores)
    off = LogitOffset(delta=delta, target_mean=target_mean)
    off.fit(p_now)
    self.offsets_.append(off)
    return self

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 the full pipeline.

Composes right-to-left: the buffer shrinks the final interval, each offset subtracts its delta on the logit scale, and the calibrator's own inverse finishes the job. Returns bounds on the model's probability output (space="probability") or their logits.

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. 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
NotImplementedError	If the wrapper was fitted with ensemble=True (K distinct maps have no single preimage).
UnattainableTargetError	If the (buffered) interval does not intersect the pipeline's output range.
ValueError	If lo, hi are not ordered in [0, 1].

Source code in src/probcal/wrapper.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 the full pipeline.

    Composes right-to-left: the buffer shrinks the final interval, each
    offset subtracts its delta on the logit scale, and the calibrator's
    own inverse finishes the job. Returns bounds on the model's
    probability output (``space="probability"``) or their logits.

    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.

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

    Raises
    ------
    NotImplementedError
        If the wrapper was fitted with ``ensemble=True`` (K distinct
        maps have no single preimage).
    UnattainableTargetError
        If the (buffered) interval does not intersect the pipeline's
        output range.
    ValueError
        If ``lo``, ``hi`` are not ordered in ``[0, 1]``.
    """
    self._check_fitted()
    if self.ensemble_:
        raise NotImplementedError(
            "interval_inverse is not defined for the ensemble flow (K distinct maps); "
            "use ensemble=False for threshold translation"
        )
    if not 0.0 <= lo <= hi <= 1.0:
        raise ValueError(f"need 0 <= lo <= hi <= 1, got lo={lo}, hi={hi}")
    lo_b, hi_b = float(lo), float(hi)
    if buffer_logit > 0.0:
        if lo > 0.0:
            lo_b = float(expit(np.array([logit(np.array([lo]))[0] + buffer_logit]))[0])
        if hi < 1.0:
            hi_b = float(expit(np.array([logit(np.array([hi]))[0] - buffer_logit]))[0])
        if lo_b > hi_b:
            raise UnattainableTargetError(
                f"buffer_logit={buffer_logit} empties the calibrated interval [{lo}, {hi}]"
            )
    total_delta = sum(off.delta_ for off in self.offsets_)
    if total_delta != 0.0:
        if lo_b > 0.0:
            lo_b = float(expit(np.array([logit(np.array([lo_b]))[0] - total_delta]))[0])
        if hi_b < 1.0:
            hi_b = float(expit(np.array([logit(np.array([hi_b]))[0] - total_delta]))[0])
    return self.calibrator_.interval_inverse(lo_b, hi_b, space=space, buffer_logit=0.0)

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.
    """
    self._check_fitted()
    if self.ensemble_:
        parts = [cal.interpret() for _, cal in self.ensemble_]
    else:
        parts = [self.calibrator_.interpret()]
    parts += [off.interpret() for off in self.offsets_]
    names: tuple[str, ...] = ()
    values: tuple[float, ...] = ()
    messages: tuple[str, ...] = ()
    for part in parts:
        names += part.param_names
        values += part.param_values
        messages += part.messages
    return Interpretation(
        method=f"CalibratedModel[{', '.join(p.method for p in parts)}]",
        param_names=names,
        param_values=values,
        messages=messages,
    )

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=...). 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=...)``. 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"
        )
    from . import __version__

    ref_model = self.model_ if self.model_ is not None else self.model
    model_params: object = None
    if hasattr(ref_model, "get_params"):
        try:
            candidate = ref_model.get_params()
            json.dumps(candidate)
            model_params = candidate
        except (TypeError, ValueError):
            model_params = None
    return {
        "probcal_schema": SCHEMA_VERSION,
        "probcal_version": __version__,
        "class": type(self).__name__,
        "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": {
                "class_name": type(ref_model).__name__,
                "model_id": self.model_id,
                "params": model_params,
            },
        },
        "fit_meta": dict(getattr(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_schema(d)
    if d.get("class") != cls.__name__:
        raise ValueError(f"payload was written by {d.get('class')!r}, not {cls.__name__}")
    from ._registry import load

    params = d.get("params", {})
    state = d.get("state", {})
    calibrator = load(state["calibrator"])
    obj = cls(
        model,
        calibrator,  # type: ignore[arg-type]
        flow=params.get("flow", "prefit"),
        cv=params.get("cv", 5),
        ensemble=params.get("ensemble", False),
        random_state=params.get("random_state", 42),
        model_id=params.get("model_id"),
    )
    obj.calibrator_ = calibrator  # type: ignore[assignment]
    obj.offsets_ = [LogitOffset.from_dict(o) for o in state.get("offsets", [])]
    obj.ensemble_ = []
    obj.model_ = model
    obj._cal_scores = None  # type: ignore[assignment]
    obj.fit_meta_ = dict(d.get("fit_meta", {}))
    obj.fitted_ = True
    return obj

to_json

to_json(path: str | PathLike[str] | None = None, *, indent: int = 2) -> str | None

Serialize to JSON text, or to path when given (returns None then).

Source code in src/probcal/wrapper.py

def to_json(
    self, path: "str | os.PathLike[str] | None" = None, *, indent: int = 2
) -> str | None:
    """Serialize to JSON text, or to ``path`` when given (returns None then)."""
    text = json.dumps(self.to_dict(), indent=indent)
    if path is None:
        return text
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return None

from_json classmethod

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

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) -> "CalibratedModel":
    """Load from a JSON string or a filesystem path (see :meth:`from_dict`)."""
    text = str(path_or_str)
    if not text.lstrip().startswith("{"):
        with open(text, encoding="utf-8") as fh:
            text = fh.read()
    return cls.from_dict(json.loads(text), model=model)

fingerprint

fingerprint() -> str

SHA-256 of the canonical serialized form, version- and timestamp-blind.

Source code in src/probcal/wrapper.py

def fingerprint(self) -> str:
    """SHA-256 of the canonical serialized form, version- and timestamp-blind."""
    return fingerprint_of_dict(self.to_dict())

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: 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).

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. 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

fit

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

Run the nested selection and refit the winner on all data.

PARAMETER	DESCRIPTION
s	Raw scores/probabilities in [0, 1]. TYPE: array_like
y	Binary outcomes in {0, 1}; both classes must be present. TYPE: array_like
sample_weight	Positive observation weights. TYPE: array_like or None DEFAULT: None

RETURNS	DESCRIPTION
CalibratorSelector	self, with best_name_, best_calibrator_, and report_ set.

RAISES	DESCRIPTION
ValueError	If scoring is not one of the accepted criteria (plain ECE and Hosmer–Lemeshow are refused as selection criteria).

Source code in src/probcal/selection.py

def fit(self, s: object, y: object, sample_weight: object = None) -> "CalibratorSelector":
    """Run the nested selection and refit the winner on all data.

    Parameters
    ----------
    s : array_like
        Raw scores/probabilities in ``[0, 1]``.
    y : array_like
        Binary outcomes in ``{0, 1}``; both classes must be present.
    sample_weight : array_like or None
        Positive observation weights.

    Returns
    -------
    CalibratorSelector
        ``self``, with ``best_name_``, ``best_calibrator_``, and
        ``report_`` set.

    Raises
    ------
    ValueError
        If ``scoring`` is not one of the accepted criteria (plain ECE
        and Hosmer–Lemeshow are refused as selection criteria).
    """
    super().fit(s, y, sample_weight)
    return self

interpret

interpret() -> Interpretation

Delegate to the refitted winner.

Source code in src/probcal/selection.py

def interpret(self) -> Interpretation:
    """Delegate to the refitted winner."""
    return self.best_calibrator_.interpret()

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)

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

ATTRIBUTE	DESCRIPTION
frac	Cumulative fraction of observations, 1/n .. 1. 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)
    idx, m = _bin_index(p_arr, n_bins, strategy)
    w_sum = np.bincount(idx, weights=w, minlength=m)
    keep = w_sum > 0
    wy = np.bincount(idx, weights=w * y_arr, minlength=m)[keep]
    wp = np.bincount(idx, weights=w * p_arr, minlength=m)[keep]
    counts = np.bincount(idx, minlength=m)[keep]
    w_kept = w_sum[keep]
    pred_mean = wp / w_kept
    event_rate = wy / w_kept
    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	Validated (must match y in length) but not used: the LOESS fit itself is unweighted. 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
        Validated (must match ``y`` in length) but not used: the LOESS fit
        itself is unweighted.

    Returns
    -------
    SmoothReliabilityCurve
        Grid coordinates (probability and logit scale) and the smoothed
        event rate at each point.
    """
    y_arr, p_arr, _ = _prep(y, p, sample_weight)
    grid = _grid(p_arr, grid_size)
    rate = np.clip(loess(p_arr, y_arr, frac=frac, xeval=grid), 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 non-negative 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 non-negative 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, level: 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._lattice / _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).

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
sample_weight	Optional non-negative 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
level	Nominal coverage level of the ribbon; must satisfy 0 < level < 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 level 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,
    level: 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._lattice`` / ``_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``).

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    sample_weight : array_like or None, keyword-only
        Optional non-negative 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.
    level : float, keyword-only
        Nominal coverage level of the ribbon; must satisfy ``0 < level < 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 ``level`` 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
    """
    if not (0.0 < level < 1.0):
        raise ValueError("level must satisfy 0 < level < 1")
    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, m, width, t_lo, b = _smece_solve(t, mass, bins)
    lattice = None if m is None or b is None else (width, t_lo, b)

    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 - level) / 2.0
        ci_low = np.minimum(np.quantile(boot_rate, a, axis=0), event_rate)
        ci_high = np.maximum(np.quantile(boot_rate, 1.0 - a, axis=0), 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', level: 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 non-negative 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"
level	Nominal coverage level of the bands; must satisfy 0 < level < 1. TYPE: (float, keyword - only) DEFAULT: 0.9
n_resamples	Number of resamples used to build the bands. 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 level is not in (0, 1).

Notes

Bands are pointwise: at each grid point, level 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",
    level: 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 non-negative 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).
    level : float, keyword-only
        Nominal coverage level of the bands; must satisfy ``0 < level < 1``.
    n_resamples : int, keyword-only
        Number of resamples used to build the bands.
    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 ``level`` is not in ``(0, 1)``.

    Notes
    -----
    Bands are pointwise: at each grid point, ``level`` 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')
    if not (0.0 < level < 1.0):
        raise ValueError("level must satisfy 0 < level < 1")
    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, level, 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=level,
        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, mirroring metrics.ecce exactly so stat_max agrees with the metric. sd_null is the pointwise standard deviation of the walk under calibration — an envelope for reading, not a simultaneous band.

PARAMETER	DESCRIPTION
y	Outcomes and predicted probabilities. TYPE: array_like
p	Outcomes and predicted probabilities. TYPE: array_like
sample_weight	Optional non-negative 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, mirroring
    ``metrics.ecce`` exactly so ``stat_max`` agrees with the metric.
    ``sd_null`` is the pointwise standard deviation of the walk under
    calibration — an envelope for reading, not a simultaneous band.

    Parameters
    ----------
    y, p : array_like
        Outcomes and predicted probabilities.
    sample_weight : array_like or None, keyword-only
        Optional non-negative 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)
    order = np.argsort(p_arr, kind="stable")
    n = len(p_arr)
    wsum = w.sum()
    cumdev = np.cumsum(w[order] * (y_arr[order] - p_arr[order])) / wsum
    # Pointwise H0 SD; reduces to sqrt(cumsum(p(1-p)))/n for unit weights.
    sd_null = np.sqrt(np.cumsum(w[order] ** 2 * p_arr[order] * (1.0 - p_arr[order]))) / wsum
    frac = np.arange(1, n + 1) / n
    k = int(np.argmax(np.abs(cumdev)))
    return EcceCurve(
        frac=frac,
        cumdev=cumdev,
        sd_null=sd_null,
        stat_max=float(np.abs(cumdev[k])),
        argmax_frac=float(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), then draws pointwise confidence bands from the information-matrix ellipsoid — a Wald approximation of the LR-region inversion. The associated p-value tests the fitted polynomial against the identity. Where the band excludes the diagonal, the data reject calibration in that region.

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 non-negative weights, same length as y. TYPE: (array_like or None, keyword - only) DEFAULT: None

RETURNS	DESCRIPTION
BeltResult	Grid coordinates, both confidence bands, 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), then draws pointwise
    confidence bands from the information-matrix ellipsoid — a Wald
    approximation of the LR-region inversion. The
    associated p-value tests the fitted polynomial against the identity.
    Where the band excludes the diagonal, the data reject calibration in
    that region.

    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 non-negative weights, same length as ``y``.

    Returns
    -------
    BeltResult
        Grid coordinates, both confidence bands, selected polynomial degree,
        and the associated calibration-test p-value.
    """
    y_arr, p_arr, w = _prep(y, p, sample_weight)
    z = logit(p_arr)

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

    def loglik(beta: np.ndarray, deg: int) -> float:
        prob = np.clip(expit(design(deg, z) @ beta), 1e-12, 1.0 - 1e-12)
        return float(np.sum(w * (y_arr * np.log(prob) + (1.0 - y_arr) * np.log1p(-prob))))

    # 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 = loglik(fit.beta, 1)
    while degree < 4 and not fit.separation:
        cand = irls_logistic(design(degree + 1, z), y_arr, w=w)
        if cand.separation:
            break
        ll_cand = loglik(cand.beta, degree + 1)
        lr = max(2.0 * (ll_cand - ll), 0.0)
        p_add = 1.0 - float(gammainc_lower(0.5, lr / 2.0))  # chi-square df=1
        if p_add >= 0.05:
            break
        degree += 1
        fit, ll = cand, ll_cand

    # Associated calibration test: fitted polynomial vs the identity map.
    ll_null = float(
        np.sum(
            w
            * (
                y_arr * np.log(np.clip(p_arr, 1e-12, 1))
                + (1.0 - y_arr) * np.log(np.clip(1.0 - p_arr, 1e-12, 1))
            )
        )
    )
    lr_cal = max(2.0 * (ll - ll_null), 0.0)
    df = degree + 1
    p_value = 1.0 - float(gammainc_lower(df / 2.0, lr_cal / 2.0))

    # Pointwise bands from the information-matrix ellipsoid.
    X = design(degree, z)
    mu = expit(np.clip(X @ fit.beta, -30.0, 30.0))
    info = (X * (w * mu * (1.0 - mu))[:, None]).T @ X
    info_inv = np.linalg.inv(info + 1e-10 * np.eye(df))
    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 = {}
    for conf in confidence:
        radius = np.sqrt(chi2_ppf(conf, float(df)) * se_sq)
        bands[conf] = (expit(eta - radius), expit(eta + radius))
    lo_80, hi_80 = bands[confidence[0]]
    lo_95, hi_95 = bands[confidence[1]]
    return BeltResult(
        grid_p=grid_p,
        grid_logit=grid_z,
        lower_80=lo_80,
        upper_80=hi_80,
        lower_95=lo_95,
        upper_95=hi_95,
        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. Theory: docs/concepts/visualization.md.

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 = 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).

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 the 0.2.0 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 restores 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 (matching :func:probcal.metrics.evaluate's by= convention) 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. Each panel is drawn by a recursive call with rug=False, annotate=False (light default panels; pass stats=True for a per-panel stats box), 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). "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. Ignored 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. TYPE: (bool or MetricReport, 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 y/p are not given together, or risk_dist is not one of "rug", "split", None; or if by is given without both y and p, or with a length that does not match y.

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 = 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).

    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 the
    0.2.0 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`` restores 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`` (matching :func:`probcal.metrics.evaluate`'s
    ``by=`` convention) 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. Each panel is drawn by a recursive
    call with ``rug=False, annotate=False`` (light default panels; pass
    ``stats=True`` for a per-panel stats box), 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`).
    ``"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``.
        Ignored when ``by`` is given (a new figure of panels is always
        created).
    stats : bool or MetricReport, 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.
    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 ``y``/``p`` are not given together, or ``risk_dist`` is not one
        of ``"rug"``, ``"split"``, ``None``; or if ``by`` is given without
        both ``y`` and ``p``, or with a length that does not match ``y``.

    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()
    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")
        return _plot_reliability_faceted(y, p, by, scale=scale)
    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(6.5, 6))
        if scale == "logit":
            keep = (curve.event_rate > 0.0) & (curve.event_rate < 1.0)
            x, ylo = curve.pred_mean_logit[keep], logit(curve.ci_low[keep])
            yv, yhi = logit(curve.event_rate[keep]), logit(curve.ci_high[keep])
            diag = np.linspace(x.min() - 0.5, x.max() + 0.5, 50)
            ax.plot(diag, diag, ls="--", c=_GREY, lw=1, label="identity")
            ax.errorbar(
                x,
                yv,
                yerr=[np.maximum(yv - ylo, 0.0), np.maximum(yhi - yv, 0.0)],
                fmt="o",
                ms=4,
                capsize=2,
                color=_BLUE,
                label="binned",
            )
            if isinstance(smooth, KernelReliabilityCurve):
                _draw_kernel_curve(ax, smooth, "logit")
            elif smooth is not None:
                ax.plot(
                    smooth.grid_logit, logit(smooth.event_rate), lw=1.5, c=_ORANGE, label="smoothed"
                )
            _logit_axis(ax)
            ax.set_xlabel("predicted probability (logit scale)")
            ax.set_ylabel("event rate (logit scale)")
        else:
            ax.plot([0, 1], [0, 1], ls="--", c=_GREY, lw=1, label="identity")
            ax.errorbar(
                curve.pred_mean,
                curve.event_rate,
                yerr=[curve.event_rate - curve.ci_low, curve.ci_high - curve.event_rate],
                fmt="o",
                ms=4,
                capsize=2,
                color=_BLUE,
                label="binned",
            )
            if isinstance(smooth, KernelReliabilityCurve):
                _draw_kernel_curve(ax, smooth, "probability")
            elif smooth is not None:
                ax.plot(smooth.grid_p, smooth.event_rate, lw=1.5, c=_ORANGE, label="smoothed")
            ax.set_xlabel("predicted probability")
            ax.set_ylabel("event rate")

        boxed = False
        if y is not None and p is not None:
            y_arr = np.asarray(y, dtype=np.float64)
            p_arr = np.asarray(p, dtype=np.float64)
            show_density = rug and risk_dist is not None
            if show_density and risk_dist == "rug":
                ev = _rug_subsample(p_arr[y_arr == 1.0])
                ne = _rug_subsample(p_arr[y_arr == 0.0])
                if scale == "logit":
                    ev, ne = logit(ev), logit(ne)
                tf = ax.get_xaxis_transform()
                ax.plot(
                    ev, np.full(len(ev), 0.99), transform=tf,
                    ls="none", marker="|", ms=7, c=_RED, alpha=0.25,
                )  # fmt: skip
                ax.plot(
                    ne, np.full(len(ne), 0.01), transform=tf,
                    ls="none", marker="|", ms=7, c="#777777", alpha=0.18,
                )  # fmt: skip
            elif show_density and risk_dist == "split":
                _draw_split_risk_dist(ax, y_arr, p_arr, scale)
            if stats:
                txt = _stats_box_text(stats, y_arr, p_arr)
                ax.text(0.03, 0.97, txt, transform=ax.transAxes, va="top", fontsize=9, bbox=_BOX)
                boxed = True
            elif annotate:
                s = reliability_summary(y_arr, p_arr)
                txt = (
                    f"n = {s.n:,}\n"
                    f"events = {s.events:,}\n"
                    f"intercept = {s.intercept:+.3f}\n"
                    f"slope = {s.slope:.3f}\n"
                    f"ICI = {s.ici:.4f}\n"
                    f"E90 = {s.e90:.4f}\n"
                    f"Spiegelhalter p = {s.spiegelhalter_p:.3f}"
                )
                ax.text(0.03, 0.97, txt, transform=ax.transAxes, va="top", fontsize=9, bbox=_BOX)
                boxed = True
        if counts:
            # Count margin as a twin bar strip along the x-axis.
            ax2 = ax.twinx()
            xs = curve.pred_mean_logit if scale == "logit" else curve.pred_mean
            ax2.bar(xs, curve.count, width=np.ptp(xs) / (3 * len(xs) + 1), alpha=0.15, color=_GREY)
            ax2.set_yticks([])
        ax.legend(loc="lower right" if boxed else "upper left")
        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()
    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(6.5, 6))
        if scale == "logit":
            x = belt.grid_logit
            ax.plot(x, x, ls="--", c=_GREY, lw=1)
            ax.fill_between(
                x, logit(belt.lower_95), logit(belt.upper_95), color=_BLUE, alpha=0.2, label="95%"
            )
            ax.fill_between(
                x, logit(belt.lower_80), logit(belt.upper_80), color=_BLUE, alpha=0.35, label="80%"
            )
            _logit_axis(ax)
        else:
            x = belt.grid_p
            ax.plot(x, x, ls="--", c=_GREY, lw=1)
            ax.fill_between(x, belt.lower_95, belt.upper_95, color=_BLUE, alpha=0.2, label="95%")
            ax.fill_between(x, belt.lower_80, belt.upper_80, color=_BLUE, alpha=0.35, label="80%")
        ax.set_title(f"calibration belt (degree {belt.degree}, p = {belt.p_value:.3g})")
        ax.set_xlabel("predicted probability")
        ax.set_ylabel("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()
    with _plt.rc_context(_STYLE):
        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
        ):
            plot_reliability(curve, scale=scale, ax=ax)
            # Recolor the binned series to the panel's before/after semantics.
            for line in ax.lines:
                if line.get_label() == "binned":
                    line.set_color(color)
            for container in ax.containers:
                for artist in container.get_children():
                    artist.set_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. 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.
    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 _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(6.5, 4.5))
        p0, p1 = intervals[:, 0], intervals[:, 1]
        ax.fill_between(s, p0, p1, color=_BLUE, alpha=0.3, label="Venn–Abers interval")
        ax.plot(s, 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 _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(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 out of scope for this release.

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 out of scope for this release.

    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 _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(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 hasattr(result, "p_value") 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 _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(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")
    lo_g = float(logit(np.array([0.001]))[0])
    hi_g = float(logit(np.array([0.5]))[0])
    t = np.linspace(lo_g, hi_g, 200)
    lp = float(logit(np.array([offset.pre_mean_]))[0])
    lq = float(logit(np.array([offset.post_mean_]))[0])
    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(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(float(logit(np.array([offset.target_mean]))[0]), 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_}"
        )
        ax.text(0.03, 0.97, txt, transform=ax.transAxes, va="top", fontsize=9, bbox=_BOX)
        _logit_axis(ax)
        ax.set_xlabel("input probability (logit scale)")
        ax.set_ylabel("shifted probability (logit scale)")
        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. Additive: the default (False) call is pixel-identical to 0.2.0, 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. Additive: the default (``False``) call is
        pixel-identical to 0.2.0, 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 _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(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

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()
    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 _tr(x: np.ndarray) -> np.ndarray:
        if scale == "logit":
            return logit(np.clip(x, 1e-12, 1.0 - 1e-12))
        return np.asarray(x, dtype=np.float64)

    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(6.5, 6))

        grid = np.linspace(0.0, 1.0, 400)
        ybar_t = float(_tr(np.array([ybar]))[0])
        ax.plot(_tr(grid), _tr(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(_tr(grid), _tr(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(_tr(grid), _tr(y_lower), _tr(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(
                _tr(curve.pred_mean),
                _tr(curve.event_rate),
                s=sizes,
                color=_BLUE,
                zorder=5,
                label="binned",
            )
        else:
            result = corp_reliability(y_arr, p_arr, sample_weight=w, bands=None)
            lo, hi, level = result.block_lo, result.block_hi, result.block_level
            x_edges = np.empty(2 * len(lo))
            x_edges[0::2] = lo
            x_edges[1::2] = hi
            y_levels = np.repeat(level, 2)
            ax.step(_tr(x_edges), _tr(y_levels), where="post", color=_BLUE, lw=2, label="PAV fit")

        if scale == "logit":
            _logit_axis(ax)
            ax.set_xlabel("predicted probability (logit scale)")
            ax.set_ylabel("observed event rate (logit scale)")
        else:
            ax.set_xlabel("predicted probability")
            ax.set_ylabel("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()

    def _tr(x: np.ndarray) -> np.ndarray:
        if scale == "logit":
            return logit(np.clip(x, 1e-12, 1.0 - 1e-12))
        return np.asarray(x, dtype=np.float64)

    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(6.5, 6))
        lo, hi, level, weight = (
            result.block_lo,
            result.block_hi,
            result.block_level,
            result.block_weight,
        )
        domain = _tr(np.array([lo[0], hi[-1]]))
        ax.plot(domain, domain, ls="--", c=_GREY, lw=1, label="identity")

        if len(result.band_grid) > 0:
            ax.fill_between(
                _tr(result.band_grid),
                _tr(result.band_low),
                _tr(result.band_high),
                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 = np.empty(2 * len(lo))
        x_edges[0::2] = lo
        x_edges[1::2] = hi
        y_levels = np.repeat(level, 2)
        ax.step(_tr(x_edges), _tr(y_levels), where="post", color=_BLUE, lw=2, label="PAV fit")

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

        if scale == "logit":
            _logit_axis(ax)
            ax.set_xlabel("predicted probability (logit scale)")
            ax.set_ylabel("PAV-recalibrated probability (logit scale)")
        else:
            ax.set_xlabel("predicted probability")
            ax.set_ylabel("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}"
            )
            ax.text(0.03, 0.97, txt, transform=ax.transAxes, va="top", fontsize=9, bbox=_BOX)

        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 (probcal >= 0.3) 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"; if a SelectionReport is given without mcb/dsc columns (fitted before probcal 0.3); 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 (probcal >= 0.3) 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"``; if a
        ``SelectionReport`` is given without ``mcb``/``dsc`` columns
        (fitted before probcal 0.3); 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):
        if candidates.mcb is None or candidates.dsc is None or candidates.unc is None:
            raise ValueError("report has no mcb/dsc columns; refit with probcal>=0.3")
        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] = r.brier_mcb if score == "brier" else r.log_loss_mcb
            dsc[i] = r.brier_dsc if score == "brier" else r.log_loss_dsc
            unc = r.brier_unc if score == "brier" else r.log_loss_unc

    mean_score = mcb - dsc + unc

    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(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()
    with _plt.rc_context(_STYLE):
        if ax is None:
            _, ax = _plt.subplots(figsize=(6.5, 6))

        if diff:
            if not isinstance(curves, Mapping):
                raise ValueError(
                    "diff=True needs a mapping of exactly two {name: (y, p)} raw-data pairs"
                )
            pairs = list(curves.items())
            if len(pairs) != 2:
                raise ValueError(
                    "diff=True needs a mapping of exactly two {name: (y, p)} raw-data pairs"
                )
            name_a, pair_a = pairs[0]
            name_b, pair_b = pairs[1]
            if (
                not isinstance(pair_a, tuple)
                or not isinstance(pair_b, tuple)
                or len(pair_a) != 2
                or len(pair_b) != 2
            ):
                raise ValueError(
                    "diff=True needs a mapping of exactly two {name: (y, p)} raw-data pairs"
                )
            y_a, p_a = pair_a
            y_b, p_b = pair_b
            y_a, p_a, _w_a = _prep(y_a, p_a, None)
            y_b, p_b, _w_b = _prep(y_b, p_b, None)
            if len(y_a) != len(y_b):
                raise ValueError("diff=True needs paired (y, p) of equal length")

            theta = murphy_curve(y_a, p_a).thresholds
            score_a = murphy_curve(y_a, p_a, thresholds=theta).score
            score_b = murphy_curve(y_b, p_b, thresholds=theta).score
            delta = score_a - score_b

            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
            lo = np.percentile(boot, 5, axis=0)
            hi = np.percentile(boot, 95, axis=0)

            ax.fill_between(theta, lo, 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:
            for name, curve in curves.items():
                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"
                    )
                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

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", or if format="markdown" is given without path.
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"``, or if
        ``format="markdown"`` is given without ``path``.
    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 = np.asarray(y, dtype=np.float64)
    p_arr = np.asarray(p, dtype=np.float64)
    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 grades is not None and hasattr(grades, "fingerprint") and hasattr(grades, "assign"):
        fp_pairs.append(("masterscale", grades.fingerprint()))  # type: ignore[attr-defined]
    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)

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; affine_logit_coeffs_ (when not None) enables the exact affine path. TYPE: fitted calibrator
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
        Any object with ``predict_proba``; ``affine_logit_coeffs_`` (when not
        None) enables the exact affine path.
    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}")
    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))

    target = g_work(s)

    if use_affine:
        assert coeffs is not None
        a, b = coeffs
        phi_adj = a * phi_arr
        base_adj = a * base_arr + b
        method_used = "affine-exact"
    else:
        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.

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). Bands are inverted as closed intervals; the scale's own assign is half-open (lo <= p < hi), which differs only at a shared edge. 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. 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.

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.

    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). Bands are
        inverted as closed intervals; the scale's own ``assign`` is half-open
        (``lo <= p < hi``), which differs only at a shared edge.
    space : {"probability", "logit"}, keyword-only
        Scale of the returned bounds.
    buffer_logit : float, keyword-only
        Robustness margin applied in logit space before inversion.

    Returns
    -------
    dict
        Mapping of grade label to ``(raw_lo, raw_hi)`` bounds, on the scale
        requested by ``space``.
    """
    if hasattr(bands, "bands"):  # a Masterscale
        bands = bands.bands  # type: ignore[attr-defined]
    return {
        grade: calibrated_interval_to_raw(
            calibrator, lo, hi, space=space, buffer_logit=buffer_logit
        )
        for grade, (lo, hi) in bands.items()  # type: ignore[attr-defined]
    }

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. 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 (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, or names.

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.
    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 (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``, or ``names``.
    """
    y_arr = validate_binary_y(y)
    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))
    if n_grades < 1:
        raise ValueError("n_grades must be >= 1")
    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, int(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(Masterscale.from_edges(edges, names=labels).bands, provenance=provenance)

datasets

Synthetic dataset generators (make_pd_portfolio).

PdPortfolio dataclass

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

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). 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 absorbs intercept plus a portfolio-level anchor solved so that mean(p_true) == event_rate (unique by monotonicity, via bisection). With slope=1, asymmetry=0, intercept=0 the scores are exactly calibrated.

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	Additional log-odds shift applied before the mean anchor is solved. 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`` absorbs ``intercept`` plus a
    portfolio-level anchor solved so that ``mean(p_true) == event_rate``
    (unique by monotonicity, via bisection). With ``slope=1, asymmetry=0,
    intercept=0`` the scores are exactly calibrated.

    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
        Additional log-odds shift applied before the mean anchor is solved.
    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
    core = a_lo * np.log(s) - a_hi * np.log1p(-s) + intercept

    identity_case = slope == 1.0 and asymmetry == 0.0 and intercept == 0.0
    if identity_case:
        p_true = s.copy()
    else:

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

        c_anchor = bisect(gap, -60.0, 60.0, tol=1e-14)
        p_true = expit(core + c_anchor)
    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])

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, as before.

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``, as before.
    """
    self.calibrator_.fit(s, y, sample_weight=sample_weight)
    p = self.calibrator_.predict_proba(s)
    for off in self.offsets_:
        off.fit(p, sample_weight=sample_weight, y=y)
        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, arg-type]
    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()
    from .base import UnattainableTargetError

    if not 0.0 <= lo <= hi <= 1.0:
        raise ValueError(f"need 0 <= lo <= hi <= 1, got lo={lo}, hi={hi}")
    lo_b, hi_b = float(lo), float(hi)
    if buffer_logit > 0.0:
        if lo > 0.0:
            lo_b = float(expit(np.array([logit(np.array([lo]))[0] + buffer_logit]))[0])
        if hi < 1.0:
            hi_b = float(expit(np.array([logit(np.array([hi]))[0] - buffer_logit]))[0])
        if lo_b > hi_b:
            raise UnattainableTargetError(
                f"buffer_logit={buffer_logit} empties the calibrated interval [{lo}, {hi}]"
            )
    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()
    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_]
    names: tuple[str, ...] = ()
    values: tuple[float, ...] = ()
    messages: tuple[str, ...] = ()
    for part in parts:
        names += tuple(f"{part.method}.{n}" for n in part.param_names)
        values += part.param_values
        messages += part.messages
    return Interpretation(
        method=f"Chain[{', '.join(p.method for p in parts)}]",
        param_names=names,
        param_values=values,
        messages=messages,
    )

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()
    from . import __version__

    return {
        "probcal_schema": SCHEMA_VERSION,
        "probcal_version": __version__,
        "class": type(self).__name__,
        "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_schema(d)
    if d.get("class") != cls.__name__:
        raise ValueError(f"payload was written by {d.get('class')!r}, not {cls.__name__}")
    stages = [load(sd) for sd in d["state"]["stages"]]
    return cls(stages)

to_json

to_json(path: str | PathLike[str] | None = None, *, indent: int = 2) -> str | None

Serialize to JSON text, or to path when given (returns None then).

Source code in src/probcal/chain.py

def to_json(
    self, path: "str | os.PathLike[str] | None" = None, *, indent: int = 2
) -> str | None:
    """Serialize to JSON text, or to ``path`` when given (returns None then)."""
    text = json.dumps(self.to_dict(), indent=indent)
    if path is None:
        return text
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return None

from_json classmethod

from_json(path_or_str: object) -> Chain

Load from a JSON string or a filesystem path.

Source code in src/probcal/chain.py

@classmethod
def from_json(cls, path_or_str: object) -> "Chain":
    """Load from a JSON string or a filesystem path."""
    text = str(path_or_str)
    if not text.lstrip().startswith("{"):
        with open(text, encoding="utf-8") as fh:
            text = fh.read()
    return cls.from_dict(json.loads(text))

fingerprint

fingerprint() -> str

SHA-256 of the canonical serialized form (stages included).

Source code in src/probcal/chain.py

def fingerprint(self) -> str:
    """SHA-256 of the canonical serialized form (stages included)."""
    return fingerprint_of_dict(self.to_dict())

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)

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``.
    """
    from .. import __version__

    return {
        "probcal_schema": SCHEMA_VERSION,
        "probcal_version": __version__,
        "class": type(self).__name__,
        "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.

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.

    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_schema(d)
    if d.get("class") != cls.__name__:
        raise ValueError(f"payload was written by {d.get('class')!r}, not {cls.__name__}")
    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"]),
    )

to_json

to_json(path: str | PathLike[str] | None = None, *, indent: int = 2) -> str | None

Serialize to JSON text, or to path when given (returns None then).

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

def to_json(
    self, path: "str | os.PathLike[str] | None" = None, *, indent: int = 2
) -> str | None:
    """Serialize to JSON text, or to ``path`` when given (returns None then)."""
    text = json.dumps(self.to_dict(), indent=indent)
    if path is None:
        return text
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return None

from_json classmethod

from_json(path_or_str: object, *, model: object = None) -> AppliedAction

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

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

@classmethod
def from_json(cls, path_or_str: object, *, model: object = None) -> "AppliedAction":
    """Load from a JSON string or a filesystem path (see :meth:`from_dict`)."""
    text = str(path_or_str)
    if not text.lstrip().startswith("{"):
        with open(text, encoding="utf-8") as fh:
            text = fh.read()
    return cls.from_dict(json.loads(text), model=model)

fingerprint

fingerprint() -> str

SHA-256 of the canonical serialized form (version/timestamp blind).

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

def fingerprint(self) -> str:
    """SHA-256 of the canonical serialized form (version/timestamp blind)."""
    return fingerprint_of_dict(self.to_dict())

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')

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 (per-grade processes join automatically when grade arrays are passed). TYPE: tuple of {"offset", "shape"} DEFAULT: ('offset', 'shape')
grades	Optional explicit grade universe; None discovers grades from the grade arrays. TYPE: tuple or None DEFAULT: None
mixture_grid	Positive 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. TYPE: tuple(lo, hi, count) DEFAULT: (-3.0, 3.0, 241)
min_history	Number of past batches 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 used by the plug-ins, and by the recommendation rule when recommendation_window="trailing"; None uses all past batches. 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 toplug_in_window, which predates any alarm. Whenplug_in_windowis also set, the window starts at the LATER of the two starts (max(onset_idx, n_batches - plug_in_window)), so a shortplug_in_windowstill bounds how far back the since-onset window can reach."trailing"is the escape hatch restoring 0.2.0 behaviour exactly for those diagnostic INPUTS: it ignores the onset estimate and usesplug_in_window(or all past batches) instead, unconditionally.onset_labeland the onset sentence inreasoningare populated under both modes — only the diagnostic window differs. When onset is unavailable (any step loaded from a pre-0.3 payload carries no increment), both modes fall back to"trailing"andonset_labelisNone``. 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

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:
    if not 0.0 < alpha < 1.0:
        raise ValueError(f"alpha must lie in (0, 1), got {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}"
        )
    self.alpha = alpha
    self.components = tuple(components)
    self.grades = grades
    self.mixture_grid = tuple(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; non-uniform weights break the exact martingale property and warn once (reporting parity). TYPE: array_like or None DEFAULT: None
grade	Optional per-observation grade labels, or a :class:probcal.Masterscale that assigns them from p; either activates the per-grade offset processes. The first masterscale seen is recorded as masterscale_fingerprint_ and serialized; 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.

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; non-uniform weights break the exact
        martingale property and warn once (reporting parity).
    grade : array_like, Masterscale, or None
        Optional per-observation grade labels, or a
        :class:`probcal.Masterscale` that assigns them from ``p``;
        either activates the per-grade offset processes. The first
        masterscale seen is recorded as ``masterscale_fingerprint_`` and
        serialized; 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.
    """
    y_arr = _validate_outcomes(y)
    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))
    if not self._warned_weights and not np.all(w_arr == w_arr[0]):
        warnings.warn(
            "non-uniform sample weights break the exact martingale property; "
            "the e-values remain reported for parity but the type-I guarantee "
            "is approximate",
            UserWarning,
            stacklevel=2,
        )
        self._warned_weights = True
    g_arr: np.ndarray | None = None
    if grade is not None:
        if hasattr(grade, "assign") and hasattr(grade, "fingerprint"):
            fp = grade.fingerprint()  # type: ignore[attr-defined]
            if self.masterscale_fingerprint_ is None:
                self.masterscale_fingerprint_ = fp
            elif fp != self.masterscale_fingerprint_:
                raise ValueError(
                    "this monitor was started with masterscale "
                    f"{self.masterscale_fingerprint_[:12]}...; a different masterscale "
                    f"({fp[:12]}...) changes the grade universe. Start a new monitor."
                )
            g_arr = np.asarray(grade.assign(p_arr)).astype(str)  # type: ignore[attr-defined]
        else:
            g_arr = np.asarray(grade).astype(str)
        if len(g_arr) != len(p_arr):
            raise ValueError("grade and p must have equal length")
    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), w_arr)
        self._log_shape += shape_inc
    log_e_increment = offset_inc + shape_inc

    if g_arr is not None:
        for g 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_log:
                self._grade_cs_log[g] = np.zeros(len(self._cs_grid))
                self._grade_cs_max[g] = np.zeros(len(self._cs_grid))
            mask = g_arr == g
            if history_ready:
                gz, gy, gw = self._past(grade=g)
                d_g = plug_in_delta(gz, gy, gw)
            else:
                d_g = 0.0
            zg, pg, yg, wg = z[mask], p_arr[mask], y_arr[mask], w_arr[mask]
            self._grade_procs[g].update(zg, pg, yg, wg, d_g)

            # Per-grade confidence sequence: same construction as the
            # global one, restricted to this grade's own batch slice and
            # using its own plug-in as the alternative.
            gq_alt = np.clip(pg if d_g == 0.0 else expit(zg + d_g), 1e-12, 1.0 - 1e-12)
            glog_q = np.log(gq_alt)
            glog_1mq = np.log1p(-gq_alt)
            gp0 = np.clip(expit(zg[None, :] + self._cs_grid[:, None]), 1e-12, 1.0 - 1e-12)
            gterms = yg * (glog_q[None, :] - np.log(gp0)) + (1.0 - yg) * (
                glog_1mq[None, :] - np.log1p(-gp0)
            )
            self._grade_cs_log[g] = self._grade_cs_log[g] + (wg[None, :] * gterms).sum(axis=1)
            self._grade_cs_max[g] = np.maximum(self._grade_cs_max[g], self._grade_cs_log[g])

    # Confidence sequence: e-process per shifted null, plug-in alternative.
    q_alt = np.clip(p_arr if delta_hat == 0.0 else expit(z + delta_hat), 1e-12, 1.0 - 1e-12)
    log_q = np.log(q_alt)
    log_1mq = np.log1p(-q_alt)
    p0 = np.clip(expit(z[None, :] + self._cs_grid[:, None]), 1e-12, 1.0 - 1e-12)
    terms = y_arr * (log_q[None, :] - np.log(p0)) + (1.0 - y_arr) * (
        log_1mq[None, :] - np.log1p(-p0)
    )
    self._cs_log += (w_arr[None, :] * terms).sum(axis=1)
    self._cs_max = np.maximum(self._cs_max, self._cs_log)

    # 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)
    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 {}
    if alarm_at is None:
        return MonitorReport(
            steps=tuple(self.steps_),
            alarm_at=None,
            recommendation="none",
            reasoning=("no alarm: the global e-process never reached 1/alpha",),
            alpha=self.alpha,
            grade_table=grade_table,
        )
    onset_idx = self._onset_index()
    onset_label = self.steps_[onset_idx].label if onset_idx is not None else None
    # Shared with apply_recommendation() (_onset_index,
    # _recommendation_window_start), so the diagnostic window here and
    # the action window there never disagree.
    start = self._recommendation_window_start(onset_idx)
    pz, py, pw = self._since(start)
    delta_now = plug_in_delta(pz, py, pw)
    e_shape = self.steps_[-1].e_shape
    lo, hi = self._slope_ci(pz, py, pw)
    slope_ok = lo <= 1.0 <= hi
    # Residual-shape check: does the 2-parameter Cox correction explain the
    # trailing window materially better than the offset-only correction?
    # (The shape e-process itself also fires under pure level drift — its
    # alternative family contains the intercept — so it cannot separate
    # the two failure modes on its own.)
    resid_lr = self._residual_shape_lr(pz, py, pw, delta_now)
    shape_needed = resid_lr > 3.841  # chi-square(1) at 5%
    reasoning = [
        f"alarm at {alarm_at!r}; trailing-window offset {delta_now:+.3f} log-odds",
        f"shape e-process {e_shape:.3g} vs 1/alpha = {1.0 / self.alpha:.1f} "
        "(reported; fires under level drift too, so not decisive alone)",
        f"trailing-window Cox slope 95% bootstrap CI [{lo:.3f}, {hi:.3f}] "
        + ("contains" if slope_ok else "excludes")
        + " 1",
        f"Cox-vs-offset residual LR on the trailing window {resid_lr:.2f} "
        + ("exceeds" if shape_needed else "is within")
        + " the chi-square(1) 5% bound 3.84",
        (
            f"estimated drift onset at {onset_label} (backward-CUSUM argmax of the "
            "plug-in log-LR increments — an estimate, not a test)"
            if onset_idx is not None
            else "drift onset unavailable: steps recorded before 0.3.0 carry no log-e "
            "increments (trailing window used)"
        ),
        "the recommendation is a diagnostic, not a test — see the monitoring chapter",
    ]
    recommendation = "re-offset" if (slope_ok and not shape_needed) else "re-fit"
    return MonitorReport(
        steps=tuple(self.steps_),
        alarm_at=alarm_at,
        recommendation=recommendation,
        reasoning=tuple(reasoning),
        alpha=self.alpha,
        grade_table=grade_table,
        onset_label=onset_label,
    )

apply_recommendation

apply_recommendation(target: object = None) -> object

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 from the recommendation window onward (:meth:_onset_index and :meth:_recommendation_window_start -- the same window :meth:report uses for its trailing-window diagnostics; the onset index is recomputed directly rather than looked up by rep.onset_label, since batch labels are opaque and may repeat, and is unavailable altogether when any step came from a pre-0.3 payload -- the window is then the trailing one, as in :meth:report), 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, and continuing to accumulate it would test a null nobody deploys any more -- the same reasoning the monitoring chapter gives for starting a new monitor after any re-calibration.

"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 -- a deep copy of target with the offset appended via .offset_to(delta=est.delta); 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: :meth:report and the estimation below read only the retained batch arrays; 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) -> object:
    """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 from the
    recommendation window onward (:meth:`_onset_index` and
    :meth:`_recommendation_window_start` -- the same window
    :meth:`report` uses for its trailing-window diagnostics; the onset
    index is recomputed directly rather than looked up by
    ``rep.onset_label``, since batch labels are opaque and may repeat,
    and is unavailable altogether when any step came from a pre-0.3
    payload -- the window is then the trailing one, as in
    :meth:`report`),
    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, and continuing to accumulate it
    would test a null nobody deploys any more -- the same reasoning
    the monitoring chapter gives for starting a new monitor after any
    re-calibration.

    ``"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` -- a deep copy of
      ``target`` with the offset appended via
      ``.offset_to(delta=est.delta)``; ``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: :meth:`report` and the estimation below
    read only the retained batch arrays; 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 ..chain import Chain
    from ..offset import LogitOffset, estimate_offset
    from ..wrapper import CalibratedModel
    from ._actions import AppliedAction

    if target is not None and not isinstance(target, (Chain, CalibratedModel)):
        raise TypeError(
            "target must be None, a Chain, or a CalibratedModel, got "
            f"{type(target).__name__}"
        )
    old_target_fp = target.fingerprint() if target is not None else None

    rep = self.report()
    kind = rep.recommendation
    old_fp = self.fingerprint()

    if kind == "none":
        audit: dict[str, Any] = {
            "alarm_at": rep.alarm_at,
            "onset_label": rep.onset_label,
            "old_monitor_fingerprint": old_fp,
            "new_monitor_fingerprint": old_fp,
            "offset_fingerprint": None,
            "old_target_fingerprint": old_target_fp,
            "new_target_fingerprint": old_target_fp,
            "delta": None,
            "se": None,
        }
        return AppliedAction(
            kind=kind, offset=None, composed=None, monitor=None, window=(), audit=audit
        )

    # By index, not label: labels are opaque and may repeat (_onset_index).
    onset_idx = self._onset_index()
    start = self._recommendation_window_start(onset_idx)
    labels = tuple(s.label for s in self.steps_[start:])

    if kind == "re-fit":
        audit = {
            "alarm_at": rep.alarm_at,
            "onset_label": rep.onset_label,
            "old_monitor_fingerprint": old_fp,
            "new_monitor_fingerprint": old_fp,
            "offset_fingerprint": None,
            "old_target_fingerprint": old_target_fp,
            "new_target_fingerprint": old_target_fp,
            "delta": None,
            "se": None,
        }
        return AppliedAction(
            kind=kind, offset=None, composed=None, monitor=None, window=labels, audit=audit
        )

    # kind == "re-offset"
    z_w, y_w, w_w = self._since(start)
    p_w = expit(z_w)
    est = estimate_offset(y_w, p_w, sample_weight=w_w)
    offset = LogitOffset(delta=est.delta).fit(p_w)

    composed: object | None = None
    new_target_fp = old_target_fp
    if isinstance(target, Chain):
        composed = Chain([target.calibrator_, *target.offsets_, offset])
        new_target_fp = composed.fingerprint()
    elif isinstance(target, CalibratedModel):
        composed = copy.deepcopy(target).offset_to(delta=est.delta)
        new_target_fp = composed.fingerprint()

    fresh = type(self)(**self._ctor_params())

    audit = {
        "alarm_at": rep.alarm_at,
        "onset_label": rep.onset_label,
        "old_monitor_fingerprint": old_fp,
        "new_monitor_fingerprint": fresh.fingerprint(),
        "offset_fingerprint": offset.fingerprint(),
        "old_target_fingerprint": old_target_fp,
        "new_target_fingerprint": new_target_fp,
        "delta": float(est.delta),
        "se": float(est.se),
    }
    return AppliedAction(
        kind=kind,
        offset=offset,
        composed=composed,
        monitor=fresh,
        window=labels,
        audit=audit,
    )

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)."""
    from .. import __version__

    p = self._ctor_params()
    state = self._state_dict()
    return {
        "probcal_schema": SCHEMA_VERSION,
        "probcal_version": __version__,
        "class": type(self).__name__,
        "params": {
            "alpha": p["alpha"],
            "components": list(p["components"]),
            "grades": list(p["grades"]) if p["grades"] is not None else None,
            "mixture_grid": list(p["mixture_grid"]),
            "delta_ci_grid": list(p["delta_ci_grid"]),
            "min_history": p["min_history"],
            "plug_in_window": p["plug_in_window"],
            "recommendation_window": p["recommendation_window"],
        },
        "state": state,
        "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_schema(d)
    if d.get("class") != cls.__name__:
        raise ValueError(f"payload was written by {d.get('class')!r}, not {cls.__name__}")
    params = dict(d["params"])
    params["components"] = tuple(params["components"])
    params["grades"] = tuple(params["grades"]) if params["grades"] is not None else None
    params["mixture_grid"] = tuple(params["mixture_grid"])
    params["delta_ci_grid"] = tuple(params["delta_ci_grid"])
    params["recommendation_window"] = params.get("recommendation_window", "since_onset")
    mon = cls(**params)
    st = d["state"]
    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).astype(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_log = np.asarray(st["cs_log"], dtype=np.float64)
    mon._cs_max = np.asarray(st["cs_max"], dtype=np.float64)
    mon._grade_cs_log = {
        g: np.asarray(a, dtype=np.float64) for g, a in st.get("grade_cs_log", {}).items()
    }
    mon._grade_cs_max = {
        g: np.asarray(a, dtype=np.float64) for g, a in st.get("grade_cs_max", {}).items()
    }
    mon._max_log_global = float(st["max_log_global"])
    mon._alarmed = bool(st["alarmed"])
    mon._warned_weights = bool(st["warned_weights"])
    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["e_grades"] = dict(sd["e_grades"])
        sd["grade_delta_ci"] = {
            g: (tuple(v) if v is not None else None)
            for g, v in sd.get("grade_delta_ci", {}).items()
        }
        mon.steps_.append(MonitorStep(**sd))
    return mon

to_json

to_json(path: str | PathLike[str] | None = None, *, indent: int = 2) -> str | None

Serialize to JSON text, or to path when given (returns None then).

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

def to_json(
    self, path: "str | os.PathLike[str] | None" = None, *, indent: int = 2
) -> str | None:
    """Serialize to JSON text, or to ``path`` when given (returns None then)."""
    text = json.dumps(self.to_dict(), indent=indent)
    if path is None:
        return text
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return None

from_json classmethod

from_json(path_or_str: object) -> CalibrationMonitor

Load from a JSON string or a filesystem path.

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

@classmethod
def from_json(cls, path_or_str: object) -> "CalibrationMonitor":
    """Load from a JSON string or a filesystem path."""
    text = str(path_or_str)
    if not text.lstrip().startswith("{"):
        with open(text, encoding="utf-8") as fh:
            text = fh.read()
    return cls.from_dict(json.loads(text))

fingerprint

fingerprint() -> str

SHA-256 of the canonical serialized state (version-blind).

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

def fingerprint(self) -> str:
    """SHA-256 of the canonical serialized state (version-blind)."""
    return fingerprint_of_dict(self.to_dict())

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, and None as well when any step carries no increment (a pre-0.3 payload). 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."""
    rows = [asdict(s) for s in self.steps]
    try:
        import pandas as pd
    except ImportError:
        return rows
    return pd.DataFrame(rows)

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] = dict(), log_e_increment: float | None = None)

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; absent for steps loaded from a pre-0.3 payload). TYPE: dict[str, tuple[float, float] | None]
log_e_increment	This batch's additive plug-in log-LR increment: the offset plug-in's bern_log_lr contribution (0.0 when delta_hat == 0) plus the shape plug-in's (0.0 when its plug-in is the identity). 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. Steps written by :meth:CalibrationMonitor.update always carry a float; steps loaded from a pre-0.3 payload carry None (that payload records no increments), and a monitor holding any such step reports no onset at all. TYPE: float or None

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 state (mon._cs_grid[mon._cs_max < -log(1 - level)]) and take their max. This needs the live monitor object (its _cs_grid/_cs_max arrays), 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)
>>> for seed in range(3):
...     d = make_pd_portfolio(n=500, random_state=seed)
...     rng = np.random.default_rng(seed)
...     y = (rng.random(500) < d.scores).astype(float)  # drift injected
...     _ = 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 state
      (``mon._cs_grid[mon._cs_max < -log(1 - level)]``) and take their
      max. This needs the live monitor object (its ``_cs_grid``/``_cs_max``
      arrays), 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)
    >>> for seed in range(3):
    ...     d = make_pd_portfolio(n=500, random_state=seed)
    ...     rng = np.random.default_rng(seed)
    ...     y = (rng.random(500) < d.scores).astype(float)  # drift injected
    ...     _ = mon.update(y, d.scores, label=f"b{seed}")
    >>> off = moc_offset(mon)
    >>> off.delta_ >= mon.steps_[-1].delta_hat
    True
    """
    if not isinstance(monitor_or_report, (CalibrationMonitor, MonitorReport)):
        raise TypeError(
            "moc_offset requires a CalibrationMonitor or a MonitorReport, got "
            f"{type(monitor_or_report).__name__}"
        )
    mon = monitor_or_report if isinstance(monitor_or_report, CalibrationMonitor) else None

    steps: Sequence[MonitorStep]
    if isinstance(monitor_or_report, CalibrationMonitor):
        steps = monitor_or_report.steps_
    else:
        steps = monitor_or_report.steps
    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 _cs_grid/_cs_max arrays, 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"
            )
        if not 0.0 < level < 1.0:
            raise ValueError("level must lie in (0, 1)")
        threshold = -np.log(1.0 - level)
        surviving = mon._cs_grid[mon._cs_max < threshold]
        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 k = sum(w * y) and n = sum(w) -- 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 ``k = sum(w * y)``
    and ``n = sum(w)`` -- 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)
    if not 0.0 < level < 1.0:
        raise ValueError("level must lie in (0, 1)")
    k = float(np.sum(w_arr * y_arr))
    n = float(np.sum(w_arr))
    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, rational approximations) 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), 1e-12, 1.0 - 1e-12)
    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) keeps closed intervals, since boundary points have measure zero on the raw scale; a validator reconciling counts uses the assignment rule above.

GradeTable dataclass

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

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)

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. Serialized with the object and shown by :meth:interpret. 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 keeps closed intervals; the two conventions differ only at a shared edge, which has measure zero on the raw scale.

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}")
        lo, hi = float(bounds[0]), float(bounds[1])
        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)
    object.__setattr__(
        self, "_provenance", dict(provenance) if provenance is not None else None
    )

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).

from_edges classmethod

from_edges(edges: object, names: object = None, lo: float = 0.0, hi: float = 1.0) -> 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

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
) -> "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.
    """
    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)})

index

index(p: object) -> ndarray

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

p goes through :func:probcal._validation.validate_scores, so a two-column predict_proba matrix is accepted.

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`` goes through :func:`probcal._validation.validate_scores`, so a
    two-column ``predict_proba`` matrix is accepted.

    Raises
    ------
    ValueError
        If any ``p`` lies outside ``[edges[0], edges[-1]]``.
    """
    p_arr = validate_scores(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_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))
    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.

Source code in src/probcal/masterscale.py

def interpret(self) -> Interpretation:
    """The band table in words, with the boundary convention and provenance."""
    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 keeps closed intervals"
    )
    if self._provenance is not None:
        messages.append(
            "built by build_masterscale: "
            + ", ".join(f"{k}={v}" for k, v in self._provenance.items())
        )
    return Interpretation(
        method="Masterscale",
        param_names=self._names,
        param_values=tuple(float(hi) for _, (_, hi) in self._bands),
        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)."""
    from . import __version__

    return {
        "probcal_schema": SCHEMA_VERSION,
        "probcal_version": __version__,
        "class": type(self).__name__,
        "params": {"bands": {n: [lo, hi] for n, (lo, hi) in self._bands}},
        "state": {"provenance": 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_schema(d)
    if d.get("class") != cls.__name__:
        raise ValueError(f"payload was written by {d.get('class')!r}, not {cls.__name__}")
    bands = {str(n): (float(b[0]), float(b[1])) for n, b in d["params"]["bands"].items()}
    return cls(bands, provenance=d.get("state", {}).get("provenance"))

to_json

to_json(path: str | PathLike[str] | None = None, *, indent: int = 2) -> str | None

Serialize to JSON text, or to path when given (returns None then).

Source code in src/probcal/masterscale.py

def to_json(
    self, path: "str | os.PathLike[str] | None" = None, *, indent: int = 2
) -> str | None:
    """Serialize to JSON text, or to ``path`` when given (returns None then)."""
    text = json.dumps(self.to_dict(), indent=indent)
    if path is None:
        return text
    with open(path, "w", encoding="utf-8") as fh:
        fh.write(text)
    return None

from_json classmethod

from_json(path_or_str: object) -> Masterscale

Load from a JSON string or a filesystem path.

Source code in src/probcal/masterscale.py

@classmethod
def from_json(cls, path_or_str: object) -> "Masterscale":
    """Load from a JSON string or a filesystem path."""
    text = str(path_or_str)
    if not text.lstrip().startswith("{"):
        with open(text, encoding="utf-8") as fh:
            text = fh.read()
    return cls.from_dict(json.loads(text))

fingerprint

fingerprint() -> str

SHA-256 of the canonical serialized form, blind to the writing version.

Source code in src/probcal/masterscale.py

def fingerprint(self) -> str:
    """SHA-256 of the canonical serialized form, blind to the writing version."""
    return fingerprint_of_dict(self.to_dict())