Fixes to Market Data connections

Author: ChristianJStarr

examples/README.md

@@ -44,3 +44,189 @@ Some examples require additional Python packages:
 - All examples connect to the demo environment by default. To use the live environment, set the `PROJECTX_ENVIRONMENT` environment variable to "live".
 - The examples are designed to be interactive and will prompt for user input when needed.
 - Make sure your API credentials have the necessary permissions for the actions demonstrated in each example.
+
+## Running the Examples
+
+To run these examples, you'll need:
+
+1. A ProjectX account with API access
+2. Your username and API key
+
+Set these as environment variables:
+
+```bash
+export PROJECTX_USERNAME=your_username
+export PROJECTX_API_KEY=your_api_key
+```
+
+Then run an example:
+
+```bash
+python realtime_example.py
+```
+
+## Troubleshooting Realtime Connections
+
+If you're experiencing issues with realtime connections, follow this troubleshooting guide:
+
+### Common Issues and Solutions
+
+#### No events being received
+
+If you've subscribed to market events but aren't receiving any callbacks:
+
+1. **Check Authentication**: Ensure your API key is valid and has appropriate permissions.
+   ```python
+   # Verify authentication is working
+   accounts = client.accounts.get_accounts()
+   print(accounts)  # If this fails, you have an authentication issue
+   ```
+
+2. **Verify Connection Establishment**: Add debug logging to check connection status.
+   ```python
+   import logging
+   logging.basicConfig(level=logging.DEBUG)
+   ```
+
+3. **Check Method Names**: The exact method names must match between client and server.
+   - SubscribeContractQuotes
+   - SubscribeContractTrades
+   - SubscribeContractMarketDepth
+
+4. **Check Contract IDs**: Verify the contract ID is valid and properly formatted.
+   ```python
+   # Get all available contracts
+   contracts = client.contracts.search_contracts()
+   valid_ids = [c["id"] for c in contracts["contracts"]]
+   print(valid_ids)
+   ```
+
+5. **Enable Signal-R Debug Mode**: Directly configure the SignalR connection:
+   ```python
+   # Add more debug logging
+   import signalrcore.hub.errors
+   signalrcore.hub.errors.HubError.verbose = True
+   ```
+
+6. **Verify Event Names**: The events from the server must match these exactly:
+   - GatewayQuote
+   - GatewayTrade
+   - GatewayDepth
+
+7. **Check Callback Function Signatures**: Your callbacks should accept 2 parameters:
+   ```python
+   def handle_quote(contract_id: str, data: dict):
+       # Both parameters are required
+       print(f"Contract: {contract_id}, Data: {data}")
+   ```
+
+8. **Connection Timing**: Ensure the connection is fully established before subscribing:
+   ```python
+   # Start first, then subscribe
+   realtime.start()
+   time.sleep(2)  # Give it time to connect
+   market_hub.subscribe_quotes(contract_id, handle_quote)
+   ```
+
+#### Connection Drops or Errors
+
+If the connection is unstable or drops frequently:
+
+1. **Check Network Stability**: Realtime connections require stable internet.
+
+2. **Implement Reconnect Logic**: Use the built-in reconnect functionality:
+   ```python
+   # The SDK handles reconnection automatically
+   # You can also manually reconnect
+   realtime.stop()
+   time.sleep(1)
+   realtime.start()  # This will reestablish connections and resubscribe
+   ```
+
+3. **Check for Credential Expiration**: Ensure your token hasn't expired.
+   The SDK should handle this automatically, but you can verify:
+   ```python
+   # Force token refresh
+   client.auth.validate_token()
+   ```
+
+### Debugging Tools
+
+1. **Connection Testing Script**:
+   ```python
+   import time
+   from projectx_sdk import ProjectXClient
+   from projectx_sdk.realtime import RealtimeService
+
+   # Enable verbose logging
+   import logging
+   logging.basicConfig(level=logging.DEBUG)
+
+   client = ProjectXClient(username="...", api_key="...", environment="demo")
+   realtime = RealtimeService(client)
+
+   # Connection state callbacks
+   def on_connect():
+       print("CONNECTED!")
+
+   def on_disconnect():
+       print("DISCONNECTED!")
+
+   # Manually attach to connection events
+   realtime.market._connection.on_open(on_connect)
+   realtime.market._connection.on_close(on_disconnect)
+
+   # Start the connection
+   realtime.start()
+
+   # Keep alive
+   try:
+       while True:
+           time.sleep(1)
+   except KeyboardInterrupt:
+       realtime.stop()
+   ```
+
+2. **Data Structure Inspection**:
+   ```python
+   def debug_callback(contract_id, data):
+       import json
+       print(f"CONTRACT: {contract_id}")
+       print(f"DATA: {json.dumps(data, indent=2)}")
+
+   market_hub.subscribe_quotes("CON.F.US.ENQ.H25", debug_callback)
+   ```
+
+### Advanced Troubleshooting
+
+If basic troubleshooting doesn't resolve your issue:
+
+1. **Direct SignalR Connection**: Bypass the SDK abstraction layer:
+   ```python
+   from signalrcore.hub_connection_builder import HubConnectionBuilder
+
+   token = client.auth.get_token()
+   hub_url = "https://gateway-api-demo.s2f.projectx.com/hubs/market"
+
+   connection = (
+       HubConnectionBuilder()
+       .with_url(f"{hub_url}?access_token={token}")
+       .with_automatic_reconnect({
+           "type": "raw",
+           "keep_alive_interval": 10,
+           "reconnect_interval": 5,
+           "max_attempts": 10,
+       })
+       .build()
+   )
+
+   connection.on("GatewayQuote", lambda contract_id, data:
+       print(f"QUOTE: {contract_id} - {data}"))
+
+   connection.start()
+   connection.invoke("SubscribeContractQuotes", "CON.F.US.ENQ.H25")
+   ```
+
+2. **Network Traffic Analysis**: Use tools like Wireshark to inspect the WebSocket traffic.
+
+3. **Check Server Status**: Verify the ProjectX API status and whether there are any known issues.

