mahmoud20138 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 │
└──────────────┴──────────────┴─────────────┴──────────────────┘
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
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
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)
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
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"]}
…(truncated)