ftdt-quant-lab/sim/maker.py

"""
Market maker quoting logic.

Generates bid/ask quotes based on microprice, inventory, volatility,
and spread constraints. Uses Avellaneda-Stoikov optimal control framework.
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from typing import Optional


@dataclass
class Quote:
    """A pair of maker quotes."""

    bid: float
    ask: float
    bid_size: float
    ask_size: float
    reservation: float   # optimal price given inventory
    spread_bps: float
    timestamp: float = 0.0


@dataclass
class MakerConfig:
    """Configuration for a market-making strategy."""

    gamma: float = 0.1          # risk aversion
    k: float = 1.5              # orderbook liquidity parameter
    tau: float = 1.0            # time horizon (hours)
    min_spread_bps: float = 1.0 # minimum spread in bps
    max_spread_bps: float = 20.0
    base_size: float = 0.001    # base quote size
    max_inventory: float = 0.005
    skew_factor: float = 0.5    # how aggressively to skew with inventory
    volatility_window: int = 100


class AvellanedaStoikovMaker:
    """Market maker using Avellaneda-Stoikov stochastic control.

    Generates bid/ask quotes that balance spread capture against
    inventory risk via a reservation price.

    Usage:
        maker = AvellanedaStoikovMaker(MakerConfig())
        maker.observe(100000.0)  # feed mid prices
        quote = maker.quote(100000.0, inventory=0.001, elapsed=0.5)
    """

    def __init__(self, config: MakerConfig | None = None):
        self._cfg = config or MakerConfig()
        self._prices: list[float] = []
        self._sigma: float = 0.02  # annualized volatility estimate

    def observe(self, mid_price: float):
        """Feed a new mid price observation for volatility estimation."""
        if mid_price <= 0:
            return
        self._prices.append(mid_price)
        if len(self._prices) > self._cfg.volatility_window:
            self._prices = self._prices[-self._cfg.volatility_window:]

        if len(self._prices) >= 2:
            returns = [
                math.log(self._prices[i] / self._prices[i - 1])
                for i in range(1, len(self._prices))
                if self._prices[i - 1] > 0
            ]
            if returns:
                mean = sum(returns) / len(returns)
                var = sum((r - mean) ** 2 for r in returns) / max(len(returns) - 1, 1)
                self._sigma = max(math.sqrt(var * 365 * 24), 0.001)  # annualize

    def quote(
        self,
        mid_price: float,
        inventory: float,
        elapsed_hours: float,
    ) -> Quote:
        """Generate bid/ask quotes given current state.

        Args:
            mid_price: current mid price
            inventory: current signed inventory (+ = long, - = short)
            elapsed_hours: elapsed time in this session (for T-t decay)
        """
        s = self._sigma
        gamma = self._cfg.gamma
        tau_remaining = self._cfg.tau - elapsed_hours
        tau_remaining = max(tau_remaining, 0.01)

        sigma_sq = s * s
        r = mid_price - inventory * gamma * sigma_sq * tau_remaining

        optimal_spread = gamma * sigma_sq * tau_remaining + (2.0 / gamma) * math.log(
            1.0 + gamma / self._cfg.k
        )
        optimal_spread = max(optimal_spread, mid_price * self._cfg.min_spread_bps / 10000)
        optimal_spread = min(optimal_spread, mid_price * self._cfg.max_spread_bps / 10000)

        half = optimal_spread / 2.0
        bid = r - half
        ask = r + half

        bid = max(bid, 1.0)
        ask = max(ask, bid + mid_price * self._cfg.min_spread_bps / 10000)

        spread_bps = (ask - bid) / mid_price * 10000 if mid_price > 0 else 0

        return Quote(
            bid=round(bid, 2),
            ask=round(ask, 2),
            bid_size=self._cfg.base_size,
            ask_size=self._cfg.base_size,
            reservation=round(r, 2),
            spread_bps=round(spread_bps, 2),
        )

    def quote_with_skew(
        self,
        mid_price: float,
        inventory: float,
        elapsed_hours: float,
        target_inventory: float = 0.0,
    ) -> Quote:
        """Quote with additional inventory skew toward target."""
        base = self.quote(mid_price, inventory, elapsed_hours)
        inv_deviation = (inventory - target_inventory) / max(self._cfg.max_inventory, 0.0001)
        skew = inv_deviation * self._cfg.skew_factor * base.spread_bps / 10000 * mid_price

        if inventory > target_inventory:
            return Quote(
                bid=round(base.bid - skew, 2),
                ask=round(base.ask - skew, 2),
                bid_size=base.bid_size * 0.5,
                ask_size=base.ask_size * 1.5,
                reservation=base.reservation,
                spread_bps=base.spread_bps,
            )
        else:
            return Quote(
                bid=round(base.bid - skew, 2),
                ask=round(base.ask - skew, 2),
                bid_size=base.bid_size * 1.5,
                ask_size=base.ask_size * 0.5,
                reservation=base.reservation,
                spread_bps=base.spread_bps,
            )

    @property
    def sigma(self) -> float:
        return self._sigma

    @property
    def config(self) -> MakerConfig:
        return self._cfg


class GridMaker:
    """Simple grid market maker — places orders at evenly-spaced levels."""

    def __init__(
        self,
        grid_levels: int = 5,
        spacing_bps: float = 5.0,
        size_per_level: float = 0.001,
    ):
        self._levels = grid_levels
        self._spacing = spacing_bps
        self._size = size_per_level

    def quotes(self, mid_price: float) -> list[dict]:
        """Generate grid quotes around mid."""
        quotes = []
        for i in range(1, self._levels + 1):
            offset = mid_price * self._spacing * i / 10000
            quotes.append({"side": "bid", "price": round(mid_price - offset, 2), "size": self._size})
            quotes.append({"side": "ask", "price": round(mid_price + offset, 2), "size": self._size})
        return quotes