Parsing Streaming Tick Data

From Raw Ticker to Structured Data

A "tick" is the smallest possible price movement in a market. A stream of tick data represents every individual change in price or volume.

The ib_insync library simplifies this for us. Instead of processing every tiny raw message, it provides a stateful Ticker object that is updated whenever new information arrives. Our job, then, is not to build a tick-by-tick stream from scratch, but to parse this Ticker object each time it's updated.

The core principle is to create an abstraction layer. Our DataHandler will act as a translator, converting the broker-specific Ticker object into our own internal format. This makes our system modular: if we ever decide to switch brokers, we would only need to update the data handler, leaving our strategy logic untouched.

The following resource uses the native ibapi to demonstrate a similar pattern. It creates a dedicated class to receive tick data and populates a Pandas DataFrame. While we will use a different data structure, the concept of a dedicated handler that parses and stores incoming data is identical.

Designing Our TickData Structure

Before we build the parser, let's define what we want the output to look like. We need a simple structure to hold the essential information from a market data update. Python's dataclasses are an excellent tool for this, as they provide a clean and explicit way to define data-centric objects.

We'll create a TickData class in a new file, framework/data_types.py. This keeps our custom data structures organized.

# In a new file: framework/data_types.py
from dataclasses import dataclass
from datetime import datetime

@dataclass
class TickData:
    """
    Represents a single snapshot of market data for a contract.
    """
    timestamp: datetime
    contract_id: int
    bid_price: float
    ask_price: float
    last_price: float
    bid_size: int
    ask_size: int
    last_size: int
    volume: int

This TickData object is our internal standard. Any component in our system that needs market data will expect to receive an object of this type. It's clean, self-contained, and completely independent of Interactive Brokers or ib_insync.

Implementing the DataHandler

Now we can create the handler itself. This will be a new class, MarketDataHandler, in our existing framework/market_data.py file. Its job is to perform the translation from ib_insync.Ticker to our TickData format.

# In framework/market_data.py
# Add these imports at the top
from .data_types import TickData 
from datetime import datetime

# ... (The MarketDataSubscriber class from the previous lesson remains here) ...

class MarketDataHandler:
    """
    Parses incoming Ticker objects into a structured TickData format.
    """
    def process_incoming_ticker(self, ticker: Ticker):
        """
        Parses an ib_insync Ticker object and handles the resulting TickData.
        
        :param ticker: The updated Ticker object from ib_insync.
        """
        # The Ticker object's time attribute can be naive, so we should
        # make it timezone-aware (or use UTC) for consistency.
        # For simplicity, we'll use the current UTC time when the tick is processed.
        # A more robust solution might involve timezone conversions based on the market.
        ts = datetime.utcnow()

        # Handle potential NaN values for fields that may not be available yet.
        # Replace NaN with -1 or 0 as appropriate for numeric types.
        bid_price = ticker.bid if ticker.bid == ticker.bid else -1.0
        ask_price = ticker.ask if ticker.ask == ticker.ask else -1.0
        last_price = ticker.last if ticker.last == ticker.last else -1.0
        
        bid_size = ticker.bidSize if ticker.bidSize == ticker.bidSize else 0
        ask_size = ticker.askSize if ticker.askSize == ticker.askSize else 0
        last_size = ticker.lastSize if ticker.lastSize == ticker.lastSize else 0
        
        volume = ticker.volume if ticker.volume == ticker.volume else 0

        # Create our standardized TickData object
        tick_event = TickData(
            timestamp=ts,
            contract_id=ticker.contract.conId,
            bid_price=bid_price,
            ask_price=ask_price,
            last_price=last_price,
            bid_size=bid_size,
            ask_size=ask_size,
            last_size=last_size,
            volume=volume
        )
        
        # For now, we just print the structured data.
        # In the next lesson, this is where we'll push the tick_event
        # into a central event queue for other components to consume.
        print(f"Parsed Tick: {tick_event}")

Key points in the MarketDataHandler:

• Timestamping: We capture the timestamp using datetime.utcnow(). While the Ticker object has a time attribute, it can sometimes be naive or inconsistent, so using a unified timestamp upon processing is a more robust practice.
• Handling NaN: Market data feeds don't always provide every field in every update. ib_insync uses NaN (Not a Number) for missing values. Our TickData class expects floats and integers. The check ticker.bid == ticker.bid is a clever Python trick to test for NaN, because NaN is the only value not equal to itself. We replace missing values with sensible defaults like -1.0 for price and 0 for size.
• Output: The method culminates in creating a TickData instance. This structured object is the final product of our parsing logic.

