Quant Trading Systems: Architecture & Infrastructure

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


@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,  # Will calculate
        "low": 0,  # Will calculate
        "close": 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

from abc import ABC, abstractmethod
from enum import Enum


class Signal(Enum):
    """Trading signal types."""

    LONG = 1
    NEUTRAL = 0
    SHORT = -1


class StrategyBase(ABC):
    """Abstract base class for trading strategies."""

    def __init__(self, symbols: list[str]):
        self.symbols = symbols
        self.positions: dict[str, float] = {s: 0 for s in symbols}
        self.signals: dict[str, Signal] = {s: Signal.NEUTRAL for s in symbols}

    @abstractmethod
    def on_data(self, data: dict) -> dict[str, Signal]:
        """Process new data and generate signals."""
        pass

    @abstractmethod
    def calculate_target_positions(self) -> dict[str, float]:
        """Convert signals to target position sizes."""
        pass


class MomentumStrategy(StrategyBase):
    """
    Simple momentum strategy for demonstration.
    Goes long when returns exceed threshold, short when below.
    """

    def __init__(
        self, symbols: list[str], lookback: int = 20, threshold: float = 0.02
    ):
        super().__init__(symbols)
        self.lookback = lookback
        self.threshold = threshold
        self.price_history: dict[str, list] = {s: [] for s in symbols}

    def on_data(self, data: dict) -> dict[str, Signal]:
        """Update price history and generate signals."""
        signals = {}

        for symbol in self.symbols:
            if symbol not in data:
                signals[symbol] = Signal.NEUTRAL
                continue

            # Update price history
            self.price_history[symbol].append(data[symbol])

            # Keep only lookback period
            if len(self.price_history[symbol]) > self.lookback:
                self.price_history[symbol] = self.price_history[symbol][
                    -self.lookback :
                ]

            # Calculate signal if we have enough history
            if len(self.price_history[symbol]) >= self.lookback:
                prices = np.array(self.price_history[symbol])
                momentum = (prices[-1] - prices[0]) / prices[0]

                if momentum > self.threshold:
                    signals[symbol] = Signal.LONG
                elif momentum < -self.threshold:
                    signals[symbol] = Signal.SHORT
                else:
                    signals[symbol] = Signal.NEUTRAL
            else:
                signals[symbol] = Signal.NEUTRAL

        self.signals = signals
        return signals

    def calculate_target_positions(self) -> dict[str, float]:
        """Equal weight positions based on signals."""
        n_active = sum(1 for s in self.signals.values() if s != Signal.NEUTRAL)
        if n_active == 0:
            return {s: 0 for s in self.symbols}

        weight = 1.0 / n_active
        return {
            symbol: weight * signal.value
            for symbol, signal in self.signals.items()
        }

# Simulate strategy execution
strategy = MomentumStrategy(["SPY", "QQQ"], lookback=5, threshold=0.01)

# Simulate 10 days of price data
np.random.seed(42)
spy_prices = 450 * np.cumprod(1 + np.random.randn(10) * 0.01)
qqq_prices = 380 * np.cumprod(1 + np.random.randn(10) * 0.015)

execution_log = []
for day in range(10):
    data = {"SPY": spy_prices[day], "QQQ": qqq_prices[day]}
    signals = strategy.on_data(data)
    positions = strategy.calculate_target_positions()

    log_entry = (
        f"{day + 1:3d} | {spy_prices[day]:9.2f} | {qqq_prices[day]:9.2f} | "
        f"{signals['SPY'].name:10s} | {signals['QQQ'].name:10s} | "
        f"SPY: {positions['SPY']:+.2f}, QQQ: {positions['QQQ']:+.2f}"
    )
    execution_log.append(log_entry)

import json
from pathlib import Path


@dataclass
class StrategyState:
    """Complete state snapshot for a strategy."""

    timestamp: datetime
    positions: dict[str, float]
    signals: dict[str, int]
    open_orders: list[dict]
    pnl_today: float
    pnl_total: float
    max_drawdown: float

    def to_dict(self) -> dict:
        return {
            "timestamp": self.timestamp.isoformat(),
            "positions": self.positions,
            "signals": self.signals,
            "open_orders": self.open_orders,
            "pnl_today": self.pnl_today,
            "pnl_total": self.pnl_total,
            "max_drawdown": self.max_drawdown,
        }

    @classmethod
    def from_dict(cls, data: dict) -> "StrategyState":
        data["timestamp"] = datetime.fromisoformat(data["timestamp"])
        return cls(**data)


