What Cross-Exchange Statistical Arbitrage Actually Is
Statistical arbitrage — stat arb for short — is one of the oldest systematic trading strategies in existence. Renaissance Technologies built a multi-billion dollar empire on it. Knowing how to build a statistical arbitrage bot in Python puts you in the same conceptual territory, just with crypto pairs and REST APIs instead of equities and Bloomberg terminals.
The core idea: find two assets whose prices move together over time. When they temporarily diverge beyond a statistically meaningful threshold, bet on convergence. Long the underperformer, short the outperformer, wait for prices to snap back, collect the spread.
Cross-exchange stat arb adds another layer. You're not just trading BTC/ETH on Binance — you're trading the same asset across Binance and Bybit, or trading highly correlated assets where the relationship is enforced by capital flows between venues. Price discrepancies between exchanges are often short-lived, but the correlation coefficient between pairs can remain stable for months, giving you something to systematically exploit.
This guide walks through the entire build: data collection, cointegration testing, signal generation, order execution, and risk controls. It's not a toy script. It's a working framework you can extend.
Prerequisites: Python 3.11+, familiarity with pandas and numpy, at least one exchange API key, and an understanding of what mean reversion trading means in practice.
Step 1: Set Up Your Environment and Data Pipeline
Start with a clean virtual environment and install the core dependencies:
pip install ccxt pandas numpy statsmodels scipy matplotlib python-dotenv asyncio aiohttp
ccxt is the workhorse here — it gives you a unified interface to 100+ exchanges. You'll use it to pull OHLCV data and submit orders without writing exchange-specific REST clients from scratch.
Create your project structure:
stat_arb_bot/
├── config/
│ ├── .env
│ └── pairs.yaml
├── data/
│ ├── fetcher.py
│ └── storage.py
├── analysis/
│ ├── cointegration.py
│ └── signals.py
├── execution/
│ ├── orders.py
│ └── risk.py
├── main.py
└── backtest.py
Your .env file holds API credentials — never hardcode these:
BINANCE_API_KEY=your_key_here
BINANCE_SECRET=your_secret_here
BYBIT_API_KEY=your_key_here
BYBIT_SECRET=your_secret_here
Now build the data fetcher. The goal is pulling synchronized OHLCV data from two exchanges simultaneously:
import ccxt
import pandas as pd
import asyncio
from datetime import datetime, timedelta
import os
from dotenv import load_dotenv
load_dotenv()
class MultiExchangeFetcher:
def __init__(self):
self.exchanges = {
'binance': ccxt.binance({
'apiKey': os.getenv('BINANCE_API_KEY'),
'secret': os.getenv('BINANCE_SECRET'),
'enableRateLimit': True,
}),
'bybit': ccxt.bybit({
'apiKey': os.getenv('BYBIT_API_KEY'),
'secret': os.getenv('BYBIT_SECRET'),
'enableRateLimit': True,
})
}
def fetch_ohlcv(self, exchange_id: str, symbol: str,
timeframe: str = '1h', limit: int = 500) -> pd.DataFrame:
exchange = self.exchanges[exchange_id]
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
def fetch_synchronized_prices(self, symbol: str,
timeframe: str = '1h', limit: int = 500) -> pd.DataFrame:
"""Fetch close prices from both exchanges, aligned on timestamp."""
dfs = {}
for ex_id in self.exchanges:
df = self.fetch_ohlcv(ex_id, symbol, timeframe, limit)
dfs[ex_id] = df['close'].rename(ex_id)
combined = pd.concat(dfs.values(), axis=1).dropna()
return combined
One thing most tutorials skip: timestamp alignment matters enormously. A 5-second mismatch between exchange data can make a cointegrated pair look non-stationary. Always use the dropna() call after aligning on index.
Step 2: Cointegration Testing — the Real Foundation
Correlation is not cointegration. This is where most retail stat arb attempts fail. Two assets can be 90% correlated on a rolling 30-day basis and still have a spread that random-walks to infinity. Cointegration means the spread is stationary — it has a stable long-run equilibrium it reverts to.
Think of it like two dogs on a leash. They might wander apart, but the leash always pulls them back. High correlation just means they're running in the same direction. Cointegration means the leash exists.
Use the Engle-Granger two-step test:
import numpy as np
from statsmodels.tsa.stattools import coint, adfuller
from statsmodels.regression.linear_model import OLS
from statsmodels.tools import add_constant
import pandas as pd
class CointegrationAnalyzer:
def test_cointegration(self, series_a: pd.Series, series_b: pd.Series,
significance: float = 0.05) -> dict:
"""
Run Engle-Granger cointegration test.
Returns test stats, p-value, and hedge ratio.
"""
# Run cointegration test
score, pvalue, critical_values = coint(series_a, series_b)
# Estimate hedge ratio via OLS regression
X = add_constant(series_b)
model = OLS(series_a, X).fit()
hedge_ratio = model.params[series_b.name]
# Compute spread
spread = series_a - hedge_ratio * series_b
# ADF test on spread to confirm stationarity
adf_result = adfuller(spread, autolag='AIC')
return {
'coint_pvalue': pvalue,
'is_cointegrated': pvalue < significance,
'hedge_ratio': hedge_ratio,
'spread': spread,
'adf_pvalue': adf_result[1],
'adf_stationary': adf_result[1] < significance,
'critical_values': critical_values
}
def scan_pairs(self, price_df: pd.DataFrame,
significance: float = 0.05) -> pd.DataFrame:
"""Scan all asset combinations for cointegrated pairs."""
cols = price_df.columns
results = []
for i in range(len(cols)):
for j in range(i + 1, len(cols)):
asset_a, asset_b = cols[i], cols[j]
result = self.test_cointegration(
price_df[asset_a],
price_df[asset_b],
significance
)
result['pair'] = f"{asset_a}/{asset_b}"
result['asset_a'] = asset_a
result['asset_b'] = asset_b
results.append(result)
return pd.DataFrame(results).sort_values('coint_pvalue')
The hedge ratio output from the OLS regression tells you how many units of asset B to hold for every unit of asset A. This is your position sizing anchor. Get it wrong and your "market neutral" position bleeds directional exposure.
Important: Re-run this test every 24–48 hours. Pairs that were cointegrated last month may not be today. Regime changes, correlation risk, and structural breaks in markets will kill a static pairs selection.
Step 3: Z-Score Signal Generation
Once you have a cointegrated pair and a spread, you need to know when the spread is "wide enough" to trade. The z-score does this: it normalizes the current spread deviation relative to its historical mean and standard deviation.
class SignalGenerator:
def __init__(self, entry_threshold: float = 2.0,
exit_threshold: float = 0.5,
lookback: int = 60):
self.entry_threshold = entry_threshold
self.exit_threshold = exit_threshold
self.lookback = lookback # periods for rolling stats
def compute_zscore(self, spread: pd.Series) -> pd.Series:
"""Rolling z-score of the spread."""
rolling_mean = spread.rolling(window=self.lookback).mean()
rolling_std = spread.rolling(window=self.lookback).std()
zscore = (spread - rolling_mean) / rolling_std
return zscore
def generate_signals(self, spread: pd.Series) -> pd.DataFrame:
zscore = self.compute_zscore(spread)
signals = pd.DataFrame(index=spread.index)
signals['spread'] = spread
signals['zscore'] = zscore
signals['position'] = 0
# Entry signals
signals.loc[zscore > self.entry_threshold, 'position'] = -1 # Short spread
signals.loc[zscore < -self.entry_threshold, 'position'] = 1 # Long spread
# Exit signals (mean reversion complete)
signals.loc[abs(zscore) < self.exit_threshold, 'position'] = 0
# Forward-fill to hold positions
signals['position'] = signals['position'].replace(0, np.nan)
signals['position'] = signals['position'].ffill().fillna(0)
# But override exits with explicit 0
signals.loc[abs(zscore) < self.exit_threshold, 'position'] = 0
return signals
Most guides set the entry threshold at exactly 2.0 standard deviations, which is fine in theory. In practice, crypto spreads can gap through 2.0 fast during volatile sessions. I've seen traders get better results with 2.2–2.5 for entries and 0.25–0.5 for exits, reducing trade frequency but improving per-trade quality. The right number depends entirely on your pair's historical spread behavior — which is why backtesting isn't optional.
Step 4: Backtesting the Strategy
Before a single dollar goes live, run a proper backtesting strategy that includes realistic transaction costs. The guide on how to backtest a crypto trading strategy using Python covers the broader framework — here's how to apply it specifically to stat arb.
class StatArbBacktester:
def __init__(self, maker_fee: float = 0.001,
taker_fee: float = 0.001,
slippage: float = 0.0005):
self.maker_fee = maker_fee
self.taker_fee = taker_fee
self.slippage = slippage
def run(self, signals: pd.DataFrame,
price_a: pd.Series,
price_b: pd.Series,
hedge_ratio: float,
initial_capital: float = 10000) -> pd.DataFrame:
position = signals['position']
trades = position.diff().fillna(0)
# P&L from spread
spread_returns = signals['spread'].diff()
strategy_returns = position.shift(1) * spread_returns
# Apply transaction costs on trade entries/exits
trade_costs = abs(trades) * (self.taker_fee + self.slippage) * price_a
net_returns = strategy_returns - trade_costs
cumulative_pnl = net_returns.cumsum() + initial_capital
results = pd.DataFrame({
'gross_pnl': strategy_returns.cumsum(),
'net_pnl': net_returns.cumsum(),
'portfolio_value': cumulative_pnl,
'position': position,
'zscore': signals['zscore']
})
# Performance metrics
sharpe = self._sharpe_ratio(net_returns)
max_dd = self._max_drawdown(cumulative_pnl)
print(f"Sharpe Ratio: {sharpe:.2f}")
print(f"Max Drawdown: {max_dd:.2%}")
print(f"Total Net P&L: ${net_returns.sum():.2f}")
print(f"Number of Trades: {(abs(trades) > 0).sum()}")
return results
def _sharpe_ratio(self, returns: pd.Series,
periods_per_year: int = 8760) -> float:
"""Annualized Sharpe using hourly returns."""
if returns.std() == 0:
return 0
return (returns.mean() / returns.std()) * np.sqrt(periods_per_year)
def _max_drawdown(self, portfolio_values: pd.Series) -> float:
rolling_max = portfolio_values.cummax()
drawdown = (portfolio_values - rolling_max) / rolling_max
return drawdown.min()
Two warnings about overfitting in machine learning — and this applies equally to stat arb parameter selection:
- Don't optimize your z-score thresholds on the full dataset — you'll find spurious "optimal" parameters that collapse on live data.
- Always use walk-forward analysis: train on months 1–6, test on month 7, train on months 1–7, test on month 8, repeat. This mimics how the strategy will actually perform.
Step 5: Live Execution Engine
The backtest looks good? Now for the hard part. Live execution requires managing order placement across two exchanges simultaneously, handling partial fills, and recovering gracefully from network errors.
import ccxt
import time
import logging
from typing import Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ExecutionEngine:
def __init__(self, exchange_a: ccxt.Exchange,
exchange_b: ccxt.Exchange,
max_position_usd: float = 1000):
self.exchange_a = exchange_a
self.exchange_b = exchange_b
self.max_position_usd = max_position_usd
self.open_positions = {}
def place_pair_trade(self, symbol: str,
direction: int, # 1 = long spread, -1 = short spread
hedge_ratio: float,
price_a: float,
price_b: float) -> dict:
"""
Place simultaneous orders on both exchanges.
direction=1: BUY asset_a, SELL asset_b (long spread)
direction=-1: SELL asset_a, BUY asset_b (short spread)
"""
# Calculate position sizes
size_a = self.max_position_usd / price_a
size_b = (self.max_position_usd * hedge_ratio) / price_b
side_a = 'buy' if direction == 1 else 'sell'
side_b = 'sell' if direction == 1 else 'buy'
try:
logger.info(f"Placing {side_a} {size_a:.4f} {symbol} on exchange A")
order_a = self.exchange_a.create_order(
symbol, 'market', side_a, size_a
)
logger.info(f"Placing {side_b} {size_b:.4f} {symbol} on exchange B")
order_b = self.exchange_b.create_order(
symbol, 'market', side_b, size_b
)
position_id = f"{symbol}_{int(time.time())}"
self.open_positions[position_id] = {
'symbol': symbol,
'direction': direction,
'size_a': size_a,
'size_b': size_b,
'entry_time': time.time(),
'order_a': order_a,
'order_b': order_b
}
return self.open_positions[position_id]
except ccxt.NetworkError as e:
logger.error(f"Network error placing orders: {e}")
# Attempt to flatten any partial fills
self._emergency_flatten(symbol)
raise
except ccxt.ExchangeError as e:
logger.error(f"Exchange error: {e}")
raise
def _emergency_flatten(self, symbol: str):
"""Close any open positions on both exchanges in an emergency."""
logger.warning(f"Emergency flatten triggered for {symbol}")
for ex in [self.exchange_a, self.exchange_b]:
try:
balance = ex.fetch_balance()
# Implementation depends on your specific position tracking
pass
except Exception as e:
logger.error(f"Emergency flatten failed: {e}")
Critical: Never place an order on exchange A and assume the exchange B order will fill at the same price. Slippage and execution risk during the lag between orders can turn a profitable signal into a losing trade instantly. On major pairs like BTC/USDT with high liquidity depth, this risk is manageable. On smaller altcoin pairs, it can be severe.
The maker vs taker fees decision also matters here. Using limit orders to capture maker rebates sounds attractive, but if your limit order on leg A fills while leg B's limit misses, you're sitting on a naked directional position. Many serious stat arb desks use market orders for entry despite the higher taker fees — the certainty of fill outweighs the cost difference.
Step 6: Risk Controls and Position Management
No risk management = not a trading bot, just a donation machine. The article on AI agent risk exposure controls for autonomous on-chain positions covers the broader framework for automated systems. Here's the stat arb–specific implementation:
class RiskManager:
def __init__(self,
max_spread_hold_hours: float = 48,
max_loss_per_trade_pct: float = 0.02,
max_daily_loss_pct: float = 0.05,
max_concurrent_pairs: int = 3):
self.max_spread_hold_hours = max_spread_hold_hours
self.max_loss_per_trade_pct = max_loss_per_trade_pct
self.max_daily_loss_pct = max_daily_loss_pct
self.max_concurrent_pairs = max_concurrent_pairs
self.daily_pnl = 0.0
self.initial_capital = None
def check_entry_allowed(self, open_positions: dict) -> bool:
"""Gate checks before entering a new position."""
if len(open_positions) >= self.max_concurrent_pairs:
logger.info("Max concurrent positions reached")
return False
if self.initial_capital and \
self.daily_pnl / self.initial_capital < -self.max_daily_loss_pct:
logger.warning("Daily loss limit hit — no new entries")
return False
return True
def check_exit_required(self, position: dict,
current_pnl: float,
entry_capital: float) -> tuple[bool, str]:
"""Determine if a position must be force-closed."""
# Time-based stop: spread hasn't converged in X hours
hold_hours = (time.time() - position['entry_time']) / 3600
if hold_hours > self.max_spread_hold_hours:
return True, "max_hold_time_exceeded"
# Loss-based stop
pnl_pct = current_pnl / entry_capital
if pnl_pct < -self.max_loss_per_trade_pct:
return True, "stop_loss_triggered"
return False, ""
The time-based stop deserves special attention. Stat arb relies on mean reversion happening — but sometimes it doesn't. Positions can gap against you if one exchange delists a token, has a regulatory issue, or if you've picked a pair that's entering a structural regime change. A 48-hour hard stop prevents a bad trade from becoming an account-ending trade.
Think of it like a bakery. You bake bread expecting to sell it by end of day. If it doesn't sell, you don't keep it in the display case for a week hoping someone buys it — you discount it and move on. Position time limits work the same way.
Position sizing should tie back to volatility-adjusted position sizing. The more volatile the spread's recent history, the smaller your initial position. A simple implementation:
def calculate_position_size(spread_volatility: float,
base_capital: float,
target_vol: float = 0.01) -> float:
"""Scale position size inversely with spread volatility."""
vol_scalar = target_vol / max(spread_volatility, 1e-8)
position_size = base_capital * min(vol_scalar, 1.0) # Cap at full base capital
return position_size
Step 7: The Main Loop
Glue everything together into a continuous execution loop with proper error handling:
import time
import yaml
import logging
from analysis.cointegration import CointegrationAnalyzer
from analysis.signals import SignalGenerator
from data.fetcher import MultiExchangeFetcher
from execution.orders import ExecutionEngine
from execution.risk import RiskManager
logger = logging.getLogger(__name__)
class StatArbBot:
def __init__(self, config_path: str = 'config/pairs.yaml'):
with open(config_path) as f:
self.config = yaml.safe_load(f)
self.fetcher = MultiExchangeFetcher()
self.analyzer = CointegrationAnalyzer()
self.signal_gen = SignalGenerator(
entry_threshold=self.config.get('entry_threshold', 2.0),
exit_threshold=self.config.get('exit_threshold', 0.5),
lookback=self.config.get('lookback_periods', 60)
)
self.risk_mgr = RiskManager(
max_spread_hold_hours=self.config.get('max_hold_hours', 48),
max_loss_per_trade_pct=self.config.get('max_loss_pct', 0.02)
)
self.open_positions = {}
self.coint_results_cache = {}
self.last_coint_check = 0
def run(self, symbols: list[str], poll_interval_seconds: int = 300):
logger.info("Starting stat arb bot...")
while True:
try:
self._cycle(symbols)
time.sleep(poll_interval_seconds)
except KeyboardInterrupt:
logger.info("Shutdown requested")
break
except Exception as e:
logger.error(f"Unhandled error in main loop: {e}", exc_info=True)
time.sleep(60) # Back off before retrying
def _cycle(self, symbols: list[str]):
# Re-run cointegration every 24 hours
if time.time() - self.last_coint_check > 86400:
self._refresh_cointegration(symbols)
self.last_coint_check = time.time()
for pair_key, coint_data in self.coint_results_cache.items():
if not coint_data['is_cointegrated']:
continue
symbol = coint_data['symbol']
prices = self.fetcher.fetch_synchronized_prices(symbol, limit=120)
if prices.empty or len(prices) < 60:
continue
spread = prices.iloc[:, 0] - coint_data['hedge_ratio'] * prices.iloc[:, 1]
signals = self.signal_gen.generate_signals(spread)
latest_signal = signals['position'].iloc[-1]
current_zscore = signals['zscore'].iloc[-1]
logger.info(f"{pair_key} | z-score: {current_zscore:.2f} | signal: {latest_signal}")
# Check exits on open positions
if pair_key in self.open_positions:
self._check_exit(pair_key, signals)
# Check entries if no open position
elif latest_signal != 0 and self.risk_mgr.check_entry_allowed(self.open_positions):
logger.info(f"Entry signal for {pair_key}: direction={int(latest_signal)}")
# Place trade — implementation omitted for brevity
def _refresh_cointegration(self, symbols: list[str]):
logger.info("Refreshing cointegration analysis...")
# Fetch longer history for cointegration testing
for symbol in symbols:
prices = self.fetcher.fetch_synchronized_prices(symbol, limit=500)
if len(prices) < 200:
continue
result = self.analyzer.test_cointegration(
prices.iloc[:, 0], prices.iloc[:, 1]
)
result['symbol'] = symbol
self.coint_results_cache[symbol] = result
logger.info(f"{symbol}: cointegrated={result['is_cointegrated']}, p={result['coint_pvalue']:.4f}")
Myth vs Reality: Common Stat Arb Misconceptions
Myth: "If two assets are highly correlated, they're good stat arb candidates."
Reality: Correlation measures co-movement direction. Cointegration measures whether the spread is mean-reverting. A correlation coefficient of 0.95 tells you almost nothing about whether a trade will work. Always test for cointegration.
Myth: "More pairs = more profit."
Reality: Running 20 correlated pairs simultaneously inflates your apparent diversification while creating massive hidden concentration risk. When crypto markets crash, correlations spike toward 1.0 and every pair blows up at once. Three to five well-selected, genuinely cointegrated pairs outperforms a bloated portfolio every time.
Myth: "Stat arb is market neutral."
Reality: The hedge ratio from OLS regression is a point estimate. It drifts over time. A position that starts delta-neutral can accumulate directional exposure as the ratio shifts. Recalibrate your hedge ratios at least every 48 hours, and monitor your net delta in real time.
Comparing Pair Selection Approaches
| Method | Pros | Cons | Best For |
|---|---|---|---|
| Same asset, two exchanges | Near-perfect cointegration | Tiny spreads, razor-thin margins | High-frequency, low latency setups |
| BTC/ETH cross-exchange | Deep liquidity, stable relationship | Structural breaks possible | Medium-frequency, hourly signals |
| Sector pairs (SOL/AVAX) | Wider spreads = more profit potential | More volatile relationship | Daily signals, longer hold times |
| Stablecoin pairs (USDT/USDC) | Very stable | Tiny spread, borrow cost often exceeds profit | Niche use cases only |
For most builders working with this Python crypto pairs trading framework, the BTC/ETH or BTC/BNB cross-exchange approach offers the best balance of spread width and relationship stability.
Step 8: Paper Trading and Deployment
I can't overstate this: run the bot in paper trading mode for at least four to six weeks before deploying real capital. Paper trading catches the subtle edge cases — API rate limits you didn't anticipate, spread behavior around funding settlement windows, or your cointegration pair breaking during a major news event.
Set up logging to a persistent store (PostgreSQL or even CSV works at small scale) and track every signal, every theoretical trade, and every z-score reading. Review the logs weekly. You'll spot systematic issues that never appear in a backtest.
For deployment, a lightweight VPS (2 vCPU, 4GB RAM) running Ubuntu is sufficient for polling at 5-minute intervals on five or fewer pairs. Use systemd to run the bot as a service with automatic restarts. Set up alerting via Telegram bot or email when the daily loss limit is hit or when a cointegration check fails.
Performance Benchmarks to Aim For
A well-calibrated stat arb bot trading BTC-related pairs with realistic costs should achieve:
- Sharpe ratio: 1.5–2.5 on an annualized basis (out-of-sample)
- Win rate: 55–70% of trades (stat arb wins more often than it loses, but individual losses can be larger)
- Maximum drawdown: Below 10% of starting capital with proper stops
- Trade frequency: 3–15 trades per pair per month on hourly signals
These numbers assume you're trading liquid pairs with real taker fees around 0.05–0.10% per side and realistic slippage of 3–5 basis points. The moment you move to illiquid altcoin pairs, all these benchmarks degrade significantly. That's a feature of the alpha generation process — you're earning a premium for providing price discovery, but the friction cost goes up commensurately.
For further reading on how agent-based systems handle volatile vs stable market conditions — which directly affects your cointegration stability — the analysis in agent-based trading systems performance in volatile vs stable markets is worth your time.
Key Takeaways
- Cointegration, not correlation — run the Engle-Granger test on every pair before touching entry logic. Correlation is a distraction.
- Re-calibrate constantly — hedge ratios and cointegration relationships drift. Daily recalculation isn't excessive; it's necessary.
- Fee modeling is the strategy — a stat arb edge that's 0.15% per round trip doesn't survive 0.20% in combined fees and slippage. Know your costs before you test your signals.
- Time-based stops save accounts — when convergence doesn't happen in your expected window, the trade thesis has failed. Exit, don't average in.
- Paper trade first, always — six weeks of paper trading on live data surfaces issues that six months of backtesting misses.
- Keep your pair universe small and liquid — three great pairs beats twenty mediocre ones every single time.