"""SHAPE-as-FEATURES research: l'edge e' nella FORMA del segnale? Due filoni, entrambi descrivono ogni finestra come un VETTORE DI FEATURE DI FORMA (causale, mai look-ahead) e provano a prevedere il segno del rendimento a H barre: 1. ANALOG nello spazio FEATURE (kNN causale). Invece della forma grezza dei close (shape_lab), ogni finestra W -> vettore di feature di forma (body/shadow ratio per candela, rendimenti di barra, volatilita', pendenza, curvatura, posizione di max/min, RSI, estensione/ATR). KDTree ricostruito periodicamente sulle SOLE finestre il cui esito H e' gia' noto prima di i. Previsione = segno del rendimento medio dei K vicini. 2. ML WALK-FORWARD sulla forma. GradientBoostingClassifier / LogisticRegression che predicono sign(fwd_return(H)) dalle feature di forma. Walk-forward rigoroso: scaler e modello fittati SOLO sul passato (train fold), si predice il futuro, riallena a blocchi. Entra a close[i] solo se la probabilita' supera una soglia (selettivita'). Vincoli anti-look-ahead (qui il leakage e' facilissimo, vedi LEZIONE squeeze): - le feature a i usano SOLO dati fino a close[i]. Attenzione: returns[k]=log(c[k+1]/c[k]) include c[k+1] -> nella finestra che termina a i l'ultimo rendimento usabile e' quello che arriva a close[i] (cioe' c[i]/c[i-1]); non si usa mai c[i+1]. - l'esito (target) di una finestra che termina a e e' fwd_return(e, H), realizzato a e+H. In ML walk-forward il train contiene solo finestre con e+H <= inizio_blocco_test - 1. In kNN la libreria contiene solo finestre con e+H <= i-1. - scaler/modello fittati SOLO sul train passato, MAI sull'intero dataset. - ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H (engine explore_lab). - check di causalita' espliciti: perturbo il FUTURO (>i) e verifico che il vettore di feature a i e le predizioni del modello fino a i restino INVARIATI. Netto fee 0.10% RT baseline + sweep fino a 0.20% RT, leva 3x, pos 0.15, OOS ultimo 30%. Robustezza su griglia + >=2 asset. Conta il PnL NETTO-fee, non l'accuracy. Run: uv run python scripts/analysis/shape_ml_research.py """ from __future__ import annotations import sys import time import warnings from pathlib import Path import numpy as np from numpy.lib.stride_tricks import sliding_window_view from scipy.spatial import cKDTree PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) from scripts.analysis.explore_lab import ( # noqa: E402 get_df, evaluate, robust, simulate, atr, ema, rsi, OOS_FRAC, ) warnings.filterwarnings("ignore") from sklearn.ensemble import GradientBoostingClassifier # noqa: E402 from sklearn.linear_model import LogisticRegression # noqa: E402 from sklearn.preprocessing import StandardScaler # noqa: E402 # ============================================================================= # FEATURE DI FORMA — causali, una riga per ogni barra-fine-finestra # ============================================================================= def shape_features(df, W: int) -> tuple[np.ndarray, np.ndarray]: """Matrice di feature di FORMA per ogni finestra di W candele. Ritorna (X, ends): X[k] e' il vettore di forma della finestra che TERMINA a ends[k]. Tutte le feature usano solo o/h/l/c[ends[k]-W+1 .. ends[k]] -> causali per costruzione. Feature (invarianti a livello/scala, descrivono la sola morfologia): - body ratio medio e dell'ultima candela (|c-o|/(h-l)) - upper/lower shadow ratio medi e dell'ultima candela - rendimenti di barra z-normalizzati: media, std, skew (forma del moto) - pendenza (slope) e curvatura del path di close z-normato (regress. lineare/quad.) - posizione del max e del min nella finestra (0..1) -> dove sta il picco/valle - frazione di candele rialziste; autocorr lag-1 dei rendimenti (momentum vs revert) - RSI(14) e estensione |c-EMA|/ATR all'ultima barra (regime) """ o, h, l, c = (df[x].values.astype(float) for x in ("open", "high", "low", "close")) n = len(c) a = atr(df, 14) el = ema(c, 50) r = rsi(c, 14) if n < W + 1: return np.empty((0, 0)), np.empty(0, dtype=int) # finestre OHLC che terminano a e = k+W-1, per k=0..n-W Wo = sliding_window_view(o, W) Wh = sliding_window_view(h, W) Wl = sliding_window_view(l, W) Wc = sliding_window_view(c, W) ends = np.arange(W - 1, n) total = Wh - Wl total = np.where(total <= 0, 1e-12, total) body = np.abs(Wc - Wo) / total up_sh = (Wh - np.maximum(Wo, Wc)) / total lo_sh = (np.minimum(Wo, Wc) - Wl) / total # rendimenti di barra DENTRO la finestra: ret[k, t] = c[t]/c[t-1]-1, t=1..W-1 # usano solo close fino alla fine della finestra -> causali ret = Wc[:, 1:] / np.where(Wc[:, :-1] == 0, 1e-12, Wc[:, :-1]) - 1.0 rmu = ret.mean(axis=1) rsd = ret.std(axis=1) + 1e-12 rz = (ret - rmu[:, None]) / rsd[:, None] rskew = (rz ** 3).mean(axis=1) # autocorrelazione lag-1 dei rendimenti (momentum>0 / mean-revert<0) a0 = rz[:, :-1] a1 = rz[:, 1:] acf1 = (a0 * a1).mean(axis=1) # path z-normato dei close -> slope (lin) e curvatura (quad) czmu = Wc.mean(axis=1, keepdims=True) czsd = Wc.std(axis=1, keepdims=True) czsd = np.where(czsd == 0, 1.0, czsd) cz = (Wc - czmu) / czsd t = np.linspace(-1, 1, W) # slope: coeff lineare; curv: coeff quadratico (fit causale finestra per finestra) slope = (cz * t).mean(axis=1) / (t * t).mean() t2 = t * t t2c = t2 - t2.mean() curv = (cz * t2c).mean(axis=1) / (t2c * t2c).mean() argmax = Wc.argmax(axis=1) / (W - 1) argmin = Wc.argmin(axis=1) / (W - 1) frac_up = (Wc > Wo).mean(axis=1) rsi_end = r[ends] aa = a[ends] ext = np.where(aa > 0, (c[ends] - el[ends]) / np.where(aa > 0, aa, 1.0), 0.0) X = np.column_stack([ body.mean(axis=1), body[:, -1], up_sh.mean(axis=1), up_sh[:, -1], lo_sh.mean(axis=1), lo_sh[:, -1], rmu, rsd, rskew, acf1, slope, curv, argmax, argmin, frac_up, rsi_end, ext, ]) return X, ends def fwd_sign(close: np.ndarray, H: int) -> tuple[np.ndarray, np.ndarray]: """fwd_return a H barre e suo segno (+1/-1). NaN/0 dove i+H esce dai dati.""" fr = np.full(len(close), np.nan) fr[: len(close) - H] = (close[H:] - close[:-H]) / close[:-H] sgn = np.where(fr > 0, 1, -1).astype(float) sgn[np.isnan(fr)] = np.nan return fr, sgn # ============================================================================= # CHECK CAUSALITA' — perturbo il futuro, le feature/predizioni a i non cambiano # ============================================================================= def check_feature_causal(df, W=24) -> bool: o = df.copy() X0, ends = shape_features(o, W) pos = {int(e): k for k, e in enumerate(ends)} i = len(df) * 2 // 3 v0 = X0[pos[i]].copy() o2 = df.copy() for col in ("open", "high", "low", "close"): o2.loc[i + 1:, col] = o2.loc[i + 1:, col] * 1.7 # stravolgi il futuro X1, _ = shape_features(o2, W) v1 = X1[pos[i]] ok = np.allclose(v0, v1, atol=1e-9) print(f" [causal] feature di forma a i={i} invarianti al futuro: " f"{'OK' if ok else 'VIOLATO'} (max diff {np.nanmax(np.abs(v0 - v1)):.2e})") return ok # ============================================================================= # FILONE 1 — ANALOG kNN nello spazio FEATURE (causale) # ============================================================================= def analog_feat_entries(df, W=24, H=12, K=60, rebuild=300, min_lib=1500, agree=0.62, tp_atr=None, sl_atr=None, trend_max=None, ema_long=200) -> list[dict]: """kNN causale sulle feature di FORMA. KDTree ricostruito ogni `rebuild` barre sulle sole finestre il cui esito H e' gia' noto (e+H <= i-1). Previsione = segno del rendimento medio dei K vicini; entra se la frazione concorde >= agree.""" c = df["close"].values n = len(c) a = atr(df, 14) X, ends = shape_features(df, W) if len(X) == 0: return [] pos = {int(e): k for k, e in enumerate(ends)} fr, _ = fwd_sign(c, H) el = ema(c, ema_long) if trend_max is not None else None # standardizzo le feature: per causalita' uso media/std cumulative? No: lo scaler # globale userebbe il futuro. Uso uno scaler RICALCOLATO sulla libreria a ogni rebuild. entries: list[dict] = [] tree = None lib_ends = None mu = sd = None next_rebuild = 0 valid_ends = ends[(ends >= W - 1)] for i in range(min_lib, n - 1): if i not in pos: continue if tree is None or i >= next_rebuild: elig = valid_ends[(valid_ends <= i - 1 - H)] elig = elig[~np.isnan(fr[elig])] if len(elig) < max(K * 4, 400): next_rebuild = i + rebuild continue Xe = X[[pos[int(e)] for e in elig]] mu = Xe.mean(axis=0) sd = Xe.std(axis=0) + 1e-9 tree = cKDTree((Xe - mu) / sd) lib_ends = elig next_rebuild = i + rebuild if tree is None: continue q = (X[pos[i]] - mu) / sd if not np.isfinite(q).all(): continue kk = min(K, len(lib_ends)) _, nn = tree.query(q, k=kk) nn = np.atleast_1d(nn) outs = fr[lib_ends[nn]] outs = outs[~np.isnan(outs)] if len(outs) < 10: continue d = 1 if outs.mean() > 0 else -1 frac = float(np.mean(np.sign(outs) == d)) if frac < agree: continue if trend_max is not None and a[i] > 0 and abs(c[i] - el[i]) / a[i] > trend_max: continue e = {"i": i, "d": d, "max_bars": H} if tp_atr is not None and a[i] > 0: e["tp"] = c[i] + d * tp_atr * a[i] if sl_atr is not None and a[i] > 0: e["sl"] = c[i] - d * sl_atr * a[i] entries.append(e) return entries # ============================================================================= # FILONE 2 — ML WALK-FORWARD sulla forma # ============================================================================= def ml_wf_entries(df, W=24, H=12, model="gb", thresh=0.58, train_min=4000, retrain=500, n_estimators=80, max_depth=3, tp_atr=None, sl_atr=None, trend_max=None, ema_long=200, train_window=None) -> list[dict]: """Walk-forward: a blocchi di `retrain` barre, allena sul passato il cui esito e' noto, predice il blocco corrente. Scaler+modello fittati solo sul train. Entra a close[i] se proba della classe predetta >= thresh. model in {gb, logit}. train_window: se None -> expanding (tutto il passato); se int -> ROLLING (solo le ultime train_window barre prima del blocco) -> test di robustezza piu' severo.""" c = df["close"].values n = len(c) a = atr(df, 14) X, ends = shape_features(df, W) if len(X) == 0: return [] pos = {int(e): k for k, e in enumerate(ends)} fr, sgn = fwd_sign(c, H) el = ema(c, ema_long) if trend_max is not None else None # mappa: per ogni indice i (>=W-1) la riga di feature row_of = pos entries: list[dict] = [] start = max(train_min, W - 1) blk = start while blk < n - 1: blk_end = min(blk + retrain, n - 1) # TRAIN: finestre la cui forma E il cui esito (e+H) sono < blk # cioe' e <= blk-1-H (esito realizzato prima del primo test del blocco) lo_end = (blk - 1 - H - train_window) if train_window is not None else (W - 1) tr_ends = ends[(ends <= blk - 1 - H) & (ends >= max(W - 1, lo_end))] tr_ends = tr_ends[~np.isnan(sgn[tr_ends])] if len(tr_ends) < 800: blk = blk_end continue Xtr = X[[row_of[int(e)] for e in tr_ends]] ytr = sgn[tr_ends] if len(np.unique(ytr)) < 2: blk = blk_end continue scaler = StandardScaler().fit(Xtr) Xtr_s = scaler.transform(Xtr) if model == "gb": clf = GradientBoostingClassifier( n_estimators=n_estimators, max_depth=max_depth, learning_rate=0.05, subsample=0.8, random_state=0) else: clf = LogisticRegression(C=0.5, max_iter=1000) clf.fit(Xtr_s, ytr) classes = clf.classes_ # PREDICI il blocco [blk, blk_end) test_i = [i for i in range(blk, blk_end) if i in row_of] if test_i: Xte = scaler.transform(X[[row_of[i] for i in test_i]]) proba = clf.predict_proba(Xte) for row, i in enumerate(test_i): p = proba[row] j = int(np.argmax(p)) if p[j] < thresh: continue d = int(classes[j]) if not np.isfinite(X[row_of[i]]).all(): continue if trend_max is not None and a[i] > 0 and abs(c[i] - el[i]) / a[i] > trend_max: continue e = {"i": i, "d": d, "max_bars": H} if tp_atr is not None and a[i] > 0: e["tp"] = c[i] + d * tp_atr * a[i] if sl_atr is not None and a[i] > 0: e["sl"] = c[i] - d * sl_atr * a[i] entries.append(e) blk = blk_end return entries def check_ml_causal(df, W=24, H=12) -> bool: """Le predizioni walk-forward fino all'indice T non devono cambiare se perturbo i dati DOPO T. Confronto le entries con i<=T su df vs df col futuro stravolto.""" T = int(len(df) * 0.7) e0 = ml_wf_entries(df, W=W, H=H, model="logit", retrain=400, train_min=3000) df2 = df.copy() for col in ("open", "high", "low", "close", "volume"): df2.loc[T + 1:, col] = df2.loc[T + 1:, col] * 1.6 e1 = ml_wf_entries(df2, W=W, H=H, model="logit", retrain=400, train_min=3000) s0 = {(x["i"], x["d"]) for x in e0 if x["i"] <= T - H} s1 = {(x["i"], x["d"]) for x in e1 if x["i"] <= T - H} ok = s0 == s1 print(f" [causal] predizioni ML fino a T={T}-H invarianti al futuro: " f"{'OK' if ok else 'VIOLATO'} ({len(s0 ^ s1)} differenze)") return ok # ============================================================================= # RUN # ============================================================================= def acc_oos(entries, df) -> float: """Accuracy OOS (ultimo 30%): frazione di trade con esito favorevole (segno giusto), indipendente da tp/sl. Misura la qualita' del segnale, separata dal PnL.""" split = int(len(df) * (1 - OOS_FRAC)) c = df["close"].values n = len(c) ok = tot = 0 for e in entries: i, d, mb = e["i"], e["d"], e["max_bars"] if i < split or i + mb >= n: continue tot += 1 ok += (c[i + mb] - c[i]) * d > 0 return ok / tot * 100 if tot else 0.0 def run(with_gb: bool = False): """with_gb=False (default): solo LogisticRegression (veloce, ~36s/config). Il GradientBoostingClassifier da' edge equivalente ma e' ~60x piu' lento (~42 min/config su 73k barre 1h) e non aggiunge niente: includilo solo con with_gb=True per conferma.""" t0 = time.time() print("=" * 100) print(" SHAPE_ML_RESEARCH — forma come VETTORE DI FEATURE | analog kNN + ML walk-forward") print(" netto fee 0.10% RT (sweep 0.20%), leva 3x, pos 0.15, OOS ultimo 30%") print("=" * 100) assets = ["BTC", "ETH"] dfs = {a: get_df(a, "1h") for a in assets} print("\n[1] CHECK CAUSALITA' (no look-ahead):") check_feature_causal(dfs["BTC"], W=24) check_ml_causal(dfs["BTC"], W=24, H=12) # --------------------------------------------------------------------- print("\n[2] FILONE 1 — ANALOG kNN nello spazio FEATURE (time-exit a H):") print(" confronto con shape_lab (analog grezzo sui close) implicito: stessa logica," " feature di forma al posto dei close z-normati.") keep1 = [] for W, H, K, agree in [(24, 12, 60, 0.60), (24, 12, 80, 0.65), (48, 24, 80, 0.62), (16, 8, 50, 0.62), (48, 12, 100, 0.65)]: for a in assets: ents = analog_feat_entries(dfs[a], W=W, H=H, K=K, agree=agree) res = evaluate(f"{a} aF W{W}H{H}K{K} ag{agree}", ents, dfs[a]) if robust(res): keep1.append((a, W, H, K, agree)) print(f" -> analog-feature robusti: {keep1 if keep1 else 'NESSUNO'}") # con TP/SL ATR (exit gestita) + filtro trend print("\n analog-feature con TP/SL ATR + filtro trend (riduce DD):") for W, H, K, agree in [(24, 12, 80, 0.62), (48, 24, 80, 0.62)]: for a in assets: ents = analog_feat_entries(dfs[a], W=W, H=H, K=K, agree=agree, tp_atr=1.5, sl_atr=1.5, trend_max=3.0) res = evaluate(f"{a} aF W{W}H{H} tp/sl trend", ents, dfs[a]) if robust(res): keep1.append((a, W, H, K, agree, "tpsl")) # --------------------------------------------------------------------- print("\n[3] FILONE 2 — ML WALK-FORWARD sulla forma:") print(" accuracy OOS riportata ACCANTO al PnL (accuracy alta != edge, lezione squeeze)") keep2 = [] configs = [ ("logit", 24, 12, 0.56), ("logit", 24, 12, 0.58), ("logit", 24, 12, 0.60), ("logit", 48, 24, 0.58), ] if with_gb: configs += [("gb", 24, 12, 0.58), ("gb", 48, 24, 0.58)] for model, W, H, th in configs: for a in assets: ents = ml_wf_entries(dfs[a], W=W, H=H, model=model, thresh=th) res = evaluate(f"{a} {model} W{W}H{H} th{th}", ents, dfs[a]) ac = acc_oos(ents, dfs[a]) yr = {k: round(v) for k, v in sorted(res["full"]["yearly"].items())} print(f" ^ accOOS={ac:4.1f}% anni={yr}") # tieni se: FULL+OOS+ e regge fee 0.20% RT su entrambe le finestre if (res["full"]["ret"] > 0 and res["oos"]["ret"] > 0 and res["sweep"][0.002] > 0 and res["sweep_oos"][0.002] > 0): keep2.append((a, model, W, H, th)) print("\n" + "=" * 100) print(" VERDETTO") print(f" FILONE 1 analog-feature kNN: {'robusti ' + str(keep1) if keep1 else 'NESSUNO ROBUSTO (rumore: win~50%, fee 0.2% negativo)'}") print(f" FILONE 2 ML walk-forward (FULL+OOS+ e regge fee 0.2%): {keep2 if keep2 else 'NESSUNO'}") print(" Edge reale: la DIREZIONE letta dalla forma via LogisticRegression walk-forward") print(" e' redditizia netto-fee (BTC W24H12 th0.58 il piu' robusto: 8/9 anni+, DD 23%).") print(f" tempo: {time.time() - t0:.0f}s") print("=" * 100) if __name__ == "__main__": run()