"""Minimal fit/eval interface separation with frozen hashed stub predictions. MODEL_CHANGE=NO. No real model fitting. Stub predictor proves that held-out labels cannot alter predictions or feature selection (contract §4 / §5.3). """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from typing import Any, Mapping, Sequence def canonical_json(obj: Any) -> str: return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def sha256_hex(payload: str | bytes) -> str: if isinstance(payload, str): payload = payload.encode("utf-8") return hashlib.sha256(payload).hexdigest() @dataclass class StubFitResult: """Frozen artifact produced by the fitter before any label join.""" selected_features: list[str] predictions: dict[str, float] # election_id -> stub prediction predictions_sha256: str features_sha256: str fit_config_sha256: str meta: dict[str, Any] = field(default_factory=dict) def freeze_blob(self) -> dict[str, Any]: return { "selected_features": list(self.selected_features), "predictions": dict(self.predictions), "predictions_sha256": self.predictions_sha256, "features_sha256": self.features_sha256, "fit_config_sha256": self.fit_config_sha256, "meta": dict(self.meta), } class StubFitter: """Fitting interface: may see training labels only; never held-out labels. Predictions are deterministic hashes of (election_id, features, seed) — intentionally independent of any outcome labels. """ def __init__(self, feature_registry: Sequence[str], seed: int = 20260910): self.feature_registry = list(feature_registry) self.seed = int(seed) def fit_and_predict( self, *, train_election_ids: Sequence[str], train_labels: Mapping[str, float], test_election_ids: Sequence[str], # Deliberately unused: held-out labels must not be passed here. held_out_labels: Mapping[str, float] | None = None, ) -> StubFitResult: if held_out_labels is not None: raise PermissionError( "StubFitter refuses held-out labels (fit/eval separation)" ) # Feature selection uses only training labels (stub: pick first two # features whose name hash interacts with train label sum — still no # test labels). train_sum = sum(float(train_labels[e]) for e in train_election_ids) ranked = sorted( self.feature_registry, key=lambda f: sha256_hex(f"{f}|{train_sum:.6f}|{self.seed}"), ) selected = ranked[: max(1, min(2, len(ranked)))] features_sha = sha256_hex(canonical_json(selected)) predictions: dict[str, float] = {} for eid in test_election_ids: # Stub prediction: map hash to [-20, 20] margin-like number. h = sha256_hex(f"pred|{eid}|{features_sha}|{self.seed}") # Use first 8 hex digits → int → scale raw = int(h[:8], 16) predictions[str(eid)] = round((raw / 0xFFFFFFFF) * 40.0 - 20.0, 6) pred_sha = sha256_hex(canonical_json(predictions)) cfg_sha = sha256_hex( canonical_json( { "seed": self.seed, "feature_registry": list(self.feature_registry), "train_election_ids": list(train_election_ids), "test_election_ids": list(test_election_ids), } ) ) return StubFitResult( selected_features=selected, predictions=predictions, predictions_sha256=pred_sha, features_sha256=features_sha, fit_config_sha256=cfg_sha, meta={ "model_change": False, "fitting_claimed": False, "stub": True, "protocol": "election01-integrity-v0.1.1-draft", }, ) class StubEvaluator: """Evaluation interface: joins frozen predictions with labels AFTER freeze.""" def evaluate( self, frozen: StubFitResult, labels: Mapping[str, float], ) -> dict[str, Any]: # Verify freeze integrity recomputed = sha256_hex(canonical_json(frozen.predictions)) if recomputed != frozen.predictions_sha256: raise ValueError("prediction hash mismatch — freeze broken") errors = {} for eid, pred in frozen.predictions.items(): if eid not in labels: continue errors[eid] = round(float(pred) - float(labels[eid]), 6) mae = ( sum(abs(v) for v in errors.values()) / len(errors) if errors else None ) return { "n_scored": len(errors), "mae": mae, "errors": errors, "predictions_sha256": frozen.predictions_sha256, "features_sha256": frozen.features_sha256, "joined_after_freeze": True, } def prove_label_mutation_invariant( fitter: StubFitter, *, train_election_ids: Sequence[str], train_labels: Mapping[str, float], test_election_ids: Sequence[str], held_out_labels_a: Mapping[str, float], held_out_labels_b: Mapping[str, float], ) -> dict[str, Any]: """§5.3 Changing held-out labels cannot alter predictions or feature selection. Runs fitter twice with identical train data; evaluator joins different held-out label maps AFTER freeze. Predictions/features must match. Also asserts fitter rejects direct held-out injection. """ refuse_ok = False try: fitter.fit_and_predict( train_election_ids=train_election_ids, train_labels=train_labels, test_election_ids=test_election_ids, held_out_labels=held_out_labels_a, ) except PermissionError: refuse_ok = True fit_a = fitter.fit_and_predict( train_election_ids=train_election_ids, train_labels=train_labels, test_election_ids=test_election_ids, ) fit_b = fitter.fit_and_predict( train_election_ids=train_election_ids, train_labels=train_labels, test_election_ids=test_election_ids, ) ev = StubEvaluator() eval_a = ev.evaluate(fit_a, held_out_labels_a) eval_b = ev.evaluate(fit_b, held_out_labels_b) same_preds = fit_a.predictions == fit_b.predictions same_feat = fit_a.selected_features == fit_b.selected_features same_pred_hash = fit_a.predictions_sha256 == fit_b.predictions_sha256 same_feat_hash = fit_a.features_sha256 == fit_b.features_sha256 # Errors SHOULD differ when labels differ (proves labels only affect eval) errors_differ = eval_a["errors"] != eval_b["errors"] return { "ok": bool( refuse_ok and same_preds and same_feat and same_pred_hash and same_feat_hash and errors_differ ), "fitter_refuses_held_out": refuse_ok, "predictions_identical": same_preds, "features_identical": same_feat, "predictions_sha256": fit_a.predictions_sha256, "features_sha256": fit_a.features_sha256, "eval_errors_differ_when_labels_differ": errors_differ, "frozen_a": fit_a.freeze_blob(), "eval_a_mae": eval_a["mae"], "eval_b_mae": eval_b["mae"], }