PythagorasGoal/Old/scripts/analysis/strategy_research.py

"""Ricerca strategie fee-aware, OOS, oltre la famiglia squeeze.

Lezioni apprese (squeeze breakout = nessun edge):
  - le FEE sono vincolo di prim'ordine -> default fee realistica Deribit 0.10% RT
    (taker 0.05%/lato, maker ~0%); poche operazioni meglio di molte
  - i breakout RIENTRANO -> si esplora mean-reversion, non continuation
  - ogni numero e' NETTO dopo fee+leva, su finestra held-out + per anno

Engine realistico: ingresso a close[i] (eseguibile), uscita su TP/SL intrabar
(high/low) o time-limit, una posizione per volta (non-overlap), capitale composto.
"""
from __future__ import annotations

import sys
from pathlib import Path

import numpy as np
import pandas as pd

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from src.data.downloader import load_data

FEE_RT = 0.001          # Deribit perp realistico: taker 0.05%/lato
LEV = 3.0
POS = 0.15
OOS_FRAC = 0.30
BARS_PER_YEAR = {"15m": 35040, "1h": 8760, "4h": 2190, "1d": 365}


# ----------------------------- dati -----------------------------
def get_df(asset: str, tf: str) -> pd.DataFrame:
    """tf nativo (15m,1h) o resample da 1h (4h,1d)."""
    if tf in ("15m", "1h"):
        return load_data(asset, tf).reset_index(drop=True)
    base = load_data(asset, "1h").copy()
    base["dt"] = pd.to_datetime(base["timestamp"], unit="ms", utc=True)
    base = base.set_index("dt")
    rule = {"4h": "4h", "1d": "1D"}[tf]
    agg = base.resample(rule).agg(
        {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}
    ).dropna()
    agg["timestamp"] = agg.index.asi8 // 10**6
    return agg.reset_index(drop=True)


# --------------------------- indicatori ---------------------------
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 rsi(close: np.ndarray, n: int = 14) -> np.ndarray:
    d = np.diff(close, prepend=close[0])
    up = pd.Series(np.where(d > 0, d, 0.0)).ewm(alpha=1/n, adjust=False).mean()
    dn = pd.Series(np.where(d < 0, -d, 0.0)).ewm(alpha=1/n, adjust=False).mean()
    rs = up / dn.replace(0, np.nan)
    return (100 - 100 / (1 + rs)).values


