Calibrating Materiality Thresholds from Historical Breaks

Setting Up Dynamic Routing Rules for High-Value Exceptions assumes someone already knows the right dollar bands. In practice those bands start as a guess — a compliance officer picks a round number, ops raises it after the review queue floods, and nobody can say whether the current floor is actually catching genuine breaks or just generating noise. This page replaces the guess with a measurement: replay a labelled window of closed exceptions, sweep a grid of candidate thresholds against their true disposition, and pick the point where raising the threshold further stops buying you meaningfully less review volume for the false-auto-resolve risk it adds. The output is a frozen, versioned envelope of thresholds — one per entity and currency pair — that Threshold-Based Routing Logic can consume directly, with every calibration run left as an auditable artifact rather than a Slack message nobody can find six months later.

Prerequisites

Step 1 — Load historical breaks with their true disposition

The model is deliberately narrower than a live exception record: it only needs the amount that would have driven a routing decision and the outcome a human later confirmed. Four dispositions cover every case that matters for calibration — whether the exception was auto-resolved or escalated, and whether that call was later found correct.

python
from pydantic import BaseModel, ConfigDict, field_validator
from decimal import Decimal
from datetime import datetime, timedelta, timezone
from enum import Enum
from uuid import UUID

class Disposition(str, Enum):
    AUTO_RESOLVED_CORRECT = "AUTO_RESOLVED_CORRECT"     # closed itself, correctly
    AUTO_RESOLVED_FALSE = "AUTO_RESOLVED_FALSE"          # closed itself, should not have
    ESCALATED_GENUINE = "ESCALATED_GENUINE"              # went to review, correctly
    ESCALATED_UNNECESSARY = "ESCALATED_UNNECESSARY"      # went to review, wasted the queue

class HistoricalBreak(BaseModel):
    model_config = ConfigDict(frozen=True)

    break_id: UUID
    entity: str
    currency: str
    break_amount: Decimal       # absolute size of the discrepancy that drove routing
    disposition: Disposition    # ground truth, set only after the case closed
    resolved_at: datetime       # timezone-aware UTC
    source_hash: str

    @field_validator("break_amount")
    @classmethod
    def non_negative_two_dp(cls, v: Decimal) -> Decimal:
        if v < 0:
            raise ValueError("break_amount must be non-negative")
        return v.quantize(Decimal("0.01"))

def load_window(records: list[dict], lookback_days: int, now: datetime) -> list[HistoricalBreak]:
    cutoff = now - timedelta(days=lookback_days)
    parsed = [HistoricalBreak(**r) for r in records]
    return [b for b in parsed if b.resolved_at >= cutoff]

AUTO_RESOLVED_FALSE and ESCALATED_GENUINE together define the “genuinely material” set — breaks that, regardless of what actually happened to them, truly needed a human to look. Everything else was immaterial noise. That split is what every downstream metric is measured against.

Step 2 — Sweep candidate thresholds and score each one

For every candidate threshold, replay the historical population as if that threshold had been live: anything at or below it would have auto-resolved, anything above escalates. Two numbers characterise the outcome — the false-auto-resolve rate, the fraction of genuinely material breaks the threshold would have let through, and the review-queue volume, the fraction of all breaks that still land on a human.

python
from dataclasses import dataclass

@dataclass(frozen=True)
class ThresholdScore:
    threshold: Decimal
    false_autoresolve_rate: Decimal
    review_queue_volume: Decimal
    genuine_caught: int
    genuine_total: int

GENUINE = (Disposition.AUTO_RESOLVED_FALSE, Disposition.ESCALATED_GENUINE)

def score_threshold(breaks: list[HistoricalBreak], threshold: Decimal) -> ThresholdScore:
    genuine = [b for b in breaks if b.disposition in GENUINE]
    genuine_missed = [b for b in genuine if b.break_amount <= threshold]
    escalated_now = [b for b in breaks if b.break_amount > threshold]
    return ThresholdScore(
        threshold=threshold,
        false_autoresolve_rate=(Decimal(len(genuine_missed)) / Decimal(len(genuine))
                                 if genuine else Decimal("0")),
        review_queue_volume=(Decimal(len(escalated_now)) / Decimal(len(breaks))
                              if breaks else Decimal("0")),
        genuine_caught=len(genuine) - len(genuine_missed),
        genuine_total=len(genuine),
    )

def sweep(breaks: list[HistoricalBreak], grid: list[Decimal]) -> list[ThresholdScore]:
    return [score_threshold(breaks, t) for t in sorted(grid)]

Keep the grid itself in Decimal — comparing a float-derived threshold against a Decimal break_amount either raises TypeError or silently coerces, and a coerced comparison is exactly the kind of penny-level drift that later fails an audit reconciliation.

Step 3 — Find the knee where marginal escalation stops paying off

