import pandas as pd
import numpy as np
from statsmodels.tsa.stattools import coint
from scipy import stats
class PairsBacktester:
    @staticmethod
    def test_cointegration(series1, series2):
        """Johansen cointegration test"""
        score, p_value, _ = coint(series1, series2)
        return p_value  # p < 0.05 suggests cointegration
@staticmethod
    def calculate_hedge_ratio(series1, series2):
        """OLS regression for hedge ratio"""
        X = np.column_stack([series2, np.ones(len(series2))])
        beta, alpha = np.linalg.lstsq(X, series1, rcond=None)[0]
        return beta
def backtest_pair(self, df, asset1_col, asset2_col, entry_zscore=2.0, exit_zscore=0.5):
        """Efficient vectorized pairs backtest"""
        df = df.copy()
# Calculate spread
        prices1 = df[asset1_col]
        prices2 = df[asset2_col]
# Hedge ratio
        beta = self.calculate_hedge_ratio(prices1.values, prices2.values)
# Spread
        df['Spread'] = prices1 - (beta * prices2)
# Z-score
        df['SMA_Spread'] = df['Spread'].rolling(60).mean()
        df['Std_Spread'] = df['Spread'].rolling(60).std()
        df['Zscore'] = (df['Spread'] - df['SMA_Spread']) / df['Std_Spread']
# Signals (vectorized)
        df['Position'] = 0
        df.loc[df['Zscore'] < -entry_zscore, 'Position'] = 1      # Long spread
        df.loc[df['Zscore'] > entry_zscore, 'Position'] = -1       # Short spread
        df.loc[abs(df['Zscore']) < exit_zscore, 'Position'] = 0    # Exit
df['Position'] = df['Position'].fillna(method='ffill').fillna(0)
# Returns
        df['Return1'] = prices1.pct_change()
        df['Return2'] = prices2.pct_change()
# Pairs return: Long asset1 + Short asset2
        df['Strategy_Return'] = df['Position'].shift(1)  (df['Return1'] - beta  df['Return2']) * 0.998
df['Cumulative'] = (1 + df['Strategy_Return']).cumprod()
sr = df['Strategy_Return'].dropna()
        return {
            'Return': (df['Cumulative'].iloc[-1] - 1) * 100,
            'Sharpe': (sr.mean() / sr.std()) * np.sqrt(252) if sr.std() > 0 else 0,
            'Win_Rate': len(sr[sr > 0]) / len(sr) * 100,
            'Trades': (df['Position'].diff() != 0).sum(),
        }
class ParallelPairsBacktester:
    """Efficient multi-pair backtesting"""
def __init__(self, pair_list):
        self.pair_list = pair_list  # [(asset1, asset2), ...]
        self.results = {}
def backtest_all_pairs(self, df_prices):
        """Parallel backtest all pairs"""
        from concurrent.futures import ThreadPoolExecutor
backtester = PairsBacktester()
with ThreadPoolExecutor(max_workers=4) as executor:
            futures = {
                executor.submit(backtester.backtest_pair, df_prices, asset1, asset2): (asset1, asset2)
                for asset1, asset2 in self.pair_list
            }
for future in futures:
                pair = futures[future]
                try:
                    result = future.result()
                    self.results[pair] = result
                except Exception as e:
                    print(f"Error backtesting {pair}: {str(e)}")
return pd.DataFrame(self.results).T.sort_values('Sharpe', ascending=False)

def find_cointegrated_pairs(price_df, threshold=0.05):
    """Find all cointegrated pairs in a universe"""
    cointegrated_pairs = []
symbols = price_df.columns
    n = len(symbols)
for i in range(n):
        for j in range(i+1, n):
            p_value = PairsBacktester.test_cointegration(
                price_df[symbols[i]].values,
                price_df[symbols[j]].values
            )
if p_value < threshold:
                cointegrated_pairs.append({
                    'Asset1': symbols[i],
                    'Asset2': symbols[j],
                    'PValue': p_value,
                })
return pd.DataFrame(cointegrated_pairs).sort_values('PValue')
Find cointegrated pairs
price_df = pd.read_csv('stock_prices.csv')
pairs = find_cointegrated_pairs(price_df)
print(f"Found {len(pairs)} cointegrated pairs")

def optimize_pairs_parameters(df, asset1, asset2, lookback_range, zscore_range):
    """Grid search optimal parameters"""
    results = []
for lookback in lookback_range:
        for entry_z in zscore_range:
            df_copy = df.copy()
beta = PairsBacktester.calculate_hedge_ratio(df[asset1].values, df[asset2].values)
            df_copy['Spread'] = df[asset1] - (beta * df[asset2])
df_copy['SMA'] = df_copy['Spread'].rolling(lookback).mean()
            df_copy['Std'] = df_copy['Spread'].rolling(lookback).std()
            df_copy['Zscore'] = (df_copy['Spread'] - df_copy['SMA']) / df_copy['Std']
# Generate signals
            df_copy['Position'] = 0
            df_copy.loc[df_copy['Zscore'] < -entry_z, 'Position'] = 1
            df_copy.loc[df_copy['Zscore'] > entry_z, 'Position'] = -1
# Returns
            df_copy['Return1'] = df[asset1].pct_change()
            df_copy['Return2'] = df[asset2].pct_change()
            df_copy['Strategy_Return'] = df_copy['Position'].shift(1)  (df_copy['Return1'] - beta  df_copy['Return2'])
sr = df_copy['Strategy_Return'].dropna()
            sharpe = (sr.mean() / sr.std()) * np.sqrt(252) if sr.std() > 0 else 0
results.append({
                'Lookback': lookback,
                'Entry_ZScore': entry_z,
                'Sharpe': sharpe,
            })
return pd.DataFrame(results).sort_values('Sharpe', ascending=False)

def portfolio_pairs_trading(pairs_list, df_prices):
    """Trade multiple pairs simultaneously"""
    backtester = PairsBacktester()
    total_position = 0
for asset1, asset2 in pairs_list:
        # Calculate individual pair position
        pair_metrics = backtester.backtest_pair(df_prices, asset1, asset2)
        # Combine positions (with appropriate weighting)
        total_position += pair_metrics['Position']
return total_position