class UniverseWithRollingWindow(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2020, 8, 3) # Start date of backtest
self.SetEndDate(2020, 8, 6) # End date of backtest
self.SetCash(100000) # Amount of cash in account for backtest
self.AddUniverse(self.CoarseSelectionFunction) # Adding CoarseSelectionFunction as Universe
self.UniverseSettings.Resolution = Resolution.Minute # Sets Universe resolution to minute
self.Data = {} # Holds all of data indexed by each symbol for the OnData method
def CoarseSelectionFunction(self, coarse):
sortedByDollarVolume = sorted(coarse, key=lambda c: c.DollarVolume, reverse=True)
stocks = [x.Symbol for x in sortedByDollarVolume if x.Price >= 5 and x.DollarVolume >= 5000000]
return stocks[:10]
def OnSecuritiesChanged(self, changes):
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.Data:
self.Data[symbol] = SymbolData(self, symbol)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.Data:
symbolData = self.Data.pop(symbol, None)
self.SubscriptionManager.RemoveConsolidator(symbol, symbolData.consolidator) # Remove subcription for symbols removed from universe
def OnData(self, data): # OnData event is the primary entry point for the algorithm. Each new data point will be pumped in here.
if not self.Portfolio.Invested:
for symbol in self.Data.keys():
symbolData = self.Data[symbol]
if symbolData.Bars[0].Open < symbolData.Bars[0].Close:
self.MarketOrder(symbol, 1)
self.Debug("MarketOrder was placed for symbol " + str(symbol))
class SymbolData:
def __init__(self, algorithm, symbol):
self.algorithm = algorithm
self.symbol = symbol
self.Bars = RollingWindow[IBaseDataBar](6) # Rolling window for data bars
self.consolidator = TradeBarConsolidator(timedelta(minutes=5))
self.consolidator.DataConsolidated += self.OnDataConsolidated
algorithm.SubscriptionManager.AddConsolidator(symbol, self.consolidator)
def OnDataConsolidated(self, sender, bar):
self.Bars.Add(bar)