Question for Mia AI Bo

from AlgorithmImports import *
from datetime import datetime, timedelta, date
class ImmediateUniversePopulationAlgorithm(QCAlgorithm):
    """
    Example algorithm that

    1. Pre-loads (boot-straps) a universe of ≤900 equities in `Initialize` using
       historical CoarseFundamental data, so symbols are available the moment
       trading starts (even mid-week in live trading).
    2. Re-applies the identical coarse dollar-volume filter every Monday after
       the market closes, preparing the universe for the next week.
    3. Uses guards/flags to ensure the same selection logic never runs twice on
       the same day.
    4. Retires the bootstrap universe after the first scheduled refresh – from
       week 2 onward only the scheduled logic drives the universe.
    5. Relies exclusively on built-in QuantConnect APIs – ready for cloud live
       deployment.
    """

    # --------------------------------------------------------------------- #
    # INITIALIZATION                                                        #
    # --------------------------------------------------------------------- #
    def initialize(self) -> None:
        # Engine settings
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(100_000)

        # Coarse-selection parameters
        self.universe_settings.resolution = Resolution.DAILY
        self._max_universe_size: int = 900
        self._bootstrap_completed: bool = False          # flips after 1st refresh
        self._last_universe_update: date = date.min      # prevents duplicate runs

        # One always-present symbol used for scheduling
        self._sentinel: Symbol = self.add_equity("SPY", Resolution.DAILY).symbol

        # --------- 1) Immediate bootstrap universe -------------------------
        reference_time: datetime = self.start_date - timedelta(days=1)  # yesterday
        self._initial_symbols: list[Symbol] = self._get_top_dollar_volume_symbols(reference_time)
        self.debug(f"Boot-strapping {len(self._initial_symbols)} equities…")

        for symbol in self._initial_symbols:
            if symbol != self._sentinel:      # SPY already added
                self.add_equity(symbol.value, Resolution.DAILY)

        # --------- 2) Weekly scheduled refresh -----------------------------
        # Monday, 5 minutes after the close
        self.schedule.on(
            self.date_rules.every(DayOfWeek.MONDAY),
            self.time_rules.after_market_close(self._sentinel, 5),
            self._refresh_weekly_universe
        )

    # --------------------------------------------------------------------- #
    # COARSE SELECTION HELPERS                                              #
    # --------------------------------------------------------------------- #
    def _get_top_dollar_volume_symbols(self, reference_time: datetime) -> list[Symbol]:
        """
        Return ≤`_max_universe_size` symbols with the highest dollar volume for the
        most recent trading day ≤ `reference_time`.
        """
        if reference_time <= datetime.min:
            raise ValueError("reference_time must be a valid datetime")

        # Single-day history request for CoarseFundamental data
        coarse_history = self.history(
            CoarseFundamental,
            reference_time,
            reference_time,
            Resolution.DAILY
        )

        coarse_list: list[CoarseFundamental] = []
        if not coarse_history.empty:
            # DataFrame index: (time, symbol) – the object lives in column 'value'
            for _, row in coarse_history.iterrows():
                coarse = row["value"]
                if coarse and coarse.has_fundamental_data:
                    coarse_list.append(coarse)

        top = sorted(
            coarse_list,
            key=lambda c: c.dollar_volume if c.dollar_volume else 0.0,
            reverse=True
        )[: self._max_universe_size]

        return [c.symbol for c in top]

    # --------------------------------------------------------------------- #
    # WEEKLY UNIVERSE MAINTENANCE                                           #
    # --------------------------------------------------------------------- #
    def _refresh_weekly_universe(self) -> None:
        """Re-selects the universe every Monday after the close."""
        # Guard – ensure we run at most once per calendar day
        if self.time.date() == self._last_universe_update:
            return

        self.debug(f"Weekly universe refresh – {self.time}")

        new_symbols: list[Symbol] = self._get_top_dollar_volume_symbols(self.time)

        # Always retain the sentinel
        desired: set[Symbol] = set(new_symbols).union({self._sentinel})
        current: set[Symbol] = set(list(self.active_securities.keys))

        to_add = desired.difference(current)
        to_remove = current.difference(desired)

        # --- Add new securities ------------------------------------------
        for symbol in to_add:
            if not self.securities.contains_key(symbol):
                self.add_equity(symbol.value, Resolution.DAILY)

        # --- Remove obsolete securities ----------------------------------
        for symbol in to_remove:
            if symbol != self._sentinel:      # never drop sentinel
                self.remove_security(symbol)

        self.debug(f"Added {len(to_add)} | Removed {len(to_remove)}")

        # Mark bootstrap step as done after first scheduled run
        if not self._bootstrap_completed:
            self._bootstrap_completed = True

        self._last_universe_update = self.time.date()

    # --------------------------------------------------------------------- #
    # ONDATA – NO TRADING LOGIC INCLUDED                                    #
    # --------------------------------------------------------------------- #
    def on_data(self, data: Slice) -> None:
        # Strategy logic would go here; this demo focuses on universe management.
        pass

Key takeaways & workflow:

• The universe is immediately bootstrapped using self.history() to fetch the current top stocks by dollar volume.
• The same logic is re-applied weekly using a scheduled event after market close each Monday.
• A guard flag ensures the scheduled logic isn’t run twice during the first week or on Mondays where the algorithm starts after the close.
• A “sentinel” equity (SPY) is always present to facilitate scheduling and universe stability.
• After the first scheduled update, only the scheduled weekly refresh remains in charge of the universe.

This is best-practice for immediate and scheduled universe updates on QuantConnect Cloud/live, as per your requirements!

Sources: US Equity Coarse Universe - QuantConnect.com, Fundamental Universes - QuantConnect.com, docs writing algorithms universes dataless scheduled universes, Key Concepts - QuantConnect.com, True Beats - QuantConnect.com