Help mia

# Detailed Question for QuantConnect Expert Bot - Custom Filter & Price Issues

## 🚨 CRITICAL PROBLEMS SUMMARY

Our intraday liquidity monitoring strategy is experiencing critical failures:

### Problem #1: Strategy Fails During REGULAR Market Hours

- **IMPORTANT:** The problem occurs during NORMAL market hours (9:30 AM - 4:00 PM ET), not just extended hours

- When `self.universe_settings.extended_market_hours = True` is set at universe level, custom filters don't run AT ALL

- Result: 0 stocks selected, strategy completely non-functional

- This happens even during regular trading hours when the market is open

### Problem #2: Prices Show as 0

- When using `extended_market_hours = True` at universe_settings level

- Prices in coarse filter show as 0 or unavailable

- Causes custom filter logic to fail

### Problem #3: Custom Coarse/Fine Filters Don't Execute with extended_market_hours

- When `self.universe_settings.extended_market_hours = True` is enabled

- Custom coarse_filter_function and fine_filter_function are NEVER called

- Result: 0 stocks selected

- Without extended_market_hours: filters work perfectly and select 300 stocks

### Problem #4: Uncertainty About Correct Solution

- Is there an error in how we wrote the custom filter functions?

- Is there a correct way to use extended_market_hours with custom filters?

- Should we use a completely different approach?

---

## 🎯 STRATEGY OBJECTIVE & ARCHITECTURE

**Platform:** QuantConnect Paper Trading  

**Resources:** 2 CPU cores, 4GB RAM  

**Universe Size:** 300 US stocks  

**Strategy Type:** Monitoring only (no trades) - Null Model pattern for observation  

### Core Strategy Design:

1. **Universe Selection:** Custom coarse + fine filters to select 300 most liquid stocks

2. **Batch Rotation System:** Monitor 100 stocks at a time, rotate every 5 minutes

3. **Phased Monitoring:**

   - **Phase 1:** Light monitoring with 3 indicators (RDV, CMF, OBV)

   - **Phase 2:** Intensive monitoring with 6 indicators for stocks showing accumulation

4. **Goal:** Detect early liquidity and accumulation for day trading (3-minute scalping)

5. **Need:** Pre-market (4:00-9:30 AM) and after-hours (4:00-8:00 PM) data

### Resource Efficiency:

- Without batch rotation: 300 stocks × 6 indicators = 1,800 indicators (too much memory)

- With batch rotation: 100 stocks × 3 indicators (Phase 1) + up to 30 stocks × 6 indicators (Phase 2) = ~390 indicators

- Memory reduction: 78% less resource usage

---

## 📊 COMPLETE INDICATOR CONFIGURATION

### Phase 1 Indicators (Light Monitoring - 100 stocks per batch):

```python

# Relative Dollar Volume (RDV)

self.RDV_PERIOD = 2  # Only 2 minutes for ultra-fast response

self.RDV_ENTRY_THRESHOLD = 1.01  # Detects even 1% volume increase

self.RDV_STRONG_THRESHOLD = 1.05  # Strong signal at 5%

# Chaikin Money Flow (CMF)

self.CMF_PERIOD = 3  # 3 minutes

self.CMF_STRONG_BULLISH = 0.01  # Very sensitive (1% threshold)

# On Balance Volume (OBV)

# No period - cumulative indicator

```

### Phase 2 Indicators (Intensive Monitoring - promoted stocks only):

```python

# All Phase 1 indicators PLUS:

# Volume Weighted Average Price (VWAP)

self.VWAP_PERIOD = 5  # 5 minutes

# Volume Simple Moving Average

self.VOLUME_SMA_PERIOD = 10  # 10 minutes

self.VOLUME_SPIKE_MULTIPLIER = 1.2  # Alert on 20% volume increase

# Balance of Power (BOP)

self.BOP_BULLISH_THRESHOLD = 0.02  # 2% buying pressure

self.BOP_BEARISH_THRESHOLD = -0.02  # 2% selling pressure

# Average True Range (ATR) - for volatility

self.ATR_PERIOD = 5  # 5 minutes

```

