PythagorasGoal/scripts/s2_04_momentum_microstructure.py

"""S2-04: Momentum microstructure su 5m.
Approccio: cattura micro-trend intraday.
- Identifica breakout da consolidamento su 5m
- Conferma con volume e acceleration
- Hold breve (15-30 min), stop stretto
- Target: molti piccoli guadagni, alta frequenza
"""
from __future__ import annotations
import sys
sys.path.insert(0, ".")

import numpy as np
import pandas as pd
from src.data.downloader import load_data

FEE = 0.001
INITIAL = 1000
LEVERAGE = 3


def ema(arr: np.ndarray, period: int) -> np.ndarray:
    result = np.full(len(arr), np.nan)
    k = 2 / (period + 1)
    result[period - 1] = np.mean(arr[:period])
    for i in range(period, len(arr)):
        result[i] = arr[i] * k + result[i - 1] * (1 - k)
    return result


def atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray:
    tr = np.maximum(high - low, np.maximum(np.abs(high - np.roll(close, 1)), np.abs(low - np.roll(close, 1))))
    tr[0] = high[0] - low[0]
    return ema(tr, period)


def run_momentum(asset):
    print(f"\n{'#'*60}")
    print(f"  {asset} 5m — MOMENTUM MICROSTRUCTURE")
    print(f"{'#'*60}")

    df = load_data(asset, "5m")
    close = df["close"].values
    high = df["high"].values
    low = df["low"].values
    volume = df["volume"].values
    n = len(close)
    split = int(n * 0.7)

    timestamps = pd.to_datetime(df["timestamp"], unit="ms", utc=True)

    ema_fast = ema(close, 8)
    ema_slow = ema(close, 21)
    ema_trend = ema(close, 55)
    atr_vals = atr(high, low, close, 14)

    configs = [
        # (consolidation_bars, breakout_atr_mult, hold_bars, stop_atr, tp_atr, min_vol_mult, name)
        (12, 1.5, 3, 1.0, 2.0, 1.3, "tight_12bar"),
        (12, 1.5, 6, 1.5, 2.5, 1.2, "medium_12bar"),
        (24, 2.0, 6, 1.5, 3.0, 1.5, "wide_24bar"),
        (6, 1.2, 3, 1.0, 1.5, 1.1, "fast_6bar"),
        (12, 1.5, 3, 0.8, 2.0, 1.3, "tight_stop"),
        (18, 1.8, 4, 1.2, 2.5, 1.4, "balanced_18bar"),
    ]

    for consol_bars, brk_mult, hold_bars, stop_m, tp_m, vol_mult, name in configs:
        capital = float(INITIAL)
        correct = 0
        total = 0
        daily_trades = {}

        for i in range(max(split, 60), n - hold_bars):
            if np.isnan(ema_fast[i]) or np.isnan(ema_slow[i]) or np.isnan(atr_vals[i]) or atr_vals[i] == 0:
                continue

            day = timestamps.iloc[i].strftime("%Y-%m-%d")
            if daily_trades.get(day, 0) >= 5:
                continue

            # Consolidation: range delle ultime N barre < 1.5 ATR
            consol_range = np.max(high[i - consol_bars : i]) - np.min(low[i - consol_bars : i])
            if consol_range > 1.5 * atr_vals[i]:
                continue

            # Breakout: current bar breaks consolidation range
            consol_high = np.max(high[i - consol_bars : i])
            consol_low = np.min(low[i - consol_bars : i])

            breakout_up = close[i] > consol_high + atr_vals[i] * (brk_mult - 1)
            breakout_down = close[i] < consol_low - atr_vals[i] * (brk_mult - 1)

            if not (breakout_up or breakout_down):
                continue

            # Volume confirmation
            vol_avg = np.mean(volume[max(0, i - 24) : i])
            if vol_avg > 0 and volume[i] < vol_avg * vol_mult:
                continue

            # Trend filter: only trade in direction of trend
            if breakout_up and close[i] < ema_trend[i]:
                continue
            if breakout_down and close[i] > ema_trend[i]:
                continue

            direction = "long" if breakout_up else "short"
            entry = close[i]
            stop_price = entry - atr_vals[i] * stop_m if direction == "long" else entry + atr_vals[i] * stop_m
            tp_price = entry + atr_vals[i] * tp_m if direction == "long" else entry - atr_vals[i] * tp_m

            exit_price = close[min(i + hold_bars, n - 1)]
            for j in range(i + 1, min(i + hold_bars + 1, n)):
                if direction == "long":
                    if low[j] <= stop_price:
                        exit_price = stop_price
                        break
                    if high[j] >= tp_price:
                        exit_price = tp_price
                        break
                else:
                    if high[j] >= stop_price:
                        exit_price = stop_price
                        break
                    if low[j] <= tp_price:
                        exit_price = tp_price
                        break
                exit_price = close[j]

            if direction == "long":
                trade_ret = (exit_price - entry) / entry
            else:
                trade_ret = (entry - exit_price) / entry

            net = trade_ret * LEVERAGE - FEE * 2 * LEVERAGE
            capital += capital * 0.1 * net
            capital = max(capital, 0)

            total += 1
            if trade_ret > 0:
                correct += 1
            daily_trades[day] = daily_trades.get(day, 0) + 1

        if total < 30:
            continue

        acc = correct / total * 100
        ret = (capital - INITIAL) / INITIAL * 100
        test_days = (n - split) / (24 * 12)
        test_years = test_days / 365.25
        ann = ((capital / INITIAL) ** (1 / test_years) - 1) * 100 if test_years > 0 and capital > 0 else -100
        dpnl = (capital - INITIAL) / test_days if test_days > 0 else 0
        days_active = len(daily_trades)

        tag = "✅" if acc >= 55 and ann >= 30 else ""
        print(f"  {name:20s}: trades={total:5d} acc={acc:.1f}% ret={ret:+.1f}% ann={ann:+.1f}% €/day={dpnl:.2f} active={days_active} t/day={total/days_active:.1f} {tag}")


for asset in ["ETH", "BTC"]:
    run_momentum(asset)