foxhunt/backtesting/src/dbn_replay.rs

//! DBN (Databento Binary) Replay Engine
//!
//! Loads DBN market data and streams [`ReplayEvent`]s through an mpsc channel
//! for consumption by the backtesting pipeline.
//!
//! Two construction paths are supported:
//! - [`DbnReplayEngine::from_events`] for pre-built events (testing, pipelines)
//! - [`DbnReplayEngine::from_bytes`] for raw `.dbn` file bytes (requires the
//!   `dbn_converter` module once Task 1 is integrated)

use crate::replay_engine::ReplayEvent;
use tokio::sync::mpsc;
use trading_engine::types::events::MarketEvent;

/// Replay engine that holds a sorted sequence of [`ReplayEvent`]s and can
/// stream them through an unbounded tokio mpsc channel.
pub struct DbnReplayEngine {
    events: Vec<ReplayEvent>,
}

impl DbnReplayEngine {
    /// Load replay events from raw DBN bytes.
    ///
    /// This method requires the `data` crate's `DbnParser` and the
    /// `dbn_converter` bridge module. Until those are wired into the
    /// `backtesting` crate's dependency graph this returns an error.
    ///
    /// # Errors
    ///
    /// Returns an error because the DBN parsing pipeline is not yet
    /// integrated into the backtesting crate.
    pub fn from_bytes(
        _bytes: &[u8],
    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
        // TODO: Once the `data` crate is added to backtesting/Cargo.toml and
        // `crate::dbn_converter::processed_to_market_event` exists, replace
        // this stub with:
        //
        //   let parser = DbnParser::new()?;
        //   let processed = parser.parse_batch(bytes)?;
        //   let mut events = Vec::with_capacity(processed.len());
        //   for (seq, msg) in processed.into_iter().enumerate() {
        //       let market_event = processed_to_market_event(msg)?;
        //       let ts = market_event.timestamp();
        //       events.push(ReplayEvent {
        //           event: market_event,
        //           original_timestamp: ts,
        //           replay_timestamp: ts,
        //           source_id: "dbn".to_string(),
        //           sequence: seq as u64,
        //       });
        //   }
        //   events.sort_by_key(|e| e.original_timestamp);
        //   Ok(Self { events })
        Err("DbnReplayEngine::from_bytes is not yet available: \
             the `data` crate dependency and dbn_converter bridge \
             module must be integrated first"
            .into())
    }

    /// Create an engine from a pre-built vector of [`ReplayEvent`]s.
    ///
    /// Events are sorted by `original_timestamp` on construction so that
    /// [`start_replay`](Self::start_replay) always emits them in
    /// chronological order.
    #[must_use]
    pub fn from_events(mut events: Vec<ReplayEvent>) -> Self {
        events.sort_by_key(|e| e.original_timestamp);
        Self { events }
    }

    /// Number of events loaded into the engine.
    #[must_use]
    pub fn event_count(&self) -> usize {
        self.events.len()
    }

    /// Stream all loaded events through an unbounded mpsc channel.
    ///
    /// A background tokio task is spawned to send each event. If the
    /// receiver is dropped the task terminates gracefully.
    ///
    /// # Returns
    ///
    /// The receiving half of the channel. Events arrive in
    /// `original_timestamp` order.
    pub fn start_replay(&self) -> mpsc::UnboundedReceiver<ReplayEvent> {
        let (tx, rx) = mpsc::unbounded_channel();
        let events = self.events.clone();

        tokio::spawn(async move {
            for event in events {
                if tx.send(event).is_err() {
                    // Receiver dropped -- stop sending.
                    break;
                }
            }
        });

        rx
    }
}

/// Extract the timestamp from a [`MarketEvent`] variant.
///
/// Every `MarketEvent` variant carries a `timestamp: DateTime<Utc>` field.
/// This helper avoids having to pattern-match in calling code.
#[must_use]
pub fn extract_timestamp(event: &MarketEvent) -> chrono::DateTime<chrono::Utc> {
    event.timestamp()
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use chrono::{DateTime, TimeDelta};
    use common::{Price, Quantity, Symbol};

    /// Helper: build a synthetic `ReplayEvent` with the given sequence and
    /// timestamp offset (in seconds from a fixed base).
    fn make_replay_event(
        sequence: u64,
        offset_secs: i64,
    ) -> ReplayEvent {
        let base = DateTime::from_timestamp(1_700_000_000, 0)
            .unwrap_or_default();
        let ts = base + TimeDelta::seconds(offset_secs);

        // Use a simple Trade event -- Price/Quantity constructors are
        // fallible so we fall back to ZERO on error (will not happen for
        // these small test values).
        let event = MarketEvent::Trade {
            symbol: Symbol::new("TEST".to_string()),
            price: Price::from_f64(100.0).unwrap_or(Price::ZERO),
            size: Quantity::from_f64(1.0).unwrap_or(Quantity::ZERO),
            timestamp: ts,
            side: None,
            venue: None,
            trade_id: None,
        };

        ReplayEvent {
            event,
            original_timestamp: ts,
            replay_timestamp: ts,
            source_id: "test".to_string(),
            sequence,
        }
    }

    // -- test_from_events_empty -----------------------------------------------

    #[test]
    fn test_from_events_empty() {
        let engine = DbnReplayEngine::from_events(vec![]);
        assert_eq!(engine.event_count(), 0);
    }

    // -- test_from_events_sends_all -------------------------------------------

    #[tokio::test]
    async fn test_from_events_sends_all() {
        let events: Vec<ReplayEvent> = (0..5)
            .map(|i| make_replay_event(i, i as i64))
            .collect();

        let engine = DbnReplayEngine::from_events(events);
        assert_eq!(engine.event_count(), 5);

        let mut rx = engine.start_replay();

        let mut received = Vec::new();
        while let Some(event) = rx.recv().await {
            received.push(event);
        }

        assert_eq!(received.len(), 5);
        // Verify sequence numbers are present (0..5).
        for (idx, ev) in received.iter().enumerate() {
            assert_eq!(ev.sequence, idx as u64);
        }
    }

    // -- test_events_sorted_by_timestamp --------------------------------------

    #[tokio::test]
    async fn test_events_sorted_by_timestamp() {
        // Supply events deliberately out of order.
        let events = vec![
            make_replay_event(0, 30),  // latest
            make_replay_event(1, 10),  // earliest
            make_replay_event(2, 20),  // middle
        ];

        let engine = DbnReplayEngine::from_events(events);
        let mut rx = engine.start_replay();

        let mut received = Vec::new();
        while let Some(event) = rx.recv().await {
            received.push(event);
        }

        assert_eq!(received.len(), 3);

        // Events must arrive in ascending timestamp order.
        for pair in received.windows(2) {
            let first = pair.first();
            let second = pair.get(1);
            if let (Some(a), Some(b)) = (first, second) {
                assert!(
                    a.original_timestamp <= b.original_timestamp,
                    "Events not in chronological order: {:?} > {:?}",
                    a.original_timestamp,
                    b.original_timestamp,
                );
            }
        }
    }

    // -- test_from_bytes_stub_returns_error ------------------------------------

    #[test]
    fn test_from_bytes_stub_returns_error() {
        let result = DbnReplayEngine::from_bytes(&[]);
        assert!(result.is_err());
    }

    // -- test_extract_timestamp ------------------------------------------------

    #[test]
    fn test_extract_timestamp() {
        let ev = make_replay_event(0, 42);
        let ts = extract_timestamp(&ev.event);
        assert_eq!(ts, ev.original_timestamp);
    }
}