### Detection Thresholds:

```python

self.MIN_BULLISH_INDICATORS = 2  # Need at least 2 positive indicators for signal

```

### Batch Rotation Configuration:

```python

self.BATCH_SIZE = 100  # 100 stocks per batch

self.ROTATION_INTERVAL_MINUTES = 5  # Rotate every 5 minutes

self.MAX_PROMOTED_STOCKS = 30  # Maximum promoted to Phase 2

self.PROMOTION_THRESHOLD = 2  # Need 2 signals to promote

self.PROMOTION_BARS = 1  # Promote after 1 bar

self.DEMOTION_MINUTES = 10  # Demote after 10 minutes of no signals

```

---

## 📝 COMPLETE CODE - All Custom Functions

### 1. Universe Settings in Initialize

```python

def initialize(self):

    # Basic settings

    self.set_start_date(2024, 1, 1)

    self.set_cash(100000)

    self.set_time_zone(TimeZones.NEW_YORK)

    # Universe configuration

    self.UNIVERSE_SIZE = 300

    self.MIN_PRICE = 1.0

    self.MAX_PRICE = 30.0

    self.MIN_SHARES_OUTSTANDING = 500_000

    self.MAX_SHARES_OUTSTANDING = 50_000_000

    self.MIN_MARKET_CAP = 50_000

    self.MAX_MARKET_CAP = 5_000_000_000

    self.MIN_DOLLAR_VOLUME = 50_000

    # ===== Universe Selection Setup =====

    self.universe_settings.resolution = Resolution.MINUTE

    # ❌ THIS LINE CAUSES THE PROBLEM - filters don't run when enabled

    # self.universe_settings.extended_market_hours = True

    # Add custom universe filters

    self.add_universe(self.coarse_filter_function, self.fine_filter_function)

    # Initialize storage

    self.all_universe_stocks = []  # Stores all 300 selected stocks

    self.current_batch_index = 0

    self.active_batch_symbols = set()  # Currently monitored 100 stocks

    self.promoted_symbols = set()  # Stocks promoted to Phase 2

    # Indicator dictionaries

    self.phase1_rdv_indicators = {}

    self.phase1_cmf_indicators = {}

    self.phase1_obv_indicators = {}

    self.rdv_indicators = {}  # Phase 2

    self.cmf_indicators = {}  # Phase 2

    self.obv_indicators = {}  # Phase 2

    self.vwap_indicators = {}  # Phase 2

    self.volume_sma_indicators = {}  # Phase 2

    self.bop_indicators = {}  # Phase 2

    # Schedule batch rotation every 5 minutes

    self.schedule.on(

        self.date_rules.every_day(),

        self.time_rules.every(timedelta(minutes=self.ROTATION_INTERVAL_MINUTES)),

        self.rotate_batch

    )

```

### 2. Coarse Filter Function (Initial Filtering)

```python

def coarse_filter_function(self, coarse):

    """Initial stock filtering based on price and dollar volume"""

    try:

        # Filter basic criteria

        filtered = [

            c for c in coarse

            if c.has_fundamental_data

            and c.price > 5  # Price > $5

            and c.dollar_volume > 10_000_000  # Dollar volume > $10M

        ]

        # Sort by dollar volume (highest first)

        sorted_coarse = sorted(filtered, key=lambda c: c.dollar_volume, reverse=True)

        # Select top 2000 stocks

        top_stocks = sorted_coarse[:2000]

        self.debug(f"[CoarseFilter] {len(coarse)} input | {len(filtered)} after filter | {len(top_stocks)} top")

        return [c.symbol for c in top_stocks]

    except Exception as e:

        self.error(f"[ERROR] Error in coarse_filter_function: {str(e)}")

        return []

```

### 3. Fine Filter Function (Fundamental Filtering)

