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