# --------------------------- engine ---------------------------
def simulate(entries: list[dict], df: pd.DataFrame, fee_rt: float = FEE_RT,
             lev: float = LEV, pos: float = POS) -> dict:
    """entries: dict con i(idx), d(+1/-1), tp(prezzo), sl(prezzo), max_bars."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    n = len(c)
    cap = peak = 1000.0
    max_dd = 0.0
    fee = fee_rt * lev
    trades = wins = 0
    last_exit = -1
    bars_in = 0
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    yearly: dict[int, float] = {}

    for e in entries:
        i, d = e["i"], e["d"]
        if i <= last_exit or i + 1 >= n:
            continue
        entry = c[i]
        tp, sl, mb = e["tp"], e["sl"], e["max_bars"]
        exit_p = c[min(i + mb, n - 1)]
        for k in range(1, mb + 1):
            j = i + k
            if j >= n:
                exit_p = c[n - 1]; 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 k == mb:
                exit_p = c[j]
        ret = (exit_p - entry) / entry * d * lev - fee
        cb = cap
        cap = max(cb + cb * pos * ret, 10.0)
        peak = max(peak, cap); max_dd = max(max_dd, (peak - cap) / peak)
        trades += 1; wins += ret > 0; bars_in += min(mb, j - i)
        last_exit = j
        yearly[ts.iloc[i].year] = yearly.get(ts.iloc[i].year, 0.0) + ret * 100
    return {
        "trades": trades,
        "win": wins / trades * 100 if trades else 0.0,
        "ret": (cap / 1000 - 1) * 100,
        "dd": max_dd * 100,
        "yearly": yearly,
        "exposure": bars_in / n * 100,
    }


# --------------------------- strategie ---------------------------
def bollinger_fade(df, n=20, k=2.0, sl_atr=2.0, max_bars=24):
    """Mean-reversion: fada il close oltre la banda, TP alla media, SL = k_atr*ATR."""
    c = df["close"].values
    ma = pd.Series(c).rolling(n).mean().values
    sd = pd.Series(c).rolling(n).std().values
    a = atr(df, 14)
    up, lo = ma + k * sd, ma - k * sd
    ents = []
    for i in range(n + 14, len(c)):
        if np.isnan(up[i]) or np.isnan(a[i]):
            continue
        if c[i] < lo[i] and c[i - 1] >= lo[i - 1]:          # appena sotto la banda
            ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
        elif c[i] > up[i] and c[i - 1] <= up[i - 1]:
            ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
    return ents


def rsi_revert(df, n=14, lo=30, hi=70, sl_atr=2.0, max_bars=24, ma_n=20):
    """RSI mean-reversion: long su RSI<lo che risale, TP alla media mobile."""
    c = df["close"].values
    r = rsi(c, n)
    ma = pd.Series(c).rolling(ma_n).mean().values
    a = atr(df, 14)
    ents = []
    for i in range(max(n, ma_n) + 1, len(c)):
        if np.isnan(r[i]) or np.isnan(ma[i]) or np.isnan(a[i]):
            continue
        if r[i - 1] < lo <= r[i]:
            ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
        elif r[i - 1] > hi >= r[i]:
            ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
    return ents


def donchian_trend(df, n=20, sl_atr=2.0, tp_atr=6.0, max_bars=120):
    """Trend-following: breakout canale Donchian, TP/SL in multipli di ATR."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    hh = pd.Series(h).rolling(n).max().shift(1).values
    ll = pd.Series(l).rolling(n).min().shift(1).values
    a = atr(df, 14)
    ents = []
    for i in range(n + 14, len(c)):
        if np.isnan(hh[i]) or np.isnan(a[i]):
            continue
        if c[i] > hh[i]:
            ents.append({"i": i, "d": 1, "tp": c[i] + tp_atr * a[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
        elif c[i] < ll[i]:
            ents.append({"i": i, "d": -1, "tp": c[i] - tp_atr * a[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
    return ents


STRATS = {
    "BOLL_fade k2 m24":   (bollinger_fade, dict(n=20, k=2.0, sl_atr=2.0, max_bars=24)),
    "BOLL_fade k2.5 m24": (bollinger_fade, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24)),
    "RSI_revert 30/70":   (rsi_revert, dict(n=14, lo=30, hi=70, sl_atr=2.0, max_bars=24)),
    "RSI_revert 25/75":   (rsi_revert, dict(n=14, lo=25, hi=75, sl_atr=2.0, max_bars=24)),
    "DONCH_trend n20":    (donchian_trend, dict(n=20, sl_atr=2.0, tp_atr=6.0, max_bars=120)),
    "DONCH_trend n50":    (donchian_trend, dict(n=50, sl_atr=2.0, tp_atr=8.0, max_bars=200)),
}


def deep_dive():
    print("\n" + "#" * 120)
    print("  APPROFONDIMENTO BOLLINGER FADE (mean-reversion) — l'unica famiglia con edge netto")
    print("#" * 120)

    cases = [("BTC", "1h"), ("ETH", "1h"), ("BTC", "4h"), ("ETH", "4h")]
    base = dict(n=20, k=2.5, sl_atr=2.0, max_bars=24)

    # --- per anno (config base k2.5/n20) ---
    print(f"\n  [1] PnL NETTO per anno — n=20 k=2.5 sl=2ATR | fee {FEE_RT*100:.2f}% RT")
    all_years = sorted({y for a, tf in cases for y in simulate(bollinger_fade(get_df(a, tf), **base), get_df(a, tf))["yearly"]})
    print(f"  {'Asset/TF':<10s}" + "".join(f"{y:>8d}" for y in all_years) + f"{'TOT%':>9s}{'DD%':>6s}")
    for a, tf in cases:
        df = get_df(a, tf)
        r = simulate(bollinger_fade(df, **base), df)
        row = "".join(f"{r['yearly'].get(y, 0):>+8.0f}" for y in all_years)
        print(f"  {a+' '+tf:<10s}{row}{r['ret']:>+9.0f}{r['dd']:>6.0f}")

    # --- sensibilita' fee ---
    print(f"\n  [2] SENSIBILITA' FEE — Ret% FULL / OOS (n=20 k=2.5)")
    fees = [0.0, 0.0005, 0.001, 0.002]
    print(f"  {'Asset/TF':<10s}" + "".join(f"{f'{f*100:.2f}%RT':>22s}" for f in fees))
    print(f"  {'':<10s}" + "".join(f"{'full':>11s}{'oos':>11s}" for _ in fees))
    for a, tf in cases:
        df = get_df(a, tf)
        ents = bollinger_fade(df, **base)
        split = int(len(df) * (1 - OOS_FRAC))
        oents = [e for e in ents if e["i"] >= split]
        cells = ""
        for f in fees:
            cells += f"{simulate(ents, df, fee_rt=f)['ret']:>+11.0f}{simulate(oents, df, fee_rt=f)['ret']:>+11.0f}"
        print(f"  {a+' '+tf:<10s}{cells}")

    # --- griglia parametri (robustezza) su BTC/ETH 1h ---
    print(f"\n  [3] GRIGLIA PARAMETRI — Ret%OOS (DD%) | fee {FEE_RT*100:.2f}% RT, deve essere stabile")
    for a in ["BTC", "ETH"]:
        df = get_df(a, "1h")
        split = int(len(df) * (1 - OOS_FRAC))
        print(f"\n  {a} 1h     " + "".join(f"{f'k={k}':>16s}" for k in [2.0, 2.5, 3.0]))
        for n in [14, 20, 30, 50]:
            cells = ""
            for k in [2.0, 2.5, 3.0]:
                ents = [e for e in bollinger_fade(df, n=n, k=k, sl_atr=2.0, max_bars=24) if e["i"] >= split]
                r = simulate(ents, df)
                cell = f"{r['ret']:+.0f}({r['dd']:.0f})"
                cells += f"{cell:>16s}"
            print(f"   n={n:<4d}{cells}")


def main():
    print("=" * 120)
    print(f"  RICERCA STRATEGIE — NETTO dopo fee {FEE_RT*100:.2f}% RT | leva {LEV:.0f}x | pos {POS*100:.0f}% "
          f"| OOS = ultimo {int(OOS_FRAC*100)}%")
    print("=" * 120)
    print(f"  {'Strategia':<20s}{'Asset':>5s}{'TF':>5s}{'Trd':>6s}{'Tr/yr':>7s}{'Win%':>7s}"
          f"{'Ret%FULL':>10s}{'Ret%OOS':>10s}{'DD%':>7s}{'Exp%':>7s}{'AnniPos':>9s}")
    print("  " + "-" * 116)

    for label, (fn, params) in STRATS.items():
        for asset in ["BTC", "ETH"]:
            for tf in ["1h", "4h"]:
                df = get_df(asset, tf)
                ents = fn(df, **params)
                full = simulate(ents, df)
                split = int(len(df) * (1 - OOS_FRAC))
                oos = simulate([e for e in ents if e["i"] >= split], df)
                yrs = full["yearly"]
                pos_yrs = sum(1 for v in yrs.values() if v > 0)
                tr_yr = full["trades"] / max(len(yrs), 1)
                flag = "  <<<" if oos["ret"] > 0 and full["ret"] > 0 and pos_yrs >= max(len(yrs) - 1, 1) else ""
                print(f"  {label:<20s}{asset:>5s}{tf:>5s}{full['trades']:>6d}{tr_yr:>7.0f}{full['win']:>7.1f}"
                      f"{full['ret']:>+10.1f}{oos['ret']:>+10.1f}{full['dd']:>7.1f}{full['exposure']:>7.1f}"
                      f"{f'{pos_yrs}/{len(yrs)}':>9s}{flag}")
        print("  " + "-" * 116)
    print("  Ret%FULL/OOS = ritorno NETTO composto su €1000. AnniPos = anni con PnL netto>0.")
    print("  <<< = positivo full+OOS e robusto (quasi tutti gli anni positivi).")
    deep_dive()


if __name__ == "__main__":
    main()