```python

def fine_filter_function(self, fine):

    """Fine filtering based on fundamentals and liquidity"""

    try:

        # Filter by market cap and fundamentals

        filtered = [

            f for f in fine

            if f.market_cap > 2_000_000_000  # Market cap > $2B

            and f.valuation_ratios.pe_ratio > 0

            and f.valuation_ratios.pe_ratio < 50

            and f.earning_reports.basic_average_shares.three_months > 0

            and f.earning_reports.basic_eps.twelve_months > 0

        ]

        # Sort by market cap (highest first)

        sorted_fine = sorted(filtered, key=lambda f: f.market_cap, reverse=True)

        # Select exactly 300 stocks

        selected = sorted_fine[:300]

        self.debug(f"[FineFilter] {len(fine)} input | {len(filtered)} after filter | {len(selected)} selected")

        return [f.symbol for f in selected]

    except Exception as e:

        self.error(f"[ERROR] Error in fine_filter_function: {str(e)}")

        return []

```

### 4. OnSecuritiesChanged - Stock Addition/Removal Handler

```python

def OnSecuritiesChanged(self, changes):

    """Handle universe changes - add/remove stocks"""

    try:

        self.debug("[DIAG] OnSecuritiesChanged START")

        # Process added securities

        if changes.added_securities:

            self.debug(f"[DIAG] Processing {len(changes.added_securities)} added securities")

            for added in changes.added_securities:

                try:

                    # Add to universe list

                    if added.symbol not in self.all_universe_stocks:

                        self.all_universe_stocks.append(added.symbol)

                    # ATTEMPTED SOLUTION: Add extended hours per security

                    # This creates 300 additional subscriptions (600 total)

                    try:

                        self.add_equity(added.symbol, Resolution.MINUTE,

                                      extended_market_hours=True,

                                      fill_forward=False)

                        self.debug(f"[Extended-Hours-Sub] {added.symbol.value} | Extended hours enabled")

                    except Exception as ex:

                        self.debug(f"[Extended-Hours-Sub] Failed for {added.symbol.value}: {str(ex)}")

                    # Verify price is available

                    if added.symbol in self.securities:

                        sec = self.securities[added.symbol]

                        has_price = sec.price > 0

                        self.debug(f"[SUB-VERIFY] {added.symbol.value} | HasPrice: {has_price} | Price: ${sec.price:.2f}")

                    if self.ENABLE_DETAILED_LOGGING and self.securities[added.symbol].price > 0:

                        self.log(f"[Added] {added.symbol.value} @ ${self.securities[added.symbol].price:.2f}")

                except Exception as e:

                    self.log(f"[ERROR] Error adding {added.symbol.value}: {str(e)}")

            # Initialize first batch once we have enough stocks

            if len(self.all_universe_stocks) >= self.BATCH_SIZE and not self.first_batch_initialized:

                self.debug("[DIAG] Initializing first batch")

                self.initialize_batch(0)

                self.first_batch_initialized = True

                self.log(f"[Universe] First batch initialized: {len(self.active_batch_symbols)} stocks")

                self.debug("[DIAG] First batch initialized successfully")

            self.log(f"[Universe] Total stocks in universe: {len(self.all_universe_stocks)}")

            self.debug(f"[DIAG] Added {len(changes.added_securities)} securities")

        # Process removed securities

        if changes.removed_securities:

            for removed in changes.removed_securities:

                try:

                    # Remove from universe list

                    if removed.symbol in self.all_universe_stocks:

                        self.all_universe_stocks.remove(removed.symbol)

                    # Clean up indicators

                    self.cleanup_symbol_indicators(removed.symbol)

                    # Clean up state tracking

                    self.active_batch_symbols.discard(removed.symbol)

                    self.promoted_symbols.discard(removed.symbol)

                    self.symbol_phases.pop(removed.symbol, None)

                    self.liquidity_states.pop(removed.symbol, None)

                    self.state_entry_time.pop(removed.symbol, None)

                    if self.ENABLE_DETAILED_LOGGING:

                        self.log(f"[Removed] {removed.symbol.value}")

                except Exception as e:

                    self.error(f"[ERROR] Error removing {removed.symbol.value}: {str(e)}")

        self.debug("[DIAG] OnSecuritiesChanged completed successfully")

    except Exception as e:

        self.error(f"[ERROR] General error in OnSecuritiesChanged: {str(e)}")

```

