Cross-Exchange Crypto Arbitrage: Exploit Price Differences Across Venues
Price discrepancies across centralized exchanges create arbitrage opportunities. This guide covers building systems to identify and execute profitable trades across multiple venues.Understanding Cross-Exchange Arbitrage
Different exchanges maintain different orderbooks for the same trading pair. Price differences arise from:- Geographic separation and different local demand
- Trading volumes and market maker activity
- Network effects and regional preferences
- Time zone differences
Building a Multi-Exchange Price Aggregator
``python
import ccxt
import asyncio
import pandas as pd
import numpy as np
from datetime import datetime
import logging
from typing import Dict, List, Tuple
import aiohttp
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class MultiExchangePriceAggregator:
"""Monitor prices across multiple exchanges in real-time"""
def __init__(self, exchange_configs: Dict[str, Dict]):
"""
Initialize aggregator with multiple exchanges
Args:
exchange_configs: Dict with exchange names and API credentials
"""
self.exchanges = {}
self.price_history = {}
self.bid_ask_spreads = {}
for exchange_name, config in exchange_configs.items():
try:
exchange_class = getattr(ccxt, exchange_name)
self.exchanges[exchange_name] = exchange_class(config)
self.price_history[exchange_name] = []
logger.info(f"Initialized {exchange_name}")
except Exception as e:
logger.error(f"Failed to initialize {exchange_name}: {e}")
async def fetch_multi_exchange_prices(self, symbol: str,
timeframe: str = '1h') -> Dict[str, float]:
"""Fetch prices from all exchanges simultaneously"""
prices = {}
tasks = []
for exchange_name, exchange in self.exchanges.items():
tasks.append(self._fetch_exchange_price(exchange_name, exchange, symbol))
results = await asyncio.gather(*tasks, return_exceptions=True)
for exchange_name, price in zip(self.exchanges.keys(), results):
if isinstance(price, Exception):
logger.warning(f"Error fetching from {exchange_name}: {price}")
else:
prices[exchange_name] = price
return prices
async def _fetch_exchange_price(self, exchange_name: str,
exchange: ccxt.Exchange,
symbol: str) -> float:
"""Fetch price from single exchange"""
try:
ticker = await exchange.fetch_ticker(symbol)
return ticker['last']
except Exception as e:
logger.warning(f"Failed to fetch {symbol} from {exchange_name}: {e}")
return None
async def fetch_bid_ask_data(self, symbol: str) -> Dict[str, Dict]:
"""Fetch best bid/ask prices from all exchanges"""
bid_ask = {}
tasks = []
for exchange_name, exchange in self.exchanges.items():
tasks.append(self._fetch_exchange_bid_ask(exchange_name, exchange, symbol))
results = await asyncio.gather(*tasks, return_exceptions=True)
for exchange_name, data in zip(self.exchanges.keys(), results):
if isinstance(data, Exception):
logger.warning(f"Error fetching bid/ask from {exchange_name}: {data}")
else:
bid_ask[exchange_name] = data
return bid_ask
async def _fetch_exchange_bid_ask(self, exchange_name: str,
exchange: ccxt.Exchange,
symbol: str) -> Dict:
"""Fetch bid/ask from single exchange"""
try:
orderbook = await exchange.fetch_order_book(symbol, limit=1)
return {
'bid': orderbook['bids'][0][0] if orderbook['bids'] else None,
'ask': orderbook['asks'][0][0] if orderbook['asks'] else None,
'bid_size': orderbook['bids'][0][1] if orderbook['bids'] else 0,
'ask_size': orderbook['asks'][0][1] if orderbook['asks'] else 0,
'timestamp': datetime.now()
}
except Exception as e:
logger.warning(f"Failed to fetch orderbook from {exchange_name}: {e}")
return None
def calculate_price_spread(self, exchange_prices: Dict[str, float]) -> Dict:
"""Calculate spread between highest and lowest prices"""
valid_prices = {k: v for k, v in exchange_prices.items() if v is not None}
if not valid_prices or len(valid_prices) < 2:
return None
highest_price = max(valid_prices.values())
lowest_price = min(valid_prices.values())
spread = (highest_price - lowest_price) / lowest_price
return {
'lowest_exchange': [k for k, v in valid_prices.items() if v == lowest_price][0],
'lowest_price': lowest_price,
'highest_exchange': [k for k, v in valid_prices.items() if v == highest_price][0],
'highest_price': highest_price,
'spread_percent': spread * 100,
'spread_absolute': highest_price - lowest_price,
'timestamp': datetime.now()
}
`
Detecting Profitable Arbitrage Opportunities
`python
class ArbitrageOpportunityDetector:
"""Identify profitable arbitrage opportunities"""
def __init__(self, aggregator: MultiExchangePriceAggregator,
min_spread: float = 0.005, transaction_costs: float = 0.001):
"""
Initialize detector
Args:
aggregator: MultiExchangePriceAggregator instance
min_spread: Minimum spread to consider (0.5%)
transaction_costs: Total transaction costs (0.1% per side)
"""
self.aggregator = aggregator
self.min_spread = min_spread
self.transaction_costs = transaction_costs
self.opportunity_history = []
async def find_arbitrage_opportunities(self, symbol: str) -> List[Dict]:
"""Find all current arbitrage opportunities"""
opportunities = []
# Check simple two-exchange arbitrage
opportunities.extend(await self._find_two_exchange_arbs(symbol))
# Check triangle arbitrage
opportunities.extend(await self._find_triangle_arbs(symbol))
# Sort by profit potential
opportunities = sorted(opportunities,
key=lambda x: x['net_profit_percent'],
reverse=True)
return opportunities
async def _find_two_exchange_arbs(self, symbol: str) -> List[Dict]:
"""Find simple buy-low-sell-high opportunities"""
prices = await self.aggregator.fetch_multi_exchange_prices(symbol)
bid_ask = await self.aggregator.fetch_bid_ask_data(symbol)
if not prices or len(prices) < 2:
return []
opportunities = []
for exchange_buy in prices.keys():
for exchange_sell in prices.keys():
if exchange_buy == exchange_sell:
continue
buy_exchange = exchange_buy
sell_exchange = exchange_sell
# Get actual prices from bid/ask data
if exchange_buy in bid_ask and bid_ask[exchange_buy]:
buy_price = bid_ask[exchange_buy]['ask']
else:
buy_price = prices[exchange_buy]
if exchange_sell in bid_ask and bid_ask[exchange_sell]:
sell_price = bid_ask[exchange_sell]['bid']
else:
sell_price = prices[exchange_sell]
if not buy_price or not sell_price:
continue
spread = (sell_price - buy_price) / buy_price
# Calculate net profit after costs
gross_profit = spread * 100
net_profit = gross_profit - (self.transaction_costs 2 100)
if net_profit > self.min_spread * 100:
opportunity = {
'type': 'two_exchange',
'symbol': symbol,
'buy_exchange': buy_exchange,
'buy_price': buy_price,
'sell_exchange': sell_exchange,
'sell_price': sell_price,
'gross_spread_percent': spread * 100,
'transaction_costs_percent': self.transaction_costs 2 100,
'net_profit_percent': net_profit,
'timestamp': datetime.now()
}
opportunities.append(opportunity)
logger.info(f"Found arb: {buy_exchange} -> {sell_exchange} "
f"{symbol} net profit: {net_profit:.3f}%")
return opportunities
async def _find_triangle_arbs(self, symbol: str) -> List[Dict]:
"""Find triangle arbitrage (e.g., BTC-USDT-ETH-BTC)"""
# Implementation for more complex triangle arbitrage
return []
def estimate_profit_on_size(self, opportunity: Dict, trade_size: float,
price: float) -> float:
"""Estimate absolute profit on trade size"""
position_value = trade_size * price
profit_percent = opportunity['net_profit_percent'] / 100
return position_value * profit_percent
`
Executing Cross-Exchange Arbitrage
`python
class CrossExchangeArbitrageExecutor:
"""Execute arbitrage trades across exchanges"""
def __init__(self, exchanges: Dict[str, ccxt.Exchange],
max_slippage: float = 0.002):
"""
Initialize executor
Args:
exchanges: Dict of exchange_name -> exchange instance
max_slippage: Maximum acceptable slippage (0.2%)
"""
self.exchanges = exchanges
self.max_slippage = max_slippage
self.executed_trades = []
self.execution_log = []
async def execute_arbitrage(self, opportunity: Dict, trade_size: float) -> Dict:
"""Execute two-exchange arbitrage trade"""
logger.info(f"Executing arbitrage: {trade_size} {opportunity['symbol']} "
f"from {opportunity['buy_exchange']} to {opportunity['sell_exchange']}")
try:
# Step 1: Place buy order on first exchange
buy_result = await self._execute_buy(opportunity['buy_exchange'],
opportunity['symbol'],
trade_size,
opportunity['buy_price'])
if not buy_result['success']:
logger.error(f"Buy failed on {opportunity['buy_exchange']}")
return {'success': False, 'error': 'Buy failed'}
# Step 2: Place sell order on second exchange
sell_result = await self._execute_sell(opportunity['sell_exchange'],
opportunity['symbol'],
trade_size,
opportunity['sell_price'])
if not sell_result['success']:
logger.error(f"Sell failed on {opportunity['sell_exchange']}")
# Need to unwind buy position
await self._unwind_position(opportunity['buy_exchange'],
opportunity['symbol'],
trade_size)
return {'success': False, 'error': 'Sell failed'}
# Calculate actual profit
actual_profit = (sell_result['price'] - buy_result['price']) * trade_size
actual_profit_percent = ((sell_result['price'] - buy_result['price']) /
buy_result['price']) * 100
execution = {
'timestamp': datetime.now(),
'symbol': opportunity['symbol'],
'buy_exchange': opportunity['buy_exchange'],
'buy_price': buy_result['price'],
'sell_exchange': opportunity['sell_exchange'],
'sell_price': sell_result['price'],
'trade_size': trade_size,
'gross_profit': actual_profit,
'profit_percent': actual_profit_percent,
'buy_order_id': buy_result['order_id'],
'sell_order_id': sell_result['order_id']
}
self.executed_trades.append(execution)
logger.info(f"Arbitrage executed successfully: ${actual_profit:.2f} "
f"({actual_profit_percent:.3f}%)")
return {'success': True, 'execution': execution}
except Exception as e:
logger.error(f"Arbitrage execution failed: {e}")
return {'success': False, 'error': str(e)}
async def _execute_buy(self, exchange_name: str, symbol: str,
amount: float, limit_price: float) -> Dict:
"""Execute buy order on exchange"""
try:
exchange = self.exchanges[exchange_name]
# Use limit order slightly above market to ensure execution
order_price = limit_price * (1 + 0.0001) # 0.01% above market
order = await exchange.create_limit_buy_order(symbol, amount, order_price)
# Check for slippage
actual_price = order.get('average', order_price)
slippage = (actual_price - limit_price) / limit_price
if slippage > self.max_slippage:
logger.warning(f"High slippage on {exchange_name}: {slippage*100:.2f}%")
return {
'success': True,
'order_id': order['id'],
'price': actual_price,
'amount': amount
}
except Exception as e:
logger.error(f"Failed to execute buy on {exchange_name}: {e}")
return {'success': False, 'error': str(e)}
async def _execute_sell(self, exchange_name: str, symbol: str,
amount: float, limit_price: float) -> Dict:
"""Execute sell order on exchange"""
try:
exchange = self.exchanges[exchange_name]
# Use limit order slightly below market
order_price = limit_price * (1 - 0.0001)
order = await exchange.create_limit_sell_order(symbol, amount, order_price)
actual_price = order.get('average', order_price)
return {
'success': True,
'order_id': order['id'],
'price': actual_price,
'amount': amount
}
except Exception as e:
logger.error(f"Failed to execute sell on {exchange_name}: {e}")
return {'success': False, 'error': str(e)}
async def _unwind_position(self, exchange_name: str, symbol: str,
amount: float):
"""Close position if arbitrage fails"""
try:
exchange = self.exchanges[exchange_name]
order = await exchange.create_market_sell_order(symbol, amount)
logger.warning(f"Unwound position at market price on {exchange_name}")
except Exception as e:
logger.error(f"Failed to unwind position: {e}")
`
Risk Management in Cross-Exchange Arbitrage
`python
class CrossExchangeRiskManager:
"""Manage risks specific to cross-exchange trading"""
def __init__(self, max_position_value: float = 10000,
max_execution_time: int = 300):
"""
Initialize risk manager
Args:
max_position_value: Max value per trade ($)
max_execution_time: Max time to execute both sides (seconds)
"""
self.max_position_value = max_position_value
self.max_execution_time = max_execution_time
self.exchange_balance_limits = {}
self.rate_limit_status = {}
def is_arb_tradeable(self, opportunity: Dict) -> bool:
"""Check if arbitrage meets all safety criteria"""
# Check if profit exceeds minimum
if opportunity['net_profit_percent'] < 0.3:
return False
# Check exchange health
if not self._check_exchange_health(opportunity['buy_exchange']):
return False
if not self._check_exchange_health(opportunity['sell_exchange']):
return False
# Check rate limits
if self._check_rate_limits(opportunity['buy_exchange']):
return False
if self._check_rate_limits(opportunity['sell_exchange']):
return False
return True
def _check_exchange_health(self, exchange_name: str) -> bool:
"""Check if exchange is healthy"""
# Would check actual exchange status
return True
def _check_rate_limits(self, exchange_name: str) -> bool:
"""Check if rate limits are exhausted"""
# Would check actual rate limit status
return False
def calculate_max_trade_size(self, opportunity: Dict,
available_balance: float) -> float:
"""Calculate position size respecting limits"""
# Balance limit
max_by_balance = min(available_balance, self.max_position_value)
# Exchange limit
buy_exchange_limit = self.exchange_balance_limits.get(
opportunity['buy_exchange'], float('inf')
)
return min(max_by_balance, buy_exchange_limit)
`
Monitoring and Alerting System
`python
class ArbitrageMonitor:
"""Continuously monitor and execute arbitrage opportunities"""
def __init__(self, detector: ArbitrageOpportunityDetector,
executor: CrossExchangeArbitrageExecutor,
min_profit_threshold: float = 0.003):
self.detector = detector
self.executor = executor
self.min_profit_threshold = min_profit_threshold
self.running = False
async def start_monitoring(self, symbol: str, check_interval: int = 5):
"""Start monitoring loop"""
self.running = True
logger.info(f"Starting arbitrage monitor for {symbol}")
while self.running:
try:
# Find opportunities
opportunities = await self.detector.find_arbitrage_opportunities(symbol)
# Filter by minimum profit
tradeable_opps = [o for o in opportunities
if o['net_profit_percent'] > self.min_profit_threshold * 100]
if tradeable_opps:
# Execute best opportunity
best_opp = tradeable_opps[0]
logger.info(f"Best opportunity: {best_opp['net_profit_percent']:.3f}% profit")
# Execute with conservative size
trade_size = 0.1 # Start small
result = await self.executor.execute_arbitrage(best_opp, trade_size)
if result['success']:
logger.info(f"Trade executed: {result['execution']}")
await asyncio.sleep(check_interval)
except Exception as e:
logger.error(f"Monitor error: {e}")
await asyncio.sleep(check_interval)
def stop_monitoring(self):
"""Stop monitoring loop"""
self.running = False
``