diff --git a/ml/src/hyperopt/adapters/dqn.rs b/ml/src/hyperopt/adapters/dqn.rs index 7ea99c765..9f4169669 100644 --- a/ml/src/hyperopt/adapters/dqn.rs +++ b/ml/src/hyperopt/adapters/dqn.rs @@ -3044,15 +3044,9 @@ impl HyperparameterOptimizable for DQNTrainer { }; // Component 2: HFT activity score (25% weight) - // P0 FIX (2025-11-20): TEMPORARILY DISABLED due to missing action counters - // Root cause: DQN trainer never populates buy_count/sell_count/hold_count in additional_metrics - // This causes false -5.0 penalty, degrading Sharpe from 0.77 to 0.29 (62% loss) - // Proper fix requires 1-2h to implement action counting in ml/src/trainers/dqn.rs - // TODO: Re-enable after implementing buy/sell/hold action counting - let hft_activity = 0.0; // DISABLED: was calculate_hft_activity_score_wave10(...) - - // Log warning about disabled activity penalty - tracing::warn!("P0 FIX: Activity penalty DISABLED (action counters not implemented)"); + // Re-enabled: DQN trainer now populates buy_count/sell_count/hold_count + // in additional_metrics from the 45-action factored space. + let hft_activity = calculate_hft_activity_score_wave10(buy_pct, sell_pct, hold_pct); // Component 3: Stability penalty (15% weight) // Penalizes gradient explosion (>50.0) and Q-value volatility (>100.0) diff --git a/ml/src/trainers/dqn/trainer.rs b/ml/src/trainers/dqn/trainer.rs index cb22f99f5..f5c2d5033 100644 --- a/ml/src/trainers/dqn/trainer.rs +++ b/ml/src/trainers/dqn/trainer.rs @@ -1128,6 +1128,18 @@ impl DQNTrainer { metrics.add_metric("top5_coverage_pct", top5_coverage_pct); metrics.add_metric("total_actions", total_actions as f64); + + // Aggregate buy/sell/hold counts from 45-action factored space + // Layout: index = exposure*9 + order*3 + urgency + // Sell (Short100 + Short50): exposure 0-1, indices 0..18 + // Hold (Flat): exposure 2, indices 18..27 + // Buy (Long50 + Long100): exposure 3-4, indices 27..45 + let sell_count: usize = total_action_counts.iter().take(18).sum(); + let hold_count: usize = total_action_counts.iter().skip(18).take(9).sum(); + let buy_count: usize = total_action_counts.iter().skip(27).sum(); + metrics.add_metric("buy_count", buy_count as f64); + metrics.add_metric("sell_count", sell_count as f64); + metrics.add_metric("hold_count", hold_count as f64); } if early_stopped { diff --git a/services/ml_training_service/build.rs b/services/ml_training_service/build.rs index 001d075a2..86ebce916 100644 --- a/services/ml_training_service/build.rs +++ b/services/ml_training_service/build.rs @@ -3,6 +3,7 @@ fn main() -> Result<(), Box> { let config = tonic_prost_build::configure(); config + .clone() .build_server(true) .build_client(true) .compile_well_known_types(true) @@ -13,7 +14,22 @@ fn main() -> Result<(), Box> { .client_mod_attribute(".", "#[allow(unused_qualifications)]") .compile_protos(&["proto/ml_training.proto"], &["proto"])?; + // Compile TLI BacktestingService proto (client only -- for validation pipeline) + config + .build_server(false) + .build_client(true) + .compile_well_known_types(true) + .extern_path(".google.protobuf", "::prost_types") + .type_attribute(".", "#[derive(serde::Serialize, serde::Deserialize)]") + .server_mod_attribute(".", "#[allow(unused_qualifications)]") + .client_mod_attribute(".", "#[allow(unused_qualifications)]") + .compile_protos( + &["../../tli/proto/trading.proto"], + &["../../tli/proto"], + )?; + println!("cargo:rerun-if-changed=proto/ml_training.proto"); + println!("cargo:rerun-if-changed=../../tli/proto/trading.proto"); Ok(()) } diff --git a/services/ml_training_service/src/lib.rs b/services/ml_training_service/src/lib.rs index d405bf5cd..0ce7139a2 100644 --- a/services/ml_training_service/src/lib.rs +++ b/services/ml_training_service/src/lib.rs @@ -36,6 +36,11 @@ pub mod trial_executor; pub mod tuning_manager; pub mod validation_pipeline; +// TLI BacktestingService proto (client only -- for validation pipeline) +pub mod backtesting_proto { + tonic::include_proto!("foxhunt.tli"); +} + // Re-export proto module for test access pub use service::proto; diff --git a/services/ml_training_service/src/service.rs b/services/ml_training_service/src/service.rs index bb4943dd6..1565cb2be 100644 --- a/services/ml_training_service/src/service.rs +++ b/services/ml_training_service/src/service.rs @@ -35,9 +35,10 @@ use proto::{ TrainingStatusUpdate as ProtoStatusUpdate, }; +use crate::batch_tuning_manager::{BatchJobStatus, BatchTuningManager}; use crate::grpc_tuning_handlers::TuningHandlers; use crate::orchestrator::{JobStatus, TrainingOrchestrator, TrainingStatusUpdate}; -use crate::tuning_manager::TuningManager; +use crate::tuning_manager::{TuningJobStatus, TuningManager}; use config::MLConfig; use ml::training_pipeline::{ ModelArchitectureConfig, ProductionTrainingConfig, TrainingHyperparameters, @@ -47,6 +48,7 @@ use ml::training_pipeline::{ pub struct MLTrainingServiceImpl { orchestrator: Arc, tuning_handlers: Arc, + batch_tuning_manager: Arc, #[allow(dead_code)] config: MLConfig, } @@ -58,10 +60,13 @@ impl MLTrainingServiceImpl { tuning_manager: Arc, config: MLConfig, ) -> Self { - let tuning_handlers = Arc::new(TuningHandlers::new(tuning_manager)); + let tuning_handlers = Arc::new(TuningHandlers::new(Arc::clone(&tuning_manager))); + let working_dir = std::env::var("TUNING_WORKING_DIR").unwrap_or_else(|_| ".".to_string()); + let batch_tuning_manager = Arc::new(BatchTuningManager::new(tuning_manager, working_dir)); Self { orchestrator, tuning_handlers, + batch_tuning_manager, config, } } @@ -78,6 +83,31 @@ impl MLTrainingServiceImpl { } } + /// Convert internal batch job status to protobuf batch status + fn convert_batch_status(status: &BatchJobStatus) -> proto::BatchTuningStatus { + match status { + BatchJobStatus::Pending => proto::BatchTuningStatus::BatchPending, + BatchJobStatus::Running => proto::BatchTuningStatus::BatchRunning, + BatchJobStatus::Completed => proto::BatchTuningStatus::BatchCompleted, + BatchJobStatus::PartiallyCompleted => { + proto::BatchTuningStatus::BatchPartiallyCompleted + }, + BatchJobStatus::Failed => proto::BatchTuningStatus::BatchFailed, + BatchJobStatus::Stopped => proto::BatchTuningStatus::BatchStopped, + } + } + + /// Convert internal tuning job status to protobuf tuning status + fn convert_tuning_job_status(status: &TuningJobStatus) -> proto::TuningJobStatus { + match status { + TuningJobStatus::Pending => proto::TuningJobStatus::TuningPending, + TuningJobStatus::Running => proto::TuningJobStatus::TuningRunning, + TuningJobStatus::Completed => proto::TuningJobStatus::TuningCompleted, + TuningJobStatus::Failed => proto::TuningJobStatus::TuningFailed, + TuningJobStatus::Stopped => proto::TuningJobStatus::TuningStopped, + } + } + /// Convert protobuf hyperparameters to internal config fn convert_hyperparameters( &self, @@ -783,34 +813,209 @@ impl MlTrainingService for MLTrainingServiceImpl { /// Start batch tuning job for multiple models async fn batch_start_tuning_jobs( &self, - _request: Request, + request: Request, ) -> Result, Status> { - // TODO: Implement batch tuning orchestration - Err(Status::unimplemented( - "Batch tuning is not yet implemented. Use individual StartTuningJob calls instead.", - )) + let req = request.into_inner(); + + info!( + "Starting batch tuning job for models: {:?} with {} trials each", + req.model_types, req.trials_per_model + ); + + // Validate inputs + if req.model_types.is_empty() { + return Err(Status::invalid_argument( + "At least one model type must be specified", + )); + } + + if req.trials_per_model == 0 { + return Err(Status::invalid_argument( + "Number of trials per model must be > 0", + )); + } + + if req.config_path.is_empty() { + return Err(Status::invalid_argument( + "Configuration path cannot be empty", + )); + } + + // Determine data source path from the proto DataSource + let data_source_path = req.data_source.and_then(|ds| { + ds.source.map(|s| match s { + proto::data_source::Source::FilePath(p) => p, + proto::data_source::Source::HistoricalDbQuery(q) => q, + proto::data_source::Source::RealTimeStreamTopic(t) => t, + }) + }); + + // Determine YAML export path (use custom or default) + let yaml_export_path = if req.yaml_export_path.is_empty() { + None + } else { + Some(req.yaml_export_path) + }; + + // Start batch tuning via manager + let batch_id = self + .batch_tuning_manager + .start_batch_tuning( + req.model_types, + req.trials_per_model, + req.config_path, + data_source_path, + req.auto_export_yaml, + yaml_export_path, + ) + .await + .map_err(|e| Status::internal(format!("Failed to start batch tuning: {}", e)))?; + + // Get the job to return the execution order + let job = self + .batch_tuning_manager + .get_batch_status(batch_id) + .await + .map_err(|e| { + Status::internal(format!("Batch started but failed to retrieve status: {}", e)) + })?; + + info!( + "Batch tuning job started: {} with execution order: {:?}", + batch_id, job.execution_order + ); + + let response = proto::BatchStartTuningJobsResponse { + batch_id: batch_id.to_string(), + execution_order: job.execution_order, + message: format!("Batch tuning started for {} models", job.models.len()), + status: Self::convert_batch_status(&job.status) as i32, + }; + + Ok(Response::new(response)) } /// Get batch tuning job status async fn get_batch_tuning_status( &self, - _request: Request, + request: Request, ) -> Result, Status> { - // TODO: Implement batch status retrieval - Err(Status::unimplemented( - "Batch tuning status is not yet implemented.", - )) + let req = request.into_inner(); + + let batch_id = Uuid::parse_str(&req.batch_id) + .map_err(|_| Status::invalid_argument("Invalid batch ID format"))?; + + debug!("Getting batch tuning status for: {}", batch_id); + + let job = self + .batch_tuning_manager + .get_batch_status(batch_id) + .await + .map_err(|e| Status::not_found(format!("Batch job not found: {}", e)))?; + + // Convert model results to proto format + let results: Vec = job + .results + .iter() + .map(|r| proto::ModelTuningResult { + model_type: r.model_type.clone(), + job_id: r.job_id.to_string(), + status: Self::convert_tuning_job_status(&r.status) as i32, + best_params: r.best_params.clone(), + best_metrics: r.best_metrics.clone(), + trials_completed: r.trials_completed, + started_at: r.started_at.timestamp(), + completed_at: r.completed_at.map(|dt| dt.timestamp()).unwrap_or(0), + error_message: r.error_message.clone().unwrap_or_default(), + }) + .collect(); + + // Determine the current model name + let current_model = job + .execution_order + .get(job.current_model_index) + .cloned() + .unwrap_or_default(); + + let response = proto::GetBatchTuningStatusResponse { + batch_id: job.batch_id.to_string(), + status: Self::convert_batch_status(&job.status) as i32, + current_model_index: job.current_model_index as u32, + total_models: job.execution_order.len() as u32, + results, + current_model, + started_at: job.started_at.timestamp(), + updated_at: job.updated_at.timestamp(), + estimated_completion_time: 0, // Not yet estimated + yaml_export_path: job.yaml_export_path.clone(), + }; + + Ok(Response::new(response)) } /// Stop a running batch tuning job async fn stop_batch_tuning_job( &self, - _request: Request, + request: Request, ) -> Result, Status> { - // TODO: Implement batch job cancellation - Err(Status::unimplemented( - "Batch tuning stop is not yet implemented.", - )) + let req = request.into_inner(); + + let batch_id = Uuid::parse_str(&req.batch_id) + .map_err(|_| Status::invalid_argument("Invalid batch ID format"))?; + + info!( + "Stopping batch tuning job: {} (reason: {})", + batch_id, req.reason + ); + + // Stop the batch job + self.batch_tuning_manager + .stop_batch_job(batch_id, req.reason.clone()) + .await + .map_err(|e| Status::internal(format!("Failed to stop batch job: {}", e)))?; + + // Get final status + let job = self + .batch_tuning_manager + .get_batch_status(batch_id) + .await + .map_err(|e| Status::not_found(format!("Batch job not found after stop: {}", e)))?; + + // Convert completed results + let completed_results: Vec = job + .results + .iter() + .map(|r| proto::ModelTuningResult { + model_type: r.model_type.clone(), + job_id: r.job_id.to_string(), + status: Self::convert_tuning_job_status(&r.status) as i32, + best_params: r.best_params.clone(), + best_metrics: r.best_metrics.clone(), + trials_completed: r.trials_completed, + started_at: r.started_at.timestamp(), + completed_at: r.completed_at.map(|dt| dt.timestamp()).unwrap_or(0), + error_message: r.error_message.clone().unwrap_or_default(), + }) + .collect(); + + let completed_count = job + .results + .iter() + .filter(|r| r.status == TuningJobStatus::Completed) + .count(); + + let response = proto::StopBatchTuningJobResponse { + success: true, + message: format!( + "Batch job stopped. {} of {} models completed before stop.", + completed_count, + job.execution_order.len() + ), + final_status: Self::convert_batch_status(&job.status) as i32, + completed_results, + }; + + Ok(Response::new(response)) } } diff --git a/services/ml_training_service/src/validation_pipeline.rs b/services/ml_training_service/src/validation_pipeline.rs index 985d9346d..a6714f852 100644 --- a/services/ml_training_service/src/validation_pipeline.rs +++ b/services/ml_training_service/src/validation_pipeline.rs @@ -9,10 +9,17 @@ use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; +use std::collections::HashMap; use std::path::Path; -use tracing::{debug, error, info}; +use tokio::sync::Mutex; +use tonic::transport::Channel; +use tracing::{debug, error, info, warn}; use uuid::Uuid; +use crate::backtesting_proto::backtesting_service_client::BacktestingServiceClient; +use crate::backtesting_proto::{ + GetBacktestResultsRequest, GetBacktestStatusRequest, StartBacktestRequest, +}; use crate::orchestrator::TrainingJob; /// Validation pipeline configuration @@ -121,6 +128,8 @@ pub struct ValidationResult { /// Model validation pipeline pub struct ValidationPipeline { config: ValidationConfig, + /// Optional gRPC client for BacktestingService (falls back to mock when None) + backtesting_client: Option>>, } impl ValidationPipeline { @@ -150,7 +159,16 @@ impl ValidationPipeline { )); } - Ok(Self { config }) + Ok(Self { + config, + backtesting_client: None, + }) + } + + /// Create a validation pipeline with a BacktestingService gRPC client + pub fn with_backtesting_client(mut self, client: BacktestingServiceClient) -> Self { + self.backtesting_client = Some(Mutex::new(client)); + self } /// Get validation configuration @@ -339,29 +357,166 @@ impl ValidationPipeline { Ok(bars) } - /// Run backtest using BacktestingService + /// Run backtest using BacktestingService gRPC (falls back to mock when unavailable) pub async fn run_backtest( &self, - _training_job: &TrainingJob, - _data_path: &str, + training_job: &TrainingJob, + data_path: &str, ) -> Result { info!( "Running backtest for validation (duration: {} days)", self.config.backtest_duration_days ); - // TODO: Integrate with BacktestingService gRPC client - // For now, generate realistic mock metrics for testing - let mock_metrics = self.generate_mock_backtest_results().await; + let metrics = match &self.backtesting_client { + Some(client_mutex) => { + match self + .run_grpc_backtest(client_mutex, training_job, data_path) + .await + { + Ok(m) => m, + Err(e) => { + warn!( + "BacktestingService gRPC call failed, falling back to mock: {}", + e + ); + self.generate_mock_backtest_results().await + } + } + } + None => { + warn!("BacktestingService not configured, using mock backtest results"); + self.generate_mock_backtest_results().await + } + }; info!( "Backtest completed: Sharpe={:.2}, Win Rate={:.2}%, Drawdown={:.2}%", - mock_metrics.sharpe_ratio, - mock_metrics.win_rate * 100.0, - mock_metrics.max_drawdown * 100.0 + metrics.sharpe_ratio, + metrics.win_rate * 100.0, + metrics.max_drawdown * 100.0 ); - Ok(mock_metrics) + Ok(metrics) + } + + /// Execute backtest via gRPC BacktestingService + async fn run_grpc_backtest( + &self, + client_mutex: &Mutex>, + training_job: &TrainingJob, + data_path: &str, + ) -> Result { + let mut client = client_mutex.lock().await; + + // Calculate time range from backtest_duration_days + let now = Utc::now(); + let start = now - chrono::Duration::days(i64::from(self.config.backtest_duration_days)); + + let mut parameters = HashMap::new(); + parameters.insert("model_type".to_string(), training_job.model_type.clone()); + parameters.insert("data_path".to_string(), data_path.to_string()); + + let request = tonic::Request::new(StartBacktestRequest { + strategy_name: format!("ml_validation_{}", training_job.model_type), + symbols: vec!["ES.FUT".to_string()], + start_date_unix_nanos: start.timestamp_nanos_opt().unwrap_or(0), + end_date_unix_nanos: now.timestamp_nanos_opt().unwrap_or(0), + initial_capital: 100_000.0, + parameters, + save_results: false, + description: format!( + "Validation backtest for training job {}", + training_job.id + ), + }); + + let start_resp = client + .start_backtest(request) + .await + .context("StartBacktest gRPC call failed")? + .into_inner(); + + if !start_resp.success { + return Err(anyhow::anyhow!( + "BacktestingService rejected request: {}", + start_resp.message + )); + } + + let backtest_id = start_resp.backtest_id; + info!("Backtest started with id: {}", backtest_id); + + // Poll for completion (up to 5 minutes) + let timeout_at = tokio::time::Instant::now() + tokio::time::Duration::from_secs(300); + loop { + if tokio::time::Instant::now() >= timeout_at { + return Err(anyhow::anyhow!( + "Backtest {} timed out after 5 minutes", + backtest_id + )); + } + + tokio::time::sleep(tokio::time::Duration::from_secs(2)).await; + + let status_resp = client + .get_backtest_status(tonic::Request::new(GetBacktestStatusRequest { + backtest_id: backtest_id.clone(), + })) + .await + .context("GetBacktestStatus gRPC call failed")? + .into_inner(); + + // BacktestStatus: 3 = COMPLETED, 4 = FAILED + match status_resp.status { + 3 => { + info!("Backtest {} completed", backtest_id); + break; + } + 4 => { + return Err(anyhow::anyhow!( + "Backtest {} failed: {}", + backtest_id, + status_resp.error_message.unwrap_or_default() + )); + } + _ => { + debug!( + "Backtest {} progress: {:.1}%", + backtest_id, status_resp.progress_percentage + ); + } + } + } + + // Fetch results + let results_resp = client + .get_backtest_results(tonic::Request::new(GetBacktestResultsRequest { + backtest_id: backtest_id.clone(), + include_trades: false, + include_metrics: true, + })) + .await + .context("GetBacktestResults gRPC call failed")? + .into_inner(); + + let proto_metrics = results_resp + .metrics + .ok_or_else(|| anyhow::anyhow!("BacktestingService returned no metrics"))?; + + Ok(ValidationMetrics { + sharpe_ratio: proto_metrics.sharpe_ratio, + win_rate: proto_metrics.win_rate, + max_drawdown: proto_metrics.max_drawdown, + total_trades: proto_metrics.total_trades, + avg_profit_per_trade: if proto_metrics.total_trades > 0 { + proto_metrics.total_return / proto_metrics.total_trades as f64 + } else { + 0.0 + }, + profit_factor: proto_metrics.profit_factor, + total_return: proto_metrics.total_return, + }) } /// Generate mock backtest results (temporary until BacktestingService integration) diff --git a/services/trading_service/src/core/risk_manager.rs b/services/trading_service/src/core/risk_manager.rs index e889789a7..c53150c66 100644 --- a/services/trading_service/src/core/risk_manager.rs +++ b/services/trading_service/src/core/risk_manager.rs @@ -1169,13 +1169,121 @@ impl RiskManager { } async fn recalculate_symbol_var(&self, symbol: &str) -> Result<(), RiskError> { - // TODO: Implement VaR recalculation for accounts holding this symbol - // Currently a placeholder - production requires: - // 1. Query all accounts holding position in 'symbol' - // 2. Recalculate portfolio VaR for each affected account - // 3. Update exposure tracking with new VaR values - // 4. Trigger compliance events if VaR limits exceeded - let _ = symbol; // Acknowledge parameter until implementation complete + // 1. Collect affected account IDs (those holding a position in this symbol) + let affected_accounts: Vec = { + let exposures = self.exposures.read().await; + exposures + .iter() + .filter(|(_, exp)| exp.concentration_by_symbol.contains_key(symbol)) + .map(|(account_id, _)| account_id.clone()) + .collect() + }; + + if affected_accounts.is_empty() { + return Ok(()); + } + + // 2. Build portfolio return series for each affected account + let return_history = self.return_history.read().await; + + for account_id in &affected_accounts { + // Get account exposure snapshot + let exposure = { + let exposures = self.exposures.read().await; + match exposures.get(account_id) { + Some(exp) => exp.clone(), + None => continue, + } + }; + + // Calculate weighted portfolio returns using historical simulation + let min_history_len = exposure + .concentration_by_symbol + .keys() + .filter_map(|sym| return_history.get(sym).map(|r| r.len())) + .min() + .unwrap_or(0); + + if min_history_len < 30 { + // Not enough data points for reliable VaR; skip this account + continue; + } + + let mut portfolio_returns = Vec::with_capacity(min_history_len); + for i in 0..min_history_len { + let mut period_return = 0.0_f64; + let mut total_weight = 0.0_f64; + + for (sym, &weight) in &exposure.concentration_by_symbol { + if let Some(sym_returns) = return_history.get(sym) { + if let Some(&ret) = sym_returns.get(i) { + if ret.is_finite() { + period_return += weight * ret; + total_weight += weight.abs(); + } + } + } + } + + // Normalize if we have valid weights + if total_weight > 0.0 { + let normalized = period_return / total_weight; + if normalized.is_finite() { + portfolio_returns.push(normalized); + } + } + } + + if portfolio_returns.len() < 30 { + continue; + } + + // 3. Sort returns for percentile-based historical VaR + portfolio_returns + .sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + + // 4. Calculate 95% VaR (5th percentile of sorted returns) + let var_index_1d = (portfolio_returns.len() as f64 * 0.05).floor() as usize; + let var_return_1d = portfolio_returns + .get(var_index_1d) + .copied() + .unwrap_or(-0.02); // Conservative fallback + + // 1-day 95% VaR as positive loss amount (negate the negative percentile return) + let var_1d = -(var_return_1d * exposure.total_exposure); + let var_1d = if var_1d.is_finite() { var_1d } else { 0.0 }; + + // 10-day VaR via square-root-of-time scaling + let var_10d = var_1d * (10.0_f64).sqrt(); + + // 5. Update exposure tracking with new VaR values + { + let mut exposures = self.exposures.write().await; + if let Some(exp) = exposures.get_mut(account_id) { + exp.var_1d = var_1d; + exp.var_10d = var_10d; + exp.last_update_ns = HardwareTimestamp::now().as_nanos(); + } + } + + // 6. Check VaR against limits and trigger compliance events if breached + let var_limit = + self.fixed_to_price(self.limits.var_limit_1d.load(Ordering::Relaxed)); + + if var_limit > 0.0 && var_1d > var_limit { + let violation = RiskViolation::VarLimitExceeded { + var_1d, + limit: var_limit, + }; + self.record_violation(violation).await; + + debug!( + "VaR limit breached for account {}: 1d VaR {:.2} > limit {:.2}", + account_id, var_1d, var_limit + ); + } + } + Ok(()) } diff --git a/services/trading_service/src/services/trading.rs b/services/trading_service/src/services/trading.rs index 18d45bcf8..d7a165414 100644 --- a/services/trading_service/src/services/trading.rs +++ b/services/trading_service/src/services/trading.rs @@ -1157,11 +1157,9 @@ impl TradingServiceImpl { let event = TradingEvent::new(event_type_internal, order_id.to_string(), payload); - // NOTE: EventPublisher publish() requires &mut self, not available through Arc - // For now, log event instead of publishing - // TODO: Refactor EventPublisher to use interior mutability - let _ = event; // Suppress unused warning - // Event publishing disabled - needs refactoring + if let Err(e) = self.state.event_publisher.publish(event).await { + warn!("Failed to publish event: {}", e); + } } /// Convert TradingEvent to OrderEvent proto diff --git a/trading_engine/src/compliance/mod.rs b/trading_engine/src/compliance/mod.rs index 11b9ee8c8..c43c67882 100644 --- a/trading_engine/src/compliance/mod.rs +++ b/trading_engine/src/compliance/mod.rs @@ -453,6 +453,12 @@ impl ComplianceEngine { } /// Assess `MAR` compliance + /// + /// Performs basic market surveillance checks including: + /// - Unusual volume detection (large order quantities) + /// - Price deviation analysis (limit price vs market volatility) + /// - Suspicious pattern detection (off-hours trading, stressed market activity) + #[allow(clippy::float_arithmetic)] async fn assess_mar_compliance( &self, context: &ComplianceContext, @@ -462,23 +468,200 @@ impl ComplianceEngine { return Ok(ComplianceStatus::NotApplicable); } - // TODO: Market surveillance analysis (when module is implemented) - // Market surveillance analysis - if let Some(_order) = &context.order_info { - // Placeholder - market surveillance module not yet implemented - findings.push(ComplianceFinding { - id: format!("MAR-TODO-{}", Uuid::new_v4()), - regulation: "Market Abuse Regulation".to_owned(), - severity: ComplianceSeverity::Info, - description: "Market surveillance module not yet implemented".to_owned(), - remediation: "Implement market surveillance analysis".to_owned(), - due_date: Some(Utc::now() + Duration::days(30)), - status: FindingStatus::Open, - }); + let mut warnings: Vec = Vec::new(); + let mut has_violation = false; + + // --- Check 1: Unusual volume detection --- + if let Some(order) = &context.order_info { + let qty = order.quantity.to_f64(); + + // Flag extraordinarily large orders as potential market manipulation + // (layering/spoofing indicator). In production this threshold would be + // instrument-specific and derived from historical ADV; here we use a + // conservative static bound as a baseline surveillance heuristic. + const LARGE_ORDER_THRESHOLD: f64 = 100_000.0; + const VERY_LARGE_ORDER_THRESHOLD: f64 = 500_000.0; + + if qty > VERY_LARGE_ORDER_THRESHOLD { + findings.push(ComplianceFinding { + id: format!("MAR-VOL-CRIT-{}", Uuid::new_v4()), + regulation: "MAR Article 16 - Suspicious Order Detection".to_owned(), + severity: ComplianceSeverity::High, + description: format!( + "Exceptionally large order detected: {qty:.2} units on {} ({}). \ + Order size exceeds critical threshold ({VERY_LARGE_ORDER_THRESHOLD:.0}) \ + and may indicate market manipulation or erroneous order entry.", + order.symbol, order.side + ), + remediation: "Investigate order origin. Verify client intent and \ + review for potential layering or spoofing activity." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(4)), + status: FindingStatus::Open, + }); + has_violation = true; + } else if qty > LARGE_ORDER_THRESHOLD { + findings.push(ComplianceFinding { + id: format!("MAR-VOL-{}", Uuid::new_v4()), + regulation: "MAR Article 16 - Unusual Volume".to_owned(), + severity: ComplianceSeverity::Medium, + description: format!( + "Large order detected: {qty:.2} units on {} ({}). \ + Exceeds surveillance threshold ({LARGE_ORDER_THRESHOLD:.0}).", + order.symbol, order.side + ), + remediation: + "Monitor subsequent order flow for this symbol and client." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(24)), + status: FindingStatus::Open, + }); + warnings.push(format!("Large order volume on {}", order.symbol)); + } + + // --- Check 2: Price deviation analysis --- + // When a limit price is set, check for anomalies. A price of zero + // is treated as potentially erroneous. With market context we also + // flag large limit orders during high volatility. + if let Some(limit_price) = &order.price { + let price_val = limit_price.to_f64(); + + // Zero or near-zero limit price is suspicious + if price_val < f64::EPSILON { + findings.push(ComplianceFinding { + id: format!("MAR-PRC-ZERO-{}", Uuid::new_v4()), + regulation: "MAR Article 12 - Price Manipulation".to_owned(), + severity: ComplianceSeverity::High, + description: format!( + "Order on {} has a zero or near-zero limit price ({price_val:.8}). \ + This may indicate an erroneous order or attempted price manipulation.", + order.symbol + ), + remediation: + "Reject or hold order pending manual review of price validity." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(1)), + status: FindingStatus::Open, + }); + has_violation = true; + } + + // In high-volatility markets, large limit orders carry + // elevated manipulation risk + if let Some(mkt) = &context.market_context { + if mkt.volatility > 0.05 && qty > LARGE_ORDER_THRESHOLD { + findings.push(ComplianceFinding { + id: format!("MAR-PRC-HVOL-{}", Uuid::new_v4()), + regulation: "MAR Article 12 - Price Manipulation Risk".to_owned(), + severity: ComplianceSeverity::Medium, + description: format!( + "Large limit order ({qty:.2} units @ {price_val:.4}) on {} \ + during high volatility (vol={:.4}). Elevated manipulation risk.", + order.symbol, mkt.volatility + ), + remediation: "Cross-reference with recent price movements. \ + Verify order is not part of a layering pattern." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(8)), + status: FindingStatus::Open, + }); + warnings.push("High-vol limit order".to_owned()); + } + } + } } - // Ok variant - Ok(ComplianceStatus::Compliant) + // --- Check 3: Suspicious pattern detection --- + // Trading during off-hours or market stress can indicate insider + // knowledge or manipulative intent. + if let Some(mkt) = &context.market_context { + match mkt.session { + TradingSession::PreMarket | TradingSession::AfterHours => { + if let Some(order) = &context.order_info { + let qty = order.quantity.to_f64(); + // Large off-hours orders are more suspicious + if qty > 10_000.0 { + findings.push(ComplianceFinding { + id: format!("MAR-PAT-OOH-{}", Uuid::new_v4()), + regulation: "MAR Article 16 - Off-Hours Surveillance".to_owned(), + severity: ComplianceSeverity::Medium, + description: format!( + "Significant order ({qty:.2} units) on {} placed during \ + {:?} session. Off-hours large orders warrant additional \ + scrutiny for potential insider trading.", + order.symbol, mkt.session + ), + remediation: "Review client trading history for pattern of \ + off-hours activity preceding material events." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(24)), + status: FindingStatus::Open, + }); + warnings.push("Off-hours large order".to_owned()); + } + } + } + TradingSession::Closed => { + // Orders submitted while market is closed are abnormal + if context.order_info.is_some() { + findings.push(ComplianceFinding { + id: format!("MAR-PAT-CLOSED-{}", Uuid::new_v4()), + regulation: "MAR Article 16 - Closed Market Order".to_owned(), + severity: ComplianceSeverity::High, + description: + "Order submitted during closed market session. \ + This is abnormal and may indicate system error or \ + attempted manipulation." + .to_owned(), + remediation: + "Verify order source and system clock synchronization." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(1)), + status: FindingStatus::Open, + }); + has_violation = true; + } + } + TradingSession::Regular => {} + } + + // Stressed market conditions warrant heightened surveillance + if matches!(mkt.conditions, MarketConditions::Stress) { + if let Some(order) = &context.order_info { + findings.push(ComplianceFinding { + id: format!("MAR-PAT-STRESS-{}", Uuid::new_v4()), + regulation: "MAR Article 12 - Stressed Market Activity".to_owned(), + severity: ComplianceSeverity::Medium, + description: format!( + "Order on {} submitted during market stress conditions. \ + Enhanced surveillance required per MAR guidelines.", + order.symbol + ), + remediation: + "Apply enhanced monitoring. Review for potential exploitation \ + of stressed conditions." + .to_owned(), + due_date: Some(Utc::now() + Duration::hours(12)), + status: FindingStatus::Open, + }); + warnings.push("Stressed market order".to_owned()); + } + } + } + + // --- Determine MAR status --- + if has_violation { + Ok(ComplianceStatus::Violation( + warnings + .into_iter() + .chain(std::iter::once("MAR violation detected".to_owned())) + .collect(), + )) + } else if !warnings.is_empty() { + Ok(ComplianceStatus::Warning(warnings)) + } else { + Ok(ComplianceStatus::Compliant) + } } /// Assess data protection compliance diff --git a/trading_engine/src/compliance/sox_compliance.rs b/trading_engine/src/compliance/sox_compliance.rs index 06b5e1f19..39c3e476b 100644 --- a/trading_engine/src/compliance/sox_compliance.rs +++ b/trading_engine/src/compliance/sox_compliance.rs @@ -1584,14 +1584,16 @@ pub enum ReviewStatus { } /// `SOX` Audit Logger -#[derive(Debug)] -/// `SOXAuditLogger` /// -/// Auto-generated documentation placeholder - enhance with specifics +/// Maintains an in-memory audit trail and asynchronously persists events +/// to a JSONL file via a background tokio task for durable SOX compliance. +#[derive(Debug)] #[allow(dead_code)] pub struct SOXAuditLogger { audit_trail: Vec, retention_policy: AuditRetentionPolicy, + /// Async channel sender for persisting audit events to disk + audit_writer: Option>, } /// `SOX` audit event @@ -2108,6 +2110,45 @@ impl AccessControlMatrix { impl SOXAuditLogger { pub fn new(retention_days: &u32) -> Self { + let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::(); + + tokio::spawn(async move { + use tokio::io::AsyncWriteExt; + + // Ensure the audit directory exists + if let Err(e) = tokio::fs::create_dir_all("audit").await { + tracing::error!("Failed to create audit directory: {e}"); + return; + } + + let file = tokio::fs::OpenOptions::new() + .create(true) + .append(true) + .open("audit/sox_audit_trail.jsonl") + .await; + + let mut file = match file { + Ok(f) => f, + Err(e) => { + tracing::error!("Failed to open SOX audit trail file: {e}"); + return; + } + }; + + while let Some(event) = rx.recv().await { + let line = match serde_json::to_string(&event) { + Ok(json) => format!("{json}\n"), + Err(e) => { + tracing::error!("Failed to serialize SOX audit event: {e}"); + continue; + } + }; + if let Err(e) = file.write_all(line.as_bytes()).await { + tracing::error!("Failed to write SOX audit event: {e}"); + } + } + }); + Self { audit_trail: Vec::new(), retention_policy: AuditRetentionPolicy { @@ -2116,6 +2157,7 @@ impl SOXAuditLogger { compression_enabled: true, encryption_required: true, }, + audit_writer: Some(tx), } } @@ -2124,10 +2166,12 @@ impl SOXAuditLogger { // Add to in-memory trail for immediate access self.audit_trail.push(event.clone()); - // TODO: Implement high-performance async persistence - // - Use lock-free ring buffer for HFT compatibility - // - Batch events for efficient disk writes - // - Compress and encrypt per retention policy + // Persist asynchronously via background writer task + if let Some(ref tx) = self.audit_writer { + if let Err(e) = tx.send(event) { + tracing::error!("SOX audit persistence channel closed: {e}"); + } + } Ok(()) }