As the threshold rises, review volume falls and false-auto-resolve rate climbs — a straightforward trade curve. The right pick is not the lowest error rate (that pushes review volume back toward 100%) and not the lowest volume (that pushes error toward 100%). It’s the point where the next increase in threshold buys back very little review relief for the risk it adds — the knee. First filter to the max_false_autoresolve ceiling from the configuration table below, then walk the eligible points for the one with the best marginal trade-off.

python
def find_knee(scores: list[ThresholdScore], max_false_autoresolve: Decimal) -> ThresholdScore:
    eligible = [s for s in scores if s.false_autoresolve_rate <= max_false_autoresolve]
    if not eligible:
        raise ValueError("OVERFIT_THRESHOLD: no candidate satisfies max_false_autoresolve")

    best = eligible[0]
    best_gain = Decimal("-1")
    for prev, cur in zip(eligible, eligible[1:]):
        d_rate = cur.false_autoresolve_rate - prev.false_autoresolve_rate
        d_volume = prev.review_queue_volume - cur.review_queue_volume
        gain = d_volume - d_rate  # marginal queue relief net of marginal risk added
        if gain > best_gain:
            best_gain = gain
            best = cur
    return best
Threshold sweep: false-auto-resolve rate vs review-queue volume, knee marked at $350 A plot frame spans the chart with five candidate thresholds on the x-axis: $50, $150, $350, $750 and $1500. Two curves are drawn against a shared 0 to 45 percent y-axis. The false-auto-resolve rate curve, in pink, starts near zero at $50 and rises steadily to fifteen percent at $1500. The review-queue volume curve, in blue, starts at forty-one percent at $50 and falls to five percent at $1500. The two curves cross in the middle of the chart. A dashed vertical guide line at $350 connects down from a violet callout box labelled chosen threshold $350, knee, marking the point past which the false-auto-resolve rate begins climbing faster than the review-queue volume falls. A legend below the frame identifies the pink line as false-auto-resolve rate, the blue line as review-queue volume, and the violet marker as the chosen threshold. 0% 15% 30% 45% $50 $150 $350 $750 $1500 Candidate materiality threshold $350 · knee false-auto-resolve rate review-queue volume chosen threshold

Step 4 — Express bands per entity and currency

A single global threshold flattens real differences: a thinly-traded currency pair generates fewer, larger breaks than a high-volume domestic feed, and pooling them biases the knee toward whichever population dominates the sample. Fit a threshold per (entity, currency) pair, but only where there is enough labelled history to trust the fit — below min_samples, fall back to the global band rather than let a handful of records dictate a routing decision.

python
from collections import defaultdict

def bands_by_segment(
    breaks: list[HistoricalBreak],
    grid: list[Decimal],
    max_false_autoresolve: Decimal,
    min_samples: int,
    global_band: ThresholdScore,
) -> dict[tuple[str, str], ThresholdScore]:
    segments: dict[tuple[str, str], list[HistoricalBreak]] = defaultdict(list)
    for b in breaks:
        segments[(b.entity, b.currency)].append(b)

    result: dict[tuple[str, str], ThresholdScore] = {}
    for key, seg_breaks in segments.items():
        if len(seg_breaks) < min_samples:
            result[key] = global_band          # INSUFFICIENT_HISTORY fallback
            continue
        seg_scores = sweep(seg_breaks, grid)
        result[key] = find_knee(seg_scores, max_false_autoresolve)
    return result

The fallback is not a compromise — it is the correct behaviour. A per-entity threshold fit on 12 samples is a fit to noise, and shipping it into Threshold-Based Routing Logic as if it were reliable is worse than sharing the global band across every thin segment.

Step 5 — Snapshot the chosen envelope for audit

Once every band is chosen, freeze the whole set into a single immutable envelope, hash the underlying dataset so the calibration run is reproducible, and emit a structured audit event before the envelope is allowed to feed live routing. Genuine breaks that clear the new threshold still need a deadline once they escalate — SLA-Gated Escalation Rules picks up from here.

python
import hashlib
import logging

log = logging.getLogger("materiality.calibration")

class ThresholdEnvelope(BaseModel):
    model_config = ConfigDict(frozen=True)

    envelope_id: UUID
    effective_from: datetime
    lookback_days: int
    bands: dict[str, Decimal]     # "entity|currency" -> chosen threshold
    max_false_autoresolve: Decimal
    dataset_hash: str

