PythagorasGoal/Old/scripts/analysis/honest_improve.py

"""Miglioramenti ONESTI: alzare Acc, ridurre DD, migliorare PnL senza overfitting.

Leve usate (tutte robuste e documentate, niente tuning sui singoli anni):
  1. ABSOLUTE-MOMENTUM overlay (dual momentum): vai in CASH quando il "mercato"
     (BTC) e' sotto la sua media di lungo periodo -> taglia i bear (2022/2026).
  2. VOL-TARGETING: scala l'esposizione per puntare a una volatilita' costante
     -> riduce il DD e liscia la PnL.
  3. TRAILING STOP ad ATR per il trend (TR01) -> blocca i profitti.
Confronto base vs migliorata su FULL + OOS + DD pieno + per-anno.
"""
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 scripts.analysis.honest_lab import atr, ema, get_df, available_assets, FEE_RT
from scripts.analysis.honest_rotation import build_panel

LEV, POS = 3.0, 0.15


def _dd(eq: np.ndarray) -> float:
    peak = eq[0]; mx = 0.0
    for v in eq:
        peak = max(peak, v); mx = max(mx, (peak - v) / peak if peak > 0 else 0.0)
    return mx * 100


# ============================================================================
# ROT01 migliorata: dual-momentum (cash se BTC < SMA) + vol-target
# ============================================================================
def rot_improved(lookback=60, top_k=2, gross=0.45, regime_n=100,
                 target_vol=0.0, vol_n=20, fee_rt=FEE_RT, oos_frac=0.0):
    panel = build_panel(available_assets(), "1d")
    cols = list(panel.columns)
    P = panel.values; T, N = P.shape
    rets = np.zeros_like(P); rets[1:] = P[1:] / P[:-1] - 1
    years = panel.index.year.values
    btc = P[:, cols.index("BTC")]
    use_regime = regime_n and regime_n > 1
    btc_ma = pd.Series(btc).rolling(max(regime_n, 2)).mean().values
    # vol realizzata del portafoglio equal-weight come proxy di scala
    mkt_ret = rets.mean(axis=1)
    rv = pd.Series(mkt_ret).rolling(vol_n).std().values * np.sqrt(365)
    start = max(lookback + 1, (regime_n + 1) if use_regime else 0, int(T * (1 - oos_frac)) if oos_frac else 0)
    cap = 1000.0; w = np.zeros(N)
    eq = [cap]; yearly: dict[int, float] = {}; pos_days = {}; days = {}; reb = {}
    for i in range(start, T - 1):
        if use_regime:
            risk_on = btc[i] > btc_ma[i] if not np.isnan(btc_ma[i]) else False
        else:
            risk_on = True
        mom = P[i] / P[i - lookback] - 1
        order = np.argsort(mom)[::-1]
        chosen = [j for j in order if mom[j] > 0][:top_k] if risk_on else []
        g = gross
        if target_vol > 0 and not np.isnan(rv[i]) and rv[i] > 0:
            g = min(gross, gross * target_vol / rv[i])     # solo riduzione (no leva extra)
        new_w = np.zeros(N)
        for j in chosen:
            new_w[j] = g / len(chosen)
        turnover = np.abs(new_w - w).sum()
        if turnover > 1e-9:
            cap -= cap * turnover * (fee_rt / 2)
        w = new_w
        pr = float(np.dot(w, rets[i + 1]))
        cap = max(cap * (1 + pr), 10.0)
        eq.append(cap)
        y = int(years[i])
        yearly[y] = yearly.get(y, 0.0) + pr * 100
        pos_days[y] = pos_days.get(y, 0) + (pr > 0); days[y] = days.get(y, 0) + 1
        reb[y] = reb.get(y, 0) + (turnover > 1e-9)
    return {"ret": (cap / 1000 - 1) * 100, "dd": _dd(np.array(eq)), "yearly": yearly,
            "pos_years": sum(1 for v in yearly.values() if v > 0), "n_years": len(yearly),
            "pos_days": pos_days, "days": days, "reb": reb}


