Prompt One Shot Failure: Universe Selector Argument is not Really a List

[AlexCatarino]

Prompt

Build an algorithm that chains an ETF constituents universe with a fundamental universe to select SPY constituents trading above their 200-day SMA. In the fundamental callback, maintain a per-symbol helper class holding a 200-period SMA warmed up from history, updated with the adjusted price; remove indicators when symbols leave. Schedule an equal-weighted rebalance daily, 1 minute after market open. Seed initial prices so newly-added members can be traded immediately. Plot the count of possible vs selected symbols. Backtest from September 1, 2024 to December 31, 2024.

Error

The backtest ended with a runtime error.

Runtime Error: object of type 'OfTypeIterator[Fundamental]' has no len()
at _fundamental_filter
possible = len(fundamentals)
^^^^^^^^^^^^^^^^^
in main.py: line 70

object of type 'OfTypeIterator[Fundamental]' has no len()
at _fundamental_filter
possible = len(fundamentals)
^^^^^^^^^^^^^^^^^
in main.py: line 70

Code

main.py

from AlgorithmImports import *
from typing import List


class SpyConstituentsSmaAlgorithm(QCAlgorithm):

    def initialize(self) -> None:
        self.set_start_date(2024, 9, 1)
        self.set_end_date(2024, 12, 31)
        self.set_cash(1_000_000)

        self.settings.seed_initial_prices = True

        # Chain ETF constituents universe with a fundamental filter.
        self._etf_universe = self.add_universe(
            self.universe.etf("SPY", Market.USA, self.universe_settings,
                              self._etf_filter),
            self._fundamental_filter
        )

        # Per-symbol SMA helper instances keyed by Symbol.
        self._symbol_data: dict[Symbol, SymbolData] = {}

        # Schedule rebalance daily, 1 min after market open.
        self.schedule.on(
            self.date_rules.every_day("SPY"),
            self.time_rules.after_market_open("SPY", 1),
            self._rebalance
        )

        # Charts
        chart = Chart("Universe Count")
        chart.add_series(Series("Possible", SeriesType.LINE, 0))
        chart.add_series(Series("Selected", SeriesType.LINE, 0))
        self.add_chart(chart)

    # ------------------------------------------------------------------
    # Universe filters
    # ------------------------------------------------------------------

    def _etf_filter(self, constituents: List[ETFConstituentUniverse]) -> List[Symbol]:
        """Keep all SPY constituents."""
        return [c.symbol for c in constituents]

    def _fundamental_filter(self, fundamentals: List[Fundamental]) -> List[Symbol]:
        """
        Select constituents whose adjusted close is above their 200-day SMA.
        Maintain SymbolData helpers; remove stale ones.
        """
        incoming = {f.symbol for f in fundamentals}

        # Remove symbols no longer in the universe.
        for sym in list(self._symbol_data.keys()):
            if sym not in incoming:
                del self._symbol_data[sym]

        selected = []
        for f in fundamentals:
            sym = f.symbol
            if sym not in self._symbol_data:
                self._symbol_data[sym] = SymbolData(self, sym)
            sd = self._symbol_data[sym]
            price = f.adjusted_price
            if price and price > 0:
                sd.update(self.time, price)
            if sd.is_ready and price and sd.sma.current.value > 0:
                if price > sd.sma.current.value:
                    selected.append(sym)

        possible = len(fundamentals)
        self.plot("Universe Count", "Possible", possible)
        self.plot("Universe Count", "Selected", len(selected))

        return selected if selected else [fundamentals[0].symbol]

    # ------------------------------------------------------------------
    # Rebalance
    # ------------------------------------------------------------------

    def _rebalance(self) -> None:
        selected = list(self._etf_universe.selected)
        if not selected:
            return
        weight = 1.0 / len(selected)
        targets = [PortfolioTarget(sym, weight) for sym in selected]
        self.set_holdings(targets, liquidate_existing_holdings=True)


# ----------------------------------------------------------------------
# Helper class
# ----------------------------------------------------------------------

class SymbolData:
    """Holds a 200-period SMA for a single symbol, warmed up from history."""

    SMA_PERIOD = 200

    def __init__(self, algorithm: QCAlgorithm, symbol: Symbol) -> None:
        self._algorithm = algorithm
        self.sma = SimpleMovingAverage(self.SMA_PERIOD)

        # Warm up from daily history.
        history = algorithm.history(symbol, self.SMA_PERIOD, Resolution.DAILY)
        if not history.empty:
            for row in history.itertuples():
                close = getattr(row, "close", None)
                if close and close > 0:
                    self.sma.update(row.Index[1] if isinstance(row.Index, tuple) else row.Index, close)

    def update(self, time, price: float) -> None:
        self.sma.update(time, price)

    @property
    def is_ready(self) -> bool:
        return self.sma.is_ready