Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(analysis): miglioramenti - ROT02 dual-momentum + portafoglio (DD 12%)

Browse Source 
Obiettivo: alzare Acc, ridurre DD, migliorare PnL. Leve oneste, no tuning per-anno.

- ROT02: overlay absolute-momentum (cash se BTC<SMA100) su ROT01. Domina su tutte
  le metriche: FULL +679->+1095%, OOS +44->+98%, DD 53->40%.
- DIP01 market-gate (variante low-DD): alza Acc (ETH 52->57, SOL 49->52) e dimezza
  il DD (ETH 53->23), al costo di PnL. De-risking opzionale; su BTC il gate va evitato.
- PORT01: portafoglio equal-weight giornaliero delle 3 sleeve anti-correlate
  (DIP01+TR01+ROT02). DD 12% (sotto ogni sleeve), CAGR 45%, 2022 bear -1% (era -30%).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
AdrianoDev
2026-05-28 23:49:14 +02:00
parent ad141f080c
commit 783fa5546f
5 changed files with 505 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/analysis/honest_improve.py
+175
View File
@@ -0,0 +1,175 @@
"""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}")
scripts/analysis/honest_improve2.py
+184
View File
@@ -0,0 +1,184 @@
"""Miglioramenti v2: market-regime gate su DIP01 + PORTAFOGLIO combinato.
- DIP01 con gate di mercato: compra i dip solo quando BTC e' risk-on (BTC>SMA),
cosi' si evitano le capitolazioni dei bear (2018/2022) che peggiorano Acc/DD/PnL.
- Portafoglio: equal-weight giornaliero delle 3 strategie migliorate -> la
diversificazione taglia il DD mantenendo la PnL (migliora il risk-adjusted).
Tutto NETTO, con DD pieno e 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_improve import rot_improved, _dd
LEV, POS = 3.0, 0.15
def _daily_equity(ts_list, cap_list, idx):
"""serie di equity giornaliera (ffill) su un DatetimeIndex comune."""
s = pd.Series(cap_list, index=pd.to_datetime(ts_list, utc=True))
s = s[~s.index.duplicated(keep="last")].sort_index()
daily = s.resample("1D").last().reindex(idx).ffill().bfill()
return daily
# ---------- DIP01 con market-regime gate ----------
def dip_market_gated(asset, n=50, z_in=2.5, sl_atr=2.5, max_bars=24,
market_n=100, fee_rt=FEE_RT, oos_frac=0.0, return_equity=False):
df = get_df(asset, "1h")
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)
# regime di mercato: BTC 1h > SMA(market_n in giorni -> *24 barre)
btc = get_df("BTC", "1h")
bser = pd.Series(btc["close"].values,
index=pd.to_datetime(btc["timestamp"], unit="ms", utc=True))
bser = bser[~bser.index.duplicated()]
bma = bser.rolling(market_n * 24).mean()
risk_on = (bser > bma).reindex(ts, method="ffill").fillna(False).values
fee = fee_rt * LEV
cap = 1000.0; last_exit = -1
eq_ts, eq_v = [], []
yt: dict[int, list] = {}; ypnl: dict[int, float] = {}
split = int(N * (1 - oos_frac)) if oos_frac else 0
for i in range(n + 14, 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
if market_n and not risk_on[i]:
continue
if i <= last_exit or i + 1 >= N:
continue
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 * POS * ret, 10.0)
last_exit = j
y = ts.iloc[i].year
rec = yt.setdefault(y, [0, 0]); rec[0] += 1; rec[1] += ret > 0
ypnl[y] = ypnl.get(y, 0.0) + ret * 100
eq_ts.append(ts.iloc[j]); eq_v.append(cap)
t = sum(v[0] for v in yt.values()); w = sum(v[1] for v in yt.values())
out = {"ret": (cap / 1000 - 1) * 100, "dd": _dd(np.array(eq_v)) if eq_v else 0.0,
"trades": t, "acc": w / t * 100 if t else 0.0, "yt": yt, "ypnl": ypnl,
"pos_years": sum(1 for v in ypnl.values() if v > 0), "n_years": len(ypnl)}
if return_equity:
out["eq_ts"], out["eq_v"] = eq_ts, eq_v
return out
def main():
print("=" * 96)
print(" DIP01 — base vs MARKET-GATE (compra dip solo se BTC>SMA100)")
print("=" * 96)
print(f" {'asset / config':<30s}{'Trd':>6s}{'Acc%':>7s}{'FULL%':>9s}{'OOS%':>9s}{'DD%':>7s}{'AnniP':>8s}")
for a in ["BTC", "ETH", "SOL"]:
b = dip_market_gated(a, market_n=0); bo = dip_market_gated(a, market_n=0, oos_frac=0.30)
g = dip_market_gated(a, market_n=100); go = dip_market_gated(a, market_n=100, oos_frac=0.30)
print(f" {a+' base':<30s}{b['trades']:>6d}{b['acc']:>7.1f}{b['ret']:>+9.0f}{bo['ret']:>+9.0f}"
f"{b['dd']:>7.0f}{str(b['pos_years'])+'/'+str(b['n_years']):>8s}")
print(f" {a+' +gate100':<30s}{g['trades']:>6d}{g['acc']:>7.1f}{g['ret']:>+9.0f}{go['ret']:>+9.0f}"
f"{g['dd']:>7.0f}{str(g['pos_years'])+'/'+str(g['n_years']):>8s}")
# ---------- PORTAFOGLIO combinato (3 sleeve diversificate) ----------
print("\n" + "=" * 96)
print(" PORTAFOGLIO equal-weight giornaliero (ribilanciato): DIP01 + TR01-basket + ROT02")
print("=" * 96)
idx = pd.date_range("2021-01-01", "2026-05-26", freq="1D", tz="UTC")
# sleeve 1: DIP01 base su BTC (la migliore)
d = dip_market_gated("BTC", market_n=0, return_equity=True)
eq_dip = _norm(_daily_equity(d["eq_ts"], d["eq_v"], idx))
# sleeve 2: TR01 equal-weight su {BNB,BTC,DOGE,SOL,XRP}
eq_tr = _norm(_tr_basket_daily(["BNB", "BTC", "DOGE", "SOL", "XRP"], idx))
# sleeve 3: ROT02 dual-momentum
eq_rot = _norm(_rot_daily_equity(idx))
members = {"DIP01_BTC": eq_dip, "TR01_basket": eq_tr, "ROT02_dualmom": eq_rot}
# ribilanciamento giornaliero equal-weight: media dei rendimenti giornalieri
drets = pd.DataFrame({k: v.pct_change().fillna(0) for k, v in members.items()})
port_ret = drets.mean(axis=1)
combo = (1 + port_ret).cumprod()
print(f" Periodo {idx[0].date()} -> {idx[-1].date()} (leva/pos gia' incluse nelle sleeve)")
print(f" {'sleeve':<16s}{'ret%':>9s}{'DD%':>7s}{'CAGR%':>8s}")
yrs = (idx[-1] - idx[0]).days / 365.25
for name, s in members.items():
r = (s.iloc[-1] / s.iloc[0] - 1) * 100
cagr = ((s.iloc[-1] / s.iloc[0]) ** (1 / yrs) - 1) * 100
print(f" {name:<16s}{r:>+9.0f}{_dd(s.values):>7.0f}{cagr:>8.0f}")
r = (combo.iloc[-1] / combo.iloc[0] - 1) * 100
cagr = ((combo.iloc[-1] / combo.iloc[0]) ** (1 / yrs) - 1) * 100
print(f" {'PORTAFOGLIO':<16s}{r:>+9.0f}{_dd(combo.values):>7.0f}{cagr:>8.0f} <-- DD molto piu' basso, CAGR solida")
# per-anno del portafoglio
pa = (port_ret.groupby(port_ret.index.year).apply(lambda x: ((1 + x).prod() - 1) * 100))
print(" Portafoglio per-anno: " + " ".join(f"{y}:{v:+.0f}%" for y, v in pa.items()))
def _norm(s):
return s / s.iloc[0]
def _tr_basket_daily(assets, idx):
"""equity giornaliera media di TR01 (EMA20/100 long-only, 4h) sul paniere."""
eqs = []
for a in assets:
df = get_df(a, "4h"); c = df["close"].values; n = len(c)
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
ef, es = ema(c, 20), ema(c, 100)
sig = np.where(ef > es, 1.0, 0.0); sig[:100] = 0.0
cap = 1000.0; cur = 0.0; fee = FEE_RT / 2 * LEV
tl, cl = [], []
for i in range(n - 1):
s = sig[i]
if s != cur:
cap -= cap * POS * fee * abs(s - cur); cur = s
cap = max(cap * (1 + POS * LEV * (c[i + 1] - c[i]) / c[i] * cur), 10.0)
tl.append(ts.iloc[i]); cl.append(cap)
eqs.append(_norm(_daily_equity(tl, cl, idx)))
return _norm(pd.concat(eqs, axis=1).mean(axis=1))
def _rot_daily_equity(idx):
"""equity giornaliera della ROT01 dual-momentum (ricostruita bar-by-bar)."""
from scripts.analysis.honest_rotation import build_panel
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
btc = P[:, cols.index("BTC")]; bma = pd.Series(btc).rolling(100).mean().values
cap = 1000.0; w = np.zeros(N); ts_list = []; cap_list = []
for i in range(101, T - 1):
risk_on = btc[i] > bma[i] if not np.isnan(bma[i]) else False
mom = P[i] / P[i - 60] - 1; order = np.argsort(mom)[::-1]
chosen = [j for j in order if mom[j] > 0][:2] if risk_on else []
nw = np.zeros(N)
for j in chosen:
nw[j] = 0.45 / len(chosen)
cap -= cap * np.abs(nw - w).sum() * (FEE_RT / 2); w = nw
cap = max(cap * (1 + float(np.dot(w, rets[i + 1]))), 10.0)
ts_list.append(panel.index[i]); cap_list.append(cap)
s = _daily_equity(ts_list, cap_list, idx); return s / s.iloc[0]
if __name__ == "__main__":
main()