Crypto Exchange Integration

Crypto Exchange Integration

Overview

Exchange integration involves three distinct concerns: REST (order management, account state), WebSocket (real-time data, order updates), and error handling (rate limits, disconnections, rejections). Treat each separately.

When to Use

• Implementing CCXT-based exchange connectors
• Building WebSocket data feeds with reconnection
• Managing rate limits across multiple API calls
• Handling exchange-specific order types and quirks
• Implementing order lifecycle tracking (new → filled → closed)

CCXT Unified Pattern

import ccxt.async_support as ccxt
import asyncio
from typing import Optional

class ExchangeConnector:
    def __init__(self, exchange_id: str, api_key: str, secret: str,
                 sandbox: bool = True):
        exchange_class = getattr(ccxt, exchange_id)
        self.exchange = exchange_class({
            'apiKey': api_key,
            'secret': secret,
            'sandbox': sandbox,
            'enableRateLimit': True,   # REQUIRED — uses CCXT's built-in limiter
            'options': {
                'defaultType': 'future',  # 'spot' | 'future' | 'margin'
                'adjustForTimeDifference': True,
            },
        })

    async def safe_fetch_ohlcv(self, symbol: str, timeframe: str = '1h',
                                limit: int = 500) -> list:
        """Fetch OHLCV with retry on rate limit."""
        for attempt in range(3):
            try:
                return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
            except ccxt.RateLimitExceeded:
                wait = 2 ** attempt
                await asyncio.sleep(wait)
            except ccxt.NetworkError as e:
                if attempt == 2:
                    raise
                await asyncio.sleep(1)
        return []

    async def place_order(self, symbol: str, side: str, order_type: str,
                          amount: float, price: Optional[float] = None,
                          params: dict = None) -> dict:
        """Place order with exchange-normalised error handling."""
        params = params or {}
        try:
            if order_type == 'market':
                return await self.exchange.create_market_order(symbol, side, amount, params)
            elif order_type == 'limit':
                return await self.exchange.create_limit_order(symbol, side, amount, price, params)
        except ccxt.InsufficientFunds as e:
            raise ValueError(f"Insufficient funds for {side} {amount} {symbol}") from e
        except ccxt.InvalidOrder as e:
            raise ValueError(f"Invalid order params: {e}") from e
        except ccxt.ExchangeError as e:
            # Log and re-raise — do not silently swallow
            raise

    async def close(self):
        await self.exchange.close()

WebSocket Feed with Reconnection

import websockets
import json
import asyncio
from datetime import datetime

class BinanceWebSocketFeed:
    """Robust WebSocket with exponential backoff reconnection."""

    def __init__(self, symbols: list[str], on_tick_callback):
        self.symbols = symbols
        self.on_tick = on_tick_callback
        self.running = False
        self._reconnect_delay = 1

    def _build_ws_url(self) -> str:
        streams = '/'.join(f"{s.lower()}@trade" for s in self.symbols)
        return f"wss://stream.binance.com:9443/stream?streams={streams}"

    async def start(self):
        self.running = True
        while self.running:
            try:
                async with websockets.connect(
                    self._build_ws_url(),
                    ping_interval=20,
                    ping_timeout=10,
                    close_timeout=5,
                ) as ws:
                    self._reconnect_delay = 1  # reset on successful connect
                    async for message in ws:
                        data = json.loads(message)
                        await self.on_tick(data['data'])
            except (websockets.ConnectionClosed,
                    websockets.WebSocketException,
                    ConnectionRefusedError) as e:
                if not self.running:
                    break
                # Exponential backoff, capped at 60s
                await asyncio.sleep(min(self._reconnect_delay, 60))
                self._reconnect_delay = min(self._reconnect_delay * 2, 60)

    def stop(self):
        self.running = False

Rate Limit Management

import time
from collections import deque

class RateLimiter:
    """Token bucket rate limiter for exchange APIs."""

    def __init__(self, max_calls: int, period_seconds: float):
        self.max_calls = max_calls
        self.period = period_seconds
        self.calls = deque()

    def wait(self):
        """Block until a request slot is available."""
        now = time.monotonic()
        # Remove calls outside the window
        while self.calls and self.calls[0] <= now - self.period:
            self.calls.popleft()

        if len(self.calls) >= self.max_calls:
            sleep_time = self.period - (now - self.calls[0])
            if sleep_time > 0:
                time.sleep(sleep_time)

        self.calls.append(time.monotonic())

# Exchange rate limits (requests per minute)
RATE_LIMITS = {
    'binance_spot':   {'max_calls': 1200, 'period': 60},
    'binance_future': {'max_calls': 2400, 'period': 60},
    'kraken':         {'max_calls': 60,   'period': 60},
}

Exchange-Specific Quirks

Binance Futures

# Binance requires separate endpoint for futures
exchange = ccxt.binance({
    'options': {'defaultType': 'future'},
})
# Set leverage before placing any order
await exchange.fapiPrivate_post_leverage({'symbol': 'BTCUSDT', 'leverage': 5})

# Position mode: one-way (default) or hedge
# Check with:
pos_mode = await exchange.fapiPrivateGetPositionSideDual()

