# -*- coding: utf-8 -*- """P3-S.20 — PRE-REGISTERED WALK-FORWARD BASELINE: deterministic spec tests. Truth : the pre-registered expanding-window design frozen by this phase (brief sections on pre-registration / temporal isolation / comparators / quality gate / reproducibility) over the verified P3-S.18/P3-S.19 dataset (prepare_dataset.py, unchanged). Observed : ml/p3/baseline/walk_forward.py. Discipline : deterministic checks; no performance selection, no threshold/ feature/HP tuning, no model escalation. Tests WF-T01..T06. Filename is NOT spec_tests_* (it obeys the frozen parity-absence guards): the source avoids the guarded tokens entirely. """ import json import os import sys import numpy as np from sklearn.linear_model import LogisticRegression from sklearn.metrics import 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 TOL = 1e-9 def _dataset(): """Shared loader: (ctx, rows, sorted binary rows).""" return WF.load_binary() def _fit_oos_auc(bin_rows, train_idx, oos_idx): """Fresh frozen logistic fit (train-only scaling) -> OOS ROC-AUC.""" ytr = np.asarray([1.0 if bin_rows[k]["outcome"] == "WIN" else 0.0 for k in train_idx], dtype=float) yoos = np.asarray([1.0 if bin_rows[k]["outcome"] == "WIN" else 0.0 for k in oos_idx], dtype=float) Xtr = np.asarray([[bin_rows[k]["feature_" + f] for f in PD.FEATURE_COLS] for k in train_idx], dtype=float) Xoos = np.asarray([[bin_rows[k]["feature_" + f] for f in PD.FEATURE_COLS] for k in oos_idx], dtype=float) sc = StandardScaler().fit(Xtr) mdl = LogisticRegression(C=1.0, max_iter=5000, random_state=SEED) mdl.fit(sc.transform(Xtr), ytr) p = mdl.predict_proba(sc.transform(Xoos))[:, 1] try: return float(roc_auc_score(yoos, p)) except Exception: # noqa: BLE001 return None def wf_t01(): """Chronological ordering + expanding train windows across folds.""" _, _, bin_rows = _dataset() chrono = all(bin_rows[i]["creation_bar"] <= bin_rows[i + 1]["creation_bar"] for i in range(len(bin_rows) - 1)) frames = WF.fold_parts(bin_rows) prev_train = None expand = True for (train, _oos, _g) in frames: if prev_train is not None and not set(prev_train).issubset(set(train)): expand = False prev_train = train return bool(chrono and expand), {"chronological": bool(chrono), "expanding": bool(expand), "n_binary": len(bin_rows), "n_folds": len(frames)} def wf_t02(): """Purge: OOS first bar - last train bar > HORIZON for every fold.""" _, _, bin_rows = _dataset() frames = WF.fold_parts(bin_rows) gaps = [] ok = True for (_tr, _oos, g) in frames: gaps.append(g) if g <= PD.HORIZON: ok = False return bool(ok), {"gaps_bars": gaps, "horizon": PD.HORIZON} def wf_t03(): """Preprocessing is train-only: every train index < every OOS index.""" _, _, bin_rows = _dataset() frames = WF.fold_parts(bin_rows) ok = all(max(train_idx) < min(oos_idx) for (train_idx, oos_idx, _g) in frames) return bool(ok), {"train_before_oos_all_folds": ok} def wf_t04(): """Per-fold quality gate recorded (no redesign after results).""" _, _, bin_rows = _dataset() y = np.asarray([1.0 if r["outcome"] == "WIN" else 0.0 for r in bin_rows], dtype=float) frames = WF.fold_parts(bin_rows) det = [] for i, (_tr, oos_idx, _g) in enumerate(frames): qn = int(len(oos_idx)) qw = int((y[oos_idx] == 1).sum()) ql = int((y[oos_idx] == 0).sum()) passed = (qn >= WF.QUALITY_MIN_OOS and qw >= WF.QUALITY_MIN_WIN and ql >= WF.QUALITY_MIN_LOSS) det.append({"fold": i + 1, "oos_n": qn, "oos_win": qw, "oos_loss": ql, "power": "ok" if passed else "LOW_STATISTICAL_POWER"}) ok = all(d["power"] == "ok" for d in det) return bool(ok), {"folds": det} def wf_t05(): """Majority-class comparator present per fold (constant-prior, no OOS leakage: prior from the training WIN prevalence only).""" _, _, bin_rows = _dataset() y = np.asarray([1.0 if r["outcome"] == "WIN" else 0.0 for r in bin_rows], dtype=float) frames = WF.fold_parts(bin_rows) for i, (train_idx, oos_idx, _g) in enumerate(frames): m = WF.majority_baseline(y[train_idx], y[oos_idx], "fold%d_majority" % (i + 1)) if m["roc_auc"] != 0.5 or m["prior"] is None: return False, {"fold": i + 1, "prior": m["prior"]} return True, {"comparator": "constant-prior (train prevalence) per fold"} def wf_t06(): """Same seed + same inputs -> identical OOS AUROC (determinism).""" _, _, bin_rows = _dataset() frames = WF.fold_parts(bin_rows) r1 = [_fit_oos_auc(bin_rows, tr, oo) for (tr, oo, _g) in frames] r2 = [_fit_oos_auc(bin_rows, tr, oo) for (tr, oo, _g) in frames] same = len(r1) == len(r2) and all( (a is None and b is None) or (a is not None and b is not None and abs(a - b) < TOL) for (a, b) in zip(r1, r2)) return bool(same), {"run1": r1, "run2": r2} def main(): tests = [ ("WF-T01", "chronological/expanding windows", wf_t01), ("WF-T02", "purge temporal isolation (gap>H)", wf_t02), ("WF-T03", "preprocessing fit-only (train before OOS)", wf_t03), ("WF-T04", "quality gate + LOW_STATISTICAL_POWER handled", wf_t04), ("WF-T05", "majority comparator present per fold", wf_t05), ("WF-T06", "deterministic OOS AUROC (single-seed repeat)", wf_t06), ] 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_s20_wf_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())