# 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

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())

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}")

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]}")

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']}")

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}")

# 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)