Quant Trading Systems: Architecture & 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

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"],
        }
    )

# Speed of light calculation for different media
speed_of_light_vacuum = 299792458  # meters per second
fiber_refractive_index = 1.47  # Light slows down in fiber
microwave_speed = speed_of_light_vacuum * 0.99  # Nearly vacuum speed

# Chicago to New Jersey distances
fiber_distance_km = 1200  # Typical fiber route (not straight line)
straight_line_km = 1140  # Approximate straight line

# Calculate one-way latency
fiber_latency_ms = (
    (fiber_distance_km * 1000)
    / (speed_of_light_vacuum / fiber_refractive_index)
    * 1000
)
microwave_latency_ms = (straight_line_km * 1000) / microwave_speed * 1000
advantage_ms = fiber_latency_ms - microwave_latency_ms
advantage_us = advantage_ms * 1000

from enum import Enum


class OrderStatus(Enum):
    PENDING = "PENDING"  # Created but not sent
    SUBMITTED = "SUBMITTED"  # Sent to venue
    PARTIAL = "PARTIAL"  # Partially filled
    FILLED = "FILLED"  # Completely filled
    CANCELLED = "CANCELLED"  # Cancelled by user or system
    REJECTED = "REJECTED"  # Rejected by venue


@dataclass
class OrderRecord:
    """Complete record of an order and its fills."""

    order_id: str
    symbol: str
    side: str
    quantity: float
    order_type: str
    limit_price: Optional[float]
    status: OrderStatus
    created_time: datetime
    submitted_time: Optional[datetime] = None
    filled_quantity: float = 0
    average_fill_price: float = 0
    fills: list = None

    def __post_init__(self):
        if self.fills is None:
            self.fills = []

    @property
    def remaining_quantity(self) -> float:
        return self.quantity - self.filled_quantity

    @property
    def is_complete(self) -> bool:
        return self.status in [
            OrderStatus.FILLED,
            OrderStatus.CANCELLED,
            OrderStatus.REJECTED,
        ]


class OrderManagementSystem:
    """Tracks orders through their complete lifecycle."""

    def __init__(self):
        self.orders: dict[str, OrderRecord] = {}
        self.order_counter = 0

    def create_order(
        self,
        symbol: str,
        side: str,
        quantity: float,
        order_type: str,
        limit_price: Optional[float] = None,
    ) -> str:
        """Create a new order and return its ID."""
        self.order_counter += 1
        order_id = f"ORD-{self.order_counter:06d}"

        order = OrderRecord(
            order_id=order_id,
            symbol=symbol,
            side=side,
            quantity=quantity,
            order_type=order_type,
            limit_price=limit_price,
            status=OrderStatus.PENDING,
            created_time=datetime.now(),
        )
        self.orders[order_id] = order
        return order_id

    def submit_order(self, order_id: str) -> bool:
        """Mark order as submitted to venue."""
        if order_id not in self.orders:
            return False
        order = self.orders[order_id]
        if order.status != OrderStatus.PENDING:
            return False
        order.status = OrderStatus.SUBMITTED
        order.submitted_time = datetime.now()
        return True

    def on_fill(self, order_id: str, fill_qty: float, fill_price: float):
        """Process a fill notification from the venue."""
        if order_id not in self.orders:
            return

        order = self.orders[order_id]

        # Update average price
        total_value = (
            order.average_fill_price * order.filled_quantity
            + fill_price * fill_qty
        )
        order.filled_quantity += fill_qty
        order.average_fill_price = total_value / order.filled_quantity

        # Record fill
        order.fills.append(
            {"quantity": fill_qty, "price": fill_price, "time": datetime.now()}
        )

        # Update status
        if order.filled_quantity >= order.quantity:
            order.status = OrderStatus.FILLED
        else:
            order.status = OrderStatus.PARTIAL

    def get_open_orders(self) -> list[OrderRecord]:
        """Get all non-complete orders."""
        return [o for o in self.orders.values() if not o.is_complete]

    def get_order_summary(self) -> dict:
        """Get summary statistics of all orders."""
        total = len(self.orders)
        by_status = {}
        for order in self.orders.values():
            status = order.status.value
            by_status[status] = by_status.get(status, 0) + 1
        return {"total": total, "by_status": by_status}

# Demonstrate order management
oms = OrderManagementSystem()

# Create and process some orders
order1_id = oms.create_order("AAPL", "BUY", 100, "LIMIT", 185.50)
order2_id = oms.create_order("GOOGL", "BUY", 50, "MARKET")
order3_id = oms.create_order("MSFT", "SELL", 200, "LIMIT", 380.00)

# Submit orders
for oid in [order1_id, order2_id, order3_id]:
    oms.submit_order(oid)

# Simulate fills
oms.on_fill(order1_id, 60, 185.48)  # Partial fill
oms.on_fill(order1_id, 40, 185.50)  # Complete fill
oms.on_fill(order2_id, 50, 141.25)  # Complete fill

summary = oms.get_order_summary()

@dataclass
class Position:
    """Represents a position in a single security."""

    symbol: str
    quantity: float
    average_cost: float
    current_price: float = 0

    @property
    def market_value(self) -> float:
        return self.quantity * self.current_price

    @property
    def cost_basis(self) -> float:
        return self.quantity * self.average_cost

    @property
    def unrealized_pnl(self) -> float:
        return self.market_value - self.cost_basis

    @property
    def unrealized_pnl_pct(self) -> float:
        if self.cost_basis == 0:
            return 0
        return self.unrealized_pnl / abs(self.cost_basis)


class PortfolioManager:
    """Manages portfolio positions and P&L tracking."""

    def __init__(self, initial_cash: float):
        self.cash = initial_cash
        self.positions: dict[str, Position] = {}
        self.realized_pnl = 0

    def update_position(
        self,
        symbol: str,
        quantity_change: float,
        price: float,
        is_fill: bool = True,
    ):
        """Update position from a trade fill or market data update."""
        if symbol not in self.positions:
            self.positions[symbol] = Position(symbol, 0, 0, price)

        pos = self.positions[symbol]

        if is_fill:
            # This is a trade - update cash and average cost
            trade_value = quantity_change * price
            self.cash -= trade_value

            if quantity_change > 0:  # Buying
                # Update average cost
                total_cost = pos.cost_basis + trade_value
                new_quantity = pos.quantity + quantity_change
                pos.average_cost = (
                    total_cost / new_quantity if new_quantity != 0 else 0
                )
                pos.quantity = new_quantity
            else:  # Selling
                # Realize P&L
                sold_qty = abs(quantity_change)
                realized = sold_qty * (price - pos.average_cost)
                self.realized_pnl += realized
                pos.quantity += quantity_change

        # Update current price
        pos.current_price = price

        # Clean up zero positions
        if abs(pos.quantity) < 0.001:
            del self.positions[symbol]

    def update_prices(self, prices: dict[str, float]):
        """Update current prices for all positions."""
        for symbol, price in prices.items():
            if symbol in self.positions:
                self.positions[symbol].current_price = price

    def get_portfolio_summary(self) -> dict:
        """Get complete portfolio summary."""
        total_market_value = sum(
            p.market_value for p in self.positions.values()
        )
        total_unrealized = sum(
            p.unrealized_pnl for p in self.positions.values()
        )
        nav = self.cash + total_market_value

        return {
            "cash": self.cash,
            "market_value": total_market_value,
            "nav": nav,
            "unrealized_pnl": total_unrealized,
            "realized_pnl": self.realized_pnl,
            "total_pnl": total_unrealized + self.realized_pnl,
            "num_positions": len(self.positions),
        }