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"); + } + } +} 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..24fa72bba 100644 --- a/backtesting/src/lib.rs +++ b/backtesting/src/lib.rs @@ -66,7 +66,10 @@ 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 model_loader; pub mod replay_engine; pub mod slippage; pub mod strategies; @@ -74,6 +77,7 @@ pub mod strategy_runner; pub mod strategy_tester; // Re-export types for public API +pub use model_loader::{load_models_for_backtest, BacktestMLModel, ModelSpec}; pub use metrics::{MetricsCalculator, PerformanceAnalytics}; pub use replay_engine::{MarketReplay, ReplayConfig}; pub use strategies::{DQNAction, DQNReplayStrategy, PositionState, TradingActionType}; diff --git a/backtesting/src/model_loader.rs b/backtesting/src/model_loader.rs new file mode 100644 index 000000000..561924ac1 --- /dev/null +++ b/backtesting/src/model_loader.rs @@ -0,0 +1,341 @@ +//! Model loading and MLModel wrapper for backtesting. +//! +//! Provides [`BacktestMLModel`], a bridge that wraps a +//! [`ModelInferenceAdapter`] (ensemble-level inference) and exposes +//! it through the global [`MLModel`] trait so it can be registered +//! in the [`ModelRegistry`]. + +use std::collections::HashMap; +use std::sync::Arc; + +use async_trait::async_trait; + +use ml::dqn::dqn::DQNConfig; +use ml::ensemble::adapters::dqn::DqnInferenceAdapter; +use ml::ensemble::adapters::ppo::PpoInferenceAdapter; +use ml::ensemble::inference_adapter::{FeatureVector, ModelInferenceAdapter}; +use ml::ppo::ppo::PPOConfig; +use ml::{Features, MLError, MLModel, MLResult, ModelMetadata, ModelPrediction, ModelType}; + +/// Wraps a [`ModelInferenceAdapter`] to satisfy the [`MLModel`] trait +/// expected by the global [`ModelRegistry`]. +/// +/// The conversion path is: +/// `Features` -> `FeatureVector` -> adapter `predict` -> `EnsemblePrediction` -> `ModelPrediction` +pub struct BacktestMLModel { + adapter: Box, + model_type: ModelType, +} + +impl std::fmt::Debug for BacktestMLModel { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BacktestMLModel") + .field("model_name", &self.adapter.model_name()) + .field("model_type", &self.model_type) + .field("is_ready", &self.adapter.is_ready()) + .finish() + } +} + +impl BacktestMLModel { + /// Create a new `BacktestMLModel` from an inference adapter and its type. + pub fn new(adapter: Box, model_type: ModelType) -> Self { + Self { + adapter, + model_type, + } + } + + /// Delegates to the underlying adapter's readiness check. + pub fn is_ready(&self) -> bool { + self.adapter.is_ready() + } +} + +#[async_trait] +impl MLModel for BacktestMLModel { + fn name(&self) -> &str { + self.adapter.model_name() + } + + fn model_type(&self) -> ModelType { + self.model_type + } + + async fn predict(&self, features: &Features) -> MLResult { + // Convert Features -> FeatureVector + let fv = FeatureVector { + values: features.values.clone(), + timestamp: features.timestamp as i64, + }; + + // Run inference through the adapter + let ensemble_pred = self.adapter.predict(&fv)?; + + // Build metadata map from the prediction metadata + let mut metadata = HashMap::new(); + metadata.insert( + "direction".to_string(), + serde_json::json!(ensemble_pred.direction), + ); + metadata.insert( + "latency_us".to_string(), + serde_json::json!(ensemble_pred.metadata.latency_us), + ); + if let Some(ref q_values) = ensemble_pred.metadata.q_values { + metadata.insert("q_values".to_string(), serde_json::json!(q_values)); + } + if let Some(ref quantiles) = ensemble_pred.metadata.quantiles { + metadata.insert("quantiles".to_string(), serde_json::json!(quantiles)); + } + if let Some(ref attn) = ensemble_pred.metadata.attention_weights { + metadata.insert("attention_weights".to_string(), serde_json::json!(attn)); + } + + Ok(ModelPrediction { + value: ensemble_pred.direction, + confidence: ensemble_pred.confidence, + metadata, + timestamp: features.timestamp, + model_id: ensemble_pred.model_name, + }) + } + + fn get_confidence(&self) -> f64 { + 0.8 + } + + fn is_ready(&self) -> bool { + self.adapter.is_ready() + } + + fn get_metadata(&self) -> ModelMetadata { + ModelMetadata::new( + self.model_type, + "1.0.0".to_string(), + 51, // canonical 51-dim feature vector + 0.0, + ) + } + + fn validate_features(&self, features: &Features) -> MLResult<()> { + if features.values.is_empty() { + return Err(MLError::ValidationError { + message: "Empty feature vector".to_string(), + }); + } + Ok(()) + } +} + +/// Specification for loading a model into the backtest registry. +/// +/// Each `ModelSpec` describes one model to instantiate: its type +/// (e.g. `"DQN"`, `"PPO"`), an optional checkpoint path, and the +/// ensemble weight used by the aggregation layer. +#[derive(Debug, Clone)] +pub struct ModelSpec { + /// Model type identifier (`"DQN"`, `"PPO"`). + pub model_type: String, + /// Path to a `.safetensors` checkpoint. Empty string = random weights. + pub checkpoint_path: String, + /// Ensemble weight (used downstream by the aggregation layer). + pub weight: f64, +} + +/// Load models described by `specs` and register them in the global +/// [`ModelRegistry`](ml::ModelRegistry). +/// +/// Call this **before** running a backtest so that the strategy layer +/// can look up models by name. +/// +/// # Errors +/// +/// Returns `MLError` if a model fails to initialise or register. +/// Unknown `model_type` values are logged and skipped. +pub async fn load_models_for_backtest(specs: &[ModelSpec]) -> Result<(), MLError> { + let registry = ml::get_global_registry(); + + for spec in specs { + let adapter: Box = match spec.model_type.as_str() { + "DQN" => { + let config = DQNConfig { + state_dim: 51, + num_actions: 45, + hidden_dims: vec![128, 128], + ..Default::default() + }; + if spec.checkpoint_path.is_empty() { + Box::new(DqnInferenceAdapter::new(config)?) + } else { + Box::new(DqnInferenceAdapter::from_checkpoint( + config, + &spec.checkpoint_path, + )?) + } + } + "PPO" => { + let config = PPOConfig { + state_dim: 51, + num_actions: 45, + policy_hidden_dims: vec![128, 64], + value_hidden_dims: vec![128, 64], + ..Default::default() + }; + Box::new(PpoInferenceAdapter::new(config)?) + } + other => { + tracing::warn!("Unknown model type: {}, skipping", other); + continue; + } + }; + + let model_type = match spec.model_type.as_str() { + "DQN" => ModelType::DQN, + "PPO" => ModelType::PPO, + _ => { + // Already skipped above via `continue`, but guard against + // future additions. + tracing::warn!("Unmapped model type: {}, skipping registration", spec.model_type); + continue; + } + }; + + let ml_model = BacktestMLModel::new(adapter, model_type); + registry.register(Arc::new(ml_model)).await?; + } + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + use ml::dqn::dqn::DQNConfig; + use ml::ensemble::adapters::dqn::DqnInferenceAdapter; + + fn test_dqn_config() -> DQNConfig { + DQNConfig { + state_dim: 51, + num_actions: 45, + hidden_dims: vec![64, 64], + ..Default::default() + } + } + + #[tokio::test] + async fn test_backtest_ml_model_wraps_dqn() { + let adapter = DqnInferenceAdapter::new(test_dqn_config()) + .expect("DqnInferenceAdapter creation should succeed in test"); + let model = BacktestMLModel::new(Box::new(adapter), ModelType::DQN); + + // Build a 51-dim feature set + let features = Features::new(vec![0.1; 51], (0..51).map(|i| format!("f{i}")).collect()); + + let prediction = model + .predict(&features) + .await + .expect("predict should succeed in test"); + + // Direction must be in [-1, 1] + assert!( + prediction.value >= -1.0 && prediction.value <= 1.0, + "direction {} out of [-1, 1]", + prediction.value, + ); + // Confidence must be in [0, 1] + assert!( + prediction.confidence >= 0.0 && prediction.confidence <= 1.0, + "confidence {} out of [0, 1]", + prediction.confidence, + ); + // Model id should be "DQN" + assert_eq!(prediction.model_id, "DQN"); + } + + #[tokio::test] + async fn test_backtest_ml_model_name_and_type() { + let adapter = DqnInferenceAdapter::new(test_dqn_config()) + .expect("DqnInferenceAdapter creation should succeed in test"); + let model = BacktestMLModel::new(Box::new(adapter), ModelType::DQN); + + assert_eq!(model.name(), "DQN"); + assert_eq!(model.model_type(), ModelType::DQN); + assert!(model.is_ready()); + } + + #[tokio::test] + async fn test_backtest_ml_model_validate_empty_features() { + let adapter = DqnInferenceAdapter::new(test_dqn_config()) + .expect("DqnInferenceAdapter creation should succeed in test"); + let model = BacktestMLModel::new(Box::new(adapter), ModelType::DQN); + + let empty_features = Features::new(vec![], vec![]); + let result = model.validate_features(&empty_features); + assert!(result.is_err(), "should reject empty features"); + } + + #[test] + fn test_model_spec_creation() { + let spec = ModelSpec { + model_type: "DQN".to_string(), + checkpoint_path: String::new(), + weight: 0.6, + }; + assert_eq!(spec.model_type, "DQN"); + assert!(spec.checkpoint_path.is_empty()); + assert!((spec.weight - 0.6).abs() < f64::EPSILON); + } + + #[tokio::test] + async fn test_load_dqn_into_registry() { + let specs = vec![ModelSpec { + model_type: "DQN".to_string(), + checkpoint_path: String::new(), + weight: 1.0, + }]; + + load_models_for_backtest(&specs) + .await + .expect("load_models_for_backtest should succeed in test"); + + let registry = ml::get_global_registry(); + let model = registry.get("DQN").await; + assert!(model.is_some(), "DQN should be registered in the global registry"); + + let model = model.expect("already checked Some above"); + assert_eq!(model.name(), "DQN"); + assert_eq!(model.model_type(), ModelType::DQN); + assert!(model.is_ready()); + } + + #[tokio::test] + async fn test_load_ppo_into_registry() { + let specs = vec![ModelSpec { + model_type: "PPO".to_string(), + checkpoint_path: String::new(), + weight: 0.5, + }]; + + load_models_for_backtest(&specs) + .await + .expect("load_models_for_backtest should succeed for PPO in test"); + + let registry = ml::get_global_registry(); + let model = registry.get("PPO").await; + assert!(model.is_some(), "PPO should be registered in the global registry"); + } + + #[tokio::test] + async fn test_load_unknown_model_type_skipped() { + let specs = vec![ModelSpec { + model_type: "UNKNOWN_MODEL".to_string(), + checkpoint_path: String::new(), + weight: 1.0, + }]; + + // Should succeed (unknown types are skipped, not errored) + let result = load_models_for_backtest(&specs).await; + assert!(result.is_ok(), "unknown model types should be skipped without error"); + } +} diff --git a/backtesting/src/strategy_runner.rs b/backtesting/src/strategy_runner.rs index f68fe1db7..5503b3787 100644 --- a/backtesting/src/strategy_runner.rs +++ b/backtesting/src/strategy_runner.rs @@ -7,6 +7,7 @@ use anyhow::Result; use async_trait::async_trait; +use common::ml_strategy::ProductionFeatureExtractor225; use common::Order; use common::{OrderSide, OrderStatus, Position, Price, Quantity, Symbol}; use rust_decimal::prelude::ToPrimitive; @@ -15,6 +16,7 @@ use trading_engine::types::events::MarketEvent; // Use canonical types from ML module and real ML registry use chrono::{DateTime, Utc}; use dashmap::DashMap; +use ml::features::ProductionFeatureExtractorAdapter; use ml::{get_global_registry, Features, ModelPrediction}; use parking_lot::RwLock; use serde::{Deserialize, Serialize}; @@ -27,7 +29,8 @@ use tracing::{debug, info, warn}; use std::arch::x86_64::*; use crate::strategy_tester::{ - SignalType, Strategy, StrategyConfig, StrategyContext, StrategyResult, TradingSignal, + PerformanceSnapshot, SignalType, Strategy, StrategyConfig, StrategyContext, StrategyResult, + TradeRecord, TradingSignal, }; /// Adaptive strategy runner that integrates ML models with backtesting @@ -40,8 +43,10 @@ pub struct AdaptiveStrategyRunner { predictions_cache: Arc>, /// Performance tracking performance_tracker: Arc>, - /// Feature extractor + /// Feature extractor (legacy 3-5 dim fallback) feature_extractor: Arc, + /// Production 51-dimension feature extractor from ml::features + production_extractor: Arc>, /// Risk manager risk_manager: Arc, } @@ -178,6 +183,19 @@ impl Default for MarketState { } } +/// Tracks an open (or partially open) entry for PnL accounting. +#[derive(Debug, Clone)] +struct OpenEntry { + /// Side of the original entry + side: OrderSide, + /// Average entry price + entry_price: Price, + /// Remaining quantity in the entry + quantity: Decimal, + /// Timestamp when the position was opened + entry_time: DateTime, +} + /// Performance tracking for the strategy #[derive(Debug, Clone)] struct PerformanceTracker { @@ -193,9 +211,17 @@ struct PerformanceTracker { current_drawdown: Decimal, /// Peak portfolio value peak_value: Decimal, + /// Initial capital for equity curve baseline + initial_capital: Decimal, /// Model prediction accuracy #[allow(dead_code)] model_accuracy: HashMap, + /// Completed round-trip trade records + trade_records: Vec, + /// Equity snapshots: (timestamp, portfolio_value) + equity_snapshots: Vec<(DateTime, Decimal)>, + /// Open position entries keyed by symbol for PnL accounting + open_entries: HashMap, } impl Default for PerformanceTracker { @@ -207,7 +233,11 @@ impl Default for PerformanceTracker { max_drawdown: Decimal::ZERO, current_drawdown: Decimal::ZERO, peak_value: Decimal::ZERO, + initial_capital: Decimal::ZERO, model_accuracy: HashMap::new(), + trade_records: Vec::new(), + equity_snapshots: Vec::new(), + open_entries: HashMap::new(), } } } @@ -685,6 +715,7 @@ impl AdaptiveStrategyRunner { pub fn new(config: AdaptiveStrategyConfig) -> Self { let feature_extractor = Arc::new(FeatureExtractor::new(config.feature_settings.clone())); let risk_manager = Arc::new(RiskManager::new(config.risk_settings.clone())); + let production_extractor = Arc::new(RwLock::new(ProductionFeatureExtractorAdapter::new())); Self { config, @@ -692,6 +723,7 @@ impl AdaptiveStrategyRunner { predictions_cache: Arc::new(DashMap::new()), performance_tracker: Arc::new(RwLock::new(PerformanceTracker::default())), feature_extractor, + production_extractor, risk_manager, } } @@ -856,6 +888,7 @@ impl Strategy for AdaptiveStrategyRunner { { let mut tracker = self.performance_tracker.write(); tracker.peak_value = initial_capital; + tracker.initial_capital = initial_capital; } // Verify models are available @@ -906,40 +939,102 @@ impl Strategy for AdaptiveStrategyRunner { } } - // Extract features and get prediction - let market_state = self.market_state.read().clone(); + // Update production extractor with new price data + let price_f64 = price.to_decimal()?.to_f64().unwrap_or(0.0); + // Volume not available in MarketEvent::Trade; use a default + let volume_f64 = 1000.0; + { + let mut prod_ext = self.production_extractor.write(); + if let Err(e) = prod_ext.update(price_f64, volume_f64, *timestamp) { + debug!("Production extractor update failed: {}", e); + } + } - if market_state.price_history.len() >= 10 { - // Minimum data for prediction - match self.feature_extractor.extract_features(&market_state).await { - Ok(features) => { - match self.get_ensemble_prediction(&features).await { - Ok(prediction) => { - if let Some(signal) = self.generate_signal( - &prediction, - symbol.clone(), - price.to_decimal()?, + // Try production 51-dim feature extraction first + let production_features = { + let mut prod_ext = self.production_extractor.write(); + prod_ext.extract_features() + }; + + match production_features { + Ok(feat_values) if feat_values.len() == 51 => { + // Build feature names for the 51-dim vector + let feature_names: Vec = (0..51) + .map(|i| format!("prod_feature_{}", i)) + .collect(); + + let ts_micros = timestamp + .timestamp_micros(); + let features = Features { + values: feat_values, + names: feature_names, + timestamp: ts_micros as u64, + symbol: Some(symbol.to_string()), + }; + + match self.get_ensemble_prediction(&features).await { + Ok(prediction) => { + if let Some(signal) = self.generate_signal( + &prediction, + symbol.clone(), + price.to_decimal()?, + context.account_balance, + )? { + let current_position = context.positions.get(symbol); + if self.risk_manager.validate_trade( + &signal, + current_position, context.account_balance, )? { - // Validate trade with risk manager - let current_position = context.positions.get(symbol); - if self.risk_manager.validate_trade( - &signal, - current_position, - context.account_balance, - )? { - signals.push(signal); + signals.push(signal); + } + } + } + Err(e) => { + debug!("Prediction failed: {}", e); + } + } + } + Ok(_) | Err(_) => { + // Production extractor not warmed up yet, fall back to legacy extractor + let market_state = self.market_state.read().clone(); + + if market_state.price_history.len() >= 10 { + match self + .feature_extractor + .extract_features(&market_state) + .await + { + Ok(features) => { + match self.get_ensemble_prediction(&features).await { + Ok(prediction) => { + if let Some(signal) = self.generate_signal( + &prediction, + symbol.clone(), + price.to_decimal()?, + context.account_balance, + )? { + let current_position = + context.positions.get(symbol); + if self.risk_manager.validate_trade( + &signal, + current_position, + context.account_balance, + )? { + signals.push(signal); + } + } + } + Err(e) => { + debug!("Prediction failed: {}", e); } } - }, + } Err(e) => { - debug!("Prediction failed: {}", e); - }, + debug!("Feature extraction failed: {}", e); + } } - }, - Err(e) => { - debug!("Feature extraction failed: {}", e); - }, + } } } }, @@ -951,18 +1046,125 @@ impl Strategy for AdaptiveStrategyRunner { Ok(signals) } - async fn on_order_update(&mut self, order: &Order, _context: &StrategyContext) -> Result<()> { + async fn on_order_update(&mut self, order: &Order, context: &StrategyContext) -> Result<()> { if order.status == OrderStatus::Filled { + let fill_price = order + .average_price + .unwrap_or_else(|| order.price.unwrap_or(Price::ZERO)); + let fill_qty = order.quantity.to_decimal().unwrap_or(Decimal::ZERO); + let symbol_key = order.symbol.to_string(); + let mut tracker = self.performance_tracker.write(); - tracker.total_trades += 1; + + // Check if this fill closes (or reduces) an existing open entry + let mut closed = false; + if let Some(entry) = tracker.open_entries.get(&symbol_key).cloned() { + // A Buy closes a short entry; a Sell closes a long entry + let is_closing = matches!( + (&entry.side, &order.side), + (OrderSide::Buy, OrderSide::Sell) | (OrderSide::Sell, OrderSide::Buy) + ); + + if is_closing { + let closed_qty = entry.quantity.min(fill_qty); + let entry_dec = + entry.entry_price.to_decimal().unwrap_or(Decimal::ZERO); + let exit_dec = fill_price.to_decimal().unwrap_or(Decimal::ZERO); + + // PnL: long close = (exit - entry) * qty + // short close = (entry - exit) * qty + let pnl = if entry.side == OrderSide::Buy { + (exit_dec - entry_dec) * closed_qty + } else { + (entry_dec - exit_dec) * closed_qty + }; + + let return_pct = if entry_dec != Decimal::ZERO { + pnl / (entry_dec * closed_qty) * Decimal::from(100) + } else { + Decimal::ZERO + }; + + let trade = TradeRecord { + trade_id: uuid::Uuid::new_v4().to_string(), + symbol: order.symbol.clone(), + side: entry.side.clone(), + entry_price: entry.entry_price, + exit_price: fill_price, + quantity: order.quantity, + entry_time: entry.entry_time, + exit_time: context.current_time, + pnl, + return_pct, + commission: Decimal::ZERO, + }; + + tracker.total_trades += 1; + tracker.total_pnl += pnl; + if pnl > Decimal::ZERO { + tracker.winning_trades += 1; + } + tracker.trade_records.push(trade); + + // Update or remove the open entry + let remaining = entry.quantity - closed_qty; + if remaining <= Decimal::ZERO { + tracker.open_entries.remove(&symbol_key); + } else { + if let Some(e) = tracker.open_entries.get_mut(&symbol_key) { + e.quantity = remaining; + } + } + + // If the fill is larger than the old entry, open a new entry + // for the remainder (position flip) + let excess = fill_qty - closed_qty; + if excess > Decimal::ZERO { + tracker.open_entries.insert( + symbol_key.clone(), + OpenEntry { + side: order.side.clone(), + entry_price: fill_price, + quantity: excess, + entry_time: context.current_time, + }, + ); + } + + closed = true; + } + } + + if !closed { + // Opening a new position or adding to existing + let entry = tracker + .open_entries + .entry(symbol_key) + .or_insert_with(|| OpenEntry { + side: order.side.clone(), + entry_price: fill_price, + quantity: Decimal::ZERO, + entry_time: context.current_time, + }); + entry.quantity += fill_qty; + // Update average entry price (weighted average) + // This simple version just keeps the latest; for proper + // accounting the PositionTracker in strategy_tester handles it + } + + // Take an equity snapshot + tracker + .equity_snapshots + .push((context.current_time, context.account_balance)); info!( - "Order filled: {} {} @ {}", + "Order filled: {} {} @ {} | total_pnl={} trades={} wins={}", order.side, order.quantity, - order - .average_price - .unwrap_or_else(|| order.price.unwrap_or(Price::ZERO)) + fill_price, + tracker.total_pnl, + tracker.total_trades, + tracker.winning_trades, ); } @@ -987,13 +1189,18 @@ impl Strategy for AdaptiveStrategyRunner { if context.account_balance > tracker.peak_value { tracker.peak_value = context.account_balance; tracker.current_drawdown = Decimal::ZERO; - } else { + } else if tracker.peak_value > Decimal::ZERO { tracker.current_drawdown = (tracker.peak_value - context.account_balance) / tracker.peak_value; if tracker.current_drawdown > tracker.max_drawdown { tracker.max_drawdown = tracker.current_drawdown; } } + + // Record equity snapshot on every position update + tracker + .equity_snapshots + .push((context.current_time, context.account_balance)); } Ok(()) @@ -1002,8 +1209,14 @@ impl Strategy for AdaptiveStrategyRunner { async fn finalize(&mut self, context: &StrategyContext) -> Result { let tracker = self.performance_tracker.read(); - let total_return = if tracker.peak_value > Decimal::ZERO { - (context.account_balance - tracker.peak_value) / tracker.peak_value + let baseline = if tracker.initial_capital > Decimal::ZERO { + tracker.initial_capital + } else { + tracker.peak_value + }; + + let total_return = if baseline > Decimal::ZERO { + (context.account_balance - baseline) / baseline } else { Decimal::ZERO }; @@ -1021,6 +1234,31 @@ impl Strategy for AdaptiveStrategyRunner { Decimal::ZERO }; + // Build performance timeline from equity snapshots + let performance_timeline: Vec = tracker + .equity_snapshots + .iter() + .map(|(ts, value)| { + let realized = tracker.total_pnl; + let dd = if tracker.peak_value > Decimal::ZERO { + (tracker.peak_value - *value) / tracker.peak_value + } else { + Decimal::ZERO + }; + PerformanceSnapshot { + timestamp: *ts, + portfolio_value: *value, + cash_balance: *value, + unrealized_pnl: Decimal::ZERO, + realized_pnl: realized, + open_positions: 0, + drawdown: if dd > Decimal::ZERO { dd } else { Decimal::ZERO }, + } + }) + .collect(); + + let trades = tracker.trade_records.clone(); + Ok(StrategyResult { strategy_name: "adaptive_ml_strategy".to_string(), total_return, @@ -1035,8 +1273,8 @@ impl Strategy for AdaptiveStrategyRunner { Decimal::ZERO }, final_value: context.account_balance, - trades: vec![], // Would be populated with detailed trade records - performance_timeline: vec![], // Would be populated with performance snapshots + trades, + performance_timeline, }) } @@ -1121,4 +1359,363 @@ mod tests { let features = extractor.extract_features(&market_state).await; assert!(features.is_ok()); } + + #[test] + fn test_production_features_have_51_dimensions() { + let mut extractor = ProductionFeatureExtractorAdapter::new(); + // Feed 55 price updates (past the warmup period of 50) + for i in 0..55 { + let price = 100.0 + i as f64 * 0.1; + let volume = 1000.0; + let timestamp = Utc::now(); + extractor + .update(price, volume, timestamp) + .unwrap_or_else(|e| { + panic!("Production extractor update failed at step {}: {}", i, e) + }); + } + let features = extractor + .extract_features() + .unwrap_or_else(|e| panic!("Production extractor extract_features failed: {}", e)); + assert_eq!( + features.len(), + 51, + "Production extractor should produce exactly 51 features, got {}", + features.len() + ); + // Validate all features are finite + for (i, val) in features.iter().enumerate() { + assert!( + val.is_finite(), + "Feature {} should be finite, found {}", + i, val + ); + } + } + + #[test] + fn test_production_extractor_warmup_skips_gracefully() { + let mut extractor = ProductionFeatureExtractorAdapter::new(); + // Feed only 5 price updates (below warmup threshold) + for i in 0..5 { + let price = 100.0 + i as f64 * 0.1; + let _ = extractor.update(price, 1000.0, Utc::now()); + } + // Should either return an error or return a short vector + // (during warmup, extraction may fail) + let result = extractor.extract_features(); + // We accept either an error (warmup not ready) or a valid 51-dim result + match result { + Ok(features) => { + assert_eq!(features.len(), 51); + } + Err(_) => { + // Expected during warmup - this is fine + } + } + } + + #[test] + fn test_adaptive_strategy_has_production_extractor() { + let strategy = create_adaptive_strategy(); + // Verify the production extractor is initialized + let ext = strategy.production_extractor.read(); + // Should be able to access the extractor without issues + drop(ext); + } + + #[test] + fn test_performance_tracker_records_trade() { + use crate::strategy_tester::TradeRecord; + + let mut tracker = PerformanceTracker::default(); + tracker.initial_capital = Decimal::from(100_000); + tracker.peak_value = Decimal::from(100_000); + + // Simulate opening a long entry + tracker.open_entries.insert( + "AAPL".to_string(), + OpenEntry { + side: OrderSide::Buy, + entry_price: Price::from_f64(100.0).unwrap(), + quantity: Decimal::from(10), + entry_time: Utc::now(), + }, + ); + + // Simulate a winning close: sold at 110 + let pnl = (Decimal::from(110) - Decimal::from(100)) * Decimal::from(10); // +100 + tracker.total_trades += 1; + tracker.total_pnl += pnl; + if pnl > Decimal::ZERO { + tracker.winning_trades += 1; + } + tracker.trade_records.push(TradeRecord { + trade_id: "test-1".to_string(), + symbol: Symbol::from("AAPL"), + side: OrderSide::Buy, + entry_price: Price::from_f64(100.0).unwrap(), + exit_price: Price::from_f64(110.0).unwrap(), + quantity: Quantity::from_f64(10.0).unwrap(), + entry_time: Utc::now(), + exit_time: Utc::now(), + pnl, + return_pct: Decimal::from(10), + commission: Decimal::ZERO, + }); + tracker.open_entries.remove("AAPL"); + + assert_eq!(tracker.total_trades, 1); + assert_eq!(tracker.winning_trades, 1); + assert_eq!(tracker.total_pnl, Decimal::from(100)); + assert_eq!(tracker.trade_records.len(), 1); + assert_eq!(tracker.trade_records.first().map(|t| &t.symbol), Some(&Symbol::from("AAPL"))); + } + + #[test] + fn test_performance_tracker_losing_trade() { + let mut tracker = PerformanceTracker::default(); + tracker.initial_capital = Decimal::from(100_000); + tracker.peak_value = Decimal::from(100_000); + + // Simulate opening a long entry + tracker.open_entries.insert( + "TSLA".to_string(), + OpenEntry { + side: OrderSide::Buy, + entry_price: Price::from_f64(200.0).unwrap(), + quantity: Decimal::from(5), + entry_time: Utc::now(), + }, + ); + + // Simulate a losing close: sold at 180 + let pnl = (Decimal::from(180) - Decimal::from(200)) * Decimal::from(5); // -100 + tracker.total_trades += 1; + tracker.total_pnl += pnl; + if pnl > Decimal::ZERO { + tracker.winning_trades += 1; + } + tracker.trade_records.push(TradeRecord { + trade_id: "test-2".to_string(), + symbol: Symbol::from("TSLA"), + side: OrderSide::Buy, + entry_price: Price::from_f64(200.0).unwrap(), + exit_price: Price::from_f64(180.0).unwrap(), + quantity: Quantity::from_f64(5.0).unwrap(), + entry_time: Utc::now(), + exit_time: Utc::now(), + pnl, + return_pct: Decimal::from(-10), + commission: Decimal::ZERO, + }); + tracker.open_entries.remove("TSLA"); + + assert_eq!(tracker.total_trades, 1); + assert_eq!(tracker.winning_trades, 0); + assert_eq!(tracker.total_pnl, Decimal::from(-100)); + assert_eq!(tracker.trade_records.len(), 1); + } + + #[tokio::test] + async fn test_finalize_populates_trades() { + let mut runner = create_adaptive_strategy(); + let initial_capital = Decimal::from(100_000); + + // Initialize the runner so peak_value / initial_capital are set + runner + .initialize(initial_capital, StrategyConfig::default()) + .await + .unwrap(); + + // Manually inject trade records and equity snapshots into the tracker + { + let mut tracker = runner.performance_tracker.write(); + tracker.total_trades = 2; + tracker.winning_trades = 1; + tracker.total_pnl = Decimal::from(50); // net +50 + + tracker.trade_records.push(TradeRecord { + trade_id: "finalize-1".to_string(), + symbol: Symbol::from("AAPL"), + side: OrderSide::Buy, + entry_price: Price::from_f64(100.0).unwrap(), + exit_price: Price::from_f64(110.0).unwrap(), + quantity: Quantity::from_f64(10.0).unwrap(), + entry_time: Utc::now(), + exit_time: Utc::now(), + pnl: Decimal::from(100), + return_pct: Decimal::from(10), + commission: Decimal::ZERO, + }); + tracker.trade_records.push(TradeRecord { + trade_id: "finalize-2".to_string(), + symbol: Symbol::from("MSFT"), + side: OrderSide::Buy, + entry_price: Price::from_f64(300.0).unwrap(), + exit_price: Price::from_f64(290.0).unwrap(), + quantity: Quantity::from_f64(5.0).unwrap(), + entry_time: Utc::now(), + exit_time: Utc::now(), + pnl: Decimal::from(-50), + return_pct: Decimal::new(-333, 2), + commission: Decimal::ZERO, + }); + + // Add equity snapshots + let now = Utc::now(); + tracker.equity_snapshots.push((now, Decimal::from(100_100))); + tracker.equity_snapshots.push((now, Decimal::from(100_050))); + } + + // Build a context with the final balance + let context = StrategyContext { + current_time: Utc::now(), + account_balance: Decimal::from(100_050), + buying_power: Decimal::from(100_050), + positions: HashMap::new(), + open_orders: HashMap::new(), + market_prices: HashMap::new(), + performance: crate::strategy_tester::PerformanceMetrics::default(), + }; + + let result = runner.finalize(&context).await.unwrap(); + + // Trades should be populated + assert_eq!(result.trades.len(), 2, "finalize() must return trade records"); + assert_eq!(result.trades.first().map(|t| t.trade_id.as_str()), Some("finalize-1")); + + // Performance timeline should be populated from equity snapshots + assert_eq!( + result.performance_timeline.len(), + 2, + "finalize() must return performance timeline" + ); + + // Verify aggregated stats + assert_eq!(result.total_trades, 2); + assert_eq!(result.win_rate, Decimal::from(1) / Decimal::from(2)); + assert_eq!(result.avg_trade_return, Decimal::from(25)); // 50 / 2 + assert_eq!(result.final_value, Decimal::from(100_050)); + } + + #[tokio::test] + async fn test_on_order_update_tracks_pnl() { + let mut runner = create_adaptive_strategy(); + let initial_capital = Decimal::from(100_000); + + runner + .initialize(initial_capital, StrategyConfig::default()) + .await + .unwrap(); + + let now = Utc::now(); + let context = StrategyContext { + current_time: now, + account_balance: initial_capital, + buying_power: initial_capital, + positions: HashMap::new(), + open_orders: HashMap::new(), + market_prices: HashMap::new(), + performance: crate::strategy_tester::PerformanceMetrics::default(), + }; + + // Step 1: simulate a Buy fill at $100 (opening) + let buy_order = Order { + id: "order_1".to_string().into(), + client_order_id: None, + broker_order_id: None, + account_id: None, + symbol: Symbol::from("AAPL"), + side: OrderSide::Buy, + order_type: common::OrderType::Market, + status: OrderStatus::Filled, + time_in_force: common::TimeInForce::Day, + quantity: Quantity::from_f64(10.0).unwrap(), + price: Some(Price::from_f64(100.0).unwrap()), + stop_price: None, + filled_quantity: Quantity::from_f64(10.0).unwrap(), + remaining_quantity: Quantity::ZERO, + average_price: Some(Price::from_f64(100.0).unwrap()), + avg_fill_price: None, + average_fill_price: None, + exchange_order_id: None, + parent_id: None, + execution_algorithm: None, + execution_params: serde_json::json!({}), + stop_loss: None, + take_profit: None, + created_at: common::HftTimestamp::now_or_zero(), + updated_at: None, + expires_at: None, + metadata: serde_json::json!({}), + }; + + runner.on_order_update(&buy_order, &context).await.unwrap(); + + // After a buy open, no round-trip trade yet + { + let tracker = runner.performance_tracker.read(); + assert_eq!(tracker.total_trades, 0, "Opening buy should not count as a completed trade"); + assert!(tracker.open_entries.contains_key("AAPL")); + } + + // Step 2: simulate a Sell fill at $110 (closing the long) + let sell_order = Order { + id: "order_2".to_string().into(), + client_order_id: None, + broker_order_id: None, + account_id: None, + symbol: Symbol::from("AAPL"), + side: OrderSide::Sell, + order_type: common::OrderType::Market, + status: OrderStatus::Filled, + time_in_force: common::TimeInForce::Day, + quantity: Quantity::from_f64(10.0).unwrap(), + price: Some(Price::from_f64(110.0).unwrap()), + stop_price: None, + filled_quantity: Quantity::from_f64(10.0).unwrap(), + remaining_quantity: Quantity::ZERO, + average_price: Some(Price::from_f64(110.0).unwrap()), + avg_fill_price: None, + average_fill_price: None, + exchange_order_id: None, + parent_id: None, + execution_algorithm: None, + execution_params: serde_json::json!({}), + stop_loss: None, + take_profit: None, + created_at: common::HftTimestamp::now_or_zero(), + updated_at: None, + expires_at: None, + metadata: serde_json::json!({}), + }; + + let close_context = StrategyContext { + current_time: now, + account_balance: Decimal::from(100_100), // gained 100 + buying_power: Decimal::from(100_100), + positions: HashMap::new(), + open_orders: HashMap::new(), + market_prices: HashMap::new(), + performance: crate::strategy_tester::PerformanceMetrics::default(), + }; + + runner + .on_order_update(&sell_order, &close_context) + .await + .unwrap(); + + // After close, we should have a completed trade with PnL = +100 + { + let tracker = runner.performance_tracker.read(); + assert_eq!(tracker.total_trades, 1, "Closing sell should create a completed trade"); + assert_eq!(tracker.winning_trades, 1); + assert_eq!(tracker.total_pnl, Decimal::from(100)); + assert_eq!(tracker.trade_records.len(), 1); + assert!(!tracker.open_entries.contains_key("AAPL"), "Entry should be removed after close"); + // Equity snapshots recorded on each order update + assert!(tracker.equity_snapshots.len() >= 2); + } + } } diff --git a/backtesting/src/strategy_tester.rs b/backtesting/src/strategy_tester.rs index 8099effbd..f1a674ea3 100644 --- a/backtesting/src/strategy_tester.rs +++ b/backtesting/src/strategy_tester.rs @@ -820,24 +820,180 @@ impl PositionTracker { .collect() } + /// Update position accounting for a filled order. + /// + /// Handles opening new positions, adding to existing ones, reducing positions, + /// closing positions (generating a TradeRecord with realized PnL), and flipping + /// from long to short or vice-versa. pub async fn update_position( &self, - _symbol: &Symbol, - _order: &Order, - _execution_price: Price, + symbol: &Symbol, + order: &Order, + execution_price: Price, ) -> Result<()> { - // Position update logic would be implemented here - // This is a simplified version + let exec_price_dec = execution_price.to_decimal().unwrap_or(Decimal::ZERO); + let order_qty_dec = order.quantity.to_decimal().unwrap_or(Decimal::ZERO); + + // Signed fill quantity: positive for Buy, negative for Sell + let signed_fill_qty = match order.side { + OrderSide::Buy => order_qty_dec, + OrderSide::Sell => -order_qty_dec, + }; + + let now = Utc::now(); + + if let Some(mut pos_ref) = self.positions.get_mut(symbol) { + let old_qty = pos_ref.quantity; + let old_avg = pos_ref.avg_price; + let new_qty = old_qty + signed_fill_qty; + + // Determine if this fill reduces / closes the position + // A reduction happens when old_qty and signed_fill_qty have opposite signs + let is_reducing = (old_qty > Decimal::ZERO && signed_fill_qty < Decimal::ZERO) + || (old_qty < Decimal::ZERO && signed_fill_qty > Decimal::ZERO); + + if is_reducing { + // Quantity being closed is the smaller of |old_qty| and |signed_fill_qty| + let closed_qty = old_qty.abs().min(order_qty_dec); + // PnL: for a long close: (exit - entry) * qty + // for a short close: (entry - exit) * qty + let pnl = if old_qty > Decimal::ZERO { + (exec_price_dec - old_avg) * closed_qty + } else { + (old_avg - exec_price_dec) * closed_qty + }; + + let entry_side = if old_qty > Decimal::ZERO { + OrderSide::Buy + } else { + OrderSide::Sell + }; + + // Record the round-trip trade + let trade = TradeRecord { + trade_id: Uuid::new_v4().to_string(), + symbol: symbol.clone(), + side: entry_side, + entry_price: Price::from_decimal(old_avg), + exit_price: execution_price, + quantity: Quantity::from_f64( + closed_qty.try_into().unwrap_or(0.0), + ) + .unwrap_or(Quantity::ZERO), + entry_time: pos_ref.created_at, + exit_time: now, + pnl, + return_pct: if old_avg != Decimal::ZERO { + pnl / (old_avg * closed_qty) * Decimal::from(100) + } else { + Decimal::ZERO + }, + commission: Decimal::ZERO, // Commission tracked separately in record_trade + }; + + let mut records = self.trade_records.write().await; + records.push(trade); + + if new_qty == Decimal::ZERO { + // Position fully closed -- archive and remove + let closed_pos = pos_ref.clone(); + drop(pos_ref); + let mut history = self.position_history.write().await; + history.push(closed_pos); + self.positions.remove(symbol); + } else if (new_qty > Decimal::ZERO) != (old_qty > Decimal::ZERO) { + // Position flipped direction -- close old fully, open new with remainder + let remainder_qty = new_qty.abs(); + + // Archive the closed position + let closed_pos = pos_ref.clone(); + // Update to the new flipped position + pos_ref.quantity = new_qty; + pos_ref.avg_price = exec_price_dec; + pos_ref.avg_cost = exec_price_dec; + pos_ref.average_price = exec_price_dec; + pos_ref.basis = remainder_qty * exec_price_dec; + pos_ref.unrealized_pnl = Decimal::ZERO; + pos_ref.realized_pnl += pnl; + pos_ref.updated_at = now; + pos_ref.last_updated = now; + + drop(pos_ref); + let mut history = self.position_history.write().await; + history.push(closed_pos); + } else { + // Partial reduction, same direction + pos_ref.quantity = new_qty; + pos_ref.realized_pnl += pnl; + pos_ref.updated_at = now; + pos_ref.last_updated = now; + } + } else { + // Adding to existing position (same direction) -- compute weighted avg price + let total_cost = old_avg * old_qty.abs() + exec_price_dec * order_qty_dec; + let total_qty = old_qty.abs() + order_qty_dec; + let new_avg = if total_qty != Decimal::ZERO { + total_cost / total_qty + } else { + exec_price_dec + }; + + pos_ref.quantity = new_qty; + pos_ref.avg_price = new_avg; + pos_ref.avg_cost = new_avg; + pos_ref.average_price = new_avg; + pos_ref.basis = total_qty * new_avg; + pos_ref.updated_at = now; + pos_ref.last_updated = now; + } + } else { + // No existing position -- open a new one + let pos = Position::new(symbol.as_str().to_owned(), signed_fill_qty, exec_price_dec); + self.positions.insert(symbol.clone(), pos); + } + Ok(()) } + /// Record a trade fill for audit / reporting. + /// + /// This creates a TradeRecord from the raw fill data and appends it to the + /// internal trade log. For round-trip PnL records, see `update_position`. pub async fn record_trade( &self, - _order: &Order, - _execution_price: Price, - _commission: Decimal, + order: &Order, + execution_price: Price, + commission: Decimal, ) { - // Trade recording logic would be implemented here + let now = Utc::now(); + let trade = TradeRecord { + trade_id: Uuid::new_v4().to_string(), + symbol: order.symbol.clone(), + side: order.side.clone(), + entry_price: execution_price, + exit_price: execution_price, + quantity: order.quantity, + entry_time: order.created_at.to_datetime(), + exit_time: now, + pnl: Decimal::ZERO, // Raw fill; PnL computed in update_position + return_pct: Decimal::ZERO, + commission, + }; + + let mut records = self.trade_records.write().await; + records.push(trade); + } + + /// Get a snapshot of all recorded trades. + pub async fn get_trades(&self) -> Vec { + let records = self.trade_records.read().await; + records.clone() + } + + /// Sum of realized PnL across all recorded round-trip trades. + pub async fn get_realized_pnl(&self) -> Decimal { + let records = self.trade_records.read().await; + records.iter().map(|t| t.pnl).sum() } } @@ -934,4 +1090,230 @@ mod tests { let tester = StrategyTester::new(strategy, config, market_replay, initial_capital); assert_eq!(tester.strategy.name(), "test_strategy"); } + + /// Helper: build a minimal Order for testing position tracker logic. + fn make_test_order(symbol: &str, side: OrderSide, qty_f64: f64) -> Order { + Order { + id: format!("test_{}", Uuid::new_v4()).into(), + client_order_id: None, + broker_order_id: None, + account_id: None, + symbol: Symbol::from(symbol), + side, + order_type: OrderType::Market, + status: OrderStatus::Filled, + time_in_force: TimeInForce::Day, + quantity: Quantity::from_f64(qty_f64).unwrap_or(Quantity::ZERO), + price: None, + stop_price: None, + filled_quantity: Quantity::from_f64(qty_f64).unwrap_or(Quantity::ZERO), + remaining_quantity: Quantity::ZERO, + average_price: None, + avg_fill_price: None, + average_fill_price: None, + exchange_order_id: None, + parent_id: None, + execution_algorithm: None, + execution_params: serde_json::json!({}), + stop_loss: None, + take_profit: None, + created_at: common::HftTimestamp::now_or_zero(), + updated_at: None, + expires_at: None, + metadata: serde_json::json!({}), + } + } + + #[tokio::test] + async fn test_position_tracker_records_trade() { + let tracker = PositionTracker::new(); + let order = make_test_order("AAPL", OrderSide::Buy, 10.0); + let price = Price::from_f64(150.0).unwrap(); + let commission = Decimal::new(15, 1); // 1.5 + + tracker.record_trade(&order, price, commission).await; + + let trades = tracker.get_trades().await; + assert_eq!(trades.len(), 1); + let trade = &trades[0]; + assert_eq!(trade.symbol, Symbol::from("AAPL")); + assert_eq!(trade.side, OrderSide::Buy); + assert_eq!(trade.commission, Decimal::new(15, 1)); + assert_eq!(trade.quantity, order.quantity); + } + + #[tokio::test] + async fn test_position_tracker_update_position_open_and_close() { + let tracker = PositionTracker::new(); + let sym = Symbol::from("AAPL"); + + // Step 1: Buy 10 shares at $100 + let buy_order = make_test_order("AAPL", OrderSide::Buy, 10.0); + let buy_price = Price::from_f64(100.0).unwrap(); + tracker + .update_position(&sym, &buy_order, buy_price) + .await + .unwrap(); + + // Verify the position exists with correct quantity + let positions = tracker.get_all_positions().await; + assert_eq!(positions.len(), 1); + let pos = positions.get(&sym).unwrap(); + assert!(pos.quantity > Decimal::ZERO, "Position should be long"); + + // Step 2: Sell 10 shares at $110 -- closes the position + let sell_order = make_test_order("AAPL", OrderSide::Sell, 10.0); + let sell_price = Price::from_f64(110.0).unwrap(); + tracker + .update_position(&sym, &sell_order, sell_price) + .await + .unwrap(); + + // Position should be removed after full close + let positions = tracker.get_all_positions().await; + assert!( + positions.is_empty(), + "Position should be removed after full close" + ); + + // A round-trip trade record should have been created with correct PnL + let trades = tracker.get_trades().await; + assert_eq!(trades.len(), 1, "One round-trip trade expected"); + let trade = &trades[0]; + // PnL = (110 - 100) * 10 = 100 + assert_eq!(trade.pnl, Decimal::from(100)); + assert_eq!(trade.side, OrderSide::Buy); // entry side was Buy + } + + #[tokio::test] + async fn test_position_tracker_short_position_pnl() { + let tracker = PositionTracker::new(); + let sym = Symbol::from("TSLA"); + + // Sell short 5 shares at $200 + let sell_order = make_test_order("TSLA", OrderSide::Sell, 5.0); + let sell_price = Price::from_f64(200.0).unwrap(); + tracker + .update_position(&sym, &sell_order, sell_price) + .await + .unwrap(); + + let positions = tracker.get_all_positions().await; + let pos = positions.get(&sym).unwrap(); + assert!(pos.quantity < Decimal::ZERO, "Position should be short"); + + // Cover (buy) 5 shares at $180 -- profit on short + let buy_order = make_test_order("TSLA", OrderSide::Buy, 5.0); + let buy_price = Price::from_f64(180.0).unwrap(); + tracker + .update_position(&sym, &buy_order, buy_price) + .await + .unwrap(); + + let positions = tracker.get_all_positions().await; + assert!(positions.is_empty(), "Short position should be closed"); + + let trades = tracker.get_trades().await; + assert_eq!(trades.len(), 1); + // Short PnL = (entry 200 - exit 180) * 5 = 100 + assert_eq!(trades[0].pnl, Decimal::from(100)); + } + + #[tokio::test] + async fn test_position_tracker_get_realized_pnl() { + let tracker = PositionTracker::new(); + + // Trade 1: Buy 10 AAPL at $100, sell at $110 => PnL = +100 + let sym1 = Symbol::from("AAPL"); + let buy1 = make_test_order("AAPL", OrderSide::Buy, 10.0); + let sell1 = make_test_order("AAPL", OrderSide::Sell, 10.0); + tracker + .update_position(&sym1, &buy1, Price::from_f64(100.0).unwrap()) + .await + .unwrap(); + tracker + .update_position(&sym1, &sell1, Price::from_f64(110.0).unwrap()) + .await + .unwrap(); + + // Trade 2: Buy 5 MSFT at $300, sell at $290 => PnL = -50 + let sym2 = Symbol::from("MSFT"); + let buy2 = make_test_order("MSFT", OrderSide::Buy, 5.0); + let sell2 = make_test_order("MSFT", OrderSide::Sell, 5.0); + tracker + .update_position(&sym2, &buy2, Price::from_f64(300.0).unwrap()) + .await + .unwrap(); + tracker + .update_position(&sym2, &sell2, Price::from_f64(290.0).unwrap()) + .await + .unwrap(); + + // Total realized PnL should be 100 + (-50) = 50 + let total_pnl = tracker.get_realized_pnl().await; + assert_eq!(total_pnl, Decimal::from(50)); + + let trades = tracker.get_trades().await; + assert_eq!(trades.len(), 2, "Two round-trip trades expected"); + } + + #[tokio::test] + async fn test_position_tracker_partial_close() { + let tracker = PositionTracker::new(); + let sym = Symbol::from("GOOG"); + + // Buy 20 at $50 + let buy = make_test_order("GOOG", OrderSide::Buy, 20.0); + tracker + .update_position(&sym, &buy, Price::from_f64(50.0).unwrap()) + .await + .unwrap(); + + // Sell 10 at $60 -- partial close + let sell = make_test_order("GOOG", OrderSide::Sell, 10.0); + tracker + .update_position(&sym, &sell, Price::from_f64(60.0).unwrap()) + .await + .unwrap(); + + // 10 shares should remain + let positions = tracker.get_all_positions().await; + assert_eq!(positions.len(), 1); + let pos = positions.get(&sym).unwrap(); + assert_eq!(pos.quantity, Decimal::from(10)); + + // Realized PnL from the closed 10 shares: (60-50)*10 = 100 + let pnl = tracker.get_realized_pnl().await; + assert_eq!(pnl, Decimal::from(100)); + } + + #[tokio::test] + async fn test_position_tracker_add_to_position() { + let tracker = PositionTracker::new(); + let sym = Symbol::from("AMZN"); + + // Buy 10 at $100 + let buy1 = make_test_order("AMZN", OrderSide::Buy, 10.0); + tracker + .update_position(&sym, &buy1, Price::from_f64(100.0).unwrap()) + .await + .unwrap(); + + // Buy 10 more at $120 -- adding to position + let buy2 = make_test_order("AMZN", OrderSide::Buy, 10.0); + tracker + .update_position(&sym, &buy2, Price::from_f64(120.0).unwrap()) + .await + .unwrap(); + + let positions = tracker.get_all_positions().await; + let pos = positions.get(&sym).unwrap(); + assert_eq!(pos.quantity, Decimal::from(20)); + // Weighted avg: (100*10 + 120*10) / 20 = 110 + assert_eq!(pos.avg_price, Decimal::from(110)); + + // No realized PnL yet (no close) + let pnl = tracker.get_realized_pnl().await; + assert_eq!(pnl, Decimal::ZERO); + } } diff --git a/backtesting/tests/dbn_backtest_integration.rs b/backtesting/tests/dbn_backtest_integration.rs new file mode 100644 index 000000000..87abc8539 --- /dev/null +++ b/backtesting/tests/dbn_backtest_integration.rs @@ -0,0 +1,392 @@ +//! End-to-end integration tests for the backtesting vertical slice. +//! +//! Validates the full pipeline: synthetic data -> model loading -> replay -> prediction. +//! No real .dbn files, no PostgreSQL, no trained checkpoints required. + +// Integration tests live outside the crate, so the crate-level deny lints don't apply. +// Tests are allowed to use .unwrap(). + +use backtesting::dbn_converter::processed_to_market_event; +use backtesting::dbn_replay::DbnReplayEngine; +use backtesting::model_loader::{load_models_for_backtest, BacktestMLModel, ModelSpec}; +use backtesting::replay_engine::ReplayEvent; + +use chrono::{DateTime, TimeDelta, Utc}; +use common::{OrderSide, Price, Quantity, Symbol}; +use data::providers::databento::dbn_parser::ProcessedMessage; +use ml::dqn::dqn::DQNConfig; +use ml::ensemble::adapters::dqn::DqnInferenceAdapter; +use ml::features::ProductionFeatureExtractorAdapter; +use ml::{get_global_registry, Features, MLModel, ModelType}; +use rust_decimal::Decimal; +use trading_engine::timing::{HardwareTimestamp, TimingSource}; +use trading_engine::types::events::MarketEvent; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// 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, + } +} + +/// Create a synthetic `MarketEvent::Trade`. +fn make_trade_event(symbol: &str, price_f64: f64, volume: f64, ts: DateTime) -> MarketEvent { + MarketEvent::Trade { + symbol: Symbol::new(symbol.to_string()), + price: Price::from_f64(price_f64).unwrap_or(Price::ZERO), + size: Quantity::from_f64(volume).unwrap_or(Quantity::ZERO), + timestamp: ts, + side: Some(OrderSide::Buy), + venue: None, + trade_id: None, + } +} + +/// Wrap a `MarketEvent` into a `ReplayEvent` with a given sequence and timestamp. +fn make_replay_event(event: MarketEvent, sequence: u64, ts: DateTime) -> ReplayEvent { + ReplayEvent { + event, + original_timestamp: ts, + replay_timestamp: ts, + source_id: "synthetic".to_string(), + sequence, + } +} + +// --------------------------------------------------------------------------- +// Test 1: DBN converter roundtrip +// --------------------------------------------------------------------------- + +#[test] +fn test_dbn_converter_roundtrip() { + let now = Utc::now(); + let hw_ts = ts_from_dt(now); + let price = Price::from_f64(150.25).unwrap(); + + let msg = ProcessedMessage::Trade { + symbol: "AAPL".to_string(), + timestamp: hw_ts, + price, + size: Decimal::new(200, 0), // 200 shares + side: OrderSide::Buy, + trade_id: Some("T42".to_string()), + conditions: vec![], + }; + + let event = processed_to_market_event(&msg); + assert!(event.is_some(), "Trade ProcessedMessage should convert to MarketEvent"); + + let event = event.unwrap(); + if let MarketEvent::Trade { + symbol, + price: p, + size, + side, + trade_id, + .. + } = &event + { + assert_eq!(symbol.as_str(), "AAPL"); + assert_eq!(*p, price); + assert!( + (size.to_f64() - 200.0).abs() < f64::EPSILON, + "size should be 200, got {}", + size.to_f64() + ); + assert_eq!(*side, Some(OrderSide::Buy)); + assert_eq!(*trade_id, Some("T42".to_string())); + } else { + panic!("Expected MarketEvent::Trade variant"); + } +} + +// --------------------------------------------------------------------------- +// Test 2: Replay engine streams events in timestamp order +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn test_replay_engine_streams_events() { + let base = DateTime::from_timestamp(1_700_000_000, 0).unwrap(); + + // Create 10 synthetic events with trending price 100 -> 109 + let events: Vec = (0..10) + .map(|i| { + let ts = base + TimeDelta::seconds(i as i64); + let price = 100.0 + i as f64; + let event = make_trade_event("SYNTH", price, 50.0, ts); + make_replay_event(event, i as u64, ts) + }) + .collect(); + + let engine = DbnReplayEngine::from_events(events); + assert_eq!(engine.event_count(), 10); + + let mut rx = engine.start_replay(); + + let mut received = Vec::new(); + while let Some(ev) = rx.recv().await { + received.push(ev); + } + + assert_eq!(received.len(), 10, "Should receive all 10 events"); + + // Verify chronological order + for pair in received.windows(2) { + let a = pair.first().unwrap(); + let b = pair.get(1).unwrap(); + assert!( + a.original_timestamp <= b.original_timestamp, + "Events not in chronological order: {:?} > {:?}", + a.original_timestamp, + b.original_timestamp, + ); + } + + // Verify price trend + if let MarketEvent::Trade { price: first_price, .. } = &received.first().unwrap().event { + if let MarketEvent::Trade { price: last_price, .. } = &received.last().unwrap().event { + assert!( + last_price > first_price, + "Price should trend upward: first={}, last={}", + first_price, last_price + ); + } + } +} + +// --------------------------------------------------------------------------- +// Test 3: DQN model loads with random weights and predicts +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn test_model_loads_and_predicts() { + let config = DQNConfig { + state_dim: 51, + num_actions: 45, + hidden_dims: vec![64, 64], + ..Default::default() + }; + + let adapter = DqnInferenceAdapter::new(config).unwrap(); + let model = BacktestMLModel::new(Box::new(adapter), ModelType::DQN); + + assert!(model.is_ready(), "Model should be ready after creation"); + assert_eq!(model.name(), "DQN"); + assert_eq!(model.model_type(), ModelType::DQN); + + // Build a 51-dim feature vector + let features = Features::new( + vec![0.1; 51], + (0..51).map(|i| format!("f{i}")).collect(), + ); + + let prediction = model.predict(&features).await.unwrap(); + + // Direction must be in [-1, 1] + assert!( + prediction.value >= -1.0 && prediction.value <= 1.0, + "direction {} out of [-1, 1]", + prediction.value, + ); + // Confidence must be in [0, 1] + assert!( + prediction.confidence >= 0.0 && prediction.confidence <= 1.0, + "confidence {} out of [0, 1]", + prediction.confidence, + ); + // Model id should be "DQN" + assert_eq!(prediction.model_id, "DQN"); +} + +// --------------------------------------------------------------------------- +// Test 4: Production feature extractor produces exactly 51 dimensions +// --------------------------------------------------------------------------- + +#[test] +fn test_production_features_51_dim() { + use common::ml_strategy::ProductionFeatureExtractor225; + + let mut extractor = ProductionFeatureExtractorAdapter::new(); + + // Feed 55 price updates (past the warmup period of 50) + for i in 0..55 { + let price = 100.0 + i as f64 * 0.1; + let volume = 1000.0; + let timestamp = Utc::now(); + extractor + .update(price, volume, timestamp) + .unwrap_or_else(|e| { + panic!("Production extractor update failed at step {}: {}", i, e) + }); + } + + let features = extractor + .extract_features() + .unwrap_or_else(|e| panic!("extract_features failed: {}", e)); + + assert_eq!( + features.len(), + 51, + "Production extractor should produce exactly 51 features, got {}", + features.len() + ); + + // Validate all features are finite + for (i, val) in features.iter().enumerate() { + assert!( + val.is_finite(), + "Feature {} should be finite, found {}", + i, val + ); + } +} + +// --------------------------------------------------------------------------- +// Test 5: Full pipeline -- events -> feature extractor -> model prediction +// --------------------------------------------------------------------------- + +#[tokio::test] +async fn test_full_pipeline_synthetic_backtest() { + use common::ml_strategy::ProductionFeatureExtractor225; + + // --- Step 1: Register a DQN model in the global registry --- + let specs = vec![ModelSpec { + model_type: "DQN".to_string(), + checkpoint_path: String::new(), // random weights + weight: 1.0, + }]; + + load_models_for_backtest(&specs) + .await + .unwrap_or_else(|e| panic!("load_models_for_backtest failed: {}", e)); + + let registry = get_global_registry(); + let model = registry.get("DQN").await; + assert!(model.is_some(), "DQN should be registered in the global registry"); + + // --- Step 2: Create 100 synthetic ReplayEvents with trending prices --- + let base = DateTime::from_timestamp(1_700_000_000, 0).unwrap(); + let events: Vec = (0..100) + .map(|i| { + let ts = base + TimeDelta::seconds(i as i64); + let price = 100.0 + (i as f64); // 100 -> 199 + let volume = 500.0 + (i as f64) * 2.0; + let event = make_trade_event("SYNTH", price, volume, ts); + make_replay_event(event, i as u64, ts) + }) + .collect(); + + // --- Step 3: Create replay engine and stream events --- + let engine = DbnReplayEngine::from_events(events); + assert_eq!(engine.event_count(), 100); + + let mut rx = engine.start_replay(); + + // --- Step 4: Consume events, update feature extractor, run predictions --- + let mut extractor = ProductionFeatureExtractorAdapter::new(); + let mut events_processed: u64 = 0; + let mut predictions_made: u64 = 0; + let warmup_period: u64 = 55; // extractor needs ~50 bars for warmup + + while let Some(replay_event) = rx.recv().await { + events_processed += 1; + + // Extract price and volume from the MarketEvent + if let MarketEvent::Trade { + price, size, timestamp, .. + } = &replay_event.event + { + let price_f64 = price.to_decimal().unwrap_or(Decimal::ZERO); + let volume_f64 = size.to_f64(); + + // Update the feature extractor + let price_f64_val: f64 = price_f64.try_into().unwrap_or(0.0); + let _ = extractor.update(price_f64_val, volume_f64, *timestamp); + + // After warmup, extract features and predict + if events_processed >= warmup_period { + match extractor.extract_features() { + Ok(feat_values) if feat_values.len() == 51 => { + // Verify exactly 51 dimensions + assert_eq!(feat_values.len(), 51, "Feature vector should be 51-dim"); + + let feature_names: Vec = (0..51) + .map(|i| format!("prod_feature_{}", i)) + .collect(); + + let features = Features { + values: feat_values, + names: feature_names, + timestamp: timestamp.timestamp_micros() as u64, + symbol: Some("SYNTH".to_string()), + }; + + // Run prediction through the global registry + let results = registry + .predict_selected(&["DQN".to_string()], &features) + .await; + + for result in results { + match result { + Ok(pred) => { + predictions_made += 1; + // Validate prediction bounds + assert!( + pred.value >= -1.0 && pred.value <= 1.0, + "Prediction value {} out of [-1, 1]", + pred.value + ); + assert!( + pred.confidence >= 0.0 && pred.confidence <= 1.0, + "Prediction confidence {} out of [0, 1]", + pred.confidence + ); + } + Err(e) => { + panic!("DQN prediction failed: {:?}", e); + } + } + } + } + Ok(feat) => { + // Feature extractor not yet warmed up, fewer than 51 features + eprintln!( + "Event {}: extractor returned {} features (expected 51, still warming up)", + events_processed, + feat.len() + ); + } + Err(e) => { + // Expected during warmup + eprintln!( + "Event {}: extractor not ready yet: {}", + events_processed, e + ); + } + } + } + } + } + + // --- Step 5: Verify results --- + assert_eq!(events_processed, 100, "Should have processed all 100 events"); + assert!( + predictions_made > 0, + "Should have made at least one prediction after warmup (made {})", + predictions_made, + ); + + println!( + "Full pipeline test passed: {} events processed, {} predictions made", + events_processed, predictions_made + ); +}