High-Frequency Trading Analytics Pipeline

-- High-Frequency Trading Data Schema
CREATE TABLE market_ticks (
    tick_id BIGSERIAL PRIMARY KEY,
    symbol VARCHAR(20) NOT NULL,
    timestamp TIMESTAMPTZ NOT NULL,
    bid DECIMAL(19,6),
    ask DECIMAL(19,6),
    last_price DECIMAL(19,6),
    volume BIGINT,
    trade_count INTEGER,
    created_at TIMESTAMPTZ DEFAULT NOW()
) PARTITION BY RANGE (timestamp);

-- Create partitions for each day
CREATE TABLE market_ticks_2024_01_01 PARTITION OF market_ticks
    FOR VALUES FROM ('2024-01-01') TO ('2024-01-02');

-- Order book snapshots
CREATE TABLE order_book_snapshots (
    snapshot_id BIGSERIAL PRIMARY KEY,
    symbol VARCHAR(20) NOT NULL,
    timestamp TIMESTAMPTZ NOT NULL,
    bid_levels JSONB, -- Array of {price, quantity}
    ask_levels JSONB, -- Array of {price, quantity}
    spread DECIMAL(19,6),
    mid_price DECIMAL(19,6)
) PARTITION BY RANGE (timestamp);

-- Position tracking
CREATE TABLE positions (
    position_id BIGSERIAL PRIMARY KEY,
    account_id BIGINT NOT NULL,
    symbol VARCHAR(20) NOT NULL,
    quantity DECIMAL(19,6) NOT NULL,
    avg_price DECIMAL(19,6) NOT NULL,
    current_price DECIMAL(19,6),
    unrealized_pnl DECIMAL(19,6),
    last_updated TIMESTAMPTZ DEFAULT NOW()
);

-- Create indexes for ultra-fast queries
CREATE INDEX CONCURRENTLY idx_ticks_symbol_time ON market_ticks(symbol, timestamp DESC);
CREATE INDEX CONCURRENTLY idx_ticks_time ON market_ticks(timestamp DESC);
CREATE INDEX CONCURRENTLY idx_positions_account_symbol ON positions(account_id, symbol);

// Ultra-low-latency market data processor
@Slf4j
public class UltraLowLatencyProcessor {
    
    private final RingBuffer<MarketTick> tickBuffer;
    private final AtomicLong sequence = new AtomicLong(0);
    private final ExecutorService executor;
    
    public UltraLowLatencyProcessor(int bufferSize) {
        this.tickBuffer = RingBuffer.createSingleProducer(
            MarketTick::new, bufferSize, 
            new YieldingWaitStrategy()
        );
        this.executor = Executors.newSingleThreadExecutor(r -> {
            Thread t = new Thread(r, "MarketDataProcessor");
            t.setPriority(Thread.MAX_PRIORITY);
            return t;
        });
    }
    
    public void processTick(MarketTick tick) {
        long sequence = tickBuffer.next();
        try {
            MarketTick event = tickBuffer.get(sequence);
            event.copyFrom(tick);
        } finally {
            tickBuffer.publish(sequence);
        }
    }
    
    // Zero-copy tick processing
    public void onTick(MarketTick tick, long sequence, boolean endOfBatch) {
        // Process tick with minimal object creation
        updateOrderBook(tick);
        calculateIndicators(tick);
        checkRiskLimits(tick);
        
        if (endOfBatch) {
            publishSignals();
        }
    }
    
    private void updateOrderBook(MarketTick tick) {
        // Lock-free order book update
        orderBook.updateBidAsk(tick.getBid(), tick.getAsk());
        orderBook.updateDepth(tick.getBidLevels(), tick.getAskLevels());
    }
    
    private void calculateIndicators(MarketTick tick) {
        // Real-time technical indicators
        vwap.update(tick.getLastPrice(), tick.getVolume());
        rsi.update(tick.getLastPrice());
        bollingerBands.update(tick.getLastPrice());
    }
    
    private void checkRiskLimits(MarketTick tick) {
        // Ultra-fast risk checks
        if (riskEngine.checkPositionLimits(tick.getSymbol())) {
            riskEngine.triggerCircuitBreaker(tick.getSymbol());
        }
    }
}

# Real-time feature engineering for trading
import numpy as np
from typing import Dict, List
import asyncio
from dataclasses import dataclass

@dataclass
class MarketFeatures:
    symbol: str
    timestamp: float
    vwap: float
    rsi: float
    bollinger_upper: float
    bollinger_lower: float
    order_book_imbalance: float
    volume_profile: Dict[str, float]