def snapshot_envelope(
    bands: dict[tuple[str, str], ThresholdScore],
    lookback_days: int,
    max_false_autoresolve: Decimal,
    breaks: list[HistoricalBreak],
    trace_id: UUID,
) -> ThresholdEnvelope:
    dataset_hash = hashlib.sha256(
        "|".join(sorted(b.source_hash for b in breaks)).encode("utf-8")
    ).hexdigest()
    envelope = ThresholdEnvelope(
        envelope_id=trace_id,
        effective_from=datetime.now(timezone.utc),
        lookback_days=lookback_days,
        bands={f"{k[0]}|{k[1]}": v.threshold for k, v in bands.items()},
        max_false_autoresolve=max_false_autoresolve,
        dataset_hash=dataset_hash,
    )
    log.info(
        "materiality.threshold_calibrated",
        extra={
            "trace_id": str(trace_id),
            "source_hash": dataset_hash,
            "match_decision": "ENVELOPE_SNAPSHOTTED",
            "band_count": len(envelope.bands),
            "max_false_autoresolve": str(max_false_autoresolve),
        },
    )
    return envelope

Treat the envelope as a deploy artifact, not a live-mutable config row: version it, review it, and gate its promotion the same way schema and rule changes are gated in CI/CD Gating for Financial Controls, so a bad calibration run can never reach production without a second set of eyes on the diff.

Configuration boundary table

Parameter Default Valid range Notes
lookback_days 180 30–730 Window of closed, labelled breaks to replay
threshold_grid $25–$5000, 25 steps log-spaced Candidate materiality levels swept per band
max_false_autoresolve 0.005 (0.5%) 0.001–0.02 Ceiling on genuine breaks the chosen threshold may let through
per_entity true boolean Fit an independent band per entity/currency pair
min_samples 40 20–500 Below this, a segment falls back to the global band

Verification and testing

Freeze a small labelled fixture with a known answer and assert the calibration recovers it — specifically, that the chosen threshold never exceeds the configured error ceiling even when the grid offers a more permissive option.

python
def test_knee_respects_error_ceiling():
    grid = [Decimal(x) for x in ("50", "150", "350", "750", "1500")]
    now = datetime.now(timezone.utc)
    breaks = [
        HistoricalBreak(break_id=UUID(int=1), entity="ACME", currency="USD",
                         break_amount=Decimal("40.00"), disposition=Disposition.AUTO_RESOLVED_CORRECT,
                         resolved_at=now, source_hash="h1"),
        HistoricalBreak(break_id=UUID(int=2), entity="ACME", currency="USD",
                         break_amount=Decimal("120.00"), disposition=Disposition.AUTO_RESOLVED_FALSE,
                         resolved_at=now, source_hash="h2"),
        HistoricalBreak(break_id=UUID(int=3), entity="ACME", currency="USD",
                         break_amount=Decimal("300.00"), disposition=Disposition.ESCALATED_GENUINE,
                         resolved_at=now, source_hash="h3"),
        HistoricalBreak(break_id=UUID(int=4), entity="ACME", currency="USD",
                         break_amount=Decimal("900.00"), disposition=Disposition.ESCALATED_UNNECESSARY,
                         resolved_at=now, source_hash="h4"),
    ]
    scores = sweep(breaks, grid)
    chosen = find_knee(scores, max_false_autoresolve=Decimal("0.5"))
    assert chosen.false_autoresolve_rate <= Decimal("0.5")
    assert chosen.threshold in grid

In a real fixture, scale each segment past min_samples and add a second window from a different quarter — asserting the chosen threshold is stable across windows is a stronger signal than any single-window fit, and catches a threshold that only looks good because it memorised one quarter’s noise.

Troubleshooting

  • INSUFFICIENT_HISTORY — a segment has fewer than min_samples labelled breaks, so its knee is unreliable. Root cause is usually a low-volume entity/currency pair or a lookback_days window that’s too short. Fix by widening lookback_days or accepting the global_band fallback from Step 4 rather than trusting a thin fit.
  • OVERFIT_THRESHOLD — find_knee raises because no candidate in the grid satisfies max_false_autoresolve, or the chosen point looks excellent on the calibration window but degrades badly on the next one. Root cause is a threshold grid fit to a single window’s noise. Fix by validating across at least two non-overlapping historical windows and keeping only thresholds that hold up on both.
  • DRIFTED_DISTRIBUTION — a threshold that scored well six months ago is now letting through far more genuine breaks than the audit expects. Root cause is a shift in the underlying break-amount distribution — a new business line, an FX repricing, a volume surge — that the frozen envelope no longer reflects. Fix by re-running the sweep on a rolling schedule and comparing distributions between windows before trusting the old envelope.
  • FLOAT_MATERIALITY — the chosen threshold is off by fractions of a cent, or comparisons against break_amount behave inconsistently near the boundary. Root cause is a float leaking into the threshold grid or the historical break_amount field upstream. Fix by constructing every grid value and every stored amount as Decimal, and quantizing to Decimal("0.01") at the validation boundary as in Step 1.

Part of Threshold-Based Routing Logic within Exception Routing & Human-in-the-Loop Workflows.