fix: Real-time data processing for E-mini contracts and low-volume periods (v3.1.8)

## Fixed
- Real-time data not being processed for E-mini contracts (NQ/ES) that resolve to different exchange symbols
- Symbol matching now properly handles contract resolution (e.g., NQ→ENQ, ES→EP)
- Stores both user-specified instrument and resolved symbol ID for accurate matching

## Added
- Bar timer mechanism that creates empty bars during low-volume periods
- Ensures consistent bar generation at regular intervals regardless of trading activity
- Empty bars (volume=0) maintain price continuity using the last close price
- Particularly important for low-volume contracts and after-hours trading

## Technical Details
- RealtimeDataManager now tracks `instrument_symbol_id` alongside `instrument`
- Symbol validation checks both original and resolved symbols
- Bar timer runs periodically based on shortest configured timeframe
- Empty bars are distinguishable by volume=0

This fixes issues where NQ and other E-mini contracts were not updating in real-time.

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <[email protected]>

Author: TexasCoding

CHANGELOG.md

@@ -14,6 +14,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Migration guides will be provided for all breaking changes
 - Semantic versioning (MAJOR.MINOR.PATCH) is strictly followed
 
+## [3.1.8] - 2025-08-12
+
+### Fixed
+- **🔧 Real-time Data Processing**: Fixed real-time data not being processed for E-mini contracts (NQ/ES)
+  - Symbol matching now handles contract resolution (e.g., NQ resolves to ENQ)
+  - Stores both original instrument and resolved symbol ID for proper matching
+  - Affects all contracts where user symbol differs from exchange symbol
+
+### Added
+- **⏱️ Bar Timer Mechanism**: Automatic bar creation during low-volume periods
+  - Creates empty bars (volume=0) at regular intervals when no ticks arrive
+  - Ensures consistent bar generation for all instruments regardless of trading activity
+  - Particularly important for low-volume contracts and after-hours trading
+  - Empty bars maintain price continuity using the last close price
+
+### Improved
+- Enhanced symbol validation to support both user-specified and exchange-resolved symbols
+- Better handling of futures contract name resolution (NQ→ENQ, ES→EP, etc.)
+- More robust real-time data pipeline for all futures contracts
+
 ## [3.1.7] - 2025-08-12
 
 ### Changed

CLAUDE.md

@@ -2,7 +2,7 @@
 
 This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository.
 
-## Project Status: v3.1.7 - Stable Production Release
+## Project Status: v3.1.8 - Stable Production Release
 
 **IMPORTANT**: This project uses a fully asynchronous architecture. All APIs are async-only, optimized for high-performance futures trading.
 
@@ -288,7 +288,13 @@ async with ProjectX.from_env() as client:
 
 ## Recent Changes
 
-### v3.1.7 - Latest Release
+### v3.1.8 - Latest Release
+- **Fixed**: Real-time data processing for E-mini contracts (NQ/ES) that resolve to different symbols
+- **Added**: Bar timer mechanism to create empty bars during low-volume periods
+- **Improved**: Symbol matching to handle contract resolution (e.g., NQ→ENQ)
+- **Enhanced**: Real-time data manager now properly processes all futures contracts
+
+### v3.1.7 - Previous Release
 - Minor updates and improvements
 - Documentation enhancements
 

docs/conf.py

@@ -23,8 +23,8 @@
 project = "project-x-py"
 copyright = "2025, Jeff West"
 author = "Jeff West"
-release = "3.1.7"
-version = "3.1.7"
+release = "3.1.8"
+version = "3.1.8"
 
 # -- General configuration ---------------------------------------------------
 

examples/realtime_data_manager/01_events_with_on.py

@@ -7,8 +7,10 @@
 
 async def main():
     print("Creating TradingSuite...")
+    # Note: Use "MNQ" for Micro E-mini Nasdaq-100 futures
+    # "NQ" resolves to E-mini Nasdaq (ENQ) which may have different data characteristics
     suite = await TradingSuite.create(
-        instrument="MNQ",
+        instrument="ES",  # Works best with MNQ for consistent real-time updates
         timeframes=["15sec"],
     )
     print("TradingSuite created!")
@@ -19,7 +21,7 @@ async def main():
     # Set up signal handler for clean exit
     shutdown_event = asyncio.Event()
 
