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

"""
Grid Market Making — structural spread capture without directional signal.

Places a symmetric grid of limit orders above and below the current mid-price.
When a buy fills, a sell is immediately placed one grid level above. When a sell
fills, a buy is placed one grid level below. Captures the spread repeatedly
during ranging / oscillating markets.

Architecture:
  - Backtest: simulate fills from candle high/low ranges
  - Live/Paper: submit real POST-ONLY limit orders and manage order lifecycle

Grid params:
  - grid_levels: number of levels on each side (default 10)
  - grid_spacing_pct: spacing between levels as % of price (default 0.1%)
  - order_size: fixed size per grid level (default 0.001 BTC)
  - rebalance_every: recenter grid every N bars (default 20)
  - maker_fee: fee for limit orders
"""
from __future__ import annotations

import logging
from collections import deque
from typing import Any

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 GridMMNT(BaseHlStrategy):
    """Grid market making — places symmetrical buy/sell grid around mid-price."""

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

        self._grid_levels = config.params.get("grid_levels", 10)
        self._grid_spacing_pct = config.params.get("grid_spacing_pct", 0.001)  # 0.1%
        self._rebalance_every = config.params.get("rebalance_every", 20)
        self._maker_fee = config.maker_fee

        # Virtual order book: {price: {"side": "BUY"/"SELL", "size": float, "filled": bool}}
        self._grid: dict[float, dict] = {}
        self._fills: list[dict] = []
        self._inventory: float = 0.0
        self._cumulative_pnl: float = 0.0
        self._bar_count: int = 0
        self._last_mid: float = 0.0

    # ── Core logic ─────────────────────────────────────────────

    def on_bar(self, bar: Bar):
        self._bar_count += 1
        mid = float(bar.close)
        high = float(bar.high)
        low = float(bar.low)

        # Initialise or rebalance grid
        if not self._grid or self._bar_count % self._rebalance_every == 0:
            self._build_grid(mid)

        # Check fills: compare candle range against grid levels
        filled_buys = []
        filled_sells = []

        for price, order in self._grid.items():
            if order["filled"]:
                continue
            if order["side"] == "BUY" and low <= price:
                order["filled"] = True
                filled_buys.append((price, order))
            elif order["side"] == "SELL" and high >= price:
                order["filled"] = True
                filled_sells.append((price, order))

        # Process fills
        for px, order in filled_buys:
            self._inventory += order["size"]
            self._cumulative_pnl -= order["size"] * px * self._maker_fee
            # Place matching sell one grid level up
            sell_px = px * (1 + self._grid_spacing_pct)
            self._grid[sell_px] = {"side": "SELL", "size": order["size"], "filled": False}
            self._fills.append({
                "side": "BUY", "price": round(px, 1), "size": order["size"],
                "fee": round(order["size"] * px * self._maker_fee, 6),
                "bar": self._bar_count,
            })

        for px, order in filled_sells:
            # Profit = spread capture minus fees
            spread_pnl = order["size"] * px * self._grid_spacing_pct
            fee = order["size"] * px * self._maker_fee
            self._inventory -= order["size"]
            self._cumulative_pnl += spread_pnl - fee
            # Place matching buy one grid level down
            buy_px = px * (1 - self._grid_spacing_pct)
            self._grid[buy_px] = {"side": "BUY", "size": order["size"], "filled": False}
            self._fills.append({
                "side": "SELL", "price": round(px, 1), "size": order["size"],
                "pnl": round(spread_pnl - fee, 6), "bar": self._bar_count,
            })

        self._last_mid = mid

    def _build_grid(self, mid: float):
        """Rebuild grid from scratch around current mid price."""
        self._grid.clear()
        size = self._cfg.order_size
        spacing = self._grid_spacing_pct

        for i in range(1, self._grid_levels + 1):
            buy_px = mid * (1 - i * spacing)
            sell_px = mid * (1 + i * spacing)
            self._grid[round(buy_px, 6)] = {"side": "BUY", "size": size, "filled": False}
            self._grid[round(sell_px, 6)] = {"side": "SELL", "size": size, "filled": False}

    # ── Signal for paper trader / live ─────────────────────────

    def compute_signal(self, price: float | None = None,
                       orderbook: dict | None = None) -> dict | None:
        """Return grid quotes for paper trader / deploy orchestrator.

        Returns the full grid of bid/ask prices for the execution layer
        to submit as limit orders.
        """
        if price is None or price <= 0:
            return None

        mid = price
        self._build_grid(mid)

        bids = [(p, o["size"]) for p, o in sorted(self._grid.items(), reverse=True)
                 if o["side"] == "BUY"]
        asks = [(p, o["size"]) for p, o in sorted(self._grid.items())
                 if o["side"] == "SELL"]

        return {
            "signal": "GRID",
            "strength": 1.0,
            "bids": bids[:5],
            "asks": asks[:5],
            "levels": self._grid_levels,
            "spacing_pct": self._grid_spacing_pct * 100,
        }

    # ── Metrics ─────────────────────────────────────────────────

    @property
    def pnl(self) -> float:
        return self._cumulative_pnl

    @property
    def total_fills(self) -> int:
        return len(self._fills)

    @property
    def inventory(self) -> float:
        return self._inventory

    def handle_signal(self, signal: dict):
        """Grid MM doesn't use single-side signals — handled by on_bar directly."""
        pass