class StateManager:
    """Manages strategy state persistence and recovery."""

    def __init__(self, state_dir: str = "./state"):
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(exist_ok=True)

    def save_state(self, strategy_id: str, state: StrategyState):
        """Persist state to disk."""
        filepath = self.state_dir / f"{strategy_id}_state.json"
        with open(filepath, "w") as f:
            json.dump(state.to_dict(), f, indent=2)

    def load_state(self, strategy_id: str) -> Optional[StrategyState]:
        """Load state from disk if it exists."""
        filepath = self.state_dir / f"{strategy_id}_state.json"
        if not filepath.exists():
            return None
        with open(filepath, "r") as f:
            return StrategyState.from_dict(json.load(f))

    def checkpoint(self, strategy_id: str, state: StrategyState):
        """
        Create a timestamped checkpoint for recovery.
        Useful for debugging and audit trails.
        """
        timestamp_str = state.timestamp.strftime("%Y%m%d_%H%M%S")
        filepath = (
            self.state_dir / f"{strategy_id}_checkpoint_{timestamp_str}.json"
        )
        with open(filepath, "w") as f:
            json.dump(state.to_dict(), f, indent=2)

@dataclass
class Order:
    """Represents a trading order."""

    symbol: str
    side: str  # 'BUY' or 'SELL'
    quantity: float
    order_type: str  # 'MARKET', 'LIMIT', etc.
    limit_price: Optional[float] = None


class RiskLimits:
    """Configuration for risk limits."""

    max_position_size: float = 100000  # Max notional per position
    max_order_size: float = 50000  # Max notional per order
    max_portfolio_exposure: float = 500000  # Total long + short notional
    max_concentration: float = 0.2  # Max weight in single position
    max_daily_loss: float = 10000  # Stop trading if exceeded
    max_drawdown: float = 0.05  # 5% drawdown limit


class PreTradeRiskChecker:
    """Validates orders against risk limits before submission."""

    def __init__(self, limits: RiskLimits):
        self.limits = limits

    def check_order(
        self,
        order: Order,
        current_position: float,
        portfolio_value: float,
        current_price: float,
        total_exposure: float,
        daily_pnl: float,
    ) -> tuple[bool, str]:
        """
        Check if order passes all pre-trade risk checks.
        Returns (approved, reason).
        """
        order_notional = order.quantity * current_price

        # Check order size limit
        if order_notional > self.limits.max_order_size:
            return (
                False,
                f"Order size {order_notional:.0f} exceeds limit {self.limits.max_order_size:.0f}",
            )

        # Check resulting position size
        if order.side == "BUY":
            new_position = current_position + order.quantity
        else:
            new_position = current_position - order.quantity

        new_position_notional = abs(new_position * current_price)
        if new_position_notional > self.limits.max_position_size:
            return (
                False,
                f"Resulting position {new_position_notional:.0f} exceeds limit",
            )

        # Check concentration limit
        concentration = new_position_notional / portfolio_value
        if concentration > self.limits.max_concentration:
            return (
                False,
                f"Position concentration {concentration:.1%} exceeds {self.limits.max_concentration:.1%}",
            )

        # Check portfolio exposure
        new_exposure = total_exposure + order_notional
        if new_exposure > self.limits.max_portfolio_exposure:
            return (
                False,
                f"Portfolio exposure {new_exposure:.0f} would exceed limit",
            )

        # Check daily loss limit
        if daily_pnl < -self.limits.max_daily_loss:
            return False, "Daily loss limit breached, trading suspended"

        return True, "Order approved"

# Demonstrate pre-trade risk checks
limits = RiskLimits()
risk_checker = PreTradeRiskChecker(limits)

# Test various orders
test_orders = [
    (Order("AAPL", "BUY", 100, "MARKET"), 0, 185.0, "Normal order"),
    (Order("AAPL", "BUY", 1000, "MARKET"), 0, 185.0, "Large order"),
    (
        Order("AAPL", "BUY", 100, "MARKET"),
        500,
        185.0,
        "Would exceed position limit",
    ),
]

portfolio_value = 500000
total_exposure = 200000
daily_pnl = -2000

risk_results = []
for order, current_pos, price, description in test_orders:
    approved, reason = risk_checker.check_order(
        order, current_pos, portfolio_value, price, total_exposure, daily_pnl
    )
    status = "✓ APPROVED" if approved else "✗ REJECTED"
    risk_results.append(f"{description:30s} | {status:12s} | {reason}")

from scipy import stats


