import pandas as pd
import numpy as np
def identify_correlated_pairs(symbols_prices, min_correlation=0.8):
    """Find symbol pairs with high correlation"""
    correlation_matrix = symbols_prices.corr()
pairs = []
    for i in range(len(correlation_matrix)):
        for j in range(i+1, len(correlation_matrix)):
            corr = correlation_matrix.iloc[i, j]
            if corr > min_correlation:
                pairs.append({
                    'symbol1': correlation_matrix.index[i],
                    'symbol2': correlation_matrix.index[j],
                    'correlation': corr
                })
return pairs
def calculate_spread(price1, price2, hedge_ratio=None):
    """Calculate normalized spread"""
    if hedge_ratio is None:
        hedge_ratio = np.mean(price1) / np.mean(price2)
return price1 - hedge_ratio * price2
def simple_spread_signal(spread, window=20, entry_std=1.5, exit_std=0.5):
    """Generate buy/sell signals based on spread z-score"""
    mean = spread.rolling(window).mean()
    std = spread.rolling(window).std()
    z_score = (spread - mean) / std
signals = pd.Series(0, index=spread.index)
    signals[z_score > entry_std] = -1  # Spread too wide, short it
    signals[z_score < -entry_std] = 1  # Spread too tight, long it
return signals
Example
pairs_df = pd.DataFrame({
    'JCI': [100, 101, 99, 100, 102],  # Stock A
    'DBC': [200, 201, 200, 199, 201]  # Correlated commodity
})
pairs = identify_correlated_pairs(pairs_df)
print(f"Found pairs: {pairs}")
spread = calculate_spread(pairs_df['JCI'], pairs_df['DBC'])
signals = simple_spread_signal(spread)

class BeginnerPairsBacktest:
    """Simple pairs trading backtest"""
def __init__(self, prices1, prices2, initial_capital=100000):
        self.prices1 = prices1
        self.prices2 = prices2
        self.capital = initial_capital
        self.trades = []
# Calculate spread
        self.hedge_ratio = np.mean(prices1) / np.mean(prices2)
        self.spread = prices1 - self.hedge_ratio * prices2
def run(self, window=20, entry_zscore=1.5, exit_zscore=0.5):
        """Execute backtest"""
        position = None
for i in range(window, len(self.spread)):
            # Calculate z-score
            spread_window = self.spread.iloc[i-window:i]
            mean = spread_window.mean()
            std = spread_window.std()
            z_score = (self.spread.iloc[i] - mean) / std
# Exit on mean reversion
            if position and abs(z_score) < exit_zscore:
                price1 = self.prices1.iloc[i]
                price2 = self.prices2.iloc[i]
if position == 1:  # Long spread
                    pnl = (price1 - position['p1']) - self.hedge_ratio * (price2 - position['p2'])
                else:  # Short spread
                    pnl = -(price1 - position['p1']) + self.hedge_ratio * (price2 - position['p2'])
self.capital += pnl
                self.trades.append({'pnl': pnl})
                position = None
# Entry on divergence
            if not position and abs(z_score) > entry_zscore:
                position = {
                    'p1': self.prices1.iloc[i],
                    'p2': self.prices2.iloc[i],
                    'type': 1 if z_score > 0 else -1
                }
return {
            'total_trades': len(self.trades),
            'total_return': (self.capital - initial_capital) / initial_capital,
            'avg_trade': np.mean([t['pnl'] for t in self.trades])
        }
Example
gold_prices = pd.Series([150, 151, 149, 152, 150, 148, 149])
silver_prices = pd.Series([20, 20.5, 19.8, 20.3, 20, 19.5, 19.7])
backtest = BeginnerPairsBacktest(gold_prices, silver_prices)
results = backtest.run(window=5, entry_zscore=1.5, exit_zscore=0.5)
print(f"Trades: {results['total_trades']}")
print(f"Return: {results['total_return']:.2%}")