High-Frequency Trading Infrastructure: Latency Optimization

// BAD: Traditional socket-based trading (milliseconds)
int traditional_send(int sock, char* data, size_t len) {
    struct msghdr msg = {0};
    struct iovec iov = {data, len};
    msg.msg_iov = &iov;
    msg.msg_iovlen = 1;
    
    // Context switch, kernel copy, interrupt handling
    return sendmsg(sock, &msg, 0);  // 10-100μs latency
}

// GOOD: DPDK-based kernel bypass (sub-microsecond)
struct rte_mbuf* create_packet(struct rte_mempool* pool) {
    struct rte_mbuf* pkt = rte_pktmbuf_alloc(pool);
    char* payload = rte_pktmbuf_mtod(pkt, char*);
    
    // Direct memory access, no kernel involvement
    memcpy(payload, trading_data, payload_size);
    pkt->data_len = payload_size;
    
    return pkt;  // < 1μs latency
}

#!/usr/bin/env python3
"""DPDK packet processing for trading."""

import ctypes
from dataclasses import dataclass
from typing import List

class DPDKTrading:
    """Kernel bypass networking for HFT."""
    
    def __init__(self, port_id: int = 0, rx_queues: int = 1):
        self.port_id = port_id
        self.rx_queues = rx_queues
        self.running = False
    
    def initialize(self):
        """Initialize DPDK for trading."""
        args = "-n 4 --socket-mem 512,512 --"
        print(f"Initializing DPDK with args: {args}")
        
        # EAL initialization
        # ret = rte_eal_init(len(args), args)
        
        # Port configuration
        # port_config = rte_eth_conf()
        # port_config.rxmode.split_hash = True
        
        print("DPDK initialized for trading")
    
    def allocate_buffers(self, nb_mbuf: int = 8192):
        """Allocate packet buffers."""
        # mbuf_pool = rte_pktmbuf_pool_create(
        #     "TRADING_POOL",
        #     nb_mbuf,
        #     256,  # cache size
        #     0,    # priv size
        #     1518, # mbuf size
        #     rte_socket_id()
        # )
        print(f"Allocated {nb_mbuf} mbuf entries")
    
    def send_order(self, symbol: str, side: str, 
                   quantity: int, price: float):
        """Send order with minimal latency."""
        # Direct memory write to NIC
        order_packet = self.build_order_packet(
            symbol, side, quantity, price
        )
        
        # Bypass kernel, direct NIC write
        # rte_eth_tx_buffer(self.port_id, 0, self.tx_buffer, order_packet)
        
        # Target: < 500ns from function call to NIC
        print(f"Order sent: {side} {quantity} {symbol} @ {price}")
    
    def build_order_packet(self, symbol: str, side: str,
                           quantity: int, price: float) -> bytes:
        """Build FIX/ITCH order packet."""
        # Proprietary binary format for speed
        packet = bytearray(64)
        
        # Message type: New Order Single
        packet[0] = 0x35  # 'D' in FIX
        
        # Symbol (8 bytes, left-padded)
        symbol_bytes = symbol.encode('ascii')[:8]
        packet[8:16] = symbol_bytes
        
        # Side: 1=Buy, 2=Sell
        packet[16] = 1 if side.upper() == 'BUY' else 2
        
        # Quantity (8 bytes, big-endian)
        packet[17:25] = quantity.to_bytes(8, 'big')
        
        # Price (8 bytes, big-endian, 4 decimal places)
        price_int = int(price * 10000)
        packet[25:33] = price_int.to_bytes(8, 'big')
        
        return bytes(packet)

// FPGA-based order entry (UltraScale+)
// Target: < 100ns latency

