189 lines
6.9 KiB
Python
189 lines
6.9 KiB
Python
"""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())
|