Market making and high-frequency trading: quote generation, inventory management, order book analysis, latency tracking, microstructure signals, spoofing detection, optimal execution (TWAP/VWAP), Avellaneda-Stoikov model, order flow toxicity, market impact estimation. USE FOR: market making, market maker, HFT, high frequency, order book, bid ask, spread, inventory, spoofing, microstructure, latency, TWAP, VWAP, order flow, toxicity, tick data, limit order.
Scanned 6/3/2026
Install via CLI
Skills Directoryopenskills install mahmoud20138/Tradecraft---
name: market-making-hft
description: "Market making and high-frequency trading: quote generation, inventory management, order book analysis, latency tracking, microstructure signals, spoofing detection, optimal execution (TWAP/VWAP), Avellaneda-Stoikov model, order flow toxicity, market impact estimation. USE FOR: market making, market maker, HFT, high frequency, order book, bid ask, spread, inventory, spoofing, microstructure, latency, TWAP, VWAP, order flow, toxicity, tick data, limit order."
related_skills:
- execution-algo-trading
- liquidity-analysis
- tick-data-storage
tags:
- trading
- execution
- market-making
- hft
- inventory
- quote-generation
skill_level: expert
kind: reference
category: trading/execution
status: active
---
> **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(
level.volume / (1 + abs(level.price - mid) / mid * 100)
for level in bids
)
ask_vol = sum(
level.volume / (1 + abs(level.price - mid) / mid * 100)
for level in asks
)
else:
bid_vol = sum(level.volume for level in bids)
ask_vol = sum(level.volume for level in asks)
total = bid_vol + ask_vol
if total == 0:
return 0.0
return (bid_vol - ask_vol) / total
@staticmethod
def book_depth_profile(
bid_levels: List[OrderBookLevel],
ask_levels: List[OrderBookLevel],
pct_range: float = 1.0,
) -> dict:
"""
Analyze order book depth within a percentage range of mid price.
Args:
pct_range: Percentage of mid price to include (e.g., 1.0 = ±1%)
"""
if not bid_levels or not ask_levels:
return {"error": "Empty order book"}
mid = (bid_levels[0].price + ask_levels[0].price) / 2
lower = mid * (1 - pct_range / 100)
upper = mid * (1 + pct_range / 100)
bids_in_range = [l for l in bid_levels if l.price >= lower]
asks_in_range = [l for l in ask_levels if l.price <= upper]
bid_depth = sum(l.volume * l.price for l in bids_in_range)
ask_depth = sum(l.volume * l.price for l in asks_in_range)
# Find large walls (levels > 3x average)
all_volumes = [l.volume for l in bids_in_range + asks_in_range]
avg_vol = np.mean(all_volumes) if all_volumes else 0
bid_walls = [
{"price": l.price, "volume": l.volume, "usd_value": l.volume * l.price}
for l in bids_in_range
if l.volume > avg_vol * 3
]
ask_walls = [
{"price": l.price, "volume": l.volume, "usd_value": l.volume * l.price}
for l in asks_in_range
if l.volume > avg_vol * 3
]
return {
"mid_price": round(mid, 8),
"bid_depth_usd": round(bid_depth, 2),
"ask_depth_usd": round(ask_depth, 2),
"depth_ratio": round(bid_depth / ask_depth, 4) if ask_depth > 0 else float("inf"),
"bid_levels_count": len(bids_in_range),
"ask_levels_count": len(asks_in_range),
"bid_walls": bid_walls[:5],
"ask_walls": ask_walls[:5],
"imbalance": round(
(bid_depth - ask_depth) / (bid_depth + ask_depth), 4
) if (bid_depth + ask_depth) > 0 else 0,
}
@staticmethod
def detect_spoofing(
order_updates: List[dict],
threshold_cancel_ratio: float = 0.9,
min_size_multiple: float = 3.0,
window_seconds: float = 60.0,
) -> List[dict]:
"""
Detect potential spoofing activity in order book updates.
Spoofing indicators:
- Large orders placed then quickly cancelled (>90% cancel rate)
- Orders significantly larger than average
- Rapid placement/cancellation cycles
Args:
order_updates: List of dicts with keys: order_id, action (place/cancel/fill),
price, volume, timestamp, side
threshold_cancel_ratio: Min cancel/place ratio to flag
min_size_multiple: Min multiple of average size to flag
"""
if not order_updates:
return []
# Group by order ID
orders: Dict[str, List[dict]] = {}
for update in order_updates:
oid = update.get("order_id", "")
if oid not in orders:
orders[oid] = []
orders[oid].append(update)
# Calculate average order size
all_volumes = [
u.get("volume", 0)
for u in order_updates
if u.get("action") == "place"
]
avg_volume = np.mean(all_volumes) if all_volumes else 0
suspects = []
for oid, updates in orders.items():
placed = [u for u in updates if u.get("action") == "place"]
cancelled = [u for u in updates if u.get("action") == "cancel"]
filled = [u for u in updates if u.get("action") == "fill"]
if not placed:
continue
volume = placed[0].get("volume", 0)
is_large = volume > avg_volume * min_size_multiple if avg_volume > 0 else False
# Check if quickly cancelled
if cancelled and placed:
place_time = placed[0].get("timestamp")
cancel_time = cancelled[0].get("timestamp")
if place_time and cancel_time:
lifetime = (cancel_time - place_time).total_seconds()
else:
lifetime = window_seconds
else:
lifetime = None
# Flag as suspicious
cancel_ratio = len(cancelled) / len(placed) if placed else 0
if (
is_large
and cancel_ratio >= threshold_cancel_ratio
and not filled
and lifetime is not None
and lifetime < window_seconds
):
suspects.append({
"order_id": oid,
"side": placed[0].get("side", "unknown"),
"price": placed[0].get("price", 0),
"volume": volume,
"volume_multiple": round(volume / avg_volume, 1) if avg_volume > 0 else 0,
"lifetime_seconds": round(lifetime, 2) if lifetime else None,
"cancel_ratio": round(cancel_ratio, 2),
"confidence": min(1.0, (volume / avg_volume / 5 + cancel_ratio) / 2) if avg_volume > 0 else 0.5,
})
return sorted(suspects, key=lambda x: x.get("confidence", 0), reverse=True)
@staticmethod
def vwap_from_book(
levels: List[OrderBookLevel],
target_volume: float,
) -> Optional[float]:
"""
Calculate VWAP for executing a target volume through the book.
This gives the expected average execution price for a market order
of the given size.
"""
if not levels or target_volume <= 0:
return None
remaining = target_volume
total_cost = 0.0
for level in levels:
fill = min(remaining, level.volume)
total_cost += fill * level.price
remaining -= fill
if remaining <= 0:
break
filled_volume = target_volume - remaining
if filled_volume == 0:
return None
return total_cost / filled_volume
@staticmethod
def market_impact_estimate(
levels: List[OrderBookLevel],
order_size: float,
) -> dict:
"""
Estimate market impact of a market order.
Returns:
Dict with vwap, slippage, price levels consumed, etc.
"""
if not levels:
return {"error": "Empty book"}
best_price = levels[0].price
vwap = OrderBookAnalyzer.vwap_from_book(levels, order_size)
if vwap is None:
return {
"error": "Insufficient liquidity",
"available_volume": sum(l.volume for l in levels),
"requested_volume": order_size,
}
slippage = abs(vwap - best_price)
slippage_bps = slippage / best_price * 10000 if best_price > 0 else 0
# Count levels consumed
remaining = order_size
levels_consumed = 0
for level in levels:
remaining -= level.volume
levels_consumed += 1
if remaining <= 0:
break
return {
"best_price": round(best_price, 8),
"vwap": round(vwap, 8),
"slippage": round(slippage, 8),
"slippage_bps": round(slippage_bps, 2),
"levels_consumed": levels_consumed,
"total_levels": len(levels),
"order_size": order_size,
}
```
---
## Latency Tracker & Execution Monitor
# Latency Tracker & Execution Monitor
```python
class LatencyTracker:
"""
Track and analyze execution latency for HFT operations.
Monitors:
- Order submission to acknowledgement latency
- Market data feed latency
- Round-trip latency
- Latency distribution (percentiles)
- Latency anomaly detection
"""
def __init__(self, max_samples: int = 10000):
self._measurements: List[float] = []
self._max_samples = max_samples
self._labels: Dict[str, List[float]] = {}
def record(
self,
latency_seconds: float,
label: str = "default",
) -> None:
"""Record a latency measurement."""
self._measurements.append(latency_seconds)
if len(self._measurements) > self._max_samples:
self._measurements = self._measurements[-self._max_samples:]
if label not in self._labels:
self._labels[label] = []
self._labels[label].append(latency_seconds)
if len(self._labels[label]) > self._max_samples:
self._labels[label] = self._labels[label][-self._max_samples:]
def record_timestamps(
self,
sent_time: float,
received_time: float,
label: str = "default",
) -> float:
"""Record latency from two timestamps. Returns latency."""
latency = received_time - sent_time
self.record(latency, label)
return latency
def stats(self, label: Optional[str] = None) -> dict:
"""
Get latency statistics.
Returns percentiles, mean, std, min, max in milliseconds.
"""
measurements = self._labels.get(label, self._measurements) if label else self._measurements
if not measurements:
return {"error": "No measurements recorded"}
arr = np.array(measurements) * 1000 # Convert to ms
return {
"label": label or "all",
"count": len(measurements),
"mean_ms": round(float(np.mean(arr)), 3),
"std_ms": round(float(np.std(arr)), 3),
"min_ms": round(float(np.min(arr)), 3),
"max_ms": round(float(np.max(arr)), 3),
"p50_ms": round(float(np.percentile(arr, 50)), 3),
"p90_ms": round(float(np.percentile(arr, 90)), 3),
"p95_ms": round(float(np.percentile(arr, 95)), 3),
"p99_ms": round(float(np.percentile(arr, 99)), 3),
"p999_ms": round(float(np.percentile(arr, 99.9)), 3),
}
def detect_anomalies(
self,
label: Optional[str] = None,
threshold_std: float = 3.0,
) -> List[dict]:
"""Detect latency anomalies (values beyond threshold standard deviations)."""
measurements = self._labels.get(label, self._measurements) if label else self._measurements
if len(measurements) < 10:
return []
arr = np.array(measurements)
mean = np.mean(arr)
std = np.std(arr)
if std == 0:
return []
anomalies = []
for i, val in enumerate(measurements):
z_score = (val - mean) / std
if abs(z_score) > threshold_std:
anomalies.append({
"index": i,
"latency_ms": round(val * 1000, 3),
"z_score": round(z_score, 2),
"severity": "HIGH" if abs(z_score) > 5 else "MODERATE",
})
return anomalies
def jitter(self, label: Optional[str] = None) -> dict:
"""Calculate jitter (variation in latency)."""
measurements = self._labels.get(label, self._measurements) if label else self._measurements
if len(measurements) < 2:
return {"error": "Need at least 2 measurements"}
diffs = np.diff(np.array(measurements)) * 1000 # ms
return {
"mean_jitter_ms": round(float(np.mean(np.abs(diffs))), 3),
"max_jitter_ms": round(float(np.max(np.abs(diffs))), 3),
"std_jitter_ms": round(float(np.std(diffs)), 3),
}
# ═════════════════════════════════════════════════════════════
# OPTIMAL EXECUTION ALGORITHMS
# ═════════════════════════════════════════════════════════════
class OptimalExecution:
"""
Optimal execution algorithms for minimizing market impact.
Implements:
- TWAP (Time-Weighted Average Price)
- VWAP (Volume-Weighted Average Price)
- Implementation Shortfall minimization
"""
@staticmethod
def twap_schedule(
total_quantity: float,
duration_minutes: int,
interval_minutes: int = 1,
) -> List[dict]:
"""
Generate a TWAP execution schedule.
Splits the total quantity into equal slices over time.
Args:
total_quantity: Total quantity to execute
duration_minutes: Total execution window
interval_minutes: Time between child orders
"""
n_slices = max(1, duration_minutes // interval_minutes)
slice_qty = total_quantity / n_slices
schedule = []
for i in range(n_slices):
schedule.append({
"slice": i + 1,
"time_offset_minutes": i * interval_minutes,
"quantity": round(slice_qty, 8),
"cumulative_quantity": round(slice_qty * (i + 1), 8),
"pct_complete": round((i + 1) / n_slices * 100, 1),
})
return schedule
@staticmethod
def vwap_schedule(
total_quantity: float,
volume_profile: List[float],
duration_minutes: int,
) -> List[dict]:
"""
Generate a VWAP execution schedule.
Distributes quantity proportional to historical volume profile.
Args:
total_quantity: Total quantity to execute
volume_profile: Historical volume for each time bucket (normalized)
duration_minutes: Total execution window
"""
if not volume_profile:
return OptimalExecution.twap_schedule(total_quantity, duration_minutes)
total_volume = sum(volume_profile)
if total_volume == 0:
return OptimalExecution.twap_schedule(total_quantity, duration_minutes)
interval = duration_minutes / len(volume_profile)
schedule = []
cumulative = 0.0
for i, vol in enumerate(volume_profile):
weight = vol / total_volume
qty = total_quantity * weight
cumulative += qty
schedule.append({
"slice": i + 1,
"time_offset_minutes": round(i * interval, 1),
"quantity": round(qty, 8),
"cumulative_quantity": round(cumulative, 8),
"volume_weight": round(weight, 4),
"pct_complete": round(cumulative / total_quantity * 100, 1),
})
return schedule
@staticmethod
def implementation_shortfall(
decision_price: float,
execution_prices: List[float],
execution_quantities: List[float],
side: str = "buy",
) -> dict:
"""
Calculate implementation shortfall.
Measures the cost of delayed/imperfect execution vs the decision price.
Args:
decision_price: Price at time of decision
execution_prices: List of fill prices
execution_quantities: List of fill quantities
side: "buy" or "sell"
"""
if not execution_prices or not execution_quantities:
return {"error": "No executions"}
total_qty = sum(execution_quantities)
if total_qty == 0:
return {"error": "Zero total quantity"}
avg_exec_price = (
sum(p * q for p, q in zip(execution_prices, execution_quantities))
/ total_qty
)
if side == "buy":
shortfall = (avg_exec_price - decision_price) * total_qty
shortfall_bps = (avg_exec_price / decision_price - 1) * 10000
else:
shortfall = (decision_price - avg_exec_price) * total_qty
shortfall_bps = (1 - avg_exec_price / decision_price) * 10000
return {
"decision_price": round(decision_price, 8),
"avg_execution_price": round(avg_exec_price, 8),
"total_quantity": round(total_qty, 8),
"shortfall_usd": round(shortfall, 2),
"shortfall_bps": round(shortfall_bps, 2),
"num_fills": len(execution_prices),
"slippage_direction": "adverse" if shortfall > 0 else "favorable",
}
```
---
## Order Flow Toxicity (VPIN)
# Order Flow Toxicity (VPIN)
```python
class OrderFlowToxicity:
"""
Volume-Synchronized Probability of Informed Trading (VPIN).
Measures the proportion of "toxic" (informed) order flow. High VPIN
indicates adverse selection risk for market makers.
Based on Easley, Lopez de Prado, and O'Hara (2012).
"""
@staticmethod
def calculate_vpin(
trades: List[dict],
bucket_volume: float,
n_buckets: int = 50,
) -> dict:
"""
Calculate VPIN from trade data.
Args:
trades: List of dicts with keys: price, volume, timestamp
bucket_volume: Volume per bucket (e.g., average daily volume / 50)
n_buckets: Number of buckets for the rolling window
Returns:
Dict with VPIN value and toxicity assessment
"""
if not trades or bucket_volume <= 0:
return {"error": "Invalid input"}
# Classify trades as buy/sell using tick rule
classified = OrderFlowToxicity._classify_trades(trades)
# Create volume buckets
buckets = []
current_buy = 0.0
current_sell = 0.0
current_total = 0.0
for trade in classified:
vol = trade["volume"]
if trade["side"] == "buy":
current_buy += vol
else:
current_sell += vol
current_total += vol
while current_total >= bucket_volume:
overflow = current_total - bucket_volume
# Proportionally reduce
if current_total > 0:
factor = bucket_volume / (current_total)
else:
factor = 1.0
buckets.append({
"buy_volume": current_buy * factor,
"sell_volume": current_sell * factor,
"total_volume": bucket_volume,
"order_imbalance": abs(current_buy * factor - current_sell * factor),
})
# Carry overflow
current_buy = current_buy * (1 - factor)
current_sell = current_sell * (1 - factor)
current_total = overflow
if len(buckets) < n_buckets:
return {
"vpin": 0.0,
"warning": f"Only {len(buckets)} buckets (need {n_buckets})",
"toxicity": "UNKNOWN",
}
# Calculate VPIN over rolling window
recent_buckets = buckets[-n_buckets:]
total_imbalance = sum(b["order_imbalance"] for b in recent_buckets)
total_volume = sum(b["total_volume"] for b in recent_buckets)
vpin = total_imbalance / total_volume if total_volume > 0 else 0
# Toxicity assessment
if vpin > 0.7:
toxicity = "EXTREME"
action = "Widen spreads significantly or stop quoting"
elif vpin > 0.5:
toxicity = "HIGH"
action = "Widen spreads and reduce position sizes"
elif vpin > 0.3:
toxicity = "ELEVATED"
action = "Monitor closely, slight spread widening"
else:
toxicity = "NORMAL"
action = "Standard market making parameters"
return {
"vpin": round(vpin, 4),
"toxicity": toxicity,
"action": action,
"n_buckets_used": n_buckets,
"total_buckets_available": len(buckets),
}
@staticmethod
def _classify_trades(trades: List[dict]) -> List[dict]:
"""Classify trades as buy/sell using the tick rule."""
classified = []
prev_price = trades[0]["price"] if trades else 0
for trade in trades:
price = trade["price"]
if price > prev_price:
side = "buy"
elif price < prev_price:
side = "sell"
else:
side = classified[-1]["side"] if classified else "buy"
classified.append({
**trade,
"side": side,
})
prev_price = price
return classified
@staticmethod
def kyle_lambda(
price_changes: List[float],
volume_changes: List[float],
) -> dict:
"""
Estimate Kyle's lambda (price impact coefficient).
Lambda measures how much prices move per unit of order flow.
Higher lambda = lower liquidity / higher informed trading.
Uses simple OLS regression: delta_price = lambda * signed_volume + epsilon
"""
if len(price_changes) != len(volume_changes) or len(price_changes) < 10:
return {"error": "Need at least 10 matching observations"}
x = np.array(volume_changes)
y = np.array(price_changes)
# OLS: lambda = Cov(y, x) / Var(x)
cov = np.cov(y, x)[0, 1]
var_x = np.var(x)
if var_x == 0:
return {"error": "Zero variance in volume"}
lambda_coeff = cov / var_x
# R-squared
y_pred = lambda_coeff * x
ss_res = np.sum((y - y_pred) ** 2)
ss_tot = np.sum((y - np.mean(y)) ** 2)
r_squared = 1 - ss_res / ss_tot if ss_tot > 0 else 0
return {
"kyle_lambda": round(lambda_coeff, 8),
"r_squared": round(r_squared, 4),
"interpretation": (
"High price impact per unit volume — low liquidity"
if abs(lambda_coeff) > np.std(y) / np.std(x)
else "Normal price impact — adequate liquidity"
),
"n_observations": len(price_changes),
}
```
---
## HFT Risk Manager
# HFT Risk Manager
```python
class HFTRiskManager:
"""
Real-time risk management for high-frequency market making.
Monitors:
- Position limits
- Loss limits (daily, per-trade)
- Inventory age
- Quote sanity checks
- Kill switch triggers
"""
def __init__(
self,
max_position: float = 10.0,
max_daily_loss: float = 1000.0,
max_trade_loss: float = 100.0,
max_inventory_age_seconds: float = 300.0,
max_spread_bps: float = 100.0,
min_spread_bps: float = 0.5,
):
self.max_position = max_position
self.max_daily_loss = max_daily_loss
self.max_trade_loss = max_trade_loss
self.max_inventory_age_seconds = max_inventory_age_seconds
self.max_spread_bps = max_spread_bps
self.min_spread_bps = min_spread_bps
self._daily_pnl = 0.0
self._current_position = 0.0
self._position_opened_at: Optional[datetime] = None
self._kill_switch = False
self._warnings: List[str] = []
def check_quote(self, quote: Quote, mid_price: float) -> dict:
"""
Validate a quote before submission.
Returns dict with approved flag and any warnings/rejections.
"""
issues = []
if self._kill_switch:
return {"approved": False, "reason": "KILL SWITCH ACTIVE", "issues": ["Kill switch triggered"]}
# Spread sanity
if quote.spread_bps > self.max_spread_bps:
issues.append(f"Spread too wide: {quote.spread_bps:.1f} bps > {self.max_spread_bps}")
if quote.spread_bps < self.min_spread_bps:
issues.append(f"Spread too tight: {quote.spread_bps:.1f} bps < {self.min_spread_bps}")
# Price sanity (quotes shouldn't be too far from mid)
bid_deviation = abs(quote.bid_price - mid_price) / mid_price * 10000
ask_deviation = abs(quote.ask_price - mid_price) / mid_price * 10000
if bid_deviation > 200 or ask_deviation > 200:
issues.append(f"Quote too far from mid: bid={bid_deviation:.0f}bps, ask={ask_deviation:.0f}bps")
# Negative spread (crossed market)
if quote.ask_price <= quote.bid_price:
issues.append("Crossed market: ask <= bid")
# Position limit
if abs(self._current_position + quote.bid_size) > self.max_position:
issues.append(f"Would exceed position limit: {abs(self._current_position + quote.bid_size):.2f} > {self.max_position}")
# Daily loss limit
if self._daily_pnl <= -self.max_daily_loss:
issues.append(f"Daily loss limit reached: {self._daily_pnl:.2f}")
self._kill_switch = True
approved = len(issues) == 0
return {
"approved": approved,
"issues": issues,
"quote": {
"bid": quote.bid_price,
"ask": quote.ask_price,
"spread_bps": quote.spread_bps,
},
}
def on_fill(self, side: str, price: float, quantity: float, pnl: float = 0.0) -> dict:
"""
Process a fill and update risk state.
Returns risk status after the fill.
"""
if side == "buy":
self._current_position += quantity
else:
self._current_position -= quantity
self._daily_pnl += pnl
if self._position_opened_at is None and self._current_position != 0:
self._position_opened_at = datetime.now(timezone.utc)
elif self._current_position == 0:
self._position_opened_at = None
# Check limits
warnings = []
if abs(self._current_position) > self.max_position * 0.8:
warnings.append(f"Position at {abs(self._current_position) / self.max_position:.0%} of limit")
if self._daily_pnl < -self.max_daily_loss * 0.8:
warnings.append(f"Daily PnL at {self._daily_pnl:.2f} ({abs(self._daily_pnl / self.max_daily_loss):.0%} of limit)")
if abs(pnl) > self.max_trade_loss:
warnings.append(f"Trade loss {pnl:.2f} exceeds limit {self.max_trade_loss}")
return {
"position": round(self._current_position, 8),
"daily_pnl": round(self._daily_pnl, 2),
"kill_switch": self._kill_switch,
"warnings": warnings,
}
def check_inventory_age(self) -> Optional[str]:
"""Check if inventory has been held too long."""
if self._position_opened_at is None or self._current_position == 0:
return None
age = (datetime.now(timezone.utc) - self._position_opened_at).total_seconds()
if age > self.max_inventory_age_seconds:
return (
f"Inventory age {age:.0f}s exceeds limit "
f"{self.max_inventory_age_seconds:.0f}s — consider unwinding"
)
return None
def trigger_kill_switch(self, reason: str = "manual") -> None:
"""Activate kill switch — reject all new quotes."""
self._kill_switch = True
self._warnings.append(f"Kill switch triggered: {reason}")
def reset_kill_switch(self) -> None:
"""Deactivate kill switch."""
self._kill_switch = False
def reset_daily(self) -> None:
"""Reset daily counters (call at start of new trading day)."""
self._daily_pnl = 0.0
self._kill_switch = False
self._warnings.clear()
@property
def status(self) -> dict:
return {
"position": round(self._current_position, 8),
"position_pct": round(abs(self._current_position) / self.max_position * 100, 1) if self.max_position > 0 else 0,
"daily_pnl": round(self._daily_pnl, 2),
"daily_pnl_pct_of_limit": round(abs(self._daily_pnl) / self.max_daily_loss * 100, 1) if self.max_daily_loss > 0 else 0,
"kill_switch": self._kill_switch,
"inventory_age": self.check_inventory_age(),
"warnings": list(self._warnings),
}
```
---
## Related Skills
- [Execution Algo Trading](../execution-algo-trading.md)
- [Market Microstructure](../market-microstructure.md)
- [Liquidity Analysis](../liquidity-analysis.md)
- [Tick Data Storage](../tick-data-storage.md)
No comments yet. Be the first to comment!
