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 <noreply@anthropic.com>

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<RetrainingTrigger>,
+    /// 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<ModelRollingMetrics>;
+
+    /// Get confidence-bucketed metrics for gate optimization
+    fn get_gate_buckets(&self) -> Vec<GateBucketMetrics>;
+
+    /// Get current model weights
+    fn get_current_weights(&self) -> HashMap<String, f64>;
+
+    /// 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<Mutex<WeightOptimizer>>,
+    gate_optimizer: Arc<Mutex<GateOptimizer>>,
+    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(&current_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<Self>,
+        metrics: Arc<dyn MetricsProvider>,
+    ) -> 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<RetrainingTrigger> {
+        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<ModelRollingMetrics>,
+        gate_buckets: Vec<GateBucketMetrics>,
+        weights: HashMap<String, f64>,
+        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<ModelRollingMetrics> {
+            self.model_metrics.clone()
+        }
+        fn get_gate_buckets(&self) -> Vec<GateBucketMetrics> {
+            self.gate_buckets.clone()
+        }
+        fn get_current_weights(&self) -> HashMap<String, f64> {
+            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);
+    }
+}

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;

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<dyn trading_service::feedback_loop::MetricsProvider> =
+            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<ml::ensemble::weight_optimizer::ModelRollingMetrics> {
+        Vec::new()
+    }
+    fn get_gate_buckets(&self) -> Vec<ml::ensemble::gate_optimizer::GateBucketMetrics> {
+        Vec::new()
+    }
+    fn get_current_weights(&self) -> std::collections::HashMap<String, f64> {
+        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)