//! Production-Ready DBN (Databento Binary) Format Parser //! //! High-performance, zero-copy parser for Databento's binary format with SIMD optimizations //! for ultra-low latency HFT market data processing. Targets <1μs processing per tick. //! //! ## Performance Features //! //! - **Zero-Copy Deserialization**: Direct memory mapping with minimal allocations //! - **SIMD Optimizations**: Vectorized processing for batch operations //! - **Lock-Free Processing**: Atomic operations for concurrent access //! - **Hardware Timestamps**: RDTSC-based timing for latency measurement //! - **Memory Prefetching**: Cache-optimized data access patterns //! //! ## DBN Format Support //! //! - **Trade Messages**: Fast trade tick processing with price/size/conditions //! - **Quote Messages**: L1 BBO with bid/ask spreads //! - **Order Book**: L2/L3 depth with incremental updates //! - **Statistics**: OHLCV bars and session statistics //! - **Status Messages**: Market status and trading halts use crate::error::{DataError, Result}; use trading_engine::{ lockfree::{LockFreeRingBuffer, HftMessage, message_types}, simd::{SafeSimdDispatcher, SimdMarketDataOps, AlignedPrices, AlignedVolumes}, timing::HardwareTimestamp, types::prelude::*, events::{TradingEvent, EventProcessor}, }; use serde::{Deserialize, Serialize}; use std::sync::{Arc, atomic::{AtomicU64, AtomicBool, Ordering}}; use std::mem::{size_of, MaybeUninit}; use std::slice; use tracing::{debug, warn, error, instrument}; /// DBN message header - optimized for zero-copy parsing #[repr(C, packed)] #[derive(Debug, Clone, Copy)] pub struct DbnMessageHeader { /// Message length in bytes pub length: u16, /// Record type identifier pub rtype: u8, /// Publisher ID pub publisher_id: u8, /// Instrument ID pub instrument_id: u32, /// Timestamp (nanoseconds since Unix epoch) pub ts_event: u64, } /// DBN trade message - zero-copy optimized #[repr(C, packed)] #[derive(Debug, Clone, Copy)] pub struct DbnTradeMessage { pub header: DbnMessageHeader, /// Trade price (scaled integer) pub price: i64, /// Trade size pub size: u32, /// Trade action (A=Add, C=Cancel, M=Modify, T=Trade, F=Fill) pub action: u8, /// Trade side (A=Ask, B=Bid, N=None) pub side: u8, /// Trade flags pub flags: u16, /// Depth level pub depth: u8, /// Sequence number pub sequence: u32, /// Reserved padding _padding: u8, } /// DBN quote message - L1 BBO data #[repr(C, packed)] #[derive(Debug, Clone, Copy)] pub struct DbnQuoteMessage { pub header: DbnMessageHeader, /// Bid price (scaled integer) pub bid_px: i64, /// Ask price (scaled integer) pub ask_px: i64, /// Bid size pub bid_sz: u32, /// Ask size pub ask_sz: u32, /// Bid count pub bid_ct: u8, /// Ask count pub ask_ct: u8, /// Flags pub flags: u16, /// Sequence number pub sequence: u32, /// Reserved padding _padding: [u8; 2], } /// DBN order book message - L2/L3 depth #[repr(C, packed)] #[derive(Debug, Clone, Copy)] pub struct DbnOrderBookMessage { pub header: DbnMessageHeader, /// Order ID pub order_id: u64, /// Price (scaled integer) pub price: i64, /// Size pub size: u32, /// Flags pub flags: u16, /// Channel ID pub channel_id: u8, /// Order count pub order_count: u8, /// Action (A=Add, C=Cancel, M=Modify, T=Trade, F=Fill) pub action: u8, /// Side (A=Ask, B=Bid) pub side: u8, /// Sequence number pub sequence: u32, /// Reserved padding _padding: [u8; 2], } /// DBN OHLCV bar message #[repr(C, packed)] #[derive(Debug, Clone, Copy)] pub struct DbnOhlcvMessage { pub header: DbnMessageHeader, /// Open price (scaled integer) pub open: i64, /// High price (scaled integer) pub high: i64, /// Low price (scaled integer) pub low: i64, /// Close price (scaled integer) pub close: i64, /// Volume pub volume: u64, } /// DBN message types #[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum DbnMessageType { Trade = 0x54, // 'T' Quote = 0x51, // 'Q' OrderBook = 0x4F, // 'O' Ohlcv = 0x42, // 'B' (Bar) Status = 0x53, // 'S' Error = 0x45, // 'E' } impl From for DbnMessageType { fn from(value: u8) -> Self { match value { 0x54 => Self::Trade, 0x51 => Self::Quote, 0x4F => Self::OrderBook, 0x42 => Self::Ohlcv, 0x53 => Self::Status, 0x45 => Self::Error, _ => Self::Error, // Default to error for unknown types } } } /// High-performance DBN parser with SIMD optimizations pub struct DbnParser { /// SIMD dispatcher for optimal performance simd_dispatcher: SafeSimdDispatcher, /// Market data operations simd_ops: Option, /// Performance metrics metrics: Arc, /// Symbol mapping for instrument IDs symbol_map: Arc>>, /// Price scaling factors per instrument price_scales: Arc>>, /// Event processor for integration event_processor: Option>, /// Lock-free message buffer for high-frequency processing message_buffer: Arc>, } impl DbnParser { /// Create new high-performance DBN parser pub fn new() -> Result { let simd_dispatcher = SafeSimdDispatcher::new(); let simd_ops = simd_dispatcher.create_market_data_ops().ok(); if simd_ops.is_none() { warn!("AVX2 not available - falling back to scalar processing"); } else { debug!("DBN parser initialized with AVX2 SIMD optimizations"); } let message_buffer = Arc::new( LockFreeRingBuffer::new(32768) .map_err(|e| DataError::InitializationError(format!("Failed to create message buffer: {}", e)))? ); Ok(Self { simd_dispatcher, simd_ops, metrics: Arc::new(DbnParserMetrics::new()), symbol_map: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())), price_scales: Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())), event_processor: None, message_buffer, }) } /// Set event processor for integration with core event system pub fn set_event_processor(&mut self, processor: Arc) { self.event_processor = Some(processor); } /// Update symbol mapping for instrument IDs pub fn update_symbol_map(&self, mapping: std::collections::HashMap) { let mut symbol_map = self.symbol_map.write().unwrap(); symbol_map.extend(mapping); debug!("Updated symbol map with {} instruments", symbol_map.len()); } /// Update price scaling factors pub fn update_price_scales(&self, scales: std::collections::HashMap) { let mut price_scales = self.price_scales.write().unwrap(); price_scales.extend(scales); debug!("Updated price scales for {} instruments", price_scales.len()); } /// Parse DBN binary data with zero-copy optimization #[instrument(skip(self, data), level = "trace")] pub fn parse_batch(&self, data: &[u8]) -> Result> { let start_time = HardwareTimestamp::now(); let mut messages = Vec::new(); let mut offset = 0; // Pre-allocate for common batch sizes messages.reserve(1000); while offset + size_of::() <= data.len() { // Zero-copy header parsing let header = unsafe { std::ptr::read_unaligned(data.as_ptr().add(offset) as *const DbnMessageHeader) }; // Validate message length let header_length = header.length; // Copy field to avoid unaligned reference if header_length == 0 || offset + header_length as usize > data.len() { warn!("Invalid message length: {} at offset {}", header_length, offset); break; } // Parse message based on type let message_data = &data[offset..offset + header_length as usize]; match self.parse_single_message(message_data, header)? { Some(msg) => messages.push(msg), None => { // Unknown message type - skip self.metrics.increment_unknown_messages(); } } offset += header.length as usize; self.metrics.increment_messages_parsed(); } // SIMD batch processing for trade and quote messages if messages.len() >= 4 && self.simd_ops.is_some() { self.simd_batch_process(&mut messages)?; } let parse_time = HardwareTimestamp::now(); let latency_ns = parse_time.latency_ns(&start_time); self.metrics.record_parse_latency(latency_ns); // Check if we met the <1μs per tick target if messages.len() > 0 { let per_tick_latency = latency_ns / messages.len() as u64; if per_tick_latency > 1000 { warn!("Parse latency {}ns/tick exceeds 1μs target", per_tick_latency); } self.metrics.record_per_tick_latency(per_tick_latency); } Ok(messages) } /// Parse single DBN message with zero-copy fn parse_single_message(&self, data: &[u8], header: DbnMessageHeader) -> Result> { let message_type = DbnMessageType::from(header.rtype); let timestamp = HardwareTimestamp::from_ns(header.ts_event); match message_type { DbnMessageType::Trade => { if data.len() < size_of::() { return Err(DataError::InvalidFormat("Trade message too short".to_string())); } let trade_msg = unsafe { std::ptr::read_unaligned(data.as_ptr() as *const DbnTradeMessage) }; let symbol = self.get_symbol(header.instrument_id); let price = self.scale_price(trade_msg.price, header.instrument_id); let size = Decimal::from(trade_msg.size); let processed = ProcessedMessage::Trade { symbol, timestamp, price, size, side: match trade_msg.side { b'A' => OrderSide::Sell, // Ask side b'B' => OrderSide::Buy, // Bid side _ => OrderSide::Buy, // Default }, trade_id: Some(trade_msg.sequence.to_string()), conditions: vec![], // Parse from flags if needed }; self.metrics.increment_trades_processed(); Ok(Some(processed)) } DbnMessageType::Quote => { if data.len() < size_of::() { return Err(DataError::InvalidFormat("Quote message too short".to_string())); } let quote_msg = unsafe { std::ptr::read_unaligned(data.as_ptr() as *const DbnQuoteMessage) }; let symbol = self.get_symbol(header.instrument_id); let bid_price = self.scale_price(quote_msg.bid_px, header.instrument_id); let ask_price = self.scale_price(quote_msg.ask_px, header.instrument_id); let bid_size = Decimal::from(quote_msg.bid_sz); let ask_size = Decimal::from(quote_msg.ask_sz); let processed = ProcessedMessage::Quote { symbol, timestamp, bid: Some(bid_price), ask: Some(ask_price), bid_size: Some(bid_size), ask_size: Some(ask_size), exchange: Some(format!("pub_{}", header.publisher_id)), }; self.metrics.increment_quotes_processed(); Ok(Some(processed)) } DbnMessageType::OrderBook => { if data.len() < size_of::() { return Err(DataError::InvalidFormat("OrderBook message too short".to_string())); } let ob_msg = unsafe { std::ptr::read_unaligned(data.as_ptr() as *const DbnOrderBookMessage) }; let symbol = self.get_symbol(header.instrument_id); let price = self.scale_price(ob_msg.price, header.instrument_id); let size = Decimal::from(ob_msg.size); let processed = ProcessedMessage::OrderBook { symbol, timestamp, price, size, side: match ob_msg.side { b'A' => OrderSide::Sell, b'B' => OrderSide::Buy, _ => OrderSide::Buy, }, action: match ob_msg.action { b'A' => OrderBookAction::Add, b'C' => OrderBookAction::Cancel, b'M' => OrderBookAction::Modify, b'T' => OrderBookAction::Trade, _ => OrderBookAction::Add, }, level: ob_msg.channel_id as usize, order_id: Some(ob_msg.order_id.to_string()), }; self.metrics.increment_orderbook_processed(); Ok(Some(processed)) } DbnMessageType::Ohlcv => { if data.len() < size_of::() { return Err(DataError::InvalidFormat("OHLCV message too short".to_string())); } let ohlcv_msg = unsafe { std::ptr::read_unaligned(data.as_ptr() as *const DbnOhlcvMessage) }; let symbol = self.get_symbol(header.instrument_id); let open = self.scale_price(ohlcv_msg.open, header.instrument_id); let high = self.scale_price(ohlcv_msg.high, header.instrument_id); let low = self.scale_price(ohlcv_msg.low, header.instrument_id); let close = self.scale_price(ohlcv_msg.close, header.instrument_id); let volume = Decimal::from(ohlcv_msg.volume); let processed = ProcessedMessage::Ohlcv { symbol, timestamp, open, high, low, close, volume, }; self.metrics.increment_bars_processed(); Ok(Some(processed)) } DbnMessageType::Status | DbnMessageType::Error => { // Handle status and error messages let processed = ProcessedMessage::Status { timestamp, message: format!("Status message type: {:?}", message_type), }; Ok(Some(processed)) } } } /// SIMD batch processing for performance optimization fn simd_batch_process(&self, messages: &mut [ProcessedMessage]) -> Result<()> { if let Some(ref simd_ops) = self.simd_ops { // Group messages by type for SIMD processing let mut trade_prices = Vec::new(); let mut trade_volumes = Vec::new(); for msg in messages.iter() { if let ProcessedMessage::Trade { price, size, .. } = msg { trade_prices.push(price.to_f64().unwrap_or(0.0)); trade_volumes.push(size.to_f64().unwrap_or(0.0)); } } // Calculate VWAP using SIMD if we have enough trades if trade_prices.len() >= 4 { let aligned_prices = AlignedPrices::from_slice(&trade_prices); let aligned_volumes = AlignedVolumes::from_slice(&trade_volumes); unsafe { let vwap = simd_ops.calculate_vwap_aligned(&aligned_prices, &aligned_volumes); debug!("Batch VWAP calculated: {:.4}", vwap); self.metrics.record_vwap(vwap); } } } Ok(()) } /// Get symbol name for instrument ID fn get_symbol(&self, instrument_id: u32) -> String { self.symbol_map .read() .unwrap() .get(&instrument_id) .cloned() .unwrap_or_else(|| format!("UNKNOWN_{}", instrument_id)) } /// Scale integer price to decimal using instrument-specific scaling fn scale_price(&self, price: i64, instrument_id: u32) -> Price { let scale = self.price_scales .read() .unwrap() .get(&instrument_id) .copied() .unwrap_or(4); // Default to 4 decimal places Price::from(price) / Price::from(10_i64.pow(scale as u32)) } /// Send processed messages to event system pub async fn send_to_event_system(&self, messages: Vec) -> Result<()> { if let Some(ref processor) = self.event_processor { for msg in messages { let trading_event = self.convert_to_trading_event(msg)?; // Capture event with sub-microsecond latency if let Err(e) = processor.capture_event(trading_event).await { error!("Failed to capture trading event: {}", e); self.metrics.increment_event_errors(); } } } Ok(()) } /// Convert processed message to trading event fn convert_to_trading_event(&self, msg: ProcessedMessage) -> Result { match msg { ProcessedMessage::Trade { symbol, timestamp, price, size, side, trade_id, .. } => { Ok(TradingEvent::TradeExecuted { symbol, timestamp, price, quantity: size, side, trade_id: trade_id.unwrap_or_default(), }) } ProcessedMessage::Quote { symbol, timestamp, bid, ask, bid_size, ask_size, .. } => { Ok(TradingEvent::QuoteUpdated { symbol, timestamp, bid, ask, bid_size, ask_size, }) } ProcessedMessage::OrderBook { symbol, timestamp, price, size, side, action, .. } => { Ok(TradingEvent::OrderBookUpdated { symbol, timestamp, side, price, quantity: size, action: format!("{:?}", action), }) } _ => { Err(DataError::ConversionError("Unsupported message type for trading event".to_string())) } } } /// Get performance metrics pub fn get_metrics(&self) -> DbnParserMetricsSnapshot { self.metrics.get_snapshot() } } /// Processed message types from DBN parsing #[derive(Debug, Clone)] pub enum ProcessedMessage { Trade { symbol: String, timestamp: HardwareTimestamp, price: Price, size: Decimal, side: OrderSide, trade_id: Option, conditions: Vec, }, Quote { symbol: String, timestamp: HardwareTimestamp, bid: Option, ask: Option, bid_size: Option, ask_size: Option, exchange: Option, }, OrderBook { symbol: String, timestamp: HardwareTimestamp, price: Price, size: Decimal, side: OrderSide, action: OrderBookAction, level: usize, order_id: Option, }, Ohlcv { symbol: String, timestamp: HardwareTimestamp, open: Price, high: Price, low: Price, close: Price, volume: Decimal, }, Status { timestamp: HardwareTimestamp, message: String, }, } /// Order book actions #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum OrderBookAction { Add, Cancel, Modify, Trade, } /// Performance metrics for DBN parser #[derive(Debug)] pub struct DbnParserMetrics { messages_parsed: AtomicU64, trades_processed: AtomicU64, quotes_processed: AtomicU64, orderbook_processed: AtomicU64, bars_processed: AtomicU64, unknown_messages: AtomicU64, event_errors: AtomicU64, parse_latency_sum_ns: AtomicU64, parse_latency_count: AtomicU64, per_tick_latency_sum_ns: AtomicU64, per_tick_latency_count: AtomicU64, vwap_sum: AtomicU64, // Store as u64 (scaled by 10000) vwap_count: AtomicU64, } impl DbnParserMetrics { pub fn new() -> Self { Self { messages_parsed: AtomicU64::new(0), trades_processed: AtomicU64::new(0), quotes_processed: AtomicU64::new(0), orderbook_processed: AtomicU64::new(0), bars_processed: AtomicU64::new(0), unknown_messages: AtomicU64::new(0), event_errors: AtomicU64::new(0), parse_latency_sum_ns: AtomicU64::new(0), parse_latency_count: AtomicU64::new(0), per_tick_latency_sum_ns: AtomicU64::new(0), per_tick_latency_count: AtomicU64::new(0), vwap_sum: AtomicU64::new(0), vwap_count: AtomicU64::new(0), } } pub fn increment_messages_parsed(&self) { self.messages_parsed.fetch_add(1, Ordering::Relaxed); } pub fn increment_trades_processed(&self) { self.trades_processed.fetch_add(1, Ordering::Relaxed); } pub fn increment_quotes_processed(&self) { self.quotes_processed.fetch_add(1, Ordering::Relaxed); } pub fn increment_orderbook_processed(&self) { self.orderbook_processed.fetch_add(1, Ordering::Relaxed); } pub fn increment_bars_processed(&self) { self.bars_processed.fetch_add(1, Ordering::Relaxed); } pub fn increment_unknown_messages(&self) { self.unknown_messages.fetch_add(1, Ordering::Relaxed); } pub fn increment_event_errors(&self) { self.event_errors.fetch_add(1, Ordering::Relaxed); } pub fn record_parse_latency(&self, latency_ns: u64) { self.parse_latency_sum_ns.fetch_add(latency_ns, Ordering::Relaxed); self.parse_latency_count.fetch_add(1, Ordering::Relaxed); } pub fn record_per_tick_latency(&self, latency_ns: u64) { self.per_tick_latency_sum_ns.fetch_add(latency_ns, Ordering::Relaxed); self.per_tick_latency_count.fetch_add(1, Ordering::Relaxed); } pub fn record_vwap(&self, vwap: f64) { let scaled_vwap = (vwap * 10000.0) as u64; self.vwap_sum.fetch_add(scaled_vwap, Ordering::Relaxed); self.vwap_count.fetch_add(1, Ordering::Relaxed); } pub fn get_snapshot(&self) -> DbnParserMetricsSnapshot { let parse_count = self.parse_latency_count.load(Ordering::Relaxed); let avg_parse_latency_ns = if parse_count > 0 { self.parse_latency_sum_ns.load(Ordering::Relaxed) / parse_count } else { 0 }; let tick_count = self.per_tick_latency_count.load(Ordering::Relaxed); let avg_per_tick_latency_ns = if tick_count > 0 { self.per_tick_latency_sum_ns.load(Ordering::Relaxed) / tick_count } else { 0 }; let vwap_count = self.vwap_count.load(Ordering::Relaxed); let avg_vwap = if vwap_count > 0 { (self.vwap_sum.load(Ordering::Relaxed) / vwap_count) as f64 / 10000.0 } else { 0.0 }; DbnParserMetricsSnapshot { messages_parsed: self.messages_parsed.load(Ordering::Relaxed), trades_processed: self.trades_processed.load(Ordering::Relaxed), quotes_processed: self.quotes_processed.load(Ordering::Relaxed), orderbook_processed: self.orderbook_processed.load(Ordering::Relaxed), bars_processed: self.bars_processed.load(Ordering::Relaxed), unknown_messages: self.unknown_messages.load(Ordering::Relaxed), event_errors: self.event_errors.load(Ordering::Relaxed), avg_parse_latency_ns, avg_per_tick_latency_ns, avg_vwap, } } } /// Snapshot of DBN parser metrics #[derive(Debug, Clone, Serialize, Deserialize)] pub struct DbnParserMetricsSnapshot { pub messages_parsed: u64, pub trades_processed: u64, pub quotes_processed: u64, pub orderbook_processed: u64, pub bars_processed: u64, pub unknown_messages: u64, pub event_errors: u64, pub avg_parse_latency_ns: u64, pub avg_per_tick_latency_ns: u64, pub avg_vwap: f64, } #[cfg(test)] mod tests { use super::*; #[test] fn test_dbn_message_sizes() { // Verify packed struct sizes for zero-copy parsing assert_eq!(size_of::(), 16); assert_eq!(size_of::(), 32); assert_eq!(size_of::(), 48); assert_eq!(size_of::(), 48); assert_eq!(size_of::(), 56); } #[test] fn test_dbn_parser_creation() { let parser = DbnParser::new(); assert!(parser.is_ok()); let parser = parser.unwrap(); let metrics = parser.get_metrics(); assert_eq!(metrics.messages_parsed, 0); } #[tokio::test] async fn test_symbol_mapping() { let parser = DbnParser::new().unwrap(); let mut mapping = std::collections::HashMap::new(); mapping.insert(1, "AAPL".to_string()); mapping.insert(2, "MSFT".to_string()); parser.update_symbol_map(mapping); assert_eq!(parser.get_symbol(1), "AAPL"); assert_eq!(parser.get_symbol(2), "MSFT"); assert_eq!(parser.get_symbol(999), "UNKNOWN_999"); } #[test] fn test_price_scaling() { let parser = DbnParser::new().unwrap(); let mut scales = std::collections::HashMap::new(); scales.insert(1, 4); // 4 decimal places scales.insert(2, 2); // 2 decimal places parser.update_price_scales(scales); let price1 = parser.scale_price(123450, 1); // Should be 12.3450 let price2 = parser.scale_price(12345, 2); // Should be 123.45 assert_eq!(price1, Price::new(123450, 4)); assert_eq!(price2, Price::new(12345, 2)); } }