From 4084008642fccd61804947eed552562bfd10312d Mon Sep 17 00:00:00 2001 From: Pietro Giacobazzi Date: Wed, 17 Jun 2026 07:48:26 +0000 Subject: [PATCH] PaPP v2 Modello: self-test (reproducibilita', anti-leakage, varianti) Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- PaPP v2/Modello/src/selftest.py | 80 +++++++++++++++++++++++++++++++++ 1 file changed, 80 insertions(+) create mode 100644 PaPP v2/Modello/src/selftest.py diff --git a/PaPP v2/Modello/src/selftest.py b/PaPP v2/Modello/src/selftest.py new file mode 100644 index 0000000..62d55a8 --- /dev/null +++ b/PaPP v2/Modello/src/selftest.py @@ -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)