hypercall/read_cache/

market_stats.rs

//! Market statistics cache for 24h volume and open interest.

use anyhow::Result;
use hypercall_types::{to_human_readable_decimal, utils::is_option_symbol};
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info, warn};

use crate::messaging::EventBusTrait;
use crate::portfolio::PortfolioService;
use crate::shared::topics::TOPIC_FILLS;
use hypercall_types::EngineMessage;

/// Number of hourly buckets for 24h rolling volume
const HOURLY_BUCKETS: usize = 24;
const UNIX_MS_THRESHOLD: u64 = 10_000_000_000;
const DAY_MS: u128 = 86_400_000;

fn fill_notional_volume(
    symbol: &str,
    size: Decimal,
    price: Decimal,
    underlying_notional: Option<Decimal>,
) -> Option<Decimal> {
    if is_option_symbol(symbol) {
        return underlying_notional;
    }

    Some(to_human_readable_decimal(symbol, size).abs() * price)
}

fn market_stats_cutoff_timestamp() -> i64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_millis().saturating_sub(DAY_MS) as i64)
        .unwrap_or(0)
}

fn timestamp_to_seconds(timestamp: u64) -> u64 {
    if timestamp >= UNIX_MS_THRESHOLD {
        timestamp / 1000
    } else {
        timestamp
    }
}

/// Statistics for a single instrument
#[derive(Debug, Clone)]
struct InstrumentStats {
    /// Hourly volume buckets for rolling 24h calculation
    /// Index 0 is the current hour, index 1 is 1 hour ago, etc.
    volume_buckets: [Decimal; HOURLY_BUCKETS],
    /// The Unix timestamp (truncated to hour) of the current bucket
    current_bucket_hour: u64,
}

impl InstrumentStats {
    fn new() -> Self {
        Self {
            volume_buckets: [dec!(0); HOURLY_BUCKETS],
            current_bucket_hour: 0,
        }
    }

    /// Get the current hour from a Unix timestamp
    fn timestamp_to_hour(timestamp: u64) -> u64 {
        timestamp_to_seconds(timestamp) / 3600
    }

    /// Add volume to the appropriate bucket, rolling over if needed
    fn add_volume(&mut self, timestamp: u64, notional_volume: Decimal) {
        let current_hour = Self::timestamp_to_hour(timestamp);

        if self.current_bucket_hour == 0 {
            // First fill - initialize
            self.current_bucket_hour = current_hour;
            self.volume_buckets[0] = notional_volume;
            return;
        }

        let hours_elapsed = current_hour.saturating_sub(self.current_bucket_hour);

        if hours_elapsed == 0 {
            // Same hour - add to current bucket
            self.volume_buckets[0] += notional_volume;
        } else if hours_elapsed < HOURLY_BUCKETS as u64 {
            // Roll buckets forward
            // Shift existing buckets to the right by hours_elapsed positions
            let shift = hours_elapsed as usize;
            for i in (shift..HOURLY_BUCKETS).rev() {
                self.volume_buckets[i] = self.volume_buckets[i - shift];
            }
            // Zero out the newly exposed buckets
            for i in 0..shift {
                self.volume_buckets[i] = dec!(0);
            }
            // Add new volume to current bucket
            self.volume_buckets[0] = notional_volume;
            self.current_bucket_hour = current_hour;
        } else {
            // More than 24 hours elapsed - reset all buckets
            self.volume_buckets = [dec!(0); HOURLY_BUCKETS];
            self.volume_buckets[0] = notional_volume;
            self.current_bucket_hour = current_hour;
        }
    }

    /// Get the total 24h volume
    fn get_volume_24h(&self) -> Decimal {
        self.volume_buckets.iter().sum()
    }

