//! Unified Feature Extractor //! //! Consistent feature engineering across training, trading, and backtesting systems. //! Integrates market data from Databento and news events from Benzinga to create //! comprehensive feature vectors for ML model training and inference. use crate::error::Result; use crate::features::{ FeatureCategory, FeatureMetadata, FeatureVector, MicrostructureAnalyzer, PortfolioAnalyzer, PricePoint, RegimeDetector, TechnicalIndicators, TemporalFeatures, }; use crate::providers::common::NewsEvent; use chrono::{DateTime, Duration, Utc}; use common::MarketDataEvent; use config::data_config::{ DataMicrostructureConfig as MicrostructureConfig, DataRegimeDetectionConfig as RegimeDetectionConfig, DataTechnicalIndicatorsConfig as TechnicalIndicatorsConfig, TrainingFeatureEngineeringConfig as FeatureEngineeringConfig, }; use num_traits::ToPrimitive; use serde::{Deserialize, Serialize}; use std::collections::{BTreeMap, HashMap, VecDeque}; use std::sync::Arc; use tokio::sync::RwLock; use tracing::info; /// Unified feature extraction configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct UnifiedFeatureExtractorConfig { /// Feature engineering configuration pub feature_config: FeatureEngineeringConfig, /// News analysis configuration pub news_config: NewsAnalysisConfig, /// Feature aggregation settings pub aggregation: AggregationConfig, /// Output configuration pub output: OutputConfig, } /// News analysis configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct NewsAnalysisConfig { /// Enable sentiment analysis pub sentiment_analysis: bool, /// News impact window (minutes) pub impact_window_minutes: u32, /// Minimum importance threshold (0.0-1.0) pub min_importance: f64, /// News categories to include pub categories: Vec, /// Weight different news types pub news_type_weights: HashMap, /// Enable event clustering pub event_clustering: bool, /// Maximum news events per symbol per period pub max_events_per_period: u32, } /// Feature aggregation configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct AggregationConfig { /// Primary timeframe for features (minutes) pub primary_timeframe_minutes: u32, /// Secondary timeframes for multi-scale features pub secondary_timeframes: Vec, /// Lookback periods for historical features pub lookback_periods: Vec, /// Enable cross-symbol features pub cross_symbol_features: bool, /// Maximum symbols for cross-correlation pub max_correlation_symbols: u32, /// Maximum market data buffer size per symbol pub max_buffer_size: usize, } /// Output configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct OutputConfig { /// Include feature metadata pub include_metadata: bool, /// Feature scaling method pub scaling_method: ScalingMethod, /// Handle missing values pub missing_value_strategy: MissingValueStrategy, /// Feature selection criteria pub feature_selection: FeatureSelectionConfig, } /// Feature scaling methods #[derive(Debug, Clone, Serialize, Deserialize)] pub enum ScalingMethod { /// No scaling None, /// Min-max normalization MinMax, /// Z-score standardization StandardScore, /// Robust scaling (median and IQR) Robust, /// Quantile transformation Quantile, } /// Missing value handling strategies #[derive(Debug, Clone, Serialize, Deserialize)] pub enum MissingValueStrategy { /// Forward fill ForwardFill, /// Backward fill BackwardFill, /// Linear interpolation Interpolate, /// Use zero/neutral values Zero, /// Use mean values Mean, /// Drop incomplete records Drop, } /// Feature selection configuration #[derive(Debug, Clone, Serialize, Deserialize)] pub struct FeatureSelectionConfig { /// Enable feature selection pub enabled: bool, /// Maximum number of features pub max_features: Option, /// Minimum correlation threshold pub min_correlation: f64, /// Maximum correlation for removal pub max_correlation: f64, /// Feature importance threshold pub importance_threshold: f64, } /// Unified feature extractor pub struct UnifiedFeatureExtractor { /// Configuration config: UnifiedFeatureExtractorConfig, /// Technical indicators calculator technical_indicators: Arc>, /// Microstructure analyzer microstructure: Arc>, /// Regime detector regime_detector: Arc>, /// Portfolio analyzer portfolio_analyzer: Arc>, /// News event buffer news_buffer: Arc>>>, /// Market data buffer market_data_buffer: Arc>>>, /// Feature cache feature_cache: Arc>>, /// Feature statistics for scaling and imputation feature_stats: Arc>>, } /// Cached feature vector with timestamp #[derive(Debug, Clone)] pub struct CachedFeatureVector { /// Feature vector pub features: FeatureVector, /// Cache timestamp pub cached_at: DateTime, /// Time to live (minutes) pub ttl_minutes: u32, } /// Multi-modal feature set combining market and news data #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MultiModalFeatures { /// Market-based features pub market_features: HashMap, /// News-based features pub news_features: HashMap, /// Cross-modal features (market-news interactions) pub cross_modal_features: HashMap, /// Temporal features pub temporal_features: HashMap, /// Regime features pub regime_features: HashMap, } /// News impact analysis result #[derive(Debug, Clone)] pub struct NewsImpactAnalysis { /// Symbol pub symbol: String, /// Analysis timestamp pub timestamp: DateTime, /// Overall sentiment score (-1.0 to 1.0) pub overall_sentiment: f64, /// News volume (number of events) pub news_volume: u32, /// Average importance pub avg_importance: f64, /// Event type distribution pub event_type_distribution: HashMap, /// Recent high-impact events pub recent_events: Vec, } /// Price reaction to news events #[derive(Debug, Clone)] pub struct PriceReaction { /// Average price reaction (percentage change) pub avg_reaction: f64, /// Volatility of price reactions pub volatility: f64, /// Direction of reactions (-1: mostly negative, 0: mixed, 1: mostly positive) pub direction: f64, } /// Running statistics for a feature #[derive(Debug, Clone)] pub struct FeatureStats { /// Running mean pub mean: f64, /// Running variance (for standard deviation calculation) pub variance: f64, /// Minimum value seen pub min: f64, /// Maximum value seen pub max: f64, /// Sample count pub count: usize, /// Last observed value (for forward fill) pub last_value: Option, } impl Default for UnifiedFeatureExtractorConfig { fn default() -> Self { Self { feature_config: FeatureEngineeringConfig { enable_normalization: true, enable_scaling: true, enable_log_returns: true, lookback_window: 100, technical_indicators: TechnicalIndicatorsConfig { enable_moving_averages: true, enable_momentum: true, enable_volatility: true, window_sizes: vec![5, 10, 20, 50, 200], ma_periods: vec![5, 10, 20, 50, 200], rsi_periods: vec![14, 21], bollinger_periods: vec![20], macd: config::data_config::DataMACDConfig { fast_period: 12, slow_period: 26, signal_period: 9, enabled: true, }, }, microstructure: MicrostructureConfig { enable_bid_ask_spread: true, enable_order_flow: true, tick_size: 0.01, lot_size: 100.0, bid_ask_spread: true, volume_imbalance: true, price_impact: true, kyle_lambda: true, amihud_ratio: true, }, // tlob config moved to microstructure section // temporal config not part of TrainingFeatureEngineeringConfig // temporal: TemporalConfig { // enable_time_features: true, // enable_seasonal: true, // market_session: true, // holiday_effects: true, // expiration_effects: true, // }, regime_detection: RegimeDetectionConfig { enable_hmm: true, enable_clustering: true, window_size: 50, n_states: 3, volatility_regime: true, trend_regime: true, volume_regime: true, correlation_regime: true, lookback_period: 100, }, }, news_config: NewsAnalysisConfig { sentiment_analysis: true, impact_window_minutes: 60, min_importance: 0.3, categories: vec![ "Earnings".to_string(), "Analyst Rating".to_string(), "Breaking".to_string(), "FDA".to_string(), "M&A".to_string(), ], news_type_weights: { let mut weights = HashMap::new(); weights.insert("Earnings".to_string(), 1.0); weights.insert("Rating".to_string(), 0.8); weights.insert("News".to_string(), 0.6); weights.insert("Economic".to_string(), 0.4); weights }, event_clustering: true, max_events_per_period: 10, }, aggregation: AggregationConfig { primary_timeframe_minutes: 1, secondary_timeframes: vec![5, 15, 60], lookback_periods: vec![10, 50, 200], cross_symbol_features: true, max_correlation_symbols: 20, max_buffer_size: 10000, }, output: OutputConfig { include_metadata: true, scaling_method: ScalingMethod::StandardScore, missing_value_strategy: MissingValueStrategy::ForwardFill, feature_selection: FeatureSelectionConfig { enabled: true, max_features: Some(1000), min_correlation: 0.01, max_correlation: 0.95, importance_threshold: 0.001, }, }, } } } impl UnifiedFeatureExtractor { /// Create a new unified feature extractor pub fn new(config: UnifiedFeatureExtractorConfig) -> Result { info!("Initializing unified feature extractor"); let technical_indicators = Arc::new(RwLock::new(TechnicalIndicators::new( config.feature_config.technical_indicators.clone(), ))); let microstructure = Arc::new(RwLock::new(MicrostructureAnalyzer::new( config.feature_config.microstructure.clone(), ))); let regime_detector = Arc::new(RwLock::new(RegimeDetector::new( crate::features::RegimeDetectorConfig { lookback_periods: 20, volatility_threshold: 0.02, trend_threshold: 0.7, correlation_threshold: 0.7, rebalance_frequency: 5, }, ))); let portfolio_analyzer = Arc::new(RwLock::new(PortfolioAnalyzer::new( crate::features::PortfolioAnalyzerConfig { risk_free_rate: 0.02, target_return: 0.15, rebalance_threshold: 0.05, max_position_size: 0.10, diversification_target: 10, }, ))); Ok(Self { config, technical_indicators, microstructure, regime_detector, portfolio_analyzer, news_buffer: Arc::new(RwLock::new(BTreeMap::new())), market_data_buffer: Arc::new(RwLock::new(BTreeMap::new())), feature_cache: Arc::new(RwLock::new(HashMap::new())), feature_stats: Arc::new(RwLock::new(HashMap::new())), }) } /// Update with new market data pub async fn update_market_data(&self, symbol: &str, event: MarketDataEvent) -> Result<()> { let mut buffer = self.market_data_buffer.write().await; let symbol_buffer = buffer .entry(symbol.to_string()) .or_insert_with(VecDeque::new); symbol_buffer.push_back(event.clone()); // Keep only recent data (configurable window) let max_buffer_size = self.config.aggregation.max_buffer_size; while symbol_buffer.len() > max_buffer_size { symbol_buffer.pop_front(); } // Update technical indicators if let MarketDataEvent::Bar(bar_event) = event { let price_point = PricePoint { timestamp: bar_event.end_timestamp, open: ToPrimitive::to_f64(&bar_event.open).unwrap_or(0.0), high: ToPrimitive::to_f64(&bar_event.high).unwrap_or(0.0), low: ToPrimitive::to_f64(&bar_event.low).unwrap_or(0.0), close: ToPrimitive::to_f64(&bar_event.close).unwrap_or(0.0), }; let mut indicators = self.technical_indicators.write().await; indicators.update_price(symbol, price_point); } // Invalidate cache for this symbol self.invalidate_cache(symbol).await; Ok(()) } /// Update with new news event pub async fn update_news(&self, news_event: NewsEvent) -> Result<()> { let mut buffer = self.news_buffer.write().await; // Add event to all relevant symbols for symbol in &news_event.symbols { let symbol_buffer = buffer .entry(symbol.to_string()) .or_insert_with(VecDeque::new); symbol_buffer.push_back(news_event.clone()); // Keep only recent events (configurable window) let max_age = Duration::minutes(self.config.news_config.impact_window_minutes as i64 * 4); let cutoff_time = Utc::now() - max_age; while let Some(front_event) = symbol_buffer.front() { if front_event.timestamp < cutoff_time { symbol_buffer.pop_front(); } else { break; } } // Invalidate cache for this symbol self.invalidate_cache(symbol.as_ref()).await; } Ok(()) } /// Extract comprehensive features for a symbol pub async fn extract_features( &self, symbol: &str, timestamp: DateTime, ) -> Result { // Check cache first if let Some(cached) = self.get_cached_features(symbol, timestamp).await? { return Ok(cached.features); } info!("Extracting features for symbol: {}", symbol); // Extract multi-modal features let multi_modal = self.extract_multimodal_features(symbol, timestamp).await?; // Combine all features let mut all_features = HashMap::new(); all_features.extend(multi_modal.market_features); all_features.extend(multi_modal.news_features); all_features.extend(multi_modal.cross_modal_features); all_features.extend(multi_modal.temporal_features); all_features.extend(multi_modal.regime_features); // Apply scaling and missing value handling let processed_features = self.post_process_features(all_features).await?; // Create metadata let metadata = self.create_feature_metadata(&processed_features); let feature_vector = FeatureVector { timestamp, symbol: symbol.to_string(), features: processed_features, metadata, }; // Cache the result self.cache_features(symbol, feature_vector.clone()).await; Ok(feature_vector) } /// Extract features for multiple symbols (batch processing) pub async fn extract_features_batch( &self, symbols: &[String], timestamp: DateTime, ) -> Result> { let mut results = Vec::new(); // Process in parallel (if configured) for symbol in symbols { let features = self.extract_features(symbol, timestamp).await?; results.push(features); } Ok(results) } /// Extract multi-modal features combining market and news data async fn extract_multimodal_features( &self, symbol: &str, timestamp: DateTime, ) -> Result { // Extract market features let market_features = self.extract_market_features(symbol, timestamp).await?; // Extract news features let news_features = self.extract_news_features(symbol, timestamp).await?; // Extract temporal features let temporal_features = TemporalFeatures::extract_features(timestamp); // Extract regime features let regime_features = self.extract_regime_features(symbol).await?; // Extract cross-modal features let cross_modal_features = self .extract_cross_modal_features(symbol, &market_features, &news_features, timestamp) .await?; Ok(MultiModalFeatures { market_features, news_features, cross_modal_features, temporal_features, regime_features, }) } /// Extract market-based features async fn extract_market_features( &self, symbol: &str, _timestamp: DateTime, ) -> Result> { let mut features = HashMap::new(); // Technical indicators let indicators = self.technical_indicators.read().await; let ta_features = indicators.calculate_features(symbol); features.extend(ta_features); // Microstructure features let microstructure = self.microstructure.read().await; let micro_features = microstructure.calculate_features(symbol); features.extend(micro_features); // Add volume and volatility features if let Some(recent_bars) = self.get_recent_market_data(symbol, 20).await? { features.extend(self.calculate_volatility_features(&recent_bars)); features.extend(self.calculate_volume_features(&recent_bars)); } Ok(features) } /// Extract news-based features async fn extract_news_features( &self, symbol: &str, timestamp: DateTime, ) -> Result> { let mut features = HashMap::new(); let news_analysis = self.analyze_news_impact(symbol, timestamp).await?; // Basic news features features.insert( "news_sentiment_1h".to_string(), news_analysis.overall_sentiment, ); features.insert( "news_volume_1h".to_string(), news_analysis.news_volume as f64, ); features.insert( "news_avg_importance_1h".to_string(), news_analysis.avg_importance, ); // Event type features for (event_type, count) in news_analysis.event_type_distribution { features.insert( format!("news_{}_count_1h", event_type.to_lowercase()), count as f64, ); } // Recent high-impact events let high_impact_count = news_analysis .recent_events .iter() .filter(|event| event.importance > 0.7) .count(); features.insert( "news_high_impact_count_1h".to_string(), high_impact_count as f64, ); // Time-based news features (different windows) for &window_minutes in &[5, 15, 60, 240] { let window_analysis = self .analyze_news_impact_window(symbol, timestamp, window_minutes) .await?; let window_suffix = format!("{}m", window_minutes); features.insert( format!("news_sentiment_{}", window_suffix), window_analysis.overall_sentiment, ); features.insert( format!("news_volume_{}", window_suffix), window_analysis.news_volume as f64, ); } Ok(features) } /// Extract cross-modal features (market-news interactions) async fn extract_cross_modal_features( &self, symbol: &str, market_features: &HashMap, news_features: &HashMap, _timestamp: DateTime, ) -> Result> { let mut features = HashMap::new(); // Sentiment-momentum interaction if let (Some(&sentiment), Some(&momentum)) = ( news_features.get("news_sentiment_1h"), market_features.get("rsi_14"), ) { features.insert( "sentiment_momentum_interaction".to_string(), sentiment * momentum, ); } // News volume vs price volatility if let (Some(&news_vol), Some(&volatility)) = ( news_features.get("news_volume_1h"), market_features.get("bb_bandwidth_20"), ) { features.insert( "news_volume_volatility_ratio".to_string(), news_vol / (volatility + 1e-6), ); } // Sentiment divergence from technical indicators if let (Some(&sentiment), Some(&rsi)) = ( news_features.get("news_sentiment_1h"), market_features.get("rsi_14"), ) { let rsi_normalized = (rsi - 50.0) / 50.0; // Normalize RSI to -1 to 1 features.insert( "sentiment_technical_divergence".to_string(), sentiment - rsi_normalized, ); } // Calculate price reaction to news features.extend(self.calculate_news_price_reaction(symbol).await?); Ok(features) } /// Extract regime-based features async fn extract_regime_features(&self, symbol: &str) -> Result> { let mut features = HashMap::new(); // Get recent market data for regime analysis let lookback = self.config.feature_config.regime_detection.lookback_period; let recent_data = self.get_recent_market_data(symbol, lookback).await?; if let Some(data) = recent_data { // Extract prices and volumes for regime analysis let prices: Vec = data .iter() .filter_map(|event| { if let MarketDataEvent::Bar(bar) = event { ToPrimitive::to_f64(&bar.close) } else { None } }) .collect(); let volumes: Vec = data .iter() .filter_map(|event| { if let MarketDataEvent::Bar(bar) = event { bar.volume.to_f64() } else { None } }) .collect(); if prices.len() >= 20 { // 1. Volatility Regime Detection let volatility_regime = self.detect_volatility_regime(&prices); features.insert("volatility_regime".to_string(), volatility_regime); // 2. Trend Regime Detection let trend_regime = self.detect_trend_regime(&prices); features.insert("trend_regime".to_string(), trend_regime); // 3. Volume Regime Detection if volumes.len() >= 20 { let volume_regime = self.detect_volume_regime(&volumes); features.insert("volume_regime".to_string(), volume_regime); } // 4. Market State Features let (volatility_percentile, trend_strength) = self.calculate_regime_metrics(&prices); features.insert("volatility_percentile".to_string(), volatility_percentile); features.insert("trend_strength".to_string(), trend_strength); } } // Default to neutral regime if insufficient data features .entry("volatility_regime".to_string()) .or_insert(0.0); features.entry("trend_regime".to_string()).or_insert(0.0); features.entry("volume_regime".to_string()).or_insert(0.0); Ok(features) } /// Detect volatility regime: -1 (low), 0 (normal), 1 (high) fn detect_volatility_regime(&self, prices: &[f64]) -> f64 { if prices.len() < 20 { return 0.0; } // Calculate returns let returns: Vec = prices .windows(2) .filter_map(|w| { let ret = (w[1] / w[0]).ln(); if ret.is_finite() { Some(ret) } else { None } }) .collect(); if returns.is_empty() { return 0.0; } // Calculate realized volatility (standard deviation of returns) let mean = returns.iter().sum::() / returns.len() as f64; let variance = returns.iter().map(|r| (r - mean).powi(2)).sum::() / returns.len() as f64; let volatility = variance.sqrt(); // Annualize volatility (assuming daily data, multiply by sqrt(252)) let annualized_vol = volatility * (252.0_f64).sqrt(); // Classify regime based on threshold (using reasonable defaults) let vol_threshold = 0.20; // 20% annualized volatility as baseline if annualized_vol > vol_threshold * 1.5 { 1.0 // High volatility regime } else if annualized_vol < vol_threshold * 0.5 { -1.0 // Low volatility regime } else { 0.0 // Normal volatility regime } } /// Detect trend regime: -1 (downtrend), 0 (sideways), 1 (uptrend) fn detect_trend_regime(&self, prices: &[f64]) -> f64 { if prices.len() < 20 { return 0.0; } // Calculate short-term and long-term moving averages let short_window = 10; let long_window = 20.min(prices.len()); let short_ma = prices[prices.len() - short_window..].iter().sum::() / short_window as f64; let long_ma = prices[prices.len() - long_window..].iter().sum::() / long_window as f64; // Calculate trend strength let trend_pct = (short_ma - long_ma) / long_ma; let threshold = 0.01; // 1% trend threshold if trend_pct > threshold { 1.0 // Uptrend } else if trend_pct < -threshold { -1.0 // Downtrend } else { 0.0 // Sideways/neutral } } /// Detect volume regime: -1 (low), 0 (normal), 1 (high) fn detect_volume_regime(&self, volumes: &[f64]) -> f64 { if volumes.len() < 20 { return 0.0; } // Calculate average volume let avg_volume = volumes.iter().sum::() / volumes.len() as f64; let recent_volume = volumes.last().unwrap_or(&0.0); // Volume ratio relative to average let volume_ratio = recent_volume / (avg_volume + 1e-6); if volume_ratio > 1.5 { 1.0 // High volume regime } else if volume_ratio < 0.5 { -1.0 // Low volume regime } else { 0.0 // Normal volume regime } } /// Calculate regime metrics fn calculate_regime_metrics(&self, prices: &[f64]) -> (f64, f64) { if prices.len() < 20 { return (0.5, 0.0); } // Calculate returns for volatility let returns: Vec = prices .windows(2) .filter_map(|w| { let ret = (w[1] / w[0]).ln(); if ret.is_finite() { Some(ret) } else { None } }) .collect(); // Volatility percentile (normalized to 0-1) let mean = returns.iter().sum::() / returns.len() as f64; let variance = returns.iter().map(|r| (r - mean).powi(2)).sum::() / returns.len() as f64; let volatility = variance.sqrt(); let volatility_percentile = (volatility * 100.0).min(1.0).max(0.0); // Trend strength (linear regression slope) let n = prices.len() as f64; let x_mean = (n - 1.0) / 2.0; let y_mean = prices.iter().sum::() / n; let mut numerator = 0.0; let mut denominator = 0.0; for (i, &price) in prices.into_iter().enumerate() { let x_diff = i as f64 - x_mean; numerator += x_diff * (price - y_mean); denominator += x_diff * x_diff; } let slope = if denominator > 1e-10 { numerator / denominator } else { 0.0 }; // Normalize trend strength to -1 to 1 range let trend_strength = (slope / y_mean).clamp(-1.0, 1.0); (volatility_percentile, trend_strength) } /// Analyze news impact for a symbol async fn analyze_news_impact( &self, symbol: &str, timestamp: DateTime, ) -> Result { let window_minutes = self.config.news_config.impact_window_minutes as i64; self.analyze_news_impact_window(symbol, timestamp, window_minutes as u32) .await } /// Analyze news impact within a specific time window async fn analyze_news_impact_window( &self, symbol: &str, timestamp: DateTime, window_minutes: u32, ) -> Result { let buffer = self.news_buffer.read().await; let window_start = timestamp - Duration::minutes(window_minutes as i64); let relevant_events: Vec = buffer .get(symbol) .map(|events| { events .iter() .filter(|event| { event.timestamp >= window_start && event.timestamp <= timestamp && event.importance >= self.config.news_config.min_importance }) .cloned() .collect() }) .unwrap_or_default(); let overall_sentiment = if relevant_events.is_empty() { 0.0 } else { let weighted_sentiment: f64 = relevant_events .iter() .filter_map(|event| { event.sentiment_score.map(|s| { let weight = self .config .news_config .news_type_weights .get(&format!("{:?}", event.event_type)) .unwrap_or(&1.0); s * event.importance * weight }) }) .sum(); let total_weight: f64 = relevant_events .iter() .filter(|event| event.sentiment_score.is_some()) .map(|event| { let weight = self .config .news_config .news_type_weights .get(&format!("{:?}", event.event_type)) .unwrap_or(&1.0); event.importance * weight }) .sum(); if total_weight > 0.0 { weighted_sentiment / total_weight } else { 0.0 } }; let avg_importance = if relevant_events.is_empty() { 0.0 } else { relevant_events.iter().map(|e| e.importance).sum::() / relevant_events.len() as f64 }; let mut event_type_distribution = HashMap::new(); for event in &relevant_events { let event_type_str = format!("{:?}", event.event_type); *event_type_distribution.entry(event_type_str).or_insert(0) += 1; } Ok(NewsImpactAnalysis { symbol: symbol.to_string(), timestamp, overall_sentiment, news_volume: relevant_events.len() as u32, avg_importance, event_type_distribution, recent_events: relevant_events, }) } /// Calculate volatility features from recent market data fn calculate_volatility_features(&self, bars: &[MarketDataEvent]) -> HashMap { let mut features = HashMap::new(); let returns: Vec = bars .windows(2) .filter_map(|window| { if let (MarketDataEvent::Bar(bar1), MarketDataEvent::Bar(bar2)) = (&window[0], &window[1]) { let ret = (ToPrimitive::to_f64(&bar2.close).unwrap_or(0.0) / ToPrimitive::to_f64(&bar1.close).unwrap_or(1.0) - 1.0) .ln(); if ret.is_finite() { Some(ret) } else { None } } else { None } }) .collect(); if returns.len() > 1 { let mean_return = returns.iter().sum::() / returns.len() as f64; let variance = returns .iter() .map(|r| (r - mean_return).powi(2)) .sum::() / (returns.len() - 1) as f64; let volatility = variance.sqrt(); features.insert("volatility_realized".to_string(), volatility); features.insert("mean_return".to_string(), mean_return); // Skewness and kurtosis if volatility > 0.0 { let skewness = returns .iter() .map(|r| ((r - mean_return) / volatility).powi(3)) .sum::() / returns.len() as f64; let kurtosis = returns .iter() .map(|r| ((r - mean_return) / volatility).powi(4)) .sum::() / returns.len() as f64; features.insert("return_skewness".to_string(), skewness); features.insert("return_kurtosis".to_string(), kurtosis); } } features } /// Calculate volume features from recent market data fn calculate_volume_features(&self, bars: &[MarketDataEvent]) -> HashMap { let mut features = HashMap::new(); let volumes: Vec = bars .iter() .filter_map(|bar| { if let MarketDataEvent::Bar(bar_event) = bar { bar_event.volume.to_f64() } else { None } }) .collect(); if !volumes.is_empty() { let avg_volume = volumes.iter().sum::() / volumes.len() as f64; let current_volume = volumes.last().unwrap_or(&0.0); features.insert( "volume_ratio".to_string(), current_volume / (avg_volume + 1e-6), ); // Volume trend if volumes.len() >= 2 { let recent_avg = volumes[volumes.len() / 2..].iter().sum::() / (volumes.len() / 2) as f64; let early_avg = volumes[..volumes.len() / 2].iter().sum::() / (volumes.len() / 2) as f64; features.insert( "volume_trend".to_string(), (recent_avg - early_avg) / (early_avg + 1e-6), ); } } features } /// Calculate price reaction to news events async fn calculate_news_price_reaction(&self, symbol: &str) -> Result> { let mut features = HashMap::new(); // Get recent news events for this symbol let news_buffer = self.news_buffer.read().await; let recent_news = news_buffer.get(symbol); if let Some(news_events) = recent_news { // Get market data buffer let market_buffer = self.market_data_buffer.read().await; let market_data = market_buffer.get(symbol); if let Some(bars) = market_data { // Analyze price reaction at different time windows: 5m, 15m, 1h let windows = vec![(5, "5m"), (15, "15m"), (60, "1h")]; for (window_minutes, suffix) in windows { let reaction = self.calculate_price_reaction_window(news_events, bars, window_minutes); features.insert( format!("news_price_reaction_{}", suffix), reaction.avg_reaction, ); features.insert( format!("news_price_volatility_{}", suffix), reaction.volatility, ); features.insert( format!("news_price_direction_{}", suffix), reaction.direction, ); } } } // Default values if no news or data features .entry("news_price_reaction_5m".to_string()) .or_insert(0.0); features .entry("news_price_reaction_15m".to_string()) .or_insert(0.0); features .entry("news_price_reaction_1h".to_string()) .or_insert(0.0); Ok(features) } /// Calculate price reaction within a specific time window after news events fn calculate_price_reaction_window( &self, news_events: &VecDeque, market_data: &VecDeque, window_minutes: i64, ) -> PriceReaction { let mut reactions = Vec::new(); // For each news event, find price changes before and after for news_event in news_events.into_iter().rev().take(10) { // Take most recent 10 news events if let Some(reaction) = self.calculate_single_event_reaction(news_event, market_data, window_minutes) { reactions.push(reaction); } } if reactions.is_empty() { return PriceReaction { avg_reaction: 0.0, volatility: 0.0, direction: 0.0, }; } // Aggregate reactions let avg_reaction = reactions.iter().sum::() / reactions.len() as f64; // Calculate volatility of reactions let mean = avg_reaction; let variance = reactions.iter().map(|r| (r - mean).powi(2)).sum::() / reactions.len() as f64; let volatility = variance.sqrt(); // Determine direction (positive or negative) let positive_count = reactions.iter().filter(|&r| *r > 0.0).count(); let half_len = reactions.len() as f64 * 0.5; let positive_f64 = positive_count as f64; let direction = if positive_f64 > half_len { 1.0 // Mostly positive reactions } else if positive_f64 < half_len { -1.0 // Mostly negative reactions } else { 0.0 // Mixed reactions }; PriceReaction { avg_reaction, volatility, direction, } } /// Calculate price reaction for a single news event fn calculate_single_event_reaction( &self, news_event: &NewsEvent, market_data: &VecDeque, window_minutes: i64, ) -> Option { let news_time = news_event.timestamp; let window_duration = Duration::minutes(window_minutes); // Find price before news event (within 5 minutes before) let before_window_start = news_time - Duration::minutes(5); let before_price = market_data.iter().rev().find_map(|event| { if let MarketDataEvent::Bar(bar) = event { if bar.end_timestamp >= before_window_start && bar.end_timestamp < news_time { return ToPrimitive::to_f64(&bar.close); } } None }); // Find price after news event (at end of window) let after_window_end = news_time + window_duration; let after_price = market_data.iter().rev().find_map(|event| { if let MarketDataEvent::Bar(bar) = event { if bar.end_timestamp > news_time && bar.end_timestamp <= after_window_end { return ToPrimitive::to_f64(&bar.close); } } None }); // Calculate percentage change match (before_price, after_price) { (Some(before), Some(after)) if before > 0.0 => { let pct_change = ((after - before) / before) * 100.0; // Weight by news importance Some(pct_change * news_event.importance) }, _ => None, } } /// Get recent market data for a symbol async fn get_recent_market_data( &self, symbol: &str, count: usize, ) -> Result>> { let buffer = self.market_data_buffer.read().await; if let Some(data) = buffer.get(symbol) { let recent: Vec = data.iter().rev().take(count).cloned().collect(); if recent.is_empty() { Ok(None) } else { Ok(Some(recent)) } } else { Ok(None) } } /// Post-process features (scaling, missing values, etc.) async fn post_process_features( &self, mut features: HashMap, ) -> Result> { // Update feature statistics first self.update_feature_statistics(&features).await; // Handle missing values match self.config.output.missing_value_strategy { MissingValueStrategy::Zero => { // Replace NaN/infinite values with 0 for value in features.values_mut() { if !value.is_finite() { *value = 0.0; } } }, MissingValueStrategy::Mean => { // Implement mean imputation based on historical data let stats = self.feature_stats.read().await; for (feature_name, value) in features.iter_mut() { if !value.is_finite() { if let Some(stat) = stats.get(feature_name) { *value = stat.mean; } else { *value = 0.0; // Fallback to zero if no stats available } } } }, MissingValueStrategy::ForwardFill => { // Implement forward fill using last known values let stats = self.feature_stats.read().await; for (feature_name, value) in features.iter_mut() { if !value.is_finite() { if let Some(stat) = stats.get(feature_name) { if let Some(last_val) = stat.last_value { *value = last_val; } else { *value = 0.0; // Fallback if no previous value } } else { *value = 0.0; } } } }, _ => { // For other strategies, just replace non-finite values with 0 for value in features.values_mut() { if !value.is_finite() { *value = 0.0; } } }, } // Apply scaling match self.config.output.scaling_method { ScalingMethod::StandardScore => { // Implement z-score standardization with running statistics let stats = self.feature_stats.read().await; for (feature_name, value) in features.iter_mut() { if let Some(stat) = stats.get(feature_name) { if stat.count > 1 && stat.variance > 0.0 { let std_dev = stat.variance.sqrt(); *value = (*value - stat.mean) / std_dev; } } } }, ScalingMethod::MinMax => { // Implement min-max scaling to [0, 1] range let stats = self.feature_stats.read().await; for (feature_name, value) in features.iter_mut() { if let Some(stat) = stats.get(feature_name) { let range = stat.max - stat.min; if range > 1e-10 { *value = (*value - stat.min) / range; } else { *value = 0.5; // Center value if no range } } } }, ScalingMethod::None => { // No scaling needed }, _ => { // Default to no scaling }, } Ok(features) } /// Update running statistics for features async fn update_feature_statistics(&self, features: &HashMap) { let mut stats = self.feature_stats.write().await; for (feature_name, &value) in features { if !value.is_finite() { continue; // Skip non-finite values for statistics } let stat = stats.entry(feature_name.clone()).or_insert(FeatureStats { mean: 0.0, variance: 0.0, min: value, max: value, count: 0, last_value: None, }); // Update running statistics using Welford's online algorithm stat.count += 1; let delta = value - stat.mean; stat.mean += delta / stat.count as f64; let delta2 = value - stat.mean; stat.variance += delta * delta2; // Update min/max if value < stat.min { stat.min = value; } if value > stat.max { stat.max = value; } // Update last value for forward fill stat.last_value = Some(value); // Convert variance to sample variance if stat.count > 1 { stat.variance = stat.variance / (stat.count - 1) as f64; } } } /// Create feature metadata fn create_feature_metadata(&self, features: &HashMap) -> FeatureMetadata { let mut feature_descriptions = HashMap::new(); let mut feature_categories = HashMap::new(); let mut quality_indicators = HashMap::new(); for feature_name in features.keys() { // Categorize features based on naming patterns let category = if feature_name.contains("sma") || feature_name.contains("ema") || feature_name.contains("rsi") || feature_name.contains("macd") || feature_name.contains("bb_") { FeatureCategory::TechnicalIndicator } else if feature_name.contains("news_") { FeatureCategory::TLOB // Using TLOB as placeholder for news features } else if feature_name.contains("volume") { FeatureCategory::Volume } else if feature_name.contains("price") || feature_name.contains("close") || feature_name.contains("return") { FeatureCategory::Price } else if feature_name.contains("hour") || feature_name.contains("day") || feature_name.contains("session") { FeatureCategory::Temporal } else if feature_name.contains("regime") || feature_name.contains("volatility") { FeatureCategory::Regime } else if feature_name.contains("spread") || feature_name.contains("imbalance") { FeatureCategory::Microstructure } else { FeatureCategory::Price // Default category }; feature_descriptions.insert( feature_name.clone(), format!("Auto-generated: {}", feature_name), ); feature_categories.insert(feature_name.clone(), category); quality_indicators.insert(feature_name.clone(), 1.0); // Default quality } // Collect categories before moving feature_categories let categories: Vec = feature_categories.values().cloned().collect(); FeatureMetadata { feature_descriptions, feature_categories, quality_indicators, symbol: "".to_string(), timestamp: Utc::now(), feature_count: features.len(), categories, } } /// Check cache for features async fn get_cached_features( &self, symbol: &str, timestamp: DateTime, ) -> Result> { let cache = self.feature_cache.read().await; let cache_key = format!("{}_{}", symbol, timestamp.format("%Y%m%d_%H%M")); if let Some(cached) = cache.get(&cache_key) { let age_minutes = (Utc::now() - cached.cached_at).num_minutes() as u32; if age_minutes < cached.ttl_minutes { return Ok(Some(cached.clone())); } } Ok(None) } /// Cache features async fn cache_features(&self, symbol: &str, features: FeatureVector) { let cache_key = format!("{}_{}", symbol, features.timestamp.format("%Y%m%d_%H%M")); let cached = CachedFeatureVector { features, cached_at: Utc::now(), ttl_minutes: 5, // Cache for 5 minutes }; let mut cache = self.feature_cache.write().await; cache.insert(cache_key, cached); // Cleanup old cache entries if cache.len() > 1000 { let cutoff = Utc::now() - Duration::minutes(60); cache.retain(|_, v| v.cached_at > cutoff); } } /// Invalidate cache for a symbol async fn invalidate_cache(&self, symbol: &str) { let mut cache = self.feature_cache.write().await; cache.retain(|key, _| !key.starts_with(symbol)); } } // Placeholder implementations REMOVED - duplicates removed // These impls are already defined in features.rs with proper configs #[cfg(test)] #[allow(clippy::assertions_on_result_states)] mod tests { use super::*; #[test] fn test_config_creation() { let config = UnifiedFeatureExtractorConfig::default(); assert!(config.news_config.sentiment_analysis); assert!(!config .feature_config .technical_indicators .ma_periods .is_empty()); } #[tokio::test] async fn test_extractor_creation() { let config = UnifiedFeatureExtractorConfig::default(); let extractor = UnifiedFeatureExtractor::new(config); assert!(extractor.is_ok()); } #[test] fn test_news_analysis_config() { let config = NewsAnalysisConfig { sentiment_analysis: true, impact_window_minutes: 60, min_importance: 0.3, categories: vec!["Earnings".to_string()], news_type_weights: HashMap::new(), event_clustering: false, max_events_per_period: 10, }; assert_eq!(config.impact_window_minutes, 60); assert_eq!(config.min_importance, 0.3); } #[test] fn test_aggregation_config() { let config = AggregationConfig { primary_timeframe_minutes: 60, secondary_timeframes: vec![300, 900], lookback_periods: vec![10, 20, 50], cross_symbol_features: true, max_correlation_symbols: 5, max_buffer_size: 10000, }; assert_eq!(config.secondary_timeframes.len(), 2); assert!(config.cross_symbol_features); assert_eq!(config.primary_timeframe_minutes, 60); } #[test] fn test_output_config() { let config = OutputConfig { include_metadata: true, scaling_method: ScalingMethod::StandardScore, missing_value_strategy: MissingValueStrategy::ForwardFill, feature_selection: FeatureSelectionConfig { enabled: true, max_features: Some(100), min_correlation: 0.01, max_correlation: 0.95, importance_threshold: 0.001, }, }; assert!(matches!( config.scaling_method, ScalingMethod::StandardScore )); assert!(matches!( config.missing_value_strategy, MissingValueStrategy::ForwardFill )); assert!(config.include_metadata); } #[test] fn test_feature_selection_config() { let config = FeatureSelectionConfig { enabled: true, max_features: Some(100), min_correlation: 0.01, max_correlation: 0.95, importance_threshold: 0.001, }; assert!(config.enabled); assert_eq!(config.min_correlation, 0.01); assert_eq!(config.max_features, Some(100)); } #[test] fn test_cached_feature_vector() { let features = HashMap::new(); let metadata = FeatureMetadata { symbol: "AAPL".to_string(), timestamp: Utc::now(), feature_count: 0, categories: vec![], feature_descriptions: HashMap::new(), feature_categories: HashMap::new(), quality_indicators: HashMap::new(), }; let cached = CachedFeatureVector { features: FeatureVector { timestamp: Utc::now(), symbol: "AAPL".to_string(), features, metadata, }, cached_at: Utc::now(), ttl_minutes: 60, }; assert_eq!(cached.ttl_minutes, 60); assert!(cached.cached_at <= Utc::now()); } #[test] fn test_multi_modal_features_empty() { let features = MultiModalFeatures { market_features: HashMap::new(), news_features: HashMap::new(), cross_modal_features: HashMap::new(), temporal_features: HashMap::new(), regime_features: HashMap::new(), }; assert!(features.market_features.is_empty()); assert!(features.news_features.is_empty()); assert!(features.cross_modal_features.is_empty()); } #[test] fn test_multi_modal_features_populated() { let mut features = MultiModalFeatures { market_features: HashMap::new(), news_features: HashMap::new(), cross_modal_features: HashMap::new(), temporal_features: HashMap::new(), regime_features: HashMap::new(), }; features.market_features.insert("close".to_string(), 100.0); features .market_features .insert("volume".to_string(), 1000.0); features.news_features.insert("sentiment".to_string(), 0.7); assert_eq!(features.market_features.len(), 2); assert_eq!(features.news_features.len(), 1); assert!(features.cross_modal_features.is_empty()); } #[test] fn test_news_impact_analysis() { let analysis = NewsImpactAnalysis { symbol: "AAPL".to_string(), timestamp: Utc::now(), overall_sentiment: 0.7, news_volume: 5, avg_importance: 0.8, event_type_distribution: HashMap::from([ ("Earnings".to_string(), 3), ("M&A".to_string(), 2), ]), recent_events: vec![], }; assert_eq!(analysis.overall_sentiment, 0.7); assert_eq!(analysis.avg_importance, 0.8); assert_eq!(analysis.news_volume, 5); assert_eq!(analysis.event_type_distribution.len(), 2); } #[test] fn test_scaling_methods() { assert!(matches!( ScalingMethod::StandardScore, ScalingMethod::StandardScore )); assert!(matches!(ScalingMethod::MinMax, ScalingMethod::MinMax)); assert!(matches!(ScalingMethod::Robust, ScalingMethod::Robust)); assert!(matches!(ScalingMethod::None, ScalingMethod::None)); } #[test] fn test_missing_value_strategies() { assert!(matches!( MissingValueStrategy::Zero, MissingValueStrategy::Zero )); assert!(matches!( MissingValueStrategy::Mean, MissingValueStrategy::Mean )); assert!(matches!( MissingValueStrategy::ForwardFill, MissingValueStrategy::ForwardFill )); assert!(matches!( MissingValueStrategy::BackwardFill, MissingValueStrategy::BackwardFill )); assert!(matches!( MissingValueStrategy::Interpolate, MissingValueStrategy::Interpolate )); } #[test] fn test_portfolio_analyzer_creation() { let config = crate::features::PortfolioAnalyzerConfig { risk_free_rate: 0.02, target_return: 0.15, rebalance_threshold: 0.05, max_position_size: 0.10, diversification_target: 10, }; let analyzer = PortfolioAnalyzer::new(config); assert!(analyzer.positions.is_empty()); assert!(analyzer.pnl_history.is_empty()); } #[test] fn test_regime_detector_creation() { let config = crate::features::RegimeDetectorConfig { lookback_periods: 50, volatility_threshold: 0.02, trend_threshold: 0.01, correlation_threshold: 0.7, rebalance_frequency: 100, }; let detector = RegimeDetector::new(config); assert!(detector.price_history.is_empty()); } #[tokio::test] async fn test_cache_cleanup() { let config = UnifiedFeatureExtractorConfig::default(); let extractor = UnifiedFeatureExtractor::new(config).unwrap(); // Test cache starts empty let cache = extractor.feature_cache.read().await; assert!(cache.is_empty()); } #[tokio::test] async fn test_cache_invalidation() { let config = UnifiedFeatureExtractorConfig::default(); let extractor = UnifiedFeatureExtractor::new(config).unwrap(); // Add a cached entry { let mut cache = extractor.feature_cache.write().await; let features = HashMap::from([("close".to_string(), 150.0)]); let metadata = FeatureMetadata { symbol: "AAPL".to_string(), timestamp: Utc::now(), feature_count: 1, categories: vec![FeatureCategory::Price], feature_descriptions: HashMap::new(), feature_categories: HashMap::new(), quality_indicators: HashMap::new(), }; cache.insert( "AAPL_60".to_string(), CachedFeatureVector { features: FeatureVector { timestamp: Utc::now(), symbol: "AAPL".to_string(), features, metadata, }, cached_at: Utc::now(), ttl_minutes: 60, }, ); } // Verify cache has entry { let cache = extractor.feature_cache.read().await; assert_eq!(cache.len(), 1); } // Invalidate cache for symbol extractor.invalidate_cache("AAPL").await; // Verify cache is empty let cache = extractor.feature_cache.read().await; assert_eq!(cache.len(), 0); } #[test] fn test_default_config() { let config = UnifiedFeatureExtractorConfig::default(); assert!(config.news_config.sentiment_analysis); assert!(config.aggregation.cross_symbol_features); assert!(matches!( config.output.scaling_method, ScalingMethod::StandardScore )); } }