Adriano
f42fec9fac
Nuova strategia MT01: squeeze 15m + momentum EMA 1h BTC 15m: 82.7% acc, 503 trades, DD 5.9%, 9/9 anni, worst 72% ETH 15m: 81.2% acc, 404 trades, DD 2.9%, 9/9 anni, worst 73% Strategie testate e scartate (waste W23-W28): IB01 inside bar (58.7%, no edge) DC01 donchian (48%, sotto random) SB01 retest (52%, no edge) MR01 mean reversion RSI (62.9%, DD 29%) VO01 volume spike (64.2%, DD 34%) HY01 squeeze+MR (64.6%, DD 14.5%) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
132 lines
4.1 KiB
Python
132 lines
4.1 KiB
Python
"""IB01 — Inside Bar Breakout.
|
|
|
|
Pattern di compressione a singola candela: quando una barra ha high < prev high
|
|
E low > prev low, il prezzo si sta comprimendo. Al breakout del range della
|
|
inside bar, segui la direzione.
|
|
|
|
17% delle candele 15m sono inside bars → frequenza altissima.
|
|
|
|
IN:
|
|
- OHLCV DataFrame
|
|
- Parametri: min_consecutive (N inside bars consecutivi),
|
|
volume_filter, breakout_confirm
|
|
|
|
OUT:
|
|
- Signal al breakout del range dell'inside bar
|
|
- BacktestResult
|
|
|
|
Logica:
|
|
1. Identifica N inside bars consecutivi (compressione)
|
|
2. Quando il prezzo rompe il range → entra nella direzione del breakout
|
|
3. Filtro: volume al breakout > media
|
|
4. Hold fisso
|
|
"""
|
|
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
|
|
|
|
|
|
class InsideBarBreakout(Strategy):
|
|
name = "IB01_inside_bar"
|
|
description = "Inside bar breakout — compressione a singola candela"
|
|
default_assets = ["BTC", "ETH"]
|
|
default_timeframes = ["15m", "1h"]
|
|
fee_rt = 0.002
|
|
|
|
def generate_signals(self, df, ts, **params):
|
|
c = df["close"].values
|
|
h = df["high"].values
|
|
l = df["low"].values
|
|
v = df["volume"].values
|
|
n = len(c)
|
|
|
|
min_consec = params.get("min_consecutive", 2)
|
|
use_vol = params.get("vol_filter", False)
|
|
min_range_pct = params.get("min_range_pct", 0.002)
|
|
|
|
# Volume media
|
|
vol_ma = np.full(n, np.nan)
|
|
for i in range(20, n):
|
|
vol_ma[i] = np.mean(v[i - 20:i])
|
|
|
|
signals = []
|
|
consec = 0
|
|
mother_high = 0.0
|
|
mother_low = 0.0
|
|
|
|
for i in range(1, n - 1):
|
|
is_inside = h[i] <= h[i - 1] and l[i] >= l[i - 1]
|
|
|
|
if is_inside:
|
|
if consec == 0:
|
|
mother_high = h[i - 1]
|
|
mother_low = l[i - 1]
|
|
consec += 1
|
|
else:
|
|
if consec >= min_consec:
|
|
range_pct = (mother_high - mother_low) / mother_low if mother_low > 0 else 0
|
|
if range_pct < min_range_pct:
|
|
consec = 0
|
|
continue
|
|
|
|
# Breakout detection sulla barra corrente
|
|
if c[i] > mother_high:
|
|
direction = 1
|
|
elif c[i] < mother_low:
|
|
direction = -1
|
|
else:
|
|
consec = 0
|
|
continue
|
|
|
|
# Volume filter
|
|
if use_vol and not np.isnan(vol_ma[i]):
|
|
if v[i] < vol_ma[i] * 1.2:
|
|
consec = 0
|
|
continue
|
|
|
|
signals.append(Signal(
|
|
idx=i, direction=direction, entry_price=c[i],
|
|
metadata={"consec": consec, "range_pct": round(range_pct * 100, 3)},
|
|
))
|
|
|
|
consec = 0
|
|
|
|
return signals
|
|
|
|
|
|
if __name__ == "__main__":
|
|
strategy = InsideBarBreakout()
|
|
|
|
configs = [
|
|
("2ib", {"min_consecutive": 2}),
|
|
("3ib", {"min_consecutive": 3}),
|
|
("4ib", {"min_consecutive": 4}),
|
|
("2ib+vol", {"min_consecutive": 2, "vol_filter": True}),
|
|
("3ib+vol", {"min_consecutive": 3, "vol_filter": True}),
|
|
("2ib r>0.3%", {"min_consecutive": 2, "min_range_pct": 0.003}),
|
|
("3ib r>0.3%", {"min_consecutive": 3, "min_range_pct": 0.003}),
|
|
]
|
|
|
|
all_results = []
|
|
for label, params in configs:
|
|
for asset in ["BTC", "ETH"]:
|
|
for tf in ["15m", "1h"]:
|
|
for hold in [3, 6]:
|
|
r = strategy.backtest(asset, tf, hold=hold, **params)
|
|
if r and r.trades >= 30:
|
|
r.strategy_name = f"IB01 {label} h={hold}"
|
|
all_results.append(r)
|
|
|
|
all_results.sort(key=lambda r: r.accuracy, reverse=True)
|
|
print(f"\n{'=' * 120}")
|
|
print(f" IB01 INSIDE BAR BREAKOUT — TOP 20")
|
|
print(f"{'=' * 120}")
|
|
for r in all_results[:20]:
|
|
r.print_summary()
|
|
if all_results:
|
|
all_results[0].print_yearly()
|