Eliminating Latency and Ingestion Hallucinations in High-Frequency Order Feeds with Pydantic v2 Core

                    ┌────────────────────────────────────────┐
                    │ Raw Market Data / Broker FIX Feeds     │
                    │ (ITCH, OUCH, FIX 4.4, JSON WebSockets) │
                    └───────────────────┬────────────────────┘
                                        │
                                        ▼
             ┌──────────────────────────────────────────────────────┐
             │       Pydantic v2 Validation Perimeter Layer         │
             │       (Rust-backed pydantic-core parsing)            │
             ├──────────────────────────────────────────────────────┤
             │ • Memory-view parsing & Type Coercion (strict=True)  │
             │ • Mathematical Constraints (Decimal, Precision <= 6) │
             │ • Cross-Field Invariant Check (Bid < Ask, Spreads)   │
             │ • Bi-Temporal Nanosecond Sequencing                  │
             └───────────┬──────────────────────────────┬───────────┘
                         │                              │
         [Valid Payloads]│                              │[Corrupted / Out-of-Order]
                         ▼                              ▼
┌─────────────────────────────────┐    ┌───────────────────────────────────┐
│ Apache Kafka / Redpanda Cluster │    │ Dead-Letter Queue (DLQ) & Audit   │
│ - high-throughput execution log │    │ - Immutable audit log             │
│ - Partitioned by Symbol / Venue │    │ - Real-time compliance alerting   │
└────────────────┬────────────────┘    └───────────────────────────────────┘
                 │
                 ▼
┌────────────────────────────────────────────────────────────────────────┐
│ Downstream Consumers: Core Matching Engine, C++ Trading Gateway,       │
│ Apache Iceberg / Delta Lake Bi-Temporal Trade Reconstruction Lakehouse │
└────────────────────────────────────────────────────────────────────────┘

"""
Production Market Data Integrity Contract (Pydantic v2)
Author: ChemData & AI Consulting Ltd
Compatibility: Python 3.10+, Pydantic v2.6+
"""

from decimal import Decimal
from typing import Annotated
from enum import Enum
from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    field_validator,
    model_validator,
)


class InstrumentClass(str, Enum):
    EQUITY = "EQUITY"
    FX_SPOT = "FX_SPOT"
    COMMODITY_FUTURE = "COMMODITY_FUTURE"
    CRYPTO_DERIVATIVE = "CRYPTO_DERIVATIVE"


class OrderSide(str, Enum):
    BUY = "BUY"
    SELL = "SELL"


class LimitOrderBookTick(BaseModel):
    """
    High-throughput Level 2 market data tick representation.
    Enforces strict decimal bounds, positive spread invariants, and
    bi-temporal timestamp coherence.
    """

    model_config = ConfigDict(
        strict=True,
        frozen=True,
        str_strip_whitespace=True,
        use_enum_values=True,
        extra="forbid",
    )

    # Identifiers
    symbol: Annotated[str, Field(pattern=r"^[A-Z0-9]{2,10}(/[A-Z0-9]{2,10})?$")]
    venue: Annotated[str, Field(min_length=3, max_length=12)]
    instrument_type: InstrumentClass

    # Bi-Temporal Timestamps (Unix Epoch in Nanoseconds)
    exchange_timestamp_ns: Annotated[int, Field(gt=0)]
    arrival_timestamp_ns: Annotated[int, Field(gt=0)]

    # Pricing & Depths (Strict Decimals to prevent IEEE-754 precision leakage)
    bid_price: Annotated[Decimal, Field(gt=Decimal("0.00000000"))]
    bid_size: Annotated[Decimal, Field(gt=Decimal("0.00000000"))]
    ask_price: Annotated[Decimal, Field(gt=Decimal("0.00000000"))]
    ask_size: Annotated[Decimal, Field(gt=Decimal("0.00000000"))]
    sequence_number: Annotated[int, Field(ge=0)]

    @field_validator("bid_price", "ask_price", mode="before")
    @classmethod
    def convert_and_round_decimal(cls, value: object) -> Decimal:
        """
        Guarantees incoming numeric values are parsed into exact Decimal
        representations without floating-point intermediary contamination.
        """
        if isinstance(value, float):
            return Decimal(str(value))
        if isinstance(value, (str, int)):
            return Decimal(value)
        raise ValueError(f"Unacceptable price type: {type(value)}")

    @model_validator(mode="after")
    def validate_market_invariants(self) -> "LimitOrderBookTick":
        """
        Cross-field business invariants:
        1. Bid price must strictly be lower than ask price (No inverted/crossed book).
        2. Local arrival timestamp must be >= exchange publication timestamp.
        3. Maximum tick spread cannot exceed 15% of mid-price (anomalous spike filter).
        """
        # 1. Crossed Book Prevention
        if self.bid_price >= self.ask_price:
            raise ValueError(
                f"Crossed/Inverted book detected for {self.symbol}: "
                f"bid={self.bid_price} >= ask={self.ask_price}"
            )

        # 2. Causality Verification (Arrival cannot precede generation)
        # Allows up to 5ms negative margin for cross-datacenter NTP clock skew
        clock_skew_ns = 5_000_000
        if (self.exchange_timestamp_ns - self.arrival_timestamp_ns) > clock_skew_ns:
            raise ValueError(
                f"Causal violation: Exchange timestamp ({self.exchange_timestamp_ns}) "
                f"exceeds arrival ({self.arrival_timestamp_ns}) beyond skew threshold."
            )

        # 3. Anomalous Spread Breach Filter
        mid_price = (self.bid_price + self.ask_price) / Decimal("2")
        spread = self.ask_price - self.bid_price
        if (spread / mid_price) > Decimal("0.15"):
            raise ValueError(
                f"Anomalous spread percentage breach ({spread / mid_price:.4%}) "
                f"detected for {self.symbol}"
            )

        return self