//! Autonomous Feedback Loop Orchestrator //! //! Runs as a background tokio task. Periodically: //! 1. Queries rolling performance metrics (from QuestDB or in-memory) //! 2. Runs weight optimizer → applies weight adjustments //! 3. Runs gate optimizer → applies threshold adjustments //! 4. Checks retraining triggers (Sharpe < threshold, accuracy < threshold) //! 5. Monitors ensemble Sharpe for kill switch activation use ml::ensemble::conviction_gates::ConvictionGateConfig; use ml::ensemble::gate_optimizer::{ GateBucketMetrics, GateOptimizationResult, GateOptimizer, GateOptimizerConfig, }; use ml::ensemble::weight_optimizer::{ ModelRollingMetrics, OptimizationResult, WeightOptimizer, WeightOptimizerConfig, }; use std::collections::HashMap; use std::sync::Arc; use std::time::Duration; use tokio::sync::Mutex; use tracing::{error, info}; /// Retraining trigger reasons #[derive(Debug, Clone)] pub enum RetrainingTrigger { /// Sharpe ratio dropped below threshold LowSharpe { model_id: String, sharpe: f64 }, /// Accuracy dropped below threshold LowAccuracy { model_id: String, accuracy: f64 }, /// Models are too correlated HighCorrelation { pair: (String, String), correlation: f64 }, } /// Configuration for the feedback loop #[derive(Debug, Clone)] pub struct FeedbackLoopConfig { /// How often to run the optimization cycle pub cycle_interval: Duration, /// Kill switch: if ensemble Sharpe drops below this, freeze all optimizers pub kill_switch_threshold: f64, /// Sharpe ratio threshold for retraining trigger pub retrain_sharpe_threshold: f64, /// Accuracy threshold for retraining trigger pub retrain_accuracy_threshold: f64, /// Correlation threshold for retraining trigger pub retrain_correlation_threshold: f64, } impl Default for FeedbackLoopConfig { fn default() -> Self { Self { cycle_interval: Duration::from_secs(24 * 3600), // 24h kill_switch_threshold: -1.0, retrain_sharpe_threshold: -0.5, retrain_accuracy_threshold: 0.45, retrain_correlation_threshold: 0.90, } } } /// Result of a single feedback loop cycle #[derive(Debug, Clone)] pub struct CycleResult { /// Weight adjustments applied (if any) pub weight_result: OptimizationResult, /// Gate adjustments applied (if any) pub gate_result: GateOptimizationResult, /// Retraining triggers detected pub retraining_triggers: Vec, /// Whether kill switch was activated this cycle pub kill_switch_activated: bool, } /// Trait for providing rolling metrics to the feedback loop. /// In production, this queries QuestDB. In tests, it's mocked. pub trait MetricsProvider: Send + Sync { /// Get rolling model-level metrics for weight optimization fn get_model_metrics(&self) -> Vec; /// Get confidence-bucketed metrics for gate optimization fn get_gate_buckets(&self) -> Vec; /// Get current model weights fn get_current_weights(&self) -> HashMap; /// Get current conviction gate config fn get_gate_config(&self) -> ConvictionGateConfig; /// Get 7-day ensemble Sharpe ratio fn get_ensemble_sharpe_7d(&self) -> f64; /// Get model pairwise correlations (for retraining triggers) fn get_model_correlations(&self) -> Vec<((String, String), f64)>; } /// Autonomous feedback loop orchestrator pub struct FeedbackLoop { weight_optimizer: Arc>, gate_optimizer: Arc>, config: FeedbackLoopConfig, } impl FeedbackLoop { pub fn new(config: FeedbackLoopConfig) -> Self { Self { weight_optimizer: Arc::new(Mutex::new(WeightOptimizer::new( WeightOptimizerConfig::default(), ))), gate_optimizer: Arc::new(Mutex::new(GateOptimizer::new( GateOptimizerConfig::default(), ))), config, } } /// Create with custom optimizer configs pub fn with_optimizers( config: FeedbackLoopConfig, weight_config: WeightOptimizerConfig, gate_config: GateOptimizerConfig, ) -> Self { Self { weight_optimizer: Arc::new(Mutex::new(WeightOptimizer::new(weight_config))), gate_optimizer: Arc::new(Mutex::new(GateOptimizer::new(gate_config))), config, } } /// Run one optimization cycle. Returns the results for observability. pub async fn run_cycle(&self, metrics: &dyn MetricsProvider) -> CycleResult { // 1. Check kill switch let ensemble_sharpe = metrics.get_ensemble_sharpe_7d(); let kill_switch_activated = ensemble_sharpe < self.config.kill_switch_threshold; if kill_switch_activated { error!( ensemble_sharpe = ensemble_sharpe, threshold = self.config.kill_switch_threshold, "Kill switch activated — freezing all optimizers" ); self.weight_optimizer.lock().await.freeze(); self.gate_optimizer.lock().await.freeze(); return CycleResult { weight_result: OptimizationResult::KillSwitchActive, gate_result: GateOptimizationResult::KillSwitchActive, retraining_triggers: vec![], kill_switch_activated: true, }; } // 2. Run weight optimization let model_metrics = metrics.get_model_metrics(); let current_weights = metrics.get_current_weights(); let weight_result = self .weight_optimizer .lock() .await .optimize(¤t_weights, &model_metrics); if let OptimizationResult::Adjusted(ref adjustments) = weight_result { info!( count = adjustments.len(), "Feedback loop: weight adjustments proposed" ); } // 3. Run gate optimization let gate_config = metrics.get_gate_config(); let buckets = metrics.get_gate_buckets(); let gate_result = self .gate_optimizer .lock() .await .optimize(&gate_config, &buckets); if let GateOptimizationResult::Adjusted(ref adjustments) = gate_result { info!( count = adjustments.len(), "Feedback loop: gate adjustments proposed" ); } // 4. Check retraining triggers let retraining_triggers = self.check_retraining_triggers(metrics); if !retraining_triggers.is_empty() { info!( count = retraining_triggers.len(), "Feedback loop: retraining triggers detected" ); } CycleResult { weight_result, gate_result, retraining_triggers, kill_switch_activated: false, } } /// Spawn the feedback loop as a background tokio task. /// /// Returns a handle that can be used to abort the loop. pub fn spawn( self: Arc, metrics: Arc, ) -> tokio::task::JoinHandle<()> { let interval = self.config.cycle_interval; tokio::spawn(async move { let mut ticker = tokio::time::interval(interval); // Skip the first immediate tick ticker.tick().await; loop { ticker.tick().await; info!("Feedback loop: starting optimization cycle"); let result = self.run_cycle(metrics.as_ref()).await; if result.kill_switch_activated { error!("Feedback loop: kill switch active, pausing until manual intervention"); // Sleep longer when kill switch is active to avoid log spam tokio::time::sleep(Duration::from_secs(3600)).await; } } }) } /// Freeze both optimizers (manual kill switch) pub async fn freeze(&self) { self.weight_optimizer.lock().await.freeze(); self.gate_optimizer.lock().await.freeze(); } /// Unfreeze both optimizers (requires human acknowledgment) pub async fn unfreeze(&self) { self.weight_optimizer.lock().await.unfreeze(); self.gate_optimizer.lock().await.unfreeze(); } /// Check if either optimizer is frozen pub async fn is_frozen(&self) -> bool { self.weight_optimizer.lock().await.is_frozen() || self.gate_optimizer.lock().await.is_frozen() } fn check_retraining_triggers(&self, metrics: &dyn MetricsProvider) -> Vec { let mut triggers = Vec::new(); // Check per-model metrics for m in &metrics.get_model_metrics() { if m.sharpe_30d < self.config.retrain_sharpe_threshold { triggers.push(RetrainingTrigger::LowSharpe { model_id: m.model_id.clone(), sharpe: m.sharpe_30d, }); } if m.prediction_accuracy < self.config.retrain_accuracy_threshold { triggers.push(RetrainingTrigger::LowAccuracy { model_id: m.model_id.clone(), accuracy: m.prediction_accuracy, }); } } // Check correlations for ((a, b), corr) in &metrics.get_model_correlations() { if *corr > self.config.retrain_correlation_threshold { triggers.push(RetrainingTrigger::HighCorrelation { pair: (a.clone(), b.clone()), correlation: *corr, }); } } triggers } } impl std::fmt::Debug for FeedbackLoop { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("FeedbackLoop") .field("config", &self.config) .finish() } } #[cfg(test)] mod tests { use super::*; use std::time::Instant; /// Mock metrics provider for testing struct MockMetrics { model_metrics: Vec, gate_buckets: Vec, weights: HashMap, gate_config: ConvictionGateConfig, ensemble_sharpe: f64, correlations: Vec<((String, String), f64)>, } impl MockMetrics { fn healthy() -> Self { Self { model_metrics: vec![ ModelRollingMetrics { model_id: "dqn".into(), sharpe_30d: 1.5, win_rate_30d: 0.58, prediction_accuracy: 0.62, trade_count: 200, deployed_at: Instant::now() - Duration::from_secs(30 * 24 * 3600), }, ModelRollingMetrics { model_id: "ppo".into(), sharpe_30d: 0.8, win_rate_30d: 0.55, prediction_accuracy: 0.58, trade_count: 200, deployed_at: Instant::now() - Duration::from_secs(30 * 24 * 3600), }, ], gate_buckets: vec![ GateBucketMetrics { confidence_lower: 0.50, confidence_upper: 0.60, win_rate: 0.56, trade_count: 100, avg_pnl: 0.002, }, GateBucketMetrics { confidence_lower: 0.60, confidence_upper: 0.70, win_rate: 0.58, trade_count: 100, avg_pnl: 0.003, }, GateBucketMetrics { confidence_lower: 0.70, confidence_upper: 0.80, win_rate: 0.62, trade_count: 100, avg_pnl: 0.005, }, ], weights: { let mut w = HashMap::new(); w.insert("dqn".into(), 0.5); w.insert("ppo".into(), 0.5); w }, gate_config: ConvictionGateConfig::default(), ensemble_sharpe: 1.2, correlations: vec![], } } } impl MetricsProvider for MockMetrics { fn get_model_metrics(&self) -> Vec { self.model_metrics.clone() } fn get_gate_buckets(&self) -> Vec { self.gate_buckets.clone() } fn get_current_weights(&self) -> HashMap { self.weights.clone() } fn get_gate_config(&self) -> ConvictionGateConfig { self.gate_config.clone() } fn get_ensemble_sharpe_7d(&self) -> f64 { self.ensemble_sharpe } fn get_model_correlations(&self) -> Vec<((String, String), f64)> { self.correlations.clone() } } #[tokio::test] async fn test_healthy_cycle_adjusts_weights() { let config = FeedbackLoopConfig { cycle_interval: Duration::from_millis(100), ..Default::default() }; let feedback = FeedbackLoop::with_optimizers( config, WeightOptimizerConfig { cooldown: Duration::ZERO, ..Default::default() }, GateOptimizerConfig { cooldown: Duration::ZERO, ..Default::default() }, ); let metrics = MockMetrics::healthy(); let result = feedback.run_cycle(&metrics).await; // Should produce weight adjustments (models have different Sharpe) assert!(matches!(result.weight_result, OptimizationResult::Adjusted(_))); assert!(!result.kill_switch_activated); assert!(result.retraining_triggers.is_empty()); } #[tokio::test] async fn test_kill_switch_freezes_optimizers() { let feedback = FeedbackLoop::new(FeedbackLoopConfig { kill_switch_threshold: -1.0, ..Default::default() }); let mut metrics = MockMetrics::healthy(); metrics.ensemble_sharpe = -2.0; // Below kill switch let result = feedback.run_cycle(&metrics).await; assert!(result.kill_switch_activated); assert!(matches!( result.weight_result, OptimizationResult::KillSwitchActive )); assert!(matches!( result.gate_result, GateOptimizationResult::KillSwitchActive )); assert!(feedback.is_frozen().await); } #[tokio::test] async fn test_manual_freeze_unfreeze() { let feedback = FeedbackLoop::new(FeedbackLoopConfig::default()); assert!(!feedback.is_frozen().await); feedback.freeze().await; assert!(feedback.is_frozen().await); let metrics = MockMetrics::healthy(); let result = feedback.run_cycle(&metrics).await; assert!(matches!( result.weight_result, OptimizationResult::KillSwitchActive )); feedback.unfreeze().await; assert!(!feedback.is_frozen().await); } #[tokio::test] async fn test_retraining_trigger_low_sharpe() { let config = FeedbackLoopConfig { retrain_sharpe_threshold: -0.5, ..Default::default() }; let feedback = FeedbackLoop::new(config); let mut metrics = MockMetrics::healthy(); metrics.model_metrics[1].sharpe_30d = -1.0; // PPO tanking let result = feedback.run_cycle(&metrics).await; assert_eq!(result.retraining_triggers.len(), 1); if let RetrainingTrigger::LowSharpe { ref model_id, .. } = result.retraining_triggers[0] { assert_eq!(model_id, "ppo"); } else { panic!("Expected LowSharpe trigger"); } } #[tokio::test] async fn test_retraining_trigger_low_accuracy() { let config = FeedbackLoopConfig { retrain_accuracy_threshold: 0.45, ..Default::default() }; let feedback = FeedbackLoop::new(config); let mut metrics = MockMetrics::healthy(); metrics.model_metrics[0].prediction_accuracy = 0.40; // DQN accuracy dropped let result = feedback.run_cycle(&metrics).await; let accuracy_triggers: Vec<_> = result .retraining_triggers .iter() .filter(|t| matches!(t, RetrainingTrigger::LowAccuracy { .. })) .collect(); assert_eq!(accuracy_triggers.len(), 1); } #[tokio::test] async fn test_retraining_trigger_high_correlation() { let config = FeedbackLoopConfig { retrain_correlation_threshold: 0.90, ..Default::default() }; let feedback = FeedbackLoop::new(config); let mut metrics = MockMetrics::healthy(); metrics.correlations = vec![ (("dqn".into(), "ppo".into()), 0.95), // Too correlated ]; let result = feedback.run_cycle(&metrics).await; let corr_triggers: Vec<_> = result .retraining_triggers .iter() .filter(|t| matches!(t, RetrainingTrigger::HighCorrelation { .. })) .collect(); assert_eq!(corr_triggers.len(), 1); } #[tokio::test] async fn test_no_triggers_healthy_system() { let feedback = FeedbackLoop::new(FeedbackLoopConfig::default()); let metrics = MockMetrics::healthy(); let result = feedback.run_cycle(&metrics).await; assert!(result.retraining_triggers.is_empty()); assert!(!result.kill_switch_activated); } }