exchange-health

---

name: exchange-health compatibility: opencode completeness: 95 content-types:

• code
• guidance
• config
• do-dont description: '"Provides Exchange system health monitoring and connectivity status tracking"' license: MIT maturity: stable metadata: domain: trading output-format: code related-skills: ai-order-flow-analysis, data-alternative-data role: implementation scope: implementation triggers: connectivity, exchange health, exchange-health, monitoring, system, cloudwatch, logging archetypes:
  • tactical anti_triggers:
  • brainstorming
  • vague ideation
  • no risk management response_profile: verbosity: low directive_strength: high abstraction_level: operational version: "1.0.0"

---

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

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.

• Exchange Status Tutorial
• Exchange Uptime Monitoring
• Trading System Health Checks
• Market Status Indicators
• Exchange API Latency Monitoring