Price is the effect. The order book is the cause.
At 9:30:00 AM ET on a typical trading day, Apple's order book sits in relative equilibrium — bid sizes roughly matching ask sizes, the spread compressed to a few cents. Then at 9:30:01, 50,000 shares hit the bid simultaneously. The pressure ratio spikes from 1.05 to 3.47. The spread widens to $0.08. And three seconds later, the stock is up 0.4% on 2.3x average volume.
This article answers the question every systematic trader eventually faces: how do you compress multi-level order book data into a single, backtestable signal that captures institutional accumulation before price follows?
We will build a production-grade implementation using TickDB's depth channel, derive the weighted pressure ratio factor, implement dynamic thresholds calibrated to each symbol's historical baseline, and integrate everything into a backtestable signal generation framework.
The Microstructure Problem: Why Simple Ratios Fail
Most traders begin with the naive pressure ratio:
Naive Pressure Ratio = Σ(Bid Size, L1) / Σ(Ask Size, L1)
This fails in three critical ways.
First, L1 snapshots are noisy. A single large order at the best bid can spike the ratio to 2.5 for one second before being cancelled. The signal whipsaws, generating false entries.
Second, the baseline varies by symbol. A pressure ratio of 1.5 means entirely different things for a large-cap liquid name like Microsoft versus a mid-cap with $50M average daily volume. A fixed threshold of 1.5 produces too many signals on liquid names and too few on illiquid ones.
Third, order book depth is hierarchical. The first level of the book reflects the most aggressive participants. Deeper levels — L2 through L5 — reveal the "defense line" where larger orders sit. A complete picture requires weighting across multiple levels.
The solution is a three-part architecture:
- Weighted pressure ratio using exponentially decaying weights across L1–L5
- Symbol-specific baseline calibration using rolling historical z-scores
- Signal confirmation filter requiring sustained pressure (not a single snapshot)
Weighted Pressure Ratio: The Mathematics
The weighted pressure ratio (WPR) assigns exponentially decaying weights to each order book level:
WPR = Σ(w_i × Bid_i) / Σ(w_i × Ask_i)
where w_i = α^(i-1), α = 0.7, and i ranges from 1 to 5
At L1, weight = 1.0. At L2, weight = 0.7. At L3, weight = 0.49. At L4, weight = 0.343. At L5, weight = 0.240.
This weighting scheme captures the intuition that the best bid/ask levels are most predictive of imminent price movement, while deeper levels provide context without dominating the signal.
Implementation note: The depth channel on TickDB supports L1 for US equities. For HK and crypto markets, L1–L10 is available. The implementation below uses L1 but includes the weighted calculation structure so you can extend to multi-level depth when available.
Signal Architecture: Three-Phase Detection
The complete signal generation pipeline operates in three phases:
Phase 1 — Baseline Calibration (pre-market or rolling)
- Compute rolling 20-day WPR mean and standard deviation during liquid hours (10:00–15:30 ET)
- Store z-score thresholds per symbol
- Recalibrate weekly to account for float changes, index rebalancing
Phase 2 — Real-Time Detection
- Subscribe to TickDB
depthchannel via WebSocket - Compute WPR on each snapshot
- Calculate z-score:
z = (WPR_current − WPR_mean) / WPR_std - Flag signal when z-score exceeds dynamic threshold
Phase 3 — Confirmation Filter
- Require WPR to exceed threshold for N consecutive snapshots (N = 3 for liquid names, N = 5 for illiquid)
- Record signal timestamp, peak WPR, and spread width at trigger
Production-Grade Code: TickDB Depth Channel Integration
The following implementation includes all production requirements: WebSocket heartbeat with ping/pong, exponential backoff with jitter on reconnect, rate-limit handling, timeout enforcement, and API key management via environment variable.
import os
import json
import time
import random
import statistics
from datetime import datetime, timedelta
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Callable
import threading
# Third-party WebSocket library — use websockets (pip install websockets)
try:
import websockets
except ImportError:
raise ImportError("Install websockets: pip install websockets")
@dataclass
class WPRConfig:
"""Configuration for weighted pressure ratio signal generation."""
symbol: str
alpha: float = 0.7 # Exponential decay weight
levels: int = 1 # US equities: L1 only via TickDB
z_threshold: float = 2.0 # Z-score threshold for signal
confirmation_count: int = 3 # Consecutive snapshots required
calibration_window: int = 20 # Trading days for baseline
api_key: str = field(default_factory=lambda: os.environ.get("TICKDB_API_KEY"))
base_url: str = "api.tickdb.ai"
ws_path: str = "v1/market/depth"
def __post_init__(self):
if not self.api_key:
raise ValueError(
"TICKDB_API_KEY environment variable is not set. "
"Get your key at https://tickdb.ai/dashboard"
)
class WPRCalculator:
"""Computes weighted pressure ratio with baseline normalization."""
def __init__(self, config: WPRConfig):
self.config = config
self.history = deque(maxlen=config.calibration_window * 390) # ~390 1-sec snaps per trading day
def compute_wpr(self, depth_data: dict) -> Optional[float]:
"""
Compute weighted pressure ratio from depth snapshot.
depth_data format from TickDB:
{
"symbol": "AAPL.US",
"asks": [[price, size], ...], # Sorted ascending
"bids": [[price, size], ...], # Sorted descending
"timestamp": 1234567890123
}
"""
bids = depth_data.get("bids", [])
asks = depth_data.get("asks", [])
if not bids or not asks:
return None
weighted_bid = 0.0
weighted_ask = 0.0
# Compute weighted sum across available levels
for i, (price, size) in enumerate(bids[:self.config.levels]):
weight = self.config.alpha ** i
weighted_bid += weight * size
for i, (price, size) in enumerate(asks[:self.config.levels]):
weight = self.config.alpha ** i
weighted_ask += weight * size
if weighted_ask == 0:
return None
return weighted_bid / weighted_ask
def update_history(self, wpr: float):
"""Add WPR to rolling history for baseline calibration."""
self.history.append(wpr)
def compute_z_score(self, current_wpr: float) -> Optional[float]:
"""Compute z-score relative to rolling baseline."""
if len(self.history) < 10: # Require minimum sample
return None
mean = statistics.mean(self.history)
stdev = statistics.stdev(self.history)
if stdev == 0:
return None
return (current_wpr - mean) / stdev
class TickDBDepthSubscriber:
"""
Production-grade WebSocket subscriber for TickDB depth channel.
Features:
- Heartbeat via ping/pong
- Exponential backoff with jitter on reconnect
- Rate-limit handling (code 3001)
- Thread-safe signal callbacks
"""
def __init__(self, config: WPRConfig):
self.config = config
self.wpr_calc = WPRCalculator(config)
self.running = False
self.websocket = None
self.signal_callback: Optional[Callable] = None
self._reconnect_attempts = 0
self._max_reconnect_attempts = 10
self._consecutive_signals = 0
# Exponential backoff parameters
self._base_delay = 1.0
self._max_delay = 60.0
async def _build_url(self) -> str:
"""Build authenticated WebSocket URL."""
return f"wss://{self.config.base_url}/{self.config.ws_path}?api_key={self.config.api_key}&symbol={self.config.symbol}"
async def _connect(self):
"""Establish WebSocket connection with authentication."""
url = await self._build_url()
self.websocket = await websockets.connect(
url,
ping_interval=20, # Send ping every 20 seconds
ping_timeout=10, # Wait 10 seconds for pong
close_timeout=5 # Allow 5 seconds for graceful close
)
self._reconnect_attempts = 0
print(f"[{datetime.now().isoformat()}] Connected to TickDB depth channel for {self.config.symbol}")
async def _handle_message(self, message: str):
"""Process incoming depth snapshot."""
try:
data = json.loads(message)
# Handle TickDB error responses
if "code" in data and data["code"] != 0:
await self._handle_error(data)
return
# Compute WPR
wpr = self.wpr_calc.compute_wpr(data)
if wpr is None:
return
# Update baseline history
self.wpr_calc.update_history(wpr)
# Compute z-score
z_score = self.wpr_calc.compute_z_score(wpr)
if z_score is None:
return
# Check for signal
await self._check_signal(wpr, z_score, data.get("timestamp"))
except json.JSONDecodeError as e:
print(f"[WARN] Failed to parse message: {e}")
except Exception as e:
print(f"[ERROR] Unexpected error in message handler: {e}")
async def _handle_error(self, error_data: dict):
"""Handle TickDB API errors with appropriate recovery."""
code = error_data.get("code", 0)
message = error_data.get("message", "Unknown error")
if code in (1001, 1002):
raise ValueError(f"Authentication failed: {message}. Check TICKDB_API_KEY.")
elif code == 2002:
raise KeyError(f"Symbol {self.config.symbol} not found. Verify via /v1/symbols/available")
elif code == 3001:
# Rate limited — extract Retry-After
retry_after = int(error_data.get("retry_after", 5))
print(f"[WARN] Rate limited. Waiting {retry_after} seconds.")
await asyncio.sleep(retry_after)
else:
print(f"[WARN] API error {code}: {message}")
async def _check_signal(self, wpr: float, z_score: float, timestamp: int):
"""Apply confirmation filter and emit signal."""
if z_score >= self.config.z_threshold:
self._consecutive_signals += 1
if self._consecutive_signals >= self.config.confirmation_count:
print(f"[SIGNAL] BUY PRESSURE DETECTED | WPR: {wpr:.3f} | Z: {z_score:.2f} | Time: {timestamp}")
if self.signal_callback:
self.signal_callback({
"symbol": self.config.symbol,
"wpr": wpr,
"z_score": z_score,
"timestamp": timestamp,
"signal_type": "BUY_PRESSURE"
})
elif z_score <= -self.config.z_threshold:
self._consecutive_signals += 1
if self._consecutive_signals >= self.config.confirmation_count:
print(f"[SIGNAL] SELL PRESSURE DETECTED | WPR: {wpr:.3f} | Z: {z_score:.2f} | Time: {timestamp}")
if self.signal_callback:
self.signal_callback({
"symbol": self.config.symbol,
"wpr": wpr,
"z_score": z_score,
"timestamp": timestamp,
"signal_type": "SELL_PRESSURE"
})
else:
self._consecutive_signals = 0
async def _reconnect_with_backoff(self):
"""Reconnect with exponential backoff and jitter."""
self._reconnect_attempts += 1
if self._reconnect_attempts > self._max_reconnect_attempts:
print(f"[ERROR] Max reconnection attempts ({self._max_reconnect_attempts}) exceeded.")
self.running = False
return
# Exponential backoff: delay = min(base × 2^attempt, max_delay)
delay = min(self._base_delay * (2 ** self._reconnect_attempts), self._max_delay)
# Add jitter: ±10% to prevent thundering herd
jitter = random.uniform(-delay * 0.1, delay * 0.1)
total_delay = delay + jitter
print(f"[WARN] Reconnecting in {total_delay:.2f}s (attempt {self._reconnect_attempts}/{self._max_reconnect_attempts})")
await asyncio.sleep(total_delay)
try:
await self._connect()
except Exception as e:
print(f"[ERROR] Reconnection failed: {e}")
await self._reconnect_with_backoff()
async def subscribe(self, callback: Callable):
"""
Main subscription loop with automatic reconnection.
Args:
callback: Function to call when a signal is generated.
Signature: callback(signal_dict)
"""
self.signal_callback = callback
self.running = True
while self.running:
try:
await self._connect()
async for message in self.websocket:
await self._handle_message(message)
except websockets.exceptions.ConnectionClosed as e:
print(f"[WARN] Connection closed: {e.code} — {e.reason}")
if self.running:
await self._reconnect_with_backoff()
except Exception as e:
print(f"[ERROR] Unexpected error: {e}")
if self.running:
await self._reconnect_with_backoff()
def stop(self):
"""Gracefully stop the subscriber."""
print(f"[{datetime.now().isoformat()}] Stopping depth subscriber for {self.config.symbol}")
self.running = False
# ⚠️ Engineering Warning: This implementation is designed for research and
# strategy development. For live trading, you must:
# 1. Add position sizing and risk management logic
# 2. Implement order execution with slippage modeling
# 3. Add transaction cost analysis
# 4. Use async/await properly with an event loop (see asyncio.run() example below)
if __name__ == "__main__":
import asyncio
def on_signal(signal: dict):
"""Handle incoming pressure signals."""
print(f"\n{'='*60}")
print(f" SIGNAL TRIGGERED: {signal['signal_type']}")
print(f" Symbol: {signal['symbol']}")
print(f" WPR: {signal['wpr']:.4f}")
print(f" Z-Score: {signal['z_score']:.2f}")
print(f" Timestamp: {signal['timestamp']}")
print(f"{'='*60}\n")
async def main():
config = WPRConfig(
symbol="AAPL.US",
z_threshold=2.0,
confirmation_count=3,
calibration_window=20
)
subscriber = TickDBDepthSubscriber(config)
# Graceful shutdown handler
try:
await subscriber.subscribe(on_signal)
except KeyboardInterrupt:
print("\n[INFO] Interrupt received — shutting down.")
subscriber.stop()
# Run the subscriber
asyncio.run(main())
Key implementation details:
- WebSocket authentication: The API key is passed as a URL parameter (
?api_key=) for WebSocket connections, not as a header. - Ping/pong heartbeat: Configured via
ping_interval=20andping_timeout=10. If the server does not respond with a pong within 10 seconds, the connection is terminated. - Reconnection logic: After a disconnect, the subscriber waits with exponential backoff (1s, 2s, 4s, 8s...) capped at 60s, with ±10% jitter to prevent synchronized reconnection storms.
- Rate-limit handling: When the server returns
code: 3001, the subscriber reads theretry_aftervalue and sleeps accordingly before retrying. - Thread-safe callbacks: Signals are emitted via a callback function, allowing integration with your portfolio management or execution system.
Backtest Framework Integration
The following backtest module uses historical kline data to pre-calibrate baseline statistics, then evaluates the pressure ratio signal against realized returns.
import os
import requests
import time
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
import statistics
class WPRBacktester:
"""
Backtester for weighted pressure ratio strategy.
Uses TickDB /v1/market/kline endpoint for historical OHLCV data
and simulates order book pressure from OHLCV characteristics.
Note: True order book backtesting requires tick-level data.
This implementation uses OHLCV-derived proxies for signal generation.
"""
API_KEY = os.environ.get("TICKDB_API_KEY")
BASE_URL = "https://api.tickdb.ai"
def __init__(self, symbol: str, start_date: str, end_date: str):
self.symbol = symbol
self.start_date = start_date
self.end_date = end_date
self.signals = []
self.trades = []
def _fetch_kline_data(self, interval: str = "1h", limit: int = 1000) -> List[dict]:
"""
Fetch historical OHLCV data from TickDB.
Uses header authentication: X-API-Key
"""
url = f"{self.BASE_URL}/v1/market/kline"
headers = {"X-API-Key": self.API_KEY}
params = {
"symbol": self.symbol,
"interval": interval,
"start_time": int(datetime.fromisoformat(self.start_date).timestamp() * 1000),
"end_time": int(datetime.fromisoformat(self.end_date).timestamp() * 1000),
"limit": limit
}
response = requests.get(
url,
headers=headers,
params=params,
timeout=(3.05, 10) # Connect timeout, read timeout
)
if response.status_code != 200:
raise RuntimeError(f"HTTP {response.status_code}: {response.text}")
data = response.json()
if data.get("code") == 3001:
retry_after = int(response.headers.get("Retry-After", 5))
time.sleep(retry_after)
return self._fetch_kline_data(interval, limit)
if data.get("code") != 0:
raise RuntimeError(f"API error {data.get('code')}: {data.get('message')}")
return data.get("data", [])
def _estimate_pressure_from_ohlcv(self, candles: List[dict]) -> List[float]:
"""
Estimate buy/sell pressure from OHLCV data.
Proxy heuristic:
- If close > open: Buyers were more aggressive (positive pressure)
- If close < open: Sellers were more aggressive (negative pressure)
- Magnitude scaled by (high - low) / close (intraday range)
This is a proxy. For production backtests, use true order book data
or alternative data sources with tick-level granularity.
"""
pressures = []
for candle in candles:
open_price = candle.get("open", 0)
close_price = candle.get("close", 0)
high_price = candle.get("high", 0)
low_price = candle.get("low", 0)
if high_price == low_price or open_price == 0:
pressures.append(1.0) # Neutral
continue
direction = (close_price - open_price) / open_price
range_ratio = (high_price - low_price) / close_price
# Estimated WPR: >1 means buy pressure, <1 means sell pressure
estimated_wpr = 1.0 + (direction * 10) / range_ratio
estimated_wpr = max(0.1, min(estimated_wpr, 10.0)) # Clamp to reasonable range
pressures.append(estimated_wpr)
return pressures
def compute_z_scores(self, pressures: List[float], window: int = 20) -> List[float]:
"""Compute rolling z-scores for pressure series."""
z_scores = []
for i in range(len(pressures)):
if i < window:
z_scores.append(0.0)
continue
window_data = pressures[i-window:i]
mean = statistics.mean(window_data)
stdev = statistics.stdev(window_data)
if stdev == 0:
z_scores.append(0.0)
continue
z = (pressures[i] - mean) / stdev
z_scores.append(z)
return z_scores
def generate_signals(self, z_threshold: float = 2.0) -> List[Dict]:
"""Generate signals from z-scores with confirmation filter."""
candles = self._fetch_kline_data(interval="5m")
pressures = self._estimate_pressure_from_ohlcv(candles)
z_scores = self.compute_z_scores(pressures, window=20)
signals = []
consecutive_count = 0
confirmation_required = 3
for i, (candle, z) in enumerate(zip(candles, z_scores)):
if z >= z_threshold:
consecutive_count += 1
if consecutive_count >= confirmation_required:
signals.append({
"timestamp": candle.get("open_time"),
"type": "BUY",
"z_score": z,
"wpr_estimate": pressures[i],
"price": candle.get("close")
})
consecutive_count = 0
elif z <= -z_threshold:
consecutive_count += 1
if consecutive_count >= confirmation_required:
signals.append({
"timestamp": candle.get("open_time"),
"type": "SELL",
"z_score": z,
"wpr_estimate": pressures[i],
"price": candle.get("close")
})
consecutive_count = 0
else:
consecutive_count = 0
return signals
def run_backtest(self, z_threshold: float = 2.0, hold_hours: int = 4) -> Dict:
"""
Run complete backtest and return performance metrics.
Args:
z_threshold: Z-score threshold for signal generation
hold_hours: Hours to hold position after signal
Returns:
Dictionary with performance metrics
"""
signals = self.generate_signals(z_threshold)
if not signals:
return {"error": "No signals generated — check data availability"}
# Simplified P&L calculation
# In production, use proper fill modeling with slippage
total_return = 0.0
wins = 0
losses = 0
for i, signal in enumerate(signals):
if signal["type"] == "BUY" and i + 1 < len(signals):
entry_price = signal["price"]
# Find exit signal
for j in range(i + 1, len(signals)):
if signals[j]["type"] == "SELL":
exit_price = signals[j]["price"]
ret = (exit_price - entry_price) / entry_price
total_return += ret
if ret > 0:
wins += 1
else:
losses += 1
break
total_trades = wins + losses
win_rate = wins / total_trades if total_trades > 0 else 0
avg_return = total_return / total_trades if total_trades > 0 else 0
return {
"symbol": self.symbol,
"period": f"{self.start_date} to {self.end_date}",
"total_signals": len(signals),
"executed_trades": total_trades,
"wins": wins,
"losses": losses,
"win_rate": f"{win_rate:.2%}",
"avg_return": f"{avg_return:.4%}",
"total_return": f"{total_return:.2%}",
"z_threshold": z_threshold
}
if __name__ == "__main__":
# Initialize backtester
backtester = WPRBacktester(
symbol="AAPL.US",
start_date="2025-01-01",
end_date="2025-04-01"
)
# Run backtest
results = backtester.run_backtest(z_threshold=2.0, hold_hours=4)
print("\n" + "="*60)
print(" WPR Strategy Backtest Results")
print("="*60)
for key, value in results.items():
print(f" {key}: {value}")
print("="*60)
Depth Channel Data Reference
TickDB's depth channel provides real-time order book snapshots. The following table summarizes availability by asset class:
| Asset class | Depth levels available | Update frequency | Use case |
|---|---|---|---|
| US equities | L1 (best bid/ask) | Real-time push | Pressure ratio, spread analysis |
| HK equities | L1–L10 | Real-time push | Multi-level depth strategies |
| Crypto | L1–L10 | Real-time push | High-frequency signals |
| Forex | Not available | — | — |
| Precious metals | Not available | — | — |
| Indices | Not available | — | — |
Critical limitation: The trades endpoint does not cover US equities or A-shares. For tick-level trade data on US names, an alternative data source is required. TickDB's strength lies in its cross-asset OHLCV coverage (10+ years for US equities) combined with real-time depth for HK and crypto markets.
Deployment Configuration by User Segment
| Segment | Recommended configuration | Notes |
|---|---|---|
| Individual quant (research) | Free tier, single symbol, 5-minute kline backtest | Suitable for strategy validation |
| Active individual trader | Free or Starter tier, 3–5 symbols, real-time depth for HK/crypto | Focus on HK equities where L1–L10 depth is available |
| Quant team | Professional tier, batch historical data, multiple symbol subscriptions | Leverage 10+ years of US equity OHLCV for cross-cycle backtesting |
| Institutional | Enterprise tier, full API access, dedicated support, custom data feeds | Discuss latency requirements and coverage gaps with the enterprise team |
Next Steps
If you want to validate this strategy on historical data:
- Sign up at tickdb.ai (free tier available, no credit card required)
- Generate an API key in your dashboard
- Set
export TICKDB_API_KEY="your_key_here" - Run the backtest module above against your target symbols
If you need multi-level depth (L1–L10) for HK equities or crypto:
- These asset classes support full depth in the
depthchannel - Update the
levelsparameter inWPRConfigto 5 or 10 - Extend the weighted calculation to leverage the deeper book
If you're interested in institutional-grade coverage:
- US equity OHLCV: 10+ years available via
/v1/market/kline - Multi-asset portfolio: Crypto + HK equities + US equities in a single API
- Contact enterprise@tickdb.ai for custom data requirements
If you use AI coding assistants:
- Search for and install the
tickdb-market-dataSKILL in your AI tool's marketplace - The skill provides pre-built prompts for TickDB API integration
Disclaimer: This article does not constitute investment advice. The strategies and code examples presented are for educational and research purposes. Markets involve risk; past performance does not guarantee future results. Backtested results are based on historical simulation and do not reflect actual trading outcomes. Slippage, market impact, and liquidity constraints can significantly affect live performance. Always conduct thorough out-of-sample validation and consult with qualified financial professionals before deploying any trading strategy.