rusty_backtest/features/

order_flow.rs

//! Order Flow Analysis - High-performance implementation of OFI and VPIN
//!
//! Based on patterns from Binance Tardis reference implementation with
//! Rust-specific optimizations for HFT performance.

use super::{OrderBookSnapshot, TradeSide, TradeTick, decimal_to_f64_or_nan};
use rust_decimal::prelude::ToPrimitive;

/// Calculate basic Order Flow Imbalance (OFI)
///
/// OFI measures the net order flow at the best bid/ask level between two snapshots.
/// Positive values indicate buying pressure, negative values indicate selling pressure.
#[inline(always)]
pub fn calculate_ofi(prev_book: &OrderBookSnapshot, curr_book: &OrderBookSnapshot) -> f64 {
    let mut ofi = 0.0;

    // Best bid/ask analysis
    if let (Some(prev_bid), Some(curr_bid)) = (prev_book.bids.first(), curr_book.bids.first())
        && prev_bid.price == curr_bid.price
    {
        ofi += (curr_bid.quantity - prev_bid.quantity)
            .to_f64()
            .unwrap_or(0.0);
    }

    if let (Some(prev_ask), Some(curr_ask)) = (prev_book.asks.first(), curr_book.asks.first())
        && prev_ask.price == curr_ask.price
    {
        ofi -= (curr_ask.quantity - prev_ask.quantity)
            .to_f64()
            .unwrap_or(0.0);
    }

    ofi
}

/// Calculate multi-level weighted OFI
///
/// Extends basic OFI to multiple levels with distance-based weighting.
/// Levels closer to mid-price receive higher weights.
#[inline(always)]
pub fn calculate_weighted_ofi(
    prev_book: &OrderBookSnapshot,
    curr_book: &OrderBookSnapshot,
    max_levels: usize,
) -> f64 {
    let mut weighted_ofi = 0.0;
    let levels = max_levels
        .min(prev_book.bids.len())
        .min(curr_book.bids.len());

    for i in 0..levels {
        let weight = 1.0 / (i as f64 + 1.0); // Distance-based weighting

        // Bid side
        if let (Some(prev_bid), Some(curr_bid)) = (prev_book.bids.get(i), curr_book.bids.get(i))
            && prev_bid.price == curr_bid.price
        {
            weighted_ofi += weight
                * (curr_bid.quantity - prev_bid.quantity)
                    .to_f64()
                    .unwrap_or(0.0);
        }

        // Ask side
        if let (Some(prev_ask), Some(curr_ask)) = (prev_book.asks.get(i), curr_book.asks.get(i))
            && prev_ask.price == curr_ask.price
        {
            weighted_ofi -= weight
                * (curr_ask.quantity - prev_ask.quantity)
                    .to_f64()
                    .unwrap_or(0.0);
        }
    }

    weighted_ofi
}

/// Calculate Volume-Synchronized Probability of Informed Trading (VPIN)
///
/// VPIN measures the imbalance of trade volume between buy and sell orders
/// over fixed volume buckets, providing early warning of informed trading.
pub fn calculate_vpin(trades: &[TradeTick], bucket_size: usize) -> f64 {
    if trades.len() < bucket_size {
        return 0.0;
    }

    let recent_trades = &trades[trades.len() - bucket_size..];

    let mut buy_volume = 0.0;
    let mut sell_volume = 0.0;

    for trade in recent_trades {
        match trade.side {
            TradeSide::Buy => buy_volume += decimal_to_f64_or_nan(trade.quantity),
            TradeSide::Sell => sell_volume += decimal_to_f64_or_nan(trade.quantity),
        }
    }

    let total_volume = buy_volume + sell_volume;
    if total_volume > 0.0 {
        (buy_volume - sell_volume).abs() / total_volume
    } else {
        0.0
    }
}

/// High-performance VPIN calculation with rolling window
///
/// Optimized version using circular buffer for continuous VPIN computation
/// without reallocating memory.
pub struct VPINCalculator {
    bucket_size: usize,
    buy_volumes: Vec<f64>,
    sell_volumes: Vec<f64>,
    current_index: usize,
    is_full: bool,
}

impl VPINCalculator {
    /// Create new VPIN calculator with specified bucket size
    #[must_use]
    pub fn new(bucket_size: usize) -> Self {
        Self {
            bucket_size,
            buy_volumes: vec![0.0; bucket_size],
            sell_volumes: vec![0.0; bucket_size],
            current_index: 0,
            is_full: false,
        }
    }

    /// Add trade and get updated VPIN value
    #[inline(always)]
    pub fn add_trade(&mut self, trade: &TradeTick) -> Option<f64> {
        // Reset current bucket
        self.buy_volumes[self.current_index] = 0.0;
        self.sell_volumes[self.current_index] = 0.0;

        // Add trade to current bucket
        match trade.side {
            TradeSide::Buy => {
                self.buy_volumes[self.current_index] += decimal_to_f64_or_nan(trade.quantity)
            }
            TradeSide::Sell => {
                self.sell_volumes[self.current_index] += decimal_to_f64_or_nan(trade.quantity)
            }
        }

        // Move to next bucket
        self.current_index = (self.current_index + 1) % self.bucket_size;
        if self.current_index == 0 {
            self.is_full = true;
        }

        // Calculate VPIN if we have enough data
        if self.is_full {
            Some(self.calculate_current_vpin())
        } else {
            None
        }
    }
    /// Alias for calculate_current_vpin() to maintain backward compatibility
    pub fn calculate(&self) -> f64 {
        self.calculate_current_vpin()
    }

