diff --git a/ml/src/ensemble/coordinator.rs b/ml/src/ensemble/coordinator.rs index 83b53a1ff..e9e042af5 100644 --- a/ml/src/ensemble/coordinator.rs +++ b/ml/src/ensemble/coordinator.rs @@ -10,7 +10,8 @@ use crate::ensemble::conviction_gates::{ use crate::ensemble::inference_adapter::{FeatureVector, ModelInferenceAdapter}; use crate::ensemble::{EnsembleDecision, ModelVote, ModelWeight, TradingAction}; use crate::{Features, MLError, MLResult, ModelPrediction}; -use chrono::{Timelike, Utc}; +use chrono::{DateTime, TimeZone, Timelike, Utc}; +use chrono_tz::America::New_York; use std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; @@ -235,6 +236,16 @@ impl EnsembleCoordinator { }; match adapter.predict(&fv) { Ok(ensemble_pred) => { + if !ensemble_pred.direction.is_finite() + || !ensemble_pred.confidence.is_finite() + { + warn!( + "Adapter {} returned non-finite prediction \ + (direction={}, confidence={}), skipping", + model_id, ensemble_pred.direction, ensemble_pred.confidence + ); + continue; + } let prediction = ModelPrediction::new( model_id.clone(), ensemble_pred.direction, @@ -377,12 +388,18 @@ impl EnsembleCoordinator { agreeing as f64 / decision.model_votes.len() as f64 } - /// Determine current trading session (Eastern Time, simplified UTC-5) + /// Determine current trading session (Eastern Time, DST-aware) fn current_trading_session() -> TradingSession { - let now = Utc::now(); - let et_hour = (now.hour() + 24 - 5) % 24; - let et_minute = now.minute(); - let et_time = et_hour * 60 + et_minute; + Self::trading_session_at(Utc::now()) + } + + /// Determine trading session for an arbitrary UTC timestamp. + /// + /// Uses `chrono-tz` America/New_York so the boundaries are correct + /// during both EST (UTC-5, Nov-Mar) and EDT (UTC-4, Mar-Nov). + fn trading_session_at(utc_time: DateTime) -> TradingSession { + let et = utc_time.with_timezone(&New_York); + let et_time = et.hour() * 60 + et.minute(); match et_time { t if t < 240 => TradingSession::AfterHours, // 00:00-04:00 ET @@ -513,12 +530,17 @@ impl SignalAggregator { }); } + // Normalize effective weights so they sum to 1.0 before any calculations. + // Raw effective_weight = static_weight * dynamic_weight can range 0.5-1.5 + // per model, so the sum across models is not guaranteed to be 1.0. + let normalized = Self::normalize_effective_weights(&predictions, weights); + // Calculate weighted average signal let (weighted_signal, _total_weight) = - self.calculate_weighted_signal(&predictions, weights); + self.calculate_weighted_signal(&predictions, &normalized); // Calculate ensemble confidence - let confidence = self.calculate_ensemble_confidence(&predictions, weights); + let confidence = self.calculate_ensemble_confidence(&predictions, &normalized); // Calculate disagreement rate let disagreement_rate = self.calculate_disagreement_rate(&predictions); @@ -526,8 +548,8 @@ impl SignalAggregator { // Determine trading action let action = TradingAction::from_signal(weighted_signal, self.signal_threshold); - // Build model votes - let model_votes = self.build_model_votes(&predictions, weights); + // Build model votes (uses normalized weights) + let model_votes = self.build_model_votes(&predictions, &normalized); let decision = EnsembleDecision::new( action, @@ -540,20 +562,62 @@ impl SignalAggregator { Ok(decision) } - /// Calculate weighted average signal + /// Normalize effective weights across all predictions so they sum to 1.0. + /// + /// Models without an explicit weight entry receive equal share (1/N). + /// If total weight is zero or non-finite, all models get uniform 1/N weight. + fn normalize_effective_weights( + predictions: &[ModelPrediction], + weights: &HashMap, + ) -> HashMap { + let n = predictions.len(); + let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 }; + + // Collect raw effective weights + let raw: Vec<(String, f64)> = predictions + .iter() + .map(|pred| { + let w = weights + .get(&pred.model_id) + .map(|mw| mw.effective_weight()) + .unwrap_or(uniform); + (pred.model_id.clone(), w) + }) + .collect(); + + let total: f64 = raw.iter().map(|(_, w)| w).sum(); + + let mut normalized = HashMap::new(); + if total > 0.0 && total.is_finite() { + for (id, w) in raw { + normalized.insert(id, w / total); + } + } else { + // Fallback: uniform weights + for (id, _) in raw { + normalized.insert(id, uniform); + } + } + + normalized + } + + /// Calculate weighted average signal using pre-normalized weights fn calculate_weighted_signal( &self, predictions: &[ModelPrediction], - weights: &HashMap, + normalized_weights: &HashMap, ) -> (f64, f64) { + let n = predictions.len(); + let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 }; let mut weighted_sum = 0.0; let mut total_weight = 0.0; for pred in predictions { - let weight = weights + let weight = normalized_weights .get(&pred.model_id) - .map(|w| w.effective_weight()) - .unwrap_or(1.0 / predictions.len() as f64); + .copied() + .unwrap_or(uniform); weighted_sum += pred.value * pred.confidence * weight; total_weight += weight * pred.confidence; @@ -568,20 +632,22 @@ impl SignalAggregator { (signal, total_weight) } - /// Calculate ensemble confidence + /// Calculate ensemble confidence using pre-normalized weights fn calculate_ensemble_confidence( &self, predictions: &[ModelPrediction], - weights: &HashMap, + normalized_weights: &HashMap, ) -> f64 { + let n = predictions.len(); + let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 }; let mut confidence_sum = 0.0; let mut weight_sum = 0.0; for pred in predictions { - let weight = weights + let weight = normalized_weights .get(&pred.model_id) - .map(|w| w.effective_weight()) - .unwrap_or(1.0 / predictions.len() as f64); + .copied() + .unwrap_or(uniform); confidence_sum += pred.confidence * weight; weight_sum += weight; @@ -613,19 +679,21 @@ impl SignalAggregator { disagreements as f64 / predictions.len() as f64 } - /// Build model votes map + /// Build model votes map using pre-normalized weights fn build_model_votes( &self, predictions: &[ModelPrediction], - weights: &HashMap, + normalized_weights: &HashMap, ) -> HashMap { + let n = predictions.len(); + let uniform = if n > 0 { 1.0 / n as f64 } else { 0.0 }; let mut votes = HashMap::new(); for pred in predictions { - let weight = weights + let weight = normalized_weights .get(&pred.model_id) - .map(|w| w.effective_weight()) - .unwrap_or(1.0 / predictions.len() as f64); + .copied() + .unwrap_or(uniform); let vote = ModelVote::new(pred.model_id.clone(), pred.value, pred.confidence, weight); @@ -879,6 +947,184 @@ mod tests { assert_eq!(decision.model_count(), 1); } + #[tokio::test] + async fn test_ensemble_filters_nan_inf_predictions() { + use crate::ensemble::inference_adapter::{ + EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta, + }; + + struct NanAdapter; + impl ModelInferenceAdapter for NanAdapter { + fn model_name(&self) -> &str { + "NAN_MODEL" + } + fn predict( + &self, + _features: &FeatureVector, + ) -> crate::MLResult { + Ok(EnsemblePrediction { + model_name: "NAN_MODEL".to_string(), + direction: f64::NAN, + confidence: 0.9, + metadata: PredictionMeta::default(), + }) + } + fn is_ready(&self) -> bool { + true + } + } + unsafe impl Send for NanAdapter {} + unsafe impl Sync for NanAdapter {} + + struct InfAdapter; + impl ModelInferenceAdapter for InfAdapter { + fn model_name(&self) -> &str { + "INF_MODEL" + } + fn predict( + &self, + _features: &FeatureVector, + ) -> crate::MLResult { + Ok(EnsemblePrediction { + model_name: "INF_MODEL".to_string(), + direction: 0.5, + confidence: f64::INFINITY, + metadata: PredictionMeta::default(), + }) + } + fn is_ready(&self) -> bool { + true + } + } + unsafe impl Send for InfAdapter {} + unsafe impl Sync for InfAdapter {} + + struct GoodAdapter; + impl ModelInferenceAdapter for GoodAdapter { + fn model_name(&self) -> &str { + "GOOD_MODEL" + } + fn predict( + &self, + _features: &FeatureVector, + ) -> crate::MLResult { + Ok(EnsemblePrediction { + model_name: "GOOD_MODEL".to_string(), + direction: 0.6, + confidence: 0.85, + metadata: PredictionMeta::default(), + }) + } + fn is_ready(&self) -> bool { + true + } + } + unsafe impl Send for GoodAdapter {} + unsafe impl Sync for GoodAdapter {} + + let mut coordinator = EnsembleCoordinator::new(); + coordinator.add_adapter(Box::new(NanAdapter)); + coordinator.add_adapter(Box::new(InfAdapter)); + coordinator.add_adapter(Box::new(GoodAdapter)); + + coordinator + .register_model("NAN_MODEL".to_string(), 0.33) + .await + .unwrap(); + coordinator + .register_model("INF_MODEL".to_string(), 0.33) + .await + .unwrap(); + coordinator + .register_model("GOOD_MODEL".to_string(), 0.34) + .await + .unwrap(); + + let features = Features::new( + vec![0.5; 5], + vec!["f1", "f2", "f3", "f4", "f5"] + .into_iter() + .map(String::from) + .collect(), + ); + + let decision = coordinator.predict(&features).await.unwrap(); + + // Only the good model should survive filtering + assert_eq!( + decision.model_count(), + 1, + "NaN and Inf predictions should be filtered out" + ); + assert!( + decision.confidence.is_finite(), + "Ensemble confidence must be finite" + ); + assert!( + decision.signal.is_finite(), + "Ensemble signal must be finite" + ); + } + + #[tokio::test] + async fn test_ensemble_all_nan_returns_error() { + use crate::ensemble::inference_adapter::{ + EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta, + }; + + struct AllNanAdapter { + name: &'static str, + } + impl ModelInferenceAdapter for AllNanAdapter { + fn model_name(&self) -> &str { + self.name + } + fn predict( + &self, + _features: &FeatureVector, + ) -> crate::MLResult { + Ok(EnsemblePrediction { + model_name: self.name.to_string(), + direction: f64::NAN, + confidence: f64::NAN, + metadata: PredictionMeta::default(), + }) + } + fn is_ready(&self) -> bool { + true + } + } + unsafe impl Send for AllNanAdapter {} + unsafe impl Sync for AllNanAdapter {} + + let mut coordinator = EnsembleCoordinator::new(); + coordinator.add_adapter(Box::new(AllNanAdapter { name: "A" })); + coordinator.add_adapter(Box::new(AllNanAdapter { name: "B" })); + + coordinator + .register_model("A".to_string(), 0.5) + .await + .unwrap(); + coordinator + .register_model("B".to_string(), 0.5) + .await + .unwrap(); + + let features = Features::new( + vec![0.5; 5], + vec!["f1", "f2", "f3", "f4", "f5"] + .into_iter() + .map(String::from) + .collect(), + ); + + let result = coordinator.predict(&features).await; + assert!( + result.is_err(), + "All-NaN predictions should produce an error" + ); + } + #[tokio::test] async fn test_ensemble_rejects_no_adapters() { let coordinator = EnsembleCoordinator::new(); @@ -907,4 +1153,176 @@ mod tests { err_msg ); } + + #[tokio::test] + async fn test_normalized_weights_sum_to_one() { + let aggregator = SignalAggregator::new(); + + let predictions = vec![ + ModelPrediction::new("DQN".to_string(), 0.8, 0.9), + ModelPrediction::new("PPO".to_string(), 0.7, 0.85), + ModelPrediction::new("TFT".to_string(), 0.6, 0.8), + ]; + + // Create weights with dynamic adjustments that cause unnormalized sums. + // DQN: high perf => dynamic ~1.2, effective = 0.5 * 1.2 = 0.6 + // PPO: low perf => dynamic ~0.6, effective = 0.3 * 0.6 = 0.18 + // TFT: default => dynamic 1.0, effective = 0.2 * 1.0 = 0.2 + // Raw sum = 0.98, NOT 1.0 + let mut weights = HashMap::new(); + + let mut dqn_w = ModelWeight::new("DQN".to_string(), 0.5); + dqn_w.performance_metrics.sharpe_ratio = 2.0; + dqn_w.performance_metrics.accuracy = 0.7; + dqn_w.update_dynamic_weight(); + weights.insert("DQN".to_string(), dqn_w); + + let mut ppo_w = ModelWeight::new("PPO".to_string(), 0.3); + ppo_w.performance_metrics.sharpe_ratio = 0.3; + ppo_w.performance_metrics.accuracy = 0.3; + ppo_w.update_dynamic_weight(); + weights.insert("PPO".to_string(), ppo_w); + + let mut tft_w = ModelWeight::new("TFT".to_string(), 0.2); + // Keep default performance (sharpe=1.0, accuracy=0.5) => dynamic ~0.95 + tft_w.update_dynamic_weight(); + weights.insert("TFT".to_string(), tft_w); + + // Verify raw effective weights do NOT sum to 1.0 + let raw_sum: f64 = weights.values().map(|w| w.effective_weight()).sum(); + assert!( + (raw_sum - 1.0).abs() > 0.01, + "Raw effective weights should NOT sum to 1.0 (got {}), \ + otherwise the test does not exercise normalization", + raw_sum + ); + + // After aggregation, model_votes weights must sum to 1.0 + let decision = aggregator + .aggregate(predictions, &weights) + .await + .ok() + .expect("aggregation should succeed"); + + let vote_weight_sum: f64 = decision.model_votes.values().map(|v| v.weight).sum(); + assert!( + (vote_weight_sum - 1.0).abs() < 1e-10, + "Normalized vote weights must sum to 1.0, got {}", + vote_weight_sum + ); + + // Each individual weight must be positive + for (model_id, vote) in &decision.model_votes { + assert!( + vote.weight > 0.0, + "Weight for {} must be positive, got {}", + model_id, + vote.weight + ); + } + } + + #[test] + fn test_normalize_effective_weights_uniform_fallback() { + // When weights map is empty, all models get 1/N + let predictions = vec![ + ModelPrediction::new("A".to_string(), 0.5, 0.8), + ModelPrediction::new("B".to_string(), 0.3, 0.7), + ]; + let weights = HashMap::new(); + + let normalized = SignalAggregator::normalize_effective_weights(&predictions, &weights); + + assert_eq!(normalized.len(), 2); + let sum: f64 = normalized.values().sum(); + assert!( + (sum - 1.0).abs() < 1e-10, + "Uniform fallback weights must sum to 1.0, got {}", + sum + ); + for (_, w) in &normalized { + assert!( + (*w - 0.5).abs() < 1e-10, + "Each weight should be 0.5 for 2 models, got {}", + w + ); + } + } + + #[test] + fn test_normalize_effective_weights_single_model() { + let predictions = vec![ModelPrediction::new("SOLO".to_string(), 0.9, 0.95)]; + let mut weights = HashMap::new(); + weights.insert( + "SOLO".to_string(), + ModelWeight::new("SOLO".to_string(), 0.7), + ); + + let normalized = SignalAggregator::normalize_effective_weights(&predictions, &weights); + + assert_eq!(normalized.len(), 1); + let w = normalized.get("SOLO").copied().unwrap_or(0.0); + assert!( + (w - 1.0).abs() < 1e-10, + "Single model must get weight 1.0, got {}", + w + ); + } + + /// Verify DST-aware trading session boundaries. + /// + /// June 15 2026 is during EDT (UTC-4), so 13:30 UTC = 09:30 ET (Regular). + /// The old hardcoded UTC-5 would have mapped 13:30 UTC to 08:30 ET (PreMarket) -- wrong. + #[test] + fn test_trading_session_dst_correctness() { + // --- EDT (summer): UTC-4 --- + // June 15 2026 13:30 UTC => 09:30 ET => Regular + let june_1330 = Utc.with_ymd_and_hms(2026, 6, 15, 13, 30, 0).unwrap(); + assert_eq!( + EnsembleCoordinator::trading_session_at(june_1330), + TradingSession::Regular, + "13:30 UTC in June (EDT) should be 09:30 ET = Regular, not PreMarket" + ); + + // June 15 2026 20:00 UTC => 16:00 ET => AfterHours + let june_2000 = Utc.with_ymd_and_hms(2026, 6, 15, 20, 0, 0).unwrap(); + assert_eq!( + EnsembleCoordinator::trading_session_at(june_2000), + TradingSession::AfterHours, + "20:00 UTC in June (EDT) should be 16:00 ET = AfterHours" + ); + + // June 15 2026 08:00 UTC => 04:00 ET => PreMarket + let june_0800 = Utc.with_ymd_and_hms(2026, 6, 15, 8, 0, 0).unwrap(); + assert_eq!( + EnsembleCoordinator::trading_session_at(june_0800), + TradingSession::PreMarket, + "08:00 UTC in June (EDT) should be 04:00 ET = PreMarket" + ); + + // --- EST (winter): UTC-5 --- + // January 15 2026 14:30 UTC => 09:30 ET => Regular + let jan_1430 = Utc.with_ymd_and_hms(2026, 1, 15, 14, 30, 0).unwrap(); + assert_eq!( + EnsembleCoordinator::trading_session_at(jan_1430), + TradingSession::Regular, + "14:30 UTC in January (EST) should be 09:30 ET = Regular" + ); + + // January 15 2026 21:00 UTC => 16:00 ET => AfterHours + let jan_2100 = Utc.with_ymd_and_hms(2026, 1, 15, 21, 0, 0).unwrap(); + assert_eq!( + EnsembleCoordinator::trading_session_at(jan_2100), + TradingSession::AfterHours, + "21:00 UTC in January (EST) should be 16:00 ET = AfterHours" + ); + + // January 15 2026 03:00 UTC => 22:00 ET (prev day) => AfterHours + let jan_0300 = Utc.with_ymd_and_hms(2026, 1, 15, 3, 0, 0).unwrap(); + assert_eq!( + EnsembleCoordinator::trading_session_at(jan_0300), + TradingSession::AfterHours, + "03:00 UTC in January (EST) should be 22:00 ET = AfterHours" + ); + } }