ETFConstituentsUniverseSelectionModel with 52weeks

Here's the corrected implementation for filtering stocks within 5% of their 52-week high using the ETFConstituentsUniverseSelectionModel:

class MyAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2023, 1, 1)
        self.set_cash(100000)
        self.universe_settings.resolution = Resolution.DAILY
        self.add_universe_selection(
            ETFConstituentsUniverseSelectionModel("SPY", self.universe_settings, self.etf_constituents_filter)
        )
        self.schedule.on(self.date_rules.every_day(), self.time_rules.before_market_close("SPY", 30), self.rebalance)

    def etf_constituents_filter(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]:
        # Get historical data for the last 52 weeks
        history = self.history([c.symbol for c in constituents], 252, Resolution.DAILY)

        selected_symbols = []
        for constituent in constituents:
            symbol = constituent.symbol
            if symbol not in history.index:
                continue

            # Calculate the 52-week high
            prices = history.loc[symbol]['close']
            high_52_week = prices.max()

            # Get the current price
            current_price = prices.iloc[-1]

            # Check if the current price is within 5% of the 52-week high
            if current_price >= 0.95 * high_52_week:
                selected_symbols.append(symbol)

        return selected_symbols

    def rebalance(self) -> None:
        # Logic to rebalance the portfolio based on the selected universe
        pass

Key Changes:

• Universe Settings: Set resolution instead of asynchronous as it is read-only.
• Scheduling: Changed before_market_open to before_market_close since before_market_open is not available.
• History Method: Ensure the history method is used correctly with the appropriate parameters.

Sources: ETF Constituents Universes - QuantConnect.com, E T F Constituents Framework Algorithm, ETF Constituents Universes - QuantConnect.com, US ETF Constituents - QuantConnect.com, E T F Constituent Universe R S I Alpha Model Algorithm

What is the rebalance implementation ? why we need ? how to implement add equity which are in the selected_symbol and how to implement on_data to iterate through only selected_symbols only ?

Hi Dharmesh

You don't necessarily need the rebalance scheduled event. Instead using on_data is perfectly fine. You don't need the add_equity method as well, since the data subscription of the universe members are handled automatically: when the security is in the universe, it will be subscribed; if it leaves the universe, it won't be subscribed anymore.

If you need to track or iterate the member of the universe, you can do the below:

class MyAlgorithm(QCAlgorithm):
    def initialize(self) -> None:
        self.set_start_date(2023, 1, 1)
        self.set_cash(100000)
        self.universe_settings.resolution = Resolution.DAILY
        self._universe = self.add_universe(self.universe.etf("SPY", Market.USA, self.universe_settings, self.etf_constituents_filter))
        
    def etf_constituents_filter(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]:
        # Get historical data for the last 52 weeks
        history = self.history([c.symbol for c in constituents], 252, Resolution.DAILY)
        selected_symbols = []
        
        for constituent in constituents:
            symbol = constituent.symbol
            if symbol not in history.index:
                continue
                
            # Calculate the 52-week high
            prices = history.loc[symbol]['close']
            high_52_week = prices.max()
            # Get the current price
            current_price = prices.iloc[-1]
            
            # Check if the current price is within 5% of the 52-week high
            if current_price >= 0.95 * high_52_week:
                selected_symbols.append(symbol)
                
        return selected_symbols
        
    def on_data(self, slice):
        # iterate the universe members
        for symbol in self._universe.members.keys():
            pass

Best
Louis