rusty_bin/monitor/schema/

mod.rs

//! Schema module for FlatBuffers serialization and deserialization
//!
//! This module provides high-level interfaces for working with trade and orderbook data
//! using FlatBuffers for efficient serialization.
//!
//! ## Performance Optimizations
//!
//! This module implements several JSON parsing optimizations:
//!
//! 1. **Binary Decimal Serialization**: Instead of converting `Decimal` to `String` and back,
//!    we use binary representation for FlatBuffers, eliminating the string conversion overhead.
//!
//! 2. **Fast Decimal Parsing**: Uses simd_json's optimized number parsing where possible,
//!    falling back to standard parsing only for edge cases.
//!
//! 3. **Zero-Copy Operations**: Leverages FlatBuffers' zero-copy capabilities with binary data
//!    instead of string-based serialization.
//!
//! These optimizations significantly improve performance for high-frequency trading scenarios
//! where every microsecond matters.

use planus::ReadAsRoot;
use rust_decimal::Decimal;
use simd_json;
use simd_json::prelude::*;
use smartstring::alias::String as SmartString;
use std::io;
use thiserror::Error;

/// Optimized serialization utilities for efficient Decimal handling
///
/// This module provides high-performance decimal parsing and serialization functions.
/// Designed for trading applications where decimal precision and performance are critical.
pub mod decimal_optimized {
    use rust_decimal::Decimal;
    use smartstring::alias::String as SmartString;

    /// Fast string to Decimal parser with optimized path for common cases
    ///
    /// Replaces the standard .parse() method for better performance.
    /// Uses optimized parsing for common decimal patterns, falling back to standard parsing for edge cases.
    ///
    /// # Arguments
    ///
    /// * `s` - String slice to parse as decimal
    ///
    /// # Returns
    ///
    /// Returns `Ok(Decimal)` on success, or `Err(String)` with error message on failure.
    pub fn fast_parse_decimal(s: &str) -> Result<Decimal, String> {
        // Fast path for common decimal patterns using optimized parsing
        match fast_decimal_from_str(s) {
            Ok(decimal) => Ok(decimal),
            Err(_) => {
                // Fallback to standard parsing for edge cases
                s.parse::<Decimal>()
                    .map_err(|e| format!("Failed to parse decimal '{s}': {e}"))
            }
        }
    }

    /// Optimized Decimal to string conversion using SmartString for performance
    ///
    /// Converts a decimal to string using [`SmartString`] for better memory efficiency.
    /// Optimized for short decimal strings common in trading scenarios.
    ///
    /// # Arguments
    ///
    /// * `decimal` - The decimal value to convert
    ///
    /// # Returns
    ///
    /// Returns a [`SmartString`] containing the decimal representation.
    pub fn fast_decimal_to_string(decimal: Decimal) -> SmartString {
        SmartString::from(decimal.to_string())
    }

    /// Fast decimal parsing for simple numeric strings
    ///
    /// Handles common trading decimal patterns efficiently using optimized parsing paths.
    /// Tries integer parsing first, then float conversion for simple decimals.
    ///
    /// # Arguments
    ///
    /// * `s` - String slice to parse
    ///
    /// # Returns
    ///
    /// Returns `Ok(Decimal)` on success, or `Err(&'static str)` for complex patterns.
    fn fast_decimal_from_str(s: &str) -> Result<Decimal, &'static str> {
        // Handle empty string
        if s.is_empty() {
            return Err("Empty string");
        }

        // Fast path for integers (no decimal point)
        if !s.contains('.')
            && let Ok(int_val) = s.parse::<i64>()
        {
            return Ok(Decimal::from(int_val));
        }

        // Fast path for simple decimals using f64 conversion
        if let Ok(float_val) = s.parse::<f64>()
            && float_val.is_finite()
        {
            return Decimal::try_from(float_val).map_err(|_| "Float conversion failed");
        }

