PythagorasGoal/Old/scripts/analysis/strategy_research_v2.py

"""Ricerca v2 — nuove strategie oltre MR01, stessa metodologia fee-aware OOS.

Lezioni ereditate (vedi strategy_research.py / oos_validation.py):
  - mean-reversion ha edge, continuation/trend NO (i breakout rientrano)
  - fee = vincolo di prim'ordine -> default Deribit 0.10% RT, poche operazioni meglio
  - ingresso ESEGUIBILE a close[i] (mai look-ahead con direzione da barra i)
  - ogni numero NETTO dopo fee+leva, su finestra held-out (OOS=ultimo 30%) + per anno

Nuovi candidati (tutti fade/mean-reversion con ingresso onesto):
  MR02 donchian_fade  - fade rottura canale Donchian (opposto del trend che muore)
  MR03 keltner_fade   - fade canale Keltner (ATR), TP alla EMA media
  MR04 zscore_revert  - fade deviazione z-score estrema, TP alla media
  MR05 boll_fade_adx  - Bollinger fade con filtro regime ADX (solo mercato laterale)

Engine identico a strategy_research.simulate (ingresso close[i], exit TP/SL intrabar
high/low o time-limit, 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))

# riusa engine, dati e indicatori gia' validati
from scripts.analysis.strategy_research import (
    FEE_RT, LEV, POS, OOS_FRAC, get_df, atr, rsi, simulate,
)


# --------------------------- indicatori extra ---------------------------
def ema(x: np.ndarray, n: int) -> np.ndarray:
    return pd.Series(x).ewm(span=n, adjust=False).mean().values


def adx(df: pd.DataFrame, n: int = 14) -> np.ndarray:
    """Average Directional Index: misura la forza del trend (alto=trend, basso=range)."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    up = h - np.roll(h, 1)
    dn = np.roll(l, 1) - l
    up[0] = dn[0] = 0.0
    plus_dm = np.where((up > dn) & (up > 0), up, 0.0)
    minus_dm = np.where((dn > up) & (dn > 0), dn, 0.0)
    pc = np.roll(c, 1); pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    atr_n = pd.Series(tr).ewm(alpha=1/n, adjust=False).mean().values
    pdi = 100 * pd.Series(plus_dm).ewm(alpha=1/n, adjust=False).mean().values / np.where(atr_n == 0, np.nan, atr_n)
    mdi = 100 * pd.Series(minus_dm).ewm(alpha=1/n, adjust=False).mean().values / np.where(atr_n == 0, np.nan, atr_n)
    dx = 100 * np.abs(pdi - mdi) / np.where((pdi + mdi) == 0, np.nan, pdi + mdi)
    return pd.Series(dx).ewm(alpha=1/n, adjust=False).mean().values


