1. Opening
"Price is the effect. The order book is the cause."
Every quantitative trading system depends on a foundational truth: the data feeding it is accurate, complete, and delivered within an acceptable latency window. When that truth breaks — when a primary market data feed delivers stale prices, drops ticks during volatile periods, or introduces phantom trades — the consequences cascade silently through backtests, risk models, and live execution before anyone notices.
The problem is not that primary data sources are unreliable. Most institutional-grade feeds maintain 99.9%+ uptime. The problem is the definition of "reliable." A feed that delivers 99.9% of ticks correctly is still wrong 0.1% of the time — and in a high-frequency strategy processing 50,000 ticks per second, that translates to 50 corrupted data points every second, or 180,000 per hour. At $0.01 per tick of slippage, that is $1,800 in hidden cost per hour.
Dual-source cross-validation addresses this gap. By subscribing to a second independent market data feed and continuously comparing its output against the primary source, a trading system can detect latency spikes, identify data gaps, flag anomalous price deviations, and trigger automated alerts before errors propagate into trading decisions.
This article walks through the architecture, algorithms, and production-grade Python implementation for building a dual-source validation monitor using TickDB as the secondary feed.
2. The Pain-Point Landscape: Why Primary Feeds Fail
Understanding why dual-source validation matters requires examining the specific failure modes that plague primary market data feeds.
2.1 Latency Skew
Market data arrives from multiple exchange collocation points through various network paths. A primary feed may route through a data center in New Jersey, while the secondary feed routes through Chicago. During peak volatility, network congestion can introduce asymmetric latency — the primary feed lags by 50–200 ms while the secondary remains current. A mean-reversion strategy relying on the primary feed would consistently enter positions 100 ms late, systematically bleeding alpha.
2.2 Data Dropout
Exchange matching engines occasionally experience brief connectivity hiccups — a 50 ms gap in the primary feed's WebSocket connection may not trigger a reconnect alarm, but it creates a blind spot in the order book state. The strategy continues operating on stale depth data, potentially executing against a bid-ask spread that no longer exists.
2.3 Price Anomalies
Fat-finger errors, erroneous prints, and exchange cancellation messages can push a primary feed to deliver prices that deviate 10–50% from fair value for a brief window. Without a secondary reference point, a strategy may interpret these anomalies as genuine market signals and enter losing positions.
2.4 The Quantification Imperative
A data quality monitoring system must quantify these risks to justify its operational overhead. The following table summarizes the typical failure modes and their business impact:
| Failure Mode | Typical Frequency | Detection Difficulty | Estimated Cost per Incident |
|---|---|---|---|
| Latency skew > 50 ms | 3–5 events/day during volatile sessions | Moderate — requires cross-feed timestamp comparison | $500–$5,000 (slippage + missed opportunities) |
| Data dropout < 500 ms | 1–2 events/day | Low — reconnect logic may mask the gap | $200–$2,000 (stale book positions) |
| Price anomaly > 5% | 0–1 events/month | High — requires deviation threshold calibration | $5,000–$50,000 (bad fills) |
| Feed outage > 5 sec | 0–2 events/year | Low — obvious system failure | $10,000+ (extended downtime) |
The data makes the case clear: even a modest dual-source validation system can pay for itself within weeks by catching the high-frequency, low-severity incidents that erode performance silently.
3. Architecture Overview: The Dual-Source Validation System
3.1 System Design Principles
A dual-source validation system must satisfy three non-negotiable requirements:
Independence: The secondary feed must not share infrastructure, network paths, or authentication systems with the primary feed. If both feeds route through the same colocation facility, a network switch failure would blind both simultaneously.
Low overhead: The validation logic must introduce less latency than the gap it is designed to detect. A 5 ms validation check that itself adds 10 ms of latency is counterproductive.
Deterministic alerting: Alert thresholds must be calibrated to minimize both false positives (alert fatigue) and false negatives (missed incidents).
3.2 High-Level Architecture
┌─────────────────────────────────────────────────────────────────────────────┐
│ Dual-Source Validation Monitor │
├─────────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ WebSocket ┌──────────────┐ WebSocket │
│ │ Primary │ ◄───────────────── │ Secondary │ ◄───────────────── │
│ │ Feed (e.g., │ (tick stream) │ Feed │ (tick stream) │
│ │ Polygon) │ │ (TickDB) │ │
│ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │
│ │ Timestamp Sync │ │
│ │ (NTP-aligned clocks) │ │
│ └──────────────┬─────────────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────────────┐ │
│ │ Cross-Validation │ │
│ │ Engine (Python) │ │
│ │ │ │
│ │ • Price comparison │ │
│ │ • Latency tracking │ │
│ │ • Gap detection │ │
│ │ • Anomaly scoring │ │
│ └──────────┬──────────┘ │
│ │ │
│ ┌───────────────┼───────────────┐ │
│ ▼ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ ┌────────────┐ │
│ │ Metrics │ │ Alert │ │ Incident │ │
│ │ Dashboard │ │ Webhook │ │ Log │ │
│ │ (InfluxDB)│ │ (Slack) │ │ (Postgres)│ │
│ └────────────┘ └────────────┘ └────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
3.3 Data Flow
- Both feeds subscribe to the same instrument (e.g., AAPL.US) and receive tick streams over WebSocket connections.
- Each incoming tick is timestamped with the local receive time (not the exchange publication time, which may not be available on the secondary feed).
- The cross-validation engine buffers ticks from both feeds and performs pairwise comparison at configurable intervals (default: 1 second).
- Deviation metrics are computed and compared against thresholds.
- If a threshold is breached, an alert is dispatched to the configured webhook endpoint.
4. Production-Grade Implementation
The following Python implementation provides a complete, production-ready dual-source validation monitor. All code follows the mandatory production standards: heartbeat, exponential backoff with jitter, rate-limit handling, timeouts, and environment-variable-based authentication.
4.1 Core Dependencies and Configuration
import os
import json
import time
import asyncio
import logging
import statistics
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Optional, Callable
from collections import deque
import httpx
import websockets
from websockets.exceptions import ConnectionClosed, WebSocketException
# ⚠️ For production HFT workloads, consider replacing httpx with aiohttp
# and websockets with asyncio-based connection pooling.
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("dual_source_monitor")
@dataclass
class ValidationConfig:
"""Configuration for dual-source validation thresholds."""
symbol: str = "AAPL.US"
# Price deviation threshold (fraction of mid-price)
price_deviation_threshold: float = 0.005 # 0.5%
# Latency skew threshold (milliseconds)
latency_skew_threshold_ms: float = 100.0
# Minimum sample size before validation activates
warmup_ticks: int = 100
# Maximum acceptable gap duration (seconds)
max_gap_duration_sec: float = 2.0
# Comparison window (seconds)
comparison_window_sec: float = 1.0
# Alert cooldown (seconds) to prevent alert storms
alert_cooldown_sec: float = 30.0
# Primary feed WebSocket URL
primary_ws_url: str = "wss://your-primary-feed.example.com/ws"
# TickDB WebSocket URL (secondary feed)
secondary_ws_url: str = "wss://api.tickdb.ai/ws/v1/market"
# Alert webhook URL
webhook_url: Optional[str] = os.environ.get("ALERT_WEBHOOK_URL")
# Rate limit configuration
rate_limit_code: int = 3001
base_retry_delay: float = 1.0
max_retry_delay: float = 60.0
@dataclass
class TickData:
"""Normalized tick data structure."""
symbol: str
price: float
volume: float
side: str # "buy" or "sell"
timestamp: float # Unix timestamp (seconds)
receive_time: float = field(default_factory=time.time)
source: str = "unknown"
@dataclass
class ValidationMetrics:
"""Aggregated metrics for the comparison window."""
primary_count: int = 0
secondary_count: int = 0
price_deviations: list = field(default_factory=list)
latency_skews_ms: list = field(default_factory=list)
gaps_detected: int = 0
anomalies_detected: int = 0
window_start: float = field(default_factory=time.time)
4.2 WebSocket Connection Manager with Reconnection Logic
class WebSocketFeed:
"""
WebSocket connection manager with heartbeat, exponential backoff,
jitter, and rate-limit handling.
Supports both a generic primary feed and TickDB as the secondary feed.
"""
def __init__(
self,
url: str,
name: str,
api_key: Optional[str] = None,
subscribe_message: Optional[dict] = None,
):
self.url = url
self.name = name
self.api_key = api_key
self.subscribe_message = subscribe_message or {}
self._ws: Optional[websockets.WebSocketClientProtocol] = None
self._retry_count: int = 0
self._connected: bool = False
self._last_message_time: float = 0.0
# Heartbeat interval (seconds)
self._heartbeat_interval: float = 20.0
self._last_heartbeat: float = 0.0
def _build_url(self) -> str:
"""
Build WebSocket URL with authentication.
TickDB uses URL parameter authentication; adjust for your primary feed.
"""
if "tickdb" in self.url:
# TickDB authentication via URL parameter
separator = "&" if "?" in self.url else "?"
return f"{self.url}{separator}api_key={self.api_key}"
return self.url
async def connect(self) -> None:
"""Establish WebSocket connection with timeout and error handling."""
try:
url = self._build_url()
self._ws = await websockets.connect(
url,
ping_interval=self._heartbeat_interval,
ping_timeout=10.0,
open_timeout=10.0,
close_timeout=5.0,
)
self._connected = True
self._retry_count = 0
logger.info(f"[{self.name}] Connected to {url}")
# Send subscription message if provided
if self.subscribe_message:
await self._send_subscription()
except WebSocketException as e:
logger.error(f"[{self.name}] WebSocket connection failed: {e}")
raise
async def _send_subscription(self) -> None:
"""Send subscription request for the target symbol."""
if self._ws and self.subscribe_message:
await self._ws.send(json.dumps(self.subscribe_message))
logger.info(f"[{self.name}] Subscription sent: {self.subscribe_message}")
async def _send_heartbeat(self) -> None:
"""Send heartbeat (ping) to keep connection alive."""
if self._ws and self._connected:
try:
# TickDB uses JSON ping: {"cmd": "ping"}
# Adjust format for your primary feed
if "tickdb" in self.url:
await self._ws.send(json.dumps({"cmd": "ping"}))
else:
await self._ws.ping()
self._last_heartbeat = time.time()
except Exception as e:
logger.warning(f"[{self.name}] Heartbeat failed: {e}")
async def _handle_rate_limit(self, retry_after: Optional[int] = None) -> None:
"""Handle rate limit (code 3001) with configurable retry delay."""
delay = retry_after or int(self._calculate_backoff())
logger.warning(f"[{self.name}] Rate limited. Retrying after {delay}s")
await asyncio.sleep(delay)
def _calculate_backoff(self) -> float:
"""
Calculate retry delay with exponential backoff and jitter.
delay = min(base * (2 ** retry), max_delay) + random(0, delay * 0.1)
"""
base_delay = ValidationConfig.base_retry_delay
max_delay = ValidationConfig.max_retry_delay
retry_count = self._retry_count
delay = min(base_delay * (2**retry_count), max_delay)
# Add jitter to prevent thundering herd
jitter = statistics.uniform(0, delay * 0.1)
return delay + jitter
async def reconnect(self) -> bool:
"""
Attempt reconnection with exponential backoff.
Returns True if reconnection succeeded, False otherwise.
"""
self._connected = False
self._retry_count += 1
delay = self._calculate_backoff()
logger.warning(
f"[{self.name}] Reconnecting in {delay:.2f}s "
f"(attempt {self._retry_count})"
)
await asyncio.sleep(delay)
try:
await self.connect()
return True
except Exception as e:
logger.error(f"[{self.name}] Reconnection failed: {e}")
return False
async def receive(self) -> Optional[dict]:
"""
Receive and parse a message from the WebSocket feed.
Returns None on disconnect, raises exception on error.
"""
if not self._ws or not self._connected:
return None
try:
message = await asyncio.wait_for(
self._ws.recv(), timeout=30.0
)
self._last_message_time = time.time()
data = json.loads(message)
# Check for rate limit response
if isinstance(data, dict) and data.get("code") == ValidationConfig.rate_limit_code:
retry_after = int(data.get("retry_after", 5))
await self._handle_rate_limit(retry_after)
return None
return data
except asyncio.TimeoutError:
logger.warning(f"[{self.name}] Receive timeout — sending heartbeat")
await self._send_heartbeat()
return None
except ConnectionClosed as e:
logger.warning(f"[{self.name}] Connection closed: {e}")
self._connected = False
return None
async def close(self) -> None:
"""Gracefully close the WebSocket connection."""
if self._ws:
await self._ws.close()
self._connected = False
logger.info(f"[{self.name}] Connection closed")
4.3 TickDB Depth Channel Subscription
def build_tickdb_subscription(symbol: str, channels: list[str]) -> dict:
"""
Build TickDB subscription message for specified channels.
The 'depth' channel provides L1 order book data for US equities.
Note: TickDB 'depth' support varies by market:
- US equities: L1 (best bid/ask only)
- HK equities and crypto: L1–L10 (up to 10 price levels)
- Forex, precious metals, indices: depth not supported
"""
return {
"cmd": "subscribe",
"params": {
"channels": channels, # e.g., ["kline_1m", "depth"]
"symbols": [symbol],
},
}
def parse_tickdb_depth(data: dict) -> Optional[TickData]:
"""
Parse TickDB 'depth' channel data into a normalized TickData structure.
TickDB depth message format:
{
"type": "depth",
"symbol": "AAPL.US",
"bids": [[price, size], ...],
"asks": [[price, size], ...],
"timestamp": 1700000000000
}
"""
try:
if data.get("type") != "depth":
return None
symbol = data.get("symbol", "UNKNOWN")
bids = data.get("bids", [])
asks = data.get("asks", [])
if not bids or not asks:
return None
best_bid = float(bids[0][0])
best_ask = float(asks[0][0])
mid_price = (best_bid + best_ask) / 2.0
# Compute total volume at best levels
bid_volume = sum(float(b[1]) for b in bids[:3])
ask_volume = sum(float(a[1]) for a in asks[:3])
# Determine dominant side
side = "buy" if bid_volume > ask_volume else "sell"
volume = max(bid_volume, ask_volume)
# Convert nanosecond timestamp to seconds
timestamp_sec = data.get("timestamp", 0) / 1_000_000_000
return TickData(
symbol=symbol,
price=mid_price,
volume=volume,
side=side,
timestamp=timestamp_sec,
receive_time=time.time(),
source="tickdb",
)
except (KeyError, IndexError, ValueError) as e:
logger.error(f"Failed to parse TickDB depth data: {e}")
return None
4.4 Primary Feed Adapter
def parse_primary_feed(data: dict, symbol: str) -> Optional[TickData]:
"""
Parse primary feed data into a normalized TickData structure.
Adapt this function to match your primary feed's message format.
Example primary feed format (adjust to match your vendor):
{
"type": "trade",
"symbol": "AAPL",
"price": 185.42,
"size": 100,
"side": "S",
"timestamp": 1700000000123
}
"""
try:
# Normalize symbol format (e.g., "AAPL" -> "AAPL.US")
feed_symbol = data.get("symbol", "")
if not feed_symbol.endswith(".US") and feed_symbol.endswith(".us"):
feed_symbol = feed_symbol.split(".")[0] + ".US"
if feed_symbol != symbol:
return None
price = float(data.get("price", 0))
volume = float(data.get("size", 0))
raw_side = data.get("side", "").upper()
side = "buy" if raw_side == "B" or raw_side == "BUY" else "sell"
timestamp_sec = data.get("timestamp", 0) / 1000 # ms to seconds
if price <= 0:
return None
return TickData(
symbol=feed_symbol,
price=price,
volume=volume,
side=side,
timestamp=timestamp_sec,
receive_time=time.time(),
source="primary",
)
except (KeyError, ValueError) as e:
logger.error(f"Failed to parse primary feed data: {e}")
return None
4.5 Cross-Validation Engine
class CrossValidationEngine:
"""
Core engine for dual-source cross-validation.
Compares tick streams from primary and secondary feeds,
computes deviation metrics, and triggers alerts.
"""
def __init__(self, config: ValidationConfig):
self.config = config
self.primary_buffer: deque = deque(maxlen=1000)
self.secondary_buffer: deque = deque(maxlen=1000)
self.metrics = ValidationMetrics()
self.last_alert_time: float = 0.0
self.tick_count: int = 0
self._alert_callback: Optional[Callable] = None
def set_alert_callback(self, callback: Callable[[dict], None]) -> None:
"""Register a callback function for alert dispatch."""
self._alert_callback = callback
def ingest_primary_tick(self, tick: TickData) -> None:
"""Ingest a tick from the primary feed."""
self.primary_buffer.append(tick)
self.tick_count += 1
if self.tick_count > self.config.warmup_ticks:
self._check_for_gap(tick, source="primary")
def ingest_secondary_tick(self, tick: TickData) -> None:
"""Ingest a tick from the secondary feed (TickDB)."""
self.secondary_buffer.append(tick)
self.tick_count += 1
if self.tick_count > self.config.warmup_ticks:
self._check_for_gap(tick, source="secondary")
def _check_for_gap(self, tick: TickData, source: str) -> None:
"""Detect data gaps based on inter-arrival time."""
buffer = (
self.primary_buffer
if source == "primary"
else self.secondary_buffer
)
if len(buffer) < 2:
return
prev_tick = buffer[-2]
inter_arrival = tick.receive_time - prev_tick.receive_time
if inter_arrival > self.config.max_gap_duration_sec:
self.metrics.gaps_detected += 1
logger.warning(
f"[Gap Alert] {source.upper()} feed: "
f"{inter_arrival:.2f}s gap detected (threshold: "
f"{self.config.max_gap_duration_sec}s)"
)
self._dispatch_alert(
alert_type="data_gap",
source=source,
inter_arrival_sec=inter_arrival,
last_tick_time=prev_tick.receive_time,
current_tick_time=tick.receive_time,
)
def compare_and_compute(self) -> ValidationMetrics:
"""
Compare ticks from both feeds within the comparison window.
Computes price deviation and latency skew metrics.
"""
now = time.time()
window_start = now - self.config.comparison_window_sec
# Filter ticks within the comparison window
primary_in_window = [
t for t in self.primary_buffer if t.receive_time >= window_start
]
secondary_in_window = [
t for t in self.secondary_buffer if t.receive_time >= window_start
]
# Update counts
self.metrics.primary_count = len(primary_in_window)
self.metrics.secondary_count = len(secondary_in_window)
self.metrics.window_start = window_start
# Clear previous window's deviation lists
self.metrics.price_deviations.clear()
self.metrics.latency_skews_ms.clear()
if not primary_in_window or not secondary_in_window:
logger.debug("Insufficient ticks in comparison window")
return self.metrics
# Compute metrics using the most recent ticks from each feed
primary_latest = primary_in_window[-1]
secondary_latest = secondary_in_window[-1]
# Price deviation: |primary - secondary| / mid-price
mid_price = (primary_latest.price + secondary_latest.price) / 2.0
if mid_price > 0:
price_deviation = abs(
primary_latest.price - secondary_latest.price
) / mid_price
self.metrics.price_deviations.append(price_deviation)
# Latency skew: difference in receive timestamps
latency_skew_ms = abs(
primary_latest.receive_time - secondary_latest.receive_time
) * 1000
self.metrics.latency_skews_ms.append(latency_skew_ms)
# Check for anomalies
if (
self.metrics.price_deviations
and self.metrics.price_deviations[-1]
> self.config.price_deviation_threshold
):
self.metrics.anomalies_detected += 1
logger.warning(
f"[Anomaly Alert] Price deviation: "
f"{self.metrics.price_deviations[-1]:.4%} "
f"(threshold: {self.config.price_deviation_threshold:.4%})"
)
self._dispatch_alert(
alert_type="price_anomaly",
primary_price=primary_latest.price,
secondary_price=secondary_latest.price,
deviation=self.metrics.price_deviations[-1],
)
if (
self.metrics.latency_skews_ms
and self.metrics.latency_skews_ms[-1]
> self.config.latency_skew_threshold_ms
):
logger.warning(
f"[Latency Alert] Skew: "
f"{self.metrics.latency_skews_ms[-1]:.1f}ms "
f"(threshold: {self.config.latency_skew_threshold_ms}ms)"
)
self._dispatch_alert(
alert_type="latency_skew",
skew_ms=self.metrics.latency_skews_ms[-1],
threshold_ms=self.config.latency_skew_threshold_ms,
)
return self.metrics
def _dispatch_alert(self, alert_type: str, **kwargs) -> None:
"""Dispatch an alert with cooldown enforcement."""
now = time.time()
# Enforce alert cooldown to prevent alert storms
if now - self.last_alert_time < self.config.alert_cooldown_sec:
logger.debug(
f"Alert suppressed (cooldown): {alert_type} — "
f"{self.config.alert_cooldown_sec - (now - self.last_alert_time):.1f}s remaining"
)
return
alert = {
"type": alert_type,
"timestamp": datetime.fromtimestamp(now, tz=timezone.utc).isoformat(),
"symbol": self.config.symbol,
**kwargs,
}
logger.info(f"[Alert Dispatch] {alert_type}: {alert}")
# Call registered callback (for webhook dispatch)
if self._alert_callback:
try:
self._alert_callback(alert)
except Exception as e:
logger.error(f"Alert callback failed: {e}")
self.last_alert_time = now
4.6 Alert Webhook Dispatcher
class AlertDispatcher:
"""HTTP webhook dispatcher for alert notifications."""
def __init__(self, webhook_url: str, timeout: float = 5.0):
self.webhook_url = webhook_url
self.timeout = timeout
async def send(self, alert: dict) -> bool:
"""
Send alert payload to webhook endpoint.
Returns True on success, False on failure.
"""
if not self.webhook_url:
logger.debug("No webhook URL configured — alert not dispatched")
return False
try:
async with httpx.AsyncClient() as client:
response = await client.post(
self.webhook_url,
json=alert,
headers={"Content-Type": "application/json"},
timeout=self.timeout,
)
response.raise_for_status()
logger.info(f"[Webhook] Alert sent successfully: {alert['type']}")
return True
except httpx.HTTPStatusError as e:
logger.error(f"[Webhook] HTTP error {e.response.status_code}: {e}")
return False
except httpx.RequestError as e:
logger.error(f"[Webhook] Request failed: {e}")
return False
4.7 Main Orchestration Loop
async def run_validation_monitor(config: ValidationConfig) -> None:
"""
Main orchestration loop for the dual-source validation monitor.
Establishes connections to both feeds, ingests ticks, and runs
comparison cycles at configurable intervals.
"""
# Initialize WebSocket feeds
primary_feed = WebSocketFeed(
url=config.primary_ws_url,
name="primary",
api_key=os.environ.get("PRIMARY_API_KEY"),
subscribe_message={
"cmd": "subscribe",
"params": {"channels": ["trade"], "symbols": [config.symbol]},
},
)
# TickDB as secondary feed — use depth channel for L1 order book
tickdb_feed = WebSocketFeed(
url=config.secondary_ws_url,
name="tickdb",
api_key=os.environ.get("TICKDB_API_KEY"),
subscribe_message=build_tickdb_subscription(
symbol=config.symbol, channels=["depth"]
),
)
# Initialize validation engine
engine = CrossValidationEngine(config)
# Initialize alert dispatcher if webhook is configured
dispatcher = None
if config.webhook_url:
dispatcher = AlertDispatcher(config.webhook_url)
engine.set_alert_callback(lambda alert: asyncio.create_task(dispatcher.send(alert)))
# Comparison cycle interval (seconds)
comparison_interval: float = config.comparison_window_sec
logger.info("Starting dual-source validation monitor...")
logger.info(f" Primary: {config.primary_ws_url}")
logger.info(f" Secondary (TickDB): {config.secondary_ws_url}")
logger.info(f" Symbol: {config.symbol}")
logger.info(f" Price deviation threshold: {config.price_deviation_threshold:.4%}")
logger.info(f" Latency skew threshold: {config.latency_skew_threshold_ms:.1f}ms")
try:
# Connect to both feeds
await primary_feed.connect()
await tickdb_feed.connect()
while True:
# Receive from primary feed
primary_data = await primary_feed.receive()
if primary_data:
tick = parse_primary_feed(primary_data, config.symbol)
if tick:
engine.ingest_primary_tick(tick)
# Receive from secondary feed (TickDB)
secondary_data = await tickdb_feed.receive()
if secondary_data:
tick = parse_tickdb_depth(secondary_data)
if tick:
engine.ingest_secondary_tick(tick)
# Run comparison cycle
metrics = engine.compare_and_compute()
if metrics.primary_count > 0 and metrics.secondary_count > 0:
logger.debug(
f"[Metrics] Primary: {metrics.primary_count} ticks, "
f"TickDB: {metrics.secondary_count} ticks, "
f"Gaps: {metrics.gaps_detected}, "
f"Anomalies: {metrics.anomalies_detected}"
)
# Sleep until next comparison cycle
await asyncio.sleep(comparison_interval)
except KeyboardInterrupt:
logger.info("Shutdown signal received — closing connections")
except Exception as e:
logger.error(f"Fatal error in validation loop: {e}", exc_info=True)
finally:
await primary_feed.close()
await tickdb_feed.close()
logger.info("Dual-source validation monitor stopped")
if __name__ == "__main__":
# Load configuration from environment variables
config = ValidationConfig(
symbol=os.environ.get("MONITOR_SYMBOL", "AAPL.US"),
primary_ws_url=os.environ.get(
"PRIMARY_WS_URL", "wss://your-primary-feed.example.com/ws"
),
secondary_ws_url="wss://api.tickdb.ai/ws/v1/market",
webhook_url=os.environ.get("ALERT_WEBHOOK_URL"),
price_deviation_threshold=float(
os.environ.get("PRICE_DEVIATION_THRESHOLD", "0.005")
),
latency_skew_threshold_ms=float(
os.environ.get("LATENCY_SKEW_THRESHOLD_MS", "100.0")
),
)
# Verify required API keys
if not os.environ.get("TICKDB_API_KEY"):
raise EnvironmentError(
"TICKDB_API_KEY environment variable is not set. "
"Sign up at https://tickdb.ai to obtain an API key."
)
# Run the validation monitor
asyncio.run(run_validation_monitor(config))
5. Core Algorithm: Buy/Sell Pressure Ratio
Beyond basic price comparison, the depth channel enables a more sophisticated validation metric: the buy/sell pressure ratio. This ratio compares the aggregate bid-side volume against the aggregate ask-side volume across the top N price levels, providing insight into order book imbalance.
5.1 Algorithm Definition
Buy Pressure Ratio (BPR) = Σ(bid_size[i]) for i in [1, N] / Σ(ask_size[i]) for i in [1, N]
Interpretation:
- BPR > 1.0: Buying pressure dominates (bid side is larger)
- BPR < 1.0: Selling pressure dominates (ask side is larger)
- BPR ≈ 1.0: Balanced order book
5.2 Cross-Source Pressure Ratio Comparison
When comparing pressure ratios between the primary feed and TickDB, discrepancies reveal structural differences in liquidity perception:
| Scenario | Primary BPR | TickDB BPR | Interpretation |
|---|---|---|---|
| Both > 1.0, aligned | 1.85 | 1.79 | Consistent buying pressure — low risk |
| Both > 1.0, diverging | 1.85 | 1.12 | Primary feed signals stronger bid depth |
| Primary < 1.0, TickDB > 1.0 | 0.72 | 1.34 | Sign reversal — investigate primary feed data |
| Both < 1.0 | 0.58 | 0.61 | Consistent selling pressure |
5.3 Pressure Ratio Implementation
def compute_pressure_ratio(bids: list[list], asks: list[list], levels: int = 3) -> float:
"""
Compute buy/sell pressure ratio from order book levels.
Args:
bids: List of [price, size] pairs for bid levels
asks: List of [price, size] pairs for ask levels
levels: Number of price levels to include (default: top 3)
Returns:
Pressure ratio (bid_volume / ask_volume)
"""
bid_volume = sum(float(b[1]) for b in bids[:levels])
ask_volume = sum(float(a[1]) for a in asks[:levels])
if ask_volume == 0:
return float("inf") # Theoretical infinite buying pressure
return bid_volume / ask_volume
def compare_pressure_ratios(
primary_bpr: float, tickdb_bpr: float, threshold: float = 0.25
) -> dict:
"""
Compare pressure ratios from two sources and flag significant discrepancies.
Args:
primary_bpr: Pressure ratio from primary feed
tickdb_bpr: Pressure ratio from TickDB
threshold: Maximum acceptable relative difference (default: 25%)
Returns:
Dictionary with comparison result and interpretation
"""
if primary_bpr == 0 or tickdb_bpr == 0:
return {"status": "invalid", "reason": "Zero pressure ratio"}
relative_diff = abs(primary_bpr - tickdb_bpr) / ((primary_bpr + tickdb_bpr) / 2)
return {
"status": "aligned" if relative_diff <= threshold else "diverging",
"primary_bpr": round(primary_bpr, 3),
"tickdb_bpr": round(tickdb_bpr, 3),
"relative_difference": round(relative_diff, 4),
"threshold": threshold,
"interpretation": _interpret_bpr_comparison(primary_bpr, tickdb_bpr),
}
def _interpret_bpr_comparison(primary: float, secondary: float) -> str:
"""Generate human-readable interpretation of BPR comparison."""
if primary > 1.0 and secondary > 1.0:
direction = "buying"
elif primary < 1.0 and secondary < 1.0:
direction = "selling"
elif primary > 1.0 > secondary:
return "Sign reversal: primary signals buying pressure, TickDB signals selling pressure"
elif secondary > 1.0 > primary:
return "Sign reversal: TickDB signals buying pressure, primary signals selling pressure"
else:
return "Both feeds show near-balanced order book"
avg_ratio = (primary + secondary) / 2
return f"Both feeds indicate {direction} pressure (avg BPR: {avg_ratio:.2f})"
6. Threshold Calibration Guide
Calibrating detection thresholds is the most critical — and most overlooked — step in building a dual-source validation system. Overly tight thresholds generate false positives; overly loose thresholds miss real incidents.
6.1 Recommended Starting Points by Asset Class
| Asset Class | Price Deviation Threshold | Latency Skew Threshold | Max Gap Duration |
|---|---|---|---|
| US Large-Cap Equities | 0.3% (30 bps) | 50 ms | 1.0 sec |
| US Small-Cap Equities | 0.5% (50 bps) | 100 ms | 2.0 sec |
| Crypto (major) | 0.2% (20 bps) | 30 ms | 0.5 sec |
| Crypto (alt) | 0.5% (50 bps) | 100 ms | 2.0 sec |
| HK Equities | 0.4% (40 bps) | 75 ms | 1.5 sec |
6.2 Calibration Process
Baseline collection (7 days): Run the validation monitor in logging-only mode (no alerts dispatched). Collect metrics on normal operating conditions.
Distribution analysis: Compute the 95th and 99th percentiles of price deviation and latency skew during normal conditions. Set thresholds at 1.5× the 95th percentile for initial calibration.
False positive tuning (14 days): Monitor alert frequency. If alerts fire more than 5 times per day per symbol, increase the threshold by 10–20%.
False negative review (monthly): Conduct manual review of primary feed logs during known incident periods. Measure whether the system would have caught them.
7. Deployment Guide by User Segment
| User Segment | Recommended Deployment | Infrastructure Requirements |
|---|---|---|
| Individual quant trader | Single-process Python script, cron-scheduled or systemd service | 1 vCPU, 512 MB RAM, Linux or macOS |
| Small trading team (2–5 strategies) | Containerized deployment (Docker), centralized logging | 2 vCPU, 2 GB RAM, persistent volume for incident logs |
| Institutional team (10+ strategies) | Kubernetes deployment with horizontal pod autoscaling, Prometheus metrics | 4+ vCPU, 8 GB RAM, managed logging (Datadog/Splunk), Slack webhook integration |
7.1 Environment Variable Quick Reference
| Variable | Required | Description |
|---|---|---|
TICKDB_API_KEY |
Yes | TickDB API key for secondary feed authentication |
PRIMARY_API_KEY |
Yes (if primary feed requires auth) | Primary feed API key |
PRIMARY_WS_URL |
No | Primary feed WebSocket URL (default: placeholder) |
MONITOR_SYMBOL |
No | Symbol to monitor (default: AAPL.US) |
ALERT_WEBHOOK_URL |
No | Slack/Teams webhook URL for alerts |
PRICE_DEVIATION_THRESHOLD |
No | Price deviation threshold (default: 0.005) |
LATENCY_SKEW_THRESHOLD_MS |
No | Latency skew threshold in ms (default: 100.0) |
8. Closing
"Price is the effect. The order book is the cause."
A trading system that trusts its primary data feed without verification is a trading system that trusts an unverified assumption. Every quantitative strategy — whether it runs on 1-minute mean-reversion or 4-hour trend-following — depends on the same foundational truth: the data is accurate, the timing is correct, and the delivery is complete.
Dual-source cross-validation does not eliminate all data quality risks. It introduces a systematic, automated process for detecting the 0.1% of ticks that slip through standard monitoring — the ones that silently erode edge, corrupt backtests, and trigger phantom signals.
The architecture and code provided in this article give you a production-ready foundation. Deploy it against your primary feed, calibrate thresholds to your asset class and strategy frequency, and extend it with pressure ratio analysis, order flow imbalance metrics, or custom anomaly detection as your monitoring sophistication grows.
Next Steps
If you're an individual quant trader running a single strategy, start with the single-process Python script. Set TICKDB_API_KEY and MONITOR_SYMBOL, point it at your primary feed, and review the first week's metrics before calibrating thresholds.
If you're a trading team evaluating dual-source validation for the first time, deploy the Docker container with centralized logging. Run it for 30 days in logging-only mode before enabling alert dispatch to establish a baseline.
If you need 10+ years of historical OHLCV data for backtesting your strategy, TickDB provides cleaned, aligned US equity kline data covering multiple market cycles — essential for validating that your strategy's edge survived past volatility regimes. Reach out to enterprise@tickdb.ai for institutional plans.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for faster integration of TickDB endpoints into your monitoring workflows.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. Data validation systems reduce but do not eliminate operational risk.