### 5. Batch Initialization Function

```python

def initialize_batch(self, batch_index):

    """Initialize a new batch of stocks with Phase 1 indicators"""

    try:

        start_idx = batch_index * self.BATCH_SIZE

        end_idx = min(start_idx + self.BATCH_SIZE, len(self.all_universe_stocks))

        batch_symbols = self.all_universe_stocks[start_idx:end_idx]

        for symbol in batch_symbols:

            if symbol not in self.promoted_symbols:

                # Create Phase 1 indicators for this stock

                self.create_phase1_indicators(symbol)

                self.active_batch_symbols.add(symbol)

                self.symbol_phases[symbol] = 1

                self.phase1_signal_count[symbol] = 0

        self.log(f"[BatchInit] Batch {batch_index + 1} initialized: {len(batch_symbols)} stocks | " +

                f"Range: {start_idx}-{end_idx} | " +

                f"Promoted continuing: {len(self.promoted_symbols)}")

    except Exception as e:

        self.log(f"[ERROR] Error in initialize_batch: {str(e)}")

```

### 6. Batch Rotation Function

```python

def rotate_batch(self):

    """Rotate batches - replace current batch with next batch"""

    self.debug("[DIAG] rotate_batch START")

    try:

        if not self.first_batch_initialized:

            self.debug("[DIAG] Skipping rotation - first batch not yet initialized")

            return

        if len(self.all_universe_stocks) == 0:

            self.debug("[DIAG] Skipping rotation - no stocks in universe")

            return

        self.batch_rotation_count += 1

        # Clean up old batch (excluding promoted stocks)

        old_batch_symbols = self.active_batch_symbols - self.promoted_symbols

        for symbol in old_batch_symbols:

            self.cleanup_phase1_indicators(symbol)

        # Calculate next batch index

        import math

        num_batches = max(1, math.ceil(len(self.all_universe_stocks) / self.BATCH_SIZE))

        self.current_batch_index = (self.current_batch_index + 1) % num_batches

        # Initialize new batch

        self.active_batch_symbols = set(self.promoted_symbols)

        self.initialize_batch(self.current_batch_index)

        total_active = len(self.active_batch_symbols)

        batch_count = total_active - len(self.promoted_symbols)

        self.log(f"[BatchRotation #{self.batch_rotation_count}] " +

                f"Batch {self.current_batch_index + 1}/{num_batches} | " +

                f"New stocks: {batch_count} | " +

                f"Promoted continuing: {len(self.promoted_symbols)} | " +

                f"Total active: {total_active}")

        self.check_for_demotions()

        self.debug("[DIAG] rotate_batch completed successfully")

    except Exception as e:

        self.error(f"[ERROR] Error in rotate_batch: {str(e)}")

```

### 7. Indicator Creation Functions

