"""AD01 — Adaptive Squeeze Threshold. Problema SQ02: sq_threshold fisso (0.8) non si adatta al regime di volatilità. Soluzione: threshold adattivo basato su volatilità recente. Logica: - Calcola volatilità rolling (std dei rendimenti su finestra 100 barre) - Confronta con percentile storico (rolling 500 barre) - Alta vol (>70° percentile) → soglia BASSA (0.65) — squeeze più "lenti" - Bassa vol (<30° percentile) → soglia ALTA (0.90) — squeeze "stretti" - Vol media → soglia standard (0.80) Razionale: in mercati calmi, il BB si stringe molto → sq_threshold alto cattura segnali migliori. In mercati volatili, bastano squeeze minori per essere significativi. Anti-overfitting: solo 3 parametri (low_thr, mid_thr, high_thr), logica deterministica. Eredita antifakeout + volume da SQ02. """ 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, BacktestResult, YearlyStats, TF_MINUTES from src.strategies.indicators import keltner_ratio, ema from src.data.downloader import load_data def _adaptive_sq_threshold(close: np.ndarray, vol_window: int = 100, regime_window: int = 500, low_thr: float = 0.65, mid_thr: float = 0.80, high_thr: float = 0.90) -> np.ndarray: """Calcola sq_threshold adattivo per ogni barra.""" n = len(close) lr = np.diff(np.log(np.where(close <= 0, 1e-10, close))) vol = np.full(n, np.nan) for i in range(vol_window, n): vol[i] = np.std(lr[i - vol_window:i]) # Percentile rolling della volatilità thresh = np.full(n, mid_thr) for i in range(regime_window, n): if np.isnan(vol[i]): continue hist = vol[i - regime_window:i] hist = hist[~np.isnan(hist)] if len(hist) < 10: continue p30 = np.percentile(hist, 30) p70 = np.percentile(hist, 70) if vol[i] < p30: thresh[i] = high_thr # vol bassa → soglia alta elif vol[i] > p70: thresh[i] = low_thr # vol alta → soglia bassa else: thresh[i] = mid_thr return thresh def _detect_adaptive_squeezes(close, high, low, kcr, adaptive_thr, min_dur: int = 5) -> list[dict]: """Squeeze con threshold adattivo per ogni barra.""" events = [] in_sq = False sq_start = 0 for i in range(1, len(close)): if np.isnan(kcr[i]) or np.isnan(adaptive_thr[i]): continue thr = adaptive_thr[i] is_sq = kcr[i] < thr 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 < min_dur: continue events.append({ "idx": i, "dur": dur, "sq_start": sq_start, "kcr_at_release": kcr[i], "thr_used": adaptive_thr[i], }) return events class AdaptiveSqueeze(Strategy): name = "AD01_adaptive_squeeze" description = "Squeeze con threshold adattivo a regime volatilità" default_assets = ["BTC", "ETH"] default_timeframes = ["15m", "1h"] fee_rt = 0.002 leverage = 3.0 position_size = 0.15 initial_capital = 1000.0 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) bb_w = params.get("bb_window", 14) low_thr = params.get("low_thr", 0.65) mid_thr = params.get("mid_thr", 0.80) high_thr = params.get("high_thr", 0.90) retrace_limit = params.get("retrace_limit", 0.6) vol_mult = params.get("vol_multiplier", 1.3) use_vol = params.get("use_vol", True) vol_window = params.get("vol_window", 100) regime_window = params.get("regime_window", 500) kcr = keltner_ratio(c, h, l, bb_w) adaptive_thr = _adaptive_sq_threshold( c, vol_window, regime_window, low_thr, mid_thr, high_thr ) events = _detect_adaptive_squeezes(c, h, l, kcr, adaptive_thr) 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 direction = 1 if first_ret > 0 else -1 # Anti-fakeout br = h[i] - l[i] if br > 0: if direction == 1 and (h[i] - c[i]) / br > retrace_limit: continue elif direction == -1 and (c[i] - l[i]) / br > retrace_limit: continue # Volume confirm if use_vol: sq_start = ev["sq_start"] avg_sq_v = np.mean(v[sq_start:i]) if avg_sq_v > 0 and v[i] <= avg_sq_v * vol_mult: continue signals.append(Signal( idx=i, direction=direction, entry_price=c[i - 1], metadata={ "dur": ev["dur"], "thr_used": ev.get("thr_used", mid_thr), }, )) return signals if __name__ == "__main__": strategy = AdaptiveSqueeze() configs = [ # low_thr, mid_thr, high_thr, use_vol {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": True}, {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": False}, {"low_thr": 0.60, "mid_thr": 0.78, "high_thr": 0.92, "use_vol": True}, {"low_thr": 0.70, "mid_thr": 0.82, "high_thr": 0.90, "use_vol": True}, {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.95, "use_vol": True}, {"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": True, "vol_multiplier": 1.2}, ] all_results = [] for cfg in configs: for asset in ["BTC", "ETH"]: for tf in ["15m", "1h"]: for hold in [3, 6]: r = strategy.backtest(asset, tf, hold=hold, **cfg) if r and r.trades >= 20: lbl = (f"AD01 lt={cfg['low_thr']} ht={cfg['high_thr']} " f"v={cfg['use_vol']} h={hold}") r.strategy_name = lbl all_results.append(r) all_results.sort(key=lambda r: r.accuracy, reverse=True) print(f"\n{'=' * 130}") print(" AD01 ADAPTIVE SQUEEZE THRESHOLD — TOP 20") print(f"{'=' * 130}") print(f" {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} " f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} " f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}") print(f" {'─' * 120}") for r in all_results[:20]: r.print_summary() if all_results: all_results[0].print_yearly() print(f"\n BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, €5.23/day, 9 anni") print(f" BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")