PythagorasGoal/Old/scripts/analysis/risk_management.py

"""Gestione del rischio sulle fade (MR01/MR02/MR03/MR07): alzare Acc, ridurre DD.

Due analisi, ognuna misurata FULL e OOS (ultimo 30%) per non illudersi:

  (A) SCREENING LEVE — confronta su ogni strategia le leve di rischio:
      - vol-target sizing (size ~ 1/distanza-SL)         -> SCARTATA (peggiora)
      - skip alta volatilita' (ATR% in coda alta)        -> SCARTATA (peggiora)
      - filtro trend (|close-EMA200|/ATR oltre soglia)   -> ADOTTATA (Acc+ DD-)
      - combinazione di tutte

  (B) FILTRO TREND + PORTAFOGLIO:
      - sweep della soglia trend (assoluta in ATR, regola unica = no overfit)
      - portafoglio equipesato su sotto-conti indipendenti: curve poco correlate
        -> DD aggregato << DD del singolo sleeve (vera leva anti-drawdown)

Engine fedele: ingresso close[i], exit TP/SL intrabar (high/low) o time-limit,
non-overlap, capitale composto. Numeri NETTI fee 0.10% RT, leva 3x.
"""
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
from scripts.analysis.strategy_research import bollinger_fade, atr
from scripts.analysis.strategy_research_v2 import donchian_fade, return_reversal

FEE_RT, LEV, POS, INIT, OOS_FRAC = 0.001, 3.0, 0.15, 1000.0, 0.30

# config base di ogni strategia (come strategies.yml).
# NB: MR03 keltner_fade spostata in scripts/waste/ (fade piu' debole, ridondante
# con MR01); la funzione keltner_fade resta in strategy_research_v2 come record.
STRATS = {
    "MR01": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)),
    "MR02": (donchian_fade,  dict(n=20, sl_atr=2.0, max_bars=24)),
    "MR07": (return_reversal,dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
}
STRATS_ETH = dict(STRATS)


def strats_for(asset: str) -> dict:
    return STRATS_ETH if asset == "ETH" else STRATS


# ============================ (A) SCREENING LEVE ============================
def add_context(ents, df, ema_long=200):
    """Aggiunge a ogni entry: sl_dist, atr_pct, trend_dist (|close-EMA|/ATR)."""
    c = df["close"].values
    a = atr(df, 14)
    el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
    apct = a / c
    for e in ents:
        i = e["i"]
        e["sl_dist"] = abs(c[i] - e["sl"]) / c[i]
        e["atr_pct"] = apct[i]
        e["trend_dist"] = abs(c[i] - el[i]) / a[i] if a[i] else 0.0
    return ents


def simulate(ents, df, fee_rt=FEE_RT, lev=LEV, split=-1,
             sizer=None, vol_skip=None, trend_skip=None, max_size=0.30):
    """sizer: funzione(entry)->frazione capitale; default POS fisso.
    vol_skip: soglia atr_pct sopra cui salto. trend_skip: soglia trend_dist sopra cui salto."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    n = len(c)
    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    cap = peak = INIT
    dd = 0.0; last = -1; trd = wins = 0
    fee = fee_rt * lev
    yearly = {}; rets = []
    for e in ents:
        i, d = e["i"], e["d"]
        if i <= last or i + 1 >= n or i < split:
            continue
        if vol_skip is not None and e["atr_pct"] > vol_skip:
            continue
        if trend_skip is not None and e["trend_dist"] > trend_skip:
            continue
        entry = c[i]; tp, sl, mb = e["tp"], e["sl"], e["max_bars"]
        exit_p = c[min(i + mb, n - 1)]; j = min(i + mb, n - 1)
        for k in range(1, mb + 1):
            j = i + k
            if j >= n:
                exit_p = c[n - 1]; break
            hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
            ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
            if hs: exit_p = sl; break
            if ht: exit_p = tp; break
            if k == mb: exit_p = c[j]
        ret = (exit_p - entry) / entry * d * lev - fee
        size = POS if sizer is None else min(sizer(e), max_size)
        cap = max(cap + cap * size * ret, 10.0)
        peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
        trd += 1; wins += ret > 0; last = j; rets.append(ret * size)
        y = ts.iloc[i].year; yearly[y] = yearly.get(y, 0.0) + ret * size * INIT
    sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
    return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
                ret=(cap / INIT - 1) * 100, dd=dd * 100, yearly=yearly, sharpe=sharpe)


def vol_target_sizer(target=0.015):
    """size t.c. rischio (size*lev*sl_dist) ~ target; piu' largo lo stop, meno size."""
    return lambda e: target / (LEV * max(e["sl_dist"], 1e-4))


