import numpy as np ### <summary> ### Basic template algorithm simply initializes the date range and cash. This is a skeleton ### framework you can use for designing an algorithm. ### </summary> class BasicTemplateAlgorithm(QCAlgorithm): def Initialize(self): '''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.''' self.SetStartDate(2014,1,1) #Set Start Date self.SetEndDate(2015,1,1) #Set End Date self.SetCash(50000) #Set Strategy Cash self.SetBrokerageModel(BrokerageName.InteractiveBrokersBrokerage) # what resolution should the data *added* to the universe be? self.UniverseSettings.Resolution = Resolution.Daily # this add universe method accepts two parameters: # - coarse selection function: accepts an IEnumerable<CoarseFundamental> and returns an IEnumerable<Symbol> # - fine selection function: accepts an IEnumerable<FineFundamental> and returns an IEnumerable<Symbol> self.AddUniverse(self.CoarseSelectionFunction, self.FineSelectionFunction) self.__numberOfSymbols = 5 self.__numberOfSymbolsFine = 2 self._changes = None self.stateData = {} # sort the data by daily dollar volume and take the top 'NumberOfSymbols' def CoarseSelectionFunction(self, coarse): # sort descending by daily dollar volume sortedByDollarVolume = sorted(coarse, key=lambda x: x.DollarVolume, reverse=True) for cf in sortedByDollarVolume: if cf.Symbol not in self.stateData: self.stateData[cf.Symbol] = SelectionData(cf.Symbol, 20, 4) # update the SelectionData object with current EOD price breaches = self.stateData[cf.Symbol] breaches.update(cf.EndTime, cf.AdjustedPrice) # Filter the values of the dict: we only want positive breaches values = list(filter(lambda x: x.pos_breach, self.stateData.Values())) # look for upper breach of the keltner channel # return the symbol objects of the top entries from our sorted collection return [ x.Symbol for x in values[:self.__numberOfSymbols] ] # sort the data by P/E ratio and take the top 'NumberOfSymbolsFine' def FineSelectionFunction(self, fine): # sort descending by P/E ratio sortedByPeRatio = sorted(fine, key=lambda x: x.ValuationRatios.PERatio, reverse=True) # take the top entries from our sorted collection return [ x.Symbol for x in sortedByPeRatio[:self.__numberOfSymbolsFine] ] def OnData(self, data): self.Log(f"OnData({self.UtcTime}): Keys: {', '.join([key.Value for key in self.Securities.Keys if self.Securities[key].Invested])}") # if we have no changes, do nothing if self._changes is None: return # liquidate removed securities for security in self._changes.RemovedSecurities: if security.Invested: self.Liquidate(security.Symbol) self.Log(f"Liquidated: {security.Symbol}") # we want 20% allocation in each security in our universe for security in self._changes.AddedSecurities: self.SetHoldings(security.Symbol, 0.2) self.Log(f"SetHoldings: {security.Symbol}") self._changes = None # this event fires whenever we have changes to our universe def OnSecuritiesChanged(self, changes): self._changes = changes class SelectionData(object): def __init__(self, symbol, period, k): self.symbol = symbol self.keltner = KeltnerChannels(period, k) self.pos_breach = False self.neg_breach = False def update(self, time, price): if self.keltner.Update(time, price): if price > self.keltner: self.pos_breach = True self.neg_breach = False elif price < self.keltner: self.pos_breach = False self.neg_breach = True else: self.pos_breach = False self.neg_breach = False

I am trying to learn how to use the AddUniverse method. I am using a combination of coarse and fine selection functions.  I have created SelectionData class which includes the KeltnerChannel indicator.  When I run the algorithm, the compiler reports an error stating the KelternChannel should be updated using IBaseDataBar. I am calling the SymbolData update method from a CoarseSelectionFunction.  The coarse parameter does not seem to contain a Bar property.  How can I access IBaseDataBar from within the CoarseSelectionFunction?

Author