```python

def create_phase1_indicators(self, symbol):

    """Create Phase 1 indicators (3 lightweight indicators)"""

    try:

        self.phase1_rdv_indicators[symbol] = self.rdv(symbol, period=self.RDV_PERIOD, resolution=Resolution.MINUTE)

        self.phase1_cmf_indicators[symbol] = self.cmf(symbol, self.CMF_PERIOD, Resolution.MINUTE)

        self.phase1_obv_indicators[symbol] = self.obv(symbol, Resolution.MINUTE)

        self.debug(f"[IndicatorCreation] {symbol.value} | Phase1 indicators created (RDV, CMF, OBV)")

    except Exception as e:

        self.error(f"[ERROR] Error creating Phase 1 indicators for {symbol.value}: {str(e)}")

def promote_to_phase2(self, symbol):

    """Promote stock to Phase 2 - add 3 more indicators"""

    try:

        # Create Phase 2 indicators (in addition to Phase 1)

        self.rdv_indicators[symbol] = self.rdv(symbol, period=self.RDV_PERIOD, resolution=Resolution.MINUTE)

        self.cmf_indicators[symbol] = self.cmf(symbol, self.CMF_PERIOD, Resolution.MINUTE)

        self.obv_indicators[symbol] = self.obv(symbol, Resolution.MINUTE)

        self.vwap_indicators[symbol] = self.vwap(symbol, self.VWAP_PERIOD, Resolution.MINUTE)

        self.volume_sma_indicators[symbol] = self.sma(symbol, self.VOLUME_SMA_PERIOD, Resolution.MINUTE, Field.VOLUME)

        self.bop_indicators[symbol] = self.bop(symbol, Resolution.MINUTE)

        self.promoted_symbols.add(symbol)

        self.symbol_phases[symbol] = 2

        self.phase2_entry_time[symbol] = self.time

        self.promotion_count += 1

        self.log(f"[Promotion #{self.promotion_count}] {symbol.value} promoted to Phase 2 | " +

                f"Total promoted: {len(self.promoted_symbols)}/{self.MAX_PROMOTED_STOCKS}")

    except Exception as e:

        self.error(f"[ERROR] Error promoting {symbol.value} to Phase 2: {str(e)}")

```

---

## ❓ SPECIFIC QUESTIONS FOR EXPERT BOT

### Question #1: Are the Custom Filter Functions Written Correctly?

- Are the coarse_filter_function and fine_filter_function above written correctly?

- Are there any logic errors or misuse of the QuantConnect API?

- Is the sorting and filtering approach correct?

- Should we be checking for other conditions (like `c.price > 0` in coarse filter)?

### Question #2: Why Do Prices Show as 0 in Some Cases?

- When does `c.price` in CoarseFilter equal 0?

- Is this related to extended_market_hours setting?

- Is there a way to verify price availability before filtering?

- Should we add explicit price checks in our filters?

### Question #3: Why Does extended_market_hours at Universe Level Stop Custom Filters?

```python

# ❌ This causes: 0 stocks selected, filters never called

self.universe_settings.extended_market_hours = True

self.add_universe(self.coarse_filter_function, self.fine_filter_function)

```

- **Why does this behavior occur?**

- Is this a bug or expected behavior?

- Does extended_market_hours only work with built-in universe methods like `universe.dollar_volume.top(100)`?

- Is there any way to use extended_market_hours with custom coarse/fine filters?

### Question #4: What's the Correct Way to Get Extended Hours Data with Custom Filters?

**Option A: Individual subscriptions (current approach)**

```python

# In OnSecuritiesChanged, after universe adds stocks

for symbol in added_securities:

    self.add_equity(symbol, Resolution.MINUTE, extended_market_hours=True)

```

- Is this approach correct?

- Does this cause excessive memory usage (600 subscriptions instead of 300)?

- Does it work reliably in Paper Trading?

- Are there any downsides?

**Option B: Different approach?**

- Is there a better method we haven't tried?

- Can extended_market_hours be used at universe level with custom filters somehow?

- Is there a special configuration or pattern we're missing?

### Question #5: Memory Consumption Issues

- When adding 300 manual extended_market_hours subscriptions

- Plus 300 subscriptions from Universe = 600 total subscriptions

- Does this cause Out of Memory in Paper Trading (4GB RAM limit)?

- Or is the OOM issue only in Backtest (fast processing) and not Paper Trading (real-time)?

- What's the recommended approach for monitoring 300 stocks with extended hours?

### Question #6: Do Custom Filters Need Modification for Extended Hours?

- Should we check market hours inside the filter functions?

- Should we handle zero prices differently?

- Are there special best practices for custom filters with extended hours?

- Do we need to use different data fields or methods?

### Question #7: Universe Selection Timing

- When exactly do coarse and fine filters execute?

- Is it always at midnight UTC?