    /// Roll buckets to current time (for accurate reads even without fills)
    fn roll_to_current(&mut self, current_timestamp: u64) {
        let current_hour = Self::timestamp_to_hour(current_timestamp);

        if self.current_bucket_hour == 0 {
            return; // No data yet
        }

        let hours_elapsed = current_hour.saturating_sub(self.current_bucket_hour);

        if hours_elapsed > 0 && hours_elapsed < HOURLY_BUCKETS as u64 {
            // Roll buckets forward
            let shift = hours_elapsed as usize;
            for i in (shift..HOURLY_BUCKETS).rev() {
                self.volume_buckets[i] = self.volume_buckets[i - shift];
            }
            // Zero out the newly exposed buckets
            for i in 0..shift {
                self.volume_buckets[i] = dec!(0);
            }
            self.current_bucket_hour = current_hour;
        } else if hours_elapsed >= HOURLY_BUCKETS as u64 {
            // More than 24 hours elapsed - reset all buckets
            self.volume_buckets = [dec!(0); HOURLY_BUCKETS];
            self.current_bucket_hour = current_hour;
        }
    }
}

/// Cache for market statistics (volume_24h and open_interest)
pub struct MarketStatsCache {
    /// Per-symbol volume statistics
    stats: Arc<RwLock<HashMap<String, InstrumentStats>>>,
    /// Reference to portfolio service for open interest calculation
    portfolio_service: Arc<dyn PortfolioService + Send + Sync>,
}

impl MarketStatsCache {
    /// Create a new MarketStatsCache.
    ///
    /// # Arguments
    /// * `portfolio_service` - The portfolio service to derive open interest from
    pub fn new(portfolio_service: Arc<dyn PortfolioService + Send + Sync>) -> Self {
        Self {
            stats: Arc::new(RwLock::new(HashMap::new())),
            portfolio_service,
        }
    }

    /// Initialize the cache and start listening to events (legacy, no shutdown support).
    pub async fn initialize(self: Arc<Self>, event_bus: Arc<dyn EventBusTrait>) -> Result<()> {
        // Create a dummy shutdown channel that never fires.
        let (tx, rx) = tokio::sync::broadcast::channel::<()>(1);
        std::mem::forget(tx);
        self.initialize_with_shutdown(event_bus, rx).await
    }