def _line(label, full, oos):
    print(f"  {label:<28s}{full['trades']:>6d}{full['acc']:>7.1f}{full['ret']:>+10.0f}{full['dd']:>7.1f}{full['sharpe']:>7.2f}"
          f"  | {oos['trades']:>5d}{oos['acc']:>7.1f}{oos['ret']:>+9.0f}{oos['dd']:>7.1f}{oos['sharpe']:>7.2f}")


def screen_levers():
    print("=" * 110)
    print("  (A) SCREENING LEVE — vol-target / vol-skip / filtro-trend | NETTO fee 0.10% RT, leva 3x")
    print("=" * 110)
    for asset in ["BTC", "ETH"]:
        df = load_data(asset, "1h")
        split = int(len(df) * (1 - OOS_FRAC))
        print(f"\n  {asset} 1h")
        print(f"  {'config':<28s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}{'Shrp':>7s}"
              f"  | {'oTrd':>5s}{'oAcc':>7s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
        print("  " + "-" * 106)
        for nm, (fn, params) in strats_for(asset).items():
            ents = add_context(fn(df, **params), df)
            p85 = float(np.quantile([e["atr_pct"] for e in ents], 0.85))
            t90 = float(np.quantile([e["trend_dist"] for e in ents], 0.90))
            _line(f"{nm} base", simulate(ents, df), simulate(ents, df, split=split))
            _line(f"{nm} +volTarget", simulate(ents, df, sizer=vol_target_sizer()),
                  simulate(ents, df, split=split, sizer=vol_target_sizer()))
            _line(f"{nm} +volSkip(p85)", simulate(ents, df, vol_skip=p85),
                  simulate(ents, df, split=split, vol_skip=p85))
            _line(f"{nm} +trendSkip(p90)", simulate(ents, df, trend_skip=t90),
                  simulate(ents, df, split=split, trend_skip=t90))
            _line(f"{nm} +ALL", simulate(ents, df, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90),
                  simulate(ents, df, split=split, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90))
            print("  " + "-" * 106)
    print("\n  Esito: vol-target e vol-skip PEGGIORANO; il filtro trend e' l'unica leva utile.")


# ===================== (B) FILTRO TREND + PORTAFOGLIO =====================
def build_trades(ents, df, lev=LEV, fee_rt=FEE_RT, trend_max=None, ema_long=200):
    """Lista trade non-overlap: (entry_idx, exit_idx, ret_netto). Filtro trend opzionale."""
    h, l, c = df["high"].values, df["low"].values, df["close"].values
    n = len(c); a = atr(df, 14)
    el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
    fee = fee_rt * lev
    out = []; last = -1
    for e in ents:
        i, d = e["i"], e["d"]
        if i <= last or i + 1 >= n:
            continue
        if trend_max is not None and a[i] and abs(c[i] - el[i]) / a[i] > trend_max:
            continue
        entry = c[i]; tp, sl, mb = e["tp"], e["sl"], e["max_bars"]
        exit_p = c[min(i + mb, n - 1)]; j = min(i + mb, n - 1)
        for k in range(1, mb + 1):
            j = i + k
            if j >= n:
                exit_p = c[n - 1]; break
            hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
            ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
            if hs: exit_p = sl; break
            if ht: exit_p = tp; break
            if k == mb: exit_p = c[j]
        ret = (exit_p - entry) / entry * d * lev - fee
        out.append((i, j, ret)); last = j
    return out


def metrics_single(trades, pos=POS, split=-1):
    cap = peak = INIT; dd = 0.0; trd = wins = 0; rets = []
    for i, j, ret in trades:
        if i < split:
            continue
        cap = max(cap + cap * pos * ret, 10.0)
        peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
        trd += 1; wins += ret > 0; rets.append(ret * pos)
    sh = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
    return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
                ret=(cap / INIT - 1) * 100, dd=dd * 100, sharpe=sh)


