Universe Warmup Python by Stephen Hyer

Stephen Hyer INVESTOR

| Reward Discussion

I've read through everything I can on the forums but I'm still having difficulty getting my universe to warm up using the History() method. In the attached algorithm, what I believe is happening is:

from clr import AddReference AddReference("System") AddReference("QuantConnect.Algorithm") AddReference("QuantConnect.Indicators") AddReference("QuantConnect.Common") AddReference("QuantConnect.Algorithm.Framework") from System import * from QuantConnect import * from QuantConnect.Data import * from QuantConnect.Algorithm import * from QuantConnect.Indicators import * from System.Collections.Generic import List class PublicHelp(QCAlgorithm): def Initialize(self): self.SetStartDate(2017,1,1) #Set Start Date self.SetEndDate(datetime.now().date() - timedelta(1)) #Set End Date #self.SetEndDate(2013,1,1) #Set End Date self.SetCash(150000) #Set Strategy Cash self.UniverseSettings.Resolution = Resolution.Hour self.averages = { }; self.AddEquity("SPY", Resolution.Hour) self.AddUniverse(self.CoarseSelectionFunction) self.Schedule.On(self.DateRules.EveryDay("SPY"), self.TimeRules.At(9,31), self.BuyFunc)

I'm initializing and importing most of the basic stuff. I've set my Universe resolution to Hours, so the CoarseSelectionFunction should be run each hour, yeah?

So going forward, I have my coarse filter which filters down by volume and uses the EMACross tutorial code that you can find in the Universe selection portion of the Documentation.

#Universe Filter # sort the data by volume and price, apply the moving average crossver, and take the top 24 sorted results based on breakout magnitude def CoarseSelectionFunction(self, coarse): filtered = [ x for x in coarse if (x.DollarVolume > 50000000) ] # We are going to use a dictionary to refer the object that will keep the moving averages for cf in filtered: if cf.Symbol not in self.averages: self.averages[cf.Symbol] = SymbolData(cf.Symbol) # Updates the SymbolData object with current EOD price avg = self.averages[cf.Symbol] history = self.History(cf.Symbol, 16) avg.WarmUpIndicators(history.iloc[cf.Symbol]) avg.update(cf.EndTime, cf.AdjustedPrice) # Filter the values of the dict: we only want up-trending securities values = list(filter(lambda x: x.is_uptrend, self.averages.values())) # Sorts the values of the dict: we want those with greater difference between the moving averages values.sort(key=lambda x: x.scale, reverse=True) for x in values[:200]: self.Log('symbol: ' + str(x.symbol.Value) + ' scale: ' + str(x.scale)) # we need to return only the symbol objects return [ x.symbol for x in values[:200] ] # this event fires whenever we have changes to our universe def OnSecuritiesChanged(self, changes): self.changes = changes # liquidate removed securities for security in changes.RemovedSecurities: if security.Invested: self.Liquidate(security.Symbol) #EMA Crossover Class class SymbolData(object): def __init__(self, symbol): self.symbol = symbol self.fast = ExponentialMovingAverage(50) self.slow = ExponentialMovingAverage(200) self.is_uptrend = False self.scale = None def update(self, time, value): if self.fast.Update(time, value) and self.slow.Update(time, value): fast = self.fast.Current.Value slow = self.slow.Current.Value self.is_uptrend = (fast / slow) > 1.00 if self.is_uptrend: self.scale = (fast - slow) / ((fast + slow) / 2.0) def WarmUpIndicators(self, history): for tuple in history.itertuples(): self.fast.Update(tuple.index, tuple.close) self.slow.Update(tuple.index, tuple.close)

Here I'm running into my main problem. I don't know where to/how to apply the history loop to warmup each indicator as its added to the universe. In this example I'm getting the error:

