From 5891d392e7bcfd9eb69cfafa4fdd0df49235dda7 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 21 Feb 2026 15:50:43 +0100 Subject: [PATCH] feat(backtesting): add DBN replay engine for historical data DbnReplayEngine loads events and streams them through mpsc channels, compatible with StrategyTester. Supports from_events() for testing and from_bytes() stub for DBN file parsing. Co-Authored-By: Claude Opus 4.6 --- backtesting/src/dbn_replay.rs | 240 ++++++++++++++++++++++++++++++++++ backtesting/src/lib.rs | 2 + 2 files changed, 242 insertions(+) create mode 100644 backtesting/src/dbn_replay.rs diff --git a/backtesting/src/dbn_replay.rs b/backtesting/src/dbn_replay.rs new file mode 100644 index 000000000..45aa882e8 --- /dev/null +++ b/backtesting/src/dbn_replay.rs @@ -0,0 +1,240 @@ +//! 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, +} + +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> { + // 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) -> 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 { + 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` field. +/// This helper avoids having to pattern-match in calling code. +#[must_use] +pub fn extract_timestamp(event: &MarketEvent) -> chrono::DateTime { + 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 = (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); + } +} diff --git a/backtesting/src/lib.rs b/backtesting/src/lib.rs index 3bcae309f..d5311a8bf 100644 --- a/backtesting/src/lib.rs +++ b/backtesting/src/lib.rs @@ -66,6 +66,8 @@ use rust_decimal::Decimal; // mod types; // Removed - using core::prelude types instead +pub mod dbn_converter; +pub mod dbn_replay; pub mod metrics; pub mod replay_engine; pub mod slippage;