From 22c60808737a3145712ffdf28e7bf5d2c32e3c00 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Fri, 29 May 2026 00:08:24 +0200 Subject: [PATCH] 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) --- CLAUDE.md | 2 +- scripts/analysis/best_yearly.py | 309 ------------- scripts/analysis/compare_strategies.py | 559 ----------------------- scripts/analysis/final_report.py | 298 ------------ scripts/analysis/honest_candidates.py | 175 ------- scripts/analysis/honest_diag.py | 73 --- scripts/analysis/honest_diag2.py | 64 --- scripts/analysis/honest_yearly.py | 188 -------- scripts/analysis/honest_yearly2.py | 74 --- scripts/analysis/risk_improvements.py | 139 ------ scripts/analysis/risk_management.py | 260 +++++++++++ scripts/analysis/risk_portfolio.py | 163 ------- scripts/analysis/yearly_market_report.py | 169 ------- 13 files changed, 261 insertions(+), 2212 deletions(-) delete mode 100644 scripts/analysis/best_yearly.py delete mode 100644 scripts/analysis/compare_strategies.py delete mode 100644 scripts/analysis/final_report.py delete mode 100644 scripts/analysis/honest_candidates.py delete mode 100644 scripts/analysis/honest_diag.py delete mode 100644 scripts/analysis/honest_diag2.py delete mode 100644 scripts/analysis/honest_yearly.py delete mode 100644 scripts/analysis/honest_yearly2.py delete mode 100644 scripts/analysis/risk_improvements.py create mode 100644 scripts/analysis/risk_management.py delete mode 100644 scripts/analysis/risk_portfolio.py delete mode 100644 scripts/analysis/yearly_market_report.py diff --git a/CLAUDE.md b/CLAUDE.md index 46150f1..5bd69d5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -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`, `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%, -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. Leva non robusta scartate: vol-target sizing e skip-alta-volatilità (peggiorano). diff --git a/scripts/analysis/best_yearly.py b/scripts/analysis/best_yearly.py deleted file mode 100644 index 8b83b99..0000000 --- a/scripts/analysis/best_yearly.py +++ /dev/null @@ -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%") diff --git a/scripts/analysis/compare_strategies.py b/scripts/analysis/compare_strategies.py deleted file mode 100644 index 143e326..0000000 --- a/scripts/analysis/compare_strategies.py +++ /dev/null @@ -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}") diff --git a/scripts/analysis/final_report.py b/scripts/analysis/final_report.py deleted file mode 100644 index 8323a73..0000000 --- a/scripts/analysis/final_report.py +++ /dev/null @@ -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") diff --git a/scripts/analysis/honest_candidates.py b/scripts/analysis/honest_candidates.py deleted file mode 100644 index 8abaeb6..0000000 --- a/scripts/analysis/honest_candidates.py +++ /dev/null @@ -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}%") diff --git a/scripts/analysis/honest_diag.py b/scripts/analysis/honest_diag.py deleted file mode 100644 index 8132981..0000000 --- a/scripts/analysis/honest_diag.py +++ /dev/null @@ -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 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")) diff --git a/scripts/analysis/honest_diag2.py b/scripts/analysis/honest_diag2.py deleted file mode 100644 index 0d66aea..0000000 --- a/scripts/analysis/honest_diag2.py +++ /dev/null @@ -1,64 +0,0 @@ -"""Diag2: long-MR sempre + short-MR SOLO in downtrend confermato (close 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 -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)}") diff --git a/scripts/analysis/honest_yearly.py b/scripts/analysis/honest_yearly.py deleted file mode 100644 index 1983939..0000000 --- a/scripts/analysis/honest_yearly.py +++ /dev/null @@ -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()) diff --git a/scripts/analysis/honest_yearly2.py b/scripts/analysis/honest_yearly2.py deleted file mode 100644 index 2f0c6d7..0000000 --- a/scripts/analysis/honest_yearly2.py +++ /dev/null @@ -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) diff --git a/scripts/analysis/risk_improvements.py b/scripts/analysis/risk_improvements.py deleted file mode 100644 index e66c946..0000000 --- a/scripts/analysis/risk_improvements.py +++ /dev/null @@ -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() diff --git a/scripts/analysis/risk_management.py b/scripts/analysis/risk_management.py new file mode 100644 index 0000000..be7e10e --- /dev/null +++ b/scripts/analysis/risk_management.py @@ -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() diff --git a/scripts/analysis/risk_portfolio.py b/scripts/analysis/risk_portfolio.py deleted file mode 100644 index a336c2a..0000000 --- a/scripts/analysis/risk_portfolio.py +++ /dev/null @@ -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() diff --git a/scripts/analysis/yearly_market_report.py b/scripts/analysis/yearly_market_report.py deleted file mode 100644 index 20e343c..0000000 --- a/scripts/analysis/yearly_market_report.py +++ /dev/null @@ -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()