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

Universe Selection

ETF Constituents Universes

Introduction

The ETFConstituentsUniverseSelectionModel selects a universe of US Equities based on the constituents of an ETF. These Universe Selection models rely on the US ETF Constituents dataset. They run on a daily schedule by default. To adjust the selection schedule, see Schedule.

Add ETF Constituents Universe Selection

To add an ETFConstituentsUniverseSelectionModel to your algorithm, in the initialize method, call the add_universe_selection method. The ETFConstituentsUniverseSelectionModel constructor expects an ETF ticker.

Select Language: 
# Run universe selection asynchronously to speed up your algorithm.
self.universe_settings.asynchronous = True
self.add_universe_selection(ETFConstituentsUniverseSelectionModel("SPY"))

The following table describes the arguments the model accepts:

ArgumentData TypeDescriptionDefault Value
etf_tickerstringTicker of the ETF to get constituents for. To view the available ETFs, see Supported ETFs.
universe_settingsUniverseSettingsThe universe settings. If you don't provide an argument, the model uses the algorithm.universe_settings by default.None
universe_filter_funcCallable[[list[ETFConstituentUniverse]], list[Symbol]]Function to filter ETF constituents. If you don't provide an argument, the model selects all of the ETF constituents by default.None

If you provide a universe_filter_func argument, you can use the following attributes of the ETFConstituentUniverse objects to select your universe:

ETFConstituentUniverse Select Language:

The following example shows how to select the 10 Equities with the largest weight in the SPY ETF:

Select Language: 
# Initialize asynchronous settings for speed and use the ETFConstituentsUniverseSelectionModel 
# to select the top 10 SPY constituents by weight, focusing on blue-chip stocks with minimal risk.
def initialize(self) -> None:
    self.universe_settings.asynchronous = True   
    self.add_universe_selection(
        ETFConstituentsUniverseSelectionModel("SPY", universe_filter_func=self._etf_constituents_filter)
    )

def _etf_constituents_filter(self, constituents: list[ETFConstituentUniverse]) -> list[Symbol]:
    # Select the 10 largest Equities in the ETF.
    selected = sorted(
        [c for c in constituents if c.weight],
        key=lambda c: c.weight, reverse=True
    )[:10]
    return [c.symbol for c in selected]

To move the ETF Symbol and the selection function outside of the algorithm class, create a universe selection model that inherits the ETFConstituentsUniverseSelectionModel class.

Select Language: 
# Initialize asynchronous settings for speed and use the LargestWeightSPYETFUniverseSelectionModel 
# to select the top 10 blue-chip SPY constituents by weight, focusing on stocks with minimal risk.
self.universe_settings.asynchronous = True
self.add_universe_selection(LargestWeightSPYETFUniverseSelectionModel())

# Outside of the algorithm class
class LargestWeightSPYETFUniverseSelectionModel(ETFConstituentsUniverseSelectionModel):
    
    def __init__(self) -> None:
        super().__init__('SPY', universe_filter_func=self._etf_constituents_filter)

    def _etf_constituents_filter(self, constituents: list[ETFConstituentUniverse]) -> list[Symbol]:
        # Select the 10 largest Equities in the ETF.
        selected = sorted(
            [c for c in constituents if c.weight],
            key=lambda c: c.weight, reverse=True
        )[:10]
        return [c.symbol for c in selected]

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.

Examples

The following examples demonstrate some common practices for implementing the framework ETF constituent universe selection model.

Example 1: Weighted ETF Constituents

A subset of the SPY constituents outperform the SPY while many underperform the overall index. In an attempt to buy the ETF constituents that outperform the index, the following algorithm buys the top 50 weighted assets in the ETF. In this example, we will pass a function to the ETFConstituentsUniverseSelectionModel for selection.

