Real-time Data Validation

The "Why": Understanding Data Corruption

Before we write any code, it's crucial to understand what kind of errors we're looking for. Tick data can be corrupted in numerous ways.

To get a solid overview of the problem space, please read the following article from QuantPedia. It outlines the common types of errors found in high-frequency data.

From the article and considering our specific use case (real-time options data), here are the key validation checks we need to perform:

• Sanity Checks: Prices, volumes, and implied volatility must not be negative.
• Logical Consistency: The bid price must not be greater than the ask price.
• Domain-Specific Range Checks: As you know from your options experience, Greeks have expected ranges. For example, a call option's delta should be between 0 and 1, and a put's between -1 and 0. Vega and Gamma should be positive.
• Staleness: A check to ensure the data isn't from a significant delay in the feed.

The "How": An Anti-Corruption Layer with Pydantic

Now that we know what to validate, how do we implement it efficiently? We could write a series of if/else statements, but that quickly becomes cumbersome and hard to maintain. A much cleaner, more professional approach is to define the "schema" of our data declaratively and let a specialized library handle the validation.

For this, we will use Pydantic, a standard library in modern Python development for data validation. We'll use it to create an Anti-Corruption Layer (ACL)—an architectural pattern that isolates our clean, internal system from messy external data. The dev.to article below explains this concept and why Pydantic is the perfect tool for the job.

The core ideas are:

1. We will replace our simple TickData dataclass with a pydantic.BaseModel.
2. This model will declaratively define the data types and validation rules for every field.
3. When new data arrives, we will attempt to parse it using this model. If validation passes, the data is clean.
4. If validation fails, Pydantic will raise a ValidationError. We will catch this, log the bad data for later analysis (the "Quarantine Pattern"), and prevent it from being published to our strategies.

Implementation: Upgrading Our Data Handler

Let's put this into practice. We'll be modifying two files: framework/data_types.py and framework/market_data.py.

Step 1: Replace TickData with a Pydantic Model

First, make sure you have Pydantic installed: pip install pydantic.

Now, open framework/data_types.py and replace the existing TickData dataclass with this new Pydantic-based version.

# In framework/data_types.py
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Optional

# (Keep the Position and CreditSpreadPosition classes if they are in this file)

class PydanticTickData(BaseModel):
    """
    A Pydantic model to represent and validate a single tick of market data.
    This acts as our Anti-Corruption Layer.
    """
    timestamp: datetime
    contract_id: int
    
    # Use Field to add validation rules. Prices must be non-negative.
    bid_price: float = Field(ge=0)
    ask_price: float = Field(ge=0)
    last_price: Optional[float] = Field(default=None, ge=0)
    
    # Volumes must be non-negative integers
    bid_size: int = Field(ge=0)
    ask_size: int = Field(ge=0)
    last_size: Optional[int] = Field(default=None, ge=0)
    
    volume: int = Field(ge=0)
    
    # Greeks and IV validation
    delta: Optional[float] = Field(default=None, le=1.0, ge=-1.0)
    gamma: Optional[float] = Field(default=None, ge=0)
    vega: Optional[float] = Field(default=None, ge=0)
    theta: Optional[float] = None # Can be negative
    implied_volatility: Optional[float] = Field(default=None, ge=0)

    # A custom validator to ensure the bid/ask spread is logical
    @field_validator('ask_price')
    def check_bid_ask_spread(cls, ask_price, values):
        # The 'values' dict contains the fields that have already been validated.
        # Pydantic v2 passes a ValidationInfo object, from which we can get the data dict.
        data = values.data
        if 'bid_price' in data and ask_price > 0 and data['bid_price'] > 0:
            if data['bid_price'] > ask_price:
                raise ValueError('Bid price cannot be greater than ask price.')
        return ask_price

    class Config:
        # Forbid any extra fields not defined in the model.
        # This helps catch unexpected changes in the data source API.
        extra = 'forbid'

# You can now remove the old TickData dataclass.
# We also need to update the MarketDataEvent to expect this new type.
# In framework/event_system.py, change the MarketDataEvent class:
#
# class MarketDataEvent(Event):
#    def __init__(self, tick_data: PydanticTickData):  <-- Update this type hint
#        self.type = 'MARKET_DATA'
#        self.tick_data = tick_data

This model is far more powerful than our old dataclass. It declaratively enforces that prices and volumes are non-negative, delta is within its [-1, 1] range, and that the bid is not higher than the ask. The extra = 'forbid' config is particularly useful, as it will raise an error if the incoming data contains unexpected new fields, alerting us to a potential schema change from the broker.

Step 2: Implement the Quarantine Pattern in the MarketDataHandler

Now, let's update MarketDataHandler to use our new PydanticTickData model and implement the quarantine logic.

