# Technical Market Microstructure > "Implements order book dynamics and order flow analysis for risk management and algorithmic trading execution." - Skill: `paulpas/technical-market-microstructure` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/technical-market-microstructure` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/technical-market-microstructure/raw - Safety review: pending - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/technical-market-microstructure --- **Role:** Analyze order book depth, spread, and trade execution patterns **Philosophy:** Order book reflects real-time supply and demand; microstructure reveals hidden liquidity ## Key Principles 1. **Spread as Liquidity Indicator**: Tight spreads indicate high liquidity 2. **Order Book Imbalance**:Buy/sell pressure visible in depth 3. **Hidden Orders**: Large orders may be partially visible (iceberg orders) 4. **Quote Stuffing Detection**: Rapid order cancellations may indicate manipulation 5. **Latency Arbitrage**: Speed advantage in order execution ## Implementation Guidelines ### Structure - Core logic: technical_analysis/microstructure.py - Helper functions: technical_analysis/book_analysis.py - Tests: tests/test_microstructure.py ### Patterns to Follow - Process order book snapshots efficiently - Track order flow delta (bid-ask imbalance) - Monitor latency metrics per exchange ## Adherence Checklist Before completing your task, verify: - [ ] Order book updates processed within 100ms - [ ] Spread widening alerts trigger when > 3x average - [ ] Hidden liquidity estimated using multiple methods - [ ] Order flow imbalance calculated per price level - [ ] Market impact estimates include slippage modeling Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory. ## Python Implementation ```python import numpy as np import pandas as pd from typing import Dict, List, Tuple, Optional from dataclasses import dataclass from collections import deque import time @dataclass class OrderBookState: """Current state of the order book.""" timestamp: float bid_levels: List[Tuple[float, float]] # (price, quantity) ask_levels: List[Tuple[float, float]] mid_price: float bid_ask_spread: float depth_imbalance: float @dataclass class OrderFlow: """Aggregated order flow data.""" timestamp: float buy_volume: float sell_volume: float net_flow: float trade_count: int avg_trade_size: float class MarketMicrostructure: """Analyzes order book dynamics and order flow.""" def __init__(self, max_book_size: int = 10): self.max_book_size = max_book_size self.order_flow_buffer = deque(maxlen=1000) def process_order_book_snapshot( self, bids: List[Tuple[float, float]], asks: List[Tuple[float, float]] ) -> OrderBookState: """Process raw order book data into structured state.""" # Ensure ascending order for asks, descending for bids bids = sorted(bids, key=lambda x: x[0], reverse=True)[:self.max_book_size] asks = sorted(asks, key=lambda x: x[0])[:self.max_book_size] # Calculate mid price if bids and asks: mid = (bids[0][0] + asks[0][0]) / 2 spread = asks[0][0] - bids[0][0] else: mid, spread = 0, 0 # Calculate depth imbalance bid_depth = sum(q for _, q in bids) ask_depth = sum(q for _, q in asks) depth_imbalance = (bid_depth - ask_depth) / (bid_depth + ask_depth + 1e-8) return OrderBookState( timestamp=time.time(), bid_levels=bids, ask_levels=asks, mid_price=mid, bid_ask_spread=spread, depth_imbalance=depth_imbalance ) def calculate_order_flow_delta( self, trades: List[Dict], previous_books: Dict[str, OrderBookState] ) -> OrderFlow: """Calculate order flow based on trades and book changes.""" buy_volume = sum(t['size'] for t in trades if t['side'] == 'buy') sell_volume = sum(t['size'] for t in trades if t['side'] == 'sell') net_flow = buy_volume - sell_volume return OrderFlow( timestamp=time.time(), buy_volume=buy_volume, sell_volume=sell_volume, net_flow=net_flow, trade_count=len(trades), avg_trade_size=np.mean([t['size'] for t in trades]) if trades else 0 ) def detect_iceberg_orders( self, order_book: OrderBookState, min_size: float = 1000 ) -> Dict[str, float]: """Detect potential iceberg orders in the order book.""" icebergs = {} # Look for suspiciously large orders at price levels for side, levels in [('bid', order_book.bid_levels), ('ask', order_book.ask_levels)]: for i, (price, qty) in enumerate(levels): # Check if this level has unusually large size if qty > min_size: # Compare to adjacent levels for consistency adjacent_avg = 0 count = 0 for j in range(max(0, i-3), min(len(levels), i+4)): if j != i: adjacent_avg += levels[j][1] count += 1 if count > 0 and qty > 2 * adjacent_avg / count: icebergs[price] = qty return icebergs def calculate_tape_metrics( self, trades: List[Dict], window: int = 50 ) -> Dict[str, float]: """Calculate tape-based metrics for short-term signals.""" if len(trades) < window: window = len(trades) if window == 0: return { 'buy_pressure': 0, 'sell_pressure': 0, 'trade_imbalance': 0, 'velocity': 0 } recent_trades = trades[-window:] # Calculate pressure buys = sum(t['size'] for t in recent_trades if t['side'] == 'buy') sells = sum(t['size'] for t in recent_trades if t['side'] == 'sell') total = buys + sells # Calculate velocity (trades per second) if len(recent_trades) >= 2: time_span = recent_trades[-1]['timestamp'] - recent_trades[0]['timestamp'] velocity = len(recent_trades) / time_span if time_span > 0 else 0 else: velocity = 0 return { 'buy_pressure': buys / total if total > 0 else 0, 'sell_pressure': sells / total if total > 0 else 0, 'trade_imbalance': (buys - sells) / total if total > 0 else 0, 'velocity': velocity } def estimate_hidden_liquidity( self, order_book: OrderBookState, price_levels: int = 5 ) -> float: """Estimate hidden liquidity using statistical methods.""" if not order_book.bid_levels or not order_book.ask_levels: return 0 # Method: Compare visible depth to historical average visible_bid = sum(q for _, q in order_book.bid_levels[:price_levels]) visible_ask = sum(q for _, q in order_book.ask_levels[:price_levels]) # Historical average (simplified - in practice would use rolling mean) avg_bid_depth = visible_bid * 1.5 # Assume 50% hidden on average avg_ask_depth = visible_ask * 1.5 hidden_bid = max(0, avg_bid_depth - visible_bid) hidden_ask = max(0, avg_ask_depth - visible_ask) return hidden_bid + hidden_ask def detect_quote_cramming(self, order_updates: List[Dict], window: float = 1.0) -> bool: """Detect rapid order submissions/cancellations that may indicate manipulation.""" if len(order_updates) < 50: return False # Count updates within time window end_time = order_updates[-1]['timestamp'] start_time = end_time - window recent_updates = [ u for u in order_updates if start_time <= u['timestamp'] <= end_time ] # Flag if high frequency of updates with low execution ratio execution_rate = len([u for u in recent_updates if u['type'] == 'execution']) / len(recent_updates) return len(recent_updates) > 100 and execution_rate < 0.1 ``` --- --- ### Pattern 2: Risk-Managed Trading Logic with Validation ```python from __future__ import annotations import logging from dataclasses import dataclass from typing import Optional logger = logging.getLogger(__name__) @dataclass(frozen=True) class TradeSignal: """Immutable trade signal with all required validation constraints.""" symbol: str side: str # "buy" or "sell" price: float quantity: float confidence: float # 0.0 to 1.0 reason: str def validate(self) -> bool: """Validate that the trade signal meets all business constraints.""" if self.quantity <= 0: raise ValueError(f"Quantity must be positive, got {self.quantity}") if self.price <= 0: raise ValueError(f"Price must be positive, got {self.price}") if not 0.0 <= self.confidence <= 1.0: raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}") return True def generate_trade_signal( symbol: str, side: str, price: float, quantity: float, confidence: float, reason: str, ) -> TradeSignal: """Generate a validated trade signal with guard clause checks.""" if side not in ("buy", "sell"): raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'") signal = TradeSignal( symbol=symbol, side=side, price=price, quantity=quantity, confidence=confidence, reason=reason, ) signal.validate() logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)", symbol, side, quantity, price, confidence) return signal def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict: """Execute a trade signal after applying risk management checks.""" adjusted_quantity = signal.quantity if signal.side == "buy" and signal.quantity > max_position_pct: logger.warning("Position %s exceeds max %.1f%% — capping to %.4f", signal.symbol, max_position_pct * 100, max_position_pct) adjusted_quantity = max_position_pct return { "symbol": signal.symbol, "side": signal.side, "price": signal.price, "quantity": adjusted_quantity, "capped": adjusted_quantity < signal.quantity, "confidence": signal.confidence, "status": "submitted", } ``` ## Constraints ### MUST DO - Implement indicator calculations using rolling windows with explicit lookback periods; never use full-history data for online indicators - Validate signal generation by confirming alignment across multiple independent indicators before acting on a single signal - Calculate all price-based indicators (SMA, EMA, RSI) on closing prices unless specifically designed for tick data - Include proper handling of missing/NaN candles in indicator pipelines — forward-fill only within session boundaries - Log signal generation with the full context window of indicator values that led to each signal ### MUST NOT DO - Do not use look-ahead bias: never reference future bars or prices when calculating indicators during backtesting - Avoid recalculating all indicators from scratch on every tick — maintain running state for efficiency - Never combine indicators with different timeframes without explicit resampling and clear documentation of the alignment logic - Do not generate signals based on a single indicator crossover; require confirmation from price action or volume - Avoid hardcoding parameter values (e.g., RSI period = 14) without testing regime-specific optima ## Live References > Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content. - [Market Microstructure Overview](https://en.wikipedia.org/wiki/Market_microstructure) - [Limit Order Book Dynamics](https://en.wikipedia.org/wiki/Limit_order_book) - [Bid-Ask Spread Analysis](https://www.investopedia.com/terms/b/bid-ask-spread.asp) - [High-Frequency Trading Microstructure](https://arxiv.org/abs/0802.2576) - [Market Maker Behavior and Inventory](https://en.wikipedia.org/wiki/Market_microstructure)