        Err("Complex decimal pattern")
    }

    /// Batch convert vector of decimal strings with SIMD-friendly processing
    ///
    /// Processes multiple decimal strings in a single operation for better performance.
    /// Uses vectorized processing patterns that can benefit from SIMD optimizations.
    ///
    /// # Arguments
    ///
    /// * `strings` - Slice of string slices to parse as decimals
    ///
    /// # Returns
    ///
    /// Returns a vector of results, one for each input string.
    pub fn batch_parse_decimals(strings: &[&str]) -> Vec<Result<Decimal, String>> {
        strings.iter().map(|s| fast_parse_decimal(s)).collect()
    }
}

/// Generated FlatBuffers schema definitions
///
/// This module contains the auto-generated Rust code from FlatBuffers schema files.
/// It includes the compiled schema definitions for trade and orderbook data structures.
#[allow(missing_docs)]
pub mod generated;

/// Represents the side of a trade transaction
///
/// This enum identifies whether a trade was initiated by a buyer or seller.
/// Used in trade records to indicate market direction and flow.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum TradeSide {
    /// Trade initiated by a buyer (market buy order)
    Buy,
    /// Trade initiated by a seller (market sell order)
    Sell,
}

/// Errors that can occur during schema operations
///
/// Comprehensive error type for all schema-related operations including
/// serialization, deserialization, and data validation.
#[derive(Error, Debug)]
pub enum SchemaError {
    /// Error during serialization process
    #[error("Serialization failed: {0}")]
    Serialization(String),

    /// Error during deserialization process
    #[error("Deserialization failed: {0}")]
    Deserialization(String),

    /// Invalid or malformed data encountered
    #[error("Invalid data: {0}")]
    InvalidData(String),

