From 0ab3b5698a43e89aa03e9090ccd38ff5b51ada09 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Wed, 27 May 2026 11:12:47 +0200 Subject: [PATCH] docs: confronto migliori strategie S1/S2 per anno, dati reali Co-Authored-By: Claude Opus 4.7 (1M context) --- scripts/best_strategies_yearly.py | 309 ++++++++++++++++++++++++++++++ 1 file changed, 309 insertions(+) create mode 100644 scripts/best_strategies_yearly.py diff --git a/scripts/best_strategies_yearly.py b/scripts/best_strategies_yearly.py new file mode 100644 index 0000000..8b83b99 --- /dev/null +++ b/scripts/best_strategies_yearly.py @@ -0,0 +1,309 @@ +"""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%")