Integrating the Handler with the Subscriber

Now, we need to connect this new handler to the MarketDataSubscriber we built in the last lesson. We'll modify our main application logic to instantiate the handler and pass its processing method to the subscriber.

Here’s how the updated MarketDataSubscriber and the integration would look. Notice we are simply connecting the pieces we've already built.

# In framework/market_data.py

# ... (keep MarketDataHandler and data_types import) ...

class MarketDataSubscriber:
    """
    Manages real-time market data subscriptions using ib_insync.
    """
    def __init__(self, ib_client: IB):
        self.ib = ib_client
        self.tickers: Dict[int, Ticker] = {}
        # The data_handler will now be our MarketDataHandler's method
        self.data_handler: Callable[[Ticker], None] = None

    def subscribe_contracts(self, contracts: List[Option]):
        for contract in contracts:
            if contract.conId in self.tickers:
                print(f"Already subscribed to {contract.localSymbol}.")
                continue

            print(f"Subscribing to market data for: {contract.localSymbol}")
            
            ticker = self.ib.reqMktData(contract, genericTickList='', snapshot=False, regulatorySnapshot=False)
            
            self.tickers[contract.conId] = ticker
            
            # Here's the key: we hook the on_ticker_update event
            ticker.updateEvent += self.on_ticker_update

    def on_ticker_update(self, ticker: Ticker):
        """
        This method is called by ib_insync whenever a subscribed ticker is updated.
        It delegates the processing to the registered data_handler.
        """
        if self.data_handler:
            self.data_handler(ticker)

    def set_data_handler(self, handler: Callable[[Ticker], None]):
        """
        Sets the function to handle incoming ticker data.
        
        :param handler: A function that takes a Ticker object as an argument.
        """
        self.data_handler = handler

# --- Example of integration in your main file ---
# if __name__ == '__main__':
#     ib = IB()
#     ib.connect('127.0.0.1', 7497, clientId=3) # Use a new client ID
#
#     # Define contracts to monitor
#     spy_call = Option('SPY', '20241220', 550, 'C', 'SMART')
#     ib.qualifyContracts(spy_call)
#
#     # 1. Instantiate the handler
#     data_handler = MarketDataHandler()
#
#     # 2. Instantiate the subscriber
#     subscriber = MarketDataSubscriber(ib)
#
#     # 3. Connect the handler to the subscriber
#     subscriber.set_data_handler(data_handler.process_incoming_ticker)
#
#     # 4. Subscribe to contracts
#     subscriber.subscribe_contracts([spy_call])
#
#     print("Subscription is live. Waiting for parsed market data...")
#     try:
#         ib.run()
#     except (KeyboardInterrupt, SystemExit):
#         print("System shutdown requested.")
#         ib.disconnect()

This design beautifully separates concerns. The MarketDataSubscriber knows how to get data from IBKR. The MarketDataHandler knows how to parse it. Neither needs to know about the other's internal workings.

Conclusion

In this lesson, we created a critical piece of our trading framework's foundation: the DataHandler. By parsing the raw, broker-specific Ticker objects into a clean, standardized TickData dataclass, we have successfully decoupled our future strategy logic from our data source.

Key Takeaways:

• A DataHandler acts as a translator, creating an essential abstraction layer in a trading system.
• Defining a standardized internal data format (like our TickData dataclass) makes the system modular and easier to maintain.
• Parsing real-world data requires handling missing values (NaN) and being mindful of details like timestamps.
• Separating the Subscriber (what gets data) from the Handler (what parses data) is a robust design pattern that promotes clean, reusable code.

Preview of the Next Lesson:
Our DataHandler currently just prints the TickData object. This confirms our parsing works, but the data isn't going anywhere useful. To make it available to our strategies, we need a distribution mechanism. In the next lesson, "Design an event-driven system to publish market data updates to the relevant strategy instances," we will implement an event queue. Our DataHandler will place the TickData objects into this queue, and other components, like our strategy classes, will be able to subscribe to it and receive the data they need to make trading decisions.