diff --git a/services/trading_agent_service/build.rs b/services/trading_agent_service/build.rs index 4db2aa7ae..757a1191b 100644 --- a/services/trading_agent_service/build.rs +++ b/services/trading_agent_service/build.rs @@ -1,5 +1,18 @@ fn main() -> Result<(), Box> { // 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(()) } diff --git a/services/trading_agent_service/src/autonomous_scaling.rs b/services/trading_agent_service/src/autonomous_scaling.rs index 8c6090f33..5ad408d18 100644 --- a/services/trading_agent_service/src/autonomous_scaling.rs +++ b/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>>, } 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 { + let mut result = HashMap::new(); + + let client_result: Result<&crate::proto::ml::ml_service_client::MlServiceClient, 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 { + // Batch-fetch ensemble confidences for all candidate symbols + let symbols: Vec = 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; diff --git a/services/trading_agent_service/src/lib.rs b/services/trading_agent_service/src/lib.rs index 8f90802f7..c55aef827 100644 --- a/services/trading_agent_service/src/lib.rs +++ b/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;