From ad141f080c1f79837556b0e6f1502f228d31b537 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Thu, 28 May 2026 23:42:04 +0200 Subject: [PATCH] feat(analysis): report per-anno (Trade/Acc/DD/PnL) delle 3 strategie Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/analysis/honest_yearly.py | 188 ++++++++++++++++++++++++++++++ 1 file changed, 188 insertions(+) create mode 100644 scripts/analysis/honest_yearly.py diff --git a/scripts/analysis/honest_yearly.py b/scripts/analysis/honest_yearly.py new file mode 100644 index 0000000..1983939 --- /dev/null +++ b/scripts/analysis/honest_yearly.py @@ -0,0 +1,188 @@ +"""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())