409 lines
14 KiB
Python
409 lines
14 KiB
Python
"""Strategia 13: Squeeze + ML ibrida.
|
|
Squeeze breakout come PRE-FILTRO (quando tradare),
|
|
ML come CONFERMA DIREZIONALE (quale direzione).
|
|
|
|
Pipeline:
|
|
1. Rileva squeeze release (Bollinger esce da Keltner)
|
|
2. Estrai features frattali/strutturali dalla finestra
|
|
3. ML predice direzione con confidenza
|
|
4. Trade SOLO se squeeze + ML concordano
|
|
|
|
Obiettivo: accuracy squeeze (>80%) + volume trade ML.
|
|
"""
|
|
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.data.downloader import load_data
|
|
from src.fractal.patterns import encode_candles
|
|
|
|
FEE = 0.001
|
|
INITIAL = 1000
|
|
|
|
|
|
def keltner_ratio(close, high, low, window=20):
|
|
n = len(close)
|
|
result = np.full(n, np.nan)
|
|
for i in range(window, n):
|
|
wc = close[i-window:i]
|
|
wh = high[i-window:i]
|
|
wl = 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_r = (ma + 1.5*atr) - (ma - 1.5*atr)
|
|
bb_r = (ma + 2*bb_std) - (ma - 2*bb_std)
|
|
if kc_r > 0:
|
|
result[i] = bb_r / kc_r
|
|
return result
|
|
|
|
|
|
def detect_squeeze_releases(close, high, low, volume, bb_w, sq_thr, min_duration=5):
|
|
kcr = keltner_ratio(close, high, low, bb_w)
|
|
events = []
|
|
in_sq = False
|
|
sq_start = 0
|
|
|
|
for i in range(bb_w + 1, len(close)):
|
|
if np.isnan(kcr[i]):
|
|
continue
|
|
is_sq = kcr[i] < sq_thr
|
|
if is_sq and not in_sq:
|
|
in_sq = True
|
|
sq_start = i
|
|
elif not is_sq and in_sq:
|
|
in_sq = False
|
|
duration = i - sq_start
|
|
if duration < min_duration:
|
|
continue
|
|
|
|
avg_vol = np.mean(volume[sq_start:i])
|
|
events.append({
|
|
"idx": i,
|
|
"squeeze_start": sq_start,
|
|
"duration": duration,
|
|
"avg_vol_squeeze": avg_vol,
|
|
"kcr_at_release": kcr[i],
|
|
})
|
|
return events
|
|
|
|
|
|
def build_features_at(df, i, squeeze_info):
|
|
"""Features per il punto di squeeze release."""
|
|
if i < 100:
|
|
return None
|
|
|
|
o = df["open"].values
|
|
h = df["high"].values
|
|
l = df["low"].values
|
|
c = df["close"].values
|
|
v = df["volume"].values
|
|
|
|
feats = []
|
|
|
|
# Structural features multi-window
|
|
for w in [12, 24, 48]:
|
|
win_c = c[i-w:i]
|
|
win_o = o[i-w:i]
|
|
win_h = h[i-w:i]
|
|
win_l = l[i-w:i]
|
|
win_v = v[i-w:i]
|
|
|
|
mn, mx = win_l.min(), max(win_h.max(), win_c.max())
|
|
rng = mx - mn if mx - mn > 0 else 1e-10
|
|
|
|
total = win_h - win_l
|
|
total = np.where(total == 0, 1e-10, total)
|
|
body = np.abs(win_c - win_o) / total
|
|
direction = np.sign(win_c - win_o)
|
|
|
|
log_c = np.log(np.where(win_c == 0, 1e-10, win_c))
|
|
rets = np.diff(log_c)
|
|
|
|
v_mean = np.mean(win_v)
|
|
|
|
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):]),
|
|
(win_c[-1] - mn) / rng,
|
|
win_v[-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,
|
|
])
|
|
|
|
# Squeeze-specific features
|
|
sq = squeeze_info
|
|
feats.extend([
|
|
sq["duration"],
|
|
sq["duration"] / 24, # durata in giorni
|
|
sq["kcr_at_release"],
|
|
v[i-1] / sq["avg_vol_squeeze"] if sq["avg_vol_squeeze"] > 0 else 1,
|
|
np.mean(v[i:min(i+3, len(v))]) / sq["avg_vol_squeeze"] if sq["avg_vol_squeeze"] > 0 else 1,
|
|
])
|
|
|
|
# Price position
|
|
h48 = np.max(h[max(0, i-48):i])
|
|
l48 = np.min(l[max(0, i-48):i])
|
|
r48 = h48 - l48
|
|
feats.append((c[i-1] - l48) / r48 if r48 > 0 else 0.5)
|
|
|
|
# ATR normalized
|
|
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))))
|
|
atr = np.mean(tr[1:])
|
|
feats.append(atr / c[i-1] if c[i-1] > 0 else 0)
|
|
|
|
# First bar momentum (la barra di breakout)
|
|
first_ret = (c[i] - c[i-1]) / c[i-1] if c[i-1] > 0 else 0
|
|
feats.append(first_ret)
|
|
|
|
return np.nan_to_num(np.array(feats), nan=0, posinf=1e6, neginf=-1e6)
|
|
|
|
|
|
def run_hybrid(asset, tf, bb_w, sq_thr, brk_bars, leverage, pos_pct):
|
|
print(f"\n{'='*65}")
|
|
print(f" {asset} {tf} — HYBRID Squeeze+ML (BBw={bb_w}, sq={sq_thr}, brk={brk_bars})")
|
|
print(f" Leverage: {leverage}x, Position: {pos_pct*100:.0f}%")
|
|
print(f"{'='*65}")
|
|
|
|
df = load_data(asset, tf)
|
|
close = df["close"].values
|
|
high = df["high"].values
|
|
low = df["low"].values
|
|
volume = df["volume"].values
|
|
n = len(df)
|
|
|
|
events = detect_squeeze_releases(close, high, low, volume, bb_w, sq_thr)
|
|
print(f" Squeeze releases totali: {len(events)}")
|
|
|
|
# Build dataset: solo ai punti di squeeze
|
|
X_all, y_all, ev_all = [], [], []
|
|
for ev in events:
|
|
i = ev["idx"]
|
|
if i + brk_bars >= n or i < 100:
|
|
continue
|
|
feats = build_features_at(df, i, ev)
|
|
if feats is None:
|
|
continue
|
|
actual_ret = (close[i + brk_bars - 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:
|
|
print(" Troppi pochi campioni.")
|
|
return None
|
|
|
|
X = np.array(X_all)
|
|
y = np.array(y_all)
|
|
|
|
print(f" Samples: {len(X)}, Up: {np.mean(y)*100:.1f}%")
|
|
|
|
# Walk-forward
|
|
TRAIN_SIZE = max(int(len(X) * 0.5), 50)
|
|
STEP_SIZE = max(int(len(X) * 0.1), 10)
|
|
|
|
results = {}
|
|
|
|
for ml_thr in [0.50, 0.55, 0.60, 0.65, 0.70]:
|
|
capital = float(INITIAL)
|
|
equity = [capital]
|
|
trades_list = []
|
|
correct = 0
|
|
total = 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 = X[start:train_end]
|
|
y_tr = y[start:train_end]
|
|
X_te = X[train_end:test_end]
|
|
y_te = y[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_bars - 1] - close[i - 1]) / close[i - 1]
|
|
|
|
# ML decide direction
|
|
direction = None
|
|
if p_up >= ml_thr:
|
|
direction = "long"
|
|
elif p_up <= (1 - ml_thr):
|
|
direction = "short"
|
|
|
|
if direction is None:
|
|
continue
|
|
|
|
is_correct = (direction == "long" and actual_ret > 0) or (direction == "short" and actual_ret < 0)
|
|
total += 1
|
|
if is_correct:
|
|
correct += 1
|
|
|
|
trade_ret = actual_ret if direction == "long" else -actual_ret
|
|
net = trade_ret * leverage - FEE * 2 * leverage
|
|
pnl = capital * pos_pct * net
|
|
capital += pnl
|
|
capital = max(capital, 0)
|
|
equity.append(capital)
|
|
|
|
trades_list.append({
|
|
"idx": i,
|
|
"direction": direction,
|
|
"p_up": p_up,
|
|
"actual_ret": actual_ret,
|
|
"correct": is_correct,
|
|
"pnl": pnl,
|
|
})
|
|
|
|
start += STEP_SIZE
|
|
|
|
if total == 0:
|
|
continue
|
|
|
|
acc = correct / total * 100
|
|
|
|
# Max drawdown
|
|
peak = equity[0]
|
|
max_dd = 0
|
|
for v in equity:
|
|
if v > peak:
|
|
peak = v
|
|
dd = (peak - v) / peak if peak > 0 else 0
|
|
max_dd = max(max_dd, dd)
|
|
|
|
# Annualized
|
|
first_ev = ev_all[TRAIN_SIZE] if TRAIN_SIZE < len(ev_all) else ev_all[0]
|
|
last_ev = ev_all[-1]
|
|
test_candles = last_ev["idx"] - first_ev["idx"]
|
|
if tf == "1h":
|
|
test_days = test_candles / 24
|
|
elif tf == "15m":
|
|
test_days = test_candles / (24 * 4)
|
|
else:
|
|
test_days = test_candles / 24
|
|
test_years = test_days / 365.25 if test_days > 0 else 1
|
|
ann = ((capital / INITIAL) ** (1 / test_years) - 1) * 100 if test_years > 0 and capital > 0 else -100
|
|
daily_pnl = (capital - INITIAL) / test_days if test_days > 0 else 0
|
|
trades_yr = total / test_years if test_years > 0 else 0
|
|
|
|
tag = ""
|
|
if acc >= 80:
|
|
tag = " ✅✅"
|
|
elif acc >= 70:
|
|
tag = " ✅"
|
|
|
|
print(f" ml_thr={ml_thr:.2f}: trades={total:4d} acc={acc:.1f}%{tag} ret={(capital-INITIAL)/INITIAL*100:+.1f}% ann={ann:+.1f}% dd={max_dd*100:.1f}% sharpe= trades/yr={trades_yr:.0f} €/day={daily_pnl:.2f}")
|
|
|
|
results[ml_thr] = {
|
|
"trades": total, "accuracy": acc, "capital": capital,
|
|
"annualized": ann, "max_dd": max_dd * 100, "daily_pnl": daily_pnl,
|
|
"trades_yr": trades_yr,
|
|
}
|
|
|
|
# Modalità "squeeze puro" come baseline
|
|
capital_sq = float(INITIAL)
|
|
correct_sq = 0
|
|
total_sq = 0
|
|
split = int(len(X) * 0.5)
|
|
|
|
for k in range(split, len(X)):
|
|
ev = ev_all[k]
|
|
i = ev["idx"]
|
|
if i + brk_bars >= n:
|
|
continue
|
|
first_ret = (close[i] - close[i-1]) / close[i-1]
|
|
if abs(first_ret) < 0.001:
|
|
continue
|
|
direction = 1 if first_ret > 0 else -1
|
|
actual_ret = (close[i + brk_bars - 1] - close[i - 1]) / close[i - 1]
|
|
is_correct = (direction == 1 and actual_ret > 0) or (direction == -1 and actual_ret < 0)
|
|
total_sq += 1
|
|
if is_correct:
|
|
correct_sq += 1
|
|
trade_ret = actual_ret * direction
|
|
net = trade_ret * leverage - FEE * 2 * leverage
|
|
capital_sq += capital_sq * pos_pct * net
|
|
capital_sq = max(capital_sq, 0)
|
|
|
|
if total_sq > 0:
|
|
acc_sq = correct_sq / total_sq * 100
|
|
print(f" BASELINE squeeze puro: trades={total_sq:4d} acc={acc_sq:.1f}% ret={(capital_sq-INITIAL)/INITIAL*100:+.1f}%")
|
|
|
|
return results
|
|
|
|
|
|
# ===== MAIN: test su multiple configurazioni =====
|
|
print("=" * 70)
|
|
print(" STRATEGIA 13: SQUEEZE + ML IBRIDA")
|
|
print("=" * 70)
|
|
|
|
configs = [
|
|
# (asset, tf, bb_w, sq_thr, brk_bars, leverage, pos_pct)
|
|
("ETH", "1h", 20, 0.8, 3, 3, 0.2),
|
|
("ETH", "1h", 30, 0.9, 3, 3, 0.2),
|
|
("ETH", "1h", 14, 0.8, 3, 3, 0.2),
|
|
("ETH", "1h", 20, 0.9, 3, 3, 0.2),
|
|
("BTC", "1h", 14, 0.8, 3, 3, 0.2),
|
|
("BTC", "1h", 20, 0.8, 3, 3, 0.2),
|
|
("BTC", "1h", 14, 0.9, 6, 3, 0.2),
|
|
("ETH", "15m", 14, 0.8, 3, 3, 0.15),
|
|
("ETH", "15m", 20, 0.9, 3, 3, 0.15),
|
|
("BTC", "15m", 14, 0.9, 3, 3, 0.15),
|
|
# Aggressive
|
|
("ETH", "1h", 20, 0.8, 3, 5, 0.3),
|
|
("ETH", "1h", 30, 0.9, 3, 5, 0.3),
|
|
]
|
|
|
|
all_results = []
|
|
for cfg in configs:
|
|
r = run_hybrid(*cfg)
|
|
if r:
|
|
for thr, data in r.items():
|
|
all_results.append({
|
|
"config": f"{cfg[0]} {cfg[1]} BBw={cfg[2]} sq={cfg[3]} brk={cfg[4]} lev={cfg[5]} pos={cfg[6]}",
|
|
"ml_thr": thr,
|
|
**data,
|
|
})
|
|
|
|
# Sort by accuracy
|
|
print("\n\n" + "=" * 70)
|
|
print(" CLASSIFICA PER ACCURACY (top 20)")
|
|
print("=" * 70)
|
|
sorted_acc = sorted(all_results, key=lambda x: x["accuracy"], reverse=True)
|
|
for r in sorted_acc[:20]:
|
|
tag = "✅✅" if r["accuracy"] >= 80 else "✅" if r["accuracy"] >= 70 else ""
|
|
print(f" {r['config']:55s} ml={r['ml_thr']:.2f} → acc={r['accuracy']:.1f}% {tag:4s} trades={r['trades']:4d} ret={(r['capital']-INITIAL)/INITIAL*100:+.1f}% ann={r['annualized']:+.1f}% dd={r['max_dd']:.1f}% €/day={r['daily_pnl']:.2f}")
|
|
|
|
print("\n" + "=" * 70)
|
|
print(" CLASSIFICA PER ROI ANNUO (top 20, min 20 trades)")
|
|
print("=" * 70)
|
|
sorted_roi = sorted([r for r in all_results if r["trades"] >= 20], key=lambda x: x["annualized"], reverse=True)
|
|
for r in sorted_roi[:20]:
|
|
tag = "✅✅" if r["accuracy"] >= 80 else "✅" if r["accuracy"] >= 70 else ""
|
|
print(f" {r['config']:55s} ml={r['ml_thr']:.2f} → acc={r['accuracy']:.1f}% {tag:4s} ann={r['annualized']:+.1f}% trades={r['trades']:4d} dd={r['max_dd']:.1f}% €/day={r['daily_pnl']:.2f}")
|
|
|
|
print("\n" + "=" * 70)
|
|
print(" SWEET SPOT: acc>=75% AND ann>=20% AND trades>=15")
|
|
print("=" * 70)
|
|
sweet = [r for r in all_results if r["accuracy"] >= 75 and r["annualized"] >= 20 and r["trades"] >= 15]
|
|
sweet.sort(key=lambda x: x["accuracy"] * x["annualized"], reverse=True)
|
|
for r in sweet:
|
|
print(f" {r['config']:55s} ml={r['ml_thr']:.2f} → acc={r['accuracy']:.1f}% ann={r['annualized']:+.1f}% trades={r['trades']:4d} dd={r['max_dd']:.1f}% €/day={r['daily_pnl']:.2f}")
|