market-making-hft

> **Skill:** Market Making Hft | **Domain:** trading | **Category:** execution | **Level:** expert > **Tags:** `trading`, `execution`, `market-making`, `hft`, `inventory`, `quote-generation` --- ## Market Making Core Engine # Market Making Core Engine ## Overview Complete market making strategy implementation covering quote generation with inventory skew, volatility adjustment, and multiple market making models (basic, Avellaneda-Stoikov, grid-based). Designed for both crypto exchanges and forex/CFD markets. ## Architecture ``` ┌───────────────────────────────────────────────────────────────┐ │ Market Making Engine │ ├──────────────┬──────────────┬─────────────┬──────────────────┤ │ Quote │ Inventory │ Volatility │ Avellaneda- │ │ Generator │ Manager │ Estimator │ Stoikov Model │ └──────────────┴──────────────┴─────────────┴──────────────────┘ ``` ```python import numpy as np import pandas as pd from dataclasses import dataclass, field from typing import Dict, List, Optional, Tuple from datetime import datetime, timezone, timedelta import math from enum import Enum # ═════════════════════════════════════════════════════════════ # CORE DATA TYPES # ═════════════════════════════════════════════════════════════ @dataclass class Quote: """A two-sided market quote (bid and ask).""" bid_price: float ask_price: float bid_size: float ask_size: float mid_price: float spread: float spread_bps: float # Spread in basis points timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) @property def is_valid(self) -> bool: return ( self.bid_price > 0 and self.ask_price > self.bid_price and self.bid_size > 0 and self.ask_size > 0 ) @dataclass class OrderBookLevel: """A single level in the order book.""" price: float volume: float order_count: int = 0 side: str = "" # "bid" or "ask" @dataclass class OrderBook: """Complete order book snapshot.""" bids: List[OrderBookLevel] asks: List[OrderBookLevel] timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc)) exchange: str = "" symbol: str = "" @property def best_bid(self) -> Optional[float]: return self.bids[0].price if self.bids else None @property def best_ask(self) -> Optional[float]: return self.asks[0].price if self.asks else None @property def mid_price(self) -> Optional[float]: if self.best_bid and self.best_ask: return (self.best_bid + self.best_ask) / 2 return None @property def spread(self) -> Optional[float]: if self.best_bid and self.best_ask: return self.best_ask - self.best_bid return None @property def spread_bps(self) -> Optional[float]: if self.mid_price and self.spread: return self.spread / self.mid_price * 10000 return None @dataclass class InventoryState: """Current inventory position.""" quantity: float = 0.0 avg_entry_price: float = 0.0 max_quantity: float = 1.0 unrealized_pnl: float = 0.0 realized_pnl: float = 0.0 trades_count: int = 0 @property def utilization(self) -> float: """Inventory utilization: -1 (max short) to +1 (max long).""" if self.max_quantity == 0: return 0.0 return self.quantity / self.max_quantity @property def is_flat(self) -> bool: return abs(self.quantity) < 1e-10 # ═════════════════════════════════════════════════════════════ # BASIC MARKET MAKER # ═════════════════════════════════════════════════════════════ class BasicMarketMaker: """ Simple symmetric market maker with inventory skew. Generates two-sided quotes around mid price with: - Base spread (configurable target) - Inventory skew (widen spread away from inventory direction) - Volatility adjustment (wider spread in high vol) - Order size scaling (reduce size at inventory limits) """ def __init__( self, spread_target_bps: float = 10.0, max_inventory: float = 1.0, skew_factor: float = 0.5, vol_multiplier: float = 2.0, base_order_size: float = 0.1, ): """ Args: spread_target_bps: Target spread in basis points max_inventory: Maximum position size (in base currency units) skew_factor: How aggressively to skew quotes (0-1) vol_multiplier: How much volatility widens spread base_order_size: Default order size """ self.spread_target_bps = spread_target_bps self.max_inventory = max_inventory self.skew_factor = skew_factor self.vol_multiplier = vol_multiplier self.base_order_size = base_order_size self.inventory = InventoryState(max_quantity=max_inventory) def generate_quote( self, mid_price: float, volatility: float = 0.0, order_book_imbalance: float = 0.0, ) -> Quote: """ Generate a two-sided quote. Args: mid_price: Current mid-market price volatility: Annualized volatility (e.g., 0.2 = 20%) order_book_imbalance: -1 (ask heavy) to +1 (bid heavy) Returns: Quote with bid/ask prices and sizes """ # Base half-spread half_spread_bps = self.spread_target_bps / 2 half_spread = mid_price * half_spread_bps / 10000 # Inventory skew: shift quotes to reduce inventory inventory_ratio = self.inventory.utilization inventory_skew = inventory_ratio * half_spread * self.skew_factor # Volatility adjustment vol_adjustment = volatility * mid_price * self.vol_multiplier / 10000 # Order book imbalance adjustment (lean into imbalance) imbalance_adj = order_book_imbalance * half_spread * 0.3 # Calculate prices bid = mid_price - half_spread - inventory_skew - vol_adjustment + imbalance_adj ask = mid_price + half_spread - inventory_skew + vol_adjustment + imbalance_adj # Size scaling based on inventory bid_size_scale = max(0.1, 1 - max(0, inventory_ratio)) ask_size_scale = max(0.1, 1 + min(0, inventory_ratio)) bid_size = self.base_order_size * bid_size_scale ask_size = self.base_order_size * ask_size_scale spread = ask - bid spread_bps = spread / mid_price * 10000 if mid_price > 0 else 0 return Quote( bid_price=round(bid, 8), ask_price=round(ask, 8), bid_size=round(bid_size, 8), ask_size=round(ask_size, 8), mid_price=mid_price, spread=round(spread, 8), spread_bps=round(spread_bps, 2), ) def on_fill( self, side: str, price: float, quantity: float, ) -> None: """Process a trade fill and update inventory.""" if side == "buy": new_qty = self.inventory.quantity + quantity # Update average price if self.inventory.quantity >= 0: total_cost = ( self.inventory.avg_entry_price * self.inventory.quantity + price * quantity ) self.inventory.avg_entry_price = ( total_cost / new_qty if new_qty > 0 else price ) self.inventory.quantity = new_qty elif side == "sell": # Realize PnL on sells if self.inventory.quantity > 0: pnl = (price - self.inventory.avg_entry_price) * min(quantity, self.inventory.quantity) self.inventory.realized_pnl += pnl self.inventory.quantity -= quantity self.inventory.trades_count += 1 def update_unrealized_pnl(self, current_price: float) -> float: """Update and return unrealized PnL.""" if self.inventory.quantity == 0: self.inventory.unrealized_pnl = 0.0 else: self.inventory.unrealized_pnl = ( (current_price - self.inventory.avg_entry_price) * self.inventory.quantity ) return self.inventory.unrealized_pnl # ═════════════════════════════════════════════════════════════ # AVELLANEDA-STOIKOV MODEL # ═════════════════════════════════════════════════════════════ class AvellanedaStoikovMM: """ Avellaneda-Stoikov optimal market making model. From "High-frequency trading in a limit order book" (2008). Provides theoretically optimal quotes given risk aversion, volatility, and time horizon. Key formula: reservation_price = s - q * gamma * sigma^2 * (T - t) optimal_spread = gamma * sigma^2 * (T - t) + 2/gamma * ln(1 + gamma/kappa) Where: s = mid price q = inventory gamma = risk aversion parameter sigma = volatility T - t = time remaining kappa = order arrival rate """ def __init__( self, gamma: float = 0.1, kappa: float = 1.5, sigma: float = 0.02, time_horizon_seconds: float = 3600, max_inventory: float = 1.0, tick_size: float = 0.01, ): """ Args: gamma: Risk aversion (higher = more risk-averse, tighter inventory) kappa: Order arrival intensity (higher = more frequent fills expected) sigma: Volatility per second (annualized_vol / sqrt(252 * 24 * 3600)) time_horizon_seconds: Trading session length in seconds max_inventory: Maximum inventory limit tick_size: Minimum price increment """ self.gamma = gamma self.kappa = kappa self.sigma = sigma self.time_horizon = time_horizon_seconds self.max_inventory = max_inventory self.tick_size = tick_size self.inventory = 0.0 self._start_time = datetime.now(timezone.utc) def reservation_price( self, mid_price: float, time_remaining: Optional[float] = None, ) -> float: """ Calculate the reservation price (indifference price). The price at which the market maker is indifferent between holding and not holding one more unit, given their current inventory. """ if time_remaining is None: elapsed = (datetime.now(timezone.utc) - self._start_time).total_seconds() time_remaining = max(0.001, self.time_horizon - elapsed) # r = s - q * gamma * sigma^2 * tau return mid_price - self.inventory * self.gamma * self.sigma ** 2 * time_remaining def optimal_spread( self, time_remaining: Optional[float] = None, ) -> float: """ Calculate the optimal spread (ask - bid). Wider spread when: - Higher risk aversion (gamma) - Higher volatility (sigma) - More time remaining (tau) - Lower order arrival rate (kappa) """ if time_remaining is None: elapsed = (datetime.now(timezone.utc) - self._start_time).total_seconds() time_remaining = max(0.001, self.time_horizon - elapsed) # delta = gamma * sigma^2 * tau + 2/gamma * ln(1 + gamma/kappa) spread = ( self.gamma * self.sigma ** 2 * time_remaining + (2 / self.gamma) * math.log(1 + self.gamma / self.kappa) ) # Round to tick size return max(spread, self.tick_size * 2) def generate_quote( self, mid_price: float, time_remaining: Optional[float] = None, ) -> Quote: """Generate optimal quotes using Avellaneda-Stoikov model.""" r = self.reservation_price(mid_price, time_remaining) spread = self.optimal_spread(time_remaining) bid = r - spread / 2 ask = r + spread / 2 # Round to tick size bid = math.floor(bid / self.tick_size) * self.tick_size ask = math.ceil(ask / self.tick_size) * self.tick_size # Size based on inventory proximity to limit inv_ratio = abs(self.inventory) / self.max_inventory if self.max_inventory > 0 else 0 base_size = 1.0 - inv_ratio * 0.8 # Reduce size near limits # Favor reducing inventory side if self.inventory > 0: ask_size = base_size * 1.5 bid_size = base_size * 0.5 elif self.inventory < 0: bid_size = base_size * 1.5 ask_size = base_size * 0.5 else: bid_size = base_size ask_size = base_size return Quote( bid_price=round(bid, 8), ask_price=round(ask, 8), bid_size=round(max(0.01, bid_size), 4), ask_size=round(max(0.01, ask_size), 4), mid_price=mid_price, spread=round(ask - bid, 8), spread_bps=round((ask - bid) / mid_price * 10000, 2) if mid_price > 0 else 0, ) def on_fill(self, side: str, quantity: float) -> None: """Update inventory on fill.""" if side == "buy": self.inventory += quantity elif side == "sell": self.inventory -= quantity def reset_session(self) -> None: """Reset for a new trading session.""" self._start_time = datetime.now(timezone.utc) self.inventory = 0.0 ``` --- ## Order Book Analyzer # Order Book Analyzer ```python class OrderBookAnalyzer: """ Deep order book analysis for microstructure intelligence. Provides: - Order flow imbalance (directional predictor) - Book depth analysis - Spoofing detection - Support/resistance from order clusters - VWAP calculation from book - Liquidity heatmap generation """ @staticmethod def order_flow_imbalance( bid_levels: List[OrderBookLevel], ask_levels: List[OrderBookLevel], depth: int = 10, weighted: bool = True, ) -> float: """ Calculate order flow imbalance (OFI). A directional signal: positive = buy pressure, negative = sell pressure. Args: bid_levels: List of bid levels (best to worst) ask_levels: List of ask levels (best to worst) depth: Number of levels to consider weighted: If True, weight levels by inverse distance from mid Returns: Imbalance from -1 (all sell) to +1 (all buy) """ bids = bid_levels[:depth] asks = ask_levels[:depth] if not bids or not asks: return 0.0 if weighted: mid = (bids[0].price + asks[0].price) / 2 if bids and asks else 0 if mid == 0: return 0.0 bid_vol = sum(