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