gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'feature/ml-inference-cleanup'

Browse Source
This commit is contained in:
jgrusewski
2026-03-01 19:09:28 +01:00
parent 169821b3da e9177095b7
commit 378438f9a1
23 changed files with 497 additions and 7392 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
crates/common/Cargo.toml
View File

@@ -80,8 +80,4 @@ tempfile = "3.8"
[features]
default = ["database"]
database = ["sqlx"]
questdb = ["questdb-rs"]
[[bench]]
name = "ml_strategy_bench"
harness = false
questdb = ["questdb-rs"]

525
crates/common/benches/ml_strategy_bench.rs
View File

@@ -1,525 +0,0 @@
//! Performance Benchmarks for 25-Feature ML Strategy System
//!
//! Agent A13 - Comprehensive latency and memory profiling for:
//! - Individual technical indicators (RSI, MACD, BB, ATR, Stochastic, ADX, CCI)
//! - Full 25-feature extraction end-to-end
//! - Memory usage analysis
//!
//! ## Targets
//! - Individual indicators: <5μs per update
//! - Full 25-feature extraction: <100μs per bar
//! - Memory: <500 bytes per symbol state
//!
//! ## Run Benchmarks
//! ```bash
//! cargo bench -p common --bench ml_strategy_bench
//! ```
use chrono::Utc;
use common::ml_strategy::MLFeatureExtractor;
use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion};
use std::time::Duration;
// ============================================================================
// Test Data Generator
// ============================================================================
/// Generate realistic market data for benchmarking
fn generate_market_data(num_bars: usize, seed: u64) -> Vec<(f64, f64)> {
use std::f64::consts::PI;
let mut rng = fastrand::Rng::with_seed(seed);
let mut data = Vec::with_capacity(num_bars);
let mut price = 100.0;
for i in 0..num_bars {
// Combine trend, cycle, and noise
let trend = (i as f64 * 0.01) % 10.0 - 5.0;
let cycle = (i as f64 * 0.1 * PI).sin() * 2.0;
let noise = (rng.f64() - 0.5) * 0.5;
price += trend * 0.01 + cycle * 0.05 + noise;
price = price.clamp(50.0, 150.0);
let volume = 10000.0 + (i as f64 * 0.5 * PI).sin().abs() * 5000.0 + rng.f64() * 2000.0;
data.push((price, volume));
}
data
}
// ============================================================================
// Individual Indicator Benchmarks
// ============================================================================
/// Benchmark RSI (14-period) incremental update
fn bench_rsi_update(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_rsi");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 42);
// Warm up extractor with 20 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(20) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 20;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features[23]); // RSI is at index 23
});
});
group.finish();
}
/// Benchmark MACD incremental update (EMA-12, EMA-26, Signal-9)
fn bench_macd_update(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_macd");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 43);
// Warm up extractor
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(26) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 26;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features[24]); // MACD line
black_box(features[25]); // MACD signal
});
});
group.finish();
}
/// Benchmark Bollinger Bands (20-period SMA + 2σ)
fn bench_bollinger_bands(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_bollinger_bands");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 44);
// Warm up with 20 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(20) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 20;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features[19]); // BB position at index 19
});
});
group.finish();
}
/// Benchmark Stochastic Oscillator (%K and %D)
fn bench_stochastic(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_stochastic");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 45);
// Warm up with 14 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(14) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 14;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features[20]); // Stochastic %K
black_box(features[21]); // Stochastic %D
});
});
group.finish();
}
/// Benchmark ADX (Average Directional Index, 14-period)
fn bench_adx(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_adx");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 46);
// Warm up with 14 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(14) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 14;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features[18]); // ADX at index 18
});
});
group.finish();
}
/// Benchmark CCI (Commodity Channel Index, 20-period)
fn bench_cci(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_cci");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 47);
// Warm up with 20 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(20) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 20;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features[22]); // CCI at index 22
});
});
group.finish();
}
/// Benchmark ATR (Average True Range) - part of ADX calculation
fn bench_atr(c: &mut Criterion) {
let mut group = c.benchmark_group("indicator_atr");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 48);
// Warm up with 14 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(14) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_update", |b| {
let mut ext = extractor.clone();
let mut idx = 14;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
// ATR is internal state, accessed via ADX feature
black_box(features[18]); // ADX uses ATR internally
});
});
group.finish();
}
// ============================================================================
// End-to-End Feature Extraction Benchmarks
// ============================================================================
/// Benchmark full 25-feature extraction (cold start)
fn bench_full_extraction_cold(c: &mut Criterion) {
let mut group = c.benchmark_group("full_extraction_cold");
group.measurement_time(Duration::from_secs(10));
let data = generate_market_data(30, 50);
group.bench_function("30_bars_cold_start", |b| {
let timestamp = Utc::now();
b.iter(|| {
let mut extractor = MLFeatureExtractor::new(30);
for (price, volume) in &data {
let features =
extractor.extract_features(black_box(*price), black_box(*volume), timestamp);
black_box(features);
}
});
});
group.finish();
}
/// Benchmark full 25-feature extraction (warm state, single update)
fn bench_full_extraction_warm(c: &mut Criterion) {
let mut group = c.benchmark_group("full_extraction_warm");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(1000, 51);
// Warm up extractor with 30 bars
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(30) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("single_bar_warm", |b| {
let mut ext = extractor.clone();
let mut idx = 30;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features);
});
});
group.finish();
}
/// Benchmark throughput: bars processed per second
fn bench_extraction_throughput(c: &mut Criterion) {
let mut group = c.benchmark_group("extraction_throughput");
group.measurement_time(Duration::from_secs(10));
for batch_size in [10, 100, 1000] {
let data = generate_market_data(batch_size, 52);
group.bench_with_input(
BenchmarkId::from_parameter(batch_size),
&batch_size,
|b, _| {
let timestamp = Utc::now();
b.iter(|| {
let mut extractor = MLFeatureExtractor::new(30);
for (price, volume) in &data {
let features = extractor.extract_features(
black_box(*price),
black_box(*volume),
timestamp,
);
black_box(features);
}
});
},
);
}
group.finish();
}
/// Benchmark feature extraction with different lookback windows
fn bench_lookback_impact(c: &mut Criterion) {
let mut group = c.benchmark_group("lookback_window_impact");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(100, 53);
for lookback in [20, 30, 50, 100] {
group.bench_with_input(
BenchmarkId::from_parameter(lookback),
&lookback,
|b, &lb| {
let timestamp = Utc::now();
b.iter(|| {
let mut extractor = MLFeatureExtractor::new(lb);
// Process all bars
for (price, volume) in &data {
let features = extractor.extract_features(
black_box(*price),
black_box(*volume),
timestamp,
);
black_box(features);
}
});
},
);
}
group.finish();
}
// ============================================================================
// Memory Benchmarks
// ============================================================================
/// Memory usage analysis for MLFeatureExtractor
fn bench_memory_usage(c: &mut Criterion) {
let mut group = c.benchmark_group("memory_usage");
group.measurement_time(Duration::from_secs(3));
group.bench_function("extractor_size", |b| {
b.iter(|| {
let extractor = MLFeatureExtractor::new(black_box(30));
black_box(std::mem::size_of_val(&extractor));
});
});
// Measure memory after warmup
group.bench_function("extractor_warm_size", |b| {
let data = generate_market_data(30, 54);
let timestamp = Utc::now();
b.iter(|| {
let mut extractor = MLFeatureExtractor::new(30);
// Fill with data
for (price, volume) in &data {
extractor.extract_features(*price, *volume, timestamp);
}
black_box(std::mem::size_of_val(&extractor));
});
});
group.finish();
}
// ============================================================================
// Latency Distribution Analysis
// ============================================================================
/// Measure P50/P95/P99 latencies for feature extraction
fn bench_latency_distribution(c: &mut Criterion) {
let mut group = c.benchmark_group("latency_distribution");
group.measurement_time(Duration::from_secs(10));
group.sample_size(1000); // Increase sample size for better percentile accuracy
let data = generate_market_data(1000, 55);
// Warm up extractor
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
for (price, volume) in data.iter().take(30) {
extractor.extract_features(*price, *volume, timestamp);
}
group.bench_function("p50_p95_p99_latency", |b| {
let mut ext = extractor.clone();
let mut idx = 30;
b.iter(|| {
let (price, volume) = data[idx % data.len()];
let features = ext.extract_features(black_box(price), black_box(volume), timestamp);
idx += 1;
black_box(features);
});
});
group.finish();
}
// ============================================================================
// Comparative Benchmarks
// ============================================================================
/// Compare feature extraction with/without oscillators
fn bench_oscillator_overhead(c: &mut Criterion) {
let mut group = c.benchmark_group("oscillator_overhead");
group.measurement_time(Duration::from_secs(5));
let data = generate_market_data(100, 56);
let timestamp = Utc::now();
// Benchmark: Extract only first 7 base features (price, volume, time)
group.bench_function("base_features_7", |b| {
b.iter(|| {
let mut extractor = MLFeatureExtractor::new(30);
for (price, volume) in &data {
let features =
extractor.extract_features(black_box(*price), black_box(*volume), timestamp);
// Access only base features
black_box(&features[0..7]);
}
});
});
// Benchmark: Full 26-feature extraction (7 base + 3 oscillators + 3 volume + 5 EMA + 8 new)
group.bench_function("full_features_26", |b| {
b.iter(|| {
let mut extractor = MLFeatureExtractor::new(30);
for (price, volume) in &data {
let features =
extractor.extract_features(black_box(*price), black_box(*volume), timestamp);
black_box(features);
}
});
});
group.finish();
}
// ============================================================================
// Criterion Configuration
// ============================================================================
criterion_group!(
benches,
// Individual indicators
bench_rsi_update,
bench_macd_update,
bench_bollinger_bands,
bench_stochastic,
bench_adx,
bench_cci,
bench_atr,
// End-to-end extraction
bench_full_extraction_cold,
bench_full_extraction_warm,
bench_extraction_throughput,
bench_lookback_impact,
// Memory analysis
bench_memory_usage,
// Latency distribution
bench_latency_distribution,
// Comparative analysis
bench_oscillator_overhead,
);
criterion_main!(benches);

4
crates/common/src/lib.rs
View File

@@ -78,8 +78,8 @@ pub mod trading;
// Re-export shared ML strategy types
pub use ml_strategy::{
MLFeatureExtractor, MLModelAdapter, MLModelPerformance, MLPrediction,
SharedMLStrategy, SimpleDQNAdapter,
MLModelAdapter, MLModelPerformance, MLPrediction,
SharedMLStrategy,
};
// Re-export canonical financial types (aliased to avoid collision with old types:: re-exports)

21
crates/common/src/metrics/server.rs
View File

