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

"""
Avellaneda-Stoikov Market Making NautilusTrader strategy.

Inventory-aware dual-sided quoting with stochastic control.
Adapted from the production ASMarketMaker (strategies/as_quoter.py).

Key insight: AS tells you WHEN to quote each side, not what price.
We always quote at best bid/ask — the AS formula controls which sides
are active based on inventory risk and reservation price.

When long → reservation price drops → stop quoting bid side
When short → reservation price rises → stop quoting ask side
When flat → quote both sides

For backtesting: simulate maker fills when price reaches our levels.
For live: submit POST-ONLY limit orders at best bid/ask.
"""
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 ASMarketMakingNT(BaseHlStrategy):
    """A-S stochastic control market making — side selection, not price selection."""

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

        # A-S parameters
        self._gamma = config.params.get("gamma", 0.1)
        self._tau = config.params.get("tau", 1.0)       # Session length (hours)
        self._max_inventory = config.params.get("max_inventory", config.order_size * 10)
        self._gamma_scale = config.params.get("gamma_scale", 500000)
        self._vol_window = config.params.get("vol_window", 300)

        # Vol estimation
        self._sigma_prices: deque[float] = deque(maxlen=self._vol_window)
        self._sigma: float = 0.01

        # Inventory tracking
        self._inventory: float = 0.0
        self._last_mid: float = 0.0
        self._tick_count: int = 0

        # Fill simulation (backtest mode)
        self._fills: list[dict] = []
        self._cumulative_pnl: float = 0.0

    def on_bar(self, bar: Bar):
        mid = float(bar.close)
        self._sigma_prices.append(mid)

        self._update_vol()
        self._tick_count += 1

        # Simulate bid/ask from candle high/low
        bid = float(bar.low)
        ask = float(bar.high)

        # T elapsed for this bar (approximate)
        t = (self._tick_count * 1.0) / (self._tau * 3600)  # Simplified

        selection = self._should_quote(mid, bid, ask, t)

        if not selection.get("quote_bid") and not selection.get("quote_ask"):
            return  # No quoting — circuit breaker active

        # Simulate fill: if we quoted bid and price went down past our level
        if selection.get("quote_bid"):
            # Check if candle low dipped below our bid level
            if float(bar.low) <= bid:
                self._simulate_fill(OrderSide.BUY, bid)

        if selection.get("quote_ask"):
            if float(bar.high) >= ask:
                self._simulate_fill(OrderSide.SELL, ask)

    def _update_vol(self):
        if len(self._sigma_prices) >= 10:
            prices = list(self._sigma_prices)
            returns = [(prices[i] - prices[i - 1]) / prices[i - 1] for i in range(1, len(prices))]
            mu = np.mean(returns)
            var = np.mean([(r - mu) ** 2 for r in returns])
            self._sigma = max(math.sqrt(var) if var > 0 else 0.01, 0.001)

    def _should_quote(self, mid: float, best_bid: float, best_ask: float, t: float) -> dict:
        # Hard inventory bounds
        if abs(self._inventory) >= self._max_inventory:
            if self._inventory > 0:
                return {"quote_bid": False, "quote_ask": True, "reservation": mid, "sigma": self._sigma}
            else:
                return {"quote_bid": True, "quote_ask": False, "reservation": mid, "sigma": self._sigma}

        # Circuit breaker: skip if vol is extremely high (> 3x normal)
        if len(self._sigma_prices) >= 5:
            recent = list(self._sigma_prices)[-5:]
            move_pct = abs(recent[-1] - recent[0]) / (recent[0] + 1e-8)
            if move_pct > 3 * self._sigma * math.sqrt(5):
                return {"quote_bid": False, "quote_ask": False, "reservation": mid, "sigma": self._sigma}

        # Reservation price from A-S formula
        q_notional = self._inventory * mid
        gamma_eff = self._gamma * self._gamma_scale
        tau_rem = max(self._tau - t, 0.01)
        sigma_sq = max(self._sigma ** 2, 0.000001)
        reservation = mid - q_notional * gamma_eff * sigma_sq * tau_rem

        # Quote sides based on reservation vs market
        quote_bid = reservation >= best_bid or abs(self._inventory) < self._max_inventory * 0.1
        quote_ask = reservation <= best_ask or abs(self._inventory) < self._max_inventory * 0.1

        return {
            "quote_bid": quote_bid,
            "quote_ask": quote_ask,
            "reservation": reservation,
            "sigma": self._sigma,
        }

    def _simulate_fill(self, side: OrderSide, price: float):
        """Simulate a fill in backtest mode."""
        size = self._cfg.order_size
        fee_rate = self._cfg.maker_fee if self._cfg.fee_model == "maker" else self._cfg.taker_fee
        fee = size * price * fee_rate

        # Update inventory + PnL
        if side == OrderSide.BUY:
            self._inventory += size
            # PnL from spread capture
            self._cumulative_pnl -= fee
        else:
            self._inventory -= size
            self._cumulative_pnl -= fee

        # Assume we close immediately at same price (simplification for backtest)
        # In production, fills are tracked by the real exchange
        self._fills.append({
            "side": "BUY" if side == OrderSide.BUY else "SELL",
            "size": size,
            "price": price,
            "fee": round(fee, 6),
            "inventory": round(self._inventory, 8),
            "cumulative_pnl": round(self._cumulative_pnl, 4),
        })

    def compute_signal(self, price: float | None = None) -> dict | None:
        """External signal compute for paper trade orchestrator."""
        if price is None or price <= 0:
            return None

        self._sigma_prices.append(price)
        self._update_vol()

        # Return quoting decision as a signal
        selection = self._should_quote(price, price * 0.999, price * 1.001, 0.5)

        if selection.get("quote_bid") and selection.get("quote_ask"):
            return {"signal": "DUAL", "strength": 1.0, "reservation": selection.get("reservation", price)}
        elif selection.get("quote_bid"):
            return {"signal": "BID_ONLY", "strength": 1.0, "reservation": selection.get("reservation", price)}
        elif selection.get("quote_ask"):
            return {"signal": "ASK_ONLY", "strength": 1.0, "reservation": selection.get("reservation", price)}
        return None

    def handle_signal(self, signal: dict):
        sig = signal.get("signal", "")
        if "DUAL" in sig:
            self._submit_order(OrderSide.BUY)
            self._submit_order(OrderSide.SELL)
        elif "BID" in sig:
            self._submit_order(OrderSide.BUY)
        elif "ASK" in sig:
            self._submit_order(OrderSide.SELL)