PythagorasGoal/scripts/waste/CM01_cross_market_momentum.py

"""CM01 — Cross-Market Momentum Filter.

Squeeze su asset primario, entra SOLO se l'altro asset (BTC↔ETH)
mostra momentum short-term nella STESSA direzione.

Differenza da MT01: MT01 usa EMA slope su 1h (trend lento).
CM01 usa rendimento grezzo degli ultimi 3-6 bar sull'asset cross
(momentum veloce, stesso timeframe).

Razionale: BTC e ETH sono altamente correlati ma non perfettamente.
Se BTC fa squeeze breakout UP e anche ETH sta salendo (momentum 3-6 bar),
la probabilità di continuazione è maggiore perché c'è consenso di mercato.

Anti-overfitting: 1 parametro chiave (cross_bars 3-6), logica deterministica.
Eredita antifakeout + volume da SQ02.
"""
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
from src.strategies.indicators import keltner_ratio, detect_squeezes
from src.data.downloader import load_data


class CrossMarketMomentum(Strategy):
    name = "CM01_cross_momentum"
    description = "Squeeze + cross-asset short-term momentum filter"
    default_assets = ["BTC", "ETH"]
    default_timeframes = ["15m", "1h"]
    fee_rt = 0.002
    leverage = 3.0
    position_size = 0.15
    initial_capital = 1000.0

    # Map asset → cross asset
    _CROSS = {"BTC": "ETH", "ETH": "BTC"}

    def generate_signals(self, df: pd.DataFrame, ts: pd.DatetimeIndex,
                         **params) -> list[Signal]:
        """Genera segnali con cross-market momentum."""
        c = df["close"].values
        h = df["high"].values
        l = df["low"].values
        v = df["volume"].values
        n = len(c)
        ts_ms = df["timestamp"].values

        asset = params.get("asset", "BTC")
        tf = params.get("tf", "15m")
        bb_w = params.get("bb_window", 14)
        sq_thr = params.get("sq_threshold", 0.8)
        retrace_limit = params.get("retrace_limit", 0.6)
        vol_mult = params.get("vol_multiplier", 1.3)
        use_vol = params.get("use_vol", True)
        cross_bars = params.get("cross_bars", 4)  # barre momentum cross
        mom_min = params.get("mom_min", 0.0)      # momentum minimo (0 = solo direzione)

        # Carica cross asset
        cross_asset = self._CROSS.get(asset)
        if cross_asset is None:
            return []

        try:
            df_cross = load_data(cross_asset, tf)
        except Exception:
            return []

        c_cross = df_cross["close"].values
        ts_cross_ms = df_cross["timestamp"].values
        n_cross = len(c_cross)

        # Momentum cross: rendimento log su cross_bars barre
        cross_mom = np.full(n_cross, np.nan)
        for i in range(cross_bars, n_cross):
            if c_cross[i - cross_bars] > 0:
                cross_mom[i] = np.log(c_cross[i] / c_cross[i - cross_bars])

        kcr = keltner_ratio(c, h, l, bb_w)
        events = detect_squeezes(c, h, l, kcr, sq_thr)

        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
            direction = 1 if first_ret > 0 else -1

            # Anti-fakeout
            br = h[i] - l[i]
            if br > 0:
                if direction == 1 and (h[i] - c[i]) / br > retrace_limit:
                    continue
                elif direction == -1 and (c[i] - l[i]) / br > retrace_limit:
                    continue

            # Volume confirm
            if use_vol:
                sq_start = ev["sq_start"]
                avg_sq_v = np.mean(v[sq_start:i])
                if avg_sq_v > 0 and v[i] <= avg_sq_v * vol_mult:
                    continue

            # Cross-market momentum: trova indice cross corrispondente
            i_cross = np.searchsorted(ts_cross_ms, ts_ms[i]) - 1
            if i_cross < cross_bars or i_cross >= n_cross:
                continue
            mom = cross_mom[i_cross]
            if np.isnan(mom):
                continue

            # Filtra per direzione concordante
            if direction == 1 and mom <= mom_min:
                continue
            if direction == -1 and mom >= -mom_min:
                continue

            signals.append(Signal(
                idx=i,
                direction=direction,
                entry_price=c[i - 1],
                metadata={
                    "dur": ev["dur"],
                    "cross_mom": float(mom),
                },
            ))
        return signals


if __name__ == "__main__":
    strategy = CrossMarketMomentum()

    configs = [
        # cross_bars, mom_min, use_vol
        {"cross_bars": 3, "mom_min": 0.0, "use_vol": True},
        {"cross_bars": 4, "mom_min": 0.0, "use_vol": True},
        {"cross_bars": 6, "mom_min": 0.0, "use_vol": True},
        {"cross_bars": 4, "mom_min": 0.001, "use_vol": True},
        {"cross_bars": 4, "mom_min": 0.002, "use_vol": True},
        {"cross_bars": 4, "mom_min": 0.0, "use_vol": False},
        {"cross_bars": 3, "mom_min": 0.001, "use_vol": False},
        {"cross_bars": 6, "mom_min": 0.001, "use_vol": True},
    ]

    all_results = []
    for cfg in configs:
        for asset in ["BTC", "ETH"]:
            for tf in ["15m", "1h"]:
                for hold in [3, 6]:
                    r = strategy.backtest(asset, tf, hold=hold,
                                          cross_bars=cfg["cross_bars"],
                                          mom_min=cfg["mom_min"],
                                          use_vol=cfg["use_vol"])
                    if r and r.trades >= 20:
                        lbl = (f"CM01 cb={cfg['cross_bars']} "
                               f"mm={cfg['mom_min']} v={cfg['use_vol']} h={hold}")
                        r.strategy_name = lbl
                        all_results.append(r)

    all_results.sort(key=lambda r: r.accuracy, reverse=True)

    print(f"\n{'=' * 130}")
    print("  CM01 CROSS-MARKET MOMENTUM — TOP 20")
    print(f"{'=' * 130}")
    print(f"  {'Nome':<50s} {'A/T':>7s} {'Trades':>6s} {'Acc':>6s} "
          f"{'PnL€':>10s} {'DD%':>6s} {'€/day':>7s} "
          f"{'Mkt%':>5s} {'Dur':>5s} {'Anni':>4s}")
    print(f"  {'─' * 120}")
    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%, €5.23/day, 9 anni")
    print(f"  BENCHMARK MT01: 82.7% acc, 503t, DD 5.9%")