docs: add deferred stubs implementation plan

6-task plan: proto change, client compilation, retrain wiring, ensemble confidence RPC, portfolio positions DB query, verification. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-01 21:21:53 +01:00
parent e006518ab0
commit 6157ea0f77
1 changed files with 615 additions and 0 deletions
--- a/docs/plans/2026-03-01-deferred-stubs-implementation.md
+++ b/docs/plans/2026-03-01-deferred-stubs-implementation.md
@@ -0,0 +1,615 @@
+# Deferred Stubs Implementation Plan
+
+> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
+
+**Goal:** Replace 3 deferred stubs with real implementations: retrain_model gRPC wiring, portfolio positions from DB, and ML confidence from ensemble RPC.
+
+**Architecture:** Each stub is independent. Retrain wires trading_service → ml_training_service via proto additive change. Portfolio positions queries existing `broker_positions` DB table. ML confidence calls existing `GetEnsembleVote` RPC with fallback. All changes are additive with graceful fallback on service unavailability.
+
+**Tech Stack:** Rust, tonic gRPC, SQLX (offline mode), protobuf3, tokio
+
+---
+
+### Task 1: Add TrainingMode to ml_training.proto
+
+**Files:**
+- Modify: `services/ml_training_service/proto/ml_training.proto:71-78` — add fields to StartTrainingRequest
+
+**Step 1: Add TrainingMode enum and fields to StartTrainingRequest**
+
+In `services/ml_training_service/proto/ml_training.proto`, add a `TrainingMode` enum
+right before the `StartTrainingRequest` message (before line 71), and add 3 new fields
+to `StartTrainingRequest`:
+
+```protobuf
+// Training mode selection
+enum TrainingMode {
+  TRAINING_MODE_FULL = 0;        // Full training from scratch (default, backward-compatible)
+  TRAINING_MODE_FINE_TUNE = 1;   // Fine-tune from existing checkpoint
+}
+
+message StartTrainingRequest {
+  string model_type = 1;
+  DataSource data_source = 2;
+  Hyperparameters hyperparameters = 3;
+  bool use_gpu = 4;
+  string description = 5;
+  map<string, string> tags = 6;
+  // New fields for fine-tune support
+  TrainingMode mode = 7;                    // FULL (default) or FINE_TUNE
+  string resume_checkpoint_path = 8;        // Path to checkpoint for fine-tune
+  uint32 max_epochs = 9;                    // Override epoch count (0 = use default)
+}
+```
+
+Note: Fields 7-9 are additive — existing callers that don't set them get FULL mode with
+defaults, so this is backward compatible.
+
+**Step 2: Verify proto compiles**
+
+Run: `SQLX_OFFLINE=true cargo check -p ml_training_service 2>&1 | head -20`
+Expected: no proto-related errors. The generated Rust code will include the new enum and fields.
+
+**Step 3: Verify trading_service still compiles**
+
+The trading_service does NOT import ml_training.proto — it has its own `proto/ml.proto`.
+Run: `SQLX_OFFLINE=true cargo check -p trading_service 2>&1 | head -20`
+Expected: clean compile (no change to its proto).
+
+**Step 4: Commit**
+
+```
+git add services/ml_training_service/proto/ml_training.proto
+git commit -m "proto(ml_training): add TrainingMode enum for fine-tune support"
+```
+
+---
+
+### Task 2: Add ml_training.proto client to trading_service
+
+The trading_service needs to call ml_training_service's `StartTraining` RPC.
+It already compiles 5 protos via `tonic_prost_build`. We add ml_training.proto
+as a 6th, client-only.
+
+**Files:**
+- Modify: `services/trading_service/build.rs` — compile ml_training.proto (client-only)
+- Modify: `services/trading_service/src/lib.rs:28-53` — add `pub mod ml_training` to proto module
+
+**Step 1: Update build.rs to compile ml_training.proto**
+
+In `services/trading_service/build.rs`, add a second compilation pass after the existing
+`compile_protos` call. The ml_training.proto lives in the ml_training_service directory,
+so we reference it via relative path:
+
+```rust
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Existing: compile trading_service's own protos (server+client)
+    tonic_prost_build::configure()
+        .build_server(true)
+        .build_client(true)
+        .server_mod_attribute(".", "#[allow(unused_qualifications)]")
+        .client_mod_attribute(".", "#[allow(unused_qualifications)]")
+        .compile_protos(
+            &[
+                "proto/trading.proto",
+                "proto/risk.proto",
+                "proto/ml.proto",
+                "proto/config.proto",
+                "proto/monitoring.proto",
+            ],
+            &["proto"],
+        )?;
+
+    // NEW: compile ml_training.proto (client-only, for retrain_model forwarding)
+    tonic_prost_build::configure()
+        .build_server(false)
+        .build_client(true)
+        .client_mod_attribute(".", "#[allow(unused_qualifications)]")
+        .compile_protos(
+            &["../ml_training_service/proto/ml_training.proto"],
+            &["../ml_training_service/proto"],
+        )?;
+
+    println!("cargo:rerun-if-changed=../ml_training_service/proto/ml_training.proto");
+
+    Ok(())
+}
+```
+
+**Step 2: Add ml_training module to lib.rs proto block**
+
+In `services/trading_service/src/lib.rs`, inside the `pub mod proto { ... }` block
+(after the `monitoring` module, before the closing `}`), add:
+
+```rust
+    /// ML Training Service client definitions (for retrain forwarding)
+    pub mod ml_training {
+        tonic::include_proto!("ml_training");
+    }
+```
+
+**Step 3: Verify compilation**
+
+Run: `SQLX_OFFLINE=true cargo check -p trading_service 2>&1 | head -30`
+Expected: compiles clean. The new module generates `MlTrainingServiceClient` under
+`crate::proto::ml_training::ml_training_service_client`.
+
+**Step 4: Commit**
+
+```
+git add services/trading_service/build.rs services/trading_service/src/lib.rs
+git commit -m "feat(trading_service): compile ml_training.proto client for retrain forwarding"
+```
+
+---
+
+### Task 3: Wire retrain_model to ml_training_service
+
+**Files:**
+- Modify: `services/trading_service/src/services/enhanced_ml.rs:1018-1033` — replace stub with real gRPC call
+- Modify: `services/trading_service/src/services/enhanced_ml.rs:179-214` — add client field to struct
+
+**Step 1: Add MlTrainingServiceClient to EnhancedMLServiceImpl**
+
+In `services/trading_service/src/services/enhanced_ml.rs`, add imports at the top
+(after the existing `use` block near line 28):
+
+```rust
+use tokio::sync::OnceCell;
+use tonic::transport::Channel;
+```
+
+Add a new field to `EnhancedMLServiceImpl` struct (after line ~213, before the closing `}`):
+
+```rust
+    // gRPC client for forwarding retrain requests to ml_training_service
+    ml_training_client: Arc<OnceCell<crate::proto::ml_training::ml_training_service_client::MlTrainingServiceClient<Channel>>>,
+```
+
+**Step 2: Initialize the OnceCell in the constructor**
+
+Find the `EnhancedMLServiceImpl::new()` method (or wherever the struct is constructed).
+Add `ml_training_client: Arc::new(OnceCell::new())` to the field initialization.
+
+**Step 3: Add a helper to lazily connect**
+
+Add a private method on `EnhancedMLServiceImpl`:
+
+```rust
+    /// Lazily connect to ml_training_service. Endpoint from ML_TRAINING_SERVICE_URL env var.
+    async fn get_training_client(
+        &self,
+    ) -> Result<crate::proto::ml_training::ml_training_service_client::MlTrainingServiceClient<Channel>, Status> {
+        self.ml_training_client
+            .get_or_try_init(|| async {
+                let url = std::env::var("ML_TRAINING_SERVICE_URL")
+                    .unwrap_or_else(|_| "http://ml-training-service:50052".to_string());
+                let channel = Channel::from_shared(url)
+                    .map_err(|e| Status::internal(format!("invalid training service URL: {e}")))?
+                    .connect()
+                    .await
+                    .map_err(|e| Status::unavailable(format!("ml_training_service unreachable: {e}")))?;
+                Ok(crate::proto::ml_training::ml_training_service_client::MlTrainingServiceClient::new(channel))
+            })
+            .await
+            .cloned()
+    }
+```
+
+Note: `OnceCell::get_or_try_init` initializes at most once. After first successful connect,
+subsequent calls return the cached client instantly. On failure, it retries next call.
+
+**Step 4: Replace the retrain_model stub**
+
+Replace the entire `retrain_model` method body (lines 1018-1033) with:
+
+```rust
+    async fn retrain_model(
+        &self,
+        request: Request<RetrainModelRequest>,
+    ) -> Result<Response<RetrainModelResponse>, Status> {
+        let req = request.into_inner();
+        info!(model_name = %req.model_name, "retrain_model: forwarding to ml_training_service");
+
+        // Look up model to validate it exists and get its ModelType
+        let models = self.models.read().await;
+        let model_info = models.get(&req.model_name).ok_or_else(|| {
+            Status::not_found(format!("model '{}' not registered", req.model_name))
+        })?;
+        let model_type_str = model_info.model_type.as_str().to_string();
+        drop(models);
+
+        // Connect to ml_training_service
+        let mut client = self.get_training_client().await?;
+
+        // Build fine-tune request
+        use crate::proto::ml_training::{StartTrainingRequest as MlStartReq, TrainingMode};
+        let training_req = MlStartReq {
+            model_type: model_type_str,
+            data_source: None, // Service uses default data path
+            hyperparameters: None,
+            use_gpu: true,
+            description: format!("Fine-tune {} via retrain_model RPC", req.model_name),
+            tags: req.parameters.clone(),
+            mode: TrainingMode::TrainingModeFinetune.into(),
+            resume_checkpoint_path: String::new(), // Service finds latest checkpoint
+            max_epochs: 10,
+        };
+
+        let resp = client
+            .start_training(tonic::Request::new(training_req))
+            .await
+            .map_err(|e| {
+                warn!(error = %e, "ml_training_service StartTraining failed");
+                Status::unavailable(format!("training service error: {e}"))
+            })?
+            .into_inner();
+
+        Ok(Response::new(RetrainModelResponse {
+            success: true,
+            message: resp.message,
+            job_id: Some(resp.job_id),
+            started_at: std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .map(|d| d.as_secs() as i64)
+                .unwrap_or(0),
+        }))
+    }
+```
+
+**Step 5: Verify compilation**
+
+Run: `SQLX_OFFLINE=true cargo check -p trading_service 2>&1 | head -30`
+
+Fix any issues:
+- If `TrainingMode::TrainingModeFineture` doesn't match the generated enum variant name,
+  check the generated code. Prost converts `TRAINING_MODE_FINE_TUNE` → `TrainingModeFineture`
+  or similar. Use the actual variant. You can find it by grepping:
+  `grep -rn "TrainingMode" services/trading_service/target/` or look at the generated code.
+- If `OnceCell::get_or_try_init` signature doesn't match, use the `tokio::sync::OnceCell`
+  version (not `std::sync::OnceLock` which is sync-only).
+
+**Step 6: Run tests**
+
+Run: `SQLX_OFFLINE=true cargo test -p trading_service --lib 2>&1 | tail -5`
+Expected: existing tests pass (the test suite mocks the gRPC layer, so the client won't connect).
+
+**Step 7: Commit**
+
+```
+git add services/trading_service/src/services/enhanced_ml.rs
+git commit -m "feat(trading_service): wire retrain_model to ml_training_service via gRPC"
+```
+
+---
+
+### Task 4: Wire ML confidence to ensemble RPC
+
+**Files:**
+- Modify: `services/trading_agent_service/src/autonomous_scaling.rs:355-358` — add optional ML client field
+- Modify: `services/trading_agent_service/src/autonomous_scaling.rs:650-666` — replace heuristic with RPC + fallback
+
+The `trading_agent_service` compiles `proto/trading_agent.proto`, but the `GetEnsembleVote`
+RPC lives in `bin/fxt/proto/ml.proto`. We need to either:
+(a) add ml.proto compilation to trading_agent_service's build.rs, or
+(b) use the `ml` crate's in-process ensemble directly.
+
+Since the `ml` crate is already a dependency with `minimal-inference` feature, and the
+`AutonomousUniverseManager` doesn't run in the hot path, option (b) is simpler. But the
+design says gRPC — so we'll add the proto compilation.
+
+**Step 1: Add ml.proto compilation to trading_agent_service build.rs**
+
+Replace `services/trading_agent_service/build.rs` content with:
+
+```rust
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    // Existing: compile trading_agent.proto
+    tonic_prost_build::compile_protos("proto/trading_agent.proto")?;
+
+    // NEW: compile ml.proto (client-only, for GetEnsembleVote)
+    tonic_prost_build::configure()
+        .build_server(false)
+        .build_client(true)
+        .client_mod_attribute(".", "#[allow(unused_qualifications)]")
+        .compile_protos(
+            &["../../bin/fxt/proto/ml.proto"],
+            &["../../bin/fxt/proto"],
+        )?;
+
+    println!("cargo:rerun-if-changed=../../bin/fxt/proto/ml.proto");
+
+    Ok(())
+}
+```
+
+**Step 2: Add proto module for ml types**
+
+Find where the trading_agent_service declares its proto module (likely in `src/lib.rs` or
+`src/main.rs`). Add:
+
+```rust
+pub mod ml_proto {
+    tonic::include_proto!("ml");
+}
+```
+
+If there's no lib.rs, add it in the file that declares `mod autonomous_scaling`.
+
+**Step 3: Add optional MlServiceClient to AutonomousUniverseManager**
+
+In `autonomous_scaling.rs`, add imports:
+
+```rust
+use tokio::sync::OnceCell;
+use tonic::transport::Channel;
+use std::sync::Arc;
+```
+
+Modify the struct (line 355-358):
+
+```rust
+pub struct AutonomousUniverseManager {
+    constraints: SystemConstraints,
+    pool: PgPool,
+    ml_client: Arc<OnceCell<crate::ml_proto::ml_service_client::MlServiceClient<Channel>>>,
+}
+```
+
+Update `new()` and `with_constraints()` to initialize `ml_client: Arc::new(OnceCell::new())`.
+
+**Step 4: Add ensemble confidence helper with caching**
+
+Add a private method:
+
+```rust
+    /// Get ML confidence for symbols from ensemble service, with 5s cache per call.
+    /// Falls back to liquidity heuristic on any error.
+    async fn get_ensemble_confidences(
+        &self,
+        symbols: &[String],
+    ) -> HashMap<String, f64> {
+        let mut result = HashMap::new();
+
+        // Try gRPC ensemble
+        let client_result = 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::ml_proto::ml_service_client::MlServiceClient::new(channel))
+            })
+            .await;
+
+        if let Ok(client) = client_result {
+            let mut client = client.clone();
+            let request = crate::ml_proto::EnsembleRequest {
+                symbols: symbols.to_vec(),
+                model_names: vec![], // all models
+                method: 0,           // WEIGHTED_AVERAGE
+            };
+            match client.get_ensemble_vote(tonic::Request::new(request)).await {
+                Ok(resp) => {
+                    for vote in &resp.into_inner().votes {
+                        result.insert(vote.symbol.clone(), vote.confidence);
+                    }
+                    return result;
+                }
+                Err(e) => {
+                    tracing::warn!(error = %e, "ensemble RPC failed, using liquidity fallback");
+                }
+            }
+        }
+
+        result
+    }
+```
+
+**Step 5: Update score_symbols_mock to use ensemble confidences**
+
+In `score_symbols_mock()` (line 650-680), change the method signature to async and
+update the ml_confidence line:
+
+```rust
+    async fn score_symbols(
+        &self,
+        instruments: &[Instrument],
+        tier: &CapitalScalingTier,
+    ) -> Vec<SymbolScore> {
+        // Batch-fetch ensemble confidences for all 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| {
+                inst.liquidity_score >= (tier.min_liquidity / 5_000_000.0)
+                    && inst.avg_daily_volume >= tier.min_liquidity
+            })
+            .map(|inst| {
+                // Use ensemble confidence if available, fallback to liquidity heuristic
+                let ml_confidence = ensemble_confidences
+                    .get(&inst.symbol.to_string())
+                    .copied()
+                    .unwrap_or_else(|| inst.liquidity_score * 0.9 + 0.1);
+
+                // ... rest unchanged
+```
+
+Note: Rename from `score_symbols_mock` to `score_symbols` since it's no longer mock.
+
+**Step 6: Update all callers of score_symbols_mock**
+
+Grep for `score_symbols_mock` in the file and update call sites to `.await` the new async method.
+
+**Step 7: Verify compilation**
+
+Run: `SQLX_OFFLINE=true cargo check -p trading_agent_service 2>&1 | head -30`
+
+**Step 8: Run tests**
+
+Run: `SQLX_OFFLINE=true cargo test -p trading_agent_service --lib 2>&1 | tail -5`
+
+**Step 9: Commit**
+
+```
+git add services/trading_agent_service/
+git commit -m "feat(trading_agent): wire ML confidence to ensemble GetEnsembleVote with fallback"
+```
+
+---
+
+### Task 5: Implement portfolio positions from DB
+
+**Files:**
+- Modify: `crates/risk-data/src/var.rs:641-649` — replace stub with DB query
+
+The `broker_positions` table already exists (migration 047). The `VarRepositoryImpl`
+already has `db_pool: PgPool`. This is the simplest change — just query the existing table.
+
+**Step 1: Understand the mapping**
+
+`broker_positions` table columns → `PortfolioPosition` struct:
+- `symbol` → `symbol`
+- `quantity` → `quantity`
+- `market_value` → `market_value`
+- Account/portfolio mapping: `account_id` in DB, `portfolio_id` in the API.
+  For now, treat `portfolio_id` as `account_id` (same concept in our single-account system).
+- `currency` → not in table, default "USD" (futures are USD-denominated)
+- `asset_class` → not in table, default "futures" (all our instruments are futures)
+
+**Step 2: Replace the stub with a DB query**
+
+In `crates/risk-data/src/var.rs`, replace `get_portfolio_positions()` (lines 641-649) with:
+
+```rust
+    async fn get_portfolio_positions(
+        &self,
+        portfolio_id: &str,
+    ) -> RiskDataResult<Vec<PortfolioPosition>> {
+        info!("Getting portfolio positions for {}", portfolio_id);
+
+        // Query broker_positions table (populated by fill events from trading engine)
+        let rows = sqlx::query_as!(
+            PortfolioPositionRow,
+            r#"SELECT symbol, quantity, market_value
+               FROM broker_positions
+               WHERE account_id = $1 AND quantity != 0
+               ORDER BY symbol"#,
+            portfolio_id
+        )
+        .fetch_all(&self.db_pool)
+        .await
+        .map_err(|e| RiskDataError::Database(format!("failed to fetch positions: {e}")))?;
+
+        if rows.is_empty() {
+            warn!(
+                portfolio_id = %portfolio_id,
+                "no positions found in broker_positions table"
+            );
+        }
+
+        let positions = rows
+            .into_iter()
+            .map(|row| PortfolioPosition {
+                symbol: row.symbol,
+                quantity: row.quantity,
+                market_value: row.market_value.unwrap_or_default(),
+                currency: "USD".to_string(),
+                asset_class: "futures".to_string(),
+            })
+            .collect();
+
+        Ok(positions)
+    }
+```
+
+**Step 3: Add the helper row struct**
+
+Add near the `PortfolioPosition` struct definition (around line 129):
+
+```rust
+/// Internal row type for SQLX query mapping from broker_positions table
+#[derive(Debug, sqlx::FromRow)]
+struct PortfolioPositionRow {
+    symbol: String,
+    quantity: Decimal,
+    market_value: Option<Decimal>,
+}
+```
+
+**Step 4: Update SQLX offline metadata**
+
+Since we use `query_as!` macro, SQLX needs offline metadata. We can't run `cargo sqlx prepare`
+without a live DB, so instead use `query_as_unchecked!` or a manual `sqlx::query_as`:
+
+Replace `sqlx::query_as!` with a runtime query (no compile-time check needed for offline mode):
+
+```rust
+        let rows: Vec<PortfolioPositionRow> = sqlx::query_as(
+            "SELECT symbol, quantity, market_value \
+             FROM broker_positions \
+             WHERE account_id = $1 AND quantity != 0 \
+             ORDER BY symbol"
+        )
+        .bind(portfolio_id)
+        .fetch_all(&self.db_pool)
+        .await
+        .map_err(|e| RiskDataError::Database(format!("failed to fetch positions: {e}")))?;
+```
+
+This avoids the `query_as!` macro which requires offline metadata files.
+
+**Step 5: Verify compilation**
+
+Run: `SQLX_OFFLINE=true cargo check -p risk-data 2>&1 | head -20`
+
+**Step 6: Run tests**
+
+Run: `SQLX_OFFLINE=true cargo test -p risk-data --lib 2>&1 | tail -5`
+Expected: existing tests pass. The DB query won't execute in unit tests (no PgPool).
+
+**Step 7: Commit**
+
+```
+git add crates/risk-data/src/var.rs
+git commit -m "feat(risk-data): wire get_portfolio_positions to broker_positions DB table"
+```
+
+---
+
+### Task 6: Full workspace verification
+
+**Step 1: Workspace build**
+
+Run: `SQLX_OFFLINE=true cargo check --workspace 2>&1 | tail -10`
+Expected: zero errors
+
+**Step 2: Clippy on affected crates**
+
+Run: `SQLX_OFFLINE=true cargo clippy -p trading_service -p trading_agent_service -p risk-data -p ml_training_service --lib -- -D warnings 2>&1 | tail -20`
+Expected: zero warnings
+
+**Step 3: Test affected crates**
+
+Run:
+```bash
+SQLX_OFFLINE=true cargo test -p trading_service --lib 2>&1 | tail -3
+SQLX_OFFLINE=true cargo test -p trading_agent_service --lib 2>&1 | tail -3
+SQLX_OFFLINE=true cargo test -p risk-data --lib 2>&1 | tail -3
+SQLX_OFFLINE=true cargo test -p ml_training_service --lib 2>&1 | tail -3
+```
+
+**Step 4: Verify all 3 stubs are replaced**
+
+```bash
+grep -rn "Status::unavailable.*not yet wired" services/trading_service/src/services/enhanced_ml.rs
+grep -rn "STUB.*get_portfolio_positions" crates/risk-data/src/var.rs
+grep -rn "TODO.*Wire to real ensemble" services/trading_agent_service/src/autonomous_scaling.rs
+```
+
+Expected: zero matches for all three greps (stubs are gone).