def sleeve_equity(trades, n_bars, pos=POS, split=-1):
    """Equity di uno sleeve su sotto-conto indipendente (capitale INIT, pos fissa)."""
    eq = np.full(n_bars, INIT, dtype=float)
    cap = INIT
    for i, j, ret in sorted(trades, key=lambda t: t[1]):
        if i < split:
            continue
        cap = max(cap + cap * pos * ret, 10.0)
        eq[j:] = cap
    return eq


def metrics_portfolio(strat_trades, n_bars, pos=POS, split=-1):
    """Portafoglio equipesato: media di N sotto-conti indipendenti. DD sull'aggregata."""
    sleeves = [sleeve_equity(tr, n_bars, pos=pos, split=split) for tr in strat_trades.values()]
    agg = np.mean(sleeves, axis=0)
    agg = agg[max(split, 0):]
    peak = np.maximum.accumulate(agg)
    dd = float(np.max((peak - agg) / peak) * 100)
    trd = sum(1 for tr in strat_trades.values() for i, _, _ in tr if i >= split)
    wins = sum(1 for tr in strat_trades.values() for i, _, r in tr if i >= split and r > 0)
    return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0, ret=(agg[-1] / INIT - 1) * 100, dd=dd)


def trend_and_portfolio():
    # --- sweep soglia trend ---
    print("\n" + "=" * 104)
    print("  (B1) FILTRO TREND |close-EMA200|/ATR > soglia -> SALTA | NETTO fee 0.10% RT, leva 3x")
    print("=" * 104)
    print(f"  {'Strat/Asset':<14s}{'soglia':>8s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>9s}{'DD%':>7s}"
          f"  | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
    print("  " + "-" * 100)
    for asset in ["BTC", "ETH"]:
        df = load_data(asset, "1h"); split = int(len(df) * (1 - OOS_FRAC))
        for nm, (fn, params) in strats_for(asset).items():
            ents = fn(df, **params)
            for thr in [None, 4.0, 3.0, 2.5, 2.0]:
                tr = build_trades(ents, df, trend_max=thr)
                f = metrics_single(tr); o = metrics_single(tr, split=split)
                lab = "base" if thr is None else f"{thr}ATR"
                print(f"  {nm+' '+asset:<14s}{lab:>8s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+9.0f}{f['dd']:>7.1f}"
                      f"  | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
            print("  " + "-" * 100)

    # --- portafoglio equipesato (filtro trend 3.0 ATR) ---
    print("\n" + "=" * 104)
    print("  (B2) PORTAFOGLIO equipesato: N sotto-conti indipendenti (pos 0.15, filtro trend 3.0 ATR)")
    print("=" * 104)
    print(f"  {'Universo':<26s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}"
          f"  | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}")
    print("  " + "-" * 100)
    all_trades = {}
    for asset in ["BTC", "ETH"]:
        df = load_data(asset, "1h"); split = int(len(df) * (1 - OOS_FRAC)); n = len(df)
        st = {f"{nm}_{asset}": build_trades(fn(df, **p), df, trend_max=3.0) for nm, (fn, p) in strats_for(asset).items()}
        all_trades.update(st)
        f = metrics_portfolio(st, n); o = metrics_portfolio(st, n, split=split)
        print(f"  {'Portafoglio '+asset+' (4 strat)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}"
              f"  | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}")
    df0 = load_data("BTC", "1h"); split0 = int(len(df0) * (1 - OOS_FRAC))
    f = metrics_portfolio(all_trades, len(df0)); o = metrics_portfolio(all_trades, len(df0), split=split0)
    print("  " + "-" * 100)
    print(f"  {'GLOBALE BTC+ETH (8 sleeve)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}"
          f"  | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}")
    print("\n  Curve poco correlate => DD aggregato molto piu' basso del singolo sleeve.")


def main():
    screen_levers()
    trend_and_portfolio()


if __name__ == "__main__":
    main()