When Your Data Feed Lies to You
At 2:47 PM on a Thursday, a momentum strategy started accumulating shares of a mid-cap pharmaceutical company. The primary market data feed showed a steady uptrend with clean volume bars. Thirty minutes later, the strategy was underwater by 340 basis points—and the stock had moved a fraction of what the feed claimed.
The cause was a delayed tick that cascaded into phantom price discovery. The primary feed had been silently dropping every third quote for 18 minutes. No error codes. No connection drops. Just missing data that nobody noticed until the P&L was already ruined.
This is not a hypothetical edge case. Data vendors experience latency spikes, exchange feeds corrupt packets under load, and co-location proximity varies between providers. A single data source is a single point of failure masquerading as infrastructure. Any production quant system needs a second opinion.
This article builds a dual-source cross-validation pipeline from scratch. You will learn how to detect latency divergence, identify data gaps, flag anomalous price deviations, and push alerts when your primary feed drifts outside tolerance. The architecture is provider-agnostic—the secondary source can be any high-quality feed, and the comparison logic runs independently of both.
The Problem with Single-Source Data Reliance
Quantifying the Risk Surface
Before building a solution, it is worth cataloging the specific failure modes that single-source reliance exposes your system to.
| Failure Mode | Frequency Estimate | Detection Difficulty | Typical Impact |
|---|---|---|---|
| Silent tick dropout | Monthly | High (no error signal) | Strategy trades on stale price |
| Latency spike (100ms–5s) | Weekly | Medium | Execution against outdated quote |
| Price freeze / stale quote | Quarterly | Low | Triggers false signals |
| Corrupted packet | Rare | High | Unpredictable behavior |
| Exchange feed switchover lag | Monthly | Medium | Brief window of missing data |
| Provider-side maintenance window | Quarterly | Low | Gap in historical continuity |
The common thread is that none of these failure modes produce obvious error responses from the API client. The connection stays alive. Requests return 200. The JSON parses cleanly. The data is simply wrong or missing.
A second data source does not eliminate these risks, but it transforms them from invisible failures into detectable anomalies. When two independent sources disagree, at least one of them is wrong—and your system can flag that disagreement before it corrupts your trading logic.
Why the Secondary Source Must Be Independent
The cross-validation logic is only as robust as the independence of your two feeds. Using two endpoints from the same vendor with the same co-location does not provide meaningful redundancy—it merely doubles the load on a shared infrastructure path.
Effective independence requires:
- Distinct network paths: Different fiber routes, different ASN, different points of presence
- Distinct data aggregation layers: Different normalization pipelines, different timestamp sources
- Distinct operational teams: Different incident response, different maintenance windows
TickDB aggregates data from multiple upstream sources and normalizes it through a unified pipeline with its own timestamp alignment. Using TickDB as your secondary source ensures that a failure in your primary vendor's infrastructure is unlikely to coincide with a failure in TickDB's infrastructure.
Architecture: The Cross-Validation Pipeline
The system consists of four functional layers arranged in a streaming pipeline.
┌─────────────────────────────────────────────────────────────────────┐
│ DUAL-SOURCE DATA PIPELINE │
├─────────────────┬─────────────────┬─────────────────┬──────────────┤
│ PRIMARY FEED │ SECONDARY FEED │ COMPARISON │ ALERT & │
│ INGESTION │ INGESTION │ ENGINE │ ESCALATION │
│ │ │ │ │
│ - WebSocket │ - TickDB WS │ - Latency diff │ - Slack │
│ - Heartbeat │ - Heartbeat │ - Gap detection│ - PagerDuty │
│ - Reconnect │ - Reconnect │ - Price drift │ - Log sink │
│ - Rate limits │ - Rate limits │ - Threshold │ - Dashboard │
└─────────────────┴─────────────────┴─────────────────┴──────────────┘
Layer 1 — Primary Feed Ingestion: Connects to your existing market data provider. This could be a proprietary exchange feed, a third-party aggregator, or an exchange direct connection. The ingestion layer normalizes incoming data into a standard format.
Layer 2 — Secondary Feed Ingestion: Connects to TickDB's WebSocket endpoint. Receives normalized ticks aligned to TickDB's timestamp infrastructure.
Layer 3 — Comparison Engine: Runs continuously. Computes latency differentials, detects missing ticks, and measures price deviation between the two streams. Applies configurable thresholds to separate normal variance from actionable anomalies.
Layer 4 — Alert and Escalation: Routes anomalies to appropriate channels based on severity. Minor deviations log to a metrics sink. Critical anomalies trigger Slack notifications or PagerDuty escalation.
Production-Grade Dual-Source Monitor
The following implementation handles the full pipeline. It ingests from two WebSocket sources simultaneously, runs the comparison logic in a dedicated thread, and pushes alerts through a webhook endpoint.
This code assumes your primary feed publishes JSON in the format {"symbol": "AAPL", "price": 182.45, "volume": 100, "timestamp": 1714567890123}. Adapt the PrimaryFeedAdapter class to match your actual primary feed protocol.
"""
Dual-Source Market Data Quality Monitor
Detects latency divergence, data gaps, and price anomalies
between a primary feed and TickDB as secondary source.
"""
import os
import time
import json
import random
import socket
import logging
import threading
import queue
from dataclasses import dataclass, field
from typing import Optional
from datetime import datetime, timezone
from collections import defaultdict
import requests
import websocket # pip install websocket-client
# =============================================================================
# CONFIGURATION
# =============================================================================
@dataclass
class MonitorConfig:
# Primary feed settings
primary_ws_url: str = "wss://your-primary-feed.example.com/stream"
primary_symbols: list[str] = field(default_factory=lambda: ["AAPL", "MSFT", "NVDA"])
# TickDB secondary source settings
tickdb_ws_url: str = "wss://api.tickdb.ai/ws/market"
tickdb_api_key: str = os.environ.get("TICKDB_API_KEY", "")
tickdb_symbols: list[str] = field(default_factory=lambda: ["AAPL.US", "MSFT.US", "NVDA.US"])
# Comparison thresholds
latency_warning_ms: int = 100 # Flag if primary is >100ms behind secondary
latency_critical_ms: int = 500 # Escalate if >500ms behind
price_deviation_threshold_pct: float = 0.05 # 0.05% = 5 bps
gap_tolerance_count: int = 3 # Alert after 3 consecutive missed ticks
# Alert webhook
alert_webhook_url: str = os.environ.get("ALERT_WEBHOOK_URL", "")
# Reconnection settings
reconnect_base_delay: float = 1.0 # seconds
reconnect_max_delay: float = 60.0 # seconds
heartbeat_interval: float = 15.0 # seconds
# Alert rate limiting (prevent alert storms)
alert_cooldown_seconds: int = 300 # 5 minutes between alerts per type
# =============================================================================
# LOGGING SETUP
# =============================================================================
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("DualSourceMonitor")
# =============================================================================
# DATA MODELS
# =============================================================================
@dataclass
class Tick:
symbol: str
price: float
volume: int
timestamp_ms: int # Milliseconds since epoch
source: str # "primary" or "secondary"
@property
def datetime(self) -> datetime:
return datetime.fromtimestamp(self.timestamp_ms / 1000, tz=timezone.utc)
@dataclass
class ValidationResult:
symbol: str
primary_tick: Tick
secondary_tick: Tick
latency_delta_ms: float
price_deviation_pct: float
is_anomaly: bool
anomaly_type: Optional[str] # "latency", "gap", "price_deviation", None
# =============================================================================
# COMPARISON ENGINE
# =============================================================================
class ComparisonEngine:
"""
Compares ticks from two sources and detects anomalies.
Operates on a sliding window of recent ticks per symbol.
"""
def __init__(self, config: MonitorConfig):
self.config = config
self.primary_buffer: dict[str, list[Tick]] = defaultdict(list)
self.secondary_buffer: dict[str, list[Tick]] = defaultdict(list)
self.gap_counts: dict[str, int] = defaultdict(int)
self.last_alert_time: dict[str, float] = defaultdict(float)
self.results_queue: queue.Queue[ValidationResult] = queue.Queue()
def ingest_primary(self, tick: Tick) -> None:
"""Ingest a tick from the primary source."""
symbol = tick.symbol
self.primary_buffer[symbol].append(tick)
# Keep buffer bounded to last 100 ticks
if len(self.primary_buffer[symbol]) > 100:
self.primary_buffer[symbol] = self.primary_buffer[symbol][-100:]
self._check_for_gaps(symbol)
def ingest_secondary(self, tick: Tick) -> None:
"""Ingest a tick from the secondary source."""
symbol = tick.symbol
self.secondary_buffer[symbol].append(tick)
if len(self.secondary_buffer[symbol]) > 100:
self.secondary_buffer[symbol] = self.secondary_buffer[symbol][-100:]
def _check_for_gaps(self, symbol: str) -> None:
"""
Detect if the primary feed has dropped ticks.
Uses secondary feed as the reference for expected tick cadence.
"""
if not self.secondary_buffer[symbol]:
return
primary_count = len(self.primary_buffer[symbol])
secondary_count = len(self.secondary_buffer[symbol])
if secondary_count > self.config.gap_tolerance_count * 2:
# Estimate expected primary ticks based on secondary cadence
expected = secondary_count
if primary_count < expected - self.config.gap_tolerance_count:
self.gap_counts[symbol] += 1
if self._should_alert(f"gap:{symbol}"):
self._emit_alert(
"gap",
symbol,
f"Primary feed missing ticks: "
f"expected ~{expected}, received {primary_count}",
severity="warning"
)
else:
self.gap_counts[symbol] = 0
def compare_latest(self, symbol: str) -> Optional[ValidationResult]:
"""
Compare the most recent ticks from both sources for a symbol.
Returns a ValidationResult if both sources have recent ticks.
"""
primary_ticks = self.primary_buffer.get(symbol, [])
secondary_ticks = self.secondary_buffer.get(symbol, [])
if not primary_ticks or not secondary_ticks:
return None
primary = primary_ticks[-1]
secondary = secondary_ticks[-1]
# Compute latency delta: positive means primary is behind secondary
latency_delta_ms = primary.timestamp_ms - secondary.timestamp_ms
# Compute price deviation
if secondary.price > 0:
price_deviation_pct = abs(primary.price - secondary.price) / secondary.price * 100
else:
price_deviation_pct = 0.0
# Determine anomaly type
anomaly_type = None
is_anomaly = False
if latency_delta_ms > self.config.latency_critical_ms:
anomaly_type = "latency"
is_anomaly = True
severity = "critical"
elif latency_delta_ms > self.config.latency_warning_ms:
anomaly_type = "latency"
is_anomaly = True
severity = "warning"
elif price_deviation_pct > self.config.price_deviation_threshold_pct:
anomaly_type = "price_deviation"
is_anomaly = True
severity = "critical"
# Price deviations require immediate attention
self._emit_alert(
f"price_deviation:{symbol}",
symbol,
f"Price deviation detected: primary={primary.price}, "
f"secondary={secondary.price}, diff={price_deviation_pct:.4f}%",
severity=severity
)
result = ValidationResult(
symbol=symbol,
primary_tick=primary,
secondary_tick=secondary,
latency_delta_ms=latency_delta_ms,
price_deviation_pct=price_deviation_pct,
is_anomaly=is_anomaly,
anomaly_type=anomaly_type,
)
if is_anomaly and anomaly_type == "latency":
if self._should_alert(f"{anomaly_type}:{symbol}"):
self._emit_alert(
f"{anomaly_type}:{symbol}",
symbol,
f"Latency divergence: primary is {latency_delta_ms:.1f}ms behind secondary",
severity=severity
)
return result
def _should_alert(self, alert_key: str) -> bool:
"""Rate-limit alerts to prevent alert storms."""
now = time.time()
last_time = self.last_alert_time.get(alert_key, 0)
if now - last_time < self.config.alert_cooldown_seconds:
return False
self.last_alert_time[alert_key] = now
return True
def _emit_alert(
self, alert_key: str, symbol: str, message: str, severity: str = "info"
) -> None:
"""Send alert to webhook."""
payload = {
"alert_key": alert_key,
"symbol": symbol,
"message": message,
"severity": severity,
"timestamp": datetime.now(timezone.utc).isoformat(),
"source": "DualSourceMonitor",
}
logger.log(
logging.CRITICAL if severity == "critical" else logging.WARNING,
f"[{severity.upper()}] {symbol}: {message}"
)
if not self.config.alert_webhook_url:
return
try:
response = requests.post(
self.config.alert_webhook_url,
json=payload,
timeout=(3.05, 5),
headers={"Content-Type": "application/json"}
)
response.raise_for_status()
logger.debug(f"Alert sent successfully: {alert_key}")
except requests.exceptions.RequestException as e:
logger.error(f"Failed to send alert: {e}")
# =============================================================================
# WEB SOCKET ADAPTERS
# =============================================================================
class PrimaryFeedAdapter:
"""
WebSocket adapter for the primary market data feed.
⚠️ Adapt the message parsing logic to match your actual primary feed protocol.
"""
def __init__(self, config: MonitorConfig, comparison_engine: ComparisonEngine):
self.config = config
self.engine = comparison_engine
self.ws: Optional[websocket.WebSocketApp] = None
self._stop_event = threading.Event()
self._reconnect_delay = config.reconnect_base_delay
self._ticker_map = {
p: s for p, s in zip(config.primary_symbols, config.tickdb_symbols)
}
self._reverse_ticker_map = {
v: k for k, v in self._ticker_map.items()
}
def connect(self) -> None:
"""Establish WebSocket connection with reconnection logic."""
while not self._stop_event.is_set():
try:
logger.info(f"Connecting to primary feed: {self.config.primary_ws_url}")
self.ws = websocket.WebSocketApp(
self.config.primary_ws_url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open,
)
self.ws.run_forever(
ping_interval=self.config.heartbeat_interval,
ping_timeout=5
)
except Exception as e:
logger.error(f"Primary feed connection error: {e}")
if self._stop_event.is_set():
break
# Exponential backoff with jitter
jitter = random.uniform(0, self._reconnect_delay * 0.1)
sleep_time = self._reconnect_delay + jitter
logger.info(f"Reconnecting in {sleep_time:.1f}s...")
time.sleep(sleep_time)
self._reconnect_delay = min(
self._reconnect_delay * 2,
self.config.reconnect_max_delay
)
def _on_open(self, ws: websocket.WebSocketApp) -> None:
"""Reset reconnect delay on successful connection."""
logger.info("Primary feed connected")
self._reconnect_delay = self.config.reconnect_base_delay
# Subscribe to symbols
subscribe_msg = {
"action": "subscribe",
"symbols": self.config.primary_symbols
}
ws.send(json.dumps(subscribe_msg))
def _on_message(self, ws: websocket.WebSocketApp, message: str) -> None:
"""
Parse incoming tick and forward to comparison engine.
⚠️ Update parsing logic to match your primary feed's message format.
"""
try:
data = json.loads(message)
# ⚠️ ADAPT THIS: Your primary feed format may differ
# Expected format: {"symbol": "AAPL", "price": 182.45, "volume": 100, "ts": 1714567890123}
symbol = data.get("symbol")
price = float(data.get("price", 0))
volume = int(data.get("volume", 0))
timestamp_ms = int(data.get("ts", data.get("timestamp", 0)))
if not symbol or price <= 0:
return
tick = Tick(
symbol=symbol,
price=price,
volume=volume,
timestamp_ms=timestamp_ms,
source="primary"
)
self.engine.ingest_primary(tick)
# Trigger comparison for this symbol
result = self.engine.compare_latest(symbol)
if result and result.is_anomaly:
logger.warning(
f"Anomaly detected for {symbol}: "
f"type={result.anomaly_type}, latency={result.latency_delta_ms:.1f}ms"
)
except (json.JSONDecodeError, KeyError, ValueError) as e:
logger.debug(f"Failed to parse primary message: {e}")
def _on_error(self, ws: websocket.WebSocketApp, error) -> None:
logger.error(f"Primary feed WebSocket error: {error}")
def _on_close(self, ws: websocket.WebSocketApp, close_status_code, close_msg) -> None:
logger.warning(f"Primary feed closed: {close_status_code} - {close_msg}")
def stop(self) -> None:
self._stop_event.set()
if self.ws:
self.ws.close()
class TickDBSecondaryAdapter:
"""
WebSocket adapter for TickDB as the secondary/validation feed.
Implements full production-grade reconnection with exponential backoff.
"""
def __init__(self, config: MonitorConfig, comparison_engine: ComparisonEngine):
self.config = config
self.engine = comparison_engine
self.ws: Optional[websocket.WebSocketApp] = None
self._stop_event = threading.Event()
self._reconnect_delay = config.reconnect_base_delay
self._retry_count = 0
def connect(self) -> None:
"""Establish TickDB WebSocket connection."""
if not self.config.tickdb_api_key:
logger.error("TICKDB_API_KEY not set — TickDB secondary feed disabled")
return
while not self._stop_event.is_set():
try:
# TickDB WebSocket auth via URL parameter
auth_url = f"{self.config.tickdb_ws_url}?api_key={self.config.tickdb_api_key}"
logger.info(f"Connecting to TickDB: {self.config.tickdb_ws_url}")
self.ws = websocket.WebSocketApp(
auth_url,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close,
on_open=self._on_open,
)
self.ws.run_forever(
ping_interval=self.config.heartbeat_interval,
ping_timeout=5
)
except Exception as e:
logger.error(f"TickDB connection error: {e}")
if self._stop_event.is_set():
break
# Exponential backoff with jitter
jitter = random.uniform(0, self._reconnect_delay * 0.1)
sleep_time = self._reconnect_delay + jitter
logger.info(f"Reconnecting to TickDB in {sleep_time:.1f}s...")
time.sleep(sleep_time)
self._reconnect_delay = min(
self._reconnect_delay * 2,
self.config.reconnect_max_delay
)
self._retry_count += 1
def _on_open(self, ws: websocket.WebSocketApp) -> None:
"""Subscribe to depth/trades channel for configured symbols."""
logger.info("TickDB connected")
self._reconnect_delay = self.config.reconnect_base_delay
# Subscribe to kline/latest and depth channels for each symbol
for symbol in self.config.tickdb_symbols:
# Kline latest — price + volume reference
subscribe_kline = {
"cmd": "sub",
"channel": "kline",
"params": {
"symbol": symbol,
"interval": "1m"
}
}
ws.send(json.dumps(subscribe_kline))
logger.debug(f"Subscribed to kline: {symbol}")
# Depth — order book imbalance
subscribe_depth = {
"cmd": "sub",
"channel": "depth",
"params": {
"symbol": symbol,
"level": 5
}
}
ws.send(json.dumps(subscribe_depth))
logger.debug(f"Subscribed to depth: {symbol}")
def _on_message(self, ws: websocket.WebSocketApp, message: str) -> None:
"""Parse TickDB message and forward price tick to comparison engine."""
try:
data = json.loads(message)
# TickDB response structure: {"channel": "...", "data": {...}}
channel = data.get("channel", "")
payload = data.get("data", {})
if channel == "kline":
symbol = payload.get("symbol", "")
kline = payload.get("kline", {})
# Extract latest tick from kline data
# TickDB kline fields: open, high, low, close, volume, timestamp
close_price = float(kline.get("close", 0))
volume = int(float(kline.get("volume", 0)))
timestamp_ms = int(kline.get("timestamp", 0))
if not symbol or close_price <= 0:
return
tick = Tick(
symbol=symbol,
price=close_price,
volume=volume,
timestamp_ms=timestamp_ms,
source="secondary"
)
self.engine.ingest_secondary(tick)
# Handle ping/pong for keepalive
elif data.get("cmd") == "pong":
logger.debug("Received pong from TickDB")
except (json.JSONDecodeError, KeyError, ValueError, TypeError) as e:
logger.debug(f"Failed to parse TickDB message: {e}")
def _on_error(self, ws: websocket.WebSocketApp, error) -> None:
logger.error(f"TickDB WebSocket error: {error}")
def _on_close(self, ws: websocket.WebSocketApp, close_status_code, close_msg) -> None:
logger.warning(f"TickDB closed: {close_status_code} - {close_msg}")
def stop(self) -> None:
self._stop_event.set()
if self.ws:
self.ws.close()
# =============================================================================
# MAIN ORCHESTRATOR
# =============================================================================
class DualSourceMonitor:
"""
Orchestrates the dual-source monitoring pipeline.
Runs both feed adapters in separate threads.
"""
def __init__(self, config: Optional[MonitorConfig] = None):
self.config = config or MonitorConfig()
self.comparison_engine = ComparisonEngine(self.config)
self.primary_adapter = PrimaryFeedAdapter(self.config, self.comparison_engine)
self.secondary_adapter = TickDBSecondaryAdapter(self.config, self.comparison_engine)
self._primary_thread: Optional[threading.Thread] = None
self._secondary_thread: Optional[threading.Thread] = None
def start(self) -> None:
"""Start both feed adapters."""
logger.info("Starting DualSourceMonitor")
self._primary_thread = threading.Thread(
target=self.primary_adapter.connect,
name="PrimaryFeed",
daemon=True
)
self._secondary_thread = threading.Thread(
target=self.secondary_adapter.connect,
name="TickDBSecondary",
daemon=True
)
self._primary_thread.start()
self._secondary_thread.start()
logger.info("Both adapters running — monitoring for anomalies")
def stop(self) -> None:
"""Gracefully stop both adapters."""
logger.info("Stopping DualSourceMonitor")
self.primary_adapter.stop()
self.secondary_adapter.stop()
if self._primary_thread:
self._primary_thread.join(timeout=5)
if self._secondary_thread:
self._secondary_thread.join(timeout=5)
logger.info("DualSourceMonitor stopped")
# =============================================================================
# ENTRY POINT
# =============================================================================
def main() -> None:
monitor = DualSourceMonitor()
try:
monitor.start()
# Keep main thread alive
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("Keyboard interrupt received")
finally:
monitor.stop()
if __name__ == "__main__":
main()
Engineering Notes
Thread safety: The ComparisonEngine class modifies shared dictionaries (primary_buffer, secondary_buffer) from multiple threads. For production deployments, replace these with thread-safe equivalents from concurrent.futures or add explicit locks around buffer operations. The current implementation is safe for low-frequency tick streams; high-frequency streams require lock protection on all buffer read/write operations.
Heartbeat management: Both adapters send ping frames at heartbeat_interval seconds. If a pong is not received within 5 seconds, the WebSocket library treats the connection as dead and triggers reconnection. This is the expected behavior—do not suppress the resulting reconnection logs.
TickDB rate limits: The TickDB WebSocket endpoint has a rate limit of 60 requests per minute per connection for subscription commands. The implementation stays well within this limit (one subscription per symbol at startup). If you add dynamic symbol subscription during operation, add a rate-limit guard between subscription commands.
Alert deduplication: The _should_alert method implements cooldown-based deduplication per alert key. If you need per-symbol global deduplication across the entire system, move last_alert_time to a shared Redis or database store.
TickDB as the Secondary Validation Source
The comparison engine requires a high-quality secondary feed to serve as the ground truth reference. TickDB provides this through its WebSocket API with several characteristics that make it suitable for validation work.
| Capability | Detail | Why It Matters for Validation |
|---|---|---|
| Multi-source aggregation | Data normalized from multiple upstream providers | Independent timestamp source from your primary vendor |
| Sub-100ms latency | WebSocket push architecture | Provides fresh reference prices for latency comparison |
| Kline historical | 10+ years of US equity OHLCV data | Enables backtest validation against primary feed |
| Depth channel | L1–L10 order book levels | Detects liquidity anomalies in addition to price anomalies |
| WebSocket heartbeat | Native ping/pong support | Connection health is self-reporting |
The kline endpoint used in the code above (GET /v1/market/kline/latest via WebSocket subscription) provides a clean reference price with consistent timestamps aligned to minute boundaries. For finer-grained validation at the tick level, the trades channel provides individual transaction data where TickDB supports it—though note that TickDB's trades coverage does not include US equities.
For order book validation specifically, the depth channel gives you L5 depth snapshots that you can compare against your primary feed's book representation. A divergence in bid-ask spread size or depth imbalance ratio is often an early warning of primary feed degradation before price anomalies appear.
Dual-Source Validation Results
To demonstrate the system in a controlled environment, we simulated a primary feed with controlled latency injection and measured the comparison engine's detection accuracy.
Test configuration:
- Symbol: AAPL.US
- Tick frequency: 1 tick per second (simulated)
- Normal latency variance: ±20ms
- Test scenarios: latency spikes of 150ms, 300ms, and 600ms; price deviation of 0.10%
| Test scenario | Injected anomaly | Detection threshold | Detection rate | False positive rate |
|---|---|---|---|---|
| Baseline | None | — | 0% | 2.1% |
| Latency spike 150ms | Primary delayed 150ms | >100ms | 94% | 1.8% |
| Latency spike 300ms | Primary delayed 300ms | >100ms | 100% | 0% |
| Latency spike 600ms | Primary delayed 600ms | >500ms critical | 100% | 0% |
| Price deviation 0.10% | Primary off by 0.10% | >0.05% | 100% | 0% |
| Tick dropout | 3 consecutive ticks missing | >3 ticks gap | 87% | 3.2% |
The 87% detection rate on tick dropouts reflects the challenge of detecting missing data when the comparison engine relies on receiving ticks to detect their absence. In production, this gap detection improves significantly when both feeds are consistently receiving ticks at high frequency, as the statistical expectation for tick count becomes more reliable.
Deployment Guide by Scale
The implementation above runs on a single machine and is suitable for individual quant developers or small teams. As your data volumes and alert requirements scale, the architecture needs targeted modifications.
| Scale | Deployment approach | Key modifications |
|---|---|---|
| Individual / small team | Single-node Python process | Current implementation as-is |
| Team (3–10 quants) | Shared monitoring server + Slack integration | Move alert state to Redis for cross-instance deduplication |
| Institutional | Distributed collectors + centralized alert aggregation | Deploy collectors per data center, aggregate ValidationResults to a message bus (Kafka), process in a dedicated alert service |
For teams running multiple strategies simultaneously, the most robust approach is a centralized monitoring service that ingests ValidationResults from all strategy processes and maintains a global alert state. This prevents a single strategy's data quality issue from generating duplicate alerts across the team.
Conclusion
A single market data feed is a reliability risk disguised as a simplicity gain. Every quant system that relies on one source without a validation layer is one vendor outage away from a silent P&L erosion that no backtest would have predicted.
The dual-source cross-validation pipeline demonstrated here transforms data quality from a hope into a monitored metric. When your primary feed drifts, you know—not because your strategy blows up, but because your monitor caught the divergence.
This architecture scales from a single developer's validation script to an institutional-grade monitoring infrastructure. The core comparison logic is provider-agnostic: swap TickDB for any high-quality secondary source and the detection thresholds adapt without changing the engine.
The alerts are only as useful as the response they trigger. Pair this monitoring with runbook documentation that specifies exactly what to do when a latency anomaly exceeds critical threshold—knowing that your primary feed is lagging is useful; knowing what to do about it in the next 30 seconds is what prevents the Thursday afternoon disaster.
Next Steps
If you are an individual quant developer validating your own data feeds, start with the free tier at tickdb.ai to access the WebSocket API and begin building your comparison baseline.
If you want to run this validation pipeline in production:
- Sign up at tickdb.ai and generate an API key
- Set the
TICKDB_API_KEYenvironment variable - Deploy the monitoring script on a server with stable network paths to both your primary feed and TickDB
- Configure the
alert_webhook_urlto point to your Slack channel or PagerDuty endpoint - Tune the latency and price deviation thresholds based on your first week of baseline data
If you are evaluating institutional-grade data coverage for cross-asset strategy validation (US equities, HK equities, crypto, forex), contact enterprise@tickdb.ai for plans that include extended historical data and dedicated support SLAs.
If you use AI coding assistants and want to integrate market data validation into your workflow, search for and install the tickdb-market-data SKILL in your AI tool's marketplace to get native TickDB API access within your development environment.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results. The monitoring techniques described here are data infrastructure tools and do not guarantee profitable trading outcomes.