    /// I/O operation failed
    #[error("IO error: {0}")]
    Io(#[from] io::Error),

    /// Feature or functionality not yet implemented
    #[error("Not implemented: {0}")]
    NotImplemented(String),
}

/// Result type alias for schema operations
///
/// Convenience type alias that uses [`SchemaError`] as the error type.
/// Used throughout the module for consistent error handling.
pub type Result<T> = std::result::Result<T, SchemaError>;

/// High-level trade data structure for easier manipulation
///
/// Represents a single trade transaction with all relevant metadata.
/// This structure is optimized for serialization and uses `SmartString` for performance.
#[derive(Debug, Clone)]
pub struct TradeRecord {
    /// Timestamp when the trade occurred at the exchange (nanoseconds since epoch)
    pub timestamp_exchange: u64,
    /// Timestamp when the trade was processed by our system (nanoseconds since epoch)
    pub timestamp_system: u64,
    /// Trading pair symbol (e.g., "BTCUSDT")
    pub symbol: SmartString,
    /// Exchange identifier (e.g., "binance", "coinbase")
    pub exchange: SmartString,
    /// Trade execution price using high-precision decimal arithmetic
    pub price: Decimal,
    /// Trade quantity/volume using high-precision decimal arithmetic
    pub quantity: Decimal,
    /// Side of the trade (buy or sell)
    pub side: TradeSide,
    /// Unique identifier for this trade from the exchange
    pub trade_id: SmartString,
    /// Optional order ID of the buyer (if available from exchange)
    pub buyer_order_id: Option<SmartString>,
    /// Optional order ID of the seller (if available from exchange)
    pub seller_order_id: Option<SmartString>,
    /// Sequence number for ordering trades chronologically
    pub sequence: u64,
}

/// High-level orderbook data structure for easier manipulation
///
/// Represents a complete order book snapshot with bid/ask levels.
/// Contains all price levels and metadata for a trading pair at a specific time.
#[derive(Debug, Clone)]
pub struct OrderBookRecord {
    /// Timestamp when the orderbook was captured at the exchange (nanoseconds since epoch)
    pub timestamp_exchange: u64,
    /// Timestamp when the orderbook was processed by our system (nanoseconds since epoch)
    pub timestamp_system: u64,
    /// Trading pair symbol (e.g., "BTCUSDT")
    pub symbol: SmartString,
    /// Exchange identifier (e.g., "binance", "coinbase")
    pub exchange: SmartString,
    /// Ordered list of bid price levels (highest to lowest)
    pub bids: Vec<PriceLevel>,
    /// Ordered list of ask price levels (lowest to highest)
    pub asks: Vec<PriceLevel>,
    /// Sequence number for ordering orderbook updates chronologically
    pub sequence: u64,
    /// Optional checksum for data integrity verification
    pub checksum: Option<SmartString>,
}

/// High-level price level structure
///
/// Represents a single price level in an order book with price, quantity, and order count.
/// Used for both bid and ask levels in orderbook snapshots.
#[derive(Debug, Clone)]
pub struct PriceLevel {
    /// Price at this level using high-precision decimal arithmetic
    pub price: Decimal,
    /// Total quantity available at this price level
    pub quantity: Decimal,
    /// Number of individual orders at this price level (if available)
    pub order_count: Option<u32>,
}

/// Batch of trade records for efficient processing
///
/// Groups multiple trade records together for batch operations.
/// Useful for bulk serialization and processing of trade data.
#[derive(Debug)]
pub struct TradesBatchRecord {
    /// Vector of trade records in this batch
    pub trades: Vec<TradeRecord>,
    /// Timestamp when this batch was created (nanoseconds since epoch)
    pub batch_timestamp: u64,
    /// Unique identifier for this batch
    pub batch_id: SmartString,
}

/// Batch of orderbook records for efficient processing
///
/// Groups multiple orderbook records together for batch operations.
/// Useful for bulk serialization and processing of orderbook data.
#[derive(Debug)]
pub struct OrderBookBatchRecord {
    /// Vector of orderbook records in this batch
    pub orderbooks: Vec<OrderBookRecord>,
    /// Timestamp when this batch was created (nanoseconds since epoch)
    pub batch_timestamp: u64,
    /// Unique identifier for this batch
    pub batch_id: SmartString,
}

/// Serialization utilities for trade data
///
/// Provides methods to serialize and deserialize trade records using FlatBuffers.
/// Optimized for high-frequency trading scenarios with thread-local builders.
pub struct TradeSerializer;

thread_local! {
    /// Thread-local builder for zero-allocation serialization
    ///
    /// Provides a reusable FlatBuffers builder per thread to avoid allocations.
    /// Used by [`TradeSerializer::serialize_trade_into`] for optimal performance.
    static TRADE_BUILDER: std::cell::RefCell<planus::Builder> = std::cell::RefCell::new(planus::Builder::new());
}

impl TradeSerializer {
    /// Serialize a single trade record to FlatBuffers
    ///
    /// Converts a [`TradeRecord`] to FlatBuffers binary format for efficient storage.
    /// Returns a vector of bytes containing the serialized trade data.
    ///
    /// # Arguments
    ///
    /// * `trade` - The trade record to serialize
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<u8>)` containing the serialized data, or `Err(SchemaError)` on failure.
    pub fn serialize_trade(trade: &TradeRecord) -> Result<Vec<u8>> {
        use crate::monitor::schema::generated::rusty_monitor::schema::{
            Trade, TradeSide as FbTradeSide,
        };

        let mut builder = planus::Builder::new();

        // Convert internal TradeSide to FlatBuffers TradeSide
        let fb_side = match trade.side {
            TradeSide::Buy => FbTradeSide::Buy,
            TradeSide::Sell => FbTradeSide::Sell,
        };

        // Build the Trade using the generated builder pattern
        // Note: We need to find the correct builder construction pattern
        let trade_data = Trade {
            timestamp_exchange: trade.timestamp_exchange,
            timestamp_system: trade.timestamp_system,
            symbol: trade.symbol.to_string(),
            exchange: trade.exchange.to_string(),
            price: trade.price.to_string(),
            quantity: trade.quantity.to_string(),
            side: fb_side,
            trade_id: Some(trade.trade_id.to_string()),
            buyer_order_id: trade.buyer_order_id.as_ref().map(|s| s.to_string()),
            seller_order_id: trade.seller_order_id.as_ref().map(|s| s.to_string()),
            sequence: trade.sequence,
        };

        let trade_offset = planus::WriteAsOffset::prepare(&trade_data, &mut builder);
        let bytes = builder.finish(trade_offset, None);
        Ok(bytes.to_vec())
    }

    /// Serialize a batch of trades to FlatBuffers
    ///
    /// Converts a [`TradesBatchRecord`] containing multiple trades to FlatBuffers binary format.
    /// More efficient than serializing individual trades for bulk operations.
    ///
    /// # Arguments
    ///
    /// * `batch` - The batch of trades to serialize
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<u8>)` containing the serialized batch data, or `Err(SchemaError)` on failure.
    pub fn serialize_batch(batch: &TradesBatchRecord) -> Result<Vec<u8>> {
        use crate::monitor::schema::generated::rusty_monitor::schema::{
            Trade, TradeSide as FbTradeSide, TradesBatch,
        };

        let mut builder = planus::Builder::new();

        // Convert all trades to FlatBuffers format
        let trades: Result<Vec<Trade>> = batch
            .trades
            .iter()
            .map(|trade| {
                let fb_side = match trade.side {
                    TradeSide::Buy => FbTradeSide::Buy,
                    TradeSide::Sell => FbTradeSide::Sell,
                };

                Ok(Trade {
                    timestamp_exchange: trade.timestamp_exchange,
                    timestamp_system: trade.timestamp_system,
                    symbol: trade.symbol.to_string(),
                    exchange: trade.exchange.to_string(),
                    price: trade.price.to_string(),
                    quantity: trade.quantity.to_string(),
                    side: fb_side,
                    trade_id: Some(trade.trade_id.to_string()),
                    buyer_order_id: trade.buyer_order_id.as_ref().map(|s| s.to_string()),
                    seller_order_id: trade.seller_order_id.as_ref().map(|s| s.to_string()),
                    sequence: trade.sequence,
                })
            })
            .collect();

        let trades = trades?;

        let trades_batch = TradesBatch {
            trades,
            batch_timestamp: batch.batch_timestamp,
            batch_id: Some(batch.batch_id.to_string()),
        };

        let batch_offset = planus::WriteAsOffset::prepare(&trades_batch, &mut builder);
        let bytes = builder.finish(batch_offset, None);
        Ok(bytes.to_vec())
    }