# ============================================================================
# DIP01 migliorata: filtro regime (no dip in bear forte) + vol-target sizing
# ============================================================================
def dip_improved(asset, tf="1h", n=50, z_in=2.5, sl_atr=2.5, max_bars=24,
                 regime_n=200, vol_target=0.0, fee_rt=FEE_RT, oos_frac=0.0):
    df = get_df(asset, tf)
    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)
    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)
    sma_r = pd.Series(c).rolling(regime_n).mean().values
    atr_pct = a / c           # volatilita' relativa
    base_vol = np.nanmedian(atr_pct[regime_n:regime_n * 2]) if N > regime_n * 2 else np.nanmedian(atr_pct)
    fee = fee_rt * LEV
    cap = 1000.0; last_exit = -1
    eq = [cap]; yt: dict[int, list] = {}
    start = max(n + 14, regime_n + 1) if regime_n else n + 14
    split = int(N * (1 - oos_frac)) if oos_frac else 0
    for i in range(start, N):
        if i < split or np.isnan(z[i]) or np.isnan(a[i]):
            continue
        if not (z[i] <= -z_in and z[i - 1] > -z_in):
            continue
        # filtro regime: salta i dip in bear forte (prezzo molto sotto SMA lunga)
        if regime_n and not np.isnan(sma_r[i]) and c[i] < sma_r[i] * 0.90:
            continue
        if i <= last_exit or i + 1 >= N:
            continue
        # vol-target: riduci posizione se ATR% > base (no leva extra)
        psize = POS
        if vol_target > 0 and not np.isnan(atr_pct[i]) and atr_pct[i] > 0:
            psize = POS * min(1.0, base_vol / atr_pct[i])
        entry = c[i]; tp, sl, mb = ma[i], c[i] - sl_atr * a[i], 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:
                j = N - 1; exit_p = c[j]; break
            if l[j] <= sl:
                exit_p = sl; break
            if h[j] >= tp:
                exit_p = tp; break
            if k == mb:
                exit_p = c[j]
        ret = (exit_p - entry) / entry * LEV - fee
        cap = max(cap + cap * psize * ret, 10.0)
        last_exit = j
        y = ts.iloc[i].year
        rec = yt.setdefault(y, [0, 0]); rec[0] += 1; rec[1] += ret > 0
        eq.append(cap)
    t = sum(v[0] for v in yt.values()); w = sum(v[1] for v in yt.values())
    return {"ret": (cap / 1000 - 1) * 100, "dd": _dd(np.array(eq)),
            "trades": t, "acc": w / t * 100 if t else 0.0,
            "yt": yt, "pos_years": sum(1 for v in yt.values() if v[1] / max(v[0],1) and v[1]>v[0]*0 and (v[1]>0)), "n_years": len(yt)}


def dip_acc_pnl(asset, **kw):
    """ritorna anche FULL e OOS."""
    full = dip_improved(asset, **kw)
    oos = dip_improved(asset, oos_frac=0.30, **kw)
    return full, oos


if __name__ == "__main__":
    print("=" * 92)
    print("  ROT01 — BASE vs MIGLIORATA (dual-momentum cash + vol-target)")
    print("=" * 92)
    print(f"  {'config':<40s}{'FULL%':>9s}{'OOS%':>9s}{'DD%pieno':>10s}{'AnniP':>8s}")
    b = rot_improved(regime_n=0); bo = rot_improved(regime_n=0, oos_frac=0.30)
    print(f"  {'BASE (no overlay)':<40s}{b['ret']:>+9.0f}{bo['ret']:>+9.0f}{b['dd']:>10.0f}"
          f"{str(b['pos_years'])+'/'+str(b['n_years']):>8s}")
    for rn in [100, 150, 200]:
        f = rot_improved(regime_n=rn); o = rot_improved(regime_n=rn, oos_frac=0.30)
        print(f"  {'+ dual-mom cash (BTC<SMA'+str(rn)+')':<40s}{f['ret']:>+9.0f}{o['ret']:>+9.0f}"
              f"{f['dd']:>10.0f}{str(f['pos_years'])+'/'+str(f['n_years']):>8s}")
    for tv in [0.6, 0.8]:
        f = rot_improved(regime_n=150, target_vol=tv); o = rot_improved(regime_n=150, target_vol=tv, oos_frac=0.30)
        print(f"  {'+ dual-mom150 + volTarget'+str(tv):<40s}{f['ret']:>+9.0f}{o['ret']:>+9.0f}"
              f"{f['dd']:>10.0f}{str(f['pos_years'])+'/'+str(f['n_years']):>8s}")

    print("\n" + "=" * 92)
    print("  DIP01 — BASE vs MIGLIORATA (filtro regime + vol-target)")
    print("=" * 92)
    print(f"  {'asset / config':<34s}{'Trd':>6s}{'Acc%':>7s}{'FULL%':>9s}{'OOS%':>9s}{'DD%pieno':>10s}")
    for a in ["BTC", "ETH", "SOL"]:
        for label, kw in [("base", dict(regime_n=0, vol_target=0)),
                          ("+regime+volTgt", dict(regime_n=200, vol_target=0.5))]:
            f, o = dip_acc_pnl(a, **kw)
            print(f"  {a+' '+label:<34s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+9.0f}"
                  f"{o['ret']:>+9.0f}{f['dd']:>10.0f}")