- Does extended_market_hours affect the execution timing?

- In Paper Trading, when can we expect universe selection to run?

### Question #8: Batch Rotation with Extended Hours

- Our batch rotation system monitors 100 stocks at a time

- Does batch rotation interfere with extended hours subscriptions?

- Should we add/remove extended hours subscriptions during batch rotation?

- Or is it better to subscribe all 300 stocks to extended hours upfront?

---

## 📊 OBSERVED BEHAVIOR

### ✅ WITHOUT extended_market_hours (WORKS):

```

[CoarseFilter] 8000 input | 2699 after filter | 2000 top

[FineFilter] 2000 input | 789 after filter | 300 selected

[DIAG] Added 300 securities

[BatchInit] Batch 1 initialized: 100 stocks

[OnData-ENTRY] Called! Total bars in slice: 95

```

**Result:** Filters work perfectly, 300 stocks selected, strategy runs normally

### ❌ WITH extended_market_hours at Universe Level (FAILS):

```

[DIAG] AddUniverse PRE

[DIAG] AddUniverse POST

[DIAG] Universe selection configured

... NO messages from CoarseFilter or FineFilter ...

[DIAG] Added 0 securities

```

**Result:** Filters NEVER called, 0 stocks selected, strategy completely broken

**NOTE:** This happens even during regular market hours (9:30 AM - 4:00 PM ET)

### ⚠️ WITH Individual extended_market_hours Subscriptions (WORKS but OOM in Backtest):

```

[CoarseFilter] 2699 after filter

[FineFilter] 300 selected

[Extended-Hours-Sub] AAPL | Extended hours enabled

[Extended-Hours-Sub] MSFT | Extended hours enabled

... (300 times)

[BatchInit] Batch 1 initialized: 100 stocks

[OnData-ENTRY] Time: 2024-01-02 18:16:00  ← after-hours timestamp!

[OnData-ENTRY] Time: 2024-01-02 20:00:00  ← 8:00 PM ET

... works for a while then ...

RuntimeError: System.OutOfMemoryException (in Backtest only, at 01/03/2024 14:31:00 UTC)

```

**Result:**

- Filters work correctly

- Extended hours data flows (timestamps 18:16, 20:00 prove it)

- OnData called during extended hours

- Eventually crashes with OOM in Backtest (but user says Backtest has separate resources from Paper Trading)

---

## 🆘 WHAT WE NEED FROM YOU

Please help us understand:

1. **Root Cause Diagnosis:**

   - WHY do custom coarse/fine filters fail when `extended_market_hours = True` at universe level?

   - WHY can prices be 0 in coarse filter?

   - Is there an error in how we wrote our custom filter functions?

2. **Correct Solution:**

   - What is the RELIABLE way to achieve:

     - 300 stocks selected via custom coarse/fine filters

     - Extended market hours data (pre-market & after-hours)

     - Reasonable memory usage (4GB RAM limit in Paper Trading)

   - Should we use individual `add_equity()` calls with extended_market_hours?

   - Or is there a better approach?

3. **Best Practices:**

   - How should custom filters be written to work reliably with extended hours?

   - What's the optimal way to manage subscriptions for 300 stocks?

   - Are there any QuantConnect-specific patterns we should follow?

4. **Memory Considerations:**

   - Is 600 subscriptions (300 universe + 300 extended hours) too much for 4GB RAM?

   - Does Paper Trading handle memory differently than Backtest?

   - Should we use a different architecture?

Thank you for your help! 🙏

---

## 📎 ADDITIONAL CONTEXT

- **Strategy runs continuously in Paper Trading** (real-time)

- **No actual trades executed** - monitoring only (Null Model pattern)

- **Goal is early detection** of liquidity for 3-minute day trading scalps

- **Batch rotation reduces memory** by 78% (390 vs 1,800 indicators)

- **User needs extended hours** because significant liquidity often starts in pre-market

- **Problem existed during regular hours FIRST**, extended hours was added later as diagnostic