paulpas Role: Monitor exchange API health and availability to ensure reliable trading operations
Philosophy: Exchange availability is critical; early detection of issues prevents trading failures and potential losses
Key Principles
- Multi-Metric Health Check: Combine latency, error rate, and heartbeat monitoring
- Health Score Calculation: Aggregate metrics into single health score
- Alerting Thresholds: Configurable thresholds with escalation paths
- Graceful Degradation: Reduce functionality when exchanges are unhealthy
- Circuit State Tracking: Maintain state for automatic recovery coordination
Implementation Guidelines
Structure
- Core logic: health_monitor/monitor.py
- Metrics collection: health_monitor/metrics.py
- Tests: tests/test_exchange_health.py
Patterns to Follow
- Use async health checks for parallel monitoring
- Implement sliding window for rate calculations
- Support pluggable alert backends
- Track historical health data
Adherence Checklist
Before completing your task, verify:
- Health checks run at configurable intervals
- Alerts are sent before service disruption
- Circuit breaker integration is active
- Historical health data is persisted
- Multiple alert channels are supported
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
Python Implementation
import asyncio
import time
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
from collections import deque
import logging
class HealthStatus(Enum):
UNKNOWN = "unknown"
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
@dataclass
class HealthMetric:
"""Single health metric observation."""
name: str
value: float
timestamp: float
labels: Dict[str, str] = field(default_factory=dict)
@dataclass
class ExchangeHealth:
"""Health status for a single exchange."""
exchange_id: str
status: HealthStatus = HealthStatus.UNKNOWN
last_check: float = 0
metrics: Dict[str, float] = field(default_factory=dict)
alerts: List[str] = field(default_factory=list)
latency_p50: float = 0
latency_p95: float = 0
error_rate: float = 0
heartbeat_lag: float = 0
class SlidingWindow:
"""Sliding window for time-series data."""
def __init__(self, window_seconds: float, max_samples: int = 1000):
self.window = window_seconds
self.max_samples = max_samples
self.samples: deque = deque()
self.lock = asyncio.Lock()
async def add(self, value: float, timestamp: float = None):
"""Add sample to window."""
async with self.lock:
timestamp = timestamp or time.time()
self.samples.append((timestamp, value))
# Trim old samples
cutoff = timestamp - self.window
while self.samples and self.samples[0][0] < cutoff:
self.samples.popleft()
# Trim to max size
while len(self.samples) > self.max_samples:
self.samples.popleft()
async def get(self) -> List[float]:
"""Get all samples."""
async with self.lock:
return [v for _, v in self.samples]
async def get_average(self) -> float:
"""Get average of samples."""
samples = await self.get()
return sum(samples) / len(samples) if samples else 0
async def get_percentile(self, percentile: float) -> float:
"""Get percentile of samples."""
samples = await self.get()
if not samples:
return 0
sorted_samples = sorted(samples)
index = int(len(sorted_samples) * percentile / 100)
return sorted_samples[min(index, len(sorted_samples) - 1)]
class HealthMonitor:
"""Monitors health of multiple exchanges."""
def __init__(
self,
check_interval: float = 10.0,
alert_threshold: float = 0.7, # 70% health triggers alert
unhealthy_threshold: float = 0.5, # 50% health = unhealthy
alert_cooldown: float = 300.0 # 5 minutes between alerts
):
self.check_interval = check_interval
self.alert_threshold = alert_threshold
self.unhealthy_threshold = unhealthy_threshold
self.alert_cooldown = alert_cooldown
self.exchanges: Dict[str, ExchangeHealth] = {}
self.latency_windows: Dict[str, SlidingWindow] = {}
self.error_windows: Dict[str, SlidingWindow] = {}
self.heartbeat_windows: Dict[str, SlidingWindow] = {}
self.alert_callbacks: List[Callable] = []
self.last_alert_time: float = 0
self.health_callbacks: List[Callable] = []
def register_exchange(self, exchange_id: str):
"""Register exchange for health monitoring."""
if exchange_id not in self.exchanges:
self.exchanges[exchange_id] = ExchangeHealth(exchange_id=exchange_id)
self.latency_windows[exchange_id] = SlidingWindow(60.0)
self.error_windows[exchange_id] = SlidingWindow(60.0)
self.heartbeat_windows[exchange_id] = SlidingWindow(60.0)
async def record_latency(self, exchange_id: str, latency_ms: float):
"""Record latency measurement."""
self.register_exchange(exchange_id)
await self.latency_windows[exchange_id].add(latency_ms)
health = self.exchanges[exchange_id]
health.latency_p50 = await self.latency_windows[exchange_id].get_percentile(50)
health.latency_p95 = await self.latency_windows[exchange_id].get_percentile(95)
async def record_error(self, exchange_id: str):
"""Record error occurrence."""
self.register_exchange(exchange_id)
await self.error_windows[exchange_id].add(1.0)
# Calculate error rate
samples = await self.error_windows[exchange_id].get()
if samples:
error_rate = sum(samples) / len(samples)
self.exchanges[exchange_id].error_rate = error_rate
async def record_success(self, exchange_id: str):
"""Record successful request."""
self.register_exchange(exchange_id)
await self.error_windows[exchange_id].add(0.0)
async def record_heartbeat(self, exchange_id: str, lag_ms: float):
"""Record WebSocket heartbeat lag."""
self.register_exchange(exchange_id)
await self.heartbeat_windows[exchange_id].add(lag_ms)
health = self.exchanges[exchange_id]
health.heartbeat_lag = await self.heartbeat_windows[exchange_id].get_average()
async def check_health(self, exchange_id: str) -> ExchangeHealth:
"""Perform health check and update status."""
health = self.exchanges.get(exchange_id)
if not health:
return health
health.last_check = time.time()
# Calculate health score (0-1, higher is better)
score = 1.0
# Latency penalty
if health.latency_p95 > 1000: # > 1 second
score -= 0.3
elif health.latency_p95 > 500: # > 500ms
score -= 0.1
# Error rate penalty
score -= health.error_rate * 2
# Heartbeat lag penalty
if health.heartbeat_lag > 5000: # > 5 seconds
score -= 0.2
elif health.heartbeat_lag > 1000: # > 1 second
score -= 0.05
score = max(0.0, min(1.0, score))
health.metrics["health_score"] = score
# Determine status
if score >= self.alert_threshold:
health.status = HealthStatus.HEALTHY
elif score >= self.unhealthy_threshold:
health.status = HealthStatus.DEGRADED
health.alerts.append("Exchange performance degraded")
else:
health.status = HealthStatus.UNHEALTHY
health.alerts.append("Exchange health critical")
# Check for alerts
await self._check_alerts(health)
return health
async def _check_alerts(self, health: ExchangeHealth):
"""Check if alerts should be sent."""
if health.status == HealthStatus.UNHEALTHY:
if time.time() - self.last_alert_time > self.alert_cooldown:
self.last_alert_time = time.time()
for callback in self.alert_callbacks:
try:
await callback(health)
except Exception as e:
logging.error(f"Alert callback error: {e}")
# Trigger health change callbacks
for callback in self.health_callbacks:
try:
callback(health)
except Exception as e:
logging.error(f"Health callback error: {e}")
def register_alert_callback(self, callback: Callable):
"""Register alert callback."""
self.alert_callbacks.append(callback)
def register_health_callback(self, callback: Callable):
"""Register health status change callback."""
self.health_callbacks.append(callback)
def get_exchange_status(self, exchange_id: str) -> Optional[ExchangeHealth]:
"""Get current health status for exchange."""
return self.exchanges.get(exchange_id)
def get_all_statuses(self) -> Dict[str, ExchangeHealth]:
"""Get health status for all exchanges."""
return self.exchanges.copy()
async def run_periodic_checks(self):
"""Run periodic health checks."""
while True:
for exchange_id in self.exchanges:
await self.check_health(exchange_id)
await asyncio.sleep(self.check_interval)
def is_exchange_healthy(self, exchange_id: str) -> bool:
"""Check if exchange is currently healthy."""
health = self.exchanges.get(exchange_id)
if not health:
return False
return health.status == HealthStatus.HEALTHY
class CircuitHealthIntegrator:
"""Integrates health monitoring with circuit breaker."""
def __init__(self, health_monitor: HealthMonitor, circuit_breaker):
self.health = health_monitor
self.circuit = circuit_breaker
async def on_health_change(self, health: ExchangeHealth):
"""Handle health status changes for circuit management."""
if health.status == HealthStatus.UNHEALTHY:
self.circuit.record_failure()
logging.warning(f"Circuit breaker triggered for {health.exchange_id}")
elif health.status == HealthStatus.HEALTHY:
self.circuit.record_success()
Pattern 2: Exchange Health Check with Multi-Metric Scoring
from __future__ import annotations
import logging
import time
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
logger = logging.getLogger(__name__)
class HealthStatus(str, Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
UNHEALTHY = "unhealthy"
UNKNOWN = "unknown"
@dataclass
class ExchangeHealthReport:
"""Comprehensive health report for an exchange connection."""
exchange_id: str
status: HealthStatus = HealthStatus.UNKNOWN
avg_latency_ms: float = 0.0
error_rate_1m: float = 0.0 # Errors / total requests in last minute
uptime_ratio: float = 1.0 # Fraction of time reachable in last 5 minutes
ws_connected: bool = False
order_submit_latency_ms: float = 0.0
last_check_at: float = field(default_factory=time.monotonic)
health_score: float = 100.0 # 0-100 composite score
@property
def is_tradable(self) -> bool:
return self.status in (HealthStatus.HEALTHY, HealthStatus.DEGRADED)
class ExchangeHealthMonitor:
"""Monitors exchange health using multiple metrics with composite scoring."""
def __init__(self, exchange_id: str, check_interval: float = 10.0):
self.exchange_id = exchange_id
self._interval = check_interval
self._history: list[ExchangeHealthReport] = []
self._latency_samples: list[float] = []
async def run_health_check(self) -> ExchangeHealthReport:
"""Execute a comprehensive health check against the exchange."""
start = time.monotonic()
# Measure API latency via ping
try:
await self._ping_exchange()
latency_ms = (time.monotonic() - start) * 1000
except Exception as e:
logger.error("Health check ping failed for %s: %s", self.exchange_id, e)
return ExchangeHealthReport(
exchange_id=self.exchange_id, status=HealthStatus.UNHEALTHY,
last_check_at=time.monotonic(), health_score=0.0,
)
# Measure WebSocket connectivity
ws_connected = await self._check_websocket()
# Compute composite health score (0-100)
latency_penalty = min(40, latency_ms / 5) # Lose up to 40 points for high latency
ws_penalty = 30 if not ws_connected else 0
score = max(0, 100 - latency_penalty - ws_penalty)
if score >= 80:
status = HealthStatus.HEALTHY
elif score >= 50:
status = HealthStatus.DEGRADED
else:
status = HealthStatus.UNHEALTHY
report = ExchangeHealthReport(
exchange_id=self.exchange_id,
status=status,
avg_latency_ms=round(latency_ms, 2),
ws_connected=ws_connected,
health_score=round(score, 1),
last_check_at=time.monotonic(),
)
self._history.append(report)
self._latency_samples.append(latency_ms)
if len(self._latency_samples) > 100:
self._latency_samples.pop(0)
logger.info("Health check %s: status=%s score=%.1f latency=%.0fms",
self.exchange_id, status.value, score, latency_ms)
return report
async def _ping_exchange(self) -> None:
"""Lightweight API call to measure response time."""
pass # Actual implementation varies by exchange
async def _check_websocket(self) -> bool:
"""Verify WebSocket connection is alive and receiving updates."""
return True # Placeholder — actual WS health check
Constraints
MUST DO
- Implement a unified adapter interface across all exchange integrations to standardize order placement, cancellation, and querying
- Handle rate limiting proactively with token bucket or leaky bucket algorithms — never wait for 429 responses before slowing down
- Maintain local order state as the source of truth; reconcile with exchange state periodically via webhook events and polling
- Implement heartbeat monitoring per exchange connection with automatic failover to a secondary data feed on timeout
- Log all API interactions including request/response IDs, timing, and status codes for audit and debugging
MUST NOT DO
- Do not trust exchange-reported order states without local confirmation — always reconcile after every state change
- Avoid sending multiple orders for the same position simultaneously across different adapters or sessions
- Never store API keys or secrets in code — use environment variables or a secrets manager with automatic rotation
- Do not assume all exchanges support the same order types — implement graceful degradation with clear capability negotiation
- Avoid polling-based price updates when WebSocket/streaming APIs are available — polling creates unnecessary load and latency
Live References
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.