chore(analysis): pulizia e accorpamento script di analisi (25 -> 15 file)

- accorpa risk_improvements.py + risk_portfolio.py -> risk_management.py
  (sezione A screening leve, sezione B filtro trend + portafoglio)
- rimuove 4 script legacy della famiglia squeeze (ormai in waste, non
  referenziati): compare_strategies, best_yearly, final_report, yearly_market_report
- rimuove 5 script honest_* di diagnostica/iterazione superati da honest_matrix
  (consolidato) e non importati: honest_diag, honest_diag2, honest_candidates,
  honest_yearly, honest_yearly2
- mantiene il core honest (lab/improve/improve2/rotation/trend) + canonici
  (final/matrix), tutta la ricerca fade (strategy_research[_v2]), validazione
  (oos_validation, validate_worker_mr01), intrabar_test (lezione squeeze)
- aggiorna riferimento in CLAUDE.md. Import-check: 14/14 moduli OK.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-29 00:08:24 +02:00
parent fd547748dd
commit 22c6080873
13 changed files with 261 additions and 2212 deletions
+1 -1
View File
@@ -110,7 +110,7 @@ esteso rispetto al trend di fondo (`|close EMA(ema_long)| / ATR(14) > trend_
cioè quando si starebbe fadando un trend/crollo estremo. Con `trend_max=3.0`, cioè quando si starebbe fadando un trend/crollo estremo. Con `trend_max=3.0`,
`ema_long=200` (default in `strategies.yml`): accuratezza su tutti gli sleeve `ema_long=200` (default in `strategies.yml`): accuratezza su tutti gli sleeve
e DD giù drasticamente su ETH (MR01 71%→26%, MR02 42%→25%, MR03 66%→34%, e DD giù drasticamente su ETH (MR01 71%→26%, MR02 42%→25%, MR03 66%→34%,
MR07 46%→21%), edge OOS confermato (vedi `scripts/analysis/risk_portfolio.py`). MR07 46%→21%), edge OOS confermato (vedi `scripts/analysis/risk_management.py`).
Unica eccezione: MR03 BTC, dove il filtro peggiora entrambe → lasciato disattivo. Unica eccezione: MR03 BTC, dove il filtro peggiora entrambe → lasciato disattivo.
Leva non robusta scartate: vol-target sizing e skip-alta-volatilità (peggiorano). Leva non robusta scartate: vol-target sizing e skip-alta-volatilità (peggiorano).
-309
View File
@@ -1,309 +0,0 @@
"""Confronto migliori strategie S1 e S2 — andamento per anno."""
from __future__ import annotations
import sys
sys.path.insert(0, ".")
import numpy as np
import pandas as pd
from src.data.downloader import load_data
from src.fractal.patterns import encode_candles
FEE_PERP = 0.002 # 0.1% taker roundtrip perpetual
FEE_OPT = 0.0052 # options roundtrip
INITIAL = 1000
LEVERAGE = 3
def keltner_ratio(close, high, low, window=14):
n = len(close)
r = np.full(n, np.nan)
for i in range(window, n):
wc, wh, wl = close[i-window:i], high[i-window:i], low[i-window:i]
ma = np.mean(wc)
bb_std = np.std(wc)
tr = np.maximum(wh-wl, np.maximum(np.abs(wh-np.roll(wc,1)), np.abs(wl-np.roll(wc,1))))
atr = np.mean(tr[1:])
kc = (ma+1.5*atr)-(ma-1.5*atr)
bb = (ma+2*bb_std)-(ma-2*bb_std)
if kc > 0:
r[i] = bb/kc
return r
def rv_ann(close, window):
lr = np.diff(np.log(np.where(close==0, 1e-10, close)))
r = np.full(len(close), np.nan)
for i in range(window, len(lr)):
r[i+1] = np.std(lr[i-window:i]) * np.sqrt(24*365)
return r
def rsi(close, period=14):
delta = np.diff(close)
gain = np.where(delta>0, delta, 0)
loss = np.where(delta<0, -delta, 0)
result = np.full(len(close), 50.0)
if len(gain) < period:
return result
ag = np.mean(gain[:period])
al = np.mean(loss[:period])
for i in range(period, len(delta)):
ag = (ag*(period-1)+gain[i])/period
al = (al*(period-1)+loss[i])/period
result[i+1] = 100 if al == 0 else 100-100/(1+ag/al)
return result
def ema(arr, period):
r = np.full(len(arr), np.nan)
k = 2/(period+1)
r[period-1] = np.mean(arr[:period])
for i in range(period, len(arr)):
r[i] = arr[i]*k + r[i-1]*(1-k)
return r
# =====================================================================
# S1 BEST: Squeeze Breakout ETH 1h (BBw=14, sq=0.8, brk=3)
# =====================================================================
def run_s1_squeeze(asset, tf):
df = load_data(asset, tf)
c, h, l, v = df["close"].values, df["high"].values, df["low"].values, df["volume"].values
n = len(c)
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
kcr = keltner_ratio(c, h, l, 14)
yearly = {}
in_sq = False
sq_start = 0
for i in range(15, n):
if np.isnan(kcr[i]):
continue
is_sq = kcr[i] < 0.8
if is_sq and not in_sq:
in_sq = True
sq_start = i
elif not is_sq and in_sq:
in_sq = False
if i - sq_start < 5 or i + 3 >= n:
continue
first_ret = (c[i] - c[i-1]) / c[i-1]
if abs(first_ret) < 0.001:
continue
direction = 1 if first_ret > 0 else -1
actual = (c[i+2] - c[i-1]) / c[i-1]
trade_ret = actual * direction
net = trade_ret * LEVERAGE - FEE_PERP * LEVERAGE
year = ts.iloc[i].year
if year not in yearly:
yearly[year] = {"pnls": [], "wins": 0, "total": 0}
yearly[year]["pnls"].append(net)
yearly[year]["total"] += 1
if trade_ret > 0:
yearly[year]["wins"] += 1
return yearly
# =====================================================================
# S1 BEST ALT: Squeeze+ML hybrid ETH 15m
# =====================================================================
# Troppo complesso da ricalcolare (serve ML training). Uso i dati S1 squeeze puro.
# =====================================================================
# S2 BEST: VRP ETH 48h (con IV stimata, unico disponibile su 8 anni)
# =====================================================================
def run_s2_vrp(asset, dte=48):
df = load_data(asset, "1h")
c = df["close"].values
n = len(c)
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
rv_24 = rv_ann(c, 24)
rv_168 = rv_ann(c, 168)
yearly = {}
for i in range(170, n - dte):
if ts.iloc[i].hour != 8:
continue
rv_s, rv_l = rv_24[i], rv_168[i]
if np.isnan(rv_s) or np.isnan(rv_l) or rv_s < 0.05 or rv_l < 0.05:
continue
regime = rv_s / rv_l
iv_pf = 0.9 if regime > 2 else (1.0 if regime > 1.5 else (1.1 if regime > 1 else 1.2))
iv = rv_l * iv_pf
prem = iv * np.sqrt(dte/(24*365)) * 0.8
spot = c[i]
move = abs(c[min(i+dte, n-1)] - spot) / spot
pos = 0.10
raw = (prem - move) * pos if move <= prem else max(-(move-prem)*pos, -pos*0.05)
net = raw - FEE_OPT * pos
year = ts.iloc[i].year
if year not in yearly:
yearly[year] = {"pnls": [], "wins": 0, "total": 0}
yearly[year]["pnls"].append(net)
yearly[year]["total"] += 1
if raw > 0:
yearly[year]["wins"] += 1
return yearly
# =====================================================================
# S2 BEST PERPETUAL: Multi-TF 15m+1h BTC
# =====================================================================
def run_s2_multitf(asset):
df_1h = load_data(asset, "1h")
df_15m = load_data(asset, "15m")
c1h = df_1h["close"].values
ts1h = pd.to_datetime(df_1h["timestamp"], unit="ms", utc=True)
c15 = df_15m["close"].values
ts15 = df_15m["timestamp"].values
n15 = len(c15)
ema_50 = ema(c1h, 50)
rsi_15m = rsi(c15, 14)
yearly = {}
daily_done = set()
for i in range(100, n15 - 12):
ts_dt = pd.Timestamp(ts15[i], unit="ms", tz="UTC")
day = ts_dt.strftime("%Y-%m-%d")
if day in daily_done:
continue
if rsi_15m[i] > 35 and rsi_15m[i] < 65:
continue
h_idx = np.searchsorted(ts1h.values.astype("int64"), ts15[i]) - 1
if h_idx < 50 or h_idx >= len(c1h) or np.isnan(ema_50[h_idx]):
continue
direction = None
if rsi_15m[i] < 30 and c1h[h_idx] > ema_50[h_idx]:
direction = "long"
elif rsi_15m[i] > 70 and c1h[h_idx] < ema_50[h_idx]:
direction = "short"
if direction is None:
continue
entry = c15[i]
exit_price = c15[min(i+12, n15-1)]
trade_ret = (exit_price-entry)/entry if direction == "long" else (entry-exit_price)/entry
net = trade_ret * LEVERAGE - FEE_PERP * LEVERAGE
year = ts_dt.year
if year not in yearly:
yearly[year] = {"pnls": [], "wins": 0, "total": 0}
yearly[year]["pnls"].append(net)
yearly[year]["total"] += 1
if trade_ret > 0:
yearly[year]["wins"] += 1
daily_done.add(day)
return yearly
# =====================================================================
# REPORT
# =====================================================================
strategies = {
"S1: Squeeze BTC 1h": run_s1_squeeze("BTC", "1h"),
"S1: Squeeze ETH 1h": run_s1_squeeze("ETH", "1h"),
"S1: Squeeze ETH 15m": run_s1_squeeze("ETH", "15m"),
"S2: VRP ETH 48h (IV est)": run_s2_vrp("ETH", 48),
"S2: VRP BTC 48h (IV est)": run_s2_vrp("BTC", 48),
"S2: MultiTF BTC 15m+1h": run_s2_multitf("BTC"),
"S2: MultiTF ETH 15m+1h": run_s2_multitf("ETH"),
}
all_years = sorted(set(y for v in strategies.values() for y in v))
print("=" * 120)
print(" MIGLIORI STRATEGIE — ANDAMENTO PER ANNO")
print(" Fee reali. PnL su €1000 flat (no compounding). Dati OHLCV reali 2018-2026.")
print(" ⚠ VRP usa IV STIMATA (non reale) — fidarsi solo dei dati perpetual per backtest lungo")
print("=" * 120)
# Header
hdr = f" {'Anno':>6s}"
for name in strategies:
short = name.split(": ")[1][:18]
hdr += f" | {short:>18s}"
print(hdr)
print(f" {'-' * (len(hdr) - 2)}")
# Per anno: accuracy / PnL totale
for year in all_years:
row_acc = f" {year:>6d}"
row_pnl = f" {'':>6s}"
for name, yearly in strategies.items():
if year in yearly:
d = yearly[year]
acc = d["wins"]/d["total"]*100 if d["total"] > 0 else 0
pnl = sum(d["pnls"]) * INITIAL
tag = "" if acc >= 75 else "" if acc >= 65 else "" if acc >= 55 else " "
row_acc += f" | {acc:>5.1f}% {tag} {d['total']:>3d}t"
row_pnl += f" | €{pnl:>+8.0f} "
else:
row_acc += f" | {'':>18s}"
row_pnl += f" | {'':>18s}"
print(row_acc)
print(row_pnl)
# Totali
print(f" {'-' * (len(hdr) - 2)}")
row_tot = f" {'TOT':>6s}"
for name, yearly in strategies.items():
all_pnls = [p for d in yearly.values() for p in d["pnls"]]
all_wins = sum(d["wins"] for d in yearly.values())
all_total = sum(d["total"] for d in yearly.values())
acc = all_wins/all_total*100 if all_total > 0 else 0
pnl = sum(all_pnls) * INITIAL
row_tot += f" | {acc:>5.1f}% {all_total:>4d}t"
print(row_tot)
row_pnl_tot = f" {'€TOT':>6s}"
for name, yearly in strategies.items():
all_pnls = [p for d in yearly.values() for p in d["pnls"]]
pnl = sum(all_pnls) * INITIAL
row_pnl_tot += f" | €{pnl:>+8.0f} "
print(row_pnl_tot)
# Compounding
print(f"\n {'':>6s}", end="")
for name in strategies:
short = name.split(": ")[1][:18]
print(f" | {short:>18s}", end="")
print()
row_comp = f" {'COMP':>6s}"
for name, yearly in strategies.items():
cap = float(INITIAL)
for year in sorted(yearly):
for pnl in yearly[year]["pnls"]:
cap += cap * pnl
cap = max(cap, 10)
row_comp += f" | €{cap:>12,.0f} "
print(row_comp)
# Drawdown
row_dd = f" {'MAXDD':>6s}"
for name, yearly in strategies.items():
cap = float(INITIAL)
peak = cap
mdd = 0
for year in sorted(yearly):
for pnl in yearly[year]["pnls"]:
cap += cap * pnl
cap = max(cap, 10)
if cap > peak: peak = cap
dd = (peak - cap) / peak
mdd = max(mdd, dd)
row_dd += f" | {mdd*100:>12.1f}% "
print(row_dd)
# Legenda
print(f"\n Legenda: ▓ ≥75% acc ▒ ≥65% acc ░ ≥55% acc")
print(f" ⚠ S2 VRP: IV stimata (rv_long × 1.0-1.2), NON dati reali opzioni")
print(f" S1 Squeeze e S2 MultiTF: dati OHLCV reali al 100%")
-559
View File
@@ -1,559 +0,0 @@
"""Analisi finale — S1 (squeeze puro) vs Script 13 (squeeze+ML GBM).
Metriche: PnL, num trades, DD max, tempo medio a mercato, descrizione.
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")
import numpy as np
import pandas as pd
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.preprocessing import StandardScaler
from src.data.downloader import load_data
from src.fractal.patterns import encode_candles
FEE_PERP = 0.002
FEE_ML = 0.001
INITIAL = 1000
LEVERAGE = 3
TF_MINUTES = {"1m": 1, "5m": 5, "15m": 15, "1h": 60, "4h": 240, "1d": 1440}
# ── helpers ──────────────────────────────────────────────────────────
def keltner_ratio(close, high, low, window=14):
n = len(close)
r = np.full(n, np.nan)
for i in range(window, n):
wc, wh, wl = close[i-window:i], high[i-window:i], low[i-window:i]
ma = np.mean(wc)
bb_std = np.std(wc)
tr = np.maximum(wh-wl, np.maximum(np.abs(wh-np.roll(wc,1)), np.abs(wl-np.roll(wc,1))))
atr = np.mean(tr[1:])
kc = (ma+1.5*atr)-(ma-1.5*atr)
bb = (ma+2*bb_std)-(ma-2*bb_std)
if kc > 0:
r[i] = bb/kc
return r
def detect_squeezes(close, high, low, kcr, sq_thr=0.8, min_dur=5):
events = []
in_sq = False
sq_start = 0
for i in range(1, len(close)):
if np.isnan(kcr[i]):
continue
is_sq = kcr[i] < sq_thr
if is_sq and not in_sq:
in_sq = True
sq_start = i
elif not is_sq and in_sq:
in_sq = False
dur = i - sq_start
if dur < min_dur:
continue
events.append({"idx": i, "dur": dur, "sq_start": sq_start,
"avg_vol_squeeze": np.mean(close[sq_start:i]),
"kcr_at_release": kcr[i]})
return events
def _build_result(yearly, capital, max_dd, all_t, all_w, time_pct, avg_dur_h):
acc = all_w / all_t * 100
tot_pnl = sum(p for d in yearly.values() for p in d["pnls"])
years_active = len(yearly)
all_pnls = [p for d in yearly.values() for p in d["pnls"]]
sharpe = np.mean(all_pnls) / np.std(all_pnls) * np.sqrt(252) if len(all_pnls) > 1 and np.std(all_pnls) > 0 else 0
year_details = {}
for y in sorted(yearly):
d = yearly[y]
ya = d["w"] / d["t"] * 100 if d["t"] > 0 else 0
yp = sum(d["pnls"])
year_details[y] = {"trades": d["t"], "acc": ya, "pnl": yp}
valid_years = {y: d for y, d in year_details.items() if d["trades"] >= 10}
if valid_years:
worst_y = min(valid_years, key=lambda y: valid_years[y]["acc"])
worst_acc = valid_years[worst_y]["acc"]
elif year_details:
worst_y = min(year_details, key=lambda y: year_details[y]["acc"])
worst_acc = year_details[worst_y]["acc"]
else:
worst_y = "N/A"
worst_acc = 0
daily_pnl = tot_pnl / (years_active * 365) if years_active > 0 else 0
return {
"trades": all_t, "acc": acc, "pnl": tot_pnl, "capital": capital,
"max_dd": max_dd * 100, "sharpe": sharpe, "daily_pnl": daily_pnl,
"time_in_market_pct": time_pct, "avg_dur_h": avg_dur_h,
"years_active": years_active, "worst_year": str(worst_y),
"worst_acc": worst_acc, "year_details": year_details,
}
# ── S1: Squeeze breakout puro ────────────────────────────────────────
def run_s1_squeeze(asset, tf, hold=3):
df = load_data(asset, tf)
c, h, l, v = df["close"].values, df["high"].values, df["low"].values, df["volume"].values
n = len(c)
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
kcr = keltner_ratio(c, h, l, 14)
events = detect_squeezes(c, h, l, kcr)
yearly = {}
capital = float(INITIAL)
peak = capital
max_dd = 0
total_bars = 0
for ev in events:
i = ev["idx"]
if i + hold + 1 >= n:
continue
first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0
if abs(first_ret) < 0.001:
continue
direction = 1 if first_ret > 0 else -1
entry = c[i-1]
exit_price = c[min(i + hold - 1, n - 1)]
actual = (exit_price - entry) / entry * direction
net = actual * LEVERAGE - FEE_PERP * LEVERAGE
capital += capital * 0.15 * net
capital = max(capital, 10)
if capital > peak: peak = capital
dd = (peak - capital) / peak
max_dd = max(max_dd, dd)
total_bars += hold
year = ts.iloc[i].year
if year not in yearly:
yearly[year] = {"w": 0, "t": 0, "pnls": []}
yearly[year]["t"] += 1
if actual > 0: yearly[year]["w"] += 1
yearly[year]["pnls"].append(net * INITIAL)
all_t = sum(d["t"] for d in yearly.values())
all_w = sum(d["w"] for d in yearly.values())
if all_t == 0: return None
return _build_result(yearly, capital, max_dd, all_t, all_w,
total_bars / n * 100, hold * TF_MINUTES.get(tf, 60) / 60)
def run_s1_antifake_vol(asset, tf, hold=3):
df = load_data(asset, tf)
c, h, l, v = df["close"].values, df["high"].values, df["low"].values, df["volume"].values
n = len(c)
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
kcr = keltner_ratio(c, h, l, 14)
events = detect_squeezes(c, h, l, kcr)
yearly = {}
capital = float(INITIAL)
peak = capital
max_dd = 0
total_bars = 0
for ev in events:
i = ev["idx"]
if i + hold + 1 >= n:
continue
first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0
if abs(first_ret) < 0.001:
continue
br = h[i] - l[i]
if br > 0:
if c[i] > c[i-1]:
if (h[i] - c[i]) / br > 0.6:
continue
else:
if (c[i] - l[i]) / br > 0.6:
continue
avg_v = np.mean(v[ev["sq_start"]:i])
if avg_v > 0 and v[i] <= avg_v * 1.3:
continue
direction = 1 if first_ret > 0 else -1
entry = c[i-1]
exit_price = c[min(i + hold - 1, n - 1)]
actual = (exit_price - entry) / entry * direction
net = actual * LEVERAGE - FEE_PERP * LEVERAGE
capital += capital * 0.15 * net
capital = max(capital, 10)
if capital > peak: peak = capital
dd = (peak - capital) / peak
max_dd = max(max_dd, dd)
total_bars += hold
year = ts.iloc[i].year
if year not in yearly:
yearly[year] = {"w": 0, "t": 0, "pnls": []}
yearly[year]["t"] += 1
if actual > 0: yearly[year]["w"] += 1
yearly[year]["pnls"].append(net * INITIAL)
all_t = sum(d["t"] for d in yearly.values())
all_w = sum(d["w"] for d in yearly.values())
if all_t == 0: return None
return _build_result(yearly, capital, max_dd, all_t, all_w,
total_bars / n * 100, hold * TF_MINUTES.get(tf, 60) / 60)
# ── Script 13: Squeeze + ML ibrida (GBM walk-forward) ────────────────
def build_features_at(df, i, squeeze_info):
if i < 100:
return None
o = df["open"].values
h = df["high"].values
l = df["low"].values
c = df["close"].values
v = df["volume"].values
feats = []
for w in [12, 24, 48]:
win_c = c[i-w:i]
win_o = o[i-w:i]
win_h = h[i-w:i]
win_l = l[i-w:i]
win_v = v[i-w:i]
mn, mx = win_l.min(), max(win_h.max(), win_c.max())
rng = mx - mn if mx - mn > 0 else 1e-10
total = win_h - win_l
total = np.where(total == 0, 1e-10, total)
body = np.abs(win_c - win_o) / total
direction = np.sign(win_c - win_o)
log_c = np.log(np.where(win_c == 0, 1e-10, win_c))
rets = np.diff(log_c)
v_mean = np.mean(win_v)
feats.extend([
np.mean(rets) if len(rets) > 0 else 0,
np.std(rets) if len(rets) > 0 else 0,
np.sum(rets) if len(rets) > 0 else 0,
float(pd.Series(rets).skew()) if len(rets) > 2 else 0,
float(pd.Series(rets).kurtosis()) if len(rets) > 3 else 0,
np.mean(body), np.std(body),
np.mean(direction), np.mean(direction[-min(3, w):]),
(win_c[-1] - mn) / rng,
win_v[-1] / v_mean if v_mean > 0 else 1,
np.corrcoef(rets[:-1], rets[1:])[0, 1] if len(rets) > 1 and np.std(rets) > 0 else 0,
])
sq = squeeze_info
feats.extend([
sq["dur"], sq["dur"] / 24, sq["kcr_at_release"],
v[i-1] / sq["avg_vol_squeeze"] if sq["avg_vol_squeeze"] > 0 else 1,
np.mean(v[i:min(i+3, len(v))]) / sq["avg_vol_squeeze"] if sq["avg_vol_squeeze"] > 0 else 1,
])
h48 = np.max(h[max(0, i-48):i])
l48 = np.min(l[max(0, i-48):i])
r48 = h48 - l48
feats.append((c[i-1] - l48) / r48 if r48 > 0 else 0.5)
tr = np.maximum(h[i-14:i] - l[i-14:i],
np.maximum(np.abs(h[i-14:i] - np.roll(c[i-14:i], 1)),
np.abs(l[i-14:i] - np.roll(c[i-14:i], 1))))
atr = np.mean(tr[1:])
feats.append(atr / c[i-1] if c[i-1] > 0 else 0)
first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0
feats.append(first_ret)
return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
def run_s13_hybrid(asset, tf, bb_w, sq_thr, brk_bars, leverage, pos_pct, ml_thr):
df = load_data(asset, tf)
close = df["close"].values
high = df["high"].values
low = df["low"].values
volume = df["volume"].values
n = len(df)
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
kcr = keltner_ratio(close, high, low, bb_w)
events = detect_squeezes(close, high, low, kcr, sq_thr)
X_all, y_all, ev_all = [], [], []
for ev in events:
i = ev["idx"]
if i + brk_bars >= n or i < 100:
continue
feats = build_features_at(df, i, ev)
if feats is None:
continue
actual_ret = (close[i + brk_bars - 1] - close[i - 1]) / close[i - 1]
X_all.append(feats)
y_all.append(1 if actual_ret > 0 else 0)
ev_all.append(ev)
if len(X_all) < 50:
return None
X = np.array(X_all)
y = np.array(y_all)
TRAIN_SIZE = max(int(len(X) * 0.5), 50)
STEP_SIZE = max(int(len(X) * 0.1), 10)
yearly = {}
capital = float(INITIAL)
peak = capital
max_dd = 0
total_bars = 0
all_t = 0
all_w = 0
start = 0
while start + TRAIN_SIZE + STEP_SIZE <= len(X):
train_end = start + TRAIN_SIZE
test_end = min(train_end + STEP_SIZE, len(X))
X_tr, y_tr = X[start:train_end], y[start:train_end]
X_te = X[train_end:test_end]
if len(np.unique(y_tr)) < 2:
start += STEP_SIZE
continue
scaler = StandardScaler()
X_tr_s = scaler.fit_transform(X_tr)
X_te_s = scaler.transform(X_te)
model = GradientBoostingClassifier(
n_estimators=150, max_depth=4, min_samples_leaf=10,
learning_rate=0.05, subsample=0.8, random_state=42,
)
model.fit(X_tr_s, y_tr)
up_idx = list(model.classes_).index(1) if 1 in model.classes_ else -1
if up_idx < 0:
start += STEP_SIZE
continue
for j in range(len(X_te)):
proba = model.predict_proba(X_te_s[j:j+1])[0]
p_up = proba[up_idx]
ev = ev_all[train_end + j]
i = ev["idx"]
actual_ret = (close[i + brk_bars - 1] - close[i - 1]) / close[i - 1]
direction = None
if p_up >= ml_thr:
direction = 1
elif p_up <= (1 - ml_thr):
direction = -1
if direction is None:
continue
is_correct = (direction == 1 and actual_ret > 0) or (direction == -1 and actual_ret < 0)
trade_ret = actual_ret * direction
net = trade_ret * leverage - FEE_ML * 2 * leverage
capital += capital * pos_pct * net
capital = max(capital, 10)
if capital > peak: peak = capital
dd = (peak - capital) / peak
max_dd = max(max_dd, dd)
total_bars += brk_bars
all_t += 1
if is_correct: all_w += 1
year = ts.iloc[i].year
if year not in yearly:
yearly[year] = {"w": 0, "t": 0, "pnls": []}
yearly[year]["t"] += 1
if is_correct: yearly[year]["w"] += 1
yearly[year]["pnls"].append(net * INITIAL)
start += STEP_SIZE
if all_t == 0:
return None
return _build_result(yearly, capital, max_dd, all_t, all_w,
total_bars / n * 100, brk_bars * TF_MINUTES.get(tf, 60) / 60)
# ═══════════════════════════════════════════════════════════════════
# ESECUZIONE
# ═══════════════════════════════════════════════════════════════════
print("Calcolo in corso...\n")
strategies = []
def add(name, desc, cat, result):
if result and result["trades"] >= 20:
strategies.append({"name": name, "desc": desc, "cat": cat, **result})
# ── S1: Squeeze puro ────────────────────────────────────────────
add("S1 Squeeze BTC 15m", "Squeeze breakout puro, BBw=14, hold 3×15m, leva 3x",
"S1", run_s1_squeeze("BTC", "15m"))
add("S1 Squeeze ETH 15m", "Squeeze breakout puro, BBw=14, hold 3×15m, leva 3x",
"S1", run_s1_squeeze("ETH", "15m"))
add("S1 Squeeze BTC 1h", "Squeeze breakout puro, BBw=14, hold 3×1h, leva 3x",
"S1", run_s1_squeeze("BTC", "1h"))
add("S1 Squeeze ETH 1h", "Squeeze breakout puro, BBw=14, hold 3×1h, leva 3x",
"S1", run_s1_squeeze("ETH", "1h"))
add("S1 AF+Vol BTC 15m", "Squeeze + antifakeout + volume confirm >1.3× media",
"S1", run_s1_antifake_vol("BTC", "15m"))
add("S1 AF+Vol ETH 15m", "Squeeze + antifakeout + volume confirm >1.3× media",
"S1", run_s1_antifake_vol("ETH", "15m"))
add("S1 AF+Vol BTC 1h", "Squeeze + antifakeout + volume confirm >1.3× media",
"S1", run_s1_antifake_vol("BTC", "1h"))
add("S1 AF+Vol ETH 1h", "Squeeze + antifakeout + volume confirm >1.3× media",
"S1", run_s1_antifake_vol("ETH", "1h"))
# ── Script 13: Squeeze + ML (GBM walk-forward) ─────────────────
print(" Training ML models...")
add("S13 ETH 15m bb14 ml70", "Squeeze+GBM walk-forward, BBw=14 sq=0.8 ml≥0.70, 3x leva 15% pos",
"S13", run_s13_hybrid("ETH", "15m", 14, 0.8, 3, 3, 0.15, 0.70))
add("S13 ETH 15m bb14 ml65", "Squeeze+GBM walk-forward, BBw=14 sq=0.8 ml≥0.65, 3x leva 15% pos",
"S13", run_s13_hybrid("ETH", "15m", 14, 0.8, 3, 3, 0.15, 0.65))
add("S13 ETH 15m bb20 ml70", "Squeeze+GBM walk-forward, BBw=20 sq=0.9 ml≥0.70, 3x leva 15% pos",
"S13", run_s13_hybrid("ETH", "15m", 20, 0.9, 3, 3, 0.15, 0.70))
add("S13 BTC 15m bb14 ml70", "Squeeze+GBM walk-forward, BBw=14 sq=0.9 ml≥0.70, 3x leva 15% pos",
"S13", run_s13_hybrid("BTC", "15m", 14, 0.9, 3, 3, 0.15, 0.70))
add("S13 BTC 15m bb14 ml65", "Squeeze+GBM walk-forward, BBw=14 sq=0.9 ml≥0.65, 3x leva 15% pos",
"S13", run_s13_hybrid("BTC", "15m", 14, 0.9, 3, 3, 0.15, 0.65))
add("S13 BTC 1h bb14 ml70", "Squeeze+GBM walk-forward, BBw=14 sq=0.8 ml≥0.70, 3x leva 20% pos",
"S13", run_s13_hybrid("BTC", "1h", 14, 0.8, 3, 3, 0.20, 0.70))
add("S13 BTC 1h bb14 ml65", "Squeeze+GBM walk-forward, BBw=14 sq=0.8 ml≥0.65, 3x leva 20% pos",
"S13", run_s13_hybrid("BTC", "1h", 14, 0.8, 3, 3, 0.20, 0.65))
add("S13 ETH 1h bb14 ml70", "Squeeze+GBM walk-forward, BBw=14 sq=0.8 ml≥0.70, 3x leva 20% pos",
"S13", run_s13_hybrid("ETH", "1h", 14, 0.8, 3, 3, 0.20, 0.70))
strategies.sort(key=lambda x: x["acc"], reverse=True)
# ═══════════════════════════════════════════════════════════════════
# TABELLA 1: Classifica
# ═══════════════════════════════════════════════════════════════════
W = 150
print("=" * W)
print(" S1 (SQUEEZE PURO) vs S13 (SQUEEZE + GBM) — CLASSIFICA FINALE")
print(f" Fee: 0.2% RT. Dati OHLCV reali 2018-2026. Position 15%. Leva 3x.")
print("=" * W)
hdr = (f" {'#':>2s} {'Cat':>3s} {'Nome':<26s} {'Trades':>6s} {'Acc':>6s} "
f"{'PnL€':>9s} {'DD%':>6s} {'€/day':>7s} {'Sharpe':>7s} "
f"{'Mkt%':>5s} {'Dur':>6s} {'Worst':>12s} {'Anni':>4s}")
print(hdr)
print(f" {''*(W-4)}")
for idx, s in enumerate(strategies, 1):
worst = f"{s['worst_year']}({s['worst_acc']:.0f}%)"
dur_str = f"{s['avg_dur_h']:.0f}h" if s['avg_dur_h'] >= 1 else f"{s['avg_dur_h']*60:.0f}m"
tag = " ★★" if s["acc"] >= 78 else "" if s["acc"] >= 76 else ""
print(f" {idx:>2d} {s['cat']:>3s} {s['name']:<26s} {s['trades']:>6d} {s['acc']:>5.1f}% "
f"{s['pnl']:>+8.0f} {s['max_dd']:>5.1f}% {s['daily_pnl']:>+6.2f} {s['sharpe']:>7.2f} "
f"{s['time_in_market_pct']:>4.1f}% {dur_str:>6s} {worst:>12s} {s['years_active']:>4d}{tag}")
# ═══════════════════════════════════════════════════════════════════
# TABELLA 2: Descrizione
# ═══════════════════════════════════════════════════════════════════
print(f"\n\n{'=' * W}")
print(" DESCRIZIONE")
print(f"{'=' * W}")
print(f" {'#':>2s} {'Nome':<26s} {'Descrizione'}")
print(f" {''*(W-4)}")
for idx, s in enumerate(strategies, 1):
print(f" {idx:>2d} {s['name']:<26s} {s['desc']}")
# ═══════════════════════════════════════════════════════════════════
# TABELLA 3: Breakdown per anno
# ═══════════════════════════════════════════════════════════════════
top_n = min(12, len(strategies))
top = strategies[:top_n]
all_years = sorted(set(y for s in top for y in s["year_details"]))
print(f"\n\n{'=' * W}")
print(f" BREAKDOWN PER ANNO — TOP {top_n} (accuracy% / trades)")
print(f"{'=' * W}")
header = f" {'Nome':<26s}"
for y in all_years:
header += f" {y:>10d}"
print(header)
print(f" {''*(W-4)}")
for s in top:
line = f" {s['name']:<26s}"
for y in all_years:
if y in s["year_details"]:
d = s["year_details"][y]
line += f" {d['acc']:>4.0f}%/{d['trades']:<4d}"
else:
line += f" {'':>10s}"
print(line)
# ═══════════════════════════════════════════════════════════════════
# TABELLA 4: Robustezza
# ═══════════════════════════════════════════════════════════════════
print(f"\n\n{'=' * W}")
print(f" ANALISI ROBUSTEZZA")
print(f"{'=' * W}")
print(f" {'#':>2s} {'Nome':<26s} {'MinAcc':>7s} {'MaxAcc':>7s} {'Spread':>7s} "
f"{'AnniOK':>7s} {'€/trade':>8s} {'Verdict':<12s}")
print(f" {''*90}")
for idx, s in enumerate(strategies, 1):
yd = s["year_details"]
valid = {y: d for y, d in yd.items() if d["trades"] >= 10}
accs = [d["acc"] for d in (valid if valid else yd).values()]
if not accs:
continue
min_a, max_a = min(accs), max(accs)
spread = max_a - min_a
years_ok = sum(1 for a in accs if a >= 70)
avg_pnl = s["pnl"] / s["trades"] if s["trades"] > 0 else 0
n_valid = len(valid if valid else yd)
if n_valid < 4:
verdict = "⚠ CORTO"
elif min_a < 60:
verdict = "⚠ FRAGILE"
elif min_a >= 72 and s["acc"] >= 77:
verdict = "✅ SOLIDO"
elif min_a >= 65 and s["acc"] >= 74:
verdict = "~ BUONO"
else:
verdict = "~ OK"
print(f" {idx:>2d} {s['name']:<26s} {min_a:>6.1f}% {max_a:>6.1f}% {spread:>6.1f}% "
f"{years_ok:>3d}/{n_valid:<3d}{avg_pnl:>+7.1f} {verdict:<12s}")
# ═══════════════════════════════════════════════════════════════════
# VERDETTO
# ═══════════════════════════════════════════════════════════════════
print(f"\n\n{'=' * W}")
print(f" VERDETTO FINALE")
print(f"{'=' * W}")
solidi = [s for s in strategies if s["trades"] >= 200 and s["years_active"] >= 5 and s["worst_acc"] >= 65]
solidi_s1 = [s for s in solidi if s["cat"] == "S1"]
solidi_ml = [s for s in solidi if s["cat"] == "S13"]
solidi_s1.sort(key=lambda x: x["acc"], reverse=True)
solidi_ml.sort(key=lambda x: x["daily_pnl"], reverse=True)
if solidi_s1:
b = solidi_s1[0]
print(f"\n MIGLIORE S1 (regole pure, facile da deployare):")
print(f" {b['name']}{b['acc']:.1f}% acc, {b['trades']} trades, DD {b['max_dd']:.1f}%, €{b['daily_pnl']:+.2f}/day, Sharpe {b['sharpe']:.2f}")
if solidi_ml:
m = solidi_ml[0]
print(f"\n MIGLIORE S13 (squeeze+GBM, più complesso):")
print(f" {m['name']}{m['acc']:.1f}% acc, {m['trades']} trades, DD {m['max_dd']:.1f}%, €{m['daily_pnl']:+.2f}/day, Sharpe {m['sharpe']:.2f}")
max_pnl = max(strategies, key=lambda x: x["pnl"])
print(f"\n MAX PnL: {max_pnl['name']} — €{max_pnl['pnl']:+,.0f}")
-298
View File
@@ -1,298 +0,0 @@
"""Report finale: TOP 5 metodi + simulazione crescita capitale €1000 → €50/giorno."""
from __future__ import annotations
import sys
sys.path.insert(0, ".")
import numpy as np
from src.data.downloader import load_data
print("=" * 70)
print(" REPORT FINALE — TOP 5 METODI")
print(" Target: accuracy >80%, ROI annuo >30%, €50/giorno da €1000")
print("=" * 70)
# Metodo 1: Squeeze Breakout ETH 1h (BBw=20, sqThr=0.8, volume confirmed)
# Metodo 2: Squeeze Breakout ETH 1h (BBw=30, sqThr=0.9, senza vol filter)
# Metodo 3: Squeeze Breakout BTC+ETH combinato
# Metodo 4: Squeeze Breakout 15m (alta frequenza)
# Metodo 5: GBM Structural + Squeeze filter (ibrido ML + strutturale)
FEE = 0.001
LEVERAGE = 3
INITIAL = 1000
def bollinger_bandwidth(close, window=20):
n = len(close)
result = np.full(n, np.nan)
for i in range(window, n):
w = close[i-window:i]
ma = np.mean(w)
std = np.std(w)
if ma > 0:
result[i] = (2 * 2 * std) / ma
return result
def keltner_ratio(close, high, low, window=20):
n = len(close)
result = np.full(n, np.nan)
for i in range(window, n):
wc = close[i-window:i]
wh = high[i-window:i]
wl = low[i-window:i]
ma = np.mean(wc)
bb_std = np.std(wc)
tr = np.maximum(wh - wl, np.maximum(np.abs(wh - np.roll(wc,1)), np.abs(wl - np.roll(wc,1))))
atr = np.mean(tr[1:])
kc_r = (ma + 1.5*atr) - (ma - 1.5*atr)
bb_r = (ma + 2*bb_std) - (ma - 2*bb_std)
if kc_r > 0:
result[i] = bb_r / kc_r
return result
def run_squeeze_backtest(close, high, low, volume, bb_w, sq_thr, brk_bars, vol_filter, split_pct=0.7, leverage=3, pos_pct=0.2):
n = len(close)
split = int(n * split_pct)
kcr = keltner_ratio(close, high, low, bb_w)
in_sq = False
sq_start = 0
capital = float(INITIAL)
equity = [capital]
trades = []
for i in range(bb_w + 1, n):
if np.isnan(kcr[i]):
continue
is_sq = kcr[i] < sq_thr
if is_sq and not in_sq:
in_sq = True
sq_start = i
elif not is_sq and in_sq:
in_sq = False
duration = i - sq_start
if duration < 5 or i < split or i + brk_bars >= n:
continue
# Volume check
if vol_filter:
avg_v = np.mean(volume[sq_start:i])
brk_v = np.mean(volume[i:i+brk_bars])
if avg_v > 0 and brk_v < avg_v * 1.3:
continue
first_ret = (close[i] - close[i-1]) / close[i-1]
if abs(first_ret) < 0.001:
continue
direction = 1 if first_ret > 0 else -1
actual = (close[i+brk_bars-1] - close[i-1]) / close[i-1]
is_correct = (direction == 1 and actual > 0) or (direction == -1 and actual < 0)
trade_ret = actual * direction
net = trade_ret * leverage - FEE * 2 * leverage
pnl = capital * pos_pct * net
capital += pnl
capital = max(capital, 0)
equity.append(capital)
trades.append({
"correct": is_correct,
"actual_ret": actual,
"net_pnl": pnl,
"capital_after": capital,
})
if not trades:
return None
correct = sum(1 for t in trades if t["correct"])
acc = correct / len(trades) * 100
total_ret = (capital - INITIAL) / INITIAL * 100
test_candles = n - split
test_days = test_candles / 24
test_years = test_days / 365.25
ann = ((capital / INITIAL) ** (1/test_years) - 1) * 100 if test_years > 0 and capital > 0 else -100
daily_pnl = (capital - INITIAL) / test_days if test_days > 0 else 0
peak = equity[0]
max_dd = 0
for v in equity:
if v > peak: peak = v
dd = (peak - v) / peak if peak > 0 else 0
max_dd = max(max_dd, dd)
return {
"trades": len(trades),
"accuracy": acc,
"total_return": total_ret,
"annualized": ann,
"max_drawdown": max_dd * 100,
"final_capital": capital,
"daily_pnl": daily_pnl,
"trades_per_year": len(trades) / test_years if test_years > 0 else 0,
}
methods = []
# --- Metodo 1: ETH 1h, BBw=20, sqThr=0.8, vol confirmed ---
df_eth = load_data("ETH", "1h")
r1 = run_squeeze_backtest(df_eth["close"].values, df_eth["high"].values, df_eth["low"].values, df_eth["volume"].values,
bb_w=20, sq_thr=0.8, brk_bars=3, vol_filter=True)
methods.append(("M1: ETH 1h Squeeze+Vol (BBw=20,sq=0.8)", r1))
# --- Metodo 2: ETH 1h, BBw=30, sqThr=0.9, no vol ---
r2 = run_squeeze_backtest(df_eth["close"].values, df_eth["high"].values, df_eth["low"].values, df_eth["volume"].values,
bb_w=30, sq_thr=0.9, brk_bars=3, vol_filter=False)
methods.append(("M2: ETH 1h Squeeze (BBw=30,sq=0.9)", r2))
# --- Metodo 3: BTC+ETH combinato ---
df_btc = load_data("BTC", "1h")
r3a = run_squeeze_backtest(df_btc["close"].values, df_btc["high"].values, df_btc["low"].values, df_btc["volume"].values,
bb_w=14, sq_thr=0.8, brk_bars=3, vol_filter=False, pos_pct=0.1)
r3b = run_squeeze_backtest(df_eth["close"].values, df_eth["high"].values, df_eth["low"].values, df_eth["volume"].values,
bb_w=20, sq_thr=0.8, brk_bars=3, vol_filter=False, pos_pct=0.1)
if r3a and r3b:
combined_trades = r3a["trades"] + r3b["trades"]
combined_correct = int(r3a["accuracy"]/100 * r3a["trades"]) + int(r3b["accuracy"]/100 * r3b["trades"])
combined_acc = combined_correct / combined_trades * 100 if combined_trades > 0 else 0
# Simulate portfolio
cap = float(INITIAL)
# Rough estimate: alternate between assets
for r in [r3a, r3b]:
ret_per_trade = r["total_return"] / 100 / r["trades"] if r["trades"] > 0 else 0
for _ in range(r["trades"]):
cap *= (1 + ret_per_trade * 0.5)
r3 = {
"trades": combined_trades,
"accuracy": combined_acc,
"total_return": (cap - INITIAL) / INITIAL * 100,
"annualized": r3a["annualized"] * 0.5 + r3b["annualized"] * 0.5,
"max_drawdown": max(r3a["max_drawdown"], r3b["max_drawdown"]),
"final_capital": cap,
"daily_pnl": r3a["daily_pnl"] + r3b["daily_pnl"],
"trades_per_year": r3a["trades_per_year"] + r3b["trades_per_year"],
}
methods.append(("M3: BTC+ETH 1h Portafoglio Squeeze", r3))
# --- Metodo 4: BTC 15m alta frequenza ---
df_btc_15 = load_data("BTC", "15m")
r4 = run_squeeze_backtest(df_btc_15["close"].values, df_btc_15["high"].values, df_btc_15["low"].values, df_btc_15["volume"].values,
bb_w=14, sq_thr=0.9, brk_bars=3, vol_filter=True)
methods.append(("M4: BTC 15m Squeeze+Vol alta freq", r4))
# --- Metodo 5: ETH 1h squeeze aggressivo ---
r5 = run_squeeze_backtest(df_eth["close"].values, df_eth["high"].values, df_eth["low"].values, df_eth["volume"].values,
bb_w=20, sq_thr=0.8, brk_bars=3, vol_filter=False, leverage=3)
methods.append(("M5: ETH 1h Squeeze aggressivo (no vol)", r5))
# --- Print results ---
print("\n")
for i, (name, r) in enumerate(methods, 1):
if r is None:
print(f" {name}: NO TRADES")
continue
print(f" {'='*65}")
print(f" #{i}{name}")
print(f" {'='*65}")
print(f" Trades: {r['trades']}")
print(f" Accuracy: {r['accuracy']:.1f}% {'' if r['accuracy'] >= 80 else '⚠️' if r['accuracy'] >= 70 else ''}")
print(f" Return totale: {r['total_return']:+.1f}%")
print(f" Return annuo: {r['annualized']:+.1f}% {'' if r['annualized'] >= 30 else '⚠️' if r['annualized'] >= 15 else ''}")
print(f" Max Drawdown: {r['max_drawdown']:.1f}%")
print(f" Capitale finale: €{r['final_capital']:.0f}")
print(f" €/giorno media: €{r['daily_pnl']:.2f}")
print(f" Trades/anno: {r['trades_per_year']:.0f}")
print()
# --- Simulazione crescita 6 mesi ---
print("\n" + "=" * 70)
print(" SIMULAZIONE CRESCITA CAPITALE — 6 MESI")
print(" Metodo: M1 (ETH 1h Squeeze+Vol) — il più preciso (83.9%)")
print("=" * 70)
# M1 params: ~87 trades in ~2.5 anni test = ~35 trades/anno = ~3 al mese
# Accuracy: 83.9%, average return per trade with 3x leverage
# Simulo con dati reali: prendo i trade dal test period
close = df_eth["close"].values
high = df_eth["high"].values
low = df_eth["low"].values
volume = df_eth["volume"].values
n = len(close)
split = int(n * 0.7)
kcr = keltner_ratio(close, high, low, 20)
in_sq = False
sq_start = 0
all_trade_rets = []
for i in range(21, n):
if np.isnan(kcr[i]):
continue
is_sq = kcr[i] < 0.8
if is_sq and not in_sq:
in_sq = True
sq_start = i
elif not is_sq and in_sq:
in_sq = False
if i - sq_start < 5 or i < split or i + 3 >= n:
continue
avg_v = np.mean(volume[sq_start:i])
brk_v = np.mean(volume[i:i+3])
if avg_v > 0 and brk_v < avg_v * 1.3:
continue
first_ret = (close[i] - close[i-1]) / close[i-1]
if abs(first_ret) < 0.001:
continue
direction = 1 if first_ret > 0 else -1
actual = (close[i+2] - close[i-1]) / close[i-1]
trade_ret = actual * direction
all_trade_rets.append(trade_ret)
avg_win = np.mean([r for r in all_trade_rets if r > 0]) if any(r > 0 for r in all_trade_rets) else 0
avg_loss = np.mean([r for r in all_trade_rets if r <= 0]) if any(r <= 0 for r in all_trade_rets) else 0
win_rate = sum(1 for r in all_trade_rets if r > 0) / len(all_trade_rets)
print(f"\n Statistiche trade:")
print(f" Win rate: {win_rate*100:.1f}%")
print(f" Avg win: {avg_win*100:.2f}%")
print(f" Avg loss: {avg_loss*100:.2f}%")
print(f" Trades totali nel test: {len(all_trade_rets)}")
print(f" Trades/mese stimati: ~{len(all_trade_rets) / 30:.0f}")
print(f"\n Crescita simulata mese per mese (€1000 iniziali, leva 3x, 20% per trade):")
capital = 1000.0
monthly_trades = max(len(all_trade_rets) // 30, 3)
# Shuffle trades to simulate different sequences
np.random.seed(42)
for month in range(1, 7):
n_trades = monthly_trades
month_rets = np.random.choice(all_trade_rets, size=n_trades, replace=True)
for ret in month_rets:
net = ret * LEVERAGE - FEE * 2 * LEVERAGE
capital += capital * 0.2 * net
capital = max(capital, 10)
daily_pnl = capital * 0.003 # stima conservativa 0.3% daily basata su performance
print(f" Mese {month}: capitale €{capital:.0f}, €/giorno stima: €{daily_pnl:.1f}")
print(f"\n Capitale dopo 6 mesi: €{capital:.0f}")
print(f" €/giorno necessari: €50")
print(f" €/giorno ottenibili (0.5% daily su capitale): €{capital * 0.005:.1f}")
if capital * 0.005 >= 50:
print(f"\n ✅ TARGET RAGGIUNGIBILE: con €{capital:.0f} di capitale, 0.5% daily = €{capital*0.005:.0f}/giorno")
else:
needed = 50 / 0.005
print(f"\n ⚠️ Servono €{needed:.0f} di capitale per €50/giorno al 0.5% daily")
print(f" Raggiungibile estendendo il periodo di crescita a ~{int(np.log(needed/1000) / np.log(1 + 0.15) + 0.5)} mesi")
-175
View File
@@ -1,175 +0,0 @@
"""Strategie candidate ONESTE + sweep multi-asset/tf con verdetto.
Ogni generatore restituisce una lista di entries {i,d,tp,sl,max_bars} usando
SOLO dati fino a close[i]. L'engine (honest_lab.simulate) entra a close[i].
Famiglie testate (meccanismi distinti, per diversificazione):
MR mean-reversion single-asset (Bollinger fade, RSI revert, Z-score)
XS cross-sectional relative-value (fade della divergenza vs paniere)
MOM time-series momentum / trend su timeframe alto
SES seasonality (ora del giorno UTC)
"""
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 ( # noqa: E402
atr, rsi, ema, get_df, simulate, oos_split, verdict,
available_assets, FEE_RT,
)
# ============================================================================
# MR — mean reversion single-asset
# ============================================================================
def bollinger_fade(df, n=50, k=2.5, sl_atr=2.0, max_bars=24):
c = df["close"].values
ma = pd.Series(c).rolling(n).mean().values
sd = pd.Series(c).rolling(n).std().values
a = atr(df, 14)
up, lo = ma + k * sd, ma - k * sd
ents = []
for i in range(n + 14, len(c)):
if np.isnan(up[i]) or np.isnan(a[i]):
continue
if c[i] < lo[i] and c[i - 1] >= lo[i - 1]:
ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
elif c[i] > up[i] and c[i - 1] <= up[i - 1]:
ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
return ents
def rsi_revert(df, n=14, lo=25, hi=75, sl_atr=2.5, max_bars=24, ma_n=20):
c = df["close"].values
r = rsi(c, n)
ma = pd.Series(c).rolling(ma_n).mean().values
a = atr(df, 14)
ents = []
for i in range(max(n, ma_n) + 1, len(c)):
if np.isnan(r[i]) or np.isnan(ma[i]) or np.isnan(a[i]):
continue
if r[i - 1] < lo <= r[i]:
ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
elif r[i - 1] > hi >= r[i]:
ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
return ents
def zscore_revert(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24):
"""Entra quando close e' a |z|>z_in std dalla media; TP alla media."""
c = df["close"].values
ma = pd.Series(c).rolling(n).mean().values
sd = pd.Series(c).rolling(n).std().values
a = atr(df, 14)
z = (c - ma) / sd
ents = []
for i in range(n + 14, len(c)):
if np.isnan(z[i]) or np.isnan(a[i]) or sd[i] == 0:
continue
if z[i] <= -z_in and z[i - 1] > -z_in:
ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
elif z[i] >= z_in and z[i - 1] < z_in:
ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
return ents
# ============================================================================
# MOM — time-series momentum / trend (timeframe alto, niente breakout intrabar)
# ============================================================================
def ema_trend(df, fast=20, slow=50, sl_atr=3.0, tp_atr=10.0, max_bars=240):
"""Trend following: cross EMA fast/slow deciso a close[i], TP/SL ad ATR."""
c = df["close"].values
ef, es = ema(c, fast), ema(c, slow)
a = atr(df, 14)
ents = []
for i in range(slow + 14, len(c)):
if np.isnan(a[i]):
continue
cross_up = ef[i] > es[i] and ef[i - 1] <= es[i - 1]
cross_dn = ef[i] < es[i] and ef[i - 1] >= es[i - 1]
if cross_up:
ents.append({"i": i, "d": 1, "tp": c[i] + tp_atr * a[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
elif cross_dn:
ents.append({"i": i, "d": -1, "tp": c[i] - tp_atr * a[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
return ents
# ============================================================================
# SES — seasonality (ora del giorno UTC). Direzione fissa decisa solo dall'ora.
# ============================================================================
def time_of_day(df, hour_long=None, hour_short=None, hold=6):
"""Entra a close della candela all'ora UTC indicata, esce dopo `hold` barre
(no TP/SL: tp/sl messi a +-inf cosi' esce solo a time-limit)."""
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
c = df["close"].values
hours = ts.dt.hour.values
hour_long = set(hour_long or [])
hour_short = set(hour_short or [])
ents = []
for i in range(1, len(c)):
if hours[i] in hour_long:
ents.append({"i": i, "d": 1, "tp": np.inf, "sl": -np.inf, "max_bars": hold})
elif hours[i] in hour_short:
ents.append({"i": i, "d": -1, "tp": -np.inf, "sl": np.inf, "max_bars": hold})
return ents
# ============================================================================
# sweep
# ============================================================================
def run_sweep(generators: dict, assets: list[str], tfs: list[str]):
print("=" * 130)
print(f" HONEST LAB — NETTO fee {FEE_RT*100:.2f}% RT | leva 3x | pos 15% | OOS ultimo 30%")
print("=" * 130)
print(f" {'Strategia':<26s}{'Asset':>5s}{'TF':>5s}{'Trd':>6s}{'Win%':>7s}"
f"{'FULL%':>9s}{'OOS%':>9s}{'DD%':>6s}{'Exp%':>6s}{'AnniPos':>9s}{'OK':>4s}")
print(" " + "-" * 126)
survivors = []
for label, (fn, params) in generators.items():
for asset in assets:
for tf in tfs:
try:
df = get_df(asset, tf)
except Exception:
continue
ents = fn(df, **params)
if len(ents) < 30:
continue
full = simulate(ents, df)
_, oos_e = oos_split(ents, df)
oos = simulate(oos_e, df)
ok = verdict(full, oos)
flag = " OK" if ok else ""
print(f" {label:<26s}{asset:>5s}{tf:>5s}{full.trades:>6d}{full.win:>7.1f}"
f"{full.ret:>+9.0f}{oos.ret:>+9.0f}{full.dd:>6.0f}{full.exposure:>6.0f}"
f"{f'{full.pos_years}/{full.n_years}':>9s}{flag:>4s}")
if ok:
survivors.append((label, asset, tf, full, oos))
print(" " + "-" * 126)
return survivors
GENERATORS = {
"MR_boll n50 k2.5": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)),
"MR_boll n20 k2.5": (bollinger_fade, dict(n=20, k=2.5, sl_atr=2.0, max_bars=24)),
"MR_rsi 25/75": (rsi_revert, dict(n=14, lo=25, hi=75, sl_atr=2.5, max_bars=24)),
"MR_zscore z2.5": (zscore_revert, dict(n=50, z_in=2.5, sl_atr=2.5, max_bars=24)),
"MR_zscore z3": (zscore_revert, dict(n=50, z_in=3.0, sl_atr=2.5, max_bars=24)),
"MOM_ema 20/50": (ema_trend, dict(fast=20, slow=50, sl_atr=3.0, tp_atr=10.0, max_bars=240)),
}
if __name__ == "__main__":
assets = available_assets()
print("Asset disponibili:", assets)
survivors = run_sweep(GENERATORS, assets, ["1h", "4h"])
print(f"\n SOPRAVVISSUTI (FULL+OOS+anni+DD): {len(survivors)}")
for label, a, tf, full, oos in survivors:
print(f" {label:<26s} {a} {tf} FULL {full.ret:+.0f}% OOS {oos.ret:+.0f}% DD {full.dd:.0f}%")
-73
View File
@@ -1,73 +0,0 @@
"""Diagnostica: perche' la mean-reversion simmetrica perde su asset trending?
Test: long-only vs short-only, e MR FILTRATA DAL TREND (buy-dip in uptrend,
sell-rip in downtrend) per evitare di fadeare i trend forti.
"""
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 ( # noqa: E402
atr, ema, get_df, simulate, oos_split, available_assets, FEE_RT,
)
def zscore_entries(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24,
trend_n=0, side="both"):
"""Z-score revert con filtro trend opzionale.
trend_n>0: EMA di lungo periodo. Long solo se close>EMA (uptrend),
short solo se close<EMA (downtrend).
side: 'both' | 'long' | 'short'
"""
c = df["close"].values
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)
et = ema(c, trend_n) if trend_n > 0 else None
start = max(n + 14, trend_n + 1 if trend_n else 0)
ents = []
for i in range(start, len(c)):
if np.isnan(z[i]) or np.isnan(a[i]):
continue
long_ok = (et is None or c[i] > et[i]) and side in ("both", "long")
short_ok = (et is None or c[i] < et[i]) and side in ("both", "short")
if z[i] <= -z_in and z[i - 1] > -z_in and long_ok:
ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
elif z[i] >= z_in and z[i - 1] < z_in and short_ok:
ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
return ents
def row(label, df, ents):
if len(ents) < 20:
print(f" {label:<34s} {'<20 trd':>50s}")
return None
full = simulate(ents, df)
_, oe = oos_split(ents, df)
oos = simulate(oe, df)
print(f" {label:<34s}{full.trades:>6d}{full.win:>7.1f}{full.ret:>+9.0f}"
f"{oos.ret:>+9.0f}{full.dd:>6.0f}{f'{full.pos_years}/{full.n_years}':>8s}")
return full, oos
if __name__ == "__main__":
assets = available_assets()
print(f"HONEST DIAG — z-score revert, fee {FEE_RT*100:.2f}% RT, leva 3x | OOS 30%")
for tf in ["1h"]:
for a in assets:
df = get_df(a, tf)
print(f"\n === {a} {tf} === {'Trd':>5s}{'Win%':>7s}{'FULL%':>8s}{'OOS%':>8s}{'DD%':>6s}{'AnniP':>8s}")
base = dict(n=50, z_in=2.5, sl_atr=2.5, max_bars=24)
row("both, no filter", df, zscore_entries(df, **base, side="both"))
row("long-only, no filter", df, zscore_entries(df, **base, side="long"))
row("short-only, no filter", df, zscore_entries(df, **base, side="short"))
row("both + trend200 filter", df, zscore_entries(df, **base, trend_n=200, side="both"))
row("both + trend500 filter", df, zscore_entries(df, **base, trend_n=500, side="both"))
row("long + trend200 filter", df, zscore_entries(df, **base, trend_n=200, side="long"))
-64
View File
@@ -1,64 +0,0 @@
"""Diag2: long-MR sempre + short-MR SOLO in downtrend confermato (close<EMA_t).
Idea: il dip-buying funziona su tutti gli asset (drift rialzista crypto); lo
short funziona solo quando il trend e' gia' giu' -> shortare i rimbalzi in
downtrend, mai i rimbalzi in bull-run.
"""
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 ( # noqa: E402
atr, ema, get_df, simulate, oos_split, available_assets, FEE_RT,
)
def regime_mr(df, n=50, z_in=2.5, sl_atr=2.5, max_bars=24, trend_n=200,
allow_short=True):
"""Long su z<=-z_in SEMPRE. Short su z>=+z_in solo se close<EMA(trend_n)."""
c = df["close"].values
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)
et = ema(c, trend_n)
start = max(n + 14, trend_n + 1)
ents = []
for i in range(start, len(c)):
if np.isnan(z[i]) or np.isnan(a[i]):
continue
if z[i] <= -z_in and z[i - 1] > -z_in:
ents.append({"i": i, "d": 1, "tp": ma[i], "sl": c[i] - sl_atr * a[i], "max_bars": max_bars})
elif allow_short and z[i] >= z_in and z[i - 1] < z_in and c[i] < et[i]:
ents.append({"i": i, "d": -1, "tp": ma[i], "sl": c[i] + sl_atr * a[i], "max_bars": max_bars})
return ents
def show(label, df, ents):
if len(ents) < 20:
print(f" {label:<30s} <20 trd"); return None
full = simulate(ents, df); _, oe = oos_split(ents, df); oos = simulate(oe, df)
print(f" {label:<30s}{full.trades:>6d}{full.win:>7.1f}{full.ret:>+9.0f}"
f"{oos.ret:>+9.0f}{full.dd:>6.0f}{f'{full.pos_years}/{full.n_years}':>8s}")
return full, oos
if __name__ == "__main__":
assets = available_assets()
print(f"DIAG2 — regime MR (long sempre + short in downtrend) fee {FEE_RT*100:.2f}% leva3x OOS30%")
surv = 0
for a in assets:
df = get_df(a, "1h")
print(f"\n === {a} 1h === {'Trd':>5s}{'Win%':>7s}{'FULL%':>8s}{'OOS%':>8s}{'DD%':>6s}{'AnniP':>8s}")
show("long-only", df, regime_mr(df, allow_short=False))
r = show("long + short@downtrend200", df, regime_mr(df, trend_n=200))
show("long + short@downtrend500", df, regime_mr(df, trend_n=500))
if r and r[0].ret > 0 and r[1].ret > 0:
surv += 1
print(f"\n Asset con regime200 positivo FULL+OOS: {surv}/{len(assets)}")
-188
View File
@@ -1,188 +0,0 @@
"""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())
-74
View File
@@ -1,74 +0,0 @@
"""Tabella per-anno (PnL% e DD% intra-anno) delle versioni MIGLIORATE:
ROT02 (dual-momentum), le 3 sleeve e il PORTAFOGLIO combinato.
Tutto NETTO. Riusa gli engine di honest_improve / honest_improve2.
"""
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_improve2 import ( # noqa: E402
dip_market_gated, _daily_equity, _norm, _tr_basket_daily, _rot_daily_equity,
)
def _year_dd(eq: pd.Series) -> dict[int, float]:
out = {}
for y, g in eq.groupby(eq.index.year):
peak = g.iloc[0]; dd = 0.0
for v in g:
peak = max(peak, v); dd = max(dd, (peak - v) / peak if peak > 0 else 0.0)
out[int(y)] = dd * 100
return out
def _year_pnl(eq: pd.Series) -> dict[int, float]:
out = {}
for y, g in eq.groupby(eq.index.year):
out[int(y)] = (g.iloc[-1] / g.iloc[0] - 1) * 100
return out
def table(name, eq):
eq = _norm(eq)
dd = _year_dd(eq); pnl = _year_pnl(eq)
print(f"\n {name}")
print(f" {'Anno':>6s}{'PnL%':>9s}{'DD%':>7s}")
print(" " + "-" * 22)
for y in sorted(pnl):
print(f" {y:>6d}{pnl[y]:>+9.0f}{dd[y]:>7.0f}")
tot = (eq.iloc[-1] / eq.iloc[0] - 1) * 100
print(f" {'TOT':>6s}{tot:>+9.0f}{_year_dd(eq) and max(_year_dd(eq).values()):>7.0f}(max anno)")
if __name__ == "__main__":
print("=" * 60)
print(" RISULTATI PER ANNO — versioni migliorate (NETTO)")
print("=" * 60)
# ROT02 dal 2020 (dati paniere)
idx_rot = pd.date_range("2020-09-01", "2026-05-26", freq="1D", tz="UTC")
eq_rot = _rot_daily_equity(idx_rot)
table("ROT02 — dual-momentum rotation (1d)", eq_rot)
# sleeve + portafoglio dal 2021
idx = pd.date_range("2021-01-01", "2026-05-26", freq="1D", tz="UTC")
d = dip_market_gated("BTC", market_n=0, return_equity=True)
eq_dip = _norm(_daily_equity(d["eq_ts"], d["eq_v"], idx))
eq_tr = _norm(_tr_basket_daily(["BNB", "BTC", "DOGE", "SOL", "XRP"], idx))
eq_r2 = _norm(_rot_daily_equity(idx))
table("Sleeve DIP01 — BTC (1h)", eq_dip)
table("Sleeve TR01 — basket (4h)", eq_tr)
table("Sleeve ROT02 (1d)", eq_r2)
drets = pd.DataFrame({"DIP": eq_dip.pct_change().fillna(0),
"TR": eq_tr.pct_change().fillna(0),
"ROT": eq_r2.pct_change().fillna(0)})
combo = (1 + drets.mean(axis=1)).cumprod()
table("PORTAFOGLIO equal-weight (daily rebal)", combo)
-139
View File
@@ -1,139 +0,0 @@
"""Migliorare Acc e ridurre DD sulle fade (MR01/MR02/MR03/MR07) senza overfit.
Leve testate, ognuna misurata FULL e OOS (ultimo 30%) per non illudersi:
- vol-target sizing: size per trade ~ 1/distanza-SL -> rischio costante, DD piu' liscio
- filtro vol regime: salta i trade con ATR% in coda alta (periodi caotici)
- filtro anti-trend: non fadare contro un trend forte (|close-EMA_long|/ATR grande)
- portfolio: equity curve combinata delle 4 strategie su un conto unico
Engine fedele (ingresso close[i], exit TP/SL intrabar o time-limit, non-overlap,
capitale composto) con sizing per-trade. Numeri NETTI fee 0.10% RT, leva 3x.
"""
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 src.data.downloader import load_data
from scripts.analysis.strategy_research import bollinger_fade, atr
from scripts.analysis.strategy_research_v2 import donchian_fade, keltner_fade, return_reversal
FEE_RT, LEV, POS, INIT, OOS_FRAC = 0.001, 3.0, 0.15, 1000.0, 0.30
# config base di ogni strategia (come strategies.yml)
STRATS = {
"MR01": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)),
"MR02": (donchian_fade, dict(n=20, sl_atr=2.0, max_bars=24)),
"MR03": (keltner_fade, dict(n=30, k=2.0, sl_atr=2.0, max_bars=24)),
"MR07": (return_reversal,dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
}
STRATS_ETH3 = dict(STRATS); STRATS_ETH3["MR03"] = (keltner_fade, dict(n=50, k=2.0, sl_atr=2.0, max_bars=24))
def add_context(ents, df, ema_long=200):
"""Aggiunge a ogni entry: sl_dist_pct, atr_pct, trend_dist (|close-EMA|/ATR)."""
c = df["close"].values
a = atr(df, 14)
el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
apct = a / c
for e in ents:
i = e["i"]
e["sl_dist"] = abs(c[i] - e["sl"]) / c[i]
e["atr_pct"] = apct[i]
e["trend_dist"] = abs(c[i] - el[i]) / a[i] if a[i] else 0.0
return ents
def simulate(ents, df, fee_rt=FEE_RT, lev=LEV, split=-1,
sizer=None, vol_skip=None, trend_skip=None, max_size=0.30):
"""sizer: funzione(entry)->frazione capitale; default POS fisso.
vol_skip: soglia atr_pct sopra cui salto. trend_skip: soglia trend_dist sopra cui salto."""
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)
cap = peak = INIT
dd = 0.0; last = -1; trd = wins = 0
fee = fee_rt * lev
yearly = {}; rets = []
for e in ents:
i, d = e["i"], e["d"]
if i <= last or i + 1 >= n or i < split:
continue
if vol_skip is not None and e["atr_pct"] > vol_skip:
continue
if trend_skip is not None and e["trend_dist"] > trend_skip:
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:
exit_p = c[n - 1]; break
hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
if hs: exit_p = sl; break
if ht: exit_p = tp; break
if k == mb: exit_p = c[j]
ret = (exit_p - entry) / entry * d * lev - fee
size = POS if sizer is None else min(sizer(e), max_size)
cap = max(cap + cap * size * ret, 10.0)
peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
trd += 1; wins += ret > 0; last = j; rets.append(ret * size)
y = ts.iloc[i].year; yearly[y] = yearly.get(y, 0.0) + ret * size * INIT
sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
ret=(cap / INIT - 1) * 100, dd=dd * 100, yearly=yearly, sharpe=sharpe)
def vol_target_sizer(target=0.015):
"""size t.c. rischio (size*lev*sl_dist) ~ target; piu' largo lo stop, meno size."""
return lambda e: target / (LEV * max(e["sl_dist"], 1e-4))
def line(label, full, oos):
print(f" {label:<28s}{full['trades']:>6d}{full['acc']:>7.1f}{full['ret']:>+10.0f}{full['dd']:>7.1f}{full['sharpe']:>7.2f}"
f" | {oos['trades']:>5d}{oos['acc']:>7.1f}{oos['ret']:>+9.0f}{oos['dd']:>7.1f}{oos['sharpe']:>7.2f}")
def main():
for asset in ["BTC", "ETH"]:
df = load_data(asset, "1h")
split = int(len(df) * (1 - OOS_FRAC))
table = STRATS_ETH3 if asset == "ETH" else STRATS
# quantili vol globali per la soglia (p90)
print("\n" + "=" * 110)
print(f" {asset} 1h — leve di riduzione DD / aumento Acc | NETTO fee 0.10% RT, leva 3x")
print("=" * 110)
print(f" {'config':<28s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}{'Shrp':>7s}"
f" | {'oTrd':>5s}{'oAcc':>7s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
print(" " + "-" * 106)
for nm, (fn, params) in table.items():
ents = add_context(fn(df, **params), df)
apct = np.array([e["atr_pct"] for e in ents])
p85 = float(np.quantile(apct, 0.85))
tdist = np.array([e["trend_dist"] for e in ents])
t90 = float(np.quantile(tdist, 0.90))
base_f = simulate(ents, df); base_o = simulate(ents, df, split=split)
line(f"{nm} base", base_f, base_o)
vt_f = simulate(ents, df, sizer=vol_target_sizer()); vt_o = simulate(ents, df, split=split, sizer=vol_target_sizer())
line(f"{nm} +volTarget", vt_f, vt_o)
vs_f = simulate(ents, df, vol_skip=p85); vs_o = simulate(ents, df, split=split, vol_skip=p85)
line(f"{nm} +volSkip(p85)", vs_f, vs_o)
ts_f = simulate(ents, df, trend_skip=t90); ts_o = simulate(ents, df, split=split, trend_skip=t90)
line(f"{nm} +trendSkip(p90)", ts_f, ts_o)
allf = simulate(ents, df, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90)
allo = simulate(ents, df, split=split, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90)
line(f"{nm} +ALL", allf, allo)
print(" " + "-" * 106)
print("\n Shrp = Sharpe annuo-naive sui ritorni per-trade. oXxx = stessa metrica su OOS (ultimo 30%).")
if __name__ == "__main__":
main()
+260
View File
@@ -0,0 +1,260 @@
"""Gestione del rischio sulle fade (MR01/MR02/MR03/MR07): alzare Acc, ridurre DD.
Due analisi, ognuna misurata FULL e OOS (ultimo 30%) per non illudersi:
(A) SCREENING LEVE — confronta su ogni strategia le leve di rischio:
- vol-target sizing (size ~ 1/distanza-SL) -> SCARTATA (peggiora)
- skip alta volatilita' (ATR% in coda alta) -> SCARTATA (peggiora)
- filtro trend (|close-EMA200|/ATR oltre soglia) -> ADOTTATA (Acc+ DD-)
- combinazione di tutte
(B) FILTRO TREND + PORTAFOGLIO:
- sweep della soglia trend (assoluta in ATR, regola unica = no overfit)
- portafoglio equipesato su sotto-conti indipendenti: curve poco correlate
-> DD aggregato << DD del singolo sleeve (vera leva anti-drawdown)
Engine fedele: ingresso close[i], exit TP/SL intrabar (high/low) o time-limit,
non-overlap, capitale composto. Numeri NETTI fee 0.10% RT, leva 3x.
"""
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 src.data.downloader import load_data
from scripts.analysis.strategy_research import bollinger_fade, atr
from scripts.analysis.strategy_research_v2 import donchian_fade, keltner_fade, return_reversal
FEE_RT, LEV, POS, INIT, OOS_FRAC = 0.001, 3.0, 0.15, 1000.0, 0.30
# config base di ogni strategia (come strategies.yml); su ETH MR03 usa n=50
STRATS = {
"MR01": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)),
"MR02": (donchian_fade, dict(n=20, sl_atr=2.0, max_bars=24)),
"MR03": (keltner_fade, dict(n=30, k=2.0, sl_atr=2.0, max_bars=24)),
"MR07": (return_reversal,dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
}
STRATS_ETH = dict(STRATS); STRATS_ETH["MR03"] = (keltner_fade, dict(n=50, k=2.0, sl_atr=2.0, max_bars=24))
def strats_for(asset: str) -> dict:
return STRATS_ETH if asset == "ETH" else STRATS
# ============================ (A) SCREENING LEVE ============================
def add_context(ents, df, ema_long=200):
"""Aggiunge a ogni entry: sl_dist, atr_pct, trend_dist (|close-EMA|/ATR)."""
c = df["close"].values
a = atr(df, 14)
el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
apct = a / c
for e in ents:
i = e["i"]
e["sl_dist"] = abs(c[i] - e["sl"]) / c[i]
e["atr_pct"] = apct[i]
e["trend_dist"] = abs(c[i] - el[i]) / a[i] if a[i] else 0.0
return ents
def simulate(ents, df, fee_rt=FEE_RT, lev=LEV, split=-1,
sizer=None, vol_skip=None, trend_skip=None, max_size=0.30):
"""sizer: funzione(entry)->frazione capitale; default POS fisso.
vol_skip: soglia atr_pct sopra cui salto. trend_skip: soglia trend_dist sopra cui salto."""
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)
cap = peak = INIT
dd = 0.0; last = -1; trd = wins = 0
fee = fee_rt * lev
yearly = {}; rets = []
for e in ents:
i, d = e["i"], e["d"]
if i <= last or i + 1 >= n or i < split:
continue
if vol_skip is not None and e["atr_pct"] > vol_skip:
continue
if trend_skip is not None and e["trend_dist"] > trend_skip:
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:
exit_p = c[n - 1]; break
hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
if hs: exit_p = sl; break
if ht: exit_p = tp; break
if k == mb: exit_p = c[j]
ret = (exit_p - entry) / entry * d * lev - fee
size = POS if sizer is None else min(sizer(e), max_size)
cap = max(cap + cap * size * ret, 10.0)
peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
trd += 1; wins += ret > 0; last = j; rets.append(ret * size)
y = ts.iloc[i].year; yearly[y] = yearly.get(y, 0.0) + ret * size * INIT
sharpe = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
ret=(cap / INIT - 1) * 100, dd=dd * 100, yearly=yearly, sharpe=sharpe)
def vol_target_sizer(target=0.015):
"""size t.c. rischio (size*lev*sl_dist) ~ target; piu' largo lo stop, meno size."""
return lambda e: target / (LEV * max(e["sl_dist"], 1e-4))
def _line(label, full, oos):
print(f" {label:<28s}{full['trades']:>6d}{full['acc']:>7.1f}{full['ret']:>+10.0f}{full['dd']:>7.1f}{full['sharpe']:>7.2f}"
f" | {oos['trades']:>5d}{oos['acc']:>7.1f}{oos['ret']:>+9.0f}{oos['dd']:>7.1f}{oos['sharpe']:>7.2f}")
def screen_levers():
print("=" * 110)
print(" (A) SCREENING LEVE — vol-target / vol-skip / filtro-trend | NETTO fee 0.10% RT, leva 3x")
print("=" * 110)
for asset in ["BTC", "ETH"]:
df = load_data(asset, "1h")
split = int(len(df) * (1 - OOS_FRAC))
print(f"\n {asset} 1h")
print(f" {'config':<28s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}{'Shrp':>7s}"
f" | {'oTrd':>5s}{'oAcc':>7s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
print(" " + "-" * 106)
for nm, (fn, params) in strats_for(asset).items():
ents = add_context(fn(df, **params), df)
p85 = float(np.quantile([e["atr_pct"] for e in ents], 0.85))
t90 = float(np.quantile([e["trend_dist"] for e in ents], 0.90))
_line(f"{nm} base", simulate(ents, df), simulate(ents, df, split=split))
_line(f"{nm} +volTarget", simulate(ents, df, sizer=vol_target_sizer()),
simulate(ents, df, split=split, sizer=vol_target_sizer()))
_line(f"{nm} +volSkip(p85)", simulate(ents, df, vol_skip=p85),
simulate(ents, df, split=split, vol_skip=p85))
_line(f"{nm} +trendSkip(p90)", simulate(ents, df, trend_skip=t90),
simulate(ents, df, split=split, trend_skip=t90))
_line(f"{nm} +ALL", simulate(ents, df, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90),
simulate(ents, df, split=split, sizer=vol_target_sizer(), vol_skip=p85, trend_skip=t90))
print(" " + "-" * 106)
print("\n Esito: vol-target e vol-skip PEGGIORANO; il filtro trend e' l'unica leva utile.")
# ===================== (B) FILTRO TREND + PORTAFOGLIO =====================
def build_trades(ents, df, lev=LEV, fee_rt=FEE_RT, trend_max=None, ema_long=200):
"""Lista trade non-overlap: (entry_idx, exit_idx, ret_netto). Filtro trend opzionale."""
h, l, c = df["high"].values, df["low"].values, df["close"].values
n = len(c); a = atr(df, 14)
el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
fee = fee_rt * lev
out = []; last = -1
for e in ents:
i, d = e["i"], e["d"]
if i <= last or i + 1 >= n:
continue
if trend_max is not None and a[i] and abs(c[i] - el[i]) / a[i] > trend_max:
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:
exit_p = c[n - 1]; break
hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
if hs: exit_p = sl; break
if ht: exit_p = tp; break
if k == mb: exit_p = c[j]
ret = (exit_p - entry) / entry * d * lev - fee
out.append((i, j, ret)); last = j
return out
def metrics_single(trades, pos=POS, split=-1):
cap = peak = INIT; dd = 0.0; trd = wins = 0; rets = []
for i, j, ret in trades:
if i < split:
continue
cap = max(cap + cap * pos * ret, 10.0)
peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
trd += 1; wins += ret > 0; rets.append(ret * pos)
sh = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
ret=(cap / INIT - 1) * 100, dd=dd * 100, sharpe=sh)
def sleeve_equity(trades, n_bars, pos=POS, split=-1):
"""Equity di uno sleeve su sotto-conto indipendente (capitale INIT, pos fissa)."""
eq = np.full(n_bars, INIT, dtype=float)
cap = INIT
for i, j, ret in sorted(trades, key=lambda t: t[1]):
if i < split:
continue
cap = max(cap + cap * pos * ret, 10.0)
eq[j:] = cap
return eq
def metrics_portfolio(strat_trades, n_bars, pos=POS, split=-1):
"""Portafoglio equipesato: media di N sotto-conti indipendenti. DD sull'aggregata."""
sleeves = [sleeve_equity(tr, n_bars, pos=pos, split=split) for tr in strat_trades.values()]
agg = np.mean(sleeves, axis=0)
agg = agg[max(split, 0):]
peak = np.maximum.accumulate(agg)
dd = float(np.max((peak - agg) / peak) * 100)
trd = sum(1 for tr in strat_trades.values() for i, _, _ in tr if i >= split)
wins = sum(1 for tr in strat_trades.values() for i, _, r in tr if i >= split and r > 0)
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0, ret=(agg[-1] / INIT - 1) * 100, dd=dd)
def trend_and_portfolio():
# --- sweep soglia trend ---
print("\n" + "=" * 104)
print(" (B1) FILTRO TREND |close-EMA200|/ATR > soglia -> SALTA | NETTO fee 0.10% RT, leva 3x")
print("=" * 104)
print(f" {'Strat/Asset':<14s}{'soglia':>8s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>9s}{'DD%':>7s}"
f" | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
print(" " + "-" * 100)
for asset in ["BTC", "ETH"]:
df = load_data(asset, "1h"); split = int(len(df) * (1 - OOS_FRAC))
for nm, (fn, params) in strats_for(asset).items():
ents = fn(df, **params)
for thr in [None, 4.0, 3.0, 2.5, 2.0]:
tr = build_trades(ents, df, trend_max=thr)
f = metrics_single(tr); o = metrics_single(tr, split=split)
lab = "base" if thr is None else f"{thr}ATR"
print(f" {nm+' '+asset:<14s}{lab:>8s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+9.0f}{f['dd']:>7.1f}"
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
print(" " + "-" * 100)
# --- portafoglio equipesato (filtro trend 3.0 ATR) ---
print("\n" + "=" * 104)
print(" (B2) PORTAFOGLIO equipesato: N sotto-conti indipendenti (pos 0.15, filtro trend 3.0 ATR)")
print("=" * 104)
print(f" {'Universo':<26s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}"
f" | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}")
print(" " + "-" * 100)
all_trades = {}
for asset in ["BTC", "ETH"]:
df = load_data(asset, "1h"); split = int(len(df) * (1 - OOS_FRAC)); n = len(df)
st = {f"{nm}_{asset}": build_trades(fn(df, **p), df, trend_max=3.0) for nm, (fn, p) in strats_for(asset).items()}
all_trades.update(st)
f = metrics_portfolio(st, n); o = metrics_portfolio(st, n, split=split)
print(f" {'Portafoglio '+asset+' (4 strat)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}"
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}")
df0 = load_data("BTC", "1h"); split0 = int(len(df0) * (1 - OOS_FRAC))
f = metrics_portfolio(all_trades, len(df0)); o = metrics_portfolio(all_trades, len(df0), split=split0)
print(" " + "-" * 100)
print(f" {'GLOBALE BTC+ETH (8 sleeve)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}"
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}")
print("\n Curve poco correlate => DD aggregato molto piu' basso del singolo sleeve.")
def main():
screen_levers()
trend_and_portfolio()
if __name__ == "__main__":
main()
-163
View File
@@ -1,163 +0,0 @@
"""Affina il filtro trend (soglia assoluta ATR) e costruisce il portafoglio combinato.
Due risultati:
(1) trend filter: salta le fade quando |close-EMA200|/ATR > soglia (non fadare un
trend estremo). Soglia ASSOLUTA in multipli di ATR -> stessa regola per tutte
le strategie/asset, basso rischio di overfit. Sweep soglie, FULL e OOS.
(2) portafoglio: equity curve combinata delle 4 strategie sullo stesso conto
(rischio diviso fra N posizioni). Curve poco correlate -> DD aggregato << DD
della singola strategia. Confronto singola vs portafoglio, con/senza filtro.
"""
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 src.data.downloader import load_data
from scripts.analysis.strategy_research import bollinger_fade, atr
from scripts.analysis.strategy_research_v2 import donchian_fade, keltner_fade, return_reversal
FEE_RT, LEV, INIT, OOS_FRAC = 0.001, 3.0, 1000.0, 0.30
STRATS = {
"MR01": (bollinger_fade, dict(n=50, k=2.5, sl_atr=2.0, max_bars=24)),
"MR02": (donchian_fade, dict(n=20, sl_atr=2.0, max_bars=24)),
"MR03": (keltner_fade, dict(n=30, k=2.0, sl_atr=2.0, max_bars=24)),
"MR07": (return_reversal,dict(n=50, k=3.5, tp_atr=2.0, sl_atr=1.5, max_bars=24)),
}
STRATS_ETH = dict(STRATS); STRATS_ETH["MR03"] = (keltner_fade, dict(n=50, k=2.0, sl_atr=2.0, max_bars=24))
def build_trades(ents, df, lev=LEV, fee_rt=FEE_RT, trend_max=None, ema_long=200):
"""Ritorna lista trade non-overlap: (entry_idx, exit_idx, ret_netto). Filtro trend opzionale."""
h, l, c = df["high"].values, df["low"].values, df["close"].values
n = len(c); a = atr(df, 14)
el = pd.Series(c).ewm(span=ema_long, adjust=False).mean().values
fee = fee_rt * lev
out = []; last = -1
for e in ents:
i, d = e["i"], e["d"]
if i <= last or i + 1 >= n:
continue
if trend_max is not None and a[i] and abs(c[i] - el[i]) / a[i] > trend_max:
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:
exit_p = c[n - 1]; break
hs = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
ht = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
if hs: exit_p = sl; break
if ht: exit_p = tp; break
if k == mb: exit_p = c[j]
ret = (exit_p - entry) / entry * d * lev - fee
out.append((i, j, ret)); last = j
return out
def metrics_single(trades, ts, pos=0.15, split=-1):
cap = peak = INIT; dd = 0.0; trd = wins = 0; rets = []
for i, j, ret in trades:
if i < split:
continue
cap = max(cap + cap * pos * ret, 10.0)
peak = max(peak, cap); dd = max(dd, (peak - cap) / peak)
trd += 1; wins += ret > 0; rets.append(ret * pos)
sh = float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets))) if len(rets) > 1 and np.std(rets) > 0 else 0.0
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
ret=(cap / INIT - 1) * 100, dd=dd * 100, sharpe=sh)
def sleeve_equity(trades, n_bars, pos=0.15, split=-1):
"""Equity curve di uno sleeve su sotto-conto indipendente (capitale INIT, pos fissa).
Ritorna array lungo n_bars (step aggiornato alla chiusura di ogni trade)."""
eq = np.full(n_bars, INIT, dtype=float)
cap = INIT
for i, j, ret in sorted(trades, key=lambda t: t[1]):
if i < split:
continue
cap = max(cap + cap * pos * ret, 10.0)
eq[j:] = cap # da j in poi il sotto-conto vale cap
return eq
def metrics_portfolio(strat_trades, n_bars, ts, pos=0.15, split=-1):
"""Portafoglio equipesato: capitale diviso in N sotto-conti indipendenti, ciascuno
con la sua strategia a `pos` fisso. Equity aggregata = media dei sotto-conti (somma
normalizzata a base INIT). DD misurato sull'equity aggregata. Niente leva sovrapposta."""
sleeves = [sleeve_equity(tr, n_bars, pos=pos, split=split) for tr in strat_trades.values()]
agg = np.mean(sleeves, axis=0) # media -> base INIT, diversificazione reale
# restringi alla finestra effettiva (da split in poi se OOS)
lo = max(split, 0)
agg = agg[lo:]
peak = np.maximum.accumulate(agg)
dd = float(np.max((peak - agg) / peak) * 100)
trd = sum(1 for tr in strat_trades.values() for i, _, _ in tr if i >= split)
wins = sum(1 for tr in strat_trades.values() for i, _, r in tr if i >= split and r > 0)
return dict(trades=trd, acc=wins / trd * 100 if trd else 0.0,
ret=(agg[-1] / INIT - 1) * 100, dd=dd, sharpe=0.0)
def main():
# ---------- (1) sweep soglia trend ----------
print("=" * 104)
print(" (1) FILTRO TREND |close-EMA200|/ATR > soglia -> SALTA | NETTO fee 0.10% RT, leva 3x")
print("=" * 104)
print(f" {'Strat/Asset':<14s}{'soglia':>8s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>9s}{'DD%':>7s}"
f" | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}{'oShrp':>7s}")
print(" " + "-" * 100)
thresholds = [None, 4.0, 3.0, 2.5, 2.0]
for asset in ["BTC", "ETH"]:
df = load_data(asset, "1h"); ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
split = int(len(df) * (1 - OOS_FRAC))
table = STRATS_ETH if asset == "ETH" else STRATS
for nm, (fn, params) in table.items():
ents = fn(df, **params)
for thr in thresholds:
tr = build_trades(ents, df, trend_max=thr)
f = metrics_single(tr, ts); o = metrics_single(tr, ts, split=split)
lab = "base" if thr is None else f"{thr}ATR"
print(f" {nm+' '+asset:<14s}{lab:>8s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+9.0f}{f['dd']:>7.1f}"
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
print(" " + "-" * 100)
# ---------- (2) portafoglio combinato ----------
print("\n" + "=" * 104)
print(" (2) PORTAFOGLIO equipesato: capitale diviso in N sotto-conti indipendenti")
print(" (pos 0.15 ciascuno, filtro trend 3.0 ATR). DD aggregato vs singola strategia.")
print("=" * 104)
print(f" {'Universo':<26s}{'Trd':>6s}{'Acc%':>7s}{'Ret%':>10s}{'DD%':>7s}{'':>7s}"
f" | {'oAcc':>6s}{'oRet':>9s}{'oDD':>7s}{'':>7s}")
print(" " + "-" * 100)
all_trades = {}
for asset in ["BTC", "ETH"]:
df = load_data(asset, "1h"); ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
split = int(len(df) * (1 - OOS_FRAC)); n = len(df)
table = STRATS_ETH if asset == "ETH" else STRATS
st = {f"{nm}_{asset}": build_trades(fn(df, **p), df, trend_max=3.0) for nm, (fn, p) in table.items()}
all_trades.update(st)
f = metrics_portfolio(st, n, ts); o = metrics_portfolio(st, n, ts, split=split)
print(f" {'Portafoglio '+asset+' (4 strat)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}{f['sharpe']:>7.2f}"
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
# globale 8 sleeve
df0 = load_data("BTC", "1h"); ts0 = pd.to_datetime(df0["timestamp"], unit="ms", utc=True)
split0 = int(len(df0) * (1 - OOS_FRAC))
f = metrics_portfolio(all_trades, len(df0), ts0); o = metrics_portfolio(all_trades, len(df0), ts0, split=split0)
print(" " + "-" * 100)
print(f" {'GLOBALE BTC+ETH (8 sleeve)':<26s}{f['trades']:>6d}{f['acc']:>7.1f}{f['ret']:>+10.0f}{f['dd']:>7.1f}{f['sharpe']:>7.2f}"
f" | {o['acc']:>6.1f}{o['ret']:>+9.0f}{o['dd']:>7.1f}{o['sharpe']:>7.2f}")
print("\n Nota: ogni sleeve gira su un sotto-conto indipendente (pos 0.15); l'equity di")
print(" portafoglio e' la media dei sotto-conti. Curve poco correlate => DD aggregato")
print(" molto piu' basso del DD del singolo sleeve (la vera leva anti-drawdown).")
if __name__ == "__main__":
main()
-169
View File
@@ -1,169 +0,0 @@
"""Report accuracy per ANNO × MERCATO delle strategie migliori.
Esegue ogni strategia vincente su BTC e ETH e produce tabella
accuracy/trades per ogni anno. Permette di vedere robustezza temporale
e differenze tra mercati.
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")
import importlib.util
from pathlib import Path
STRATEGIES_DIR = Path("scripts/strategies")
def load_class(module_file, class_name):
path = STRATEGIES_DIR / f"{module_file}.py"
spec = importlib.util.spec_from_file_location(module_file, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return getattr(mod, class_name)
# (label, module, class, params, hold)
STRATEGIES = [
("SQ02 antifake+vol", "SQ02_squeeze_antifake_vol", "SqueezeAntifakeVol", {}, 3),
("MT01 ema20+vol", "MT01_squeeze_mtf_momentum", "SqueezeMTFMomentum",
{"ema_period": 20, "min_slope": 0.001, "vol_filter": True}, 3),
("PD01 vtb3 vm1.3", "PD01_price_volume_divergence", "PriceVolumeDivergence",
{}, 3),
("CM01 cb6+vol", "CM01_cross_market_momentum", "CrossMarketMomentum",
{"cross_bars": 6, "mom_min": 0.001, "use_vol": True}, 3),
("AD01 lt.65 ht.95", "AD01_adaptive_squeeze", "AdaptiveSqueeze",
{"low_thr": 0.65, "high_thr": 0.95, "use_vol": True}, 3),
]
ASSETS = ["BTC", "ETH"]
TF = "15m"
ALL_YEARS = list(range(2018, 2027))
def run():
results = {} # (label, asset) -> BacktestResult
for label, module, cls_name, params, hold in STRATEGIES:
try:
cls = load_class(module, cls_name)
except Exception as e:
print(f"SKIP {label}: {e}")
continue
strat = cls()
for asset in ASSETS:
try:
r = strat.backtest(asset, TF, hold=hold, **params)
if r:
results[(label, asset)] = r
except Exception as e:
print(f" errore {label} {asset}: {e}")
# ── Tabella ACCURACY per anno × mercato ──────────────────────────
print(f"\n{'=' * 140}")
print(f" ACCURACY PER ANNO × MERCATO — {TF} (fee 0.2% RT, leva 3x, pos 15%)")
print(f"{'=' * 140}")
header = f" {'Strategia':<22s} {'Mkt':>3s}"
for y in ALL_YEARS:
header += f" {y:>7d}"
header += f"{'TOT':>6s} {'DD%':>5s} {'Worst':>10s}"
print(header)
print(f" {'' * 136}")
for label, module, cls_name, params, hold in STRATEGIES:
for asset in ASSETS:
r = results.get((label, asset))
if not r:
continue
yd = {ys.year: ys for ys in r.yearly}
line = f" {label:<22s} {asset:>3s}"
for y in ALL_YEARS:
if y in yd:
line += f" {yd[y].accuracy:>5.0f}%↑" if yd[y].accuracy >= 80 else f" {yd[y].accuracy:>5.0f}% "
else:
line += f" {'':>7s}"
worst = r.worst_year
worst_str = f"{worst.year}({worst.accuracy:.0f}%)" if worst else "N/A"
line += f"{r.accuracy:>5.1f}% {r.max_dd:>4.1f}% {worst_str:>10s}"
print(line)
print(f" {'·' * 136}")
# ── Tabella TRADES per anno × mercato ────────────────────────────
print(f"\n{'=' * 140}")
print(f" NUMERO TRADES PER ANNO × MERCATO")
print(f"{'=' * 140}")
header = f" {'Strategia':<22s} {'Mkt':>3s}"
for y in ALL_YEARS:
header += f" {y:>7d}"
header += f"{'TOT':>6s} {'€/day':>6s}"
print(header)
print(f" {'' * 130}")
for label, module, cls_name, params, hold in STRATEGIES:
for asset in ASSETS:
r = results.get((label, asset))
if not r:
continue
yd = {ys.year: ys for ys in r.yearly}
line = f" {label:<22s} {asset:>3s}"
for y in ALL_YEARS:
if y in yd:
line += f" {yd[y].trades:>7d}"
else:
line += f" {'':>7s}"
line += f"{r.trades:>6d} {r.daily_pnl:>+6.2f}"
print(line)
print(f" {'·' * 130}")
# ── Tabella PnL per anno × mercato ──────────────────────────────
print(f"\n{'=' * 140}")
print(f" PnL € PER ANNO × MERCATO (su €1000, no compounding tra anni)")
print(f"{'=' * 140}")
header = f" {'Strategia':<22s} {'Mkt':>3s}"
for y in ALL_YEARS:
header += f" {y:>7d}"
header += f"{'TOT€':>8s}"
print(header)
print(f" {'' * 132}")
for label, module, cls_name, params, hold in STRATEGIES:
for asset in ASSETS:
r = results.get((label, asset))
if not r:
continue
yd = {ys.year: ys for ys in r.yearly}
line = f" {label:<22s} {asset:>3s}"
for y in ALL_YEARS:
if y in yd:
line += f" {yd[y].pnl:>+7.0f}"
else:
line += f" {'':>7s}"
line += f"{r.pnl:>+8.0f}"
print(line)
print(f" {'·' * 132}")
# ── Sintesi: media per anno (tutte le strategie) ────────────────
print(f"\n{'=' * 140}")
print(f" SINTESI — Accuracy media per anno (tutte le strategie, BTC+ETH)")
print(f"{'=' * 140}")
year_acc = {y: [] for y in ALL_YEARS}
for (label, asset), r in results.items():
for ys in r.yearly:
if ys.trades >= 10:
year_acc[ys.year].append(ys.accuracy)
line_y = f" {'Anno':<22s} "
line_a = f" {'Acc media':<22s} "
for y in ALL_YEARS:
accs = year_acc[y]
avg = sum(accs) / len(accs) if accs else 0
line_y += f" {y:>7d}"
line_a += f" {avg:>6.1f}%"
print(line_y)
print(line_a)
if __name__ == "__main__":
run()