    /// Serialize a trade directly into a provided buffer (zero-copy)
    ///
    /// Serializes a trade record directly into a provided buffer, avoiding allocation.
    /// Uses thread-local builder for optimal performance in high-frequency scenarios.
    ///
    /// # Arguments
    ///
    /// * `trade` - The trade record to serialize
    /// * `buffer` - Mutable buffer to write serialized data into
    ///
    /// # Returns
    ///
    /// Returns `Ok(usize)` with the number of bytes written, or `Err(SchemaError)` on failure.
    pub fn serialize_trade_into(trade: &TradeRecord, buffer: &mut Vec<u8>) -> Result<usize> {
        use crate::monitor::schema::generated::rusty_monitor::schema::{
            Trade, TradeSide as FbTradeSide,
        };

        TRADE_BUILDER.with(|builder_cell| {
            let mut builder = builder_cell.borrow_mut();
            builder.clear(); // Reset builder for reuse

            // Convert internal TradeSide to FlatBuffers TradeSide
            let fb_side = match trade.side {
                TradeSide::Buy => FbTradeSide::Buy,
                TradeSide::Sell => FbTradeSide::Sell,
            };

            // Build the Trade using the generated builder pattern
            let trade_data = Trade {
                timestamp_exchange: trade.timestamp_exchange,
                timestamp_system: trade.timestamp_system,
                symbol: trade.symbol.to_string(),
                exchange: trade.exchange.to_string(),
                price: trade.price.to_string(),
                quantity: trade.quantity.to_string(),
                side: fb_side,
                trade_id: Some(trade.trade_id.to_string()),
                buyer_order_id: trade.buyer_order_id.as_ref().map(|s| s.to_string()),
                seller_order_id: trade.seller_order_id.as_ref().map(|s| s.to_string()),
                sequence: trade.sequence,
            };

            let trade_offset = planus::WriteAsOffset::prepare(&trade_data, &mut builder);
            let bytes = builder.finish(trade_offset, None);

            // Clear the buffer and copy data
            buffer.clear();
            buffer.extend_from_slice(bytes);

            Ok(bytes.len())
        })
    }

