"""Audit evaluation evidence against an immutable generalization contract. This dependency-free artifact does not prove that a model will generalize. It turns a deployment claim into a versioned, content-addressed object and refuses evaluation evidence that does not match the population, time, model, sampling, and loss assumptions in that claim. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import math import re from typing import Any _IDENTIFIER = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:/@-]{0,127}$") _MAX_ALLOWED_RECORDS = 100_000 _MAX_ALLOWED_SLICES = 64 def _require_identifier(name: str, value: str) -> None: if not isinstance(value, str) or not _IDENTIFIER.fullmatch(value): raise ValueError(f"{name} must be a non-empty stable identifier") def _require_owner(name: str, value: str) -> None: _require_identifier(name, value) if not value.startswith("team:"): raise ValueError(f"{name} must name an accountable team: owner") def _require_int(name: str, value: int, *, minimum: int = 0) -> None: if isinstance(value, bool) or not isinstance(value, int) or value < minimum: raise ValueError(f"{name} must be an integer >= {minimum}") def _require_probability(name: str, value: float) -> None: if isinstance(value, bool) or not isinstance(value, (int, float)): raise TypeError(f"{name} must be a real number") if not math.isfinite(value): raise ValueError(f"{name} must be finite") if not 0.0 <= value <= 1.0: raise ValueError(f"{name} must be between 0 and 1") def _canonical(value: Any) -> Any: if isinstance(value, dict): return {key: _canonical(item) for key, item in 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 GeneralizationContract: contract_version: str task_id: str population_id: str decision_point: str prediction_horizon_days: int outcome_definition: str loss_id: str model_version: str evaluation_dataset_revision: str sampling_frame_id: str required_slices: tuple[str, ...] minimum_records: int minimum_slice_records: int max_records: int maximum_empirical_risk: float maximum_slice_risk: float model_owner: str data_owner: str def __post_init__(self) -> None: for name in ( "contract_version", "task_id", "population_id", "decision_point", "outcome_definition", "model_version", "evaluation_dataset_revision", "sampling_frame_id", ): _require_identifier(name, getattr(self, name)) if self.loss_id != "brier-v1": raise ValueError("this implementation supports loss_id=brier-v1 only") _require_int("prediction_horizon_days", self.prediction_horizon_days, minimum=1) _require_int("minimum_records", self.minimum_records, minimum=1) _require_int("minimum_slice_records", self.minimum_slice_records, minimum=1) _require_int("max_records", self.max_records, minimum=1) if self.max_records > _MAX_ALLOWED_RECORDS: raise ValueError(f"max_records cannot exceed {_MAX_ALLOWED_RECORDS}") if self.minimum_records > self.max_records: raise ValueError("minimum_records cannot exceed max_records") if self.minimum_slice_records > self.max_records: raise ValueError("minimum_slice_records cannot exceed max_records") if not isinstance(self.required_slices, tuple) or not self.required_slices: raise TypeError("required_slices must be a non-empty immutable tuple") if len(self.required_slices) > _MAX_ALLOWED_SLICES: raise ValueError(f"required_slices cannot exceed {_MAX_ALLOWED_SLICES}") for index, slice_id in enumerate(self.required_slices): _require_identifier(f"required_slices[{index}]", slice_id) if len(set(self.required_slices)) != len(self.required_slices): raise ValueError("required_slices must be unique") _require_probability("maximum_empirical_risk", self.maximum_empirical_risk) _require_probability("maximum_slice_risk", self.maximum_slice_risk) _require_owner("model_owner", self.model_owner) _require_owner("data_owner", self.data_owner) @property def contract_id(self) -> str: return _content_id(self.contract_version, asdict(self)) @dataclass(frozen=True) class EvaluationRecord: record_id: str entity_id: str population_id: str decision_day: int feature_as_of_day: int event_day: int | None label_as_of_day: int prediction: float outcome: int slice_ids: tuple[str, ...] @dataclass(frozen=True) class EvaluationSet: contract_id: str evaluation_id: str dataset_revision: str sampling_frame_id: str model_version: str records: tuple[EvaluationRecord, ...] @property def evidence_id(self) -> str: return _content_id("evaluation-evidence-v1", asdict(self)) @dataclass(frozen=True) class SliceRisk: slice_id: str count: int risk: float @dataclass(frozen=True) class GeneralizationAudit: contract_id: str evaluation_id: str evidence_id: str record_count: int empirical_risk: float slice_risks: tuple[SliceRisk, ...] worst_slice_id: str worst_slice_risk: float decision: str def _validate_record( contract: GeneralizationContract, record: EvaluationRecord, index: int ) -> None: if type(record) is not EvaluationRecord: raise TypeError(f"records[{index}] must be a concrete EvaluationRecord") _require_identifier(f"records[{index}].record_id", record.record_id) _require_identifier(f"records[{index}].entity_id", record.entity_id) if record.population_id != contract.population_id: raise ValueError(f"records[{index}] population_id does not match contract") _require_int(f"records[{index}].decision_day", record.decision_day) _require_int(f"records[{index}].feature_as_of_day", record.feature_as_of_day) _require_int(f"records[{index}].label_as_of_day", record.label_as_of_day) if record.feature_as_of_day > record.decision_day: raise ValueError(f"records[{index}] uses a feature from after the decision") maturity_day = record.decision_day + contract.prediction_horizon_days if record.label_as_of_day < maturity_day: raise ValueError(f"records[{index}] label is immature for the full prediction horizon") _require_probability(f"records[{index}].prediction", record.prediction) if isinstance(record.outcome, bool) or type(record.outcome) is not int: raise TypeError(f"records[{index}].outcome must be the integer 0 or 1") if record.outcome not in (0, 1): raise ValueError(f"records[{index}].outcome must be 0 or 1") if record.outcome == 0: if record.event_day is not None: raise ValueError(f"records[{index}] negative label cannot carry an event_day") else: if record.event_day is None: raise ValueError(f"records[{index}] positive label requires event_day") _require_int(f"records[{index}].event_day", record.event_day) if not record.decision_day < record.event_day <= maturity_day: raise ValueError(f"records[{index}] event_day is outside the prediction horizon") if record.event_day > record.label_as_of_day: raise ValueError(f"records[{index}] event_day is after label_as_of_day") if not isinstance(record.slice_ids, tuple) or not record.slice_ids: raise TypeError(f"records[{index}].slice_ids must be a non-empty immutable tuple") if len(record.slice_ids) > _MAX_ALLOWED_SLICES: raise ValueError(f"records[{index}].slice_ids is over the bounded limit") if len(set(record.slice_ids)) != len(record.slice_ids): raise ValueError(f"records[{index}].slice_ids must be unique") allowed = set(contract.required_slices) for slice_id in record.slice_ids: _require_identifier(f"records[{index}].slice_ids", slice_id) if slice_id not in allowed: raise ValueError(f"records[{index}] contains an undeclared slice") def audit_generalization( contract: GeneralizationContract, evidence: EvaluationSet ) -> GeneralizationAudit: """Validate evidence identity and compute bounded Brier risk by slice.""" if type(contract) is not GeneralizationContract: raise TypeError("contract must be a concrete GeneralizationContract") GeneralizationContract.__post_init__(contract) if type(evidence) is not EvaluationSet: raise TypeError("evidence must be a concrete EvaluationSet") if evidence.contract_id != contract.contract_id: raise ValueError("evidence contract_id does not match the full contract identity") _require_identifier("evaluation_id", evidence.evaluation_id) if evidence.dataset_revision != contract.evaluation_dataset_revision: raise ValueError("evidence dataset_revision does not match contract") if evidence.sampling_frame_id != contract.sampling_frame_id: raise ValueError("evidence sampling_frame_id does not match contract") if evidence.model_version != contract.model_version: raise ValueError("evidence model_version does not match contract") if not isinstance(evidence.records, tuple): raise TypeError("records must be an immutable tuple") record_count = len(evidence.records) if record_count > contract.max_records: raise ValueError("record count exceeds the contract's bounded workload") if record_count < contract.minimum_records: raise ValueError("record count is below minimum_records") seen_records: set[str] = set() seen_entities: set[str] = set() losses: list[float] = [] losses_by_slice = {slice_id: [] for slice_id in contract.required_slices} for index, record in enumerate(evidence.records): _validate_record(contract, record, index) if record.record_id in seen_records: raise ValueError("record_id values must be unique") if record.entity_id in seen_entities: raise ValueError("entity_id values must be independent in this evaluation") seen_records.add(record.record_id) seen_entities.add(record.entity_id) loss = (float(record.prediction) - float(record.outcome)) ** 2 losses.append(loss) for slice_id in record.slice_ids: losses_by_slice[slice_id].append(loss) slice_risks: list[SliceRisk] = [] for slice_id in contract.required_slices: values = losses_by_slice[slice_id] if len(values) < contract.minimum_slice_records: raise ValueError(f"slice {slice_id} is below minimum_slice_records") slice_risks.append( SliceRisk(slice_id=slice_id, count=len(values), risk=math.fsum(values) / len(values)) ) empirical_risk = math.fsum(losses) / record_count worst = max(slice_risks, key=lambda item: (item.risk, item.slice_id)) decision = ( "PASS" if empirical_risk <= contract.maximum_empirical_risk and worst.risk <= contract.maximum_slice_risk else "HOLD" ) return GeneralizationAudit( contract_id=contract.contract_id, evaluation_id=evidence.evaluation_id, evidence_id=evidence.evidence_id, record_count=record_count, empirical_risk=empirical_risk, slice_risks=tuple(slice_risks), worst_slice_id=worst.slice_id, worst_slice_risk=worst.risk, decision=decision, ) ILLUSTRATIVE_CONTRACT = GeneralizationContract( contract_version="generalization-contract-v1", task_id="illustrative-renewal-risk", population_id="illustrative-active-accounts", decision_point="weekly-renewal-review", prediction_horizon_days=30, outcome_definition="renewal-cancelled-within-horizon", loss_id="brier-v1", model_version="renewal-model-v3", evaluation_dataset_revision="renewal-eval-2026-08-20", sampling_frame_id="active-accounts-frame-2026-07", required_slices=("high-volume", "low-volume"), minimum_records=4, minimum_slice_records=2, max_records=1000, maximum_empirical_risk=0.08, maximum_slice_risk=0.13, model_owner="team:retention-modeling", data_owner="team:account-data", ) ILLUSTRATIVE_EVIDENCE = EvaluationSet( contract_id=ILLUSTRATIVE_CONTRACT.contract_id, evaluation_id="renewal-evaluation-run-0042", dataset_revision=ILLUSTRATIVE_CONTRACT.evaluation_dataset_revision, sampling_frame_id=ILLUSTRATIVE_CONTRACT.sampling_frame_id, model_version=ILLUSTRATIVE_CONTRACT.model_version, records=( EvaluationRecord("record-1", "account-1", "illustrative-active-accounts", 100, 100, None, 135, 0.1, 0, ("high-volume",)), EvaluationRecord("record-2", "account-2", "illustrative-active-accounts", 100, 99, 119, 135, 0.8, 1, ("high-volume",)), EvaluationRecord("record-3", "account-3", "illustrative-active-accounts", 100, 98, None, 135, 0.3, 0, ("low-volume",)), EvaluationRecord("record-4", "account-4", "illustrative-active-accounts", 100, 100, 129, 135, 0.6, 1, ("low-volume",)), ), ) def format_example() -> str: audit = audit_generalization(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_EVIDENCE) return "\n".join( ( "example=illustrative_only", f"contract_version={ILLUSTRATIVE_CONTRACT.contract_version}", f"evaluation={audit.evaluation_id}", f"evidence_id={audit.evidence_id}", f"records={audit.record_count}", f"empirical_risk={audit.empirical_risk:.3f}", f"worst_slice={audit.worst_slice_id}:{audit.worst_slice_risk:.3f}", f"decision={audit.decision}", ) ) if __name__ == "__main__": print(format_example())