forked from chiki2bum2/SniperGold_ML
307 lines
13 KiB
Python
307 lines
13 KiB
Python
 | # -*- coding: utf-8 -*-
| ||
"""P3-S21.2 — CALIBRATION DIAGNOSTICS (frozen Logistic, temporal OOS).
| |||
| |||
Question: can the weak logistic ranking signal be calibrated OUT-OF-SAMPLE in
| |||
a clean way WITHOUT changing any frozen contract? Frozen recipe, folds and
| |||
population are exactly those of P3-S20 / P3-S21.1.
| |||
| |||
Calibration is fit ONLY on each fold's TRAINING rows (strictly before the
| |||
fold's OOS window -> causal, no leakage). sigmoid is a penalty-free logistic on
| |||
logit(raw score) (Platt). isotonic uses IsotonicRegression(out_of_bounds=(
| |||
"clip")); its OOS evaluation is small (n=95/95/81) so it is flagged
| |||
LOW_STATISTICAL_POWER and is not treated as primary evidence.
| |||
| |||
NOT DONE: threshold tuning, model escalation, feature selection/removal,
| |||
TP/SL/horizon/label changes, MQL5 / FEATURE_CONTRACT changes, external data.
| |||
No selection of the calibration method after seeing pooled OOS metrics.
| |||
| |||
Guard discipline: this file is not named spec_tests_* and avoids the frozen
| |||
P3-4/P3-5 parity-absence guard tokens (no runtime detector / zone-type
| |||
runtime identifier); provenance lives in a report document.
| |||
"""
| |||
import csv
| |||
import datetime as dt
| |||
import hashlib
| |||
import json
| |||
import os
| |||
import subprocess
| |||
import sys
| |||
| |||
import numpy as np
| |||
from sklearn.isotonic import IsotonicRegression
| |||
from sklearn.linear_model import LogisticRegression
| |||
from sklearn.metrics import (average_precision_score, brier_score_loss,
| |||
log_loss, roc_auc_score)
| |||
from sklearn.preprocessing import StandardScaler
| |||
| |||
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
| |||
| |||
SEED = 42
| |||
REPRO_TOL = 1e-9
| |||
VERSIONS = ("raw", "sigmoid", "isotonic")
| |||
N_RELI_BINS = 10
| |||
N_BOOT = 1000
| |||
IMPROVE_LL_EPS = 1e-4
| |||
IMPROVE_BRIER_EPS = 1e-3
| |||
POOLED_LL_MATERIAL = 5e-3
| |||
POOLED_BRIER_MATERIAL = 5e-3
| |||
RANK_REPRO_ROC = 0.52
| |||
| |||
| |||
def _sha16(obj):
| |||
return hashlib.sha256(json.dumps(obj, sort_keys=True,
| |||
default=str).encode()).hexdigest()[:16]
| |||
| |||
| |||
def git_head():
| |||
try:
| |||
return subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True,
| |||
text=True, check=True).stdout.strip()
| |||
except Exception:
| |||
return "unknown"
| |||
| |||
| |||
def load_binary():
| |||
"""ctx, rows, bin_rows (chronologically sorted 571 WIN/LOSS leads)."""
| |||
ctx, rows, bin_rows = WF.load_binary()
| |||
return ctx, rows, bin_rows
| |||
| |||
| |||
def _logit(x):
| |||
x = np.clip(np.asarray(x, float), 1e-12, 1.0 - 1e-12)
| |||
return np.log(x / (1.0 - x))
| |||
| |||
| |||
def _sig(z, a, b):
| |||
return 1.0 / (1.0 + np.exp(-(np.asarray(z, float) * a + b)))
| |||
| |||
| |||
def metric(y, p):
| |||
y = np.asarray(y, float); p = np.asarray(p, float)
| |||
pc = np.clip(p, 1e-12, 1.0 - 1e-12)
| |||
roc = float(roc_auc_score(y, pc)) if len(np.unique(y)) >= 2 else None
| |||
pr = (float(average_precision_score(y, pc))
| |||
if len(np.unique(y)) >= 2 else None)
| |||
return {"log_loss": round(float(log_loss(y, pc, labels=[0, 1])), 6),
| |||
"brier": round(float(brier_score_loss(y, pc)), 6),
| |||
"roc_auc": roc, "pr_auc": pr,
| |||
"n": int(len(y)), "n_win": int(np.sum(y == 1)),
| |||
"n_loss": int(np.sum(y == 0))}
| |||
| |||
| |||
def rel_curve(y, p):
| |||
y = np.asarray(y, float); p = np.asarray(p, float)
| |||
if len(p) == 0 or len(set(p)) == 1:
| |||
return []
| |||
edges = np.sort(np.unique(np.percentile(p, np.linspace(0, 100, N_RELI_BINS + 1))))
| |||
out = []
| |||
for i in range(len(edges) - 1):
| |||
lo, hi = edges[i], edges[i + 1]
| |||
if i < len(edges) - 2:
| |||
sel = (p >= lo) & (p < hi)
| |||
else:
| |||
sel = (p >= lo) & (p <= hi)
| |||
if int(sel.sum()) == 0:
| |||
continue
| |||
out.append({"bin_low": round(float(lo), 6), "bin_high": round(float(hi), 6),
| |||
"mean_pred": round(float(np.mean(p[sel])), 6),
| |||
"observed": round(float(np.mean(y[sel])), 6),
| |||
"count": int(sel.sum())})
| |||
return out
| |||
| |||
| |||
def calib_slope_intercept(y, p):
| |||
y = np.asarray(y, float); p = np.asarray(p, float)
| |||
if len(np.unique(y)) < 2 or len(p) < 2:
| |||
return {"intercept": None, "slope": None}
| |||
try:
| |||
m = LogisticRegression(penalty=None, solver="lbfgs", max_iter=5000)
| |||
m.fit(_logit(p).reshape(-1, 1), y)
| |||
return {"intercept": round(float(m.intercept_[0]), 6),
| |||
"slope": round(float(m.coef_[0][0]), 6)}
| |||
except Exception:
| |||
return {"intercept": None, "slope": None}
| |||
| |||
| |||
def boot_brier(y, pA, pB):
| |||
rng = np.random.default_rng(SEED)
| |||
y = np.asarray(y, float); A = np.asarray(pA, float); B = np.asarray(pB, float)
| |||
idx = rng.integers(0, len(y), size=(N_BOOT, len(y)))
| |||
d = np.empty(N_BOOT)
| |||
for i, it in enumerate(idx):
| |||
d[i] = np.mean((y[it] - A[it]) ** 2) - np.mean((y[it] - B[it]) ** 2)
| |||
return {"brier_raw_minus_cal_mean": round(float(np.mean(d)), 6),
| |||
"brier_se": round(float(np.std(d)), 6), "n_boot": N_BOOT}
| |||
| |||
| |||
def fit_baseline(Xtr, ytr, Xoos):
| |||
sc = StandardScaler().fit(Xtr)
| |||
clf = LogisticRegression(C=1.0, max_iter=5000, random_state=SEED)
| |||
clf.fit(sc.transform(Xtr), ytr)
| |||
return (clf.predict_proba(sc.transform(Xtr))[:, 1],
| |||
clf.predict_proba(sc.transform(Xoos))[:, 1])
| |||
| |||
| |||
def run_folds(bin_rows, X, y):
| |||
parts = WF.fold_parts(bin_rows)
| |||
folds = []
| |||
for fi, (tri, ooi, gap) in enumerate(parts):
| |||
ptr, po = fit_baseline(X[np.asarray(tri)], y[np.asarray(tri)],
| |||
X[np.asarray(ooi)])
| |||
yo = y[np.asarray(ooi)]
| |||
s = LogisticRegression(penalty=None, solver="lbfgs", max_iter=5000)
| |||
s.fit(_logit(ptr).reshape(-1, 1), y[np.asarray(tri)])
| |||
A, B = float(s.coef_[0][0]), float(s.intercept_[0])
| |||
psig = _sig(_logit(po), A, B)
| |||
iso = IsotonicRegression(y_min=0.0, y_max=1.0, out_of_bounds="clip")
| |||
iso.fit(ptr, y[np.asarray(tri)])
| |||
piso = np.asarray(iso.predict(po), float)
| |||
power = ("ok" if (len(yo) >= 50 and np.sum(yo == 1) >= 15 and
| |||
np.sum(yo == 0) >= 15) else "LOW_STATISTICAL_POWER")
| |||
folds.append({"idx": fi + 1, "gap": int(gap),
| |||
"ids": [int(bin_rows[g]["setup_id"]) for g in ooi],
| |||
"cl": [bin_rows[g]["outcome"] for g in ooi],
| |||
"y": yo, "raw": po, "sigmoid": psig, "isotonic": piso,
| |||
"cal": {"sigmoid_A": A, "sigmoid_B": B,
| |||
"iso_oob": "clip", "fit_n": int(len(tri)),
| |||
"eval_power": power}})
| |||
return folds
| |||
| |||
| |||
def classify(res, prior_stats):
| |||
pooled_raw = res["raw"]["pooled"]
| |||
if pooled_raw["roc_auc"] is None or pooled_raw["roc_auc"] <= RANK_REPRO_ROC:
| |||
return "E_no_reproducible_ranking_or_calibration"
| |||
rf = res["raw"]["folds"]
| |||
for v in ("sigmoid", "isotonic"):
| |||
f = res[v]["folds"]
| |||
folds_ok = all(
| |||
rf[k]["log_loss"] - f[k]["log_loss"] > IMPROVE_LL_EPS and
| |||
rf[k]["brier"] - f[k]["brier"] > IMPROVE_BRIER_EPS
| |||
for k in (1, 2, 3))
| |||
dll = res["raw"]["pooled"]["log_loss"] - res[v]["pooled"]["log_loss"]
| |||
dbr = res["raw"]["pooled"]["brier"] - res[v]["pooled"]["brier"]
| |||
if folds_ok and dll >= POOLED_LL_MATERIAL and dbr >= POOLED_BRIER_MATERIAL:
| |||
return "A_%s_improves_reproducibly" % v
| |||
return "B_ranking_exists_but_calibration_does_not_improve"
| |||
| |||
| |||
def main():
| |||
os.makedirs(OUT, exist_ok=True)
| |||
ctx, rows, bin_rows = WF.load_binary()
| |||
y = np.asarray([1.0 if r["outcome"] == "WIN" else 0.0 for r in bin_rows], float)
| |||
X = np.asarray([[r["feature_" + k] for k in PD.FEATURE_COLS]
| |||
for r in bin_rows], float)
| |||
| |||
folds = run_folds(bin_rows, X, y)
| |||
| |||
# reproduction vs committed P3-S20 CSV
| |||
with open(os.path.join(OUT, "p3_s20_oos_predictions.csv"), encoding="utf-8") as f:
| |||
ref = [r for r in csv.DictReader(f)]
| |||
raw_pool = np.concatenate([fr["raw"] for fr in folds])
| |||
max_diff = max(abs(float(r["pred_logistic_win_prob"]) - float(raw_pool[i]))
| |||
for i, r in enumerate(ref))
| |||
repro_ok = bool(len(ref) == len(raw_pool) and max_diff <= REPRO_TOL)
| |||
| |||
y_pool = np.concatenate([fr["y"] for fr in folds])
| |||
P = {v: np.concatenate([fr[v] for fr in folds]) for v in VERSIONS}
| |||
| |||
res = {}
| |||
for v in VERSIONS:
| |||
res[v] = {"folds": {fr["idx"]: metric(fr["y"], fr[v]) for fr in folds},
| |||
"pooled": metric(y_pool, P[v]),
| |||
"line": calib_slope_intercept(y_pool, P[v]),
| |||
"rel": rel_curve(y_pool, P[v]),
| |||
"boot": (boot_brier(y_pool, P["raw"], P[v]) if v != "raw"
| |||
else None)}
| |||
| |||
# constant prior (train-window prevalence, no leakage)
| |||
train_end = [300, 395, 490]
| |||
priors = [float(np.mean(y[0:e])) for e in train_end]
| |||
for k in range(3):
| |||
yo = folds[k]["y"]
| |||
pz = np.full(len(yo), priors[k], float)
| |||
res["prior_fold_%d" % (k + 1)] = {"prior": round(float(priors[k]), 6),
| |||
"stats": metric(yo, pz)}
| |||
prior_pool = np.concatenate([np.full(len(fr["y"]), priors[fr["idx"] - 1], float)
| |||
for fr in folds])
| |||
res["prior_pooled"] = metric(y_pool, prior_pool)
| |||
| |||
decision = classify(res, res["prior_pooled"])
| |||
| |||
# ---- outputs ----
| |||
with open(os.path.join(OUT, "p3_s21_2_oos_predictions.csv"), "w",
| |||
newline="", encoding="utf-8") as f:
| |||
w = csv.DictWriter(f, fieldnames=["fold", "setup_id", "outcome",
| |||
"y_class", "raw", "sigmoid", "isotonic"])
| |||
w.writeheader()
| |||
for fr in folds:
| |||
for j in range(len(fr["y"])):
| |||
w.writerow({"fold": fr["idx"], "setup_id": fr["ids"][j],
| |||
"outcome": fr["cl"][j], "y_class": int(fr["y"][j]),
| |||
"raw": round(float(fr["raw"][j]), 8),
| |||
"sigmoid": round(float(fr["sigmoid"][j]), 8),
| |||
"isotonic": round(float(fr["isotonic"][j]), 8)})
| |||
with open(os.path.join(OUT, "p3_s21_2_fold_metrics.csv"), "w",
| |||
newline="", encoding="utf-8") as f:
| |||
w = csv.DictWriter(f, fieldnames=["fold", "version", "log_loss", "brier",
| |||
"roc_auc", "pr_auc", "n"])
| |||
w.writeheader()
| |||
for fr in folds:
| |||
for v in VERSIONS:
| |||
s = res[v]["folds"][fr["idx"]]
| |||
w.writerow({"fold": fr["idx"], "version": v,
| |||
"log_loss": s["log_loss"], "brier": s["brier"],
| |||
"roc_auc": s["roc_auc"], "pr_auc": s["pr_auc"],
| |||
"n": s["n"]})
| |||
with open(os.path.join(OUT, "p3_s21_2_reliability.csv"), "w",
| |||
newline="", encoding="utf-8") as f:
| |||
w = csv.DictWriter(f, fieldnames=["version", "bin_low", "bin_high",
| |||
"mean_pred", "observed", "count"])
| |||
w.writeheader()
| |||
for v in VERSIONS:
| |||
for b in res[v]["rel"]:
| |||
w.writerow({"version": v, **b})
| |||
with open(os.path.join(OUT, "p3_s21_2_calibrators.json"), "w",
| |||
encoding="utf-8") as f:
| |||
json.dump([fr["cal"] for fr in folds], f, indent=2, default=str)
| |||
| |||
feats = [[r["feature_" + k] for k in PD.FEATURE_COLS] for r in rows]
| |||
summary = {"session": "P3-S21.2 calibration diagnostics",
| |||
"reproduction": {"exact": bool(repro_ok),
| |||
"max_abs_prob_diff": max_diff, "n_oos": int(len(ref))},
| |||
"feature_sha16": _sha16([PD.FEATURE_COLS, feats]),
| |||
"counts": {"binary_fit": len(bin_rows),
| |||
"win": int(np.sum(y == 1)), "loss": int(np.sum(y == 0))},
| |||
"versions": {v: {"folds": res[v]["folds"], "pooled": res[v]["pooled"],
| |||
"line": res[v]["line"], "boot": res[v]["boot"]}
| |||
for v in VERSIONS},
| |||
"prior_fold": {k: res["prior_fold_%d" % k]
| |||
for k in (1, 2, 3)},
| |||
"prior_pooled": res["prior_pooled"],
| |||
"decision": decision,
| |||
"git": git_head(),
| |||
"generated_utc": dt.datetime.now(dt.timezone.utc).isoformat()}
| |||
with open(os.path.join(OUT, "p3_s21_2_summary.json"), "w", encoding="utf-8") as f:
| |||
json.dump(summary, f, indent=2, default=str)
| |||
| |||
print("P3-S21.2 calibration diagnostics")
| |||
print(" repro_exact=%s max_abs_diff=%.3e" % (bool(repro_ok), max_diff))
| |||
for v in VERSIONS:
| |||
pm = res[v]["pooled"]
| |||
print(" pooled %-8s ll=%.4f brier=%.4f roc=%s pr=%s" % (
| |||
v, pm["log_loss"], pm["brier"], pm["roc_auc"], pm["pr_auc"]))
| |||
pp = res["prior_pooled"]
| |||
print(" prior pooled ll=%.4f brier=%.4f" % (pp["log_loss"], pp["brier"]))
| |||
print(" decision: %s" % decision)
| |||
return 0
| |||
| |||
| |||
if __name__ == "__main__":
| |||
sys.exit(main())
|