"""FASE 1 — triage dei 2 superstiti su BTC/ETH, sull'harness onesto (research_lab). Sul feed pulito solo SH01 (shape-ML) e frammenti HONEST mostravano segnale residuo. Delle HONEST solo DIP (dip-reversion) è testabile su BTC/ETH (TR01/ROT02 richiedono alt esclusi). Qui ri-implemento DIP e SH01-shape-ML come SERIE DI POSIZIONE e li passo ai gate onesti (FULL/OOS-VAL, vs buy&hold, null p-value, sweep fee, griglia). Hold-out 2025+ resta BLOCCATO. uv run python scripts/analysis/phase1_survivors.py """ from __future__ import annotations import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) import numpy as np import pandas as pd from sklearn.linear_model import LogisticRegression from sklearn.preprocessing import StandardScaler from src.data.downloader import load_data from scripts.analysis.research_lab import ( backtest, buy_hold, mc_pvalue, report, VAL_START, HOLDOUT_START, FEE_RT, ) # ----------------------------- DIP reversion (long-only) ----------------------------- def dip_signal(df, n=50, k=2.0, z_exit=0.0, max_bars=72): """Long-only: entra (pos=1) quando lo z-score causale del prezzo vs MA(n) <= -k (dip), esce quando z>=z_exit o dopo max_bars. Decisione a close[i] (z[i] usa close[i]), guadagna close[i]->close[i+1]. Niente fill su estremi di candela.""" c = df["close"].values.astype(float) s = pd.Series(c) ma = s.rolling(n).mean().values sd = s.rolling(n).std().values z = np.where(sd > 0, (c - ma) / sd, np.nan) pos = np.zeros(len(c)) inpos = False held = 0 for i in range(len(c)): if not inpos: if not np.isnan(z[i]) and z[i] <= -k: inpos, held = True, 0 pos[i] = 1.0 else: held += 1 if (not np.isnan(z[i]) and z[i] >= z_exit) or held >= max_bars: inpos = False # esce al close[i]: pos[i]=0 else: pos[i] = 1.0 return pos # ----------------------------- SH01 shape-ML (walk-forward) ----------------------------- def _shape_features(df, W): """~12 feature di FORMA causali per barra, dalla finestra che termina a i (usa solo <=i).""" o = df["open"].values.astype(float); h = df["high"].values.astype(float) l = df["low"].values.astype(float); c = df["close"].values.astype(float) s = pd.Series(c) ret1 = s.pct_change() rng = (h - l) / np.where(c > 0, c, np.nan) body = (c - o) / np.where(h - l > 0, h - l, np.nan) up_sh = (h - np.maximum(o, c)) / np.where(h - l > 0, h - l, np.nan) dn_sh = (np.minimum(o, c) - l) / np.where(h - l > 0, h - l, np.nan) # RSI(14) d = s.diff() gain = d.clip(lower=0).rolling(14).mean() loss = (-d.clip(upper=0)).rolling(14).mean() rsi = 100 - 100 / (1 + gain / loss.replace(0, np.nan)) hi_w = pd.Series(h).rolling(W).max(); lo_w = pd.Series(l).rolling(W).min() feat = { "mom_w": s / s.shift(W) - 1.0, # rendimento sulla finestra "mom_half": s / s.shift(W // 2) - 1.0, # accelerazione "vol_w": ret1.rolling(W).std(), "rsi": rsi / 100.0, "ma_dist": (c - s.rolling(W).mean()) / s.rolling(W).std(), "pos_in_range": (c - lo_w) / (hi_w - lo_w).replace(0, np.nan), # dove sta il close nel range W "range": pd.Series(rng).rolling(3).mean(), "body": pd.Series(body).rolling(3).mean(), "up_shadow": pd.Series(up_sh).rolling(3).mean(), "dn_shadow": pd.Series(dn_sh).rolling(3).mean(), "ret1": ret1, "skew_w": ret1.rolling(W).skew(), } X = pd.DataFrame(feat).values return X def shape_ml_signal(df, W=24, H=12, th=0.55, refit=750, warmup=3000, long_short=True): """LogisticRegression walk-forward sulla forma. Label = segno del rendimento a H barre. Al tempo di decisione i si allena SOLO su campioni j con esito già realizzato (j+H <= i): strettamente causale, nessun leak. Rifit ogni `refit` barre (velocità). pos = +1 se P(up)>th, -1 se P(up)<1-th (long_short), altrimenti 0.""" c = df["close"].values.astype(float) n = len(c) X = _shape_features(df, W) fwd = np.full(n, np.nan) fwd[:n - H] = c[H:] / c[:n - H] - 1.0 y = (fwd > 0).astype(float) valid = ~np.isnan(X).any(axis=1) pos = np.zeros(n) model = scaler = None start = max(warmup, W + H + 200) for i in range(start, n): if model is None or (i - start) % refit == 0: # campioni di training: feature valide E label realizzata entro i (j+H <= i) tr = np.where(valid & (np.arange(n) + H <= i) & (np.arange(n) >= W))[0] tr = tr[tr < i - H] if len(tr) >= 500 and len(np.unique(y[tr])) == 2: scaler = StandardScaler().fit(X[tr]) model = LogisticRegression(max_iter=200, C=1.0).fit(scaler.transform(X[tr]), y[tr]) if model is not None and valid[i]: p_up = float(model.predict_proba(scaler.transform(X[i:i + 1]))[0, 1]) pos[i] = 1.0 if p_up > th else (-1.0 if (long_short and p_up < 1 - th) else 0.0) return pos # ----------------------------------- run ----------------------------------- def main(): TF = "1h" print("=" * 90) print(f" FASE 1 — triage superstiti su BTC/ETH {TF} | netto fee 0.10% RT | hold-out {HOLDOUT_START}+ BLOCCATO") print("=" * 90) data = {a: load_data(a, TF) for a in ("BTC", "ETH")} # ---------- DIP: griglia robustezza (plateau?) ---------- print("\n" + "#" * 90) print(" DIP reversion (long-only) — griglia FULL Sharpe (plateau = robusto, picco = overfit)") print("#" * 90) GRID = [(n, k) for n in (30, 50, 100) for k in (1.5, 2.0, 2.5)] for a in ("BTC", "ETH"): df = data[a] print(f"\n {a}: " + " ".join( f"n{n}k{k}→{backtest(df, dip_signal(df, n=n, k=k), TF).sharpe:>5.2f}" for n, k in GRID)) # report onesto sulla config centrale for a in ("BTC", "ETH"): report(f"DIP {a} (n50 k2.0)", data[a], dip_signal(data[a], n=50, k=2.0), TF) # ---------- SH01 shape-ML: config record + paio di varianti ---------- print("\n" + "#" * 90) print(" SH01 shape-ML (walk-forward LogReg) — long/short") print("#" * 90) for a in ("BTC", "ETH"): df = data[a] pos = shape_ml_signal(df, W=24, H=12, th=0.55, long_short=True) report(f"SH-ML {a} (W24 H12 th.55 L/S)", df, pos, TF) # variante long-only (meno fee) pos_lo = shape_ml_signal(df, W=24, H=12, th=0.55, long_short=False) report(f"SH-ML {a} (W24 H12 th.55 LONG-only)", df, pos_lo, TF) print("\n" + "=" * 90) print(" VERDETTO: un edge è REALE solo se FULL e OOS-VAL Sharpe > 0, regge il sweep fee,") print(" e BATTE il null (p<0.05). Altrimenti = rumore, si chiude.") print("=" * 90) if __name__ == "__main__": main()