    /// Initialize the cache with shutdown support.
    ///
    /// This method starts the event listener for fills. For better startup performance,
    /// use `initialize_with_backfill` which loads last 24h from DB first.
    pub async fn initialize_with_shutdown(
        self: Arc<Self>,
        event_bus: Arc<dyn EventBusTrait>,
        shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<()> {
        info!("Initializing MarketStatsCache (no DB backfill)");

        // Start event listener for fills
        self.start_event_listeners_with_shutdown(event_bus, shutdown_rx)
            .await?;

        info!("MarketStatsCache initialized");
        Ok(())
    }

    /// Initialize with DB backfill for the last 24 hours of fills.
    ///
    /// This is the recommended initialization method (Phase 3c):
    /// 1. Load last 24h of fills from database
    /// 2. Then subscribe to event bus for real-time updates
    ///
    /// This avoids replaying all historical fills.
    pub async fn initialize_with_backfill(
        self: Arc<Self>,
        diesel_db: &dyn hypercall_db::AnalyticsReader,
        event_bus: Arc<dyn EventBusTrait>,
        shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<()> {
        info!("Initializing MarketStatsCache with DB backfill");

        // 1. Load last 24h of fills from database
        let fills_loaded = self.backfill_from_db(diesel_db).await?;
        info!(
            "MarketStatsCache backfilled {} fills from last 24h",
            fills_loaded
        );

        // 2. Subscribe to event bus for real-time updates
        // TODO: There is a small data gap between the DB backfill query completing and
        // the event subscription starting. Fills in that window are missed.
        // For 24h volume stats this is negligible (ms-level gap). A proper fix would
        // subscribe first (buffering events), backfill, then replay the buffer.
        self.start_event_listeners_with_shutdown(event_bus, shutdown_rx)
            .await?;

        info!("MarketStatsCache initialized with DB backfill");
        Ok(())
    }

    async fn backfill_from_db(
        &self,
        diesel_db: &dyn hypercall_db::AnalyticsReader,
    ) -> Result<usize> {
        let cutoff_timestamp = market_stats_cutoff_timestamp();

        match diesel_db.get_fills_since_timestamp(cutoff_timestamp).await {
            Ok(fill_rows) => {
                let count = fill_rows.len();
                let mut skipped_missing_notional = 0;
                for row in fill_rows {
                    let Some(notional_volume) = fill_notional_volume(
                        &row.symbol,
                        row.size,
                        row.price,
                        row.underlying_notional,
                    ) else {
                        skipped_missing_notional += 1;
                        continue;
                    };

                    let mut stats = self.stats.write().await;
                    let instrument_stats =
                        stats.entry(row.symbol).or_insert_with(InstrumentStats::new);
                    instrument_stats.add_volume(row.timestamp as u64, notional_volume);
                }

                if skipped_missing_notional > 0 {
                    warn!(
                        "MarketStatsCache skipped {} historical option fills missing underlying_notional",
                        skipped_missing_notional
                    );
                }

                Ok(count)
            }
            Err(e) => {
                // Log warning but don't fail - DB might not have fills table in some environments
                warn!(
                    "MarketStatsCache DB backfill failed (will rely on event bus): {}",
                    e
                );
                Ok(0)
            }
        }
    }

    /// Start event listeners for fill events.
    async fn start_event_listeners_with_shutdown(
        self: Arc<Self>,
        event_bus: Arc<dyn EventBusTrait>,
        mut shutdown_rx: tokio::sync::broadcast::Receiver<()>,
    ) -> Result<()> {
        info!("Starting event listener for MarketStatsCache");

        let topics = vec![TOPIC_FILLS.to_string()];
        let mut receiver = event_bus
            .subscribe(topics)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to subscribe to fills topic: {}", e))?;

        let cache = self.clone();
        tokio::spawn(async move {
            info!("MarketStatsCache event listener started");
            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        info!("MarketStatsCache event listener received shutdown signal");
                        break;
                    }
                    maybe_message = receiver.recv() => {
                        match maybe_message {
                            Some(EngineMessage::OrderFilled { fill, .. }) => {
                                cache.handle_fill(&fill).await;
                            }
                            Some(_) => {
                                // Ignore other message types
                            }
                            None => {
                                // Channel closed
                                break;
                            }
                        }
                    }
                }
            }
            info!("MarketStatsCache event listener stopped");
        });

