Correct way of adding indicators on consolidated bars with QCAlgorithm

And since consolidated bar handler is the same for each symbol, do I have to check whether all symbols' data has been updated to the same time?

def OnDataConsolidated(...):
...
       # check if symbols' indicators' EndTimes are same
       indicator_end_times = list(map(lambda x: x.last_update_datetime, symbol_datas))
       if len(set(indicator_end_times)) > 1:
           # self.Log(f'Not all indicator end times are same: {set(indicator_end_times)}')
           return
...

Nottakumasato

If I were asked "what do I prefer to use to add indicators to consolidated bars?".

The subject of my preference would be dictionaries.

If you are satisfied with my answer, please accept it and don't forget to like it.

Hi Notta,

My 2 cents…

With just a slow and fast EMA per symbol, it does not matter much but the “SymbolData” way is generally recommended so for each symbol, in one object, you have encapsulated all your data and indicators. It is generally sound.

Now, in your first algo, in OnData, you do not check to see if there's data or not for each of your symbols before you do “symbol_data_single.consolidator.Update(data[ticker])”. You're gonna hit moments when you will Update with None.

In SymbolData.__init__, you also register self.fast twice with:

algorithm.RegisterIndicator(self.symbol, self.fast, self.consolidator)algorithm.RegisterIndicator(self.symbol, self.fast, self.consolidator)

Finally, you never call your SymbolData.update method. But you do register your indicators so you should not update them manually on top of it.

I just glanced over it but your update logic in the second one might be ok. You still don't do proper sanity checks in OnData however.

Fred

Vladimir thank you for the suggestion but I think I like the SymbolData way (seems more organized to me) 

Fred Painchaud thank you for the comments too. Would something like below work for checking data=None per ticker?

    def OnData(self, data: PythonSlice) -> None:
        for ticker in self.tickers:
            if data[ticker] is None:
                continue
            symbol_data_single = self.symbol_data[ticker]
            symbol_data_single.consolidator.Update(data[ticker])

Also as posted above, should I check whether all symbols' data has been updated to the same time?

def OnDataConsolidated(...):
...
       # check if symbols' indicators' EndTimes are same
       indicator_end_times = list(map(lambda x: x.last_update_datetime, symbol_datas))
       if len(set(indicator_end_times)) > 1:
           # self.Log(f'Not all indicator end times are same: {set(indicator_end_times)}')
           return
...

Hi Notta,

if ticker not in data or data[ticker] is None:
	continue

data[ticker] will exception if ticker is not in it.

Now that I spot it:

Also, the best would be not to map ticker strings to symboldata objects but symbols.

        for ticker in self.tickers:
            security = self.AddData(type=BinanceTradeBarData, ticker=ticker, resolution=Resolution.Minute)
            self.symbol_data[security.Symbol] = SymbolData(
                algorithm = self,
                symbol = security.Symbol,
                ema_fast = self.ema_fast, 
                ema_slow = self.ema_slow,
                consolidate_bar_count = self.consolidate_bar_count
                )

Hence the name SymbolData 😊. Tickers are to be used only at the beginning to sub to an asset, for instance. Then, it's recommended to work with symbols. When you use tickers, symbols are searched for anyway in the background in a cache.

Then you would iterate on your dictionary:

        for symbol in self.symbol_data:
            if symbol not in data or data[symbol] is None:
                continue

or if you also need values:

        for symbol, symbol_data in self.symbol_data.items():
            if symbol not in data or data[symbol] is None:
                continue

Re your other post, by “symbol_datas” I guess you mean “self.symbol_data” ? And “last_update_datetime” would be added to SymbolData.

That said, you want to do it like at the end of OnDataConsolidated? Bu that method will be called once per symbol subbed. You'd need to call that sanity check after all are updated. I would not do it. But well, later on, keep in mind your indie can become out of sync anytime. Say there is an issue with the data feed for a particular asset and thus you stop receiving new data for that asset alone…. So the sanity check in question is a good idea - I just implement all my sanity checks centrally, in a class of invariants I check anytime it makes sense. The reactions to detected errors are also centralized in that class. Anyway - that's all a matter of tastes and engineering.

Fred

Fred Painchaud added your suggestions:

• using Symbol as keys in the dictionary of SymbolData's
• checking if `symbol not in data` first, then the `data[symbol] is None` part

So `symbol_datas` are just the SymbolData instances for each symbol in universe. 

        symbol_datas = list(self.symbol_data.values())

Exactly like you said, the `last_update_datetime` is an attribute of SymbolData and updated when indicators are updated via `update_indicators()` in `OnDataConsolidated()`. 

class SymbolData(object):
    def __init__(self, algorithm:QCAlgorithm, symbol:Symbol, ema_fast:int=50, ema_slow:int=200, consolidate_bar_count:int = 5):
        self.symbol = symbol
        self.algorithm = algorithm
        # indicators
        self.fast = ExponentialMovingAverage(ema_fast)
        self.slow = ExponentialMovingAverage(ema_slow)
        # consolidator
        self.consolidator = TradeBarConsolidator(timedelta(minutes=consolidate_bar_count))
        self.consolidator.DataConsolidated += algorithm.OnDataConsolidated
        algorithm.SubscriptionManager.AddConsolidator(self.symbol, self.consolidator)

        self.signal = 0
        self.last_update_datetime = None

    def update_indicators(self, time, value):
        self.fast.Update(time, value)
        self.slow.Update(time, value)
        self.last_update_datetime = self.fast.Current.EndTime

One question though. Could you elaborate on `I just implement all my sanity checks centrally, in a class of invariants I check anytime it makes sense. The reactions to detected errors are also centralized in that class.`? If there is a small example of an algorithm you implemented this way, I would love to take a look as an example for my future algorithms!

Hey,

What I meant is that I implement my sanity checks in a SanityChecks class I define.

class SanityChecks:
	@staticmethod
	def sanity_check_1(...):
		...

	@staticmethod
	def sanity_check_2(...):
		...
	...
	@staticmethod
	def type_1(...):
		sanity_check_1(...)
		sanity_check_42(...)
	...
	@staticmethod
	def all(...):
		sanity_check_1(...)
		sanity_check_2(...)
		...
		sanity_check_42(...)
		...

That's part of the library I've been building but sharing it won't help as it all works on Numpy ndarrays, numba, ray, etc. I've re-implemented almost everything in the “middle” of LEAN. All “algo business logic”, more or less. I'm using the “data feeder” (History, Universes, manual subs) and the Orders/Broker proxies. So I'm like compatible (I can use Live nodes) but I don't use much of the class hierarchy in LEAN for algo implementation. I'd need to document it to a great extent if shared as it is documented simply for me not to get lost, so no time now. Sorry.

Fred

Nottakumasato,

--> I like the SymbolData way (seems more organized to me) 

When you're done developing with SymbolData, please don't forget 

to post a backtest to compare performance and readability.

Fred Painchaud I see, no worries at all, still a very good idea to have a class to just import your checks from. 

Vladimir yep will do. Have to fix a timeout issue when using remote files but will definitely post an example of a market neutral strategy that I have been working on.