@@ -4,7 +4,7 @@
//! existing HTTP stack. Training binaries and CLI tools that don't have an
//! HTTP server use this instead.
use std::io::{BufRead, BufReader, Write as IoWrite};
use std::io::{BufRead, BufReader, Read as IoRead, Write as IoWrite};
use std::net::TcpListener;
use super::gather_metrics;
@@ -14,6 +14,11 @@ use super::gather_metrics;
/// Responds to `GET /metrics` with the global registry output in Prometheus
/// text exposition format. Any other request gets a 404. The thread is
/// detached and dies with the process.
///
/// Safety hardening:
/// - 5-second read timeout prevents slow-loris connections
/// - 8KB request line limit prevents memory abuse
/// - Correct Content-Type charset for Prometheus scrapers
pub fn start_metrics_server(port: u16) {
std::thread::spawn(move || {
let addr = format!("0.0.0.0:{port}");
@@ -33,21 +38,25 @@ pub fn start_metrics_server(port: u16) {
continue;
};
// Read the HTTP request line (e.g. "GET /metrics HTTP/1.1")
let mut reader = BufReader::new(&stream);
// 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 mut request_line = String::new();
if reader.read_line(&mut request_line).is_err() {
if BufReader::new((&stream).take(8192))
.read_line(&mut request_line)
.is_err()
{
continue;
}
let is_metrics = request_line.starts_with("GET /metrics");
drop(reader);
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\r\n\
Content-Type: text/plain; version=0.0.4; charset=utf-8\r\n\
Content-Length: {}\r\n\r\n\
{}",
body.len(),

2518
crates/common/src/ml_strategy.rs
View File

File diff suppressed because it is too large Load Diff

472
crates/common/tests/macd_tests.rs
View File

@@ -1,472 +0,0 @@
//! MACD (Moving Average Convergence Divergence) Unit Tests
//! Agent A2 - Wave 19 - TDD Implementation
//!
//! Tests 2 MACD features: MACD line and MACD Signal line
//! Validates:
//! - Correct EMA periods (12, 26, 9)
//! - Convergence/divergence detection
//! - Zero crossover behavior
//! - Signal line smoothing
//! - Normalization to [-1, 1]
//! - O(1) incremental updates
//! - Performance (<8μs target)
use chrono::Utc;
use common::ml_strategy::MLFeatureExtractor;
use std::time::Instant;
#[test]
fn test_macd_feature_count() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Build up sufficient history (need 26+ bars for MACD, 34+ for signal)
for i in 0..50 {
let price = 4500.0 + (i as f64 * 0.25);
let volume = 100_000.0;
let features = extractor.extract_features(price, volume, timestamp);
// After sufficient warmup (50 bars), verify MACD features are present
if i >= 49 {
// Expected features:
// 0-17: Original 18 features
// 18: ADX (Agent A6)
// 19: Bollinger Bands Position (Agent A3)
// 20: Stochastic %K (Agent A5)
// 21: Stochastic %D (Agent A5)
// 22: CCI (Agent A7)
// 23: RSI (Agent A1)
// 24: MACD line (EMA12 - EMA26, normalized) - Agent A2
// 25: MACD Signal line (EMA9 of MACD, normalized) - Agent A2
// Wave C (4 features):
// 26: OBV Momentum (10-period ROC)
// 27: Volume Oscillator (5/20-period)
// 28: A/D Line (Accumulation/Distribution)
// 29: EMA Ratio (EMA-10 / EMA-50)
// Total: 30 features (26 Wave A + 4 Wave C)
assert_eq!(
features.len(),
30,
"Expected 30 features with Wave A + Wave C, got {} at iteration {}",
features.len(),
i
);
// MACD line (index 24)
let macd_line = features[24];
assert!(
macd_line.is_finite() && (-1.0..=1.0).contains(&macd_line),
"MACD line out of range: {} at iteration {}",
macd_line,
i
);
// MACD Signal line (index 25)
let macd_signal = features[25];
assert!(
macd_signal.is_finite() && (-1.0..=1.0).contains(&macd_signal),
"MACD Signal out of range: {} at iteration {}",
macd_signal,
i
);
}
}
}
#[test]
fn test_macd_convergence_bullish() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Phase 1: Downtrend (30 bars) - creates divergence
for i in 0..30 {
let price = 4600.0 - (i as f64 * 2.0); // Price declining
extractor.extract_features(price, 100_000.0, timestamp);
}
// Phase 2: Uptrend (30 bars) - MACD should converge (bullish)
for i in 0..30 {
let price = 4540.0 + (i as f64 * 1.5); // Price rising
let features = extractor.extract_features(price, 100_000.0, timestamp);
if i >= 25 && features.len() >= 26 {
let macd_line = features[24];
let macd_signal = features[25];
// During bullish convergence, MACD should be positive and rising
// MACD line should eventually cross above signal line
println!(
"Bar {}: MACD={:.6}, Signal={:.6}, Diff={:.6}",
i,
macd_line,
macd_signal,
macd_line - macd_signal
);
// MACD should be positive during uptrend (or approaching zero)
assert!(
macd_line.is_finite() && macd_signal.is_finite(),
"MACD values should be finite during convergence"
);
}
}
}
#[test]
fn test_macd_divergence_bearish() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Phase 1: Uptrend (30 bars) - creates convergence
for i in 0..30 {
let price = 4400.0 + (i as f64 * 2.0); // Price rising
extractor.extract_features(price, 100_000.0, timestamp);
}
// Phase 2: Downtrend (30 bars) - MACD should diverge (bearish)
for i in 0..30 {
let price = 4460.0 - (i as f64 * 1.5); // Price falling
let features = extractor.extract_features(price, 100_000.0, timestamp);
if i >= 25 && features.len() >= 26 {
let macd_line = features[24];
let macd_signal = features[25];
// During bearish divergence, MACD should be negative and falling
// MACD line should eventually cross below signal line
println!(
"Bar {}: MACD={:.6}, Signal={:.6}, Diff={:.6}",
i,
macd_line,
macd_signal,
macd_line - macd_signal
);
// MACD should be negative during downtrend (or approaching zero)
assert!(
macd_line.is_finite() && macd_signal.is_finite(),
"MACD values should be finite during divergence"
);
}
}
}
#[test]
fn test_macd_zero_crossover() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Phase 1: Establish flat market
for _ in 0..20 {
extractor.extract_features(4500.0, 100_000.0, timestamp);
}
// Phase 2: Sharp uptrend (crosses zero from below)
let mut macd_values = Vec::new();
let mut signal_values = Vec::new();
for i in 0..40 {
let price = 4500.0 + (i as f64 * 3.0); // Strong uptrend
let features = extractor.extract_features(price, 100_000.0, timestamp);
if i >= 20 && features.len() >= 26 {
let macd = features[24];
let signal = features[25];
macd_values.push(macd);
signal_values.push(signal);
println!(
"Bar {}: Price={:.2}, MACD={:.6}, Signal={:.6}",
i, price, macd, signal
);
}
}
// Verify MACD eventually becomes positive during strong uptrend
let positive_macd_count = macd_values.iter().filter(|&&m| m > 0.0).count();
assert!(
positive_macd_count > 5,
"MACD should show positive values during uptrend, got {} positive out of {}",
positive_macd_count,
macd_values.len()
);
}
#[test]
fn test_macd_signal_line_smoothing() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Create volatile price action
let mut macd_values = Vec::new();
let mut signal_values = Vec::new();
for i in 0..60 {
let price = 4500.0 + ((i as f64 / 3.0).sin() * 50.0); // Sinusoidal volatility
let features = extractor.extract_features(price, 100_000.0, timestamp);
if i >= 35 && features.len() >= 26 {
let macd = features[24];
let signal = features[25];
macd_values.push(macd);
signal_values.push(signal);
}
}
// Calculate volatility of MACD vs Signal
let macd_volatility = calculate_volatility(&macd_values);
let signal_volatility = calculate_volatility(&signal_values);
println!(
"MACD volatility: {:.6}, Signal volatility: {:.6}",
macd_volatility, signal_volatility
);
// Signal line should be smoother (less volatile) than MACD line
// This validates the EMA-9 smoothing
assert!(
signal_volatility < macd_volatility * 1.2,
"Signal line should be smoother than MACD line: signal_vol={:.6}, macd_vol={:.6}",
signal_volatility,
macd_volatility
);
}
#[test]
fn test_macd_incremental_update_performance() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Warm up with 50 bars
for i in 0..50 {
let price = 4500.0 + (i as f64 * 0.25);
extractor.extract_features(price, 100_000.0, timestamp);
}
// Benchmark MACD computation (incremental O(1) updates)
let mut total_duration = std::time::Duration::ZERO;
for i in 0..100 {
let price = 4500.0 + (50.0 + i as f64) * 0.25;
let start = Instant::now();
let _features = extractor.extract_features(price, 100_000.0, timestamp);
let duration = start.elapsed();
total_duration += duration;
}
let avg_duration = total_duration / 100;
let avg_micros = avg_duration.as_micros();
println!(
"Average MACD feature extraction time: {}μs per bar",
avg_micros
);
// Target: <8μs per update (O(1) incremental computation)
// This is much faster than recalculating full EMAs each time
assert!(
avg_micros < 50_000,
"MACD extraction too slow: {}μs (target: <50,000μs, O(1) expected: <8μs)",
avg_micros
);
}
#[test]
fn test_macd_normalization_bounds() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Test with extreme price movements
let prices = [
4000.0, 4500.0, 5000.0, 4200.0, 4800.0, // Extreme volatility
3800.0, 5200.0, 4100.0, 4900.0, 4400.0,
];
// Build up history
for _ in 0..50 {
extractor.extract_features(4500.0, 100_000.0, timestamp);
}
// Now test extreme movements
for (i, &price) in prices.iter().enumerate() {
let features = extractor.extract_features(price, 100_000.0, timestamp);
if features.len() >= 26 {
let macd = features[24];
let signal = features[25];
println!(
"Extreme price {}: Price={:.2}, MACD={:.6}, Signal={:.6}",
i, price, macd, signal
);
// MACD and Signal must remain in [-1, 1] range even with extreme prices
assert!(
(-1.0..=1.0).contains(&macd),
"MACD out of bounds with extreme price: {} (price={})",
macd,
price
);
assert!(
(-1.0..=1.0).contains(&signal),
"MACD Signal out of bounds with extreme price: {} (price={})",
signal,
price
);
}
}
}
#[test]
fn test_macd_histogram_implicit() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Build uptrend
for i in 0..50 {
let price = 4400.0 + (i as f64 * 2.0);
let features = extractor.extract_features(price, 100_000.0, timestamp);
if i >= 40 && features.len() >= 26 {
let macd = features[24];
let signal = features[25];
let histogram = macd - signal; // MACD histogram = MACD line - Signal line
println!(
"Bar {}: MACD={:.6}, Signal={:.6}, Histogram={:.6}",
i, macd, signal, histogram
);
// Histogram should be computable from MACD and Signal
// During uptrend, histogram often positive (MACD > Signal)
assert!(
histogram.is_finite(),
"MACD histogram should be finite: {}",
histogram
);
}
}
}
#[test]
fn test_macd_edge_case_zero_price() {
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Build normal prices
for i in 0..40 {
let price = 4500.0 + (i as f64 * 0.5);
extractor.extract_features(price, 100_000.0, timestamp);
}
// Test with zero price (edge case, should not crash)
let features = extractor.extract_features(0.0, 100_000.0, timestamp);
if features.len() >= 26 {
let macd = features[24];
let signal = features[25];
// Should not produce NaN or infinite values
assert!(
macd.is_finite(),
"MACD should be finite with zero price: {}",
macd
);
assert!(
signal.is_finite(),
"MACD Signal should be finite with zero price: {}",
signal
);
}
}
#[test]
fn test_macd_consistency_across_runs() {
// Create two extractors with same parameters
let mut extractor1 = MLFeatureExtractor::new(50);
let mut extractor2 = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Feed identical data to both
for i in 0..60 {
let price = 4500.0 + (i as f64 * 0.5);
let volume = 100_000.0;
let features1 = extractor1.extract_features(price, volume, timestamp);
let features2 = extractor2.extract_features(price, volume, timestamp);
if i >= 50 && features1.len() >= 22 && features2.len() >= 22 {
let macd1 = features1[20];
let signal1 = features1[21];
let macd2 = features2[20];
let signal2 = features2[21];
// MACD should be deterministic (identical across runs)
assert!(
(macd1 - macd2).abs() < 1e-10,
"MACD differs: {:.15} vs {:.15} at bar {}",
macd1,
macd2,
i
);
assert!(
(signal1 - signal2).abs() < 1e-10,
"MACD Signal differs: {:.15} vs {:.15} at bar {}",
signal1,
signal2,
i
);
}
}
}
#[test]
fn test_macd_ema_periods_correctness() {
// Validate MACD uses correct EMA periods (12, 26, 9)
let mut extractor = MLFeatureExtractor::new(50);
let timestamp = Utc::now();
// Build steady uptrend
for i in 0..60 {
let price = 4500.0 + (i as f64 * 1.0);
let features = extractor.extract_features(price, 100_000.0, timestamp);
if i >= 50 && features.len() >= 26 {
let macd = features[24];
let signal = features[25];
// During steady uptrend:
// - EMA12 rises faster than EMA26 (shorter period = more responsive)
// - MACD (EMA12 - EMA26) should be positive and increasing
// - Signal (EMA9 of MACD) should lag behind MACD
println!(
"Bar {}: Price={:.2}, MACD={:.6}, Signal={:.6}",
i,
4500.0 + (i as f64),
macd,
signal
);
assert!(
macd.is_finite() && signal.is_finite(),
"MACD values should be finite during steady uptrend"
);
}
}
}
// Helper function for volatility calculation
fn calculate_volatility(values: &[f64]) -> f64 {
if values.len() < 2 {
return 0.0;
}
let mean: f64 = values.iter().sum::<f64>() / values.len() as f64;
let variance: f64 =
values.iter().map(|&v| (v - mean).powi(2)).sum::<f64>() / values.len() as f64;
variance.sqrt()
}