module order_entry #(
    parameter SYMBOL_BITS = 8,
    parameter PRICE_BITS = 32,
    parameter QTY_BITS = 32
)(
    input wire clk,
    input wire rst,
    
    // Market data input
    input wire [SYMBOL_BITS-1:0] symbol_in,
    input wire [PRICE_BITS-1:0] price_in,
    input wire [QTY_BITS-1:0] qty_in,
    input wire valid_in,
    
    // Order output to exchange
    output reg [7:0] order_type,
    output reg [SYMBOL_BITS-1:0] symbol_out,
    output reg [PRICE_BITS-1:0] price_out,
    output reg [QTY_BITS-1:0] qty_out,
    output reg valid_out,
    
    // Control
    output reg ready
);

    // Order book state (BRAM)
    reg [PRICE_BITS-1:0] bid_price [0:255];
    reg [PRICE_BITS-1:0] ask_price [0:255];
    reg [QTY_BITS-1:0] bid_qty [0:255];
    reg [QTY_BITS-1:0] ask_qty [0:255];
    
    always @(posedge clk) begin
        if (rst) begin
            ready <= 1'b1;
            valid_out <= 1'b0;
        end else if (valid_in && ready) begin
            // Check spread immediately
            if (price_in <= ask_price[symbol_in]) begin
                // Buy order - check liquidity
                order_type <= 8'h01; // Market
                symbol_out <= symbol_in;
                price_out <= ask_price[symbol_in];
                qty_out <= qty_in;
                valid_out <= 1'b1;
            end else {
                // Post to book
                order_type <= 8'h02; // Limit
                symbol_out <= symbol_in;
                price_out <= price_in;
                qty_out <= qty_in;
                valid_out <= 1'b1;
            end
        end
    end
endmodule

#!/usr/bin/env python3
"""FPGA price feed processing with nanosecond timestamps."""

import struct
from dataclasses import dataclass
from typing import Optional

@dataclass
class PriceTick:
    """Market data tick with hardware timestamp."""
    symbol: str
    bid: float
    ask: float
    bid_size: int
    ask_size: int
    timestamp_ns: int  # Hardware timestamp in nanoseconds

class FPGAFeedHandler:
    """Process FPGA-accelerated market data."""
    
    def __init__(self):
        self.symbol_map = {}
        self.last_bid = {}
        self.last_ask = {}
    
    def parse_itc_packet(self, packet: bytes) -> Optional[PriceTick]:
        """Parse ITCH protocol packet from FPGA."""
        
        # Packet structure from FPGA
        # [4:8] timestamp_ns
        # [8:16] symbol (8 bytes, ASCII)
        # [16:24] bid_price (8 bytes, 4 decimal)
        # [24:32] ask_price (8 bytes, 4 decimal)
        # [32:40] bid_size (8 bytes)
        # [40:48] ask_size (8 bytes)
        
        if len(packet) < 48:
            return None
        
        timestamp_ns = struct.unpack('>Q', packet[4:12])[0]
        symbol = packet[12:20].decode('ascii').strip()
        bid_price = struct.unpack('>Q', packet[20:28])[0] / 10000
        ask_price = struct.unpack('>Q', packet[28:36])[0] / 10000
        bid_size = struct.unpack('>Q', packet[36:44])[0]
        ask_size = struct.unpack('>Q', packet[44:52])[0]
        
        return PriceTick(
            symbol=symbol,
            bid=bid_price,
            ask=ask_price,
            bid_size=bid_size,
            ask_size=ask_size,
            timestamp_ns=timestamp_ns
        )
    
    def calculate_spread(self, tick: PriceTick) -> float:
        """Calculate spread in basis points."""
        mid = (tick.bid + tick.ask) / 2
        spread = tick.ask - tick.bid
        bps = (spread / mid) * 10000
        return bps
    
    def detect_arbitrage(self, tick: PriceTick, 
                         exchange_prices: dict) -> Optional[dict]:
        """Detect cross-exchange arbitrage."""
        
        for exchange, ex_prices in exchange_prices.items():
            if tick.symbol not in ex_prices:
                continue
            
            ex_bid, ex_ask = ex_prices[tick.symbol]
            
            # Buy on one exchange, sell on another
            if tick.ask < ex_bid:
                profit_bps = ((ex_bid - tick.ask) / tick.ask) * 10000
                if profit_bps > 2:  # > 2 bps
                    return {
                        'buy_exchange': 'this',
                        'sell_exchange': exchange,
                        'buy_price': tick.ask,
                        'sell_price': ex_bid,
                        'profit_bps': profit_bps,
                        'latency_ns': tick.timestamp_ns
                    }
        
        return None

#!/usr/bin/env python3
"""Exchange colocation infrastructure planning."""

class ColocationPlanner:
    """Plan colocation strategy for HFT."""
    
    EXCHANGE_LOCATIONS = {
        'NYSE': {'location': 'Mahwah, NJ', 'latency_us': 0.5},
        'NASDAQ': {'location': 'Carteret, NJ', 'latency_us': 0.5},
        'CME': {'location': 'Aurora, IL', 'latency_us': 1.0},
        'ICE': {'location': 'Atlanta, GA', 'latency_us': 2.0},
        'LSE': {'location': 'London, UK', 'latency_us': 3.0},
        'EUREX': {'location': 'Frankfurt, DE', 'latency_us': 3.0},
    }
    
    def __init__(self, strategy_type: str = 'multi_asset'):
        self.strategy_type = strategy_type
        self.rack_costs = self._load_rack_costs()
    
    def _load_rack_costs(self) -> dict:
        """Monthly colocation costs per rack."""
        return {
            'Mahwah': 15000,
            'Carteret': 15000,
            'Aurora': 12000,
            'Atlanta': 10000,
            'Frankfurt': 18000,
            'London': 18000,
        }
    
    def calculate_latency_budget(self) -> dict:
        """Calculate latency budget breakdown."""
        
        # Total target: 10μs end-to-end
        return {
            'strategy_compute': 2.0,      # μs
            'risk_check': 1.0,             # μs
            'order_routing': 0.5,          # μs
            'network_to_exchange': 2.0,   # μs (varies by location)
            'exchange_processing': 1.5,   # μs
            'confirmation_return': 2.0,   # μs
            'safety_margin': 1.0,         # μs
        }
    
    def select_locations(self, target_markets: list) -> dict:
        """Select optimal colocation sites."""
        
        # Group exchanges by location
        locations = {}
        
        for exchange in target_markets:
            if exchange not in self.EXCHANGE_LOCATIONS:
                continue
            
            info = self.EXCHANGE_LOCATIONS[exchange]
            loc = info['location']
            
            if loc not in locations:
                locations[loc] = {
                    'cost_monthly': self.rack_costs.get(loc, 15000),
                    'exchanges': [],
                    'avg_latency_us': info['latency_us']
                }
            
            locations[loc]['exchanges'].append(exchange)
        
        return locations
    
    def estimate_costs(self, locations: dict, 
                       racks_per_location: int = 2) -> dict:
        """Estimate total colocation costs."""
        
        total_monthly = 0
        total_annual = 0
        
        details = []
        
        for loc, info in locations.items():
            monthly = info['cost_monthly'] * racks_per_location
            annual = monthly * 12
            
            total_monthly += monthly
            total_annual += annual
            
            details.append({
                'location': loc,
                'racks': racks_per_location,
                'monthly': monthly,
                'annual': annual,
                'exchanges': info['exchanges']
            })
        
        return {
            'details': details,
            'total_monthly': total_monthly,
            'total_annual': total_annual,
            'per_rack_monthly': total_monthly / (len(locations) * racks_per_location)
        }

// BAD: Text-based parsing (slow)
json_error_t parse_json_quote(const char* json) {
    json_t *root = json_loads(json, 0, &error);
    json_t *symbol = json_object_get(root, "symbol");
    json_t *bid = json_object_get(root, "bid");
    // JSON parsing: 10-50μs
    return error;
}

// GOOD: Binary protocol parsing (fast)
#pragma pack(push, 1)
struct ItchMessage {
    uint16_t msg_length;
    uint64_t timestamp_ns;
    uint8_t  msg_type;
    uint32_t symbol;
    uint64_t bid_price;
    uint64_t ask_price;
    uint32_t bid_size;
    uint32_t ask_size;
};
#pragma pack(pop)

// Binary parsing: 50-100ns
struct ItchMessage* parse_binary_quote(const char* data) {
    return (struct ItchMessage*)data;  // Zero-copy
}

#!/usr/bin/env python3
"""High-performance multicast market data processing."""

import socket
import struct
from typing import Dict, List
from collections import defaultdict

class MulticastFeedHandler:
    """Process multicast market data feeds."""
    
    def __init__(self, multicast_group: str, port: int):
        self.group = multicast_group
        self.port = port
        self.sock = None
        self.book_state: Dict[str, dict] = defaultdict(dict)
    
    def setup_socket(self):
        """Configure socket for low-latency multicast."""
        
        self.sock = socket.socket(
            socket.AF_INET, 
            socket.SOCK_DGRAM,
            socket.IPPROTO_UDP
        )
        
        # Reuse address for multiple processes
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        
        # Bind to multicast group
        self.sock.bind((self.group, self.port))
        
        # Join multicast group
        mreq = struct.pack("4sl", 
                          socket.inet_aton(self.group),
                          socket.INADDR_ANY)
        self.sock.setsockopt(
            socket.IPPROTO_IP, 
            socket.IP_ADD_MEMBERSHIP, 
            mreq
        )
        
        # Low-latency socket options
        self.sock.setsockopt(
            socket.SOL_SOCKET,
            socket.SO_RCVBUF,
            1024 * 1024  # 1MB buffer
        )
        
        # Disable Nagle (send immediately)
        self.sock.setsockopt(
            socket.IPPROTO_TCP,
            socket.TCP_NODELAY,
            1
        )
        
        print(f"Multicast socket configured: {self.group}:{self.port}")
    
    def process_message(self, data: bytes) -> dict:
        """Process ITCH/OUCH message."""
        
        if len(data) < 4:
            return None
        
        msg_type = chr(data[0])
        
        if msg_type == 'P':  # Add Order - MPID
            # P: Add Order (Long Form)
            # [1] msg_type
            # [2:4] time_offset
            # [4:12] order_ref (8)
            # [12:20] symbol (8)
            # [20] side (1)
            # [21:29] shares (8)
            # [29:37] price (8)
            # [37] participation (1)
            
            order_ref = struct.unpack('>Q', data[4:12])[0]
            symbol = data[12:20].decode().strip()
            side = chr(data[20])
            shares = struct.unpack('>Q', data[21:29])[0]
            price = struct.unpack('>Q', data[29:37])[0] / 10000
            
            self.book_state[symbol][order_ref] = {
                'side': side,
                'shares': shares,
                'price': price
            }
            
            return {'type': 'add', 'symbol': symbol, 
                    'ref': order_ref, 'side': side}
        
        elif msg_type == 'F':  # Add Order - MPID
            # Similar to P but shorter form
            pass
        
        elif msg_type == 'E':  # Execute Order
            order_ref = struct.unpack('>Q', data[4:12])[0]
            executed = struct.unpack('>Q', data[12:20])[0]
            
            if symbol := self._find_symbol(order_ref):
                if order_ref in self.book_state[symbol]:
                    self.book_state[symbol][order_ref]['shares'] -= executed
            
            return {'type': 'execute', 'ref': order_ref}
        
        return None
    
    def _find_symbol(self, order_ref: int) -> str:
        """Find symbol by order reference."""
        for symbol, orders in self.book_state.items():
            if order_ref in orders:
                return symbol
        return None

#!/usr/bin/env python3
"""Pre-trade risk management for HFT."""

from dataclasses import dataclass
from typing import Optional
from enum import Enum
import time

class OrderSide(Enum):
    BUY = 1
    SELL = 2

@dataclass
class Order:
    """Order details."""
    symbol: str
    side: OrderSide
    quantity: int
    price: float
    client_id: int

@dataclass
class RiskLimits:
    """Risk limits configuration."""
    max_order_size: int = 10000
    max_notional_single: float = 1000000.0
    max_daily_volume: float = 50000000.0
    max_position_per_symbol: float = 5000000.0
    max_loss_per_minute: float = 100000.0

class HFTRiskManager:
    """High-frequency trading risk manager."""
    
    def __init__(self, limits: RiskLimits):
        self.limits = limits
        self.positions: dict = {}
        self.daily_volume = 0.0
        self.minute_losses = []
        self.last_reset = time.time()
    
    def check_order(self, order: Order, 
                    current_prices: dict) -> tuple[bool, str]:
        """Pre-trade risk check - must complete in < 1μs."""
        
        # 1. Check order size
        if order.quantity > self.limits.max_order_size:
            return False, "ORDER_SIZE_EXCEEDED"
        
        # 2. Check notional value
        notional = order.quantity * order.price
        if notional > self.limits.max_notional_single:
            return False, "NOTIONAL_EXCEEDED"
        
        # 3. Check daily volume
        if self.daily_volume + notional > self.limits.max_daily_volume:
            return False, "DAILY_VOLUME_EXCEEDED"
        
        # 4. Check position limit
        current_position = self.positions.get(order.symbol, 0.0)
        
        if order.side == OrderSide.BUY:
            new_position = current_position + notional
        else:
            new_position = current_position - notional
        
        if abs(new_position) > self.limits.max_position_per_symbol:
            return False, "POSITION_LIMIT_EXCEEDED"
        
        # 5. Check loss limit
        current_loss = self._calculate_minute_loss()
        if current_loss > self.limits.max_loss_per_minute:
            return False, "LOSS_LIMIT_EXCEEDED"
        
        # All checks passed
        return True, "OK"
    
    def _calculate_minute_loss(self) -> float:
        """Calculate loss in current minute window."""
        
        now = time.time()
        
        # Reset if minute passed
        if now - self.last_reset > 60:
            self.minute_losses = []
            self.last_reset = now
        
        return sum(self.minute_losses)
    
    def update_position(self, order: Order, executed: bool = True):
        """Update positions after execution."""
        
        if not executed:
            return
        
        notional = order.quantity * order.price
        
        if order.symbol not in self.positions:
            self.positions[order.symbol] = 0.0
        
        if order.side == OrderSide.BUY:
            self.positions[order.symbol] += notional
        else:
            self.positions[order.symbol] -= notional
        
        self.daily_volume += notional
    
    def get_exposure_report(self) -> dict:
        """Generate current exposure report."""
        
        total_long = sum(p for p in self.positions.values() if p > 0)
        total_short = abs(sum(p for p in self.positions.values() if p < 0))
        
        return {
            'total_long': total_long,
            'total_short': total_short,
            'net_exposure': total_long - total_short,
            'daily_volume': self.daily_volume,
            'positions': dict(self.positions)
        }

# Network architecture for HFT
network_topology:
  core:
    - name: "Trading Switch"
      model: "Arista 7280R3"
      latency: "500ns"
      ports: 100
      
  connectivity:
    exchange_colocation:
      - exchange: "NYSE"
        location: "Mahwah"
        latency_target: "0.5μs"
        fiber_type: "single-mode"
        distance_km: 0
        
      - exchange: "NASDAQ"
        location: "Carteret"
        latency_target: "0.5μs"
        fiber_type: "single-mode"
        distance_km: 0
        
      - exchange: "CME"
        location: "Aurora"
        latency_target: "1.0μs"
        fiber_type: "single-mode"
        distance_km: 50
        
  nic_configuration:
    model: "Mellanox ConnectX-7"
    features:
      - "RDMA (RoCE v2)"
      - "Hardware timestamp"
      - "Kernel bypass"
      - "Zero-copy"
    driver: "MLX5"
    settings:
      rx_desc: 4096
      tx_desc: 4096
      mtu: 9000
      tx_queue_size: 4096

#!/usr/bin/env python3
"""HFT latency benchmarking."""

import time
import statistics
from typing import List

class LatencyBenchmark:
    """Benchmark trading system latency."""
    
    def __init__(self, iterations: int = 100000):
        self.iterations = iterations
        self.results: List[float] = []
    
    def benchmark_order_send(self) -> dict:
        """Benchmark order send latency."""
        
        latencies = []
        
        for _ in range(self.iterations):
            start = time.perf_counter_ns()
            
            # Simulate order send
            # In reality: DMA to NIC, no syscalls
            
            end = time.perf_counter_ns()
            latencies.append(end - start)
        
        return {
            'mean_ns': statistics.mean(latencies),
            'p50_ns': statistics.median(latencies),
            'p99_ns': sorted(latencies)[int(len(latencies) * 0.99)],
            'p999_ns': sorted(latencies)[int(len(latencies) * 0.999)],
            'max_ns': max(latencies),
            'min_ns': min(latencies)
        }
    
    def benchmark_feed_parsing(self) -> dict:
        """Benchmark market data parsing."""
        
        sample_packet = b'P' + b'\x00' * 63  # 64-byte packet
        
        latencies = []
        
        for _ in range(self.iterations):
            start = time.perf_counter_ns()
            
            # Binary parse
            symbol = sample_packet[12:20]
            price = int.from_bytes(sample_packet[29:37], 'big')
            
            end = time.perf_counter_ns()
            latencies.append(end - start)
        
        return {
            'mean_ns': statistics.mean(latencies),
            'p99_ns': sorted(latencies)[int(len(latencies) * 0.99)]
        }

# Example benchmark results
BENCHMARK_RESULTS = {
    'order_send_software': {
        'mean_us': 15.2,
        'p99_us': 45.3,
        'method': 'Standard sockets'
    },
    'order_send_kernel_bypass': {
        'mean_us': 0.8,
        'p99_us': 2.1,
        'method': 'DPDK'
    },
    'order_send_fpga': {
        'mean_us': 0.05,
        'p99_us': 0.12,
        'method': 'FPGA'
    },
    'feed_parse_software': {
        'mean_us': 3.2,
        'p99_us': 8.5,
        'method': 'JSON parsing'
    },
    'feed_parse_binary': {
        'mean_us': 0.15,
        'p99_us': 0.45,
        'method': 'Binary parsing'
    },
    'feed_parse_fpga': {
        'mean_us': 0.01,
        'p99_us': 0.03,
        'method': 'FPGA offload'
    }
}