"""S3-03: Ultimate Squeeze — combina TUTTI i filtri migliori. Filtri che funzionano (testati singolarmente): - Anti-fakeout (+1% acc) - Long squeeze duration (+1% acc) - Cross-asset squeeze simultaneo (+0.5%) - Timing 4-16 UTC (+0.5%) - Correlation ETH-BTC alta per ETH trades (+1%) - Volume confirmation al breakout Nuovi filtri da testare: - Volume delta: up_volume - down_volume al breakout - Momentum confirmation: breakout nella direzione del trend 1h - Volatility regime: skip in regime estremo (RV > 100%) """ from __future__ import annotations import sys sys.path.insert(0, ".") import numpy as np import pandas as pd from src.data.downloader import load_data FEE_RT = 0.002 INITIAL = 1000 LEVERAGE = 3 def keltner_ratio(close, high, low, window=14): n = len(close) r = np.full(n, np.nan) for i in range(window, n): wc, wh, wl = close[i-window:i], high[i-window:i], low[i-window:i] ma = np.mean(wc) bb_std = np.std(wc) tr = np.maximum(wh-wl, np.maximum(np.abs(wh-np.roll(wc,1)), np.abs(wl-np.roll(wc,1)))) atr = np.mean(tr[1:]) kc = (ma+1.5*atr)-(ma-1.5*atr) bb = (ma+2*bb_std)-(ma-2*bb_std) if kc > 0: r[i] = bb/kc return r def ema(arr, period): r = np.full(len(arr), np.nan) k = 2/(period+1) r[period-1] = np.mean(arr[:period]) for i in range(period, len(arr)): r[i] = arr[i]*k + r[i-1]*(1-k) return r def rv_ann(close, window): lr = np.diff(np.log(np.where(close == 0, 1e-10, close))) r = np.full(len(close), np.nan) for i in range(window, len(lr)): r[i+1] = np.std(lr[i-window:i]) * np.sqrt(24*365) return r def run_ultimate(primary, tf="15m"): secondary = "ETH" if primary == "BTC" else "BTC" df = load_data(primary, tf) c, h, l, v = df["close"].values, df["high"].values, df["low"].values, df["volume"].values n = len(df) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) df2 = load_data(secondary, tf) c2, ts2 = df2["close"].values, df2["timestamp"].values kcr = keltner_ratio(c, h, l, 14) kcr2 = keltner_ratio(c2, df2["high"].values, df2["low"].values, 14) ema_50 = ema(c, 50) rv_48 = rv_ann(c, 48) # Rolling correlation ret1 = np.diff(np.log(np.where(c == 0, 1e-10, c))) ret2 = np.diff(np.log(np.where(c2[:len(c)] == 0, 1e-10, c2[:len(c)]))) min_len = min(len(ret1), len(ret2)) ret1 = ret1[:min_len] ret2 = ret2[:min_len] corr = np.full(n, np.nan) for i in range(48, min_len): cv = np.corrcoef(ret1[i-48:i], ret2[i-48:i])[0,1] corr[i+1] = cv if np.isfinite(cv) else 0 # Detect squeezes events = [] in_sq = False sq_start = 0 for i in range(15, n): if np.isnan(kcr[i]): continue is_sq = kcr[i] < 0.8 if is_sq and not in_sq: in_sq = True sq_start = i elif not is_sq and in_sq: in_sq = False dur = i - sq_start if dur < 5 or i + 6 >= n: continue events.append({"idx": i, "dur": dur, "sq_start": sq_start}) print(f"\n{'#'*70}") print(f" {primary} {tf} — ULTIMATE SQUEEZE ({len(events)} squeeze events)") print(f"{'#'*70}") filters_map = { "antifake": lambda ev, i: not _antifake(c, h, l, i), "long_sq": lambda ev, i: ev["dur"] >= 10, "timing": lambda ev, i: 4 <= ts.iloc[i].hour <= 16, "cross": lambda ev, i: _cross_squeeze(kcr2, i, ts, ts2), "corr_high": lambda ev, i: not np.isnan(corr[i]) and abs(corr[i]) >= 0.6, "vol_confirm": lambda ev, i: _vol_confirm(v, i, ev["sq_start"]), "trend_align": lambda ev, i: _trend_align(c, ema_50, i), "low_rv": lambda ev, i: not np.isnan(rv_48[i]) and rv_48[i] < 1.5, } def _antifake(c, h, l, i): if i + 1 >= len(c): return False br = h[i] - l[i] if br <= 0: return False if c[i] > c[i-1]: return (h[i] - c[i]) / br > 0.6 return (c[i] - l[i]) / br > 0.6 def _cross_squeeze(kcr2, i, ts1, ts2_arr): i2 = np.searchsorted(ts2_arr, ts.values[i].astype("int64") // 10**6) i2 = min(i2, len(kcr2)-1) return any(not np.isnan(kcr2[j]) and kcr2[j] < 0.85 for j in range(max(0,i2-10), i2+1)) def _vol_confirm(v, i, sq_start): avg = np.mean(v[sq_start:i]) return avg > 0 and v[i] > avg * 1.3 def _trend_align(c, ema_val, i): if np.isnan(ema_val[i]): return True first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0 if first_ret > 0: return c[i] > ema_val[i] return c[i] < ema_val[i] # Test combinazioni incrementali combos = [ ("BASE", []), ("antifake", ["antifake"]), ("long_sq", ["long_sq"]), ("antifake+long", ["antifake", "long_sq"]), ("antifake+timing", ["antifake", "timing"]), ("antifake+cross", ["antifake", "cross"]), ("antifake+corr", ["antifake", "corr_high"]), ("antifake+vol", ["antifake", "vol_confirm"]), ("antifake+trend", ["antifake", "trend_align"]), ("af+long+timing", ["antifake", "long_sq", "timing"]), ("af+long+cross", ["antifake", "long_sq", "cross"]), ("af+long+corr", ["antifake", "long_sq", "corr_high"]), ("af+long+trend", ["antifake", "long_sq", "trend_align"]), ("af+long+cross+time", ["antifake", "long_sq", "cross", "timing"]), ("af+long+corr+time", ["antifake", "long_sq", "corr_high", "timing"]), ("af+long+corr+trend", ["antifake", "long_sq", "corr_high", "trend_align"]), ("ALL_NO_VOL", ["antifake", "long_sq", "cross", "timing", "corr_high", "trend_align", "low_rv"]), ("ALL", ["antifake", "long_sq", "cross", "timing", "corr_high", "vol_confirm", "trend_align", "low_rv"]), ("BEST_5", ["antifake", "long_sq", "corr_high", "trend_align", "low_rv"]), ] results = [] for combo_name, filter_names in combos: yearly = {} capital = float(INITIAL) peak = capital max_dd = 0 for ev in events: i = ev["idx"] first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0 if abs(first_ret) < 0.001: continue skip = False for fn in filter_names: if fn in filters_map and not filters_map[fn](ev, i): skip = True break if skip: continue direction = 1 if first_ret > 0 else -1 entry = c[i-1] exit_price = c[min(i+2, n-1)] actual = (exit_price - entry) / entry * direction net = actual * LEVERAGE - FEE_RT * LEVERAGE capital += capital * 0.15 * net capital = max(capital, 10) if capital > peak: peak = capital dd = (peak - capital) / peak max_dd = max(max_dd, dd) year = ts.iloc[i].year if year not in yearly: yearly[year] = {"w": 0, "t": 0, "pnls": []} yearly[year]["t"] += 1 if actual > 0: yearly[year]["w"] += 1 yearly[year]["pnls"].append(net * INITIAL) all_t = sum(d["t"] for d in yearly.values()) all_w = sum(d["w"] for d in yearly.values()) if all_t < 20: continue acc = all_w / all_t * 100 pnl = sum(p for d in yearly.values() for p in d["pnls"]) worst = min(yearly.items(), key=lambda x: x[1]["w"]/x[1]["t"] if x[1]["t"]>0 else 0) wa = worst[1]["w"]/worst[1]["t"]*100 if worst[1]["t"]>0 else 0 results.append({ "name": combo_name, "trades": all_t, "acc": acc, "pnl": pnl, "dd": max_dd*100, "capital": capital, "worst": f"{worst[0]}({wa:.0f}%)", "yearly": yearly, }) results.sort(key=lambda x: x["acc"], reverse=True) print(f"\n {'Name':.<28s} {'Trades':>6s} {'Acc':>6s} {'PnL€':>9s} {'DD%':>5s} {'Worst':>12s}") print(f" {'-'*70}") for r in results[:20]: tag = "✅✅" if r["acc"] >= 80 else "✅" if r["acc"] >= 78 else "" print(f" {r['name']:.<28s} {r['trades']:>6d} {r['acc']:>5.1f}% €{r['pnl']:>+8.0f} {r['dd']:>4.1f}% {r['worst']:>12s} {tag}") # Dettaglio migliore if results: best = results[0] print(f"\n MIGLIORE: {best['name']} → {best['acc']:.1f}% acc, DD {best['dd']:.1f}%") for y in sorted(best["yearly"]): d = best["yearly"][y] ya = d["w"]/d["t"]*100 if d["t"]>0 else 0 tag = " ← CRASH" if y in [2020,2021,2022] else "" print(f" {y}: {d['t']:4d}t {ya:5.1f}% €{sum(d['pnls']):+.0f}{tag}") return results all_r = [] for asset in ["BTC", "ETH"]: for tf in ["15m", "1h"]: r = run_ultimate(asset, tf) for x in r: all_r.append({**x, "key": f"{asset}_{tf}"}) all_r.sort(key=lambda x: x["acc"], reverse=True) print(f"\n\n{'='*70}") print(f" TOP 10 GLOBALE") print(f"{'='*70}") for r in all_r[:10]: tag = "✅✅" if r["acc"] >= 80 else "✅" if r["acc"] >= 78 else "" print(f" {r['key']:.<10s} {r['name']:.<28s} {r['trades']:>5d} {r['acc']:>5.1f}% €{r['pnl']:>+8.0f} DD {r['dd']:.1f}% {r['worst']:>12s} {tag}")