"""Audit one residual-tree partition under an immutable evidence contract. The example values are illustrative. This module intentionally audits a named candidate split; it does not search over features and then reuse the same gain as unbiased evidence for model selection. """ from __future__ import annotations from dataclasses import asdict, dataclass import hashlib import json import math import re from typing import Any, Iterable MAX_OBSERVATIONS = 10_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 _positive_int(name: str, value: object, maximum: int = MAX_OBSERVATIONS) -> 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 _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 ResidualObservation: observation_id: str entity_id: str feature_value: float label_value: float base_prediction: float sample_weight: float dataset_version: str feature_id: str feature_version: str target_id: str label_version: str base_model_version: str cohort_id: str window_id: str def __post_init__(self) -> None: for field in ( "observation_id", "entity_id", "dataset_version", "feature_id", "feature_version", "target_id", "label_version", "base_model_version", "cohort_id", "window_id", ): _identifier(field, getattr(self, field)) _finite("feature_value", self.feature_value) _finite("label_value", self.label_value) _finite("base_prediction", self.base_prediction) weight = _finite("sample_weight", self.sample_weight) if weight <= 0.0: raise ValueError("sample_weight must be positive") @property def residual(self) -> float: value = self.label_value - self.base_prediction if not math.isfinite(value): raise OverflowError("derived residual overflowed") return value @classmethod def capture( cls, *, observation_id: str, entity_id: str, feature_value: float, label_value: float, base_prediction: float, sample_weight: float, dataset_version: str, feature_id: str, feature_version: str, target_id: str, label_version: str, base_model_version: str, cohort_id: str, window_id: str, ) -> "ResidualObservation": return cls( observation_id=_identifier("observation_id", observation_id), entity_id=_identifier("entity_id", entity_id), feature_value=_finite("feature_value", feature_value), label_value=_finite("label_value", label_value), base_prediction=_finite("base_prediction", base_prediction), sample_weight=_finite("sample_weight", sample_weight), dataset_version=_identifier("dataset_version", dataset_version), feature_id=_identifier("feature_id", feature_id), feature_version=_identifier("feature_version", feature_version), target_id=_identifier("target_id", target_id), label_version=_identifier("label_version", label_version), base_model_version=_identifier( "base_model_version", base_model_version ), cohort_id=_identifier("cohort_id", cohort_id), window_id=_identifier("window_id", window_id), ) @dataclass(frozen=True) class PartitionContract: contract_version: str dataset_version: str feature_id: str feature_version: str target_id: str label_version: str cohort_id: str window_id: str base_model_version: str residual_definition: str loss_metric: str loss_cost_version: str gain_convention: str threshold_convention: str candidate_threshold: float minimum_child_count: int minimum_child_weight: float minimum_gain: float gain_roundoff_tolerance: float numeric_precision_convention: str minimum_numeric_resolution: float maximum_sample_weight: float maximum_gain: float maximum_absolute_residual: float maximum_observations: int data_owner: str label_owner: str model_owner: str decision_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "dataset_version", "feature_id", "feature_version", "target_id", "label_version", "cohort_id", "window_id", "base_model_version", "loss_cost_version", ): _identifier(field, getattr(self, field)) for field in ("data_owner", "label_owner", "model_owner", "decision_owner"): _owner(field, getattr(self, field)) if self.residual_definition != "label-minus-base-prediction": raise ValueError("unsupported residual_definition") if self.loss_metric != "weighted-residual-sse": raise ValueError("unsupported loss_metric") if self.gain_convention != "parent-sse-minus-child-sse": raise ValueError("unsupported gain_convention") if self.threshold_convention != "left-less-than-or-equal": raise ValueError("unsupported threshold_convention") if ( self.numeric_precision_convention != "binary64-bounded-inputs-and-gain-v1" ): raise ValueError("unsupported numeric_precision_convention") _finite("candidate_threshold", self.candidate_threshold) _positive_int("minimum_child_count", self.minimum_child_count) maximum_observations = _positive_int( "maximum_observations", self.maximum_observations ) resolution = _finite( "minimum_numeric_resolution", self.minimum_numeric_resolution ) child_weight = _finite("minimum_child_weight", self.minimum_child_weight) gain = _finite("minimum_gain", self.minimum_gain) tolerance = _finite( "gain_roundoff_tolerance", self.gain_roundoff_tolerance ) maximum_sample_weight = _finite( "maximum_sample_weight", self.maximum_sample_weight ) maximum_gain = _finite("maximum_gain", self.maximum_gain) residual_limit = _finite( "maximum_absolute_residual", self.maximum_absolute_residual ) if not MIN_BINARY64_RESOLUTION <= resolution <= MAX_BINARY64_RESOLUTION: raise ValueError("minimum_numeric_resolution is outside the safe range") if ( maximum_sample_weight < resolution or maximum_sample_weight > 1.0 / resolution ): raise ValueError("maximum_sample_weight is outside the safe range") if ( child_weight < resolution or child_weight > maximum_sample_weight * maximum_observations ): raise ValueError("minimum_child_weight is outside the safe range") if maximum_gain < resolution or maximum_gain > 1.0 / resolution: raise ValueError("maximum_gain is outside the safe range") if gain < resolution or gain > maximum_gain: raise ValueError("minimum_gain is outside the safe range") if tolerance < 0.0 or tolerance > resolution: raise ValueError( "gain_roundoff_tolerance must be between zero and the numeric resolution" ) if residual_limit < resolution or residual_limit > 1.0 / resolution: raise ValueError("maximum_absolute_residual is outside the safe range") @property def content_id(self) -> str: return _content_id("partition-contract", asdict(self)) @dataclass(frozen=True) class ResidualBatch: evidence_id: str evidence_content_id: str contract_content_id: str observations: tuple[ResidualObservation, ...] def __post_init__(self) -> None: _identifier("evidence_id", self.evidence_id) _identifier("evidence_content_id", self.evidence_content_id) _identifier("contract_content_id", self.contract_content_id) if not isinstance(self.observations, tuple): raise TypeError("observations must be an immutable tuple") if not self.observations: raise ValueError("observations must not be empty") if len(self.observations) > MAX_OBSERVATIONS: raise ValueError("observations exceed the global safety limit") if any(type(item) is not ResidualObservation for item in self.observations): raise TypeError("observations must contain concrete ResidualObservation values") @classmethod def capture( cls, *, evidence_id: str, contract: PartitionContract, observations: Iterable[ResidualObservation], ) -> "ResidualBatch": if type(contract) is not PartitionContract: raise TypeError("contract must be a concrete PartitionContract") frozen = tuple(observations) if not frozen: raise ValueError("observations must not be empty") if len(frozen) > contract.maximum_observations: raise ValueError("observations exceed contract maximum") if any(type(item) is not ResidualObservation for item in frozen): raise TypeError("observations must contain concrete ResidualObservation values") return cls( evidence_id=_identifier("evidence_id", evidence_id), evidence_content_id=_batch_content_id( contract.content_id, _identifier("evidence_id", evidence_id), frozen, ), contract_content_id=contract.content_id, observations=frozen, ) @dataclass(frozen=True) class PartitionReport: contract_content_id: str evidence_id: str evidence_content_id: str left_count: int right_count: int parent_sse: float child_sse: float gain: float decision: str def _batch_content_id( contract_content_id: str, evidence_id: str, observations: tuple[ResidualObservation, ...], ) -> str: return _content_id( "residual-batch", { "contract_content_id": contract_content_id, "evidence_id": evidence_id, "observations": tuple(asdict(item) for item in observations), }, ) def _weighted_sse(observations: tuple[ResidualObservation, ...]) -> float: total_weight = math.fsum(item.sample_weight for item in observations) weighted_sum = math.fsum( item.sample_weight * item.residual for item in observations ) mean = weighted_sum / total_weight value = math.fsum( item.sample_weight * (item.residual - mean) ** 2 for item in observations ) if not math.isfinite(value): raise OverflowError("weighted SSE overflowed") return value def audit_partition( contract: PartitionContract, batch: ResidualBatch ) -> PartitionReport: if type(contract) is not PartitionContract: raise TypeError("contract must be a concrete PartitionContract") if type(batch) is not ResidualBatch: raise TypeError("batch must be a concrete ResidualBatch") contract.__post_init__() batch.__post_init__() if batch.contract_content_id != contract.content_id: raise ValueError("batch was captured for a different contract") expected_evidence_content_id = _batch_content_id( batch.contract_content_id, batch.evidence_id, batch.observations, ) if batch.evidence_content_id != expected_evidence_content_id: raise ValueError("batch evidence content identity is invalid") if len(batch.observations) > contract.maximum_observations: raise ValueError("observations exceed contract maximum") observation_ids = tuple(item.observation_id for item in batch.observations) entity_ids = tuple(item.entity_id for item in batch.observations) if len(observation_ids) != len(set(observation_ids)): raise ValueError("duplicate observation_id") if len(entity_ids) != len(set(entity_ids)): raise ValueError("duplicate entity_id") expected_scope = ( contract.dataset_version, contract.feature_id, contract.feature_version, contract.target_id, contract.label_version, contract.base_model_version, contract.cohort_id, contract.window_id, ) for item in batch.observations: # Revalidate exact instances at the trust boundary. Frozen dataclasses # prevent ordinary mutation, but callers can still bypass constructors. item.__post_init__() observed_scope = ( item.dataset_version, item.feature_id, item.feature_version, item.target_id, item.label_version, item.base_model_version, item.cohort_id, item.window_id, ) if observed_scope != expected_scope: raise ValueError("observation scope does not match contract") residual = item.residual if residual != 0.0 and abs(residual) < contract.minimum_numeric_resolution: raise ValueError("residual is below the contracted numeric resolution") if abs(residual) > contract.maximum_absolute_residual: raise ValueError("residual exceeds the contracted magnitude limit") if not ( contract.minimum_numeric_resolution <= item.sample_weight <= contract.maximum_sample_weight ): raise ValueError("sample_weight is outside the contracted numeric range") left = tuple( item for item in batch.observations if item.feature_value <= contract.candidate_threshold ) right = tuple( item for item in batch.observations if item.feature_value > contract.candidate_threshold ) if len(left) < contract.minimum_child_count or len(right) < contract.minimum_child_count: raise ValueError("candidate split violates minimum_child_count") left_weight = math.fsum(item.sample_weight for item in left) right_weight = math.fsum(item.sample_weight for item in right) if min(left_weight, right_weight) < contract.minimum_child_weight: raise ValueError("candidate split violates minimum_child_weight") parent_sse = _weighted_sse(batch.observations) child_sse = math.fsum((_weighted_sse(left), _weighted_sse(right))) gain = parent_sse - child_sse if not math.isfinite(gain): raise OverflowError("partition gain overflowed") if abs(gain) > contract.maximum_gain: raise ValueError("partition gain exceeds the contracted numeric range") if gain < 0.0 and math.isclose( gain, 0.0, rel_tol=0.0, abs_tol=contract.gain_roundoff_tolerance, ): gain = 0.0 decision = "ACCEPT_SPLIT" if gain >= contract.minimum_gain else "REJECT_SPLIT" return PartitionReport( contract_content_id=contract.content_id, evidence_id=batch.evidence_id, evidence_content_id=batch.evidence_content_id, left_count=len(left), right_count=len(right), parent_sse=parent_sse, child_sse=child_sse, gain=gain, decision=decision, ) def _example() -> None: contract = PartitionContract( contract_version="residual-partition-v1", dataset_version="churn-training-2026-07", feature_id="days-since-last-session", feature_version="session-features-v4", target_id="retained-at-30-days", label_version="retention-label-v2", cohort_id="paid-self-serve", window_id="train-2026-07", base_model_version="churn-base-v7", residual_definition="label-minus-base-prediction", loss_metric="weighted-residual-sse", loss_cost_version="retention-squared-loss-v1", gain_convention="parent-sse-minus-child-sse", threshold_convention="left-less-than-or-equal", candidate_threshold=7.0, minimum_child_count=2, minimum_child_weight=2.0, minimum_gain=1.0, gain_roundoff_tolerance=1e-12, numeric_precision_convention="binary64-bounded-inputs-and-gain-v1", minimum_numeric_resolution=1e-12, maximum_sample_weight=1_000_000.0, maximum_gain=1_000_000.0, maximum_absolute_residual=2.0, maximum_observations=100, data_owner="team:growth-data", label_owner="team:retention-measurement", model_owner="team:churn-modeling", decision_owner="team:growth-risk", ) rows = ( ("obs-1", "account-1", 2.0, 0.0, 1.0), ("obs-2", "account-2", 5.0, 0.25, 0.75), ("obs-3", "account-3", 12.0, 0.75, 0.25), ("obs-4", "account-4", 20.0, 1.0, 0.0), ) observations = tuple( ResidualObservation.capture( observation_id=observation_id, entity_id=entity_id, feature_value=feature_value, label_value=label_value, base_prediction=base_prediction, sample_weight=1.0, dataset_version=contract.dataset_version, feature_id=contract.feature_id, feature_version=contract.feature_version, target_id=contract.target_id, label_version=contract.label_version, base_model_version=contract.base_model_version, cohort_id=contract.cohort_id, window_id=contract.window_id, ) for observation_id, entity_id, feature_value, label_value, base_prediction in rows ) batch = ResidualBatch.capture( evidence_id="partition-window-001", contract=contract, observations=observations, ) report = audit_partition(contract, batch) print("example=illustrative_only") print(f"contract_version={contract.contract_version}") print(f"evidence={report.evidence_id}") print(f"evidence_content_id={report.evidence_content_id}") print(f"threshold={contract.candidate_threshold:.1f}") print(f"children={report.left_count}/{report.right_count}") print(f"parent_sse={report.parent_sse:.3f}") print(f"child_sse={report.child_sse:.3f}") print(f"gain={report.gain:.3f}") print(f"decision={report.decision}") if __name__ == "__main__": _example()