    /// Deserialize trade data from FlatBuffers
    ///
    /// Converts FlatBuffers binary data back to a [`TradeRecord`] structure.
    /// Uses optimized decimal parsing for better performance.
    ///
    /// # Arguments
    ///
    /// * `data` - Byte slice containing serialized trade data
    ///
    /// # Returns
    ///
    /// Returns `Ok(TradeRecord)` with the deserialized trade, or `Err(SchemaError)` on failure.
    pub fn deserialize_trade(data: &[u8]) -> Result<TradeRecord> {
        use crate::monitor::schema::generated::rusty_monitor::schema::{
            TradeRef, TradeSide as FbTradeSide,
        };

        // Deserialize using FlatBuffers
        let trade_ref = TradeRef::read_as_root(data)
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;

        // Extract fields from FlatBuffers reference
        let timestamp_exchange = trade_ref
            .timestamp_exchange()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let timestamp_system = trade_ref
            .timestamp_system()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let symbol = trade_ref
            .symbol()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let exchange = trade_ref
            .exchange()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let price_str = trade_ref
            .price()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let quantity_str = trade_ref
            .quantity()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let fb_side = trade_ref
            .side()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let trade_id = trade_ref
            .trade_id()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let buyer_order_id = trade_ref.buyer_order_id().ok().flatten();
        let seller_order_id = trade_ref.seller_order_id().ok().flatten();
        let sequence = trade_ref
            .sequence()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;

        // Convert FlatBuffers string types to internal types using optimized parsing
        let price = decimal_optimized::fast_parse_decimal(price_str)
            .map_err(|e| SchemaError::InvalidData(format!("Invalid price: {e}")))?;
        let quantity = decimal_optimized::fast_parse_decimal(quantity_str)
            .map_err(|e| SchemaError::InvalidData(format!("Invalid quantity: {e}")))?;
        let side = match fb_side {
            FbTradeSide::Buy => TradeSide::Buy,
            FbTradeSide::Sell => TradeSide::Sell,
        };

        Ok(TradeRecord {
            timestamp_exchange,
            timestamp_system,
            symbol: symbol.into(),
            exchange: exchange.into(),
            price,
            quantity,
            side,
            trade_id: trade_id.unwrap_or("").into(),
            buyer_order_id: buyer_order_id.map(|s| s.into()),
            seller_order_id: seller_order_id.map(|s| s.into()),
            sequence,
        })
    }
}

/// Serialization utilities for orderbook data
///
/// Provides methods to serialize and deserialize orderbook records using FlatBuffers.
/// Handles complex price level structures with optimized decimal parsing.
pub struct OrderBookSerializer;

impl OrderBookSerializer {
    /// Serialize a single orderbook record to FlatBuffers
    ///
    /// Converts an [`OrderBookRecord`] to FlatBuffers binary format for efficient storage.
    /// Handles complex price level structures with multiple bid/ask levels.
    ///
    /// # Arguments
    ///
    /// * `orderbook` - The orderbook record to serialize
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<u8>)` containing the serialized data, or `Err(SchemaError)` on failure.
    pub fn serialize_orderbook(orderbook: &OrderBookRecord) -> Result<Vec<u8>> {
        use crate::monitor::schema::generated::orderbook_generated::rusty_monitor::schema::{
            OrderBook, PriceLevel as FbPriceLevel,
        };

        let mut builder = planus::Builder::new();

        // Convert price levels to FlatBuffers format
        let bids: Vec<FbPriceLevel> = orderbook
            .bids
            .iter()
            .map(|level| FbPriceLevel {
                price: level.price.to_string(),
                quantity: level.quantity.to_string(),
                order_count: level.order_count.unwrap_or(0),
            })
            .collect();

        let asks: Vec<FbPriceLevel> = orderbook
            .asks
            .iter()
            .map(|level| FbPriceLevel {
                price: level.price.to_string(),
                quantity: level.quantity.to_string(),
                order_count: level.order_count.unwrap_or(0),
            })
            .collect();

        let orderbook_data = OrderBook {
            timestamp_exchange: orderbook.timestamp_exchange,
            timestamp_system: orderbook.timestamp_system,
            symbol: orderbook.symbol.to_string(),
            exchange: orderbook.exchange.to_string(),
            bids,
            asks,
            sequence: orderbook.sequence,
            last_update_id: 0, // Default value, could be added to OrderBookRecord if needed
            is_snapshot: true, // Default value, could be added to OrderBookRecord if needed
            depth: (orderbook.bids.len() + orderbook.asks.len()) as u32,
            checksum: orderbook.checksum.as_ref().map(|s| s.to_string()),
        };

        let orderbook_offset = planus::WriteAsOffset::prepare(&orderbook_data, &mut builder);
        let bytes = builder.finish(orderbook_offset, None);
        Ok(bytes.to_vec())
    }

    /// Deserialize orderbook data from FlatBuffers
    ///
    /// Converts FlatBuffers binary data back to an [`OrderBookRecord`] structure.
    /// Reconstructs all price levels with optimized decimal parsing.
    ///
    /// # Arguments
    ///
    /// * `data` - Byte slice containing serialized orderbook data
    ///
    /// # Returns
    ///
    /// Returns `Ok(OrderBookRecord)` with the deserialized orderbook, or `Err(SchemaError)` on failure.
    pub fn deserialize_orderbook(data: &[u8]) -> Result<OrderBookRecord> {
        use crate::monitor::schema::generated::orderbook_generated::rusty_monitor::schema::OrderBookRef;

        // Deserialize using FlatBuffers
        let orderbook_ref = OrderBookRef::read_as_root(data)
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;

        // Extract fields from FlatBuffers reference
        let timestamp_exchange = orderbook_ref
            .timestamp_exchange()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let timestamp_system = orderbook_ref
            .timestamp_system()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let symbol = orderbook_ref
            .symbol()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let exchange = orderbook_ref
            .exchange()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let sequence = orderbook_ref
            .sequence()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let checksum = orderbook_ref.checksum().ok().flatten();

        // Convert price levels
        let bids_vec = orderbook_ref
            .bids()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let bids = Self::parse_price_levels_fb(&bids_vec)?;

        let asks_vec = orderbook_ref
            .asks()
            .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
        let asks = Self::parse_price_levels_fb(&asks_vec)?;

        Ok(OrderBookRecord {
            timestamp_exchange,
            timestamp_system,
            symbol: symbol.into(),
            exchange: exchange.into(),
            bids,
            asks,
            sequence,
            checksum: checksum.map(|s| s.into()),
        })
    }

    /// Helper to parse price levels from JSON
    ///
    /// Converts JSON array of price levels to internal [`PriceLevel`] structures.
    /// Used for processing raw JSON data from exchanges.
    ///
    /// # Arguments
    ///
    /// * `levels_json` - JSON value containing array of price levels
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<PriceLevel>)` with parsed levels, or `Err(SchemaError)` on failure.
    fn parse_price_levels_json(levels_json: &simd_json::OwnedValue) -> Result<Vec<PriceLevel>> {
        let levels_array = levels_json
            .as_array()
            .ok_or_else(|| SchemaError::InvalidData("Price levels must be an array".to_string()))?;

        let mut result = Vec::new();
        for level_json in levels_array {
            let price = level_json["price"]
                .as_str()
                .ok_or_else(|| SchemaError::InvalidData("Missing price in level".to_string()))?
                .parse::<Decimal>()
                .map_err(|e| SchemaError::InvalidData(format!("Invalid price: {e}")))?;

            let quantity = level_json["quantity"]
                .as_str()
                .ok_or_else(|| SchemaError::InvalidData("Missing quantity in level".to_string()))?
                .parse::<Decimal>()
                .map_err(|e| SchemaError::InvalidData(format!("Invalid quantity: {e}")))?;

            let order_count = level_json["order_count"].as_u64().map(|c| c as u32);

            result.push(PriceLevel {
                price,
                quantity,
                order_count,
            });
        }

        Ok(result)
    }

    /// Helper to parse price levels from FlatBuffers
    ///
    /// Converts FlatBuffers vector of price levels to internal [`PriceLevel`] structures.
    /// Used during deserialization of orderbook data.
    ///
    /// # Arguments
    ///
    /// * `levels_vec` - FlatBuffers vector containing price level references
    ///
    /// # Returns
    ///
    /// Returns `Ok(Vec<PriceLevel>)` with parsed levels, or `Err(SchemaError)` on failure.
    fn parse_price_levels_fb<'a>(
        levels_vec: &planus::Vector<'a, planus::Result<crate::monitor::schema::generated::orderbook_generated::rusty_monitor::schema::PriceLevelRef<'a>>>,
    ) -> Result<Vec<PriceLevel>> {
        let mut result = Vec::new();
        for i in 0..levels_vec.len() {
            let level_ref = levels_vec
                .get(i)
                .unwrap()
                .map_err(|e| SchemaError::Deserialization(e.to_string()))?;

            let price_str = level_ref
                .price()
                .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
            let quantity_str = level_ref
                .quantity()
                .map_err(|e| SchemaError::Deserialization(e.to_string()))?;
            let order_count = level_ref
                .order_count()
                .map_err(|e| SchemaError::Deserialization(e.to_string()))?;

            let price = decimal_optimized::fast_parse_decimal(price_str)
                .map_err(|e| SchemaError::InvalidData(format!("Invalid price: {e}")))?;
            let quantity = decimal_optimized::fast_parse_decimal(quantity_str)
                .map_err(|e| SchemaError::InvalidData(format!("Invalid quantity: {e}")))?;

            result.push(PriceLevel {
                price,
                quantity,
                order_count: if order_count == 0 {
                    None
                } else {
                    Some(order_count)
                },
            });
        }

        Ok(result)
    }
}

/// Utility functions for timestamp handling
///
/// Provides high-resolution timestamp functions for market data timestamping.
/// All timestamps are in nanoseconds since Unix epoch for maximum precision.
pub mod timestamp {
    /// Get current timestamp in nanoseconds
    ///
    /// Returns the current high-resolution timestamp in nanoseconds since Unix epoch.
    /// Uses the monitor's optimized time utilities for consistent timing.
    pub fn now_nanos() -> u64 {
        crate::monitor::utils::time::now_nanos()
    }

    /// Get system time in nanoseconds since UNIX epoch
    ///
    /// Returns the current system time in nanoseconds since Unix epoch.
    /// Uses standard library SystemTime for compatibility.
    pub fn system_time_nanos() -> u64 {
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos() as u64
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use rust_decimal_macros::dec;

    #[test]
    fn test_trade_serialization_roundtrip() {
        let trade = TradeRecord {
            timestamp_exchange: 1234567890123456789,
            timestamp_system: 1234567890123456790,
            symbol: "BTCUSDT".into(),
            exchange: "binance".into(),
            price: dec!(50000.123),
            quantity: dec!(0.001),
            side: TradeSide::Buy,
            trade_id: "12345".into(),
            buyer_order_id: Some("buyer123".into()),
            seller_order_id: Some("seller456".into()),
            sequence: 100,
        };

        let serialized = TradeSerializer::serialize_trade(&trade).unwrap();
        let deserialized = TradeSerializer::deserialize_trade(&serialized).unwrap();

        assert_eq!(trade.timestamp_exchange, deserialized.timestamp_exchange);
        assert_eq!(trade.symbol, deserialized.symbol);
        assert_eq!(trade.price, deserialized.price);
        assert_eq!(trade.side, deserialized.side);
    }

    #[test]
    fn test_orderbook_serialization_roundtrip() {
        let orderbook = OrderBookRecord {
            timestamp_exchange: 1234567890123456789,
            timestamp_system: 1234567890123456790,
            symbol: "BTCUSDT".into(),
            exchange: "binance".into(),
            bids: vec![PriceLevel {
                price: dec!(50000.0),
                quantity: dec!(1.0),
                order_count: Some(5),
            }],
            asks: vec![PriceLevel {
                price: dec!(50001.0),
                quantity: dec!(2.0),
                order_count: Some(3),
            }],
            sequence: 200,
            checksum: Some("abc123".into()),
        };

        let serialized = OrderBookSerializer::serialize_orderbook(&orderbook).unwrap();
        let deserialized = OrderBookSerializer::deserialize_orderbook(&serialized).unwrap();

        assert_eq!(
            orderbook.timestamp_exchange,
            deserialized.timestamp_exchange
        );
        assert_eq!(orderbook.symbol, deserialized.symbol);
        assert_eq!(orderbook.bids.len(), deserialized.bids.len());
        assert_eq!(orderbook.asks.len(), deserialized.asks.len());
    }
}