paulpas 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
- Spread as Liquidity Indicator: Tight spreads indicate high liquidity
- Order Book Imbalance:Buy/sell pressure visible in depth
- Hidden Orders: Large orders may be partially visible (iceberg orders)
- Quote Stuffing Detection: Rapid order cancellations may indicate manipulation
- 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
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
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.