Help mia

from AlgorithmImports import *
import json
from datetime import timedelta
from collections import deque
class LargeUniverseNewsMonitor(QCAlgorithm):
    """
    Best-practice template that

      1. Builds a large (800-symbol) fundamental universe, stores it in
         ObjectStore, and refreshes it once a week.
      2. Subscribes to BenzingaNews for every universe member, processes
         headlines in real-time, skips duplicates, and ignores news older
         than 5 minutes.
      3. Adds/removes data subscriptions efficiently in OnSecuritiesChanged.

    Notes (see inline comments for details):
        • Data-subscription hard-limit ≈ 2000.  800 equities × (price+news)
          stays within the cap.
        • ObjectStore key payload (<100 kB) is far below the 10 MB limit.
        • De-duplication cache capped at 10 000 IDs to prevent memory bloat.
        • When live, consider batching/throttling alerts to respect provider
          and messaging limits.
    """

    # --- USER SETTINGS ---------------------------------------------------
    _EMAIL         = "user@email.com"    # change if you really want e-mails
    _universe_size = 800                 # target number of symbols
    # ---------------------------------------------------------------------

    def initialize(self) -> None:
        self.set_start_date(2021, 1, 1)
        self.set_end_date(2022, 1, 1)
        self.set_cash(1_000_000)

        # ---------- internal state ----------
        self._store_key: str                 = "filtered_universe_v1"
        self._symbols_from_store: bool       = False
        self._universe_symbols: list[Symbol] = []
        self._universe_tickers: set[str]     = set()
        self._refresh_flag: bool             = False   # set True by scheduler

        self._news_subscriptions: set[Symbol] = set()
        self._processed_news_ids: deque[str]  = deque(maxlen=10_000)

        self.universe_settings.resolution = Resolution.MINUTE

        # 1) fast boot  ––––––––––––––––––––––––––––––––––––––––––––––––––––
        self._load_universe_from_store()

        # 2) universe selection  –––––––––––––––––––––––––––––––––––––––––––
        self.add_universe(self._coarse, self._fine)

        # 3) schedule weekly refresh  ––––––––––––––––––––––––––––––––––––––
        self.schedule.on(
            self.date_rules.every(DayOfWeek.MONDAY),
            self.time_rules.before_market_open("SPY", 30),
            self._flag_universe_refresh
        )

    # ======================= ObjectStore helpers ==========================
    def _load_universe_from_store(self) -> None:
        """Fast-start: read ticker list from ObjectStore (if any)."""
        if not self.object_store.contains_key(self._store_key):
            return
        try:
            raw       = self.object_store.read(self._store_key)
            tickers   = json.loads(raw)
            if not tickers:
                return
            self.debug(f"Loaded {len(tickers):,} symbols from ObjectStore")
            self._symbols_from_store = True
            self._universe_tickers   = set(tickers)

            # Add equities & news feeds right away
            for tkr in tickers:
                symbol = self.add_equity(tkr, Resolution.MINUTE).symbol
                self._universe_symbols.append(symbol)
                self._subscribe_news(symbol)
        except Exception as err:
            self.error(f"ObjectStore load failed: {err}")

    def _save_universe_to_store(self, tickers: list[str]) -> None:
        """Persist ticker list so the next back-test/live run starts fast."""
        try:
            self.object_store.save(self._store_key, json.dumps(tickers))
            self.debug(f"Persisted {len(tickers):,} tickers to ObjectStore")
        except Exception as err:
            self.error(f"ObjectStore save failed: {err}")

    # ======================= Universe selection ===========================
    def _flag_universe_refresh(self) -> None:
        self._refresh_flag = True

    def _coarse(self, coarse: list[CoarseFundamental]) -> list[Symbol]:
        """Skip work unless a refresh is requested."""
        if not self._refresh_flag and self._universe_symbols:
            return self._universe_symbols          # no changes – super fast

        liquid = [c for c in coarse if c.has_fundamental_data and c.price > 5]
        liquid.sort(key=lambda c: c.dollar_volume, reverse=True)
        return [c.symbol for c in liquid[:1_000]]   # pass shortlist to _fine

    def _fine(self, fine: list[FineFundamental]) -> list[Symbol]:
        if not self._refresh_flag and self._universe_symbols:
            return self._universe_symbols

        picked: list[tuple[Symbol, float]] = []
        for f in fine:
            try:
                if (f.earning_reports.basic_eps.twelve_months > 0 and
                        0 < f.valuation_ratios.pe_ratio < 25 and
                        f.operation_ratios.current_ratio.one_year > 1):
                    picked.append((f.symbol, f.market_cap))
            except AttributeError:
                continue

        picked.sort(key=lambda kv: kv[1], reverse=True)
        symbols = [kv[0] for kv in picked[:self._universe_size]]
        tickers = [s.value for s in symbols]

        # cache + persist
        self._universe_symbols = symbols
        self._universe_tickers = set(tickers)
        self._save_universe_to_store(tickers)

        self._refresh_flag = False
        return symbols

    # ======================= News helpers =================================
    def _subscribe_news(self, symbol: Symbol) -> None:
        if symbol in self._news_subscriptions:
            return
        try:
            self.add_data(BenzingaNews, symbol, Resolution.MINUTE)
            self._news_subscriptions.add(symbol)
        except Exception as err:
            self.debug(f"News subscription failed for {symbol}: {err}")

    def _unsubscribe_news(self, symbol: Symbol) -> None:
        if symbol not in self._news_subscriptions:
            return
        try:
            self.remove_security(symbol)            # drops price + news
            self._news_subscriptions.remove(symbol)
        except Exception as err:
            self.debug(f"Un-subscribe failed for {symbol}: {err}")

    # ======================= Slice processing =============================
    def on_data(self, data: Slice) -> None:
        news = data.get(BenzingaNews)
        if not news:
            return

        for article in news.values():
            # 1) avoid duplicates
            if article.id in self._processed_news_ids:
                continue
            # 2) ignore headlines older than 5 min
            if (self.time - article.end_time) > timedelta(minutes=5):
                continue
            # 3) make sure the headline concerns at least one universe member
            if not any(t in self._universe_tickers for t in article.tickers):
                continue

            headline = f"{article.time:%H:%M} | {', '.join(article.tickers)} | {article.title}"
            self.debug(f"NEWS: {headline}")
            # Optional live alert:
            # self.notify.email(self._email, "Fresh Benzinga headline", headline)

            self._processed_news_ids.append(article.id)

    # ======================= Keep subscriptions in-sync ===================
    def on_securities_changed(self, changes: SecurityChanges) -> None:
        for sec in changes.added_securities:
            self._subscribe_news(sec.symbol)
        for sec in changes.removed_securities:
            self._unsubscribe_news(sec.symbol)

Best Practice Notes for مطلق:

• Universe selection: Use a custom universe for filtering, not manual AddEquity for hundreds of symbols.
• ObjectStore: Store the ticker list as JSON under a single key, e.g., "filtered_universe_v1", and reload at startup for fast launch. Update the store weekly on a timer.
• OnSecuritiesChanged: Add only subscriptions for new symbols, remove them for dropped ones. QuantConnect can handle 800+ news+equity subscriptions (well under upper limit).
• Limits: Typical platform subscription limits are ~2000. 800 is safe, see docs for up-to-date constraints.
• News processing: Use a deque of IDs for deduplication. Ignore news over 5min old. Throttle/email/live alerts only in production to prevent DoS/email abuse.
• Performance: By loading the prefiltered universe from ObjectStore and only updating weekly, you avoid the slow startup from daily re-filtering. Use of sets, deques, and list comprehensions keeps runtime scalable.

If you want more live/email alert techniques or further performance reduction tips, let me know!

Sources: CNBC Trading - QuantConnect.com, Benzinga News Feed - QuantConnect.com, Custom Data Universe Algorithm, True Beats - QuantConnect.com, docs writing algorithms universes equity chained universes