From f0bafdf30bcfb20b2cfb291c44d3ade1004b4165 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 23 Feb 2026 12:11:28 +0100 Subject: [PATCH] docs: add operational maturity implementation plan (12 tasks, 6 phases) Three pillars: 7-gate conviction system, autonomous feedback loop with kill switch, Rust-native model registry. QuestDB analytics layer. ~55 tests, 7 new files planned across ml/ and trading_service/. Co-Authored-By: Claude Opus 4.6 --- ...-23-operational-maturity-implementation.md | 1800 +++++++++++++++++ 1 file changed, 1800 insertions(+) create mode 100644 docs/plans/2026-02-23-operational-maturity-implementation.md diff --git a/docs/plans/2026-02-23-operational-maturity-implementation.md b/docs/plans/2026-02-23-operational-maturity-implementation.md new file mode 100644 index 000000000..6617323b5 --- /dev/null +++ b/docs/plans/2026-02-23-operational-maturity-implementation.md @@ -0,0 +1,1800 @@ +# Operational Maturity System Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Build a 7-gate conviction system, autonomous feedback loop with kill switch, Rust-native model registry, and QuestDB time-series analytics — closing the loop from prediction to execution to learning. + +**Architecture:** Three pillars integrated through PostgreSQL (audit, registry) and QuestDB (time-series analytics). The conviction gates sit in the critical trading path. The feedback loop and QuestDB are non-critical — trading continues if they fail. The model registry extends existing `ml_model_versions` with lifecycle tracking. + +**Tech Stack:** Rust, Candle, sqlx 0.8.6 (PostgreSQL), questdb-rs 6.x (ILP ingestion), tokio (async), prometheus (metrics) + +**Design doc:** `docs/plans/2026-02-23-operational-maturity-design.md` + +--- + +## Phase 1: Conviction Gates (Critical Path) + +### Task 1: ConvictionGateConfig and GateResult types + +**Files:** +- Create: `ml/src/ensemble/conviction_gates.rs` +- Modify: `ml/src/ensemble/mod.rs:7-23` (add module declaration) + +**Step 1: Write the failing test** + +In `ml/src/ensemble/conviction_gates.rs`, add the module with tests at the bottom: + +```rust +//! 7-Gate Conviction System +//! +//! Evaluates ensemble predictions through a series of quality gates before +//! allowing trade execution. Gates are evaluated in order; any failure +//! results in HOLD. + +use serde::{Deserialize, Serialize}; + +/// Trading session windows (Eastern Time) +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum TradingSession { + /// Pre-market: 04:00-09:30 ET + PreMarket, + /// Regular hours: 09:30-16:00 ET + Regular, + /// After-hours: 16:00-20:00 ET + AfterHours, +} + +/// Configuration for the 7-gate conviction system +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ConvictionGateConfig { + /// Gate 1: Minimum ratio of healthy models (default: 0.70) + pub model_health_threshold: f64, + /// Gate 2: Allowed trading sessions (default: Regular only) + pub allowed_sessions: Vec, + /// Gate 3: Minimum ensemble confidence (default: 0.60) + pub min_confidence: f64, + /// Gate 4: Maximum disagreement rate (default: 0.40) + pub max_disagreement: f64, + /// Gate 5: Minimum quorum ratio (default: 0.60) + pub min_quorum: f64, + /// Gate 6: Threshold tightening factor in high-vol regimes (default: 0.80) + pub regime_tightening_factor: f64, + /// Gate 6: Volatility level above which regime filter activates + pub high_vol_threshold: f64, + /// Gate 7: Enable conviction-based position sizing + pub conviction_scaling_enabled: bool, +} + +impl Default for ConvictionGateConfig { + fn default() -> Self { + Self { + model_health_threshold: 0.70, + allowed_sessions: vec![TradingSession::Regular], + min_confidence: 0.60, + max_disagreement: 0.40, + min_quorum: 0.60, + regime_tightening_factor: 0.80, + high_vol_threshold: 0.03, + conviction_scaling_enabled: true, + } + } +} + +/// Which gate rejected the decision (if any) +#[derive(Debug, Clone, PartialEq)] +pub enum GateRejection { + ModelHealth { healthy_ratio: f64, threshold: f64 }, + TimeOfDay { current_session: TradingSession }, + Confidence { confidence: f64, threshold: f64 }, + Agreement { disagreement: f64, threshold: f64 }, + Quorum { quorum_ratio: f64, threshold: f64 }, + Regime { adjusted_confidence: f64, adjusted_threshold: f64 }, +} + +/// Result of passing all conviction gates +#[derive(Debug, Clone)] +pub struct GatePassResult { + /// Conviction scalar (0.0 to 1.0) for position sizing + pub conviction_score: f64, + /// Details of each gate evaluation + pub gate_details: Vec, +} + +/// Individual gate evaluation result +#[derive(Debug, Clone)] +pub struct GateEvaluation { + pub gate_name: String, + pub value: f64, + pub threshold: f64, + pub passed: bool, +} + +/// Outcome of evaluating all gates +#[derive(Debug, Clone)] +pub enum ConvictionGateOutcome { + /// All gates passed — trade is allowed with conviction score + Passed(GatePassResult), + /// A gate rejected — trade should be HOLD + Rejected(GateRejection), +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_default_config_values() { + let config = ConvictionGateConfig::default(); + assert!((config.model_health_threshold - 0.70).abs() < 1e-10); + assert_eq!(config.allowed_sessions, vec![TradingSession::Regular]); + assert!((config.min_confidence - 0.60).abs() < 1e-10); + assert!((config.max_disagreement - 0.40).abs() < 1e-10); + assert!((config.min_quorum - 0.60).abs() < 1e-10); + assert!((config.regime_tightening_factor - 0.80).abs() < 1e-10); + assert!(config.conviction_scaling_enabled); + } + + #[test] + fn test_gate_rejection_variants() { + let rejection = GateRejection::Confidence { + confidence: 0.45, + threshold: 0.60, + }; + assert_eq!( + rejection, + GateRejection::Confidence { + confidence: 0.45, + threshold: 0.60 + } + ); + } +} +``` + +**Step 2: Register the module** + +In `ml/src/ensemble/mod.rs`, add after line 23 (`pub mod adapters;`): + +```rust +pub mod conviction_gates; +``` + +And add re-export after the existing re-exports (after line 50): + +```rust +pub use conviction_gates::{ + ConvictionGateConfig, ConvictionGateOutcome, GateEvaluation, GatePassResult, GateRejection, + TradingSession, +}; +``` + +**Step 3: Run test to verify it passes** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::conviction_gates --features cuda` +Expected: 2 tests pass + +**Step 4: Commit** + +```bash +git add ml/src/ensemble/conviction_gates.rs ml/src/ensemble/mod.rs +git commit -m "feat(ensemble): add ConvictionGateConfig types and gate result enums" +``` + +--- + +### Task 2: ConvictionGateEvaluator with gates 1-5 + +**Files:** +- Modify: `ml/src/ensemble/conviction_gates.rs` + +**Step 1: Write the failing tests** + +Add these tests to the existing `mod tests` block in `conviction_gates.rs`: + +```rust + #[test] + fn test_gate1_model_health_passes() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.20, + quorum_ratio: 0.80, + healthy_models: 8, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Passed(_))); + } + + #[test] + fn test_gate1_model_health_rejects() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.90, + healthy_models: 5, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::ModelHealth { .. }))); + } + + #[test] + fn test_gate2_time_of_day_rejects_premarket() { + let config = ConvictionGateConfig::default(); // Regular only + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.90, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::PreMarket, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::TimeOfDay { .. }))); + } + + #[test] + fn test_gate3_confidence_rejects_low() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.45, + disagreement_rate: 0.10, + quorum_ratio: 0.90, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::Confidence { .. }))); + } + + #[test] + fn test_gate4_agreement_rejects_high_disagreement() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.55, + quorum_ratio: 0.90, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::Agreement { .. }))); + } + + #[test] + fn test_gate5_quorum_rejects_low() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.40, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::Quorum { .. }))); + } + + #[test] + fn test_conviction_score_calculation() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.80, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + if let ConvictionGateOutcome::Passed(result) = evaluator.evaluate(&input) { + // conviction = confidence * (1 - disagreement) * quorum * health + // = 0.80 * 0.90 * 0.80 * 0.90 = 0.5184 + assert!(result.conviction_score > 0.0); + assert!(result.conviction_score <= 1.0); + assert!((result.conviction_score - 0.5184).abs() < 0.01); + } else { + panic!("Expected Passed outcome"); + } + } +``` + +**Step 2: Run tests to verify they fail** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::conviction_gates --features cuda` +Expected: FAIL — `ConvictionGateEvaluator` not found + +**Step 3: Implement ConvictionGateEvaluator** + +Add above the `#[cfg(test)]` block in `conviction_gates.rs`: + +```rust +/// Input data for gate evaluation +#[derive(Debug, Clone)] +pub struct GateInput { + /// Ensemble confidence (0.0-1.0) + pub confidence: f64, + /// Disagreement rate (0.0-1.0) + pub disagreement_rate: f64, + /// Ratio of models agreeing on direction (0.0-1.0) + pub quorum_ratio: f64, + /// Number of healthy models + pub healthy_models: usize, + /// Total number of models + pub total_models: usize, + /// Current trading session + pub current_session: TradingSession, + /// Current regime volatility (from feature vector) + pub regime_volatility: f64, +} + +/// Evaluates ensemble decisions through 7 conviction gates +#[derive(Debug, Clone)] +pub struct ConvictionGateEvaluator { + config: ConvictionGateConfig, +} + +impl ConvictionGateEvaluator { + pub fn new(config: ConvictionGateConfig) -> Self { + Self { config } + } + + /// Returns a mutable reference to the config for autonomous adjustment + pub fn config_mut(&mut self) -> &mut ConvictionGateConfig { + &mut self.config + } + + /// Returns a reference to the current config + pub fn config(&self) -> &ConvictionGateConfig { + &self.config + } + + /// Evaluate all 7 gates in order. Returns on first rejection. + pub fn evaluate(&self, input: &GateInput) -> ConvictionGateOutcome { + let mut details = Vec::with_capacity(7); + + // Gate 1: Model Health + let health_ratio = if input.total_models > 0 { + input.healthy_models as f64 / input.total_models as f64 + } else { + 0.0 + }; + details.push(GateEvaluation { + gate_name: "model_health".into(), + value: health_ratio, + threshold: self.config.model_health_threshold, + passed: health_ratio >= self.config.model_health_threshold, + }); + if health_ratio < self.config.model_health_threshold { + return ConvictionGateOutcome::Rejected(GateRejection::ModelHealth { + healthy_ratio: health_ratio, + threshold: self.config.model_health_threshold, + }); + } + + // Gate 2: Time-of-Day + details.push(GateEvaluation { + gate_name: "time_of_day".into(), + value: 0.0, // N/A for session check + threshold: 0.0, + passed: self.config.allowed_sessions.contains(&input.current_session), + }); + if !self.config.allowed_sessions.contains(&input.current_session) { + return ConvictionGateOutcome::Rejected(GateRejection::TimeOfDay { + current_session: input.current_session, + }); + } + + // Gate 6 prep: Regime-adjusted thresholds + let is_high_vol = input.regime_volatility > self.config.high_vol_threshold; + let tightening = if is_high_vol { + self.config.regime_tightening_factor + } else { + 1.0 + }; + // Tightening < 1.0 means thresholds become more demanding: + // - min_confidence increases (divide by tightening) + // - max_disagreement decreases (multiply by tightening) + // - min_quorum increases (divide by tightening) + let adj_min_confidence = (self.config.min_confidence / tightening).min(0.95); + let adj_max_disagreement = (self.config.max_disagreement * tightening).max(0.05); + let adj_min_quorum = (self.config.min_quorum / tightening).min(0.95); + + // Gate 3: Confidence + details.push(GateEvaluation { + gate_name: "confidence".into(), + value: input.confidence, + threshold: adj_min_confidence, + passed: input.confidence >= adj_min_confidence, + }); + if input.confidence < adj_min_confidence { + return ConvictionGateOutcome::Rejected(GateRejection::Confidence { + confidence: input.confidence, + threshold: adj_min_confidence, + }); + } + + // Gate 4: Agreement + details.push(GateEvaluation { + gate_name: "agreement".into(), + value: input.disagreement_rate, + threshold: adj_max_disagreement, + passed: input.disagreement_rate <= adj_max_disagreement, + }); + if input.disagreement_rate > adj_max_disagreement { + return ConvictionGateOutcome::Rejected(GateRejection::Agreement { + disagreement: input.disagreement_rate, + threshold: adj_max_disagreement, + }); + } + + // Gate 5: Quorum + details.push(GateEvaluation { + gate_name: "quorum".into(), + value: input.quorum_ratio, + threshold: adj_min_quorum, + passed: input.quorum_ratio >= adj_min_quorum, + }); + if input.quorum_ratio < adj_min_quorum { + return ConvictionGateOutcome::Rejected(GateRejection::Quorum { + quorum_ratio: input.quorum_ratio, + threshold: adj_min_quorum, + }); + } + + // Gate 6: Regime (already applied via threshold adjustments above) + details.push(GateEvaluation { + gate_name: "regime".into(), + value: input.regime_volatility, + threshold: self.config.high_vol_threshold, + passed: true, // Already enforced through adjusted thresholds + }); + + // Gate 7: Conviction Sizing + let conviction_score = if self.config.conviction_scaling_enabled { + input.confidence * (1.0 - input.disagreement_rate) * input.quorum_ratio * health_ratio + } else { + 1.0 + }; + details.push(GateEvaluation { + gate_name: "conviction_sizing".into(), + value: conviction_score, + threshold: 0.0, + passed: true, // Always passes, just scales position + }); + + ConvictionGateOutcome::Passed(GatePassResult { + conviction_score, + gate_details: details, + }) + } +} +``` + +**Step 4: Run tests to verify they pass** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::conviction_gates --features cuda` +Expected: 9 tests pass (2 from Task 1 + 7 new) + +**Step 5: Commit** + +```bash +git add ml/src/ensemble/conviction_gates.rs +git commit -m "feat(ensemble): implement ConvictionGateEvaluator with 7 gates" +``` + +--- + +### Task 3: Regime filter and edge case tests + +**Files:** +- Modify: `ml/src/ensemble/conviction_gates.rs` + +**Step 1: Write regime-specific tests** + +Add to `mod tests`: + +```rust + #[test] + fn test_gate6_regime_tightens_in_high_vol() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + // This confidence passes in low vol (threshold 0.60) + // but fails in high vol (threshold 0.60/0.80 = 0.75) + let input = GateInput { + confidence: 0.70, + disagreement_rate: 0.10, + quorum_ratio: 0.80, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.05, // Above high_vol_threshold (0.03) + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::Confidence { .. }))); + } + + #[test] + fn test_gate6_regime_no_effect_in_low_vol() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.65, + disagreement_rate: 0.10, + quorum_ratio: 0.80, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, // Below high_vol_threshold (0.03) + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Passed(_))); + } + + #[test] + fn test_zero_models_rejects() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.0, + quorum_ratio: 1.0, + healthy_models: 0, + total_models: 0, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Rejected(GateRejection::ModelHealth { .. }))); + } + + #[test] + fn test_conviction_scaling_disabled() { + let mut config = ConvictionGateConfig::default(); + config.conviction_scaling_enabled = false; + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.80, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + if let ConvictionGateOutcome::Passed(result) = evaluator.evaluate(&input) { + assert!((result.conviction_score - 1.0).abs() < 1e-10); + } else { + panic!("Expected Passed outcome"); + } + } + + #[test] + fn test_gate_details_count() { + let config = ConvictionGateConfig::default(); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.80, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::Regular, + regime_volatility: 0.01, + }; + if let ConvictionGateOutcome::Passed(result) = evaluator.evaluate(&input) { + assert_eq!(result.gate_details.len(), 7); + assert!(result.gate_details.iter().all(|g| g.passed)); + } else { + panic!("Expected Passed outcome"); + } + } + + #[test] + fn test_after_hours_allowed_when_configured() { + let mut config = ConvictionGateConfig::default(); + config.allowed_sessions.push(TradingSession::AfterHours); + let evaluator = ConvictionGateEvaluator::new(config); + let input = GateInput { + confidence: 0.80, + disagreement_rate: 0.10, + quorum_ratio: 0.80, + healthy_models: 9, + total_models: 10, + current_session: TradingSession::AfterHours, + regime_volatility: 0.01, + }; + let result = evaluator.evaluate(&input); + assert!(matches!(result, ConvictionGateOutcome::Passed(_))); + } +``` + +**Step 2: Run all conviction gate tests** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::conviction_gates --features cuda` +Expected: 15 tests pass + +**Step 3: Commit** + +```bash +git add ml/src/ensemble/conviction_gates.rs +git commit -m "test(ensemble): add regime filter and edge case tests for conviction gates" +``` + +--- + +### Task 4: Wire conviction gates into EnsembleCoordinator + +**Files:** +- Modify: `ml/src/ensemble/coordinator.rs:22-46` (add gate evaluator field) +- Modify: `ml/src/ensemble/coordinator.rs:106-138` (wire into predict()) + +**Step 1: Add ConvictionGateEvaluator to EnsembleCoordinator** + +In `coordinator.rs`, add a field to `EnsembleCoordinator` struct. The struct starts around line 22. Add: + +```rust +use super::conviction_gates::{ + ConvictionGateConfig, ConvictionGateEvaluator, ConvictionGateOutcome, GateInput, TradingSession, +}; +``` + +Add field to the struct: + +```rust + /// Conviction gate evaluator (optional — None means no gating) + conviction_gates: Option, +``` + +Add constructor method: + +```rust + /// Set conviction gate configuration + pub fn with_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() + } +``` + +Initialize the field as `None` in `EnsembleCoordinator::new()`. + +**Step 2: Wire gates into predict()** + +Modify the `predict()` method (line 106-138). After getting the decision (line 130), add gate evaluation: + +```rust + // 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.count_healthy_models().await, + 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() + ); + // Return decision with conviction score in metadata + 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, // Zero signal + decision.disagreement_rate, + decision.model_votes, + ); + hold_decision.metadata.insert( + "gate_rejection".into(), + serde_json::json!(format!("{:?}", rejection)), + ); + return Ok(hold_decision); + }, + } + } +``` + +**Step 3: Add helper methods** + +Add to `EnsembleCoordinator` impl block: + +```rust + /// Calculate quorum ratio (fraction of models agreeing on direction) + fn calculate_quorum_ratio(&self, decision: &EnsembleDecision) -> f64 { + if decision.model_votes.is_empty() { + return 0.0; + } + let majority_direction = decision.action.clone(); + let agreeing = decision.model_votes.values().filter(|v| { + let vote_action = TradingAction::from_signal(v.signal, 0.3); + vote_action == majority_direction + }).count(); + agreeing as f64 / decision.model_votes.len() as f64 + } + + /// Count models that have recent, valid predictions + async fn count_healthy_models(&self) -> usize { + // All registered adapters are considered healthy if they produced predictions + // This will be refined when we add health tracking per-adapter + self.adapters.len() + } + + /// Determine current trading session (Eastern Time) + fn current_trading_session() -> TradingSession { + use chrono::{Timelike, Utc}; + // UTC-5 for ET (simplified, ignoring DST) + 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 < 570 => TradingSession::PreMarket, // Before 09:30 + t if t < 960 => TradingSession::Regular, // 09:30-16:00 + _ => TradingSession::AfterHours, // After 16:00 + } + } + + /// Extract regime volatility from feature vector (dimensions 48-50) + fn extract_regime_volatility(features: &Features) -> f64 { + // Regime features are at indices 48-50 in the 51-dim vector + features.values.get(48).copied().unwrap_or(0.01).abs() + } +``` + +**Step 4: Run workspace compile check** + +Run: `SQLX_OFFLINE=true cargo check -p ml --features cuda` +Expected: 0 errors + +**Step 5: Run existing coordinator tests** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::coordinator --features cuda` +Expected: All existing tests still pass (gates are None by default) + +**Step 6: Commit** + +```bash +git add ml/src/ensemble/coordinator.rs +git commit -m "feat(ensemble): wire conviction gates into EnsembleCoordinator::predict()" +``` + +--- + +## Phase 2: QuestDB Client (Non-Critical Path) + +### Task 5: Add questdb-rs dependency and QuestDB client skeleton + +**Files:** +- Modify: `Cargo.toml` (workspace root — add questdb-rs to workspace deps) +- Modify: `common/Cargo.toml` (add questdb-rs dependency) +- Create: `common/src/questdb.rs` +- Modify: `common/src/lib.rs` (add pub mod questdb) + +**Step 1: Add workspace dependency** + +In the workspace root `Cargo.toml`, add to `[workspace.dependencies]`: + +```toml +questdb-rs = { version = "6", default-features = false, features = ["ilp-over-tcp"] } +``` + +In `common/Cargo.toml`, add under `[dependencies]`: + +```toml +questdb-rs = { workspace = true, optional = true } +``` + +Add to `[features]`: + +```toml +questdb = ["questdb-rs"] +``` + +In `common/src/lib.rs`, add: + +```rust +#[cfg(feature = "questdb")] +pub mod questdb; +``` + +**Step 2: Create QuestDB client with ring buffer** + +Create `common/src/questdb.rs`: + +```rust +//! QuestDB client with ring buffer for graceful degradation +//! +//! Non-critical path: if QuestDB is unavailable, metrics buffer locally +//! and flush when connection is restored. Trading never depends on QuestDB. + +use std::collections::VecDeque; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; +use tokio::sync::Mutex; +use tracing::{debug, info, warn}; + +/// Maximum entries in the ring buffer before oldest are dropped +const DEFAULT_BUFFER_CAPACITY: usize = 10_000; + +/// Health check interval +const HEALTH_CHECK_INTERVAL: Duration = Duration::from_secs(30); + +/// A metric entry waiting to be flushed to QuestDB +#[derive(Debug, Clone)] +pub struct MetricEntry { + pub table: String, + pub columns: Vec<(String, MetricValue)>, + pub timestamp_ns: i64, + pub created_at: Instant, +} + +/// Supported metric value types for ILP +#[derive(Debug, Clone)] +pub enum MetricValue { + Symbol(String), + Float(f64), + Int(i64), + Bool(bool), +} + +/// QuestDB connection configuration +#[derive(Debug, Clone)] +pub struct QuestDBConfig { + /// ILP host for writes (default: localhost) + pub ilp_host: String, + /// ILP port for writes (default: 9009) + pub ilp_port: u16, + /// PostgreSQL host for queries (default: localhost) + pub pg_host: String, + /// PostgreSQL port for queries (default: 8812) + pub pg_port: u16, + /// Ring buffer capacity (default: 10,000) + pub buffer_capacity: usize, +} + +impl Default for QuestDBConfig { + fn default() -> Self { + Self { + ilp_host: "localhost".into(), + ilp_port: 9009, + pg_host: "localhost".into(), + pg_port: 8812, + buffer_capacity: DEFAULT_BUFFER_CAPACITY, + } + } +} + +/// QuestDB client with graceful degradation via ring buffer +pub struct QuestDBClient { + config: QuestDBConfig, + buffer: Arc>>, + connected: Arc, + last_health_check: Arc>, +} + +impl QuestDBClient { + /// Create a new QuestDB client (does not connect immediately) + pub fn new(config: QuestDBConfig) -> Self { + Self { + buffer: Arc::new(Mutex::new(VecDeque::with_capacity(config.buffer_capacity))), + connected: Arc::new(AtomicBool::new(false)), + last_health_check: Arc::new(Mutex::new(Instant::now())), + config, + } + } + + /// Check if QuestDB is currently connected + pub fn is_connected(&self) -> bool { + self.connected.load(Ordering::Relaxed) + } + + /// Get current buffer size (metrics waiting to be flushed) + pub async fn buffer_size(&self) -> usize { + self.buffer.lock().await.len() + } + + /// Get age of oldest buffered entry + pub async fn buffer_age(&self) -> Duration { + let buf = self.buffer.lock().await; + buf.front().map(|e| e.created_at.elapsed()).unwrap_or(Duration::ZERO) + } + + /// Write a metric entry (buffers if QuestDB is unavailable) + pub async fn write(&self, entry: MetricEntry) { + let mut buf = self.buffer.lock().await; + + // If buffer is full, drop oldest entry + if buf.len() >= self.config.buffer_capacity { + buf.pop_front(); + debug!("QuestDB buffer full, dropped oldest entry"); + } + + buf.push_back(entry); + + // Try to flush if connected + if self.connected.load(Ordering::Relaxed) { + drop(buf); // Release lock before flush + if let Err(e) = self.try_flush().await { + warn!("QuestDB flush failed: {}", e); + self.connected.store(false, Ordering::Relaxed); + } + } + } + + /// Attempt to flush buffered entries to QuestDB via ILP + async fn try_flush(&self) -> Result<(), Box> { + use questdb::ingress::{Buffer as IlpBuffer, Sender, SenderBuilder}; + + let mut buf = self.buffer.lock().await; + if buf.is_empty() { + return Ok(()); + } + + let mut sender = SenderBuilder::new( + questdb::ingress::Protocol::Tcp, + &self.config.ilp_host, + self.config.ilp_port, + ) + .build()?; + + let mut ilp_buf = IlpBuffer::new(); + let entries_to_flush: Vec = buf.drain(..).collect(); + let count = entries_to_flush.len(); + + for entry in &entries_to_flush { + ilp_buf.table(&entry.table)?; + for (name, value) in &entry.columns { + match value { + MetricValue::Symbol(s) => { ilp_buf.symbol(name, s)?; }, + MetricValue::Float(f) => { ilp_buf.column_f64(name, *f)?; }, + MetricValue::Int(i) => { ilp_buf.column_i64(name, *i)?; }, + MetricValue::Bool(b) => { ilp_buf.column_bool(name, *b)?; }, + } + } + ilp_buf.at_nanos(entry.timestamp_ns)?; + } + + sender.flush(&mut ilp_buf)?; + info!("Flushed {} entries to QuestDB", count); + + Ok(()) + } + + /// Periodic health check — call from a background task + pub async fn health_check(&self) -> bool { + let mut last = self.last_health_check.lock().await; + if last.elapsed() < HEALTH_CHECK_INTERVAL { + return self.connected.load(Ordering::Relaxed); + } + *last = Instant::now(); + drop(last); + + // Try connecting via ILP + let connected = match questdb::ingress::SenderBuilder::new( + questdb::ingress::Protocol::Tcp, + &self.config.ilp_host, + self.config.ilp_port, + ) + .build() + { + Ok(_sender) => { + if !self.connected.load(Ordering::Relaxed) { + info!("QuestDB connection restored"); + } + true + }, + Err(e) => { + if self.connected.load(Ordering::Relaxed) { + warn!("QuestDB connection lost: {}", e); + } + false + }, + }; + + self.connected.store(connected, Ordering::Relaxed); + + // If reconnected, try flushing buffer + if connected { + if let Err(e) = self.try_flush().await { + warn!("QuestDB reconnect flush failed: {}", e); + } + } + + connected + } + + /// Get ILP connection string for the configured host + pub fn ilp_address(&self) -> String { + format!("{}:{}", self.config.ilp_host, self.config.ilp_port) + } + + /// Get PostgreSQL connection string for queries + pub fn pg_connection_string(&self) -> String { + format!( + "postgresql://admin:quest@{}:{}/qdb", + self.config.pg_host, self.config.pg_port + ) + } +} + +impl std::fmt::Debug for QuestDBClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("QuestDBClient") + .field("connected", &self.connected.load(Ordering::Relaxed)) + .field("config", &self.config) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn test_client_creation() { + let client = QuestDBClient::new(QuestDBConfig::default()); + assert!(!client.is_connected()); + assert_eq!(client.buffer_size().await, 0); + } + + #[tokio::test] + async fn test_buffer_stores_entries() { + let client = QuestDBClient::new(QuestDBConfig::default()); + let entry = MetricEntry { + table: "test".into(), + columns: vec![("value".into(), MetricValue::Float(1.0))], + timestamp_ns: 1234567890, + created_at: Instant::now(), + }; + client.write(entry).await; + assert_eq!(client.buffer_size().await, 1); + } + + #[tokio::test] + async fn test_buffer_capacity_limit() { + let config = QuestDBConfig { + buffer_capacity: 3, + ..Default::default() + }; + let client = QuestDBClient::new(config); + for i in 0..5 { + let entry = MetricEntry { + table: "test".into(), + columns: vec![("i".into(), MetricValue::Int(i))], + timestamp_ns: i, + created_at: Instant::now(), + }; + client.write(entry).await; + } + // Buffer capacity is 3, so oldest 2 should be dropped + assert_eq!(client.buffer_size().await, 3); + } + + #[tokio::test] + async fn test_buffer_age_empty() { + let client = QuestDBClient::new(QuestDBConfig::default()); + assert_eq!(client.buffer_age().await, Duration::ZERO); + } + + #[tokio::test] + async fn test_default_config() { + let config = QuestDBConfig::default(); + assert_eq!(config.ilp_host, "localhost"); + assert_eq!(config.ilp_port, 9009); + assert_eq!(config.pg_host, "localhost"); + assert_eq!(config.pg_port, 8812); + assert_eq!(config.buffer_capacity, 10_000); + } + + #[test] + fn test_pg_connection_string() { + let client = QuestDBClient::new(QuestDBConfig::default()); + assert_eq!( + client.pg_connection_string(), + "postgresql://admin:quest@localhost:8812/qdb" + ); + } +} +``` + +**Step 3: Run tests** + +Run: `SQLX_OFFLINE=true cargo test -p common --lib -- questdb --features questdb` +Expected: 6 tests pass + +**Step 4: Commit** + +```bash +git add Cargo.toml common/Cargo.toml common/src/questdb.rs common/src/lib.rs +git commit -m "feat(common): add QuestDB client with ring buffer and health monitoring" +``` + +--- + +## Phase 3: Weight Optimizer (Feedback Loop Core) + +### Task 6: Weight optimizer with EMA of Sharpe ratios + +**Files:** +- Create: `ml/src/ensemble/weight_optimizer.rs` +- Modify: `ml/src/ensemble/mod.rs` (add module) + +**Step 1: Write the weight optimizer** + +Create `ml/src/ensemble/weight_optimizer.rs`: + +```rust +//! Autonomous weight optimizer for ensemble models +//! +//! Adjusts model weights based on rolling Sharpe ratios using +//! exponentially-weighted moving average. Bounded by safety rails. + +use std::collections::HashMap; +use std::time::{Duration, Instant}; +use tracing::{info, warn}; + +/// Minimum weight for any model (prevents zeroing out) +const MIN_WEIGHT: f64 = 0.05; +/// Maximum weight for any model (prevents dominance) +const MAX_WEIGHT: f64 = 0.40; +/// Maximum weight change per optimization cycle +const MAX_STEP: f64 = 0.03; +/// Minimum number of trades before first adjustment +const MIN_OBSERVATIONS: u64 = 100; +/// Cooldown between adjustments +const COOLDOWN: Duration = Duration::from_secs(24 * 3600); // 24 hours +/// Grace period for newly deployed models +const GRACE_PERIOD: Duration = Duration::from_secs(7 * 24 * 3600); // 7 days + +/// Per-model rolling performance metrics (fed from QuestDB queries) +#[derive(Debug, Clone)] +pub struct ModelRollingMetrics { + pub model_id: String, + pub sharpe_30d: f64, + pub win_rate_30d: f64, + pub prediction_accuracy: f64, + pub trade_count: u64, + pub deployed_at: Instant, +} + +/// Weight optimizer configuration +#[derive(Debug, Clone)] +pub struct WeightOptimizerConfig { + pub ema_alpha: f64, + pub min_weight: f64, + pub max_weight: f64, + pub max_step: f64, + pub min_observations: u64, + pub cooldown: Duration, + pub grace_period: Duration, +} + +impl Default for WeightOptimizerConfig { + fn default() -> Self { + Self { + ema_alpha: 0.1, + min_weight: MIN_WEIGHT, + max_weight: MAX_WEIGHT, + max_step: MAX_STEP, + min_observations: MIN_OBSERVATIONS, + cooldown: COOLDOWN, + grace_period: GRACE_PERIOD, + } + } +} + +/// Proposed weight changes from the optimizer +#[derive(Debug, Clone)] +pub struct WeightAdjustment { + pub model_id: String, + pub old_weight: f64, + pub new_weight: f64, + pub reason: String, +} + +/// Result of an optimization cycle +#[derive(Debug, Clone)] +pub enum OptimizationResult { + /// Weights adjusted successfully + Adjusted(Vec), + /// Not enough observations yet + InsufficientData { total_trades: u64, required: u64 }, + /// Still in cooldown from last adjustment + Cooldown { remaining: Duration }, + /// Kill switch is active, no adjustments allowed + KillSwitchActive, +} + +/// Autonomous weight optimizer +#[derive(Debug)] +pub struct WeightOptimizer { + config: WeightOptimizerConfig, + last_adjustment: Option, + ema_sharpe: HashMap, + frozen: bool, +} + +impl WeightOptimizer { + pub fn new(config: WeightOptimizerConfig) -> Self { + Self { + config, + last_adjustment: None, + ema_sharpe: HashMap::new(), + frozen: false, + } + } + + /// Freeze all adjustments (kill switch) + pub fn freeze(&mut self) { + self.frozen = true; + warn!("Weight optimizer frozen by kill switch"); + } + + /// Unfreeze adjustments (human acknowledgment) + pub fn unfreeze(&mut self) { + self.frozen = false; + info!("Weight optimizer unfrozen"); + } + + pub fn is_frozen(&self) -> bool { + self.frozen + } + + /// Run one optimization cycle + /// + /// Takes current weights and rolling metrics, returns proposed adjustments. + pub fn optimize( + &mut self, + current_weights: &HashMap, + metrics: &[ModelRollingMetrics], + ) -> OptimizationResult { + if self.frozen { + return OptimizationResult::KillSwitchActive; + } + + // Check cooldown + if let Some(last) = self.last_adjustment { + let elapsed = last.elapsed(); + if elapsed < self.config.cooldown { + return OptimizationResult::Cooldown { + remaining: self.config.cooldown - elapsed, + }; + } + } + + // Check minimum observations + let total_trades: u64 = metrics.iter().map(|m| m.trade_count).sum(); + if total_trades < self.config.min_observations { + return OptimizationResult::InsufficientData { + total_trades, + required: self.config.min_observations, + }; + } + + // Filter out models in grace period + let eligible: Vec<&ModelRollingMetrics> = metrics + .iter() + .filter(|m| m.deployed_at.elapsed() >= self.config.grace_period) + .collect(); + + if eligible.is_empty() { + return OptimizationResult::InsufficientData { + total_trades, + required: self.config.min_observations, + }; + } + + // Update EMA of Sharpe ratios + for m in &eligible { + let prev = self.ema_sharpe.get(&m.model_id).copied().unwrap_or(m.sharpe_30d); + let new_ema = self.config.ema_alpha * m.sharpe_30d + (1.0 - self.config.ema_alpha) * prev; + self.ema_sharpe.insert(m.model_id.clone(), new_ema); + } + + // Calculate raw weights from EMA Sharpe (shift to positive range) + let min_ema = self.ema_sharpe.values().cloned().fold(f64::INFINITY, f64::min); + let shift = if min_ema < 0.0 { min_ema.abs() + 0.1 } else { 0.0 }; + + let raw_weights: HashMap = self + .ema_sharpe + .iter() + .map(|(id, &ema)| (id.clone(), (ema + shift).max(0.01))) + .collect(); + + let total_raw: f64 = raw_weights.values().sum(); + if total_raw <= 0.0 { + return OptimizationResult::InsufficientData { + total_trades, + required: self.config.min_observations, + }; + } + + // Normalize and clamp + let mut target_weights: HashMap = raw_weights + .iter() + .map(|(id, &raw)| { + let normalized = raw / total_raw; + let clamped = normalized.clamp(self.config.min_weight, self.config.max_weight); + (id.clone(), clamped) + }) + .collect(); + + // Re-normalize after clamping + let total_clamped: f64 = target_weights.values().sum(); + if total_clamped > 0.0 { + for w in target_weights.values_mut() { + *w /= total_clamped; + } + } + + // Apply max step limit + let mut adjustments = Vec::new(); + for (model_id, &target) in &target_weights { + let current = current_weights.get(model_id).copied().unwrap_or(target); + let delta = (target - current).clamp(-self.config.max_step, self.config.max_step); + let new_weight = (current + delta).clamp(self.config.min_weight, self.config.max_weight); + + if (new_weight - current).abs() > 1e-6 { + adjustments.push(WeightAdjustment { + model_id: model_id.clone(), + old_weight: current, + new_weight, + reason: format!( + "EMA Sharpe: {:.3}, target: {:.3}", + self.ema_sharpe.get(model_id).unwrap_or(&0.0), + target + ), + }); + } + } + + if !adjustments.is_empty() { + self.last_adjustment = Some(Instant::now()); + info!("Weight optimizer adjusted {} models", adjustments.len()); + } + + OptimizationResult::Adjusted(adjustments) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn make_metrics(models: &[(&str, f64, u64)]) -> Vec { + models + .iter() + .map(|(id, sharpe, trades)| ModelRollingMetrics { + model_id: id.to_string(), + sharpe_30d: *sharpe, + win_rate_30d: 0.55, + prediction_accuracy: 0.60, + trade_count: *trades, + deployed_at: Instant::now() - Duration::from_secs(30 * 24 * 3600), // 30 days ago + }) + .collect() + } + + fn equal_weights(models: &[&str]) -> HashMap { + let w = 1.0 / models.len() as f64; + models.iter().map(|id| (id.to_string(), w)).collect() + } + + #[test] + fn test_insufficient_data() { + let mut opt = WeightOptimizer::new(WeightOptimizerConfig::default()); + let weights = equal_weights(&["dqn", "ppo"]); + let metrics = make_metrics(&[("dqn", 1.0, 10), ("ppo", 0.5, 10)]); + let result = opt.optimize(&weights, &metrics); + assert!(matches!(result, OptimizationResult::InsufficientData { .. })); + } + + #[test] + fn test_cooldown_enforced() { + let mut opt = WeightOptimizer::new(WeightOptimizerConfig::default()); + let weights = equal_weights(&["dqn", "ppo"]); + let metrics = make_metrics(&[("dqn", 1.5, 200), ("ppo", 0.5, 200)]); + + // First optimization succeeds + let result = opt.optimize(&weights, &metrics); + assert!(matches!(result, OptimizationResult::Adjusted(_))); + + // Second immediately after should be cooldown + let result2 = opt.optimize(&weights, &metrics); + assert!(matches!(result2, OptimizationResult::Cooldown { .. })); + } + + #[test] + fn test_kill_switch_freezes() { + let mut opt = WeightOptimizer::new(WeightOptimizerConfig::default()); + opt.freeze(); + let weights = equal_weights(&["dqn"]); + let metrics = make_metrics(&[("dqn", 1.0, 200)]); + let result = opt.optimize(&weights, &metrics); + assert!(matches!(result, OptimizationResult::KillSwitchActive)); + } + + #[test] + fn test_unfreeze_allows_optimization() { + let mut opt = WeightOptimizer::new(WeightOptimizerConfig::default()); + opt.freeze(); + assert!(opt.is_frozen()); + opt.unfreeze(); + assert!(!opt.is_frozen()); + } + + #[test] + fn test_weights_bounded() { + let mut config = WeightOptimizerConfig::default(); + config.cooldown = Duration::ZERO; + let mut opt = WeightOptimizer::new(config); + let weights = equal_weights(&["dqn", "ppo", "tft"]); + let metrics = make_metrics(&[("dqn", 5.0, 200), ("ppo", -2.0, 200), ("tft", 0.5, 200)]); + + let result = opt.optimize(&weights, &metrics); + if let OptimizationResult::Adjusted(adjustments) = result { + for adj in &adjustments { + assert!(adj.new_weight >= MIN_WEIGHT, "Weight below minimum: {}", adj.new_weight); + assert!(adj.new_weight <= MAX_WEIGHT, "Weight above maximum: {}", adj.new_weight); + } + } + } + + #[test] + fn test_max_step_enforced() { + let mut config = WeightOptimizerConfig::default(); + config.cooldown = Duration::ZERO; + let mut opt = WeightOptimizer::new(config); + + let mut weights = HashMap::new(); + weights.insert("dqn".into(), 0.20); + weights.insert("ppo".into(), 0.80); + + let metrics = make_metrics(&[("dqn", 5.0, 200), ("ppo", -3.0, 200)]); + let result = opt.optimize(&weights, &metrics); + + if let OptimizationResult::Adjusted(adjustments) = result { + for adj in &adjustments { + let delta = (adj.new_weight - adj.old_weight).abs(); + assert!(delta <= MAX_STEP + 1e-6, "Step too large: {} for {}", delta, adj.model_id); + } + } + } + + #[test] + fn test_grace_period_excludes_new_models() { + let mut config = WeightOptimizerConfig::default(); + config.cooldown = Duration::ZERO; + let mut opt = WeightOptimizer::new(config); + + let weights = equal_weights(&["dqn", "new_model"]); + let mut metrics = make_metrics(&[("dqn", 1.0, 200)]); + // New model deployed 1 day ago (within grace period) + metrics.push(ModelRollingMetrics { + model_id: "new_model".into(), + sharpe_30d: 2.0, + win_rate_30d: 0.70, + prediction_accuracy: 0.80, + trade_count: 50, + deployed_at: Instant::now() - Duration::from_secs(24 * 3600), + }); + + let result = opt.optimize(&weights, &metrics); + if let OptimizationResult::Adjusted(adjustments) = result { + // new_model should not be in adjustments (grace period) + assert!(adjustments.iter().all(|a| a.model_id != "new_model")); + } + } + + #[test] + fn test_weights_sum_approximately_one() { + let mut config = WeightOptimizerConfig::default(); + config.cooldown = Duration::ZERO; + let mut opt = WeightOptimizer::new(config); + let weights = equal_weights(&["a", "b", "c", "d"]); + let metrics = make_metrics(&[("a", 1.0, 200), ("b", 1.5, 200), ("c", 0.5, 200), ("d", 0.8, 200)]); + + let result = opt.optimize(&weights, &metrics); + if let OptimizationResult::Adjusted(adjustments) = result { + let mut new_weights = weights.clone(); + for adj in &adjustments { + new_weights.insert(adj.model_id.clone(), adj.new_weight); + } + let sum: f64 = new_weights.values().sum(); + // Due to step limits, sum may not be exactly 1.0 but should be close + assert!((sum - 1.0).abs() < 0.2, "Weights sum to {} (expected ~1.0)", sum); + } + } +} +``` + +**Step 2: Register module in mod.rs** + +In `ml/src/ensemble/mod.rs`, add: + +```rust +pub mod weight_optimizer; +``` + +And re-export: + +```rust +pub use weight_optimizer::{ + ModelRollingMetrics, OptimizationResult, WeightAdjustment, WeightOptimizer, + WeightOptimizerConfig, +}; +``` + +**Step 3: Run tests** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::weight_optimizer --features cuda` +Expected: 8 tests pass + +**Step 4: Commit** + +```bash +git add ml/src/ensemble/weight_optimizer.rs ml/src/ensemble/mod.rs +git commit -m "feat(ensemble): add autonomous weight optimizer with EMA Sharpe and safety rails" +``` + +--- + +### Task 7: Gate threshold optimizer + +**Files:** +- Create: `ml/src/ensemble/gate_optimizer.rs` +- Modify: `ml/src/ensemble/mod.rs` + +**Step 1: Create gate optimizer** + +Create `ml/src/ensemble/gate_optimizer.rs` following the same pattern as the weight optimizer: +- Takes `Vec` (win rate per confidence bucket) +- Adjusts `ConvictionGateConfig` thresholds by ±0.03 max per cycle +- Same cooldown, freeze, and bounds (0.30-0.90) as weight optimizer +- Tests: bucket-based adjustment, bounds enforcement, cooldown, freeze + +This file mirrors `weight_optimizer.rs` structure but operates on gate thresholds instead of model weights. + +**Step 2: Register and test** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::gate_optimizer --features cuda` + +**Step 3: Commit** + +```bash +git add ml/src/ensemble/gate_optimizer.rs ml/src/ensemble/mod.rs +git commit -m "feat(ensemble): add gate threshold optimizer with win-rate bucketing" +``` + +--- + +## Phase 4: Model Registry + +### Task 8: ModelRegistry trait and PostgreSQL types + +**Files:** +- Create: `ml/src/registry/mod.rs` +- Modify: `ml/src/lib.rs` (add `pub mod registry;`) + +**Step 1: Create the registry module** + +Create `ml/src/registry/mod.rs` with: + +- `ModelStage` enum: `Candidate`, `Staging`, `Production`, `Archived` +- `TrainingRun` struct: experiment name, model_type, hyperparameters (JSON), git_commit, data_hash, started_at, finished_at +- `ModelVersion` struct: id, run info, metrics, artifact_path, stage +- `RunComparison` struct: runs with side-by-side metrics +- `ModelRegistry` trait (async): `log_run`, `log_metrics`, `promote`, `get_production_model`, `revert`, `compare_runs` +- Unit tests for types + +The trait is defined but the PostgreSQL implementation will be added when database is available. For now, provide an `InMemoryModelRegistry` for testing. + +**Step 2: Register and test** + +Run: `SQLX_OFFLINE=true cargo test -p ml --lib -- registry --features cuda` + +**Step 3: Commit** + +```bash +git add ml/src/registry/mod.rs ml/src/lib.rs +git commit -m "feat(ml): add ModelRegistry trait with lifecycle management" +``` + +--- + +## Phase 5: Attribution and Feedback Loop + +### Task 9: P&L attribution calculator + +**Files:** +- Create: `services/trading_service/src/attribution.rs` +- Modify: `services/trading_service/src/main.rs` (add mod declaration) + +**Step 1: Create attribution module** + +The attribution calculator takes: +- `EnsembleDecision` (model votes, weights, signals) +- Realized P&L from trade close +- Returns per-model `Attribution` entries + +```rust +pub struct TradeAttribution { + pub model_id: String, + pub model_weight: f64, + pub model_signal: f64, + pub signal_alignment: f64, // 1.0 or -1.0 + pub pnl_contribution: f64, +} +``` + +Formula: `signal_alignment = if sign(model_signal) == sign(realized_direction) { 1.0 } else { -1.0 }` +`pnl_contribution = model_weight × signal_alignment × realized_pnl.abs()` + +Tests: correct attribution, zero PnL, all models agree, all disagree. + +**Step 2: Test** + +Run: `SQLX_OFFLINE=true cargo test -p trading_service --lib -- attribution` + +**Step 3: Commit** + +```bash +git add services/trading_service/src/attribution.rs services/trading_service/src/main.rs +git commit -m "feat(trading): add P&L attribution calculator for per-model decomposition" +``` + +--- + +### Task 10: Feedback loop orchestrator + +**Files:** +- Create: `services/trading_service/src/feedback_loop.rs` +- Modify: `services/trading_service/src/main.rs` + +**Step 1: Create feedback loop** + +The feedback loop orchestrator: +- Runs as a `tokio::spawn` background task +- Periodic (configurable, default 24h) +- Queries QuestDB for rolling metrics (via pg wire) +- Calls `WeightOptimizer::optimize()` and applies results +- Calls `GateOptimizer::optimize()` and applies results +- Checks retraining triggers (Sharpe < -0.5, accuracy < 45%, correlation > 0.9) +- Monitors 7-day ensemble Sharpe for kill switch +- Snapshots config to PostgreSQL before applying changes + +Key struct: +```rust +pub struct FeedbackLoop { + weight_optimizer: Arc>, + gate_optimizer: Arc>, + questdb: Arc, + coordinator: Arc, + kill_switch_threshold: f64, // -1.0 +} +``` + +Tests: mock QuestDB responses, verify optimizer integration, kill switch trigger. + +**Step 2: Test and commit** + +```bash +git add services/trading_service/src/feedback_loop.rs services/trading_service/src/main.rs +git commit -m "feat(trading): add autonomous feedback loop orchestrator with kill switch" +``` + +--- + +## Phase 6: Integration and Verification + +### Task 11: Wire everything into trading_service main.rs + +**Files:** +- Modify: `services/trading_service/src/main.rs` + +After the ensemble coordinator setup (line ~405), add: +1. Create `ConvictionGateConfig` and set on coordinator via `with_conviction_gates()` +2. Create `QuestDBClient` from env var `QUESTDB_ILP_HOST` (default localhost) +3. Create `WeightOptimizer` and `GateOptimizer` +4. Spawn `FeedbackLoop` as background task +5. Spawn QuestDB health check as background task (every 30s) + +**Step 1: Add initialization code and compile** + +Run: `SQLX_OFFLINE=true cargo check -p trading_service` + +**Step 2: Commit** + +```bash +git commit -m "feat(trading): wire conviction gates, QuestDB, and feedback loop into service init" +``` + +--- + +### Task 12: Full workspace verification + +**Step 1: Compile entire workspace** + +Run: `SQLX_OFFLINE=true cargo check --workspace` +Expected: 0 errors + +**Step 2: Run all related tests** + +```bash +SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::conviction_gates --features cuda +SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::weight_optimizer --features cuda +SQLX_OFFLINE=true cargo test -p ml --lib -- ensemble::gate_optimizer --features cuda +SQLX_OFFLINE=true cargo test -p ml --lib -- registry --features cuda +SQLX_OFFLINE=true cargo test -p common --lib -- questdb --features questdb +SQLX_OFFLINE=true cargo test -p trading_service --lib -- attribution +``` + +Expected: All pass + +**Step 3: Clippy check** + +Run: `cargo clippy -p ml -p common -p trading_service -- -D warnings` +Expected: 0 warnings + +**Step 4: Final commit** + +```bash +git commit -m "chore: verify full workspace compilation and tests for operational maturity system" +``` + +--- + +## Summary + +| Phase | Tasks | New Files | Est. Tests | +|-------|-------|-----------|-----------| +| 1: Conviction Gates | 1-4 | `conviction_gates.rs` | ~15 | +| 2: QuestDB Client | 5 | `questdb.rs` | ~6 | +| 3: Weight Optimizer | 6-7 | `weight_optimizer.rs`, `gate_optimizer.rs` | ~16 | +| 4: Model Registry | 8 | `registry/mod.rs` | ~8 | +| 5: Attribution & Loop | 9-10 | `attribution.rs`, `feedback_loop.rs` | ~10 | +| 6: Integration | 11-12 | — | — | +| **Total** | **12 tasks** | **7 new files** | **~55 tests** | + +Dependencies: Phase 1 is independent. Phase 2 is independent. Phase 3 depends on Phase 1 (gate optimizer needs ConvictionGateConfig). Phase 4 is independent. Phase 5 depends on Phases 2 and 3. Phase 6 depends on all.