test(trading_service): add E2E pipeline integration test (10 models)
Proves the critical trading pipeline path works end-to-end: OHLCV -> features -> ensemble -> prediction. Uses all 10 real candle inference adapters (DQN, PPO, TFT, Mamba2, Liquid-CfC, TGGN, TLOB, KAN, xLSTM, Diffusion) with random weights, 60 synthetic OHLCV bars, and asserts confidence/action/model-count invariants. No external dependencies (no DB, no Docker). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
423
services/trading_service/tests/e2e_pipeline_test.rs
Normal file
423
services/trading_service/tests/e2e_pipeline_test.rs
Normal file
@@ -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!");
|
||||
}
|
||||
Reference in New Issue
Block a user