diff --git a/ml/src/training/unified_data_loader.rs b/ml/src/training/unified_data_loader.rs index d419e0f70..e904893f5 100644 --- a/ml/src/training/unified_data_loader.rs +++ b/ml/src/training/unified_data_loader.rs @@ -16,10 +16,8 @@ use serde::{Deserialize, Serialize}; use tokio::sync::RwLock; use tracing::{debug, info}; -// REMOVED: These types don't exist in ml::features, they're in the data crate -// TODO: Re-enable when data crate exports are fixed -// use data::unified_feature_extractor::{UnifiedFeatureExtractor, UnifiedFinancialFeatures}; - +use crate::features::extraction::{extract_ml_features, OHLCVBar}; +use crate::features::unified::{FeatureExtractionConfig, UnifiedFeatureExtractor}; use crate::safety::MLSafetyManager; use crate::{MLError, MLResult}; use common::types::{Price, Symbol, Volume}; @@ -30,14 +28,6 @@ pub struct OrderLevel { pub quantity: f64, } -// Temporary placeholder until data crate integration is complete -#[derive(Debug, Clone)] -pub struct UnifiedFinancialFeatures { - pub symbol: Symbol, - pub timestamp: DateTime, - pub features: Vec, -} - /// Configuration for the unified data loader #[derive(Debug, Clone, Serialize, Deserialize)] pub struct UnifiedDataLoaderConfig { @@ -213,8 +203,7 @@ pub struct TrainingDataset { /// Individual training sample #[derive(Debug, Clone, Serialize, Deserialize)] pub struct TrainingSample { - /// Input features using UnifiedFinancialFeatures - /// TODO: Replace with actual feature type when available + /// Input features (51-dim from extract_ml_features, or 2-dim fallback) pub features: Vec, /// Target values for supervised learning pub targets: Vec, @@ -270,8 +259,8 @@ pub struct DatasetStatistics { #[derive(Debug)] pub struct UnifiedDataLoader { config: UnifiedDataLoaderConfig, - /// TODO: Replace with actual feature extractor when available - _feature_extractor_placeholder: (), + /// Unified feature extractor for consistent training/serving feature vectors + feature_extractor: Option, safety_manager: Arc, databento_provider: DatabentoHistoricalProvider, benzinga_provider: BenzingaHistoricalProvider, @@ -368,9 +357,18 @@ impl UnifiedDataLoader { let databento_provider = DatabentoHistoricalProvider::new(config.databento_config.clone())?; let benzinga_provider = BenzingaHistoricalProvider::new(config.benzinga_config.clone())?; + let feature_extractor = if config.feature_extraction.use_unified_extractor { + Some(UnifiedFeatureExtractor::new( + FeatureExtractionConfig::default(), + Arc::clone(&safety_manager), + )) + } else { + None + }; + Ok(Self { config, - _feature_extractor_placeholder: (), + feature_extractor, safety_manager, databento_provider, benzinga_provider, @@ -493,36 +491,93 @@ impl UnifiedDataLoader { Ok(merged_data) } - /// Create training samples using UnifiedFeatureExtractor + /// Create training samples using the full 51-dimension feature extractor async fn create_training_samples( &self, containers: Vec, ) -> MLResult> { - let mut samples = Vec::new(); + if containers.is_empty() { + return Ok(Vec::new()); + } - for container in containers { - // TODO: Implement actual feature extraction when UnifiedFeatureExtractor is available - // For now, create placeholder features from price data + // Convert MarketDataContainers to OHLCVBars for feature extraction + let bars: Vec = containers + .iter() + .map(|c| OHLCVBar { + timestamp: c.timestamp, + open: c.price_data.open.as_f64(), + high: c.price_data.high.as_f64(), + low: c.price_data.low.as_f64(), + close: c.price_data.close.as_f64(), + volume: c.volume_data.volume.as_f64(), + }) + .collect(); + + // Use full feature extractor when available, otherwise fall back to basic features + if self.feature_extractor.is_some() { + match extract_ml_features(&bars) { + Ok(feature_vectors) => { + // extract_ml_features skips a warmup period, so feature_vectors.len() + // may be less than containers.len(). Align from the end so each + // feature vector corresponds to the correct container. + let offset = containers.len().saturating_sub(feature_vectors.len()); + let mut samples = Vec::with_capacity(feature_vectors.len()); + + for (fv, container) in + feature_vectors.iter().zip(containers.get(offset..).unwrap_or(&[])) + { + let features: Vec = fv.to_vec(); + let targets = self.create_target_values(container)?; + + samples.push(TrainingSample { + features, + targets, + timestamp: container.timestamp, + symbol: container.symbol.clone(), + weight: 1.0, + metadata: container.metadata.clone(), + }); + } + + info!( + "Created {} training samples with 51-dim features ({} bars skipped for warmup)", + samples.len(), + offset, + ); + return Ok(samples); + } + Err(e) => { + debug!( + "Full feature extraction failed ({}), falling back to basic features", + e + ); + } + } + } + + // Fallback: basic 2-feature extraction (close price, volume) + let mut samples = Vec::with_capacity(containers.len()); + for container in &containers { let features = vec![ container.price_data.close.as_f64(), container.volume_data.volume.as_f64(), ]; + let targets = self.create_target_values(container)?; - // Create target values (example: next price direction) - let targets = self.create_target_values(&container)?; - - let sample = TrainingSample { + samples.push(TrainingSample { features, targets, timestamp: container.timestamp, - symbol: container.symbol, + symbol: container.symbol.clone(), weight: 1.0, - metadata: container.metadata, - }; - - samples.push(sample); + metadata: container.metadata.clone(), + }); } + info!( + "Created {} training samples with basic 2-dim features (fallback)", + samples.len(), + ); Ok(samples) } @@ -534,11 +589,38 @@ impl UnifiedDataLoader { } /// Generate feature metadata - fn generate_feature_metadata(&self, _samples: &[TrainingSample]) -> MLResult { - // This would analyze the features and create metadata + fn generate_feature_metadata(&self, samples: &[TrainingSample]) -> MLResult { + let dim = samples.first().map_or(0, |s| s.features.len()); + let feature_names = if self.feature_extractor.is_some() && dim == 51 { + // 51-dimension feature names matching extraction.rs ordering + vec![ + "open", "high", "low", "close", "volume", + "returns", "log_returns", "high_low_range", "close_open_range", + "ema_5", "ema_10", "ema_20", "ema_50", + "rsi_14", "macd_line", "macd_signal", "macd_histogram", + "bb_upper", "bb_middle", "bb_lower", "bb_width", "bb_pctb", + "atr_14", + "higher_high", "lower_low", "inside_bar", + "vol_sma_20", "vol_ratio", "vol_momentum", + "hour_sin", "hour_cos", "dow_sin", "dow_cos", + "std_20", "skewness_20", "kurtosis_20", "z_score", + "roll_measure", "amihud_illiq", "corwin_schultz", + "regime_vol", "regime_trend", "regime_mean_rev", + "ofi_1", "ofi_2", "ofi_3", "ofi_4", + "ofi_5", "ofi_6", "ofi_7", "ofi_8", + ] + .into_iter() + .map(String::from) + .collect() + } else { + vec!["close".to_string(), "volume".to_string()] + }; + + let feature_types = vec!["continuous".to_string(); feature_names.len()]; + Ok(FeatureMetadata { - feature_names: vec![], // Would be populated from UnifiedFeatureExtractor - feature_types: vec![], + feature_names, + feature_types, feature_statistics: HashMap::new(), normalization_params: HashMap::new(), })