Runtime Error: TypeError : object is not callable at CoarseSelectionFunction in main.py:line 20 TypeError : object is not callable (Open Stacktrace) I've even attempted to constantly keep the indicators updated in the OnData method like so: #OnData def OnData(self, data): '''OnData event is the primary entry point for your algorithm. Each new data point will be pumped in here.''' '''Arguments: data: Slice object keyed by symbol containing the stock data''' #Constantly update the universe moving averages and if a slice does not contain any data for the security, remove it from the universe if self.IsMarketOpen("SPY"): if bool(self.averages): for x in self.Securities.Values: if data.ContainsKey(x.Symbol): if data[x.Symbol] is None: continue avg = self.averages[x.Symbol] avg.update(data[x.Symbol].EndTime, data[x.Symbol].Open) else: self.RemoveSecurity(x.Symbol)

However, this isn't working. Perhaps I don't understand entirely, but I thought each hour a slice of data of be pumped through the indicators stored in the averages dictionary and so after 200 hours the indicators would be warmed up. I've tried this on a slow indicator of only 5 hours assuming it would be ready by the final hour, but this isn't the case.

Could someone with a deeper understanding explain my errors please? I've attached the project, for what its worth. Thank you in advance.

Update Backtest

Stephen Hyer

Accepted Answer

Update Backtest

Notebook

The material on this website is provided for informational purposes only and does not constitute an offer to sell, a solicitation to buy, or a recommendation or endorsement for any security or strategy, nor does it constitute an offer to provide investment advisory services by QuantConnect. In addition, the material offers no opinion with respect to the suitability of any security or specific investment. QuantConnect makes no guarantees as to the accuracy or completeness of the views expressed in the website. The views are subject to change, and may have become unreliable for various reasons, including changes in market conditions or economic circumstances. All investments involve risk, including loss of principal. You should consult with an investment professional before making any investment decisions.

Stephen Hyer

9.9k Pro ,

I was able to get this to work. Here are the required code lines to successfully warm up indicators as they're added to your universe dictionary that gets pushed into your SymbolData class.

1) In your initialize make sure you have your dictionary assigned.

class EmaCrossUniverseSelectionAlgorithm(QCAlgorithm):



    def Initialize(self):



        self.SetStartDate(2019,1,1)  #Set Start Date

        self.SetEndDate(datetime.now().date() - timedelta(1)) #Set End Date

        #self.SetEndDate(2013,1,1)    #Set End Date

        self.SetCash(150000)      #Set Strategy Cash



        self.UniverseSettings.Resolution = Resolution.Hour

        self.AddRiskManagement( MaximumDrawdownPercentPerSecurity() )



        self.marginRemaining = self.Portfolio.MarginRemaining

        self.averages = { };



        # this add universe method accepts two parameters:

        # - coarse selection function: accepts an IEnumerable<CoarseFundamental> and returns an IEnumerable<Symbol>

        self.AddEquity("SPY", Resolution.Hour)   

        self.AddUniverse(self.CoarseSelectionFunction)

2) In your CoarseSelectionFunction (I imagine this can also be done in the fine filter as long as its happening as you add the securities to your dictionary).

#Universe Filter            

    # sort the data by volume and price, apply the moving average crossver, and take the top 24 sorted results based on breakout magnitude

    def CoarseSelectionFunction(self, coarse):

        

        filtered = [ x for x in coarse if ( x.DollarVolume > 50000000 ) ]



        # We are going to use a dictionary to refer the object that will keep the moving averages

        for cf in filtered:

            if cf.Symbol not in self.averages:

                self.averages[cf.Symbol] = SymbolData(cf.Symbol)



            self.history = self.History(cf.Symbol, 200)

            

            if not self.history.empty:

                self.historySymbol = str(cf.Symbol) 

                self.averages[cf.Symbol].WarmUpIndicators(self.history)#.loc[self.historySymbol])

                

            # Updates the SymbolData object with current EOD price

            avg = self.averages[cf.Symbol]

            avg.update(cf.EndTime, cf.AdjustedPrice)



        # Filter the values of the dict: we only want up-trending securities

        values = list(filter(lambda x: x.is_uptrend, self.averages.values()))



        # Sorts the values of the dict: we want those with greater difference between the moving averages

        values.sort(key=lambda x: x.scale, reverse=True)

        

    #    for x in values[:100]:

    #        self.Log('symbol: ' + str(x.symbol.Value) + '  scale: ' + str(x.scale))        

            

        # we need to return only the symbol objects

        return [ x.symbol for x in values[:100] ]

2.1) I also include a backup in my OnSecuritiesChanged() method so that as securites are readded they're definitely warmed up. I don't know if this is actually necessary or if I'm redundantly warming them up twice.

    def OnSecuritiesChanged(self, changes):

        self.changes = changes

        

        # liquidate removed securities

        for security in changes.RemovedSecurities:

            if security.Invested:

                self.Liquidate(security.Symbol)

                

        for security in changes.AddedSecurities:

            symbolData = self.averages.get(security.Symbol)

            if symbolData is None:

                self.averages[security.Symbol] = SymbolData(security.Symbol)

            

            self.history = self.History(security.Symbol, 200)

            

            if not self.history.empty:

                self.historySymbol = str(security.Symbol) 

                self.averages[security.Symbol].WarmUpIndicators(self.history)

3) Last but no least you need your SymbolData class and coresponding methods.

#EMA Crossover Class

class SymbolData(object):

    

    def __init__(self, symbol):

        self.symbol = symbol

        self.fast = ExponentialMovingAverage(50)

        self.slow = ExponentialMovingAverage(200)

        self.is_uptrend = False

        self.scale = None



    def update(self, time, value):

        if self.fast.Update(time, value) and self.slow.Update(time, value):

            fast = self.fast.Current.Value

            slow = self.slow.Current.Value

            self.is_uptrend = (fast / slow) > 1.00



        if self.is_uptrend:

            

            self.scale = (fast - slow) / ((fast + slow) / 2.0)

            

    def WarmUpIndicators(self, history):

        for index, row in history.loc[str(self.symbol)].iterrows():

            self.fast.Update(index, row["close"])

            self.slow.Update(index, row["close"])

My next goal is to get all of this to work by pumping in consolidated hourly history bars from a universe resolution of minute. If anyone can help by posting a solution I would greatly appreciate it.

Link Liang

6.1k Pro ,

Hi Shyer,

Thanks for sharing your algorithm and experience! In fact, UniverseSettings.Resolution determines the resolution of data subscribed with universe selection, not the frequency of executing universe selection. In fact, Universe Selection is called before the market opens every day. Currently it is not possible to make the selection at a specific time.

Moreover, please avoid updating indicators twice. It actually feeds the indicator with the same data entry two times, which will result in wrong/inaccurate indicators. We don't need a "back up" of warming indicators.

Please let us know if you have further questions. Thanks for your support!

Stephen Hyer INVESTOR

Update Backtest

Notebook

To unlock posting to the community forums please complete at least 30% of Boot Camp.
You can continue your Boot Camp training progress from the terminal. We hope to see you in the community soon!

Platform

QuantConnect Community

Alpha League Competition: $1,000 Weekly Prize Pool

Filter Discussion by Tags

Community Ambassadors

Earn free QuantConnect cloud credits by helping others, each week we airdrop credits to the most active community members.

232,045 Quants.

Online Now

Stay Connected