feat(trading-agent): wire SelectAssets and GetSelectedAssets to AssetSelector

Replace 2 stub gRPC handlers with real implementations:
- SelectAssets: scores instruments via MLFeatureExtractor, ranks with
  AssetSelector (TopN/Threshold/Quantile modes), persists to DB
- GetSelectedAssets: loads latest selection from asset_selections table
Add helper methods: score_instrument, store/load_asset_selection

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

services/trading_agent_service/src/service.rs

@@ -11,6 +11,9 @@ use tonic::{Request, Response, Status};
 use tracing::{error, info, instrument, warn};
 
 use crate::allocation::{AllocationMethod, AssetInfo, PortfolioAllocator};
+use crate::assets::{
+    self, AssetScore as InternalAssetScore, AssetSelector,
+};
 use crate::monitoring::TradingAgentMetrics;
 use crate::proto::trading_agent::*;
 use crate::strategies::{
@@ -22,7 +25,6 @@ use bigdecimal::ToPrimitive;
 use rust_decimal::Decimal;
 
 pub struct TradingAgentServiceImpl {
-    #[allow(dead_code)]
     db_pool: PgPool,
     universe_selector: UniverseSelector,
     strategy_coordinator: StrategyCoordinator,
@@ -113,6 +115,159 @@ impl TradingAgentServiceImpl {
         }
     }
 
+    /// Score a universe instrument by fetching price bars and computing factor scores.
+    ///
+    /// Returns an `InternalAssetScore` that can be fed to `AssetSelector`.
+    async fn score_instrument(
+        &self,
+        inst: &crate::universe::Instrument,
+    ) -> Result<InternalAssetScore, Status> {
+        let symbol = inst.symbol.as_str();
+
+        // Fetch recent price bars for feature extraction
+        let bars = self.fetch_recent_bars(symbol, 60).await?;
+
+        if bars.len() >= 26 {
+            // We have enough bars to build a feature vector via the common extractor.
+            // Build a quick feature vector from the most recent 26 bars (price/volume).
+            let mut extractor = common::ml_strategy::MLFeatureExtractor::new(20);
+            let mut features = Vec::new();
+            for bar in &bars {
+                features = extractor.extract_features(
+                    bar.close,
+                    bar.volume,
+                    bar.timestamp,
+                );
+            }
+
+            let momentum = assets::calculate_momentum_from_features(&features);
+            let value = assets::calculate_value_from_features(&features);
+            let quality = assets::calculate_liquidity_from_features(&features);
+
+            // ML score placeholder -- use liquidity_score from the instrument as a proxy
+            // (real ML inference would go here)
+            let ml_score = inst.liquidity_score.clamp(0.0, 1.0);
+
+            Ok(InternalAssetScore::new(
+                symbol.to_string(),
+                ml_score,
+                momentum,
+                value,
+                quality,
+            ))
+        } else {
+            // Not enough bars -- fall back to instrument metadata
+            let quality = inst.liquidity_score.clamp(0.0, 1.0);
+            let ml_score = quality; // proxy
+            let momentum = 0.5; // neutral
+            let value = if inst.volatility > 0.0 {
+                (1.0 / (1.0 + inst.volatility)).clamp(0.0, 1.0)
+            } else {
+                0.5
+            };
+
+            Ok(InternalAssetScore::new(
+                symbol.to_string(),
+                ml_score,
+                momentum,
+                value,
+                quality,
+            ))
+        }
+    }
+
+    /// Persist an asset selection to the database.
+    async fn store_asset_selection(
+        &self,
+        universe_id: &str,
+        criteria: &serde_json::Value,
+        scores: &[InternalAssetScore],
+        metrics: &serde_json::Value,
+    ) -> Result<(), Status> {
+        let scores_json = serde_json::to_value(scores)
+            .map_err(|e| Status::internal(format!("Failed to serialise scores: {e}")))?;
+
+        sqlx::query(
+            r#"
+            INSERT INTO asset_selections (universe_id, criteria, asset_scores, metrics)
+            VALUES ($1, $2, $3, $4)
+            "#,
+        )
+        .bind(universe_id)
+        .bind(criteria)
+        .bind(&scores_json)
+        .bind(metrics)
+        .execute(&self.db_pool)
+        .await
+        .map_err(|e| {
+            error!("Failed to persist asset selection: {}", e);
+            Status::internal(format!("Failed to persist asset selection: {e}"))
+        })?;
+
+        Ok(())
+    }
+
+    /// Load the most recent asset selection (optionally filtered by universe_id).
+    async fn load_latest_selection(
+        &self,
+        universe_id: Option<&str>,
+    ) -> Result<(Vec<InternalAssetScore>, serde_json::Value), Status> {
+        let row = if let Some(uid) = universe_id {
+            sqlx::query_as::<_, (serde_json::Value, serde_json::Value)>(
+                r#"
+                SELECT asset_scores, metrics
+                FROM asset_selections
+                WHERE universe_id = $1
+                ORDER BY selected_at DESC
+                LIMIT 1
+                "#,
+            )
+            .bind(uid)
+            .fetch_optional(&self.db_pool)
+            .await
+        } else {
+            sqlx::query_as::<_, (serde_json::Value, serde_json::Value)>(
+                r#"
+                SELECT asset_scores, metrics
+                FROM asset_selections
+                ORDER BY selected_at DESC
+                LIMIT 1
+                "#,
+            )
+            .fetch_optional(&self.db_pool)
+            .await
+        }
+        .map_err(|e| {
+            error!("Failed to load asset selection: {}", e);
+            Status::internal(format!("Failed to load asset selection: {e}"))
+        })?;
+
+        match row {
+            Some((scores_json, metrics_json)) => {
+                let scores: Vec<InternalAssetScore> =
+                    serde_json::from_value(scores_json).map_err(|e| {
+                        error!("Failed to deserialise asset scores: {}", e);
+                        Status::internal(format!("Failed to deserialise asset scores: {e}"))
+                    })?;
+                Ok((scores, metrics_json))
+            }
+            None => Ok((vec![], serde_json::json!({}))),
+        }
+    }
+
+    /// Convert an internal `AssetScore` to the proto `AssetScore` message.
+    fn convert_asset_score(score: &InternalAssetScore) -> AssetScore {
+        AssetScore {
+            symbol: score.symbol.clone(),
+            ml_score: score.ml_score,
+            momentum_score: score.momentum_score,
+            value_score: score.value_score,
+            quality_score: score.quality_score,
+            composite_score: score.composite_score,
+            model_scores: score.model_scores.clone(),
+        }
+    }
+
     /// Convert internal Instrument to proto
     fn convert_instrument(&self, inst: &crate::universe::Instrument) -> Instrument {
         Instrument {
@@ -293,43 +448,200 @@ impl trading_agent_service_server::TradingAgentService for TradingAgentServiceIm
     }
 
     // ============================================================================
-    // Asset Selection (Placeholder implementations)
+    // Asset Selection
     // ============================================================================
 
+    #[instrument(skip(self), fields(universe_id, max_assets))]
     async fn select_assets(
         &self,
-        _request: Request<SelectAssetsRequest>,
+        request: Request<SelectAssetsRequest>,
     ) -> Result<Response<SelectAssetsResponse>, Status> {
-        info!("SelectAssets called (placeholder)");
+        let req = request.into_inner();
+        info!(
+            "SelectAssets called for universe_id: {}, max_assets: {}",
+            req.universe_id, req.max_assets
+        );
+
+        let start = std::time::Instant::now();
+
+        // 1. Fetch the universe
+        let universe = self
+            .universe_selector
+            .get_universe(&req.universe_id)
+            .await
+            .map_err(|e| {
+                error!("Failed to get universe {}: {}", req.universe_id, e);
+                self.metrics.record_error("select_assets_universe_not_found");
+                Status::not_found(format!("Universe not found: {e}"))
+            })?;
+
+        // 2. Score each instrument
+        let total_evaluated = universe.instruments.len() as u32;
+        let mut internal_scores = Vec::with_capacity(universe.instruments.len());
+        for inst in &universe.instruments {
+            match self.score_instrument(inst).await {
+                Ok(score) => internal_scores.push(score),
+                Err(e) => {
+                    warn!(
+                        "Skipping instrument {} due to scoring error: {}",
+                        inst.symbol.as_str(),
+                        e
+                    );
+                }
+            }
+        }
+
+        // 3. Apply selection criteria via AssetSelector
+        let criteria = req.criteria.as_ref();
+        let min_ml = criteria
+            .map(|c| c.min_ml_signal_strength)
+            .unwrap_or(0.0);
+
+        // Build a threshold-aware selector from the request criteria
+        let selector = if min_ml > 0.0 {
+            AssetSelector::with_thresholds(min_ml, 0.0)
+        } else {
+            AssetSelector::new()
+        };
+
+        let mode = criteria
+            .and_then(|c| SelectionMode::try_from(c.mode).ok())
+            .unwrap_or(SelectionMode::TopN);
+
+        let selected = match mode {
+            SelectionMode::Threshold => selector.select_above_threshold(internal_scores),
+            SelectionMode::Quantile => {
+                // Use 20% quantile as default
+                selector.select_top_quantile(internal_scores, 0.20)
+            }
+            // TopN (default), Unspecified
+            _ => {
+                let n = if req.max_assets > 0 {
+                    req.max_assets as usize
+                } else {
+                    10
+                };
+                selector.select_top_n(internal_scores, n)
+            }
+        };
+
+        // 4. Compute selection metrics
+        let assets_selected = selected.len() as u32;
+        let (avg_composite, min_score, max_score) = if selected.is_empty() {
+            (0.0, 0.0, 0.0)
+        } else {
+            let sum: f64 = selected.iter().map(|s| s.composite_score).sum();
+            let min = selected
+                .iter()
+                .map(|s| s.composite_score)
+                .fold(f64::INFINITY, f64::min);
+            let max = selected
+                .iter()
+                .map(|s| s.composite_score)
+                .fold(f64::NEG_INFINITY, f64::max);
+            (sum / selected.len() as f64, min, max)
+        };
+
+        let selection_metrics = SelectionMetrics {
+            assets_evaluated: total_evaluated,
+            assets_selected,
+            avg_composite_score: avg_composite,
+            min_score,
+            max_score,
+        };
+
+        // 5. Persist to database
+        let criteria_json = serde_json::json!({
+            "min_ml_signal_strength": min_ml,
+            "mode": mode as i32,
+            "max_assets": req.max_assets,
+        });
+        let metrics_json = serde_json::json!({
+            "assets_evaluated": selection_metrics.assets_evaluated,
+            "assets_selected": selection_metrics.assets_selected,
+            "avg_composite_score": selection_metrics.avg_composite_score,
+            "min_score": selection_metrics.min_score,
+            "max_score": selection_metrics.max_score,
+        });
+
+        if let Err(e) = self
+            .store_asset_selection(&req.universe_id, &criteria_json, &selected, &metrics_json)
+            .await
+        {
+            warn!("Failed to persist asset selection (non-fatal): {}", e);
+        }
+
+        // 6. Record metrics
+        let duration_ms = start.elapsed().as_millis() as f64;
+        self.metrics
+            .record_asset_selection(duration_ms, assets_selected as u64);
+
+        info!(
+            "Asset selection complete: {}/{} assets selected in {}ms",
+            assets_selected, total_evaluated, duration_ms
+        );
+
+        // 7. Convert to proto
+        let proto_assets: Vec<AssetScore> =
+            selected.iter().map(Self::convert_asset_score).collect();
 
         Ok(Response::new(SelectAssetsResponse {
-            assets: vec![],
-            metrics: Some(SelectionMetrics {
-                assets_evaluated: 0,
-                assets_selected: 0,
-                avg_composite_score: 0.0,
-                min_score: 0.0,
-                max_score: 0.0,
-            }),
+            assets: proto_assets,
+            metrics: Some(selection_metrics),
             timestamp: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
         }))
     }
 
+    #[instrument(skip(self), fields(universe_id))]
     async fn get_selected_assets(
         &self,
-        _request: Request<GetSelectedAssetsRequest>,
+        request: Request<GetSelectedAssetsRequest>,
     ) -> Result<Response<GetSelectedAssetsResponse>, Status> {
-        info!("GetSelectedAssets called (placeholder)");
+        let req = request.into_inner();
+        let uid = req.universe_id.as_deref();
+        info!(
+            "GetSelectedAssets called (universe_id: {:?})",
+            uid
+        );
+
+        let (scores, _metrics_json) = self.load_latest_selection(uid).await?;
+
+        let proto_assets: Vec<AssetScore> =
+            scores.iter().map(Self::convert_asset_score).collect();
+
+        // Compute metrics from the returned scores
+        let count = proto_assets.len() as u32;
+        let (avg, min_s, max_s) = if proto_assets.is_empty() {
+            (0.0, 0.0, 0.0)
+        } else {
+            let sum: f64 = proto_assets.iter().map(|a| a.composite_score).sum();
+            let mn = proto_assets
+                .iter()
+                .map(|a| a.composite_score)
+                .fold(f64::INFINITY, f64::min);
+            let mx = proto_assets
+                .iter()
+                .map(|a| a.composite_score)
+                .fold(f64::NEG_INFINITY, f64::max);
+            (sum / count as f64, mn, mx)
+        };
+
+        let selection_metrics = SelectionMetrics {
+            assets_evaluated: count,
+            assets_selected: count,
+            avg_composite_score: avg,
+            min_score: min_s,
+            max_score: max_s,
+        };
+
+        info!(
+            "Returning {} selected assets",
+            proto_assets.len()
+        );
 
         Ok(Response::new(GetSelectedAssetsResponse {
-            assets: vec![],
-            metrics: Some(SelectionMetrics {
-                assets_evaluated: 0,
-                assets_selected: 0,
-                avg_composite_score: 0.0,
-                min_score: 0.0,
-                max_score: 0.0,
-            }),
+            assets: proto_assets,
+            metrics: Some(selection_metrics),
             timestamp: chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0),
         }))
     }