310 lines
10 KiB
Python
310 lines
10 KiB
Python
"""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%")
|