From 4ac87ab385d8fdb40783ac591b0c452b61e7e272 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Fri, 29 May 2026 12:09:28 +0200 Subject: [PATCH] research(shape): 5 famiglie di pattern-forma su harness onesto Harness shape_lab (analog kNN causale, no look-ahead verificato) + 5 ricerche parallele. 4/5 famiglie = RUMORE (confermano dominanza mean-reversion): - analog kNN forma grezza: solo BTC-overfit, non robusto >=2 asset - encoding candele UP/DOWN/DOJI + body/shadow: hit-rate ~50%, muore a fee - DTW + template geometrici: DTW peggiora euclidea; template overfit - PIP/pivot/zig-zag: 0/48 config robuste 1/5 = EDGE REALE: ML walk-forward (LogisticRegression) sulle feature di forma. BTC logit W24H12 th0.58: FULL +219% / OOS +42% / Sharpe 2.72 / 8-9 anni+ / regge fee 0.20% RT (+60/+26). Causalita' verificata. Da validare a fondo. Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/analysis/shape_analog_research.py | 177 ++++++++ scripts/analysis/shape_candle_research.py | 328 +++++++++++++++ scripts/analysis/shape_lab.py | 257 ++++++++++++ scripts/analysis/shape_ml_research.py | 427 +++++++++++++++++++ scripts/analysis/shape_pivot_research.py | 333 +++++++++++++++ scripts/analysis/shape_template_research.py | 443 ++++++++++++++++++++ 6 files changed, 1965 insertions(+) create mode 100644 scripts/analysis/shape_analog_research.py create mode 100644 scripts/analysis/shape_candle_research.py create mode 100644 scripts/analysis/shape_lab.py create mode 100644 scripts/analysis/shape_ml_research.py create mode 100644 scripts/analysis/shape_pivot_research.py create mode 100644 scripts/analysis/shape_template_research.py diff --git a/scripts/analysis/shape_analog_research.py b/scripts/analysis/shape_analog_research.py new file mode 100644 index 0000000..d6ab38c --- /dev/null +++ b/scripts/analysis/shape_analog_research.py @@ -0,0 +1,177 @@ +"""Ricerca sistematica edge nella FORMA (analog forecasting / kNN) — netto fee, OOS. + +Obiettivo: trovare una config di analog forecasting ROBUSTA, cioe' positiva +FULL+OOS, che regge fee 0.20% RT e ha quasi tutti gli anni positivi, su >=2 asset. +Si combatte la "morte per fee" della baseline (BTC1h W24H12K50 agree0.60: +FULL +112%/OOS +48% Sharpe 1.38 ma a 0.2% RT -> FULL -72 / OOS -18, win 49.5%, +esposizione 73.9%, 4531 trade) con SELETTIVITA': + - agree alto (0.70..0.90) -> entra solo con analoghi molto concordi + - conf_atr > 0 -> richiede |rendimento medio analoghi| >= conf_atr*ATR + - trend_max/ema_long -> salta forme in trend estremo + - tp_atr/sl_atr -> exit intrabar invece che solo a tempo + +Tutto causale: la forma usa solo close<=i, la libreria analoghi termina < i-H. +Per performance, il forecast kNN grezzo per barra si calcola UNA volta per +(W,H,K,rebuild) con analog_signals(); i filtri (agree/conf/trend/tp/sl) sono +applicati a valle con entries_from_signals() (cheap, risultato identico ad +analog_entries — verificato). Engine netto-fee + OOS da explore_lab. + +Uso: + uv run python scripts/analysis/shape_analog_research.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)) + +from scripts.analysis.shape_lab import ( # noqa: E402 + analog_signals, entries_from_signals, check_no_lookahead, +) +from scripts.analysis.explore_lab import get_df, evaluate, robust # noqa: E402 + +ROBUSTE: list[tuple] = [] +MIN_TRADES = 100 # un edge "robusto" su <100 trade e' rumore campionario, non edge + + +def _hdr(s: str) -> None: + print("\n" + "=" * 100, flush=True) + print(" " + s, flush=True) + print("=" * 100, flush=True) + + +def _eval(df, sig, asset, tf, tag, **filt): + ents = entries_from_signals(df, sig, **filt) + res = evaluate(f"[{asset} {tf}] {tag}", ents, df) + # robusto E con campione sufficiente (un edge su <100 trade non e' affidabile) + if robust(res) and res["full"]["trades"] >= MIN_TRADES: + print(f" ^^^ ROBUSTA ({asset} {tf}): {tag} filt={filt}", flush=True) + ROBUSTE.append((asset, tf, tag, dict(filt), res)) + elif robust(res): + print(f" (robust ma trade={res['full']['trades']}<{MIN_TRADES}: campione " + f"insufficiente, ignorato)", flush=True) + return res + + +def run(): + # --- 0) sanity no-lookahead --------------------------------------------- + _hdr("0) SANITY no-lookahead (forma causale)") + df_btc = get_df("BTC", "1h") + check_no_lookahead(df_btc, W=24, H=12) + + # sig base W24H12K50 (riusato per selettivita' agree/conf/tp/sl/trend) + sig0 = analog_signals(df_btc, W=24, H=12, K=50, rebuild=250) + + # --- 1) selettivita' via agree ------------------------------------------ + _hdr("1) BTC 1h — selettivita' agree (W24 H12 K50, time-exit)") + for ag in (0.60, 0.70, 0.80, 0.90): + _eval(df_btc, sig0, "BTC", "1h", f"agree{ag}", agree=ag) + + # --- 2) conf_atr (forza segnale) ---------------------------------------- + _hdr("2) BTC 1h — conf_atr (W24 H12 K50 agree0.70)") + for ca in (0.0, 0.25, 0.5, 1.0, 1.5): + _eval(df_btc, sig0, "BTC", "1h", f"ag0.70 conf{ca}", agree=0.70, conf_atr=ca) + + # --- 3) tp/sl intrabar --------------------------------------------------- + _hdr("3) BTC 1h — exit intrabar tp/sl (W24 H12 K50 agree0.70 conf0.5)") + for tp, sl in [(1.0, 1.0), (1.5, 1.0), (2.0, 1.5), (1.5, 2.0), (3.0, 2.0)]: + _eval(df_btc, sig0, "BTC", "1h", f"tp{tp}sl{sl}", + agree=0.70, conf_atr=0.5, tp_atr=tp, sl_atr=sl) + + # --- 4) filtro trend ----------------------------------------------------- + _hdr("4) BTC 1h — filtro trend_max (W24 H12 K50 agree0.70 conf0.5)") + for tm in (None, 2.0, 3.0, 4.0): + _eval(df_btc, sig0, "BTC", "1h", f"trend_max{tm}", + agree=0.70, conf_atr=0.5, trend_max=tm, ema_long=200) + + # --- 5) griglia W/H/K (agree0.80, time-exit) plateau --------------------- + # Griglia focalizzata: con agree0.80 e H>=24 i trade -> ~0 (vedi sez.1), e W>=24 + # porta OOS negativo; il segnale vive su W piccolo, H breve. Testo il plateau + # attorno a quella regione + una banda di controllo (W24/48) per confermare il bordo. + _hdr("5) BTC 1h — griglia W/H/K (agree0.80, time-exit) — plateau check") + for W in (12, 24, 48): + for H in (6, 12, 24): + for K in (30, 50, 80): + sig = analog_signals(df_btc, W=W, H=H, K=K, rebuild=250) + _eval(df_btc, sig, "BTC", "1h", f"W{W}H{H}K{K}", agree=0.80) + + # --- 6) rebuild sensitivity --------------------------------------------- + _hdr("6) BTC 1h — rebuild 250 vs 500 (W24 H12 K80 agree0.80)") + for rb in (250, 500): + sig = analog_signals(df_btc, W=24, H=12, K=80, rebuild=rb) + _eval(df_btc, sig, "BTC", "1h", f"rebuild{rb}", agree=0.80) + + # --- 7) cross-asset 1h: candidati selettivi ----------------------------- + _hdr("7) cross-asset 1h — candidati selettivi (>=2 robusti richiesto)") + # (build_kw: per analog_signals) (filt: per entries_from_signals) + # Su BTC 1h le uniche regioni con OOS positivo che regge fee0.2% sono W piccolo, + # H breve, K basso (W12H12K30: FULL+88/OOS+36, fee0.2% +69/+32, 243 trade, 8/9 anni; + # W12H6K30: +35/+11, fee0.2% +20/+7). conf0.25 con W24H12 e' il miglior in-sample + # ma OOS@fee~0. Verifico questi candidati cross-asset (>=2 robusti richiesto). + candidates = [ + ("C1 W12H12K30 ag.80", dict(W=12, H=12, K=30), dict(agree=0.80)), + ("C2 W12H6K30 ag.80", dict(W=12, H=6, K=30), dict(agree=0.80)), + ("C3 W12H12K30 ag.70", dict(W=12, H=12, K=30), dict(agree=0.70)), + ("C4 W24H12K50 ag.70 conf.25", dict(W=24, H=12, K=50), dict(agree=0.70, conf_atr=0.25)), + ("C5 W12H12K30 ag.80 trend3", dict(W=12, H=12, K=30), dict(agree=0.80, trend_max=3.0, ema_long=200)), + ("C6 W12H6K50 ag.70", dict(W=12, H=6, K=50), dict(agree=0.70)), + ] + per_cand: dict[str, int] = {} + for asset in ("BTC", "ETH", "ADA", "LTC", "SOL", "XRP"): + try: + df = get_df(asset, "1h") + except Exception as ex: + print(f" [{asset} 1h] SKIP load: {ex}", flush=True) + continue + # cache analog_signals per ogni build_kw distinto su questo asset + sig_cache: dict[tuple, dict] = {} + for tag, bkw, filt in candidates: + key = tuple(sorted(bkw.items())) + if key not in sig_cache: + sig_cache[key] = analog_signals(df, rebuild=250, **bkw) + res = _eval(df, sig_cache[key], asset, "1h", tag, **filt) + if robust(res): + per_cand[tag] = per_cand.get(tag, 0) + 1 + + # --- 8) verifica 15m dei candidati robusti su >=2 asset 1h -------------- + _hdr("8) verifica 15m dei candidati robusti su >=2 asset 1h") + good = [t for t, c in per_cand.items() if c >= 2] + if not good: + print(" Nessun candidato robusto su >=2 asset 1h -> niente verifica 15m.", flush=True) + else: + for tag in good: + _, bkw, filt = next(c for c in candidates if c[0] == tag) + for asset in ("BTC", "ETH"): + try: + df = get_df(asset, "15m") + except Exception as ex: + print(f" [{asset} 15m] SKIP load: {ex}", flush=True) + continue + sig = analog_signals(df, rebuild=250, **bkw) + _eval(df, sig, asset, "15m", f"{tag} (15m)", **filt) + + # --- VERDETTO ------------------------------------------------------------ + _hdr("VERDETTO") + if ROBUSTE: + agg: dict[str, list] = {} + for asset, tf, tag, filt, res in ROBUSTE: + agg.setdefault(tag, []).append(f"{asset}/{tf}") + print(f" {len(ROBUSTE)} sleeve robusti (FULL+OOS+ fee0.2% + anniPos):", flush=True) + edge = False + for tag, asl in agg.items(): + n_assets = len({a.split('/')[0] for a in asl}) + mark = " *** EDGE (>=2 asset)" if n_assets >= 2 else " (1 asset: non sufficiente)" + if n_assets >= 2: + edge = True + print(f" - {tag}: {asl}{mark}", flush=True) + if not edge: + print("\n CONCLUSIONE: nessuna config robusta su >=2 asset -> RUMORE.", flush=True) + else: + print(" NESSUNA config robusta. Famiglia analog/forma = RUMORE sotto fee reali.", flush=True) + return ROBUSTE + + +if __name__ == "__main__": + run() diff --git a/scripts/analysis/shape_candle_research.py b/scripts/analysis/shape_candle_research.py new file mode 100644 index 0000000..3ceba8f --- /dev/null +++ b/scripts/analysis/shape_candle_research.py @@ -0,0 +1,328 @@ +"""Edge nella FORMA discreta delle candele -> distribuzione condizionale dell'esito. + +Famiglia: encoding DISCRETO della morfologia di una finestra di L candele +(sequenza UP/DOWN/DOJI, opzionalmente arricchita con bucket di body-ratio e +shadow-ratio) -> codice intero. Per ogni codice si stima la distribuzione del +rendimento a H barre usando SOLO le occorrenze PASSATE il cui esito era gia' +realizzato prima della barra di decisione i (expanding window causale). Se un +codice mostra bias direzionale forte e statisticamente solido (n campioni >= +soglia, win-rate o |media| oltre soglia) si ENTRA a close[i] nella direzione del +bias; exit a H barre o TP/SL ATR. + +VINCOLI ANTI-LOOK-AHEAD (l'errore squeeze e' nato qui): + - il codice a i usa SOLO open/high/low/close fino alla barra i inclusa; + - la statistica condizionale a i conta SOLO occorrenze del codice terminate in + e con e+H <= i-1 -> il loro esito H e' interamente noto PRIMA di i; + - direzione decisa dal CODICE (forma fino a close[i]) + STATISTICHE PASSATE, + ingresso eseguibile a close[i]. + - check_no_lookahead() perturba il futuro: ne' il codice a i ne' le stat usate + devono cambiare. + +Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust). +Implementazione causale O(N) per codice via accumulatori incrementali (niente +ricalcolo dell'intera storia ad ogni barra). + +Asset: ADA BNB BTC DOGE ETH LTC SOL XRP (1h, 15m). Default 1h. +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np + +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, OOS_FRAC, +) + + +# --------------------------- encoding discreto della forma --------------------------- +def candle_codes(df, L: int, body_buckets: int = 1, shadow_buckets: int = 1) -> np.ndarray: + """Codice intero della forma per la finestra di L candele terminante in i. + + Componenti per ogni candela: + - direzione UP/DOWN/DOJI (sempre): 3 stati. + - bucket del body-ratio |c-o|/(h-l) (se body_buckets>1): quantizzazione fissa + in body_buckets livelli (corpo piccolo/medio/grande...). + - bucket dello shadow-ratio (h-max(o,c)-(min(o,c)-l))/(h-l) in [-1,1] + (se shadow_buckets>1): ombra sup vs inf. + + Quantizzazione a SOGLIE FISSE (non quantili): non dipende dal futuro ne' dal + dataset globale -> causale per costruzione. codes[i] dipende solo da + barre [i-L+1 .. i]. Per i < L-1 -> -1 (non valido). + """ + o = df["open"].values; c = df["close"].values + h = df["high"].values; l = df["low"].values + n = len(c) + rng = np.where((h - l) == 0, 1e-12, h - l) + + body = np.abs(c - o) / rng # [0,1] + direction = np.where(body < 0.1, 0, # DOJI + np.where(c > o, 1, 2)) # UP=1, DOWN=2 (3 stati: 0,1,2) + # shadow asymmetry in [-1,1]: >0 ombra sup dominante, <0 ombra inf + up_sh = (h - np.maximum(o, c)) / rng + lo_sh = (np.minimum(o, c) - l) / rng + shadow = up_sh - lo_sh + + # bucket body (soglie fisse su frazioni del range): 0..body_buckets-1 + if body_buckets > 1: + edges_b = np.linspace(0.0, 1.0, body_buckets + 1)[1:-1] + bbk = np.digitize(body, edges_b) # 0..body_buckets-1 + else: + bbk = np.zeros(n, dtype=int) + if shadow_buckets > 1: + edges_s = np.linspace(-1.0, 1.0, shadow_buckets + 1)[1:-1] + sbk = np.digitize(shadow, edges_s) + else: + sbk = np.zeros(n, dtype=int) + + # simbolo per candela: dir * (body_buckets*shadow_buckets) + bbk*shadow_buckets + sbk + nbb, nsb = body_buckets, shadow_buckets + per_dir = nbb * nsb + sym = direction * per_dir + bbk * nsb + sbk # 0 .. 3*per_dir-1 + base = 3 * per_dir + + codes = np.full(n, -1, dtype=np.int64) + # codice della finestra L: base-L polinomiale sui simboli [i-L+1 .. i] + acc = np.zeros(n, dtype=np.int64) + for k in range(L): + # contributo della candela a posizione (i-L+1+k): peso base**(L-1-k) + shifted = np.full(n, 0, dtype=np.int64) + shifted[L - 1 - k:] = sym[: n - (L - 1 - k)] if (L - 1 - k) > 0 else sym + acc += shifted * (base ** (L - 1 - k)) + codes[L - 1:] = acc[L - 1:] + return codes + + +def fwd_return(close: np.ndarray, H: int) -> np.ndarray: + out = np.full(len(close), np.nan) + out[: len(close) - H] = (close[H:] - close[:-H]) / close[:-H] + return out + + +# --------------------------- stima condizionale causale --------------------------- +def shape_entries(df, L=3, H=12, body_buckets=1, shadow_buckets=1, + min_n=30, edge=0.55, min_lib=500, + tp_atr=None, sl_atr=None, fade=False) -> list[dict]: + """Entries dal bias condizionale del codice di forma (causale, no look-ahead). + + L: lunghezza finestra-forma. H: orizzonte = max_bars. + body_buckets/shadow_buckets: granularita' dell'encoding (1 = solo direzione). + min_n: occorrenze passate minime del codice (con esito noto) per fidarsi. + edge: win-rate minimo (frazione di esiti concordi col segno della media) per + entrare; |edge-0.5| e' il margine direzionale. + min_lib: barre minime di storia prima di iniziare a operare. + tp_atr/sl_atr: TP/SL in multipli di ATR (None = solo time-limit H). + + Causalita': a barra di decisione i si aggiorna lo stato del codice della + finestra terminata in e = i-1-H (il cui esito fr[e] e' ora noto). Le statistiche + usate per decidere a i derivano quindi solo da occorrenze con e+H <= i-1. + """ + close = df["close"].values + n = len(close) + a = atr(df, 14) + codes = candle_codes(df, L, body_buckets, shadow_buckets) + fr = fwd_return(close, H) + + # accumulatori per codice: somma rendimenti, n positivi, n totali + from collections import defaultdict + cnt = defaultdict(int) + pos = defaultdict(int) + ssum = defaultdict(float) + + entries: list[dict] = [] + for i in range(min_lib, n - 1): + # aggiorna lo stato col codice la cui finestra termina in e = i-1-H + e = i - 1 - H + if e >= L - 1: + ce = codes[e] + re = fr[e] + if ce >= 0 and not np.isnan(re): + cnt[ce] += 1 + pos[ce] += 1 if re > 0 else 0 + ssum[ce] += re + + ci = codes[i] + if ci < 0: + continue + ntot = cnt.get(ci, 0) + if ntot < min_n: + continue + mean = ssum[ci] / ntot + wr_up = pos[ci] / ntot # frazione esiti positivi nel passato + d = 1 if mean > 0 else -1 + # win-rate nella direzione scelta + wr = wr_up if d == 1 else (1.0 - wr_up) + if wr < edge: + continue + if fade: + d = -d # FADE: entra contro il bias storico + ent = {"i": i, "d": d, "max_bars": H} + if tp_atr is not None and a[i] > 0: + ent["tp"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + ent["sl"] = close[i] - d * sl_atr * a[i] + entries.append(ent) + return entries + + +# --------------------------- verifica no look-ahead --------------------------- +def check_no_lookahead(df, L=3, H=12, body_buckets=2, shadow_buckets=2) -> bool: + """Perturbare il FUTURO (>i) non cambia (a) il codice a i, (b) le stat usate a i. + + (a) codes[i] dipende solo da barre <= i. + (b) le entries fino a i (incluse) non cambiano se stravolgo le barre > i+1. + """ + n = len(df) + i = n // 2 + codes0 = candle_codes(df, L, body_buckets, shadow_buckets) + df2 = df.copy() + fut = slice(i + 1, n) + for col in ("open", "high", "low", "close"): + df2.loc[df2.index[i + 1:], col] = df2[col].values[i + 1:] * 1.5 + codes1 = candle_codes(df2, L, body_buckets, shadow_buckets) + ok_code = bool(codes0[i] == codes1[i] and np.array_equal(codes0[: i + 1], codes1[: i + 1])) + + # entries: confronta quelle con indice <= i-1-H (decise con stat tutte note prima del futuro) + e0 = shape_entries(df, L=L, H=H, body_buckets=body_buckets, shadow_buckets=shadow_buckets, + min_n=5, edge=0.50, min_lib=200) + e1 = shape_entries(df2, L=L, H=H, body_buckets=body_buckets, shadow_buckets=shadow_buckets, + min_n=5, edge=0.50, min_lib=200) + cutoff = i - 1 - H + s0 = {(x["i"], x["d"]) for x in e0 if x["i"] <= cutoff} + s1 = {(x["i"], x["d"]) for x in e1 if x["i"] <= cutoff} + ok_ent = (s0 == s1) + print(f" no-lookahead codice a i={i}: {'OK' if ok_code else 'VIOLATO'}; " + f"entries<=i-1-H invarianti: {'OK' if ok_ent else 'VIOLATO'} " + f"({len(s0)} vs {len(s1)})") + return ok_code and ok_ent + + +# --------------------------- run riproducibile --------------------------- +def predictive_power(df, L=3, H=12, body_buckets=1, shadow_buckets=1, min_n=30, min_lib=500): + """Diagnostica ONESTA: la direzione predetta dal bias storico (causale) anticipa + il segno del rendimento realizzato? Misura hit-rate aggregato della predizione + (segno media passata del codice) vs realizzato, su tutte le barre operabili. + Niente fee: pura capacita' predittiva del codice di forma.""" + close = df["close"].values + n = len(close) + codes = candle_codes(df, L, body_buckets, shadow_buckets) + fr = fwd_return(close, H) + from collections import defaultdict + cnt = defaultdict(int); pos = defaultdict(int); ssum = defaultdict(float) + hits = tot = 0 + pred_ret = 0.0 + for i in range(min_lib, n - 1): + e = i - 1 - H + if e >= L - 1: + ce = codes[e]; re = fr[e] + if ce >= 0 and not np.isnan(re): + cnt[ce] += 1; pos[ce] += 1 if re > 0 else 0; ssum[ce] += re + ci = codes[i] + if ci < 0 or cnt.get(ci, 0) < min_n or np.isnan(fr[i]): + continue + d = 1 if ssum[ci] / cnt[ci] > 0 else -1 + actual = fr[i] + hits += (np.sign(actual) == d); tot += 1 + pred_ret += d * actual # PnL teorico senza fee + hr = hits / tot * 100 if tot else 0.0 + print(f" predittivita' L{L}H{H} b{body_buckets}s{shadow_buckets}: " + f"hit={hr:.2f}% su {tot} (50%=rumore) | PnL_grezzo_noFee={pred_ret*100:+.0f}%") + return hr + + +def run(): + print("=" * 100) + print(" SHAPE_CANDLE_RESEARCH — encoding discreto forma -> bias condizionale | netto fee, OOS") + print("=" * 100) + + df_btc = get_df("BTC", "1h") + print("\n[causalita']") + check_no_lookahead(df_btc, L=3, H=12, body_buckets=2, shadow_buckets=2) + + # ----- sweep base BTC/ETH 1h: solo direzione (body=shadow=1) ----- + print("\n[BTC/ETH 1h] solo direzione UP/DOWN/DOJI (body=1,shadow=1), time-exit a H:") + for asset in ("BTC", "ETH"): + df = get_df(asset, "1h") + print(f" -- {asset} 1h --") + for L in (2, 3, 4): + for H in (6, 12, 24): + ents = shape_entries(df, L=L, H=H, min_n=30, edge=0.55) + evaluate(f"dir L{L}H{H}", ents, df) + + # ----- encoding arricchito body+shadow ----- + print("\n[BTC/ETH 1h] encoding arricchito (body=2,shadow=2), time-exit a H:") + for asset in ("BTC", "ETH"): + df = get_df(asset, "1h") + print(f" -- {asset} 1h --") + for L in (2, 3): + for H in (6, 12, 24): + ents = shape_entries(df, L=L, H=H, body_buckets=2, shadow_buckets=2, + min_n=30, edge=0.55) + evaluate(f"rich L{L}H{H} b2s2", ents, df) + + # ----- selettivita': soglie edge piu' alte, meno trade ----- + print("\n[BTC/ETH 1h] selettivo (edge>=0.58, min_n>=50), dir-only:") + for asset in ("BTC", "ETH"): + df = get_df(asset, "1h") + print(f" -- {asset} 1h --") + for L in (3, 4, 5): + for H in (12, 24): + ents = shape_entries(df, L=L, H=H, min_n=50, edge=0.58) + evaluate(f"sel L{L}H{H} e58", ents, df) + + # ----- con TP/SL ATR (gestione rischio) sui candidati piu' attivi ----- + print("\n[BTC/ETH 1h] con TP/SL ATR (tp=2,sl=1.5), dir-only L3:") + for asset in ("BTC", "ETH"): + df = get_df(asset, "1h") + for H in (12, 24): + ents = shape_entries(df, L=3, H=H, min_n=30, edge=0.55, tp_atr=2.0, sl_atr=1.5) + evaluate(f"{asset} tpsl L3H{H}", ents, df) + + # ----- DIAGNOSTICA: il codice di forma ha QUALSIASI potere predittivo? ----- + print("\n[DIAGNOSTICA] hit-rate predizione (segno bias storico vs realizzato), senza fee:") + for asset in ("BTC", "ETH"): + df = get_df(asset, "1h") + print(f" -- {asset} 1h --") + for L in (2, 3, 4): + for H in (6, 12, 24): + predictive_power(df, L=L, H=H, min_n=30) + for L in (2, 3): + predictive_power(df, L=L, H=12, body_buckets=2, shadow_buckets=2, min_n=30) + + # ----- IPOTESI FADE: il bias e' anti-predittivo (mean-reversion)? ----- + print("\n[FADE del bias] entra CONTRO il bias storico (dir-only):") + for asset in ("BTC", "ETH"): + df = get_df(asset, "1h") + print(f" -- {asset} 1h --") + for L in (2, 3): + for H in (6, 12, 24): + ents = shape_entries(df, L=L, H=H, min_n=30, edge=0.55, fade=True) + evaluate(f"FADE L{L}H{H}", ents, df) + + +def run_extended(configs): + """Valuta config candidate su tutti gli asset 1h+15m e stampa robustezza.""" + assets = ["ADA", "BNB", "BTC", "DOGE", "ETH", "LTC", "SOL", "XRP"] + for cfg in configs: + print(f"\n[ESTESO] config {cfg}") + for tf in ("1h", "15m"): + print(f" -- timeframe {tf} --") + nrob = 0 + for asset in assets: + try: + df = get_df(asset, tf) + except Exception as ex: + print(f" {asset}: skip ({ex})") + continue + ents = shape_entries(df, **cfg) + res = evaluate(f"{asset} {tf}", ents, df) + nrob += robust(res) + print(f" -> robuste {nrob}/{len(assets)} su {tf}") + + +if __name__ == "__main__": + run() diff --git a/scripts/analysis/shape_lab.py b/scripts/analysis/shape_lab.py new file mode 100644 index 0000000..77311bc --- /dev/null +++ b/scripts/analysis/shape_lab.py @@ -0,0 +1,257 @@ +"""Harness ONESTO per pattern *di forma* -> previsione dell'andamento successivo. + +Idea (analog forecasting / nearest-neighbour sulla FORMA del prezzo): + - a ogni barra i guardo la forma recente W (closes z-normalizzati fino a close[i]); + - cerco nel PASSATO le K finestre piu' simili la cui forma si era gia' conclusa + *e* il cui esito a H barre era gia' noto PRIMA di i (nessun look-ahead); + - prevedo la direzione dei prossimi H barre = segno del rendimento medio degli + analoghi; entro a close[i] se l'accordo fra analoghi e' abbastanza forte. + +Vincoli anti-look-ahead (gli stessi della famiglia squeeze fallita): + - la forma usa SOLO closes fino a close[i]; + - la libreria di analoghi a decisione i contiene solo finestre che terminano in + e con e+H <= i-1 -> il loro esito e' interamente realizzato *prima* della barra i; + - ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H. + +Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust). +KDTree ricostruito ogni `rebuild` barre (causale): query O(log N), niente O(N^2). + +Asset: ADA BNB BTC DOGE ETH LTC SOL XRP (1h, 15m; BTC/ETH anche 5m). +""" +from __future__ import annotations + +import sys +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, _dt, OOS_FRAC, ASSETS, FEE_RT, +) + + +# --------------------------- forma normalizzata --------------------------- +def znorm_windows(close: np.ndarray, W: int) -> tuple[np.ndarray, np.ndarray]: + """Matrice delle finestre z-normalizzate per FORMA. + + Ritorna (M, ends) dove M[k] = z-norm(close[e-W+1 .. e]) e ends[k] = e. + Z-norm per forma: (w - media)/std -> invariante a livello e scala -> confronto + sulla sola morfologia. Le finestre piatte (std=0) hanno norm tutta a 0. + """ + if len(close) < W: + return np.empty((0, W)), np.empty(0, dtype=int) + wins = sliding_window_view(close, W) # (N-W+1, W), wins[k] = close[k..k+W-1] + mu = wins.mean(axis=1, keepdims=True) + sd = wins.std(axis=1, keepdims=True) + sd = np.where(sd == 0, 1.0, sd) + M = (wins - mu) / sd + ends = np.arange(W - 1, len(close)) # finestra k termina in e = k+W-1 + return M, ends + + +def fwd_return(close: np.ndarray, H: int) -> np.ndarray: + """Rendimento forward a H barre per ogni indice: (close[i+H]-close[i])/close[i]. + NaN dove i+H esce dai dati (non usabile come esito).""" + out = np.full(len(close), np.nan) + out[: len(close) - H] = (close[H:] - close[:-H]) / close[:-H] + return out + + +# --------------------------- analog forecasting causale --------------------------- +def analog_entries(df, W=24, H=12, K=50, rebuild=250, min_lib=800, + agree=0.60, conf_atr=0.0, tp_atr=None, sl_atr=None, + trend_max=None, ema_long=200) -> list[dict]: + """Entries da nearest-neighbour sulla FORMA (causale, no look-ahead). + + W: lunghezza finestra-forma. H: orizzonte previsione (= max_bars). K: n. analoghi. + rebuild: ogni quante barre si ricostruisce il KDTree (libreria cresce nel tempo). + min_lib: barre minime di storia prima di iniziare a operare. + agree: frazione minima di analoghi concordi sul segno per entrare (>0.5). + conf_atr: soglia |rendimento medio analoghi| in multipli di ATR-equivalente (0=off). + tp_atr/sl_atr: take-profit/stop in multipli di ATR (None = solo time-limit H). + trend_max: salta se |close-EMA(ema_long)|/ATR14 > trend_max (filtro trend, None=off). + """ + close = df["close"].values + n = len(close) + a = atr(df, 14) + M, ends = znorm_windows(close, W) # forme z-norm e indice di fine + end_pos = {int(e): k for k, e in enumerate(ends)} + fr = fwd_return(close, H) # esito H-barre per ogni indice + el = None + if trend_max is not None: + el = ema(close, ema_long) + + entries: list[dict] = [] + tree = None + lib_idx = None # indici e (fine finestra) nella libreria + next_rebuild = 0 + + for i in range(min_lib, n - 1): + # libreria causale: finestre la cui forma E il cui esito H sono < i + if tree is None or i >= next_rebuild: + eligible = ends[(ends <= i - 1 - H) & (ends >= W - 1)] + # esito noto e finito (fr non-NaN garantito da e+H <= i-1 < n) + eligible = eligible[~np.isnan(fr[eligible])] + if len(eligible) < max(K * 3, 200): + next_rebuild = i + rebuild + continue + tree = cKDTree(M[[end_pos[int(e)] for e in eligible]]) + lib_idx = eligible + next_rebuild = i + rebuild + + if tree is None: + continue + q = M[end_pos[i]] + if not np.isfinite(q).all(): + continue + kk = min(K, len(lib_idx)) + _, nn = tree.query(q, k=kk) + nn = np.atleast_1d(nn) + outs = fr[lib_idx[nn]] # rendimenti H-barre degli analoghi + outs = outs[~np.isnan(outs)] + if len(outs) < 5: + continue + mean_out = float(outs.mean()) + d = 1 if mean_out > 0 else -1 + frac = float(np.mean(np.sign(outs) == d)) + if frac < agree: + continue + if conf_atr > 0: + if not (abs(mean_out) * close[i] >= conf_atr * a[i]): + continue + if trend_max is not None and a[i] > 0: + if abs(close[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"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + e["sl"] = close[i] - d * sl_atr * a[i] + entries.append(e) + return entries + + +# --------------------------- kNN grezzo cacheable (perf) --------------------------- +def analog_signals(df, W=24, H=12, K=50, rebuild=250, min_lib=800) -> dict: + """Calcola UNA volta il forecast kNN grezzo per barra (causale), riusabile da + piu' filtri (agree/conf_atr/trend/tp/sl) senza ri-eseguire la query costosa. + + Ritorna dict con array allineati per le barre che hanno un forecast valido: + i : indice barra (ingresso eseguibile a close[i]) + mean_out : rendimento H-barre medio degli analoghi + frac : frazione di analoghi concordi col segno di mean_out (>=0.5) + d : segno previsto (+1/-1) + Identico, riga per riga, alla logica di analog_entries (stessa libreria causale, + stessa query, stessa soglia len(outs)>=5) ma SENZA i filtri di selezione. + """ + close = df["close"].values + n = len(close) + M, ends = znorm_windows(close, W) + end_pos = {int(e): k for k, e in enumerate(ends)} + fr = fwd_return(close, H) + + out_i: list[int] = [] + out_mean: list[float] = [] + out_frac: list[float] = [] + out_d: list[int] = [] + tree = None + lib_idx = None + next_rebuild = 0 + + for i in range(min_lib, n - 1): + if tree is None or i >= next_rebuild: + eligible = ends[(ends <= i - 1 - H) & (ends >= W - 1)] + eligible = eligible[~np.isnan(fr[eligible])] + if len(eligible) < max(K * 3, 200): + next_rebuild = i + rebuild + continue + tree = cKDTree(M[[end_pos[int(e)] for e in eligible]]) + lib_idx = eligible + next_rebuild = i + rebuild + if tree is None: + continue + q = M[end_pos[i]] + if not np.isfinite(q).all(): + continue + kk = min(K, len(lib_idx)) + _, nn = tree.query(q, k=kk) + nn = np.atleast_1d(nn) + outs = fr[lib_idx[nn]] + outs = outs[~np.isnan(outs)] + if len(outs) < 5: + continue + mean_out = float(outs.mean()) + d = 1 if mean_out > 0 else -1 + frac = float(np.mean(np.sign(outs) == d)) + out_i.append(i); out_mean.append(mean_out); out_frac.append(frac); out_d.append(d) + + return { + "i": np.asarray(out_i, dtype=int), + "mean_out": np.asarray(out_mean, dtype=float), + "frac": np.asarray(out_frac, dtype=float), + "d": np.asarray(out_d, dtype=int), + "H": H, + } + + +def entries_from_signals(df, sig: dict, agree=0.60, conf_atr=0.0, + tp_atr=None, sl_atr=None, trend_max=None, ema_long=200) -> list[dict]: + """Applica i filtri di selezione al forecast grezzo di analog_signals (cheap). + Risultato identico ad analog_entries con gli stessi parametri (stesso W/H/K/rebuild + usati per costruire sig).""" + close = df["close"].values + a = atr(df, 14) + H = sig["H"] + el = ema(close, ema_long) if trend_max is not None else None + entries: list[dict] = [] + for k in range(len(sig["i"])): + i = int(sig["i"][k]); d = int(sig["d"][k]) + if sig["frac"][k] < agree: + continue + if conf_atr > 0 and not (abs(sig["mean_out"][k]) * close[i] >= conf_atr * a[i]): + continue + if trend_max is not None and a[i] > 0 and abs(close[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"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + e["sl"] = close[i] - d * sl_atr * a[i] + entries.append(e) + return entries + + +# --------------------------- verifica no look-ahead --------------------------- +def check_no_lookahead(df, W=24, H=12) -> bool: + """La forma a i deve restare invariata se perturbo il FUTURO (>i). + Conferma che znorm_windows usa solo close fino a i.""" + close = df["close"].values.copy() + M0, ends = znorm_windows(close, W) + pos = {int(e): k for k, e in enumerate(ends)} + i = len(close) // 2 + q0 = M0[pos[i]].copy() + close2 = close.copy() + close2[i + 1:] *= 1.5 # stravolgo il futuro + M1, _ = znorm_windows(close2, W) + q1 = M1[pos[i]] + ok = np.allclose(q0, q1) + print(f" no-lookahead forma a i={i}: {'OK' if ok else 'VIOLATO'} " + f"(max diff {np.max(np.abs(q0 - q1)):.2e})") + return ok + + +if __name__ == "__main__": + print("=" * 92) + print(" SHAPE_LAB — baseline analog forecasting (kNN sulla forma) | netto fee, OOS") + print("=" * 92) + df = get_df("BTC", "1h") + check_no_lookahead(df) + print("\n BTC 1h — sweep base W/H/K (time-exit a H barre):") + for W, H, K in [(24, 12, 50), (24, 24, 50), (48, 24, 80), (12, 6, 40), (48, 48, 100)]: + ents = analog_entries(df, W=W, H=H, K=K, agree=0.60) + evaluate(f"analog W{W}H{H}K{K}", ents, df) diff --git a/scripts/analysis/shape_ml_research.py b/scripts/analysis/shape_ml_research.py new file mode 100644 index 0000000..05fc065 --- /dev/null +++ b/scripts/analysis/shape_ml_research.py @@ -0,0 +1,427 @@ +"""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) -> list[dict]: + """Walk-forward: a blocchi di `retrain` barre, allena su TUTTO il 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}.""" + 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) + tr_ends = ends[(ends <= blk - 1 - H) & (ends >= W - 1)] + 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() diff --git a/scripts/analysis/shape_pivot_research.py b/scripts/analysis/shape_pivot_research.py new file mode 100644 index 0000000..d4d07ae --- /dev/null +++ b/scripts/analysis/shape_pivot_research.py @@ -0,0 +1,333 @@ +"""Famiglia SHAPE-PIVOT: geometria a punti di svolta (PIP / pivot) -> bias futuro. + +Idea (causale, no look-ahead): + - a ogni barra i comprimo la finestra di L barre terminante a close[i] nei suoi + P punti percettivamente importanti (PIP, Perceptually Important Points: i punti + di massima deviazione dalla retta congiungente — Fu et al.); + - la sequenza di P punti e' una POLILINEA = forma geometrica grezza; + - la classifico con feature interpretabili e CAUSALI: + * trend dei pivot interni: higher-highs/higher-lows (HH/HL) vs lower-* (LH/LL); + * convergenza/divergenza delle pendenze (triangoli/cunei); + * distanza % di close[i] dall'ultimo pivot alto/basso (vicino a R / a S); + * pendenza dell'ultimo segmento (slancio recente); + - per ogni CLASSE geometrica stimo l'esito medio a H barre usando SOLO occorrenze + passate il cui esito era gia' realizzato prima di i (statistica causale rolling); + - entro a close[i] nella direzione del bias di classe se l'edge passato e' netto; + exit a H barre o TP/SL in ATR. + +VINCOLI (CLAUDE.md "metodologia obbligatoria" + "lezione squeeze look-ahead"): + - PIP/pivot calcolati SOLO su close[i-L+1 .. i]; nessun pivot "confermato dal futuro". + - ogni statistica per-classe usa solo campioni con esito (entry+H) <= i-1. + - ingresso eseguibile a close[i]; netto fee (0.10% RT base, sweep a 0.20%); leva 3x, + pos 0.15; validazione OOS (ultimo 30%) + robustezza griglia + >=2 asset. + - check di causalita' esplicito (perturbo il futuro: la forma a i non cambia). + +Riusa l'engine netto-fee + OOS di explore_lab (simulate/evaluate/robust). +""" +from __future__ import annotations + +import sys +from pathlib import Path + +import numpy as np + +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, _dt, OOS_FRAC, +) + + +# ========================================================================= +# PIP — Perceptually Important Points (causale, solo su close[a..b]) +# ========================================================================= +def pip_indices(seg: np.ndarray, p: int) -> list[int]: + """Estrae p indici PIP dalla serie `seg` (inclusi i 2 estremi). + + Algoritmo Fu et al.: parti dai 2 estremi; aggiungi iterativamente il punto a + massima distanza VERTICALE dalla retta che unisce i due PIP adiacenti, finche' + non hai p punti. Tutto sul segmento dato -> nessun look-ahead se seg=close[..i]. + """ + n = len(seg) + if p >= n: + return list(range(n)) + pts = [0, n - 1] + while len(pts) < p: + best_d, best_k = -1.0, -1 + for s in range(len(pts) - 1): + l, r = pts[s], pts[s + 1] + if r - l < 2: + continue + x1, y1 = l, seg[l] + x2, y2 = r, seg[r] + dx = x2 - x1 + # distanza verticale dalla retta (interpolazione lineare in x) + for k in range(l + 1, r): + if dx == 0: + dist = abs(seg[k] - y1) + else: + yline = y1 + (y2 - y1) * (k - x1) / dx + dist = abs(seg[k] - yline) + if dist > best_d: + best_d, best_k = dist, k + if best_k < 0: + break + # inserisci mantenendo l'ordine + for s in range(len(pts) - 1): + if pts[s] < best_k < pts[s + 1]: + pts.insert(s + 1, best_k) + break + return pts + + +# ========================================================================= +# Classe geometrica della polilinea PIP (feature causali interpretabili) +# ========================================================================= +def shape_class(seg: np.ndarray, p: int) -> tuple | None: + """Ritorna una tupla-classe discreta della forma PIP di `seg`, o None se degenere. + + Feature (tutte da seg=close[..i], causali): + - dir_seq: per ogni pivot interno, segno della variazione vs precedente + (sequenza su/giu) -> cattura HH/HL vs LH/LL e zig-zag; + - conv: convergenza pendenze inizio vs fine (triangolo/cuneo): segno di + (|slope_last| - |slope_first|) discretizzato; + - loc: posizione di close[i] nel range della finestra (vicino a max=resistenza, + vicino a min=supporto), in 3 bucket. + La classe e' invariante a livello/scala (z-norm implicito su forma). + """ + idx = pip_indices(seg, p) + if len(idx) < 3: + return None + y = seg[idx] + rng = y.max() - y.min() + if rng <= 0: + return None + yn = (y - y.min()) / rng # forma normalizzata 0..1 + # sequenza direzioni dei segmenti (su=1 / giu=0) + diffs = np.diff(yn) + dir_seq = tuple(int(x > 0) for x in diffs) + # convergenza: pendenza primo vs ultimo segmento + s_first = abs(diffs[0]) + s_last = abs(diffs[-1]) + if s_last > s_first * 1.3: + conv = 1 # divergente (slancio finale) + elif s_last < s_first * 0.77: + conv = -1 # convergente (compressione, triangolo/cuneo) + else: + conv = 0 + # posizione di close[i] (=ultimo punto) nel range: 0..1 in 3 bucket + last = yn[-1] + loc = 0 if last < 0.33 else (2 if last > 0.67 else 1) + return (dir_seq, conv, loc) + + +# ========================================================================= +# Strategia: bias per-classe stimato CAUSALMENTE (rolling, esito realizzato) +# ========================================================================= +def pivot_entries(df, L=48, P=5, H=12, min_lib=1000, min_samples=20, + edge=0.0, tp_atr=None, sl_atr=None, + trend_max=None, ema_long=200, mode="bias") -> list[dict]: + """Entries dalla geometria PIP con bias di classe causale. + + L: lunghezza finestra-forma. P: n. punti PIP. H: orizzonte (=max_bars). + min_lib: barre minime prima di operare. min_samples: campioni minimi per fidarsi + della statistica di una classe. edge: |rendimento medio classe| minimo + (frazione, es. 0.002 = 0.2%) per entrare. mode: + - "bias": entra nel verso del rendimento medio passato della classe (momentum + della forma: la classe X storicamente -> su/giu); + - "fade": entra nel verso OPPOSTO (test mean-reversion della forma). + Statistica per-classe accumulata SOLO con esiti realizzati < i (causale stretta). + """ + close = df["close"].values + high = df["high"].values + low = df["low"].values + n = len(close) + a = atr(df, 14) + el = ema(close, ema_long) if trend_max is not None else None + + # stato rolling per classe: somma rendimenti e conteggio (solo esiti < i) + cls_sum: dict[tuple, float] = {} + cls_cnt: dict[tuple, int] = {} + # coda di campioni la cui forma e' stata calcolata ma esito non ancora maturo + # pending[t] = (classe, indice_entry t) -> matura quando t+H <= i-1 + pending: list[tuple] = [] # (mature_at, cls, t) + pend_ptr = 0 + + entries: list[dict] = [] + + for i in range(min_lib, n - 1): + # 1) integra nello storico tutti i campioni il cui esito e' realizzato (< i) + # un campione formato a t matura quando t+H <= i-1 => mature_at = t+H+1 <= i + while pend_ptr < len(pending) and pending[pend_ptr][0] <= i: + _, cls_p, t = pending[pend_ptr] + ret_real = (close[t + H] - close[t]) / close[t] + cls_sum[cls_p] = cls_sum.get(cls_p, 0.0) + ret_real + cls_cnt[cls_p] = cls_cnt.get(cls_p, 0) + 1 + pend_ptr += 1 + + # 2) forma corrente (solo close fino a i) + seg = close[i - L + 1: i + 1] + cls = shape_class(seg, P) + if cls is None: + continue + # registra il campione corrente come pending (esito da realizzare in futuro) + pending.append((i + H + 1, cls, i)) + + # 3) decisione con statistica PASSATA della classe + cnt = cls_cnt.get(cls, 0) + if cnt < min_samples: + continue + mean_ret = cls_sum[cls] / cnt + if abs(mean_ret) < edge: + continue + d = 1 if mean_ret > 0 else -1 + if mode == "fade": + d = -d + # filtro trend opzionale + if trend_max is not None and a[i] > 0: + if abs(close[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"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + e["sl"] = close[i] - d * sl_atr * a[i] + entries.append(e) + return entries + + +# ========================================================================= +# Filone (c): distanza da supporto/resistenza locale (ultimo pivot alto/basso) +# ========================================================================= +def sr_entries(df, L=48, P=7, H=12, near=0.5, mode="fade", + tp_atr=None, sl_atr=None, trend_max=None, ema_long=200) -> list[dict]: + """Filone (c): close[i] vicino all'ultimo pivot alto (R) o basso (S) della forma. + + Usa i PIP per individuare l'ultimo massimo/minimo locale (resistenza/supporto) e + misura la distanza % di close[i]. Se close e' entro `near`*ATR da R -> bias short + (mode='fade': rimbalzo da R) o long (mode='break': rottura). Simmetrico per S. + Tutto causale: PIP su close[..i], decisione a close[i]. + """ + close = df["close"].values + n = len(close) + a = atr(df, 14) + el = ema(close, ema_long) if trend_max is not None else None + entries: list[dict] = [] + + for i in range(L, n - 1): + seg = close[i - L + 1: i + 1] + idx = pip_indices(seg, P) + if len(idx) < 3 or a[i] <= 0: + continue + y = seg[idx] + # pivot interni (escludi i 2 estremi e l'ultimo punto = close[i]) + inner = y[1:-1] + if len(inner) == 0: + continue + res = inner.max() # resistenza locale + sup = inner.min() # supporto locale + cur = close[i] + dist_r = (res - cur) / a[i] + dist_s = (cur - sup) / a[i] + d = None + if 0 <= dist_r <= near: # appena sotto R + d = -1 if mode == "fade" else 1 + elif 0 <= dist_s <= near: # appena sopra S + d = 1 if mode == "fade" else -1 + if d is None: + continue + if trend_max is not None and abs(cur - el[i]) / a[i] > trend_max: + continue + e = {"i": i, "d": d, "max_bars": H} + if tp_atr is not None: + e["tp"] = cur + d * tp_atr * a[i] + if sl_atr is not None: + e["sl"] = cur - d * sl_atr * a[i] + entries.append(e) + return entries + + +# ========================================================================= +# Check causalita' esplicito +# ========================================================================= +def check_no_lookahead(df, L=48, P=5) -> bool: + """La classe-forma a i non deve cambiare se perturbo il FUTURO (>i).""" + close = df["close"].values.copy() + i = len(close) // 2 + seg0 = close[i - L + 1: i + 1].copy() + c0 = shape_class(seg0, P) + close2 = close.copy() + close2[i + 1:] *= 1.7 # stravolge il futuro + seg1 = close2[i - L + 1: i + 1] + c1 = shape_class(seg1, P) + ok = (c0 == c1) + print(f" no-lookahead classe-forma a i={i}: {'OK' if ok else 'VIOLATO'} " + f"(c0={c0} c1={c1})") + # check su PIP indices + p0 = pip_indices(seg0, P) + p1 = pip_indices(seg1, P) + ok2 = (p0 == p1) + print(f" no-lookahead indici PIP: {'OK' if ok2 else 'VIOLATO'}") + return ok and ok2 + + +# ========================================================================= +# run() riproducibile +# ========================================================================= +def run(): + print("=" * 100) + print(" SHAPE-PIVOT RESEARCH — geometria PIP/pivot -> bias futuro | netto fee, OOS") + print("=" * 100) + + df_btc = get_df("BTC", "1h") + print("\n[CAUSALITA']") + check_no_lookahead(df_btc, L=48, P=5) + + assets = ["BTC", "ETH", "SOL", "ADA"] + dfs = {a: get_df(a, "1h") for a in assets} + + # ---- A) bias di classe PIP (momentum della forma) ---- + print("\n[A] BIAS di classe PIP (entra nel verso del rendimento medio passato della classe)") + print(" sweep L/P/H, edge=0.002, min_samples=25, time-exit a H") + A_grid = [(48, 5, 12), (48, 5, 24), (72, 6, 24), (36, 5, 12), (96, 7, 24), (48, 7, 12)] + for L, P, H in A_grid: + print(f" -- L{L} P{P} H{H} --") + for a in assets: + ents = pivot_entries(dfs[a], L=L, P=P, H=H, edge=0.002, min_samples=25, mode="bias") + evaluate(f"{a} bias L{L}P{P}H{H}", ents, dfs[a]) + + # ---- B) fade di classe PIP (mean-reversion della forma) ---- + print("\n[B] FADE di classe PIP (entra opposto al bias storico -> test mean-reversion)") + for L, P, H in A_grid: + print(f" -- L{L} P{P} H{H} --") + for a in assets: + ents = pivot_entries(dfs[a], L=L, P=P, H=H, edge=0.002, min_samples=25, mode="fade") + evaluate(f"{a} fade L{L}P{P}H{H}", ents, dfs[a]) + + # ---- C) supporto/resistenza locale dai pivot ---- + print("\n[C] S/R locale dai PIP — FADE (rimbalzo da R/S) vs BREAK (rottura)") + for mode in ("fade", "break"): + for near in (0.5, 1.0): + print(f" -- mode={mode} near={near} ATR, TP/SL 1.5/1.5 ATR, H=12 --") + for a in assets: + ents = sr_entries(dfs[a], L=48, P=7, H=12, near=near, mode=mode, + tp_atr=1.5, sl_atr=1.5) + evaluate(f"{a} SR-{mode} near{near}", ents, dfs[a]) + + # ---- D) miglior candidato con TP/SL ATR + filtro trend (se A o B mostra segnali) ---- + print("\n[D] FADE di classe con TP/SL ATR (2.0/1.5) + filtro trend 3.0, L48 P5 H24") + for a in assets: + ents = pivot_entries(dfs[a], L=48, P=5, H=24, edge=0.002, min_samples=25, + mode="fade", tp_atr=2.0, sl_atr=1.5, trend_max=3.0) + res = evaluate(f"{a} fadeTPSL L48P5H24", ents, dfs[a]) + if robust(res): + print(f" ^^^ {a} ROBUSTO") + + print("\n" + "=" * 100) + print(" Verdetto: cerca righe con FULL>0 E OOS>0 E fee0.2% OOS>0 su >=2 asset.") + print("=" * 100) + + +if __name__ == "__main__": + run() diff --git a/scripts/analysis/shape_template_research.py b/scripts/analysis/shape_template_research.py new file mode 100644 index 0000000..08f9b7f --- /dev/null +++ b/scripts/analysis/shape_template_research.py @@ -0,0 +1,443 @@ +"""SHAPE_TEMPLATE_RESEARCH — edge nella FORMA del prezzo: distanze alternative e template canonici. + +Due filoni, sull'harness ONESTO condiviso (shape_lab + explore_lab), netto-fee e OOS: + + 1. ANALOG con distanza di FORMA alternativa (DTW warping-invariant, correlazione/coseno) + confrontata HEAD-TO-HEAD con l'euclidea a PARITA' di selettivita' (stessa libreria, + stesso K, stessa soglia di accordo). DTW e' O(W^2): si usa una libreria SOTTOCAMPIONATA + (uno start ogni `step` barre) + W ridotto + banda di Sakoe-Chiba. + + 2. TEMPLATE di forma canonici (doppio top/bottom, testa-spalle, V-reversal, salita/discesa + lineare, U). A ogni i misuro la similarita' (correlazione di Pearson sulla finestra + z-normalizzata) fra forma recente e ogni template; se supera soglia, entro a close[i] + nella DIREZIONE ATTESA del template stimata SOLO sul passato (esito medio causale delle + occorrenze gia' concluse di quel template), exit H barre o tp/sl ATR. + +VINCOLI anti-look-ahead (verificati esplicitamente): + - la forma/match a i usa SOLO close fino a i (z-norm causale); + - la direzione attesa di ogni template e la libreria analog usano SOLO occorrenze il cui + esito a H barre e' gia' realizzato PRIMA di i (end + H <= i-1); + - ingresso eseguibile a close[i]; exit TP/SL intrabar o time-limit H. + +Netto fee 0.10% RT baseline + sweep fino a 0.20%. Leva 3x, pos 0.15. OOS ultimo 30%. + +Run riproducibile: uv run python scripts/analysis/shape_template_research.py +DTW e' costoso: usa run_in_background per gli sweep larghi (vedi --sweep). +""" +from __future__ import annotations + +import sys +import time +from pathlib import Path + +import numpy as np +from numpy.lib.stride_tricks import sliding_window_view + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +from scripts.analysis.explore_lab import get_df, evaluate, robust, simulate, atr, ema, OOS_FRAC # noqa: E402 +from scripts.analysis.shape_lab import znorm_windows, fwd_return # noqa: E402 + +RNG_SEED = 7 +SUBC_ASSETS = ["BTC", "ETH", "SOL"] + + +# ========================================================================================= +# DISTANZE DI FORMA +# ========================================================================================= +def _euclid(q: np.ndarray, lib: np.ndarray) -> np.ndarray: + """Distanza euclidea fra q (W,) e ogni riga di lib (M,W). Forme gia' z-normalizzate.""" + return np.sqrt(((lib - q) ** 2).sum(axis=1)) + + +def _corr_dist(q: np.ndarray, lib: np.ndarray) -> np.ndarray: + """Distanza = 1 - correlazione di Pearson (q,lib gia' z-norm: corr = q.lib / W).""" + # forme z-norm hanno media 0 std 1 -> dot/W e' la correlazione di Pearson + corr = (lib @ q) / q.shape[0] + return 1.0 - corr + + +def _cosine_dist(q: np.ndarray, lib: np.ndarray) -> np.ndarray: + """Distanza = 1 - coseno fra q e ogni riga di lib.""" + qn = q / (np.linalg.norm(q) + 1e-12) + ln = lib / (np.linalg.norm(lib, axis=1, keepdims=True) + 1e-12) + return 1.0 - (ln @ qn) + + +def _dtw_one(a: np.ndarray, b: np.ndarray, band: int) -> float: + """DTW 1D con banda di Sakoe-Chiba (|i-j|<=band). a,b stessa lunghezza W.""" + n = len(a) + INF = 1e18 + prev = np.full(n + 1, INF) + prev[0] = 0.0 + for i in range(1, n + 1): + cur = np.full(n + 1, INF) + jlo = max(1, i - band) + jhi = min(n, i + band) + ai = a[i - 1] + for j in range(jlo, jhi + 1): + cost = abs(ai - b[j - 1]) + m = prev[j] + if prev[j - 1] < m: + m = prev[j - 1] + if cur[j - 1] < m: + m = cur[j - 1] + cur[j] = cost + m + prev = cur + return float(prev[n]) + + +def _dtw_dist(q: np.ndarray, lib: np.ndarray, band: int) -> np.ndarray: + """DTW di q contro ogni riga di lib. O(M * W * band).""" + out = np.empty(lib.shape[0]) + for k in range(lib.shape[0]): + out[k] = _dtw_one(q, lib[k], band) + return out + + +DIST_FUNCS = {"euclid": _euclid, "corr": _corr_dist, "cosine": _cosine_dist} + + +# ========================================================================================= +# FILONE 1 — ANALOG con distanza configurabile (libreria sottocampionata, causale) +# ========================================================================================= +def analog_dist_entries(df, dist="euclid", W=24, H=12, K=40, step=5, rebuild=500, + min_lib=2000, agree=0.62, dtw_band=4, dtw_prefilter=200, + decide_step=1, tp_atr=None, sl_atr=None, + trend_max=None, ema_long=200) -> list[dict]: + """Analog kNN sulla FORMA con metrica `dist` ('euclid'|'corr'|'cosine'|'dtw'). + + Libreria SOTTOCAMPIONATA: si considerano solo finestre che terminano a indici + multipli di `step` (riduce N e rende DTW trattabile). Causalita': la libreria a + decisione i contiene solo finestre con end<=i-1-H (esito gia' realizzato). + Ricostruita ogni `rebuild` barre. Stessa firma per tutte le metriche -> confronto + head-to-head a parita' di selettivita' (stesso W,H,K,agree). + + DTW (costoso, O(W*band) per coppia in Python): si PREFILTRA con la correlazione ai + `dtw_prefilter` candidati piu' simili, poi si fa DTW-rerank solo su quelli (approccio + standard lower-bound/rerank). `decide_step`>1 valuta una barra ogni decide_step (non + cambia la causalita', riduce solo il numero di query DTW). + """ + close = df["close"].values + n = len(close) + a = atr(df, 14) + M, ends = znorm_windows(close, W) + end_pos = {int(e): k for k, e in enumerate(ends)} + fr = fwd_return(close, H) + el = ema(close, ema_long) if trend_max is not None else None + + # candidati di libreria: solo end multipli di step (sottocampionamento causale fisso) + base_ends = ends[(ends % step == 0)] + + entries: list[dict] = [] + lib_M = None + lib_idx = None + next_rebuild = 0 + + for i in range(min_lib, n - 1): + if i % decide_step != 0: + continue + if lib_M is None or i >= next_rebuild: + elig = base_ends[(base_ends <= i - 1 - H) & (base_ends >= W - 1)] + elig = elig[~np.isnan(fr[elig])] + if len(elig) < max(K * 3, 200): + next_rebuild = i + rebuild + continue + lib_M = M[[end_pos[int(e)] for e in elig]] + lib_idx = elig + next_rebuild = i + rebuild + + if lib_M is None: + continue + q = M[end_pos[i]] + if not np.isfinite(q).all(): + continue + if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max: + continue + + if dist == "dtw": + # prefiltro corr (cheap, vettoriale) -> DTW-rerank solo sui top dtw_prefilter + pre = _corr_dist(q, lib_M) + npre = min(dtw_prefilter, len(lib_idx)) + cand = np.argpartition(pre, npre - 1)[:npre] + dd_cand = _dtw_dist(q, lib_M[cand], dtw_band) + kk = min(K, len(cand)) + sub = np.argpartition(dd_cand, kk - 1)[:kk] + nn = cand[sub] + outs = fr[lib_idx[nn]] + outs = outs[~np.isnan(outs)] + if len(outs) < 5: + continue + mean_out = float(outs.mean()) + d = 1 if mean_out > 0 else -1 + frac = float(np.mean(np.sign(outs) == d)) + if frac < agree: + continue + e = {"i": i, "d": d, "max_bars": H} + if tp_atr is not None and a[i] > 0: + e["tp"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + e["sl"] = close[i] - d * sl_atr * a[i] + entries.append(e) + continue + + dd = DIST_FUNCS[dist](q, lib_M) + kk = min(K, len(lib_idx)) + nn = np.argpartition(dd, kk - 1)[:kk] + outs = fr[lib_idx[nn]] + outs = outs[~np.isnan(outs)] + if len(outs) < 5: + continue + mean_out = float(outs.mean()) + d = 1 if mean_out > 0 else -1 + frac = float(np.mean(np.sign(outs) == d)) + if frac < agree: + continue + e = {"i": i, "d": d, "max_bars": H} + if tp_atr is not None and a[i] > 0: + e["tp"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + e["sl"] = close[i] - d * sl_atr * a[i] + entries.append(e) + return entries + + +# ========================================================================================= +# FILONE 2 — TEMPLATE di forma canonici +# ========================================================================================= +def make_templates(W: int) -> dict[str, np.ndarray]: + """Template parametrici z-normalizzati di lunghezza W (forma pura, no scala/livello). + + Sono solo descrittori di FORMA recente (gli ultimi W close). La direzione attesa NON + e' decisa a priori: viene stimata causalmente sul passato (vedi template_entries). + """ + t = np.linspace(0, 1, W) + s = 0.012 # ampiezza gaussiana scalata sulla finestra (W-indipendente in t in [0,1]) + g = lambda c: np.exp(-((t - c) ** 2) / s) + raw = { + # estremi di reversione a DOPPIO picco (due massimi / minimi simmetrici) + "double_top": g(0.25) + g(0.75), # M: due cime + "double_bottom": -(g(0.25) + g(0.75)), # W: due fondi + # testa-spalle: spalla-testa-spalla (centro piu' alto) + "head_shoulders": g(0.2) + 1.7 * g(0.5) + g(0.8), + "inv_head_shoulders": -(g(0.2) + 1.7 * g(0.5) + g(0.8)), + # singola reversione + "v_bottom": np.abs(t - 0.5), + "inv_v_top": -np.abs(t - 0.5), + "u_bottom": (t - 0.5) ** 2, + "arch_top": -((t - 0.5) ** 2), + # trend lineari + "ramp_up": t, + "ramp_down": -t, + } + out = {} + for k, v in raw.items(): + v = np.asarray(v, dtype=float) + sd = v.std() + out[k] = (v - v.mean()) / (sd if sd > 0 else 1.0) + return out + + +def template_entries(df, W=24, H=12, corr_min=0.85, dir_min=0.10, min_lib=2000, + rebuild=300, tp_atr=None, sl_atr=None, trend_max=None, ema_long=200, + templates=None) -> list[dict]: + """Entries da match con template canonici, DIREZIONE stimata SOLO sul passato. + + A ogni i, per ogni template calcolo la correlazione di Pearson fra la forma recente + z-norm (close[i-W+1..i]) e il template. Prendo il template a correlazione massima; se + >= corr_min lo considero "attivo". La DIREZIONE in cui entrare e' il segno del rendimento + forward MEDIO storico delle occorrenze gia' concluse (end+H<=i-1) di quel template + (stesso criterio di match), purche' |media| in barre-equivalenti superi dir_min*media_atr-ish + -> qui dir_min e' una soglia sulla |media forward| relativa (frazione). NIENTE direzione a + priori: se il passato non e' coerente (occorrenze= W - 1] + lib_M = M[[end_pos[int(e)] for e in lib_ends]] # (L, W) + corr_mat = (lib_M @ T.T) / W # (L, NT) + best_tpl = np.argmax(corr_mat, axis=1) + best_corr = corr_mat[np.arange(len(lib_ends)), best_tpl] + lib_fr = fr[lib_ends] + lib_end_arr = lib_ends + + entries: list[dict] = [] + # cache direzione per template, ricostruita ogni rebuild barre + dir_cache: dict[int, int] = {} + next_rebuild = 0 + + for i in range(min_lib, n - 1): + q = M[end_pos[i]] + if not np.isfinite(q).all(): + continue + cq = (T @ q) / W # corr con ogni template + bt = int(np.argmax(cq)) + if cq[bt] < corr_min: + continue + if trend_max is not None and a[i] > 0 and abs(close[i] - el[i]) / a[i] > trend_max: + continue + + # direzione attesa: media forward causale delle occorrenze concluse dello stesso template + if i >= next_rebuild: + dir_cache = {} + next_rebuild = i + rebuild + if bt not in dir_cache: + mask = (lib_end_arr <= i - 1 - H) & (best_tpl == bt) & (best_corr >= corr_min) & (~np.isnan(lib_fr)) + outs = lib_fr[mask] + if len(outs) < 30: + dir_cache[bt] = 0 + else: + m = float(outs.mean()) + # soglia: |media| forward deve superare dir_min volte la std forward (edge vs rumore) + sd = float(outs.std()) + 1e-12 + dir_cache[bt] = (1 if m > 0 else -1) if abs(m) / sd >= dir_min else 0 + d = dir_cache[bt] + if d == 0: + continue + e = {"i": i, "d": d, "max_bars": H} + if tp_atr is not None and a[i] > 0: + e["tp"] = close[i] + d * tp_atr * a[i] + if sl_atr is not None and a[i] > 0: + e["sl"] = close[i] - d * sl_atr * a[i] + entries.append(e) + return entries + + +# ========================================================================================= +# CHECK CAUSALITA' espliciti +# ========================================================================================= +def check_causality_analog(df, **kw) -> bool: + """Le entries non devono cambiare se perturbo il FUTURO oltre l'ultima barra usata. + Tronco il df a una certa lunghezza L e verifico che le entries con i bool: + L = int(len(df) * 0.55) + H = kw.get("H", 12) + full = template_entries(df, **kw) + trunc = template_entries(df.iloc[:L].reset_index(drop=True), **kw) + horizon = L - H - 2 + f = {e["i"]: e["d"] for e in full if e["i"] < horizon} + t = {e["i"]: e["d"] for e in trunc if e["i"] < horizon} + ok = (f == t) + print(f" causalita' template: {'OK' if ok else 'VIOLATO'} " + f"({len(f)} entries confrontate <{horizon})") + return ok + + +# ========================================================================================= +# RUN +# ========================================================================================= +def run_head_to_head(assets=SUBC_ASSETS, W=16, H=12, K=40, step=6, agree=0.62, + decide_step=4, dtw_prefilter=120): + """Confronto HEAD-TO-HEAD delle metriche di forma a PARITA' di selettivita'. + + Tutte le metriche valutano le STESSE barre-decisione (decide_step) con lo STESSO + W/H/K/agree: l'unica variabile e' la distanza. decide_step>1 serve a rendere DTW + trattabile (pura Python ~9ms/query); applicato a tutte per equita'. + """ + print("=" * 100) + print(f" FILONE 1 — ANALOG head-to-head metriche (W{W} H{H} K{K} step{step} " + f"agree{agree} decide_step{decide_step}) | netto fee, OOS") + print("=" * 100) + results = {} + for asset in assets: + df = get_df(asset, "1h") + print(f"\n --- {asset} 1h (n={len(df)}) ---", flush=True) + for dist in ["euclid", "corr", "cosine", "dtw"]: + t0 = time.time() + ents = analog_dist_entries(df, dist=dist, W=W, H=H, K=K, step=step, agree=agree, + dtw_band=max(2, W // 5), dtw_prefilter=dtw_prefilter, + decide_step=decide_step) + dt = time.time() - t0 + res = evaluate(f"{dist:<7s}", ents, df) + results[(asset, dist)] = res + print(f" ^ time={dt:>5.1f}s robust={'YES' if robust(res) else 'no '}", flush=True) + return results + + +def run_templates(assets=SUBC_ASSETS, W=20, H=12, corr_min=0.85, dir_min=0.10): + print("=" * 100) + print(f" FILONE 2 — TEMPLATE canonici (W{W} H{H} corr>={corr_min} dir>={dir_min}) | netto fee, OOS") + print("=" * 100) + results = {} + for asset in assets: + df = get_df(asset, "1h") + print(f"\n --- {asset} 1h (n={len(df)}) ---") + for cm in [0.80, 0.85, 0.90]: + ents = template_entries(df, W=W, H=H, corr_min=cm, dir_min=dir_min) + res = evaluate(f"corr_min={cm}", ents, df) + results[(asset, cm)] = res + print(f" ^ robust={'YES' if robust(res) else 'no '}") + return results + + +def run_sweep(): + """Sweep largo (lento per via di DTW). Usa run_in_background.""" + print("=" * 100) + print(" SWEEP LARGO — analog griglia W/H/K/step x metriche + template griglia") + print("=" * 100) + for W in [16, 20, 28]: + for H in [8, 12, 24]: + print(f"\n##### W={W} H={H} #####") + run_head_to_head(W=W, H=H, K=40, step=6, agree=0.62) + for W in [16, 20, 28]: + for H in [8, 12, 24]: + print(f"\n##### TEMPLATE W={W} H={H} #####") + run_templates(W=W, H=H, corr_min=0.85, dir_min=0.10) + + +def run(): + np.random.seed(RNG_SEED) + print("#" * 100) + print(" SHAPE_TEMPLATE_RESEARCH — distanze di forma alternative + template canonici") + print("#" * 100) + # 1) check causalita' espliciti + print("\n[CAUSALITA']") + dfb = get_df("BTC", "1h") + check_causality_analog(dfb, dist="euclid", W=20, H=12, K=40, step=6, min_lib=2000) + check_causality_analog(dfb, dist="dtw", W=16, H=12, K=40, step=8, min_lib=2000, + dtw_band=3, decide_step=20) + check_causality_template(dfb, W=20, H=12, corr_min=0.85) + # 2) head-to-head metriche + print() + run_head_to_head() + # 3) template + print() + run_templates() + + +if __name__ == "__main__": + if "--sweep" in sys.argv: + run_sweep() + elif "--templates" in sys.argv: + run_templates() + elif "--h2h" in sys.argv: + run_head_to_head() + else: + run()