one of the most annoying parts of crypto algo trading is that every exchange has a different API. different auth schemes, different rate limits, different order types, different error codes. writing strategy logic for each exchange separately is a nightmare and a maintenance disaster.
spent last week building a unified exchange connectivity layer that abstracts the differences away. now my strategies talk to one interface and don’t care if the order goes to binance, kraken, or coinbase.
why ccxt isn’t enough #
yeah i know ccxt exists and it’s great for basic stuff. i use it under the hood. but ccxt gives you a lowest-common-denominator interface - it handles the API differences but doesn’t solve:
- connection management - websocket reconnection, heartbeats, auth token refresh
- order state tracking - partial fills, rejections, amendments across exchanges
- rate limit coordination - each exchange has different limits and penalty behaviors
- failover - if binance goes down mid-order, reroute to kraken
so i built a layer on top of ccxt that handles all of this.
the architecture #
import ccxt.async_support as ccxt
import asyncio
import time
import logging
from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
from collections import defaultdict
logger = logging.getLogger(__name__)
class ExchangeStatus(Enum):
HEALTHY = "healthy"
DEGRADED = "degraded"
DOWN = "down"
RATE_LIMITED = "rate_limited"
class OrderStatus(Enum):
PENDING = "pending"
OPEN = "open"
PARTIALLY_FILLED = "partially_filled"
FILLED = "filled"
CANCELLED = "cancelled"
REJECTED = "rejected"
FAILED = "failed"
@dataclass
class ExchangeHealth:
status: ExchangeStatus
latency_ms: float
last_successful_call: float
error_count_1m: int = 0
rate_limit_remaining: int = 100
consecutive_errors: int = 0
@property
def is_usable(self) -> bool:
return self.status in (ExchangeStatus.HEALTHY, ExchangeStatus.DEGRADED)
@dataclass
class UnifiedOrder:
order_id: str
exchange: str
symbol: str
side: str # "buy" or "sell"
order_type: str # "limit", "market", "stop_limit"
quantity: float
price: Optional[float]
status: OrderStatus
filled_qty: float = 0.0
avg_fill_price: float = 0.0
created_at: float = 0.0
updated_at: float = 0.0
fees: float = 0.0
exchange_order_id: str = ""
error_msg: str = ""
metadata: dict = field(default_factory=dict)
@property
def remaining_qty(self) -> float:
return self.quantity - self.filled_qty
@property
def fill_pct(self) -> float:
if self.quantity == 0:
return 0.0
return self.filled_qty / self.quantity * 100
class RateLimiter:
"""sliding window rate limiter per exchange"""
def __init__(self, max_requests: int, window_seconds: float):
self.max_requests = max_requests
self.window = window_seconds
self._timestamps: list[float] = []
async def acquire(self):
now = time.monotonic()
# remove expired timestamps
self._timestamps = [t for t in self._timestamps if now - t < self.window]
if len(self._timestamps) >= self.max_requests:
# wait until oldest request expires
sleep_time = self._timestamps[0] + self.window - now + 0.01
if sleep_time > 0:
logger.debug(f"rate limited, sleeping {sleep_time:.2f}s")
await asyncio.sleep(sleep_time)
self._timestamps.append(time.monotonic())
@property
def remaining(self) -> int:
now = time.monotonic()
active = sum(1 for t in self._timestamps if now - t < self.window)
return max(0, self.max_requests - active)
class ExchangeConnector:
"""manages connection to a single exchange"""
# exchange-specific rate limits
RATE_LIMITS = {
"binance": {"requests": 1200, "window": 60},
"kraken": {"requests": 15, "window": 1}, # kraken is strict
"coinbase": {"requests": 10, "window": 1}, # coinbase is very strict
}
def __init__(self, exchange_id: str, config: dict):
self.exchange_id = exchange_id
self.config = config
# initialize ccxt exchange
exchange_class = getattr(ccxt, exchange_id)
self.exchange = exchange_class({
"apiKey": config.get("api_key", ""),
"secret": config.get("secret", ""),
"password": config.get("passphrase", ""),
"enableRateLimit": False, # we handle this ourselves
"options": {"defaultType": "spot"},
})
limits = self.RATE_LIMITS.get(exchange_id, {"requests": 30, "window": 1})
self.rate_limiter = RateLimiter(limits["requests"], limits["window"])
self.health = ExchangeHealth(
status=ExchangeStatus.HEALTHY,
latency_ms=0.0,
last_successful_call=time.time(),
)
self._orders: dict[str, UnifiedOrder] = {}
self._error_timestamps: list[float] = []
async def place_order(
self,
symbol: str,
side: str,
order_type: str,
quantity: float,
price: Optional[float] = None,
) -> UnifiedOrder:
await self.rate_limiter.acquire()
order = UnifiedOrder(
order_id=f"{self.exchange_id}_{int(time.time()*1000)}",
exchange=self.exchange_id,
symbol=symbol,
side=side,
order_type=order_type,
quantity=quantity,
price=price,
status=OrderStatus.PENDING,
created_at=time.time(),
)
try:
start = time.monotonic()
if order_type == "market":
result = await self.exchange.create_market_order(
symbol, side, quantity
)
elif order_type == "limit":
result = await self.exchange.create_limit_order(
symbol, side, quantity, price
)
else:
raise ValueError(f"unsupported order type: {order_type}")
latency = (time.monotonic() - start) * 1000
self.health.latency_ms = latency
self.health.last_successful_call = time.time()
self.health.consecutive_errors = 0
order.exchange_order_id = result["id"]
order.status = OrderStatus.OPEN
if result.get("filled", 0) > 0:
order.filled_qty = result["filled"]
order.avg_fill_price = result.get("average", result.get("price", 0))
if order.filled_qty >= order.quantity:
order.status = OrderStatus.FILLED
else:
order.status = OrderStatus.PARTIALLY_FILLED
order.updated_at = time.time()
self._orders[order.order_id] = order
logger.info(
f"ORDER PLACED [{self.exchange_id}] {side} {quantity} {symbol} "
f"@ {'market' if order_type == 'market' else price} "
f"-> {order.status.value} ({latency:.0f}ms)"
)
return order
except ccxt.InsufficientFunds as e:
order.status = OrderStatus.REJECTED
order.error_msg = f"insufficient funds: {e}"
self._record_error()
logger.error(f"REJECTED [{self.exchange_id}]: {e}")
return order
except ccxt.RateLimitExceeded:
order.status = OrderStatus.FAILED
order.error_msg = "rate limit exceeded"
self.health.status = ExchangeStatus.RATE_LIMITED
self._record_error()
logger.warning(f"RATE LIMITED [{self.exchange_id}]")
return order
except ccxt.NetworkError as e:
order.status = OrderStatus.FAILED
order.error_msg = f"network error: {e}"
self._record_error()
logger.error(f"NETWORK ERROR [{self.exchange_id}]: {e}")
return order
except Exception as e:
order.status = OrderStatus.FAILED
order.error_msg = str(e)
self._record_error()
logger.error(f"ORDER FAILED [{self.exchange_id}]: {e}")
return order
async def cancel_order(self, order_id: str) -> bool:
if order_id not in self._orders:
return False
order = self._orders[order_id]
await self.rate_limiter.acquire()
try:
await self.exchange.cancel_order(
order.exchange_order_id, order.symbol
)
order.status = OrderStatus.CANCELLED
order.updated_at = time.time()
return True
except Exception as e:
logger.error(f"cancel failed [{self.exchange_id}]: {e}")
return False
async def get_balance(self, asset: str) -> float:
await self.rate_limiter.acquire()
try:
balance = await self.exchange.fetch_balance()
return balance.get(asset, {}).get("free", 0.0)
except Exception as e:
logger.error(f"balance check failed [{self.exchange_id}]: {e}")
return 0.0
def _record_error(self):
now = time.time()
self._error_timestamps.append(now)
# keep last 60 seconds
self._error_timestamps = [t for t in self._error_timestamps if now - t < 60]
self.health.error_count_1m = len(self._error_timestamps)
self.health.consecutive_errors += 1
if self.health.consecutive_errors >= 5:
self.health.status = ExchangeStatus.DOWN
elif self.health.error_count_1m >= 3:
self.health.status = ExchangeStatus.DEGRADED
async def close(self):
await self.exchange.close()
class UnifiedExchangeLayer:
"""
unified interface for trading across multiple exchanges
handles routing, failover, and order tracking
"""
def __init__(self, configs: dict[str, dict]):
self.connectors: dict[str, ExchangeConnector] = {}
for exchange_id, config in configs.items():
self.connectors[exchange_id] = ExchangeConnector(exchange_id, config)
# symbol -> preferred exchange mapping
self._routing: dict[str, list[str]] = {}
self._all_orders: dict[str, UnifiedOrder] = {}
def set_routing(self, symbol: str, exchanges: list[str]):
"""set exchange preference order for a symbol"""
self._routing[symbol] = exchanges
def _get_best_exchange(self, symbol: str) -> Optional[str]:
"""pick healthiest exchange that supports this symbol"""
candidates = self._routing.get(symbol, list(self.connectors.keys()))
for exchange_id in candidates:
connector = self.connectors.get(exchange_id)
if connector and connector.health.is_usable:
return exchange_id
return None
async def place_order(
self,
symbol: str,
side: str,
order_type: str,
quantity: float,
price: Optional[float] = None,
preferred_exchange: Optional[str] = None,
) -> UnifiedOrder:
exchange_id = preferred_exchange or self._get_best_exchange(symbol)
if not exchange_id:
return UnifiedOrder(
order_id=f"failed_{int(time.time()*1000)}",
exchange="none",
symbol=symbol,
side=side,
order_type=order_type,
quantity=quantity,
price=price,
status=OrderStatus.FAILED,
error_msg="no healthy exchange available",
)
connector = self.connectors[exchange_id]
order = await connector.place_order(symbol, side, order_type, quantity, price)
# if order failed and we have fallback exchanges, try them
if order.status == OrderStatus.FAILED:
candidates = self._routing.get(symbol, [])
for fallback_id in candidates:
if fallback_id == exchange_id:
continue
fb_connector = self.connectors.get(fallback_id)
if fb_connector and fb_connector.health.is_usable:
logger.info(
f"FAILOVER: {exchange_id} -> {fallback_id} for {symbol}"
)
order = await fb_connector.place_order(
symbol, side, order_type, quantity, price
)
if order.status != OrderStatus.FAILED:
break
self._all_orders[order.order_id] = order
return order
async def get_health_report(self) -> dict:
report = {}
for exchange_id, connector in self.connectors.items():
h = connector.health
report[exchange_id] = {
"status": h.status.value,
"latency_ms": round(h.latency_ms, 1),
"errors_1m": h.error_count_1m,
"rate_limit_remaining": connector.rate_limiter.remaining,
"consecutive_errors": h.consecutive_errors,
}
return report
async def close_all(self):
for connector in self.connectors.values():
await connector.close()
how i use it #
the beauty of this abstraction is my strategies don’t give a shit which exchange executes their orders:
# strategy code - clean and exchange-agnostic
async def execute_signal(layer: UnifiedExchangeLayer, signal: dict):
if signal["direction"] == "long":
order = await layer.place_order(
symbol="BTC/USDT",
side="buy",
order_type="limit",
quantity=0.01,
price=signal["entry_price"],
)
elif signal["direction"] == "short":
order = await layer.place_order(
symbol="BTC/USDT",
side="sell",
order_type="market",
quantity=0.01,
)
if order.status in (OrderStatus.OPEN, OrderStatus.FILLED):
logger.info(f"executed on {order.exchange}: {order.order_id}")
else:
logger.error(f"execution failed: {order.error_msg}")
the layer handles routing to the best exchange, failover if primary is down, rate limiting, and order tracking. strategy just says “buy BTC” and doesn’t care about the plumbing.
exchange health monitoring #
this chart shows real latency data from my monitoring. you can see binance had a latency spike around hour 42-48 (some API maintenance thing) and kraken had an outage around hour 56. during both events my failover routing kicked in automatically and orders went to the healthy exchanges.
rate limiting differences - the kraken problem #
kraken’s rate limiting is the most aggressive of the three. 15 requests per second sounds fine until you realize:
- fetching balances = 1 request
- placing an order = 1 request
- checking order status = 1 request
- getting current price = 1 request
that’s 4 requests just for a single trade cycle. at 15/s you can do maybe 3-4 trade cycles per second before getting rate limited. for my mean reversion strategy that needs fast execution, kraken is the bottleneck.
solution: use kraken for slower strategies (holding periods >1hr) and route fast strategies to binance (1200 req/min is much more generous).
the routing table handles this automatically:
layer.set_routing("BTC/USDT", ["binance", "kraken", "coinbase"]) # fast strats
layer.set_routing("ETH/USDT", ["binance", "coinbase", "kraken"])
layer.set_routing("SOL/USDT", ["binance", "coinbase"]) # kraken doesn't have good SOL liquidity
error rate by exchange #
look at kraken’s rate limit errors. SMH. that’s with my rate limiter in place too - imagine what it’d be like without it. i’ve been reading through some NexusFi exchange comparison threads to see how other algo traders handle this. consensus seems to be kraken’s API is powerful but punishing.
lessons learned #
-
don’t trust exchange status pages - binance can say “operational” while their API is returning 500s. monitor it yourself.
-
connection pooling matters more than you think - reusing TCP connections saved ~30ms per request. that adds up fast.
-
failover needs to be automatic - by the time you notice an exchange is down and manually reroute, you’ve missed your entry. automate it.
-
rate limit headroom - never run at >70% of rate limit capacity. leave room for burst operations (like closing multiple positions during a stop-loss cascade).
-
log everything - every order, every fill, every error. when something goes wrong at 3am and you’re trying to figure out what happened, logs are the only thing that matters.
the unified layer has been running for 9 days now. handled 2 exchange outages, 47 rate limit events, and ~1,200 orders without any missed executions. not bad for v1.
-AK