PaPP v2 Modello: self-test (reproducibilita', anti-leakage, varianti)

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-06-17 07:48:26 +00:00
parent fe10be5ecc
commit 4084008642
1 changed files with 80 additions and 0 deletions
@@ -0,0 +1,80 @@
+"""Self-test della pipeline Fase 3 (controlli di correttezza, non un training serio).
+
+Esegue:
+  T1 reproducibilita': due run identiche danno stesse metriche OOS
+  T2 controllo negativo anti-leakage: con label OOS mescolate l'AUC deve ~0.5
+  T3 no temporal leakage: la baseline OOS dipende SOLO dallo sviluppo
+  T4 variante logistic gira senza errori
+  T5 restrict_pairs (solo pattern robusti) gira senza errori
+  T6 le feature non contengono colonne di esito futuro
+"""
+import os, sys, copy
+import numpy as np
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, HERE)
+import train as T
+from data import load
+from labeling import RegimeBaseline
+from splits import split_oos
+
+CFG = os.path.join(HERE, "..", "config.yaml")
+ok = True
+
+def check(name, cond):
+    global ok
+    print(("PASS" if cond else "FAIL"), "-", name)
+    ok = ok and cond
+
+# T6: nessuna feature e' un esito futuro
+cfg = T.load_cfg(CFG)
+future_like = ("ret_", "mfe_", "mae_", "dir_", "rev_", "cret_", "bars_to_revert", "disc_max")
+feats = cfg["features_numeric"] + cfg["features_categorical"]
+check("T6 feature senza esiti futuri",
+      not any(f.startswith(future_like) or f in ("bars_to_revert",) for f in feats))
+
+# T1: reproducibilita'
+folds1, oos1 = T.run(CFG)
+folds2, oos2 = T.run(CFG)
+check("T1 reproducibilita' OOS AUC", abs(oos1["auc"] - oos2["auc"]) < 1e-9)
+
+# T3: baseline OOS non dipende dai dati OOS (no temporal leakage)
+crosses, bars = load(cfg["symbol"], cfg["year_min"])
+crosses = crosses.dropna(subset=[f"cret_{cfg['horizon']}"]).reset_index(drop=True)
+dev_cr, oos_cr = split_oos(crosses, cfg["oos_start_year"])
+dev_ba, oos_ba = split_oos(bars, cfg["oos_start_year"])
+rb_dev = RegimeBaseline(cfg["regime"], cfg["horizon"]).fit(dev_ba)
+lbl_a = rb_dev.label(oos_cr)
+# se cambio i dati OOS, l'edges/baseline (fittati su dev) NON cambiano
+rb_dev2 = RegimeBaseline(cfg["regime"], cfg["horizon"]).fit(dev_ba)
+lbl_b = rb_dev2.label(oos_cr)
+check("T3 baseline OOS deterministica e da solo-sviluppo", np.array_equal(lbl_a, lbl_b)
+      and np.allclose(rb_dev.edges_["cl"], rb_dev2.edges_["cl"]))
+
+# T2: controllo negativo (label mescolate -> AUC ~0.5)
+from evaluate import metrics
+from features import select_xy
+from model import make_model
+rng = np.random.default_rng(0)
+y_dev = rb_dev.label(dev_cr)
+final = make_model(cfg, cfg["features_numeric"], cfg["features_categorical"])
+final.fit(select_xy(dev_cr, cfg["features_numeric"], cfg["features_categorical"]), y_dev)
+p_oos = final.predict_proba(select_xy(oos_cr, cfg["features_numeric"], cfg["features_categorical"]))[:, 1]
+y_oos = rb_dev.label(oos_cr)
+y_shuf = y_oos.copy(); rng.shuffle(y_shuf)
+m_shuf = metrics(y_shuf, p_oos)
+check("T2 controllo negativo AUC~0.5 (0.45-0.55)", 0.45 <= m_shuf["auc"] <= 0.55)
+
+# T4: variante logistic
+cfg_log = copy.deepcopy(cfg); cfg_log["model"]["kind"] = "logistic"
+ml = make_model(cfg_log, cfg["features_numeric"], cfg["features_categorical"])
+ml.fit(select_xy(dev_cr, cfg["features_numeric"], cfg["features_categorical"]), y_dev)
+pl = ml.predict_proba(select_xy(oos_cr, cfg["features_numeric"], cfg["features_categorical"]))[:, 1]
+check("T4 logistic produce probabilita' valide", np.all((pl >= 0) & (pl <= 1)))
+
+# T5: restrict_pairs
+robust = ["MA365xMA7", "MA121xMA7", "MA121xMA3", "PRICExMA121", "MA182xMA30"]
+sub = dev_cr[dev_cr.pair.isin(robust)]
+check("T5 restrict_pairs ha campioni", len(sub) > 100)
+
+print("\n=== RISULTATO SELF-TEST:", "TUTTO OK" if ok else "CI SONO FAIL", "===")
+sys.exit(0 if ok else 1)