rusty_feeder/

feeder.rs

use std::fmt::Debug;
use std::sync::Arc;

use anyhow::Result;
use async_trait::async_trait;
use parking_lot::RwLock;
use rusty_model::{
    data::{
        bar::{Bar, BarCache, BarType},
        book_snapshot::OrderBookSnapshot,
        market_trade::MarketTrade,
        simd_orderbook::SharedSimdOrderBook,
    },
    enums::OrderSide,
    instruments::InstrumentId,
};
use smallvec::SmallVec;
use tokio::sync::mpsc;

/// Performance statistics for feed processing in HFT applications
///
/// This structure provides comprehensive statistics for monitoring, tuning, and
/// debugging high-frequency trading data feeds. It tracks latency, throughput,
/// and resource usage metrics that are critical for HFT operations.
///
/// Key features:
/// - Zero-allocation percentile calculations using fixed-size arrays
/// - Efficient EWMA (Exponentially Weighted Moving Average) for smooth metrics
/// - Cache-line alignment for optimal CPU cache efficiency
/// - Detailed memory usage tracking for resource monitoring
/// - Thread-safe and lock-free metrics updates
///
/// # Performance Characteristics
///
/// The statistics collection is designed to have minimal impact on the critical path:
/// - O(1) update operations for most metrics
/// - O(n log n) for percentile calculations but limited to small fixed-size arrays
/// - Zero heap allocations during normal operation
/// - Minimal cache contention through 64-byte alignment
///
/// # Example Usage
///
/// ```no_run
/// use rusty_feeder::feeder::FeedStats;
///
/// // Create new statistics tracker
/// let mut stats = FeedStats::default();
///
/// // Update with a latency sample
/// stats.add_latency_sample(250); // 250 nanoseconds
///
/// // Track memory usage
/// stats.update_memory_usage(1024); // 1KB
///
/// // Print current statistics
/// println!("Avg latency: {}ns, P99: {}ns, Max: {}ns",
///     stats.avg_process_latency_ns,
///     stats.p99_process_latency_ns,
///     stats.max_process_latency_ns);
/// ```
#[derive(Debug, Clone)]
#[repr(align(64))] // Cache-line alignment for better CPU cache efficiency
pub struct FeedStats {
    /// Total messages processed
    pub messages_processed: u64,

    /// Messages processed per second (smoothed)
    pub messages_per_second: f64,

    /// Average processing latency (nanoseconds)
    pub avg_process_latency_ns: u64,

    /// Maximum processing latency observed (nanoseconds)
    pub max_process_latency_ns: u64,

    /// 99th percentile processing latency (nanoseconds)
    pub p99_process_latency_ns: u64,

    /// Total dropped messages
    pub dropped_messages: u64,

    /// Last update timestamp (nanoseconds)
    pub last_update_time: u64,

    /// Memory usage for this feed (bytes)
    pub memory_usage_bytes: usize,

    /// Recent latency samples for calculating percentiles
    /// Using a fixed-size array to avoid heap allocations
    recent_latencies: [u64; 100],

    /// Current position in the recent_latencies circular buffer
    recent_latencies_pos: usize,

    /// Number of samples collected in the circular buffer
    recent_latencies_count: usize,
}

impl Default for FeedStats {
    fn default() -> Self {
        Self {
            messages_processed: 0,
            messages_per_second: 0.0,
            avg_process_latency_ns: 0,
            max_process_latency_ns: 0,
            p99_process_latency_ns: 0,
            dropped_messages: 0,
            last_update_time: 0,
            memory_usage_bytes: 0,
            recent_latencies: [0; 100],
            recent_latencies_pos: 0,
            recent_latencies_count: 0,
        }
    }
}

impl FeedStats {
    /// Add a new latency sample and update all latency statistics
    ///
    /// This method efficiently updates all latency-related metrics in one call:
    /// - Exponentially Weighted Moving Average (EWMA) for stable average latency
    /// - Maximum observed latency
    /// - 99th percentile latency using a rolling window
    ///
    /// Latency metrics are critical for high-frequency trading systems where
    /// microsecond or even nanosecond differences can impact trading performance.
    ///
    /// # Parameters
    ///
    /// * `latency_ns` - The measured latency in nanoseconds
    ///
    /// # Performance Characteristics
    ///
    /// - Updating avg and max latency is O(1)
    /// - P99 calculation is O(n log n) but is only performed when enough samples are collected
    /// - Uses fixed-size arrays (zero heap allocation)
    /// - The implementation uses an insertion sort for small arrays which is more efficient
    ///   than quicksort for arrays of size ~100 elements
    #[inline]
    pub fn add_latency_sample(&mut self, latency_ns: u64) {
        // Update exponentially weighted moving average (EWMA)
        self.avg_process_latency_ns = (self.avg_process_latency_ns * 9 + latency_ns) / 10;

        // Update max latency
        self.max_process_latency_ns = self.max_process_latency_ns.max(latency_ns);

        // Store in circular buffer for percentile calculation
        self.recent_latencies[self.recent_latencies_pos] = latency_ns;
        self.recent_latencies_pos = (self.recent_latencies_pos + 1) % 100;
        if self.recent_latencies_count < 100 {
            self.recent_latencies_count += 1;
        }

        // Update p99 latency if we have enough samples
        if self.recent_latencies_count >= 10 {
            self.update_p99_latency();
        }
    }

    /// Calculate the 99th percentile latency from the recent samples
    #[inline]
    fn update_p99_latency(&mut self) {
        // Make a copy of the recent latencies using a stack-allocated array
        let mut latencies = [0u64; 100];
        let count = self.recent_latencies_count;

        // Copy only the elements that have been initialized
        latencies[..count].copy_from_slice(&self.recent_latencies[..count]);

        // Sort the latencies in ascending order
        // For small arrays, this is faster than quicksort
        // Using insertion sort as it's efficient for small arrays
        for i in 1..count {
            let key = latencies[i];
            let mut j = i;
            while j > 0 && latencies[j - 1] > key {
                latencies[j] = latencies[j - 1];
                j -= 1;
            }
            latencies[j] = key;
        }

        // Calculate the 99th percentile index
        let p99_index = ((count as f64) * 0.99) as usize;

        // If we have enough samples, use the calculated index
        // Otherwise, use the highest value we have
        let index = if p99_index < count {
            p99_index
        } else {
            count - 1
        };

        // Update the p99 latency
        self.p99_process_latency_ns = latencies[index];
    }

    /// Increment the count of dropped messages
    ///
    /// This method should be called whenever a message is dropped due to
    /// errors, buffer overflow, or other exceptional conditions. Tracking
    /// dropped messages is critical for high-frequency trading systems
    /// to detect data quality issues.
    ///
    /// # Performance Impact
    ///
    /// This operation is O(1) and has minimal performance impact.
    #[inline]
    pub const fn increment_dropped(&mut self) {
        self.dropped_messages += 1;
    }

    /// Increment the count of processed messages
    ///
    /// This method should be called for every successfully processed message.
    /// The counter is used to calculate throughput and other performance metrics.
    ///
    /// # Performance Impact
    ///
    /// This operation is O(1) and has minimal performance impact.
    #[inline]
    pub const fn increment_processed(&mut self) {
        self.messages_processed += 1;
    }

    /// Update the memory usage estimate for resource monitoring
    #[inline]
    ///
    /// This method applies an exponentially weighted moving average (EWMA) to
    /// produce a stable estimate of memory usage over time, which is useful
    /// for detecting memory leaks and tracking resource utilization.
    ///
    /// # Parameters
    ///
    /// * `new_bytes` - The latest memory usage sample in bytes
    ///
    /// # Notes
    ///
    /// The EWMA formula used gives 90% weight to historical data and 10% to
    /// the new sample, which provides good stability while still responding
    /// to significant changes. This approach is particularly useful for
    /// high-frequency trading applications where small variations in memory
    /// usage are expected but sustained trends are important to track.
    ///
    /// Memory usage is tracked per feed, allowing for granular monitoring
    /// of different market data streams.
    pub const fn update_memory_usage(&mut self, new_bytes: usize) {
        // Using EWMA to stabilize the memory usage estimate
        self.memory_usage_bytes = (self.memory_usage_bytes * 9 + new_bytes) / 10;
    }
}

/// Configuration options for feeder performance tuning.
///
/// These options control various aspects of feed processing performance.
/// The settings are optimized for high-frequency trading scenarios.
///
/// Fields:
/// - `channel_buffer_size`: Channel buffer size for processed data.
/// - `max_depth_levels`: Maximum depth levels to process for orderbooks.
/// - `batch_size`: Batch size for processing messages (0 = no batching).
/// - `cpu_affinity`: CPU core affinity for processing threads (-1 = no affinity).
/// - `use_zero_copy`: Whether to use zero-copy processing when possible.
/// - `max_backlog`: Maximum message backlog before dropping.
///
/// Note: The "prefetch options" field has been removed in this version.
///
#[derive(Debug, Clone)]
pub struct FeederOptions {
    /// Channel buffer size for processed data
    pub channel_buffer_size: usize,

    /// Maximum depth levels to process for orderbooks
    pub max_depth_levels: usize,

    /// Batch size for processing messages (0 = no batching)
    pub batch_size: usize,

    /// CPU core affinity for processing threads (-1 = no affinity)
    pub cpu_affinity: i32,

    /// Whether to use zero-copy processing when possible
    pub use_zero_copy: bool,

    /// Maximum message backlog before dropping
    pub max_backlog: usize,
}

impl Default for FeederOptions {
    fn default() -> Self {
        Self {
            channel_buffer_size: 1024,
            max_depth_levels: 100,
            batch_size: 32,
            cpu_affinity: -1,
            use_zero_copy: true,
            max_backlog: 10000,
        }
    }
}

/// High-performance feeder interface for processing and normalizing exchange data
/// Optimized for minimal allocations and low latency
#[async_trait]
pub trait Feeder: Send + Sync + 'static + Debug {
    /// Raw message type from the exchange for depth/orderbook
    type DepthMessage: Send + 'static;

    /// Raw message type from the exchange for trades
    type TradeMessage: Send + 'static;

    /// Start processing and normalizing orderbook depth data
    /// Processes raw depth messages into standardized order book depth format
    async fn start_feed_depth(
        &self,
        instrument_id: InstrumentId,
        depth_rx: mpsc::Receiver<Self::DepthMessage>,
        options: Option<FeederOptions>,
    ) -> Result<mpsc::Receiver<OrderBookSnapshot>>;

    /// Stop processing orderbook depth data for a specific instrument
    async fn stop_feed_depth(&self, instrument_id: &InstrumentId) -> Result<()>;

    /// Start processing and normalizing trade data
    /// Processes raw trade messages into standardized trade format
    async fn start_feed_trades(
        &self,
        instrument_id: InstrumentId,
        trade_rx: mpsc::Receiver<Self::TradeMessage>,
        options: Option<FeederOptions>,
    ) -> Result<mpsc::Receiver<MarketTrade>>;

    /// Stop processing trade data for a specific instrument
    async fn stop_feed_trades(&self, instrument_id: &InstrumentId) -> Result<()>;

    /// Start generating bars from trade data
    /// Aggregates trade data into OHLCV bars of the specified type
    async fn start_feed_bars(
        &self,
        instrument_id: InstrumentId,
        bar_type: BarType,
        trade_rx: mpsc::Receiver<MarketTrade>,
        options: Option<FeederOptions>,
    ) -> Result<mpsc::Receiver<Bar>>;

    /// Stop generating bars for a specific instrument and bar type
    async fn stop_feed_bars(&self, instrument_id: &InstrumentId, bar_type: &BarType) -> Result<()>;

    /// Get a real-time shared orderbook for a specific instrument
    /// Returns a thread-safe shared orderbook that's kept up-to-date
    async fn get_shared_orderbook(
        &self,
        instrument_id: &InstrumentId,
    ) -> Result<SharedSimdOrderBook>;

    /// Get a bar cache for a specific instrument and bar type
    /// Returns a bar cache with recent bar data
    async fn get_bar_cache(
        &self,
        instrument_id: &InstrumentId,
        bar_type: &BarType,
        max_bars: usize,
    ) -> Result<Arc<RwLock<BarCache>>>;

    /// Get feed statistics for a specific instrument
    async fn get_stats(&self, _instrument_id: &InstrumentId) -> Result<FeedStats> {
        Ok(FeedStats::default())
    }

    /// Reset all feed statistics
    async fn reset_stats(&self) -> Result<()> {
        Ok(())
    }
}

/// Advanced orderbook analytics for HFT applications
pub trait OrderbookAnalytics {
    /// Calculate market impact for a given notional amount
    /// Returns the estimated price impact as a percentage
    fn calculate_market_impact(&self, notional_amount: f64, side: OrderSide) -> f64;

    /// Get spread as a percentage of mid price
    fn get_spread_percent(&self) -> f64;

    /// Get the effective price for a given quantity
    /// Returns the volume-weighted average execution price
    fn get_effective_price(&self, quantity: f64, side: OrderSide) -> Option<f64>;

    /// Get orderbook imbalance metric
    /// Returns a value between -1.0 (all asks) and 1.0 (all bids)
    fn get_imbalance(&self, levels: usize) -> f64;

    /// Calculate liquidity at specified price levels
    /// Returns total volume available within the price range
    fn calculate_liquidity(&self, price_range_percent: f64, side: OrderSide) -> f64;

    /// Calculate order book resistance levels
    /// Returns price levels with significant volume
    fn find_resistance_levels(&self, min_volume: f64, side: OrderSide)
    -> SmallVec<[(f64, f64); 8]>;

    /// Calculate weighted mid price based on volume
    /// Returns a more accurate mid price weighted by volume at each level
    fn weighted_mid_price(&self, depth_levels: usize) -> Option<f64>;

    /// Calculate stability of the orderbook
    /// Returns a stability score between 0.0 (unstable) and 1.0 (stable)
    fn stability_score(&self) -> f64;

    /// Calculate market pressure indicator
    /// Returns a value between -1.0 (sell pressure) and 1.0 (buy pressure)
    fn market_pressure(&self) -> f64;
}

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

    #[test]
    fn test_feed_stats_default() {
        let stats = FeedStats::default();
        assert_eq!(stats.messages_processed, 0);
        assert_eq!(stats.messages_per_second, 0.0);
        assert_eq!(stats.avg_process_latency_ns, 0);
        assert_eq!(stats.max_process_latency_ns, 0);
        assert_eq!(stats.p99_process_latency_ns, 0);
        assert_eq!(stats.dropped_messages, 0);
        assert_eq!(stats.last_update_time, 0);
        assert_eq!(stats.memory_usage_bytes, 0);
    }

    #[test]
    fn test_feed_stats_increment_processed() {
        let mut stats = FeedStats::default();

        // Test single increment
        stats.increment_processed();
        assert_eq!(stats.messages_processed, 1);

        // Test multiple increments
        for _ in 0..10 {
            stats.increment_processed();
        }
        assert_eq!(stats.messages_processed, 11);
    }

    #[test]
    fn test_feed_stats_increment_dropped() {
        let mut stats = FeedStats::default();

        // Test single increment
        stats.increment_dropped();
        assert_eq!(stats.dropped_messages, 1);

        // Test multiple increments
        for _ in 0..5 {
            stats.increment_dropped();
        }
        assert_eq!(stats.dropped_messages, 6);
    }

    #[test]
    fn test_feed_stats_update_memory_usage() {
        let mut stats = FeedStats::default();

        // Test initial update
        stats.update_memory_usage(1000);
        assert_eq!(stats.memory_usage_bytes, 100); // (0*9 + 1000) / 10 = 100

        // Test subsequent update with EWMA
        stats.update_memory_usage(2000);
        assert_eq!(stats.memory_usage_bytes, 290); // (100*9 + 2000) / 10 = 290

        // Test another update
        stats.update_memory_usage(3000);
        assert_eq!(stats.memory_usage_bytes, 561); // (290*9 + 3000) / 10 = 561
    }

    #[test]
    fn test_feed_stats_add_latency_sample() {
        let mut stats = FeedStats::default();

        // Test single sample
        stats.add_latency_sample(100);
        assert_eq!(stats.avg_process_latency_ns, 10); // (0*9 + 100) / 10 = 10
        assert_eq!(stats.max_process_latency_ns, 100);

        // Not enough samples for p99 yet
        assert_eq!(stats.p99_process_latency_ns, 0);

        // Add more samples to trigger p99 calculation
        for i in 1..15 {
            stats.add_latency_sample(i * 100);
        }

        // Check that max latency is updated
        assert_eq!(stats.max_process_latency_ns, 1400);

        // Check that p99 latency is calculated (should be close to max with 15 samples)
        assert!(stats.p99_process_latency_ns > 0);
        assert!(stats.p99_process_latency_ns <= stats.max_process_latency_ns);
    }

    #[test]
    fn test_feed_stats_p99_calculation() {
        let mut stats = FeedStats::default();

        // Add 100 samples with values 1 to 100
        for i in 1..=100 {
            stats.add_latency_sample(i);
        }

        // P99 should be very close to 99
        assert!(stats.p99_process_latency_ns >= 98);
        assert!(stats.p99_process_latency_ns <= 100);

        // Max should be 100
        assert_eq!(stats.max_process_latency_ns, 100);
    }

    #[test]
    fn test_feeder_options_default() {
        let options = FeederOptions::default();
        assert_eq!(options.channel_buffer_size, 1024);
        assert_eq!(options.max_depth_levels, 100);
        assert_eq!(options.batch_size, 32);
        assert_eq!(options.cpu_affinity, -1);
        assert!(options.use_zero_copy);
        assert_eq!(options.max_backlog, 10000);
    }
}