forked from chiki2bum2/SniperGold_ML
187 lines
8 KiB
Python
187 lines
8 KiB
Python
 | # -*- 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())
|