SniperGold_ML/ml/p3/baseline/test_walk_forward.py

# -*- 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())