"""Strategia 11: Volatility compression → breakout. Approccio diverso: non predire la direzione direttamente. 1. Identifica momenti di COMPRESSIONE (Bollinger squeeze, ATR basso, bassa fractal dim) 2. Quando la volatilità ESPLODE dopo compressione, segui la direzione del breakout 3. Alta precisione perché il breakout DOPO compressione ha forte momentum Target: pochi trade molto precisi, con leva. """ 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 from src.fractal.indicators import volatility_ratio FEE_PCT = 0.001 LEVERAGE = 3 INITIAL_CAPITAL = 1000 def bollinger_bandwidth(close: np.ndarray, window: int = 20) -> np.ndarray: """Bandwidth = (upper - lower) / middle.""" result = np.full(len(close), np.nan) for i in range(window, len(close)): w = close[i - window : i] ma = np.mean(w) std = np.std(w) if ma > 0: result[i] = (2 * 2 * std) / ma return result def keltner_channel_ratio(close: np.ndarray, high: np.ndarray, low: np.ndarray, window: int = 20) -> np.ndarray: """Ratio of Bollinger to Keltner — squeeze when < 1.""" result = np.full(len(close), np.nan) for i in range(window, len(close)): w_c = close[i - window : i] w_h = high[i - window : i] w_l = low[i - window : i] ma = np.mean(w_c) bb_std = np.std(w_c) bb_upper = ma + 2 * bb_std bb_lower = ma - 2 * bb_std tr = np.maximum(w_h - w_l, np.maximum(np.abs(w_h - np.roll(w_c, 1)), np.abs(w_l - np.roll(w_c, 1)))) atr = np.mean(tr[1:]) kc_upper = ma + 1.5 * atr kc_lower = ma - 1.5 * atr kc_range = kc_upper - kc_lower bb_range = bb_upper - bb_lower if kc_range > 0: result[i] = bb_range / kc_range return result def detect_squeeze_release( close: np.ndarray, high: np.ndarray, low: np.ndarray, volume: np.ndarray, bb_window: int = 20, squeeze_threshold: float = 0.8, breakout_bars: int = 3, volume_mult: float = 1.5, ) -> list[dict]: """Detect squeeze → breakout events.""" bw = bollinger_bandwidth(close, bb_window) kcr = keltner_channel_ratio(close, high, low, bb_window) events = [] in_squeeze = False squeeze_start = 0 for i in range(bb_window + 1, len(close)): if np.isnan(kcr[i]): continue is_squeeze = kcr[i] < squeeze_threshold if is_squeeze and not in_squeeze: in_squeeze = True squeeze_start = i elif not is_squeeze and in_squeeze: in_squeeze = False squeeze_duration = i - squeeze_start if squeeze_duration < 5: continue # Check breakout direction using next few bars if i + breakout_bars >= len(close): continue breakout_ret = (close[i + breakout_bars - 1] - close[i - 1]) / close[i - 1] # Volume confirmation avg_vol = np.mean(volume[squeeze_start:i]) breakout_vol = np.mean(volume[i:i + breakout_bars]) vol_confirmed = breakout_vol > avg_vol * volume_mult if avg_vol > 0 else False # Momentum confirmation mom_3 = (close[i + 2] - close[i - 1]) / close[i - 1] if i + 2 < len(close) else 0 events.append({ "idx": i, "squeeze_duration": squeeze_duration, "breakout_ret": breakout_ret, "vol_confirmed": vol_confirmed, "mom_3": mom_3, "bb_expansion": bw[i] / bw[squeeze_start] if bw[squeeze_start] > 0 else 1, }) return events def run_squeeze_strategy(asset: str, tf: str = "1h"): print(f"\n{'#'*60}") print(f" {asset} {tf} — VOLATILITY SQUEEZE BREAKOUT") print(f"{'#'*60}") df = load_data(asset, tf) close = df["close"].values high = df["high"].values low = df["low"].values volume = df["volume"].values n = len(df) split_idx = int(n * 0.7) for bb_w in [14, 20, 30]: for sq_thr in [0.7, 0.8, 0.9]: for brk_bars in [3, 6]: events = detect_squeeze_release(close, high, low, volume, bb_window=bb_w, squeeze_threshold=sq_thr, breakout_bars=brk_bars, volume_mult=1.3) test_events = [e for e in events if e["idx"] >= split_idx] if len(test_events) < 10: continue # Strategy: follow breakout direction, with volume confirmation capital = float(INITIAL_CAPITAL) correct = 0 total = 0 for e in test_events: i = e["idx"] if i + brk_bars * 2 >= n: continue # First 1-bar direction as signal first_bar_ret = (close[i] - close[i - 1]) / close[i - 1] if abs(first_bar_ret) < 0.001: continue direction = "long" if first_bar_ret > 0 else "short" # Actual result after holding for brk_bars more actual_ret = (close[i + brk_bars - 1] - close[i - 1]) / close[i - 1] is_correct = (direction == "long" and actual_ret > 0) or (direction == "short" and actual_ret < 0) total += 1 if is_correct: correct += 1 trade_ret = actual_ret if direction == "long" else -actual_ret net = trade_ret * LEVERAGE - FEE_PCT * 2 * LEVERAGE capital += capital * 0.2 * net capital = max(capital, 0) # Enhanced: volume-confirmed only if total > 0: acc = correct / total * 100 ret = (capital - INITIAL_CAPITAL) / INITIAL_CAPITAL * 100 test_candles = n - split_idx test_years = test_candles / (24 * 365.25) ann = ((capital / INITIAL_CAPITAL) ** (1 / test_years) - 1) * 100 if test_years > 0 and capital > 0 else -100 if acc >= 55 and total >= 15: print(f" BBw={bb_w:2d} sqThr={sq_thr:.1f} brk={brk_bars}: trades={total:4d} acc={acc:.1f}% ret={ret:+.1f}% ann={ann:+.1f}%") # Volume-confirmed only cap_vc = float(INITIAL_CAPITAL) correct_vc = 0 total_vc = 0 for e in test_events: if not e["vol_confirmed"]: continue i = e["idx"] if i + brk_bars * 2 >= n: continue first_bar_ret = (close[i] - close[i - 1]) / close[i - 1] if abs(first_bar_ret) < 0.001: continue direction = "long" if first_bar_ret > 0 else "short" actual_ret = (close[i + brk_bars - 1] - close[i - 1]) / close[i - 1] is_correct = (direction == "long" and actual_ret > 0) or (direction == "short" and actual_ret < 0) total_vc += 1 if is_correct: correct_vc += 1 trade_ret = actual_ret if direction == "long" else -actual_ret net = trade_ret * LEVERAGE - FEE_PCT * 2 * LEVERAGE cap_vc += cap_vc * 0.2 * net cap_vc = max(cap_vc, 0) if total_vc >= 10: acc_vc = correct_vc / total_vc * 100 ret_vc = (cap_vc - INITIAL_CAPITAL) / INITIAL_CAPITAL * 100 ann_vc = ((cap_vc / INITIAL_CAPITAL) ** (1 / (test_candles/(24*365.25))) - 1) * 100 if cap_vc > 0 else -100 if acc_vc >= 55: print(f" +VOL BBw={bb_w:2d} sqThr={sq_thr:.1f} brk={brk_bars}: trades={total_vc:4d} acc={acc_vc:.1f}% ret={ret_vc:+.1f}% ann={ann_vc:+.1f}%") for asset in ["BTC", "ETH"]: for tf in ["1h", "15m"]: run_squeeze_strategy(asset, tf)