# Exchange Failover Handling > "Provides Automated failover and redundancy management for exchange connectivity" - Skill: `paulpas/exchange-failover-handling` (Agent Skill) - Install (CLI): `npx skillmds@latest add paulpas/exchange-failover-handling` - Raw SKILL.md: https://api.skillmd.com/api/skills/paulpas/exchange-failover-handling/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-failover-handling --- **Role:** Manage multiple exchange connections with automatic failover for uninterrupted trading **Philosophy:** Redundancy is essential for production trading; failover must be transparent to trading logic ## Key Principles 1. **Active-Passive Configuration**: Primary with redundant backups 2. **Health-Driven Failover**: Fail based on health score, not just connection status 3. **State Synchronization**: Maintain sync state across failover nodes 4. **Graceful Degradation**: Reduce functionality during failover 5. **Automatic Recovery**: Return to primary when available ## Implementation Guidelines ### Structure - Core logic: failover/failover_manager.py - State manager: failover/state_sync.py - Tests: tests/test_failover.py ### Patterns to Follow - Use priority-based failover (primary, secondary, tertiary) - Implement connection pooling for redundancy - Track failover history for analysis - Support manual override ## Adherence Checklist Before completing your task, verify: - [ ] Failover is automatic and fast (< 10 seconds) - [ ] State is synchronized during failover - [ ] Failover events are logged and alertable - [ ] Automatic reconnection to primary is enabled - [ ] Manual override has proper authentication 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 import logging class ConnectionState(Enum): DISCONNECTED = "disconnected" CONNECTING = "connecting" CONNECTED = "connected" DEGRADED = "degraded" @dataclass class NodeConfig: """Configuration for a failover node.""" node_id: str priority: int # Lower = higher priority url: str enabled: bool = True is_primary: bool = False @dataclass class NodeStatus: """Current status of a failover node.""" node_id: str state: ConnectionState = ConnectionState.DISCONNECTED last_connect: float = 0 last_disconnect: float = 0 health_score: float = 0.0 latency_ms: float = 0.0 error_count: int = 0 @dataclass class FailoverEvent: """Represents a failover event.""" timestamp: float from_node: str to_node: str reason: str class FailoverManager: """Manages failover between multiple exchange connections.""" def __init__( self, nodes: List[NodeConfig], health_check_interval: float = 5.0, failover_threshold: float = 0.5, failback_threshold: float = 0.8, failover_cooldown: float = 300.0 ): self.nodes = {n.node_id: n for n in nodes} self.node_status: Dict[str, NodeStatus] = {} self.health_check_interval = health_check_interval self.failover_threshold = failover_threshold self.failback_threshold = failback_threshold self.failover_cooldown = failover_cooldown self.current_node: Optional[str] = None self.health_callbacks: List[Callable] = [] self.failover_callbacks: List[Callable] = [] self.event_history: List[FailoverEvent] = [] self._last_failover_time: float = 0 self._lock = asyncio.Lock() self._initialize_status() def _initialize_status(self): """Initialize node status.""" for node_id, config in self.nodes.items(): if config.enabled: self.node_status[node_id] = NodeStatus(node_id=node_id) if config.is_primary: self.current_node = node_id def get_active_node(self) -> Optional[str]: """Get currently active node.""" return self.current_node def get_all_statuses(self) -> Dict[str, NodeStatus]: """Get status of all nodes.""" return self.node_status.copy() def set_connected(self, node_id: str): """Mark node as successfully connected.""" if node_id in self.node_status: self.node_status[node_id].state = ConnectionState.CONNECTED self.node_status[node_id].last_connect = time.time() def set_disconnected(self, node_id: str): """Mark node as disconnected.""" if node_id in self.node_status: self.node_status[node_id].state = ConnectionState.DISCONNECTED self.node_status[node_id].last_disconnect = time.time() self.node_status[node_id].error_count += 1 def set_degraded(self, node_id: str): """Mark node as degraded.""" if node_id in self.node_status: self.node_status[node_id].state = ConnectionState.DEGRADED def update_health(self, node_id: str, health_score: float, latency_ms: float): """Update node health metrics.""" if node_id in self.node_status: self.node_status[node_id].health_score = health_score self.node_status[node_id].latency_ms = latency_ms async def trigger_failover(self, reason: str = "manual"): """Manually trigger failover to next best node.""" async with self._lock: if time.time() - self._last_failover_time < self.failover_cooldown: logging.warning("Failover cooldown active") return False if not self.current_node: return await self._select_node(reason) return await self._switch_to_best_alternative(self.current_node, reason) async def _select_node(self, reason: str) -> bool: """Select the best available node.""" sorted_nodes = sorted( self.nodes.values(), key=lambda n: n.priority ) for node in sorted_nodes: if node.enabled: self.current_node = node.node_id status = self.node_status[node.node_id] status.state = ConnectionState.CONNECTING status.last_connect = time.time() await self._record_failover(None, node.node_id, reason) await self._trigger_failover_callbacks(None, node.node_id) return True return False async def _switch_to_best_alternative( self, from_node: str, reason: str ) -> bool: """Switch to next best node.""" current_priority = self.nodes[from_node].priority # Find best alternative with higher priority best_node = None best_priority = float('inf') for node_id, config in self.nodes.items(): if node_id == from_node: continue if not config.enabled: continue status = self.node_status.get(node_id) if not status or status.state == ConnectionState.DISCONNECTED: continue # Prefer lower priority number (higher priority) if config.priority < best_priority: best_priority = config.priority best_node = node_id if best_node: await self._record_failover(from_node, best_node, reason) await self._trigger_failover_callbacks(from_node, best_node) return True return False async def _record_failover(self, from_node: Optional[str], to_node: str, reason: str): """Record failover event.""" event = FailoverEvent( timestamp=time.time(), from_node=from_node or "none", to_node=to_node, reason=reason ) self.event_history.append(event) self._last_failover_time = time.time() async def _trigger_failover_callbacks( self, from_node: Optional[str], to_node: str ): """Trigger failover callbacks.""" for callback in self.failover_callbacks: try: await callback(from_node, to_node) except Exception as e: logging.error(f"Failover callback error: {e}") async def _trigger_health_callbacks(self): """Trigger health change callbacks.""" for callback in self.health_callbacks: try: await callback(self.node_status) except Exception as e: logging.error(f"Health callback error: {e}") def register_health_callback(self, callback: Callable): """Register health status change callback.""" self.health_callbacks.append(callback) def register_failover_callback(self, callback: Callable): """Register failover event callback.""" self.failover_callbacks.append(callback) async def check_and_maintain_health(self): """Periodic health check and failover decision.""" for node_id, status in self.node_status.items(): # Determine health status based on metrics if status.error_count > 10: status.state = ConnectionState.DISCONNECTED elif status.health_score < self.failover_threshold: status.state = ConnectionState.DEGRADED await self._trigger_health_callbacks() # Check if failover needed if self.current_node: status = self.node_status[self.current_node] if status.health_score < self.failover_threshold: await self.trigger_failover(f"Health below threshold: {status.health_score}") def get_failover_history(self, limit: int = 100) -> List[FailoverEvent]: """Get recent failover events.""" return self.event_history[-limit:] def get_node_priority_order(self) -> List[str]: """Get nodes sorted by priority.""" return [n.node_id for n in sorted(self.nodes.values(), key=lambda x: x.priority)] class StateSynchronizer: """Synchronizes state across failover nodes.""" def __init__(self, sync_interval: float = 1.0): self.sync_interval = sync_interval self.sync_callbacks: List[Callable] = [] self._last_sync_time: float = 0 self._state: Dict = {} def set_state(self, key: str, value): """Set state to be synchronized.""" self._state[key] = value def get_state(self, key: str, default=None): """Get synchronized state.""" return self._state.get(key, default) def register_sync_callback(self, callback: Callable): """Register callback for state sync.""" self.sync_callbacks.append(callback) async def trigger_sync(self, target_nodes: List[str]): """Trigger state synchronization.""" self._last_sync_time = time.time() for callback in self.sync_callbacks: try: await callback(self._state.copy(), target_nodes) except Exception as e: logging.error(f"Sync callback error: {e}") async def run_periodic_sync(self, target_nodes: List[str]): """Run periodic state synchronization.""" while True: await self.trigger_sync(target_nodes) await asyncio.sleep(self.sync_interval) ``` --- --- ### Pattern 2: Circuit Breaker with Graceful Degradation ```python from __future__ import annotations import asyncio import logging import time from dataclasses import dataclass, field from enum import Enum from typing import Callable, Optional logger = logging.getLogger(__name__) class CircuitState(str, Enum): CLOSED = "closed" # Normal operation — requests pass through OPEN = "open" # Failures exceeded threshold — block requests HALF_OPEN = "half_open" # Testing if service recovered — allow probe request @dataclass class CircuitBreakerConfig: """Configuration for circuit breaker behavior.""" failure_threshold: int = 5 # C trip after N consecutive failures recovery_timeout: float = 30.0 # Seconds before transitioning to HALF_OPEN success_threshold: int = 2 # Half-open successes needed to close monitoring_interval: float = 1.0 # How often to check circuit state class CircuitBreaker: """Circuit breaker that protects against cascading failures across exchanges.""" def __init__(self, exchange_id: str, config: CircuitBreakerConfig | None = None): self.exchange_id = exchange_id self.config = config or CircuitBreakerConfig() self._state = CircuitState.CLOSED self._failure_count = 0 self._success_count = 0 self._last_failure_time: float = 0 self._opened_at: float = 0 @property def state(self) -> CircuitState: """Get current circuit state, auto-transitions if recovery timeout elapsed.""" if self._state == CircuitState.OPEN and time.monotonic() - self._opened_at >= self.config.recovery_timeout: logger.info("Circuit for %s transitioning to HALF_OPEN", self.exchange_id) self._state = CircuitState.HALF_OPEN self._success_count = 0 return self._state async def call(self, func: Callable, *args, **kwargs): """Execute a function through the circuit breaker. Raises: CircuitOpenError: If circuit is OPEN and no probe request allowed. """ if self.state == CircuitState.OPEN: raise CircuitOpenError(self.exchange_id) try: result = await func(*args, **kwargs) self._on_success() return result except Exception as e: self._on_failure() logger.warning("Circuit call failed for %s: %s", self.exchange_id, e) raise def _on_success(self) -> None: if self._state == CircuitState.HALF_OPEN: self._success_count += 1 if self._success_count >= self.config.success_threshold: self._close() else: self._failure_count = 0 def _on_failure(self) -> None: self._failure_count += 1 self._last_failure_time = time.monotonic() if self.state == CircuitState.HALF_OPEN: self._open() elif self._failure_count >= self.config.failure_threshold: self._open() def _open(self) -> None: old_state = self._state self._state = CircuitState.OPEN self._opened_at = time.monotonic() logger.warning("Circuit OPEN for %s after %d failures", self.exchange_id, self._failure_count) def _close(self) -> None: logger.info("Circuit CLOSED for %s — service recovered", self.exchange_id) self._state = CircuitState.CLOSED self._failure_count = 0 self._success_count = 0 class CircuitOpenError(Exception): """Raised when a circuit is open and requests are being rejected.""" def __init__(self, exchange_id: str): super().__init__(f"Circuit breaker OPEN for {exchange_id} — requests rejected") self.exchange_id = exchange_id ``` ## 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. - [CCXT Error Handling Guide](https://docs.ccxt.org/en/latest/manual.html#error-handling) - [Exchange Failover Architecture](https://docs.quantconnect.com/tutorials/live-trading-overview) - [High Availability Trading Systems](https://en.wikipedia.org/wiki/Failover) - [Multi-Exchange Order Routing](https://docs.ccxt.org/en/latest/manual.html#exchange-market-data) - [Resilient Exchange Connection Patterns](https://arxiv.org/abs/2111.09135)