diff --git a/backtesting/src/strategy_runner.rs b/backtesting/src/strategy_runner.rs index f68fe1db7..53aaf9f32 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}; @@ -40,8 +42,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, } @@ -685,6 +689,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 +697,7 @@ impl AdaptiveStrategyRunner { predictions_cache: Arc::new(DashMap::new()), performance_tracker: Arc::new(RwLock::new(PerformanceTracker::default())), feature_extractor, + production_extractor, risk_manager, } } @@ -906,40 +912,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); - }, + } } } }, @@ -1121,4 +1189,68 @@ 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); + } }