Real-time cryptocurrency market data powers trading strategies, risk management systems, and financial analytics. When that data arrives late, incomplete, or corrupted, the consequences can be devastating—missed trade executions, inaccurate portfolio valuations, or worse, catastrophic losses from stale quotes. This comprehensive guide walks you through building a production-grade data quality monitoring system using HolySheep AI's Tardis.dev relay, designed specifically for beginners with zero API experience.
HolySheep AI provides unified access to raw exchange feeds including Binance, Bybit, OKX, and Deribit with sub-50ms latency. Sign up here to receive free credits and start monitoring your data quality today.
What You Will Learn
- Understanding Tardis API data streams and message types
- Measuring data latency with millisecond precision
- Detecting data gaps and sequence breaks
- Building automated alerting for data integrity violations
- Implementing real-time monitoring dashboards
- Scaling your monitoring infrastructure for production environments
Understanding Tardis.dev Data Streams
Tardis.dev, integrated into HolySheep AI's infrastructure, captures every market event from supported exchanges. Think of it as a continuous video feed of market activity—each message represents a single moment in time: a trade executed, an order book updated, a funding rate changed, or a liquidation triggered.
Before diving into monitoring, let's understand the four primary message types you'll encounter:
- Trades: Individual buy/sell executions with price, size, and timestamp
- Order Book Deltas: Changes to the bid/ask ladder with sequence numbers
- Liquidations: Forced position closures with estimated loss amounts
- Funding Rates: Periodic payment exchanges between long and short positions
Who This Tutorial Is For
| This Guide Is Perfect For | This Guide Is NOT For |
|---|---|
| Developers new to cryptocurrency APIs | Experienced HFT firms with proprietary monitoring |
| Trading strategy developers needing reliable data | Those seeking historical backtesting data only |
| Risk management teams monitoring data quality | Casual hobbyists with no production systems |
| Quantitative researchers building analytics pipelines | Developers unwilling to write any code |
Setting Up Your HolySheep AI Environment
The first step is obtaining your API credentials. HolySheep AI offers a streamlined onboarding process with support for WeChat and Alipay payments, making it particularly convenient for users in Asia-Pacific regions. New registrations receive complimentary credits—enough to run substantial monitoring tests before committing to a paid plan.
Access the HolySheep AI dashboard by navigating to the API keys section. You'll receive two critical pieces of information:
- API Key: Your unique identifier (format:
hs_xxxxxxxxxxxxxxxx) - API Secret: Your authentication token (keep this private)
Your First Data Quality Check: Hello World Example
Let's start with the simplest possible example—connecting to the market data stream and verifying you receive messages. This foundational code will become the skeleton for all advanced monitoring we'll build later.
#!/usr/bin/env python3
"""
HolySheep AI - Tardis API Data Quality Monitor
Minimal working example for complete beginners
"""
import requests
import json
import time
from datetime import datetime
============================================
CONFIGURATION - Replace with your credentials
============================================
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Get from https://www.holysheep.ai/register
Supported exchanges: binance, bybit, okx, deribit
EXCHANGE = "binance"
SYMBOL = "BTC-USDT"
def get_server_time():
"""Check API connectivity and measure response latency"""
headers = {"X-API-Key": API_KEY}
start = time.perf_counter()
response = requests.get(
f"{BASE_URL}/time",
headers=headers,
timeout=10
)
latency_ms = (time.perf_counter() - start) * 1000
if response.status_code == 200:
data = response.json()
print(f"✓ Connected to HolySheep API")
print(f"✓ Server time: {data.get('server_time')}")
print(f"✓ Round-trip latency: {latency_ms:.2f}ms")
return True, latency_ms
else:
print(f"✗ Connection failed: {response.status_code}")
print(f" Response: {response.text}")
return False, latency_ms
def subscribe_to_trades():
"""Subscribe to real-time trade stream"""
headers = {
"X-API-Key": API_KEY,
"Content-Type": "application/json"
}
payload = {
"exchange": EXCHANGE,
"symbol": SYMBOL,
"channel": "trades",
"limit": 10 # Receive last 10 trades
}
print(f"\n📡 Subscribing to {EXCHANGE}/{SYMBOL} trade stream...")
response = requests.post(
f"{BASE_URL}/subscribe",
headers=headers,
json=payload,
timeout=10
)
if response.status_code == 200:
data = response.json()
trades = data.get('trades', [])
print(f"✓ Subscribed successfully")
print(f"✓ Received {len(trades)} recent trades")
for trade in trades[:3]: # Show first 3
print(f" {trade.get('timestamp')} | "
f"Price: {trade.get('price')} | "
f"Size: {trade.get('size')}")
return data
else:
print(f"✗ Subscription failed: {response.status_code}")
return None
Main execution
if __name__ == "__main__":
print("=" * 50)
print("HolySheep AI - Data Quality Monitor")
print("=" * 50)
connected, latency = get_server_time()
if connected:
subscribe_to_trades()
print("\n✅ Basic connectivity test complete!")What to expect when you run this: After replacing YOUR_HOLYSHEEP_API_KEY with your actual key, you should see output confirming connection to HolySheep's servers with sub-50ms latency (HolySheep guarantees <50ms globally). The terminal will display recent BTC-USDT trades from Binance with timestamps.
Building the Data Quality Monitor Class
Now we'll build a comprehensive monitoring system that tracks three critical metrics: latency (time between exchange event and your receipt), completeness (no missing sequence numbers), and validity (data falls within expected ranges).
#!/usr/bin/env python3
"""
HolySheep AI - Production Data Quality Monitor
Tracks latency, completeness, and data validity in real-time
"""
import requests
import json
import time
import threading
from datetime import datetime, timezone
from collections import deque
from dataclasses import dataclass, field
from typing import Optional, Dict, List
import statistics
============================================
CONFIGURATION
============================================
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class QualityMetrics:
"""Stores aggregated quality metrics"""
latency_samples: deque = field(default_factory=lambda: deque(maxlen=100))
sequence_gaps: List[int] = field(default_factory=list)
invalid_messages: List[str] = field(default_factory=list)
total_messages: int = 0
last_sequence: Optional[int] = None
last_timestamp: Optional[int] = None
@property
def avg_latency_ms(self) -> float:
return statistics.mean(self.latency_samples) if self.latency_samples else 0
@property
def p99_latency_ms(self) -> float:
if not self.latency_samples:
return 0
sorted_latencies = sorted(self.latency_samples)
idx = int(len(sorted_latencies) * 0.99)
return sorted_latencies[min(idx, len(sorted_latencies) - 1)]
@property
def gap_count(self) -> int:
return len(self.sequence_gaps)
@property
def completeness_rate(self) -> float:
if self.total_messages == 0:
return 100.0
return ((self.total_messages - self.gap_count) / self.total_messages) * 100
class DataQualityMonitor:
"""
Monitors Tardis API data streams for quality issues.
Tracks latency, sequence integrity, and message validity.
"""
def __init__(self, api_key: str, base_url: str = BASE_URL):
self.api_key = api_key
self.base_url = base_url
self.headers = {"X-API-Key": api_key}
self.metrics = QualityMetrics()
self.alerts = []
self._lock = threading.Lock()
# Thresholds for alerting (customize based on requirements)
self.latency_threshold_ms = 100 # Alert if avg latency exceeds 100ms
self.p99_threshold_ms = 200 # Alert if P99 latency exceeds 200ms
self.completeness_threshold = 99.0 # Alert if completeness drops below 99%
def process_trade(self, trade: Dict) -> None:
"""Process a single trade message and update metrics"""
with self._lock:
self.metrics.total_messages += 1
# 1. LATENCY MEASUREMENT
exchange_timestamp = trade.get('timestamp', 0)
receive_timestamp = int(time.time() * 1000)
# If exchange provides millisecond timestamp
if exchange_timestamp > 1_000_000_000_000:
latency = receive_timestamp - exchange_timestamp
elif exchange_timestamp > 1_000_000_000:
latency = receive_timestamp - (exchange_timestamp * 1000)
else:
latency = receive_timestamp - exchange_timestamp
self.metrics.latency_samples.append(max(0, latency))
# 2. SEQUENCE COMPLETENESS CHECK
sequence = trade.get('sequence')
if sequence is not None and self.metrics.last_sequence is not None:
expected = self.metrics.last_sequence + 1
if sequence != expected:
gap = sequence - expected
self.metrics.sequence_gaps.append(gap)
self._create_alert(
"SEQUENCE_GAP",
f"Gap of {gap} sequences detected (expected {expected}, got {sequence})",
severity="HIGH"
)
if sequence is not None:
self.metrics.last_sequence = sequence
# 3. DATA VALIDITY CHECKS
price = trade.get('price')
size = trade.get('size')
if price is None or price <= 0:
self._create_alert("INVALID_PRICE", f"Invalid price: {price}", severity="HIGH")
if size is None or size <= 0:
self._create_alert("INVALID_SIZE", f"Invalid size: {size}", severity="MEDIUM")
# Check for stale data (more than 5 seconds old)
if exchange_timestamp:
age_ms = receive_timestamp - (exchange_timestamp if exchange_timestamp > 1_000_000_000_000 else exchange_timestamp * 1000)
if age_ms > 5000:
self._create_alert("STALE_DATA", f"Data is {age_ms}ms old", severity="HIGH")
def _create_alert(self, alert_type: str, message: str, severity: str = "LOW") -> None:
"""Create and store an alert"""
alert = {
"type": alert_type,
"message": message,
"severity": severity,
"timestamp": datetime.now(timezone.utc).isoformat()
}
self.alerts.append(alert)
print(f"🚨 ALERT [{severity}] {alert_type}: {message}")
def check_thresholds(self) -> List[str]:
"""Check if any metrics exceed configured thresholds"""
warnings = []
if self.metrics.avg_latency_ms > self.latency_threshold_ms:
warnings.append(f"High average latency: {self.metrics.avg_latency_ms:.1f}ms")
if self.metrics.p99_latency_ms > self.p99_threshold_ms:
warnings.append(f"High P99 latency: {self.metrics.p99_latency_ms:.1f}ms")
if self.metrics.completeness_rate < self.completeness_threshold:
warnings.append(f"Low completeness: {self.metrics.completeness_rate:.2f}%")
return warnings
def get_report(self) -> Dict:
"""Generate current quality report"""
with self._lock:
return {
"total_messages": self.metrics.total_messages,
"latency": {
"avg_ms": round(self.metrics.avg_latency_ms, 2),
"p99_ms": round(self.metrics.p99_latency_ms, 2),
"samples": len(self.metrics.latency_samples)
},
"completeness": {
"rate_percent": round(self.metrics.completeness_rate, 4),
"gap_count": self.metrics.gap_count
},
"alerts": {
"total": len(self.alerts),
"recent": self.alerts[-5:] # Last 5 alerts
},
"status": "HEALTHY" if not self.check_thresholds() else "DEGRADED"
}
def stream_trades(self, exchange: str, symbol: str, duration_seconds: int = 60):
"""
Stream and monitor trades for specified duration.
This is a polling implementation - for production, use WebSocket.
"""
print(f"📡 Starting {duration_seconds}s monitoring session...")
print(f" Exchange: {exchange}")
print(f" Symbol: {symbol}")
print("-" * 50)
start_time = time.time()
poll_interval = 0.5 # Poll every 500ms
while time.time() - start_time < duration_seconds:
try:
response = requests.post(
f"{self.base_url}/subscribe",
headers=self.headers,
json={
"exchange": exchange,
"symbol": symbol,
"channel": "trades",
"limit": 100
},
timeout=5
)
if response.status_code == 200:
data = response.json()
trades = data.get('trades', [])
for trade in trades:
self.process_trade(trade)
if trades:
print(f"\r Progress: {time.time() - start_time:.0f}s | "
f"Msgs: {self.metrics.total_messages} | "
f"Latency: {self.metrics.avg_latency_ms:.1f}ms | "
f"Complete: {self.metrics.completeness_rate:.2f}%", end="")
except Exception as e:
print(f"\n⚠️ Stream error: {e}")
time.sleep(poll_interval)
print("\n" + "=" * 50)
print("📊 FINAL QUALITY REPORT")
print("=" * 50)
report = self.get_report()
print(json.dumps(report, indent=2))
return report
============================================
USAGE EXAMPLE
============================================
if __name__ == "__main__":
# Initialize monitor
monitor = DataQualityMonitor(
api_key=API_KEY,
base_url=BASE_URL
)
# Run 30-second monitoring session
report = monitor.stream_trades(
exchange="binance",
symbol="BTC-USDT",
duration_seconds=30
)
# Check for threshold violations
warnings = monitor.check_thresholds()
if warnings:
print("\n⚠️ THRESHOLD WARNINGS:")
for warning in warnings:
print(f" - {warning}")Running this monitor provides: Real-time visibility into your data quality with millisecond-precision latency tracking, automatic detection of sequence gaps that indicate missing data, and configurable alerting when metrics breach your defined thresholds. HolySheep's sub-50ms infrastructure means your latency readings should consistently stay well within acceptable ranges.
Building a Webhook Alerting System
For production deployments, you need alerts delivered to your incident management platform—Slack, PagerDuty, Discord, or custom webhooks. This section implements a flexible alerting pipeline that integrates with any notification system.
#!/usr/bin/env python3
"""
HolySheep AI - Webhook Alert System for Data Quality Monitoring
Integrates with Slack, PagerDuty, Discord, or custom endpoints
"""
import requests
import json
import hmac
import hashlib
import time
from datetime import datetime, timezone
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
class AlertSeverity(Enum):
LOW = "low"
MEDIUM = "medium"
HIGH = "high"
CRITICAL = "critical"
@dataclass
class Alert:
"""Represents a data quality alert"""
severity: AlertSeverity
title: str
description: str
metrics: Dict = None
timestamp: str = None
def __post_init__(self):
if self.timestamp is None:
self.timestamp = datetime.now(timezone.utc).isoformat()
class WebhookAlertManager:
"""
Manages alert delivery to multiple webhook endpoints.
Supports Slack, Discord, generic webhooks with signature verification.
"""
def __init__(self):
self.webhooks: List[Dict] = []
self.alert_history: List[Alert] = []
self.alert_callbacks: List[Callable] = []
# Rate limiting (prevent alert storms)
self.min_alert_interval_seconds = 60
self.last_alert_times: Dict[str, float] = {}
def add_slack_webhook(self, url: str, channel: str = None, mention_user: str = None):
"""Add a Slack incoming webhook"""
self.webhooks.append({
"type": "slack",
"url": url,
"channel": channel,
"mention_user": mention_user
})
print(f"✓ Added Slack webhook for channel: {channel or 'default'}")
def add_discord_webhook(self, url: str, username: str = "Data Monitor"):
"""Add a Discord webhook"""
self.webhooks.append({
"type": "discord",
"url": url,
"username": username
})
print(f"✓ Added Discord webhook: {username}")
def add_generic_webhook(self, url: str, secret: str = None, headers: Dict = None):
"""Add a generic webhook with optional HMAC signature"""
self.webhooks.append({
"type": "generic",
"url": url,
"secret": secret,
"headers": headers or {}
})
print(f"✓ Added generic webhook: {url}")
def add_alert_callback(self, callback: Callable[[Alert], None]):
"""Add a Python callback function to be called on each alert"""
self.alert_callbacks.append(callback)
def _create_slack_payload(self, alert: Alert) -> Dict:
"""Format alert for Slack"""
color_map = {
AlertSeverity.LOW: "#36a64f",
AlertSeverity.MEDIUM: "#ff9800",
AlertSeverity.HIGH: "#f44336",
AlertSeverity.CRITICAL: "#9c27b0"
}
fields = [
{"title": "Severity", "value": alert.severity.value.upper(), "short": True},
{"title": "Timestamp", "value": alert.timestamp, "short": True}
]
if alert.metrics:
if "latency_ms" in alert.metrics:
fields.append({
"title": "Latency",
"value": f"{alert.metrics['latency_ms']}ms",
"short": True
})
if "completeness" in alert.metrics:
fields.append({
"title": "Completeness",
"value": f"{alert.metrics['completeness']}%",
"short": True
})
return {
"attachments": [{
"color": color_map.get(alert.severity, "#808080"),
"title": f"🚨 {alert.title}",
"text": alert.description,
"fields": fields,
"footer": "HolySheep Data Quality Monitor",
"ts": int(time.time())
}]
}
def _create_discord_payload(self, alert: Alert) -> Dict:
"""Format alert for Discord"""
emoji_map = {
AlertSeverity.LOW: "ℹ️",
AlertSeverity.MEDIUM: "⚠️",
AlertSeverity.HIGH: "🚨",
AlertSeverity.CRITICAL: "💀"
}
description = f"**{alert.description}**"
if alert.metrics:
metrics_str = " | ".join([
f"{k}: {v}" for k, v in alert.metrics.items()
])
description += f"\n\n``\n{metrics_str}\n``"
return {
"username": "HolySheep Monitor",
"content": f"{emoji_map.get(alert.severity, '📊')} **{alert.title}**",
"embeds": [{
"description": description,
"timestamp": alert.timestamp,
"color": {
AlertSeverity.LOW: 0x36a64f,
AlertSeverity.MEDIUM: 0xff9800,
AlertSeverity.HIGH: 0xf44336,
AlertSeverity.CRITICAL: 0x9c27b0
}.get(alert.severity, 0x808080)
}]
}
def _sign_payload(self, payload: str, secret: str) -> str:
"""Create HMAC-SHA256 signature for webhook verification"""
signature = hmac.new(
secret.encode(),
payload.encode(),
hashlib.sha256
).hexdigest()
return f"sha256={signature}"
def send_alert(self, alert: Alert, force: bool = False) -> bool:
"""
Send alert to all configured webhooks.
Set force=True to bypass rate limiting.
"""
# Rate limiting check
if not force:
if alert.title in self.last_alert_times:
elapsed = time.time() - self.last_alert_times[alert.title]
if elapsed < self.min_alert_interval_seconds:
return False
self.last_alert_times[alert.title] = time.time()
self.alert_history.append(alert)
# Execute callbacks
for callback in self.alert_callbacks:
try:
callback(alert)
except Exception as e:
print(f"⚠️ Callback error: {e}")
# Send to webhooks
success_count = 0
for webhook in self.webhooks:
try:
if webhook["type"] == "slack":
payload = self._create_slack_payload(alert)
elif webhook["type"] == "discord":
payload = self._create_discord_payload(alert)
else:
payload = {"alert": {
"title": alert.title,
"description": alert.description,
"severity": alert.severity.value,
"timestamp": alert.timestamp,
"metrics": alert.metrics
}}
response = requests.post(
webhook["url"],
json=payload,
timeout=10
)
if response.status_code in (200, 204):
success_count += 1
else:
print(f"⚠️ Webhook failed ({webhook['type']}): {response.status_code}")
except Exception as e:
print(f"⚠️ Webhook error ({webhook['type']}): {e}")
return success_count > 0
============================================
INTEGRATION EXAMPLE
============================================
def create_data_quality_alert(monitor, alert_manager):
"""Generate alerts based on monitor thresholds"""
report = monitor.get_report()
warnings = monitor.check_thresholds()
if not warnings:
return
# Determine severity based on warning count
severity = AlertSeverity.LOW
if len(warnings) >= 2:
severity = AlertSeverity.HIGH
elif warnings:
severity = AlertSeverity.MEDIUM
alert = Alert(
severity=severity,
title="Data Quality Threshold Exceeded",
description="; ".join(warnings),
metrics={
"latency_ms": report["latency"]["avg_ms"],
"p99_latency_ms": report["latency"]["p99_ms"],
"completeness": report["completeness"]["rate_percent"],
"total_messages": report["total_messages"]
}
)
alert_manager.send_alert(alert)
Example usage:
if __name__ == "__main__":
# Initialize alert manager
manager = WebhookAlertManager()
# Add your webhooks here:
# manager.add_slack_webhook("https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK", "#alerts")
# manager.add_discord_webhook("https://discord.com/api/webhooks/YOUR/DISCORD/WEBHOOK")
# Add custom callback
def log_alert(alert: Alert):
print(f"📝 Custom log: {alert.title} - {alert.description}")
manager.add_alert_callback(log_alert)
# Test alert
test_alert = Alert(
severity=AlertSeverity.MEDIUM,
title="Test Alert",
description="This is a test alert from the monitoring system",
metrics={"latency_ms": 45.2, "completeness": 99.95}
)
print("\n" + "=" * 50)
print("Testing alert system...")
print("=" * 50)
# Note: Webhooks not configured, so only callback will fire
manager.send_alert(test_alert, force=True)
print("\n✓ Alert system initialized successfully")Monitoring Multiple Exchanges Simultaneously
Production systems typically ingest data from multiple exchanges for arbitrage, cross-exchange analysis, or redundancy. This example demonstrates coordinated monitoring across Binance, Bybit, OKX, and Deribit with unified alerting.
#!/usr/bin/env python3
"""
HolySheep AI - Multi-Exchange Data Quality Monitor
Monitors Binance, Bybit, OKX, and Deribit simultaneously
"""
import requests
import json
import time
import threading
from datetime import datetime, timezone
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from concurrent.futures import ThreadPoolExecutor, as_completed
BASE_URL = "https://api.holysheep.ai/v1"
API_KEY = "YOUR_HOLYSHEEP_API_KEY"
@dataclass
class ExchangeHealth:
"""Health status for a single exchange"""
name: str
is_connected: bool = False
messages_received: int = 0
last_message_time: Optional[float] = None
avg_latency_ms: float = 0.0
errors: List[str] = field(default_factory=list)
@property
def stale_seconds(self) -> float:
if self.last_message_time is None:
return float('inf')
return time.time() - self.last_message_time
class MultiExchangeMonitor:
"""
Coordinates data quality monitoring across multiple exchanges.
Provides unified view and cross-exchange comparisons.
"""
SUPPORTED_EXCHANGES = {
"binance": {"name": "Binance", "has_trades": True, "has_orderbook": True},
"bybit": {"name": "Bybit", "has_trades": True, "has_orderbook": True},
"okx": {"name": "OKX", "has_trades": True, "has_orderbook": True},
"deribit": {"name": "Deribit", "has_trades": True, "has_orderbook": False}
}
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {"X-API-Key": api_key}
self.exchanges: Dict[str, ExchangeHealth] = {
name: ExchangeHealth(name=name)
for name in self.SUPPORTED_EXCHANGES
}
self._lock = threading.Lock()
self.monitoring = False
self.poll_thread: Optional[threading.Thread] = None
def check_exchange(self, exchange: str, symbol: str = "BTC-USDT") -> Dict:
"""Check connectivity and latency for a single exchange"""
start = time.perf_counter()
try:
response = requests.post(
f"{BASE_URL}/subscribe",
headers=self.headers,
json={
"exchange": exchange,
"symbol": symbol,
"channel": "trades",
"limit": 10
},
timeout=5
)
latency_ms = (time.perf_counter() - start) * 1000
if response.status_code == 200:
data = response.json()
trades = data.get('trades', [])
with self._lock:
self.exchanges[exchange].is_connected = True
self.exchanges[exchange].messages_received += len(trades)
self.exchanges[exchange].last_message_time = time.time()
# Update rolling average latency
current_avg = self.exchanges[exchange].avg_latency_ms
count = self.exchanges[exchange].messages_received
new_avg = ((current_avg * (count - len(trades)) +
latency_ms * len(trades)) / count) if count > 0 else latency_ms
self.exchanges[exchange].avg_latency_ms = new_avg
return {
"exchange": exchange,
"status": "OK",
"latency_ms": round(latency_ms, 2),
"trades_received": len(trades)
}
else:
with self._lock:
self.exchanges[exchange].is_connected = False
self.exchanges[exchange].errors.append(
f"HTTP {response.status_code}"
)
return {
"exchange": exchange,
"status": "ERROR",
"error": f"HTTP {response.status_code}"
}
except requests.exceptions.Timeout:
with self._lock:
self.exchanges[exchange].is_connected = False
self.exchanges[exchange].errors.append("Timeout")
return {
"exchange": exchange,
"status": "TIMEOUT",
"error": "Connection timed out"
}
except Exception as e:
with self._lock:
self.exchanges[exchange].is_connected = False
self.exchanges[exchange].errors.append(str(e))
return {
"exchange": exchange,
"status": "ERROR",
"error": str(e)
}
def check_all_exchanges(self, symbol: str = "BTC-USDT") -> List[Dict]:
"""Check all exchanges in parallel"""
results = []
with ThreadPoolExecutor(max_workers=4) as executor:
futures = {
executor.submit(self.check_exchange, exchange, symbol): exchange
for exchange in self.SUPPORTED_EXCHANGES
}
for future in as_completed(futures):
exchange = futures[future]
try:
result = future.result()
results.append(result)
except Exception as e:
results.append({
"exchange": exchange,
"status": "ERROR",
"error": str(e)
})
return results
def get_unified_report(self) -> Dict:
"""Generate unified health report across all exchanges"""
with self._lock:
exchange_status = {}
all_latencies = []
for name, health in self.exchanges.items():
exchange_status[name] = {
"connected": health.is_connected,
"messages": health.messages_received,
"latency_ms": round(health.avg_latency_ms, 2),
"stale_seconds": round(health.stale_seconds, 1),
"errors": health.errors[-3:] # Last 3 errors
}
if health.is_connected:
all_latencies.append(health.avg_latency_ms)
return {
"timestamp": datetime.now(timezone.utc).isoformat(),
"exchanges": exchange_status,
"summary": {
"total_connected": sum(1 for h in self.exchanges.values() if h.is_connected),
"total_exchanges": len(self.exchanges),
"avg_latency_ms": round(sum(all_latencies) / len(all_latencies), 2) if all_latencies else 0,
"max_latency_ms": round(max(all_latencies), 2) if all_latencies else 0
}
}
def start_monitoring(self, interval_seconds: int = 10):
"""Start continuous monitoring in background thread"""
self.monitoring = True
def monitor_loop():
while self.monitoring:
results = self.check_all_exchanges()
# Check for issues
issues = [r for r in results if r["status"] != "OK"]
if issues:
print(f"\n⚠️ Issues detected:")
for issue in issues:
print(f" {issue['exchange']}: {issue.get('error', issue['status'])}")
# Print status
report = self.get_unified_report()
print(f"\r[{datetime.now().strftime('%H:%M:%S')}] "
f"Connected: {report['summary']['total_connected']}/"
f"{report['summary']['total_exchanges']} | "
f"Avg Latency: {report['summary']['avg_latency_ms']}ms", end="")
time.sleep(interval_seconds)
self.poll_thread = threading.Thread(target=monitor_loop, daemon=True)
self.poll_thread.start()
print("✓ Started multi-exchange monitoring")
def stop_monitoring(self):
"""Stop continuous monitoring"""
self.monitoring = False
if self.poll_thread:
self.poll_thread.join(timeout=5)
print("✓ Stopped monitoring")
============================================
USAGE EXAMPLE
============================================
if __name__ == "__main__":
monitor = MultiExchangeMonitor(api_key=API_KEY)
print("=" * 60)
print("HolySheep AI - Multi-Exchange Data Quality Monitor")
print("=" * 60)
# Run one-time check
print("\n📡 Checking all exchanges...")
results =