Automating Pairs Trading in Python
Python has become the language of quantitative finance, with libraries like pandas, numpy, and statsmodels enabling professional pairs trading implementations in <500 lines of code. If you hit import errors wiring up statsmodels for cointegration or ADF tests, our statsmodels import reference documents which import paths actually work. This guide provides production-ready Python code for identifying, backtesting, and deploying market-neutral pairs trading strategies. By the end, you'll have a complete pairs trading system from data ingestion through execution.Python Libraries for Pairs Trading
``python
Essential libraries
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
from datetime import datetime, timedelta
Statistical analysis
from statsmodels.tsa.stattools import adfuller, coint
from scipy import stats
import seaborn as sns
Data sources
import yfinance as yf # Stock data
import pandas_datareader as pdr # Alternative data sources
import ccxt # Crypto data
Backtesting
import backtrader # Full backtesting framework
or
from backtesting import Backtest, Strategy # Lightweight alternative
Order execution
import alpaca_trade_api as tradeapi # Alpaca broker
import td_ameritrade # TD Ameritrade
import ib_insync # Interactive Brokers
`
Complete Pairs Trading Implementation
Step 1: Data Acquisition and Preparation
`python
import pandas as pd
import yfinance as yf
from datetime import datetime, timedelta
class PairsDataManager:
def __init__(self, lookback_years=5):
self.lookback_years = lookback_years
self.lookback_days = lookback_years * 252
def fetch_price_data(self, symbols, start_date=None, end_date=None):
"""
Fetch historical OHLC data for multiple symbols
"""
if end_date is None:
end_date = datetime.now()
if start_date is None:
start_date = end_date - timedelta(days=self.lookback_days)
data = {}
for symbol in symbols:
print(f"Fetching {symbol}...")
df = yf.download(symbol, start=start_date, end=end_date, progress=False)
data[symbol] = df['Adj Close']
return pd.DataFrame(data)
def calculate_returns(self, prices):
"""Calculate log returns"""
return np.log(prices / prices.shift(1))
def calculate_correlation_matrix(self, returns):
"""Correlation between all symbol pairs"""
return returns.corr()
Usage
manager = PairsDataManager(lookback_years=5)
symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'NVDA']
prices = manager.fetch_price_data(symbols)
print("Price data shape:", prices.shape)
print("\nFirst few rows:")
print(prices.head())
`
Step 2: Pair Selection via Cointegration Testing
`python
from statsmodels.tsa.stattools import coint, adfuller
import itertools
class PairsFinder:
def __init__(self, min_correlation=0.7, max_pvalue=0.05):
self.min_correlation = min_correlation
self.max_pvalue = max_pvalue
def find_cointegrated_pairs(self, price_data):
"""
Screen all pairs for cointegration using Engle-Granger test
"""
symbols = price_data.columns.tolist()
cointegrated_pairs = []
# Test all combinations
for symbol1, symbol2 in itertools.combinations(symbols, 2):
prices1 = price_data[symbol1].values
prices2 = price_data[symbol2].values
# Engle-Granger cointegration test
score, pvalue, _ = coint(prices1, prices2)
# Correlation check
correlation = price_data[[symbol1, symbol2]].corr().iloc[0, 1]
# Store if cointegrated
if pvalue < self.max_pvalue and correlation > self.min_correlation:
cointegrated_pairs.append({
'symbol1': symbol1,
'symbol2': symbol2,
'pvalue': pvalue,
'correlation': correlation,
'strength': -np.log10(pvalue) # Strength metric
})
# Sort by strength
cointegrated_pairs = sorted(cointegrated_pairs,
key=lambda x: x['strength'], reverse=True)
return cointegrated_pairs
Usage
finder = PairsFinder(min_correlation=0.8, max_pvalue=0.05)
cointegrated = finder.find_cointegrated_pairs(prices)
print("Top cointegrated pairs:")
for i, pair in enumerate(cointegrated[:10], 1):
print(f"{i}. {pair['symbol1']}/{pair['symbol2']}: "
f"p-value={pair['pvalue']:.4f}, correlation={pair['correlation']:.3f}")
`
Step 3: Spread Calculation and Signal Generation
`python
import numpy as np
from scipy.optimize import minimize
class PairsSpreadCalculator:
def __init__(self, lookback_period=60):
self.lookback_period = lookback_period
def calculate_hedge_ratio(self, y, x):
"""
Calculate optimal hedge ratio using OLS regression
hedge_ratio = cov(y,x) / var(x)
"""
x_const = np.column_stack([x, np.ones(len(x))])
params = np.linalg.lstsq(x_const, y, rcond=None)[0]
return params[0] # Slope = hedge ratio
def calculate_spread(self, prices_y, prices_x, lookback=60):
"""
Calculate mean-reverting spread between two price series
"""
# Calculate hedge ratio on recent data
hedge_ratio = self.calculate_hedge_ratio(prices_y[-lookback:], prices_x[-lookback:])
# Spread = y - hedge_ratio * x
spread = prices_y - (hedge_ratio * prices_x)
return spread, hedge_ratio
def calculate_zscore(self, spread, lookback=60):
"""Z-score of spread for entry/exit signals"""
mean = spread.rolling(window=lookback).mean()
std = spread.rolling(window=lookback).std()
zscore = (spread - mean) / std
return zscore
def generate_signals(self, zscore, entry_threshold=2.0, exit_threshold=0.5):
"""
Generate trading signals based on Z-score
Entry: |zscore| > entry_threshold
Exit: |zscore| < exit_threshold
"""
signals = pd.Series(0, index=zscore.index)
# Long signal (y cheap relative to x)
signals[zscore < -entry_threshold] = 1
# Short signal (y expensive relative to x)
signals[zscore > entry_threshold] = -1
# Exit signal
signals[np.abs(zscore) < exit_threshold] = 0
return signals
Usage
calc = PairsSpreadCalculator(lookback_period=60)
symbol1, symbol2 = 'AAPL', 'MSFT'
y = prices[symbol1].values
x = prices[symbol2].values
spread, hedge_ratio = calc.calculate_spread(y, x)
zscore = calc.calculate_zscore(spread)
signals = calc.generate_signals(zscore)
print(f"Hedge ratio: {hedge_ratio:.4f}")
print(f"Current spread: {spread.iloc[-1]:.4f}")
print(f"Current Z-score: {zscore.iloc[-1]:.2f}")
print(f"Current signal: {signals.iloc[-1]}")
`
Step 4: Backtesting Implementation
`python
import pandas as pd
import numpy as np
class PairsBacktester:
def __init__(self, initial_capital=100000, transaction_cost=0.001):
self.initial_capital = initial_capital
self.transaction_cost = transaction_cost
self.trades = []
self.equity_curve = []
def backtest_pairs_strategy(self, prices1, prices2, signals, hedge_ratio):
"""
Backtest pairs trading strategy
"""
capital = self.initial_capital
position = 0 # 0 = no position, 1 = long pair, -1 = short pair
entry_price = None
results = {
'dates': [],
'equity': [],
'position': [],
'pnl': []
}
for i in range(len(signals)):
signal = signals.iloc[i]
price1 = prices1.iloc[i]
price2 = prices2.iloc[i]
# Entry signal
if signal == 1 and position == 0: # Long y, short x
position = 1
entry_price = (price1, price2)
capital *= (1 - self.transaction_cost)
elif signal == -1 and position == 0: # Short y, long x
position = -1
entry_price = (price1, price2)
capital *= (1 - self.transaction_cost)
# Exit signal
elif signal == 0 and position != 0:
if position == 1:
# Close long y, short x
pnl = ((price1 - entry_price[0]) / entry_price[0] -
hedge_ratio * (price2 - entry_price[1]) / entry_price[1])
else: # position == -1
# Close short y, long x
pnl = (-(price1 - entry_price[0]) / entry_price[0] +
hedge_ratio * (price2 - entry_price[1]) / entry_price[1])
capital = (1 + pnl) (1 - self.transaction_cost)
position = 0
self.trades.append({
'entry_date': signals.index[i-10], # Approximate
'exit_date': signals.index[i],
'pnl_pct': pnl,
'return': pnl
})
results['dates'].append(signals.index[i])
results['equity'].append(capital)
results['position'].append(position)
return pd.DataFrame(results)
def calculate_metrics(self, results):
"""Calculate backtest statistics"""
equity = np.array(results['equity'])
returns = np.diff(equity) / equity[:-1]
total_return = (equity[-1] - self.initial_capital) / self.initial_capital
annual_return = (1 + total_return) ** (252 / len(returns)) - 1
sharpe_ratio = np.mean(returns) / np.std(returns) * np.sqrt(252)
max_drawdown = (np.min(equity) - self.initial_capital) / self.initial_capital
trades_df = pd.DataFrame(self.trades)
if len(trades_df) > 0:
win_rate = len(trades_df[trades_df['return'] > 0]) / len(trades_df)
else:
win_rate = 0
return {
'total_return': total_return,
'annual_return': annual_return,
'sharpe_ratio': sharpe_ratio,
'max_drawdown': max_drawdown,
'win_rate': win_rate,
'total_trades': len(self.trades)
}
Usage
backtester = PairsBacktester(initial_capital=100000)
results = backtester.backtest_pairs_strategy(
prices[symbol1],
prices[symbol2],
signals,
hedge_ratio
)
metrics = backtester.calculate_metrics(results)
print("\nBacktest Results:")
print(f"Total Return: {metrics['total_return']:.2%}")
print(f"Annual Return: {metrics['annual_return']:.2%}")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"Max Drawdown: {metrics['max_drawdown']:.2%}")
print(f"Win Rate: {metrics['win_rate']:.2%}")
print(f"Total Trades: {metrics['total_trades']}")
`
Step 5: Live Trading (Paper Trading)
`python
import alpaca_trade_api as tradeapi
from datetime import datetime
import time
class PairsLiveTrader:
def __init__(self, api_key, secret_key, base_url='https://paper-trading.alpaca.markets'):
self.api = tradeapi.REST(api_key, secret_key, base_url, api_version='v2')
self.positions = {}
def run_trading_loop(self, symbol1, symbol2, lookback=60, interval_minutes=5):
"""
Execute pairs trading in real-time
"""
while True:
try:
# Fetch latest prices
price1 = float(self.api.get_latest_trade(symbol1).price)
price2 = float(self.api.get_latest_trade(symbol2).price)
# Calculate signals
calc = PairsSpreadCalculator(lookback_period=lookback)
spread, hedge_ratio = calc.calculate_spread(
np.array([price1]), np.array([price2])
)
zscore = calc.calculate_zscore(spread)
signals = calc.generate_signals(zscore)
signal = signals.iloc[-1]
# Execute trades
if signal == 1 and symbol1 not in self.positions:
self.open_pairs_position(symbol1, symbol2, 'LONG', hedge_ratio)
elif signal == -1 and symbol1 not in self.positions:
self.open_pairs_position(symbol1, symbol2, 'SHORT', hedge_ratio)
elif signal == 0 and symbol1 in self.positions:
self.close_pairs_position(symbol1, symbol2)
# Wait before next check
time.sleep(interval_minutes * 60)
except Exception as e:
print(f"Error in trading loop: {e}")
time.sleep(60)
def open_pairs_position(self, symbol1, symbol2, direction, hedge_ratio):
"""Open long/short paired position"""
notional = 10000 # $10,000 per leg
if direction == 'LONG':
# Long symbol1, short symbol2
self.api.submit_order(symbol1, notional / self.api.get_latest_trade(symbol1).price, 'buy')
self.api.submit_order(symbol2, (notional * hedge_ratio) / self.api.get_latest_trade(symbol2).price, 'sell')
else: # SHORT
# Short symbol1, long symbol2
self.api.submit_order(symbol1, notional / self.api.get_latest_trade(symbol1).price, 'sell')
self.api.submit_order(symbol2, (notional * hedge_ratio) / self.api.get_latest_trade(symbol2).price, 'buy')
self.positions[symbol1] = {'symbol2': symbol2, 'direction': direction}
print(f"Opened {direction} pairs position: {symbol1}/{symbol2}")
def close_pairs_position(self, symbol1, symbol2):
"""Close paired position"""
position1 = self.api.get_position(symbol1)
position2 = self.api.get_position(symbol2)
self.api.submit_order(symbol1, position1.qty, 'sell' if position1.qty > 0 else 'buy')
self.api.submit_order(symbol2, position2.qty, 'buy' if position2.qty > 0 else 'sell')
del self.positions[symbol1]
print(f"Closed pairs position: {symbol1}/{symbol2}")
`
Complete Working Example
`python
Full pairs trading pipeline
if __name__ == "__main__":
# 1. Fetch data
manager = PairsDataManager(lookback_years=5)
symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'NVDA']
prices = manager.fetch_price_data(symbols)
# 2. Find cointegrated pairs
finder = PairsFinder()
cointegrated = finder.find_cointegrated_pairs(prices)
# 3. Backtest top pair
if cointegrated:
pair = cointegrated[0]
symbol1, symbol2 = pair['symbol1'], pair['symbol2']
calc = PairsSpreadCalculator()
spread, hedge_ratio = calc.calculate_spread(prices[symbol1], prices[symbol2])
zscore = calc.calculate_zscore(spread)
signals = calc.generate_signals(zscore)
backtester = PairsBacktester()
results = backtester.backtest_pairs_strategy(
prices[symbol1], prices[symbol2], signals, hedge_ratio
)
metrics = backtester.calculate_metrics(results)
print(f"\n{symbol1}/{symbol2} Strategy Metrics:")
print(f"Sharpe Ratio: {metrics['sharpe_ratio']:.2f}")
print(f"Total Return: {metrics['total_return']:.2%}")
# 4. Deploy live trading (requires API credentials)
# trader = PairsLiveTrader(api_key, secret_key)
# trader.run_trading_loop(symbol1, symbol2)
`
Frequently Asked Questions
Q: Which Python backtesting library should I use?
A: For pairs trading, lightweight libraries like backtesting.py are ideal. For complex strategies, use Backtrader. For institutional-grade, use Zipline or QSTrader.
Q: How do I handle transaction costs in Python?
A: Track trades with transaction_cost = 0.001` (0.1% round-trip). Deduct from capital on every trade. Use realistic costs: equities 0.05-0.1%, crypto 0.1-0.3%.
Q: Can I use asyncio for parallel pair scanning?
A: Yes, use asyncio to fetch data for 100+ pairs concurrently, reducing runtime from 30 minutes to 2 minutes.
Q: How do I deploy a Python pairs trading bot to production?
A: Use Docker containers with 24/7 monitoring. Log all trades to database. Set up alerts for execution errors. Use paper trading first (1-2 months minimum).
Q: What libraries help with paper trading before going live?
A: Alpaca's paper trading, Interactive Brokers' demo accounts, or simulate manually in backtester. Paper trade minimum 500+ trades before risking real capital.