chartmill_backtest : for Technical breakout setups scanner

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*1000) # Are't we rich? :D
        self.SetStartDate(2017, 1, 1) 
        self.SetEndDate(2019, 9, 28) # if not specified, the Backtesting EndDate would be today 
        
        self.MyPortfolio = {} # {:ticker => date_invested_on}
        self.MaxHoldingPeriodInDays = 30
        self.MaxQuanityToHoldPerTicker = 100

        self.ScannerData = json.loads(self.Download("<backtest_data_url>"))
        self.Debug(self.ScannerData)
        
        self.spy = self.AddEquity("SPY", Resolution.Daily).Symbol
        self.SetBenchmark("SPY")
        self.UniverseSettings.Resolution = Resolution.Daily
        for date in self.ScannerData:
            tickers = self.ScannerData[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! :)

        # 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.Rebalance))
        
    
    # Returns [Symbol, ...]    
    def GetSymbolsFromChartMill(self):
        date = str(self.Time).split(" ")[0]
        symbols = []
        for ticker in self.ScannerData[date]:
            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):
        pass


    
    # 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, invest
    def Rebalance(self):
        #self.Debug("Inside Rebalance")
        symbols = self.GetSymbolsFromChartMill()

        # Investment
        for symbol in symbols:
            ticker = str(symbol)
            try:
                # If not bought the ticker already, buy it
                if not self.Securities[ticker].Invested:
                    #self.Debug("Investing in ticker : " + ticker)
                    self.Order(ticker, self.MaxQuanityToHoldPerTicker)
                # Update the ticker's last invested date in MyPortfolio
                self.MyPortfolio[ticker] = self.Time
            except Exception as error:
                #self.Debug("Error in investing for ticker : [" + ticker + "] ==> " + str(error))
                self.Debug("E/I[" + 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/L[" + ticker + "] ==> " + str(error))