Quant Trading Systems: Architecture & Infrastructure › Data Infrastructure

from collections import deque
from dataclasses import dataclass
from datetime import datetime
from typing import Optional


@dataclass
class MarketDataTick:
    """Represents a single market data update."""

    symbol: str
    timestamp: datetime
    bid_price: float
    bid_size: int
    ask_price: float
    ask_size: int
    last_price: Optional[float] = None
    last_size: Optional[int] = None

    @property
    def mid_price(self) -> float:
        return (self.bid_price + self.ask_price) / 2

    @property
    def spread(self) -> float:
        return self.ask_price - self.bid_price

    @property
    def spread_bps(self) -> float:
        return (self.spread / self.mid_price) * 10000


class MarketDataHandler:
    """Handles incoming market data with validation and buffering."""

    def __init__(self, buffer_size: int = 1000):
        self.current_quotes: dict[str, MarketDataTick] = {}
        self.tick_buffer: dict[str, deque] = {}
        self.buffer_size = buffer_size
        self.stale_threshold_seconds = 5.0

    def on_tick(self, tick: MarketDataTick) -> bool:
        """Process incoming tick with validation. Returns True if valid."""
        # Validate tick data
        if not self._validate_tick(tick):
            return False

        # Initialize buffer if needed
        if tick.symbol not in self.tick_buffer:
            self.tick_buffer[tick.symbol] = deque(maxlen=self.buffer_size)

        # Update current quote and buffer
        self.current_quotes[tick.symbol] = tick
        self.tick_buffer[tick.symbol].append(tick)
        return True

    def _validate_tick(self, tick: MarketDataTick) -> bool:
        """Validate tick data for common errors."""
        # Check for crossed markets (bid > ask)
        if tick.bid_price >= tick.ask_price:
            return False

        # Check for zero or negative prices
        if tick.bid_price <= 0 or tick.ask_price <= 0:
            return False

        # Check for unreasonable spread (> 10%)
        if tick.spread_bps > 1000:
            return False

        return True

    def is_data_stale(self, symbol: str, current_time: datetime) -> bool:
        """Check if data for a symbol is stale."""
        if symbol not in self.current_quotes:
            return True

        last_update = self.current_quotes[symbol].timestamp
        age = (current_time - last_update).total_seconds()
        return age > self.stale_threshold_seconds

# Demonstrate market data handling
handler = MarketDataHandler(buffer_size=100)

# Simulate incoming ticks
sample_ticks = [
    MarketDataTick(
        "AAPL",
        datetime(2024, 1, 15, 9, 30, 0),
        185.50,
        100,
        185.52,
        200,
        185.51,
        50,
    ),
    MarketDataTick(
        "AAPL",
        datetime(2024, 1, 15, 9, 30, 1),
        185.48,
        150,
        185.51,
        100,
        185.50,
        75,
    ),
    MarketDataTick(
        "AAPL", datetime(2024, 1, 15, 9, 30, 2), 185.55, 200, 185.45, 100
    ),  # Invalid: crossed
]

tick_log = []
for tick in sample_ticks:
    valid = handler.on_tick(tick)
    tick_log.append(
        f"Tick at {tick.timestamp.time()}: bid={tick.bid_price}, ask={tick.ask_price}, valid={valid}"
    )

current_aapl = None
spread_bps = 0.0
if "AAPL" in handler.current_quotes:
    current_aapl = handler.current_quotes["AAPL"]
    spread_bps = current_aapl.spread_bps

from datetime import timedelta


@dataclass
class AlternativeDataPoint:
    """Represents an alternative data observation."""

    symbol: str
    data_type: str
    observation_date: datetime  # When the data relates to
    available_date: datetime  # When the data became available
    value: float

    @property
    def publication_lag(self) -> timedelta:
        """Time between observation and availability."""
        return self.available_date - self.observation_date


class PointInTimeDatabase:
    """
    Database wrapper ensuring point-in-time correctness.
    Only returns data that was available at the query time.
    """

    def __init__(self):
        self.data: list[AlternativeDataPoint] = []

    def insert(self, data_point: AlternativeDataPoint):
        self.data.append(data_point)

    def query(
        self, symbol: str, data_type: str, as_of: datetime
    ) -> Optional[AlternativeDataPoint]:
        """
        Get the most recent data point that was available as of the given time.
        This prevents look-ahead bias in backtesting and research.
        """
        relevant = [
            d
            for d in self.data
            if d.symbol == symbol
            and d.data_type == data_type
            and d.available_date <= as_of  # Only data that was available
        ]

        if not relevant:
            return None

        # Return most recent available observation
        return max(relevant, key=lambda d: d.available_date)

# Demonstrate point-in-time correctness
pit_db = PointInTimeDatabase()

# Insert earnings surprise data with publication lag
pit_db.insert(
    AlternativeDataPoint(
        symbol="AAPL",
        data_type="earnings_surprise",
        observation_date=datetime(2024, 1, 25),  # Earnings call date
        available_date=datetime(
            2024, 1, 25, 16, 30
        ),  # Available after market close
        value=0.05,  # 5% positive surprise
    )
)

# Query at different times
before_release = datetime(2024, 1, 25, 12, 0)
after_release = datetime(2024, 1, 25, 17, 0)

result_before = pit_db.query("AAPL", "earnings_surprise", before_release)
result_after = pit_db.query("AAPL", "earnings_surprise", after_release)

import pandas as pd

# Demonstrate efficient data organization patterns
np.random.seed(42)

# Create sample OHLCV data
dates = pd.date_range("2024-01-01", "2024-01-31", freq="D")
n_days = len(dates)

ohlcv_data = pd.DataFrame(
    {
        "date": dates,
        "open": 100 + np.cumsum(np.random.randn(n_days) * 0.5),
        "high": 0.0,  # Will calculate
        "low": 0.0,  # Will calculate
        "close": 0.0,  # Will calculate
        "volume": np.random.randint(1000000, 5000000, n_days),
    }
)

# Generate realistic OHLC from open
for i in range(n_days):
    daily_range = np.random.uniform(0.5, 2.0)
    ohlcv_data.loc[i, "high"] = ohlcv_data.loc[i, "open"] + np.random.uniform(
        0, daily_range
    )
    ohlcv_data.loc[i, "low"] = ohlcv_data.loc[i, "open"] - np.random.uniform(
        0, daily_range
    )
    ohlcv_data.loc[i, "close"] = np.random.uniform(
        ohlcv_data.loc[i, "low"], ohlcv_data.loc[i, "high"]
    )

ohlcv_data = ohlcv_data.set_index("date")
mem_usage_kb = ohlcv_data.memory_usage(deep=True).sum() / 1024