foxhunt/crates/ml/src/microstructure/vpin_implementation.rs

//! # Complete VPIN Calculator Implementation
//!
//! Volume-Synchronized Probability of Informed Trading implementation
//! for detecting order flow toxicity and informed trading activity.
//!
//! ## Performance Targets
//! - Target latency: <25μs per calculation
//! - Memory: Ring buffer with zero allocations in hot path
//! - Precision: Integer arithmetic with 10,000x scaling

use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Instant;

use serde::{Deserialize, Serialize};

use crate::MLError;

/// Precision factor for integer arithmetic (10,000x scaling)
const VPIN_PRECISION_FACTOR: i64 = 10_000;

/// Maximum latency target in microseconds
const MAX_CALCULATION_LATENCY_US: u64 = 25;

/// Trade direction classification
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum TradeDirection {
    /// Buyer-initiated trade (aggressive buy)
    Buy,
    /// Seller-initiated trade (aggressive sell)
    Sell,
    /// Unknown direction (at mid-price or insufficient data)
    Unknown,
}

impl TradeDirection {
    /// Classify trade using Lee-Ready algorithm
    pub fn classify_lee_ready(trade_price: i64, bid: i64, ask: i64, prev_price: i64) -> Self {
        let mid_price = (bid + ask) / 2;

        if trade_price > mid_price {
            TradeDirection::Buy
        } else if trade_price < mid_price {
            TradeDirection::Sell
        } else {
            // At midpoint - use tick rule
            Self::classify_tick_rule(trade_price, prev_price)
        }
    }

    /// Classify trade using tick rule
    pub fn classify_tick_rule(trade_price: i64, prev_price: i64) -> Self {
        if trade_price > prev_price {
            TradeDirection::Buy
        } else if trade_price < prev_price {
            TradeDirection::Sell
        } else {
            TradeDirection::Unknown
        }
    }
}

/// Market data update for VPIN calculation
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MarketDataUpdate {
    /// Timestamp in microseconds
    pub timestamp: u64,
    /// Symbol identifier
    pub symbol: String,
    /// Trade price (scaled by PRECISION_FACTOR)
    pub price: i64,
    /// Trade volume
    pub volume: u64,
    /// Best bid price (scaled)
    pub bid: i64,
    /// Best ask price (scaled)
    pub ask: i64,
    /// Bid size
    pub bid_size: u64,
    /// Ask size
    pub ask_size: u64,
    /// Trade direction (if known)
    pub direction: Option<TradeDirection>,
}

/// VPIN calculation configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VPINConfig {
    /// Volume per bucket
    pub bucket_volume: u64,
    /// Number of buckets for rolling calculation
    pub bucket_count: usize,
    /// Toxicity threshold (scaled by PRECISION_FACTOR)
    pub toxicity_threshold: i64,
    /// Maximum age for data points (microseconds)
    pub max_age_us: u64,
}

impl Default for VPINConfig {
    fn default() -> Self {
        Self {
            bucket_volume: 10_000,
            bucket_count: 50,
            toxicity_threshold: 3_000, // 0.3 scaled
            max_age_us: 300_000_000,   // 5 minutes
        }
    }
}

/// VPIN calculation metrics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VPINMetrics {
    /// Current VPIN value (0.0 to 1.0)
    pub vpin: f64,
    /// Order flow imbalance (-1.0 to 1.0)
    pub order_flow_imbalance: f64,
    /// Toxicity score (0.0 to 1.0)
    pub toxicity_score: f64,
    /// Whether market is currently toxic
    pub is_toxic: bool,
    /// Number of completed buckets
    pub bucket_count: usize,
    /// Current bucket fill percentage (0.0 to 1.0)
    pub current_bucket_fill: f64,
}

impl Default for VPINMetrics {
    fn default() -> Self {
        Self {
            vpin: 0.0,
            order_flow_imbalance: 0.0,
            toxicity_score: 0.0,
            is_toxic: false,
            bucket_count: 0,
            current_bucket_fill: 0.0,
        }
    }
}