# --------------------------- strategie nuove ---------------------------
def donchian_fade(df, n=20, sl_atr=2.0, max_bars=24):
    """MR02 — fade rottura canale Donchian: rompe sopra max-N => short verso il mid.

    Coerente con 'i breakout rientrano': l'opposto di donchian_trend (che fallisce).
    Ingresso a close[i] sulla barra che chiude oltre il canale precedente.
    TP al centro del canale, SL = sl_atr*ATR oltre l'estremo.
    """
    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
        mid = (hh[i] + ll[i]) / 2.0
        if c[i] > hh[i] and c[i - 1] <= hh[i - 1]:          # rottura rialzista => fade short
            ents.append({"i": i, "d": -1, "tp": mid, "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
        elif c[i] < ll[i] and c[i - 1] >= ll[i - 1]:        # rottura ribassista => fade long
            ents.append({"i": i, "d": 1, "tp": mid, "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
    return ents


def keltner_fade(df, n=20, k=2.0, sl_atr=2.0, max_bars=24):
    """MR03 — fade canale Keltner (EMA +/- k*ATR), TP alla EMA media.

    Come Bollinger ma banda basata su ATR (volatilita' di range) invece che std:
    reagisce diversamente ai gap. Ingresso quando close esce dalla banda.
    """
    c = df["close"].values
    e = ema(c, n)
    a = atr(df, n)
    up, lo = e + k * a, e - k * a
    ents = []
    for i in range(n + 1, 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]:
            ents.append({"i": i, "d": 1, "tp": e[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": e[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
    return ents


def zscore_revert(df, n=50, z=2.0, sl_atr=2.5, max_bars=24):
    """MR04 — fade deviazione z-score estrema dalla media, TP alla media.

    z = (close-ma)/std. Entra quando |z| supera la soglia (close fuori); chiude
    quando torna alla media. Banda di Bollinger riparametrizzata in z (equivalente
    a k=z) ma con SL piu' largo e finestra lunga: poche operazioni, alta selettivita'.
    """
    c = df["close"].values
    ma = pd.Series(c).rolling(n).mean().values
    sd = pd.Series(c).rolling(n).std().values
    a = atr(df, 14)
    ents = []
    for i in range(n + 14, len(c)):
        if np.isnan(ma[i]) or sd[i] == 0 or np.isnan(a[i]):
            continue
        zi = (c[i] - ma[i]) / sd[i]
        zp = (c[i - 1] - ma[i - 1]) / sd[i - 1] if sd[i - 1] else 0.0
        if zi <= -z and zp > -z:
            ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
        elif zi >= z and zp < z:
            ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
    return ents


def boll_fade_adx(df, n=50, k=2.5, sl_atr=2.0, max_bars=24, adx_max=25.0):
    """MR05 — Bollinger fade SOLO in regime laterale (ADX < adx_max).

    Il fade soffre quando c'e' trend forte (il prezzo continua oltre la banda).
    Filtro ADX: opera solo quando la forza del trend e' bassa -> meno trade, edge piu' pulito.
    """
    c = df["close"].values
    ma = pd.Series(c).rolling(n).mean().values
    sd = pd.Series(c).rolling(n).std().values
    a = atr(df, 14)
    ax = adx(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]) or np.isnan(ax[i]):
            continue
        if ax[i] >= adx_max:                 # trend forte: niente fade
            continue
        if c[i] < lo[i] and c[i - 1] >= lo[i - 1]:
            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 rsi2_fade(df, rsi_n=2, lo=10, hi=90, ma_n=200, tp_atr=2.0, sl_atr=3.0, max_bars=24):
    """MR06 — Connors RSI(2) pullback in direzione del trend, TP/SL in ATR.

    Meccanismo distinto da MR01/MR03: non usa bande di prezzo ma l'oscillatore
    RSI(2), che satura su micro-estremi. Filtro di trend con SMA lunga:
      - close SOPRA la SMA (uptrend) + RSI(2) < lo (dip) -> long, target rimbalzo
      - close SOTTO la SMA (downtrend) + RSI(2) > hi (pop) -> short
    TP = tp_atr*ATR a favore, SL = sl_atr*ATR contro. Compra il ritracciamento
    nel trend, non il contro-trend.
    """
    c = df["close"].values
    r = rsi(c, rsi_n)
    ma = pd.Series(c).rolling(ma_n).mean().values
    a = atr(df, 14)
    ents = []
    for i in range(ma_n + 14, len(c)):
        if np.isnan(r[i]) or np.isnan(ma[i]) or np.isnan(a[i]):
            continue
        if r[i] < lo and c[i] > ma[i]:            # dip in uptrend -> long
            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 r[i] > hi and c[i] < ma[i]:          # pop in downtrend -> short
            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


def return_reversal(df, n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24):
    """MR07 — fade movimento di barra estremo (return reversal).

    Misura il rendimento dell'ultima barra in unita' di deviazione standard rolling
    dei rendimenti. Se |ret| > k*sigma, fada nella direzione opposta; TP/SL in ATR.
    Meccanismo distinto: usa la volatilita' dei RENDIMENTI, non i livelli di prezzo.
    Config robusta (k=3.5, tp=2ATR, sl=1.5ATR): positivo full+OOS BTC e ETH 1h,
    DD piu' contenuto (BTC 25% / ETH 46%).
    """
    c = df["close"].values
    ret = np.zeros_like(c)
    ret[1:] = np.diff(c) / c[:-1]
    sig = pd.Series(ret).rolling(n).std().values
    a = atr(df, 14)
    ents = []
    for i in range(n + 14, len(c)):
        if np.isnan(sig[i]) or sig[i] == 0 or np.isnan(a[i]):
            continue
        z = ret[i] / sig[i]
        if z <= -k:                                # crollo di barra -> fade long
            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 z >= k:                               # spike di barra -> fade short
            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


CANDIDATES = {
    "MR02 donch_fade n20":  (donchian_fade, dict(n=20, sl_atr=2.0, max_bars=24)),
    "MR02 donch_fade n50":  (donchian_fade, dict(n=50, sl_atr=2.0, max_bars=24)),
    "MR03 kelt_fade k2":    (keltner_fade, dict(n=20, k=2.0, sl_atr=2.0, max_bars=24)),
    "MR03 kelt_fade k2.5":  (keltner_fade, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24)),
    "MR04 zscore z2 n50":   (zscore_revert, dict(n=50, z=2.0, sl_atr=2.5, max_bars=24)),
    "MR04 zscore z2.5 n50": (zscore_revert, dict(n=50, z=2.5, sl_atr=2.5, max_bars=24)),
    "MR05 boll_adx n50":    (boll_fade_adx, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24, adx_max=25)),
    "MR05 boll_adx n20":    (boll_fade_adx, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24, adx_max=25)),
    "MR06 rsi2 10/90":      (rsi2_fade, dict(rsi_n=2, lo=10, hi=90, ma_n=200, tp_atr=2.0, sl_atr=3.0, max_bars=24)),
    "MR06 rsi2 5/95":       (rsi2_fade, dict(rsi_n=2, lo=5, hi=95, ma_n=200, tp_atr=2.0, sl_atr=3.0, max_bars=24)),
    "MR07 retrev k3.5":     (return_reversal, dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
    "MR07 retrev k3.0":     (return_reversal, dict(n=50, k=3.0, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
}


def table():
    print("=" * 122)
    print(f"  RICERCA v2 — NETTO dopo fee {FEE_RT*100:.2f}% RT | leva {LEV:.0f}x | pos {POS*100:.0f}% "
          f"| OOS = ultimo {int(OOS_FRAC*100)}%")
    print("=" * 122)
    print(f"  {'Strategia':<22s}{'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("  " + "-" * 118)
    for label, (fn, params) in CANDIDATES.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)
                robust = oos["ret"] > 0 and full["ret"] > 0 and pos_yrs >= max(len(yrs) - 1, 1)
                flag = "  <<<" if robust else ""
                print(f"  {label:<22s}{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("  " + "-" * 118)
    print("  <<< = positivo full+OOS e robusto (quasi tutti gli anni positivi).")


def deep_dive():
    """Robustezza dei 3 candidati promossi: fee sweep + griglia parametri OOS."""
    split_of = lambda df: int(len(df) * (1 - OOS_FRAC))
    fees = [0.0, 0.0005, 0.001, 0.002]

    print("\n" + "#" * 122)
    print("  APPROFONDIMENTO MR02 / MR03 / MR05 — robustezza fee + griglia (deve restare positivo)")
    print("#" * 122)

    # --- MR02 Donchian Fade ---
    print(f"\n  [MR02 donchian_fade] SENSIBILITA' FEE — Ret% FULL/OOS (n=20)")
    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 [("BTC", "1h"), ("ETH", "1h"), ("BTC", "4h"), ("ETH", "4h")]:
        df = get_df(a, tf); sp = split_of(df)
        ents = donchian_fade(df, n=20, sl_atr=2.0, max_bars=24)
        oents = [e for e in ents if e["i"] >= sp]
        cells = "".join(f"{simulate(ents, df, fee_rt=f)['ret']:>+11.0f}{simulate(oents, df, fee_rt=f)['ret']:>+11.0f}" for f in fees)
        print(f"  {a+' '+tf:<10s}{cells}")
    print(f"\n  [MR02] GRIGLIA n x sl_atr — Ret%OOS(DD%) | fee {FEE_RT*100:.2f}% RT")
    for a in ["BTC", "ETH"]:
        df = get_df(a, "1h"); sp = split_of(df)
        print(f"\n  {a} 1h     " + "".join(f"{f'sl={s}':>16s}" for s in [1.5, 2.0, 3.0]))
        for n in [10, 20, 30, 50]:
            cells = ""
            for s in [1.5, 2.0, 3.0]:
                r = simulate([e for e in donchian_fade(df, n=n, sl_atr=s, max_bars=24) if e["i"] >= sp], df)
                cell = "%+.0f(%.0f)" % (r["ret"], r["dd"])
                cells += f"{cell:>16s}"
            print(f"   n={n:<4d}{cells}")

    # --- MR03 Keltner Fade ---
    print(f"\n  [MR03 keltner_fade] GRIGLIA n x k — Ret%OOS(DD%) | fee {FEE_RT*100:.2f}% RT")
    for a in ["BTC", "ETH"]:
        df = get_df(a, "1h"); sp = split_of(df)
        print(f"\n  {a} 1h     " + "".join(f"{f'k={k}':>16s}" for k in [1.5, 2.0, 2.5]))
        for n in [14, 20, 30, 50]:
            cells = ""
            for k in [1.5, 2.0, 2.5]:
                r = simulate([e for e in keltner_fade(df, n=n, k=k, sl_atr=2.0, max_bars=24) if e["i"] >= sp], df)
                cell = "%+.0f(%.0f)" % (r["ret"], r["dd"])
                cells += f"{cell:>16s}"
            print(f"   n={n:<4d}{cells}")

    # --- MR05 Bollinger Fade + ADX ---
    print(f"\n  [MR05 boll_fade_adx] GRIGLIA n x adx_max — Ret%OOS(DD%) | fee {FEE_RT*100:.2f}% RT")
    for a in ["BTC", "ETH"]:
        df = get_df(a, "1h"); sp = split_of(df)
        print(f"\n  {a} 1h     " + "".join(f"{f'adx<{x}':>16s}" for x in [20, 25, 30]))
        for n in [20, 30, 50]:
            cells = ""
            for x in [20, 25, 30]:
                r = simulate([e for e in boll_fade_adx(df, n=n, k=2.5, sl_atr=2.0, max_bars=24, adx_max=x) if e["i"] >= sp], df)
                cell = "%+.0f(%.0f)" % (r["ret"], r["dd"])
                cells += f"{cell:>16s}"
            print(f"   n={n:<4d}{cells}")


if __name__ == "__main__":
    table()
    deep_dive()