forked from chiki2bum2/SniperGold_ML
392 lines
No EOL
17 KiB
Python
392 lines
No EOL
17 KiB
Python
# -*- 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()) |