/// Performance metrics for VPIN calculator
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VPINPerformanceMetrics {
    /// Total number of calculations performed
    pub total_calculations: u64,
    /// Average latency in microseconds
    pub avg_latency_us: f64,
    /// Maximum latency in microseconds
    pub max_latency_us: u64,
    /// Number of calculations exceeding latency target
    pub over_latency_count: u64,
}

impl Default for VPINPerformanceMetrics {
    fn default() -> Self {
        Self {
            total_calculations: 0,
            avg_latency_us: 0.0,
            max_latency_us: 0,
            over_latency_count: 0,
        }
    }
}

/// Volume bucket for VPIN calculation
#[derive(Debug, Clone)]
struct VolumeBucket {
    /// Bucket index
    index: usize,
    /// Buy volume (scaled)
    buy_volume: u64,
    /// Sell volume (scaled)
    sell_volume: u64,
    /// Total volume
    total_volume: u64,
    /// First trade timestamp
    start_time: u64,
    /// Last trade timestamp
    end_time: u64,
}

impl VolumeBucket {
    fn new(index: usize, timestamp: u64) -> Self {
        Self {
            index,
            buy_volume: 0,
            sell_volume: 0,
            total_volume: 0,
            start_time: timestamp,
            end_time: timestamp,
        }
    }

    fn add_trade(&mut self, volume: u64, direction: TradeDirection, timestamp: u64) {
        match direction {
            TradeDirection::Buy => self.buy_volume += volume,
            TradeDirection::Sell => self.sell_volume += volume,
            TradeDirection::Unknown => {
                // Split unknown trades equally
                self.buy_volume += volume / 2;
                self.sell_volume += volume / 2;
            },
        }
        self.total_volume += volume;
        self.end_time = timestamp;
    }

    fn is_full(&self, target_volume: u64) -> bool {
        self.total_volume >= target_volume
    }

    fn calculate_imbalance(&self) -> i64 {
        if self.total_volume == 0 {
            return 0;
        }

        let imbalance = (self.buy_volume as i64 - self.sell_volume as i64) * VPIN_PRECISION_FACTOR;
        imbalance / self.total_volume as i64
    }
}

/// Ring buffer for efficient bucket storage
#[derive(Debug, Clone)]
pub struct RingBuffer<T> {
    data: Vec<T>,
    head: usize,
    len: usize,
    capacity: usize,
}

impl<T: Clone> RingBuffer<T> {
    pub fn new(capacity: usize) -> Self {
        Self {
            data: Vec::with_capacity(capacity),
            head: 0,
            len: 0,
            capacity,
        }
    }

    pub fn push(&mut self, item: T) {
        if self.len < self.capacity {
            self.data.push(item);
            self.len += 1;
        } else {
            self.data[self.head] = item;
            self.head = (self.head + 1) % self.capacity;
        }
    }

    pub fn len(&self) -> usize {
        self.len
    }

    pub fn is_empty(&self) -> bool {
        self.len == 0
    }

    pub fn get(&self, index: usize) -> Option<&T> {
        if index >= self.len {
            return None;
        }

        if self.len < self.capacity {
            self.data.get(index)
        } else {
            let real_index = (self.head + index) % self.capacity;
            self.data.get(real_index)
        }
    }

    fn iter(&self) -> RingBufferIterator<'_, T> {
        RingBufferIterator {
            buffer: self,
            index: 0,
        }
    }
}

struct RingBufferIterator<'a, T> {
    buffer: &'a RingBuffer<T>,
    index: usize,
}

impl<'a, T: Clone> Iterator for RingBufferIterator<'a, T> {
    type Item = &'a T;

    fn next(&mut self) -> Option<Self::Item> {
        if self.index >= self.buffer.len() {
            None
        } else {
            let item = self.buffer.get(self.index);
            self.index += 1;
            item
        }
    }
}