2299
crates/common/tests/ml_strategy_integration_tests.rs
View File

File diff suppressed because it is too large Load Diff

112
crates/common/tests/shared_ml_strategy_integration_test.rs
View File

@@ -3,16 +3,71 @@
//! Validates that ONE SINGLE SYSTEM works for both trading and backtesting services.
//! NO duplication - both services use the same SharedMLStrategy instance.
use chrono::Utc;
use common::ml_strategy::{MLPrediction, SharedMLStrategy};
use ml::features::ProductionFeatureExtractorAdapter;
use anyhow::Result;
use chrono::{DateTime, Utc};
use common::ml_strategy::{
MLModelAdapter, MLPrediction, ProductionFeatureExtractor225, SharedMLStrategy,
};
use std::sync::Arc;
/// Mock adapter that produces deterministic predictions from features
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) {}
}
/// Mock extractor returning 225 features
struct MockExtractor;
impl ProductionFeatureExtractor225 for MockExtractor {
fn update(&mut self, _price: f64, _volume: f64, _timestamp: DateTime<Utc>) -> Result<()> {
Ok(())
}
fn extract_features(&mut self) -> Result<Vec<f64>> {
Ok(vec![0.1; 225])
}
}
fn make_strategy(threshold: f64) -> SharedMLStrategy {
SharedMLStrategy::new(
Box::new(MockExtractor),
vec![Box::new(MockAdapter::new("mock_v1"))],
threshold,
)
}
#[tokio::test]
async fn test_single_strategy_both_services() {
// Create ONE SINGLE SYSTEM with production feature extractor (225 features)
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = Arc::new(SharedMLStrategy::new_with_production_extractor(extractor, 0.3).unwrap());
let strategy = Arc::new(make_strategy(0.3));
// Simulate trading service using the strategy
let trading_strategy = Arc::clone(&strategy);
@@ -22,7 +77,6 @@ async fn test_single_strategy_both_services() {
.await
.expect("Trading service should get predictions");
// Calculate vote if predictions are available
if !predictions.is_empty() {
trading_strategy.calculate_ensemble_vote(&predictions);
}
@@ -38,7 +92,6 @@ async fn test_single_strategy_both_services() {
.await
.expect("Backtesting service should get predictions");
// Calculate vote if predictions are available
if !predictions.is_empty() {
backtesting_strategy.calculate_ensemble_vote(&predictions);
}
@@ -46,7 +99,6 @@ async fn test_single_strategy_both_services() {
predictions.len()
});
// Both services should succeed
let trading_count = trading_handle.await.expect("Trading task should complete");
let backtesting_count = backtesting_handle
.await
@@ -57,19 +109,14 @@ async fn test_single_strategy_both_services() {
backtesting_count > 0,
"Backtesting should generate predictions"
);
// Performance tracking would be populated after validate_predictions is called
// For now, just verify the strategy is functioning
}
#[tokio::test]
async fn test_concurrent_access_from_multiple_services() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = Arc::new(SharedMLStrategy::new_with_production_extractor(extractor, 0.5).unwrap());
let strategy = Arc::new(make_strategy(0.0));
let mut handles = Vec::new();
// Spawn 10 concurrent tasks (simulating trading + backtesting + monitoring services)
for i in 0..10 {
let strategy_clone = Arc::clone(&strategy);
let handle = tokio::spawn(async move {
@@ -84,7 +131,6 @@ async fn test_concurrent_access_from_multiple_services() {
handles.push(handle);
}
// Wait for all tasks
for handle in handles {
let predictions = handle.await.expect("Task should complete");
assert!(!predictions.is_empty(), "Should have predictions");
@@ -93,8 +139,7 @@ async fn test_concurrent_access_from_multiple_services() {
#[tokio::test]
async fn test_ensemble_vote_aggregation() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = SharedMLStrategy::new_with_production_extractor(extractor, 0.0).unwrap();
let strategy = make_strategy(0.0);
let predictions = vec![
MLPrediction {
@@ -128,7 +173,6 @@ async fn test_ensemble_vote_aggregation() {
let (vote, confidence) = result.unwrap_or_default();
// Weighted average should be between 0.6 and 0.8
assert!(
(0.6..=0.8).contains(&vote),
"Vote should be in expected range"
@@ -141,8 +185,7 @@ async fn test_ensemble_vote_aggregation() {
#[tokio::test]
async fn test_performance_tracking_across_services() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = Arc::new(SharedMLStrategy::new_with_production_extractor(extractor, 0.5).unwrap());
let strategy = Arc::new(make_strategy(0.0));
// Trading service generates signals
for _ in 0..5 {
@@ -151,7 +194,6 @@ async fn test_performance_tracking_across_services() {
.await
.expect("Should get predictions");
// Validate positive outcome
strategy.validate_predictions(&predictions, 0.05).await;
}
@@ -162,11 +204,9 @@ async fn test_performance_tracking_across_services() {
.await
.expect("Should get predictions");
// Validate negative outcome
strategy.validate_predictions(&predictions, -0.02).await;
}
// Check performance summary
let performance = strategy.get_performance_summary().await;
for (model_id, perf) in performance.iter() {
@@ -184,19 +224,14 @@ async fn test_performance_tracking_across_services() {
#[tokio::test]
async fn test_confidence_threshold_filtering() {
let high_extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let high_threshold_strategy = SharedMLStrategy::new_with_production_extractor(high_extractor, 0.95).unwrap();
let high_threshold_strategy = make_strategy(0.95);
let low_threshold_strategy = make_strategy(0.1);
let low_extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let low_threshold_strategy = SharedMLStrategy::new_with_production_extractor(low_extractor, 0.1).unwrap();
// High threshold should filter out most predictions
let high_predictions = high_threshold_strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
.expect("Should get predictions");
// Low threshold should keep most predictions
let low_predictions = low_threshold_strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
@@ -210,23 +245,20 @@ async fn test_confidence_threshold_filtering() {
#[tokio::test]
async fn test_feature_extraction_consistency() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = Arc::new(SharedMLStrategy::new_with_production_extractor(extractor, 0.5).unwrap());
let strategy = Arc::new(make_strategy(0.0));
// Generate predictions at two different times with same price/volume
let predictions1 = strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
.expect("Should get predictions");
tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
let predictions2 = strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
.expect("Should get predictions");
// Should have same number of models responding
assert_eq!(
predictions1.len(),
predictions2.len(),
@@ -236,8 +268,7 @@ async fn test_feature_extraction_consistency() {
#[tokio::test]
async fn test_empty_prediction_handling() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = SharedMLStrategy::new_with_production_extractor(extractor, 0.99).unwrap(); // Very high threshold
let strategy = make_strategy(0.99);
let predictions = vec![];
@@ -247,12 +278,11 @@ async fn test_empty_prediction_handling() {
#[tokio::test]
async fn test_model_performance_accuracy_tracking() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = SharedMLStrategy::new_with_production_extractor(extractor, 0.0).unwrap();
let strategy = make_strategy(0.0);
let prediction = MLPrediction {
model_id: "test_model".to_string(),
prediction_value: 0.7, // Predicts positive
prediction_value: 0.7,
confidence: 0.8,
features: vec![],
timestamp: Utc::now(),

142
crates/common/tests/test_sharedml_225_features.rs
View File

@@ -1,20 +1,57 @@
//! VALIDATION 1/8: Test that SharedMLStrategy extracts 225 features
//! Validation: Test that SharedMLStrategy correctly passes features from
//! ProductionFeatureExtractorAdapter to model adapters.
//!
//! This test validates that the Wave D implementation correctly extracts
//! all 225 features (201 Wave C + 24 Wave D regime detection features).
//! NOTE: ProductionFeatureExtractorAdapter currently produces 51 features
//! (43 base + 8 OFI placeholders). Full 225 features require MBP-10 data
//! and sequence-level feature engineering done at training time.
use anyhow::Result;
use chrono::Utc;
use common::ml_strategy::SharedMLStrategy;
use common::ml_strategy::{MLModelAdapter, MLPrediction, SharedMLStrategy};
use ml::features::ProductionFeatureExtractorAdapter;
#[tokio::test]
async fn test_sharedml_extracts_225_features() {
// Create SharedMLStrategy with production feature extractor (225 features)
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = SharedMLStrategy::new_with_production_extractor(extractor, 0.5).unwrap();
/// Mock adapter that captures features from predictions
struct FeatureCapturingAdapter {
id: String,
}
// Warm up the feature extractor with some historical data
// (needed to properly compute indicators like EMAs, RSI, etc.)
impl FeatureCapturingAdapter {
fn new(id: &str) -> Self {
Self {
id: id.to_string(),
}
}
}
impl MLModelAdapter for FeatureCapturingAdapter {
fn predict(&self, features: &[f64]) -> Result<MLPrediction> {
Ok(MLPrediction {
model_id: self.id.clone(),
prediction_value: 0.5,
confidence: 1.0, // Always pass threshold so we can inspect features
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) {}
}
#[tokio::test]
async fn test_production_extractor_feature_count() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = SharedMLStrategy::new(
extractor,
vec![Box::new(FeatureCapturingAdapter::new("capture_v1"))],
0.0,
);
// Warm up the feature extractor with historical data
for i in 0..100 {
let price = 100.0 + (i as f64 * 0.1);
let volume = 1000.0 + (i as f64 * 10.0);
@@ -23,13 +60,11 @@ async fn test_sharedml_extracts_225_features() {
.await;
}
// Extract features from the strategy
let predictions = strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
.expect("Should extract features successfully");
// Get features from the first prediction (all models use same features)
assert!(
!predictions.is_empty(),
"Should have at least one prediction"
@@ -37,47 +72,30 @@ async fn test_sharedml_extracts_225_features() {
let features = &predictions[0].features;
// VALIDATION 1: Verify feature count is exactly 225
// ProductionFeatureExtractorAdapter produces 51 features (43 base + 8 OFI placeholders)
assert_eq!(
features.len(),
225,
"SharedMLStrategy must extract exactly 225 features (201 Wave C + 24 Wave D), but got {}",
51,
"ProductionFeatureExtractorAdapter should produce 51 features, got {}",
features.len()
);
// VALIDATION 2: Verify no NaN values
// All features must be finite (no NaN, no Inf)
for (i, f) in features.iter().enumerate() {
assert!(
!f.is_nan(),
"Feature at index {} is NaN (value: {})",
i,
f
);
assert!(f.is_finite(), "Feature at index {} is not finite: {}", i, f);
}
// VALIDATION 3: Verify no Inf values
for (i, f) in features.iter().enumerate() {
assert!(
f.is_finite(),
"Feature at index {} is not finite (value: {}). All features must be finite numbers.",
i,
f
);
}
println!("✅ VALIDATION 1/8 PASSED");
println!(" - Feature count: {} (expected 225)", features.len());
println!(" - All features are finite");
println!(" - No NaN or Inf values detected");
}
#[tokio::test]
async fn test_feature_extraction_wave_d_breakdown() {
// Create SharedMLStrategy with production feature extractor (225 features)
async fn test_feature_extraction_all_finite() {
let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = SharedMLStrategy::new_with_production_extractor(extractor, 0.5).unwrap();
let strategy = SharedMLStrategy::new(
extractor,
vec![Box::new(FeatureCapturingAdapter::new("capture_v1"))],
0.0,
);
// Warm up the feature extractor
// Warm up
for i in 0..100 {
let price = 100.0 + (i as f64 * 0.1);
let volume = 1000.0 + (i as f64 * 10.0);
@@ -86,7 +104,6 @@ async fn test_feature_extraction_wave_d_breakdown() {
.await;
}
// Extract features
let predictions = strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
@@ -94,40 +111,15 @@ async fn test_feature_extraction_wave_d_breakdown() {
let features = &predictions[0].features;
// Verify feature breakdown (expected from Wave D documentation):
// - Wave A: 26 features (indices 0-25)
// - Wave B: 10 features (indices 26-35) [alternative bar sampling]
// - Wave C: 165 features (indices 36-200) [advanced feature engineering]
// - Wave D: 24 features (indices 201-224) [regime detection]
// Total: 225 features
assert_eq!(
features.len(),
225,
"Expected 225 total features (26 Wave A + 10 Wave B + 165 Wave C + 24 Wave D)"
);
// Verify Wave D features (indices 201-224) are present
for i in 201..225 {
let feature_value = features.get(i);
// All 51 features should be finite
for i in 0..features.len() {
let value = features.get(i);
assert!(value.is_some(), "Feature at index {} is missing", i);
assert!(
feature_value.is_some(),
"Wave D feature at index {} is missing",
i
);
let value = feature_value.unwrap();
assert!(
value.is_finite(),
"Wave D feature at index {} is not finite: {}",
value.unwrap().is_finite(),
"Feature at index {} is not finite: {}",
i,
value
value.unwrap()
);
}
println!("✅ Wave D feature breakdown validated");
println!(" - Wave A features (0-25): present");
println!(" - Wave B features (26-35): present");
println!(" - Wave C features (36-200): present");
println!(" - Wave D features (201-224): present");
}

393
crates/common/tests/volume_indicators_integration_test.rs
View File

@@ -1,393 +0,0 @@
//! Integration tests for volume-based technical indicators (OBV, MFI, VWAP)
//!
//! This test suite validates the implementation of volume indicators added in Wave 19.1.3
use chrono::Utc;
use common::ml_strategy::MLFeatureExtractor;
#[test]
fn test_obv_accumulation_uptrend() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Simulate strong uptrend with increasing volume
for i in 0..20 {
let price = 100.0 + (i as f64 * 2.0);
let volume = 1000.0 + (i as f64 * 50.0);
let features = extractor.extract_features(price, volume, timestamp);
if i >= 1 {
// OBV is at index 10 (7 base + 3 oscillators)
let obv = features[10];
// OBV should be positive in sustained uptrend
assert!(
obv > 0.0 || i == 1,
"OBV should be positive in uptrend at iteration {}, got {}",
i,
obv
);
}
}
}
#[test]
fn test_obv_distribution_downtrend() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Simulate strong downtrend
for i in 0..20 {
let price = 140.0 - (i as f64 * 2.0);
let volume = 1000.0 + (i as f64 * 50.0);
let features = extractor.extract_features(price, volume, timestamp);
if i >= 1 {
let obv = features[10];
// OBV should be negative in sustained downtrend
assert!(
obv < 0.0 || i == 1,
"OBV should be negative in downtrend at iteration {}, got {}",
i,
obv
);
}
}
}
#[test]
fn test_obv_unchanged_on_flat_price() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Extract first feature to initialize
extractor.extract_features(100.0, 1000.0, timestamp);
// Same price, different volumes - OBV should remain unchanged
let features1 = extractor.extract_features(100.0, 1500.0, timestamp);
let features2 = extractor.extract_features(100.0, 2000.0, timestamp);
let features3 = extractor.extract_features(100.0, 500.0, timestamp);
let obv1 = features1[10];
let obv2 = features2[10];
let obv3 = features3[10];
// All OBV values should be equal when price is flat
assert_eq!(obv1, obv2, "OBV should not change when price is unchanged");
assert_eq!(obv2, obv3, "OBV should not change when price is unchanged");
}
#[test]
fn test_mfi_overbought_condition() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Generate 15+ bars for MFI calculation
// Strong sustained uptrend with high volume = overbought
for i in 0..16 {
let price = 100.0 + (i as f64 * 3.0);
let volume = 1000.0 + (i as f64 * 200.0);
extractor.extract_features(price, volume, timestamp);
}
// Final strong up move
let features = extractor.extract_features(148.0, 4000.0, timestamp);
// MFI is at index 11 (7 base + 3 oscillators + OBV)
let mfi = features[11];
// MFI should be strongly positive (overbought, normalized from high MFI value)
assert!(
mfi > 0.3,
"MFI should indicate overbought condition (positive), got {}",
mfi
);
}
#[test]
fn test_mfi_oversold_condition() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Generate 15+ bars for MFI calculation
// Strong sustained downtrend with high volume = oversold
for i in 0..16 {
let price = 148.0 - (i as f64 * 3.0);
let volume = 1000.0 + (i as f64 * 200.0);
extractor.extract_features(price, volume, timestamp);
}
// Final strong down move
let features = extractor.extract_features(100.0, 4000.0, timestamp);
let mfi = features[11];
// MFI should be strongly negative (oversold, normalized from low MFI value)
assert!(
mfi < -0.3,
"MFI should indicate oversold condition (negative), got {}",
mfi
);
}
#[test]
fn test_mfi_neutral_condition() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Generate mixed market with equal buying/selling pressure
for i in 0..15 {
let price = if i % 2 == 0 { 100.0 } else { 101.0 };
let volume = 1000.0;
extractor.extract_features(price, volume, timestamp);
}
let features = extractor.extract_features(100.5, 1000.0, timestamp);
let mfi = features[11];
// MFI should be near neutral (close to 0)
assert!(
mfi.abs() < 0.5,
"MFI should be near neutral with mixed signals, got {}",
mfi
);
}
#[test]
fn test_vwap_benchmark_oscillating_market() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Trade around a base price with varying volumes
let prices = [100.0, 102.0, 98.0, 101.0, 99.0, 100.0, 103.0, 97.0];
let volumes = [1000.0, 500.0, 1500.0, 800.0, 1200.0, 1000.0, 600.0, 1400.0];
for (price, volume) in prices.iter().zip(volumes.iter()) {
extractor.extract_features(*price, *volume, timestamp);
}
let features = extractor.extract_features(100.0, 1000.0, timestamp);
// VWAP is at index 12 (7 base + 3 oscillators + OBV + MFI)
let vwap_ratio = features[12];
// VWAP ratio should be near 0 when price oscillates around average
assert!(
vwap_ratio.abs() < 0.2,
"VWAP ratio should be near 0 for oscillating prices, got {}",
vwap_ratio
);
}
#[test]
fn test_vwap_below_current_price() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// High volume at low prices, then price rises with low volume
extractor.extract_features(100.0, 5000.0, timestamp);
extractor.extract_features(101.0, 4000.0, timestamp);
extractor.extract_features(102.0, 3000.0, timestamp);
// Price jumps up with low volume
let features = extractor.extract_features(110.0, 500.0, timestamp);
let vwap_ratio = features[12];
// Price > VWAP, so ratio should be positive (bullish)
assert!(
vwap_ratio > 0.0,
"VWAP ratio should be positive when price > VWAP, got {}",
vwap_ratio
);
}
#[test]
fn test_vwap_above_current_price() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// High volume at high prices, then price drops with low volume
extractor.extract_features(110.0, 5000.0, timestamp);
extractor.extract_features(109.0, 4000.0, timestamp);
extractor.extract_features(108.0, 3000.0, timestamp);
// Price drops with low volume
let features = extractor.extract_features(100.0, 500.0, timestamp);
let vwap_ratio = features[12];
// Price < VWAP, so ratio should be negative (bearish)
assert!(
vwap_ratio < 0.0,
"VWAP ratio should be negative when price < VWAP, got {}",
vwap_ratio
);
}
#[test]
fn test_all_volume_indicators_normalized() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Generate diverse market conditions to test normalization
for i in 0..20 {
let price = 100.0 + ((i as f64 * 5.0).sin() * 20.0); // Volatile sine wave
let volume = 500.0 + (i as f64 * 100.0); // Increasing volume
extractor.extract_features(price, volume, timestamp);
}
let features = extractor.extract_features(105.0, 2500.0, timestamp);
let obv = features[10];
let mfi = features[11];
let vwap = features[12];
// All volume indicators should be in [-1, 1] range
assert!(
(-1.0..=1.0).contains(&obv),
"OBV should be normalized to [-1, 1], got {}",
obv
);
assert!(
(-1.0..=1.0).contains(&mfi),
"MFI should be normalized to [-1, 1], got {}",
mfi
);
assert!(
(-1.0..=1.0).contains(&vwap),
"VWAP should be normalized to [-1, 1], got {}",
vwap
);
}
#[test]
fn test_volume_indicators_with_extreme_values() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Test with extreme volume spikes and price movements
for i in 0..15 {
let price = if i == 10 { 150.0 } else { 100.0 }; // Price spike
let volume = if i == 10 { 50000.0 } else { 1000.0 }; // Volume spike
extractor.extract_features(price, volume, timestamp);
}
let features = extractor.extract_features(102.0, 1200.0, timestamp);
let obv = features[10];
let mfi = features[11];
let vwap = features[12];
// Even with extreme values, indicators should remain normalized
assert!(
(-1.0..=1.0).contains(&obv),
"OBV should handle extreme values, got {}",
obv
);
assert!(
(-1.0..=1.0).contains(&mfi),
"MFI should handle extreme values, got {}",
mfi
);
assert!(
(-1.0..=1.0).contains(&vwap),
"VWAP should handle extreme values, got {}",
vwap
);
}
#[test]
fn test_volume_indicators_insufficient_data() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Test with minimal data points
let features1 = extractor.extract_features(100.0, 1000.0, timestamp);
let features2 = extractor.extract_features(101.0, 1100.0, timestamp);
// OBV should work with 2 data points
assert_eq!(features1[10], 0.0, "OBV should be 0 for first data point");
// MFI should default to 0 with insufficient data (needs 15 points)
assert_eq!(features1[11], 0.0, "MFI should be 0 with insufficient data");
assert_eq!(features2[11], 0.0, "MFI should be 0 with insufficient data");
// VWAP should work with any amount of data
assert!(
features1[12] >= -1.0 && features1[12] <= 1.0,
"VWAP should be calculated even with minimal data"
);
}
#[test]
fn test_feature_vector_includes_volume_indicators() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Generate sufficient data for all indicators
for i in 0..30 {
let price = 100.0 + (i as f64 * 0.5);
let volume = 1000.0 + (i as f64 * 10.0);
extractor.extract_features(price, volume, timestamp);
}
let features = extractor.extract_features(115.0, 1300.0, timestamp);
// Total features: 30 (Wave A + Wave C)
// Wave A: 26 features (7 base + 3 oscillators + 3 volume + 5 EMA + 1 ADX + 1 BB + 2 Stoch + 1 CCI + 1 RSI + 2 MACD)
// Wave C: 4 features (OBV Momentum, Volume Oscillator, A/D Line, EMA Ratio)
assert_eq!(
features.len(),
30,
"Feature vector should include all 30 features (Wave A + Wave C)"
);
// Verify volume indicators are at correct indices
let obv = features[10];
let mfi = features[11];
let vwap = features[12];
assert!(obv.abs() <= 1.0, "OBV at index 10");
assert!(mfi.abs() <= 1.0, "MFI at index 11");
assert!(vwap.abs() <= 1.0, "VWAP at index 12");
}
#[test]
fn test_volume_indicators_provide_unique_signals() {
let mut extractor = MLFeatureExtractor::new(30);
let timestamp = Utc::now();
// Create scenario where volume indicators should diverge
// Phase 1: High volume accumulation at low prices
for i in 0..10 {
let price = 100.0 - (i as f64 * 0.5);
let volume = 1000.0 + (i as f64 * 300.0); // Increasing volume
extractor.extract_features(price, volume, timestamp);
}
// Phase 2: Price recovery with moderate volume
for i in 0..10 {
let price = 95.0 + (i as f64 * 1.0);
let volume = 1500.0; // Consistent moderate volume
extractor.extract_features(price, volume, timestamp);
}
let features = extractor.extract_features(105.0, 1600.0, timestamp);
let obv = features[10];
let mfi = features[11];
let vwap = features[12];
// All three indicators should provide different perspectives
// OBV: Should reflect volume accumulation during downturn + recovery
// MFI: Should show recent buying pressure (14-period window)
// VWAP: Should show price relative to volume-weighted average
// Verify they're not all the same (they provide unique information)
let indicators_equal = (obv - mfi).abs() < 0.01 && (mfi - vwap).abs() < 0.01;
assert!(
!indicators_equal,
"Volume indicators should provide different signals: OBV={}, MFI={}, VWAP={}",
obv, mfi, vwap
);
}

320
crates/common/tests/volume_indicators_test.rs
View File

@@ -1,320 +0,0 @@
//! Volume-based technical indicators validation tests
//!
//! Tests for OBV (On-Balance Volume), MFI (Money Flow Index), and VWAP
//! (Volume-Weighted Average Price) implementation in ML feature extraction.
use chrono::Utc;
use common::ml_strategy::MLFeatureExtractor;
#[test]
fn test_obv_accumulation_on_uptrend() {
let mut extractor = MLFeatureExtractor::new(20);
// Simulate uptrend with increasing prices and volume
let prices = [100.0, 101.0, 102.0, 103.0, 104.0];
let volumes = [1000.0, 1100.0, 1200.0, 1300.0, 1400.0];
let mut features_list = Vec::new();
for (price, volume) in prices.iter().zip(volumes.iter()) {
let features = extractor.extract_features(*price, *volume, Utc::now());
features_list.push(features);
}
// OBV should be increasing (positive accumulation)
// Feature index for OBV is 10 in Wave A feature set
let obv_feature_idx = 10;
// First data point has no previous price, so OBV should be 0
assert_eq!(features_list[0][obv_feature_idx], 0.0);
// Subsequent OBV values should be positive and increasing
for i in 1..features_list.len() {
let obv = features_list[i][obv_feature_idx];
assert!(
obv > 0.0,
"OBV should be positive in uptrend at index {}",
i
);
if i > 1 {
// Each OBV should be greater than or equal to previous (accumulation)
assert!(
obv >= features_list[i - 1][obv_feature_idx],
"OBV should increase in uptrend: {} < {}",
obv,
features_list[i - 1][obv_feature_idx]
);
}
}
}
#[test]
fn test_obv_distribution_on_downtrend() {
let mut extractor = MLFeatureExtractor::new(20);
// Simulate downtrend with decreasing prices
let prices = [104.0, 103.0, 102.0, 101.0, 100.0];
let volumes = [1000.0, 1100.0, 1200.0, 1300.0, 1400.0];
let mut features_list = Vec::new();
for (price, volume) in prices.iter().zip(volumes.iter()) {
let features = extractor.extract_features(*price, *volume, Utc::now());
features_list.push(features);
}
// OBV is at index 10 in Wave A feature set
let obv_feature_idx = 10;
// OBV should be decreasing (negative accumulation/distribution)
for i in 1..features_list.len() {
let obv = features_list[i][obv_feature_idx];
assert!(
obv < 0.0,
"OBV should be negative in downtrend at index {}",
i
);
if i > 1 {
// Each OBV should be less than or equal to previous (distribution)
assert!(
obv <= features_list[i - 1][obv_feature_idx],
"OBV should decrease in downtrend"
);
}
}
}
#[test]
fn test_mfi_overbought_signal() {
let mut extractor = MLFeatureExtractor::new(20);
// Generate 15 bars (need 15 for MFI 14-period calculation)
// Strong uptrend with high volume = overbought condition
for i in 0..15 {
let price = 100.0 + (i as f64 * 2.0); // Strong uptrend
let volume = 1000.0 + (i as f64 * 100.0); // Increasing volume
extractor.extract_features(price, volume, Utc::now());
}
// Last feature extraction should have MFI calculated
let features = extractor.extract_features(130.0, 2500.0, Utc::now());
// MFI is at index 11 in Wave A feature set
let mfi_feature_idx = 11;
let mfi_normalized = features[mfi_feature_idx];
// MFI normalized from [0, 100] to [-1, 1] via ((mfi/50) - 1).tanh()
// High MFI (>70 = overbought) should map to positive normalized value
// MFI of 100 -> (100/50 - 1).tanh() = 1.0.tanh() = 0.76
assert!(
mfi_normalized > 0.5,
"MFI should indicate overbought condition (positive normalized value): {}",
mfi_normalized
);
}
#[test]
fn test_mfi_oversold_signal() {
let mut extractor = MLFeatureExtractor::new(20);
// Generate 15 bars with strong downtrend = oversold condition
for i in 0..15 {
let price = 130.0 - (i as f64 * 2.0); // Strong downtrend
let volume = 1000.0 + (i as f64 * 100.0); // Increasing volume on decline
extractor.extract_features(price, volume, Utc::now());
}
// Last feature extraction
let features = extractor.extract_features(100.0, 2500.0, Utc::now());
// MFI is at index 11 in Wave A feature set
let mfi_feature_idx = 11;
let mfi_normalized = features[mfi_feature_idx];
// MFI normalized from [0, 100] to [-1, 1]
// Low MFI (<30 = oversold) should map to negative normalized value
// MFI of 0 -> (0/50 - 1).tanh() = -1.0.tanh() = -0.76
assert!(
mfi_normalized < -0.3,
"MFI should indicate oversold condition (negative normalized value): {}",
mfi_normalized
);
}
#[test]
fn test_vwap_price_benchmark() {
let mut extractor = MLFeatureExtractor::new(20);
// Trade at consistent price with varying volume
let prices = [100.0, 102.0, 98.0, 101.0, 99.0, 100.0];
let volumes = [1000.0, 500.0, 1500.0, 800.0, 1200.0, 1000.0];
let mut features_list = Vec::new();
for (price, volume) in prices.iter().zip(volumes.iter()) {
let features = extractor.extract_features(*price, *volume, Utc::now());
features_list.push(features);
}
// VWAP is at index 12 in Wave A feature set
let vwap_feature_idx = 12;
// Last VWAP should be close to base price (oscillating around it)
let vwap_ratio = features_list.last().unwrap()[vwap_feature_idx];
// VWAP ratio = (current_price - VWAP) / VWAP, normalized with tanh
// Since prices oscillate around 100, VWAP should be near 100, ratio near 0
assert!(
vwap_ratio.abs() < 0.3,
"VWAP ratio should be near 0 when price oscillates around average: {}",
vwap_ratio
);
}
#[test]
fn test_vwap_above_price_signal() {
let mut extractor = MLFeatureExtractor::new(20);
// Start with high volume at high prices, then drop price with low volume
// This will create VWAP above current price (bearish signal)
extractor.extract_features(110.0, 5000.0, Utc::now()); // High price, high volume
extractor.extract_features(109.0, 4000.0, Utc::now());
extractor.extract_features(108.0, 3000.0, Utc::now());
// Drop price with low volume
let features = extractor.extract_features(100.0, 500.0, Utc::now());
// VWAP is at index 12 in Wave A feature set
let vwap_feature_idx = 12;
let vwap_ratio = features[vwap_feature_idx];
// Price dropped below VWAP -> negative ratio
assert!(
vwap_ratio < 0.0,
"VWAP ratio should be negative when price drops below VWAP: {}",
vwap_ratio
);
}
#[test]
fn test_vwap_below_price_signal() {
let mut extractor = MLFeatureExtractor::new(20);
// Start with high volume at low prices, then raise price with low volume
// This will create VWAP below current price (bullish signal)
extractor.extract_features(100.0, 5000.0, Utc::now()); // Low price, high volume
extractor.extract_features(101.0, 4000.0, Utc::now());
extractor.extract_features(102.0, 3000.0, Utc::now());
// Raise price with low volume
let features = extractor.extract_features(110.0, 500.0, Utc::now());
// VWAP is at index 12 in Wave A feature set
let vwap_feature_idx = 12;
let vwap_ratio = features[vwap_feature_idx];
// Price rose above VWAP -> positive ratio
assert!(
vwap_ratio > 0.0,
"VWAP ratio should be positive when price rises above VWAP: {}",
vwap_ratio
);
}
#[test]
fn test_all_volume_indicators_normalized() {
let mut extractor = MLFeatureExtractor::new(20);
// Generate sufficient data for all indicators (15+ bars for MFI)
for i in 0..20 {
let price = 100.0 + (i as f64 * 0.5);
let volume = 1000.0 + (i as f64 * 50.0);
extractor.extract_features(price, volume, Utc::now());
}
// Final feature extraction
let features = extractor.extract_features(110.0, 2000.0, Utc::now());
// Check that OBV, MFI, VWAP are all normalized to [-1, 1]
// Volume indicators are at indices 10, 11, 12 in Wave A feature set
let obv_idx = 10;
let mfi_idx = 11;
let vwap_idx = 12;
assert!(
features[obv_idx] >= -1.0 && features[obv_idx] <= 1.0,
"OBV should be normalized to [-1, 1]: {}",
features[obv_idx]
);
assert!(
features[mfi_idx] >= -1.0 && features[mfi_idx] <= 1.0,
"MFI should be normalized to [-1, 1]: {}",
features[mfi_idx]
);
assert!(
features[vwap_idx] >= -1.0 && features[vwap_idx] <= 1.0,
"VWAP should be normalized to [-1, 1]: {}",
features[vwap_idx]
);
}
#[test]
fn test_feature_vector_length_increased() {
let mut extractor = MLFeatureExtractor::new(20);
// Generate sufficient data
for i in 0..20 {
let price = 100.0 + i as f64;
let volume = 1000.0 + (i as f64 * 10.0);
extractor.extract_features(price, volume, Utc::now());
}
let features = extractor.extract_features(120.0, 1200.0, Utc::now());
// Total features: 30 (Wave A + Wave C)
// Wave A: 26 features (7 base + 3 oscillators + 3 volume + 5 EMA + 1 ADX + 1 BB + 2 Stoch + 1 CCI + 1 RSI + 2 MACD)
// Wave C: 4 features (OBV Momentum, Volume Oscillator, A/D Line, EMA Ratio)
assert_eq!(
features.len(),
30,
"Feature vector should have 30 elements (Wave A + Wave C)"
);
}
#[test]
fn test_insufficient_data_graceful_handling() {
let mut extractor = MLFeatureExtractor::new(20);
// Only 1-2 data points (insufficient for MFI which needs 15)
let features1 = extractor.extract_features(100.0, 1000.0, Utc::now());
let features2 = extractor.extract_features(101.0, 1100.0, Utc::now());
// Volume indicators are at indices 10, 11, 12 in Wave A feature set
let obv_idx = 10;
let mfi_idx = 11;
let vwap_idx = 12;
// OBV should work with 2 data points
assert_eq!(
features1[obv_idx], 0.0,
"OBV should be 0 for first data point"
);
assert!(
features2[obv_idx] != 0.0 || features2[obv_idx] == 0.0,
"OBV should be calculated or 0 for second data point"
);
// MFI should default to 0 with insufficient data
assert_eq!(
features1[mfi_idx], 0.0,
"MFI should be 0 with insufficient data"
);
assert_eq!(
features2[mfi_idx], 0.0,
"MFI should be 0 with insufficient data"
);
// VWAP should work with any amount of data
assert!(
features1[vwap_idx] != 0.0 || features1[vwap_idx] == 0.0,
"VWAP should be calculated or 0"
);
}

2
crates/ml/src/ensemble/mod.rs
View File

@@ -24,8 +24,10 @@ pub mod adapters;
pub mod conviction_gates;
pub mod weight_optimizer;
pub mod gate_optimizer;
pub mod model_adapter;
// Re-export key types that are used across ensemble modules
pub use model_adapter::{EnsembleModelAdapter, build_production_strategy};
pub use ab_testing::{
ABGroup, ABMetricsTracker, ABTestConfig, ABTestResults, ABTestRouter, GroupMetrics,
Recommendation, StatisticalTestResult,

128
crates/ml/src/ensemble/model_adapter.rs Normal file
View File

@@ -0,0 +1,128 @@
//! Bridge between ml model registry and common::ml_strategy::MLModelAdapter trait.
//!
//! EnsembleModelAdapter wraps a model ID and implements MLModelAdapter so it can
//! be injected into SharedMLStrategy. When real checkpoint loading is wired, this
//! adapter will delegate to the loaded model for inference.
use anyhow::Result;
use chrono::Utc;
use common::ml_strategy::{MLModelAdapter, MLPrediction, SharedMLStrategy};
use crate::features::production_adapter::ProductionFeatureExtractorAdapter;
/// Adapter that will delegate predict() to a loaded model checkpoint.
///
/// Currently returns neutral predictions with zero confidence (filtered out
/// by the confidence threshold) until checkpoint loading is production-ready.
#[derive(Debug)]
pub struct EnsembleModelAdapter {
model_id: String,
}
impl EnsembleModelAdapter {
pub fn new<S: Into<String>>(model_id: S) -> 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 checkpoint loading when
// the model registry 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 model adapters for each
/// model in the 10-model ensemble.
///
/// Returns a strategy with:
/// - ProductionFeatureExtractorAdapter (51 features currently)
/// - 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());
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(extractor, models, min_confidence_threshold)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_ensemble_model_adapter_neutral_prediction() {
let adapter = EnsembleModelAdapter::new("dqn");
let features = vec![0.1; 51];
let prediction = adapter.predict(&features).unwrap();
assert_eq!(prediction.model_id, "dqn");
assert_eq!(prediction.prediction_value, 0.5);
assert_eq!(prediction.confidence, 0.0);
}
#[test]
fn test_build_production_strategy() {
let strategy = build_production_strategy(0.6);
assert_eq!(strategy.min_confidence_threshold(), 0.6);
}
#[tokio::test]
async fn test_production_strategy_filters_neutral_predictions() {
let strategy = build_production_strategy(0.5);
let predictions = strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
.unwrap();
// All 10 adapters return confidence=0.0, so threshold=0.5 filters them all
assert!(
predictions.is_empty(),
"Neutral predictions (confidence=0.0) should be filtered by threshold=0.5"
);
}
#[tokio::test]
async fn test_production_strategy_zero_threshold_keeps_all() {
let strategy = build_production_strategy(0.0);
let predictions = strategy
.get_ensemble_prediction(100.0, 1000.0, Utc::now())
.await
.unwrap();
// With threshold=0.0, all 10 neutral predictions pass through
assert_eq!(
predictions.len(),
10,
"Zero threshold should keep all 10 model predictions"
);
}
}

4
crates/ml/src/features/config.rs
View File

@@ -9,7 +9,7 @@
//! ## Architecture
//!
//! FeatureConfig provides a single source of truth for feature extraction
//! across both training (DbnSequenceLoader) and inference (MLFeatureExtractor).
//! across both training (DbnSequenceLoader) and inference (ProductionFeatureExtractorAdapter).
//! This eliminates the previous padding bug (256 features via 25x repetition).
//!
//! ## Usage
@@ -209,7 +209,7 @@ pub fn wave_d_features() -> Vec<Feature> {
/// Feature extraction configuration for Wave 19 progressive engineering
///
/// Tracks which feature groups are enabled across Wave A/B/C/D phases.
/// Used by both training (DbnSequenceLoader) and inference (MLFeatureExtractor).
/// Used by both training (DbnSequenceLoader) and inference (ProductionFeatureExtractorAdapter).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct FeatureConfig {
/// Feature engineering phase (Wave A/B/C/D)

2
crates/ml/src/features/production_adapter.rs
View File

@@ -22,7 +22,7 @@ use super::extraction::{FeatureExtractor, OHLCVBar};
/// use common::ml_strategy::SharedMLStrategy;
///
/// let extractor = Box::new(ProductionFeatureExtractorAdapter::new());
/// let strategy = SharedMLStrategy::new_with_production_extractor(extractor, 0.7);
/// let strategy = SharedMLStrategy::new(extractor, vec![], 0.7);
/// ```
#[derive(Debug)]
pub struct ProductionFeatureExtractorAdapter {

1
crates/ml/src/training.rs
View File

@@ -12,7 +12,6 @@
// Sub-modules for specialized training components
pub mod orchestrator;
pub mod push_metrics;
pub mod unified_data_loader;
pub mod unified_trainer; // NEW: Unified training trait for all models // NEW: Model-agnostic training orchestrator

213
crates/ml/src/training/push_metrics.rs
View File

@@ -1,213 +0,0 @@
//! Prometheus Pushgateway integration for training job metrics
//!
//! Training jobs are short-lived K8s Jobs. They push epoch-level metrics
//! to the Pushgateway so Prometheus can scrape them persistently.
use std::fmt::Write;
/// Pushgateway client for training metrics
#[derive(Debug)]
pub struct TrainingMetricsPusher {
pushgateway_url: String,
job_id: String,
model_name: String,
client: reqwest::Client,
}
impl TrainingMetricsPusher {
/// Create a new pusher. Falls back to in-cluster Pushgateway if `PUSHGATEWAY_URL` not set.
pub fn new(job_id: &str, model_name: &str) -> Self {
let pushgateway_url = std::env::var("PUSHGATEWAY_URL")
.unwrap_or_else(|_| "http://pushgateway.foxhunt.svc.cluster.local:9091".to_string());
Self {
pushgateway_url,
job_id: job_id.to_string(),
model_name: model_name.to_string(),
client: reqwest::Client::new(),
}
}
/// Push current training state to Pushgateway.
/// Non-fatal: logs errors but never panics.
pub async fn push(&self, state: &TrainingState) {
let mut body = String::new();
// Epoch progress
_ = writeln!(body, "# HELP foxhunt_training_current_epoch Current training epoch");
_ = writeln!(body, "# TYPE foxhunt_training_current_epoch gauge");
_ = writeln!(body, "foxhunt_training_current_epoch {}", state.epoch);
// Loss
_ = writeln!(
body,
"# HELP foxhunt_training_epoch_loss Training loss value"
);
_ = writeln!(body, "# TYPE foxhunt_training_epoch_loss gauge");
_ = writeln!(
body,
"foxhunt_training_epoch_loss{{split=\"train\"}} {}",
state.train_loss
);
if let Some(val_loss) = state.val_loss {
_ = writeln!(
body,
"foxhunt_training_epoch_loss{{split=\"val\"}} {}",
val_loss
);
}
// Accuracy
if let Some(accuracy) = state.accuracy {
_ = writeln!(
body,
"# HELP foxhunt_training_eval_accuracy Model evaluation accuracy"
);
_ = writeln!(body, "# TYPE foxhunt_training_eval_accuracy gauge");
_ = writeln!(body, "foxhunt_training_eval_accuracy {}", accuracy);
}
// Batches processed
_ = writeln!(
body,
"# HELP foxhunt_training_batches_processed Total batches processed"
);
_ = writeln!(body, "# TYPE foxhunt_training_batches_processed counter");
_ = writeln!(
body,
"foxhunt_training_batches_processed {}",
state.batches_processed
);
// Training speed
if state.batches_per_second > 0.0 {
_ = writeln!(
body,
"# HELP foxhunt_training_batches_per_second Training throughput"
);
_ = writeln!(body, "# TYPE foxhunt_training_batches_per_second gauge");
_ = writeln!(
body,
"foxhunt_training_batches_per_second {}",
state.batches_per_second
);
}
// Learning rate
_ = writeln!(
body,
"# HELP foxhunt_training_learning_rate Current learning rate"
);
_ = writeln!(body, "# TYPE foxhunt_training_learning_rate gauge");
_ = writeln!(
body,
"foxhunt_training_learning_rate {}",
state.learning_rate
);
// NaN / gradient events
if state.nan_count > 0 {
_ = writeln!(
body,
"# HELP foxhunt_training_nan_detected_total NaN gradient events"
);
_ = writeln!(body, "# TYPE foxhunt_training_nan_detected_total counter");
_ = writeln!(
body,
"foxhunt_training_nan_detected_total {}",
state.nan_count
);
}
if state.gradient_explosion_count > 0 {
_ = writeln!(
body,
"# HELP foxhunt_training_gradient_explosion_total Gradient clipping events"
);
_ = writeln!(
body,
"# TYPE foxhunt_training_gradient_explosion_total counter"
);
_ = writeln!(
body,
"foxhunt_training_gradient_explosion_total {}",
state.gradient_explosion_count
);
}
// Checkpoint saves
_ = writeln!(
body,
"# HELP foxhunt_training_checkpoint_saves_total Checkpoints saved"
);
_ = writeln!(
body,
"# TYPE foxhunt_training_checkpoint_saves_total counter"
);
_ = writeln!(
body,
"foxhunt_training_checkpoint_saves_total {}",
state.checkpoint_saves
);
let url = format!(
"{}/metrics/job/{}/model/{}",
self.pushgateway_url, self.job_id, self.model_name
);
match self.client.put(&url).body(body).send().await {
Ok(resp) if resp.status().is_success() => {}
Ok(resp) => {
eprintln!(
"Pushgateway returned {}: {}",
resp.status(),
resp.text().await.unwrap_or_default()
);
}
Err(e) => {
eprintln!("Failed to push metrics to Pushgateway: {e}");
}
}
}
/// Delete metrics for this job from Pushgateway (call on training completion).
pub async fn cleanup(&self) {
let url = format!(
"{}/metrics/job/{}/model/{}",
self.pushgateway_url, self.job_id, self.model_name
);
if let Err(e) = self.client.delete(&url).send().await {
eprintln!("Failed to delete metrics from Pushgateway: {e}");
}
}
}
/// Current training state to push
#[derive(Debug)]
pub struct TrainingState {
pub epoch: u64,
pub train_loss: f64,
pub val_loss: Option<f64>,
pub accuracy: Option<f64>,
pub batches_processed: u64,
pub batches_per_second: f64,
pub learning_rate: f64,
pub nan_count: u64,
pub gradient_explosion_count: u64,
pub checkpoint_saves: u64,
}
impl Default for TrainingState {
fn default() -> Self {
Self {
epoch: 0,
train_loss: 0.0,
val_loss: None,
accuracy: None,
batches_processed: 0,
batches_per_second: 0.0,
learning_rate: 0.001,
nan_count: 0,
gradient_explosion_count: 0,
checkpoint_saves: 0,
}
}
}

8
services/backtesting_service/src/ml_strategy_engine.rs
View File

@@ -20,8 +20,6 @@ use config::structures::BacktestingStrategyConfig;
// Import shared ML strategy (ONE SINGLE SYSTEM)
use common::ml_strategy::{MLPrediction as CommonMLPrediction, SharedMLStrategy};
// Import ProductionFeatureExtractorAdapter for 225-feature extraction
use ml::features::production_adapter::ProductionFeatureExtractorAdapter;
// Import UnifiedFeatureExtractor (256 features, production system)
use ml::features::unified::{FeatureExtractionConfig, UnifiedFeatureExtractor};
@@ -119,11 +117,7 @@ impl MLPoweredStrategy {
pub fn new(name: String, _lookback_periods: usize) -> Result<Self> {
// Use shared ML strategy (ONE SINGLE SYSTEM) with ProductionFeatureExtractorAdapter (225 features)
let min_confidence_threshold = 0.6;
let production_extractor = Box::new(ProductionFeatureExtractorAdapter::new());
let strategy = Arc::new(SharedMLStrategy::new_with_production_extractor(
production_extractor,
min_confidence_threshold,
)?);
let strategy = Arc::new(ml::ensemble::build_production_strategy(min_confidence_threshold));
// Initialize UnifiedFeatureExtractor (256 features)
let feature_config = FeatureExtractionConfig::default();

573
services/backtesting_service/tests/ml_strategy_backtest_test.rs
View File

@@ -1,573 +0,0 @@
//! ML Strategy Backtesting Tests - TDD Implementation
//!
//! Following strict TDD methodology (RED-GREEN-REFACTOR):
//! 1. RED: Write failing tests first
//! 2. GREEN: Minimal code to pass tests
//! 3. REFACTOR: Improve quality
//!
//! Tests ML ensemble predictions on historical market data.
use backtesting_service::dbn_data_source::DbnDataSource;
use backtesting_service::ml_strategy_engine::MLPoweredStrategy;
use backtesting_service::strategy_engine::{Portfolio, StrategyExecutor, TradeSide};
use common::ml_strategy::MLFeatureExtractor;
use num_traits::ToPrimitive;
use rust_decimal::Decimal;
use std::collections::HashMap;
mod helpers;
use helpers::{assert_chronological, assert_valid_ohlcv};
/// Helper: Get test data directory
fn get_test_data_dir() -> String {
let current_dir = std::env::current_dir().expect("INVARIANT: Current directory should be accessible");
let workspace_root = current_dir
.ancestors()
.find(|p| p.join("Cargo.toml").exists() && p.join("test_data").exists())
.expect("Could not find workspace root");
workspace_root
.join("test_data/real/databento")
.to_string_lossy()
.to_string()
}
/// Helper: Create DBN data source for test symbol
async fn create_test_data_source(symbol: &str) -> DbnDataSource {
let test_dir = get_test_data_dir();
let mut file_mapping = HashMap::new();
let file_path = match symbol {
"ES.FUT" => format!("{}/ES.FUT_ohlcv-1m_2024-01-02.dbn", test_dir),
"NQ.FUT" => format!("{}/NQ.FUT_ohlcv-1m_2024-01-02.dbn", test_dir),
"ZN.FUT" => format!("{}/ZN.FUT_ohlcv-1d_2024.dbn", test_dir),
_ => panic!("Unknown test symbol: {}", symbol),
};
file_mapping.insert(symbol.to_string(), file_path);
DbnDataSource::new(file_mapping)
.await
.expect("Failed to create DBN data source")
}
// =============================================================================
// TEST 1: ML Strategy Execution
// =============================================================================
#[tokio::test]
async fn test_ml_strategy_generates_predictions() {
// RED: Test ML strategy prediction generation
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
// Validate data quality
assert!(!bars.is_empty(), "No bars loaded");
assert_valid_ohlcv(&bars);
assert_chronological(&bars);
// Create ML strategy
let mut ml_strategy = MLPoweredStrategy::new("test_ml_strategy".to_string(), 20);
// Generate predictions for first 50 bars
let mut prediction_count = 0;
let portfolio = Portfolio::new(Decimal::from(100000));
let parameters: HashMap<String, String> = HashMap::new();
for bar in bars.iter().take(50) {
let predictions = ml_strategy.get_ensemble_prediction(bar).await;
if let Ok(preds) = predictions {
// Predictions may be empty if confidence threshold filters them out
// This is expected behavior - we just count non-empty predictions
if preds.is_empty() {
continue;
}
// Validate prediction structure when we have predictions
assert!(preds.len() >= 1, "Expected at least 1 model prediction");
// Validate prediction structure
for pred in &preds {
assert!(
pred.confidence >= 0.0 && pred.confidence <= 1.0,
"Confidence out of range: {}",
pred.confidence
);
assert!(
pred.prediction_value >= 0.0 && pred.prediction_value <= 1.0,
"Prediction value out of range: {}",
pred.prediction_value
);
assert!(pred.inference_latency_us > 0, "Invalid inference latency");
}
prediction_count += 1;
}
}
// Note: All predictions may be filtered by confidence threshold (0.6 default)
// This is valid behavior - the simple model may not have high confidence predictions
// We just verify the system works without errors
println!("✓ ML strategy executed on 50 bars: {} predictions passed confidence threshold ({}+ filtered)",
prediction_count, 50 - prediction_count);
// Verify system executed without errors (predictions may be 0 due to confidence filtering)
assert!(
prediction_count >= 0,
"System should execute without errors"
);
}
#[tokio::test]
async fn test_ml_strategy_ensemble_voting() {
// RED: Test ensemble voting mechanism
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
let mut ml_strategy = MLPoweredStrategy::new("test_ensemble".to_string(), 20);
// Get ensemble predictions for first bar with sufficient history
for bar in bars.iter().take(30) {
let predictions = ml_strategy.get_ensemble_prediction(bar).await.unwrap();
if predictions.len() >= 2 {
// Calculate ensemble vote
let ensemble_vote = ml_strategy.calculate_ensemble_vote(&predictions);
assert!(ensemble_vote.is_some(), "Ensemble vote should be computed");
let (ensemble_pred, ensemble_conf) = ensemble_vote.unwrap();
// Validate ensemble output
assert!(
ensemble_pred >= 0.0 && ensemble_pred <= 1.0,
"Ensemble prediction out of range: {}",
ensemble_pred
);
assert!(
ensemble_conf >= 0.0 && ensemble_conf <= 1.0,
"Ensemble confidence out of range: {}",
ensemble_conf
);
// Ensemble should be within bounds of individual predictions
let min_pred = predictions
.iter()
.map(|p| p.prediction_value)
.fold(f64::INFINITY, f64::min);
let max_pred = predictions
.iter()
.map(|p| p.prediction_value)
.fold(f64::NEG_INFINITY, f64::max);
assert!(
ensemble_pred >= min_pred && ensemble_pred <= max_pred,
"Ensemble prediction {} outside range [{}, {}]",
ensemble_pred,
min_pred,
max_pred
);
break; // Test first valid ensemble
}
}
}
// =============================================================================
// TEST 2: ML Backtest Execution
// =============================================================================
#[tokio::test]
async fn test_ml_backtest_generates_trades() {
// RED: Test ML backtest generates trades
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
let ml_strategy = MLPoweredStrategy::new("ml_backtest".to_string(), 20);
let portfolio = Portfolio::new(Decimal::from(100000));
let parameters = HashMap::new();
let mut total_signals = 0;
// Execute strategy on bars
for bar in bars.iter().take(200) {
let signals = ml_strategy.execute(bar, &portfolio, &parameters);
if let Ok(sigs) = signals {
total_signals += sigs.len();
// Validate signal structure
for sig in sigs {
assert!(
sig.strength >= Decimal::ZERO && sig.strength <= Decimal::ONE,
"Signal strength out of range"
);
assert!(sig.quantity > Decimal::ZERO, "Quantity must be positive");
assert!(!sig.reason.is_empty(), "Signal should have reason");
}
}
}
assert!(
total_signals > 0,
"ML strategy should generate at least some trade signals"
);
println!("✓ ML strategy generated {} trade signals", total_signals);
}
// =============================================================================
// TEST 3: Confidence Threshold Filtering
// =============================================================================
#[tokio::test]
async fn test_confidence_threshold_filtering() {
// RED: Test that confidence threshold filters low-confidence trades
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
// Test with low threshold (0.3) vs high threshold (0.8)
let thresholds = vec![0.3, 0.8];
let mut signal_counts = Vec::new();
for threshold in thresholds {
let ml_strategy = MLPoweredStrategy::new("ml_confidence_test".to_string(), 20);
let portfolio = Portfolio::new(Decimal::from(100000));
let mut parameters = HashMap::new();
parameters.insert("min_confidence".to_string(), threshold.to_string());
let mut signal_count = 0;
for bar in bars.iter().take(100) {
if let Ok(signals) = ml_strategy.execute(bar, &portfolio, &parameters) {
signal_count += signals.len();
}
}
signal_counts.push(signal_count);
}
// Higher threshold should generate fewer signals
assert!(
signal_counts[1] <= signal_counts[0],
"Higher confidence threshold ({}) should generate fewer signals. Got {} vs {}",
0.8,
signal_counts[1],
signal_counts[0]
);
println!(
"✓ Confidence filtering works: 0.3 threshold={} signals, 0.8 threshold={} signals",
signal_counts[0], signal_counts[1]
);
}
// =============================================================================
// TEST 4: Multi-Symbol ML Backtesting
// =============================================================================
#[tokio::test]
async fn test_ml_backtest_multi_symbol() {
// RED: Test ML backtesting across multiple symbols
let symbols = vec!["ES.FUT", "NQ.FUT"];
for symbol in symbols {
let data_source = create_test_data_source(symbol).await;
// Check if data file exists
if data_source.get_file_path(symbol).is_none() {
eprintln!("⚠️ Skipping {} - data file not found", symbol);
continue;
}
let bars_result = data_source.load_ohlcv_bars(symbol).await;
if bars_result.is_err() {
eprintln!("⚠️ Skipping {} - failed to load bars", symbol);
continue;
}
let bars = bars_result.unwrap();
if bars.is_empty() {
eprintln!("⚠️ Skipping {} - no bars loaded", symbol);
continue;
}
// Run ML backtest
let ml_strategy = MLPoweredStrategy::new(format!("ml_{}", symbol), 20);
let portfolio = Portfolio::new(Decimal::from(100000));
let parameters = HashMap::new();
let mut signal_count = 0;
for bar in bars.iter().take(50) {
if let Ok(signals) = ml_strategy.execute(bar, &portfolio, &parameters) {
signal_count += signals.len();
// Validate signals are for correct symbol
for sig in signals {
assert_eq!(sig.symbol, symbol, "Signal symbol mismatch");
}
}
}
println!(
"✓ ML backtest for {}: {} signals generated",
symbol, signal_count
);
}
}
// =============================================================================
// TEST 5: ML Performance Metrics
// =============================================================================
#[tokio::test]
async fn test_ml_backtest_performance_metrics() {
// RED: Test comprehensive performance metrics calculation
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
let ml_strategy = MLPoweredStrategy::new("ml_performance".to_string(), 20);
let portfolio = Portfolio::new(Decimal::from(100000));
let parameters = HashMap::new();
let mut equity_curve = vec![100000.0];
// Simulate simple backtest (buy signals only for testing)
for bar in bars.iter().take(100) {
if let Ok(signals) = ml_strategy.execute(bar, &portfolio, &parameters) {
for sig in signals {
if sig.side == TradeSide::Buy && portfolio.cash() > Decimal::ZERO {
// Simulate a small trade (simplified)
let trade_size = Decimal::from(100);
if trade_size < portfolio.cash() {
// Track equity (simplified - just price changes)
let current_equity = equity_curve.last().expect("INVARIANT: Collection should be non-empty");
// Prevent infinite/NaN Sharpe ratios - limit equity curve growth
if equity_curve.len() > 500 {
break;
}
let price_change = 0.01; // 1% change simulation
equity_curve.push(current_equity * (1.0 + price_change));
}
}
}
}
}
// Calculate basic performance metrics
if equity_curve.len() > 1 {
let initial_equity = equity_curve.first().expect("INVARIANT: Collection should be non-empty");
let final_equity = equity_curve.last().expect("INVARIANT: Collection should be non-empty");
let total_return = (final_equity - initial_equity) / initial_equity;
// Validate metrics exist
assert!(
equity_curve.len() >= 2,
"Equity curve should have multiple points"
);
// Calculate returns
let returns: Vec<f64> = equity_curve
.windows(2)
.map(|w| (w[1] - w[0]) / w[0])
.collect();
if !returns.is_empty() {
let mean_return = returns.iter().sum::<f64>() / returns.len() as f64;
let variance = returns
.iter()
.map(|r| (r - mean_return).powi(2))
.sum::<f64>()
/ returns.len() as f64;
let std_dev = variance.sqrt();
let sharpe_ratio = if std_dev > 1e-10 {
// Avoid division by very small numbers
mean_return / std_dev * (252.0_f64).sqrt() // Annualized
} else {
0.0
};
// Cap Sharpe ratio to realistic bounds for test stability
let sharpe_ratio = if sharpe_ratio.is_finite() {
sharpe_ratio.max(-5.0).min(10.0)
} else {
0.0
};
// Validate Sharpe ratio bounds
assert!(
sharpe_ratio >= -5.0 && sharpe_ratio <= 10.0,
"Sharpe ratio {} outside realistic bounds [-5, 10]",
sharpe_ratio
);
println!("✓ ML backtest metrics:");
println!(" Total return: {:.2}%", total_return * 100.0);
println!(" Sharpe ratio: {:.2}", sharpe_ratio);
println!(" Equity points: {}", equity_curve.len());
}
}
}
// =============================================================================
// TEST 6: ML Feature Extraction
// =============================================================================
#[tokio::test]
async fn test_ml_feature_extraction() {
// RED: Test feature extraction from market data
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
let mut feature_extractor = MLFeatureExtractor::new(20);
let mut feature_count = 0;
// Extract features from first 30 bars
for bar in bars.iter().take(30) {
let features = feature_extractor.extract_features(bar);
// Validate feature vector
assert!(!features.is_empty(), "Features should not be empty");
assert_eq!(features.len(), 7, "Expected 7 features (price momentum, MA, volatility, volume ratio, volume MA, hour, day)");
// Validate feature normalization (tanh: [-1, 1])
for (i, &f) in features.iter().enumerate() {
assert!(
f >= -1.0 && f <= 1.0,
"Feature {} = {} outside normalized range [-1, 1]",
i,
f
);
}
feature_count += 1;
}
assert_eq!(feature_count, 30, "Should extract features for all 30 bars");
println!(
"✓ Feature extraction successful: {} bars processed",
feature_count
);
}
// =============================================================================
// TEST 7: ML Model Performance Tracking
// =============================================================================
#[tokio::test]
async fn test_ml_model_performance_tracking() {
// RED: Test model performance tracking during backtest
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
let mut ml_strategy = MLPoweredStrategy::new("ml_tracking".to_string(), 20);
// Run predictions and track performance
let mut prev_price: Option<f64> = None;
let mut validation_count = 0;
for bar in bars.iter().take(50) {
let predictions = ml_strategy.get_ensemble_prediction(bar).await;
if let Ok(preds) = predictions {
// Skip empty predictions (filtered by confidence)
if preds.is_empty() {
prev_price = Some(bar.close.to_f64().unwrap_or(0.0));
continue;
}
// Validate predictions against actual returns
if let Some(prev) = prev_price {
let current_price = bar.close.to_f64().unwrap_or(0.0);
let actual_return = (current_price - prev) / prev;
ml_strategy
.validate_predictions(&preds, actual_return)
.await;
validation_count += 1;
}
prev_price = Some(bar.close.to_f64().unwrap_or(0.0));
}
}
// Get performance summary
let performance = ml_strategy.get_performance_summary();
// Performance tracking may be empty if no predictions passed confidence threshold
// This is valid behavior - just skip the detailed validation
if performance.is_empty() || validation_count == 0 {
println!("⚠️ No performance data (all predictions filtered by confidence threshold)");
return;
}
for (model_id, perf) in performance {
println!(
"✓ Model {}: {} predictions, {:.2}% accuracy, {:.3} avg confidence",
model_id, perf.total_predictions, perf.accuracy_percentage, perf.avg_confidence
);
// Validate performance metrics
assert!(perf.total_predictions > 0, "Model should have predictions");
assert!(
perf.accuracy_percentage >= 0.0 && perf.accuracy_percentage <= 100.0,
"Accuracy out of range"
);
assert!(
perf.avg_confidence >= 0.0 && perf.avg_confidence <= 1.0,
"Confidence out of range"
);
}
}
// =============================================================================
// TEST 8: ML vs Rule-Based Comparison (Placeholder)
// =============================================================================
#[tokio::test]
async fn test_ml_vs_rule_based_comparison() {
// RED: Compare ML strategy vs rule-based strategy
// This is a placeholder - full implementation requires running both strategies
let data_source = create_test_data_source("ES.FUT").await;
let bars = data_source.load_ohlcv_bars("ES.FUT").await.unwrap();
// ML strategy
let ml_strategy = MLPoweredStrategy::new("ml_comparison".to_string(), 20);
let portfolio = Portfolio::new(Decimal::from(100000));
let parameters = HashMap::new();
let mut ml_signal_count = 0;
for bar in bars.iter().take(100) {
if let Ok(signals) = ml_strategy.execute(bar, &portfolio, &parameters) {
ml_signal_count += signals.len();
}
}
// For now, just verify ML generates signals
// Full comparison would require implementing rule-based strategy backtest
assert!(
ml_signal_count >= 0,
"ML strategy should execute without errors"
);
println!(
"✓ ML strategy generated {} signals (rule-based comparison pending full implementation)",
ml_signal_count
);
}

14
services/trading_agent_service/src/assets.rs
View File

@@ -7,10 +7,8 @@
//! - Value: 20% weight
//! - Liquidity (quality): 10% weight
use common::ml_strategy::MLFeatureExtractor;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
/// Asset scoring result with multi-factor breakdown
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -141,18 +139,6 @@ impl AssetSelector {
}
}
/// 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,
}
}
/// Select top N assets by composite score
pub fn select_top_n(&self, mut assets: Vec<AssetScore>, n: usize) -> Vec<AssetScore> {
// Filter by thresholds

34
services/trading_agent_service/src/service.rs
View File

@@ -11,9 +11,7 @@ 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::assets::{AssetScore as InternalAssetScore, AssetSelector};
use crate::monitoring::TradingAgentMetrics;
use crate::orders::{OrderGenerator, PortfolioAllocation};
use crate::proto::trading_agent::*;
@@ -129,21 +127,21 @@ impl TradingAgentServiceImpl {
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);
// 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);
// ML score placeholder -- use liquidity_score from the instrument as a proxy
// (real ML inference would go here)

47
services/trading_service/tests/asset_selection_tests.rs
View File

@@ -4,12 +4,31 @@
//! and fallback behavior when ML is unavailable.
use anyhow::Result;
use chrono::Utc;
use common::ml_strategy::SharedMLStrategy;
use chrono::{DateTime, Utc};
use common::ml_strategy::{
MLModelAdapter, MLPrediction, ProductionFeatureExtractor225, SharedMLStrategy,
};
use sqlx::PgPool;
use std::sync::Arc;
use trading_service::assets::{AssetScore, AssetSelector, ScoringWeights};
/// Mock extractor for test SharedMLStrategy construction
struct TestMockExtractor;
impl ProductionFeatureExtractor225 for TestMockExtractor {
fn update(&mut self, _price: f64, _volume: f64, _timestamp: DateTime<Utc>) -> Result<()> {
Ok(())
}
fn extract_features(&mut self) -> Result<Vec<f64>> {
Ok(vec![0.1; 225])
}
}
/// Helper to create SharedMLStrategy for tests
fn make_test_strategy() -> SharedMLStrategy {
SharedMLStrategy::new(Box::new(TestMockExtractor), vec![], 0.6)
}
/// Helper to create test database pool
async fn setup_test_db() -> Result<PgPool> {
let database_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
@@ -110,7 +129,7 @@ async fn cleanup_test_data(pool: &PgPool, universe_id: &str) -> Result<()> {
#[tokio::test]
async fn test_asset_selector_creation() -> Result<()> {
let pool = setup_test_db().await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool, ml_strategy, None)?;
@@ -123,7 +142,7 @@ async fn test_asset_selector_creation() -> Result<()> {
#[tokio::test]
async fn test_asset_selector_custom_weights() -> Result<()> {
let pool = setup_test_db().await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let mut custom_weights = ScoringWeights {
ml_weight: 0.5,
@@ -144,7 +163,7 @@ async fn test_asset_selector_custom_weights() -> Result<()> {
#[tokio::test]
async fn test_select_assets_empty_universe() -> Result<()> {
let pool = setup_test_db().await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool, ml_strategy, None)?;
let universe_id = "test_empty_universe";
@@ -166,7 +185,7 @@ async fn test_select_assets_with_universe() -> Result<()> {
// Seed test data
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool, ml_strategy, None)?;
let assets = selector.select_assets(universe_id, 3).await?;
@@ -202,7 +221,7 @@ async fn test_asset_selection_persists_to_db() -> Result<()> {
// Seed test data
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool.clone(), ml_strategy, None)?;
let assets = selector.select_assets(universe_id, 5).await?;
@@ -233,7 +252,7 @@ async fn test_get_selected_assets() -> Result<()> {
// Seed test data
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool.clone(), ml_strategy, None)?;
// First, create a selection
@@ -268,7 +287,7 @@ async fn test_ml_integration_with_fallback() -> Result<()> {
// Seed test data
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool, ml_strategy, None)?;
// Select assets - should work even if ML predictions aren't perfect
@@ -304,7 +323,7 @@ async fn test_scoring_weights_affect_ranking() -> Result<()> {
liquidity_weight: 0.1,
};
let ml_strategy1 = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy1 = Arc::new(make_test_strategy());
let selector1 = AssetSelector::new(pool.clone(), ml_strategy1, Some(ml_heavy_weights))?;
let assets1 = selector1.select_assets(universe_id, 5).await?;
@@ -316,7 +335,7 @@ async fn test_scoring_weights_affect_ranking() -> Result<()> {
liquidity_weight: 0.1,
};
let ml_strategy2 = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy2 = Arc::new(make_test_strategy());
let selector2 = AssetSelector::new(pool.clone(), ml_strategy2, Some(momentum_heavy_weights))?;
let assets2 = selector2.select_assets(universe_id, 5).await?;
@@ -339,7 +358,7 @@ async fn test_ml_prediction_caching() -> Result<()> {
// Seed test data
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = Arc::new(AssetSelector::new(pool, ml_strategy, None)?);
// First selection - should query ML
@@ -376,7 +395,7 @@ async fn test_performance_target() -> Result<()> {
// Seed test data with more instruments
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = AssetSelector::new(pool, ml_strategy, None)?;
// Measure selection time
@@ -429,7 +448,7 @@ async fn test_concurrent_asset_selection() -> Result<()> {
// Seed test data
seed_test_universe(&pool, universe_id).await?;
let ml_strategy = Arc::new(SharedMLStrategy::new(20, 0.6));
let ml_strategy = Arc::new(make_test_strategy());
let selector = Arc::new(AssetSelector::new(pool, ml_strategy, None)?);
// Run multiple concurrent selections

51
services/trading_service/tests/ml_order_service_tests.rs
View File

@@ -434,10 +434,51 @@ async fn test_ml_performance_all_models() -> Result<()> {
#[tokio::test]
async fn test_shared_ml_strategy_integration() -> Result<()> {
use common::ml_strategy::SharedMLStrategy;
use common::ml_strategy::{
MLModelAdapter, MLPrediction, ProductionFeatureExtractor225, SharedMLStrategy,
};
// Arrange: Create strategy with 20-bar lookback, 60% confidence threshold
let strategy = SharedMLStrategy::new(20, 0.6);
struct MockExtractor;
impl ProductionFeatureExtractor225 for MockExtractor {
fn update(
&mut self,
_price: f64,
_volume: f64,
_timestamp: chrono::DateTime<Utc>,
) -> Result<()> {
Ok(())
}
fn extract_features(&mut self) -> Result<Vec<f64>> {
Ok(vec![0.1; 225])
}
}
struct MockAdapter;
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 pv = 1.0 / (1.0 + (-sum).exp());
Ok(MLPrediction {
model_id: "mock_v1".to_string(),
prediction_value: pv,
confidence: 0.5 + (pv - 0.5).abs() * 0.8,
features: features.to_vec(),
timestamp: Utc::now(),
inference_latency_us: 10,
})
}
fn model_id(&self) -> &str {
"mock_v1"
}
fn validate_prediction(&mut self, _prediction: &MLPrediction, _actual_outcome: bool) {}
}
// Arrange: Create strategy with mock adapter and 60% confidence threshold
let strategy = SharedMLStrategy::new(
Box::new(MockExtractor),
vec![Box::new(MockAdapter)],
0.6,
);
// Act: Get ensemble prediction with realistic market data
let predictions = strategy
@@ -448,10 +489,10 @@ async fn test_shared_ml_strategy_integration() -> Result<()> {
)
.await?;
// Assert: Should have at least 1 prediction (SimpleDQNAdapter is default fallback)
// Assert: Should have at least 1 prediction from injected model adapter
assert!(
!predictions.is_empty(),
"Should return at least 1 prediction from SimpleDQNAdapter"
"Should return at least 1 prediction from model adapter"
);
// Calculate ensemble vote