feat(ml): wire unified data loader to feature extractor

Replace the 2-feature placeholder (close, volume) in create_training_samples()
with the full 51-dimension extract_ml_features() pipeline. The UnifiedDataLoader
now converts MarketDataContainers to OHLCVBars and runs the production feature
extractor when use_unified_extractor is enabled, falling back to the basic
2-feature path if extraction fails or the extractor is disabled. Also populates
feature metadata with the 51 named features matching extraction.rs ordering.

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<Utc>,
-    pub features: Vec<f64>,
-}
-
 /// 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<f64>,
     /// Target values for supervised learning
     pub targets: Vec<f64>,
@@ -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<UnifiedFeatureExtractor>,
     safety_manager: Arc<MLSafetyManager>,
     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<MarketDataContainer>,
     ) -> MLResult<Vec<TrainingSample>> {
-        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<OHLCVBar> = 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<f64> = 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<FeatureMetadata> {
-        // This would analyze the features and create metadata
+    fn generate_feature_metadata(&self, samples: &[TrainingSample]) -> MLResult<FeatureMetadata> {
+        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(),
         })