PythagorasGoal/scripts/strategies/MT01_squeeze_mtf_momentum.py

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