/// Main VPIN calculator with high-performance implementation
#[derive(Debug)]
pub struct VPINCalculator {
    /// Configuration
    config: VPINConfig,
    /// Completed volume buckets (ring buffer)
    buckets: RingBuffer<VolumeBucket>,
    /// Current active bucket
    current_bucket: Option<VolumeBucket>,
    /// Last trade price for tick rule
    last_price: i64,
    /// Performance metrics
    performance: VPINPerformanceMetrics,
    /// Calculation counter
    calculation_count: AtomicU64,
    /// Latency accumulator
    latency_accumulator: AtomicU64,
}

impl VPINCalculator {
    /// Create new VPIN calculator with configuration
    pub fn new(config: VPINConfig) -> Self {
        Self {
            buckets: RingBuffer::new(config.bucket_count),
            current_bucket: None,
            last_price: 0,
            performance: VPINPerformanceMetrics::default(),
            calculation_count: AtomicU64::new(0),
            latency_accumulator: AtomicU64::new(0),
            config,
        }
    }

    /// Update VPIN with new market data
    pub fn update(&mut self, update: &MarketDataUpdate) -> Result<(), MLError> {
        let start_time = Instant::now();

        // Classify trade direction if not provided
        let direction = update
            .direction
            .unwrap_or_else(|| self.classify_trade(update));

        // Create new bucket if needed
        if self.current_bucket.is_none() {
            self.current_bucket = Some(VolumeBucket::new(self.buckets.len(), update.timestamp));
        }

        // Add trade to current bucket
        if let Some(ref mut bucket) = self.current_bucket {
            bucket.add_trade(update.volume, direction, update.timestamp);

            // Check if bucket is full
            if bucket.is_full(self.config.bucket_volume) {
                // Calculate remaining volume before moving bucket
                let remaining_volume = bucket.total_volume - self.config.bucket_volume;

                // Move to completed buckets
                let completed_bucket = self.current_bucket.take()
                    .ok_or_else(|| MLError::ModelError("current_bucket unexpectedly None".to_owned()))?;
                self.buckets.push(completed_bucket);

                // Start new bucket with remaining volume
                if remaining_volume > 0 {
                    let mut new_bucket = VolumeBucket::new(self.buckets.len(), update.timestamp);
                    new_bucket.add_trade(remaining_volume, direction, update.timestamp);
                    self.current_bucket = Some(new_bucket);
                }
            }
        }

        // Update last price for tick rule
        self.last_price = update.price;

        // Record performance metrics
        let latency_us = start_time.elapsed().as_micros() as u64;
        self.update_performance_metrics(latency_us);

        Ok(())
    }

    /// Classify trade direction using available information
    pub fn classify_trade(&self, update: &MarketDataUpdate) -> TradeDirection {
        // Use Lee-Ready algorithm if we have bid/ask
        if update.bid > 0 && update.ask > 0 {
            TradeDirection::classify_lee_ready(
                update.price,
                update.bid,
                update.ask,
                self.last_price,
            )
        } else if self.last_price > 0 {
            // Fall back to tick rule
            TradeDirection::classify_tick_rule(update.price, self.last_price)
        } else {
            TradeDirection::Unknown
        }
    }

    /// Get current VPIN value
    pub fn get_vpin(&self) -> f64 {
        if self.buckets.len() == 0 {
            return 0.0;
        }

        let total_imbalance: i64 = self
            .buckets
            .iter()
            .map(|bucket| bucket.calculate_imbalance().abs())
            .sum();

        let avg_imbalance = total_imbalance / (self.buckets.len() as i64);
        (avg_imbalance as f64) / (VPIN_PRECISION_FACTOR as f64)
    }

    /// Get number of completed buckets
    pub fn get_bucket_count(&self) -> usize {
        self.buckets.len()
    }

    /// Check if market is currently toxic
    pub fn is_toxic(&self) -> bool {
        let vpin_scaled = (self.get_vpin() * VPIN_PRECISION_FACTOR as f64) as i64;
        vpin_scaled > self.config.toxicity_threshold
    }

