diff --git a/ml/src/ensemble/coordinator.rs b/ml/src/ensemble/coordinator.rs index 796cbd537..83b53a1ff 100644 --- a/ml/src/ensemble/coordinator.rs +++ b/ml/src/ensemble/coordinator.rs @@ -4,9 +4,13 @@ //! from multiple ML models (DQN, PPO, TFT, MAMBA-2, Liquid, TLOB) for production trading decisions. //! Supports dynamic weighting based on performance and model diversity metrics. +use crate::ensemble::conviction_gates::{ + ConvictionGateConfig, ConvictionGateEvaluator, ConvictionGateOutcome, GateInput, TradingSession, +}; 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 std::collections::HashMap; use std::sync::Arc; use tokio::sync::RwLock; @@ -34,6 +38,9 @@ pub struct EnsembleCoordinator { /// Real model inference adapters for production predictions adapters: Vec>, + + /// Conviction gate evaluator (optional — None means no gating) + conviction_gates: Option, } impl std::fmt::Debug for EnsembleCoordinator { @@ -77,9 +84,28 @@ impl EnsembleCoordinator { model_weights: Arc::new(RwLock::new(HashMap::new())), config, adapters: Vec::new(), + conviction_gates: Some(ConvictionGateEvaluator::new(ConvictionGateConfig { + // Allow all sessions by default — trading service narrows to Regular + allowed_sessions: vec![ + TradingSession::PreMarket, + TradingSession::Regular, + TradingSession::AfterHours, + ], + ..ConvictionGateConfig::default() + })), } } + /// Replace conviction gate configuration + pub fn set_conviction_gates(&mut self, config: ConvictionGateConfig) { + self.conviction_gates = Some(ConvictionGateEvaluator::new(config)); + } + + /// Get mutable reference to conviction gate evaluator + pub fn conviction_gates_mut(&mut self) -> Option<&mut ConvictionGateEvaluator> { + self.conviction_gates.as_mut() + } + /// Add a real model inference adapter to the ensemble. /// When an adapter's model_name matches a registered model, its predictions /// are used for ensemble inference. @@ -129,6 +155,50 @@ impl EnsembleCoordinator { .aggregate(predictions, &*self.model_weights.read().await) .await?; + // Apply conviction gates if configured + if let Some(ref gates) = self.conviction_gates { + let gate_input = GateInput { + confidence: decision.confidence, + disagreement_rate: decision.disagreement_rate, + quorum_ratio: Self::calculate_quorum_ratio(&decision), + healthy_models: self.adapters.iter().filter(|a| a.is_ready()).count(), + total_models: self.adapters.len(), + current_session: Self::current_trading_session(), + regime_volatility: Self::extract_regime_volatility(features), + }; + + match gates.evaluate(&gate_input) { + ConvictionGateOutcome::Passed(pass_result) => { + info!( + "Conviction gates passed: score={:.3}, {} gates evaluated", + pass_result.conviction_score, + pass_result.gate_details.len() + ); + let mut gated_decision = decision; + gated_decision.metadata.insert( + "conviction_score".into(), + serde_json::Value::from(pass_result.conviction_score), + ); + return Ok(gated_decision); + } + ConvictionGateOutcome::Rejected(rejection) => { + info!("Conviction gate rejected: {:?} — forcing HOLD", rejection); + let mut hold_decision = EnsembleDecision::new( + TradingAction::Hold, + decision.confidence, + 0.0, + decision.disagreement_rate, + decision.model_votes, + ); + hold_decision.metadata.insert( + "gate_rejection".into(), + serde_json::Value::String(format!("{:?}", rejection)), + ); + return Ok(hold_decision); + } + } + } + info!( "Ensemble decision: {:?}, confidence: {:.3}, disagreement: {:.3}", decision.action, decision.confidence, decision.disagreement_rate @@ -292,6 +362,46 @@ impl EnsembleCoordinator { Ok(()) } + + /// Calculate quorum ratio (fraction of models agreeing on majority direction) + fn calculate_quorum_ratio(decision: &EnsembleDecision) -> f64 { + if decision.model_votes.is_empty() { + return 0.0; + } + let majority_action = &decision.action; + let agreeing = decision + .model_votes + .values() + .filter(|v| TradingAction::from_signal(v.signal, 0.3) == *majority_action) + .count(); + agreeing as f64 / decision.model_votes.len() as f64 + } + + /// Determine current trading session (Eastern Time, simplified UTC-5) + 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; + + match et_time { + t if t < 240 => TradingSession::AfterHours, // 00:00-04:00 ET + t if t < 570 => TradingSession::PreMarket, // 04:00-09:30 ET + t if t < 960 => TradingSession::Regular, // 09:30-16:00 ET + _ => TradingSession::AfterHours, // 16:00-24:00 ET + } + } + + /// Extract regime volatility from feature vector + /// Regime features are at indices 48-50 in the 51-dim standard feature vector + fn extract_regime_volatility(features: &Features) -> f64 { + features + .values + .get(48) + .copied() + .unwrap_or(0.01) + .abs() + } } impl Default for EnsembleCoordinator { diff --git a/ml/src/ensemble/mod.rs b/ml/src/ensemble/mod.rs index 50b023d78..15a699dcc 100644 --- a/ml/src/ensemble/mod.rs +++ b/ml/src/ensemble/mod.rs @@ -50,8 +50,8 @@ pub use metrics::{ pub use training_integration::EnsembleTrainingIntegration; pub use inference_adapter::{EnsemblePrediction, FeatureVector, ModelInferenceAdapter, PredictionMeta}; pub use conviction_gates::{ - ConvictionGateConfig, ConvictionGateOutcome, GateEvaluation, GatePassResult, GateRejection, - TradingSession, + ConvictionGateConfig, ConvictionGateEvaluator, ConvictionGateOutcome, GateEvaluation, + GateInput, GatePassResult, GateRejection, TradingSession, }; /// Errors that can occur in ensemble operations