Adriano Dal Pastro
9879b46688
L'analisi out-of-sample fee-aware ha dimostrato che l'intera famiglia squeeze-breakout (SQ01-04, MT01, ML01, AD01, CM01, PD01) non ha edge: le accuratezze storiche 76-82% erano un artefatto di look-ahead (ingresso a close[i-1] con direzione decisa da close[i]). Sotto ingresso onesto a close[i] e fee reali tutte perdono, anche a fee zero. - nuova MR01_bollinger_fade (mean-reversion): edge netto validato OOS, robusto su griglia parametri e fino a 0.20% fee RT. BTC 1h n50 k2.5: +201% OOS, DD 15% - 9 strategie squeeze spostate in scripts/waste/ - strategy_loader + strategies.yml: solo MR01 (BTC/ETH 1h) - signal_engine.train: validazione OOS (accuratezza test + signal precision) - scripts/analysis/strategy_research.py: harness di ricerca fee-aware NOTA: lo StrategyWorker va aggiornato per usare gli exit TP/SL passati in metadata prima di tradare MR01 dal vivo (ora esce solo a hold_bars/stop fisso). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
205 lines
7.7 KiB
Python
205 lines
7.7 KiB
Python
"""SQ04 — Ultimate Squeeze — combinazione incrementale di tutti i filtri.
|
||
|
||
Testa combinazioni di filtri (antifake, long_sq, timing, cross-asset,
|
||
correlation, volume, trend alignment, volatility regime) e classifica
|
||
per accuracy.
|
||
|
||
IN:
|
||
- OHLCV DataFrame (primario + secondario)
|
||
- Parametri: bb_window, sq_threshold, lista filtri da attivare
|
||
|
||
OUT:
|
||
- BacktestResult per ogni combinazione di filtri
|
||
- Classifica globale
|
||
|
||
Risultati tipici:
|
||
BTC 15m antifake+corr: 81.6% acc (ma concentrato 2018)
|
||
BTC 15m antifake+vol: 79.7% acc, 1250 trades — robusto
|
||
ETH 1h antifake+corr: 80.7% acc (solo 2018)
|
||
"""
|
||
from __future__ import annotations
|
||
import sys
|
||
sys.path.insert(0, ".")
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
from src.strategies.base import Strategy, Signal
|
||
from src.strategies.indicators import (
|
||
keltner_ratio, detect_squeezes, ema, rv_annualized, rolling_correlation,
|
||
)
|
||
from src.data.downloader import load_data
|
||
|
||
|
||
class SqueezeUltimate(Strategy):
|
||
name = "SQ04_ultimate"
|
||
description = "Ultimate squeeze — tutti i filtri combinabili"
|
||
default_assets = ["BTC", "ETH"]
|
||
default_timeframes = ["15m", "1h"]
|
||
|
||
FILTER_PRESETS = {
|
||
"antifake+vol": ["antifake", "vol_confirm"],
|
||
"antifake+corr": ["antifake", "corr_high"],
|
||
"af+long+corr+trend": ["antifake", "long_sq", "corr_high", "trend_align"],
|
||
"ALL": ["antifake", "long_sq", "cross", "timing", "corr_high",
|
||
"vol_confirm", "trend_align", "low_rv"],
|
||
}
|
||
|
||
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
||
**params) -> list[Signal]:
|
||
c = df["close"].values
|
||
h = df["high"].values
|
||
l = df["low"].values
|
||
v = df["volume"].values
|
||
n = len(c)
|
||
|
||
asset = params.get("asset", "BTC")
|
||
tf = params.get("tf", "15m")
|
||
filters = params.get("filters", ["antifake", "vol_confirm"])
|
||
|
||
kcr = keltner_ratio(c, h, l, 14)
|
||
events = detect_squeezes(c, h, l, kcr)
|
||
|
||
secondary = "ETH" if asset == "BTC" else "BTC"
|
||
df2 = load_data(secondary, tf)
|
||
c2 = df2["close"].values
|
||
kcr2 = keltner_ratio(c2, df2["high"].values, df2["low"].values, 14)
|
||
ts2 = df2["timestamp"].values
|
||
|
||
ema_50 = ema(c, 50)
|
||
rv_48 = rv_annualized(c, 48)
|
||
corr = rolling_correlation(c, c2)
|
||
|
||
signals = []
|
||
for ev in events:
|
||
i = ev["idx"]
|
||
if i < 1 or i >= 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
|
||
|
||
skip = False
|
||
for f in filters:
|
||
if f == "antifake":
|
||
br = h[i] - l[i]
|
||
if br > 0:
|
||
if c[i] > c[i-1] and (h[i] - c[i]) / br > 0.6:
|
||
skip = True
|
||
elif c[i] <= c[i-1] and (c[i] - l[i]) / br > 0.6:
|
||
skip = True
|
||
elif f == "long_sq":
|
||
if ev["dur"] < 10:
|
||
skip = True
|
||
elif f == "timing":
|
||
if ts.iloc[i].hour < 4 or ts.iloc[i].hour > 16:
|
||
skip = True
|
||
elif f == "cross":
|
||
i2 = np.searchsorted(ts2, ts.values[i].astype("int64") // 10**6)
|
||
i2 = min(i2, len(kcr2) - 1)
|
||
if not any(not np.isnan(kcr2[j]) and kcr2[j] < 0.85
|
||
for j in range(max(0, i2 - 10), i2 + 1)):
|
||
skip = True
|
||
elif f == "corr_high":
|
||
if np.isnan(corr[i]) or abs(corr[i]) < 0.6:
|
||
skip = True
|
||
elif f == "vol_confirm":
|
||
avg_v = np.mean(v[ev["sq_start"]:i])
|
||
if avg_v > 0 and v[i] <= avg_v * 1.3:
|
||
skip = True
|
||
elif f == "trend_align":
|
||
if not np.isnan(ema_50[i]):
|
||
if first_ret > 0 and c[i] < ema_50[i]:
|
||
skip = True
|
||
elif first_ret < 0 and c[i] > ema_50[i]:
|
||
skip = True
|
||
elif f == "low_rv":
|
||
if not np.isnan(rv_48[i]) and rv_48[i] >= 1.5:
|
||
skip = True
|
||
if skip:
|
||
break
|
||
|
||
if skip:
|
||
continue
|
||
|
||
signals.append(Signal(
|
||
idx=i,
|
||
direction=1 if first_ret > 0 else -1,
|
||
entry_price=c[i - 1],
|
||
metadata={"dur": ev["dur"], "filters": filters},
|
||
))
|
||
return signals
|
||
|
||
def backtest(self, asset: str, tf: str, hold: int = 3, **params):
|
||
params.setdefault("asset", asset)
|
||
params.setdefault("tf", tf)
|
||
df = load_data(asset, tf)
|
||
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||
signals = self.generate_signals(df, ts, **params)
|
||
# Usa il backtest della base ma passando i segnali già generati
|
||
from src.strategies.base import BacktestResult, YearlyStats, TF_MINUTES
|
||
c = df["close"].values
|
||
n = len(c)
|
||
yearly: dict[int, dict] = {}
|
||
capital = float(self.initial_capital)
|
||
peak = capital
|
||
max_dd = 0.0
|
||
total_bars = 0
|
||
for sig in signals:
|
||
i = sig.idx
|
||
if i + hold >= n or i < 1:
|
||
continue
|
||
entry = sig.entry_price
|
||
exit_price = c[min(i + hold - 1, n - 1)]
|
||
actual = (exit_price - entry) / entry * sig.direction
|
||
net = actual * self.leverage - self.fee_rt * self.leverage
|
||
capital += capital * self.position_size * 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, "pnl": 0.0}
|
||
yearly[year]["t"] += 1
|
||
if actual > 0: yearly[year]["w"] += 1
|
||
yearly[year]["pnl"] += net * self.initial_capital
|
||
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
|
||
yearly_stats = [YearlyStats(y, d["t"], d["w"], d["pnl"]) for y, d in sorted(yearly.items())]
|
||
return BacktestResult(
|
||
strategy_name=self.name, asset=asset, timeframe=tf, params=params,
|
||
trades=all_t, wins=all_w, pnl=sum(d["pnl"] for d in yearly.values()),
|
||
capital=capital, initial_capital=self.initial_capital,
|
||
max_dd=max_dd * 100, time_in_market_pct=total_bars / n * 100,
|
||
avg_trade_duration_h=hold * TF_MINUTES.get(tf, 60) / 60,
|
||
years_active=len(yearly), yearly=yearly_stats,
|
||
)
|
||
|
||
def report_all_presets(self):
|
||
"""Esegue tutte le combinazioni preset × asset × tf."""
|
||
all_results = []
|
||
for preset_name, filter_list in self.FILTER_PRESETS.items():
|
||
for asset in self.default_assets:
|
||
for tf in self.default_timeframes:
|
||
r = self.backtest(asset, tf, filters=filter_list)
|
||
if r and r.trades >= 20:
|
||
r.strategy_name = f"SQ04 {preset_name}"
|
||
all_results.append(r)
|
||
|
||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||
|
||
print(f"\n{'=' * 120}")
|
||
print(f" SQ04 ULTIMATE — TUTTI I PRESET")
|
||
print(f"{'=' * 120}")
|
||
for r in all_results:
|
||
r.print_summary()
|
||
return all_results
|
||
|
||
|
||
if __name__ == "__main__":
|
||
strategy = SqueezeUltimate()
|
||
strategy.report_all_presets()
|