    /// Get comprehensive VPIN result
    pub fn get_result(&self) -> VPINMetrics {
        let vpin = self.get_vpin();
        let order_flow_imbalance = self.calculate_order_flow_imbalance();
        let toxicity_score = vpin; // Simple mapping for now
        let current_bucket_fill = self.get_current_bucket_fill();

        VPINMetrics {
            vpin,
            order_flow_imbalance,
            toxicity_score,
            is_toxic: self.is_toxic(),
            bucket_count: self.buckets.len(),
            current_bucket_fill,
        }
    }

    /// Get performance metrics
    pub fn get_metrics(&self) -> VPINPerformanceMetrics {
        let total_calcs = self.calculation_count.load(Ordering::Relaxed);
        let total_latency = self.latency_accumulator.load(Ordering::Relaxed);

        VPINPerformanceMetrics {
            total_calculations: total_calcs,
            avg_latency_us: if total_calcs > 0 {
                total_latency as f64 / total_calcs as f64
            } else {
                0.0
            },
            max_latency_us: self.performance.max_latency_us,
            over_latency_count: self.performance.over_latency_count,
        }
    }

    /// Get current bucket fill percentage
    pub fn get_current_bucket_fill(&self) -> f64 {
        if let Some(ref bucket) = self.current_bucket {
            bucket.total_volume as f64 / self.config.bucket_volume as f64
        } else {
            0.0
        }
    }

    /// Calculate order flow imbalance
    fn calculate_order_flow_imbalance(&self) -> f64 {
        if self.buckets.len() == 0 {
            return 0.0;
        }

        let total_buy: u64 = self.buckets.iter().map(|b| b.buy_volume).sum();
        let total_sell: u64 = self.buckets.iter().map(|b| b.sell_volume).sum();
        let total_volume = total_buy + total_sell;

        if total_volume == 0 {
            0.0
        } else {
            (total_buy as f64 - total_sell as f64) / total_volume as f64
        }
    }

    /// Update performance metrics
    fn update_performance_metrics(&mut self, latency_us: u64) {
        self.calculation_count.fetch_add(1, Ordering::Relaxed);
        self.latency_accumulator
            .fetch_add(latency_us, Ordering::Relaxed);

        if latency_us > self.performance.max_latency_us {
            self.performance.max_latency_us = latency_us;
        }

        if latency_us > MAX_CALCULATION_LATENCY_US {
            self.performance.over_latency_count += 1;
        }
    }
}

impl Default for VPINCalculator {
    fn default() -> Self {
        Self::new(VPINConfig::default())
    }
}

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

    #[test]
    fn test_trade_direction_classification() {
        // Test Lee-Ready algorithm
        let direction = TradeDirection::classify_lee_ready(
            105000, // trade price (10.50)
            104000, // bid (10.40)
            106000, // ask (10.60)
            104500, // prev price (10.45)
        );
        assert_eq!(direction, TradeDirection::Buy);

        // Test tick rule
        let direction = TradeDirection::classify_tick_rule(105000, 104000);
        assert_eq!(direction, TradeDirection::Buy);
    }

    #[test]
    fn test_volume_bucket() {
        let mut bucket = VolumeBucket::new(0, 1000000);
        bucket.add_trade(500, TradeDirection::Buy, 1000001);
        bucket.add_trade(300, TradeDirection::Sell, 1000002);

        assert_eq!(bucket.total_volume, 800);
        assert!(!bucket.is_full(1000));

        let imbalance = bucket.calculate_imbalance();
        // (500 - 300) * 10000 / 800 = 2500
        assert_eq!(imbalance, 2500);
    }

    #[test]
    fn test_ring_buffer() {
        let mut buffer = RingBuffer::new(3);

        buffer.push(1);
        buffer.push(2);
        buffer.push(3);

        assert_eq!(buffer.len(), 3);
        assert_eq!(buffer.get(0), Some(&1));
        assert_eq!(buffer.get(1), Some(&2));
        assert_eq!(buffer.get(2), Some(&3));

        buffer.push(4);
        assert_eq!(buffer.len(), 3);
        assert_eq!(buffer.get(0), Some(&2));
        assert_eq!(buffer.get(1), Some(&3));
        assert_eq!(buffer.get(2), Some(&4));
    }
}