ftdt-quant-lab/strategies/nt/pairs_trading_nt.py

"""
Pairs Trading NautilusTrader strategy.

BTC/ETH ratio Z-score mean reversion. Computes the rolling ratio spread
between BTC and ETH prices and enters when Z-score exceeds threshold.

Entry: Z-score < -1.5 (buy ETH relative to BTC) or Z-score > 1.5 (sell ETH)
Exit:  Z-score reverts to 0 or crossing signal in opposite direction

This is the #1 performing live strategy (67% win rate, +$0.74).
"""
from __future__ import annotations

import logging
import math
from collections import deque

import numpy as np

from nautilus_trader.model.data import Bar
from nautilus_trader.model.enums import OrderSide

from framework.base_strategy import BaseHlStrategy
from framework.config import StrategyConfig

logger = logging.getLogger(__name__)


class PairsTradingNT(BaseHlStrategy):
    """BTC/ETH pairs trading with Z-score entry/exit rules."""

    def __init__(self, config: StrategyConfig):
        super().__init__(config)

        # Ratio tracking
        self._btc_prices: deque[float] = deque(maxlen=100)
        self._eth_prices: deque[float] = deque(maxlen=100)
        self._ratios: deque[float] = deque(maxlen=100)

        # Configurable params
        self._z_entry = config.params.get("z_entry", 1.5)
        self._z_exit = config.params.get("z_exit", 0.5)
        self._lookback = config.params.get("lookback", 20)

        # State
        self._in_trade = False
        self._trade_direction: str | None = None  # "long_eth" or "short_eth"

    def on_bar(self, bar: Bar):
        """Track both BTC and ETH prices. Signal on ETH bars."""
        symbol = str(bar.bar_type.instrument_id.symbol) if hasattr(bar, 'bar_type') else ""
        price = float(bar.close)

        if "BTC" in symbol.upper():
            self._btc_prices.append(price)
        elif "ETH" in symbol.upper():
            self._eth_prices.append(price)
            self._check_signal()

    def compute_signal(self, price: float | None = None) -> dict | None:
        """Alternative: compute signal from price feed (for paper trading)."""
        if price is not None:
            self._eth_prices.append(price)
            # Use last known BTC price from cached data
            if not self._btc_prices:
                return None
        return self._check_signal()

    def _check_signal(self) -> dict | None:
        if len(self._btc_prices) < self._lookback or len(self._eth_prices) < self._lookback:
            return None

        btc_list = list(self._btc_prices)
        eth_list = list(self._eth_prices)

        # Align BTC/ETH on common window
        ratios = []
        for i in range(-min(len(btc_list), len(eth_list)), 0):
            if eth_list[i] > 0:
                ratios.append(btc_list[i] / eth_list[i])

        if len(ratios) < self._lookback:
            return None

        self._ratios.append(ratios[-1])

        recent = ratios[-self._lookback:]
        mu = np.mean(recent)
        std = np.std(recent, ddof=1)
        if std <= 0:
            return None

        z = (ratios[-1] - mu) / std

        # Exit logic
        if self._in_trade:
            # Exit when Z-score reverts toward zero
            if abs(z) < self._z_exit:
                self._in_trade = False
                sig = "BUY_ETH" if self._trade_direction == "short_eth" else "SELL_ETH"
                self._trade_direction = None
                return {"signal": sig, "strength": abs(z), "reason": "exit_reversion"}

            # Exit on crossing
            if self._trade_direction == "long_eth" and z > self._z_entry:
                self._in_trade = False
                self._trade_direction = None
                return {"signal": "SELL_ETH", "strength": abs(z), "reason": "exit_crossing"}
            elif self._trade_direction == "short_eth" and z < -self._z_entry:
                self._in_trade = False
                self._trade_direction = None
                return {"signal": "BUY_ETH", "strength": abs(z), "reason": "exit_crossing"}
            return None

        # Entry logic
        if z < -self._z_entry:
            # BTC/ETH ratio is low → ETH is relatively expensive → buy ETH vs BTC
            self._in_trade = True
            self._trade_direction = "long_eth"
            return {"signal": "BUY_ETH", "strength": abs(z) / self._z_entry, "z_score": round(z, 3), "reason": "entry_zscore"}

        if z > self._z_entry:
            # BTC/ETH ratio is high → ETH is relatively cheap → sell ETH vs BTC
            self._in_trade = True
            self._trade_direction = "short_eth"
            return {"signal": "SELL_ETH", "strength": abs(z) / self._z_entry, "z_score": round(z, 3), "reason": "entry_zscore"}

        return None

    def handle_signal(self, signal: dict):
        side_str = signal["signal"]
        if "BUY" in side_str:
            self._submit_order(OrderSide.BUY)
        elif "SELL" in side_str:
            self._submit_order(OrderSide.SELL)