diff --git a/ml/src/ensemble/coordinator.rs b/ml/src/ensemble/coordinator.rs index 04ae84c6a..796cbd537 100644 --- a/ml/src/ensemble/coordinator.rs +++ b/ml/src/ensemble/coordinator.rs @@ -32,7 +32,7 @@ pub struct EnsembleCoordinator { /// Configuration for ensemble behavior config: EnsembleConfig, - /// Real model inference adapters (tried before mock predictions) + /// Real model inference adapters for production predictions adapters: Vec>, } @@ -82,7 +82,7 @@ impl EnsembleCoordinator { /// Add a real model inference adapter to the ensemble. /// When an adapter's model_name matches a registered model, its predictions - /// are used instead of mock predictions. + /// are used for ensemble inference. pub fn add_adapter(&mut self, adapter: Box) { info!("Added inference adapter: {}", adapter.model_name()); self.adapters.push(adapter); @@ -100,14 +100,28 @@ impl EnsembleCoordinator { } /// Make ensemble prediction from features + /// + /// Returns an error if no model adapters are registered or all adapters fail inference. + /// Production trading must never rely on simulated/mock predictions. pub async fn predict(&self, features: &Features) -> MLResult { + if self.adapters.is_empty() { + return Err(MLError::ModelError( + "No model adapters registered".into(), + )); + } + debug!( "Making ensemble prediction with {} features", features.values.len() ); - // Mock model predictions (in production, these would be real model calls) - let predictions = self.generate_mock_predictions(features).await?; + let predictions = self.generate_predictions(features).await?; + + if predictions.is_empty() { + return Err(MLError::ModelError( + "All model adapters failed inference".into(), + )); + } // Aggregate predictions let decision = self @@ -123,32 +137,23 @@ impl EnsembleCoordinator { Ok(decision) } - /// Generate predictions from real ML models + /// Generate predictions from real ML model adapters /// - /// PRODUCTION: Uses real model inference from registered models - /// For testing/demo without loaded models, falls back to mock predictions - async fn generate_mock_predictions( + /// Iterates over registered models and attempts inference via their adapters. + /// Models without a ready adapter or whose adapter fails are skipped -- + /// no mock/simulated predictions are generated on the production path. + async fn generate_predictions( &self, features: &Features, ) -> MLResult> { - // Acquire locks and collect model info, then drop locks immediately - let model_info: Vec<(String, Option)> = { - let registry = self.active_models.read().await; + let model_ids: Vec = { let weights = self.model_weights.read().await; - - weights - .iter() - .map(|(model_id, _)| { - let checkpoint = registry.active.get(model_id).cloned(); - (model_id.clone(), checkpoint) - }) - .collect() - }; // Locks are dropped here + weights.keys().cloned().collect() + }; let mut predictions = Vec::new(); - for (model_id, checkpoint_opt) in model_info { - // Try real adapter first + for model_id in model_ids { if let Some(adapter) = self .adapters .iter() @@ -166,97 +171,28 @@ impl EnsembleCoordinator { ensemble_pred.confidence, ); predictions.push(prediction); - continue; } Err(e) => { warn!( - "Adapter {} inference failed, falling back to mock: {}", + "Adapter {} inference failed, skipping: {}", model_id, e ); } } - } - - // Try to get real model from registry - if let Some(checkpoint_path) = checkpoint_opt { - debug!( - "Model {} loaded from checkpoint: {}", - model_id, checkpoint_path - ); - // In production, actual model inference would happen here - // For now, we'll use enhanced mock predictions that simulate real model behavior - let value = self.simulate_trained_model_prediction(&model_id, features); - let confidence = 0.80 + (value.abs() * 0.15); // Higher confidence for "trained" models - - let prediction = ModelPrediction::new(model_id.clone(), value, confidence); - predictions.push(prediction); } else { - // Fallback to basic mock prediction for unloaded models - let value = self.mock_model_prediction(&model_id, features); - let confidence = 0.70 + (value.abs() * 0.2); - - let prediction = ModelPrediction::new(model_id.clone(), value, confidence); - predictions.push(prediction); + warn!( + "No ready adapter for model {}, skipping", + model_id + ); } } Ok(predictions) } - /// Simulate trained model prediction (more realistic than mock) - fn simulate_trained_model_prediction(&self, model_id: &str, features: &Features) -> f64 { - let feature_sum: f64 = features.values.iter().take(5).sum(); - let feature_mean = feature_sum / 5.0; - - // Simulate different model architectures with trained-like behavior - match model_id { - "DQN" => { - // Deep Q-Network: action-value based decision - let q_buy = - (feature_mean * 0.85 + features.values.get(1).unwrap_or(&0.0) * 0.15).tanh(); - let q_sell = - (feature_mean * -0.80 + features.values.get(2).unwrap_or(&0.0) * 0.20).tanh(); - (q_buy - q_sell) / 2.0 // Normalized difference - }, - "PPO" => { - // Proximal Policy Optimization: policy gradient based - let policy_logit = - feature_mean * 0.92 + features.values.get(3).unwrap_or(&0.0) * 0.08; - policy_logit.tanh() * 0.95 // High confidence policy - }, - "TFT" => { - // Temporal Fusion Transformer: attention-based temporal patterns - let temporal_signal = features.values.iter().take(4).sum::() / 4.0; - (temporal_signal * 0.75).tanh() - }, - "MAMBA-2" => { - // State-space selective mechanism (0.80 multiplier) - let state_signal = features.values.iter().take(6).sum::() / 6.0; - let selective_weight = (state_signal.abs() * 2.0).tanh(); - (state_signal * 0.80 * selective_weight).tanh() - }, - "TFT-INT8" => { - // TFT with INT8 quantization: same architecture as TFT, memory-optimized - let temporal_signal = features.values.iter().take(4).sum::() / 4.0; - (temporal_signal * 0.75).tanh() - }, - _ => 0.0, - } - } - - /// Mock model prediction (basic fallback) - fn mock_model_prediction(&self, model_id: &str, features: &Features) -> f64 { - let feature_sum: f64 = features.values.iter().take(5).sum(); - let feature_mean = feature_sum / 5.0; - - match model_id { - "DQN" => (feature_mean * 0.8).tanh(), - "PPO" => (feature_mean * 0.9).tanh(), - "TFT" => (feature_mean * 0.7).tanh(), - "MAMBA-2" => (feature_mean * 0.85).tanh(), - "TFT-INT8" => (feature_mean * 0.7).tanh(), // Same behavior as TFT - _ => 0.0, - } + /// Check whether any registered adapter is ready for inference + pub fn has_models(&self) -> bool { + self.adapters.iter().any(|a| a.is_ready()) } /// Update model weights based on performance @@ -628,7 +564,41 @@ mod tests { #[tokio::test] async fn test_ensemble_prediction() { - let coordinator = EnsembleCoordinator::new(); + use crate::ensemble::inference_adapter::{ + EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta, + }; + + struct StubAdapter { + name: &'static str, + direction: f64, + confidence: f64, + } + impl ModelInferenceAdapter for StubAdapter { + fn model_name(&self) -> &str { + self.name + } + fn predict( + &self, + _features: &FeatureVector, + ) -> crate::MLResult { + Ok(EnsemblePrediction { + model_name: self.name.to_string(), + direction: self.direction, + confidence: self.confidence, + metadata: PredictionMeta::default(), + }) + } + fn is_ready(&self) -> bool { + true + } + } + unsafe impl Send for StubAdapter {} + unsafe impl Sync for StubAdapter {} + + let mut coordinator = EnsembleCoordinator::new(); + coordinator.add_adapter(Box::new(StubAdapter { name: "DQN", direction: 0.6, confidence: 0.9 })); + coordinator.add_adapter(Box::new(StubAdapter { name: "PPO", direction: 0.5, confidence: 0.85 })); + coordinator.add_adapter(Box::new(StubAdapter { name: "TFT", direction: 0.4, confidence: 0.8 })); coordinator .register_model("DQN".to_string(), 0.33) @@ -800,7 +770,7 @@ mod tests { } #[tokio::test] - async fn test_ensemble_graceful_no_adapters() { + async fn test_ensemble_rejects_no_adapters() { let coordinator = EnsembleCoordinator::new(); coordinator .register_model("DQN".to_string(), 0.5) @@ -818,7 +788,13 @@ mod tests { .map(String::from) .collect(), ); - let decision = coordinator.predict(&features).await.unwrap(); - assert!(decision.confidence >= 0.0); + let result = coordinator.predict(&features).await; + assert!(result.is_err(), "predict() must fail when no adapters are registered"); + let err_msg = format!("{}", result.unwrap_err()); + assert!( + err_msg.contains("No model adapters registered"), + "Expected 'No model adapters registered' error, got: {}", + err_msg + ); } }