PythagorasGoal/Old/scripts/strategies/DIP01_dip_reversion.py

"""DIP01 — Dip-Buy Z-Score Reversion (long-only).

Variante robusta e ONESTA della famiglia mean-reversion: compra SOLO i dip
(close a z<=-z_in deviazioni sotto la media mobile) e prende profitto al rientro
verso la media. Niente short: nel campione 2018-2026 shortare cripto perde OOS
sistematicamente (vedi scripts/analysis/honest_final.py).

Logica:
    1. z-score = (close - SMA(n)) / STD(n)
    2. ENTRY long quando z attraversa al ribasso -z_in (capitolazione)
    3. EXIT: take-profit alla media mobile, stop-loss a sl_atr*ATR sotto l'entry,
       o time-limit max_bars
    4. ingresso a close[i] (eseguibile dal vivo, nessun look-ahead)

Validazione (netto, fee 0.10% RT Deribit, leva 3x, OOS = ultimo 30%):
    BTC 1h: FULL +298% / OOS +59% / DD 23% / 7-9 anni positivi
    ETH 1h: FULL +190% / OOS +224% / DD 54%
    SOL 1h: FULL  +50% / OOS +13% / DD 25%
    Regge lo sweep fee fino a 0.20% RT (BTC OOS +45% anche a 0.20%).
    Robusto su BTC/ETH/SOL (asset major); sugli alt molto parabolici (DOGE/BNB)
    non ha edge -> usare solo su BTC/ETH/SOL.

Compatibile con StrategyWorker: ogni Signal porta tp/sl/max_bars in metadata.
"""
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, BacktestResult, YearlyStats, TF_MINUTES
from src.data.downloader import load_data


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


class DipReversion(Strategy):
    name = "DIP01_dip_reversion"
    description = "Long-only dip-buy z-score reversion, TP alla media"
    default_assets = ["BTC", "ETH", "SOL"]
    default_timeframes = ["1h"]
    fee_rt = 0.001
    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
        n = params.get("n", 50)
        z_in = params.get("z_in", 2.5)
        sl_atr = params.get("sl_atr", 2.5)
        max_bars = params.get("max_bars", 24)

        ma = pd.Series(c).rolling(n).mean().values
        sd = pd.Series(c).rolling(n).std().values
        a = _atr(df, 14)
        z = (c - ma) / np.where(sd == 0, np.nan, sd)

        signals: list[Signal] = []
        for i in range(n + 14, len(c)):
            if np.isnan(z[i]) or np.isnan(a[i]):
                continue
            if z[i] <= -z_in and z[i - 1] > -z_in:
                signals.append(Signal(
                    idx=i, direction=1, entry_price=c[i],
                    metadata={"tp": float(ma[i]), "sl": float(c[i] - sl_atr * a[i]),
                              "max_bars": max_bars},
                ))
        return signals

    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:
                    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,
        )


if __name__ == "__main__":
    strat = DipReversion()
    print(f"{'=' * 100}")
    print(f"  DIP01 DIP-BUY REVERSION — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x")
    print(f"{'=' * 100}")
    for asset in ["BTC", "ETH", "SOL"]:
        r = strat.backtest(asset, "1h", n=50, z_in=2.5, sl_atr=2.5, max_bars=24)
        if r:
            r.strategy_name = f"DIP01 {asset} 1h"
            r.print_summary()