From 6157ea0f779b856a6abc49f32246c6060503ff33 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 1 Mar 2026 21:21:53 +0100 Subject: [PATCH] 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 --- ...026-03-01-deferred-stubs-implementation.md | 615 ++++++++++++++++++ 1 file changed, 615 insertions(+) create mode 100644 docs/plans/2026-03-01-deferred-stubs-implementation.md diff --git a/docs/plans/2026-03-01-deferred-stubs-implementation.md b/docs/plans/2026-03-01-deferred-stubs-implementation.md new file mode 100644 index 000000000..b56c2a537 --- /dev/null +++ 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 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> { + // 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>>, +``` + +**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, 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, + ) -> Result, 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> { + // 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>>, +} +``` + +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 { + 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 { + // Batch-fetch ensemble confidences for all 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| { + 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> { + 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, +} +``` + +**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 = 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).