From dc94c0a757a9b4ba8bf575673363e40b4c9c0fd1 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 23 Feb 2026 13:49:47 +0100 Subject: [PATCH] feat(trading): add feedback loop orchestrator and wire into service init FeedbackLoop runs as a background tokio task, periodically running weight and gate optimization cycles with kill switch monitoring and retraining trigger detection. Wired into main.rs with a stub MetricsProvider until QuestDB is deployed. 7 new tests covering kill switch, freeze/unfreeze, retraining triggers. Co-Authored-By: Claude Opus 4.6 --- services/trading_service/src/feedback_loop.rs | 528 ++++++++++++++++++ services/trading_service/src/lib.rs | 3 + services/trading_service/src/main.rs | 54 ++ 3 files changed, 585 insertions(+) create mode 100644 services/trading_service/src/feedback_loop.rs diff --git a/services/trading_service/src/feedback_loop.rs b/services/trading_service/src/feedback_loop.rs new file mode 100644 index 000000000..0d93b1c5b --- /dev/null +++ b/services/trading_service/src/feedback_loop.rs @@ -0,0 +1,528 @@ +//! 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, Instant}; +use tokio::sync::Mutex; +use tracing::{error, info, warn}; + +/// 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 { + warn!( + 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() { + warn!( + 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::*; + + /// 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 mut config = FeedbackLoopConfig::default(); + config.cycle_interval = Duration::from_millis(100); + + 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); + } +} diff --git a/services/trading_service/src/lib.rs b/services/trading_service/src/lib.rs index fb6c400d2..ea79ddc8a 100644 --- a/services/trading_service/src/lib.rs +++ b/services/trading_service/src/lib.rs @@ -166,6 +166,9 @@ pub mod health; /// P&L attribution: decomposes realized trade P&L into per-model contributions pub mod attribution; +/// Autonomous feedback loop: weight/gate optimization with kill switch +pub mod feedback_loop; + // Re-export for tests pub use ensemble_coordinator::EnsembleCoordinator; pub use paper_trading_executor::PaperTradingExecutor; diff --git a/services/trading_service/src/main.rs b/services/trading_service/src/main.rs index 918150257..91d619a43 100644 --- a/services/trading_service/src/main.rs +++ b/services/trading_service/src/main.rs @@ -37,6 +37,7 @@ use trading_service::services::ml_performance_monitor::MLPerformanceMonitor; use trading_service::services::monitoring::MonitoringServiceImpl; use trading_service::services::risk::RiskServiceImpl; use trading_service::services::trading::TradingServiceImpl; +use trading_service::feedback_loop::{FeedbackLoop, FeedbackLoopConfig}; use trading_service::state::TradingServiceState; /// Default configuration values @@ -528,6 +529,34 @@ async fn main() -> Result<()> { warn!("⚠️ Ensemble coordinator not available - prediction generation loop disabled"); } + // Initialize autonomous feedback loop (weight + gate optimization with kill switch) + { + let feedback_config = FeedbackLoopConfig { + cycle_interval: Duration::from_secs( + std::env::var("FEEDBACK_LOOP_INTERVAL_SECS") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(24 * 3600), + ), + kill_switch_threshold: std::env::var("FEEDBACK_KILL_SWITCH_SHARPE") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(-1.0), + ..Default::default() + }; + + let feedback_loop = Arc::new(FeedbackLoop::new(feedback_config)); + + // Stub metrics provider: returns no data until QuestDB is configured. + // When QuestDB is deployed, replace with a real provider that queries + // rolling_model_metrics and confidence_bucket_stats tables. + let metrics_provider: Arc = + Arc::new(StubMetricsProvider); + + let _feedback_handle = Arc::clone(&feedback_loop).spawn(metrics_provider); + info!("Autonomous feedback loop spawned (cycle=env:FEEDBACK_LOOP_INTERVAL_SECS, default=24h)"); + } + // Subscribe to ML performance alerts let monitor_clone = Arc::clone(&ml_performance_monitor); tokio::spawn(async move { @@ -750,6 +779,31 @@ async fn main() -> Result<()> { Ok(()) } +/// Stub metrics provider — returns no data, causing optimizers to report InsufficientData. +/// Replace with QuestDB-backed provider when time-series DB is deployed. +struct StubMetricsProvider; + +impl trading_service::feedback_loop::MetricsProvider for StubMetricsProvider { + fn get_model_metrics(&self) -> Vec { + Vec::new() + } + fn get_gate_buckets(&self) -> Vec { + Vec::new() + } + fn get_current_weights(&self) -> std::collections::HashMap { + std::collections::HashMap::new() + } + fn get_gate_config(&self) -> ml::ensemble::conviction_gates::ConvictionGateConfig { + ml::ensemble::conviction_gates::ConvictionGateConfig::default() + } + fn get_ensemble_sharpe_7d(&self) -> f64 { + 0.0 // Neutral — won't trigger kill switch + } + fn get_model_correlations(&self) -> Vec<((String, String), f64)> { + Vec::new() + } +} + /// Initialize authentication configuration /// /// NOTE: Authentication is handled by API Gateway (Wave 70)