How to Manage Dynamic Universes and Security Changes in Your Algorithm

To build a strategy that automatically adds or removes stocks based on market changes in QuantConnect, you need to implement Dynamic Universe Selection.

This allows the Lean Engine to filter thousands of stocks daily based on criteria like volume, price, or fundamental data (P/E ratio, Market Cap, Earnings), automatically updating the list of symbols your algorithm trades.

Key Components

1. Coarse Selection: Filters the entire market (8000+ stocks) by Price and Dollar Volume. This runs fast and ensures you only look at liquid assets.
2. Fine Selection: Filters the results of the Coarse selection using Fundamental data (Morningstar data).
3. OnSecuritiesChanged: An event handler that triggers whenever the universe adds or removes a stock. This is where you initialize indicators or liquidate positions.

Strategy Example: Quality Companies in an Uptrend

The following strategy automatically selects the top 10 stocks with a Market Cap over $1 Billion and a positive P/E ratio. It then buys them if they are above their 200-day Moving Average.

# region imports
from AlgorithmImports import *
# endregion

class DynamicUniverseStrategy(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2020, 1, 1)
        self.set_end_date(2023, 1, 1)
        self.set_cash(100000)

        # 1. Configure Universe Settings
        # This setting ensures that when a stock is added, we get Daily data for it.
        self.universe_settings.resolution = Resolution.DAILY
        
        # 2. Add the Universe
        # We pass two functions: one for coarse selection (price/volume) 
        # and one for fine selection (fundamentals).
        self.add_universe(self.coarse_selection_function, self.fine_selection_function)

        # Dictionary to hold indicators for our dynamic universe
        self._indicators = {}
        # Minimum days of history to warm up indicators
        self._lookback = 200

    def coarse_selection_function(self, coarse):
        """
        Filters the complete market data (8000+ stocks).
        Goal: Select liquid stocks with fundamental data available.
        """
        # Filter for stocks with price > $10 and that have fundamental data
        filtered = [x for x in coarse if x.has_fundamental_data and x.price > 10]
        
        # Sort by Dollar Volume (Price * Volume) to get the most liquid stocks
        sorted_by_dollar_volume = sorted(filtered, key=lambda x: x.dollar_volume, reverse=True)
        
        # Return the top 50 symbols to be passed to the Fine Selection function
        return [x.symbol for x in sorted_by_dollar_volume[:50]]

    def fine_selection_function(self, fine):
        """
        Filters the results of Coarse Selection using Fundamental Data.
        Goal: Select 'Quality' companies (High Market Cap, Positive P/E).
        """
        # Filter: Market Cap > 1 Billion and P/E Ratio > 0
        filtered = [x for x in fine if x.market_cap > 1e9 and x.valuation_ratios.pe_ratio > 0]
        
        # Sort by Market Cap descending
        sorted_by_cap = sorted(filtered, key=lambda x: x.market_cap, reverse=True)
        
        # Return the top 10 symbols to be added to the algorithm
        return [x.symbol for x in sorted_by_cap[:10]]

    def on_securities_changed(self, changes):
        """
        Event handler triggered when the Universe adds or removes securities.
        """
        # 1. Handle Removed Securities
        for security in changes.removed_securities:
            symbol = security.symbol
            # Liquidate the stock if we hold it
            if self.portfolio[symbol].invested:
                self.liquidate(symbol, "Removed from Universe")
            
            # Clean up the indicator to free memory
            if symbol in self._indicators:
                del self._indicators[symbol]

        # 2. Handle Added Securities
        for security in changes.added_securities:
            symbol = security.symbol
            # Create a 200-day Simple Moving Average for the new stock
            self._indicators[symbol] = self.sma(symbol, self._lookback, Resolution.DAILY)
            
            # Warm up the indicator immediately using historical data
            history = self.history(symbol, self._lookback, Resolution.DAILY)
            if not history.empty:
                for time, row in history.loc[symbol].iterrows():
                    self._indicators[symbol].update(time, row["close"])

    def on_data(self, data):
        """
        Main trading logic loop.
        """
        # Iterate through our active indicators
        for symbol, sma in self._indicators.items():
            
            # Ensure data exists for this symbol in the current slice
            if not data.contains_key(symbol) or data[symbol] is None:
                continue

            # Ensure indicator is ready
            if not sma.is_ready:
                continue

            price = data[symbol].close
            
            # Logic: Buy if Price > 200 SMA, Sell if Price < 200 SMA
            if price > sma.current.value:
                if not self.portfolio[symbol].invested:
                    # Allocate 10% of portfolio to this stock
                    self.set_holdings(symbol, 0.10)
            
            elif price < sma.current.value:
                if self.portfolio[symbol].invested:
                    self.liquidate(symbol, "Price below SMA")

How It Works

1. add_universe: This tells Lean to run the selection functions every day at midnight.
2. coarse_selection_function: Reduces 8,000+ stocks down to 50 based on liquidity. This is crucial for simulation speed.
3. fine_selection_function: Looks at the P/E ratio and Market Cap of those 50 stocks and picks the top 10.
4. on_securities_changed:
   • When a stock drops out of the top 10 (or loses fundamental criteria), it appears in changes.removed_securities. We liquidate it immediately.
   • When a new stock enters the top 10, it appears in changes.added_securities. We create a Moving Average indicator for it dynamically.
5. on_data: We loop through the currently selected stocks and apply technical analysis (Price vs. SMA) to decide entry/exit.

Common Questions (Q&A)

Q: Why do I need x.has_fundamental_data in the coarse selection?
A: Not all assets (like ETFs or penny stocks) have Morningstar fundamental data. If you try to access x.valuation_ratios in the Fine selection for a symbol that doesn't have data, the algorithm will crash. Filtering for it in Coarse selection prevents this.

Q: Can I change how often the universe updates?
A: Yes. By default, add_universe runs daily. You can use ScheduledUniverse to run selection weekly or monthly to reduce turnover and transaction costs.

Q: What happens to my open positions if a stock is removed from the universe?
A: The data feed for that stock stops shortly after it is removed from the universe. If you do not call self.liquidate() inside on_securities_changed, you will be left holding a "ghost" position that you cannot easily close because you are no longer receiving price updates for it.

Q: How do I debug which stocks are being selected?
A: You can add self.debug(f"Added: {security.symbol}") inside the on_securities_changed method to print the selected tickers to the console log.