import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import adfuller
from scipy.stats import linregress
class PairsSelectionEngine:
    def __init__(self, lookback_years=5):
        self.lookback_years = lookback_years
        self.lookback_days = lookback_years * 252  # Trading days per year
def test_cointegration(self, price_series_1, price_series_2, significance=0.05):
        """
        Augmented Dickey-Fuller test for cointegration
        p-value < 0.05 indicates stationary (cointegrated) pair
        """
# Calculate spread (linear combination of two series)
        slope, intercept, r_value, p_value, std_err = linregress(price_series_1, price_series_2)
        spread = price_series_2 - (slope * price_series_1 + intercept)
# ADF test on spread
        adf_result = adfuller(spread, autolag='AIC')
return {
            'p_value': adf_result[1],
            'is_cointegrated': adf_result[1] < significance,
            'test_statistic': adf_result[0],
            'slope': slope,
            'intercept': intercept,
            'r_squared': r_value ** 2,
            'spread': spread
        }
def find_cointegrated_pairs(self, symbols, price_data):
        """
        Screen all symbol combinations for cointegration
        """
cointegrated_pairs = []
for i, symbol1 in enumerate(symbols):
            for symbol2 in symbols[i+1:]:
                prices1 = price_data[symbol1]
                prices2 = price_data[symbol2]
# Ensure same length and no NaNs
                if len(prices1) != len(prices2):
                    continue
result = self.test_cointegration(prices1, prices2)
if result['is_cointegrated'] and result['r_squared'] > 0.8:
                    cointegrated_pairs.append({
                        'symbol1': symbol1,
                        'symbol2': symbol2,
                        'p_value': result['p_value'],
                        'r_squared': result['r_squared'],
                        'slope': result['slope'],
                        'intercept': result['intercept']
                    })
# Sort by p-value (lower = stronger cointegration)
        return sorted(cointegrated_pairs, key=lambda x: x['p_value'])
Example: Screen S&P 500 for pairs
sp500_symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'NVDA']  # ... 500 symbols
price_data = fetch_historical_prices(sp500_symbols, years=5)
engine = PairsSelectionEngine(lookback_years=5)
cointegrated = engine.find_cointegrated_pairs(sp500_symbols, price_data)
print("Top cointegrated pairs:")
for pair in cointegrated[:10]:
    print(f"{pair['symbol1']}/{pair['symbol2']}: p-value={pair['p_value']:.4f}, R²={pair['r_squared']:.2%}")

def calculate_rolling_correlation(series1, series2, window=60):
    """
    Monitor if correlation weakens (pair degrades)
    Healthy pairs maintain >0.7 correlation
    """
correlation = series1.rolling(window).corr(series2)
    return correlation
def validate_pair_stability(symbol1, symbol2, min_correlation=0.7, lookback_days=252):
    """
    Ensure pair correlation hasn't deteriorated recently
    """
prices1 = fetch_prices(symbol1, days=lookback_days)
    prices2 = fetch_prices(symbol2, days=lookback_days)
rolling_corr = calculate_rolling_correlation(prices1, prices2, window=60)
recent_corr = rolling_corr.tail(20).mean()
    historical_corr = rolling_corr.head(-20).mean()
deterioration = historical_corr - recent_corr
return {
        'recent_correlation': recent_corr,
        'historical_correlation': historical_corr,
        'deterioration': deterioration,
        'is_stable': recent_corr > min_correlation
    }

class AutomatedPairsTrader:
    def __init__(self, account_balance=100000, risk_per_trade=0.02):
        self.balance = account_balance
        self.risk_per_trade = risk_per_trade
        self.open_pairs = {}
        self.trades = []
def calculate_spread_zscore(self, symbol1, symbol2, lookback=60):
        """
        Calculate Z-score of price ratio
        Entry at ±2.0 std devs, exit at ±0.5 std devs
        """
prices1 = fetch_prices(symbol1, days=lookback)
        prices2 = fetch_prices(symbol2, days=lookback)
# Price ratio
        ratio = prices2 / prices1
# Statistics
        ratio_mean = ratio.mean()
        ratio_std = ratio.std()
        current_ratio = ratio.iloc[-1]
# Z-score
        zscore = (current_ratio - ratio_mean) / ratio_std
return {
            'zscore': zscore,
            'ratio': current_ratio,
            'mean': ratio_mean,
            'std': ratio_std,
            'upper_band': ratio_mean + (2.0 * ratio_std),
            'lower_band': ratio_mean - (2.0 * ratio_std),
            'middle_band': ratio_mean
        }