# Binance timestamps must be within 5000ms of server time
# Use: exchange.options['adjustForTimeDifference'] = True

Kraken

# Kraken uses asset pairs with 'X' prefix for crypto, 'Z' for fiat
# BTC/USD = 'XBT/USD' on Kraken (not 'BTC/USD')
symbol = exchange.market_id('BTC/USD')  # -> 'XXBTZUSD'

# Nonce must be strictly increasing
# CCXT handles this automatically but watch for clock skew

# Kraken rate limiting: 'tier' based
# Verified accounts: 15 calls/s for private endpoints

IBKR (Interactive Brokers via ib_async)

Minimal IBKR connectivity example only — full equity order procedures (equity order types, partial-fill reconciliation, SSR/LULD handling, flatten-all) live in the equity-broker-execution skill. ib_async is the ACTIVE third-party successor to the archived ib_insync, wrapping the socket-based TWS API (it is not an official IBKR SDK).

from ib_async import IB, MarketOrder, LimitOrder

async def ibkr_place_order(contract, side: str, qty: float,
                           order_type: str = 'MKT', price: float = None):
    ib = IB()
    await ib.connectAsync('127.0.0.1', 7497, clientId=1)

    if order_type == 'MKT':
        order = MarketOrder(side, qty)
    else:
        if price is None:
            raise ValueError("limit order requires a price")
        order = LimitOrder(side, qty, price)

    trade = ib.placeOrder(contract, order)
    await ib.waitOnUpdateAsync()
    return trade.orderStatus.status

Order Lifecycle Tracking

from enum import Enum

class OrderStatus(Enum):
    PENDING   = 'pending'
    OPEN      = 'open'
    PARTIALLY_FILLED = 'partially_filled'
    FILLED    = 'filled'
    CANCELLED = 'cancelled'
    REJECTED  = 'rejected'

async def track_order(exchange, order_id: str, symbol: str,
                       timeout_s: float = 30) -> dict:
    """Poll order until terminal state or timeout."""
    start = asyncio.get_event_loop().time()
    while True:
        order = await exchange.fetch_order(order_id, symbol)
        status = order['status']

        if status in ('closed', 'canceled', 'rejected', 'expired'):
            return order

        if asyncio.get_event_loop().time() - start > timeout_s:
            raise TimeoutError(f"Order {order_id} not terminal after {timeout_s}s: {status}")

        await asyncio.sleep(0.5)

Boot Reconciliation — the crypto implementation of the runtime protocol

trading-automation-runtime defines ONE asset-neutral ReconciliationRequest → ReconciliationResult protocol (open orders, executions/fills, positions, balances; scope; snapshot watermark; pagination completeness; discrepancy list; explicit complete | incomplete | unknown). Crypto has no single boot-reconciliation surface — CCXT capabilities vary by exchange, so this adapter must check capabilities, never assume them, and translate the gaps into an honest incomplete / unknown status.

Per-exchange capability checks (read exchange.has, do not assume)

def reconciliation_capabilities(exchange):
    """CCXT capability flags vary by venue — inspect them; never assume parity."""
    has = exchange.has
    return {
        "open_orders":     bool(has.get("fetchOpenOrders")),
        "closed_orders":   bool(has.get("fetchClosedOrders") or has.get("fetchOrders")),
        "my_trades":       bool(has.get("fetchMyTrades")),      # fills / executions
        "positions":       bool(has.get("fetchPositions")),     # derivatives only on many
        "balance":         bool(has.get("fetchBalance")),
        "since_pagination": bool(has.get("fetchMyTrades")),     # time-window paginate
    }

• Composite execution keys. A crypto fill is keyed by (exchange, symbol, trade_id) and linked to its order by order id + client_order_id where the venue supports it — never by price/qty heuristics. Some venues recycle or omit ids; record what you actually received rather than what you expected.
• Pagination / completeness semantics. fetchMyTrades / fetchClosedOrders paginate by since + limit (a cursor on some venues). A reconciliation is complete only when the pages drain to the snapshot watermark; a truncated page, a rate-limit abort mid-pagination, or a venue that caps history is incomplete.
• What incomplete means per venue (capability variance). If an exchange cannot enumerate open orders, lacks a positions endpoint (spot-only), or caps trade history below the lookback the runtime needs, the result is incomplete (or unknown when even the capability is unclear) — which blocks new exposure at the runtime admission gate. Never paper over a missing capability as an empty-but-complete result.

Common Mistakes

1. Not enabling enableRateLimit — CCXT has a built-in limiter; always enable it
2. Swallowing exceptions — log every exchange error; silent failures cause position mismatches
3. Testing on mainnet first — always use sandbox/testnet; Binance Testnet: testnet.binance.vision
4. WebSocket without ping — connections drop silently without heartbeats; set ping_interval
5. Ignoring symbol normalisation — Kraken XBT/USD ≠ Binance BTC/USDT; use exchange.market()
6. Sync calls in async context — use ccxt.async_support, not ccxt in asyncio apps
7. Assuming immediate fill — always track order lifecycle; partial fills are common

Anti-Patterns