diff --git a/services/trading_service/tests/e2e_pipeline_test.rs b/services/trading_service/tests/e2e_pipeline_test.rs new file mode 100644 index 000000000..1c092e253 --- /dev/null +++ b/services/trading_service/tests/e2e_pipeline_test.rs @@ -0,0 +1,423 @@ +//! End-to-end pipeline integration test: OHLCV -> Features -> Ensemble -> Prediction +//! +//! Proves the critical trading pipeline path works with all 10 real candle +//! inference adapters (DQN, PPO, TFT, Mamba2, Liquid-CfC, TGGN, TLOB, KAN, +//! xLSTM, Diffusion) using random (untrained) weights. +//! +//! No external dependencies: no DB, no Docker, no network. + +use std::sync::Arc; + +use chrono::Utc; +use trading_service::adapter_bridge::InferenceAdapterBridge; +use trading_service::ensemble_coordinator::EnsembleCoordinator; + +/// Register all 10 inference adapters into the coordinator. +/// +/// This mirrors the exact construction pattern from `main.rs` in the +/// trading_service binary so the test exercises the real production path. +async fn register_all_10_adapters(coordinator: &EnsembleCoordinator) -> usize { + let mut registered = 0usize; + + // 1. DQN adapter (51-dim input, 45 factored actions) + match ml::ensemble::adapters::DqnInferenceAdapter::new(ml::dqn::dqn::DQNConfig { + state_dim: 51, + num_actions: 45, + hidden_dims: vec![64, 64], + ..Default::default() + }) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "DQN".to_string(), + ml::ModelType::DQN, + )); + if coordinator + .register_loaded_model("DQN".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create DQN adapter: {}", e), + } + + // 2. PPO adapter (51-dim input, 3 actions: buy/sell/hold) + match ml::ensemble::adapters::PpoInferenceAdapter::new(ml::ppo::ppo::PPOConfig { + state_dim: 51, + num_actions: 3, + policy_hidden_dims: vec![64, 64], + value_hidden_dims: vec![64, 64], + ..Default::default() + }) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "PPO".to_string(), + ml::ModelType::PPO, + )); + if coordinator + .register_loaded_model("PPO".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create PPO adapter: {}", e), + } + + // 3. TFT adapter (51-dim split: 3 static + 8 known + 40 unknown) + match ml::ensemble::adapters::TftInferenceAdapter::new( + ml::tft::TFTConfig { + input_dim: 51, + hidden_dim: 32, + num_heads: 4, + num_layers: 1, + prediction_horizon: 1, + sequence_length: 10, + num_quantiles: 3, + num_static_features: 3, + num_known_features: 8, + num_unknown_features: 40, + ..Default::default() + }, + 10, + ) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "TFT".to_string(), + ml::ModelType::TFT, + )); + if coordinator + .register_loaded_model("TFT".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create TFT adapter: {}", e), + } + + // 4. Mamba2 adapter (d_model=64, zero-pads from 51-dim features) + match ml::ensemble::adapters::Mamba2InferenceAdapter::new( + ml::mamba::Mamba2Config { + d_model: 64, + d_state: 16, + d_head: 16, + num_heads: 2, + expand: 1, + num_layers: 1, + ..Default::default() + }, + 10, + ) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "MAMBA2".to_string(), + ml::ModelType::MAMBA, + )); + if coordinator + .register_loaded_model("MAMBA2".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create MAMBA2 adapter: {}", e), + } + + // 5. Liquid CfC adapter (51-dim input, 3 output: buy/hold/sell) + match ml::ensemble::adapters::LiquidInferenceAdapter::new( + ml::liquid::candle_cfc::CfCTrainConfig { + input_size: 51, + hidden_size: 32, + output_size: 3, + backbone_hidden_sizes: vec![32], + seq_len: 1, + device: ml::liquid::candle_cfc::DeviceConfig::Cpu, + ..ml::liquid::candle_cfc::CfCTrainConfig::default() + }, + ) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "Liquid-CfC".to_string(), + ml::ModelType::LNN, + )); + if coordinator + .register_loaded_model("Liquid-CfC".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create Liquid-CfC adapter: {}", e), + } + + // 6. TGGN adapter (51-dim input, 2-layer candle projection) + match ml::ensemble::adapters::TggnInferenceAdapter::new(51, 64) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "TGGN".to_string(), + ml::ModelType::TGGN, + )); + if coordinator + .register_loaded_model("TGGN".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create TGGN adapter: {}", e), + } + + // 7. TLOB adapter (51-dim features, 3-layer MLP, seq_len=10) + match ml::ensemble::adapters::TlobInferenceAdapter::new(51, 64, 10) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "TLOB".to_string(), + ml::ModelType::TLOB, + )); + if coordinator + .register_loaded_model("TLOB".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create TLOB adapter: {}", e), + } + + // 8. KAN adapter (51-dim input, B-spline activations) + match ml::ensemble::adapters::KanInferenceAdapter::new(ml::kan::KANConfig { + layer_widths: vec![51, 32, 16, 1], + ..Default::default() + }) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "KAN".to_string(), + ml::ModelType::KAN, + )); + if coordinator + .register_loaded_model("KAN".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create KAN adapter: {}", e), + } + + // 9. xLSTM adapter (51-dim input, sLSTM+mLSTM, seq_len=10) + match ml::ensemble::adapters::XlstmInferenceAdapter::new( + ml::xlstm::XLSTMConfig { + input_dim: 51, + hidden_dim: 64, + num_blocks: 2, + num_heads: 2, + ..Default::default() + }, + 10, + ) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "xLSTM".to_string(), + ml::ModelType::XLSTM, + )); + if coordinator + .register_loaded_model("xLSTM".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create xLSTM adapter: {}", e), + } + + // 10. Diffusion adapter (51-dim features, DDPM denoiser, t=1 inference) + match ml::ensemble::adapters::DiffusionInferenceAdapter::new(ml::diffusion::DiffusionConfig { + seq_len: 1, + feature_dim: 51, + hidden_dim: 64, + num_layers: 2, + time_embed_dim: 32, + num_timesteps: 100, + sampling_steps: 5, + ..Default::default() + }) { + Ok(adapter) => { + let bridge = Arc::new(InferenceAdapterBridge::new( + Box::new(adapter), + "Diffusion".to_string(), + ml::ModelType::Diffusion, + )); + if coordinator + .register_loaded_model("Diffusion".to_string(), bridge, 0.10) + .await + .is_ok() + { + registered += 1; + } + } + Err(e) => eprintln!("Failed to create Diffusion adapter: {}", e), + } + + registered +} + +/// End-to-end pipeline test: OHLCV -> features -> ensemble (10 models) -> prediction. +/// +/// This test proves the critical trading pipeline works in-process with no +/// external dependencies. All 10 candle-backed inference adapters are +/// initialised with random weights and fed synthetic market data. +#[tokio::test(flavor = "multi_thread")] +async fn test_e2e_ohlcv_to_prediction_10_models() { + // ---- Step 1: Create ensemble coordinator and register all 10 models ---- + let coordinator = EnsembleCoordinator::new(); + + let registered = register_all_10_adapters(&coordinator).await; + let model_count = coordinator.model_count().await; + + println!( + "[E2E] Registered {}/{} adapters (coordinator reports {} models)", + registered, 10, model_count, + ); + assert_eq!( + model_count, 10, + "All 10 inference adapters must be registered" + ); + + // ---- Step 2: Feed 60 synthetic OHLCV bars to warm up feature extractors ---- + let symbol = "ES.FUT"; + let base_time = Utc::now(); + let base_price = 4500.0_f64; + + for i in 0..60 { + // Gentle uptrend with small noise + let price = base_price + (i as f64) * 0.25; + let volume = 10_000.0 + (i as f64) * 50.0; + let ts = base_time + chrono::Duration::seconds(i * 60); + + coordinator + .update_market_data(symbol, price, volume, ts) + .await + .unwrap_or_else(|e| panic!("update_market_data failed at bar {}: {}", i, e)); + } + println!("[E2E] Fed 60 OHLCV bars for {}", symbol); + + // ---- Step 3: Extract features (should be real 51-dim after warmup) ---- + let features = coordinator + .fetch_features_for_symbol(symbol) + .await + .unwrap_or_else(|e| panic!("fetch_features_for_symbol failed: {}", e)); + + assert_eq!( + features.values.len(), + 51, + "Feature vector must be 51-dimensional" + ); + assert_eq!(features.names.len(), 51, "Feature names must be 51"); + assert_eq!(features.symbol.as_deref(), Some(symbol)); + + // After warmup, at least some features should be non-zero + let non_zero = features.values.iter().filter(|v| v.abs() > 1e-12).count(); + println!( + "[E2E] Feature extraction: {}/51 non-zero features", + non_zero + ); + assert!( + non_zero > 5, + "Expected many non-zero features after 60-bar warmup, got {}", + non_zero, + ); + + // All features must be finite + for (idx, &val) in features.values.iter().enumerate() { + assert!( + val.is_finite(), + "Feature {} ({}) is NaN/Inf: {}", + idx, + features.names.get(idx).map(|s| s.as_str()).unwrap_or("?"), + val, + ); + } + + // ---- Step 4: Run ensemble prediction ---- + let decision = coordinator + .predict(&features) + .await + .unwrap_or_else(|e| panic!("Ensemble prediction failed: {}", e)); + + println!( + "[E2E] Ensemble decision: action={:?}, confidence={:.4}, signal={:.4}, \ + disagreement={:.4}, model_votes={}", + decision.action, + decision.confidence, + decision.signal, + decision.disagreement_rate, + decision.model_count(), + ); + + // Print per-model votes for debugging + for (model_id, vote) in &decision.model_votes { + println!( + " [{}] signal={:.4}, confidence={:.4}, weight={:.4}", + model_id, vote.signal, vote.confidence, vote.weight, + ); + } + + // ---- Step 5: Assertions ---- + + // Confidence must be in [0.0, 1.0] + assert!( + decision.confidence >= 0.0 && decision.confidence <= 1.0, + "Confidence {:.4} must be in [0.0, 1.0]", + decision.confidence, + ); + + // Action must be Buy, Sell, or Hold (exhaustive match proves it) + match decision.action { + ml::ensemble::TradingAction::Buy + | ml::ensemble::TradingAction::Sell + | ml::ensemble::TradingAction::Hold => {} + } + + // Model count must be 10 (all models contributed a vote) + assert_eq!( + decision.model_count(), + 10, + "All 10 models should have contributed votes, got {}", + decision.model_count(), + ); + + // Signal must be finite + assert!( + decision.signal.is_finite(), + "Signal must be finite, got {}", + decision.signal, + ); + + // Disagreement rate must be in [0.0, 1.0] + assert!( + decision.disagreement_rate >= 0.0 && decision.disagreement_rate <= 1.0, + "Disagreement rate {:.4} must be in [0.0, 1.0]", + decision.disagreement_rate, + ); + + println!("[E2E] All assertions passed -- pipeline is wired end-to-end!"); +}