refactor(strategie): tieni solo MR01 mean-reversion, squeeze -> waste
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>
This commit is contained in:
Adriano Dal Pastro
@@ -1,205 +0,0 @@
|
||||
"""AD01 — Adaptive Squeeze Threshold.
|
||||
|
||||
Problema SQ02: sq_threshold fisso (0.8) non si adatta al regime di volatilità.
|
||||
Soluzione: threshold adattivo basato su volatilità recente.
|
||||
|
||||
Logica:
|
||||
- Calcola volatilità rolling (std dei rendimenti su finestra 100 barre)
|
||||
- Confronta con percentile storico (rolling 500 barre)
|
||||
- Alta vol (>70° percentile) → soglia BASSA (0.65) — squeeze più "lenti"
|
||||
- Bassa vol (<30° percentile) → soglia ALTA (0.90) — squeeze "stretti"
|
||||
- Vol media → soglia standard (0.80)
|
||||
|
||||
Razionale: in mercati calmi, il BB si stringe molto → sq_threshold alto cattura
|
||||
segnali migliori. In mercati volatili, bastano squeeze minori per essere significativi.
|
||||
|
||||
Anti-overfitting: solo 3 parametri (low_thr, mid_thr, high_thr), logica deterministica.
|
||||
Eredita antifakeout + volume da SQ02.
|
||||
"""
|
||||
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, BacktestResult, YearlyStats, TF_MINUTES
|
||||
from src.strategies.indicators import keltner_ratio, ema
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
def _adaptive_sq_threshold(close: np.ndarray,
|
||||
vol_window: int = 100,
|
||||
regime_window: int = 500,
|
||||
low_thr: float = 0.65,
|
||||
mid_thr: float = 0.80,
|
||||
high_thr: float = 0.90) -> np.ndarray:
|
||||
"""Calcola sq_threshold adattivo per ogni barra."""
|
||||
n = len(close)
|
||||
lr = np.diff(np.log(np.where(close <= 0, 1e-10, close)))
|
||||
vol = np.full(n, np.nan)
|
||||
for i in range(vol_window, n):
|
||||
vol[i] = np.std(lr[i - vol_window:i])
|
||||
|
||||
# Percentile rolling della volatilità
|
||||
thresh = np.full(n, mid_thr)
|
||||
for i in range(regime_window, n):
|
||||
if np.isnan(vol[i]):
|
||||
continue
|
||||
hist = vol[i - regime_window:i]
|
||||
hist = hist[~np.isnan(hist)]
|
||||
if len(hist) < 10:
|
||||
continue
|
||||
p30 = np.percentile(hist, 30)
|
||||
p70 = np.percentile(hist, 70)
|
||||
if vol[i] < p30:
|
||||
thresh[i] = high_thr # vol bassa → soglia alta
|
||||
elif vol[i] > p70:
|
||||
thresh[i] = low_thr # vol alta → soglia bassa
|
||||
else:
|
||||
thresh[i] = mid_thr
|
||||
return thresh
|
||||
|
||||
|
||||
def _detect_adaptive_squeezes(close, high, low, kcr, adaptive_thr,
|
||||
min_dur: int = 5) -> list[dict]:
|
||||
"""Squeeze con threshold adattivo per ogni barra."""
|
||||
events = []
|
||||
in_sq = False
|
||||
sq_start = 0
|
||||
for i in range(1, len(close)):
|
||||
if np.isnan(kcr[i]) or np.isnan(adaptive_thr[i]):
|
||||
continue
|
||||
thr = adaptive_thr[i]
|
||||
is_sq = kcr[i] < 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,
|
||||
"kcr_at_release": kcr[i],
|
||||
"thr_used": adaptive_thr[i],
|
||||
})
|
||||
return events
|
||||
|
||||
|
||||
class AdaptiveSqueeze(Strategy):
|
||||
name = "AD01_adaptive_squeeze"
|
||||
description = "Squeeze con threshold adattivo a regime volatilità"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
fee_rt = 0.002
|
||||
leverage = 3.0
|
||||
position_size = 0.15
|
||||
initial_capital = 1000.0
|
||||
|
||||
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)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
low_thr = params.get("low_thr", 0.65)
|
||||
mid_thr = params.get("mid_thr", 0.80)
|
||||
high_thr = params.get("high_thr", 0.90)
|
||||
retrace_limit = params.get("retrace_limit", 0.6)
|
||||
vol_mult = params.get("vol_multiplier", 1.3)
|
||||
use_vol = params.get("use_vol", True)
|
||||
vol_window = params.get("vol_window", 100)
|
||||
regime_window = params.get("regime_window", 500)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
adaptive_thr = _adaptive_sq_threshold(
|
||||
c, vol_window, regime_window, low_thr, mid_thr, high_thr
|
||||
)
|
||||
events = _detect_adaptive_squeezes(c, h, l, kcr, adaptive_thr)
|
||||
|
||||
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
|
||||
direction = 1 if first_ret > 0 else -1
|
||||
|
||||
# Anti-fakeout
|
||||
br = h[i] - l[i]
|
||||
if br > 0:
|
||||
if direction == 1 and (h[i] - c[i]) / br > retrace_limit:
|
||||
continue
|
||||
elif direction == -1 and (c[i] - l[i]) / br > retrace_limit:
|
||||
continue
|
||||
|
||||
# Volume confirm
|
||||
if use_vol:
|
||||
sq_start = ev["sq_start"]
|
||||
avg_sq_v = np.mean(v[sq_start:i])
|
||||
if avg_sq_v > 0 and v[i] <= avg_sq_v * vol_mult:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i,
|
||||
direction=direction,
|
||||
entry_price=c[i - 1],
|
||||
metadata={
|
||||
"dur": ev["dur"],
|
||||
"thr_used": ev.get("thr_used", mid_thr),
|
||||
},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = AdaptiveSqueeze()
|
||||
|
||||
configs = [
|
||||
# low_thr, mid_thr, high_thr, use_vol
|
||||
{"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": True},
|
||||
{"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90, "use_vol": False},
|
||||
{"low_thr": 0.60, "mid_thr": 0.78, "high_thr": 0.92, "use_vol": True},
|
||||
{"low_thr": 0.70, "mid_thr": 0.82, "high_thr": 0.90, "use_vol": True},
|
||||
{"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.95, "use_vol": True},
|
||||
{"low_thr": 0.65, "mid_thr": 0.80, "high_thr": 0.90,
|
||||
"use_vol": True, "vol_multiplier": 1.2},
|
||||
]
|
||||
|
||||
all_results = []
|
||||
for cfg in configs:
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for tf in ["15m", "1h"]:
|
||||
for hold in [3, 6]:
|
||||
r = strategy.backtest(asset, tf, hold=hold, **cfg)
|
||||
if r and r.trades >= 20:
|
||||
lbl = (f"AD01 lt={cfg['low_thr']} ht={cfg['high_thr']} "
|
||||
f"v={cfg['use_vol']} h={hold}")
|
||||
r.strategy_name = lbl
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
|
||||
print(f"\n{'=' * 130}")
|
||||
print(" AD01 ADAPTIVE SQUEEZE THRESHOLD — TOP 20")
|
||||
print(f"{'=' * 130}")
|
||||
print(f" {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
|
||||
f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
|
||||
f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}")
|
||||
print(f" {'─' * 120}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
|
||||
print(f"\n BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, €5.23/day, 9 anni")
|
||||
print(f" BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")
|
||||
@@ -1,183 +0,0 @@
|
||||
"""CM01 — Cross-Market Momentum Filter.
|
||||
|
||||
Squeeze su asset primario, entra SOLO se l'altro asset (BTC↔ETH)
|
||||
mostra momentum short-term nella STESSA direzione.
|
||||
|
||||
Differenza da MT01: MT01 usa EMA slope su 1h (trend lento).
|
||||
CM01 usa rendimento grezzo degli ultimi 3-6 bar sull'asset cross
|
||||
(momentum veloce, stesso timeframe).
|
||||
|
||||
Razionale: BTC e ETH sono altamente correlati ma non perfettamente.
|
||||
Se BTC fa squeeze breakout UP e anche ETH sta salendo (momentum 3-6 bar),
|
||||
la probabilità di continuazione è maggiore perché c'è consenso di mercato.
|
||||
|
||||
Anti-overfitting: 1 parametro chiave (cross_bars 3-6), logica deterministica.
|
||||
Eredita antifakeout + volume da SQ02.
|
||||
"""
|
||||
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, BacktestResult, YearlyStats
|
||||
from src.strategies.indicators import keltner_ratio, detect_squeezes
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
class CrossMarketMomentum(Strategy):
|
||||
name = "CM01_cross_momentum"
|
||||
description = "Squeeze + cross-asset short-term momentum filter"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
fee_rt = 0.002
|
||||
leverage = 3.0
|
||||
position_size = 0.15
|
||||
initial_capital = 1000.0
|
||||
|
||||
# Map asset → cross asset
|
||||
_CROSS = {"BTC": "ETH", "ETH": "BTC"}
|
||||
|
||||
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
||||
**params) -> list[Signal]:
|
||||
"""Genera segnali con cross-market momentum."""
|
||||
c = df["close"].values
|
||||
h = df["high"].values
|
||||
l = df["low"].values
|
||||
v = df["volume"].values
|
||||
n = len(c)
|
||||
ts_ms = df["timestamp"].values
|
||||
|
||||
asset = params.get("asset", "BTC")
|
||||
tf = params.get("tf", "15m")
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
retrace_limit = params.get("retrace_limit", 0.6)
|
||||
vol_mult = params.get("vol_multiplier", 1.3)
|
||||
use_vol = params.get("use_vol", True)
|
||||
cross_bars = params.get("cross_bars", 4) # barre momentum cross
|
||||
mom_min = params.get("mom_min", 0.0) # momentum minimo (0 = solo direzione)
|
||||
|
||||
# Carica cross asset
|
||||
cross_asset = self._CROSS.get(asset)
|
||||
if cross_asset is None:
|
||||
return []
|
||||
|
||||
try:
|
||||
df_cross = load_data(cross_asset, tf)
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
c_cross = df_cross["close"].values
|
||||
ts_cross_ms = df_cross["timestamp"].values
|
||||
n_cross = len(c_cross)
|
||||
|
||||
# Momentum cross: rendimento log su cross_bars barre
|
||||
cross_mom = np.full(n_cross, np.nan)
|
||||
for i in range(cross_bars, n_cross):
|
||||
if c_cross[i - cross_bars] > 0:
|
||||
cross_mom[i] = np.log(c_cross[i] / c_cross[i - cross_bars])
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
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
|
||||
direction = 1 if first_ret > 0 else -1
|
||||
|
||||
# Anti-fakeout
|
||||
br = h[i] - l[i]
|
||||
if br > 0:
|
||||
if direction == 1 and (h[i] - c[i]) / br > retrace_limit:
|
||||
continue
|
||||
elif direction == -1 and (c[i] - l[i]) / br > retrace_limit:
|
||||
continue
|
||||
|
||||
# Volume confirm
|
||||
if use_vol:
|
||||
sq_start = ev["sq_start"]
|
||||
avg_sq_v = np.mean(v[sq_start:i])
|
||||
if avg_sq_v > 0 and v[i] <= avg_sq_v * vol_mult:
|
||||
continue
|
||||
|
||||
# Cross-market momentum: trova indice cross corrispondente
|
||||
i_cross = np.searchsorted(ts_cross_ms, ts_ms[i]) - 1
|
||||
if i_cross < cross_bars or i_cross >= n_cross:
|
||||
continue
|
||||
mom = cross_mom[i_cross]
|
||||
if np.isnan(mom):
|
||||
continue
|
||||
|
||||
# Filtra per direzione concordante
|
||||
if direction == 1 and mom <= mom_min:
|
||||
continue
|
||||
if direction == -1 and mom >= -mom_min:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i,
|
||||
direction=direction,
|
||||
entry_price=c[i - 1],
|
||||
metadata={
|
||||
"dur": ev["dur"],
|
||||
"cross_mom": float(mom),
|
||||
},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = CrossMarketMomentum()
|
||||
|
||||
configs = [
|
||||
# cross_bars, mom_min, use_vol
|
||||
{"cross_bars": 3, "mom_min": 0.0, "use_vol": True},
|
||||
{"cross_bars": 4, "mom_min": 0.0, "use_vol": True},
|
||||
{"cross_bars": 6, "mom_min": 0.0, "use_vol": True},
|
||||
{"cross_bars": 4, "mom_min": 0.001, "use_vol": True},
|
||||
{"cross_bars": 4, "mom_min": 0.002, "use_vol": True},
|
||||
{"cross_bars": 4, "mom_min": 0.0, "use_vol": False},
|
||||
{"cross_bars": 3, "mom_min": 0.001, "use_vol": False},
|
||||
{"cross_bars": 6, "mom_min": 0.001, "use_vol": True},
|
||||
]
|
||||
|
||||
all_results = []
|
||||
for cfg in configs:
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for tf in ["15m", "1h"]:
|
||||
for hold in [3, 6]:
|
||||
r = strategy.backtest(asset, tf, hold=hold,
|
||||
cross_bars=cfg["cross_bars"],
|
||||
mom_min=cfg["mom_min"],
|
||||
use_vol=cfg["use_vol"])
|
||||
if r and r.trades >= 20:
|
||||
lbl = (f"CM01 cb={cfg['cross_bars']} "
|
||||
f"mm={cfg['mom_min']} v={cfg['use_vol']} h={hold}")
|
||||
r.strategy_name = lbl
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
|
||||
print(f"\n{'=' * 130}")
|
||||
print(" CM01 CROSS-MARKET MOMENTUM — TOP 20")
|
||||
print(f"{'=' * 130}")
|
||||
print(f" {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
|
||||
f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
|
||||
f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}")
|
||||
print(f" {'─' * 120}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
|
||||
print(f"\n BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, €5.23/day, 9 anni")
|
||||
print(f" BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")
|
||||
@@ -1,266 +0,0 @@
|
||||
"""ML01 — Squeeze + GBM (Gradient Boosting Machine) Walk-Forward.
|
||||
|
||||
Strategia ibrida: squeeze breakout come pre-filtro (QUANDO tradare),
|
||||
GradientBoosting su features strutturali come conferma (QUALE direzione).
|
||||
|
||||
Pipeline:
|
||||
1. Rileva squeeze release (Bollinger esce da Keltner)
|
||||
2. Estrai 44 features dalla finestra (structural multi-window + squeeze
|
||||
metadata + price position + ATR + momentum breakout)
|
||||
3. GBM walk-forward: train su 50% rolling, step 10%, predice direzione
|
||||
4. Trade solo se ML ha confidenza ≥ ml_threshold
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame
|
||||
- Parametri: bb_window (14), sq_threshold (0.8), brk_bars (3),
|
||||
ml_threshold (0.70), leverage (3), position_pct (0.15)
|
||||
|
||||
OUT:
|
||||
- BacktestResult con metriche walk-forward (no data leakage)
|
||||
- Solo periodo di test (seconda metà dati)
|
||||
|
||||
Risultati tipici:
|
||||
ETH 15m bb14 ml=0.70: 76.9% acc, 1213 trades, DD 4.2%, €13.78/day
|
||||
BTC 15m bb14 ml=0.70: 78.8% acc, 1964 trades, DD 7.0%, €5.51/day
|
||||
BTC 1h bb14 ml=0.70: 77.3% acc, 617 trades, DD 6.7%, €3.85/day
|
||||
|
||||
Note:
|
||||
- GBM = GradientBoostingClassifier di scikit-learn
|
||||
- Walk-forward: nessun look-ahead, train sempre prima di test
|
||||
- Il baseline squeeze puro ha accuracy più alta (~79.5%) ma DD peggiore
|
||||
- Il valore del ML è filtrare breakout deboli → DD ridotto
|
||||
"""
|
||||
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.strategies.base import Strategy, Signal, BacktestResult, YearlyStats, TF_MINUTES
|
||||
from src.strategies.indicators import keltner_ratio, detect_squeezes
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
def _build_features(df: pd.DataFrame, i: int, squeeze_info: dict) -> np.ndarray | None:
|
||||
"""44 features per il punto di squeeze release."""
|
||||
if i < 100:
|
||||
return None
|
||||
o, h, l, c, v = (df["open"].values, df["high"].values, df["low"].values,
|
||||
df["close"].values, df["volume"].values)
|
||||
feats = []
|
||||
for w in [12, 24, 48]:
|
||||
wc, wo = c[i-w:i], o[i-w:i]
|
||||
wh, wl, wv = h[i-w:i], l[i-w:i], v[i-w:i]
|
||||
mn, mx = wl.min(), max(wh.max(), wc.max())
|
||||
rng = mx - mn if mx - mn > 0 else 1e-10
|
||||
total = np.where(wh - wl == 0, 1e-10, wh - wl)
|
||||
body = np.abs(wc - wo) / total
|
||||
direction = np.sign(wc - wo)
|
||||
log_c = np.log(np.where(wc == 0, 1e-10, wc))
|
||||
rets = np.diff(log_c)
|
||||
v_mean = np.mean(wv)
|
||||
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):]),
|
||||
(wc[-1] - mn) / rng,
|
||||
wv[-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.get("avg_vol", 1) if sq.get("avg_vol", 0) > 0 else 1,
|
||||
np.mean(v[i:min(i+3, len(v))]) / sq.get("avg_vol", 1) if sq.get("avg_vol", 0) > 0 else 1,
|
||||
])
|
||||
h48, l48 = np.max(h[max(0, i-48):i]), 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))))
|
||||
feats.append(np.mean(tr[1:]) / c[i-1] if c[i-1] > 0 else 0)
|
||||
feats.append((c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0)
|
||||
return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
|
||||
|
||||
|
||||
class SqueezeGBM(Strategy):
|
||||
name = "ML01_squeeze_gbm"
|
||||
description = "Squeeze + GBM walk-forward — ML filtra breakout deboli"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
fee_ml = 0.001
|
||||
|
||||
def generate_signals(self, df, ts, **params):
|
||||
raise NotImplementedError("ML01 usa backtest custom con walk-forward")
|
||||
|
||||
def backtest(self, asset: str, tf: str, hold: int = 3, **params) -> BacktestResult | None:
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8 if tf == "1h" else 0.9)
|
||||
brk = params.get("brk_bars", hold)
|
||||
ml_thr = params.get("ml_threshold", 0.70)
|
||||
lev = params.get("leverage", self.leverage)
|
||||
pos = params.get("position_pct", self.position_size)
|
||||
|
||||
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)
|
||||
raw_events = detect_squeezes(close, high, low, kcr, sq_thr)
|
||||
|
||||
# Aggiungi avg_vol a ogni evento
|
||||
events = []
|
||||
for ev in raw_events:
|
||||
ev["avg_vol"] = float(np.mean(volume[ev["sq_start"]:ev["idx"]]))
|
||||
events.append(ev)
|
||||
|
||||
X_all, y_all, ev_all = [], [], []
|
||||
for ev in events:
|
||||
i = ev["idx"]
|
||||
if i + brk >= n or i < 100:
|
||||
continue
|
||||
feats = _build_features(df, i, ev)
|
||||
if feats is None:
|
||||
continue
|
||||
actual_ret = (close[i + brk - 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, y = np.array(X_all), np.array(y_all)
|
||||
TRAIN_SIZE = max(int(len(X) * 0.5), 50)
|
||||
STEP_SIZE = max(int(len(X) * 0.1), 10)
|
||||
|
||||
yearly: dict[int, dict] = {}
|
||||
capital = float(self.initial_capital)
|
||||
peak = capital
|
||||
max_dd = 0.0
|
||||
total_bars = 0
|
||||
all_t = 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 - 1] - close[i - 1]) / close[i - 1]
|
||||
|
||||
if p_up >= ml_thr:
|
||||
direction = 1
|
||||
elif p_up <= (1 - ml_thr):
|
||||
direction = -1
|
||||
else:
|
||||
continue
|
||||
|
||||
is_correct = (direction == 1 and actual_ret > 0) or (direction == -1 and actual_ret < 0)
|
||||
trade_ret = actual_ret * direction
|
||||
net = trade_ret * lev - self.fee_ml * 2 * lev
|
||||
capital += capital * pos * net
|
||||
capital = max(capital, 10)
|
||||
if capital > peak:
|
||||
peak = capital
|
||||
dd = (peak - capital) / peak
|
||||
max_dd = max(max_dd, dd)
|
||||
total_bars += brk
|
||||
|
||||
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, "pnl": 0.0}
|
||||
yearly[year]["t"] += 1
|
||||
if is_correct:
|
||||
yearly[year]["w"] += 1
|
||||
yearly[year]["pnl"] += net * self.initial_capital
|
||||
|
||||
start += STEP_SIZE
|
||||
|
||||
if all_t == 0:
|
||||
return None
|
||||
|
||||
yearly_stats = [
|
||||
YearlyStats(year=y, trades=d["t"], wins=d["w"], pnl=d["pnl"])
|
||||
for y, d in sorted(yearly.items())
|
||||
]
|
||||
|
||||
return BacktestResult(
|
||||
strategy_name=self.name,
|
||||
asset=asset,
|
||||
timeframe=tf,
|
||||
params={"bb_w": bb_w, "sq_thr": sq_thr, "ml_thr": ml_thr,
|
||||
"brk": brk, "lev": lev, "pos": pos},
|
||||
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=brk * TF_MINUTES.get(tf, 60) / 60,
|
||||
years_active=len(yearly),
|
||||
yearly=yearly_stats,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeGBM()
|
||||
print("Training ML models...\n")
|
||||
results = []
|
||||
for asset in ["ETH", "BTC"]:
|
||||
for tf in ["15m", "1h"]:
|
||||
for ml_thr in [0.65, 0.70]:
|
||||
r = strategy.backtest(asset, tf, ml_threshold=ml_thr)
|
||||
if r and r.trades >= 20:
|
||||
r.strategy_name = f"ML01 {asset} {tf} ml={ml_thr}"
|
||||
results.append(r)
|
||||
results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
|
||||
print(f"{'=' * 120}")
|
||||
print(f" ML01 SQUEEZE+GBM — RISULTATI")
|
||||
print(f"{'=' * 120}")
|
||||
for r in results:
|
||||
r.print_summary()
|
||||
if results:
|
||||
results[0].print_yearly()
|
||||
@@ -0,0 +1,167 @@
|
||||
"""MR01 — Bollinger Fade (mean-reversion).
|
||||
|
||||
L'UNICA famiglia con edge netto reale dopo l'analisi out-of-sample fee-aware
|
||||
(vedi scripts/analysis/strategy_research.py). Contrario della tesi squeeze:
|
||||
i breakout RIENTRANO, quindi si fada l'estremo verso la media.
|
||||
|
||||
Logica:
|
||||
1. Bollinger Band (window n, k deviazioni) sul close
|
||||
2. ENTRY: close esce sotto la banda inferiore -> long (o sopra la superiore -> short)
|
||||
3. EXIT: take-profit alla media mobile (il rientro atteso),
|
||||
stop-loss a sl_atr*ATR oltre l'estremo, oppure time-limit max_bars
|
||||
4. ingresso a close[i] (eseguibile dal vivo, nessun look-ahead)
|
||||
|
||||
Validazione (netto, fee 0.10% RT reale Deribit, leva 3x, OOS = ultimo 30%):
|
||||
BTC 1h n=50 k=2.5: +201% OOS, DD 15%, ~tutti gli anni positivi
|
||||
ETH 1h n=50 k=2.0: +1238% OOS, DD 23%
|
||||
Robusto su TUTTA la griglia n in {14,20,30,50} x k in {2.0,2.5,3.0}
|
||||
e su tutte le fee 0.00-0.20% RT (margine di sicurezza ampio).
|
||||
|
||||
NOTA LIVE: usa TP alla media + SL ad ATR + max_bars. Lo StrategyWorker attuale
|
||||
esce solo a hold_bars/stop -2% fisso: per tradarla come validata il worker deve
|
||||
supportare gli exit TP/SL passati in metadata (vedi metadata di ogni Signal).
|
||||
"""
|
||||
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, BacktestResult, YearlyStats, TF_MINUTES
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
def _atr(df: pd.DataFrame, n: int = 14) -> np.ndarray:
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
pc = np.roll(c, 1); pc[0] = c[0]
|
||||
tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
|
||||
return pd.Series(tr).rolling(n).mean().values
|
||||
|
||||
|
||||
class BollingerFade(Strategy):
|
||||
name = "MR01_bollinger_fade"
|
||||
description = "Mean-reversion: fada la banda di Bollinger, TP alla media"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["1h"]
|
||||
fee_rt = 0.001 # Deribit perp realistico (taker 0.05%/lato)
|
||||
leverage = 3.0
|
||||
position_size = 0.15
|
||||
initial_capital = 1000.0
|
||||
|
||||
def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
|
||||
**params) -> list[Signal]:
|
||||
c = df["close"].values
|
||||
n_len = len(c)
|
||||
bb_w = params.get("bb_window", 50)
|
||||
k = params.get("k", 2.5)
|
||||
sl_atr = params.get("sl_atr", 2.0)
|
||||
max_bars = params.get("max_bars", 24)
|
||||
|
||||
ma = pd.Series(c).rolling(bb_w).mean().values
|
||||
sd = pd.Series(c).rolling(bb_w).std().values
|
||||
a = _atr(df, 14)
|
||||
up, lo = ma + k * sd, ma - k * sd
|
||||
|
||||
signals: list[Signal] = []
|
||||
for i in range(bb_w + 14, n_len):
|
||||
if np.isnan(up[i]) or np.isnan(a[i]):
|
||||
continue
|
||||
if c[i] < lo[i] and c[i - 1] >= lo[i - 1]:
|
||||
d, sl = 1, c[i] - sl_atr * a[i]
|
||||
elif c[i] > up[i] and c[i - 1] <= up[i - 1]:
|
||||
d, sl = -1, c[i] + sl_atr * a[i]
|
||||
else:
|
||||
continue
|
||||
signals.append(Signal(
|
||||
idx=i, direction=d, entry_price=c[i],
|
||||
metadata={"tp": float(ma[i]), "sl": float(sl), "max_bars": max_bars},
|
||||
))
|
||||
return signals
|
||||
|
||||
def backtest(self, asset: str, tf: str = "1h", hold: int = 3,
|
||||
**params) -> BacktestResult | None:
|
||||
"""Backtest fedele: TP alla media / SL ad ATR / time-limit, fee+leva nette."""
|
||||
df = load_data(asset, tf)
|
||||
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
signals = self.generate_signals(df, ts, **params)
|
||||
if not signals:
|
||||
return None
|
||||
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
n = len(c)
|
||||
fee = self.fee_rt * self.leverage
|
||||
capital = peak = float(self.initial_capital)
|
||||
max_dd = 0.0
|
||||
total_bars = 0
|
||||
last_exit = -1
|
||||
yearly: dict[int, dict] = {}
|
||||
|
||||
for sig in signals:
|
||||
i, d = sig.idx, sig.direction
|
||||
if i <= last_exit or i + 1 >= n:
|
||||
continue
|
||||
entry = c[i]
|
||||
tp, sl, mb = sig.metadata["tp"], sig.metadata["sl"], sig.metadata["max_bars"]
|
||||
exit_p = c[min(i + mb, n - 1)]
|
||||
j = min(i + mb, n - 1)
|
||||
for step in range(1, mb + 1):
|
||||
j = i + step
|
||||
if j >= n:
|
||||
j = n - 1; exit_p = c[j]; break
|
||||
hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
|
||||
hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
|
||||
if hit_sl:
|
||||
exit_p = sl; break
|
||||
if hit_tp:
|
||||
exit_p = tp; break
|
||||
if step == mb:
|
||||
exit_p = c[j]
|
||||
|
||||
ret = (exit_p - entry) / entry * d * self.leverage - fee
|
||||
capital = max(capital + capital * self.position_size * ret, 10.0)
|
||||
if capital > peak:
|
||||
peak = capital
|
||||
max_dd = max(max_dd, (peak - capital) / peak)
|
||||
total_bars += (j - i)
|
||||
last_exit = j
|
||||
|
||||
year = ts.iloc[i].year
|
||||
yr = yearly.setdefault(year, {"w": 0, "t": 0, "pnl": 0.0})
|
||||
yr["t"] += 1
|
||||
if ret > 0:
|
||||
yr["w"] += 1
|
||||
yr["pnl"] += ret * self.initial_capital
|
||||
|
||||
all_t = sum(v["t"] for v in yearly.values())
|
||||
all_w = sum(v["w"] for v in yearly.values())
|
||||
if all_t == 0:
|
||||
return None
|
||||
|
||||
yearly_stats = [YearlyStats(y, v["t"], v["w"], v["pnl"]) for y, v in sorted(yearly.items())]
|
||||
return BacktestResult(
|
||||
strategy_name=self.name, asset=asset, timeframe=tf, params=params,
|
||||
trades=all_t, wins=all_w, pnl=sum(v["pnl"] for v 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=total_bars / all_t * TF_MINUTES.get(tf, 60) / 60,
|
||||
years_active=len(yearly), yearly=yearly_stats,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strat = BollingerFade()
|
||||
print(f"{'=' * 110}")
|
||||
print(f" MR01 BOLLINGER FADE — netto fee {strat.fee_rt*100:.2f}% RT, leva {strat.leverage:.0f}x")
|
||||
print(f"{'=' * 110}")
|
||||
results = []
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for k in [2.0, 2.5]:
|
||||
r = strat.backtest(asset, "1h", bb_window=50, k=k, sl_atr=2.0, max_bars=24)
|
||||
if r:
|
||||
r.strategy_name = f"MR01 {asset} 1h n50 k{k}"
|
||||
results.append(r)
|
||||
for r in results:
|
||||
r.print_summary()
|
||||
if results:
|
||||
results[0].print_yearly()
|
||||
@@ -1,261 +0,0 @@
|
||||
"""MT01 — Squeeze + Multi-Timeframe Momentum.
|
||||
|
||||
Problema SQ02: entra al breakout 15m ma non sa se il trend 1h è allineato.
|
||||
Soluzione: squeeze su 15m + conferma momentum su 1h.
|
||||
|
||||
Anti-overfitting: usa solo 2 indicatori (squeeze + EMA slope),
|
||||
nessun parametro complesso.
|
||||
|
||||
IN:
|
||||
- OHLCV 15m + 1h per lo stesso asset
|
||||
- Parametri: sq_threshold, ema_period_1h, min_slope
|
||||
|
||||
OUT:
|
||||
- Signal al breakout 15m confermato da trend 1h
|
||||
- BacktestResult
|
||||
|
||||
Logica:
|
||||
1. Squeeze release su 15m (come SQ01)
|
||||
2. Antifakeout filter (come SQ02)
|
||||
3. Check 1h: EMA slope positiva per long, negativa per short
|
||||
4. Check 1h: prezzo sopra/sotto EMA per conferma trend
|
||||
5. Entra solo se 15m e 1h concordano
|
||||
"""
|
||||
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, BacktestResult, YearlyStats, TF_MINUTES
|
||||
from src.strategies.indicators import keltner_ratio, detect_squeezes, ema
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
class SqueezeMTFMomentum(Strategy):
|
||||
name = "MT01_squeeze_mtf"
|
||||
description = "Squeeze 15m + momentum trend 1h — multi-timeframe"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m"]
|
||||
fee_rt = 0.002
|
||||
|
||||
def generate_signals(self, df, ts, **params):
|
||||
"""Genera segnali squeeze 15m confermati da trend 1h."""
|
||||
c = df["close"].values
|
||||
h = df["high"].values
|
||||
l = df["low"].values
|
||||
v = df["volume"].values
|
||||
n = len(c)
|
||||
|
||||
asset = params.get("asset", "BTC")
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
ema_period = params.get("ema_period", 50)
|
||||
min_slope_val = params.get("min_slope", 0.001)
|
||||
use_antifake = params.get("antifake", True)
|
||||
use_vol = params.get("vol_filter", False)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, 14)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
df_1h = params.get("df_1h")
|
||||
if df_1h is None:
|
||||
df_1h = load_data(asset, "1h")
|
||||
c1h = df_1h["close"].values
|
||||
ts1h_ms = df_1h["timestamp"].values
|
||||
n1h = len(c1h)
|
||||
ema_1h = ema(c1h, ema_period)
|
||||
ema_slope_arr = np.full(n1h, np.nan)
|
||||
for i in range(5, n1h):
|
||||
if not np.isnan(ema_1h[i]) and not np.isnan(ema_1h[i-5]) and ema_1h[i-5] > 0:
|
||||
ema_slope_arr[i] = (ema_1h[i] - ema_1h[i-5]) / ema_1h[i-5]
|
||||
|
||||
ts_ms = df["timestamp"].values
|
||||
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
|
||||
if use_antifake:
|
||||
br = h[i] - l[i]
|
||||
if br > 0:
|
||||
if c[i] > c[i-1] and (h[i] - c[i]) / br > 0.6:
|
||||
continue
|
||||
elif c[i] <= c[i-1] and (c[i] - l[i]) / br > 0.6:
|
||||
continue
|
||||
if use_vol:
|
||||
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
|
||||
i1h = np.searchsorted(ts1h_ms, ts_ms[i]) - 1
|
||||
if i1h < ema_period or i1h >= n1h:
|
||||
continue
|
||||
if np.isnan(ema_1h[i1h]) or np.isnan(ema_slope_arr[i1h]):
|
||||
continue
|
||||
if direction == 1:
|
||||
if c1h[i1h] < ema_1h[i1h] or ema_slope_arr[i1h] < min_slope_val:
|
||||
continue
|
||||
else:
|
||||
if c1h[i1h] > ema_1h[i1h] or ema_slope_arr[i1h] > -min_slope_val:
|
||||
continue
|
||||
|
||||
signals.append(Signal(idx=i, direction=direction, entry_price=c[i-1]))
|
||||
|
||||
return signals
|
||||
|
||||
def backtest(self, asset, tf="15m", hold=3, **params):
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
ema_period = params.get("ema_period", 50)
|
||||
min_slope = params.get("min_slope", 0.001)
|
||||
use_antifake = params.get("antifake", True)
|
||||
use_vol = params.get("vol_filter", False)
|
||||
|
||||
# Carica 15m e 1h
|
||||
df_15m = load_data(asset, "15m")
|
||||
df_1h = load_data(asset, "1h")
|
||||
|
||||
c15 = df_15m["close"].values
|
||||
h15 = df_15m["high"].values
|
||||
l15 = df_15m["low"].values
|
||||
v15 = df_15m["volume"].values
|
||||
n15 = len(c15)
|
||||
ts15 = pd.to_datetime(df_15m["timestamp"], unit="ms", utc=True)
|
||||
ts15_ms = df_15m["timestamp"].values
|
||||
|
||||
c1h = df_1h["close"].values
|
||||
ts1h_ms = df_1h["timestamp"].values
|
||||
n1h = len(c1h)
|
||||
|
||||
kcr = keltner_ratio(c15, h15, l15, 14)
|
||||
events = detect_squeezes(c15, h15, l15, kcr, sq_thr)
|
||||
|
||||
# EMA su 1h
|
||||
ema_1h = ema(c1h, ema_period)
|
||||
|
||||
# EMA slope (variazione percentuale su 5 barre)
|
||||
ema_slope = np.full(n1h, np.nan)
|
||||
for i in range(5, n1h):
|
||||
if not np.isnan(ema_1h[i]) and not np.isnan(ema_1h[i - 5]) and ema_1h[i - 5] > 0:
|
||||
ema_slope[i] = (ema_1h[i] - ema_1h[i - 5]) / ema_1h[i - 5]
|
||||
|
||||
yearly = {}
|
||||
capital = float(self.initial_capital)
|
||||
peak = capital
|
||||
max_dd = 0.0
|
||||
total_bars = 0
|
||||
|
||||
for ev in events:
|
||||
i = ev["idx"]
|
||||
if i + hold + 1 >= n15 or i < 1:
|
||||
continue
|
||||
|
||||
first_ret = (c15[i] - c15[i - 1]) / c15[i - 1] if c15[i - 1] > 0 else 0
|
||||
if abs(first_ret) < 0.001:
|
||||
continue
|
||||
|
||||
# Antifake
|
||||
if use_antifake:
|
||||
br = h15[i] - l15[i]
|
||||
if br > 0:
|
||||
if c15[i] > c15[i - 1] and (h15[i] - c15[i]) / br > 0.6:
|
||||
continue
|
||||
elif c15[i] <= c15[i - 1] and (c15[i] - l15[i]) / br > 0.6:
|
||||
continue
|
||||
|
||||
# Volume filter
|
||||
if use_vol:
|
||||
avg_v = np.mean(v15[ev["sq_start"]:i])
|
||||
if avg_v > 0 and v15[i] <= avg_v * 1.3:
|
||||
continue
|
||||
|
||||
direction = 1 if first_ret > 0 else -1
|
||||
|
||||
# Trova indice 1h corrispondente
|
||||
i1h = np.searchsorted(ts1h_ms, ts15_ms[i]) - 1
|
||||
if i1h < ema_period or i1h >= n1h or np.isnan(ema_1h[i1h]) or np.isnan(ema_slope[i1h]):
|
||||
continue
|
||||
|
||||
# Conferma trend 1h
|
||||
if direction == 1:
|
||||
if c1h[i1h] < ema_1h[i1h]:
|
||||
continue
|
||||
if ema_slope[i1h] < min_slope:
|
||||
continue
|
||||
else:
|
||||
if c1h[i1h] > ema_1h[i1h]:
|
||||
continue
|
||||
if ema_slope[i1h] > -min_slope:
|
||||
continue
|
||||
|
||||
entry = c15[i - 1]
|
||||
exit_price = c15[min(i + hold - 1, n15 - 1)]
|
||||
actual = (exit_price - entry) / entry * 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 = ts15.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="15m", 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 / n15 * 100,
|
||||
avg_trade_duration_h=hold * 15 / 60, years_active=len(yearly), yearly=yearly_stats,
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeMTFMomentum()
|
||||
|
||||
configs = [
|
||||
("ema50 sl0.1%", {"ema_period": 50, "min_slope": 0.001}),
|
||||
("ema50 sl0.05%", {"ema_period": 50, "min_slope": 0.0005}),
|
||||
("ema50 sl0.2%", {"ema_period": 50, "min_slope": 0.002}),
|
||||
("ema20 sl0.1%", {"ema_period": 20, "min_slope": 0.001}),
|
||||
("ema50 sl0.1%+vol", {"ema_period": 50, "min_slope": 0.001, "vol_filter": True}),
|
||||
("ema20 sl0.1%+vol", {"ema_period": 20, "min_slope": 0.001, "vol_filter": True}),
|
||||
("ema50 noAF", {"ema_period": 50, "min_slope": 0.001, "antifake": False}),
|
||||
("ema100 sl0.05%", {"ema_period": 100, "min_slope": 0.0005}),
|
||||
]
|
||||
|
||||
all_results = []
|
||||
for label, params in configs:
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for hold in [3, 6]:
|
||||
r = strategy.backtest(asset, "15m", hold=hold, **params)
|
||||
if r and r.trades >= 30:
|
||||
r.strategy_name = f"MT01 {label} h={hold}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
print(f"\n{'=' * 130}")
|
||||
print(f" MT01 SQUEEZE + MTF MOMENTUM — TOP 20")
|
||||
print(f"{'=' * 130}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
|
||||
print(f"\n BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, 9 anni, €5.23/day")
|
||||
@@ -1,158 +0,0 @@
|
||||
"""PD01 — Price-Volume Divergence Squeeze.
|
||||
|
||||
Estende SQ02 con volume TREND come filtro:
|
||||
- Breakout UP con volume CRESCENTE (ultimi 3 bar vs media squeeze) → ENTRA
|
||||
- Breakout UP con volume CALANTE → SALTA (divergenza bearish)
|
||||
- Viceversa per short
|
||||
|
||||
Logica anti-fakeout:
|
||||
1. Squeeze rilascio (come SQ02)
|
||||
2. Anti-fakeout candela (come SQ02)
|
||||
3. Volume al breakout > media squeeze (come SQ02)
|
||||
4. NUOVO: volume trending UP nelle ultime 3 barre prima del breakout
|
||||
|
||||
Parametri semplici, nessun overfitting.
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
class PriceVolumeDivergence(Strategy):
|
||||
name = "PD01_price_vol_div"
|
||||
description = "Squeeze + antifakeout + volume trend confirmation"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
fee_rt = 0.002
|
||||
leverage = 3.0
|
||||
position_size = 0.15
|
||||
initial_capital = 1000.0
|
||||
|
||||
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)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
retrace_limit = params.get("retrace_limit", 0.6)
|
||||
vol_mult = params.get("vol_multiplier", 1.3)
|
||||
vol_trend_bars = params.get("vol_trend_bars", 3) # barre per trend volume
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
signals = []
|
||||
for ev in events:
|
||||
i = ev["idx"]
|
||||
if i < vol_trend_bars + 1 or i >= n:
|
||||
continue
|
||||
|
||||
# Direzione breakout
|
||||
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
|
||||
|
||||
# Anti-fakeout
|
||||
br = h[i] - l[i]
|
||||
if br > 0:
|
||||
if direction == 1 and (h[i] - c[i]) / br > retrace_limit:
|
||||
continue
|
||||
elif direction == -1 and (c[i] - l[i]) / br > retrace_limit:
|
||||
continue
|
||||
|
||||
# Volume al breakout > media squeeze
|
||||
sq_start = ev["sq_start"]
|
||||
avg_sq_v = np.mean(v[sq_start:i])
|
||||
if avg_sq_v <= 0 or v[i] <= avg_sq_v * vol_mult:
|
||||
continue
|
||||
|
||||
# Volume TREND: slope delle ultime vol_trend_bars barre
|
||||
# Usa regressione lineare semplice (rank correlation del volume)
|
||||
recent_v = v[i - vol_trend_bars:i + 1] # include breakout bar
|
||||
if len(recent_v) < vol_trend_bars:
|
||||
continue
|
||||
# slope: media seconda metà vs prima metà
|
||||
mid = len(recent_v) // 2
|
||||
v_early = np.mean(recent_v[:mid])
|
||||
v_late = np.mean(recent_v[mid:])
|
||||
vol_trending_up = v_late > v_early
|
||||
vol_trending_down = v_early > v_late
|
||||
|
||||
# Concordanza: long richiede volume trending up, short trending down
|
||||
if direction == 1 and not vol_trending_up:
|
||||
continue
|
||||
if direction == -1 and not vol_trending_down:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i,
|
||||
direction=direction,
|
||||
entry_price=c[i - 1],
|
||||
metadata={
|
||||
"dur": ev["dur"],
|
||||
"vol_ratio": v[i] / avg_sq_v if avg_sq_v > 0 else 0,
|
||||
"vol_trend": v_late / v_early if v_early > 0 else 1,
|
||||
},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = PriceVolumeDivergence()
|
||||
|
||||
configs = [
|
||||
{"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6,
|
||||
"vol_multiplier": 1.3, "vol_trend_bars": 3},
|
||||
{"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6,
|
||||
"vol_multiplier": 1.2, "vol_trend_bars": 3},
|
||||
{"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.6,
|
||||
"vol_multiplier": 1.3, "vol_trend_bars": 5},
|
||||
{"bb_window": 14, "sq_threshold": 0.8, "retrace_limit": 0.5,
|
||||
"vol_multiplier": 1.3, "vol_trend_bars": 3},
|
||||
{"bb_window": 14, "sq_threshold": 0.75, "retrace_limit": 0.6,
|
||||
"vol_multiplier": 1.3, "vol_trend_bars": 3},
|
||||
{"bb_window": 20, "sq_threshold": 0.8, "retrace_limit": 0.6,
|
||||
"vol_multiplier": 1.3, "vol_trend_bars": 3},
|
||||
]
|
||||
|
||||
all_results = []
|
||||
for cfg in configs:
|
||||
for asset in ["BTC", "ETH"]:
|
||||
for tf in ["15m", "1h"]:
|
||||
for hold in [3, 6]:
|
||||
r = strategy.backtest(asset, tf, hold=hold, **cfg)
|
||||
if r and r.trades >= 20:
|
||||
lbl = (f"PD01 vtb={cfg['vol_trend_bars']} "
|
||||
f"vm={cfg['vol_multiplier']} "
|
||||
f"sq={cfg['sq_threshold']} h={hold}")
|
||||
r.strategy_name = lbl
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
|
||||
print(f"\n{'=' * 130}")
|
||||
print(" PD01 PRICE-VOLUME DIVERGENCE — TOP 20")
|
||||
print(f"{'=' * 130}")
|
||||
print(f" {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
|
||||
f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
|
||||
f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}")
|
||||
print(f" {'─' * 120}")
|
||||
for r in all_results[:20]:
|
||||
r.print_summary()
|
||||
|
||||
if all_results:
|
||||
all_results[0].print_yearly()
|
||||
|
||||
print(f"\n BENCHMARK SQ02: 79.7% acc, 1250t, DD 6.5%, €5.23/day, 9 anni")
|
||||
print(f" BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")
|
||||
@@ -1,68 +0,0 @@
|
||||
"""SQ01 — Squeeze Breakout Base.
|
||||
|
||||
Strategia strutturale: rileva compressione di volatilità (Bollinger dentro
|
||||
Keltner Channel) e segue la direzione del breakout al rilascio.
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame (da load_data)
|
||||
- Parametri: bb_window (14), sq_threshold (0.8), min_squeeze_dur (5)
|
||||
|
||||
OUT:
|
||||
- Lista di Signal con direzione breakout (+1/-1)
|
||||
- BacktestResult con equity, yearly breakdown, metriche
|
||||
|
||||
Risultati tipici:
|
||||
BTC 15m: 76.7% acc, 4062 trades, DD 6.7%, €9.32/day
|
||||
ETH 15m: 76.4% acc, 2948 trades, DD 6.2%, €10.31/day
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
class SqueezeBase(Strategy):
|
||||
name = "SQ01_squeeze_base"
|
||||
description = "Squeeze breakout puro — segui direzione al rilascio"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
|
||||
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
|
||||
n = len(c)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
min_dur = params.get("min_dur", 5)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr, min_dur)
|
||||
|
||||
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
|
||||
signals.append(Signal(
|
||||
idx=i,
|
||||
direction=1 if first_ret > 0 else -1,
|
||||
entry_price=c[i - 1],
|
||||
metadata={"dur": ev["dur"], "kcr": ev["kcr_at_release"]},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeBase()
|
||||
strategy.report()
|
||||
@@ -1,87 +0,0 @@
|
||||
"""SQ02 — Squeeze Breakout + Anti-Fakeout + Volume Confirmation.
|
||||
|
||||
Migliora SQ01 con due filtri:
|
||||
1. Anti-fakeout: scarta breakout dove la candela ritraccia >60% del range
|
||||
2. Volume confirm: volume al breakout deve essere >1.3× la media durante squeeze
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame
|
||||
- Parametri: bb_window (14), sq_threshold (0.8), retrace_limit (0.6),
|
||||
vol_multiplier (1.3)
|
||||
|
||||
OUT:
|
||||
- Lista di Signal filtrati
|
||||
- BacktestResult
|
||||
|
||||
Risultati tipici:
|
||||
BTC 15m: 79.7% acc, 1250 trades, DD 6.5%, €5.23/day — SOLIDO 9/9 anni
|
||||
ETH 15m: 78.6% acc, 942 trades, DD 3.4%, €4.33/day
|
||||
BTC 1h: 78.0% acc, 473 trades, DD 3.5%, Sharpe 6.57
|
||||
"""
|
||||
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
|
||||
|
||||
|
||||
class SqueezeAntifakeVol(Strategy):
|
||||
name = "SQ02_antifake_vol"
|
||||
description = "Squeeze + antifakeout + volume confirmation"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
|
||||
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)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
retrace_limit = params.get("retrace_limit", 0.6)
|
||||
vol_mult = params.get("vol_multiplier", 1.3)
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
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
|
||||
|
||||
br = h[i] - l[i]
|
||||
if br > 0:
|
||||
if c[i] > c[i - 1]:
|
||||
if (h[i] - c[i]) / br > retrace_limit:
|
||||
continue
|
||||
else:
|
||||
if (c[i] - l[i]) / br > retrace_limit:
|
||||
continue
|
||||
|
||||
avg_v = np.mean(v[ev["sq_start"]:i])
|
||||
if avg_v > 0 and v[i] <= avg_v * vol_mult:
|
||||
continue
|
||||
|
||||
signals.append(Signal(
|
||||
idx=i,
|
||||
direction=1 if first_ret > 0 else -1,
|
||||
entry_price=c[i - 1],
|
||||
metadata={"dur": ev["dur"], "vol_ratio": v[i] / avg_v if avg_v > 0 else 0},
|
||||
))
|
||||
return signals
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeAntifakeVol()
|
||||
strategy.report()
|
||||
@@ -1,175 +0,0 @@
|
||||
"""SQ03 — Squeeze con filtri selezionabili.
|
||||
|
||||
Ogni filtro è opzionale e attivabile via parametro. Di default attiva solo
|
||||
antifake + long_squeeze (i due filtri con miglior rapporto accuracy/trade).
|
||||
Esegue tutte le combinazioni utili e classifica.
|
||||
|
||||
Filtri disponibili:
|
||||
- antifake: scarta breakout con retrace >60% (guadagna ~+1% acc)
|
||||
- long_sq: solo squeeze durata ≥10 barre (+1% acc, dimezza trade)
|
||||
- timing: solo ore 4-16 UTC (+0.5% acc)
|
||||
- cross: asset secondario in squeeze nelle ultime 10 barre (+0.5%)
|
||||
- vol: volume al breakout >1.3× media squeeze (+1% acc)
|
||||
|
||||
IN:
|
||||
- OHLCV DataFrame (primario + secondario per cross-check)
|
||||
- Parametri: filters (lista), bb_window, sq_threshold
|
||||
|
||||
OUT:
|
||||
- BacktestResult per ogni preset di filtri
|
||||
|
||||
Risultati tipici (BTC 15m):
|
||||
antifake+long: 77.3% acc, 2179 trades
|
||||
antifake+vol: 79.7% acc, 1250 trades — SOLIDO
|
||||
ALL_FILTERS: 79.2% acc, 696 trades (restrittivo)
|
||||
"""
|
||||
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, BacktestResult, YearlyStats, TF_MINUTES
|
||||
from src.strategies.indicators import keltner_ratio, detect_squeezes
|
||||
from src.data.downloader import load_data
|
||||
|
||||
|
||||
PRESETS = {
|
||||
"antifake": ["antifake"],
|
||||
"long_sq": ["long_sq"],
|
||||
"antifake+long": ["antifake", "long_sq"],
|
||||
"antifake+vol": ["antifake", "vol"],
|
||||
"antifake+timing": ["antifake", "timing"],
|
||||
"long+timing": ["long_sq", "timing"],
|
||||
"antifake+long+time": ["antifake", "long_sq", "timing"],
|
||||
"antifake+cross": ["antifake", "cross"],
|
||||
"ALL_FILTERS": ["antifake", "long_sq", "timing", "cross"],
|
||||
}
|
||||
|
||||
|
||||
class SqueezeFiltered(Strategy):
|
||||
name = "SQ03_filtered"
|
||||
description = "Squeeze + filtri selezionabili (antifake, long, timing, cross, vol)"
|
||||
default_assets = ["BTC", "ETH"]
|
||||
default_timeframes = ["15m", "1h"]
|
||||
|
||||
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)
|
||||
|
||||
bb_w = params.get("bb_window", 14)
|
||||
sq_thr = params.get("sq_threshold", 0.8)
|
||||
filters = params.get("filters", ["antifake", "long_sq"])
|
||||
asset = params.get("asset", "BTC")
|
||||
tf = params.get("tf", "15m")
|
||||
|
||||
kcr = keltner_ratio(c, h, l, bb_w)
|
||||
events = detect_squeezes(c, h, l, kcr, sq_thr)
|
||||
|
||||
kcr2 = None
|
||||
ts2 = None
|
||||
if "cross" in filters:
|
||||
secondary = "ETH" if asset == "BTC" else "BTC"
|
||||
df2 = load_data(secondary, tf)
|
||||
kcr2 = keltner_ratio(df2["close"].values, df2["high"].values,
|
||||
df2["low"].values, bb_w)
|
||||
ts2 = df2["timestamp"].values
|
||||
|
||||
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
|
||||
|
||||
if "antifake" in filters:
|
||||
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
|
||||
|
||||
if not skip and "long_sq" in filters:
|
||||
if ev["dur"] < 10:
|
||||
skip = True
|
||||
|
||||
if not skip and "timing" in filters:
|
||||
hour = ts.iloc[i].hour
|
||||
if hour < 4 or hour > 16:
|
||||
skip = True
|
||||
|
||||
if not skip and "vol" in filters:
|
||||
avg_v = np.mean(v[ev["sq_start"]:i])
|
||||
if avg_v > 0 and v[i] <= avg_v * 1.3:
|
||||
skip = True
|
||||
|
||||
if not skip and "cross" in filters and kcr2 is not None and ts2 is not None:
|
||||
i2 = np.searchsorted(ts2, ts.values[i].astype("int64") // 10**6)
|
||||
i2 = min(i2, len(kcr2) - 1)
|
||||
cross_ok = any(
|
||||
not np.isnan(kcr2[j]) and kcr2[j] < 0.85
|
||||
for j in range(max(0, i2 - 10), i2 + 1)
|
||||
)
|
||||
if not cross_ok:
|
||||
skip = True
|
||||
|
||||
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 report_all_presets(self, assets=None, timeframes=None, hold=3):
|
||||
"""Esegue tutti i preset di filtri × asset × tf."""
|
||||
assets = assets or self.default_assets
|
||||
timeframes = timeframes or self.default_timeframes
|
||||
all_results = []
|
||||
|
||||
for preset_name, filter_list in PRESETS.items():
|
||||
for asset in assets:
|
||||
for tf in timeframes:
|
||||
r = self.backtest(asset, tf, hold, filters=filter_list)
|
||||
if r and r.trades >= 20:
|
||||
r.strategy_name = f"SQ03 {preset_name}"
|
||||
all_results.append(r)
|
||||
|
||||
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
||||
|
||||
print(f"\n{'=' * 120}")
|
||||
print(f" SQ03 SQUEEZE FILTRATO — TUTTI I PRESET ({len(all_results)} config)")
|
||||
print(f" Fee: {self.fee_rt*100:.1f}% RT | Leva: {self.leverage:.0f}x | Pos: {self.position_size*100:.0f}%")
|
||||
print(f"{'=' * 120}")
|
||||
print(f" {'Nome':<30s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
|
||||
f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
|
||||
f"{'Mkt%':>5s} {'Dur':>5s} {'Worst':>12s} {'Anni':>4s}")
|
||||
print(f" {'─' * 110}")
|
||||
|
||||
for r in all_results:
|
||||
r.print_summary()
|
||||
|
||||
if all_results:
|
||||
print(f"\n MIGLIORE: ", end="")
|
||||
best = all_results[0]
|
||||
best.print_yearly()
|
||||
|
||||
return all_results
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
strategy = SqueezeFiltered()
|
||||
strategy.report_all_presets()
|
||||
@@ -1,204 +0,0 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user