chartmill-backtest.py

chartmill-backtest.py

import json
from datetime import datetime, timedelta


# Copies logic from https://www.quantconnect.com/tutorials/strategy-library/fundamental-factor-long-short-strategy
# Algo storing - https://www.quantconnect.com/docs/algorithm-framework/algorithm-scoring
class ChartMillBackTester(QCAlgorithm):

    def Initialize(self):
        self.SetCash(1000*100) # Are't we rich? :D
        self.SetStartDate(2017, 1, 1) # Starting from further before than actual data date, so that ema's can be ready by the time our data is available
        self.SetEndDate(2019, 10, 3) # if not specified, the Backtesting EndDate would be today 
        
        self.MyPortfolio = {} # {:ticker => date_invested_on}
        self.MaxHoldingPeriodInDays = 5
        self.MaxQuanityToHoldPerTicker = 10

        scannerDataUrl = "<URL>"
        scannerData = json.loads(self.Download(scannerDataUrl)) # {:long_tickers => {}, :short_tickers => {}}
        self.LongScannerData = scannerData["long_tickers"] # {:date => [ticker, ...], ...}
        self.ShortScannerData = scannerData["short_tickers"] # {:date => [ticker, ...], ...}
        
        self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
        self.SetBenchmark("SPY")
        self.UniverseSettings.Resolution = Resolution.Daily
        #self.AddUniverse(self.ChatMillScannerData_CoarseSelectionFunction,self.ChatMillScannerData_FineSelectionFunction)
        for date in self.LongScannerData:
            try:
                tickers = (self.LongScannerData[date] + self.ShortScannerData[date])
                for ticker in tickers:
                    self.AddEquity(ticker, Resolution.Daily)
                    # Constant fee model - https://www.quantconnect.com/docs/algorithm-reference/reality-modelling#Reality-Modelling
                    self.Securities[ticker].FeeModel = ConstantFeeModel(0) # Let's assume for a sec that there's no tx fee! :)
            except:
                pass

        # Schedule the rebalance function to execute at the begining of each day
        # https://www.quantconnect.com/docs/algorithm-reference/scheduled-events
        self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.AfterMarketOpen(self.spy,5), Action(self.LongShortStrategy))
        
        # create a 90 day exponential moving average
        self.SlowEma = self.EMA("SPY", 28, Resolution.Daily)        
        
        
    
    # def ChatMillScannerData_CoarseSelectionFunction(self, coarse):
    #     #self.Debug("Inside ChatMillScannerData_CoarseSelectionFunction")
    #     return self.GetSymbolsFromChartMill()


    # def ChatMillScannerData_FineSelectionFunction(self, fine):
    #     #self.Debug("Inside ChatMillScannerData_FineSelectionFunction")
    #     return self.GetSymbolsFromChartMill()
        
    
    # Returns [Symbol, ...]    
    # longOrShort => "long" / "short"
    def GetSymbolsFromChartMill(self, longOrShort):
        date = str(self.Time).split(" ")[0]
        symbols = []
        symbolsForDate = []

        try:
            symbolsForDate = self.LongScannerData[date] if (longOrShort == "long") else self.ShortScannerData[date]
        except:
            self.Debug("Error in fetching symbols for date : " + date)
            
        
        for ticker in symbolsForDate:
            try:
                symbols.append(Symbol.Create(ticker, SecurityType.Equity, Market.USA))
            except:
                self.Debug("Error in creating symbol : " + ticker)

        #self.Debug(str(symbols))        
        return symbols
        
        

    def OnData(self, data):
        # This function is not needed as we rely on the self.Schedule function to trigger the algo
        pass


    # Returns {"long" : N, "short" : N} based on price/ema relationship
    # Force return {"long" : 1, "short" : 1} to remove any long/short bias
    def GetHedgeBias(self):
        longBias = {"long" : 1, "short" : 0.7}
        shortBias = {"long" : 0.7, "short" : 1}
        neutralBias = {"long" : 1, "short" : 1}

        price = self.Securities["SPY"].Price
        slowEmaValue = self.SlowEma.Current.Value
        
        return neutralBias

        # if not self.SlowEma.IsReady or price == self.SlowEma.Current.Value:
        #     return neutralBias
        # else:
        #     if price > slowEmaValue: # long bias
        #         return longBias
        #     else: # short bias
        #         return shortBias
            

    
    # https://www.quantconnect.com/docs/algorithm-reference/securities-and-portfolio
    # https://www.quantconnect.com/docs/key-concepts/security-identifiers
    # Algo : if invested, liquidate after N days. Else, go long or short!
    def LongShortStrategy(self):
        #self.Debug("Inside Rebalance")
        longSymbols = self.GetSymbolsFromChartMill("long")
        shortSymbols = self.GetSymbolsFromChartMill("short")

        # Go Long on long symbols
        quantityToHold = self.GetHedgeBias()["long"] * self.MaxQuanityToHoldPerTicker        
        for symbol in longSymbols:
            ticker = str(symbol)
            try:
                # If not bought the ticker already, buy it
                if not self.Securities[ticker].Invested:
                    quantityToHold = self.GetHedgeBias()["long"] * self.MaxQuanityToHoldPerTicker
                    self.Order(ticker, quantityToHold)
                # Update the ticker's last invested date in MyPortfolio
                self.MyPortfolio[ticker] = self.Time
            except Exception as error:
                self.Debug("E/Long[" + ticker + "] ==> " + str(error))

        # # Go Short on short symbols
        quantityToHold = self.GetHedgeBias()["short"] * self.MaxQuanityToHoldPerTicker
        for symbol in shortSymbols:
            ticker = str(symbol)
            try:
                # If not sold the ticker already, short it
                if not self.Securities[ticker].Invested:
                    self.Order(ticker, -1 * quantityToHold)
                # Update the ticker's last invested date in MyPortfolio
                self.MyPortfolio[ticker] = self.Time
            except Exception as error:
                self.Debug("E/Short[" + ticker + "] ==> " + str(error))

        # Liquidating
        for ticker in list(self.MyPortfolio): # we need to use list() because we pop https://stackoverflow.com/a/11941855/440362   
            try:    
                # Buy the ticker and update it's holdings in MyPortfolio
                today = self.Time
                investedDate = self.MyPortfolio[ticker]
                tickerHoldingPeriod = abs((today - investedDate).days)
                if tickerHoldingPeriod > self.MaxHoldingPeriodInDays:
                    # liquidate the ticker and remove it from holding tracking
                    #self.Debug("Liquidating ticker : " + ticker)
                    self.Liquidate(ticker)
                    self.MyPortfolio.pop(ticker)
            except Exception as error:
                self.Debug("E/Liquidate[" + ticker + "] ==> " + str(error))