Fix real-time data streaming example bar data access

- Changed from trying to format Polars Series to direct dict access
- Access bar data fields directly from event.data['data'] dictionary
- Fixes 'unsupported format string passed to Series.__format__' error
- Properly displays OHLC, volume, and timestamp for new bars

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

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

Author: TexasCoding

examples/realtime_data/01_basic_realtime_data_streaming.py

@@ -1,15 +1,24 @@
 #!/usr/bin/env python
 """
-Basic real-time data streaming example
+Basic real-time data streaming example.
+
+This example demonstrates:
+- Connecting to real-time data feeds
+- Handling tick (quote) updates
+- Processing new bar events
+- Monitoring connection health
+- Displaying streaming statistics
 """
 
 import asyncio
 from datetime import datetime
 
 from project_x_py import EventType, TradingSuite
+from project_x_py.event_bus import Event
 
 
 async def main():
+    """Main function to run real-time data streaming."""
     # Create suite with real-time capabilities
     suite = await TradingSuite.create(
         ["MNQ"],
@@ -26,7 +35,8 @@ async def main():
     bar_count = 0
     last_price = None
 
-    async def on_tick(event):
+    async def on_tick(event: Event):
+        """Handle tick updates."""
         nonlocal tick_count, last_price
         tick_data = event.data
 
@@ -38,7 +48,8 @@ async def on_tick(event):
             timestamp = datetime.now().strftime("%H:%M:%S")
             print(f"[{timestamp}] Tick #{tick_count}: ${last_price:.2f}")
 
-    async def on_new_bar(event):
+    async def on_new_bar(event: Event):
+        """Handle new bar events."""
         nonlocal bar_count
         bar_count += 1
 
@@ -48,34 +59,41 @@ async def on_new_bar(event):
         event_data = event.data
         timeframe = event_data.get("timeframe", "unknown")
 
-        # The actual bar data is nested in the 'data' field
-        bar_data = event_data.get(
-            "data", event_data
-        )  # Fallback to event_data if not nested
+        # Get the bar data directly from the event
+        bar_data = event_data.get("data", {})
 
-        print(f"[{timestamp}] New {timeframe} bar #{bar_count}:")
+        if bar_data:
+            print(f"[{timestamp}] New {timeframe} bar #{bar_count}:")
+
+            # Access the bar data fields directly
+            open_price = bar_data.get("open", 0)
+            high_price = bar_data.get("high", 0)
+            low_price = bar_data.get("low", 0)
+            close_price = bar_data.get("close", 0)
+            volume = bar_data.get("volume", 0)
+            bar_timestamp = bar_data.get("timestamp", "")
 
-        # Check if bar_data has the expected fields
-        if (
-            "open" in bar_data
-            and "high" in bar_data
-            and "low" in bar_data
-            and "close" in bar_data
-        ):
             print(
-                f"  OHLC: ${bar_data['open']:.2f} / ${bar_data['high']:.2f} / ${bar_data['low']:.2f} / ${bar_data['close']:.2f}"
+                f"  OHLC: ${open_price:.2f} / ${high_price:.2f} / "
+                f"${low_price:.2f} / ${close_price:.2f}"
             )
-            print(f"  Volume: {bar_data.get('volume', 0)}")
-            print(f"  Timestamp: {bar_data.get('timestamp')}")
-
-    async def on_connection_status(event):
-        status = event.data.get("status", "unknown")
-        print(f"Connection Status: {status}")
+            print(f"  Volume: {volume}")
+            print(f"  Timestamp: {bar_timestamp}")
+
+    async def on_connection_status(event: Event):
+        """Handle connection status changes."""
+        status = event.data.get("connected", False)
+        print(f"Connection Status Changed: {status}")
+        if status:
+            print("✅ Real-time feed connected")
+        else:
+            print("❌ Real-time feed disconnected")
 
     # Register event handlers
     await mnq_context.on(EventType.QUOTE_UPDATE, on_tick)
     await mnq_context.on(EventType.NEW_BAR, on_new_bar)
     await mnq_context.on(EventType.CONNECTED, on_connection_status)
+    await mnq_context.on(EventType.DISCONNECTED, on_connection_status)
 
     print("Listening for real-time data... Press Ctrl+C to exit")
 
@@ -88,11 +106,17 @@ async def on_connection_status(event):
             connection_health = await mnq_context.data.get_health_score()
 
             print(
-                f"Status - Price: ${current_price:.2f} | Ticks: {tick_count} | Bars: {bar_count} | Health: {connection_health}"
+                f"Status - Price: ${current_price:.2f} | "
+                f"Ticks: {tick_count} | Bars: {bar_count} | "
+                f"Health: {connection_health}"
             )
 
     except KeyboardInterrupt:
         print("\nShutting down real-time stream...")
+    finally:
+        # Ensure proper cleanup
+        await suite.disconnect()
+        print("Disconnected from real-time feeds")
 
 
 if __name__ == "__main__":