paulpas 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
- Active-Passive Configuration: Primary with redundant backups
- Health-Driven Failover: Fail based on health score, not just connection status
- State Synchronization: Maintain sync state across failover nodes
- Graceful Degradation: Reduce functionality during failover
- 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
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
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.