book
Checkout our new book! Hands on AI Trading with Python, QuantConnect, and AWS Learn More arrow

Universe Selection

Fundamental Universes

Introduction

A FundamentalUniverseSelectionModel selects a universe of US Equities based on Fundamental data. Depending on the Fundamental properties you use, these universes rely on the US Equity Coarse Universe dataset, the US Fundamental dataset, or both.

These types of universes operate on daily schedule by default. In backtests, they select assets at midnight. In live trading, the selection timing depends on the data provider you use. To adjust the selection schedule, see Schedule.

If you use a fundamental Universe Selection model, the only way to unsubscribe from a security is to return a list from the selection function that doesn't include the security Symbol. The remove_security method doesn't work with these types of Universe Selection models.

Fundamental Selection

The FundamentalUniverseSelectionModel lets you select stocks based on corporate Fundamental data. You can specific the selection method, which takes a list of Fundamental objects as argument and returns a list of Symbol objects.

Select Language:
def initialize(self) -> None:
    # Run asynchronous universe selection to speed up your algorithm. 
    # In this case, you can't rely on the method or algorithm state between filter calls. 
    self.universe_settings.asynchronous = True
    self.add_universe_selection(FundamentalUniverseSelectionModel(self._fundamental_filter_function))

def _fundamental_filter_function(self, fundamental: List[Fundamental]):
    # Select symbols with fundamental data and a price above $10.
    filtered = [f for f in fundamental if f.price > 10 and not np.isnan(f.valuation_ratios.pe_ratio)]
    # Sort the assets in ascending order by their P/E ratio.
    sorted_by_pe_ratio = sorted(filtered, key=lambda f: f.valuation_ratios.pe_ratio)
    # Select the first 10 assets.
    return [f.symbol for f in sorted_by_pe_ratio[:10]]

The following table describes the arguments the model accepts:

ArgumentData TypeDescriptionDefault Value
selector Callable[[List[Fundamental]], List[Symbol]]Filter function to select assets based on fundamental data.
universe_settingsUniverseSettingsThe universe settings. If you don't provide an argument, the model uses the algorithm.universe_settings by default.None

The Fundamental objects have the following properties:

To move the selection function outside of the algorithm class, create a universe selection model that inherits the FundamentalUniverseSelectionModel class and override its select method.

Select Language:
# In the initialize method, enable asynchronous universe selection to speed up your algorithm.
self.universe_settings.asynchronous = True
# Add a custom universe selection model that selects undervalued, liquid stocks.
self.add_universe_selection(LiquidAndLowPEUniverseSelectionModel())

# Outside of the algorithm class, define the universe selection model.
class LiquidAndLowPEUniverseSelectionModel(FundamentalUniverseSelectionModel):
    def select(self, fundamental: List[Fundamental]) -> List[Symbol]:
        # Select symbols with fundamental data and a price above $1.
        filtered = [x for x in fundamental if x.price > 1 and not np.isnan(x.valuation_ratios.pe_ratio)]
        # Select the 100 assets with the greatest daily dollar volume.
        most_liquid = sorted(filtered, key=lambda x: x.dollar_volume, reverse=True)[:100]
        # Select the 10 assets with the lowest PE ratio.
        lowest_pe_ratio = sorted(most_liquid, key=lambda x: x.valuation_ratios.pe_ratio, reverse=True)[:10]
        return [x.Symbol for x in lowest_pe_ratio]

To return the current universe constituents from the selection function, return Universe.UNCHANGED.

To view the implementation of this model, see the LEAN GitHub repository.

EMA Cross Selection

The EmaCrossUniverseSelectionModel applies two exponential moving average (EMA) indicators to the price history of assets and then selects the assets that have their fast EMA furthest above their slow EMA on a percentage basis.

Select Language:
# Initialize asynchronous universe settings for faster processing and add EmaCrossUniverseSelectionModel to dynamically manage the universe based on EMA cross signals to identify trending assets.
def initialize(self) -> None:
    self.universe_settings.asynchronous = True   
    self.add_universe_selection(EmaCrossUniverseSelectionModel())

The following table describes the arguments the model accepts:

ArgumentData TypeDescriptionDefault Value
fast_periodintFast EMA period100
slow_periodintSlow EMA period300
universe_countintMaximum number of members of this universe selection500
universe_settingsUniverseSettingsThe universe settings. If you don't provide an argument, the model uses the algorithm.universe_settings by default.None

To view the implementation of this model, see the LEAN GitHub repository.

Examples

The following examples demonstrate some common practices for implementing the framework fundamental universe model.

Example 1: 50 Stocks >$10/Share and >$10M in Daily Trading Volume

The following algorithm selects the 50 most liquid US Equities above $10/share and $10M daily volume. We pass a function to the FundamentalUniverseSelectionModel to filter the fundamental data for stock selection.

