Indicators
Indicator Universes
Introduction
An indicator universe uses technical indicators to determine the constituents of the universe. Imagine a universe that only contains assets above their 10-day simple moving average. You can incorporate indicators into any of the types of universes in the Universes chapter. To create an indicator universe, define a helper class that contains the indicators and then define a universe that updates the indicators and selects assets.
Define SymbolData Objects
To make it easy to create and update indicators for each security in the universe, move the indicator logic into a class. In the universe definition, you can create an instance of this class for each security in the universe. The indicators you create in this class should be manual indicators so you can ensure they only update during universe selection.
class SymbolData(object): def __init__(self, symbol): self._symbol = symbol self.tolerance = 1.01 self.fast = ExponentialMovingAverage(100) self.slow = ExponentialMovingAverage(300) self.is_uptrend = False self.scale = 0 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 * self.tolerance if self.is_uptrend: self.scale = (fast - slow) / ((fast + slow) / 2.0)
You need to use a SymbolData
class instead of assigning the indicators to the Fundamental
object because you can't create custom attributes on Fundamental
objects like you can with Security
objects.
Define the Universe
You need to define SymbolData objects before you define the universe that selects securities.
When your universe function receives an object that contains all the possible securities, create a SymbolData
object for each new security and update the remaining SymbolData
objects with their daily price or some other data point. For example, the following universe definition selects US Equities that have the greatest difference between two moving averages.
class EmaCrossUniverseSelectionAlgorithm(QCAlgorithm): def initialize(self) -> None: '''Initialise the data and resolution required, as well as the cash and start-end dates for your algorithm. All algorithms must initialized.''' self.set_start_date(2010,1,1) #Set Start Date self.set_end_date(2015,1,1) #Set End Date self.set_cash(100000) #Set Strategy Cash self.universe_settings.asynchronous = True self.universe_settings.resolution = Resolution.DAILY self.universe_settings.leverage = 2 self.count = 10 self.averages = { } # this add universe method accepts two parameters: # - fundamental selection function: accepts an IEnumerable<Fundamental> and returns an IEnumerable<Symbol> self.add_universe(self.fundamental_selection_function) # sort the data by daily dollar volume and take the top 'NumberOfSymbols' def fundamental_selection_function(self, fundamental: List[Fundamental]) -> List[Symbol]: # We are going to use a dictionary to refer the object that will keep the moving averages for f in fundamental: if f.symbol not in self.averages: self.averages[f.symbol] = SymbolData(f.symbol) # Updates the SymbolData object with current EOD price avg = self.averages[f.symbol] avg.update(f.end_time, f.adjusted_price) # 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[:self.count]: 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[:self.count] ]
Examples
The following examples demonstrate some common practices for using indicator universes.
Example 1: Universe Selection
The following algorithm selects the stocks within 5% of the 1-year maximum price among the top 100 liquid stocks. To do so, we make use of Maximum indicator to do so. Then, we hold the stocks with price > EMA > SMA, which indicates an upward accelerating trend.
class ManualIndicatorAlgorithm(QCAlgorithm): maximum_by_symbol = {} def initialize(self) -> None: self.set_start_date(2021, 1, 1) self.set_end_date(2021, 2, 1) # Select a popularity-based universe with indicators in a Selection function. self._universe = self.add_universe(self.selection) # Set a scheduled event to rebalance daily on the daily indicator signals. self.schedule.on( self.date_rules.every_day(), self.time_rules.at(9, 31), self.rebalance ) def selection(self, fundamentals: List[Fundamental]) -> List[Symbol]: selected = [] # Initially filtered for the top 100 liquid stocks first. filtered = sorted(fundamentals, key=lambda f: f.dollar_volume, reverse=True)[:100] for f in filtered: if f.symbol not in self.maximum_by_symbol: self.maximum_by_symbol[f.symbol] = Maximum(252) # Warm up the Maximum indicator to its readiness to use immediately. history = self.history[TradeBar](f.symbol, 252, Resolution.DAILY) for bar in history: self.maximum_by_symbol[f.symbol].update(bar.end_time, bar.close) else: # Update the indicator with the last known adjusted price daily. self.maximum_by_symbol[f.symbol].update(f.end_time, f.adjusted_price) # Select to trade if the current price is within 5% of the maximum price of the last year. # Close to the maximum price provides evidence of high popularity for the fund to support the trend. if f.adjusted_price >= self.maximum_by_symbol[f.symbol].current.value * 0.95: selected.append(f.symbol) return selected def rebalance(self) -> None: def to_buy(symbol): security = self.securities[symbol] # Buy the stocks whose prices are above the EMA and the EMA is above the SMA, meaning their trend is upward accelerating. return security.price > security.ema.current.value > security.sma.current.value symbols_to_buy = [symbol for symbol in self._universe.selected if to_buy(symbol)] # Equally invest in the selected stocks to dissipate the capital risk evenly. count = len(symbols_to_buy) if count > 0: targets = [PortfolioTarget(symbol, 1 / count) for symbol in symbols_to_buy] # Liquidate the positions that are not on an upward trend or are not popular anymore. self.set_holdings(targets, liquidate_existing_holdings=True) def on_securities_changed(self, changes: SecurityChanges) -> None: for added in changes.added_securities: symbol = added.symbol # Create an EMA and an SMA manual indicator for trade signal generation. added.ema = ExponentialMovingAverage(20) added.sma = SimpleMovingAverage(20) # Warm up the indicators to ensure their readiness to use them immediately. self.warm_up_indicator(symbol, added.ema, Resolution.DAILY) self.warm_up_indicator(symbol, added.sma, Resolution.DAILY) # Subscribe to the indicators to update daily price data for updated trade signal generation. self.register_indicator(symbol, added.ema, Resolution.DAILY) self.register_indicator(symbol, added.sma, Resolution.DAILY) for removed in changes.removed_securities: # Cancel data subscription to release computation resources when leaving the universe. self.deregister_indicator(removed.ema) self.deregister_indicator(removed.sma)
Other Examples
For more examples, see the following algorithms: