"""Compare bound windows for distribution drift and feedback-loop risk. The executable fixture is illustrative. A monitor signal is evidence to investigate, not proof of a causal mechanism or permission to retrain. """ from __future__ import annotations from dataclasses import asdict, dataclass from datetime import date, timedelta import hashlib import json import math import re from typing import Any, Iterable MAX_OBSERVATIONS_PER_WINDOW = 20_000 MAX_BUCKETS = 100 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 _iso_date(name: str, value: object) -> date: if not isinstance(value, str): raise ValueError(f"{name} must be an ISO calendar date") try: parsed = date.fromisoformat(value) except ValueError as error: raise ValueError(f"{name} must be an ISO calendar date") from error if parsed.isoformat() != value: raise ValueError(f"{name} must be an ISO calendar date") return parsed def _optional_label(value: object) -> int | None: if value is None: return None if type(value) is not int or value not in (0, 1): raise ValueError("label must be None or 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 FeedbackObservation: observation_id: str entity_id: str feature_bucket: str probability: float decision: str label: int | None event_definition: str horizon: str cohort_id: str window_id: str model_version: str policy_version: str feature_version: str data_version: str label_version: str feedback_source_id: str decision_at: str label_observed_at: str | None def __post_init__(self) -> None: for field in ( "observation_id", "entity_id", "feature_bucket", "event_definition", "horizon", "cohort_id", "window_id", "model_version", "policy_version", "feature_version", "data_version", "label_version", "feedback_source_id", ): _identifier(field, getattr(self, field)) _probability("probability", self.probability) if self.decision not in ("selected", "not-selected"): raise ValueError("decision must be selected or not-selected") _optional_label(self.label) decision_at = _iso_date("decision_at", self.decision_at) if self.label is None: if self.label_observed_at is not None: raise ValueError("unlabeled observation cannot have label_observed_at") else: observed_at = _iso_date("label_observed_at", self.label_observed_at) if observed_at < decision_at: raise ValueError("label_observed_at cannot precede decision_at") @dataclass(frozen=True) class DriftFeedbackContract: contract_version: str monitor_id: str event_definition: str horizon: str cohort_id: str baseline_window: str current_window: str baseline_window_start: str baseline_window_end: str current_window_start: str current_window_end: str label_as_of: str horizon_days: int baseline_model_version: str current_model_version: str baseline_policy_version: str current_policy_version: str feature_version: str baseline_data_version: str current_data_version: str label_version: str feedback_source_id: str response_cost_version: str comparison_convention: str label_finality_convention: str bucket_ids: tuple[str, ...] distribution_metric: str performance_metric: str feedback_metric: str aggregation_convention: str numeric_precision_convention: str minimum_numeric_resolution: float maximum_total_variation: float maximum_prediction_mean_shift: float maximum_selection_rate_shift: float maximum_label_coverage_gap: float minimum_overall_label_coverage: float maximum_brier_degradation: float minimum_labeled_per_decision_group: int minimum_label_coverage_per_decision_group: float minimum_observations_per_window: int maximum_observations_per_window: int feature_owner: str model_owner: str policy_owner: str label_owner: str monitor_owner: str response_owner: str def __post_init__(self) -> None: for field in ( "contract_version", "monitor_id", "event_definition", "horizon", "cohort_id", "baseline_window", "current_window", "baseline_model_version", "current_model_version", "baseline_policy_version", "current_policy_version", "feature_version", "baseline_data_version", "current_data_version", "label_version", "feedback_source_id", "response_cost_version", ): _identifier(field, getattr(self, field)) if self.baseline_window == self.current_window: raise ValueError("baseline_window and current_window must differ") baseline_start = _iso_date( "baseline_window_start", self.baseline_window_start ) baseline_end = _iso_date("baseline_window_end", self.baseline_window_end) current_start = _iso_date("current_window_start", self.current_window_start) current_end = _iso_date("current_window_end", self.current_window_end) label_as_of = _iso_date("label_as_of", self.label_as_of) if baseline_start > baseline_end or current_start > current_end: raise ValueError("monitor window start must not exceed its end") if baseline_end >= current_start: raise ValueError("monitor windows must be ordered and non-overlapping") _positive_int("horizon_days", self.horizon_days, 3_650) if label_as_of < current_end + timedelta(days=self.horizon_days): raise ValueError("label_as_of does not provide full current-window maturity") if self.comparison_convention != "same-model-and-policy-only": raise ValueError("unsupported comparison_convention") if self.label_finality_convention != "all-labels-final-after-full-horizon": raise ValueError("unsupported label_finality_convention") if self.baseline_model_version != self.current_model_version: raise ValueError("same-model comparison requires equal model versions") if self.baseline_policy_version != self.current_policy_version: raise ValueError("same-policy comparison requires equal policy versions") for field in ( "feature_owner", "model_owner", "policy_owner", "label_owner", "monitor_owner", "response_owner", ): _owner(field, getattr(self, field)) if not isinstance(self.bucket_ids, tuple) or not self.bucket_ids: raise TypeError("bucket_ids must be a non-empty immutable tuple") if len(self.bucket_ids) > MAX_BUCKETS: raise ValueError("bucket_ids exceed the global safety limit") for bucket_id in self.bucket_ids: _identifier("bucket_id", bucket_id) if len(self.bucket_ids) != len(set(self.bucket_ids)): raise ValueError("bucket_ids must be unique") if self.distribution_metric != "total-variation-on-bound-buckets": raise ValueError("unsupported distribution_metric") if self.performance_metric != "mean-brier-on-observed-labels": raise ValueError("unsupported performance_metric") if self.feedback_metric != "decision-conditional-label-coverage-gap": raise ValueError("unsupported feedback_metric") if self.aggregation_convention != "equal-observation-weight": raise ValueError("unsupported aggregation_convention") if ( self.numeric_precision_convention != "binary64-resolved-probability-and-metric-threshold-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") for field in ( "maximum_total_variation", "maximum_prediction_mean_shift", "maximum_selection_rate_shift", "maximum_label_coverage_gap", "minimum_overall_label_coverage", "maximum_brier_degradation", ): threshold = _probability(field, getattr(self, field)) if field != "minimum_overall_label_coverage" and threshold < resolution: raise ValueError(f"{field} is below the numeric resolution") _positive_int( "minimum_labeled_per_decision_group", self.minimum_labeled_per_decision_group, MAX_OBSERVATIONS_PER_WINDOW, ) _probability( "minimum_label_coverage_per_decision_group", self.minimum_label_coverage_per_decision_group, ) _positive_int( "minimum_observations_per_window", self.minimum_observations_per_window, MAX_OBSERVATIONS_PER_WINDOW, ) _positive_int( "maximum_observations_per_window", self.maximum_observations_per_window, MAX_OBSERVATIONS_PER_WINDOW, ) if self.minimum_observations_per_window > self.maximum_observations_per_window: raise ValueError( "minimum_observations_per_window exceeds maximum_observations_per_window" ) @property def content_id(self) -> str: return _content_id("drift-feedback-contract", asdict(self)) @dataclass(frozen=True) class DriftFeedbackReport: report_id: str contract_content_id: str evidence_id: str evidence_content_id: str baseline_count: int current_count: int total_variation: float prediction_mean_shift: float selection_rate_shift: float overall_label_coverage: float decision_label_coverage_gap: float baseline_brier: float current_brier: float brier_degradation: float decision: str def __post_init__(self) -> None: for field in ( "report_id", "contract_content_id", "evidence_id", "evidence_content_id", ): _identifier(field, getattr(self, field)) _positive_int("baseline_count", self.baseline_count, MAX_OBSERVATIONS_PER_WINDOW) _positive_int("current_count", self.current_count, MAX_OBSERVATIONS_PER_WINDOW) for field in ( "total_variation", "prediction_mean_shift", "selection_rate_shift", "overall_label_coverage", "decision_label_coverage_gap", "baseline_brier", "current_brier", "brier_degradation", ): value = _finite(field, getattr(self, field)) if field != "brier_degradation" and not 0.0 <= value <= 1.0: raise ValueError(f"{field} must be between zero and one") if self.decision not in ( "MONITOR", "HOLD_DRIFT", "HOLD_FEEDBACK", "HOLD_DRIFT_AND_FEEDBACK", ): raise ValueError("invalid monitor decision") def _validate_window( contract: DriftFeedbackContract, rows: tuple[FeedbackObservation, ...], *, window_id: str, model_version: str, policy_version: str, data_version: str, window_start: str, window_end: str, ) -> None: if len(rows) < contract.minimum_observations_per_window: raise ValueError(f"{window_id} is below minimum_observations_per_window") if len(rows) > contract.maximum_observations_per_window: raise ValueError(f"{window_id} exceeds maximum_observations_per_window") expected_scope = ( contract.event_definition, contract.horizon, contract.cohort_id, window_id, model_version, policy_version, contract.feature_version, data_version, contract.label_version, contract.feedback_source_id, ) decisions: set[str] = set() start = _iso_date("window_start", window_start) end = _iso_date("window_end", window_end) label_as_of = _iso_date("label_as_of", contract.label_as_of) for item in rows: item.__post_init__() _resolved_probability( "probability", item.probability, contract.minimum_numeric_resolution ) observed_scope = ( item.event_definition, item.horizon, item.cohort_id, item.window_id, item.model_version, item.policy_version, item.feature_version, item.data_version, item.label_version, item.feedback_source_id, ) if observed_scope != expected_scope: raise ValueError("feedback observation scope does not match contract") if item.feature_bucket not in contract.bucket_ids: raise ValueError("feature_bucket is outside the contracted vocabulary") decision_at = _iso_date("decision_at", item.decision_at) if not start <= decision_at <= end: raise ValueError("decision_at is outside the contracted monitor window") if item.label_observed_at is not None: observed_at = _iso_date("label_observed_at", item.label_observed_at) if observed_at < decision_at + timedelta(days=contract.horizon_days): raise ValueError( "label was observed before its full outcome horizon matured" ) if observed_at > label_as_of: raise ValueError("label was observed after the contracted label_as_of") decisions.add(item.decision) if decisions != {"selected", "not-selected"}: raise ValueError("each window must contain both decision groups") for decision in ("selected", "not-selected"): group = tuple(item for item in rows if item.decision == decision) labeled_count = sum(item.label is not None for item in group) coverage = labeled_count / len(group) if labeled_count < contract.minimum_labeled_per_decision_group: raise ValueError( f"{window_id}/{decision} is below minimum labeled support" ) if coverage < contract.minimum_label_coverage_per_decision_group: raise ValueError( f"{window_id}/{decision} is below minimum label coverage" ) def _brier(rows: tuple[FeedbackObservation, ...]) -> float: labeled = tuple(item for item in rows if item.label is not None) if not labeled: raise ValueError("window has no observed labels") value = math.fsum((item.probability - item.label) ** 2 for item in labeled) / len(labeled) if not math.isfinite(value): raise OverflowError("Brier score overflowed") return value def monitor_drift_and_feedback( contract: DriftFeedbackContract, *, evidence_id: str, baseline: Iterable[FeedbackObservation], current: Iterable[FeedbackObservation], ) -> DriftFeedbackReport: if type(contract) is not DriftFeedbackContract: raise TypeError("contract must be a concrete DriftFeedbackContract") contract.__post_init__() checked_evidence_id = _identifier("evidence_id", evidence_id) baseline_rows = tuple(baseline) current_rows = tuple(current) if any(type(item) is not FeedbackObservation for item in baseline_rows + current_rows): raise TypeError("windows must contain concrete FeedbackObservation values") _validate_window( contract, baseline_rows, window_id=contract.baseline_window, model_version=contract.baseline_model_version, policy_version=contract.baseline_policy_version, data_version=contract.baseline_data_version, window_start=contract.baseline_window_start, window_end=contract.baseline_window_end, ) _validate_window( contract, current_rows, window_id=contract.current_window, model_version=contract.current_model_version, policy_version=contract.current_policy_version, data_version=contract.current_data_version, window_start=contract.current_window_start, window_end=contract.current_window_end, ) all_rows = baseline_rows + current_rows observation_ids = tuple(item.observation_id for item in all_rows) entity_window_ids = tuple((item.window_id, item.entity_id) for item in all_rows) if len(observation_ids) != len(set(observation_ids)): raise ValueError("duplicate observation_id across monitor windows") if len(entity_window_ids) != len(set(entity_window_ids)): raise ValueError("duplicate entity_id within a monitor window") def bucket_share(rows: tuple[FeedbackObservation, ...], bucket_id: str) -> float: return sum(item.feature_bucket == bucket_id for item in rows) / len(rows) total_variation = 0.5 * math.fsum( abs(bucket_share(current_rows, bucket) - bucket_share(baseline_rows, bucket)) for bucket in contract.bucket_ids ) baseline_prediction_mean = math.fsum(item.probability for item in baseline_rows) / len(baseline_rows) current_prediction_mean = math.fsum(item.probability for item in current_rows) / len(current_rows) prediction_mean_shift = abs(current_prediction_mean - baseline_prediction_mean) baseline_selection_rate = sum(item.decision == "selected" for item in baseline_rows) / len(baseline_rows) current_selection_rate = sum(item.decision == "selected" for item in current_rows) / len(current_rows) selection_rate_shift = abs(current_selection_rate - baseline_selection_rate) labeled_current = tuple(item for item in current_rows if item.label is not None) overall_label_coverage = len(labeled_current) / len(current_rows) selected = tuple(item for item in current_rows if item.decision == "selected") not_selected = tuple(item for item in current_rows if item.decision == "not-selected") selected_coverage = sum(item.label is not None for item in selected) / len(selected) not_selected_coverage = sum(item.label is not None for item in not_selected) / len(not_selected) decision_coverage_gap = abs(selected_coverage - not_selected_coverage) baseline_brier = _brier(baseline_rows) current_brier = _brier(current_rows) brier_degradation = current_brier - baseline_brier values = ( total_variation, prediction_mean_shift, selection_rate_shift, overall_label_coverage, decision_coverage_gap, baseline_brier, current_brier, brier_degradation, ) if not all(math.isfinite(value) for value in values): raise OverflowError("monitor metric overflowed") drift = ( total_variation > contract.maximum_total_variation or prediction_mean_shift > contract.maximum_prediction_mean_shift or selection_rate_shift > contract.maximum_selection_rate_shift or brier_degradation > contract.maximum_brier_degradation ) feedback = ( overall_label_coverage < contract.minimum_overall_label_coverage or decision_coverage_gap > contract.maximum_label_coverage_gap ) if drift and feedback: decision = "HOLD_DRIFT_AND_FEEDBACK" elif drift: decision = "HOLD_DRIFT" elif feedback: decision = "HOLD_FEEDBACK" else: decision = "MONITOR" evidence_content_id = _content_id( "drift-feedback-evidence", { "contract_content_id": contract.content_id, "baseline": tuple(asdict(item) for item in baseline_rows), "current": tuple(asdict(item) for item in current_rows), }, ) report_payload = { "contract_content_id": contract.content_id, "evidence_id": checked_evidence_id, "evidence_content_id": evidence_content_id, "baseline_count": len(baseline_rows), "current_count": len(current_rows), "total_variation": total_variation, "prediction_mean_shift": prediction_mean_shift, "selection_rate_shift": selection_rate_shift, "overall_label_coverage": overall_label_coverage, "decision_label_coverage_gap": decision_coverage_gap, "baseline_brier": baseline_brier, "current_brier": current_brier, "brier_degradation": brier_degradation, "decision": decision, } return DriftFeedbackReport( report_id=_content_id("drift-feedback-report", report_payload), **report_payload, ) def _example() -> None: contract = DriftFeedbackContract( contract_version="drift-feedback-v1", monitor_id="renewal-risk-monitor", event_definition="renewed-within-30-days", horizon="30-days-after-offer", cohort_id="paid-self-serve", baseline_window="2026-06", current_window="2026-07", baseline_window_start="2026-06-01", baseline_window_end="2026-06-30", current_window_start="2026-07-01", current_window_end="2026-07-31", label_as_of="2026-08-31", horizon_days=30, baseline_model_version="renewal-model-v4", current_model_version="renewal-model-v4", baseline_policy_version="save-offer-v2", current_policy_version="save-offer-v2", feature_version="renewal-features-v5", baseline_data_version="renewal-events-2026-06", current_data_version="renewal-events-2026-07", label_version="renewal-outcome-v3", feedback_source_id="billing-ledger-v2", response_cost_version="monitor-response-costs-v1", comparison_convention="same-model-and-policy-only", label_finality_convention="all-labels-final-after-full-horizon", bucket_ids=("low-recency", "high-recency"), distribution_metric="total-variation-on-bound-buckets", performance_metric="mean-brier-on-observed-labels", feedback_metric="decision-conditional-label-coverage-gap", aggregation_convention="equal-observation-weight", numeric_precision_convention="binary64-resolved-probability-and-metric-threshold-v1", minimum_numeric_resolution=1e-12, maximum_total_variation=0.20, maximum_prediction_mean_shift=0.08, maximum_selection_rate_shift=0.20, maximum_label_coverage_gap=0.40, minimum_overall_label_coverage=0.80, maximum_brier_degradation=0.05, minimum_labeled_per_decision_group=1, minimum_label_coverage_per_decision_group=0.25, minimum_observations_per_window=4, maximum_observations_per_window=100, feature_owner="team:growth-data", model_owner="team:renewal-modeling", policy_owner="team:retention-product", label_owner="team:billing-measurement", monitor_owner="team:model-reliability", response_owner="team:retention-risk", ) def observation( index: int, *, window: str, bucket: str, probability: float, decision: str, label: int | None, ) -> FeedbackObservation: baseline = window == contract.baseline_window return FeedbackObservation( observation_id=f"{window}-obs-{index}", entity_id=f"account-{index}", feature_bucket=bucket, probability=probability, decision=decision, label=label, event_definition=contract.event_definition, horizon=contract.horizon, cohort_id=contract.cohort_id, window_id=window, model_version=(contract.baseline_model_version if baseline else contract.current_model_version), policy_version=(contract.baseline_policy_version if baseline else contract.current_policy_version), feature_version=contract.feature_version, data_version=(contract.baseline_data_version if baseline else contract.current_data_version), label_version=contract.label_version, feedback_source_id=contract.feedback_source_id, decision_at=("2026-06-15" if baseline else "2026-07-15"), label_observed_at=( None if label is None else ("2026-07-20" if baseline else "2026-08-20") ), ) baseline = ( observation(index=1, window=contract.baseline_window, bucket="low-recency", probability=0.2, decision="not-selected", label=0), observation(index=2, window=contract.baseline_window, bucket="low-recency", probability=0.4, decision="not-selected", label=0), observation(index=3, window=contract.baseline_window, bucket="high-recency", probability=0.6, decision="selected", label=1), observation(index=4, window=contract.baseline_window, bucket="high-recency", probability=0.8, decision="selected", label=1), ) current = ( observation(index=1, window=contract.current_window, bucket="low-recency", probability=0.3, decision="not-selected", label=None), observation(index=2, window=contract.current_window, bucket="low-recency", probability=0.5, decision="not-selected", label=0), observation(index=3, window=contract.current_window, bucket="low-recency", probability=0.7, decision="selected", label=1), observation(index=4, window=contract.current_window, bucket="high-recency", probability=0.9, decision="selected", label=1), ) report = monitor_drift_and_feedback( contract, evidence_id="monitor-batch-001", baseline=baseline, current=current, ) print("example=illustrative_only") print(f"contract_version={contract.contract_version}") print(f"total_variation={report.total_variation:.3f}") print(f"prediction_mean_shift={report.prediction_mean_shift:.3f}") print(f"selection_rate_shift={report.selection_rate_shift:.3f}") print(f"label_coverage={report.overall_label_coverage:.3f}") print(f"decision_coverage_gap={report.decision_label_coverage_gap:.3f}") print(f"brier_degradation={report.brier_degradation:.3f}") print(f"decision={report.decision}") if __name__ == "__main__": _example()