    /// Calculate VPIN for current window
    #[inline(always)]
    fn calculate_current_vpin(&self) -> f64 {
        let total_buy: f64 = self.buy_volumes.iter().sum();
        let total_sell: f64 = self.sell_volumes.iter().sum();
        let total_volume = total_buy + total_sell;

        if total_volume > 0.0 {
            (total_buy - total_sell).abs() / total_volume
        } else {
            0.0
        }
    }
}

/// Order Flow Analyzer for comprehensive order flow metrics
pub struct OrderFlowAnalyzer {
    vpin_calculator: VPINCalculator,
    ofi_window: Vec<f64>,
    window_size: usize,
}

impl OrderFlowAnalyzer {
    /// Create new order flow analyzer
    #[must_use]
    pub fn new(window_size: usize, vpin_bucket_size: usize) -> Self {
        Self {
            vpin_calculator: VPINCalculator::new(vpin_bucket_size),
            ofi_window: Vec::with_capacity(window_size * 2),
            window_size,
        }
    }

    /// Process trade tick and update metrics
    pub fn process_trade(&mut self, trade: &TradeTick) -> Option<f64> {
        self.vpin_calculator.add_trade(trade)
    }

    /// Process order book update and calculate OFI
    pub fn process_orderbook_update(
        &mut self,
        prev_book: &OrderBookSnapshot,
        curr_book: &OrderBookSnapshot,
    ) -> f64 {
        let ofi = calculate_ofi(prev_book, curr_book);

        // Add to rolling window
        self.ofi_window.push(ofi);
        if self.ofi_window.len() > self.window_size * 2 {
            let drain_count = self.window_size;
            self.ofi_window.drain(0..drain_count);
        }

        ofi
    }

    /// Get average OFI over window
    pub fn get_average_ofi(&self) -> f64 {
        if self.ofi_window.is_empty() {
            return 0.0;
        }

        let recent_window =
            &self.ofi_window[self.ofi_window.len().saturating_sub(self.window_size)..];
        recent_window.iter().sum::<f64>() / recent_window.len() as f64
    }

    /// Get OFI standard deviation
    pub fn get_ofi_std(&self) -> f64 {
        if self.ofi_window.len() < 2 {
            return 0.0;
        }

        let recent_window =
            &self.ofi_window[self.ofi_window.len().saturating_sub(self.window_size)..];
        let mean = self.get_average_ofi();

        let variance = recent_window
            .iter()
            .map(|&x| (x - mean).powi(2))
            .sum::<f64>()
            / recent_window.len() as f64;

        variance.sqrt()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::features::Level;
    use rust_decimal::{Decimal, prelude::FromPrimitive};
    use rust_decimal_macros::dec;

    fn create_test_book(
        bid_price: Decimal,
        bid_qty: f64,
        ask_price: Decimal,
        ask_qty: f64,
    ) -> OrderBookSnapshot {
        OrderBookSnapshot {
            timestamp_ns: 1000000000,
            symbol: "BTC-USD".into(),
            bids: smallvec::smallvec![Level {
                price: bid_price,
                quantity: Decimal::from_f64(bid_qty).unwrap_or(Decimal::ZERO),
                order_count: 1
            }],
            asks: smallvec::smallvec![Level {
                price: ask_price,
                quantity: Decimal::from_f64(ask_qty).unwrap_or(Decimal::ZERO),
                order_count: 1
            }],
        }
    }

    #[test]
    fn test_basic_ofi_positive() {
        let prev_book = create_test_book(dec!(49999), 10.0, dec!(50001), 10.0);
        let curr_book = create_test_book(dec!(49999), 15.0, dec!(50001), 10.0); // Bid quantity increased

        let ofi = calculate_ofi(&prev_book, &curr_book);
        assert_eq!(ofi, 5.0); // Positive OFI indicates buying pressure
    }

    #[test]
    fn test_basic_ofi_negative() {
        let prev_book = create_test_book(dec!(49999), 10.0, dec!(50001), 10.0);
        let curr_book = create_test_book(dec!(49999), 10.0, dec!(50001), 15.0); // Ask quantity increased

        let ofi = calculate_ofi(&prev_book, &curr_book);
        assert_eq!(ofi, -5.0); // Negative OFI indicates selling pressure
    }

    #[test]
    fn test_vpin_balanced() {
        let trades = vec![
            TradeTick {
                timestamp_ns: 1000000000,
                symbol: "BTC-USD".into(),
                side: TradeSide::Buy,
                price: dec!(50000),
                quantity: dec!(1.0),
            },
            TradeTick {
                timestamp_ns: 1000000001,
                symbol: "BTC-USD".into(),
                side: TradeSide::Sell,
                price: dec!(50000),
                quantity: dec!(1.0),
            },
        ];

        let vpin = calculate_vpin(&trades, 2);
        assert_eq!(vpin, 0.0); // Balanced trades should give VPIN of 0
    }

    #[test]
    fn test_vpin_imbalanced() {
        let trades = vec![
            TradeTick {
                timestamp_ns: 1000000000,
                symbol: "BTC-USD".into(),
                side: TradeSide::Buy,
                price: dec!(50000),
                quantity: dec!(3.0),
            },
            TradeTick {
                timestamp_ns: 1000000001,
                symbol: "BTC-USD".into(),
                side: TradeSide::Sell,
                price: dec!(50000),
                quantity: dec!(1.0),
            },
        ];

        let vpin = calculate_vpin(&trades, 2);
        assert_eq!(vpin, 0.5); // 2.0 net buy volume out of 4.0 total = 0.5
    }
}