Backtest

Overall Statistics
Total Orders 197 Average Win 0.34% Average Loss -0.34% Compounding Annual Return -0.938% Drawdown 2.400% Expectancy -0.010 Start Equity 1000000 End Equity 992936.42 Net Profit -0.706% Sharpe Ratio -4.392 Sortino Ratio -2.679 Probabilistic Sharpe Ratio 5.913% Loss Rate 50% Win Rate 50% Profit-Loss Ratio 1.00 Alpha -0.062 Beta 0.006 Annual Standard Deviation 0.014 Annual Variance 0 Information Ratio -0.69 Tracking Error 0.181 Treynor Ratio -10.648 Total Fees $910.98 Estimated Strategy Capacity $380000000.00 Lowest Capacity Asset QQQ RIWIV7K5Z9LX Portfolio Turnover 35.52% Drawdown Recovery 11

#region imports
from AlgorithmImports import *
#endregion
from datetime import timedelta, datetime

class SMAPairsTrading(QCAlgorithm):

    def Initialize(self):
        self.SetStartDate(2022, 6, 1)   
        self.SetEndDate(2024, 6, 1)
        self.SetCash(100000)
        
        symbols = [
            Symbol.Create("QQQQ", SecurityType.Equity, Market.USA), 
            Symbol.Create("SPY", SecurityType.Equity, Market.USA)
        ]
        self.AddUniverseSelection(ManualUniverseSelectionModel(symbols))
        self.UniverseSettings.Resolution = Resolution.Hour
        self.UniverseSettings.DataNormalizationMode = DataNormalizationMode.Raw
        self.AddAlpha(PairsTradingAlphaModel())
        self.SetPortfolioConstruction(EqualWeightingPortfolioConstructionModel())
        self.SetExecution(ImmediateExecutionModel())
        
    def OnEndOfDay(self, symbol):
        self.Log("Taking a position of " + str(self.Portfolio[symbol].Quantity) + " units of symbol " + str(symbol))

class PairsTradingAlphaModel(AlphaModel):

    def __init__(self):
        self.pair = [ ]
        self.spreadMean1 = SimpleMovingAverage(12)
        self.spreadMean2 = SimpleMovingAverage(26)
        self.period = timedelta(hours=4)
        
    def Update(self, algorithm, data):
        spread = self.pair[1].Price - self.pair[0].Price
        self.spreadMean1.Update(algorithm.Time, spread)
        self.spreadMean2.Update(algorithm.Time, spread)
        
        sprmean1_t1 = self.spreadMean1[1]
        sprmean2_t1 = self.spreadMean2[1]
        sprmean1_t2 = self.spreadMean1[2]
        sprmean2_t2 = self.spreadMean2[2]
        sprmean1_t3 = self.spreadMean1[3]
        sprmean2_t3 = self.spreadMean2[3]

        indicator_valid = all([ ele is not None for ele in [sprmean1_t1, sprmean2_t1, sprmean1_t2, sprmean2_t2, sprmean1_t3, sprmean2_t3] ])

        if indicator_valid:
            buy_signal = (sprmean1_t1>sprmean2_t1) and (sprmean1_t2<sprmean2_t2) and (sprmean1_t3<sprmean2_t3)
            sell_signal = (sprmean1_t1<sprmean2_t1) and (sprmean1_t2>sprmean2_t2) and (sprmean1_t3>sprmean2_t3)
        else:
            buy_signal = False
            sell_signal = False

        if indicator_valid and buy_signal:
            return Insight.Group(
                [
                    Insight.Price(self.pair[0].Symbol, self.period, InsightDirection.Up),
                    Insight.Price(self.pair[1].Symbol, self.period, InsightDirection.Down)
                ])
        
        if indicator_valid and sell_signal:
            return Insight.Group(
                [
                    Insight.Price(self.pair[0].Symbol, self.period, InsightDirection.Down),
                    Insight.Price(self.pair[1].Symbol, self.period, InsightDirection.Up)
                ])

        return []
    
    def OnSecuritiesChanged(self, algorithm, changes):
        self.pair = [x for x in changes.AddedSecurities]
        
        #1. Call for 500 bars of history data for each symbol in the pair and save to the variable history
        history = algorithm.History([x.Symbol for x in self.pair], 500)
        #2. Unstack the Pandas data frame to reduce it to the history close price
        history = history.close.unstack(level=0)
        #3. Iterate through the history tuple and update the mean and standard deviation with historical data 
        for tuple in history.itertuples():
            self.spreadMean1.Update(tuple[0], tuple[2]-tuple[1])
            self.spreadMean2.Update(tuple[0], tuple[2]-tuple[1])