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

refactor(ml): extract observability, stress-testing, security into sub-crates

Browse Source 
- ml-observability (1.2K lines): alerts, dashboards, metrics modules.
  Depends on ml-core + common (ModelType). 4 tests passing.

- ml-stress-testing (1.3K lines): load_generator, market_simulator,
  performance_analyzer modules. Depends on ml-core + common + config.
  5 tests passing.

- ml-security (1.4K lines): anomaly_detector, prediction_validator
  modules. Depends on ml-core + ml-ensemble (EnsembleDecision,
  ModelVote, TradingAction). 17 tests passing.

Total: 18 sub-crates extracted from ml monolith.
Workspace: 0 errors, ml tests 876 + 26 in new sub-crates.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-08 14:07:07 +01:00
parent b7597a7543
commit 4676fe79e2
20 changed files with 1129 additions and 903 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
Cargo.lock generated
View File

@@ -6272,9 +6272,12 @@ dependencies = [
"ml-features",
"ml-hyperopt",
"ml-labeling",
"ml-observability",
"ml-ppo",
"ml-regime",
"ml-risk",
"ml-security",
"ml-stress-testing",
"ml-supervised",
"ml-universe",
"ml-validation",
@@ -6565,6 +6568,21 @@ dependencies = [
"uuid",
]
[[package]]
name = "ml-observability"
version = "1.0.0"
dependencies = [
"anyhow",
"common",
"ml-core",
"once_cell",
"prometheus",
"serde",
"serde_json",
"tokio",
"tracing",
]
[[package]]
name = "ml-ppo"
version = "1.0.0"
@@ -6623,6 +6641,35 @@ dependencies = [
"tracing",
]
[[package]]
name = "ml-security"
version = "1.0.0"
dependencies = [
"chrono",
"ml-core",
"ml-ensemble",
"serde",
"serde_json",
"tokio",
"tracing",
]
[[package]]
name = "ml-stress-testing"
version = "1.0.0"
dependencies = [
"anyhow",
"common",
"config",
"ml-core",
"rand 0.8.5",
"rand_distr 0.4.3",
"rust_decimal",
"serde",
"tokio",
"tracing",
]
[[package]]
name = "ml-supervised"
version = "1.0.0"

6
Cargo.toml
View File

@@ -123,9 +123,12 @@ members = [
"crates/ml-data-validation",
"crates/ml-validation",
"crates/ml-risk",
"crates/ml-security",
"crates/ml-backtesting",
"crates/ml-asset-selection",
"crates/ml-universe",
"crates/ml-observability",
"crates/ml-stress-testing",
"crates/data",
"crates/backtesting",
"crates/common",
@@ -413,9 +416,12 @@ ml-checkpoint = { path = "crates/ml-checkpoint" }
ml-data-validation = { path = "crates/ml-data-validation" }
ml-validation = { path = "crates/ml-validation" }
ml-risk = { path = "crates/ml-risk" }
ml-security = { path = "crates/ml-security" }
ml-backtesting = { path = "crates/ml-backtesting" }
ml-asset-selection = { path = "crates/ml-asset-selection" }
ml-universe = { path = "crates/ml-universe" }
ml-observability = { path = "crates/ml-observability" }
ml-stress-testing = { path = "crates/ml-stress-testing" }
common = { path = "crates/common" }
storage = { path = "crates/storage" }
market-data = { path = "crates/market-data" }

40
crates/ml-observability/Cargo.toml Normal file
View File

@@ -0,0 +1,40 @@
[package]
name = "ml-observability"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
authors.workspace = true
license.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
publish.workspace = true
keywords.workspace = true
categories.workspace = true
description = "Production observability, metrics, and alerting for Foxhunt ML systems"
[dependencies]
# Internal workspace crates
ml-core.workspace = true
common.workspace = true
# Core async
tokio.workspace = true
# Serialization and error handling
serde.workspace = true
serde_json.workspace = true
anyhow.workspace = true
tracing.workspace = true
# Metrics
prometheus.workspace = true
# Utilities
once_cell = "1.19"
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }
[lints]
workspace = true

6
crates/ml/src/observability/alerts.rs → crates/ml-observability/src/alerts.rs
View File

@@ -199,7 +199,7 @@ impl AlertManager {
async fn send_notification(&self, channel: &AlertChannel, alert: &Alert) -> Result<()> {
match channel {
AlertChannel::Console => {
println!("🚨 ALERT: {} - {}", alert.severity as u8, alert.message);
println!("ALERT: {} - {}", alert.severity as u8, alert.message);
},
AlertChannel::Slack { webhook_url: _ } => {
// Implement Slack webhook notification
@@ -253,7 +253,7 @@ pub fn create_hft_alert_rules() -> Vec<AlertRule> {
AlertRule {
name: "high_inference_latency".to_owned(),
metric_name: "ml_inference_latency_microseconds".to_owned(),
threshold: 100.0, // 100μs
threshold: 100.0, // 100us
comparison: AlertComparison::GreaterThan,
severity: AlertSeverity::Warning,
cooldown_minutes: 5,
@@ -263,7 +263,7 @@ pub fn create_hft_alert_rules() -> Vec<AlertRule> {
AlertRule {
name: "critical_inference_latency".to_owned(),
metric_name: "ml_inference_latency_microseconds".to_owned(),
threshold: 500.0, // 500μs
threshold: 500.0, // 500us
comparison: AlertComparison::GreaterThan,
severity: AlertSeverity::Critical,
cooldown_minutes: 1,

2
crates/ml/src/observability/dashboards.rs → crates/ml-observability/src/dashboards.rs
View File

@@ -96,7 +96,7 @@ pub fn create_hft_ml_dashboard() -> DashboardConfig {
widgets: vec![
DashboardWidget {
id: "inference_latency".to_owned(),
title: "Inference Latency (μs)".to_owned(),
title: "Inference Latency (us)".to_owned(),
widget_type: WidgetType::LineChart,
metrics: vec!["ml_inference_latency_microseconds".to_owned()],
time_range_minutes: 15,

17
crates/ml-observability/src/lib.rs Normal file
View File

@@ -0,0 +1,17 @@
//! Production observability and monitoring for ML systems
//!
//! This crate provides comprehensive monitoring, metrics collection, and alerting
//! for ML models in production HFT environments.
#![deny(clippy::unwrap_used, clippy::expect_used)]
#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]
// Workspace lints deny `panic` and `indexing_slicing` — allow where needed
#![allow(clippy::panic, clippy::indexing_slicing)]
// Re-export core types used by this crate's public API
pub use common::model_types::ModelType;
pub use ml_core::{MLError, MLResult, ModelPrediction};
pub mod alerts;
pub mod dashboards;
pub mod metrics;

8
crates/ml/src/observability/metrics.rs → crates/ml-observability/src/metrics.rs
View File

@@ -1,7 +1,7 @@
//! Production observability and monitoring for ML inference pipeline
//!
//! This module provides comprehensive metrics collection, monitoring, and alerting
//! for ML models in production HFT environment. Critical for maintaining sub-50μs
//! for ML models in production HFT environment. Critical for maintaining sub-50us
//! latency targets and ensuring model reliability.
use anyhow::{Context, Result};
@@ -129,8 +129,8 @@ impl MLMetricsCollector {
];
// Cardinality Optimization: Use asset_class instead of symbol
// Before: 5 model_types × 10 models × 10,000 symbols = 500,000 series
// After: 5 model_types × 10 models × 6 asset_classes = 300 series (99.94% reduction)
// Before: 5 model_types x 10 models x 10,000 symbols = 500,000 series
// After: 5 model_types x 10 models x 6 asset_classes = 300 series (99.94% reduction)
let inference_latency = HistogramVec::new(
HistogramOpts::new(
"ml_inference_latency_microseconds",
@@ -647,7 +647,7 @@ pub struct AlertThresholds {
impl Default for AlertThresholds {
fn default() -> Self {
Self {
max_latency_us: 100.0, // 100μs max latency
max_latency_us: 100.0, // 100us max latency
min_confidence: 0.7, // 70% minimum confidence
max_error_rate: 0.05, // 5% maximum error rate
min_health_score: 0.8, // 80% minimum health score

29
crates/ml-security/Cargo.toml Normal file
View File

@@ -0,0 +1,29 @@
[package]
name = "ml-security"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
authors.workspace = true
license.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
publish.workspace = true
keywords.workspace = true
categories.workspace = true
description = "ML security — prediction validation, anomaly detection, poisoning defense"
[dependencies]
ml-core = { path = "../ml-core", default-features = false }
ml-ensemble = { path = "../ml-ensemble", default-features = false }
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
chrono.workspace = true
tracing.workspace = true
tokio.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }
[lints]
workspace = true

10
crates/ml/src/security/anomaly_detector.rs → crates/ml-security/src/anomaly_detector.rs
View File

@@ -10,7 +10,7 @@ use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tracing::{debug, error, warn};
use crate::ensemble::EnsembleDecision;
use ml_ensemble::EnsembleDecision;
/// Ensemble anomaly detector with temporal pattern analysis
///
@@ -20,8 +20,8 @@ use crate::ensemble::EnsembleDecision;
/// 3. Model behavioral drift (individual model deviates from historical mean)
///
/// # Performance
/// - Detection: ~15μs per ensemble decision
/// - History update: ~5μs
/// - Detection: ~15us per ensemble decision
/// - History update: ~5us
pub struct EnsembleAnomalyDetector {
/// Historical ensemble signal distribution
signal_history: RwLock<VecDeque<f64>>,
@@ -159,7 +159,7 @@ impl EnsembleAnomalyDetector {
/// AnomalyReport with detected issues and severity
///
/// # Performance
/// ~15μs per call (includes history access and calculations)
/// ~15us per call (includes history access and calculations)
pub async fn detect_anomaly(&self, decision: &EnsembleDecision) -> AnomalyReport {
let mut anomalies = Vec::new();
@@ -424,7 +424,7 @@ pub struct DetectorStatistics {
#[cfg(test)]
mod tests {
use super::*;
use crate::ensemble::{ModelVote, TradingAction};
use ml_ensemble::{ModelVote, TradingAction};
fn create_test_decision(
signal: f64,

227
crates/ml-security/src/lib.rs Normal file
View File

@@ -0,0 +1,227 @@
#![deny(clippy::unwrap_used, clippy::expect_used)]
#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]
#![allow(dead_code)]
#![allow(missing_docs)]
#![allow(missing_debug_implementations)]
#![allow(unused_crate_dependencies)]
#![allow(clippy::float_arithmetic)]
#![allow(clippy::non_ascii_literal)]
#![allow(clippy::str_to_string)]
#![allow(clippy::partial_pub_fields)]
#![allow(clippy::multiple_inherent_impl)]
#![allow(clippy::same_name_method)]
#![allow(clippy::shadow_reuse)]
#![allow(clippy::shadow_unrelated)]
#![allow(clippy::shadow_same)]
#![allow(clippy::doc_markdown)]
#![allow(clippy::indexing_slicing)]
#![allow(clippy::missing_const_for_fn)]
#![allow(clippy::module_name_repetitions)]
#![allow(clippy::integer_division)]
#![allow(clippy::cognitive_complexity)]
#![allow(clippy::similar_names)]
#![allow(clippy::clone_on_ref_ptr)]
#![allow(clippy::too_many_lines)]
#![allow(clippy::as_conversions)]
#![allow(clippy::cast_precision_loss)]
#![allow(clippy::cast_possible_truncation)]
#![allow(clippy::default_numeric_fallback)]
#![allow(clippy::arithmetic_side_effects)]
#![allow(clippy::needless_range_loop)]
#![allow(clippy::into_iter_on_ref)]
#![allow(clippy::new_without_default)]
#![allow(clippy::manual_let_else)]
#![allow(clippy::unnecessary_wraps)]
#![allow(clippy::too_many_arguments)]
#![allow(clippy::must_use_candidate)]
#![allow(clippy::missing_errors_doc)]
#![allow(clippy::cast_sign_loss)]
#![allow(clippy::cast_possible_wrap)]
#![allow(clippy::cast_lossless)]
#![allow(clippy::unused_async)]
#![allow(clippy::match_same_arms)]
#![allow(clippy::unused_self)]
#![allow(clippy::map_err_ignore)]
#![allow(clippy::single_match_else)]
#![allow(clippy::wildcard_imports)]
#![allow(clippy::unnecessary_cast)]
//! ML Security Module
//!
//! Provides comprehensive security features for ML inference:
//! - Prediction validation and model poisoning detection
//! - Ensemble anomaly detection
//! - Security event logging
//!
//! # Architecture
//!
//! ```text
//! ┌──────────────────┐ ┌──────────────────┐
//! │ Prediction │ │ Ensemble │
//! │ Validator │──────│ Anomaly │
//! │ │ │ Detector │
//! └────────┬─────────┘ └────────┬─────────┘
//! │ │
//! └─────────┬───────────────┘
//! │
//! ▼
//! ┌──────────────────┐
//! │ Security Event │
//! │ Logger │
//! └──────────────────┘
//! ```
pub mod anomaly_detector;
pub mod prediction_validator;
pub use anomaly_detector::{
Anomaly, AnomalyDetectorConfig, AnomalyReport, AnomalySeverity, DetectorStatistics,
EnsembleAnomalyDetector,
};
pub use prediction_validator::{
PredictionStats, PredictionValidator, ValidatedPrediction, ValidationConfig, ValidationFlag,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
/// Security event types for ML system
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SecurityEventType {
// Checkpoint security
CheckpointSignatureFailure,
CheckpointSignatureMissing,
CheckpointTamperingDetected,
// Prediction validation
PredictionOutlierDetected,
PredictionOutOfBounds,
ExtremeRateExceeded,
// Ensemble anomalies
EnsembleSuddenShift,
CoordinatedAttackSuspected,
ModelBehavioralDrift,
// System events
AutomaticRollback,
ManualIntervention,
}
/// Security event for logging
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecurityEvent {
/// Event type
pub event_type: SecurityEventType,
/// Severity level
pub severity: SecuritySeverity,
/// Event timestamp
pub timestamp: DateTime<Utc>,
/// Model ID (if applicable)
pub model_id: Option<String>,
/// Checkpoint ID (if applicable)
pub checkpoint_id: Option<String>,
/// Prediction ID (if applicable)
pub prediction_id: Option<String>,
/// Human-readable description
pub description: String,
/// Additional metadata
pub metadata: serde_json::Value,
/// Action taken in response
pub action_taken: Option<String>,
}
/// Security severity levels
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum SecuritySeverity {
Low,
Medium,
High,
Critical,
}
impl SecurityEvent {
/// Create a new security event
pub fn new(
event_type: SecurityEventType,
severity: SecuritySeverity,
description: String,
) -> Self {
Self {
event_type,
severity,
timestamp: Utc::now(),
model_id: None,
checkpoint_id: None,
prediction_id: None,
description,
metadata: serde_json::Value::Null,
action_taken: None,
}
}
/// Set model ID
pub fn with_model_id(mut self, model_id: String) -> Self {
self.model_id = Some(model_id);
self
}
/// Set checkpoint ID
pub fn with_checkpoint_id(mut self, checkpoint_id: String) -> Self {
self.checkpoint_id = Some(checkpoint_id);
self
}
/// Set prediction ID
pub fn with_prediction_id(mut self, prediction_id: String) -> Self {
self.prediction_id = Some(prediction_id);
self
}
/// Set metadata
pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self {
self.metadata = metadata;
self
}
/// Set action taken
pub fn with_action(mut self, action: String) -> Self {
self.action_taken = Some(action);
self
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_security_event_builder() {
let event = SecurityEvent::new(
SecurityEventType::PredictionOutlierDetected,
SecuritySeverity::Medium,
"Test outlier detection".to_owned(),
)
.with_model_id("DQN".to_owned())
.with_action("Flagged for review".to_owned());
assert_eq!(event.severity, SecuritySeverity::Medium);
assert_eq!(event.model_id, Some("DQN".to_owned()));
assert_eq!(event.action_taken, Some("Flagged for review".to_owned()));
}
#[test]
fn test_severity_ordering() {
assert!(SecuritySeverity::Critical > SecuritySeverity::High);
assert!(SecuritySeverity::High > SecuritySeverity::Medium);
assert!(SecuritySeverity::Medium > SecuritySeverity::Low);
}
}

10
crates/ml/src/security/prediction_validator.rs → crates/ml-security/src/prediction_validator.rs
View File

@@ -10,19 +10,19 @@ use serde::{Deserialize, Serialize};
use tokio::sync::RwLock;
use tracing::{debug, warn};
use crate::MLError;
use ml_core::MLError;
/// Prediction validator with statistical bounds checking
///
/// Detects model poisoning through:
/// 1. Range validation (predictions must be in [-1.0, 1.0])
/// 2. Z-score outlier detection (>3σ from mean)
/// 2. Z-score outlier detection (>3sigma from mean)
/// 3. Confidence sanity checks
/// 4. Extreme prediction rate limiting
///
/// # Performance
/// - Validation: ~5μs per prediction
/// - Statistics update: ~2μs
/// - Validation: ~5us per prediction
/// - Statistics update: ~2us
pub struct PredictionValidator {
/// Historical prediction statistics (rolling window)
stats: RwLock<PredictionStatistics>,
@@ -255,7 +255,7 @@ impl PredictionValidator {
/// Returns error if prediction is invalid (out of bounds, extreme rate exceeded)
///
/// # Performance
/// ~5μs per call (includes statistics update)
/// ~5us per call (includes statistics update)
pub async fn validate(
&self,
prediction: f64,

33
crates/ml-stress-testing/Cargo.toml Normal file
View File

@@ -0,0 +1,33 @@
[package]
name = "ml-stress-testing"
version.workspace = true
edition.workspace = true
rust-version.workspace = true
authors.workspace = true
license.workspace = true
repository.workspace = true
homepage.workspace = true
documentation.workspace = true
publish.workspace = true
keywords.workspace = true
categories.workspace = true
description = "Stress testing framework for ML models under high-volume market data conditions"
[dependencies]
ml-core.workspace = true
common.workspace = true
config.workspace = true
serde = { workspace = true, features = ["derive"] }
anyhow.workspace = true
tracing.workspace = true
tokio.workspace = true
rust_decimal.workspace = true
rand.workspace = true
rand_distr.workspace = true
[dev-dependencies]
tokio = { workspace = true, features = ["test-util", "macros"] }
[lints]
workspace = true

701
crates/ml-stress-testing/src/lib.rs Normal file
View File

@@ -0,0 +1,701 @@
#![deny(clippy::unwrap_used, clippy::expect_used)]
#![cfg_attr(test, allow(clippy::unwrap_used, clippy::expect_used))]
#![allow(dead_code)]
#![allow(missing_docs)]
#![allow(missing_debug_implementations)]
#![allow(clippy::float_arithmetic)]
#![allow(clippy::doc_markdown)]
#![allow(clippy::indexing_slicing)]
#![allow(clippy::missing_const_for_fn)]
//! Stress testing framework for ML models under high-volume market data conditions
//!
//! This crate provides comprehensive stress testing capabilities to validate ML model
//! performance under realistic HFT market conditions with high-frequency data feeds.
// Re-export key types from ml-core that are used throughout this crate
pub use common::model_types::ModelType;
pub use ml_core::{Features, MLModel, ModelPrediction};
pub mod load_generator;
pub mod market_simulator;
pub mod performance_analyzer;
// Re-export key types that are needed by external users
pub use load_generator::{LoadGenerator, LoadProfile, TrafficPattern};
pub use market_simulator::{MarketCondition, MarketDataSimulator, SimulatorConfig};
pub use performance_analyzer::{LatencyStats, PerformanceAnalyzer, StressTestReport};
use common::types::Price;
use rust_decimal::Decimal;
use anyhow::Result;
use rust_decimal::prelude::ToPrimitive;
use serde::{Deserialize, Serialize};
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
/// Comprehensive stress test configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StressTestConfig {
/// Test duration in seconds
pub duration_seconds: u64,
/// Target requests per second
pub target_rps: u32,
/// Number of concurrent connections
pub concurrent_connections: u32,
/// Market data feed rate (updates per second)
pub market_data_rate: u32,
/// Test phases with different load patterns
pub test_phases: Vec<TestPhase>,
/// Models to test
pub models_to_test: Vec<String>,
/// Market conditions to simulate
pub market_conditions: Vec<MarketCondition>,
/// Performance requirements
pub requirements: PerformanceRequirements,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TestPhase {
pub name: String,
pub duration_seconds: u64,
pub load_multiplier: f64,
pub market_volatility: f64,
pub error_injection_rate: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceRequirements {
/// Maximum acceptable latency in microseconds
pub max_latency_us: u64,
/// 95th percentile latency requirement
pub p95_latency_us: u64,
/// 99th percentile latency requirement
pub p99_latency_us: u64,
/// Maximum acceptable error rate
pub max_error_rate: f64,
/// Minimum throughput (predictions per second)
pub min_throughput: u32,
/// Maximum memory usage in MB
pub max_memory_mb: u64,
/// Maximum CPU utilization percentage
pub max_cpu_percent: f64,
}
impl Default for PerformanceRequirements {
fn default() -> Self {
Self {
max_latency_us: 100, // 100us max latency
p95_latency_us: 50, // 50us 95th percentile
p99_latency_us: 80, // 80us 99th percentile
max_error_rate: 0.01, // 1% max error rate
min_throughput: 10000, // 10k predictions/sec
max_memory_mb: 1024, // 1GB max memory
max_cpu_percent: 80.0, // 80% max CPU
}
}
}
/// Main stress testing orchestrator
#[derive(Debug)]
pub struct StressTestOrchestrator {
config: StressTestConfig,
market_simulator: MarketDataSimulator,
load_generator: LoadGenerator,
performance_analyzer: PerformanceAnalyzer,
}
impl StressTestOrchestrator {
/// Create new stress test orchestrator with configuration-driven market simulation
pub fn new(config: StressTestConfig) -> Result<Self> {
// Use configuration-driven approach - load from config crate
let simulation_config = config::SimulationConfig::default();
// Use test fixtures for symbol configuration
let test_symbols = vec!["TEST_LARGE_1", "TEST_LARGE_2", "TEST_MID_1", "TEST_SMALL_1"];
let simulator_config = SimulatorConfig {
symbols: test_symbols.into_iter().map(|s| s.to_string()).collect(),
update_rate_hz: config.market_data_rate,
volatility: 0.02,
trend: 0.0,
simulation_config: Some(simulation_config),
};
let market_simulator = MarketDataSimulator::new(simulator_config)?;
let load_generator = LoadGenerator::new(config.target_rps, config.concurrent_connections)?;
let performance_analyzer = PerformanceAnalyzer::new();
Ok(Self {
config,
market_simulator,
load_generator,
performance_analyzer,
})
}
/// Create stress test orchestrator with custom simulation configuration
pub fn new_with_simulation_config(
config: StressTestConfig,
simulation_config: config::SimulationConfig,
) -> Result<Self> {
let symbols: Vec<String> = simulation_config
.initial_market_state
.symbols
.keys()
.cloned()
.collect();
let simulator_config = SimulatorConfig {
symbols,
update_rate_hz: config.market_data_rate,
volatility: simulation_config.parameters.base_volatility,
trend: simulation_config.parameters.trend,
simulation_config: Some(simulation_config),
};
let market_simulator = MarketDataSimulator::new(simulator_config)?;
let load_generator = LoadGenerator::new(config.target_rps, config.concurrent_connections)?;
let performance_analyzer = PerformanceAnalyzer::new();
Ok(Self {
config,
market_simulator,
load_generator,
performance_analyzer,
})
}
/// Create stress test orchestrator for testing with generated test symbols
pub fn new_for_testing(config: StressTestConfig) -> Result<Self> {
let simulation_config = config::SimulationConfig::default();
let test_symbols = MarketDataSimulator::generate_test_symbols(&simulation_config);
let simulator_config = SimulatorConfig {
symbols: test_symbols,
update_rate_hz: config.market_data_rate,
volatility: simulation_config.parameters.base_volatility,
trend: simulation_config.parameters.trend,
simulation_config: Some(simulation_config),
};
let market_simulator = MarketDataSimulator::new(simulator_config)?;
let load_generator = LoadGenerator::new(config.target_rps, config.concurrent_connections)?;
let performance_analyzer = PerformanceAnalyzer::new();
Ok(Self {
config,
market_simulator,
load_generator,
performance_analyzer,
})
}
/// Run comprehensive stress test
pub async fn run_stress_test(
&mut self,
models: Vec<std::sync::Arc<dyn MLModel>>,
) -> Result<StressTestReport> {
tracing::info!("Starting stress test with {} models", models.len());
// Create channels for communication
let (market_tx, mut market_rx) = mpsc::channel(10000);
let (prediction_tx, _prediction_rx) = mpsc::channel(10000);
// Start market data simulation
let simulator_handle = {
let mut simulator = self.market_simulator.clone();
tokio::spawn(async move { simulator.start_simulation(market_tx).await })
};
// Start performance monitoring
let analyzer = self.performance_analyzer.clone();
let monitor_handle = tokio::spawn(async move { analyzer.start_monitoring().await });
// Execute test phases
let mut phase_results = Vec::new();
let test_start = Instant::now();
// Clone the test phases to avoid borrowing conflicts
let test_phases = self.config.test_phases.clone();
for (phase_idx, phase) in test_phases.into_iter().enumerate() {
tracing::info!("Starting test phase {}: {}", phase_idx + 1, phase.name);
let phase_result = self
.run_test_phase(&phase, &models, &mut market_rx, &prediction_tx)
.await?;
phase_results.push(phase_result);
}
// Stop simulation and monitoring
simulator_handle.abort();
monitor_handle.abort();
// Generate comprehensive report
let total_duration = test_start.elapsed();
let report = self
.generate_stress_test_report(phase_results, total_duration)
.await?;
tracing::info!(
"Stress test completed in {:.2}s",
total_duration.as_secs_f64()
);
Ok(report)
}
/// Run individual test phase
async fn run_test_phase(
&mut self,
phase: &TestPhase,
models: &[std::sync::Arc<dyn MLModel>],
market_rx: &mut mpsc::Receiver<MarketDataUpdate>,
prediction_tx: &mpsc::Sender<PredictionResult>,
) -> Result<PhaseResult> {
let phase_start = Instant::now();
let phase_duration = Duration::from_secs(phase.duration_seconds);
let mut phase_stats = PhaseStats::new();
// Adjust load generator for this phase
self.load_generator
.set_load_multiplier(phase.load_multiplier);
while phase_start.elapsed() < phase_duration {
// Process market data updates
while let Ok(market_update) = market_rx.try_recv() {
// Convert market data to features
let features = self.convert_market_data_to_features(&market_update)?;
// Run predictions on all models
for model in models {
let model_start = Instant::now();
match model.predict(&features).await {
Ok(prediction) => {
let latency_us = model_start.elapsed().as_micros() as u64;
phase_stats.record_successful_prediction(latency_us);
let result = PredictionResult {
model_name: model.name().to_string(),
model_type: model.model_type(),
prediction,
latency_us,
timestamp: std::time::SystemTime::now(),
success: true,
error_message: None,
};
drop(prediction_tx.send(result).await);
},
Err(e) => {
let latency_us = model_start.elapsed().as_micros() as u64;
phase_stats.record_failed_prediction(latency_us);
let result = PredictionResult {
model_name: model.name().to_string(),
model_type: model.model_type(),
prediction: ModelPrediction::new(
model.name().to_string(),
0.0,
0.0,
),
latency_us,
timestamp: std::time::SystemTime::now(),
success: false,
error_message: Some(e.to_string()),
};
drop(prediction_tx.send(result).await);
},
}
}
}
// Small delay to prevent busy waiting
tokio::time::sleep(Duration::from_micros(100)).await;
}
Ok(PhaseResult {
phase_name: phase.name.clone(),
duration: phase_start.elapsed(),
stats: phase_stats,
})
}
/// Convert market data to ML features
fn convert_market_data_to_features(&self, market_data: &MarketDataUpdate) -> Result<Features> {
let values = vec![
market_data.price.to_f64(),
market_data.volume.to_f64().unwrap_or(0.0),
market_data.bid.to_f64(),
market_data.ask.to_f64(),
market_data.spread().to_f64(),
market_data.mid_price().to_f64(),
];
let names = vec![
"price".to_owned(),
"volume".to_owned(),
"bid".to_owned(),
"ask".to_owned(),
"spread".to_owned(),
"mid_price".to_owned(),
];
Ok(Features::new(values, names).with_symbol(market_data.symbol.clone()))
}
/// Generate comprehensive stress test report
async fn generate_stress_test_report(
&self,
phase_results: Vec<PhaseResult>,
total_duration: Duration,
) -> Result<StressTestReport> {
let mut total_predictions = 0;
let mut total_errors = 0;
let mut all_latencies = Vec::new();
for phase in &phase_results {
total_predictions +=
phase.stats.successful_predictions + phase.stats.failed_predictions;
total_errors += phase.stats.failed_predictions;
all_latencies.extend(&phase.stats.latencies);
}
let error_rate = if total_predictions > 0 {
total_errors as f64 / total_predictions as f64
} else {
0.0
};
let latency_stats = self.calculate_latency_statistics(&all_latencies);
// Check if requirements are met
let requirements_met = self.check_requirements(&latency_stats, error_rate);
let recommendations = self.generate_recommendations(&latency_stats, error_rate);
Ok(StressTestReport {
config: self.config.clone(),
total_duration,
phase_results,
total_predictions: total_predictions as u64,
total_errors: total_errors as u64,
error_rate,
latency_stats: latency_stats.clone(),
requirements_met,
throughput_achieved: total_predictions as f64 / total_duration.as_secs_f64(),
recommendations,
})
}
fn calculate_latency_statistics(&self, latencies: &[u64]) -> LatencyStats {
if latencies.is_empty() {
return LatencyStats::default();
}
let mut sorted_latencies = latencies.to_vec();
sorted_latencies.sort_unstable();
let len = sorted_latencies.len();
let mean = sorted_latencies.iter().sum::<u64>() as f64 / len as f64;
let min = sorted_latencies[0];
let max = sorted_latencies[len - 1];
let p50 = sorted_latencies[len * 50 / 100];
let p95 = sorted_latencies[len * 95 / 100];
let p99 = sorted_latencies[len * 99 / 100];
LatencyStats {
mean,
min,
max,
p50,
p95,
p99,
count: len as u64,
}
}
fn check_requirements(
&self,
latency_stats: &LatencyStats,
error_rate: f64,
) -> RequirementsCheck {
RequirementsCheck {
latency_ok: latency_stats.max <= self.config.requirements.max_latency_us,
p95_latency_ok: latency_stats.p95 <= self.config.requirements.p95_latency_us,
p99_latency_ok: latency_stats.p99 <= self.config.requirements.p99_latency_us,
error_rate_ok: error_rate <= self.config.requirements.max_error_rate,
overall_pass: latency_stats.max <= self.config.requirements.max_latency_us
&& latency_stats.p95 <= self.config.requirements.p95_latency_us
&& latency_stats.p99 <= self.config.requirements.p99_latency_us
&& error_rate <= self.config.requirements.max_error_rate,
}
}
fn generate_recommendations(
&self,
latency_stats: &LatencyStats,
error_rate: f64,
) -> Vec<String> {
let mut recommendations = Vec::new();
if latency_stats.p99 > self.config.requirements.p99_latency_us {
recommendations.push(format!(
"P99 latency ({}us) exceeds requirement ({}us). Consider model optimization or hardware upgrades.",
latency_stats.p99, self.config.requirements.p99_latency_us
));
}
if error_rate > self.config.requirements.max_error_rate {
recommendations.push(format!(
"Error rate ({:.2}%) exceeds requirement ({:.2}%). Investigate model reliability.",
error_rate * 100.0,
self.config.requirements.max_error_rate * 100.0
));
}
if latency_stats.mean > 50.0 {
recommendations
.push("Consider enabling GPU acceleration for better performance.".to_owned());
}
if recommendations.is_empty() {
recommendations
.push("All performance requirements met. System ready for production.".to_owned());
}
recommendations
}
}
/// Market data update structure
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MarketDataUpdate {
pub symbol: String,
pub price: Price,
pub volume: Decimal,
pub bid: Price,
pub ask: Price,
pub timestamp: std::time::SystemTime,
}
impl MarketDataUpdate {
pub fn spread(&self) -> Price {
Price::from_f64(self.ask.to_f64() - self.bid.to_f64()).unwrap_or_default()
}
pub fn mid_price(&self) -> Price {
Price::from_f64((self.bid.to_f64() + self.ask.to_f64()) / 2.0).unwrap_or_default()
}
}
/// Prediction result with timing information
#[derive(Debug, Clone)]
pub struct PredictionResult {
pub model_name: String,
pub model_type: ModelType,
pub prediction: ModelPrediction,
pub latency_us: u64,
pub timestamp: std::time::SystemTime,
pub success: bool,
pub error_message: Option<String>,
}
/// Phase execution statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PhaseStats {
pub successful_predictions: u64,
pub failed_predictions: u64,
pub latencies: Vec<u64>,
}
impl PhaseStats {
pub fn new() -> Self {
Self {
successful_predictions: 0,
failed_predictions: 0,
latencies: Vec::new(),
}
}
pub fn record_successful_prediction(&mut self, latency_us: u64) {
self.successful_predictions += 1;
self.latencies.push(latency_us);
}
pub fn record_failed_prediction(&mut self, latency_us: u64) {
self.failed_predictions += 1;
self.latencies.push(latency_us);
}
}
/// Phase execution result
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct PhaseResult {
pub phase_name: String,
pub duration: Duration,
pub stats: PhaseStats,
}
/// Requirements check result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RequirementsCheck {
pub latency_ok: bool,
pub p95_latency_ok: bool,
pub p99_latency_ok: bool,
pub error_rate_ok: bool,
pub overall_pass: bool,
}
/// Create default HFT stress test configuration with production-ready settings
pub fn create_hft_stress_test_config() -> StressTestConfig {
StressTestConfig {
duration_seconds: 300, // 5 minutes
target_rps: 50000, // 50k requests per second
concurrent_connections: 100,
market_data_rate: 10000, // 10k market updates per second
test_phases: vec![
TestPhase {
name: "warmup".to_owned(),
duration_seconds: 60,
load_multiplier: 0.5,
market_volatility: 0.01,
error_injection_rate: 0.0,
},
TestPhase {
name: "normal_load".to_owned(),
duration_seconds: 120,
load_multiplier: 1.0,
market_volatility: 0.02,
error_injection_rate: 0.001,
},
TestPhase {
name: "peak_load".to_owned(),
duration_seconds: 60,
load_multiplier: 2.0,
market_volatility: 0.05,
error_injection_rate: 0.005,
},
TestPhase {
name: "stress_load".to_owned(),
duration_seconds: 60,
load_multiplier: 5.0,
market_volatility: 0.1,
error_injection_rate: 0.01,
},
],
models_to_test: vec![
"TLOB_Transformer".to_owned(),
"MAMBA_SSM".to_owned(),
"DQN_Agent".to_owned(),
],
market_conditions: vec![
MarketCondition::Normal,
MarketCondition::HighVolatility,
MarketCondition::Flash,
],
requirements: PerformanceRequirements::default(),
}
}
/// Create stress test configuration optimized for testing with generic symbols
pub fn create_test_stress_test_config() -> StressTestConfig {
let mut config = create_hft_stress_test_config();
// Reduce intensity for testing
config.duration_seconds = 60; // 1 minute for testing
config.target_rps = 1000; // 1k requests per second for testing
config.market_data_rate = 100; // 100 updates per second for testing
// Simplified test phases
config.test_phases = vec![TestPhase {
name: "test_phase".to_owned(),
duration_seconds: 60,
load_multiplier: 1.0,
market_volatility: 0.02,
error_injection_rate: 0.0,
}];
config
}
/// Create stress test configuration with custom simulation parameters
pub fn create_custom_stress_test_config(
_symbols: Vec<String>,
simulation_config: config::SimulationConfig,
) -> StressTestConfig {
let mut config = create_hft_stress_test_config();
// Use simulation config parameters
config.market_data_rate = simulation_config.parameters.update_rate_hz;
// Update volatility in test phases based on simulation config
for phase in &mut config.test_phases {
phase.market_volatility = simulation_config.parameters.base_volatility;
}
config
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_stress_test_config_creation() {
let config = create_hft_stress_test_config();
assert_eq!(config.test_phases.len(), 4);
assert_eq!(config.target_rps, 50000);
}
#[test]
fn test_market_data_calculations() {
let update = MarketDataUpdate {
symbol: "TEST_LARGE_1".to_owned(),
price: Price::from_f64(150.0).unwrap(),
volume: Decimal::try_from(1000.0).unwrap(),
bid: Price::from_f64(149.95).unwrap(),
ask: Price::from_f64(150.05).unwrap(),
timestamp: std::time::SystemTime::now(),
};
assert!((update.spread().to_f64() - 0.10).abs() < 0.001);
assert!((update.mid_price().to_f64() - 150.0).abs() < 0.001);
}
#[test]
fn test_configuration_driven_simulator() {
let simulation_config = config::SimulationConfig::default();
let test_symbols = MarketDataSimulator::generate_test_symbols(&simulation_config);
assert_eq!(test_symbols.len(), simulation_config.test_symbols.count);
assert!(test_symbols[0].starts_with(&simulation_config.test_symbols.symbol_prefix));
}
#[test]
fn test_custom_stress_test_config() {
let symbols = vec!["TEST001".to_owned(), "TEST002".to_owned()];
let simulation_config = config::SimulationConfig::default();
let stress_config = create_custom_stress_test_config(symbols, simulation_config.clone());
assert_eq!(
stress_config.market_data_rate,
simulation_config.parameters.update_rate_hz
);
}
#[test]
fn test_phase_stats() {
let mut stats = PhaseStats::new();
stats.record_successful_prediction(25);
stats.record_successful_prediction(30);
stats.record_failed_prediction(100);
assert_eq!(stats.successful_predictions, 2);
assert_eq!(stats.failed_predictions, 1);
assert_eq!(stats.latencies.len(), 3);
}
}

0
crates/ml/src/stress_testing/load_generator.rs → crates/ml-stress-testing/src/load_generator.rs
View File

1
crates/ml/src/stress_testing/market_simulator.rs → crates/ml-stress-testing/src/market_simulator.rs
View File

@@ -1,6 +1,5 @@
//! Realistic market data simulation for stress testing
// Import types from crate root (lib.rs)
use common::types::Price;
use config::{ml_config::MarketCapTier, MLSymbolConfig as SymbolConfig, SimulationConfig};
use rust_decimal::Decimal;

0
crates/ml/src/stress_testing/performance_analyzer.rs → crates/ml-stress-testing/src/performance_analyzer.rs
View File

3
crates/ml/Cargo.toml
View File

@@ -81,9 +81,12 @@ ml-checkpoint.workspace = true
ml-data-validation.workspace = true
ml-validation.workspace = true
ml-risk.workspace = true
ml-security.workspace = true
ml-backtesting.workspace = true
ml-asset-selection.workspace = true
ml-universe.workspace = true
ml-observability.workspace = true
ml-stress-testing.workspace = true
config.workspace = true
common = { workspace = true, features = ["questdb"] }
risk = { path = "../risk" }

16
crates/ml/src/observability/mod.rs
View File

@@ -1,14 +1,2 @@
//! Production observability and monitoring for ML systems
//!
//! This module provides comprehensive monitoring, metrics collection, and alerting
//! for ML models in production HFT environments.
pub mod alerts;
pub mod dashboards;
pub mod metrics;
// DO NOT RE-EXPORT - Use explicit imports at usage sites
// DO NOT RE-EXPORT - Use explicit imports at usage sites
// DO NOT RE-EXPORT - Use explicit imports at usage sites
//! Observability — thin facade re-exporting from ml-observability crate.
pub use ml_observability::*;

181
crates/ml/src/security/mod.rs
View File

@@ -1,179 +1,6 @@
//! ML Security Module
//!
//! Provides comprehensive security features for ML inference:
//! - Prediction validation and model poisoning detection
//! - Ensemble anomaly detection
//! - Security event logging
//!
//! # Architecture
//!
//! ```text
//! ┌──────────────────┐ ┌──────────────────┐
//! │ Prediction │ │ Ensemble │
//! │ Validator │──────│ Anomaly │
//! │ │ │ Detector │
//! └────────┬─────────┘ └────────┬─────────┘
//! │ │
//! └─────────┬───────────────┘
//! │
//! ▼
//! ┌──────────────────┐
//! │ Security Event │
//! │ Logger │
//! └──────────────────┘
//! ```
//! ML Security — thin facade re-exporting from ml-security crate.
pub mod anomaly_detector;
pub mod prediction_validator;
pub use ml_security::*;
pub use anomaly_detector::{
Anomaly, AnomalyDetectorConfig, AnomalyReport, AnomalySeverity, DetectorStatistics,
EnsembleAnomalyDetector,
};
pub use prediction_validator::{
PredictionStats, PredictionValidator, ValidatedPrediction, ValidationConfig, ValidationFlag,
};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
/// Security event types for ML system
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum SecurityEventType {
// Checkpoint security
CheckpointSignatureFailure,
CheckpointSignatureMissing,
CheckpointTamperingDetected,
// Prediction validation
PredictionOutlierDetected,
PredictionOutOfBounds,
ExtremeRateExceeded,
// Ensemble anomalies
EnsembleSuddenShift,
CoordinatedAttackSuspected,
ModelBehavioralDrift,
// System events
AutomaticRollback,
ManualIntervention,
}
/// Security event for logging
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SecurityEvent {
/// Event type
pub event_type: SecurityEventType,
/// Severity level
pub severity: SecuritySeverity,
/// Event timestamp
pub timestamp: DateTime<Utc>,
/// Model ID (if applicable)
pub model_id: Option<String>,
/// Checkpoint ID (if applicable)
pub checkpoint_id: Option<String>,
/// Prediction ID (if applicable)
pub prediction_id: Option<String>,
/// Human-readable description
pub description: String,
/// Additional metadata
pub metadata: serde_json::Value,
/// Action taken in response
pub action_taken: Option<String>,
}
/// Security severity levels
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum SecuritySeverity {
Low,
Medium,
High,
Critical,
}
impl SecurityEvent {
/// Create a new security event
pub fn new(
event_type: SecurityEventType,
severity: SecuritySeverity,
description: String,
) -> Self {
Self {
event_type,
severity,
timestamp: Utc::now(),
model_id: None,
checkpoint_id: None,
prediction_id: None,
description,
metadata: serde_json::Value::Null,
action_taken: None,
}
}
/// Set model ID
pub fn with_model_id(mut self, model_id: String) -> Self {
self.model_id = Some(model_id);
self
}
/// Set checkpoint ID
pub fn with_checkpoint_id(mut self, checkpoint_id: String) -> Self {
self.checkpoint_id = Some(checkpoint_id);
self
}
/// Set prediction ID
pub fn with_prediction_id(mut self, prediction_id: String) -> Self {
self.prediction_id = Some(prediction_id);
self
}
/// Set metadata
pub fn with_metadata(mut self, metadata: serde_json::Value) -> Self {
self.metadata = metadata;
self
}
/// Set action taken
pub fn with_action(mut self, action: String) -> Self {
self.action_taken = Some(action);
self
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_security_event_builder() {
let event = SecurityEvent::new(
SecurityEventType::PredictionOutlierDetected,
SecuritySeverity::Medium,
"Test outlier detection".to_owned(),
)
.with_model_id("DQN".to_owned())
.with_action("Flagged for review".to_owned());
assert_eq!(event.severity, SecuritySeverity::Medium);
assert_eq!(event.model_id, Some("DQN".to_owned()));
assert_eq!(event.action_taken, Some("Flagged for review".to_owned()));
}
#[test]
fn test_severity_ordering() {
assert!(SecuritySeverity::Critical > SecuritySeverity::High);
assert!(SecuritySeverity::High > SecuritySeverity::Medium);
assert!(SecuritySeverity::Medium > SecuritySeverity::Low);
}
}
// Re-export sub-modules so existing `crate::security::X::Y` paths keep working
pub use ml_security::{anomaly_detector, prediction_validator};

695
crates/ml/src/stress_testing/mod.rs
View File

@@ -1,693 +1,2 @@
//! Stress testing framework for ML models under high-volume market data conditions
//!
//! This module provides comprehensive stress testing capabilities to validate ML model
//! performance under realistic HFT market conditions with high-frequency data feeds.
// Import types from crate root (lib.rs)
use common::types::Price;
use rust_decimal::Decimal;
pub mod load_generator;
pub mod market_simulator;
pub mod performance_analyzer;
// Re-export key types that are needed by external users
pub use load_generator::{LoadGenerator, LoadProfile, TrafficPattern};
pub use market_simulator::{MarketCondition, MarketDataSimulator, SimulatorConfig};
pub use performance_analyzer::{LatencyStats, PerformanceAnalyzer, StressTestReport};
// Types defined in this module are automatically available
// No need to re-export types defined in the same module
use anyhow::Result;
use rust_decimal::prelude::ToPrimitive;
use serde::{Deserialize, Serialize};
use std::time::{Duration, Instant};
use tokio::sync::mpsc;
use crate::{Features, MLModel, ModelPrediction, ModelType};
/// Comprehensive stress test configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StressTestConfig {
/// Test duration in seconds
pub duration_seconds: u64,
/// Target requests per second
pub target_rps: u32,
/// Number of concurrent connections
pub concurrent_connections: u32,
/// Market data feed rate (updates per second)
pub market_data_rate: u32,
/// Test phases with different load patterns
pub test_phases: Vec<TestPhase>,
/// Models to test
pub models_to_test: Vec<String>,
/// Market conditions to simulate
pub market_conditions: Vec<MarketCondition>,
/// Performance requirements
pub requirements: PerformanceRequirements,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TestPhase {
pub name: String,
pub duration_seconds: u64,
pub load_multiplier: f64,
pub market_volatility: f64,
pub error_injection_rate: f64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PerformanceRequirements {
/// Maximum acceptable latency in microseconds
pub max_latency_us: u64,
/// 95th percentile latency requirement
pub p95_latency_us: u64,
/// 99th percentile latency requirement
pub p99_latency_us: u64,
/// Maximum acceptable error rate
pub max_error_rate: f64,
/// Minimum throughput (predictions per second)
pub min_throughput: u32,
/// Maximum memory usage in MB
pub max_memory_mb: u64,
/// Maximum CPU utilization percentage
pub max_cpu_percent: f64,
}
impl Default for PerformanceRequirements {
fn default() -> Self {
Self {
max_latency_us: 100, // 100μs max latency
p95_latency_us: 50, // 50μs 95th percentile
p99_latency_us: 80, // 80μs 99th percentile
max_error_rate: 0.01, // 1% max error rate
min_throughput: 10000, // 10k predictions/sec
max_memory_mb: 1024, // 1GB max memory
max_cpu_percent: 80.0, // 80% max CPU
}
}
}
/// Main stress testing orchestrator
#[derive(Debug)]
pub struct StressTestOrchestrator {
config: StressTestConfig,
market_simulator: MarketDataSimulator,
load_generator: LoadGenerator,
performance_analyzer: PerformanceAnalyzer,
}
impl StressTestOrchestrator {
/// Create new stress test orchestrator with configuration-driven market simulation
pub fn new(config: StressTestConfig) -> Result<Self> {
// Use configuration-driven approach - load from config crate
let simulation_config = config::SimulationConfig::default();
// Use test fixtures for symbol configuration
let test_symbols = vec!["TEST_LARGE_1", "TEST_LARGE_2", "TEST_MID_1", "TEST_SMALL_1"];
let simulator_config = SimulatorConfig {
symbols: test_symbols.into_iter().map(|s| s.to_string()).collect(),
update_rate_hz: config.market_data_rate,
volatility: 0.02,
trend: 0.0,
simulation_config: Some(simulation_config),
};
let market_simulator = MarketDataSimulator::new(simulator_config)?;
let load_generator = LoadGenerator::new(config.target_rps, config.concurrent_connections)?;
let performance_analyzer = PerformanceAnalyzer::new();
Ok(Self {
config,
market_simulator,
load_generator,
performance_analyzer,
})
}
/// Create stress test orchestrator with custom simulation configuration
pub fn new_with_simulation_config(
config: StressTestConfig,
simulation_config: config::SimulationConfig,
) -> Result<Self> {
let symbols: Vec<String> = simulation_config
.initial_market_state
.symbols
.keys()
.cloned()
.collect();
let simulator_config = SimulatorConfig {
symbols,
update_rate_hz: config.market_data_rate,
volatility: simulation_config.parameters.base_volatility,
trend: simulation_config.parameters.trend,
simulation_config: Some(simulation_config),
};
let market_simulator = MarketDataSimulator::new(simulator_config)?;
let load_generator = LoadGenerator::new(config.target_rps, config.concurrent_connections)?;
let performance_analyzer = PerformanceAnalyzer::new();
Ok(Self {
config,
market_simulator,
load_generator,
performance_analyzer,
})
}
/// Create stress test orchestrator for testing with generated test symbols
pub fn new_for_testing(config: StressTestConfig) -> Result<Self> {
let simulation_config = config::SimulationConfig::default();
let test_symbols = MarketDataSimulator::generate_test_symbols(&simulation_config);
let simulator_config = SimulatorConfig {
symbols: test_symbols,
update_rate_hz: config.market_data_rate,
volatility: simulation_config.parameters.base_volatility,
trend: simulation_config.parameters.trend,
simulation_config: Some(simulation_config),
};
let market_simulator = MarketDataSimulator::new(simulator_config)?;
let load_generator = LoadGenerator::new(config.target_rps, config.concurrent_connections)?;
let performance_analyzer = PerformanceAnalyzer::new();
Ok(Self {
config,
market_simulator,
load_generator,
performance_analyzer,
})
}
/// Run comprehensive stress test
pub async fn run_stress_test(
&mut self,
models: Vec<std::sync::Arc<dyn MLModel>>,
) -> Result<StressTestReport> {
tracing::info!("Starting stress test with {} models", models.len());
// Create channels for communication
let (market_tx, mut market_rx) = mpsc::channel(10000);
let (prediction_tx, _prediction_rx) = mpsc::channel(10000);
// Start market data simulation
let simulator_handle = {
let mut simulator = self.market_simulator.clone();
tokio::spawn(async move { simulator.start_simulation(market_tx).await })
};
// Start performance monitoring
let analyzer = self.performance_analyzer.clone();
let monitor_handle = tokio::spawn(async move { analyzer.start_monitoring().await });
// Execute test phases
let mut phase_results = Vec::new();
let test_start = Instant::now();
// Clone the test phases to avoid borrowing conflicts
let test_phases = self.config.test_phases.clone();
for (phase_idx, phase) in test_phases.into_iter().enumerate() {
tracing::info!("Starting test phase {}: {}", phase_idx + 1, phase.name);
let phase_result = self
.run_test_phase(&phase, &models, &mut market_rx, &prediction_tx)
.await?;
phase_results.push(phase_result);
}
// Stop simulation and monitoring
simulator_handle.abort();
monitor_handle.abort();
// Generate comprehensive report
let total_duration = test_start.elapsed();
let report = self
.generate_stress_test_report(phase_results, total_duration)
.await?;
tracing::info!(
"Stress test completed in {:.2}s",
total_duration.as_secs_f64()
);
Ok(report)
}
/// Run individual test phase
async fn run_test_phase(
&mut self,
phase: &TestPhase,
models: &[std::sync::Arc<dyn MLModel>],
market_rx: &mut mpsc::Receiver<MarketDataUpdate>,
prediction_tx: &mpsc::Sender<PredictionResult>,
) -> Result<PhaseResult> {
let phase_start = Instant::now();
let phase_duration = Duration::from_secs(phase.duration_seconds);
let mut phase_stats = PhaseStats::new();
// Adjust load generator for this phase
self.load_generator
.set_load_multiplier(phase.load_multiplier);
while phase_start.elapsed() < phase_duration {
// Process market data updates
while let Ok(market_update) = market_rx.try_recv() {
// Convert market data to features
let features = self.convert_market_data_to_features(&market_update)?;
// Run predictions on all models
for model in models {
let model_start = Instant::now();
match model.predict(&features).await {
Ok(prediction) => {
let latency_us = model_start.elapsed().as_micros() as u64;
phase_stats.record_successful_prediction(latency_us);
let result = PredictionResult {
model_name: model.name().to_string(),
model_type: model.model_type(),
prediction,
latency_us,
timestamp: std::time::SystemTime::now(),
success: true,
error_message: None,
};
drop(prediction_tx.send(result).await);
},
Err(e) => {
let latency_us = model_start.elapsed().as_micros() as u64;
phase_stats.record_failed_prediction(latency_us);
let result = PredictionResult {
model_name: model.name().to_string(),
model_type: model.model_type(),
prediction: ModelPrediction::new(
model.name().to_string(),
0.0,
0.0,
),
latency_us,
timestamp: std::time::SystemTime::now(),
success: false,
error_message: Some(e.to_string()),
};
drop(prediction_tx.send(result).await);
},
}
}
}
// Small delay to prevent busy waiting
tokio::time::sleep(Duration::from_micros(100)).await;
}
Ok(PhaseResult {
phase_name: phase.name.clone(),
duration: phase_start.elapsed(),
stats: phase_stats,
})
}
/// Convert market data to ML features
fn convert_market_data_to_features(&self, market_data: &MarketDataUpdate) -> Result<Features> {
let values = vec![
market_data.price.to_f64(),
market_data.volume.to_f64().unwrap_or(0.0),
market_data.bid.to_f64(),
market_data.ask.to_f64(),
market_data.spread().to_f64(),
market_data.mid_price().to_f64(),
];
let names = vec![
"price".to_owned(),
"volume".to_owned(),
"bid".to_owned(),
"ask".to_owned(),
"spread".to_owned(),
"mid_price".to_owned(),
];
Ok(Features::new(values, names).with_symbol(market_data.symbol.clone()))
}
/// Generate comprehensive stress test report
async fn generate_stress_test_report(
&self,
phase_results: Vec<PhaseResult>,
total_duration: Duration,
) -> Result<StressTestReport> {
let mut total_predictions = 0;
let mut total_errors = 0;
let mut all_latencies = Vec::new();
for phase in &phase_results {
total_predictions +=
phase.stats.successful_predictions + phase.stats.failed_predictions;
total_errors += phase.stats.failed_predictions;
all_latencies.extend(&phase.stats.latencies);
}
let error_rate = if total_predictions > 0 {
total_errors as f64 / total_predictions as f64
} else {
0.0
};
let latency_stats = self.calculate_latency_statistics(&all_latencies);
// Check if requirements are met
let requirements_met = self.check_requirements(&latency_stats, error_rate);
let recommendations = self.generate_recommendations(&latency_stats, error_rate);
Ok(StressTestReport {
config: self.config.clone(),
total_duration,
phase_results,
total_predictions: total_predictions as u64,
total_errors: total_errors as u64,
error_rate,
latency_stats: latency_stats.clone(),
requirements_met,
throughput_achieved: total_predictions as f64 / total_duration.as_secs_f64(),
recommendations,
})
}
fn calculate_latency_statistics(&self, latencies: &[u64]) -> LatencyStats {
if latencies.is_empty() {
return LatencyStats::default();
}
let mut sorted_latencies = latencies.to_vec();
sorted_latencies.sort_unstable();
let len = sorted_latencies.len();
let mean = sorted_latencies.iter().sum::<u64>() as f64 / len as f64;
let min = sorted_latencies[0];
let max = sorted_latencies[len - 1];
let p50 = sorted_latencies[len * 50 / 100];
let p95 = sorted_latencies[len * 95 / 100];
let p99 = sorted_latencies[len * 99 / 100];
LatencyStats {
mean,
min,
max,
p50,
p95,
p99,
count: len as u64,
}
}
fn check_requirements(
&self,
latency_stats: &LatencyStats,
error_rate: f64,
) -> RequirementsCheck {
RequirementsCheck {
latency_ok: latency_stats.max <= self.config.requirements.max_latency_us,
p95_latency_ok: latency_stats.p95 <= self.config.requirements.p95_latency_us,
p99_latency_ok: latency_stats.p99 <= self.config.requirements.p99_latency_us,
error_rate_ok: error_rate <= self.config.requirements.max_error_rate,
overall_pass: latency_stats.max <= self.config.requirements.max_latency_us
&& latency_stats.p95 <= self.config.requirements.p95_latency_us
&& latency_stats.p99 <= self.config.requirements.p99_latency_us
&& error_rate <= self.config.requirements.max_error_rate,
}
}
fn generate_recommendations(
&self,
latency_stats: &LatencyStats,
error_rate: f64,
) -> Vec<String> {
let mut recommendations = Vec::new();
if latency_stats.p99 > self.config.requirements.p99_latency_us {
recommendations.push(format!(
"P99 latency ({}μs) exceeds requirement ({}μs). Consider model optimization or hardware upgrades.",
latency_stats.p99, self.config.requirements.p99_latency_us
));
}
if error_rate > self.config.requirements.max_error_rate {
recommendations.push(format!(
"Error rate ({:.2}%) exceeds requirement ({:.2}%). Investigate model reliability.",
error_rate * 100.0,
self.config.requirements.max_error_rate * 100.0
));
}
if latency_stats.mean > 50.0 {
recommendations
.push("Consider enabling GPU acceleration for better performance.".to_owned());
}
if recommendations.is_empty() {
recommendations
.push("All performance requirements met. System ready for production.".to_owned());
}
recommendations
}
}
/// Market data update structure
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MarketDataUpdate {
pub symbol: String,
pub price: Price,
pub volume: Decimal,
pub bid: Price,
pub ask: Price,
pub timestamp: std::time::SystemTime,
}
impl MarketDataUpdate {
pub fn spread(&self) -> Price {
Price::from_f64(self.ask.to_f64() - self.bid.to_f64()).unwrap_or_default()
}
pub fn mid_price(&self) -> Price {
Price::from_f64((self.bid.to_f64() + self.ask.to_f64()) / 2.0).unwrap_or_default()
}
}
/// Prediction result with timing information
#[derive(Debug, Clone)]
pub struct PredictionResult {
pub model_name: String,
pub model_type: ModelType,
pub prediction: ModelPrediction,
pub latency_us: u64,
pub timestamp: std::time::SystemTime,
pub success: bool,
pub error_message: Option<String>,
}
/// Phase execution statistics
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PhaseStats {
pub successful_predictions: u64,
pub failed_predictions: u64,
pub latencies: Vec<u64>,
}
impl PhaseStats {
pub fn new() -> Self {
Self {
successful_predictions: 0,
failed_predictions: 0,
latencies: Vec::new(),
}
}
pub fn record_successful_prediction(&mut self, latency_us: u64) {
self.successful_predictions += 1;
self.latencies.push(latency_us);
}
pub fn record_failed_prediction(&mut self, latency_us: u64) {
self.failed_predictions += 1;
self.latencies.push(latency_us);
}
}
/// Phase execution result
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct PhaseResult {
pub phase_name: String,
pub duration: Duration,
pub stats: PhaseStats,
}
/// Requirements check result
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RequirementsCheck {
pub latency_ok: bool,
pub p95_latency_ok: bool,
pub p99_latency_ok: bool,
pub error_rate_ok: bool,
pub overall_pass: bool,
}
/// Create default HFT stress test configuration with production-ready settings
pub fn create_hft_stress_test_config() -> StressTestConfig {
StressTestConfig {
duration_seconds: 300, // 5 minutes
target_rps: 50000, // 50k requests per second
concurrent_connections: 100,
market_data_rate: 10000, // 10k market updates per second
test_phases: vec![
TestPhase {
name: "warmup".to_owned(),
duration_seconds: 60,
load_multiplier: 0.5,
market_volatility: 0.01,
error_injection_rate: 0.0,
},
TestPhase {
name: "normal_load".to_owned(),
duration_seconds: 120,
load_multiplier: 1.0,
market_volatility: 0.02,
error_injection_rate: 0.001,
},
TestPhase {
name: "peak_load".to_owned(),
duration_seconds: 60,
load_multiplier: 2.0,
market_volatility: 0.05,
error_injection_rate: 0.005,
},
TestPhase {
name: "stress_load".to_owned(),
duration_seconds: 60,
load_multiplier: 5.0,
market_volatility: 0.1,
error_injection_rate: 0.01,
},
],
models_to_test: vec![
"TLOB_Transformer".to_owned(),
"MAMBA_SSM".to_owned(),
"DQN_Agent".to_owned(),
],
market_conditions: vec![
MarketCondition::Normal,
MarketCondition::HighVolatility,
MarketCondition::Flash,
],
requirements: PerformanceRequirements::default(),
}
}
/// Create stress test configuration optimized for testing with generic symbols
pub fn create_test_stress_test_config() -> StressTestConfig {
let mut config = create_hft_stress_test_config();
// Reduce intensity for testing
config.duration_seconds = 60; // 1 minute for testing
config.target_rps = 1000; // 1k requests per second for testing
config.market_data_rate = 100; // 100 updates per second for testing
// Simplified test phases
config.test_phases = vec![TestPhase {
name: "test_phase".to_owned(),
duration_seconds: 60,
load_multiplier: 1.0,
market_volatility: 0.02,
error_injection_rate: 0.0,
}];
config
}
/// Create stress test configuration with custom simulation parameters
pub fn create_custom_stress_test_config(
_symbols: Vec<String>,
simulation_config: config::SimulationConfig,
) -> StressTestConfig {
let mut config = create_hft_stress_test_config();
// Use simulation config parameters
config.market_data_rate = simulation_config.parameters.update_rate_hz;
// Update volatility in test phases based on simulation config
for phase in &mut config.test_phases {
phase.market_volatility = simulation_config.parameters.base_volatility;
}
config
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_stress_test_config_creation() {
let config = create_hft_stress_test_config();
assert_eq!(config.test_phases.len(), 4);
assert_eq!(config.target_rps, 50000);
}
#[test]
fn test_market_data_calculations() {
let update = MarketDataUpdate {
symbol: "TEST_LARGE_1".to_owned(),
price: Price::from_f64(150.0).unwrap(),
volume: Decimal::try_from(1000.0).unwrap(),
bid: Price::from_f64(149.95).unwrap(),
ask: Price::from_f64(150.05).unwrap(),
timestamp: std::time::SystemTime::now(),
};
assert!((update.spread().to_f64() - 0.10).abs() < 0.001);
assert!((update.mid_price().to_f64() - 150.0).abs() < 0.001);
}
#[test]
fn test_configuration_driven_simulator() {
let simulation_config = config::SimulationConfig::default();
let test_symbols = MarketDataSimulator::generate_test_symbols(&simulation_config);
assert_eq!(test_symbols.len(), simulation_config.test_symbols.count);
assert!(test_symbols[0].starts_with(&simulation_config.test_symbols.symbol_prefix));
}
#[test]
fn test_custom_stress_test_config() {
let symbols = vec!["TEST001".to_owned(), "TEST002".to_owned()];
let simulation_config = config::SimulationConfig::default();
let stress_config = create_custom_stress_test_config(symbols, simulation_config.clone());
assert_eq!(
stress_config.market_data_rate,
simulation_config.parameters.update_rate_hz
);
}
#[test]
fn test_phase_stats() {
let mut stats = PhaseStats::new();
stats.record_successful_prediction(25);
stats.record_successful_prediction(30);
stats.record_failed_prediction(100);
assert_eq!(stats.successful_predictions, 2);
assert_eq!(stats.failed_predictions, 1);
assert_eq!(stats.latencies.len(), 3);
}
}
//! Stress Testing — thin facade re-exporting from ml-stress-testing crate.
pub use ml_stress_testing::*;