# In framework/market_data.py
import logging
from pydantic import ValidationError

from .data_types import PydanticTickData # Import our new model
from .event_system import EventEngine, MarketDataEvent, TICK_EVENT_TOPIC_PREFIX
from datetime import datetime
from ib_insync import Ticker

# ... (MarketDataSubscriber class remains the same) ...

class MarketDataHandler:
    """
    Parses, validates, and publishes incoming Ticker objects as MarketDataEvents.
    Implements the Quarantine Pattern for invalid data.
    """
    def __init__(self, event_engine: EventEngine):
        self.event_engine = event_engine

    def process_incoming_ticker(self, ticker: Ticker):
        """
        Parses, validates a Ticker object, and posts a MarketDataEvent if valid.
        If invalid, the data is 'quarantined' by logging the error.
        """
        ts = datetime.utcnow()
        contract_id = ticker.contract.conId

        # 1. Create a raw data dictionary from the ticker object
        raw_data = {
            "timestamp": ts,
            "contract_id": contract_id,
            "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 None,
            "bid_size": int(ticker.bidSize) if ticker.bidSize == ticker.bidSize else 0,
            "ask_size": int(ticker.askSize) if ticker.askSize == ticker.askSize else 0,
            "last_size": int(ticker.lastSize) if ticker.lastSize == ticker.lastSize else None,
            "volume": int(ticker.volume) if ticker.volume == ticker.volume else 0,
            # Greeks are often in ticker.modelGreeks
            "delta": ticker.modelGreeks.delta if ticker.modelGreeks and ticker.modelGreeks.delta == ticker.modelGreeks.delta else None,
            "gamma": ticker.modelGreeks.gamma if ticker.modelGreeks and ticker.modelGreeks.gamma == ticker.modelGreeks.gamma else None,
            "vega": ticker.modelGreeks.vega if ticker.modelGreeks and ticker.modelGreeks.vega == ticker.modelGreeks.vega else None,
            "theta": ticker.modelGreeks.theta if ticker.modelGreeks and ticker.modelGreeks.theta == ticker.modelGreeks.theta else None,
            "implied_volatility": ticker.modelGreeks.impliedVolatility if ticker.modelGreeks and ticker.modelGreeks.impliedVolatility == ticker.modelGreeks.impliedVolatility else None,
        }

        # 2. Attempt to validate the raw data against our Pydantic model
        try:
            validated_tick = PydanticTickData(**raw_data)
            
            # 3. If validation succeeds, create and post the event
            market_data_event = MarketDataEvent(tick_data=validated_tick)
            topic = f"{TICK_EVENT_TOPIC_PREFIX}.{contract_id}"
            self.event_engine.post(market_data_event, topic)

        except ValidationError as e:
            # 4. If validation fails, quarantine the data
            logging.warning(
                f"Data validation failed for contract {contract_id}. "
                f"Quarantining data. Error: {e}. Raw Data: {raw_data}"
            )
            # The event is NOT posted, protecting the rest of the system.

With these changes, our system is now significantly more robust. The MarketDataHandler acts as a strict guard. Only data that conforms perfectly to our PydanticTickData model will ever be published as an event. Any malformed data is logged for inspection without crashing the application or polluting our strategies.

# In PydanticTickData class within framework/data_types.py

class PydanticTickData(BaseModel):
    # ... other fields ...

    implied_volatility: Optional[float] = Field(default=None, ge=0, le=5.0) # Add le=5.0

    # ... rest of the class ...

Conclusion

Excellent work today. You have successfully implemented a critical component for ensuring the reliability of your trading system. By integrating data validation directly into the data handling pipeline, you prevent a whole class of potential errors from ever affecting your strategy logic.

Key Takeaways:

• Data Integrity is paramount: A trading system is only as good as the data it uses.
• Know your enemy: We identified common types of bad data in real-time feeds, like negative prices, illogical spreads, and out-of-range values.
• Use the right tool for the job: Pydantic allows us to create a declarative, readable, and powerful Anti-Corruption Layer by defining our data schema as a BaseModel.
• Fail gracefully with the Quarantine Pattern: Instead of crashing or ignoring bad data, we catch ValidationError, log the problematic data for analysis, and continue operating with clean data.

Preview of the Next Module:
We have now completed the entire "Real-Time Market Data Handling" module. We have a pipeline that can connect to a broker, subscribe to data, parse it, validate it, and publish it through a robust event-driven system. The foundation is solid.

In our next lesson, we will begin Module 4: Options Analytics for Spreads. We will start by leveraging our clean data stream to "Integrate a financial library (e.g., py_vollib) to calculate option prices and Greeks." This will be the first step in adding the analytical intelligence our strategies need to make decisions.