Select Language: 
class FrameworkETFConstituentsUniverseSelectionAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2023, 6, 1)
        self.set_end_date(2023, 8, 1)
        self.set_cash(10000000)

        self.etf_weight_by_symbol = {}

        # Add a universe of the SPY constituents.
        self.add_universe_selection(
            ETFConstituentsUniverseSelectionModel("SPY", universe_filter_func=self._etf_constituents_filter)
        )
        # Add an Alpha model to trade based on the constituent weights.
        self.add_alpha(EtfAlphaModel(self))
        # Position sizing was handled by insight weight (from ETF weight).
        self.set_portfolio_construction(InsightWeightingPortfolioConstructionModel())

    def _etf_constituents_filter(self, constituents: list[ETFConstituentUniverse]) -> list[Symbol]:
        # Cache the constituent weights in a dictionary for filtering and position sizing.
        self.etf_weight_by_symbol = {x.symbol: x.weight for x in constituents if x.weight}
        # Select the 50 constituents with the largest weight in the ETF.
        # They should have positive excess returns.
        return [x[0] for x in sorted(self.etf_weight_by_symbol.items(), key=lambda x: x[1], reverse=True)[:50]]

class EtfAlphaModel(AlphaModel):
    def __init__(self, algorithm: FrameworkETFConstituentsUniverseSelectionAlgorithm) -> None:
        self._algorithm = algorithm
        self._day = -1

    def update(self, algorithm: QCAlgorithm, slice: Slice) -> list[Insight]:
        insights = []

        # Daily rebalance only.
        if self._day == slice.time.day:
            return insights

        for symbol, weight in self._algorithm.etf_weight_by_symbol.items():
            # Get the ETF weight of all the assets currently in the universe.  
            # To avoid trading errors, skip assets that have no price yet.
            if symbol in slice.bars:
                insights.append(Insight.price(symbol, timedelta(1), InsightDirection.UP, weight=weight))

        self._day = slice.time.day

        return insights

Example 2: Weight Trend Selection

The following algorithm trades the QQQ constituents with the upward trend of its weight constitution, indicated by an EMA indicator, suggesting its price trend is outperforming other constituents, or its weight was promoted by the NASDAQ index, which brings positive sentiment. To better utilize the EMA indicator, we inherit the ETFConstituentsUniverseSelectionModel superclass to create a custom universe selection model.

Select Language: 
class FrameworkETFConstituentsUniverseSelectionAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2023, 6, 1)
        self.set_end_date(2023, 8, 1)
        self.set_cash(10000000)

        # Add a universe of the SPY constituents.
        self.add_universe_selection(
            UptrendETFUniverseSelectionModel("QQQ")
        )
        # Add Alpha model to trade based on the selections.
        self.add_alpha(ConstantAlphaModel(InsightType.PRICE, InsightDirection.UP, timedelta(1)))
        # Equally invest in insights to dissipate the capital risk evenly.
        self.set_portfolio_construction(EqualWeightingPortfolioConstructionModel())

        # Warm up the indicators.
        self.set_warm_up(60, Resolution.DAILY)

class UptrendETFUniverseSelectionModel(ETFConstituentsUniverseSelectionModel):
    ema_by_symbol = {}

    def __init__(self, etf_ticker: str) -> None:
        super().__init__(etf_ticker, universe_filter_func=self.etf_constituent_filter)

    def etf_constituent_filter(self, constituents: list[ETFConstituentUniverse]) -> list[Symbol]:
        # Remove the ones that are not in the ETF anymore.
        to_remove = set(self.ema_by_symbol.keys()).difference(set([x.symbol for x in constituents]))
        for symbol in to_remove:
            del self.ema_by_symbol[symbol]

        for c in constituents:
            symbol = c.symbol
            if symbol in self.ema_by_symbol and c.weight:
                # Update EMA with the latest weight.
                self.ema_by_symbol[symbol].update(c.EndTime, c.weight)
            else:
                # Create an EMA to filter by trend.
                self.ema_by_symbol[symbol] = ExponentialMovingAverage(60)

        # Select the ones with the increasing trend of constituent weight.
        return [symbol for symbol, ema in self.ema_by_symbol.items() if ema.is_ready and ema.current.value > ema.previous.value]

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: