foxhunt/ML_DATABASE_CONNECTION.md

ML Database Connection Implementation Report

Date: 2025-10-16 Wave: 14.2 Agent 6 Status: ✅ IMPLEMENTATION READY - Schema Analysis Complete

---

1. Database Schema Analysis

1.1 Ensemble Predictions Table Schema

The ensemble_predictions table (Migration 022) is production-ready for storing ensemble predictions:

CREATE TABLE ensemble_predictions (
    -- Primary identifiers
    id UUID DEFAULT gen_random_uuid(),
    prediction_timestamp TIMESTAMPTZ NOT NULL DEFAULT NOW(),

    -- Trading context
    symbol VARCHAR(20) NOT NULL,
    account_id VARCHAR(64),
    strategy_id VARCHAR(100),

    -- Ensemble decision
    ensemble_action VARCHAR(10) NOT NULL, -- BUY, SELL, HOLD
    ensemble_signal DOUBLE PRECISION NOT NULL CHECK (ensemble_signal >= -1.0 AND ensemble_signal <= 1.0),
    ensemble_confidence DOUBLE PRECISION NOT NULL CHECK (ensemble_confidence >= 0.0 AND ensemble_confidence <= 1.0),
    disagreement_rate DOUBLE PRECISION NOT NULL CHECK (disagreement_rate >= 0.0 AND disagreement_rate <= 1.0),

    -- Per-model votes (DQN, PPO, MAMBA-2, TFT)
    dqn_signal DOUBLE PRECISION,
    dqn_confidence DOUBLE PRECISION,
    dqn_weight DOUBLE PRECISION,
    dqn_vote VARCHAR(10), -- BUY, SELL, HOLD

    ppo_signal DOUBLE PRECISION,
    ppo_confidence DOUBLE PRECISION,
    ppo_weight DOUBLE PRECISION,
    ppo_vote VARCHAR(10),

    mamba2_signal DOUBLE PRECISION,
    mamba2_confidence DOUBLE PRECISION,
    mamba2_weight DOUBLE PRECISION,
    mamba2_vote VARCHAR(10),

    tft_signal DOUBLE PRECISION,
    tft_confidence DOUBLE PRECISION,
    tft_weight DOUBLE PRECISION,
    tft_vote VARCHAR(10),

    -- Execution tracking
    order_id UUID REFERENCES orders(id) ON DELETE SET NULL,
    executed_price BIGINT,
    position_size BIGINT,
    pnl BIGINT,
    commission BIGINT DEFAULT 0,
    slippage_bps INTEGER,

    -- Feature snapshot
    feature_snapshot JSONB,

    -- System context
    node_id VARCHAR(50),
    inference_latency_us INTEGER,
    aggregation_latency_us INTEGER,

    -- Metadata
    metadata JSONB,

    PRIMARY KEY (id, prediction_timestamp)
);

Key Features:

• ✅ TimescaleDB hypertable (1-day chunks)
• ✅ Per-model attribution (DQN, PPO, MAMBA-2, TFT)
• ✅ Execution tracking (order_id link)
• ✅ Feature snapshot (JSONB for reproducibility)
• ✅ Performance indexes (symbol, timestamp, disagreement)

---

2. Rust Type Mapping

2.1 EnsemblePrediction Struct (NEW)

use sqlx::FromRow;
use serde::{Serialize, Deserialize};
use uuid::Uuid;
use chrono::{DateTime, Utc};

/// Ensemble prediction record for database persistence
#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
pub struct EnsemblePrediction {
    // Primary identifiers
    pub id: Uuid,
    pub prediction_timestamp: DateTime<Utc>,

    // Trading context
    pub symbol: String,
    pub account_id: Option<String>,
    pub strategy_id: Option<String>,

    // Ensemble decision
    pub ensemble_action: String, // "BUY", "SELL", "HOLD"
    pub ensemble_signal: f64,
    pub ensemble_confidence: f64,
    pub disagreement_rate: f64,

    // Per-model votes (DQN)
    pub dqn_signal: Option<f64>,
    pub dqn_confidence: Option<f64>,
    pub dqn_weight: Option<f64>,
    pub dqn_vote: Option<String>,

    // Per-model votes (PPO)
    pub ppo_signal: Option<f64>,
    pub ppo_confidence: Option<f64>,
    pub ppo_weight: Option<f64>,
    pub ppo_vote: Option<String>,

    // Per-model votes (MAMBA-2)
    pub mamba2_signal: Option<f64>,
    pub mamba2_confidence: Option<f64>,
    pub mamba2_weight: Option<f64>,
    pub mamba2_vote: Option<String>,

    // Per-model votes (TFT)
    pub tft_signal: Option<f64>,
    pub tft_confidence: Option<f64>,
    pub tft_weight: Option<f64>,
    pub tft_vote: Option<String>,

    // Execution tracking
    pub order_id: Option<Uuid>,
    pub executed_price: Option<i64>,
    pub position_size: Option<i64>,
    pub pnl: Option<i64>,
    pub commission: Option<i64>,
    pub slippage_bps: Option<i32>,

    // Feature snapshot
    pub feature_snapshot: Option<serde_json::Value>,

    // System context
    pub node_id: Option<String>,
    pub inference_latency_us: Option<i32>,
    pub aggregation_latency_us: Option<i32>,

    // Metadata
    pub metadata: Option<serde_json::Value>,
}

impl EnsemblePrediction {
    /// Create from EnsembleDecision
    pub fn from_decision(
        decision: &EnsembleDecision,
        symbol: String,
        account_id: Option<String>,
    ) -> Self {
        let ensemble_action = format!("{:?}", decision.action).to_uppercase();

        // Extract per-model votes
        let (dqn_signal, dqn_confidence, dqn_weight, dqn_vote) =
            extract_model_vote(&decision.model_votes, "DQN");
        let (ppo_signal, ppo_confidence, ppo_weight, ppo_vote) =
            extract_model_vote(&decision.model_votes, "PPO");
        let (mamba2_signal, mamba2_confidence, mamba2_weight, mamba2_vote) =
            extract_model_vote(&decision.model_votes, "MAMBA2");
        let (tft_signal, tft_confidence, tft_weight, tft_vote) =
            extract_model_vote(&decision.model_votes, "TFT");

        Self {
            id: Uuid::new_v4(),
            prediction_timestamp: Utc::now(),
            symbol,
            account_id,
            strategy_id: None,
            ensemble_action,
            ensemble_signal: decision.signal,
            ensemble_confidence: decision.confidence,
            disagreement_rate: decision.disagreement_rate,
            dqn_signal,
            dqn_confidence,
            dqn_weight,
            dqn_vote,
            ppo_signal,
            ppo_confidence,
            ppo_weight,
            ppo_vote,
            mamba2_signal,
            mamba2_confidence,
            mamba2_weight,
            mamba2_vote,
            tft_signal,
            tft_confidence,
            tft_weight,
            tft_vote,
            order_id: None,
            executed_price: None,
            position_size: None,
            pnl: None,
            commission: None,
            slippage_bps: None,
            feature_snapshot: None,
            node_id: Some(get_node_id()),
            inference_latency_us: None,
            aggregation_latency_us: None,
            metadata: None,
        }
    }
}

/// Helper: Extract model vote from HashMap
fn extract_model_vote(
    votes: &HashMap<String, ModelVote>,
    model_id: &str,
) -> (Option<f64>, Option<f64>, Option<f64>, Option<String>) {
    votes.get(model_id).map(|vote| {
        let action = format!("{:?}", vote.action(0.3)).to_uppercase();
        (
            Some(vote.signal),
            Some(vote.confidence),
            Some(vote.weight),
            Some(action),
        )
    }).unwrap_or((None, None, None, None))
}

/// Get current node ID (hostname or environment variable)
fn get_node_id() -> String {
    std::env::var("NODE_ID")
        .unwrap_or_else(|_| hostname::get()
            .unwrap_or_else(|_| "unknown".into())
            .to_string_lossy()
            .to_string())
}

---

3. Database Integration Methods

3.1 EnsembleCoordinator Extension

File: services/trading_service/src/ensemble_coordinator.rs

use sqlx::PgPool;
use anyhow::{Context, Result};

impl EnsembleCoordinator {
    /// Add database pool to coordinator
    pub fn with_db_pool(mut self, db_pool: PgPool) -> Self {
        self.db_pool = Some(db_pool);
        self
    }

    /// Save prediction to database
    pub async fn save_prediction_to_db(
        &self,
        prediction: &EnsemblePrediction,
    ) -> Result<Uuid> {
        let db_pool = self.db_pool
            .as_ref()
            .context("Database pool not configured")?;

        let start = std::time::Instant::now();

        // Insert prediction
        let row = sqlx::query!(
            r#"
            INSERT INTO ensemble_predictions (
                id, prediction_timestamp, symbol, account_id, strategy_id,
                ensemble_action, ensemble_signal, ensemble_confidence, disagreement_rate,
                dqn_signal, dqn_confidence, dqn_weight, dqn_vote,
                ppo_signal, ppo_confidence, ppo_weight, ppo_vote,
                mamba2_signal, mamba2_confidence, mamba2_weight, mamba2_vote,
                tft_signal, tft_confidence, tft_weight, tft_vote,
                feature_snapshot, node_id, inference_latency_us, aggregation_latency_us, metadata
            ) VALUES (
                $1, $2, $3, $4, $5,
                $6, $7, $8, $9,
                $10, $11, $12, $13,
                $14, $15, $16, $17,
                $18, $19, $20, $21,
                $22, $23, $24, $25,
                $26, $27, $28, $29, $30
            )
            RETURNING id
            "#,
            prediction.id,
            prediction.prediction_timestamp,
            prediction.symbol,
            prediction.account_id,
            prediction.strategy_id,
            prediction.ensemble_action,
            prediction.ensemble_signal,
            prediction.ensemble_confidence,
            prediction.disagreement_rate,
            prediction.dqn_signal,
            prediction.dqn_confidence,
            prediction.dqn_weight,
            prediction.dqn_vote,
            prediction.ppo_signal,
            prediction.ppo_confidence,
            prediction.ppo_weight,
            prediction.ppo_vote,
            prediction.mamba2_signal,
            prediction.mamba2_confidence,
            prediction.mamba2_weight,
            prediction.mamba2_vote,
            prediction.tft_signal,
            prediction.tft_confidence,
            prediction.tft_weight,
            prediction.tft_vote,
            prediction.feature_snapshot,
            prediction.node_id,
            prediction.inference_latency_us,
            prediction.aggregation_latency_us,
            prediction.metadata,
        )
        .fetch_one(db_pool)
        .await
        .context("Failed to insert ensemble prediction")?;

        let latency_us = start.elapsed().as_micros() as i64;

        tracing::info!(
            "Saved prediction {} to database: {} {} (latency: {}μs)",
            row.id,
            prediction.ensemble_action,
            prediction.symbol,
            latency_us
        );

        Ok(row.id)
    }

    /// Start background prediction loop
    pub async fn populate_predictions_continuously(
        self: Arc<Self>,
        interval_secs: u64,
    ) -> Result<()> {
        let mut interval = tokio::time::interval(
            std::time::Duration::from_secs(interval_secs)
        );

        tracing::info!(
            "Starting background prediction loop (interval: {}s)",
            interval_secs
        );

        loop {
            interval.tick().await;

            // Generate prediction for configured symbols
            for symbol in &self.config.symbols {
                match self.generate_and_save_prediction(symbol).await {
                    Ok(prediction_id) => {
                        tracing::debug!(
                            "Generated prediction {} for {}",
                            prediction_id,
                            symbol
                        );
                    }
                    Err(e) => {
                        tracing::error!(
                            "Failed to generate prediction for {}: {}",
                            symbol,
                            e
                        );
                    }
                }
            }
        }
    }

    /// Generate prediction and save to database
    async fn generate_and_save_prediction(&self, symbol: &str) -> Result<Uuid> {
        // 1. Fetch latest market features (stub - will use real feature cache)
        let features = self.fetch_features_for_symbol(symbol).await?;

        // 2. Make ensemble prediction
        let decision = self.predict(&features).await?;

        // 3. Convert to database record
        let prediction = EnsemblePrediction::from_decision(
            &decision,
            symbol.to_string(),
            Some("paper_trading_001".to_string()),
        );

        // 4. Save to database
        let prediction_id = self.save_prediction_to_db(&prediction).await?;

        Ok(prediction_id)
    }

    /// Fetch features for symbol (stub for now)
    async fn fetch_features_for_symbol(&self, symbol: &str) -> Result<Features> {
        // TODO: Replace with real feature cache query
        Ok(Features::new(
            vec![0.5; 16], // 16 features (5 OHLCV + 10 technical indicators + timestamp)
            (0..16).map(|i| format!("feature_{}", i)).collect(),
        ).with_symbol(symbol.to_string()))
    }
}

3.2 Configuration Extension

#[derive(Debug, Clone)]
pub struct EnsembleConfig {
    /// Symbols to generate predictions for
    pub symbols: Vec<String>,

    /// Prediction generation interval (seconds)
    pub prediction_interval_secs: u64,

    /// Database connection pool
    pub db_pool: Option<PgPool>,
}

impl Default for EnsembleConfig {
    fn default() -> Self {
        Self {
            symbols: vec![
                "ES.FUT".to_string(),
                "NQ.FUT".to_string(),
                "ZN.FUT".to_string(),
                "6E.FUT".to_string(),
            ],
            prediction_interval_secs: 60, // Every 60 seconds
            db_pool: None,
        }
    }
}

---

4. TDD Test Implementation Plan

4.1 Test 1: save_prediction_to_db inserts into ensemble_predictions table

#[cfg(test)]
mod tests {
    use super::*;

    #[sqlx::test]
    async fn test_save_prediction_to_db(pool: PgPool) -> Result<()> {
        // Setup: Create coordinator with database pool
        let coordinator = EnsembleCoordinator::new()
            .with_db_pool(pool.clone());

        // Setup: Create ensemble decision
        let mut votes = HashMap::new();
        votes.insert(
            "DQN".to_string(),
            ModelVote::new("DQN".to_string(), 0.75, 0.85, 0.33)
                .with_model_type("DQN".to_string()),
        );
        votes.insert(
            "PPO".to_string(),
            ModelVote::new("PPO".to_string(), 0.65, 0.80, 0.33)
                .with_model_type("PPO".to_string()),
        );
        votes.insert(
            "TFT".to_string(),
            ModelVote::new("TFT".to_string(), 0.70, 0.82, 0.34)
                .with_model_type("TFT".to_string()),
        );

        let decision = EnsembleDecision::new(
            TradingAction::Buy,
            0.82,
            0.70,
            0.15,
            votes,
        ).with_symbol("ES.FUT".to_string());

        // Create prediction record
        let prediction = EnsemblePrediction::from_decision(
            &decision,
            "ES.FUT".to_string(),
            Some("paper_trading_001".to_string()),
        );

        // Execute: Save to database
        let prediction_id = coordinator.save_prediction_to_db(&prediction).await?;

        // Verify: Prediction was inserted
        let row = sqlx::query!(
            r#"
            SELECT id, symbol, ensemble_action, ensemble_signal, ensemble_confidence,
                   disagreement_rate, dqn_signal, ppo_signal, tft_signal
            FROM ensemble_predictions
            WHERE id = $1
            "#,
            prediction_id
        )
        .fetch_one(&pool)
        .await?;

        assert_eq!(row.id, prediction_id);
        assert_eq!(row.symbol, "ES.FUT");
        assert_eq!(row.ensemble_action, "BUY");
        assert!((row.ensemble_signal - 0.70).abs() < 0.01);
        assert!((row.ensemble_confidence - 0.82).abs() < 0.01);
        assert!((row.disagreement_rate - 0.15).abs() < 0.01);

        // Verify per-model votes
        assert!(row.dqn_signal.is_some());
        assert!((row.dqn_signal.unwrap() - 0.75).abs() < 0.01);
        assert!(row.ppo_signal.is_some());
        assert!(row.tft_signal.is_some());

        Ok(())
    }
}

4.2 Test 2: Background loop runs and saves predictions periodically

#[sqlx::test(migrations = "migrations")]
async fn test_background_prediction_loop(pool: PgPool) -> Result<()> {
    // Setup: Create coordinator with loaded models
    let coordinator = Arc::new(
        create_test_coordinator_with_models()
            .await
            .with_db_pool(pool.clone())
    );

    // Start background loop in separate task
    let coordinator_clone = coordinator.clone();
    let loop_handle = tokio::spawn(async move {
        coordinator_clone
            .populate_predictions_continuously(1) // 1 second interval
            .await
    });

    // Wait for 3 seconds (should generate 3 predictions)
    tokio::time::sleep(Duration::from_secs(3)).await;

    // Verify: Check predictions were inserted
    let count = sqlx::query_scalar!(
        r#"
        SELECT COUNT(*)
        FROM ensemble_predictions
        WHERE prediction_timestamp >= NOW() - INTERVAL '5 seconds'
        "#
    )
    .fetch_one(&pool)
    .await?;

    assert!(count.unwrap() >= 3, "Expected at least 3 predictions, got {}", count.unwrap());

    // Cleanup: Stop background loop
    loop_handle.abort();

    Ok(())
}

4.3 Test 3: Paper trading executor reads predictions from table

#[sqlx::test]
async fn test_paper_trading_reads_predictions(pool: PgPool) -> Result<()> {
    // Setup: Insert test prediction
    let prediction_id = Uuid::new_v4();
    sqlx::query!(
        r#"
        INSERT INTO ensemble_predictions (
            id, symbol, ensemble_action, ensemble_signal, ensemble_confidence, disagreement_rate
        ) VALUES (
            $1, 'ES.FUT', 'BUY', 0.75, 0.85, 0.10
        )
        "#,
        prediction_id
    )
    .execute(&pool)
    .await?;

    // Setup: Create paper trading executor
    let config = PaperTradingConfig::default();
    let executor = PaperTradingExecutor::new(pool.clone(), config);

    // Execute: Fetch pending predictions
    let predictions = executor.fetch_pending_predictions().await?;

    // Verify: Prediction was fetched
    assert_eq!(predictions.len(), 1);
    assert_eq!(predictions[0].id, prediction_id);
    assert_eq!(predictions[0].symbol, "ES.FUT");
    assert_eq!(predictions[0].ensemble_action, "BUY");

    Ok(())
}

4.4 Test 4: End-to-end: ML generates prediction → saved to DB → paper trade executes

#[sqlx::test]
async fn test_e2e_ml_to_paper_trade(pool: PgPool) -> Result<()> {
    // Setup: Create coordinator with loaded models
    let coordinator = Arc::new(
        create_test_coordinator_with_models()
            .await
            .with_db_pool(pool.clone())
    );

    // Setup: Create paper trading executor
    let config = PaperTradingConfig::default();
    let executor = Arc::new(PaperTradingExecutor::new(pool.clone(), config));

    // Step 1: Generate and save prediction
    let prediction_id = coordinator
        .generate_and_save_prediction("ES.FUT")
        .await?;

    // Step 2: Paper trading executor processes prediction
    let processed_count = executor.execute_cycle().await?;

    assert_eq!(processed_count, 1, "Should have processed 1 prediction");

    // Step 3: Verify prediction was linked to order
    let row = sqlx::query!(
        r#"
        SELECT order_id
        FROM ensemble_predictions
        WHERE id = $1
        "#,
        prediction_id
    )
    .fetch_one(&pool)
    .await?;

    assert!(row.order_id.is_some(), "Prediction should be linked to order");

    // Step 4: Verify order was created
    let order_count = sqlx::query_scalar!(
        r#"
        SELECT COUNT(*)
        FROM orders
        WHERE id = $1
        "#,
        row.order_id.unwrap()
    )
    .fetch_one(&pool)
    .await?;

    assert_eq!(order_count.unwrap(), 1, "Order should exist");

    Ok(())
}

4.5 Test 5: Performance: <100ms to save prediction to database

#[sqlx::test]
async fn test_save_prediction_performance(pool: PgPool) -> Result<()> {
    // Setup: Create coordinator
    let coordinator = EnsembleCoordinator::new()
        .with_db_pool(pool.clone());

    // Create test prediction
    let decision = create_test_decision();
    let prediction = EnsemblePrediction::from_decision(
        &decision,
        "ES.FUT".to_string(),
        Some("paper_trading_001".to_string()),
    );

    // Benchmark: Save 100 predictions
    let mut latencies = Vec::new();
    for _ in 0..100 {
        let start = Instant::now();
        coordinator.save_prediction_to_db(&prediction).await?;
        let latency = start.elapsed();
        latencies.push(latency.as_micros());
    }

    // Verify: P99 latency < 100ms
    latencies.sort();
    let p99 = latencies[99];
    assert!(
        p99 < 100_000, // 100ms in microseconds
        "P99 latency {}μs exceeds 100ms threshold",
        p99
    );

    // Report performance
    let median = latencies[50];
    let p95 = latencies[95];
    println!("Save prediction latency: median={}μs, p95={}μs, p99={}μs",
        median, p95, p99);

    Ok(())
}

---

5. Integration Architecture

┌─────────────────────────────────────────────────────────────────┐
│                   Trading Service                                │
│                                                                   │
│  ┌──────────────────────────────────────────────────────────┐  │
│  │         EnsembleCoordinator                               │  │
│  │                                                            │  │
│  │  ┌────────────┐    ┌────────────┐    ┌────────────┐     │  │
│  │  │    DQN     │    │    PPO     │    │    TFT     │     │  │
│  │  └────────────┘    └────────────┘    └────────────┘     │  │
│  │         │                  │                  │           │  │
│  │         └──────────────────┴──────────────────┘           │  │
│  │                      │                                     │  │
│  │              EnsembleDecision                              │  │
│  │                      │                                     │  │
│  │                      ▼                                     │  │
│  │         save_prediction_to_db()                           │  │
│  │                      │                                     │  │
│  └──────────────────────┼─────────────────────────────────────┘
│                         │                                       │
│                         ▼                                       │
│  ┌─────────────────────────────────────────────────────────┐  │
│  │        PostgreSQL: ensemble_predictions table            │  │
│  │  - ensemble_action, signal, confidence                   │  │
│  │  - Per-model votes (DQN, PPO, MAMBA-2, TFT)            │  │
│  │  - Feature snapshot, metadata                            │  │
│  └─────────────────────────────────────────────────────────┘  │
│                         │                                       │
│                         │ (every 100ms)                         │
│                         ▼                                       │
│  ┌─────────────────────────────────────────────────────────┐  │
│  │         PaperTradingExecutor                             │  │
│  │  - Fetch unexecuted predictions (order_id IS NULL)      │  │
│  │  - Filter: confidence ≥60%, action=BUY/SELL             │  │
│  │  - Create orders in `orders` table                       │  │
│  │  - Link predictions via order_id                         │  │
│  └─────────────────────────────────────────────────────────┘  │
│                                                                   │
└───────────────────────────────────────────────────────────────────┘

---

6. Implementation Checklist

Phase 1: Core Types (1 hour)

• Create EnsemblePrediction struct with sqlx::FromRow
• Implement from_decision() converter
• Add helper functions (extract_model_vote, get_node_id)

Phase 2: Database Methods (2 hours)

• Add db_pool: Option<PgPool> to EnsembleCoordinator
• Implement with_db_pool() builder method
• Implement save_prediction_to_db() with INSERT query
• Implement populate_predictions_continuously() background loop
• Implement generate_and_save_prediction() helper
• Add fetch_features_for_symbol() stub

Phase 3: Configuration (30 minutes)

• Create EnsembleConfig struct
• Add symbols and prediction_interval_secs fields
• Implement Default trait

Phase 4: TDD Tests (3 hours)

• Test 1: test_save_prediction_to_db (basic INSERT)
• Test 2: test_background_prediction_loop (periodic generation)
• Test 3: test_paper_trading_reads_predictions (consumer integration)
• Test 4: test_e2e_ml_to_paper_trade (full pipeline)
• Test 5: test_save_prediction_performance (latency benchmark)

Phase 5: Integration (2 hours)

• Wire coordinator to trading service startup
• Configure symbols and interval via config
• Add Prometheus metrics for predictions saved
• Test with real ML models (DQN, PPO, TFT)

---

7. Performance Targets

Metric	Target	Validation Method
Save latency (P99)	<100ms	test_save_prediction_performance
Prediction interval	60s	Configuration
Background loop overhead	<5ms	Tokio runtime metrics
Database writes/min	~4-16	4 symbols × 60s interval
Model inference latency	<50ms	Existing ML benchmarks

---

8. Production Deployment Checklist

Database

• Migration 022 applied (ensemble_predictions table exists)
• TimescaleDB hypertable enabled
• Indexes created (symbol, timestamp, disagreement)
• Compression policy configured (7-day retention)

Application

• EnsembleCoordinator wired to database pool
• Background prediction loop running
• Paper trading executor consuming predictions
• Prometheus metrics exported
• Error handling and retry logic

Monitoring

• Grafana dashboard for prediction volume
• Alert: prediction latency >100ms
• Alert: background loop failure
• Alert: high disagreement rate (>50%)

---

9. Next Steps

1. Implement Core Types (Phase 1): Create EnsemblePrediction struct in services/trading_service/src/ensemble_coordinator.rs
2. Write TDD Tests (Phase 4): Implement all 5 tests in services/trading_service/tests/ensemble_coordinator_db_tests.rs
3. Implement Database Methods (Phase 2): Add save_prediction_to_db() and background loop
4. End-to-End Validation (Phase 5): Run full pipeline test with real ML models

---

10. Success Criteria

✅ Implementation Complete when:

1. All 5 TDD tests passing (100%)
2. Predictions saved to database every 60 seconds
3. Paper trading executor processes predictions (order_id linked)
4. P99 save latency <100ms
5. E2E test passes (ML → DB → Order)

---

Status: ✅ READY FOR IMPLEMENTATION Estimated Time: 8.5 hours (1 day sprint) Risk Level: LOW (schema exists, types defined, clear integration path)