def generate_pairs_signals(self, symbol1, symbol2, entry_zscore=2.0, exit_zscore=0.5):
        """
        Generate buy/sell signals based on spread Z-score
        """
spread_data = self.calculate_spread_zscore(symbol1, symbol2)
        zscore = spread_data['zscore']
if zscore > entry_zscore:
            # Pair ratio too high: short symbol2, long symbol1
            return 'SPREAD_HIGH', 'SHORT_S2_LONG_S1'
        elif zscore < -entry_zscore:
            # Pair ratio too low: long symbol2, short symbol1
            return 'SPREAD_LOW', 'LONG_S2_SHORT_S1'
        elif abs(zscore) < exit_zscore:
            # Spread reverted: close position
            return 'MEAN_REVERSION', 'EXIT'
        else:
            return None, 'HOLD'
def execute_pairs_trade(self, symbol1, symbol2, signal_type):
        """
        Execute paired long/short position with proper sizing
        """
price1 = fetch_current_price(symbol1)
        price2 = fetch_current_price(symbol2)
# Calculate position sizes for market-neutral exposure
        notional_per_leg = self.balance * self.risk_per_trade
if signal_type == 'LONG_S2_SHORT_S1':
            size1 = notional_per_leg / price1
            size2 = notional_per_leg / price2
# Market value should be equal (neutral)
            delta = (size2  price2) - (size1  price1)
            if abs(delta) / notional_per_leg > 0.05:  # Adjust if imbalanced >5%
                size1 = (notional_per_leg / (size1  price1))
trade = {
                'pair': f"{symbol1}/{symbol2}",
                'symbol1': symbol1,
                'symbol2': symbol2,
                'leg1_side': 'SHORT',
                'leg1_size': size1,
                'leg1_price': price1,
                'leg2_side': 'LONG',
                'leg2_size': size2,
                'leg2_price': price2,
                'signal': signal_type,
                'entry_time': pd.Timestamp.now()
            }
self.open_pairs[f"{symbol1}/{symbol2}"] = trade
        self.trades.append(trade)
print(f"Pairs trade: SHORT {size1:.2f} {symbol1} @ {price1}, LONG {size2:.2f} {symbol2} @ {price2}")
def manage_pairs_positions(self):
        """
        Monitor positions and close when spread reverts
        """
for pair_name, trade in list(self.open_pairs.items()):
            symbol1, symbol2 = trade['symbol1'], trade['symbol2']
signal, action = self.generate_pairs_signals(symbol1, symbol2)
if action == 'EXIT':
                # Close both legs
                pnl = self.calculate_pairs_pnl(trade)
                print(f"Closed {pair_name}: P&L = ${pnl:.2f}")
                del self.open_pairs[pair_name]
def calculate_pairs_pnl(self, trade):
        """
        Calculate unrealized P&L for paired position
        """
current_price1 = fetch_current_price(trade['symbol1'])
        current_price2 = fetch_current_price(trade['symbol2'])
# P&L on short leg (symbol1)
        pnl_short = (trade['leg1_price'] - current_price1) * trade['leg1_size']
# P&L on long leg (symbol2)
        pnl_long = (current_price2 - trade['leg2_price']) * trade['leg2_size']
total_pnl = pnl_short + pnl_long
return total_pnl

class PairsRiskManager:
    def __init__(self, max_correlation_decay=0.15):
        self.max_correlation_decay = max_correlation_decay
def monitor_pair_degradation(self, symbol1, symbol2, pair_trade):
        """
        Close pair if correlation breaks (hedge fails)
        """
# Current correlation
        prices1 = fetch_prices(symbol1, days=60)
        prices2 = fetch_prices(symbol2, days=60)
        current_corr = prices1.corr(prices2)
# Expected correlation from setup
        expected_corr = pair_trade['original_correlation']
correlation_decay = expected_corr - current_corr
if correlation_decay > self.max_correlation_decay:
            return True, f"Correlation decayed {correlation_decay:.2%}"
return False, "Correlation stable"
def limit_sector_concentration(self, open_pairs):
        """
        Don't load all pairs from same sector
        """
sector_exposure = {}
for pair in open_pairs.values():
            sector1 = get_sector(pair['symbol1'])
            sector2 = get_sector(pair['symbol2'])
sector_exposure[sector1] = sector_exposure.get(sector1, 0) + 0.5
            sector_exposure[sector2] = sector_exposure.get(sector2, 0) + 0.5
# Alert if sector concentration > 30%
        total_exposure = sum(sector_exposure.values())
        for sector, exposure in sector_exposure.items():
            if exposure / total_exposure > 0.30:
                print(f"WARNING: {sector} concentration = {exposure/total_exposure:.1%}")