# Exchange Health > "Provides Exchange system health monitoring and connectivity status tracking" - Skill: `paulpas/exchange-health` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/exchange-health` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/exchange-health/raw - Safety review: PASS (external: skill-scanner PASS, skillspector PASS) - Works with: Claude Code, Claude.ai, OpenAI Codex - Category: Coding & Dev Tools - License: MIT - Author: paulpas (https://skillmd.com/u/paulpas) - Updated: 2026-09-17 - Page: https://skillmd.com/skills/paulpas/exchange-health --- **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 1. **Multi-Metric Health Check**: Combine latency, error rate, and heartbeat monitoring 2. **Health Score Calculation**: Aggregate metrics into single health score 3. **Alerting Thresholds**: Configurable thresholds with escalation paths 4. **Graceful Degradation**: Reduce functionality when exchanges are unhealthy 5. **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 ```python 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 ```python 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. - [Exchange Status Tutorial](https://docs.quantconnect.com/tutorials/exchange-status) - [Exchange Uptime Monitoring](https://docs.binance.org/) - [Trading System Health Checks](https://docs.quantconnect.com/tutorials/live-trading-overview) - [Market Status Indicators](https://www.investopedia.com/terms/m/market-hours.asp) - [Exchange API Latency Monitoring](https://docs.ccxt.org/en/latest/manual.html#rate-limit)