class PortfolioRiskMonitor:
    """Real-time monitoring of portfolio risk metrics."""

    def __init__(self, confidence_level: float = 0.95):
        self.confidence_level = confidence_level
        self.returns_history: list[float] = []
        self.peak_value: float = 0
        self.current_drawdown: float = 0

    def update(self, portfolio_value: float, portfolio_return: float):
        """Update risk metrics with new portfolio state."""
        self.returns_history.append(portfolio_return)

        # Update peak and drawdown
        if portfolio_value > self.peak_value:
            self.peak_value = portfolio_value
        self.current_drawdown = (
            self.peak_value - portfolio_value
        ) / self.peak_value

    def calculate_var(self, horizon_days: int = 1) -> float:
        """Calculate parametric VaR assuming normal returns."""
        if len(self.returns_history) < 20:
            return float("inf")  # Not enough data

        returns = np.array(self.returns_history)
        mu = returns.mean() * horizon_days
        sigma = returns.std() * np.sqrt(horizon_days)

        var = -(mu + sigma * stats.norm.ppf(1 - self.confidence_level))
        return var

    def calculate_expected_shortfall(self, horizon_days: int = 1) -> float:
        """Calculate Expected Shortfall (CVaR)."""
        if len(self.returns_history) < 20:
            return float("inf")

        returns = np.array(self.returns_history)
        mu = returns.mean() * horizon_days
        sigma = returns.std() * np.sqrt(horizon_days)

        alpha = 1 - self.confidence_level
        var = -(mu + sigma * stats.norm.ppf(1 - self.confidence_level))
        es = -mu + sigma * stats.norm.pdf(stats.norm.ppf(alpha)) / alpha
        return es

    def get_risk_summary(self) -> dict:
        """Get current risk metric summary."""
        return {
            "current_drawdown": self.current_drawdown,
            "var_1d_95": self.calculate_var(1),
            "es_1d_95": self.calculate_expected_shortfall(1),
            "realized_vol": np.std(self.returns_history) * np.sqrt(252)
            if len(self.returns_history) > 1
            else 0,
        }

# Simulate portfolio monitoring
np.random.seed(42)
monitor = PortfolioRiskMonitor(confidence_level=0.95)

# Simulate 60 days of returns
initial_value = 1000000
portfolio_value = initial_value

for day in range(60):
    # Simulate daily return
    daily_return = np.random.normal(0.0005, 0.015)  # ~24% annual vol
    portfolio_value *= 1 + daily_return
    monitor.update(portfolio_value, daily_return)

risk_summary = monitor.get_risk_summary()

# Demonstrate FIX message structure (simplified representation)
# Real FIX uses tag=value format with SOH delimiters


class SimpleFIXMessage:
    """Simplified FIX message builder for demonstration."""

    # Common FIX tags
    TAGS = {
        "BeginString": "8",
        "MsgType": "35",
        "SenderCompID": "49",
        "TargetCompID": "56",
        "ClOrdID": "11",
        "Symbol": "55",
        "Side": "54",
        "OrderQty": "38",
        "OrdType": "40",
        "Price": "44",
        "TransactTime": "60",
    }

    # Message types
    MSG_TYPES = {
        "NewOrderSingle": "D",
        "ExecutionReport": "8",
        "OrderCancelRequest": "F",
    }

    def __init__(self, msg_type: str):
        self.fields = {
            "BeginString": "FIX.4.4",
            "MsgType": self.MSG_TYPES.get(msg_type, msg_type),
        }

    def set_field(self, name: str, value: str):
        self.fields[name] = value
        return self

    def to_string(self) -> str:
        """Convert to FIX-like string representation."""
        parts = []
        for name, value in self.fields.items():
            tag = self.TAGS.get(name, name)
            parts.append(f"{tag}={value}")
        return "|".join(parts)


# Build a sample new order message
new_order = SimpleFIXMessage("NewOrderSingle")
new_order.set_field("SenderCompID", "QUANT_FIRM")
new_order.set_field("TargetCompID", "EXCHANGE")
new_order.set_field("ClOrdID", "ORD-2024-001")
new_order.set_field("Symbol", "AAPL")
new_order.set_field("Side", "1")  # 1 = Buy
new_order.set_field("OrderQty", "100")
new_order.set_field("OrdType", "2")  # 2 = Limit
new_order.set_field("Price", "185.50")

@dataclass
class VenueQuote:
    """Quote from a specific trading venue."""

    venue_id: str
    bid_price: float
    bid_size: int
    ask_price: float
    ask_size: int
    latency_ms: float  # Expected latency to venue
    fee_per_share: float  # Negative means rebate