        Ok(())
    }

    /// Handle a fill event to update volume statistics.
    async fn handle_fill(&self, fill: &hypercall_types::Fill) {
        let symbol = &fill.symbol;

        let Some(notional_volume) =
            fill_notional_volume(symbol, fill.size, fill.price, fill.underlying_notional)
        else {
            warn!(
                "MarketStatsCache skipped option fill {} for {} missing underlying_notional",
                fill.trade_id, symbol
            );
            return;
        };

        debug!(
            "MarketStatsCache: Recording fill for {}: size={}, price={}, notional={}",
            symbol,
            to_human_readable_decimal(symbol, fill.size),
            fill.price,
            notional_volume
        );

        let mut stats = self.stats.write().await;
        let instrument_stats = stats
            .entry(symbol.clone())
            .or_insert_with(InstrumentStats::new);

        instrument_stats.add_volume(fill.timestamp, notional_volume);
    }

    /// Get the 24-hour volume for a symbol.
    pub async fn get_volume_24h(&self, symbol: &str) -> Decimal {
        let current_timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);

        let mut stats = self.stats.write().await;
        if let Some(instrument_stats) = stats.get_mut(symbol) {
            // Roll to current time before reading
            instrument_stats.roll_to_current(current_timestamp);
            instrument_stats.get_volume_24h()
        } else {
            dec!(0)
        }
    }

    /// Get the open interest for a symbol.
    ///
    /// Open interest is calculated as SUM(ABS(amount)) / 2 across all accounts.
    /// Division by 2 because each position is double-counted (taker + maker sides).
    pub async fn get_open_interest(&self, symbol: &str) -> Decimal {
        // Get all portfolios from the portfolio service
        let portfolios = self.portfolio_service.all_portfolios().await;

        let mut total_abs_amount = dec!(0);

        for portfolio in portfolios.values() {
            if let Some(position) = portfolio.positions.get(symbol) {
                total_abs_amount += position.amount.abs();
            }
        }

        // Divide by 2 because each trade creates two positions (taker + maker)
        total_abs_amount / dec!(2)
    }

    /// Get both volume_24h and open_interest for a symbol.
    pub async fn get_stats(&self, symbol: &str) -> (Decimal, Decimal) {
        let volume = self.get_volume_24h(symbol).await;
        let open_interest = self.get_open_interest(symbol).await;
        (volume, open_interest)
    }

    /// Get stats for all tracked symbols.
    pub async fn get_all_stats(&self) -> HashMap<String, (Decimal, Decimal)> {
        let current_timestamp = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0);

        let portfolios = self.portfolio_service.all_portfolios().await;
        let mut result = HashMap::new();

        // Roll and snapshot all volume buckets in one lock acquisition.
        let mut stats = self.stats.write().await;
        for (symbol, instrument_stats) in stats.iter_mut() {
            instrument_stats.roll_to_current(current_timestamp);
            result.insert(symbol.clone(), (instrument_stats.get_volume_24h(), dec!(0)));
        }
        drop(stats);

        // Aggregate open interest in a single pass through all positions.
        for portfolio in portfolios.values() {
            for (symbol, position) in &portfolio.positions {
                let entry = result.entry(symbol.clone()).or_insert((dec!(0), dec!(0)));
                entry.1 += position.amount.abs();
            }
        }

        for (_, (_, open_interest)) in result.iter_mut() {
            *open_interest /= dec!(2);
        }

        result
    }
}

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

    fn create_stats() -> InstrumentStats {
        InstrumentStats::new()
    }

    #[test]
    fn test_add_volume_first_fill() {
        let mut stats = create_stats();

        stats.add_volume(1000, dec!(100));

        assert_eq!(stats.get_volume_24h(), dec!(100));
        assert_eq!(stats.volume_buckets[0], dec!(100));
    }

    #[test]
    fn test_add_volume_same_hour() {
        let mut stats = create_stats();

        let base_timestamp = 3600; // 1 hour in seconds
        stats.add_volume(base_timestamp, dec!(100));
        stats.add_volume(base_timestamp + 1000, dec!(50)); // Same hour

        assert_eq!(stats.get_volume_24h(), dec!(150));
        assert_eq!(stats.volume_buckets[0], dec!(150));
    }

    #[test]
    fn test_add_volume_rollover() {
        let mut stats = create_stats();

        let hour_0 = 3600;
        let hour_1 = hour_0 + 3600;
        let hour_2 = hour_0 + 7200;

        stats.add_volume(hour_0, dec!(100));
        stats.add_volume(hour_1, dec!(200));
        stats.add_volume(hour_2, dec!(300));

        assert_eq!(stats.get_volume_24h(), dec!(600)); // 100 + 200 + 300
        assert_eq!(stats.volume_buckets[0], dec!(300));
        assert_eq!(stats.volume_buckets[1], dec!(200));
        assert_eq!(stats.volume_buckets[2], dec!(100));
    }

    #[test]
    fn test_volume_expires_after_24h() {
        let mut stats = create_stats();

        let hour_0 = 3600;
        stats.add_volume(hour_0, dec!(100));

        // After 24 hours + 1 second
        let after_24h = hour_0 + (24 * 3600) + 1;
        stats.add_volume(after_24h, dec!(50));

        // Old volume should be gone
        assert_eq!(stats.get_volume_24h(), dec!(50));
    }

    #[test]
    fn test_roll_to_current() {
        let mut stats = create_stats();

        let hour_0 = 3600;
        stats.add_volume(hour_0, dec!(100));
        stats.add_volume(hour_0 + 3600, dec!(200)); // Hour 1

        // Roll to hour 3
        let hour_3 = hour_0 + 3 * 3600;
        stats.roll_to_current(hour_3);

        // Data should be shifted
        assert_eq!(stats.volume_buckets[0], dec!(0)); // Current hour (no fills)
        assert_eq!(stats.volume_buckets[1], dec!(0)); // 1 hour ago (no fills)
        assert_eq!(stats.volume_buckets[2], dec!(200)); // 2 hours ago
        assert_eq!(stats.volume_buckets[3], dec!(100)); // 3 hours ago
        assert_eq!(stats.get_volume_24h(), dec!(300));
    }

    #[test]
    fn test_roll_past_24h() {
        let mut stats = create_stats();

        let hour_0 = 3600;
        stats.add_volume(hour_0, dec!(100));

        // Roll past 24 hours
        let way_later = hour_0 + 25 * 3600;
        stats.roll_to_current(way_later);

        // All data should be gone
        assert_eq!(stats.get_volume_24h(), dec!(0));
    }

    #[tokio::test]
    async fn test_cache_get_volume() {
        let portfolio_service = Arc::new(PortfolioServiceImpl::new());
        let cache = Arc::new(MarketStatsCache::new(portfolio_service));

        // Initially should be zero
        let vol = cache.get_volume_24h("BTC-20260130-100000-C").await;
        assert_eq!(vol, dec!(0));
    }

    #[tokio::test]
    async fn test_cache_get_open_interest() {
        use hypercall_types::wallet_address::test_wallet;

        let portfolio_service = Arc::new(PortfolioServiceImpl::new());

        // Apply fills to create positions. Cash is supplied separately by balance providers.
        use hypercall_types::Fill;
        use hypercall_types::Side;

        let fill = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-20260130-100000-C".to_string(),
            price: dec!(1000),
            size: dec!(10_000_000), // 10 contracts in contract units
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(2),
            fee: dec!(0),
            is_taker: true,
            timestamp: 0,
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        portfolio_service
            .apply_event(&EngineMessage::OrderFilled {
                accounting: hypercall_engine::FillAccounting::from_fill(&fill),
                fill,
            })
            .await
            .unwrap();

        let cache = Arc::new(MarketStatsCache::new(portfolio_service));

        // Open interest should be (|10| + |-10|) / 2 = 10
        let oi = cache.get_open_interest("BTC-20260130-100000-C").await;
        assert_eq!(oi, dec!(10));
    }

    #[tokio::test]
    async fn test_handle_fill() {
        use hypercall_types::wallet_address::test_wallet;

        let portfolio_service = Arc::new(PortfolioServiceImpl::new());
        let cache = Arc::new(MarketStatsCache::new(portfolio_service));

        use hypercall_types::Fill;
        use hypercall_types::Side;

        let fill = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-20260130-100000-C".to_string(),
            price: dec!(1000),
            size: dec!(10_000_000), // 10 contracts in contract units
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(2),
            fee: dec!(0),
            is_taker: true,
            timestamp: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_secs(),
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: Some(dec!(950000)),
        };

        cache.handle_fill(&fill).await;

        let vol = cache.get_volume_24h("BTC-20260130-100000-C").await;
        assert_eq!(vol, dec!(950000));
    }

    #[tokio::test]
    async fn test_handle_option_fill_missing_underlying_notional_skips_volume() {
        use hypercall_types::wallet_address::test_wallet;

        let portfolio_service = Arc::new(PortfolioServiceImpl::new());
        let cache = Arc::new(MarketStatsCache::new(portfolio_service));

        use hypercall_types::Fill;
        use hypercall_types::Side;

        let fill = Fill {
            trade_id: 1,
            taker_order_id: 100,
            maker_order_id: 101,
            symbol: "BTC-20260130-100000-C".to_string(),
            price: dec!(1000),
            size: dec!(10_000_000),
            taker_side: Side::Buy,
            taker_wallet_address: test_wallet(1),
            maker_wallet_address: test_wallet(2),
            fee: dec!(0),
            is_taker: true,
            timestamp: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_secs(),
            builder_code_address: None,
            builder_code_fee: None,
            source: Default::default(),
            taker_realized_pnl: None,
            maker_realized_pnl: None,
            underlying_notional: None,
        };

        cache.handle_fill(&fill).await;

        let vol = cache.get_volume_24h("BTC-20260130-100000-C").await;
        assert_eq!(vol, dec!(0));
    }
}