Start New Discussion 
Back
Reward Discussion 
Hello QuantConnect Community! I need help with this code. What I'm trying to do is:
a. Consolidate minute bars over longer periods.
b. Then create a SMA of consolidated bars and make a rolling window with those bars.
c. Make another rolling window of the SMA consolidated bars in order to be able to compare the price of the consolidated bars with the SMA and execute trades.
So I took as reference the Github code: https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/MultipleSymbolConsolidationAlgorithm.py and the change I made was to remove the manual universe and I was using a Coarse Selection Filter to avoid human bias. But I get this error in OnDataConsolidated function:
Runtime Error: Trying to retrieve an element from a collection using a key that does not exist in that collection throws a KeyError exception. To prevent the exception, ensure that the AAPL key exist in the collection and/or that collection is not empty. at OnDataConsolidated self.Data[bar.Symbol.Value].SMA.Update(bar.Time in main.py: line 76
I searched on the forum discussions and didn't find this KeyError addressed with a universe selections filter.
Also, I’m not sure if the Rolling window of the bars is working well in the following lines of code:
a = format(self.Data[symbol].Bars[0])
b = format(self.Data[symbol].SMA[1])
if a >= b:
self.MarketOrder(symbol, 100)
continueAnd last I don’t quite understand what “format” makes prior to:
format(self.Data[symbol].Bars[0])Any help will be appreciated. I attached the code.
from System import *
from QuantConnect import *
from QuantConnect.Data.Consolidators import *
from QuantConnect.Data.Market import *
from QuantConnect.Orders import OrderStatus
from QuantConnect.Algorithm import QCAlgorithm
from QuantConnect.Indicators import *
import numpy as np
from datetime import timedelta, datetime
class MultipleSymbolConsolidationAlgorithm(QCAlgorithm):
def Initialize(self):
self.SetStartDate(2021, 1, 26) # Set Start Date\
self.SetEndDate(2021, 4, 26)
# Holds all of our data keyed by each symbol
self.Data = {}
# Contains all of our equity symbols
self.AddUniverse(self.CoarseSelectionFilter)
self.UniverseSettings.Resolution = Resolution.Minute
self.UniverseSettings.SetDataNormalizationMode = DataNormalizationMode.Raw
self.UniverseSettings.Leverage = 1
self.__numberOfSymbols = 10
def CoarseSelectionFilter(self, coarse):
sortedByDollarVolume = sorted(coarse, key=lambda c: c.Volume, reverse=True)
filteredByPrice = [c.Symbol for c in sortedByDollarVolume if c.Price > 10 ]
return filteredByPrice[:self.__numberOfSymbols]
def OnSecuritiesChanged(self, changes):
# This is the period of bars we'll be creating
BarPeriod = TimeSpan.FromMinutes(10)
# This is the period of our sma indicators
SimpleMovingAveragePeriod = 20
# This is the number of consolidated bars we'll hold in symbol data for reference
RollingWindowSize = 10
for security in changes.AddedSecurities:
symbol = security.Symbol
if symbol not in self.Data:
self.Data[security] = SymbolData(symbol, BarPeriod, RollingWindowSize)
# loop through all our symbols and request data subscriptions and initialize indicator
for symbol, symbolData in self.Data.items():
# define the indicator
symbolData.SMA = SimpleMovingAverage(self.CreateIndicatorName(symbol.Symbol, "SMA" + str(SimpleMovingAveragePeriod), Resolution.Minute), SimpleMovingAveragePeriod)
# define a consolidator to consolidate data for this symbol on the requested period
consolidator = TradeBarConsolidator(BarPeriod)
# write up our consolidator to update the indicator
consolidator.DataConsolidated += self.OnDataConsolidated
# we need to add this consolidator so it gets auto updates
self.SubscriptionManager.AddConsolidator(symbolData.Symbol, consolidator)
for security in changes.RemovedSecurities:
symbol = security.Symbol
if symbol in self.Data:
symbolData = self.Data.pop(symbol, None)
self.SubscriptionManager.RemoveConsolidator(security.Symbol, consolidator)
if security.Invested:
self.Liquidate(security.Symbol, "Universe Removed Security")
def OnDataConsolidated(self, sender, bar):
self.Data[bar.Symbol.Value].SMA.Update(bar.Time, bar.Close)
if self.Data[bar.Symbol.Value].SMA.IsReady:
sma = self.Data[bar.Symbol.Value].SMA.Current.Value
self.Data[bar.Symbol.Value].HistoricalSMA.Add(sma)
self.Data[bar.Symbol.Value].Bars.Add(bar)
# OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.
# Argument "data": Slice object, dictionary object with your stock data
def OnData(self,data):
# loop through each symbol in our structure
for symbol in self.Data.keys():
symbolData = self.Data[symbol]
# this check proves that this symbol was JUST updated prior to this OnData function being called
if symbolData.IsReady() and symbolData.WasJustUpdated(self.Time):
self.Debug("Data rolling window".format(self.Data[symbol].SMA))
if not self.Portfolio[symbol].Invested:
self.SetHoldings(symbol, 0.1)
self.Debug("Data: {0} -> {1}".format(self.Data[symbol].Bars[1],self.Data[symbol].SMA))
'''------------------------- Trading logic ----------------------------------- '''
a = format(self.Data[symbol].Bars[0])
b = format(self.Data[symbol].SMA[1])
if a >= b:
self.MarketOrder(symbol, 100)
continue
if a <= b:
self.MarketOrder(symbol, -100)
continue
class SymbolData(object):
def __init__(self, symbol, barPeriod, windowSize):
self.Symbol = symbol
# The period used when population the Bars rolling window
self.BarPeriod = barPeriod
self.Bars = RollingWindow[IBaseDataBar](windowSize)
# The simple moving average indicator for our symbol
self.SMA = None
sma_lookback = 10
self.HistoricalSMA = RollingWindow[float](sma_lookback)
# Returns true if all the data in this instance is ready (indicators, rolling windows, ect...)
def IsReady(self):
return self.Bars.IsReady and self.SMA.IsReady
# Returns true if the most recent trade bar time matches the current time minus the bar's period, this
# indicates that update was just called on this instance
def WasJustUpdated(self, current):
return self.Bars.Count > 0 and self.Bars[0].Time == current - self.BarPeriod
5
