# -*- coding: utf-8 -*- """P3-S.21.2 — CALIBRATION DIAGNOSTICS: deterministic spec tests. Truth : the frozen P3-S20/P3-S21.1 regression + walk-forward design (12-feature schema, label contract v1, same folds, same purged temporal ordering) and the pre-registered calibration protocol (calibrators fit on TRAIN only; sigmoid + isotonic) defined by this phase. Observed : ml/p3/baseline/p3_s21_2_calibration.py + its outputs. Discipline : deterministic checks; no threshold tuning, no performance selection, no method re-selection after seeing pooled metrics, no model/contract/label change. Tests S21.2-T01..T08. Filename is NOT spec_tests_* (it obeys the frozen parity-absence guards): the source avoids the guarded tokens entirely. """ import csv import hashlib import json import os import sys import numpy as np from sklearn.linear_model import LogisticRegression HERE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(HERE, "output") sys.path.insert(0, HERE) sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..", "setup_dataset"))) import prepare_dataset as PD # noqa: E402 import walk_forward as WF # noqa: E402 import p3_s21_2_calibration as CAL # noqa: E402 SEED = 42 TOL = 1e-9 FEATURE_SHA16 = "0414e401522ea4e2" def _load(): ctx, rows, bin_rows = CAL.load_binary() y = np.asarray([1.0 if r["outcome"] == "WIN" else 0.0 for r in bin_rows], dtype=float) X = np.asarray([[r["feature_" + k] for k in PD.FEATURE_COLS] for r in bin_rows], dtype=float) return ctx, rows, bin_rows, X, y def s212_t01(): """Feature schema identical to P3-S20 (12 frozen columns + sha16).""" _, rows, bin_rows, X, _ = _load() feats = [[r["feature_" + k] for k in PD.FEATURE_COLS] for r in rows] h = hashlib.sha256(json.dumps([PD.FEATURE_COLS, feats], sort_keys=True, default=str).encode()).hexdigest()[:16] ok = (len(PD.FEATURE_COLS) == 12 and h == FEATURE_SHA16 and X.shape[1] == 12) return bool(ok), {"n_features": len(PD.FEATURE_COLS), "feature_sha16": h, "expected": FEATURE_SHA16} def s212_t02(): """Label contract unchanged (v1, H=16, classes, counts).""" _, rows, bin_rows, _, y = _load() n_win = int((y == 1).sum()); n_loss = int((y == 0).sum()) ok = (PD.CONTRACT_VERSION == "P3_S16_LABEL_CONTRACT_v1" and PD.HORIZON == 16 and PD.BINARY_CLASSES == ("WIN", "LOSS") and len(bin_rows) == 571 and n_win == 167 and n_loss == 404) return bool(ok), {"contract": PD.CONTRACT_VERSION, "horizon": PD.HORIZON, "binary_fit": len(bin_rows), "win": n_win, "loss": n_loss} def s212_t03(): """Fold boundaries + purge gaps + usage unchanged.""" _, _, bin_rows, _, _ = _load() frames = WF.fold_parts(bin_rows) bounds = [list(WF.FOLD_WINDOWS[i]) for i in range(3)] gaps = [g for (_, _, g) in frames] counts = [(len(tr), len(oo)) for (tr, oo, _) in frames] ok = (bounds == [[0, 300, 300, 395], [0, 395, 395, 490], [0, 490, 490, 571]] and gaps == [555, 472, 321] and counts == [(300, 95), (395, 95), (490, 81)]) return bool(ok), {"bounds": bounds, "gaps": gaps, "counts": counts} def s212_t04(): """Frozen Logistic + scaler configuration reproduced.""" m = LogisticRegression(C=1.0, max_iter=5000, random_state=SEED) p = m.get_params() ok = (p["C"] == 1.0 and p["max_iter"] == 5000 and p["random_state"] == SEED and p["class_weight"] is None and p["solver"] == "lbfgs" and p["fit_intercept"] is True) return bool(ok), {k: p[k] for k in ("C", "max_iter", "random_state", "class_weight", "solver", "fit_intercept")} def s212_t05(): """UN-calibrated baseline reproduced EXACTLY (== committed P3-S20 CSV).""" _, _, bin_rows, X, y = _load() folds = CAL.run_folds(bin_rows, X, y) raw = np.concatenate([fr["raw"] for fr in folds]) with open(os.path.join(OUT, "p3_s20_oos_predictions.csv"), encoding="utf-8") as f: ref = [float(r["pred_logistic_win_prob"]) for r in csv.DictReader(f)] md = max(abs(ref[i] - float(raw[i])) for i in range(len(ref))) return bool(len(ref) == len(raw) and md <= TOL), {"max_abs_diff": md, "n_oos": len(raw)} def s212_t06(): """Temporality: calibrators fit only on TRAIN (idx before OOS) + determinism (two fits -> identical sigmoid params).""" _, _, bin_rows, X, y = _load() folds1 = CAL.run_folds(bin_rows, X, y) folds2 = CAL.run_folds(bin_rows, X, y) frames = WF.fold_parts(bin_rows) ok_ord = all(max(ti) < min(oi) for (ti, oi, _) in frames) same = all(f1["cal"]["sigmoid_A"] == f2["cal"]["sigmoid_A"] and f1["cal"]["sigmoid_B"] == f2["cal"]["sigmoid_B"] for f1, f2 in zip(folds1, folds2)) return bool(ok_ord and same), {"train_before_oos": bool(ok_ord), "sigmoid_params_deterministic": bool(same), "folds": len(folds1)} def s212_t07(): """Decision-conserving reproducibility: the summary records the pre-registered decision B (ranking present, calibration not improving).""" with open(os.path.join(OUT, "p3_s21_2_summary.json"), encoding="utf-8") as f: s = json.load(f) d = s["decision"] raw_pooled = s["versions"]["raw"]["pooled"] ok = (d.startswith("B_") and raw_pooled["roc_auc"] > CAL.RANK_REPRO_ROC) return bool(ok), {"decision": d, "pooled_raw_roc": raw_pooled["roc_auc"]} def s212_t08(): """Determinism: two independent recomputations produce identical pure probability arrays (raw/sigmoid/isotonic) and an identical feature hash.""" _, rows, bin_rows, X, y = _load() f1 = CAL.run_folds(bin_rows, X, y) f2 = CAL.run_folds(bin_rows, X, y) feats = [[r["feature_" + k] for k in PD.FEATURE_COLS] for r in rows] h = hashlib.sha256(json.dumps([PD.FEATURE_COLS, feats], sort_keys=True, default=str).encode()).hexdigest()[:16] same = True for fr1, fr2 in zip(f1, f2): for v in ("raw", "sigmoid", "isotonic"): if not np.allclose(fr1[v], fr2[v], atol=TOL): same = False return bool(same and h == FEATURE_SHA16), {"hash": h, "same": same} def main(): tests = [ ("S21.2-T01", "feature schema identical to P3-S20", s212_t01), ("S21.2-T02", "label contract unchanged", s212_t02), ("S21.2-T03", "fold boundaries + purge unchanged", s212_t03), ("S21.2-T04", "frozen Logistic config reproduced", s212_t04), ("S21.2-T05", "uncalibrated baseline reproduced exactly", s212_t05), ("S21.2-T06", "calibrators fit train-only + deterministic", s212_t06), ("S21.2-T07", "committed summary decision B recorded", s212_t07), ("S21.2-T08", "recomputation deterministic + hash unchanged", s212_t08), ] results = [] for tid, title, fn in tests: try: ok, detail = fn() except Exception as e: # noqa: BLE001 ok, detail = False, {"error": repr(e)} results.append({"id": tid, "title": title, "pass": bool(ok), "detail": detail}) print(" [%s] %s %s" % ("PASS" if ok else "FAIL", tid, title)) n_pass = sum(1 for r in results if r["pass"]) import datetime as dt with open(os.path.join(OUT, "p3_s21_2_tests.json"), "w", encoding="utf-8") as f: json.dump({"tests": results, "total": len(results), "passed": n_pass, "generated_utc": dt.datetime.now(dt.timezone.utc).isoformat()}, f, indent=2) print("TOTAL=%d PASS=%d FAIL=%d" % (len(results), n_pass, len(results) - n_pass)) return 0 if n_pass == len(results) else 1 if __name__ == "__main__": sys.exit(main())