PythagorasGoal/src/strategies/fade_base.py

"""Base condivisa per strategie mean-reversion con exit TP/SL/max_bars.

Tutte le strategie fade (MR02/MR03/MR07) generano Signal con metadata
{tp, sl, max_bars} e usano lo stesso backtest fedele: ingresso a close[i]
(eseguibile dal vivo), uscita su take-profit / stop-loss intrabar (high/low)
o time-limit, una posizione per volta (non-overlap), capitale composto,
fee+leva nette. Identico all'engine di scripts/analysis/strategy_research.py.

Le sottoclassi implementano solo generate_signals().
"""
from __future__ import annotations

import numpy as np
import pandas as pd

from src.strategies.base import Strategy, BacktestResult, YearlyStats, TF_MINUTES
from src.data.downloader import load_data
from src.fractal.indicators import rolling_hurst


def hurst_skip_mask(df: pd.DataFrame, hurst_max: float | None, window: int = 100) -> np.ndarray:
    """Loss-guard Hurst: maschera bool (True = SALTA il segnale) per regime PERSISTENTE/trending,
    dove la rolling-Hurst >= hurst_max. Le fade concentrano stop-loss e perdite proprio li'
    (diagnosi: stop-rate 43% per hurst>0.55 vs 21% anti-persistente). Filtrare hurst>=0.55
    DIMEZZA il DD del PORT06 (FULL 4.10%->2.39%) alzando lo Sharpe (validato 2026-06-02).
    CAUSALE: rolling_hurst usa solo i rendimenti fino a close[i]. hurst_max=None -> nessuno skip.
    Calcolata dalle SOLE close -> nessun feed dati esterno necessario (a differenza di DVOL)."""
    n = len(df)
    if hurst_max is None:
        return np.zeros(n, dtype=bool)
    h = rolling_hurst(df["close"].values.astype(float), window=window)
    return h >= hurst_max


def atr(df: pd.DataFrame, n: int = 14) -> np.ndarray:
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    pc = np.roll(c, 1); pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    return pd.Series(tr).rolling(n).mean().values


def trend_distance(df: pd.DataFrame, ema_long: int = 200) -> np.ndarray:
    """Distanza del close dalla EMA lunga, in multipli di ATR(14).

    Misura quanto il prezzo e' esteso rispetto al trend di fondo. Le fade
    falliscono quando si oppongono a un trend estremo (crolli/parabolic): il
    filtro `trend_max` salta i segnali con distanza > soglia. Riduce DD e alza
    l'accuratezza (validato OOS: scripts/analysis/risk_portfolio.py).
    """
    c = df["close"].values
    a = atr(df, 14)
    el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
    with np.errstate(divide="ignore", invalid="ignore"):
        return np.abs(c - el) / np.where(a == 0, np.nan, a)


class FadeStrategy(Strategy):
    """Strategy con backtest intrabar TP/SL/max_bars (exit guidati dai metadata)."""

    fee_rt = 0.001          # Deribit perp realistico (taker 0.05%/lato)
    leverage = 3.0
    position_size = 0.15
    initial_capital = 1000.0

    def backtest(self, asset: str, tf: str = "1h", hold: int = 3,
                 **params) -> BacktestResult | None:
        df = load_data(asset, tf)
        ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
        signals = self.generate_signals(df, ts, **params)
        if not signals:
            return None

        h, l, c = df["high"].values, df["low"].values, df["close"].values
        n = len(c)
        fee = self.fee_rt * self.leverage
        capital = peak = float(self.initial_capital)
        max_dd = 0.0
        total_bars = 0
        last_exit = -1
        yearly: dict[int, dict] = {}

        for sig in signals:
            i, d = sig.idx, sig.direction
            if i <= last_exit or i + 1 >= n:
                continue
            entry = c[i]
            tp, sl, mb = sig.metadata["tp"], sig.metadata["sl"], sig.metadata["max_bars"]
            exit_p = c[min(i + mb, n - 1)]
            j = min(i + mb, n - 1)
            for step in range(1, mb + 1):
                j = i + step
                if j >= n:
                    j = n - 1; exit_p = c[j]; break
                hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
                hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
                if hit_sl:                          # conservativo: SL prima del TP nello stesso bar
                    exit_p = sl; break
                if hit_tp:
                    exit_p = tp; break
                if step == mb:
                    exit_p = c[j]

            ret = (exit_p - entry) / entry * d * self.leverage - fee
            capital = max(capital + capital * self.position_size * ret, 10.0)
            if capital > peak:
                peak = capital
            max_dd = max(max_dd, (peak - capital) / peak)
            total_bars += (j - i)
            last_exit = j

            year = ts.iloc[i].year
            yr = yearly.setdefault(year, {"w": 0, "t": 0, "pnl": 0.0})
            yr["t"] += 1
            if ret > 0:
                yr["w"] += 1
            yr["pnl"] += ret * self.initial_capital

        all_t = sum(v["t"] for v in yearly.values())
        all_w = sum(v["w"] for v in yearly.values())
        if all_t == 0:
            return None

        yearly_stats = [YearlyStats(y, v["t"], v["w"], v["pnl"]) for y, v in sorted(yearly.items())]
        return BacktestResult(
            strategy_name=self.name, asset=asset, timeframe=tf, params=params,
            trades=all_t, wins=all_w, pnl=sum(v["pnl"] for v in yearly.values()),
            capital=capital, initial_capital=self.initial_capital,
            max_dd=max_dd * 100, time_in_market_pct=total_bars / n * 100,
            avg_trade_duration_h=total_bars / all_t * TF_MINUTES.get(tf, 60) / 60,
            years_active=len(yearly), yearly=yearly_stats,
        )