paulpas Role: Extract actionable insights from order book dynamics and trade flow data
Philosophy: Order flow reveals hidden liquidity and order imbalances. Prioritize microsecond timing, bid-ask spread analysis, and volume profiling for predictive signals.
Key Principles
- Hidden Liquidity Detection: Identify hidden orders through trade size and timing patterns
- Order Flow Imbalance: Track buy vs sell pressure at each price level
- Liquidity Profiling: Map liquidity across price levels and time
- Trade Size Analysis: Detect large institutional orders via size anomalies
- Time and Sales Processing: Analyze tick-by-tick data for microstructure signals
Implementation Guidelines
Structure
- Core logic:
orderflow/parser.py- Order flow data parsing - Analyzer:
orderflow/analyzers.py- Order flow analysis - Detector:
orderflow/detectors.py- Order flow pattern detection - Config:
config/orderflow_config.yaml- Order flow parameters
Patterns to Follow
- Process order book snapshots and trade streams separately
- Calculate cumulative delta over multiple time scales
- Track order flow imbalance with volume-weighted pricing
- Detect order flow divergence from price action
Adherence Checklist
Before completing your task, verify:
- Order book imbalance (bid-ask pressure) calculated
- Cumulative delta tracked over time
- Hidden liquidity detected via trade size patterns
- Volume profile constructed at key levels
- Order flow divergence from price identified
Code Examples
Order Book Imbalance Calculator
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple
from dataclasses import dataclass
@dataclass
class OrderBook:
"""Order book state at a point in time."""
timestamp: int
bids: List[Tuple[float, float]] # (price, size)
asks: List[Tuple[float, float]] # (price, size)
class OrderBookImbalance:
"""Calculate order book imbalance metrics."""
def __init__(self, depth: int = 5):
self.depth = depth
def calculate_imbalance(self, order_book: OrderBook) -> Dict:
"""Calculate order book imbalance metrics."""
bids = order_book.bids[:self.depth]
asks = order_book.asks[:self.depth]
if not bids or not asks:
return {
'imbalance': 0.5,
'bid_volume': 0,
'ask_volume': 0,
'imbalance_signed': 0,
'price_weighted_imbalance': 0
}
# Calculate volumes
bid_volume = sum(size for _, size in bids)
ask_volume = sum(size for _, size in asks)
total_volume = bid_volume + ask_volume
# Basic imbalance (0 = all bids, 1 = all asks)
imbalance = ask_volume / total_volume if total_volume > 0 else 0.5
# Signed imbalance (-1 = all bids, +1 = all asks)
imbalance_signed = (ask_volume - bid_volume) / total_volume if total_volume > 0 else 0
# Price-weighted imbalance (better pressure indicator)
bid_weighted = sum(price * size for price, size in bids)
ask_weighted = sum(price * size for price, size in asks)
price_weighted_imbalance = (ask_weighted - bid_weighted) / (bid_weighted + ask_weighted) if (bid_weighted + ask_weighted) > 0 else 0
return {
'imbalance': imbalance,
'imbalance_signed': imbalance_signed,
'bid_volume': bid_volume,
'ask_volume': ask_volume,
'total_volume': total_volume,
'price_weighted_imbalance': price_weighted_imbalance,
'bid_ask_ratio': bid_volume / ask_volume if ask_volume > 0 else float('inf')
}
def calculate_imbalance_change(self, imbalance_series: np.ndarray,
window: int = 10) -> np.ndarray:
"""Calculate change in order book imbalance."""
if len(imbalance_series) < window:
return np.zeros_like(imbalance_series)
current = np.array(imbalance_series)
previous = np.roll(current, window)
previous[:window] = current[:window]
return current - previous
def calculate_cumulative_delta(self, prices: np.ndarray,
volumes: np.ndarray,
buy_signals: np.ndarray) -> np.ndarray:
"""Calculate cumulative delta (net buying volume)."""
delta = np.where(buy_signals, volumes, -volumes)
return np.cumsum(delta)
Trade Flow Analyzer
import numpy as np
import pandas as pd
from typing import List, Dict, Tuple
from dataclasses import dataclass
@dataclass
class Trade:
"""Single trade record."""
timestamp: int
price: float
size: float
side: str # 'buy' or 'sell'
class TradeFlowAnalyzer:
"""Analyze trade flow patterns for market signals."""
def __init__(self, high_frequency_threshold: float = 1000):
"""High frequency = trades faster than this threshold (ms)."""
self.high_freq_threshold = high_frequency_threshold
def detect_large_liquidity_orders(self, trades: List[Trade],
volume_threshold: float = 3.0) -> List[Dict]:
"""Detect large hidden liquidity orders."""
if not trades:
return []
volumes = [t.size for t in trades]
avg_volume = np.mean(volumes)
std_volume = np.std(volumes)
large_orders = []
for trade in trades:
z_score = (trade.size - avg_volume) / (std_volume + 1e-6)
if z_score > volume_threshold:
large_orders.append({
'timestamp': trade.timestamp,
'price': trade.price,
'size': trade.size,
'z_score': z_score,
'side': trade.side,
'type': 'large_liquidity'
})
return large_orders
def detect_aggressive_trades(self, trades: List[Trade],
spread: float = 0.01) -> List[Dict]:
"""Detect aggressive trades that hit posted orders."""
if not trades or len(trades) < 2:
return []
aggressive = []
for i in range(1, len(trades)):
prev_trade = trades[i-1]
curr_trade = trades[i]
# Check if trade crossed spread aggressively
if curr_trade.side == 'buy' and curr_trade.price >= prev_trade.price + spread:
aggressive.append({
'timestamp': curr_trade.timestamp,
'price': curr_trade.price,
'size': curr_trade.size,
'side': 'buy',
'type': 'aggressive_buy'
})
elif curr_trade.side == 'sell' and curr_trade.price <= prev_trade.price - spread:
aggressive.append({
'timestamp': curr_trade.timestamp,
'price': curr_trade.price,
'size': curr_trade.size,
'side': 'sell',
'type': 'aggressive_sell'
})
return aggressive
def calculate_volume_profile(self, trades: List[Trade],
price_bins: int = 100) -> Dict:
"""Calculate volume profile across price levels."""
if not trades:
return {}
prices = [t.price for t in trades]
volumes = [t.size for t in trades]
min_price = min(prices)
max_price = max(prices)
bin_size = (max_price - min_price) / price_bins
volume_profile = np.zeros(price_bins)
for price, volume in zip(prices, volumes):
bin_idx = int((price - min_price) / bin_size)
bin_idx = min(bin_idx, price_bins - 1)
volume_profile[bin_idx] += volume
return {
'min_price': min_price,
'max_price': max_price,
'bin_size': bin_size,
'volume_profile': volume_profile,
'price_levels': np.linspace(min_price, max_price, price_bins),
'total_volume': sum(volumes)
}
def detect_volume_spike(self, volumes: np.ndarray,
window: int = 60,
threshold: float = 2.0) -> List[Dict]:
"""Detect volume spikes relative to recent average."""
if len(volumes) < window:
return []
# Rolling statistics
rolling_mean = pd.Series(volumes).rolling(window).mean().fillna(np.mean(volumes))
rolling_std = pd.Series(volumes).rolling(window).std().fillna(np.std(volumes))
spikes = []
for i in range(len(volumes)):
if rolling_std[i] > 0:
z_score = (volumes[i] - rolling_mean[i]) / rolling_std[i]
if z_score > threshold:
spikes.append({
'timestamp': i,
'volume': volumes[i],
'z_score': z_score,
'rolling_mean': rolling_mean[i],
'rolling_std': rolling_std[i]
})
return spikes
Hidden Liquidity Detector
import numpy as np
import pandas as pd
from typing import List, Dict
from dataclasses import dataclass
@dataclass
class Trade:
timestamp: int
price: float
size: float
side: str
class HiddenLiquidityDetector:
"""Detect hidden liquidity through trade flow patterns."""
def __init__(self, min_order_size: float = 1000,
timing_window: int = 10):
self.min_order_size = min_order_size
self.timing_window = timing_window
def detect_iceberg_orders(self, trades: List[Trade]) -> List[Dict]:
"""Detect iceberg orders (large orders broken into smaller pieces)."""
if len(trades) < 3:
return []
# Filter for large individual trades
large_trades = [t for t in trades if t.size >= self.min_order_size]
if len(large_trades) < 2:
return []
# Find clusters of trades at same price
iceberg_orders = []
i = 0
while i < len(large_trades):
current_price = large_trades[i].price
current_side = large_trades[i].side
# Count consecutive trades at same price
cluster_size = 1
cluster_start = i
cluster_time = large_trades[i].timestamp
while (i + cluster_size < len(large_trades) and
large_trades[i + cluster_size].price == current_price and
large_trades[i + cluster_size].side == current_side and
large_trades[i + cluster_size].timestamp - cluster_start < self.timing_window):
cluster_size += 1
# Iceberg pattern: multiple trades of similar size
if cluster_size >= 3:
sizes = [large_trades[cluster_start + j].size for j in range(cluster_size)]
avg_size = np.mean(sizes)
# Check if sizes are consistent (iceberg signature)
size_variance = np.var(sizes)
if size_variance < avg_size * 0.5: # Low variance relative to size
iceberg_orders.append({
'start_time': large_trades[cluster_start].timestamp,
'end_time': large_trades[cluster_start + cluster_size - 1].timestamp,
'price': current_price,
'side': current_side,
'total_volume': sum(sizes),
'avg_chunk_size': avg_size,
'chunk_count': cluster_size,
'estimate_total': avg_size * 5 # Assume 5 chunks typically
})
i += max(1, cluster_size)
return iceberg_orders
def detect_sweep_orders(self, trades: List[Trade],
book_levels: int = 5) -> List[Dict]:
"""Detect sweep orders that take out multiple liquidity levels."""
if len(trades) < 2:
return []
sweeps = []
for i in range(1, len(trades)):
prev = trades[i-1]
curr = trades[i]
# Check if price moved against expected direction
if curr.side == 'buy' and curr.price > prev.price * 1.001:
# Price surged up - likely took out ask levels
sweeps.append({
'timestamp': curr.timestamp,
'type': 'sweep_up',
'price': curr.price,
'size': curr.size,
'direction': 'up',
'confidence': 0.8
})
elif curr.side == 'sell' and curr.price < prev.price * 0.999:
# Price dropped - likely took out bid levels
sweeps.append({
'timestamp': curr.timestamp,
'type': 'sweep_down',
'price': curr.price,
'size': curr.size,
'direction': 'down',
'confidence': 0.8
})
return sweeps
def detect_absorption(self, trades: List[Trade],
price_support: float = None,
price_resistance: float = None) -> List[Dict]:
"""Detect absorption (large orders absorbing liquidity at support/resistance)."""
if not trades:
return []
absptions = []
for i in range(1, len(trades)):
curr = trades[i]
prev = trades[i-1]
# Check for absorption at support
if price_support and abs(curr.price - price_support) < 0.01:
if curr.side == 'buy' and curr.size > np.mean([t.size for t in trades]):
absptions.append({
'timestamp': curr.timestamp,
'type': 'absorption_at_support',
'price': curr.price,
'size': curr.size,
'support_level': price_support
})
# Check for absorption at resistance
if price_resistance and abs(curr.price - price_resistance) < 0.01:
if curr.side == 'sell' and curr.size > np.mean([t.size for t in trades]):
absptions.append({
'timestamp': curr.timestamp,
'type': 'absorption_at_resistance',
'price': curr.price,
'size': curr.size,
'resistance_level': price_resistance
})
return absptions
Order Flow Divergence Detector
import numpy as np
import pandas as pd
from typing import List, Dict
class OrderFlowDivergenceDetector:
"""Detect divergence between order flow and price action."""
def __init__(self, window: int = 20):
self.window = window
def calculate_cumulative_volume_delta(self, prices: np.ndarray,
volumes: np.ndarray,
buy_signals: np.ndarray) -> np.ndarray:
"""Calculate cumulative volume delta."""
delta = np.where(buy_signals, volumes, -volumes)
return np.cumsum(delta)
def detect_divergence(self, prices: np.ndarray,
cumulative_delta: np.ndarray) -> List[Dict]:
"""Detect bullish and bearish divergences."""
if len(prices) < 2 * self.window:
return []
divergences = []
# Calculate price and delta trends
price_returns = np.diff(prices)
delta_changes = np.diff(cumulative_delta)
for i in range(self.window, len(prices) - self.window):
# Price high/low
price_window = prices[i-self.window:i+self.window]
current_price = prices[i]
local_max = np.max(price_window)
local_min = np.min(price_window)
# Delta high/low
delta_window = cumulative_delta[i-self.window:i+self.window]
current_delta = cumulative_delta[i]
local_delta_max = np.max(delta_window)
local_delta_min = np.min(delta_window)
# Bullish divergence: price lower, delta higher
if current_price < local_min and current_delta > local_delta_min * 0.9:
divergences.append({
'timestamp': i,
'type': 'bullish',
'price_level': current_price,
'delta_level': current_delta,
'price_min': local_min,
'delta_min': local_delta_min,
'signal': 'potential_bottom'
})
# Bearish divergence: price higher, delta lower
if current_price > local_max * 0.99 and current_delta < local_delta_max * 0.9:
divergences.append({
'timestamp': i,
'type': 'bearish',
'price_level': current_price,
'delta_level': current_delta,
'price_max': local_max,
'delta_max': local_delta_max,
'signal': 'potential_top'
})
return divergences
def calculate_order_flow_strength(self, prices: np.ndarray,
volumes: np.ndarray,
buy_signals: np.ndarray) -> np.ndarray:
"""Calculate order flow strength indicator."""
delta = np.where(buy_signals, volumes, -volumes)
# Rolling statistics
delta_sma = pd.Series(delta).rolling(self.window).mean().fillna(0)
delta_std = pd.Series(delta).rolling(self.window).std().fillna(1)
# Strength = delta / (std * volume)
strength = delta_sma / (delta_std * np.sqrt(volumes) + 1e-6)
return strength
Constraints
MUST DO
- Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations
- Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config
- Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context
- Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures
- Implement feature computation consistently between training and inference — use the same transformation pipeline for both
MUST NOT DO
- Do not train models on look-ahead biased features (e.g., using future prices or events in training data)
- Avoid deploying a new model version without shadow-testing against the current production model first
- Never retrain a model on a data window that includes regime changes without explicit regime-aware validation
- Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC
- Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference
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.