"""Report PER ANNO (Trade, Acc%, DD%, PnL%) delle 3 strategie oneste. Acc: DIP01/TR01 = win-rate dei trade chiusi (episodi); ROT01 = % giorni positivi. DD : drawdown massimo dell'equity DENTRO l'anno solare. PnL: variazione % dell'equity nell'anno (composta). Tutto NETTO (fee 0.10% RT, leva 3x, pos 15%). Replica gli engine di honest_*. """ 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_final import dip_entries from scripts.analysis.honest_rotation import build_panel LEV, POS = 3.0, 0.15 def _yearly_dd(years: np.ndarray, equity: np.ndarray) -> dict[int, float]: """DD massimo intra-anno da una serie di equity etichettata per anno.""" out: dict[int, float] = {} for y in np.unique(years): eq = equity[years == y] peak = eq[0]; dd = 0.0 for v in eq: peak = max(peak, v) dd = max(dd, (peak - v) / peak if peak > 0 else 0.0) out[int(y)] = dd * 100 return out def _print(title, header, rows): print("\n" + "=" * 78) print(f" {title}") print("=" * 78) print(" " + header) print(" " + "-" * 74) for r in rows: print(" " + r) # --------------------------- DIP01 (trade-based) --------------------------- def dip_yearly(asset, tf="1h"): df = get_df(asset, tf) ents = dip_entries(df) 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) fee = FEE_RT * LEV cap = 1000.0 last_exit = -1 eq_y, eq_v = [], [] yt: dict[int, list] = {} # year -> [trades, wins, pnl_start_cap, pnl_end_cap] for e in ents: i, d = e["i"], e["d"] if i <= last_exit or i + 1 >= n: 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: j = n - 1; exit_p = c[j]; break if (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl): exit_p = sl; break if (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp): exit_p = tp; break if k == mb: exit_p = c[j] ret = (exit_p - entry) / entry * d * LEV - fee cap = max(cap + cap * POS * ret, 10.0) last_exit = j y = ts.iloc[i].year rec = yt.setdefault(y, [0, 0, None, None]) rec[0] += 1; rec[1] += ret > 0 eq_y.append(y); eq_v.append(cap) dd = _yearly_dd(np.array(eq_y), np.array(eq_v)) # PnL% anno: da equity prima/dopo rows = [] prev = 1000.0 yrs = sorted(yt) cum = {} cprev = 1000.0 # ricostruisci equity di fine anno end_cap = {} for y, v in zip(eq_y, eq_v): end_cap[y] = v for y in yrs: t, w = yt[y][0], yt[y][1] ec = end_cap[y] pnl = (ec / cprev - 1) * 100 cprev = ec rows.append(f"{y:>6d}{t:>8d}{(w/t*100 if t else 0):>8.1f}{dd.get(y,0):>8.1f}{pnl:>+10.1f}") return rows # --------------------------- TR01 (position episodes) --------------------------- def tr_yearly(asset, tf="4h", fast=20, slow=100): df = get_df(asset, tf) c = df["close"].values; n = len(c) ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) ef, es = ema(c, fast), ema(c, slow) sig = np.where(ef > es, 1.0, 0.0); sig[:slow] = 0.0 cap = 1000.0; cur = 0.0 fee = FEE_RT / 2 * LEV ep_start_cap = None; ep_year = None yt: dict[int, list] = {} eq_y, eq_v = [], [] for i in range(n - 1): s = sig[i] if s != cur: cap -= cap * POS * fee * abs(s - cur) if s == 1.0: # apertura long ep_start_cap = cap; ep_year = ts.iloc[i].year elif cur == 1.0 and ep_start_cap is not None: # chiusura long rec = yt.setdefault(ep_year, [0, 0]) rec[0] += 1; rec[1] += cap > ep_start_cap ep_start_cap = None cur = s pr = (c[i + 1] - c[i]) / c[i] cap = max(cap * (1 + POS * LEV * pr * cur), 10.0) eq_y.append(ts.iloc[i].year); eq_v.append(cap) if cur == 1.0 and ep_start_cap is not None: rec = yt.setdefault(ep_year, [0, 0]); rec[0] += 1; rec[1] += cap > ep_start_cap dd = _yearly_dd(np.array(eq_y), np.array(eq_v)) end_cap = {} for y, v in zip(eq_y, eq_v): end_cap[y] = v rows = []; cprev = 1000.0 for y in sorted(end_cap): t, w = yt.get(y, [0, 0]) pnl = (end_cap[y] / cprev - 1) * 100; cprev = end_cap[y] rows.append(f"{y:>6d}{t:>8d}{(w/t*100 if t else 0):>8.1f}{dd.get(y,0):>8.1f}{pnl:>+10.1f}") return rows # --------------------------- ROT01 (daily portfolio) --------------------------- def rot_yearly(lookback=60, top_k=2, gross=0.45): panel = build_panel(available_assets(), "1d") P = panel.values; T, N = P.shape rets = np.zeros_like(P); rets[1:] = P[1:] / P[:-1] - 1 years = panel.index.year.values cap = 1000.0; w = np.zeros(N) yt: dict[int, list] = {} # year -> [rebal, pos_days, days] eq_y, eq_v = [], [] for i in range(lookback + 1, T - 1): mom = P[i] / P[i - lookback] - 1 order = np.argsort(mom)[::-1] chosen = [j for j in order if mom[j] > 0][:top_k] new_w = np.zeros(N) for j in chosen: new_w[j] = gross / 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) y = int(years[i]) rec = yt.setdefault(y, [0, 0, 0]) rec[0] += turnover > 1e-9; rec[1] += pr > 0; rec[2] += 1 eq_y.append(y); eq_v.append(cap) dd = _yearly_dd(np.array(eq_y), np.array(eq_v)) end_cap = {} for y, v in zip(eq_y, eq_v): end_cap[y] = v rows = []; cprev = 1000.0 for y in sorted(end_cap): reb, pos, days = yt[y] pnl = (end_cap[y] / cprev - 1) * 100; cprev = end_cap[y] rows.append(f"{y:>6d}{reb:>8d}{(pos/days*100 if days else 0):>8.1f}{dd.get(y,0):>8.1f}{pnl:>+10.1f}") return rows if __name__ == "__main__": H = f"{'Anno':>6s}{'Trade':>8s}{'Acc%':>8s}{'DD%':>8s}{'PnL%':>10s}" for a in ["BTC", "ETH", "SOL"]: _print(f"DIP01 — {a} 1h (Acc = win-rate trade)", H, dip_yearly(a)) for a in ["BNB", "BTC", "DOGE", "SOL", "XRP"]: _print(f"TR01 — {a} 4h (Trade = episodi long, Acc = win-rate episodi)", H, tr_yearly(a)) _print("ROT01 — paniere 8 crypto 1d (Trade = ribilanciamenti, Acc = % giorni positivi)", H, rot_yearly())