PythagorasGoal/scripts/strategies/PD01_price_volume_divergence.py

"""PD01 — Price-Volume Divergence Squeeze.

Estende SQ02 con volume TREND come filtro:
  - Breakout UP con volume CRESCENTE (ultimi 3 bar vs media squeeze) → ENTRA
  - Breakout UP con volume CALANTE → SALTA (divergenza bearish)
  - Viceversa per short

Logica anti-fakeout:
  1. Squeeze rilascio (come SQ02)
  2. Anti-fakeout candela (come SQ02)
  3. Volume al breakout > media squeeze (come SQ02)
  4. NUOVO: volume trending UP nelle ultime 3 barre prima del breakout

Parametri semplici, nessun overfitting.
"""
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, detect_squeezes


class PriceVolumeDivergence(Strategy):
    name = "PD01_price_vol_div"
    description = "Squeeze + antifakeout + volume trend confirmation"
    default_assets = ["BTC", "ETH"]
    default_timeframes = ["15m", "1h"]
    fee_rt = 0.002
    leverage = 3.0
    position_size = 0.15
    initial_capital = 1000.0

    def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
                         **params) -> list[Signal]:
        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)
        retrace_limit = params.get("retrace_limit", 0.6)
        vol_mult = params.get("vol_multiplier", 1.3)
        vol_trend_bars = params.get("vol_trend_bars", 3)  # barre per trend volume

        kcr = keltner_ratio(c, h, l, bb_w)
        events = detect_squeezes(c, h, l, kcr, sq_thr)

        signals = []
        for ev in events:
            i = ev["idx"]
            if i < vol_trend_bars + 1 or i >= n:
                continue

            # Direzione breakout
            first_ret = (c[i] - c[i - 1]) / c[i - 1] if c[i - 1] > 0 else 0
            if abs(first_ret) < 0.001:
                continue
            direction = 1 if first_ret > 0 else -1

            # Anti-fakeout
            br = h[i] - l[i]
            if br > 0:
                if direction == 1 and (h[i] - c[i]) / br > retrace_limit:
                    continue
                elif direction == -1 and (c[i] - l[i]) / br > retrace_limit:
                    continue

            # Volume al breakout > media squeeze
            sq_start = ev["sq_start"]
            avg_sq_v = np.mean(v[sq_start:i])
            if avg_sq_v <= 0 or v[i] <= avg_sq_v * vol_mult:
                continue

            # Volume TREND: slope delle ultime vol_trend_bars barre
            # Usa regressione lineare semplice (rank correlation del volume)
            recent_v = v[i - vol_trend_bars:i + 1]  # include breakout bar
            if len(recent_v) < vol_trend_bars:
                continue
            # slope: media seconda metà vs prima metà
            mid = len(recent_v) // 2
            v_early = np.mean(recent_v[:mid])
            v_late = np.mean(recent_v[mid:])
            vol_trending_up = v_late > v_early
            vol_trending_down = v_early > v_late

            # Concordanza: long richiede volume trending up, short trending down
            if direction == 1 and not vol_trending_up:
                continue
            if direction == -1 and not vol_trending_down:
                continue

            signals.append(Signal(
                idx=i,
                direction=direction,
                entry_price=c[i - 1],
                metadata={
                    "dur": ev["dur"],
                    "vol_ratio": v[i] / avg_sq_v if avg_sq_v > 0 else 0,
                    "vol_trend": v_late / v_early if v_early > 0 else 1,
                },
            ))
        return signals


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

    configs = [
        {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6,
         "vol_multiplier": 1.3, "vol_trend_bars": 3},
        {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6,
         "vol_multiplier": 1.2, "vol_trend_bars": 3},
        {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6,
         "vol_multiplier": 1.3, "vol_trend_bars": 5},
        {"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.5,
         "vol_multiplier": 1.3, "vol_trend_bars": 3},
        {"bb_window": 14, "sq_threshold": 0.75, "retrace_limit": 0.6,
         "vol_multiplier": 1.3, "vol_trend_bars": 3},
        {"bb_window": 20, "sq_threshold": 0.8, "retrace_limit": 0.6,
         "vol_multiplier": 1.3, "vol_trend_bars": 3},
    ]

    all_results = []
    for cfg in configs:
        for asset in ["BTC", "ETH"]:
            for tf in ["15m", "1h"]:
                for hold in [3, 6]:
                    r = strategy.backtest(asset, tf, hold=hold, **cfg)
                    if r and r.trades >= 20:
                        lbl = (f"PD01 vtb={cfg['vol_trend_bars']} "
                               f"vm={cfg['vol_multiplier']} "
                               f"sq={cfg['sq_threshold']} h={hold}")
                        r.strategy_name = lbl
                        all_results.append(r)

    all_results.sort(key=lambda r: r.accuracy, reverse=True)

    print(f"\n{'=' * 130}")
    print("  PD01 PRICE-VOLUME DIVERGENCE — TOP 20")
    print(f"{'=' * 130}")
    print(f"  {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
          f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
          f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}")
    print(f"  {'─' * 120}")
    for r in all_results[:20]:
        r.print_summary()

    if all_results:
        all_results[0].print_yearly()

    print(f"\n  BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, €5.23/day, 9 anni")
    print(f"  BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")