PythagorasGoal/scripts/waste/W26_mean_reversion_rsi.py

"""MR01 — Mean Reversion da estremi RSI.

Approccio opposto allo squeeze: quando il prezzo va troppo lontano troppo veloce,
scommetti che torni indietro. Autocorrelazione lag-1 negativa (-0.21 BTC, -0.35 ETH)
conferma che il mercato a 15m è mean-reverting.

IN:
    - OHLCV DataFrame
    - Parametri: rsi_period, rsi_oversold, rsi_overbought, hold_bars,
                 volume_filter (volume > N× media), atr_filter (move > N×ATR)

OUT:
    - Signal: long quando RSI < oversold, short quando RSI > overbought
    - BacktestResult con metriche

Logica:
    1. RSI scende sotto soglia oversold → LONG (prezzo tornerà su)
    2. RSI sale sopra soglia overbought → SHORT (prezzo tornerà giù)
    3. Filtro opzionale: volume spike conferma l'eccesso
    4. Filtro opzionale: move recente > N×ATR (eccesso di prezzo)
    5. Hold fisso, poi chiudi
"""
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


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


class MeanReversionRSI(Strategy):
    name = "MR01_mean_reversion_rsi"
    description = "Mean reversion da estremi RSI — fade eccessi direzionali"
    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)

        rsi_period = params.get("rsi_period", 14)
        oversold = params.get("rsi_oversold", 25)
        overbought = params.get("rsi_overbought", 75)
        use_vol_filter = params.get("vol_filter", False)
        use_atr_filter = params.get("atr_filter", False)
        cooldown = params.get("cooldown", 4)

        rsi_vals = rsi(c, rsi_period)

        # Volume media rolling
        vol_ma = np.full(n, np.nan)
        for i in range(20, n):
            vol_ma[i] = np.mean(v[i - 20:i])

        # ATR
        tr = np.maximum(h[1:] - l[1:],
                        np.maximum(np.abs(h[1:] - c[:-1]), np.abs(l[1:] - c[:-1])))
        atr_vals = np.full(n, np.nan)
        for i in range(15, len(tr)):
            atr_vals[i + 1] = np.mean(tr[i - 14:i])

        signals = []
        last_signal_idx = -cooldown

        for i in range(20, n):
            if i - last_signal_idx < cooldown:
                continue

            direction = 0
            if rsi_vals[i] < oversold:
                direction = 1  # oversold → long
            elif rsi_vals[i] > overbought:
                direction = -1  # overbought → short

            if direction == 0:
                continue

            # Volume filter
            if use_vol_filter and not np.isnan(vol_ma[i]):
                if v[i] < vol_ma[i] * 1.5:
                    continue

            # ATR filter: il move recente deve essere > 1.5× ATR
            if use_atr_filter and not np.isnan(atr_vals[i]):
                recent_move = abs(c[i] - c[max(0, i - 3)]) / c[max(0, i - 3)]
                if recent_move < atr_vals[i] / c[i] * 1.5:
                    continue

            signals.append(Signal(
                idx=i, direction=direction, entry_price=c[i],
                metadata={"rsi": float(rsi_vals[i])},
            ))
            last_signal_idx = i

        return signals


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

    configs = [
        ("RSI25/75", {}),
        ("RSI20/80", {"rsi_oversold": 20, "rsi_overbought": 80}),
        ("RSI25/75+vol", {"vol_filter": True}),
        ("RSI20/80+vol", {"rsi_oversold": 20, "rsi_overbought": 80, "vol_filter": True}),
        ("RSI25/75+atr", {"atr_filter": True}),
        ("RSI20/80+vol+atr", {"rsi_oversold": 20, "rsi_overbought": 80, "vol_filter": True, "atr_filter": True}),
    ]

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

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