PythagorasGoal/Old/scripts/waste/REF_13_squeeze_ml_hybrid.py

"""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}")