diff --git a/Cargo.lock b/Cargo.lock index f38a61bc5..a92524cde 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/Cargo.toml b/Cargo.toml index 7e4807cac..f25db3156 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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" } diff --git a/crates/ml-observability/Cargo.toml b/crates/ml-observability/Cargo.toml new file mode 100644 index 000000000..fbb5511a5 --- /dev/null +++ b/crates/ml-observability/Cargo.toml @@ -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 diff --git a/crates/ml/src/observability/alerts.rs b/crates/ml-observability/src/alerts.rs similarity index 98% rename from crates/ml/src/observability/alerts.rs rename to crates/ml-observability/src/alerts.rs index 741264b5e..738960c5c 100644 --- a/crates/ml/src/observability/alerts.rs +++ b/crates/ml-observability/src/alerts.rs @@ -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 { 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 { 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, diff --git a/crates/ml/src/observability/dashboards.rs b/crates/ml-observability/src/dashboards.rs similarity index 98% rename from crates/ml/src/observability/dashboards.rs rename to crates/ml-observability/src/dashboards.rs index 3f95a1fe4..f316e4e27 100644 --- a/crates/ml/src/observability/dashboards.rs +++ b/crates/ml-observability/src/dashboards.rs @@ -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, diff --git a/crates/ml-observability/src/lib.rs b/crates/ml-observability/src/lib.rs new file mode 100644 index 000000000..152caacb1 --- /dev/null +++ b/crates/ml-observability/src/lib.rs @@ -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; diff --git a/crates/ml/src/observability/metrics.rs b/crates/ml-observability/src/metrics.rs similarity index 98% rename from crates/ml/src/observability/metrics.rs rename to crates/ml-observability/src/metrics.rs index 9343f3fc8..4d0e7b3a9 100644 --- a/crates/ml/src/observability/metrics.rs +++ b/crates/ml-observability/src/metrics.rs @@ -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 diff --git a/crates/ml-security/Cargo.toml b/crates/ml-security/Cargo.toml new file mode 100644 index 000000000..79637fa1c --- /dev/null +++ b/crates/ml-security/Cargo.toml @@ -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 diff --git a/crates/ml/src/security/anomaly_detector.rs b/crates/ml-security/src/anomaly_detector.rs similarity index 98% rename from crates/ml/src/security/anomaly_detector.rs rename to crates/ml-security/src/anomaly_detector.rs index 3b6e6c10f..4d89e10b5 100644 --- a/crates/ml/src/security/anomaly_detector.rs +++ b/crates/ml-security/src/anomaly_detector.rs @@ -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>, @@ -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, diff --git a/crates/ml-security/src/lib.rs b/crates/ml-security/src/lib.rs new file mode 100644 index 000000000..3c9b7a810 --- /dev/null +++ b/crates/ml-security/src/lib.rs @@ -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, + + /// Model ID (if applicable) + pub model_id: Option, + + /// Checkpoint ID (if applicable) + pub checkpoint_id: Option, + + /// Prediction ID (if applicable) + pub prediction_id: Option, + + /// Human-readable description + pub description: String, + + /// Additional metadata + pub metadata: serde_json::Value, + + /// Action taken in response + pub action_taken: Option, +} + +/// 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); + } +} diff --git a/crates/ml/src/security/prediction_validator.rs b/crates/ml-security/src/prediction_validator.rs similarity index 98% rename from crates/ml/src/security/prediction_validator.rs rename to crates/ml-security/src/prediction_validator.rs index 85e0e3b95..18d53892f 100644 --- a/crates/ml/src/security/prediction_validator.rs +++ b/crates/ml-security/src/prediction_validator.rs @@ -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, @@ -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, diff --git a/crates/ml-stress-testing/Cargo.toml b/crates/ml-stress-testing/Cargo.toml new file mode 100644 index 000000000..ed4374075 --- /dev/null +++ b/crates/ml-stress-testing/Cargo.toml @@ -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 diff --git a/crates/ml-stress-testing/src/lib.rs b/crates/ml-stress-testing/src/lib.rs new file mode 100644 index 000000000..3028fea90 --- /dev/null +++ b/crates/ml-stress-testing/src/lib.rs @@ -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, + /// Models to test + pub models_to_test: Vec, + /// Market conditions to simulate + pub market_conditions: Vec, + /// 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 { + // 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 { + let symbols: Vec = 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 { + 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>, + ) -> Result { + 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], + market_rx: &mut mpsc::Receiver, + prediction_tx: &mpsc::Sender, + ) -> Result { + 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 { + 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, + total_duration: Duration, + ) -> Result { + 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::() 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 { + 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, +} + +/// Phase execution statistics +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct PhaseStats { + pub successful_predictions: u64, + pub failed_predictions: u64, + pub latencies: Vec, +} + +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, + 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); + } +} diff --git a/crates/ml/src/stress_testing/load_generator.rs b/crates/ml-stress-testing/src/load_generator.rs similarity index 100% rename from crates/ml/src/stress_testing/load_generator.rs rename to crates/ml-stress-testing/src/load_generator.rs diff --git a/crates/ml/src/stress_testing/market_simulator.rs b/crates/ml-stress-testing/src/market_simulator.rs similarity index 99% rename from crates/ml/src/stress_testing/market_simulator.rs rename to crates/ml-stress-testing/src/market_simulator.rs index d6bf34320..068e7732b 100644 --- a/crates/ml/src/stress_testing/market_simulator.rs +++ b/crates/ml-stress-testing/src/market_simulator.rs @@ -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; diff --git a/crates/ml/src/stress_testing/performance_analyzer.rs b/crates/ml-stress-testing/src/performance_analyzer.rs similarity index 100% rename from crates/ml/src/stress_testing/performance_analyzer.rs rename to crates/ml-stress-testing/src/performance_analyzer.rs diff --git a/crates/ml/Cargo.toml b/crates/ml/Cargo.toml index 8880391cf..ae3c2f247 100644 --- a/crates/ml/Cargo.toml +++ b/crates/ml/Cargo.toml @@ -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" } diff --git a/crates/ml/src/observability/mod.rs b/crates/ml/src/observability/mod.rs index cf7b91222..116dea06b 100644 --- a/crates/ml/src/observability/mod.rs +++ b/crates/ml/src/observability/mod.rs @@ -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::*; diff --git a/crates/ml/src/security/mod.rs b/crates/ml/src/security/mod.rs index 49c2b8aa2..2fd6376e5 100644 --- a/crates/ml/src/security/mod.rs +++ b/crates/ml/src/security/mod.rs @@ -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, - - /// Model ID (if applicable) - pub model_id: Option, - - /// Checkpoint ID (if applicable) - pub checkpoint_id: Option, - - /// Prediction ID (if applicable) - pub prediction_id: Option, - - /// Human-readable description - pub description: String, - - /// Additional metadata - pub metadata: serde_json::Value, - - /// Action taken in response - pub action_taken: Option, -} - -/// 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}; diff --git a/crates/ml/src/stress_testing/mod.rs b/crates/ml/src/stress_testing/mod.rs index dc0174a57..79a4c5237 100644 --- a/crates/ml/src/stress_testing/mod.rs +++ b/crates/ml/src/stress_testing/mod.rs @@ -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, - /// Models to test - pub models_to_test: Vec, - /// Market conditions to simulate - pub market_conditions: Vec, - /// 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 { - // 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 { - let symbols: Vec = 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 { - 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>, - ) -> Result { - 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], - market_rx: &mut mpsc::Receiver, - prediction_tx: &mpsc::Sender, - ) -> Result { - 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 { - 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, - total_duration: Duration, - ) -> Result { - 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::() 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 { - 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, -} - -/// Phase execution statistics -#[derive(Debug, Clone, Serialize, Deserialize)] -pub struct PhaseStats { - pub successful_predictions: u64, - pub failed_predictions: u64, - pub latencies: Vec, -} - -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, - 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::*;