diff --git a/Cargo.lock b/Cargo.lock index 5871852c9..2ff91f74e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1507,6 +1507,7 @@ dependencies = [ "crossbeam-channel", "csv", "dashmap 6.1.0", + "data", "fastrand", "futures", "ml", diff --git a/backtesting/Cargo.toml b/backtesting/Cargo.toml index 00c857544..48e47ab6f 100644 --- a/backtesting/Cargo.toml +++ b/backtesting/Cargo.toml @@ -36,6 +36,7 @@ rust_decimal_macros = { workspace = true } trading_engine.workspace = true ml.workspace = true # Use real ML models for backtesting common = { path = "../common" } +data.workspace = true tracing = { workspace = true } diff --git a/backtesting/src/dbn_converter.rs b/backtesting/src/dbn_converter.rs new file mode 100644 index 000000000..fa35e9595 --- /dev/null +++ b/backtesting/src/dbn_converter.rs @@ -0,0 +1,278 @@ +//! DBN to MarketEvent converter +//! +//! Converts `ProcessedMessage` values from the Databento DBN parser +//! into the canonical `MarketEvent` type used throughout the trading engine. + +use chrono::{DateTime, Utc}; +use common::{Quantity, Symbol}; +use data::providers::databento::dbn_parser::ProcessedMessage; +use trading_engine::timing::HardwareTimestamp; +use trading_engine::types::events::MarketEvent; + +/// Convert a `HardwareTimestamp` to `DateTime`. +/// +/// Uses the nanoseconds-since-epoch stored in the timestamp. +/// Returns `None` if the nanoseconds value cannot represent a valid datetime +/// (only possible for values outside the i64 range, which is practically +/// impossible for real timestamps). +fn hw_ts_to_datetime(ts: &HardwareTimestamp) -> Option> { + let nanos_i64 = i64::try_from(ts.as_nanos()).ok()?; + Some(DateTime::from_timestamp_nanos(nanos_i64)) +} + +/// Convert a `ProcessedMessage` from the DBN parser into a `MarketEvent`. +/// +/// Returns `Some(MarketEvent)` for `Trade`, `Ohlcv`, and `Quote` variants. +/// Returns `None` for `OrderBook` and `Status` variants, which do not have +/// a direct one-to-one mapping to the `MarketEvent` enum (order book updates +/// would need aggregation into a full snapshot, and status messages are +/// system-level events). +/// +/// # Arguments +/// +/// * `msg` - A reference to the `ProcessedMessage` to convert. +/// +/// # Returns +/// +/// `Some(MarketEvent)` if the message can be converted, `None` otherwise. +pub fn processed_to_market_event(msg: &ProcessedMessage) -> Option { + match msg { + ProcessedMessage::Trade { + symbol, + timestamp, + price, + size, + side, + trade_id, + conditions: _, + } => { + let dt = hw_ts_to_datetime(timestamp)?; + let qty = Quantity::from_decimal(*size).ok()?; + Some(MarketEvent::Trade { + symbol: Symbol::new(symbol.clone()), + price: *price, + size: qty, + timestamp: dt, + side: Some(*side), + venue: None, + trade_id: trade_id.clone(), + }) + } + ProcessedMessage::Ohlcv { + symbol, + timestamp, + open, + high, + low, + close, + volume, + } => { + let dt = hw_ts_to_datetime(timestamp)?; + let vol = Quantity::from_decimal(*volume).ok()?; + Some(MarketEvent::Bar { + symbol: Symbol::new(symbol.clone()), + open: *open, + high: *high, + low: *low, + close: *close, + volume: vol, + timestamp: dt, + interval: String::new(), + venue: None, + }) + } + ProcessedMessage::Quote { + symbol, + timestamp, + bid, + ask, + bid_size, + ask_size, + exchange, + } => { + let dt = hw_ts_to_datetime(timestamp)?; + let bid_price = (*bid)?; + let ask_price = (*ask)?; + let bq = Quantity::from_decimal((*bid_size)?).ok()?; + let aq = Quantity::from_decimal((*ask_size)?).ok()?; + Some(MarketEvent::Quote { + symbol: Symbol::new(symbol.clone()), + bid_price, + bid_size: bq, + ask_price, + ask_size: aq, + timestamp: dt, + venue: exchange.clone(), + }) + } + ProcessedMessage::OrderBook { .. } | ProcessedMessage::Status { .. } => None, + } +} + +#[cfg(test)] +mod tests { + use super::*; + use common::{OrderSide, Price}; + use rust_decimal::Decimal; + use trading_engine::timing::{HardwareTimestamp, TimingSource}; + + /// Helper: build a `HardwareTimestamp` from a `DateTime`. + fn ts_from_dt(dt: DateTime) -> HardwareTimestamp { + #[allow(clippy::cast_sign_loss)] + let nanos = dt.timestamp_nanos_opt().unwrap_or(0) as u64; + HardwareTimestamp { + cycles: 0, + nanos, + source: TimingSource::SystemClock, + validation_passed: true, + } + } + + #[test] + fn test_convert_trade() { + let now = Utc::now(); + let ts = ts_from_dt(now); + let price = Price::from_f64(123.45).ok(); + let price = price.unwrap_or(Price::ZERO); + + let msg = ProcessedMessage::Trade { + symbol: "AAPL".to_string(), + timestamp: ts, + price, + size: Decimal::new(100, 0), + side: OrderSide::Buy, + trade_id: Some("t1".to_string()), + conditions: vec![], + }; + + let event = processed_to_market_event(&msg); + assert!(event.is_some(), "Trade should convert to MarketEvent"); + + let event = event.unwrap_or_else(|| { + MarketEvent::Control { + command: String::new(), + parameters: std::collections::HashMap::new(), + timestamp: Utc::now(), + } + }); + + if let MarketEvent::Trade { + symbol, + price: p, + size, + side, + trade_id, + .. + } = &event + { + assert_eq!(symbol.as_str(), "AAPL"); + assert_eq!(*p, price); + assert_eq!(size.to_f64(), 100.0); + assert_eq!(*side, Some(OrderSide::Buy)); + assert_eq!(*trade_id, Some("t1".to_string())); + } else { + // Should not reach here + assert!(matches!(event, MarketEvent::Trade { .. }), "Expected MarketEvent::Trade variant"); + } + } + + #[test] + fn test_convert_ohlcv_to_bar() { + let now = Utc::now(); + let ts = ts_from_dt(now); + let open = Price::from_f64(100.0).unwrap_or(Price::ZERO); + let high = Price::from_f64(110.0).unwrap_or(Price::ZERO); + let low = Price::from_f64(95.0).unwrap_or(Price::ZERO); + let close = Price::from_f64(105.0).unwrap_or(Price::ZERO); + + let msg = ProcessedMessage::Ohlcv { + symbol: "MSFT".to_string(), + timestamp: ts, + open, + high, + low, + close, + volume: Decimal::new(5000, 0), + }; + + let event = processed_to_market_event(&msg); + assert!(event.is_some(), "OHLCV should convert to MarketEvent::Bar"); + + let event = event.unwrap_or_else(|| { + MarketEvent::Control { + command: String::new(), + parameters: std::collections::HashMap::new(), + timestamp: Utc::now(), + } + }); + + if let MarketEvent::Bar { + symbol, + open: o, + high: h, + low: l, + close: c, + volume, + .. + } = &event + { + assert_eq!(symbol.as_str(), "MSFT"); + assert_eq!(*o, open); + assert_eq!(*h, high); + assert_eq!(*l, low); + assert_eq!(*c, close); + assert_eq!(volume.to_f64(), 5000.0); + } else { + assert!(matches!(event, MarketEvent::Bar { .. }), "Expected MarketEvent::Bar variant"); + } + } + + #[test] + fn test_convert_quote() { + let now = Utc::now(); + let ts = ts_from_dt(now); + let bid = Price::from_f64(99.0).unwrap_or(Price::ZERO); + let ask = Price::from_f64(101.0).unwrap_or(Price::ZERO); + + let msg = ProcessedMessage::Quote { + symbol: "TSLA".to_string(), + timestamp: ts, + bid: Some(bid), + ask: Some(ask), + bid_size: Some(Decimal::new(200, 0)), + ask_size: Some(Decimal::new(150, 0)), + exchange: Some("NASDAQ".to_string()), + }; + + let event = processed_to_market_event(&msg); + assert!(event.is_some(), "Quote should convert to MarketEvent::Quote"); + + let event = event.unwrap_or_else(|| { + MarketEvent::Control { + command: String::new(), + parameters: std::collections::HashMap::new(), + timestamp: Utc::now(), + } + }); + + if let MarketEvent::Quote { + symbol, + bid_price, + ask_price, + bid_size, + ask_size, + venue, + .. + } = &event + { + assert_eq!(symbol.as_str(), "TSLA"); + assert_eq!(*bid_price, bid); + assert_eq!(*ask_price, ask); + assert_eq!(bid_size.to_f64(), 200.0); + assert_eq!(ask_size.to_f64(), 150.0); + assert_eq!(*venue, Some("NASDAQ".to_string())); + } else { + assert!(matches!(event, MarketEvent::Quote { .. }), "Expected MarketEvent::Quote variant"); + } + } +}