PythagorasGoal/Old/scripts/waste/W27_volume_spike.py

"""VO01 — Volume Spike Reversal.

Quando il volume esplode (>3× media) con un forte move direzionale,
il mercato è in eccesso → fade il move (mean reversion).

Diverso dallo squeeze: non cerca compressione, cerca ECCESSO.
Il volume spike indica panico/euforia → reversal probabile.

IN:
    - OHLCV DataFrame
    - Parametri: vol_mult (3), move_threshold (0.005), hold

OUT:
    - Signal: fade la direzione del volume spike
    - BacktestResult

Logica:
    1. Volume > vol_mult × media 20 periodi
    2. Move nella candela > move_threshold (0.5%)
    3. Direzione: opposta al move (mean reversion)
    4. Filtro: non entrare se già in trend forte (EMA slope)
"""
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


class VolumeSpikeReversal(Strategy):
    name = "VO01_vol_spike_reversal"
    description = "Volume spike reversal — fade eccessi di volume/prezzo"
    default_assets = ["BTC", "ETH"]
    default_timeframes = ["15m", "1h"]
    fee_rt = 0.002

    def generate_signals(self, df, ts, **params):
        c = df["close"].values
        o = df["open"].values
        h = df["high"].values
        l = df["low"].values
        v = df["volume"].values
        n = len(c)

        vol_mult = params.get("vol_mult", 3.0)
        move_thr = params.get("move_threshold", 0.005)
        use_trend_filter = params.get("trend_filter", False)
        cooldown = params.get("cooldown", 4)

        # 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])

        # EMA per trend filter
        ema_20 = np.full(n, np.nan)
        k = 2 / 21
        ema_20[19] = np.mean(c[:20])
        for i in range(20, n):
            ema_20[i] = c[i] * k + ema_20[i - 1] * (1 - k)

        signals = []
        last_idx = -cooldown

        for i in range(21, n):
            if i - last_idx < cooldown:
                continue
            if np.isnan(vol_ma[i]):
                continue

            # Volume spike
            if v[i] < vol_ma[i] * vol_mult:
                continue

            # Price move
            move = (c[i] - o[i]) / o[i] if o[i] > 0 else 0
            if abs(move) < move_thr:
                continue

            # Fade: opposto al move
            direction = -1 if move > 0 else 1

            # Trend filter: non fare mean reversion contro trend forte
            if use_trend_filter and not np.isnan(ema_20[i]):
                ema_slope = (ema_20[i] - ema_20[max(0, i - 5)]) / ema_20[max(0, i - 5)]
                if direction == -1 and ema_slope > 0.005:
                    continue
                if direction == 1 and ema_slope < -0.005:
                    continue

            signals.append(Signal(
                idx=i, direction=direction, entry_price=c[i],
                metadata={"vol_ratio": float(v[i] / vol_ma[i]), "move_pct": round(move * 100, 3)},
            ))
            last_idx = i

        return signals


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

    configs = [
        ("v3x m0.5%", {"vol_mult": 3.0, "move_threshold": 0.005}),
        ("v3x m1%", {"vol_mult": 3.0, "move_threshold": 0.01}),
        ("v4x m0.5%", {"vol_mult": 4.0, "move_threshold": 0.005}),
        ("v4x m1%", {"vol_mult": 4.0, "move_threshold": 0.01}),
        ("v3x m0.5%+tf", {"vol_mult": 3.0, "move_threshold": 0.005, "trend_filter": True}),
        ("v3x m1%+tf", {"vol_mult": 3.0, "move_threshold": 0.01, "trend_filter": True}),
        ("v5x m1%", {"vol_mult": 5.0, "move_threshold": 0.01}),
        ("v5x m1%+tf", {"vol_mult": 5.0, "move_threshold": 0.01, "trend_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"VO01 {label} h={hold}"
                        all_results.append(r)

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