docs: confronto migliori strategie S1/S2 per anno, dati reali

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

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%")