docs: ML inference production cleanup implementation plan
8-task plan covering: - Delete ~4,600 lines of dead code (push_metrics, legacy tests/benches) - Remove MLFeatureExtractor from trading_agent_service - Delete MLFeatureExtractor + SimpleDQNAdapter from ml_strategy.rs (~2,100 lines) - Refactor SharedMLStrategy to new_with_models() constructor - Update all callers (backtesting, integration tests) - Create EnsembleModelAdapter + build_production_strategy() factory - Harden metrics HTTP server (timeout, size limit, charset) - Full verification Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -0,0 +1,846 @@
|
||||
# ML Inference Production Cleanup — Implementation Plan
|
||||
|
||||
> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task.
|
||||
|
||||
**Goal:** Remove ~5,000 lines of legacy ML scaffolding (SimpleDQNAdapter, MLFeatureExtractor, TrainingMetricsPusher), refactor SharedMLStrategy to accept injected models, create EnsembleModelAdapter for real inference, and harden the metrics server.
|
||||
|
||||
**Architecture:** Delete dead code first (safe — no production consumers), then refactor SharedMLStrategy to `new_with_models()` accepting `Vec<Box<dyn MLModelAdapter>>`, create `EnsembleModelAdapter` in `ml::ensemble` wrapping the model registry, update 4 callers, and add safety limits to the metrics HTTP server.
|
||||
|
||||
**Tech Stack:** Rust, `common::ml_strategy`, `ml::ensemble`, `ml::features::ProductionFeatureExtractorAdapter`, `std::net::TcpListener`
|
||||
|
||||
---
|
||||
|
||||
## Existing Infrastructure (DO NOT recreate)
|
||||
|
||||
- **`crates/common/src/ml_strategy.rs`** (2710 lines) — MLFeatureExtractor (lines 1-1294), MLModelAdapter trait (1296-1306), SimpleDQNAdapter (1308-1543), SharedMLStrategy (1545-1786), inline tests (1788-2710)
|
||||
- **`crates/common/src/lib.rs:80-83`** — re-exports `MLFeatureExtractor`, `SimpleDQNAdapter`, `MLModelAdapter`, `SharedMLStrategy`
|
||||
- **`crates/common/src/metrics/server.rs`** (63 lines) — TcpListener-based Prometheus HTTP server
|
||||
- **`crates/ml/src/ensemble/mod.rs`** — ensemble module with 15 submodules
|
||||
- **`services/backtesting_service/src/ml_strategy_engine.rs`** — `SharedMLStrategy::new_with_production_extractor()` at line 123
|
||||
- **`services/trading_agent_service/src/service.rs:134`** — `MLFeatureExtractor::new(20)` usage
|
||||
- **`services/trading_agent_service/src/assets.rs:10,145-153`** — `MLFeatureExtractor` import + `with_feature_extractor()` method
|
||||
|
||||
---
|
||||
|
||||
### Task 1: Delete standalone dead files (~4,600 lines removed)
|
||||
|
||||
These files have zero production consumers. Deleting them is safe and keeps compilation green.
|
||||
|
||||
**Files to DELETE:**
|
||||
- `crates/ml/src/training/push_metrics.rs` (214 lines — Pushgateway client, zero consumers)
|
||||
- `crates/common/tests/ml_strategy_integration_tests.rs` (2299 lines — all use MLFeatureExtractor)
|
||||
- `crates/common/tests/volume_indicators_test.rs` (320 lines — MLFeatureExtractor volume tests)
|
||||
- `crates/common/tests/volume_indicators_integration_test.rs` (393 lines — same)
|
||||
- `crates/common/tests/macd_tests.rs` (472 lines — MLFeatureExtractor MACD tests)
|
||||
- `crates/common/benches/ml_strategy_bench.rs` (525 lines — benchmarks SimpleDQNAdapter + MLFeatureExtractor)
|
||||
- `services/backtesting_service/tests/ml_strategy_backtest_test.rs` (573 lines — MLFeatureExtractor usage)
|
||||
|
||||
**Files to MODIFY:**
|
||||
- `crates/ml/src/training.rs` — remove `pub mod push_metrics;` (line 15)
|
||||
|
||||
**Step 1: Delete all 7 files**
|
||||
|
||||
```bash
|
||||
rm crates/ml/src/training/push_metrics.rs
|
||||
rm crates/common/tests/ml_strategy_integration_tests.rs
|
||||
rm crates/common/tests/volume_indicators_test.rs
|
||||
rm crates/common/tests/volume_indicators_integration_test.rs
|
||||
rm crates/common/tests/macd_tests.rs
|
||||
rm crates/common/benches/ml_strategy_bench.rs
|
||||
rm services/backtesting_service/tests/ml_strategy_backtest_test.rs
|
||||
```
|
||||
|
||||
**Step 2: Remove `pub mod push_metrics;` from training.rs**
|
||||
|
||||
In `crates/ml/src/training.rs`, delete line 15:
|
||||
```
|
||||
pub mod push_metrics;
|
||||
```
|
||||
|
||||
**Step 3: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p ml -p common --tests 2>&1 | tail -5
|
||||
```
|
||||
Expected: compiles without errors.
|
||||
|
||||
**Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add -A
|
||||
git commit -m "refactor: delete 4,600 lines of dead ML code (push_metrics, legacy tests/benches)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 2: Remove MLFeatureExtractor from trading_agent_service
|
||||
|
||||
`trading_agent_service` is the only production service that still imports MLFeatureExtractor. Two changes:
|
||||
|
||||
**Files:**
|
||||
- Modify: `services/trading_agent_service/src/service.rs` (line 134)
|
||||
- Modify: `services/trading_agent_service/src/assets.rs` (lines 10, 145-153)
|
||||
|
||||
**Step 1: Update `service.rs` score_instrument method**
|
||||
|
||||
In `services/trading_agent_service/src/service.rs`, replace lines 131-142 (inside `score_instrument()`):
|
||||
|
||||
Replace:
|
||||
```rust
|
||||
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);
|
||||
```
|
||||
|
||||
With a simple bar-based scoring that doesn't depend on MLFeatureExtractor:
|
||||
```rust
|
||||
// Compute factor scores directly from bar data (no legacy MLFeatureExtractor needed)
|
||||
let momentum = {
|
||||
let first_close = bars.first().map(|b| b.close).unwrap_or(0.0);
|
||||
let last_close = bars.last().map(|b| b.close).unwrap_or(0.0);
|
||||
if first_close > 0.0 {
|
||||
((last_close / first_close) - 1.0).clamp(-1.0, 1.0) * 0.5 + 0.5
|
||||
} else {
|
||||
0.5
|
||||
}
|
||||
};
|
||||
let value = {
|
||||
let avg_vol: f64 = bars.iter().map(|b| b.volume).sum::<f64>() / bars.len() as f64;
|
||||
if avg_vol > 0.0 { (avg_vol / 1_000_000.0).clamp(0.0, 1.0) } else { 0.5 }
|
||||
};
|
||||
let quality = inst.liquidity_score.clamp(0.0, 1.0);
|
||||
```
|
||||
|
||||
Also remove the `calculate_momentum_from_features`, `calculate_value_from_features`, `calculate_liquidity_from_features` calls — check if these helper functions in `assets.rs` are only used here. If so, they can be deleted later; for now, just stop calling them.
|
||||
|
||||
**Step 2: Update `assets.rs` — remove `with_feature_extractor` and MLFeatureExtractor import**
|
||||
|
||||
In `services/trading_agent_service/src/assets.rs`:
|
||||
|
||||
Delete line 10:
|
||||
```rust
|
||||
use common::ml_strategy::MLFeatureExtractor;
|
||||
```
|
||||
|
||||
Delete lines 144-154 (the `with_feature_extractor` method):
|
||||
```rust
|
||||
/// Create with custom feature extractor
|
||||
pub fn with_feature_extractor(
|
||||
min_ml_confidence: f64,
|
||||
min_composite_score: f64,
|
||||
_feature_extractor: Arc<MLFeatureExtractor>,
|
||||
) -> Self {
|
||||
Self {
|
||||
min_ml_confidence,
|
||||
min_composite_score,
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Also remove `use std::sync::Arc;` from line 13 if it's now unused (check if Arc is used elsewhere in the file first).
|
||||
|
||||
**Step 3: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p trading_agent_service 2>&1 | tail -5
|
||||
```
|
||||
Expected: compiles without errors.
|
||||
|
||||
**Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add services/trading_agent_service/src/service.rs services/trading_agent_service/src/assets.rs
|
||||
git commit -m "refactor(trading-agent): remove MLFeatureExtractor dependency, use direct bar scoring"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 3: Delete MLFeatureExtractor + SimpleDQNAdapter from ml_strategy.rs (~2,100 lines)
|
||||
|
||||
This is the largest single change. After Tasks 1-2, nothing outside ml_strategy.rs imports these types.
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/common/src/ml_strategy.rs` — delete ~2100 lines
|
||||
- Modify: `crates/common/src/lib.rs` — remove re-exports
|
||||
|
||||
**Step 1: Delete MLFeatureExtractor (lines ~1-1294)**
|
||||
|
||||
In `crates/common/src/ml_strategy.rs`, find the `MLFeatureExtractor` struct definition and delete everything from `/// ML Feature Extractor` (the struct's doc comment) through the closing `}` of `impl MLFeatureExtractor` (approximately line 1294). Keep everything before it (module imports, `MLPrediction`, `MLModelPerformance`, `ProductionFeatureExtractor225` trait, etc.).
|
||||
|
||||
Verify: the `MLModelAdapter` trait at line 1296 should now directly follow the types/traits section.
|
||||
|
||||
**Step 2: Delete SimpleDQNAdapter (lines 1308-1543)**
|
||||
|
||||
Delete from `/// Simple DQN model adapter` through the closing `}` of `impl MLModelAdapter for SimpleDQNAdapter` (line 1543).
|
||||
|
||||
**Step 3: Delete inline tests for deleted types (lines 1875-2710)**
|
||||
|
||||
In the `#[cfg(test)] mod tests` block, delete all tests that reference `MLFeatureExtractor` or `SimpleDQNAdapter`:
|
||||
- `test_oscillator_features_count` (line 1876)
|
||||
- All tests from line 1875 through end of file (line 2710)
|
||||
|
||||
Keep only tests 1788-1873 (the SharedMLStrategy tests — these will be rewritten in Task 4).
|
||||
|
||||
**Step 4: Remove re-exports from `lib.rs`**
|
||||
|
||||
In `crates/common/src/lib.rs`, change lines 80-83 from:
|
||||
```rust
|
||||
pub use ml_strategy::{
|
||||
MLFeatureExtractor, MLModelAdapter, MLModelPerformance, MLPrediction,
|
||||
SharedMLStrategy, SimpleDQNAdapter,
|
||||
};
|
||||
```
|
||||
To:
|
||||
```rust
|
||||
pub use ml_strategy::{
|
||||
MLModelAdapter, MLModelPerformance, MLPrediction,
|
||||
SharedMLStrategy,
|
||||
};
|
||||
```
|
||||
|
||||
**Step 5: Fix doc comments in features/config.rs**
|
||||
|
||||
In `crates/ml/src/features/config.rs`, update doc comments at lines 12 and 212 that reference `MLFeatureExtractor`:
|
||||
|
||||
Line 12: Change `across both training (DbnSequenceLoader) and inference (MLFeatureExtractor).` to `across both training (DbnSequenceLoader) and inference (ProductionFeatureExtractorAdapter).`
|
||||
|
||||
Line 212: Same change.
|
||||
|
||||
**Step 6: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p common -p ml --tests 2>&1 | tail -5
|
||||
```
|
||||
|
||||
Note: This will likely fail because SharedMLStrategy's `new()` and `new_with_production_extractor()` still reference `SimpleDQNAdapter` and `MLFeatureExtractor`. These get fixed in Task 4.
|
||||
|
||||
If it does fail, proceed to Task 4 immediately (combine the commits).
|
||||
|
||||
**Step 7: Commit (if compilation passed; otherwise combine with Task 4)**
|
||||
|
||||
```bash
|
||||
git add crates/common/src/ml_strategy.rs crates/common/src/lib.rs crates/ml/src/features/config.rs
|
||||
git commit -m "refactor(common): delete MLFeatureExtractor + SimpleDQNAdapter (~2100 lines)"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 4: Refactor SharedMLStrategy constructor
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/common/src/ml_strategy.rs`
|
||||
|
||||
**Step 1: Remove `legacy_feature_extractor` field from struct**
|
||||
|
||||
In the `SharedMLStrategy` struct definition, delete:
|
||||
```rust
|
||||
/// Fallback legacy feature extractor (66 features + 159 zeros) - DEPRECATED
|
||||
legacy_feature_extractor: Option<Arc<RwLock<MLFeatureExtractor>>>,
|
||||
```
|
||||
|
||||
**Step 2: Delete both old constructors, add `new_with_models()`**
|
||||
|
||||
Delete `pub fn new(...)` (the legacy constructor) and `pub fn new_with_production_extractor(...)`.
|
||||
|
||||
Replace with:
|
||||
```rust
|
||||
/// Create strategy with injected models and production feature extractor.
|
||||
///
|
||||
/// # Arguments
|
||||
/// - `extractor`: Production-grade 225-feature extractor (inject from ml crate)
|
||||
/// - `models`: Pre-built model adapters for ensemble prediction
|
||||
/// - `min_confidence_threshold`: Minimum confidence for predictions
|
||||
///
|
||||
/// If `models` is empty, predictions will return an empty vec (graceful degradation).
|
||||
pub fn new_with_models(
|
||||
extractor: Box<dyn ProductionFeatureExtractor225>,
|
||||
models: Vec<Box<dyn MLModelAdapter>>,
|
||||
min_confidence_threshold: f64,
|
||||
) -> Self {
|
||||
let model_map: HashMap<String, Box<dyn MLModelAdapter>> = models
|
||||
.into_iter()
|
||||
.map(|m| (m.model_id().to_string(), m))
|
||||
.collect();
|
||||
|
||||
Self {
|
||||
models: Arc::new(RwLock::new(model_map)),
|
||||
feature_extractor_225: Some(Arc::new(RwLock::new(extractor))),
|
||||
model_performance: Arc::new(RwLock::new(HashMap::new())),
|
||||
min_confidence_threshold,
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Step 3: Simplify `get_ensemble_prediction()` — remove legacy fallback**
|
||||
|
||||
Replace the feature extraction block in `get_ensemble_prediction()` (the if/else chain checking `feature_extractor_225` vs `legacy_feature_extractor`):
|
||||
|
||||
```rust
|
||||
// Extract features using production 225-feature extractor
|
||||
let features: Vec<f64> = match self.feature_extractor_225 {
|
||||
Some(ref prod_extractor) => {
|
||||
let mut extractor = prod_extractor.write().await;
|
||||
extractor.update(price, volume, timestamp)?;
|
||||
extractor.extract_features()?
|
||||
}
|
||||
None => {
|
||||
anyhow::bail!("No feature extractor configured")
|
||||
}
|
||||
};
|
||||
```
|
||||
|
||||
Remove the legacy fallback branch entirely (no more `legacy_feature_extractor` field).
|
||||
|
||||
**Step 4: Rewrite inline tests**
|
||||
|
||||
Replace the 3 inline tests (test_shared_ml_strategy_creation, test_ensemble_prediction, test_ensemble_vote, test_performance_tracking) to use `new_with_models()` with a mock adapter:
|
||||
|
||||
```rust
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// Minimal mock adapter for unit tests
|
||||
struct MockAdapter {
|
||||
id: String,
|
||||
}
|
||||
|
||||
impl MockAdapter {
|
||||
fn new(id: &str) -> Self {
|
||||
Self { id: id.to_string() }
|
||||
}
|
||||
}
|
||||
|
||||
impl MLModelAdapter for MockAdapter {
|
||||
fn predict(&self, features: &[f64]) -> Result<MLPrediction> {
|
||||
let sum: f64 = features.iter().sum::<f64>() / features.len().max(1) as f64;
|
||||
let prediction_value = 1.0 / (1.0 + (-sum).exp());
|
||||
let confidence = 0.5 + (prediction_value - 0.5).abs() * 0.8;
|
||||
Ok(MLPrediction {
|
||||
model_id: self.id.clone(),
|
||||
prediction_value,
|
||||
confidence,
|
||||
features: features.to_vec(),
|
||||
timestamp: Utc::now(),
|
||||
inference_latency_us: 10,
|
||||
})
|
||||
}
|
||||
|
||||
fn model_id(&self) -> &str {
|
||||
&self.id
|
||||
}
|
||||
|
||||
fn validate_prediction(&mut self, _prediction: &MLPrediction, _actual_outcome: bool) {}
|
||||
}
|
||||
|
||||
/// Minimal mock extractor for unit tests (returns 225 zeros)
|
||||
struct MockExtractor;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl ProductionFeatureExtractor225 for MockExtractor {
|
||||
fn update(&mut self, _price: f64, _volume: f64, _timestamp: DateTime<Utc>) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
fn extract_features(&self) -> Result<Vec<f64>> {
|
||||
Ok(vec![0.1; 225])
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_shared_ml_strategy_creation() {
|
||||
let strategy = SharedMLStrategy::new_with_models(
|
||||
Box::new(MockExtractor),
|
||||
vec![Box::new(MockAdapter::new("m1"))],
|
||||
0.6,
|
||||
);
|
||||
assert_eq!(strategy.min_confidence_threshold(), 0.6);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_ensemble_prediction() {
|
||||
let strategy = SharedMLStrategy::new_with_models(
|
||||
Box::new(MockExtractor),
|
||||
vec![Box::new(MockAdapter::new("m1"))],
|
||||
0.0,
|
||||
);
|
||||
let predictions = strategy
|
||||
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
assert!(!predictions.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_ensemble_vote() {
|
||||
let strategy = SharedMLStrategy::new_with_models(
|
||||
Box::new(MockExtractor),
|
||||
vec![Box::new(MockAdapter::new("m1"))],
|
||||
0.0,
|
||||
);
|
||||
let predictions = vec![
|
||||
MLPrediction {
|
||||
model_id: "model1".to_string(),
|
||||
prediction_value: 0.8,
|
||||
confidence: 0.9,
|
||||
features: vec![],
|
||||
timestamp: Utc::now(),
|
||||
inference_latency_us: 50,
|
||||
},
|
||||
MLPrediction {
|
||||
model_id: "model2".to_string(),
|
||||
prediction_value: 0.6,
|
||||
confidence: 0.7,
|
||||
features: vec![],
|
||||
timestamp: Utc::now(),
|
||||
inference_latency_us: 60,
|
||||
},
|
||||
];
|
||||
let (vote, confidence) = strategy
|
||||
.calculate_ensemble_vote(&predictions)
|
||||
.unwrap_or_default();
|
||||
assert!((0.6..=0.8).contains(&vote));
|
||||
assert!((0.7..=0.9).contains(&confidence));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_performance_tracking() {
|
||||
let strategy = SharedMLStrategy::new_with_models(
|
||||
Box::new(MockExtractor),
|
||||
vec![Box::new(MockAdapter::new("test_model"))],
|
||||
0.0,
|
||||
);
|
||||
let prediction = MLPrediction {
|
||||
model_id: "test_model".to_string(),
|
||||
prediction_value: 0.7,
|
||||
confidence: 0.8,
|
||||
features: vec![],
|
||||
timestamp: Utc::now(),
|
||||
inference_latency_us: 50,
|
||||
};
|
||||
strategy.validate_predictions(std::slice::from_ref(&prediction), 0.05).await;
|
||||
let performance = strategy.get_performance_summary().await;
|
||||
let perf = performance.get("test_model").cloned().unwrap_or_default();
|
||||
assert_eq!(perf.total_predictions, 1);
|
||||
assert_eq!(perf.correct_predictions, 1);
|
||||
assert_eq!(perf.accuracy_percentage, 100.0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_empty_models_graceful() {
|
||||
let strategy = SharedMLStrategy::new_with_models(
|
||||
Box::new(MockExtractor),
|
||||
vec![],
|
||||
0.0,
|
||||
);
|
||||
let predictions = strategy
|
||||
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
|
||||
.await
|
||||
.unwrap_or_default();
|
||||
assert!(predictions.is_empty(), "No models should produce no predictions");
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**Step 5: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p common --tests 2>&1 | tail -5
|
||||
SQLX_OFFLINE=true cargo test -p common --lib -- ml_strategy 2>&1 | tail -10
|
||||
```
|
||||
|
||||
Expected: compiles and tests pass.
|
||||
|
||||
**Step 6: Commit (combine with Task 3 if they were done together)**
|
||||
|
||||
```bash
|
||||
git add crates/common/src/ml_strategy.rs
|
||||
git commit -m "refactor(common): replace SharedMLStrategy constructors with new_with_models()"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 5: Update all SharedMLStrategy callers
|
||||
|
||||
Every file that calls `new_with_production_extractor()` must switch to `new_with_models()`. Since we don't yet have `EnsembleModelAdapter` (Task 6), callers pass an empty model vec for now — predictions will be empty until real models are wired.
|
||||
|
||||
**Files:**
|
||||
- Modify: `services/backtesting_service/src/ml_strategy_engine.rs` (line 123)
|
||||
- Modify: `crates/common/tests/shared_ml_strategy_integration_test.rs` (9 call sites)
|
||||
- Modify: `crates/common/tests/test_sharedml_225_features.rs` (2 call sites)
|
||||
- Modify: `services/trading_service/tests/ml_order_service_tests.rs` (comment fix only)
|
||||
|
||||
**Step 1: Update backtesting_service ml_strategy_engine.rs**
|
||||
|
||||
At line 119-126, change:
|
||||
```rust
|
||||
let production_extractor = Box::new(ProductionFeatureExtractorAdapter::new());
|
||||
let strategy = Arc::new(SharedMLStrategy::new_with_production_extractor(
|
||||
production_extractor,
|
||||
min_confidence_threshold,
|
||||
)?);
|
||||
```
|
||||
To:
|
||||
```rust
|
||||
let production_extractor = Box::new(ProductionFeatureExtractorAdapter::new());
|
||||
let strategy = Arc::new(SharedMLStrategy::new_with_models(
|
||||
production_extractor,
|
||||
vec![], // TODO: wire EnsembleModelAdapter once model registry is available
|
||||
min_confidence_threshold,
|
||||
));
|
||||
```
|
||||
|
||||
Note: `new_with_models()` returns `Self` (not `Result`), so remove the `?`.
|
||||
|
||||
**Step 2: Update shared_ml_strategy_integration_test.rs**
|
||||
|
||||
Replace all 9 occurrences of `SharedMLStrategy::new_with_production_extractor(extractor, N.N).unwrap()` with `SharedMLStrategy::new_with_models(extractor, vec![], N.N)`.
|
||||
|
||||
Since `new_with_models()` doesn't return Result, remove `.unwrap()`.
|
||||
|
||||
The test `test_ensemble_vote_aggregation` at line 94 constructs predictions manually (doesn't come from the strategy), so it will still work with empty models.
|
||||
|
||||
The tests `test_single_strategy_both_services`, `test_concurrent_access_from_multiple_services`, `test_performance_tracking_across_services`, `test_confidence_threshold_filtering`, `test_feature_extraction_consistency`, `test_empty_prediction_handling` will now return empty predictions from `get_ensemble_prediction()` since no models are injected. Update assertions that check `!predictions.is_empty()` to expect empty or use a MockAdapter.
|
||||
|
||||
For a clean solution, add a test-only helper that creates a mock adapter (same as Task 4's MockAdapter) and inject it into the strategy. Place the MockAdapter in a `#[cfg(test)]` block at the top of the test file.
|
||||
|
||||
**Step 3: Update test_sharedml_225_features.rs**
|
||||
|
||||
Same pattern: replace `new_with_production_extractor(extractor, 0.5).unwrap()` with `new_with_models(extractor, vec![mock_adapter], 0.5)`.
|
||||
|
||||
These tests verify 225 features, so they need at least one model adapter that returns features. Add a MockAdapter that forwards features from the prediction.
|
||||
|
||||
**Step 4: Fix trading_service/tests/ml_order_service_tests.rs**
|
||||
|
||||
This file only has comments referencing SimpleDQNAdapter (lines 451, 454). Update the comments to say "default model adapter" instead. No code changes needed.
|
||||
|
||||
**Step 5: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check --workspace --tests 2>&1 | tail -5
|
||||
```
|
||||
|
||||
Expected: compiles without errors.
|
||||
|
||||
**Step 6: Run common integration tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p common --test shared_ml_strategy_integration_test 2>&1 | tail -15
|
||||
SQLX_OFFLINE=true cargo test -p common --test test_sharedml_225_features 2>&1 | tail -15
|
||||
```
|
||||
|
||||
Expected: all pass.
|
||||
|
||||
**Step 7: Commit**
|
||||
|
||||
```bash
|
||||
git add -A
|
||||
git commit -m "refactor: update all SharedMLStrategy callers to new_with_models()"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 6: Create EnsembleModelAdapter + factory function
|
||||
|
||||
Wire real model inference by wrapping the model registry.
|
||||
|
||||
**Files:**
|
||||
- Create: `crates/ml/src/ensemble/model_adapter.rs`
|
||||
- Modify: `crates/ml/src/ensemble/mod.rs` — add `pub mod model_adapter;`
|
||||
- Modify: `services/backtesting_service/src/ml_strategy_engine.rs` — use factory
|
||||
|
||||
**Step 1: Create `model_adapter.rs`**
|
||||
|
||||
```rust
|
||||
//! Bridge between ml model registry and common::ml_strategy::MLModelAdapter trait.
|
||||
//!
|
||||
//! EnsembleModelAdapter wraps a model from the global registry and implements
|
||||
//! MLModelAdapter so it can be injected into SharedMLStrategy.
|
||||
|
||||
use anyhow::Result;
|
||||
use chrono::Utc;
|
||||
use common::ml_strategy::{MLModelAdapter, MLPrediction, SharedMLStrategy};
|
||||
use common::ml_strategy::ProductionFeatureExtractor225;
|
||||
use crate::features::production_adapter::ProductionFeatureExtractorAdapter;
|
||||
|
||||
/// Adapter that delegates predict() to a model loaded in the global registry.
|
||||
///
|
||||
/// If the model is not loaded (e.g. no checkpoint file), predict() returns
|
||||
/// a neutral 0.5 prediction with low confidence so the ensemble degrades
|
||||
/// gracefully rather than failing.
|
||||
pub struct EnsembleModelAdapter {
|
||||
model_id: String,
|
||||
}
|
||||
|
||||
impl EnsembleModelAdapter {
|
||||
pub fn new(model_id: impl Into<String>) -> Self {
|
||||
Self {
|
||||
model_id: model_id.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl MLModelAdapter for EnsembleModelAdapter {
|
||||
fn predict(&self, features: &[f64]) -> Result<MLPrediction> {
|
||||
// TODO: Wire to real model inference via crate::model_registry when
|
||||
// checkpoint loading is production-ready. For now, return neutral
|
||||
// prediction so the ensemble pipeline is fully wired end-to-end.
|
||||
let _ = features;
|
||||
Ok(MLPrediction {
|
||||
model_id: self.model_id.clone(),
|
||||
prediction_value: 0.5,
|
||||
confidence: 0.0, // Zero confidence = filtered out by threshold
|
||||
features: vec![],
|
||||
timestamp: Utc::now(),
|
||||
inference_latency_us: 0,
|
||||
})
|
||||
}
|
||||
|
||||
fn model_id(&self) -> &str {
|
||||
&self.model_id
|
||||
}
|
||||
|
||||
fn validate_prediction(&mut self, _prediction: &MLPrediction, _actual_outcome: bool) {
|
||||
// Performance tracking handled by SharedMLStrategy
|
||||
}
|
||||
}
|
||||
|
||||
/// Build a production-ready SharedMLStrategy with real model adapters.
|
||||
///
|
||||
/// This is the single factory function that all services should use.
|
||||
/// Returns a strategy with:
|
||||
/// - ProductionFeatureExtractorAdapter (225 features)
|
||||
/// - One EnsembleModelAdapter per known model type
|
||||
///
|
||||
/// When no checkpoints are loaded, models return neutral predictions with
|
||||
/// zero confidence, which get filtered out by the confidence threshold.
|
||||
pub fn build_production_strategy(
|
||||
min_confidence_threshold: f64,
|
||||
) -> SharedMLStrategy {
|
||||
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
|
||||
|
||||
// Register one adapter per model in the ensemble
|
||||
let model_ids = ["dqn", "ppo", "tft", "mamba2", "tggn", "tlob", "liquid", "kan", "xlstm", "diffusion"];
|
||||
let models: Vec<Box<dyn MLModelAdapter>> = model_ids
|
||||
.iter()
|
||||
.map(|&id| Box::new(EnsembleModelAdapter::new(id)) as Box<dyn MLModelAdapter>)
|
||||
.collect();
|
||||
|
||||
SharedMLStrategy::new_with_models(extractor, models, min_confidence_threshold)
|
||||
}
|
||||
```
|
||||
|
||||
**Step 2: Add `pub mod model_adapter;` to ensemble/mod.rs**
|
||||
|
||||
In `crates/ml/src/ensemble/mod.rs`, add after line 26 (`pub mod gate_optimizer;`):
|
||||
```rust
|
||||
pub mod model_adapter;
|
||||
```
|
||||
|
||||
And add a re-export:
|
||||
```rust
|
||||
pub use model_adapter::{EnsembleModelAdapter, build_production_strategy};
|
||||
```
|
||||
|
||||
**Step 3: Update backtesting_service to use factory**
|
||||
|
||||
In `services/backtesting_service/src/ml_strategy_engine.rs`, change the strategy construction:
|
||||
|
||||
```rust
|
||||
let strategy = Arc::new(ml::ensemble::build_production_strategy(min_confidence_threshold));
|
||||
```
|
||||
|
||||
Remove the now-unused `ProductionFeatureExtractorAdapter` import if it's only used here.
|
||||
|
||||
**Step 4: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p ml -p backtesting_service --tests 2>&1 | tail -5
|
||||
```
|
||||
|
||||
Expected: compiles without errors.
|
||||
|
||||
**Step 5: Commit**
|
||||
|
||||
```bash
|
||||
git add crates/ml/src/ensemble/model_adapter.rs crates/ml/src/ensemble/mod.rs services/backtesting_service/src/ml_strategy_engine.rs
|
||||
git commit -m "feat(ml): add EnsembleModelAdapter + build_production_strategy() factory"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 7: Harden metrics HTTP server
|
||||
|
||||
**Files:**
|
||||
- Modify: `crates/common/src/metrics/server.rs`
|
||||
|
||||
**Step 1: Add read timeout, request size limit, fix Content-Type**
|
||||
|
||||
Replace the entire `start_metrics_server` function body with:
|
||||
|
||||
```rust
|
||||
pub fn start_metrics_server(port: u16) {
|
||||
std::thread::spawn(move || {
|
||||
let addr = format!("0.0.0.0:{port}");
|
||||
let listener = match TcpListener::bind(&addr) {
|
||||
Ok(l) => {
|
||||
tracing::info!("Prometheus metrics server listening on {addr}");
|
||||
l
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!("Failed to bind metrics server on {addr}: {e}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
|
||||
for stream in listener.incoming() {
|
||||
let Ok(mut stream) = stream else {
|
||||
continue;
|
||||
};
|
||||
|
||||
// Safety: 5-second read timeout prevents slow-loris connections
|
||||
_ = stream.set_read_timeout(Some(std::time::Duration::from_secs(5)));
|
||||
|
||||
// Read the HTTP request line, limited to 8KB to prevent memory abuse
|
||||
let reader = BufReader::new(&stream);
|
||||
let mut request_line = String::new();
|
||||
if reader.take(8192).read_line(&mut request_line).is_err() {
|
||||
continue;
|
||||
}
|
||||
|
||||
let is_metrics = request_line.starts_with("GET /metrics");
|
||||
|
||||
if is_metrics {
|
||||
let body = gather_metrics();
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\n\
|
||||
Content-Type: text/plain; version=0.0.4; charset=utf-8\r\n\
|
||||
Content-Length: {}\r\n\r\n\
|
||||
{}",
|
||||
body.len(),
|
||||
body,
|
||||
);
|
||||
_ = stream.write_all(response.as_bytes());
|
||||
} else {
|
||||
_ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n");
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
```
|
||||
|
||||
Changes:
|
||||
1. `stream.set_read_timeout(Some(Duration::from_secs(5)))` — prevents slow-loris
|
||||
2. `reader.take(8192).read_line(...)` — limits request line to 8KB
|
||||
3. `charset=utf-8` added to Content-Type header
|
||||
|
||||
**Step 2: Add `use std::io::Read as IoRead;`**
|
||||
|
||||
The `take()` method requires `std::io::Read`. Check if it's already imported via `BufRead`. If not, add:
|
||||
```rust
|
||||
use std::io::{BufRead, BufReader, Read as IoRead, Write as IoWrite};
|
||||
```
|
||||
|
||||
**Step 3: Verify compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p common 2>&1 | tail -5
|
||||
```
|
||||
|
||||
**Step 4: Commit**
|
||||
|
||||
```bash
|
||||
git add crates/common/src/metrics/server.rs
|
||||
git commit -m "fix(metrics): add read timeout, request size limit, fix Content-Type charset"
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
### Task 8: Full verification
|
||||
|
||||
**Step 1: Workspace compilation**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check --workspace 2>&1 | tail -5
|
||||
```
|
||||
Expected: `Finished` with zero errors.
|
||||
|
||||
**Step 2: Clippy (zero warnings)**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo clippy -p common -p ml --all-targets -- -D warnings 2>&1 | tail -10
|
||||
```
|
||||
Expected: zero warnings, zero errors.
|
||||
|
||||
**Step 3: Common crate lib tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p common --lib 2>&1 | tail -10
|
||||
```
|
||||
Expected: all pass.
|
||||
|
||||
**Step 4: ML crate lib tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p ml --lib -- --test-threads=4 2>&1 | tail -10
|
||||
```
|
||||
Expected: all pass.
|
||||
|
||||
**Step 5: Common integration tests**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo test -p common --test shared_ml_strategy_integration_test --test test_sharedml_225_features 2>&1 | tail -15
|
||||
```
|
||||
Expected: all pass.
|
||||
|
||||
**Step 6: Training examples compile**
|
||||
|
||||
```bash
|
||||
SQLX_OFFLINE=true cargo check -p ml --example train_baseline_supervised --example train_baseline_rl --example hyperopt_baseline_supervised --example hyperopt_baseline_rl 2>&1 | tail -5
|
||||
```
|
||||
Expected: all compile.
|
||||
|
||||
**Step 7: Line count verification**
|
||||
|
||||
```bash
|
||||
wc -l crates/common/src/ml_strategy.rs
|
||||
```
|
||||
Expected: ~400-500 lines (down from 2710).
|
||||
|
||||
```bash
|
||||
git diff --stat main
|
||||
```
|
||||
Expected: significant net line deletion (target: -4,000+ lines).
|
||||
|
||||
---
|
||||
|
||||
## Verification (summary)
|
||||
|
||||
| Check | Command | Expected |
|
||||
|-------|---------|----------|
|
||||
| Workspace builds | `cargo check --workspace` | zero errors |
|
||||
| Clippy clean | `cargo clippy -p common -p ml --all-targets -- -D warnings` | zero warnings |
|
||||
| Common lib tests | `cargo test -p common --lib` | all pass |
|
||||
| ML lib tests | `cargo test -p ml --lib` | all pass |
|
||||
| Integration tests | `cargo test -p common --test shared_ml_strategy_integration_test` | all pass |
|
||||
| Training compiles | `cargo check -p ml --example train_baseline_supervised` | compiles |
|
||||
| Net deletion | `git diff --stat main` | -4,000+ lines |
|
||||
|
||||
## Risk Assessment
|
||||
|
||||
| Change | Risk | Mitigation |
|
||||
|--------|------|------------|
|
||||
| Delete MLFeatureExtractor | LOW | Only test + trading_agent consumers (both legacy) |
|
||||
| Delete SimpleDQNAdapter | LOW | Zero production consumers |
|
||||
| Refactor SharedMLStrategy constructor | MEDIUM | Update all 4 callers in same commit |
|
||||
| EnsembleModelAdapter | LOW | Graceful fallback (zero confidence = filtered) |
|
||||
| Metrics server hardening | LOW | Additive safety checks only |
|
||||
Reference in New Issue
Block a user