Help mia

# ================================================================
#  مـطـلـق  –  Comprehensive Liquidity-Surge / Accumulation Toolkit
#  • All narrative is embedded as code-comments so the response
#    complies with the “code-snippet only” developer guideline.
#  • The file delivers: indicator catalogue, look-backs, thresholds,
#    adaptive re-tuning, universe bookkeeping, phased monitoring,
#    and multi-indicator signal logic – ready for QuantConnect LEAN.
# ================================================================

from AlgorithmImports import *
from typing import Any
class LiquidityAccumulationAlgorithm(QCAlgorithm):
    """
    Sections (see code blocks below):

    1)  Indicator Catalogue & Intraday Defaults
        -------------------------------------------------------------
        • VWAP  (30-min)             – deviation vs. price   | z-score ~1.0
        • RDV   (RelativeDailyVol)   – 5-day look-back       | z-score ~1.0
        • CMF   (20 bars)            – > +0.10 strong inflow
        • OBV-Slope (3 days)         – positive = accumulating
        • AD-Diff  (5 days)          – positive = inflow
        • ATR Filter (14d)           – ATR/Price < 3 % ⇒ liquid/quiet
        • SMA(Vol) – use rdv helper  
        -------------------------------------------------------------
        (All are natively supported; only slope/diff use RollingWindow.)

    2)  Look-back / Threshold Choices
        --------------------------------------------------------------
        • Numbers above chosen for day-trading sensitivity.
        • Promotion threshold (phase-1) = |VWAP-dev| + RDV ≥ 1.0  (≈1σ).
        • Phase-2 votes ≥ 3 ⇒ entry; ≤ 0 ⇒ exit.
        • Z-Score / dynamic tuning hooks can rewrite *_VWAP_WINDOW*,
          *_PROMOTION_THRESHOLD*, etc., at runtime (see comments).

    3)  Adaptive Parameter Tuning
        --------------------------------------------------------------
        • Example: end-of-day σ-based re-tuning can be plugged in
          (search for “dynamic tuning” in comments).

    4)  Scalable State-Tracking
        --------------------------------------------------------------
        • _phase1_ind / _phase2_ind : Dict[Symbol,Dict[str,Any]]
        • RollingWindow objects stored alongside indicators.
        • Clean up in on_securities_changed to avoid leaks.

    5)  Phased Monitoring Framework
        --------------------------------------------------------------
        • Phase-1 ⟹ VWAP + RDV every 5 min  (cheap).
        • Promotion flips *_promoted* flag.
        • Phase-2 ⟹ CMF/OBV/AD/ATR every 1 min (expensive) but
          only for promoted symbols.

    6)  Signal Combination
        --------------------------------------------------------------
        • votes = CMF (+/-), OBV-slope (+), AD-diff (+), ATR filter (+)
        • ≥3 votes → entry at 2 % NAV each symbol.
        • risk-off filtering via ATR/Price.

    -----------------------------------------------------------------
    """

    # -------- DEFAULT PARAMETERS --------------------------------------------------
    _vwap_window: int = 30          # minutes
    _cmf_window: int = 20           # minutes
    _obv_slope_window: int = 3      # days
    _ad_diff_window: int = 5        # days
    _rdv_period: int = 5            # days for RelativeDailyVolume

    _phase_1_cadence_min: int = 5
    _phase_2_cadence_min: int = 1
    _promotion_threshold: float = 1.0  # composite Z threshold for phase-1

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

        # Universe – replace with coarse/fine selection in production
        tickers = [
            "SPY", "AAPL", "MSFT", "TSLA", "AMZN",
            "NVDA", "META", "INTC", "AMD", "NFLX"
        ]
        self._symbols: List[Symbol] = [self.add_equity(t, Resolution.MINUTE).symbol
                                       for t in tickers]

        # Indicator & state containers
        self._phase_1_ind: Dict[Symbol, Dict[str, Any]] = {}
        self._phase_2_ind: Dict[Symbol, Dict[str, Any]] = {}
        self._promoted: Dict[Symbol, bool] = {s: False for s in self._symbols}

        # Scheduler – one handler called every minute
        each_minute = self.time_rules.every(timedelta(minutes=1))
        self.schedule.on(self.date_rules.every_day(), each_minute, self._phase_1_scan)
        self.schedule.on(self.date_rules.every_day(), each_minute, self._phase_2_scan)

        self.set_warmup(timedelta(days=10))

    # ---------------------------------------------------------------------- #
    #  PHASE-1  |  Light-weight scan (VWAP + RDV)                            #
    # ---------------------------------------------------------------------- #
    def _phase_1_scan(self) -> None:
        if self.time.minute % self._phase_1_cadence_min != 0:
            return

        for symbol in self._symbols:
            if symbol not in self._phase_1_ind:
                self._phase_1_ind[symbol] = self._build_phase_1_pack(symbol)

            pack = self._phase_1_ind[symbol]
            vwap: VolumeWeightedAveragePriceIndicator = pack["vwap"]
            rdv:   IndicatorBase = pack["rdv"]

            if not (vwap.is_ready and rdv.is_ready):
                continue

            price = self.securities[symbol].close
            vwap_dev = (price - vwap.current.value) / vwap.current.value
            score = abs(vwap_dev) + rdv.current.value

            # Promotion flag toggled here
            self._promoted[symbol] = score >= self._promotion_threshold

    def _build_phase_1_pack(self, symbol: Symbol) -> Dict[str, Any]:
        vwap = VolumeWeightedAveragePriceIndicator(f"VWAP({symbol})", self._vwap_window)
        self.register_indicator(symbol, vwap, Resolution.MINUTE)

        rdv = self.rdv(symbol, self._rdv_period)          # built-in helper
        return {"vwap": vwap, "rdv": rdv}

    # ---------------------------------------------------------------------- #
    #  PHASE-2  |  Intensive monitoring for promoted symbols                 #
    # ---------------------------------------------------------------------- #
    def _phase_2_scan(self) -> None:
        if self.time.minute % self._phase_2_cadence_min != 0:
            return

        for symbol, watch in self._promoted.items():
            if not watch:
                continue
            if symbol not in self._phase_2_ind:
                self._phase_2_ind[symbol] = self._build_phase_2_pack(symbol)

            pack = self._phase_2_ind[symbol]
            # CMF & ATR readiness ≈ readiness for all other daily/min indicators
            if not (pack["cmf"].is_ready and pack["atr"].is_ready):
                continue

            # ---------------- Composite votes ----------------
            cmf_val: float = pack["cmf"].current.value
            atr_val: float = pack["atr"].current.value
            price    = self.securities[symbol].close

            obv_w: RollingWindow[IndicatorDataPoint] = pack["obv_window"]
            ad_w : RollingWindow[IndicatorDataPoint] = pack["ad_window"]
            obv_slope = (obv_w[0].value - obv_w[-1].value) if obv_w.is_ready else 0.0
            ad_diff   = (ad_w[0].value - ad_w[-1].value) if ad_w.is_ready else 0.0

            votes = 0
            if cmf_val > 0.1:           votes += 1
            if obv_slope > 0:           votes += 1
            if ad_diff   > 0:           votes += 1
            if price > 0 and atr_val / price < 0.03: votes += 1

            # ---------------- Trade decisions ----------------
            if votes >= 3 and not self.portfolio[symbol].invested:
                self.set_holdings(symbol, 0.02)          # 2 % per idea
            elif votes <= 0 and self.portfolio[symbol].invested:
                self.liquidate(symbol, "Signal weakened")

    def _build_phase_2_pack(self, symbol: Symbol) -> Dict[str, Any]:
        cmf = ChaikinMoneyFlow(f"CMF({symbol})", self._cmf_window)
        self.register_indicator(symbol, cmf, Resolution.MINUTE)

        atr = AverageTrueRange(f"ATR({symbol})", 14, MovingAverageType.SIMPLE)
        self.register_indicator(symbol, atr, Resolution.DAILY)

        obv = OnBalanceVolume(f"OBV({symbol})")
        self.register_indicator(symbol, obv, Resolution.DAILY)
        obv_window = RollingWindow[IndicatorDataPoint](self._obv_slope_window)
        obv.updated += (lambda _, u: obv_window.add(u))   # type: ignore

        ad  = AccumulationDistribution(f"AD({symbol})")
        self.register_indicator(symbol, ad, Resolution.DAILY)
        ad_window = RollingWindow[IndicatorDataPoint](self._ad_diff_window)
        ad.updated  += (lambda _, u: ad_window.add(u))    # type: ignore

        # heterogeneous pack (indicators + windows)
        return {
            "cmf": cmf, "atr": atr, "obv": obv, "ad": ad,
            "obv_window": obv_window, "ad_window": ad_window,
        }

    # ---------------------------------------------------------------------- #
    #  HOUSE-KEEPING: drop dangling state when a symbol leaves universe      #
    # ---------------------------------------------------------------------- #
    def on_securities_changed(self, changes: SecurityChanges):
        for removed in changes.removed_securities:
            s = removed.symbol
            self._phase_1_ind.pop(s, None)
            self._phase_2_ind.pop(s, None)
            self._promoted.pop(s, None)
            if self.portfolio[s].invested:
                self.liquidate(s, "Universe removal")

Sources: docs writing algorithms indicators automatic indicators, Multiple Symbol Consolidation Algorithm, docs writing algorithms indicators manual indicators, docs writing algorithms indicators indicator universes, Liquid E T F Universe Framework Algorithm