foxhunt/services/trading_agent_service/src/tick_processor.rs

//! Real-time tick processor for Databento MBP-10 -> microstructure features.
//!
//! Feeds each MBP-10 event and trade into OFICalculator + MicrostructureState.
//! At bar close, snapshots 20 features for the DQN forward pass.
//! Between bars, provides intermediate snapshots for speculative compute.

use data::providers::databento::mbp10::Mbp10Snapshot;
use ml_features::ofi_calculator::{MicrostructureState, OFICalculator, OFIFeatures};
use tokio::sync::watch;
use tracing::info;

/// Real-time tick processor for live trading.
///
/// Wraps `OFICalculator` (8 base features) and `MicrostructureState` (12 extended
/// features) to produce a 20-feature vector per bar from streaming MBP-10 ticks.
pub struct TickProcessor {
    /// OFI calculator (base 8 features)
    ofi: OFICalculator,
    /// Microstructure state (12 extended features)
    micro: MicrostructureState,
    /// Last computed OFI features (cached for mid-bar snapshots)
    last_ofi: OFIFeatures,
    /// Current bar start timestamp (nanoseconds)
    bar_start_ns: u64,
    /// Bar duration (nanoseconds, typically 60_000_000_000 for 60s)
    bar_duration_ns: u64,
    /// Channel to broadcast feature snapshots at bar close
    feature_tx: watch::Sender<[f64; 20]>,
    /// Tick count in current bar
    tick_count: u64,
}

impl TickProcessor {
    /// Create a new `TickProcessor`.
    ///
    /// Returns the processor and a `watch::Receiver` that receives the 20-feature
    /// snapshot every time a bar closes.
    pub fn new(bar_duration_ns: u64) -> (Self, watch::Receiver<[f64; 20]>) {
        let (tx, rx) = watch::channel([0.0; 20]);
        let now_ns = now_unix_ns();
        (
            Self {
                ofi: OFICalculator::new(),
                micro: MicrostructureState::new(now_ns, bar_duration_ns),
                last_ofi: OFIFeatures::zeros(),
                bar_start_ns: now_ns,
                bar_duration_ns,
                feature_tx: tx,
                tick_count: 0,
            },
            rx,
        )
    }

    /// Process an MBP-10 snapshot event.
    ///
    /// Updates both the OFI calculator and microstructure state. The OFI
    /// features are cached so `current_features()` can be called without
    /// requiring the snapshot again.
    pub fn on_snapshot(&mut self, snapshot: &Mbp10Snapshot) {
        // Update microstructure state (spread dynamics, queue depletion, etc.)
        self.micro.update_snapshot(snapshot);
        self.tick_count += 1;

        // Compute OFI features from snapshot and cache the result
        if let Ok(ofi_features) = self.ofi.calculate(snapshot) {
            self.micro
                .update_ofi_derived(ofi_features.ofi_level1, ofi_features.vpin);
            self.last_ofi = ofi_features;
        }
    }

    /// Process a trade event.
    ///
    /// Feeds trade data to both the OFI calculator (VPIN, Kyle's Lambda,
    /// trade imbalance) and microstructure state (Hawkes intensity, aggression).
    pub fn on_trade(&mut self, price: f64, volume: u64, is_buy: bool, timestamp_ns: u64) {
        self.ofi.feed_trade(price, volume, is_buy);
        self.micro.update_trade(price, volume, is_buy, timestamp_ns);
    }

    /// Snapshot current features (non-mutating, safe to call mid-bar for speculative compute).
    ///
    /// Returns 20 features: `[0..8]` = base OFI, `[8..20]` = microstructure state.
    pub fn current_features(&self) -> [f64; 20] {
        let base_8 = self.last_ofi.to_array();
        let micro_12 = self.micro.snapshot();
        let mut out = [0.0; 20];
        out[..8].copy_from_slice(&base_8);
        out[8..20].copy_from_slice(&micro_12);
        out
    }

    /// Called at bar close -- snapshot final features, broadcast, and reset for next bar.
    ///
    /// Returns the 20-feature vector that was broadcast.
    pub fn bar_close(&mut self) -> [f64; 20] {
        let features = self.current_features();
        let _ = self.feature_tx.send(features);

        info!(
            tick_count = self.tick_count,
            "Bar close: {} ticks processed, features broadcast",
            self.tick_count
        );

        // Reset for next bar
        let now_ns = now_unix_ns();
        self.bar_start_ns = now_ns;
        self.micro.reset(now_ns, self.bar_duration_ns);
        self.ofi = OFICalculator::new();
        self.last_ofi = OFIFeatures::zeros();
        self.tick_count = 0;

        features
    }

    /// Returns the tick count in the current bar.
    pub const fn tick_count(&self) -> u64 {
        self.tick_count
    }

    /// Returns the current bar start timestamp in nanoseconds.
    pub const fn bar_start_ns(&self) -> u64 {
        self.bar_start_ns
    }
}

/// Helper: current wall-clock time as nanoseconds since Unix epoch.
fn now_unix_ns() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map_or(0, |d| d.as_nanos() as u64)
}

#[cfg(test)]
mod tests {
    use super::*;
    use data::providers::databento::mbp10::BidAskPair;

    fn make_snapshot(bid_px: i64, ask_px: i64, bid_sz: u32, ask_sz: u32) -> Mbp10Snapshot {
        let level = BidAskPair {
            bid_px,
            bid_sz,
            bid_ct: 10,
            ask_px,
            ask_sz,
            ask_ct: 10,
        };
        Mbp10Snapshot {
            symbol: "ES.FUT".to_string(),
            timestamp: now_unix_ns(),
            levels: vec![level; 10],
            sequence: 1,
            trade_count: 0,
        }
    }

    #[test]
    fn test_new_returns_zero_features() {
        let (proc, rx) = TickProcessor::new(60_000_000_000);
        assert_eq!(*rx.borrow(), [0.0; 20]);
        assert_eq!(proc.tick_count(), 0);
    }

    #[test]
    fn test_on_snapshot_increments_tick_count() {
        let (mut proc, _rx) = TickProcessor::new(60_000_000_000);
        let snap = make_snapshot(5000_000_000_000, 5001_000_000_000, 100, 100);
        proc.on_snapshot(&snap);
        assert_eq!(proc.tick_count(), 1);
    }

    #[test]
    fn test_bar_close_resets_state() {
        let (mut proc, rx) = TickProcessor::new(60_000_000_000);
        let snap = make_snapshot(5000_000_000_000, 5001_000_000_000, 100, 100);
        proc.on_snapshot(&snap);
        proc.on_snapshot(&snap);
        assert_eq!(proc.tick_count(), 2);

        let features = proc.bar_close();
        assert_eq!(proc.tick_count(), 0);
        assert_eq!(*rx.borrow(), features);
    }

    #[test]
    fn test_current_features_is_20_dim() {
        let (proc, _rx) = TickProcessor::new(60_000_000_000);
        let features = proc.current_features();
        assert_eq!(features.len(), 20);
    }

    #[test]
    fn test_on_trade_feeds_through() {
        let (mut proc, _rx) = TickProcessor::new(60_000_000_000);
        // Should not panic
        proc.on_trade(5000.25, 10, true, now_unix_ns());
        proc.on_trade(5000.50, 5, false, now_unix_ns());
    }
}