-    def signal_handler(signum, frame):
+    def signal_handler(_signum, _frame):
         print("\n\nReceived interrupt signal. Shutting down gracefully...")
         shutdown_event.set()
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "project-x-py"
-version = "3.1.7"
+version = "3.1.8"
 description = "High-performance Python SDK for futures trading with real-time WebSocket data, technical indicators, order management, and market depth analysis"
 readme = "README.md"
 license = { text = "MIT" }

src/project_x_py/__init__.py

@@ -95,7 +95,7 @@
 
 from project_x_py.client.base import ProjectXBase
 
-__version__ = "3.1.7"
+__version__ = "3.1.8"
 __author__ = "TexasCoding"
 
 # Core client classes - renamed from Async* to standard names

src/project_x_py/indicators/__init__.py

@@ -202,7 +202,7 @@
 )
 
 # Version info
-__version__ = "3.1.7"
+__version__ = "3.1.8"
 __author__ = "TexasCoding"
 
 

src/project_x_py/realtime_data_manager/core.py

@@ -117,6 +117,7 @@ async def on_new_bar(data):
 """
 
 import asyncio
+import contextlib
 import time
 from collections import defaultdict
 from datetime import datetime
@@ -399,6 +400,8 @@ def __init__(
 
         # Contract ID for real-time subscriptions
         self.contract_id: str | None = None
+        # Actual symbol ID from the resolved instrument (e.g., "ENQ" when user specifies "NQ")
+        self.instrument_symbol_id: str | None = None
 
         # Memory management settings are set in _apply_config_defaults()
         self.last_cleanup: float = time.time()
@@ -431,6 +434,9 @@ def __init__(
         # Background cleanup task
         self._cleanup_task: asyncio.Task[None] | None = None
 
+        # Background bar timer task for low-volume periods
+        self._bar_timer_task: asyncio.Task[None] | None = None
+
         self.logger.info(
             "RealtimeDataManager initialized", extra={"instrument": instrument}
         )
@@ -529,6 +535,16 @@ async def initialize(self, initial_days: int = 1) -> bool:
             # Store the exact contract ID for real-time subscriptions
             self.contract_id = instrument_info.id
 
+            # Store the actual symbol ID for matching (e.g., "ENQ" when user specifies "NQ")
+            # Extract from symbolId like "F.US.ENQ" -> "ENQ"
+            if instrument_info.symbolId and "." in instrument_info.symbolId:
+                parts = instrument_info.symbolId.split(".")
+                self.instrument_symbol_id = (
+                    parts[-1] if parts else instrument_info.symbolId
+                )
+            else:
+                self.instrument_symbol_id = instrument_info.symbolId or self.instrument
+
             # Load initial data for all timeframes
             async with self.data_lock:
                 for tf_key, tf_config in self.timeframes.items():
@@ -668,6 +684,9 @@ async def on_new_bar(data):
             # Start cleanup task
             self.start_cleanup_task()
 
+            # Start bar timer task for low-volume periods
+            self._start_bar_timer_task()
+
             self.logger.debug(
                 LogMessages.DATA_SUBSCRIBE,
                 extra={"status": "feed_started", "instrument": self.instrument},
@@ -690,6 +709,9 @@ async def stop_realtime_feed(self) -> None:
             # Cancel cleanup task
             await self.stop_cleanup_task()
 
+            # Cancel bar timer task
+            await self._stop_bar_timer_task()
+
             # Unsubscribe from market data
             # Note: unsubscribe_market_data will be implemented in ProjectXRealtimeClient
             if self.contract_id:
@@ -730,3 +752,134 @@ async def cleanup(self) -> None:
                     dom_attr[_k] = []
 
         self.logger.info("✅ RealtimeDataManager cleanup completed")
+
+    def _start_bar_timer_task(self) -> None:
+        """Start the bar timer task for creating bars during low-volume periods."""
+        if self._bar_timer_task is None or self._bar_timer_task.done():
+            self._bar_timer_task = asyncio.create_task(self._bar_timer_loop())
+            self.logger.debug("Bar timer task started")
+
+    async def _stop_bar_timer_task(self) -> None:
+        """Stop the bar timer task."""
+        if self._bar_timer_task and not self._bar_timer_task.done():
+            self._bar_timer_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._bar_timer_task
+            self.logger.debug("Bar timer task stopped")
+
+    async def _bar_timer_loop(self) -> None:
+        """
+        Periodic task to create empty bars during low-volume periods.
+
+        This ensures bars are created at regular intervals even when
+        there's no trading activity (important for low-volume instruments).
+        """
+        try:
+            # Find the shortest timeframe interval to check
+            min_seconds = float("inf")
+            for tf_config in self.timeframes.values():
+                interval = tf_config["interval"]
+                unit = tf_config["unit"]
+
+                # Convert to seconds
+                if unit == "sec":
+                    seconds = interval
+                elif unit == "min":
+                    seconds = interval * 60
+                elif unit == "hr":
+                    seconds = interval * 3600
+                else:
+                    continue
+
+                min_seconds = min(min_seconds, seconds)
+
+            # Check at least every 5 seconds, but no more than the shortest interval
+            check_interval = min(5.0, min_seconds / 3)
+
+            self.logger.debug(f"Bar timer checking every {check_interval} seconds")
+
+            while self.is_running:
+                await asyncio.sleep(check_interval)
+
+                if not self.is_running:
+                    break
+
+                # Check each timeframe for stale bars
+                await self._check_and_create_empty_bars()
+
+        except asyncio.CancelledError:
+            self.logger.debug("Bar timer task cancelled")
+            raise
+        except Exception as e:
+            self.logger.error(f"Error in bar timer loop: {e}")
+
+    async def _check_and_create_empty_bars(self) -> None:
+        """
+        Check each timeframe and create empty bars if needed.
+
+        This handles low-volume periods where no ticks are coming in,
+        ensuring bars are still created at the proper intervals.
+        """
+        try:
+            current_time = datetime.now(self.timezone)
+            events_to_trigger = []
+
+            async with self.data_lock:
+                for tf_key, tf_config in self.timeframes.items():
+                    if tf_key not in self.data:
+                        continue
+
+                    current_data = self.data[tf_key]
+                    if current_data.height == 0:
+                        continue
+
+                    # Get the last bar time
+                    last_bar_time = (
+                        current_data.select(pl.col("timestamp")).tail(1).item()
+                    )
+
+                    # Calculate what the current bar time should be
+                    expected_bar_time = self._calculate_bar_time(
+                        current_time, tf_config["interval"], tf_config["unit"]
+                    )
+
+                    # If we're missing bars, create empty ones
+                    if expected_bar_time > last_bar_time:
+                        # Get the last close price to use for empty bars
+                        last_close = current_data.select(pl.col("close")).tail(1).item()
+
+                        # Create empty bar with last close as OHLC, volume=0
+                        new_bar = pl.DataFrame(
+                            {
+                                "timestamp": [expected_bar_time],
+                                "open": [last_close],
+                                "high": [last_close],
+                                "low": [last_close],
+                                "close": [last_close],
+                                "volume": [0],  # Zero volume for empty bars
+                            }
+                        )
+
+                        self.data[tf_key] = pl.concat([current_data, new_bar])
+                        self.last_bar_times[tf_key] = expected_bar_time
+
+                        self.logger.debug(
+                            f"Created empty bar for {tf_key} at {expected_bar_time} "
+                            f"(low volume period)"
+                        )
+
+                        # Prepare event to trigger
+                        events_to_trigger.append(
+                            {
+                                "timeframe": tf_key,
+                                "bar_time": expected_bar_time,
+                                "data": new_bar.to_dicts()[0],
+                            }
+                        )
+
+            # Trigger events outside the lock (non-blocking)
+            for event in events_to_trigger:
+                asyncio.create_task(self._trigger_callbacks("new_bar", event))
+
+        except Exception as e:
+            self.logger.error(f"Error checking/creating empty bars: {e}")

src/project_x_py/realtime_data_manager/validation.py

@@ -248,8 +248,19 @@ def _symbol_matches_instrument(
         else:
             base_symbol = symbol
 
-        # Compare with our instrument (case-insensitive)
-        return base_symbol.upper() == self.instrument.upper()
+        # Compare with both our original instrument and the resolved symbol ID
+        # This handles cases like NQ -> ENQ resolution
+        base_upper = base_symbol.upper()
+
+        # Check against original instrument (e.g., "NQ")
+        if base_upper == self.instrument.upper():
+            return True
+
+        # Check against resolved symbol ID (e.g., "ENQ" when user specified "NQ")
+        if hasattr(self, "instrument_symbol_id") and self.instrument_symbol_id:
+            return base_upper == self.instrument_symbol_id.upper()
+
+        return False
 
     def get_realtime_validation_status(
         self: "RealtimeDataManagerProtocol",