feat(trading_agent): wire ML confidence to ensemble GetEnsembleVote with fallback

Replace the liquidity heuristic in score_symbols with a real
GetEnsembleVote gRPC call to trading-service. Each symbol gets a
concurrent RPC; on failure the original liquidity*0.9+0.1 heuristic
is used as fallback. The ML client is lazily initialized via OnceCell
and the service URL is configurable via ML_SERVICE_URL env var.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

services/trading_agent_service/build.rs

@@ -1,5 +1,18 @@
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Compile proto files for Trading Agent Service
     tonic_prost_build::compile_protos("proto/trading_agent.proto")?;
+
+    // ML service client for GetEnsembleVote (ensemble confidence)
+    tonic_prost_build::configure()
+        .build_server(false)
+        .build_client(true)
+        .client_mod_attribute(".", "#[allow(unused_qualifications)]")
+        .compile_protos(
+            &["../trading_service/proto/ml.proto"],
+            &["../trading_service/proto"],
+        )?;
+
+    println!("cargo:rerun-if-changed=../trading_service/proto/ml.proto");
+
     Ok(())
 }

services/trading_agent_service/src/autonomous_scaling.rs

@@ -15,7 +15,11 @@ use bigdecimal::BigDecimal;
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use sqlx::PgPool;
+use std::collections::HashMap;
 use std::str::FromStr;
+use std::sync::Arc;
+use tokio::sync::OnceCell;
+use tonic::transport::Channel;
 use uuid::Uuid;
 
 use crate::universe::{Instrument, UniverseError};
@@ -355,6 +359,7 @@ impl SymbolScore {
 pub struct AutonomousUniverseManager {
     constraints: SystemConstraints,
     pool: PgPool,
+    ml_client: Arc<OnceCell<crate::proto::ml::ml_service_client::MlServiceClient<Channel>>>,
 }
 
 impl AutonomousUniverseManager {
@@ -363,6 +368,7 @@ impl AutonomousUniverseManager {
         Self {
             constraints: SystemConstraints::default(),
             pool,
+            ml_client: Arc::new(OnceCell::new()),
         }
     }
 
@@ -371,6 +377,7 @@ impl AutonomousUniverseManager {
         Self {
             constraints,
             pool,
+            ml_client: Arc::new(OnceCell::new()),
         }
     }
 
@@ -541,8 +548,8 @@ impl AutonomousUniverseManager {
         // Get candidate instruments
         let candidates = self.get_candidate_instruments().await?;
 
-        // Score symbols (simplified - in production would use ML ensemble)
-        let scored = self.score_symbols_mock(&candidates, &tier);
+        // Score symbols using ML ensemble confidence (falls back to liquidity heuristic)
+        let scored = self.score_symbols(&candidates, &tier).await;
 
         // Select top N symbols
         let mut selected: Vec<_> = scored
@@ -647,12 +654,90 @@ impl AutonomousUniverseManager {
         ])
     }
 
-    /// Score symbols (mock implementation - production would use ML ensemble)
-    fn score_symbols_mock(
+    /// Fetch ML confidence from ensemble service for each symbol.
+    /// Falls back to empty map on failure (callers use liquidity heuristic).
+    async fn get_ensemble_confidences(&self, symbols: &[String]) -> HashMap<String, f64> {
+        let mut result = HashMap::new();
+
+        let client_result: Result<&crate::proto::ml::ml_service_client::MlServiceClient<Channel>, String> = self
+            .ml_client
+            .get_or_try_init(|| async {
+                let url = std::env::var("ML_SERVICE_URL")
+                    .unwrap_or_else(|_| "http://trading-service:50051".to_string());
+                let channel = Channel::from_shared(url)
+                    .map_err(|e| format!("invalid ML service URL: {e}"))?
+                    .connect()
+                    .await
+                    .map_err(|e| format!("ML service unreachable: {e}"))?;
+                Ok(crate::proto::ml::ml_service_client::MlServiceClient::new(
+                    channel,
+                ))
+            })
+            .await;
+
+        let client = match client_result {
+            Ok(c) => c,
+            Err(e) => {
+                tracing::warn!(error = %e, "ML client init failed, using liquidity fallback");
+                return result;
+            }
+        };
+
+        // Fan out one RPC per symbol concurrently (service API is single-symbol)
+        let futs: Vec<_> = symbols
+            .iter()
+            .map(|symbol| {
+                let mut client = client.clone();
+                let sym = symbol.clone();
+                async move {
+                    let request = crate::proto::ml::GetEnsembleVoteRequest {
+                        symbol: sym.clone(),
+                        horizon_minutes: None,
+                        model_names: vec![],
+                    };
+                    match client
+                        .get_ensemble_vote(tonic::Request::new(request))
+                        .await
+                    {
+                        Ok(resp) => {
+                            let inner = resp.into_inner();
+                            Some((sym, inner.overall_confidence))
+                        }
+                        Err(e) => {
+                            tracing::warn!(symbol = %sym, error = %e, "ensemble RPC failed for symbol");
+                            None
+                        }
+                    }
+                }
+            })
+            .collect();
+
+        let results = futures::future::join_all(futs).await;
+        for pair in results.into_iter().flatten() {
+            result.insert(pair.0, pair.1);
+        }
+
+        if !result.is_empty() {
+            tracing::info!(
+                count = result.len(),
+                "fetched ensemble confidence for {} symbols",
+                result.len()
+            );
+        }
+
+        result
+    }
+
+    /// Score symbols using ML ensemble confidence with liquidity-heuristic fallback.
+    async fn score_symbols(
         &self,
         instruments: &[Instrument],
         tier: &CapitalScalingTier,
     ) -> Vec<SymbolScore> {
+        // Batch-fetch ensemble confidences for all candidate symbols
+        let symbols: Vec<String> = instruments.iter().map(|i| i.symbol.to_string()).collect();
+        let ensemble_confidences = self.get_ensemble_confidences(&symbols).await;
+
         instruments
             .iter()
             .filter(|inst| {
@@ -661,9 +746,11 @@ impl AutonomousUniverseManager {
                     && inst.avg_daily_volume >= tier.min_liquidity
             })
             .map(|inst| {
-                // ml_confidence approximated from liquidity when ML ensemble service unavailable.
-                // TODO: Wire to real ensemble via GetEnsembleVote gRPC when service integration is done.
-                let ml_confidence = inst.liquidity_score * 0.9 + 0.1;
+                // Use ensemble confidence when available, fall back to liquidity heuristic
+                let ml_confidence = ensemble_confidences
+                    .get(&inst.symbol.to_string())
+                    .copied()
+                    .unwrap_or(inst.liquidity_score * 0.9 + 0.1);
 
                 // Normalize scores
                 let liquidity_score = inst.liquidity_score;

services/trading_agent_service/src/lib.rs

@@ -9,6 +9,12 @@ pub mod proto {
     pub mod trading_agent {
         tonic::include_proto!("trading_agent");
     }
+
+    /// ML service client definitions (for ensemble confidence)
+    #[allow(clippy::mixed_attributes_style)]
+    pub mod ml {
+        tonic::include_proto!("ml");
+    }
 }
 
 pub mod allocation;