class SmartOrderRouter:
    """Simple smart order router for demonstration."""

    def __init__(self, venues: list[str]):
        self.venues = venues
        self.venue_quotes: dict[str, VenueQuote] = {}

    def update_quote(self, quote: VenueQuote):
        """Update quote for a venue."""
        self.venue_quotes[quote.venue_id] = quote

    def route_order(
        self, side: str, quantity: int, strategy: str = "best_price"
    ) -> list[tuple[str, int]]:
        """
        Route order across venues.
        Returns list of (venue_id, quantity) allocations.
        """
        if not self.venue_quotes:
            return []

        if strategy == "best_price":
            return self._route_best_price(side, quantity)
        elif strategy == "minimize_impact":
            return self._route_minimize_impact(side, quantity)
        else:
            raise ValueError(f"Unknown routing strategy: {strategy}")

    def _route_best_price(
        self, side: str, quantity: int
    ) -> list[tuple[str, int]]:
        """Route to venues with best prices, sweeping through order book."""
        allocations = []
        remaining = quantity

        if side == "BUY":
            # Sort venues by ask price (lowest first), then by fee
            sorted_venues = sorted(
                self.venue_quotes.values(),
                key=lambda v: (v.ask_price, v.fee_per_share),
            )

            for venue in sorted_venues:
                if remaining <= 0:
                    break
                fill_qty = min(remaining, venue.ask_size)
                if fill_qty > 0:
                    allocations.append((venue.venue_id, fill_qty))
                    remaining -= fill_qty
        else:
            # Sort by bid price (highest first)
            sorted_venues = sorted(
                self.venue_quotes.values(),
                key=lambda v: (-v.bid_price, v.fee_per_share),
            )

            for venue in sorted_venues:
                if remaining <= 0:
                    break
                fill_qty = min(remaining, venue.bid_size)
                if fill_qty > 0:
                    allocations.append((venue.venue_id, fill_qty))
                    remaining -= fill_qty

        return allocations

    def _route_minimize_impact(
        self, side: str, quantity: int
    ) -> list[tuple[str, int]]:
        """Spread order across venues to minimize market impact."""
        total_liquidity = sum(
            v.ask_size if side == "BUY" else v.bid_size
            for v in self.venue_quotes.values()
        )

        if total_liquidity == 0:
            return []

        allocations = []
        for venue in self.venue_quotes.values():
            venue_liquidity = (
                venue.ask_size if side == "BUY" else venue.bid_size
            )
            proportion = venue_liquidity / total_liquidity
            venue_qty = int(quantity * proportion)
            if venue_qty > 0:
                allocations.append((venue.venue_id, venue_qty))

        return allocations

# Demonstrate smart order routing
router = SmartOrderRouter(["NYSE", "NASDAQ", "BATS", "IEX"])

# Set up venue quotes
quotes = [
    VenueQuote("NYSE", 185.48, 500, 185.50, 300, 0.5, 0.003),
    VenueQuote("NASDAQ", 185.47, 200, 185.49, 400, 0.3, -0.002),  # Rebate
    VenueQuote("BATS", 185.48, 300, 185.50, 200, 0.4, 0.001),
    VenueQuote("IEX", 185.47, 400, 185.51, 600, 0.8, 0.0009),
]

for quote in quotes:
    router.update_quote(quote)

# Route a buy order using different strategies
buy_qty = 500
allocations_best = router.route_order("BUY", buy_qty, "best_price")
allocations_impact = router.route_order("BUY", buy_qty, "minimize_impact")

# Latency budget analysis for different strategy types
latency_budgets = {
    "HFT Market Making": {
        "target_total_us": 10,  # microseconds
        "network_us": 2,
        "processing_us": 5,
        "serialization_us": 1,
        "buffer_us": 2,
        "typical_language": "C++/FPGA",
        "infrastructure": "Co-located servers, direct feeds",
    },
    "Statistical Arbitrage": {
        "target_total_us": 1000,  # 1 millisecond
        "network_us": 100,
        "processing_us": 700,
        "serialization_us": 50,
        "buffer_us": 150,
        "typical_language": "C++/Java",
        "infrastructure": "Low-latency data center",
    },
    "Intraday Momentum": {
        "target_total_us": 100000,  # 100 milliseconds
        "network_us": 5000,
        "processing_us": 80000,
        "serialization_us": 1000,
        "buffer_us": 14000,
        "typical_language": "Python/Java",
        "infrastructure": "Cloud or standard data center",
    },
    "Daily Rebalancing": {
        "target_total_us": 60000000,  # 60 seconds
        "network_us": 50000,
        "processing_us": 59000000,
        "serialization_us": 10000,
        "buffer_us": 940000,
        "typical_language": "Python",
        "infrastructure": "Any",
    },
}

budget_summary = []
for strategy, budget in latency_budgets.items():
    total = budget["target_total_us"]
    if total >= 1000000:
        total_str = f"{total / 1000000:.0f} seconds"
    elif total >= 1000:
        total_str = f"{total / 1000:.0f} ms"
    else:
        total_str = f"{total} μs"

    budget_summary.append(
        {
            "strategy": strategy,
            "total_str": total_str,
            "language": budget["typical_language"],
            "infrastructure": budget["infrastructure"],
        }
    )