"""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+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}")