--- title: "Indicator Universes" description: "An indicator universe uses technical indicators to determine the constituents of the universe. Image a universe of assets all above their 10-day SMA." canonical: https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/indicator-universes markdown: https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/indicator-universes.md language: python source: https://github.com/QuantConnect/Documentation/tree/master/03%20Writing%20Algorithms/28%20Indicators/08%20Indicator%20Universes modified: 2026-10-30 index: https://www.quantconnect.com/llms.txt --- # Indicator Universes [Documentation](https://www.quantconnect.com/docs/v2.md) > [Writing Algorithms](https://www.quantconnect.com/docs/v2/writing-algorithms.md) > Indicators > Indicator Universes > Code examples are shown in Python. Append `?lang=csharp` to this URL for the C# version of this page. ## 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](https://www.quantconnect.com/docs/v2/writing-algorithms/universes/key-concepts.md). 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](https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/manual-indicators.md) so you can ensure they only update during universe selection. ```python 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](https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/indicator-universes.md#02-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. ```python 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(2024, 9, 1) self.set_end_date(2024, 12, 31) self.set_cash(100000) 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 and returns an IEnumerable 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](https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/supported-indicators/maximum.md) indicator to do so. Then, we hold the stocks with price > [EMA](https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/supported-indicators/exponential-moving-average.md) > [SMA](https://www.quantconnect.com/docs/v2/writing-algorithms/indicators/supported-indicators/simple-moving-average.md), which indicates an upward accelerating trend. ```python class ManualIndicatorAlgorithm(QCAlgorithm): maximum_by_symbol = {} def initialize(self) -> None: self.set_start_date(2024, 9, 1) self.set_end_date(2024, 12, 31) # 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) ``` ### Example 2: Reset Indicators on Corporate Actions The following example shows how to select a universe of US Equities based on their price and SMA indicator. The `SelectionData` class keeps track of the SMA indicator for each stock in the universe dataset. When a split or dividend occurs for a stock, the data in its indicator becomes invalid because it doesn't account for the price adjustments that the split or dividend causes. The `SelectionData` class resets and warms up the indicator with the `SCALED_RAW` [data normalization mode](https://www.quantconnect.com/docs/v2/writing-algorithms/securities/asset-classes/us-equity/requesting-data.md#11-Data-Normalization), which gives you accurate indicator values to use in your universe selection after each corporate action. ```python class EquityIndicatorUniverseSelectionAlgorithm(QCAlgorithm): def initialize(self) -> None: self.set_start_date(2024, 9, 1) self.set_end_date(2024, 12, 31) self.settings.seed_initial_prices = True # Add a universe of US Equities based on an indicator. self._selection_data_by_symbol = {} self._universe = self.add_universe(self._select_assets) # Add a warm-up period to warm up the indicators in the universe selection. self.set_warm_up(timedelta(60)) def _select_assets(self, fundamentals): # Update the indicator of all stocks in the universe dataset and # get the subset of stocks that have their indicator ready. ready_stocks = [ f for f in fundamentals if self._selection_data_by_symbol.setdefault(f.symbol, SelectionData(self, f)).update(f) ] # As assests leave the Fundamental dataset, delete their SelectionData object. for symbol in self._selection_data_by_symbol.keys() - {f.symbol for f in fundamentals}: del self._selection_data_by_symbol[symbol] # During warm-up, keep the universe empty. if self.is_warming_up: return [] # Select a subset of the stocks based on the indicator. # Example: 10 stocks furthest above their SMA. factor_by_symbol = { f.symbol: f.price / self._selection_data_by_symbol[f.symbol].indicator.current.value for f in ready_stocks } return sorted( {k: v for k, v in factor_by_symbol.items() if v > 0}, key=lambda symbol: factor_by_symbol[symbol] )[-100:] class SelectionData: def __init__(self, algorithm, f): self._algorithm = algorithm self._price_scale_factor = f.price_scale_factor self.indicator = SimpleMovingAverage(21) def update(self, f): # If there hasn't been a split or dividend since the last trading # day, just update the indicator like normal. if f.price_scale_factor == self._price_scale_factor: return self.indicator.update(f.end_time, f.price) # Otherwise, reset the indicator and warm it up with the new # adjusted history. self._price_scale_factor = f.price_scale_factor self.indicator.reset() history = self._algorithm.history[TradeBar]( f.symbol, self.indicator.warm_up_period, Resolution.DAILY, data_normalization_mode=DataNormalizationMode.SCALED_RAW ) for bar in history: self.indicator.update(bar) return self.indicator.is_ready ``` ### Other Examples For more examples, see the following algorithms: **Demonstration Algorithms** - [EmaCrossUniverseSelectionAlgorithm.py](https://github.com/QuantConnect/Lean/blob/master/Algorithm.Python/EmaCrossUniverseSelectionAlgorithm.py) (Python)