"""SQ04 — Ultimate Squeeze — combinazione incrementale di tutti i filtri. Testa combinazioni di filtri (antifake, long_sq, timing, cross-asset, correlation, volume, trend alignment, volatility regime) e classifica per accuracy. IN: - OHLCV DataFrame (primario + secondario) - Parametri: bb_window, sq_threshold, lista filtri da attivare OUT: - BacktestResult per ogni combinazione di filtri - Classifica globale Risultati tipici: BTC 15m antifake+corr: 81.6% acc (ma concentrato 2018) BTC 15m antifake+vol: 79.7% acc, 1250 trades — robusto ETH 1h antifake+corr: 80.7% acc (solo 2018) """ from __future__ import annotations import sys sys.path.insert(0, ".") import numpy as np import pandas as pd from src.strategies.base import Strategy, Signal from src.strategies.indicators import ( keltner_ratio, detect_squeezes, ema, rv_annualized, rolling_correlation, ) from src.data.downloader import load_data class SqueezeUltimate(Strategy): name = "SQ04_ultimate" description = "Ultimate squeeze — tutti i filtri combinabili" default_assets = ["BTC", "ETH"] default_timeframes = ["15m", "1h"] FILTER_PRESETS = { "antifake+vol": ["antifake", "vol_confirm"], "antifake+corr": ["antifake", "corr_high"], "af+long+corr+trend": ["antifake", "long_sq", "corr_high", "trend_align"], "ALL": ["antifake", "long_sq", "cross", "timing", "corr_high", "vol_confirm", "trend_align", "low_rv"], } def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex, **params) -> list[Signal]: c = df["close"].values h = df["high"].values l = df["low"].values v = df["volume"].values n = len(c) asset = params.get("asset", "BTC") tf = params.get("tf", "15m") filters = params.get("filters", ["antifake", "vol_confirm"]) kcr = keltner_ratio(c, h, l, 14) events = detect_squeezes(c, h, l, kcr) secondary = "ETH" if asset == "BTC" else "BTC" df2 = load_data(secondary, tf) c2 = df2["close"].values kcr2 = keltner_ratio(c2, df2["high"].values, df2["low"].values, 14) ts2 = df2["timestamp"].values ema_50 = ema(c, 50) rv_48 = rv_annualized(c, 48) corr = rolling_correlation(c, c2) signals = [] for ev in events: i = ev["idx"] if i < 1 or i >= n: continue 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 f in filters: if f == "antifake": br = h[i] - l[i] if br > 0: if c[i] > c[i-1] and (h[i] - c[i]) / br > 0.6: skip = True elif c[i] <= c[i-1] and (c[i] - l[i]) / br > 0.6: skip = True elif f == "long_sq": if ev["dur"] < 10: skip = True elif f == "timing": if ts.iloc[i].hour < 4 or ts.iloc[i].hour > 16: skip = True elif f == "cross": i2 = np.searchsorted(ts2, ts.values[i].astype("int64") // 10**6) i2 = min(i2, len(kcr2) - 1) if not any(not np.isnan(kcr2[j]) and kcr2[j] < 0.85 for j in range(max(0, i2 - 10), i2 + 1)): skip = True elif f == "corr_high": if np.isnan(corr[i]) or abs(corr[i]) < 0.6: skip = True elif f == "vol_confirm": avg_v = np.mean(v[ev["sq_start"]:i]) if avg_v > 0 and v[i] <= avg_v * 1.3: skip = True elif f == "trend_align": if not np.isnan(ema_50[i]): if first_ret > 0 and c[i] < ema_50[i]: skip = True elif first_ret < 0 and c[i] > ema_50[i]: skip = True elif f == "low_rv": if not np.isnan(rv_48[i]) and rv_48[i] >= 1.5: skip = True if skip: break if skip: continue signals.append(Signal( idx=i, direction=1 if first_ret > 0 else -1, entry_price=c[i - 1], metadata={"dur": ev["dur"], "filters": filters}, )) return signals def backtest(self, asset: str, tf: str, hold: int = 3, **params): params.setdefault("asset", asset) params.setdefault("tf", tf) df = load_data(asset, tf) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) signals = self.generate_signals(df, ts, **params) # Usa il backtest della base ma passando i segnali già generati from src.strategies.base import BacktestResult, YearlyStats, TF_MINUTES c = df["close"].values n = len(c) yearly: dict[int, dict] = {} capital = float(self.initial_capital) peak = capital max_dd = 0.0 total_bars = 0 for sig in signals: i = sig.idx if i + hold >= n or i < 1: continue entry = sig.entry_price exit_price = c[min(i + hold - 1, n - 1)] actual = (exit_price - entry) / entry * sig.direction net = actual * self.leverage - self.fee_rt * self.leverage capital += capital * self.position_size * net capital = max(capital, 10) if capital > peak: peak = capital dd = (peak - capital) / peak max_dd = max(max_dd, dd) total_bars += hold year = ts.iloc[i].year if year not in yearly: yearly[year] = {"w": 0, "t": 0, "pnl": 0.0} yearly[year]["t"] += 1 if actual > 0: yearly[year]["w"] += 1 yearly[year]["pnl"] += net * self.initial_capital all_t = sum(d["t"] for d in yearly.values()) all_w = sum(d["w"] for d in yearly.values()) if all_t == 0: return None yearly_stats = [YearlyStats(y, d["t"], d["w"], d["pnl"]) for y, d in sorted(yearly.items())] return BacktestResult( strategy_name=self.name, asset=asset, timeframe=tf, params=params, trades=all_t, wins=all_w, pnl=sum(d["pnl"] for d in yearly.values()), capital=capital, initial_capital=self.initial_capital, max_dd=max_dd * 100, time_in_market_pct=total_bars / n * 100, avg_trade_duration_h=hold * TF_MINUTES.get(tf, 60) / 60, years_active=len(yearly), yearly=yearly_stats, ) def report_all_presets(self): """Esegue tutte le combinazioni preset × asset × tf.""" all_results = [] for preset_name, filter_list in self.FILTER_PRESETS.items(): for asset in self.default_assets: for tf in self.default_timeframes: r = self.backtest(asset, tf, filters=filter_list) if r and r.trades >= 20: r.strategy_name = f"SQ04 {preset_name}" all_results.append(r) all_results.sort(key=lambda r: r.accuracy, reverse=True) print(f"\n{'=' * 120}") print(f" SQ04 ULTIMATE — TUTTI I PRESET") print(f"{'=' * 120}") for r in all_results: r.print_summary() return all_results if __name__ == "__main__": strategy = SqueezeUltimate() strategy.report_all_presets()