PythagorasGoal/scripts/waste/REF_11_volatility_breakout.py

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