Select Language:
class FrameworkFundamentalUniverseSelectionAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2019, 1, 1)
        self.set_end_date(2019, 12, 1)
        self.set_cash(100000)
        
        # Configure the universe to update at the start of each month. Most of the top 500 don't change very frequently.
        self.universe_settings.schedule.on(self.date_rules.month_start())
        # Add a universe with custom selection rules for filtering.
        self.add_universe_selection(FundamentalUniverseSelectionModel(
            lambda fundamental: [
                x.symbol for x in sorted(
                    # Large-cap non-penny stocks have a more stable trend and demand.
                    [f for f in fundamental if f.price > 10 and f.dollar_volume > 10_000_000], 
                    key=lambda f: f.dollar_volume
                )[-500:]
            ]
        ))

        # Sent insights on buying and holding the most liquid cryptos for a week.
        self.add_alpha(ConstantAlphaModel(InsightType.PRICE, InsightDirection.UP, timedelta(30)))
        # Evenly dissipate the capital risk among selected cryptos.
        self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())

Example 2: 10 Stocks Above Their 200-Day EMA With >$1B of Daily Trading Volume

Another common request is to filter the universe by a technical indicator, such as only picking stocks above their 200-day Exponential Moving Average (EMA). The Fundamental object has adjusted price and volume information so that you can do any price-related analysis. The following algorithm defines a separate class to contain the indicator of each asset. We pass a function to the FundamentalUniverseSelectionModel to filter the fundamental data for stock selection.

Select Language:
class FrameworkFundamentalUniverseSelectionAlgorithm(QCAlgorithm):
	# Create a dictionary to store the EMA data for universe selection.
    _selection_data_by_symbol = {}
	
    def initialize(self) -> None:
        self.set_start_date(2019, 1, 1)
        self.set_end_date(2019, 12, 1)
        self.set_cash(100000)
        
        # Add a universe with custom selection rules for filtering.
        self.add_universe_selection(FundamentalUniverseSelectionModel(self._select_assets))

        # Sent insights on buying and holding the most liquid cryptos for a week.
        self.add_alpha(ConstantAlphaModel(InsightType.PRICE, InsightDirection.UP, timedelta(1)))
        # Evenly dissipate the capital risk among selected cryptos.
        self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())

    def _select_assets(self, fundamental: List[Fundamental]) -> List[Symbol]:
        for f in fundamental:
            # Create/Update the EMA indicators of each stock.
            if f.symbol not in self._selection_data_by_symbol:
                self._selection_data_by_symbol[f.symbol] = SelectionData(self, f.symbol, 200)
            self._selection_data_by_symbol[f.symbol].update(f.end_time, f.adjusted_price, f.dollar_volume)
        
        # Select the Equities above their EMA and have a daily volume of $1B.
        # These assets are in an uptrend and are very liquid.
        selected = [x for x in self._selection_data_by_symbol.values() if x.is_above_ema and x.volume > 1_000_000_000]
            
        # Select the 10 most liquid Equities to avoid extra slippage.    
        return [ x.symbol for x in sorted(selected, key=lambda x: x.volume)[-10:] ]

# Create a separate class to contain the EMA information of each asset.
class SelectionData(object):
    def __init__(self, algorithm: QCAlgorithm, symbol: Symbol, period: int):
        # Create an EMA indicator for trend estimation and filtering.
        self.symbol = symbol
        self._ema = ExponentialMovingAverage(period)
        self.is_above_ema = False
        self.volume = 0

        # Warm up the EMA indicator.
        algorithm.warm_up_indicator(symbol, self._ema, Resolution.Daily);

    # Update your variables and indicators with the latest data.
    # You may also want to use the History API here to warm up the indicator.
    def update(self, time: datetime, price: float, volume: float) -> None:
        self.volume = volume
        if self._ema.update(time, price):
            self.is_above_ema = price > self._ema.current.value

In this example, the SelectionData class group variables for the universe selection and update the indicator of each asset. We highly recommend you follow this pattern to keep your algorithm tidy and bug-free. The following snippet shows an example implementation of the SelectionData class, but you can make whatever you need to store your custom universe filters. Note that the preceding SelectionData class uses a manual EMA indicator instead of the automatic version. For more information about universes that select assets based on indicators, see Indicator Universes. You need to use a SelectionData class instead of assigning the EMA to the Fundamental object because you can't create custom attributes on Fundamental objects like you can with Security objects.

Example 3: 10 Stocks Furthest Above their 10-day SMA of Volume

The process to get the 10-day Simple Moving Average (SMA) stock volume is the same process as in Example 2. First, define a SelectionData class that performs the averaging. This class tracks the ratio of today's volume relative to historical volumes. You can use this ratio to select assets above their 10-day SMA and sort the results by the Equities that have had the most significant jump since yesterday.

We inherit the FundamentalUniverseSelectionModel to create a custom universe selection model. Overriding the , we filter the fundamental data for stocks selection.

Select Language:
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel
      
class FrameworkFundamentalUniverseSelectionAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2019, 1, 1)
        self.set_end_date(2019, 12, 1)
        self.set_cash(100000)

        # Add a universe with custom selection rules for filtering.
        self.add_universe_selection(VolumeSMAUniverseSelectionModel(10))

        # Sent insights on buying and holding the selected securities.
        self.add_alpha(ConstantAlphaModel(InsightType.PRICE, InsightDirection.UP, timedelta(1)))
        # Evenly dissipate the capital risk among selected securities.
        self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())

class VolumeSMAUniverseSelectionModel(FundamentalUniverseSelectionModel):
	# Create a dictionary to store the EMA data for universe selection.
    _selection_data_by_symbol = {}

    def __init__(self, period: int) -> None:
        self.period = period

    def select(self, algorithm: QCAlgorithm, fundamental: List[Fundamental]) -> List[Symbol]:
        # Create/Update the volume SMA indicator of each stock.
        for f in fundamental:
            if f.symbol not in self._selection_data_by_symbol:
                self._selection_data_by_symbol[f.symbol] = self.SelectionData(algorithm, f.symbol, self.period)
            self._selection_data_by_symbol[f.symbol].update(f.end_time, f.adjusted_price, f.dollar_volume)

        # Select the Equities with higher trading volume than their SMA, indicating higher capital flow.
        selected = [sd for sd in self._selection_data_by_symbol.values() if sd.volume_ratio > 1]
            
        # Select the 10 Equities with the highest relative volume since they have the highest capacity
        # for scalp-trading or intra-day movement.
        return [ x.symbol for x in sorted(selected, key=lambda x: x.volume_ratio)[-10:] ]

    # Create a separate class to contain the EMA information of each asset.
    class SelectionData(object):
        def __init__(self, algorithm: QCAlgorithm, symbol: Symbol, period: int) -> None:
            self.symbol = symbol
            self.volume_ratio = 0
            # Create an SMA of volume to track the popularity of the stock.
            self._sma = SimpleMovingAverage(period)

            # Warm up the volume SMA indicator.
            algorithm.warm_up_indicator(symbol, self._sma, Resolution.DAILY, lambda data: data.volume)

        def update(self, time: datetime, price: float, volume: float) -> None:
            # Update the SMA with today's data and calculate the relative volume position for filtering.
            ready = self._sma.update(time, volume)
            sma = self._sma.current.value
            if ready and sma > 0:
                self.volume_ratio = volume / sma
                return ready
            return False

Example 4: 10 "Fastest Moving" Stocks With a 50-Day EMA > 200 Day EMA

You can construct complex universe filters with the SelectionData helper class pattern. To view a full example of this algorithm, see the EmaCrossUniverseSelectionAlgorithm in the LEAN GitHub repository or take the related Boot Camp lesson.

Select Language:
from Selection.FundamentalUniverseSelectionModel import FundamentalUniverseSelectionModel

class FrameworkFundamentalUniverseSelectionAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2019, 1, 1)
        self.set_end_date(2019, 12, 1)
        self.set_cash(100000)

        # Add a universe with custom selection rules for filtering.
        self.add_universe_selection(EmaUniverseSelectionModel(10))

        # Sent insights on buying and holding the selected securities.
        self.add_alpha(ConstantAlphaModel(InsightType.PRICE, InsightDirection.UP, timedelta(1)))
        # Evenly dissipate the capital risk among selected securities.
        self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())

class EmaUniverseSelectionModel(FundamentalUniverseSelectionModel):
	# Create a dictionary to store the EMA data for universe selection.
    averages = {}

    def __init__(self, selection_number: int) -> None:
        super().__init__()
        self.selection_number = selection_number

    def select(self, algorithm: QCAlgorithm, fundamental: List[Fundamental]) -> List[Symbol]:
        # We are going to use a dictionary to refer to the object that will keep the moving averages
        for f in fundamental:
            if f.symbol not in self.averages:
                self.averages[f.symbol] = self.SelectionData(algorithm, 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 differences between the moving average
        values.sort(key=lambda x: x.scale, reverse=True)

        # we need to return only the symbol objects
        return [ x.symbol for x in values[:self.selection_number] ]

    # class used to improve the readability of the fundamental selection function
    class SelectionData(object):
        def __init__(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
            self.symbol = symbol
            self.tolerance = 1.01
            self.fast = ExponentialMovingAverage(100)
            self.slow = ExponentialMovingAverage(300)
            self.is_uptrend = False
            self.scale = 0

            # Warm up the indicators.
            algorithm.warm_up_indicator(symbol, self.fast, Resolution.DAILY)
            algorithm.warm_up_indicator(symbol, self.slow, Resolution.DAILY)

        def update(self, time: datetime, price: float) -> None:
            if self.fast.update(time, price) and self.slow.update(time, price):
                fast = self.fast.current.value
                slow = self.slow.current.value
                self.is_uptrend = fast > slow * self.tolerance

            if self.is_uptrend:
                # computes an object score of how much larger the fast is than the slow
                self.scale = (fast - slow) / ((fast + slow) / 2.0)

Other Examples

For more examples, see the following algorithms:

You can also see our Videos. You can also get in touch with us via Discord.

Did you find this page helpful?

Contribute to the documentation: