"""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()