The Moment Your Data Betrays You
At 09:42:17.331 on a Tuesday morning, your quant team's mean-reversion strategy entered a position based on what it believed was a 2.3% price dislocation in Tesla. The signal was clean. The backtest Sharpe was 1.84. The live execution was catastrophic: a $340,000 drawdown in 90 seconds, reversed when the strategy was killed.
The root cause was not the model. It was not the execution layer. It was a 7-second data gap from the primary market data vendor — a gap invisible to the trading system because the feed continued transmitting heartbeats throughout. The strategy saw a stale price, interpreted the frozen data as equilibrium, and entered a position that was structurally flawed from the moment of submission.
This is not an edge case. In production trading systems, primary data source failures manifest in three primary forms:
- Latency spikes: Data arrives late but not missing, silently corrupting time-sensitive calculations.
- Data gaps: Feed interruptions that the downstream system does not detect because it continues receiving keepalives.
- Stale values: Specific fields update while others do not, creating internally inconsistent snapshots.
Detecting these failures from inside the system consuming the data is impossible. You cannot observe your own blindness. The only reliable approach is dual-source cross-validation: a second, independent data feed that continuously compares against the primary source and raises an alert when divergence exceeds a defined threshold.
This article provides a production-grade implementation of this architecture using TickDB as the secondary monitoring source.
Why a Second Source Is Not Optional
The Trust Problem
A market data feed that appears to be working — transmitting heartbeats, returning values, never throwing an error code — is not necessarily a trustworthy feed. The distinction between "the feed is connected" and "the data is accurate" is one that engineers learn the hard way.
Consider the failure modes that a heartbeat-only monitor misses:
| Failure Mode | Heartbeat Status | Data Status | Downstream Impact |
|---|---|---|---|
| Server-side staleness | Connected | Stale values | Strategy operates on delayed data |
| Partial snapshot | Connected | Incomplete fields | Internal state inconsistency |
| Timestamp drift | Connected | Misaligned timestamps | Cross-instrument correlation errors |
| Dropped messages | Connected | Gap in sequence numbers | Missing ticks, distorted OHLCV |
| Vendor-side throttle | Connected | Throttled to 1 update/sec | Strategy starved of signal |
None of these failure modes break the connection. All of them break the data.
The Regulatory Argument
For institutional trading operations, the ability to demonstrate data quality controls is increasingly a regulatory expectation. MiFID II, SEC Rule 15c3-5, and equivalent frameworks require that firms have "reasonable basis" for their market data. Dual-source validation provides an auditable record of data source health — a secondary log that proves the primary feed was operating within tolerance at any given timestamp.
Architecture: Dual-Source Validation System
System Design
The validation system runs as an independent monitoring process, separate from the trading engine. It maintains two concurrent data subscriptions:
- Primary feed: Your existing market data vendor (Polygon, Alpaca, IEX Cloud, or proprietary).
- Secondary feed: TickDB WebSocket stream, subscribed to the same symbols.
The monitor continuously compares snapshots from both sources. When deviation exceeds a configured threshold, it triggers an alert through a webhook, Slack message, or PagerDuty integration.
┌─────────────────────────────────────────────────────────────────┐
│ Dual-Source Monitor │
│ │
│ ┌──────────────┐ ┌──────────────┐ │
│ │ Primary Feed │ │ TickDB Feed │ │
│ │ (existing) │ │ (secondary) │ │
│ └──────┬───────┘ └──────┬───────┘ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Cross-Validation Engine │ │
│ │ • Timestamp alignment │ │
│ │ • Price deviation calculation │ │
│ │ • Sequence continuity check │ │
│ │ • Gap detection │ │
│ └────────────────────┬─────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────────────┐ │
│ │ Alert Router │ │
│ │ • Threshold breach → Webhook/Slack/PagerDuty│ │
│ │ • Data gap → Critical alert │ │
│ │ • Latency spike → Warning │ │
│ └──────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘
What TickDB Brings to the Architecture
TickDB serves as the secondary source for three reasons:
- Independent infrastructure: TickDB's servers are geographically and operationally separate from your primary vendor. Correlated failures are minimized.
- WebSocket push model: TickDB delivers data in real time, matching the cadence of the primary feed. No polling delay introduces noise into the comparison.
- Depth channel availability: For order-book-sensitive strategies, TickDB's
depthchannel provides L1–L10 level data on HK and crypto markets, enabling structural comparison beyond price alone.
The Cross-Validation Algorithm
Core Comparison Logic
The validation engine performs three checks on every tick:
Check 1: Timestamp Alignment
Compare the timestamps from both sources. If the primary source reports a tick at T but the TickDB tick for the same symbol arrives at T + Δ, flag if Δ exceeds the configured max_latency_threshold_ms.
def check_latency(primary_tick, secondary_tick, max_latency_ms=500):
"""
Verify that the primary and secondary sources delivered data
within the acceptable latency window.
"""
latency_ms = (secondary_tick['timestamp'] - primary_tick['timestamp']) * 1000
if abs(latency_ms) > max_latency_ms:
return {
'status': 'WARNING',
'metric': 'latency',
'value_ms': latency_ms,
'threshold_ms': max_latency_ms,
'message': f"Primary source is {latency_ms:.1f}ms behind secondary"
}
return {'status': 'OK'}
Check 2: Price Deviation
For the same timestamp window, compare the last traded price from both sources. Compute the deviation in basis points.
def check_price_deviation(primary_price, secondary_price, max_deviation_bps=10):
"""
Calculate the basis-point deviation between primary and secondary prices.
A deviation above max_deviation_bps indicates a stale primary feed,
a data error, or an arbitrage opportunity worth investigating.
"""
if primary_price is None or secondary_price is None:
return {'status': 'INSUFFICIENT_DATA'}
deviation_bps = abs(primary_price - secondary_price) / secondary_price * 10000
if deviation_bps > max_deviation_bps:
return {
'status': 'ALERT',
'metric': 'price_deviation',
'value_bps': deviation_bps,
'threshold_bps': max_deviation_bps,
'primary_price': primary_price,
'secondary_price': secondary_price,
'message': f"Price deviation {deviation_bps:.2f} bps exceeds threshold"
}
return {'status': 'OK', 'value_bps': deviation_bps}
Check 3: Sequence Continuity
Track the sequence number or tick count from each source. If the primary source reports fewer ticks than expected over a window, flag a potential data gap.
def check_sequence_continuity(primary_tick_count, expected_tick_count,
window_seconds=60, min_coverage_pct=95):
"""
Verify that the primary source delivered a statistically complete
set of ticks over the monitoring window.
"""
coverage_pct = (primary_tick_count / expected_tick_count) * 100 if expected_tick_count > 0 else 0
if coverage_pct < min_coverage_pct:
return {
'status': 'CRITICAL',
'metric': 'data_gap',
'coverage_pct': coverage_pct,
'expected': expected_tick_count,
'received': primary_tick_count,
'message': f"Primary feed coverage {coverage_pct:.1f}% below {min_coverage_pct}% threshold"
}
return {'status': 'OK', 'coverage_pct': coverage_pct}
Threshold Calibration
The thresholds in the algorithm above are conservative defaults. Calibrating them for your specific environment requires understanding three factors:
| Factor | How to measure | Typical range |
|---|---|---|
| Normal inter-source latency | Measure median Δ between primary and secondary timestamps over 1 hour of stable trading | 50–200 ms |
| Normal price deviation | Measure bps deviation between sources during non-volatile periods | 0–5 bps |
| Expected tick rate | Count ticks per minute from the secondary source during the monitoring window | Varies by symbol and liquidity |
Set your thresholds at the 99th percentile of normal behavior plus a 20% safety margin. Revisit calibration quarterly or after any change to your primary data vendor.
Production-Grade Implementation
The following code implements the full dual-source validation system. It uses a configurable primary source adapter (stubbed for any vendor) and TickDB as the secondary source.
import os
import json
import time
import logging
import threading
import statistics
from datetime import datetime, timezone
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Callable, Dict, Any, List
import websocket # websocket-client library
import requests
# ─────────────────────────────────────────────────────────────────
# Configuration
# ─────────────────────────────────────────────────────────────────
@dataclass
class MonitorConfig:
"""Configuration for the dual-source validation monitor."""
# TickDB (secondary source) settings
tickdb_api_key: str = field(default_factory=lambda: os.environ.get("TICKDB_API_KEY", ""))
tickdb_ws_url: str = "wss://api.tickdb.ai/ws/market"
tickdb_symbols: List[str] = field(default_factory=lambda: ["AAPL.US", "TSLA.US"])
# Primary source settings (configure for your vendor)
primary_source_name: str = "PrimaryVendor"
primary_source_url: str = "wss://your-primary-vendor.com/stream"
primary_api_key: str = field(default_factory=lambda: os.environ.get("PRIMARY_API_KEY", ""))
# Threshold configuration
max_latency_ms: int = 500
max_price_deviation_bps: float = 10.0
min_tick_coverage_pct: float = 95.0
monitoring_window_seconds: int = 60
# Alerting configuration
webhook_url: str = field(default_factory=lambda: os.environ.get("ALERT_WEBHOOK_URL", ""))
alert_cooldown_seconds: int = 300 # Prevent alert spam
# Logging
log_level: str = "INFO"
# ─────────────────────────────────────────────────────────────────
# Alert Router
# ─────────────────────────────────────────────────────────────────
@dataclass
class Alert:
severity: str # INFO, WARNING, CRITICAL
metric: str
message: str
timestamp: datetime
details: Dict[str, Any] = field(default_factory=dict)
class AlertRouter:
"""
Routes alerts to configured destinations (webhook, log, etc.).
Implements cooldown logic to prevent alert flooding.
"""
def __init__(self, config: MonitorConfig):
self.config = config
self.logger = logging.getLogger("AlertRouter")
self.last_alert_time: Dict[str, float] = {}
self.cooldown_seconds = config.alert_cooldown_seconds
def send(self, alert: Alert) -> bool:
"""Send an alert, respecting cooldown periods."""
current_time = time.time()
cooldown_key = f"{alert.severity}:{alert.metric}"
# Check cooldown
if cooldown_key in self.last_alert_time:
elapsed = current_time - self.last_alert_time[cooldown_key]
if elapsed < self.cooldown_seconds:
self.logger.debug(
f"Alert {cooldown_key} suppressed (cooldown: {elapsed:.1f}s remaining)"
)
return False
self.last_alert_time[cooldown_key] = current_time
# Log all alerts
log_method = {
"INFO": self.logger.info,
"WARNING": self.logger.warning,
"CRITICAL": self.logger.critical
}.get(alert.severity, self.logger.info)
log_method(f"[{alert.severity}] {alert.metric}: {alert.message}")
if alert.details:
self.logger.debug(f" Details: {json.dumps(alert.details)}")
# Send to webhook if configured
if self.config.webhook_url:
self._send_webhook(alert)
return True
def _send_webhook(self, alert: Alert) -> None:
"""Dispatch alert to configured webhook endpoint."""
payload = {
"severity": alert.severity,
"metric": alert.metric,
"message": alert.message,
"timestamp": alert.timestamp.isoformat(),
"details": alert.details,
"source": "dual-source-monitor"
}
try:
response = requests.post(
self.config.webhook_url,
json=payload,
headers={"Content-Type": "application/json"},
timeout=(3.05, 10)
)
response.raise_for_status()
self.logger.debug(f"Webhook delivered successfully: {response.status_code}")
except requests.exceptions.RequestException as e:
self.logger.error(f"Webhook delivery failed: {e}")
# ─────────────────────────────────────────────────────────────────
# TickDB WebSocket Client (Secondary Source)
# ─────────────────────────────────────────────────────────────────
class TickDBClient:
"""
Production-grade TickDB WebSocket client with heartbeat,
exponential backoff reconnection, and rate-limit handling.
Note: API key is passed as URL parameter for WebSocket auth.
"""
def __init__(self, api_key: str, ws_url: str, symbols: List[str],
on_tick: Callable[[dict], None], on_error: Callable[[str], None]):
self.api_key = api_key
self.ws_url = ws_url
self.symbols = symbols
self.on_tick = on_tick
self.on_error = on_error
self.ws: Optional[websocket.WebSocketApp] = None
self._running = False
self._reconnect_delay = 1.0
self._max_reconnect_delay = 60.0
self._reconnect_attempt = 0
self.logger = logging.getLogger("TickDBClient")
def connect(self) -> None:
"""Establish WebSocket connection with authentication."""
if self._running:
return
# API key passed as URL parameter for WebSocket auth
auth_url = f"{self.ws_url}?api_key={self.api_key}"
self.ws = websocket.WebSocketApp(
auth_url,
on_open=self._on_open,
on_message=self._on_message,
on_error=self._on_error,
on_close=self._on_close
)
self._running = True
self._reconnect_delay = 1.0 # Reset backoff on manual connect
# Run in background thread
thread = threading.Thread(target=self._run, daemon=True)
thread.start()
self.logger.info(f"Connecting to TickDB WebSocket: {self.ws_url}")
def _run(self) -> None:
"""WebSocket event loop."""
while self._running and self.ws:
try:
self.ws.run_forever(ping_interval=30, ping_timeout=10)
except Exception as e:
self.logger.error(f"WebSocket error: {e}")
if self._running:
self._schedule_reconnect()
def _schedule_reconnect(self) -> None:
"""Exponential backoff with jitter for reconnection."""
import random
self._reconnect_attempt += 1
delay = min(self._reconnect_delay * (2 ** (self._reconnect_attempt - 1)),
self._max_reconnect_delay)
# Add jitter: ±10% to prevent thundering herd
jitter = random.uniform(-delay * 0.1, delay * 0.1)
actual_delay = max(0.1, delay + jitter)
self.logger.warning(
f"Reconnecting in {actual_delay:.2f}s "
f"(attempt {self._reconnect_attempt})"
)
time.sleep(actual_delay)
def _on_open(self, ws) -> None:
"""Subscribe to symbols upon connection open."""
self._reconnect_attempt = 0
self._reconnect_delay = 1.0
self.logger.info("TickDB WebSocket connected. Subscribing to symbols...")
for symbol in self.symbols:
subscribe_msg = {
"cmd": "subscribe",
"params": {
"symbol": symbol,
"channels": ["ticker", "kline"]
}
}
ws.send(json.dumps(subscribe_msg))
self.logger.debug(f"Subscribed to {symbol}")
def _on_message(self, ws, message: str) -> None:
"""Process incoming TickDB messages."""
try:
data = json.loads(message)
# Handle ping for heartbeat
if data.get("cmd") == "pong":
self.logger.debug("Heartbeat acknowledged")
return
# Route market data ticks
if "data" in data:
for tick in data["data"]:
self.on_tick(tick)
except json.JSONDecodeError as e:
self.logger.error(f"Failed to parse TickDB message: {e}")
except Exception as e:
self.logger.error(f"Error processing TickDB tick: {e}")
def _on_error(self, ws, error) -> None:
error_msg = str(error)
self.logger.error(f"TickDB WebSocket error: {error_msg}")
self.on_error(error_msg)
def _on_close(self, ws, close_status_code, close_msg) -> None:
self.logger.warning(
f"TickDB WebSocket closed: {close_status_code} — {close_msg}"
)
def disconnect(self) -> None:
"""Gracefully disconnect the WebSocket."""
self._running = False
if self.ws:
self.ws.close()
self.logger.info("TickDB client disconnected")
# ─────────────────────────────────────────────────────────────────
# Primary Source Adapter (Stub — configure for your vendor)
# ─────────────────────────────────────────────────────────────────
class PrimarySourceAdapter:
"""
Adapter for the primary market data vendor.
Replace the stub implementation with your vendor's WebSocket or REST client.
The adapter must implement:
- connect(): Establish connection
- disconnect(): Close connection
- on_tick: Callback that receives ticks with {'symbol', 'price', 'timestamp', 'sequence'}
"""
def __init__(self, config: MonitorConfig,
on_tick: Callable[[dict], None],
on_error: Callable[[str], None]):
self.config = config
self.on_tick = on_tick
self.on_error = on_error
self.logger = logging.getLogger("PrimarySource")
self._running = False
def connect(self) -> None:
"""
Connect to the primary vendor's feed.
Replace this stub with your actual vendor integration.
"""
self.logger.info(
f"Connecting to primary source: {self.config.primary_source_name} "
f"at {self.config.primary_source_url}"
)
# ⚠️ STUB: Replace with actual vendor WebSocket client initialization
# Example for a generic WebSocket vendor:
# self.ws = websocket.WebSocketApp(
# self.config.primary_source_url,
# header={"Authorization": f"Bearer {self.config.primary_api_key}"},
# on_message=self._handle_message,
# on_error=self._handle_error
# )
# self.ws.run_forever()
self._running = True
self.logger.info("Primary source adapter initialized (STUB — configure vendor)")
def disconnect(self) -> None:
self._running = False
self.logger.info("Primary source adapter disconnected")
def _handle_message(self, ws, message: str) -> None:
"""Process and normalize primary source ticks."""
try:
data = json.loads(message)
# ⚠️ STUB: Normalize to standard tick format
# Replace with actual field mapping for your vendor
normalized_tick = {
"symbol": data.get("symbol", ""),
"price": float(data.get("last_price", 0)),
"timestamp": data.get("timestamp", time.time()),
"sequence": data.get("seq", 0)
}
self.on_tick(normalized_tick)
except Exception as e:
self.logger.error(f"Primary source tick parse error: {e}")
# ─────────────────────────────────────────────────────────────────
# Dual-Source Validation Monitor
# ─────────────────────────────────────────────────────────────────
class DualSourceMonitor:
"""
Core validation engine that compares primary and secondary data sources
and triggers alerts when thresholds are breached.
"""
def __init__(self, config: MonitorConfig):
self.config = config
self.alert_router = AlertRouter(config)
self.logger = logging.getLogger("DualSourceMonitor")
# Tick buffers for comparison
self.tick_buffers: Dict[str, deque] = {
"primary": deque(maxlen=1000),
"secondary": deque(maxlen=1000)
}
# Tick counters for sequence continuity
self.tick_counts: Dict[str, Dict[str, int]] = {
"primary": {},
"secondary": {}
}
self._running = False
self._lock = threading.Lock()
def start(self) -> None:
"""Start both data source clients and the monitoring loop."""
self._running = True
# Initialize tick counters
for symbol in self.config.tickdb_symbols:
self.tick_counts["primary"][symbol] = 0
self.tick_counts["secondary"][symbol] = 0
# Start primary source
self.primary = PrimarySourceAdapter(
self.config,
on_tick=lambda t: self._on_primary_tick(t),
on_error=lambda e: self._on_error("primary", e)
)
self.primary.connect()
# Start secondary source (TickDB)
self.secondary = TickDBClient(
api_key=self.config.tickdb_api_key,
ws_url=self.config.tickdb_ws_url,
symbols=self.config.tickdb_symbols,
on_tick=lambda t: self._on_secondary_tick(t),
on_error=lambda e: self._on_error("secondary", e)
)
self.secondary.connect()
# Start monitoring loop
monitor_thread = threading.Thread(target=self._monitoring_loop, daemon=True)
monitor_thread.start()
self.logger.info(
f"Dual-source monitor started. "
f"Monitoring: {self.config.tickdb_symbols}"
)
def stop(self) -> None:
"""Gracefully stop the monitor."""
self._running = False
if hasattr(self, 'primary'):
self.primary.disconnect()
if hasattr(self, 'secondary'):
self.secondary.disconnect()
self.logger.info("Dual-source monitor stopped")
def _on_primary_tick(self, tick: dict) -> None:
"""Buffer and count primary ticks."""
with self._lock:
symbol = tick["symbol"]
self.tick_buffers["primary"].append(tick)
self.tick_counts["primary"][symbol] = \
self.tick_counts["primary"].get(symbol, 0) + 1
def _on_secondary_tick(self, tick: dict) -> None:
"""Buffer and count secondary (TickDB) ticks."""
with self._lock:
symbol = tick.get("symbol")
if symbol:
self.tick_buffers["secondary"].append(tick)
self.tick_counts["secondary"][symbol] = \
self.tick_counts["secondary"].get(symbol, 0) + 1
def _on_error(self, source: str, error: str) -> None:
"""Handle data source errors."""
self.alert_router.send(Alert(
severity="CRITICAL",
metric="source_error",
message=f"{source.capitalize()} source error: {error}",
timestamp=datetime.now(timezone.utc),
details={"source": source, "error": error}
))
def _monitoring_loop(self) -> None:
"""Periodic validation check — runs every monitoring_window_seconds."""
while self._running:
time.sleep(self.config.monitoring_window_seconds)
with self._lock:
self._run_validation_cycle()
def _run_validation_cycle(self) -> None:
"""
Execute one validation cycle: check latency, price deviation,
and sequence continuity across all monitored symbols.
"""
for symbol in self.config.tickdb_symbols:
# Filter ticks for this symbol
primary_ticks = [t for t in self.tick_buffers["primary"]
if t.get("symbol") == symbol]
secondary_ticks = [t for t in self.tick_buffers["secondary"]
if t.get("symbol") == symbol]
if not secondary_ticks:
continue
latest_secondary = secondary_ticks[-1]
latest_primary = None
# Find the closest primary tick by timestamp
if primary_ticks:
secondary_ts = latest_secondary.get("timestamp", 0)
latest_primary = min(
primary_ticks,
key=lambda t: abs(t.get("timestamp", 0) - secondary_ts)
)
# ── Check 1: Latency ──────────────────────────────────
if latest_primary:
result = self._check_latency(latest_primary, latest_secondary)
if result["status"] != "OK":
self.alert_router.send(Alert(
severity=result["status"],
metric="latency",
message=result["message"],
timestamp=datetime.now(timezone.utc),
details={"symbol": symbol, **result}
))
# ── Check 2: Price Deviation ──────────────────────────
if latest_primary and latest_secondary:
result = self._check_price_deviation(
latest_primary.get("price"),
latest_secondary.get("close") or latest_secondary.get("last"),
symbol
)
if result["status"] != "OK":
self.alert_router.send(Alert(
severity=result["status"],
metric="price_deviation",
message=result["message"],
timestamp=datetime.now(timezone.utc),
details={"symbol": symbol, **result}
))
# ── Check 3: Sequence Continuity ──────────────────────
result = self._check_sequence_continuity(
symbol,
self.tick_counts["primary"].get(symbol, 0),
self.tick_counts["secondary"].get(symbol, 0)
)
if result["status"] != "OK":
self.alert_router.send(Alert(
severity=result["status"],
metric="data_gap",
message=result["message"],
timestamp=datetime.now(timezone.utc),
details={"symbol": symbol, **result}
))
def _check_latency(self, primary_tick: dict, secondary_tick: dict) -> dict:
"""Check timestamp alignment between sources."""
primary_ts = primary_tick.get("timestamp", 0)
secondary_ts = secondary_tick.get("timestamp", 0)
latency_ms = abs(primary_ts - secondary_ts) * 1000
if latency_ms > self.config.max_latency_ms:
return {
"status": "WARNING",
"latency_ms": latency_ms,
"threshold_ms": self.config.max_latency_ms,
"message": f"Latency {latency_ms:.1f}ms exceeds {self.config.max_latency_ms}ms threshold"
}
return {"status": "OK"}
def _check_price_deviation(self, primary_price: Optional[float],
secondary_price: Optional[float],
symbol: str) -> dict:
"""Check price deviation between sources."""
if primary_price is None or secondary_price is None:
return {"status": "INSUFFICIENT_DATA"}
if secondary_price == 0:
return {"status": "INSUFFICIENT_DATA"}
deviation_bps = abs(primary_price - secondary_price) / secondary_price * 10000
if deviation_bps > self.config.max_price_deviation_bps:
return {
"status": "ALERT",
"deviation_bps": deviation_bps,
"threshold_bps": self.config.max_price_deviation_bps,
"primary_price": primary_price,
"secondary_price": secondary_price,
"message": f"Price deviation {deviation_bps:.2f} bps exceeds threshold"
}
return {"status": "OK", "deviation_bps": deviation_bps}
def _check_sequence_continuity(self, symbol: str,
primary_count: int,
secondary_count: int) -> dict:
"""Check tick count coverage against secondary as baseline."""
if secondary_count == 0:
return {"status": "INSUFFICIENT_DATA"}
coverage_pct = (primary_count / secondary_count) * 100
if coverage_pct < self.config.min_tick_coverage_pct:
return {
"status": "CRITICAL",
"coverage_pct": coverage_pct,
"primary_count": primary_count,
"secondary_count": secondary_count,
"message": f"Tick coverage {coverage_pct:.1f}% below {self.config.min_tick_coverage_pct}% threshold"
}
return {"status": "OK", "coverage_pct": coverage_pct}
# ─────────────────────────────────────────────────────────────────
# Entry Point
# ─────────────────────────────────────────────────────────────────
def main():
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
)
logger = logging.getLogger("main")
# Load configuration from environment
config = MonitorConfig(
tickdb_api_key=os.environ.get("TICKDB_API_KEY", ""),
tickdb_symbols=os.environ.get("MONITOR_SYMBOLS", "AAPL.US,TSLA.US").split(","),
primary_source_name=os.environ.get("PRIMARY_SOURCE_NAME", "Polygon"),
primary_source_url=os.environ.get("PRIMARY_SOURCE_URL", ""),
primary_api_key=os.environ.get("PRIMARY_API_KEY", ""),
webhook_url=os.environ.get("ALERT_WEBHOOK_URL", ""),
max_latency_ms=int(os.environ.get("MAX_LATENCY_MS", "500")),
max_price_deviation_bps=float(os.environ.get("MAX_PRICE_DEVIATION_BPS", "10.0")),
min_tick_coverage_pct=float(os.environ.get("MIN_TICK_COVERAGE_PCT", "95.0"))
)
# Validate required configuration
if not config.tickdb_api_key:
logger.error("TICKDB_API_KEY environment variable is required")
return
# Start the monitor
monitor = DualSourceMonitor(config)
try:
logger.info("Starting dual-source validation monitor...")
monitor.start()
# Keep the main thread alive
while True:
time.sleep(60)
logger.debug("Heartbeat — monitor running")
except KeyboardInterrupt:
logger.info("Keyboard interrupt received — shutting down")
finally:
monitor.stop()
if __name__ == "__main__":
main()
Threshold Calibration Guide
Default thresholds will generate false positives in high-volatility conditions. Use this procedure to calibrate for your environment.
Step 1: Collect Baseline Data
Run the monitor in observation mode (logging only, no alerts) for at least one full trading day. Record the distribution of three metrics:
# Run in observation mode (set WEBHOOK_URL to empty to suppress alerts)
export TICKDB_API_KEY="your_key_here"
export PRIMARY_API_KEY="your_primary_key"
export MONITOR_SYMBOLS="AAPL.US,MSFT.US,NVDA.US"
export ALERT_WEBHOOK_URL="" # Empty = observation mode
export MAX_LATENCY_MS="999999" # Disable latency alerts
export MAX_PRICE_DEVIATION_BPS="999999" # Disable deviation alerts
python dual_source_monitor.py
After 4+ hours, extract the logged value_ms, value_bps, and coverage_pct from your logs.
Step 2: Compute Percentiles
import statistics
latencies_ms = [...] # Your collected latency data
deviations_bps = [...] # Your collected deviation data
coverages_pct = [...] # Your collected coverage data
def percentile(data, p):
sorted_data = sorted(data)
index = (len(sorted_data) - 1) * p / 100
floor, ceil = int(index // 1), int(index // 1) + 1
if ceil >= len(sorted_data):
return sorted_data[floor]
return sorted_data[floor] * (ceil - index) + sorted_data[ceil] * (index - floor)
print(f"Latency 99th percentile: {percentile(latencies_ms, 99):.1f} ms")
print(f"Deviation 99th percentile: {percentile(deviations_bps, 99):.2f} bps")
print(f"Coverage 1st percentile: {percentile(coverages_pct, 1):.1f}%")
Step 3: Set Thresholds
| Metric | Recommended threshold | Safety margin |
|---|---|---|
max_latency_ms |
99th percentile + 20% | Prevents alerts during normal jitter |
max_price_deviation_bps |
99th percentile + 20% | Prevents alerts during normal spread |
min_tick_coverage_pct |
1st percentile − 5% | Catches genuine gaps, not sampling noise |
Deployment Recommendations by Scale
| Deployment size | Recommended configuration | TickDB tier |
|---|---|---|
| Individual quant / single strategy | 1 monitor instance, 3–5 symbols, Slack webhook | Free tier (5 symbols, 1 req/sec) |
| Small team / 3–5 strategies | 1 monitor instance, 10–20 symbols, Slack + PagerDuty | Starter ($49/mo, 20 symbols) |
| Institutional / multi-strategy | 2–3 redundant monitor instances, full symbol universe, enterprise webhook | Professional / Enterprise |
For production deployments, run the monitor on a separate host from the trading engine. This ensures that a host-level failure does not take down both the monitor and the system it is monitoring.
Closing: Know When Your Data Lies
A trading system that cannot detect its own data failures is a trading system waiting to fail at the worst possible moment. The 7-second gap that destroyed your mean-reversion strategy is not a black swan. It is a predictable consequence of relying on a single data source without external verification.
Dual-source cross-validation does not eliminate risk. It converts invisible risk into visible signals. When the alert fires, you gain a window — whether to halt the strategy, switch to a fallback data source, or simply log the incident for later analysis.
The code in this article is production-ready. Calibrate the thresholds to your environment, connect your alert routing, and deploy on infrastructure separate from your trading engine. Then, when your primary source next fails silently, you will know.
Next Steps
If you want to implement dual-source validation today:
- Sign up at tickdb.ai (free tier available, no credit card required)
- Generate an API key in the dashboard
- Set the
TICKDB_API_KEYenvironment variable, then copy-paste the code from this article - Replace the
PrimarySourceAdapterstub with your actual vendor's WebSocket client
If you need higher symbol limits and lower latency for institutional deployment, reach out to enterprise@tickdb.ai for Professional and Enterprise plans with dedicated infrastructure and SLA guarantees.
If you use AI coding assistants, search for and install the tickdb-market-data SKILL in your AI tool's marketplace for integrated TickDB API access within your development workflow.
This article does not constitute investment advice. Markets involve risk; past performance does not guarantee future results.