class RealTimeFeatureEngine:
    def __init__(self, lookback_window: int = 100):
        self.lookback_window = lookback_window
        self.price_history = {}
        self.volume_history = {}
        self.order_book_history = {}
        
    async def compute_features(self, market_data: Dict) -> MarketFeatures:
        """Compute real-time features with minimal latency"""
        symbol = market_data['symbol']
        
        # Update historical data
        self._update_history(symbol, market_data)
        
        # Compute features in parallel
        features = await asyncio.gather(
            self._compute_vwap(symbol),
            self._compute_rsi(symbol),
            self._compute_bollinger_bands(symbol),
            self._compute_order_book_imbalance(symbol),
            self._compute_volume_profile(symbol)
        )
        
        return MarketFeatures(
            symbol=symbol,
            timestamp=market_data['timestamp'],
            vwap=features[0],
            rsi=features[1],
            bollinger_upper=features[2][0],
            bollinger_lower=features[2][1],
            order_book_imbalance=features[3],
            volume_profile=features[4]
        )
    
    def _compute_vwap(self, symbol: str) -> float:
        """Compute Volume Weighted Average Price"""
        prices = self.price_history.get(symbol, [])
        volumes = self.volume_history.get(symbol, [])
        
        if len(prices) < 2:
            return prices[-1] if prices else 0.0
            
        # Use vectorized operations for speed
        prices_array = np.array(prices[-self.lookback_window:])
        volumes_array = np.array(volumes[-self.lookback_window:])
        
        return np.sum(prices_array * volumes_array) / np.sum(volumes_array)
    
    def _compute_rsi(self, symbol: str, period: int = 14) -> float:
        """Compute Relative Strength Index"""
        prices = self.price_history.get(symbol, [])
        if len(prices) < period + 1:
            return 50.0  # Neutral RSI
            
        # Calculate price changes
        deltas = np.diff(prices[-period-1:])
        gains = np.where(deltas > 0, deltas, 0)
        losses = np.where(deltas < 0, -deltas, 0)
        
        # Exponential moving averages
        avg_gain = np.mean(gains)
        avg_loss = np.mean(losses)
        
        if avg_loss == 0:
            return 100.0
            
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        
        return rsi

-- Risk management and compliance tables
CREATE TABLE risk_limits (
    limit_id BIGSERIAL PRIMARY KEY,
    account_id BIGINT NOT NULL,
    limit_type VARCHAR(50) NOT NULL, -- POSITION, EXPOSURE, VAR, etc.
    symbol VARCHAR(20),
    limit_value DECIMAL(19,6) NOT NULL,
    current_value DECIMAL(19,6) DEFAULT 0,
    breach_threshold DECIMAL(19,6) NOT NULL,
    status VARCHAR(20) DEFAULT 'ACTIVE',
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

CREATE TABLE risk_breaches (
    breach_id BIGSERIAL PRIMARY KEY,
    limit_id BIGINT NOT NULL,
    breach_type VARCHAR(50) NOT NULL,
    breach_value DECIMAL(19,6) NOT NULL,
    limit_value DECIMAL(19,6) NOT NULL,
    breach_timestamp TIMESTAMPTZ DEFAULT NOW(),
    action_taken VARCHAR(100),
    resolved_at TIMESTAMPTZ,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Real-time risk monitoring view
CREATE VIEW v_risk_dashboard AS
SELECT 
    a.account_name,
    rl.limit_type,
    rl.symbol,
    rl.limit_value,
    rl.current_value,
    rl.breach_threshold,
    CASE 
        WHEN rl.current_value > rl.breach_threshold THEN 'BREACH'
        WHEN rl.current_value > rl.limit_value * 0.8 THEN 'WARNING'
        ELSE 'SAFE'
    END as risk_status,
    ROUND((rl.current_value / rl.limit_value) * 100, 2) as utilization_pct
FROM risk_limits rl
JOIN accounts a ON rl.account_id = a.account_id
WHERE rl.status = 'ACTIVE'
ORDER BY risk_status DESC, utilization_pct DESC;

-- VaR calculation function
CREATE OR REPLACE FUNCTION calculate_var(
    p_account_id BIGINT,
    p_confidence_level DECIMAL(3,2) DEFAULT 0.95
) RETURNS DECIMAL(19,6) AS $$
DECLARE
    var_value DECIMAL(19,6);
BEGIN
    -- Calculate portfolio VaR using historical simulation
    SELECT percentile_cont(p_confidence_level) WITHIN GROUP (ORDER BY daily_pnl)
    INTO var_value
    FROM (
        SELECT 
            DATE_TRUNC('day', timestamp) as trade_date,
            SUM(unrealized_pnl) as daily_pnl
        FROM positions
        WHERE account_id = p_account_id
        GROUP BY DATE_TRUNC('day', timestamp)
        ORDER BY trade_date DESC
        LIMIT 252  -- One trading year
    ) daily_returns;
    
    RETURN ABS(var_value);
END;
$$ LANGUAGE plpgsql;