diff --git a/scripts/13_squeeze_ml_hybrid.py b/scripts/13_squeeze_ml_hybrid.py new file mode 100644 index 0000000..c4c9c63 --- /dev/null +++ b/scripts/13_squeeze_ml_hybrid.py @@ -0,0 +1,408 @@ +"""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}")