Created with Highcharts 12.1.2EquityDec 2023Dec…Jan 2024Feb 2024Mar 2024Apr 2024May 2024Jun 2024Jul 2024Aug 2024Sep 2024Oct 2024Nov 2024Dec 2024800k900k1,000k1,100k-20-100012-1010500M1,000M010M20M02550
Overall Statistics
Total Orders
44
Average Win
2.09%
Average Loss
-2.48%
Compounding Annual Return
-5.141%
Drawdown
10.400%
Expectancy
-0.080
Start Equity
1000000
End Equity
948654.73
Net Profit
-5.135%
Sharpe Ratio
-0.967
Sortino Ratio
-0.808
Probabilistic Sharpe Ratio
5.157%
Loss Rate
50%
Win Rate
50%
Profit-Loss Ratio
0.84
Alpha
-0.082
Beta
-0.031
Annual Standard Deviation
0.09
Annual Variance
0.008
Information Ratio
-1.862
Tracking Error
0.138
Treynor Ratio
2.83
Total Fees
$603.04
Estimated Strategy Capacity
$690000000.00
Lowest Capacity Asset
MSFT R735QTJ8XC9X
Portfolio Turnover
6.19%
import numpy as np
import pandas as pd
import statsmodels.api as sm
from AlgorithmImports import *

class PairsTradingStrategy(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2023, 12, 1)
        self.SetEndDate(2024, 12, 1)
        self.SetCash(1_000_000)  # $1M initial capital
        
        self.lookback = 20  # Lookback period for z-score calculation
        self.entry_threshold = 2
        self.exit_threshold = 0
        self.risk_threshold = 3
        self.pair = ["NVDA", "MSFT"]  # Replace with the selected pair with lowest ADF
        
        # Add securities
        self.symbols = [self.AddEquity(ticker, Resolution.Daily).Symbol for ticker in self.pair]
        
        # Store historical data
        self.history = {sym: [] for sym in self.symbols}
        self.portfolio_state = 0  # 1: long, -1: short, 0: no position
        
        # Track current hedge ratio
        self.hedge_ratio = None
    
    def OnData(self, data):
        if not all(symbol in data for symbol in self.symbols):
            return
        
        # Store latest closing prices
        for symbol in self.symbols:
            self.history[symbol].append(data[symbol].Close)
            if len(self.history[symbol]) > self.lookback:
                self.history[symbol].pop(0)  # Maintain rolling window
        
        if len(self.history[self.symbols[0]]) < self.lookback:
            return  # Wait until we have enough data
        
        # Perform linear regression on log prices
        log_prices_P = np.log(self.history[self.symbols[0]])
        log_prices_Q = np.log(self.history[self.symbols[1]])
        
        model = sm.OLS(log_prices_P, sm.add_constant(log_prices_Q)).fit()
        self.hedge_ratio = model.params[1]  # Beta coefficient
        
        # Compute spread and z-score
        spread = log_prices_P - self.hedge_ratio * log_prices_Q
        mean_spread = np.mean(spread)
        std_spread = np.std(spread)
        zscore = (spread[-1] - mean_spread) / std_spread if std_spread != 0 else 0
        
        self.wt1 = 1/(1+self.hedge_ratio)
        self.wt2 = self.hedge_ratio/(1+self.hedge_ratio)


        # **Risk Management: Liquidate all positions if |zscore| > 3**
        if abs(zscore) > self.risk_threshold:
            self.Liquidate()
            self.portfolio_state = 0
            return


        # Entry conditions
        if self.portfolio_state == 0:
            if zscore < -self.entry_threshold:
                # Long P, short Q
                self.SetHoldings(self.symbols[0], self.wt1)
                self.SetHoldings(self.symbols[1], -self.wt2)
                self.portfolio_state = 1
            
            elif zscore > self.entry_threshold:
                # Short P, long Q
                self.SetHoldings(self.symbols[0], -self.wt1)
                self.SetHoldings(self.symbols[1], self.wt2)
                self.portfolio_state = -1
        
        # Exit conditions
        elif self.portfolio_state == 1 and zscore > self.exit_threshold:
            self.Liquidate()  # Close all positions
            self.portfolio_state = 0
        elif self.portfolio_state == -1 and zscore < self.exit_threshold:
            self.Liquidate()  # Close all positions
            self.portfolio_state = 0