Aliwkarimi98/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 1
forked from chiki2bum2/SniperGold_ML
Code Pull requests Activity
SniperGold_ML/ml/p3/baseline/walk_forward.py

392 lines
17 KiB
Python
RawPermalinkNormal ViewHistory

test: define pre-registered walk-forward baseline
2026-08-24 08:39:27 +07:00
# -*- coding: utf-8 -*-
"""P3-S.20 — PRE-REGISTERED EXPANDING-WINDOW WALK-FORWARD BASELINE.
Validation phase: does the weak LOGISTIC signal from P3-S.18 (test ROC-AUC
~0.609) survive predefined temporal out-of-sample testing? LOGISTIC ONLY.
Frozen design (pre-registered BEFORE computing any final OOS metric):
- Expanded-window temporal walk-forward over the chronologically sorted
571 WIN/LOSS binary candidate-setup rows (leads, TP-before-SL v1).
- Fold boundaries (row indices): F1[0:300]/[300:395]; F2[0:395]/[395:490];
F3[0:490]/[490:571] (Fold1 train ~300 OOS ~95; then expanding; OOS ~95/81)
- purge gap >= HORIZON(16) bars enforced at every fold boundary.
- LogisticRegression(C=1.0, max_iter=5000, random_state=42) identical to
P3-S.18; StandardScaler fitted on TRAINING only, applied unchanged to OOS.
- Majority-class (trivial class-prevalence) baseline per fold, fit from the
TRAIN prevalence (no OOS leakage).
- Quality gate per fold: OOS >= 45 obs (pref >= 100), >= 20 WIN, >= 20 LOSS;
else LOW_STATISTICAL_POWER and preserved (no redesign after results).
- Threshold fixed at 0.5 ONLY for confusion / precision / recall tables.
Reuses ml/p3/baseline/prepare_dataset.py UNCHANGED (12 causal features
schema, identity/feature/label separation, label contract v1.0) and the
metric helpers from evaluate_baselines.py. No tree/boost/MLP/LSTM/Informer/
regime, no feature/label/TP/SL/horizon change, no threshold/HP tuning.
Guard discipline (frozen P3-S.4/S.5 parity-absence guards scan ml/**/*.py and
exempt files named spec_tests_*; this non-spec_tests_* file must NOT contain
the guarded tokens; docs/JSON/CSV may name concepts freely).
"""
import csv
import datetime as dt
import hashlib
import json
import os
import subprocess
import sys
import numpy as np
from sklearn.linear_model import LogisticRegression
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 evaluate_baselines as EV # noqa: E402
SEED = 42
N_FOLDS = 3
# ---- pre-registered expanding-window boundaries over sorted binary rows ----
# (frozen BEFORE final OOS metrics; derived only from the brief's target
# sizes: Fold1 train ~300 / OOS ~90-100; Fold2/3 expanding, OOS ~80-100)
FOLD_WINDOWS = ( # (train_start, train_end, oos_start, oos_end)
(0, 300, 300, 395),
(0, 395, 395, 490),
(0, 490, 490, 571),
)
QUALITY_MIN_OOS = 50
QUALITY_MIN_WIN = 20
QUALITY_MIN_LOSS = 20
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: # noqa: BLE001
return "unknown"
def load_binary():
"""Chronologically sorted 571 WIN/LOSS lead rows (binary fit)."""
ctx = PD.load_verified_population()
rows = PD.build_rows(ctx)
bin_rows = [r for r in rows if r["lead"] and r["outcome"] in
PD.BINARY_CLASSES]
bin_rows.sort(key=lambda r: (r["creation_bar"], r["setup_id"]))
return ctx, rows, bin_rows
def fold_parts(bin_rows):
"""Per-fold (train_idx, oos_idx, purge_gap) with the purge asserted.
Purge: OOS first creation_bar - last TRAIN creation_bar > HORIZON bars.
In the P3-S20 design the de-overlap naturally gives large gaps; the assert
prevents a silent boundary collapse (no boundary redesign after results)."""
parts = []
for (ts, te, os_, oe) in FOLD_WINDOWS:
train = list(range(ts, te))
oos = list(range(os_, oe))
last_tr = bin_rows[train[-1]]["creation_bar"]
first_os = bin_rows[oos[0]]["creation_bar"]
gap = int(first_os - last_tr)
assert gap >= PD.HORIZON, "pre-registered purge gap violated"
parts.append((train, oos, gap))
return parts
def metrics_block(y_arr, p_arr, tag):
"""Metric set for the logistic path.
EV.metrics() supplies ROC-AUC / PR-AUC / log-loss / Brier /
precision / recall / counts. Its reported 'balanced_acc' is plain
accuracy (inherited P3-S18 convention), so we recompute a true balanced
accuracy (macro-average recall, candidates: wins and losses) and keep the
raw accuracy alongside for transparency."""
m = EV.metrics(y_arr, p_arr)
y = np.asarray(y_arr); p = np.asarray(p_arr)
pred = (p >= 0.5).astype(int)
tl = (y == 1).astype(int)
n_win = int((tl == 1).sum()); n_loss = int((tl == 0).sum())
tp = int(((pred == 1) & (tl == 1)).sum())
tn = int(((pred == 0) & (tl == 0)).sum())
acc = float((tp + tn)) / max(int(len(y)), 1)
rec_win = float(tp) / max(n_win, 1)
rec_loss = float(tn) / max(n_loss, 1)
balacc = 0.5 * (rec_win + rec_loss) if (n_win and n_loss) else \
(rec_win if n_win else rec_loss)
m["balanced_acc"] = balacc
m["accuracy"] = acc
m["tag"] = tag
return m
def majority_baseline(y_tr, y_oos, tag):
"""Majority/constant-prior baseline: score = train WIN prevalence for
every OOS row (train-derived; no OOS leakage)."""
p_prior = float(np.mean(y_tr)) if len(y_tr) else 0.5
n = int(len(y_oos))
n_win = int(np.sum(y_oos == 1))
n_loss = int(np.sum(y_oos == 0))
if n == 0:
return {"tag": tag, "prior": p_prior, "roc_auc": 0.5, "pr_auc": None,
"log_loss": None, "brier": None, "balanced_acc": None,
"win_precision": 0.0, "win_recall": 0.0, "n": 0,
"n_win": 0, "n_loss": 0}
eps = 1e-12
prob = np.full(n, p_prior, dtype=float)
brier = float(np.mean((np.asarray(y_oos) - p_prior) ** 2))
logl = -float(np.mean(np.where(y_oos == 1,
np.log(max(p_prior, eps)),
np.log(max(1.0 - p_prior, eps)))))
pred_maj = (prob >= 0.5).astype(int)
tl = (y_oos == 1).astype(int)
tn = int(np.sum((pred_maj == 0) & (tl == 0)))
tp = int(np.sum((pred_maj == 1) & (tl == 1)))
pred_pos = int(np.sum(pred_maj == 1))
win_prec = float(tp) / max(pred_pos, 1)
win_rec = float(tp) / max(n_win, 1)
if n_win and n_loss:
# true balanced accuracy = macro-average recall (TN/n_loss, TP/n_win)
balacc = 0.5 * (float(tn) / n_loss + float(tp) / n_win)
else:
balacc = 0.5
return {"tag": tag, "prior": p_prior, "roc_auc": 0.5, "pr_auc": p_prior,
"pr_auc_note": "constant-prior baseline PR-AUC = OOS prevalence",
"log_loss": logl, "brier": brier, "balanced_acc": balacc,
"win_precision": win_prec, "win_recall": win_rec,
"n": n, "n_win": n_win, "n_loss": n_loss}
def classify(fold_results, pooled, gates_ok):
"""Pre-registered decision rule (brief sections 15 + 30).
D if the quality gates fail or no usable OOS AUC; A if every fold is
above 0.5 with a meaningful (>0.02) departure and pooled > 0.5;
B if >= 2 folds show meaningful departure in either direction; else C."""
if not gates_ok:
return "D_INCONCLUSIVE_DATA_TOO_SMALL"
aucs = [fr["logistic"]["oos"]["roc_auc"] for fr in fold_results]
if len(aucs) != len(fold_results) or not aucs or any(
a is None for a in aucs):
return "D_INCONCLUSIVE_DATA_TOO_SMALL"
above = sum(1 for a in aucs if a > 0.5)
strong = sum(1 for a in aucs if abs(a - 0.5) > 0.02)
pooled_roc = pooled["roc_auc"] if pooled else None
pooled_ok = pooled_roc is None or pooled_roc > 0.5
if above == len(aucs) and strong == len(aucs) and pooled_ok:
return "A_STABLE_WEAK_SIGNAL"
if strong >= 2:
return "B_WEAK_UNSTABLE_SIGNAL"
return "C_NO_REPRODUCIBLE_SIGNAL_DETECTED"
def main():
os.makedirs(OUT, exist_ok=True)
ctx, rows, bin_rows = load_binary()
bars = np.asarray([r["creation_bar"] for r in bin_rows], dtype=int)
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)
parts = fold_parts(bin_rows)
oos_pool_rows = []
fold_results = []
conf_rows = []
gates_ok = True
for fi, (train_idx, oos_idx, gap) in enumerate(parts):
tag = "fold%d" % (fi + 1)
Xtr = X[np.asarray(train_idx)]
ytr = y[np.asarray(train_idx)]
Xoos = X[np.asarray(oos_idx)]
yoos = y[np.asarray(oos_idx)]
scaler = StandardScaler().fit(Xtr) # train only
Xtr_s = scaler.transform(Xtr)
Xoos_s = scaler.transform(Xoos) # applied unchanged
clf = LogisticRegression(C=1.0, max_iter=5000, random_state=SEED)
clf.fit(Xtr_s, ytr)
ptr = clf.predict_proba(Xtr_s)[:, 1]
poos = clf.predict_proba(Xoos_s)[:, 1]
m_tr = metrics_block(ytr, ptr, tag + "_train")
m_oos = metrics_block(yoos, poos, tag + "_oos")
m_maj = majority_baseline(ytr, yoos, tag + "_majority")
q_n = int(yoos.size)
q_win = int(np.sum(yoos == 1))
q_loss = int(np.sum(yoos == 0))
q_ok = (q_n >= QUALITY_MIN_OOS and q_win >= QUALITY_MIN_WIN
and q_loss >= QUALITY_MIN_LOSS)
gates_ok = gates_ok and q_ok
pred_bin = (poos >= 0.5).astype(int)
tl = (yoos == 1).astype(int)
c_ww = int(np.sum((pred_bin == 1) & (tl == 1))) # WIN pred WIN
c_wl = int(np.sum((pred_bin == 0) & (tl == 1))) # WIN pred LOSS
c_lw = int(np.sum((pred_bin == 1) & (tl == 0))) # LOSS pred WIN
c_ll = int(np.sum((pred_bin == 0) & (tl == 0))) # LOSS pred LOSS
conf_rows.append({"fold": tag, "oos_WIN_predWIN": c_ww,
"oos_WIN_predLOSS": c_wl,
"oos_LOSS_predWIN": c_lw,
"oos_LOSS_predLOSS": c_ll})
for j, oos_global in enumerate(oos_idx):
oos_pool_rows.append({
"fold": fi + 1, "setup_id": bin_rows[oos_global]["setup_id"],
"creation_bar": int(bin_rows[oos_global]["creation_bar"]),
"outcome": bin_rows[oos_global]["outcome"],
"y_class": int(yoos[j]),
"pred_logistic_win_prob": float(poos[j]),
"majority_prior": float(m_maj["prior"]),
})
fold_results.append({
"fold": fi + 1,
"train_bars": (int(bars[train_idx[0]]), int(bars[train_idx[-1]])),
"oos_bars": (int(bars[oos_idx[0]]), int(bars[oos_idx[-1]])),
"gap_bars": gap,
"train_count": int(len(train_idx)),
"oos_count": int(len(oos_idx)),
"quality": {"oos_n": q_n, "oos_win": q_win, "oos_loss": q_loss,
"pass": bool(q_ok),
"power": "ok" if q_ok else "LOW_STATISTICAL_POWER"},
"logistic": {"train": m_tr, "oos": m_oos},
"majority": m_maj,
})
pool_y = np.asarray([r_["y_class"] for r_ in oos_pool_rows], dtype=float)
pool_p = np.asarray([r_["pred_logistic_win_prob"] for r_ in oos_pool_rows],
dtype=float)
pooled_metrics = metrics_block(pool_y, pool_p, "pooled_oos")
pool_maj = majority_baseline(pool_y, pool_y, "pooled_majority")
aucs = [fr["logistic"]["oos"]["roc_auc"] for fr in fold_results]
decision = classify(fold_results, pooled_metrics, gates_ok)
fold_csv = []
for fr in fold_results:
Lt = fr["logistic"]["train"]; Lo = fr["logistic"]["oos"]
M = fr["majority"]
fold_csv.append({
"fold": fr["fold"], "train_count": fr["train_count"],
"oos_count": fr["oos_count"], "gap_bars": fr["gap_bars"],
"train_roc_auc": Lt["roc_auc"], "train_logloss": Lt["log_loss"],
"oos_roc_auc": Lo["roc_auc"], "oos_pr_auc": Lo["pr_auc"],
"oos_logloss": Lo["log_loss"], "oos_brier": Lo["brier"],
"oos_balanced_acc": Lo["balanced_acc"],
"oos_win_precision": Lo["win_precision"],
"oos_win_recall": Lo["win_recall"],
"oos_n": Lo["n"], "oos_n_win": Lo["n_win"],
"oos_n_loss": Lo["n_loss"],
"majority_roc_auc": M["roc_auc"],
"majority_logloss": M["log_loss"], "majority_brier": M["brier"],
"majority_prior": M["prior"],
"quality_pass": bool(fr["quality"]["pass"]),
"quality_power": fr["quality"]["power"],
})
feats = [[r["feature_" + k] for k in PD.FEATURE_COLS] for r in rows]
labels = [r["outcome"] for r in rows]
feature_sha16 = _sha16([PD.FEATURE_COLS, feats])
label_sha16 = _sha16(labels)
schema_sha16 = _sha16([PD.CONTRACT_VERSION, PD.FEATURE_VERSION,
PD.FEATURE_COLS])
manifest = {
"experiment": "P3-S20 pre-registered expanding-window walk-forward",
"contract_version": PD.CONTRACT_VERSION,
"feature_version": PD.FEATURE_VERSION,
"model": {"primary": "LogisticRegression(C=1.0, L2, max_iter=5000)",
"scaler": "StandardScaler fit on train only",
"seed": SEED},
"feature_sha16": feature_sha16,
"label_sha16": label_sha16,
"schema_sha16": schema_sha16,
"counts": {"all_setup": len(rows),
"leads": int(sum(1 for r in rows if r["lead"])),
"binary_fit": len(bin_rows),
"win": int((y == 1).sum()),
"loss": int((y == 0).sum())},
"frozen_bounds": [list(t) for t in FOLD_WINDOWS],
"num_folds": N_FOLDS,
"purge_gap_bars": PD.HORIZON,
"quality_gate": {"min_oos": QUALITY_MIN_OOS,
"min_win": QUALITY_MIN_WIN,
"min_loss": QUALITY_MIN_LOSS},
"temporal_boundaries": {"first_bar": int(bars[0]),
"last_bar": int(bars[-1])},
"aggregation": "pooled OOS + per-fold mean/median/std (sample-aware)",
"git_commit": git_head(),
"generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
}
with open(os.path.join(OUT, "p3_s20_walkforward_manifest.json"), "w",
encoding="utf-8") as f:
json.dump(manifest, f, indent=2, default=str)
with open(os.path.join(OUT, "p3_s20_fold_results.csv"), "w", newline="",
encoding="utf-8") as f:
w = csv.DictWriter(f, fieldnames=list(fold_csv[0].keys()))
w.writeheader()
for r in fold_csv:
w.writerow(r)
with open(os.path.join(OUT, "p3_s20_oos_predictions.csv"), "w",
newline="", encoding="utf-8") as f:
wn = list(oos_pool_rows[0].keys())
w = csv.DictWriter(f, fieldnames=wn)
w.writeheader()
for r in oos_pool_rows:
w.writerow(r)
summary = {
"final_decision": decision,
"quality_gates_all_pass": gates_ok,
"fold_results": fold_results,
"pooled_oos_logistic": pooled_metrics,
"pooled_oos_majority": pool_maj,
"confusion_oos_at_05": conf_rows,
"fold_roc_mean": float(np.mean(aucs)),
"fold_roc_median": float(np.median(aucs)),
"fold_roc_std": float(np.std(aucs)),
"pooled_oos_roc_auc": pooled_metrics["roc_auc"],
"pooled_oos_pr_auc": pooled_metrics["pr_auc"],
"pooled_oos_logloss": pooled_metrics["log_loss"],
"pooled_oos_brier": pooled_metrics["brier"],
"note": "Decision is evidence-based; no fold/feature/HP selection. "
"Threshold 0.5 only for confusion/precision/recall tables.",
"generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
}
with open(os.path.join(OUT, "p3_s20_summary.json"), "w", encoding="utf-8") as f:
json.dump(summary, f, indent=2, default=str)
print("P3-S20 walk-forward: %d binary rows, %d folds" %
(len(bin_rows), N_FOLDS))
for fr in fold_results:
o = fr["logistic"]["oos"]
print(" %s train=%d oos=%d roc_auc=%s pr_auc=%s logloss=%s "
"brier=%s (win=%d loss=%d)" % (
fr["fold"], fr["train_count"], fr["oos_count"],
o["roc_auc"], o["pr_auc"], o["log_loss"], o["brier"],
o["n_win"], o["n_loss"]))
print(" pooled OOS n=%d roc_auc=%s pr_auc=%s logloss=%s brier=%s" % (
pooled_metrics["n"], pooled_metrics["roc_auc"],
pooled_metrics["pr_auc"], pooled_metrics["log_loss"],
pooled_metrics["brier"]))
print(" decision: %s" % decision)
print("[saved] p3_s20_walkforward_manifest.json / "
"p3_s20_fold_results.csv / p3_s20_oos_predictions.csv / "
"p3_s20_summary.json")
return 0
if __name__ == "__main__":
sys.exit(main())
Copy permalink