"""Qualify calibrated probabilities before applying a cost-bound threshold. All example probabilities, labels, and costs are invented. The implementation is deliberately binary and refuses to treat discrimination metrics as evidence that probabilities are calibrated for a decision policy. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import math import re from typing import Any, Iterable MAX_OBSERVATIONS = 20_000 MAX_BINS = 1_000 MIN_BINARY64_RESOLUTION = 1e-15 MAX_BINARY64_RESOLUTION = 1e-3 _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,127}$") def _identifier(name: str, value: object) -> str: if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value): raise ValueError(f"{name} must be a stable identifier") return value def _owner(name: str, value: object) -> str: checked = _identifier(name, value) if not checked.startswith("team:"): raise ValueError(f"{name} must identify an accountable team owner") return checked def _finite(name: str, value: object) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise ValueError(f"{name} must be a real number") checked = float(value) if not math.isfinite(checked): raise ValueError(f"{name} must be finite") return checked def _probability(name: str, value: object) -> float: checked = _finite(name, value) if checked < 0.0 or checked > 1.0: raise ValueError(f"{name} must be between zero and one") return checked def _resolved_probability(name: str, value: object, resolution: float) -> float: checked = _probability(name, value) if checked not in (0.0, 1.0) and min(checked, 1.0 - checked) < resolution: raise ValueError(f"{name} is below the contracted numeric resolution") return checked def _positive_int(name: str, value: object, maximum: int) -> int: if isinstance(value, bool) or not isinstance(value, int) or value <= 0: raise ValueError(f"{name} must be a positive integer") if value > maximum: raise ValueError(f"{name} exceeds {maximum}") return value def _binary_label(value: object) -> int: if type(value) is not int or value not in (0, 1): raise ValueError("label must be the integer 0 or 1") return value def _canonical(value: Any) -> Any: if isinstance(value, dict): return {key: _canonical(item) for key, item in sorted(value.items())} if isinstance(value, tuple): return [_canonical(item) for item in value] return value def _content_id(prefix: str, payload: dict[str, Any]) -> str: encoded = json.dumps( _canonical(payload), sort_keys=True, separators=(",", ":"), allow_nan=False ).encode("utf-8") return f"{prefix}@sha256:{hashlib.sha256(encoded).hexdigest()}" @dataclass(frozen=True) class CalibrationObservation: observation_id: str entity_id: str probability: float label: int event_definition: str population: str horizon: str cohort_id: str evaluation_id: str evaluation_window: str model_version: str calibrator_version: str data_version: str label_version: str def __post_init__(self) -> None: for field in ( "observation_id", "entity_id", "event_definition", "population", "horizon", "cohort_id", "evaluation_id", "evaluation_window", "model_version", "calibrator_version", "data_version", "label_version", ): _identifier(field, getattr(self, field)) _probability("probability", self.probability) _binary_label(self.label) @classmethod def capture(cls, **values: object) -> "CalibrationObservation": return cls( observation_id=_identifier("observation_id", values["observation_id"]), entity_id=_identifier("entity_id", values["entity_id"]), probability=_probability("probability", values["probability"]), label=_binary_label(values["label"]), event_definition=_identifier( "event_definition", values["event_definition"] ), population=_identifier("population", values["population"]), horizon=_identifier("horizon", values["horizon"]), cohort_id=_identifier("cohort_id", values["cohort_id"]), evaluation_id=_identifier("evaluation_id", values["evaluation_id"]), evaluation_window=_identifier( "evaluation_window", values["evaluation_window"] ), model_version=_identifier("model_version", values["model_version"]), calibrator_version=_identifier( "calibrator_version", values["calibrator_version"] ), data_version=_identifier("data_version", values["data_version"]), label_version=_identifier("label_version", values["label_version"]), ) @dataclass(frozen=True) class ThresholdContract: contract_version: str event_definition: str population: str horizon: str cohort_id: str evaluation_id: str evaluation_window: str decision_window: str model_version: str calibrator_version: str data_version: str label_version: str cost_table_version: str calibration_metric: str gap_metric: str binning_convention: str threshold_convention: str boundary_convention: str positive_action: str negative_action: str false_positive_cost: float false_negative_cost: float minimum_cost_ratio: float numeric_precision_convention: str minimum_numeric_resolution: float tie_tolerance: float tie_action: str bin_count: int minimum_bin_count: int minimum_observations: int maximum_observations: int maximum_brier_score: float maximum_calibration_gap: float model_owner: str calibration_owner: str label_owner: str cost_owner: str decision_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "event_definition", "population", "horizon", "cohort_id", "evaluation_id", "evaluation_window", "decision_window", "model_version", "calibrator_version", "data_version", "label_version", "cost_table_version", "positive_action", "negative_action", "tie_action", ): _identifier(field, getattr(self, field)) for field in ( "model_owner", "calibration_owner", "label_owner", "cost_owner", "decision_owner", ): _owner(field, getattr(self, field)) if self.positive_action == self.negative_action: raise ValueError("positive_action and negative_action must differ") if self.tie_action not in (self.positive_action, self.negative_action): raise ValueError("tie_action must name a declared action") if self.calibration_metric != "mean-brier-binary": raise ValueError("unsupported calibration_metric") if self.gap_metric != "maximum-absolute-equal-width-bin-gap": raise ValueError("unsupported gap_metric") if self.binning_convention != "left-closed-final-right-closed": raise ValueError("unsupported binning_convention") if self.threshold_convention != "binary-expected-cost-equality": raise ValueError("unsupported threshold_convention") if ( self.boundary_convention != "bounded-probability-envelope-then-scaled-cost-v1" ): raise ValueError("unsupported boundary_convention") if ( self.numeric_precision_convention != "binary64-resolved-probability-and-scaled-cost-v1" ): raise ValueError("unsupported numeric_precision_convention") resolution = _finite( "minimum_numeric_resolution", self.minimum_numeric_resolution ) if not MIN_BINARY64_RESOLUTION <= resolution <= MAX_BINARY64_RESOLUTION: raise ValueError("minimum_numeric_resolution is outside the safe range") false_positive = _finite("false_positive_cost", self.false_positive_cost) false_negative = _finite("false_negative_cost", self.false_negative_cost) if not ( resolution <= false_positive <= 1.0 / resolution and resolution <= false_negative <= 1.0 / resolution ): raise ValueError("misclassification costs are outside the safe numeric range") minimum_cost_ratio = _probability( "minimum_cost_ratio", self.minimum_cost_ratio ) if minimum_cost_ratio < resolution: raise ValueError("minimum_cost_ratio is below the numeric resolution") observed_cost_ratio = min(false_positive, false_negative) / max( false_positive, false_negative ) if observed_cost_ratio < minimum_cost_ratio: raise ValueError("misclassification cost ratio is below the policy minimum") threshold = self._scaled_threshold() if not 0.0 < threshold < 1.0: raise ValueError("cost ratio does not produce a representable interior threshold") tie_tolerance = _finite("tie_tolerance", self.tie_tolerance) if ( tie_tolerance < 0.0 or tie_tolerance > resolution or tie_tolerance >= min(threshold, 1.0 - threshold) ): raise ValueError("tie_tolerance reaches outside the interior boundary") _positive_int("bin_count", self.bin_count, MAX_BINS) _positive_int( "minimum_bin_count", self.minimum_bin_count, MAX_OBSERVATIONS ) _positive_int( "minimum_observations", self.minimum_observations, MAX_OBSERVATIONS ) _positive_int( "maximum_observations", self.maximum_observations, MAX_OBSERVATIONS ) if self.minimum_observations > self.maximum_observations: raise ValueError("minimum_observations exceeds maximum_observations") maximum_brier = _probability( "maximum_brier_score", self.maximum_brier_score ) maximum_gap = _probability( "maximum_calibration_gap", self.maximum_calibration_gap ) if maximum_brier < resolution or maximum_gap < resolution: raise ValueError("qualification thresholds are below numeric resolution") @property def threshold(self) -> float: return self._scaled_threshold() def _scaled_threshold(self) -> float: scale = max(self.false_positive_cost, self.false_negative_cost) scaled_false_positive = self.false_positive_cost / scale scaled_false_negative = self.false_negative_cost / scale denominator = scaled_false_positive + scaled_false_negative if not math.isfinite(denominator) or denominator <= 0.0: raise OverflowError("scaled cost threshold denominator is invalid") return scaled_false_positive / denominator @property def content_id(self) -> str: return _content_id("threshold-contract", asdict(self)) @dataclass(frozen=True) class CalibrationReport: report_id: str contract_content_id: str evidence_id: str observation_digest: str observation_count: int brier_score: float maximum_calibration_gap: float bin_counts: tuple[int, ...] bin_gaps: tuple[float, ...] status: str def __post_init__(self) -> None: for field in ( "report_id", "contract_content_id", "evidence_id", "observation_digest", ): _identifier(field, getattr(self, field)) _positive_int("observation_count", self.observation_count, MAX_OBSERVATIONS) _probability("brier_score", self.brier_score) _probability("maximum_calibration_gap", self.maximum_calibration_gap) if not isinstance(self.bin_counts, tuple) or not self.bin_counts: raise TypeError("bin_counts must be a non-empty immutable tuple") if not isinstance(self.bin_gaps, tuple) or len(self.bin_gaps) != len( self.bin_counts ): raise TypeError("bin_gaps must align with immutable bin_counts") for count in self.bin_counts: _positive_int("bin_count support", count, MAX_OBSERVATIONS) for gap in self.bin_gaps: _probability("bin_gap", gap) if self.status not in ("QUALIFIED", "HOLD_CALIBRATION"): raise ValueError("invalid calibration status") @dataclass(frozen=True) class ScoreEvidence: score_id: str entity_id: str probability: float event_definition: str population: str horizon: str cohort_id: str decision_window: str model_version: str calibrator_version: str data_version: str def __post_init__(self) -> None: for field in ( "score_id", "entity_id", "event_definition", "population", "horizon", "cohort_id", "decision_window", "model_version", "calibrator_version", "data_version", ): _identifier(field, getattr(self, field)) _probability("probability", self.probability) @property def content_id(self) -> str: return _content_id("score-evidence", asdict(self)) @dataclass(frozen=True) class ThresholdDecision: contract_content_id: str calibration_report_id: str score_id: str score_content_id: str threshold: float positive_expected_cost: float negative_expected_cost: float action: str def _observation_digest( observations: tuple[CalibrationObservation, ...] ) -> str: payload = {"observations": tuple(asdict(item) for item in observations)} return _content_id("calibration-observations", payload) def qualify_calibration( contract: ThresholdContract, *, evidence_id: str, observations: Iterable[CalibrationObservation], ) -> CalibrationReport: if type(contract) is not ThresholdContract: raise TypeError("contract must be a concrete ThresholdContract") contract.__post_init__() checked_evidence_id = _identifier("evidence_id", evidence_id) frozen = tuple(observations) if len(frozen) < contract.minimum_observations: raise ValueError("calibration evidence is below minimum_observations") if len(frozen) > contract.maximum_observations: raise ValueError("calibration evidence exceeds maximum_observations") if any(type(item) is not CalibrationObservation for item in frozen): raise TypeError( "observations must contain concrete CalibrationObservation values" ) observation_ids = tuple(item.observation_id for item in frozen) entity_ids = tuple(item.entity_id for item in frozen) if len(observation_ids) != len(set(observation_ids)): raise ValueError("duplicate calibration observation_id") if len(entity_ids) != len(set(entity_ids)): raise ValueError("duplicate calibration entity_id") expected_scope = ( contract.event_definition, contract.population, contract.horizon, contract.cohort_id, contract.evaluation_id, contract.evaluation_window, contract.model_version, contract.calibrator_version, contract.data_version, contract.label_version, ) for item in frozen: item.__post_init__() _resolved_probability( "probability", item.probability, contract.minimum_numeric_resolution ) observed_scope = ( item.event_definition, item.population, item.horizon, item.cohort_id, item.evaluation_id, item.evaluation_window, item.model_version, item.calibrator_version, item.data_version, item.label_version, ) if observed_scope != expected_scope: raise ValueError("calibration observation scope does not match contract") brier_score = math.fsum( (item.probability - item.label) ** 2 for item in frozen ) / len(frozen) bins: list[list[CalibrationObservation]] = [ [] for _ in range(contract.bin_count) ] for item in frozen: index = min(int(item.probability * contract.bin_count), contract.bin_count - 1) bins[index].append(item) if any(len(bucket) < contract.minimum_bin_count for bucket in bins): raise ValueError("a contracted calibration bin is below minimum_bin_count") bin_gaps = tuple( abs( math.fsum(item.probability for item in bucket) / len(bucket) - math.fsum(item.label for item in bucket) / len(bucket) ) for bucket in bins ) maximum_gap = max(bin_gaps) if not math.isfinite(brier_score) or not math.isfinite(maximum_gap): raise OverflowError("calibration metrics overflowed") status = ( "QUALIFIED" if brier_score <= contract.maximum_brier_score and maximum_gap <= contract.maximum_calibration_gap else "HOLD_CALIBRATION" ) digest = _observation_digest(frozen) report_payload = { "contract_content_id": contract.content_id, "evidence_id": checked_evidence_id, "observation_digest": digest, "observation_count": len(frozen), "brier_score": brier_score, "maximum_calibration_gap": maximum_gap, "bin_counts": tuple(len(bucket) for bucket in bins), "bin_gaps": bin_gaps, "status": status, } return CalibrationReport( report_id=_content_id("calibration-report", report_payload), **report_payload, ) def decide_threshold( contract: ThresholdContract, report: CalibrationReport, score: ScoreEvidence, ) -> ThresholdDecision: if type(contract) is not ThresholdContract: raise TypeError("contract must be a concrete ThresholdContract") if type(report) is not CalibrationReport: raise TypeError("report must be a concrete CalibrationReport") if type(score) is not ScoreEvidence: raise TypeError("score must be a concrete ScoreEvidence") contract.__post_init__() report.__post_init__() score.__post_init__() _resolved_probability( "probability", score.probability, contract.minimum_numeric_resolution ) if report.contract_content_id != contract.content_id: raise ValueError("calibration report belongs to another contract") expected_report_id = _content_id( "calibration-report", { "contract_content_id": report.contract_content_id, "evidence_id": report.evidence_id, "observation_digest": report.observation_digest, "observation_count": report.observation_count, "brier_score": report.brier_score, "maximum_calibration_gap": report.maximum_calibration_gap, "bin_counts": report.bin_counts, "bin_gaps": report.bin_gaps, "status": report.status, }, ) if report.report_id != expected_report_id: raise ValueError("calibration report content identity is invalid") if report.status != "QUALIFIED": raise ValueError("calibration evidence is not qualified for decisions") if len(report.bin_counts) != contract.bin_count: raise ValueError("calibration report bin contract is invalid") score_bin = min( int(score.probability * contract.bin_count), contract.bin_count - 1 ) if report.bin_counts[score_bin] < contract.minimum_bin_count: raise ValueError("score bin lacks contracted calibration support") expected_score_scope = ( contract.event_definition, contract.population, contract.horizon, contract.cohort_id, contract.decision_window, contract.model_version, contract.calibrator_version, contract.data_version, ) observed_score_scope = ( score.event_definition, score.population, score.horizon, score.cohort_id, score.decision_window, score.model_version, score.calibrator_version, score.data_version, ) if observed_score_scope != expected_score_scope: raise ValueError("score scope does not match threshold contract") scale = max(contract.false_positive_cost, contract.false_negative_cost) scaled_false_positive = contract.false_positive_cost / scale scaled_false_negative = contract.false_negative_cost / scale scaled_positive_cost = (1.0 - score.probability) * scaled_false_positive scaled_negative_cost = score.probability * scaled_false_negative if not math.isfinite(scaled_positive_cost) or not math.isfinite( scaled_negative_cost ): raise OverflowError("scaled expected decision cost overflowed") positive_cost = scaled_positive_cost * scale negative_cost = scaled_negative_cost * scale if not math.isfinite(positive_cost) or not math.isfinite(negative_cost): raise OverflowError("expected decision cost overflowed") if math.isclose( score.probability, contract.threshold, rel_tol=0.0, abs_tol=contract.tie_tolerance, ): action = contract.tie_action elif scaled_positive_cost < scaled_negative_cost: action = contract.positive_action else: action = contract.negative_action return ThresholdDecision( contract_content_id=contract.content_id, calibration_report_id=report.report_id, score_id=score.score_id, score_content_id=score.content_id, threshold=contract.threshold, positive_expected_cost=positive_cost, negative_expected_cost=negative_cost, action=action, ) def _example() -> None: contract = ThresholdContract( contract_version="calibrated-threshold-v1", event_definition="confirmed-payment-abuse", population="authenticated-card-payments", horizon="30-days-after-payment", cohort_id="consumer-us", evaluation_id="abuse-calibration-eval-2026-08", evaluation_window="2026-07", decision_window="2026-08", model_version="abuse-model-v5", calibrator_version="isotonic-v2", data_version="payment-outcomes-v8", label_version="confirmed-abuse-v3", cost_table_version="abuse-costs-2026-08", calibration_metric="mean-brier-binary", gap_metric="maximum-absolute-equal-width-bin-gap", binning_convention="left-closed-final-right-closed", threshold_convention="binary-expected-cost-equality", boundary_convention="bounded-probability-envelope-then-scaled-cost-v1", positive_action="manual-review", negative_action="auto-clear", false_positive_cost=2.0, false_negative_cost=8.0, minimum_cost_ratio=1e-12, numeric_precision_convention="binary64-resolved-probability-and-scaled-cost-v1", minimum_numeric_resolution=1e-12, tie_tolerance=1e-12, tie_action="manual-review", bin_count=2, minimum_bin_count=4, minimum_observations=8, maximum_observations=100, maximum_brier_score=0.15, maximum_calibration_gap=0.05, model_owner="team:abuse-modeling", calibration_owner="team:risk-measurement", label_owner="team:abuse-operations", cost_owner="team:risk-finance", decision_owner="team:payment-risk", ) values = ((0.1, 0), (0.2, 0), (0.3, 0), (0.4, 1), (0.6, 0), (0.7, 1), (0.8, 1), (0.9, 1)) observations = tuple( CalibrationObservation.capture( observation_id=f"cal-{index}", entity_id=f"payment-{index}", probability=probability, label=label, event_definition=contract.event_definition, population=contract.population, horizon=contract.horizon, cohort_id=contract.cohort_id, evaluation_id=contract.evaluation_id, evaluation_window=contract.evaluation_window, model_version=contract.model_version, calibrator_version=contract.calibrator_version, data_version=contract.data_version, label_version=contract.label_version, ) for index, (probability, label) in enumerate(values, start=1) ) report = qualify_calibration( contract, evidence_id="calibration-batch-001", observations=observations ) score = ScoreEvidence( score_id="score-1042", entity_id="payment-1042", probability=0.35, event_definition=contract.event_definition, population=contract.population, horizon=contract.horizon, cohort_id=contract.cohort_id, decision_window=contract.decision_window, model_version=contract.model_version, calibrator_version=contract.calibrator_version, data_version=contract.data_version, ) decision = decide_threshold(contract, report, score) print("example=illustrative_only") print(f"contract_version={contract.contract_version}") print(f"calibration_status={report.status}") print(f"brier={report.brier_score:.3f}") print(f"max_gap={report.maximum_calibration_gap:.3f}") print(f"threshold={decision.threshold:.3f}") print( "expected_costs=" f"{contract.negative_action}:{decision.negative_expected_cost:.3f}," f"{contract.positive_action}:{decision.positive_expected_cost:.3f}" ) print(f"action={decision.action}") if __name__ == "__main__": _example()