"""Fit a small linear baseline and report debugging evidence. The implementation is intentionally bounded and dependency-free. It is not a general-purpose regression library. It binds rows to an immutable diagnostic contract, fits ordinary or ridge least squares, and compares the result with a train-mean baseline on an identity-separated evaluation split. """ 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_ROWS = 100_000 _MAX_ALLOWED_FEATURES = 16 _MAX_ALLOWED_SLICES = 64 _SOLVER_ID = "normal-equations-partial-pivot-v1" 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_finite(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") def _require_nonnegative(name: str, value: float) -> None: _require_finite(name, value) if value < 0.0: raise ValueError(f"{name} must be non-negative") 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()}" def _checked_product(left: float, right: float, context: str) -> float: product = left * right if not math.isfinite(product): raise FloatingPointError(f"{context} overflowed") return product @dataclass(frozen=True) class LinearDiagnosticContract: contract_version: str task_id: str target_id: str target_unit: str feature_names: tuple[str, ...] dataset_revision: str feature_snapshot_id: str train_split_id: str evaluation_split_id: str required_evaluation_slices: tuple[str, ...] minimum_train_rows: int minimum_evaluation_rows: int minimum_rows_per_slice: int max_rows: int ridge_lambda: float solver_id: str singular_pivot_tolerance: float minimum_evaluation_improvement: float minimum_slice_improvement: float maximum_slice_rmse: float maximum_generalization_gap: float maximum_condition_proxy: float data_owner: str target_owner: str def __post_init__(self) -> None: for name in ( "contract_version", "task_id", "target_id", "target_unit", "dataset_revision", "feature_snapshot_id", "train_split_id", "evaluation_split_id", ): _require_identifier(name, getattr(self, name)) if self.train_split_id == self.evaluation_split_id: raise ValueError("train and evaluation split IDs must differ") if not isinstance(self.feature_names, tuple) or not self.feature_names: raise TypeError("feature_names must be a non-empty immutable tuple") if len(self.feature_names) > _MAX_ALLOWED_FEATURES: raise ValueError(f"feature_names cannot exceed {_MAX_ALLOWED_FEATURES}") for index, feature in enumerate(self.feature_names): _require_identifier(f"feature_names[{index}]", feature) if len(set(self.feature_names)) != len(self.feature_names): raise ValueError("feature_names must be unique") if not isinstance(self.required_evaluation_slices, tuple) or not self.required_evaluation_slices: raise TypeError("required_evaluation_slices must be a non-empty immutable tuple") for index, slice_id in enumerate(self.required_evaluation_slices): _require_identifier(f"required_evaluation_slices[{index}]", slice_id) if len(set(self.required_evaluation_slices)) != len(self.required_evaluation_slices): raise ValueError("required_evaluation_slices must be unique") if len(self.required_evaluation_slices) > _MAX_ALLOWED_SLICES: raise ValueError(f"required_evaluation_slices cannot exceed {_MAX_ALLOWED_SLICES}") _require_int("minimum_train_rows", self.minimum_train_rows, minimum=2) _require_int("minimum_evaluation_rows", self.minimum_evaluation_rows, minimum=1) _require_int("minimum_rows_per_slice", self.minimum_rows_per_slice, minimum=1) _require_int("max_rows", self.max_rows, minimum=3) if self.max_rows > _MAX_ALLOWED_ROWS: raise ValueError(f"max_rows cannot exceed {_MAX_ALLOWED_ROWS}") if self.minimum_train_rows + self.minimum_evaluation_rows > self.max_rows: raise ValueError("minimum split coverage cannot exceed max_rows") required_slice_rows = self.minimum_rows_per_slice * len(self.required_evaluation_slices) if required_slice_rows > self.minimum_evaluation_rows: raise ValueError("minimum_evaluation_rows must cover every required slice") _require_nonnegative("ridge_lambda", self.ridge_lambda) if self.solver_id != _SOLVER_ID: raise ValueError(f"solver_id must be {_SOLVER_ID}") _require_finite("singular_pivot_tolerance", self.singular_pivot_tolerance) if self.singular_pivot_tolerance <= 0.0: raise ValueError("singular_pivot_tolerance must be positive") _require_finite("minimum_evaluation_improvement", self.minimum_evaluation_improvement) if not -1.0 <= self.minimum_evaluation_improvement <= 1.0: raise ValueError("minimum_evaluation_improvement must be between -1 and 1") _require_finite("minimum_slice_improvement", self.minimum_slice_improvement) if not -1.0 <= self.minimum_slice_improvement <= 1.0: raise ValueError("minimum_slice_improvement must be between -1 and 1") _require_nonnegative("maximum_slice_rmse", self.maximum_slice_rmse) _require_nonnegative("maximum_generalization_gap", self.maximum_generalization_gap) _require_finite("maximum_condition_proxy", self.maximum_condition_proxy) if self.maximum_condition_proxy < 1.0: raise ValueError("maximum_condition_proxy must be at least 1") _require_owner("data_owner", self.data_owner) _require_owner("target_owner", self.target_owner) @property def contract_id(self) -> str: return _content_id(self.contract_version, asdict(self)) @dataclass(frozen=True) class RegressionRow: row_id: str entity_id: str split_id: str dataset_revision: str feature_snapshot_id: str feature_values: tuple[float, ...] target: float slice_id: str @dataclass(frozen=True) class DiagnosticDataset: contract_id: str dataset_id: str target_id: str target_unit: str dataset_revision: str feature_snapshot_id: str rows: tuple[RegressionRow, ...] @property def evidence_id(self) -> str: return _content_id("linear-diagnostic-evidence-v1", asdict(self)) @dataclass(frozen=True) class LinearDiagnostic: contract_id: str dataset_id: str evidence_id: str coefficients: tuple[float, ...] train_rmse: float evaluation_rmse: float mean_baseline_rmse: float evaluation_improvement: float generalization_gap: float condition_proxy: float lowest_improvement_slice_id: str lowest_slice_improvement: float highest_rmse_slice_id: str highest_slice_rmse: float decision: str def _validate_row( contract: LinearDiagnosticContract, row: RegressionRow, index: int ) -> None: if type(row) is not RegressionRow: raise TypeError(f"rows[{index}] must be a concrete RegressionRow") _require_identifier(f"rows[{index}].row_id", row.row_id) _require_identifier(f"rows[{index}].entity_id", row.entity_id) _require_identifier(f"rows[{index}].slice_id", row.slice_id) if row.split_id not in (contract.train_split_id, contract.evaluation_split_id): raise ValueError(f"rows[{index}].split_id is not declared by the contract") if row.dataset_revision != contract.dataset_revision: raise ValueError(f"rows[{index}].dataset_revision does not match contract") if row.feature_snapshot_id != contract.feature_snapshot_id: raise ValueError(f"rows[{index}].feature_snapshot_id does not match contract") if not isinstance(row.feature_values, tuple): raise TypeError(f"rows[{index}].feature_values must be an immutable tuple") if len(row.feature_values) != len(contract.feature_names): raise ValueError(f"rows[{index}].feature_values has the wrong shape") for feature_index, value in enumerate(row.feature_values): _require_finite(f"rows[{index}].feature_values[{feature_index}]", value) _require_finite(f"rows[{index}].target", row.target) def _solve_normal_equations( rows: tuple[RegressionRow, ...], feature_count: int, ridge_lambda: float, singular_pivot_tolerance: float, ) -> tuple[tuple[float, ...], float]: width = feature_count + 1 design = [(1.0, *tuple(float(value) for value in row.feature_values)) for row in rows] targets = [float(row.target) for row in rows] matrix: list[list[float]] = [] vector: list[float] = [] for left in range(width): matrix_row: list[float] = [] for right in range(width): value = math.fsum( _checked_product(sample[left], sample[right], "normal equation") for sample in design ) if left == right and left > 0: value += ridge_lambda if not math.isfinite(value): raise FloatingPointError("normal equation accumulation overflowed") matrix_row.append(value) matrix.append(matrix_row) value = math.fsum( _checked_product(sample[left], target, "target projection") for sample, target in zip(design, targets) ) if not math.isfinite(value): raise FloatingPointError("target projection accumulation overflowed") vector.append(value) pivots: list[float] = [] for column in range(width): pivot_row = max(range(column, width), key=lambda row_index: abs(matrix[row_index][column])) pivot = matrix[pivot_row][column] if abs(pivot) <= singular_pivot_tolerance: raise ValueError("linear system is singular; add ridge or remove redundant features") if pivot_row != column: matrix[column], matrix[pivot_row] = matrix[pivot_row], matrix[column] vector[column], vector[pivot_row] = vector[pivot_row], vector[column] pivot = matrix[column][column] pivots.append(abs(pivot)) for row_index in range(column + 1, width): factor = matrix[row_index][column] / pivot for right in range(column, width): matrix[row_index][right] -= factor * matrix[column][right] vector[row_index] -= factor * vector[column] coefficients = [0.0] * width for row_index in range(width - 1, -1, -1): remainder = math.fsum( matrix[row_index][column] * coefficients[column] for column in range(row_index + 1, width) ) coefficients[row_index] = (vector[row_index] - remainder) / matrix[row_index][row_index] if not math.isfinite(coefficients[row_index]): raise FloatingPointError("coefficient solve produced a non-finite value") condition_proxy = max(pivots) / min(pivots) return tuple(coefficients), condition_proxy def _prediction(coefficients: tuple[float, ...], row: RegressionRow) -> float: value = coefficients[0] + math.fsum( _checked_product(weight, float(feature), "prediction") for weight, feature in zip(coefficients[1:], row.feature_values) ) if not math.isfinite(value): raise FloatingPointError("prediction produced a non-finite value") return value def _rmse(coefficients: tuple[float, ...], rows: tuple[RegressionRow, ...]) -> float: squared = [] for row in rows: residual = _prediction(coefficients, row) - float(row.target) squared.append(_checked_product(residual, residual, "squared residual")) value = math.sqrt(math.fsum(squared) / len(rows)) if not math.isfinite(value): raise FloatingPointError("RMSE produced a non-finite value") return value def diagnose_linear_baseline( contract: LinearDiagnosticContract, dataset: DiagnosticDataset ) -> LinearDiagnostic: """Validate identity, fit on train, and diagnose on isolated evaluation rows.""" if type(contract) is not LinearDiagnosticContract: raise TypeError("contract must be a concrete LinearDiagnosticContract") LinearDiagnosticContract.__post_init__(contract) if type(dataset) is not DiagnosticDataset: raise TypeError("dataset must be a concrete DiagnosticDataset") if dataset.contract_id != contract.contract_id: raise ValueError("dataset contract_id does not match the full contract identity") _require_identifier("dataset_id", dataset.dataset_id) if dataset.target_id != contract.target_id: raise ValueError("dataset target_id does not match contract") if dataset.target_unit != contract.target_unit: raise ValueError("dataset target_unit does not match contract") if dataset.dataset_revision != contract.dataset_revision: raise ValueError("dataset_revision does not match contract") if dataset.feature_snapshot_id != contract.feature_snapshot_id: raise ValueError("feature_snapshot_id does not match contract") if not isinstance(dataset.rows, tuple): raise TypeError("rows must be an immutable tuple") if len(dataset.rows) > contract.max_rows: raise ValueError("row count exceeds the contract's bounded workload") seen_rows: set[str] = set() seen_entities: set[str] = set() train: list[RegressionRow] = [] evaluation: list[RegressionRow] = [] for index, row in enumerate(dataset.rows): _validate_row(contract, row, index) if row.row_id in seen_rows: raise ValueError("row_id values must be unique") if row.entity_id in seen_entities: raise ValueError("entity_id values must be independent across the diagnostic") seen_rows.add(row.row_id) seen_entities.add(row.entity_id) if row.split_id == contract.train_split_id: train.append(row) else: evaluation.append(row) if len(train) < contract.minimum_train_rows: raise ValueError("train split is below minimum_train_rows") if len(evaluation) < contract.minimum_evaluation_rows: raise ValueError("evaluation split is below minimum_evaluation_rows") train_rows = tuple(train) evaluation_rows = tuple(evaluation) slice_rows: dict[str, list[RegressionRow]] = { slice_id: [] for slice_id in contract.required_evaluation_slices } for row in evaluation_rows: if row.slice_id not in slice_rows: raise ValueError("evaluation row contains an undeclared slice") slice_rows[row.slice_id].append(row) for slice_id, rows in slice_rows.items(): if len(rows) < contract.minimum_rows_per_slice: raise ValueError(f"slice {slice_id} is below minimum_rows_per_slice") coefficients, condition_proxy = _solve_normal_equations( train_rows, len(contract.feature_names), contract.ridge_lambda, contract.singular_pivot_tolerance, ) train_rmse = _rmse(coefficients, train_rows) evaluation_rmse = _rmse(coefficients, evaluation_rows) train_mean = math.fsum(float(row.target) for row in train_rows) / len(train_rows) mean_coefficients = (train_mean, *([0.0] * len(contract.feature_names))) mean_baseline_rmse = _rmse(mean_coefficients, evaluation_rows) if mean_baseline_rmse == 0.0: evaluation_improvement = 0.0 if evaluation_rmse == 0.0 else -1.0 else: evaluation_improvement = 1.0 - evaluation_rmse / mean_baseline_rmse generalization_gap = max(0.0, evaluation_rmse - train_rmse) slice_diagnostics = [] for slice_id, rows in slice_rows.items(): immutable_rows = tuple(rows) slice_rmse = _rmse(coefficients, immutable_rows) slice_mean_rmse = _rmse(mean_coefficients, immutable_rows) if slice_mean_rmse == 0.0: slice_improvement = 0.0 if slice_rmse == 0.0 else -1.0 else: slice_improvement = 1.0 - slice_rmse / slice_mean_rmse slice_diagnostics.append((slice_id, slice_rmse, slice_improvement)) lowest_improvement_slice_id, _, lowest_slice_improvement = min( slice_diagnostics, key=lambda item: (item[2], -item[1], item[0]) ) highest_rmse_slice_id, highest_slice_rmse, _ = max( slice_diagnostics, key=lambda item: (item[1], item[0]) ) decision = ( "BASELINE_ESTABLISHED" if evaluation_improvement >= contract.minimum_evaluation_improvement and generalization_gap <= contract.maximum_generalization_gap and condition_proxy <= contract.maximum_condition_proxy and all( improvement >= contract.minimum_slice_improvement and rmse <= contract.maximum_slice_rmse for _, rmse, improvement in slice_diagnostics ) else "INVESTIGATE_DATA_OR_TARGET" ) return LinearDiagnostic( contract_id=contract.contract_id, dataset_id=dataset.dataset_id, evidence_id=dataset.evidence_id, coefficients=coefficients, train_rmse=train_rmse, evaluation_rmse=evaluation_rmse, mean_baseline_rmse=mean_baseline_rmse, evaluation_improvement=evaluation_improvement, generalization_gap=generalization_gap, condition_proxy=condition_proxy, lowest_improvement_slice_id=lowest_improvement_slice_id, lowest_slice_improvement=lowest_slice_improvement, highest_rmse_slice_id=highest_rmse_slice_id, highest_slice_rmse=highest_slice_rmse, decision=decision, ) ILLUSTRATIVE_CONTRACT = LinearDiagnosticContract( contract_version="linear-diagnostic-v1", task_id="illustrative-resolution-time", target_id="hours-to-resolution", target_unit="hours", feature_names=("queue-age-hours",), dataset_revision="support-cases-2026-08-20", feature_snapshot_id="support-features-at-triage-v2", train_split_id="train-before-day-100", evaluation_split_id="evaluate-days-100-130", required_evaluation_slices=("established", "new-account"), minimum_train_rows=4, minimum_evaluation_rows=2, minimum_rows_per_slice=1, max_rows=1000, ridge_lambda=0.0, solver_id=_SOLVER_ID, singular_pivot_tolerance=1e-12, minimum_evaluation_improvement=0.90, minimum_slice_improvement=0.90, maximum_slice_rmse=0.10, maximum_generalization_gap=0.10, maximum_condition_proxy=100.0, data_owner="team:support-data", target_owner="team:support-operations", ) ILLUSTRATIVE_DATASET = DiagnosticDataset( contract_id=ILLUSTRATIVE_CONTRACT.contract_id, dataset_id="linear-baseline-run-0042", target_id=ILLUSTRATIVE_CONTRACT.target_id, target_unit=ILLUSTRATIVE_CONTRACT.target_unit, dataset_revision=ILLUSTRATIVE_CONTRACT.dataset_revision, feature_snapshot_id=ILLUSTRATIVE_CONTRACT.feature_snapshot_id, rows=( RegressionRow("row-1", "case-1", "train-before-day-100", "support-cases-2026-08-20", "support-features-at-triage-v2", (0.0,), 1.0, "established"), RegressionRow("row-2", "case-2", "train-before-day-100", "support-cases-2026-08-20", "support-features-at-triage-v2", (1.0,), 3.0, "established"), RegressionRow("row-3", "case-3", "train-before-day-100", "support-cases-2026-08-20", "support-features-at-triage-v2", (2.0,), 5.0, "established"), RegressionRow("row-4", "case-4", "train-before-day-100", "support-cases-2026-08-20", "support-features-at-triage-v2", (3.0,), 7.0, "established"), RegressionRow("row-5", "case-5", "evaluate-days-100-130", "support-cases-2026-08-20", "support-features-at-triage-v2", (4.0,), 9.0, "established"), RegressionRow("row-6", "case-6", "evaluate-days-100-130", "support-cases-2026-08-20", "support-features-at-triage-v2", (5.0,), 11.0, "new-account"), ), ) def format_example() -> str: result = diagnose_linear_baseline(ILLUSTRATIVE_CONTRACT, ILLUSTRATIVE_DATASET) coefficients = ",".join(f"{value:.3f}" for value in result.coefficients) return "\n".join( ( "example=illustrative_only", f"contract_version={ILLUSTRATIVE_CONTRACT.contract_version}", f"dataset={result.dataset_id}", f"evidence_id={result.evidence_id}", f"coefficients=intercept,{ILLUSTRATIVE_CONTRACT.feature_names[0]}:{coefficients}", f"train_rmse={result.train_rmse:.3f}", f"evaluation_rmse={result.evaluation_rmse:.3f}", f"mean_baseline_rmse={result.mean_baseline_rmse:.3f}", f"evaluation_improvement={result.evaluation_improvement:.3f}", f"lowest_improvement_slice={result.lowest_improvement_slice_id}:{result.lowest_slice_improvement:.3f}", f"highest_rmse_slice={result.highest_rmse_slice_id}:{result.highest_slice_rmse:.3f}", f"decision={result.decision}", ) ) if __name__ == "__main__": print(format_example())