examples/realtime_example.py

@@ -0,0 +1,154 @@
+"""
+Example of using ProjectX SDK realtime features.
+
+This example demonstrates how to properly set up and use the MarketHub
+to subscribe to quote, trade, and market depth events.
+"""
+
+import logging
+import os
+import signal
+import sys
+import time
+from typing import Any, Dict
+
+from projectx_sdk import ProjectXClient
+from projectx_sdk.realtime import RealtimeService
+
+# Set up logging
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("projectx_example")
+
+# Global flag to control execution
+running = True
+
+
+def setup_signal_handlers():
+    """Set up signal handlers to gracefully handle termination."""
+
+    def signal_handler(sig, frame):
+        global running
+        logger.info("Received shutdown signal, closing connections...")
+        running = False
+
+    signal.signal(signal.SIGINT, signal_handler)
+    signal.signal(signal.SIGTERM, signal_handler)
+
+
+def handle_quote(contract_id: str, data: Dict[str, Any]):
+    """
+    Handle quote updates.
+
+    Args:
+        contract_id: The contract ID
+        data: Quote data dictionary
+    """
+    bid = data.get("bid")
+    ask = data.get("ask")
+    last = data.get("last")
+    timestamp = data.get("timestamp", "N/A")
+
+    logger.info(f"Quote [{contract_id}]: Bid={bid}, Ask={ask}, Last={last} @ {timestamp}")
+
+
+def handle_trade(contract_id: str, data: Dict[str, Any]):
+    """
+    Handle trade updates.
+
+    Args:
+        contract_id: The contract ID
+        data: Trade data dictionary
+    """
+    price = data.get("price")
+    size = data.get("size")
+    timestamp = data.get("timestamp", "N/A")
+
+    logger.info(f"Trade [{contract_id}]: Price={price}, Size={size} @ {timestamp}")
+
+
+def handle_depth(contract_id: str, data: Dict[str, Any]):
+    """
+    Handle market depth updates.
+
+    Args:
+        contract_id: The contract ID
+        data: Market depth data dictionary
+    """
+    bids = data.get("bids", [])
+    asks = data.get("asks", [])
+
+    top_bid = bids[0] if bids else {"price": "N/A", "size": "N/A"}
+    top_ask = asks[0] if asks else {"price": "N/A", "size": "N/A"}
+
+    logger.info(
+        f"Depth [{contract_id}]: Top Bid={top_bid['price']} x {top_bid['size']}, "
+        f"Top Ask={top_ask['price']} x {top_ask['size']}"
+    )
+
+
+def main():
+    """Run the example."""
+    # Set up signal handlers for graceful shutdown
+    setup_signal_handlers()
+
+    # Get credentials from environment variables
+    username = os.environ.get("PROJECTX_USERNAME")
+    api_key = os.environ.get("PROJECTX_API_KEY")
+    environment = os.environ.get("PROJECTX_ENVIRONMENT")
+
+    if not username or not api_key:
+        logger.error("Please set PROJECTX_USERNAME and PROJECTX_API_KEY environment variables")
+        return 1
+
+    # Contract to subscribe to
+    contract_id = "CON.F.US.ENQ.H25"  # E-mini NASDAQ-100 March 2025
+
+    try:
+        # Create and authenticate client
+        logger.info("Initializing client...")
+        client = ProjectXClient(username=username, api_key=api_key, environment=environment)
+
+        # Create realtime service
+        logger.info("Setting up realtime connections...")
+        realtime = RealtimeService(client)
+
+        # Access the market hub (lazy initialization)
+        market_hub = realtime.market
+
+        # Subscribe to market data events
+        logger.info(f"Subscribing to market data for {contract_id}...")
+        market_hub.subscribe_quotes(contract_id, handle_quote)
+        market_hub.subscribe_trades(contract_id, handle_trade)
+        market_hub.subscribe_market_depth(contract_id, handle_depth)
+
+        # Start the connection
+        logger.info("Starting realtime connection...")
+        realtime.start()
+
+        # Main loop - keep the program running
+        logger.info("Listening for market events. Press Ctrl+C to exit.")
+        while running:
+            time.sleep(1)
+
+    except KeyboardInterrupt:
+        logger.info("Keyboard interrupt received, shutting down...")
+    except Exception as e:
+        logger.exception(f"Error in realtime example: {e}")
+        return 1
+    finally:
+        # Clean up
+        if "realtime" in locals():
+            logger.info("Stopping realtime connections...")
+            try:
+                realtime.stop()
+            except Exception as e:
+                logger.error(f"Error stopping realtime service: {e}")
+
+    logger.info("Example completed successfully")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())