Prompt One Shot Failure: ETFConstituentsUniverseDefinitionExtensions Doesn't Exist

[AlexCatarino]

Prompt

Build an algorithm that selects the 10 largest-weighted QQQ constituents and adds front-month at-the-money call options on each. The ETF filter sorts constituents by weight descending and returns the top 10. Chain an option universe on top, restricted to front-month, calls-only, with strikes within roughly two steps of the ATM strike. Use asynchronous selection and raw data normalization for fair strike comparison. In data handling, iterate each option chain and read each contract's last price. Set $100,000 cash. Backtest from September 1, 2024 to December 31, 2024.

Error

The backtest ended with a runtime error.

During the algorithm initialization, the following exception has occurred: name 'ETFConstituentsUniverseDefinitionExtensions' is not defined
at initialize
ETFConstituentsUniverseDefinitionExtensions.etf(qqq, self.universe_settings, self._qqq_constituent_selector)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
in main.py: line 20
name 'ETFConstituentsUniverseDefinitionExtensions' is not defined

name 'ETFConstituentsUniverseDefinitionExtensions' is not defined
at initialize
ETFConstituentsUniverseDefinitionExtensions.etf(qqq, self.universe_settings, self._qqq_constituent_selector)
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
in main.py: line 20
name 'ETFConstituentsUniverseDefinitionExtensions' is not defined

Code

main.py

from AlgorithmImports import *
from typing import List


class QQQTopTenCallOptions(QCAlgorithm):

    def initialize(self):
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(100_000)

        self.settings.seed_initial_prices = True

        self.universe_settings.resolution = Resolution.DAILY
        self.universe_settings.asynchronous = True
        self.universe_settings.data_normalization_mode = DataNormalizationMode.RAW

        qqq = Symbol.create("QQQ", SecurityType.EQUITY, Market.USA)
        self._equity_universe = self.add_universe(
            ETFConstituentsUniverseDefinitionExtensions.etf(qqq, self.universe_settings, self._qqq_constituent_selector)
        )

        # Track option subscriptions per underlying
        self._option_symbols: dict[Symbol, Symbol] = {}

    def _qqq_constituent_selector(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]:
        """Return the top 10 QQQ constituents sorted by weight descending."""
        sorted_by_weight = sorted(
            [c for c in constituents if c.weight is not None],
            key=lambda c: c.weight,
            reverse=True
        )
        return [c.symbol for c in sorted_by_weight[:10]]

    def on_securities_changed(self, changes: SecurityChanges):
        # Add option chains for newly added equities
        for security in changes.added_securities:
            underlying = security.symbol
            if underlying.security_type != SecurityType.EQUITY:
                continue
            if underlying not in self._option_symbols:
                option = self.add_option(underlying.value, resolution=Resolution.DAILY)
                option.set_filter(self._option_filter)
                option.price_model = OptionPriceModels.black_scholes()
                self._option_symbols[underlying] = option.symbol

        # Remove option subscriptions for removed equities
        for security in changes.removed_securities:
            underlying = security.symbol
            if underlying in self._option_symbols:
                canonical = self._option_symbols.pop(underlying)
                self.remove_option_contract(canonical)

    def _option_filter(self, option_filter_universe: OptionFilterUniverse) -> OptionFilterUniverse:
        """Filter to front-month ATM calls within ~2 strikes."""
        return (
            option_filter_universe
            .front_month()
            .calls_only()
            .strikes(-2, 2)
        )

    def on_data(self, data: Slice):
        for canonical, chain in data.option_chains.items():
            for contract in chain:
                self.log(
                    f"{self.time} | {contract.symbol.value} | "
                    f"Last: {contract.last_price:.2f} | "
                    f"Strike: {contract.strike} | "
                    f"Expiry: {contract.expiry.date()}"
                )