PythagorasGoal/scripts/waste/W28_squeeze_mr.py

"""HY01 — Squeeze + Mean Reversion Ibrida.

Insight: durante lo squeeze (bassa volatilità), il prezzo mean-reverte
DENTRO il range compresso. Autocorrelazione negativa a 15m conferma.
Invece di aspettare il BREAKOUT, tradi la MEAN REVERSION dentro lo squeeze.

Completamente diverso da SQ01-SQ04 che aspettano il RILASCIO.

IN:
    - OHLCV DataFrame
    - Parametri: bb_window, sq_threshold, rsi_period, rsi_levels,
                 vol_filter, bb_touch (prezzo tocca banda Bollinger)

OUT:
    - Signal: long quando RSI oversold DURANTE squeeze, short quando overbought
    - BacktestResult

Logica:
    1. Verifica che siamo IN squeeze (BB dentro KC)
    2. Prezzo tocca banda inferiore BB → LONG (tornerà alla media)
    3. Prezzo tocca banda superiore BB → SHORT (tornerà alla media)
    4. Conferma RSI: deve essere estremo nella direzione
    5. Hold corto (2-3 barre) — target: ritorno alla media
    6. Stop: se prezzo rompe lo squeeze → chiudi subito
"""
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


def rsi(close, period=14):
    delta = np.diff(close)
    gain = np.where(delta > 0, delta, 0)
    loss = np.where(delta < 0, -delta, 0)
    result = np.full(len(close), 50.0)
    if len(gain) < period:
        return result
    ag = np.mean(gain[:period])
    al = np.mean(loss[:period])
    for i in range(period, len(delta)):
        ag = (ag * (period - 1) + gain[i]) / period
        al = (al * (period - 1) + loss[i]) / period
        result[i + 1] = 100 if al == 0 else 100 - 100 / (1 + ag / al)
    return result


def bollinger(close, window=14):
    n = len(close)
    upper = np.full(n, np.nan)
    lower = np.full(n, np.nan)
    mid = np.full(n, np.nan)
    for i in range(window, n):
        wc = close[i - window:i]
        m = np.mean(wc)
        s = np.std(wc)
        mid[i] = m
        upper[i] = m + 2 * s
        lower[i] = m - 2 * s
    return upper, mid, lower


class SqueezeMeanReversion(Strategy):
    name = "HY01_squeeze_mr"
    description = "Mean reversion DENTRO lo squeeze — fade estremi in range compresso"
    default_assets = ["BTC", "ETH"]
    default_timeframes = ["15m", "1h"]
    fee_rt = 0.002

    def generate_signals(self, df, ts, **params):
        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)
        sq_thr = params.get("sq_threshold", 0.8)
        rsi_period = params.get("rsi_period", 14)
        rsi_low = params.get("rsi_oversold", 30)
        rsi_high = params.get("rsi_overbought", 70)
        use_bb_touch = params.get("bb_touch", True)
        cooldown = params.get("cooldown", 3)

        kcr = keltner_ratio(c, h, l, bb_w)
        rsi_vals = rsi(c, rsi_period)
        bb_upper, bb_mid, bb_lower = bollinger(c, bb_w)

        signals = []
        last_idx = -cooldown

        for i in range(bb_w + 1, n):
            if i - last_idx < cooldown:
                continue
            if np.isnan(kcr[i]) or np.isnan(bb_lower[i]):
                continue

            # Must be IN squeeze
            if kcr[i] >= sq_thr:
                continue

            direction = 0

            if use_bb_touch:
                # Prezzo tocca/rompe BB lower → long (mean reversion up)
                if c[i] <= bb_lower[i] and rsi_vals[i] < rsi_low:
                    direction = 1
                # Prezzo tocca/rompe BB upper → short (mean reversion down)
                elif c[i] >= bb_upper[i] and rsi_vals[i] > rsi_high:
                    direction = -1
            else:
                # Solo RSI
                if rsi_vals[i] < rsi_low:
                    direction = 1
                elif rsi_vals[i] > rsi_high:
                    direction = -1

            if direction == 0:
                continue

            signals.append(Signal(
                idx=i, direction=direction, entry_price=c[i],
                metadata={
                    "rsi": float(rsi_vals[i]),
                    "kcr": float(kcr[i]),
                    "bb_pos": "lower" if direction == 1 else "upper",
                },
            ))
            last_idx = i

        return signals


if __name__ == "__main__":
    strategy = SqueezeMeanReversion()

    configs = [
        ("bb+rsi30/70", {"bb_touch": True, "rsi_oversold": 30, "rsi_overbought": 70}),
        ("bb+rsi25/75", {"bb_touch": True, "rsi_oversold": 25, "rsi_overbought": 75}),
        ("bb+rsi35/65", {"bb_touch": True, "rsi_oversold": 35, "rsi_overbought": 65}),
        ("rsi30/70 only", {"bb_touch": False, "rsi_oversold": 30, "rsi_overbought": 70}),
        ("rsi25/75 only", {"bb_touch": False, "rsi_oversold": 25, "rsi_overbought": 75}),
        ("sq<0.7 bb+rsi30", {"bb_touch": True, "sq_threshold": 0.7, "rsi_oversold": 30, "rsi_overbought": 70}),
        ("sq<0.9 bb+rsi30", {"bb_touch": True, "sq_threshold": 0.9, "rsi_oversold": 30, "rsi_overbought": 70}),
        ("sq<0.9 rsi35/65", {"bb_touch": False, "sq_threshold": 0.9, "rsi_oversold": 35, "rsi_overbought": 65}),
    ]

    all_results = []
    for label, params in configs:
        for asset in ["BTC", "ETH"]:
            for tf in ["15m", "1h"]:
                for hold in [2, 3, 4]:
                    r = strategy.backtest(asset, tf, hold=hold, **params)
                    if r and r.trades >= 30:
                        r.strategy_name = f"HY01 {label} h={hold}"
                        all_results.append(r)

    all_results.sort(key=lambda r: r.accuracy, reverse=True)
    print(f"\n{'=' * 130}")
    print(f"  HY01 SQUEEZE MEAN REVERSION — TOP 25")
    print(f"{'=' * 130}")
    for r in all_results[:25]:
        r.print_summary()
    if all_results:
        all_results[0].print_yearly()