From 2c100a36fa652b6138e1e92c5f344b0fa5650c36 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 22 Feb 2026 01:16:53 +0100 Subject: [PATCH] safety: fix compliance stubs, audit fallback, and dummy ML features - Compliance: return Uuid::nil() + warn! when features disabled instead of random untraceable UUIDs (SOX, MiFID II, position monitoring, best execution analysis) - Audit queue: change fallback path from /tmp/ to /var/lib/foxhunt/, upgrade fallback log from info to warn for alerting visibility - Enhanced ML: wire get_ensemble_vote gRPC handler to real ensemble coordinator instead of hardcoded [0.1, 0.2, -0.05, 0.8] features - Paper trading: change stub confidence from 0.5 to 0.0 (below 0.6 threshold, preventing accidental orders) and add warn! log Co-Authored-By: Claude Opus 4.6 --- .../trading_service/src/async_audit_queue.rs | 10 +- .../trading_service/src/compliance_service.rs | 12 ++- .../src/ensemble_coordinator.rs | 2 +- .../src/paper_trading_executor.rs | 13 ++- .../src/services/enhanced_ml.rs | 99 +++++++++++++++++-- 5 files changed, 118 insertions(+), 18 deletions(-) diff --git a/services/trading_service/src/async_audit_queue.rs b/services/trading_service/src/async_audit_queue.rs index d524853a4..2670ff0f9 100644 --- a/services/trading_service/src/async_audit_queue.rs +++ b/services/trading_service/src/async_audit_queue.rs @@ -46,7 +46,7 @@ impl Default for AuditQueueConfig { buffer_size: 10_000, batch_size: 100, flush_interval: Duration::from_secs(1), - fallback_path: "/tmp/foxhunt_audit_fallback.jsonl".to_string(), + fallback_path: "/var/lib/foxhunt/audit_fallback.jsonl".to_string(), max_retries: 3, } } @@ -308,10 +308,10 @@ async fn write_batch_to_fallback( match write_to_disk(batch, fallback_path).await { Ok(_) => { metrics.record_fallback(); - info!( - "Wrote {} events to fallback file: {}", - batch.len(), - fallback_path + warn!( + fallback_path = %fallback_path, + event_count = batch.len(), + "Primary audit queue failed, writing to fallback file. Investigate queue health immediately." ); } Err(e) => { diff --git a/services/trading_service/src/compliance_service.rs b/services/trading_service/src/compliance_service.rs index 90f646800..dcde3f6c8 100644 --- a/services/trading_service/src/compliance_service.rs +++ b/services/trading_service/src/compliance_service.rs @@ -129,7 +129,8 @@ impl ComplianceService { /// Required for Sarbanes-Oxley Section 404 internal controls pub async fn log_sox_trade_audit(&self, data: &TradeExecutionData) -> Result { if !self.config.enable_sox_audit { - return Ok(Uuid::new_v4()); // Return dummy ID if disabled + warn!(feature = "sox_audit", "Compliance feature disabled — returning nil audit ID. Configure enable_sox_audit=true for production."); + return Ok(Uuid::nil()); } let start_time = Instant::now(); @@ -172,7 +173,8 @@ impl ComplianceService { /// Required for European Markets in Financial Instruments Directive II pub async fn create_mifid_report(&self, data: &MiFidTransactionData) -> Result { if !self.config.enable_mifid_reporting { - return Ok(Uuid::new_v4()); // Return dummy ID if disabled + warn!(feature = "mifid_reporting", "Compliance feature disabled — returning nil report ID. Configure enable_mifid_reporting=true for production."); + return Ok(Uuid::nil()); } let start_time = Instant::now(); @@ -210,7 +212,8 @@ impl ComplianceService { /// Required for MiFID II Article 57 position limits pub async fn check_position_limits(&self, data: &PositionLimitData) -> Result<(Uuid, bool)> { if !self.config.enable_position_monitoring { - return Ok((Uuid::new_v4(), false)); // Return no breach if disabled + warn!(feature = "position_monitoring", "Compliance feature disabled — returning nil audit ID. Configure enable_position_monitoring=true for production."); + return Ok((Uuid::nil(), false)); } let start_time = Instant::now(); @@ -345,7 +348,8 @@ impl ComplianceService { /// Analyze best execution for MiFID II Article 27 compliance pub async fn analyze_best_execution(&self, data: &BestExecutionData) -> Result { if !self.config.enable_best_execution_analysis { - return Ok(Uuid::new_v4()); // Return dummy ID if disabled + warn!(feature = "best_execution_analysis", "Compliance feature disabled — returning nil analysis ID. Configure enable_best_execution_analysis=true for production."); + return Ok(Uuid::nil()); } let start_time = Instant::now(); diff --git a/services/trading_service/src/ensemble_coordinator.rs b/services/trading_service/src/ensemble_coordinator.rs index a9595b557..7242594df 100644 --- a/services/trading_service/src/ensemble_coordinator.rs +++ b/services/trading_service/src/ensemble_coordinator.rs @@ -676,7 +676,7 @@ impl EnsembleCoordinator { /// (minimum ~51 updates), extraction will fail gracefully and a zero-filled /// feature vector is returned with a warning log so the ensemble can still /// produce a low-confidence prediction. - async fn fetch_features_for_symbol(&self, symbol: &str) -> Result { + pub async fn fetch_features_for_symbol(&self, symbol: &str) -> Result { use common::ml_strategy::ProductionFeatureExtractor225; let names: Vec = FEATURE_NAMES_51.iter().map(|s| (*s).to_string()).collect(); diff --git a/services/trading_service/src/paper_trading_executor.rs b/services/trading_service/src/paper_trading_executor.rs index 3dc4a4ca7..5cba70ba0 100644 --- a/services/trading_service/src/paper_trading_executor.rs +++ b/services/trading_service/src/paper_trading_executor.rs @@ -187,14 +187,23 @@ impl PaperTradingExecutor { } /// Generate ML signal from market data using SharedMLStrategy + /// + /// **WARNING**: This method currently returns a synthetic Hold/0.0-confidence + /// stub because the `SharedMLStrategy` feature-extraction pipeline has not + /// been wired to the ensemble coordinator yet. The zero confidence ensures + /// downstream consumers (e.g. `convert_signal_to_order`) will never treat + /// this as a real, actionable signal. pub async fn generate_ml_signal( &self, _market_data: &[(f64, f64, f64, f64, f64)], ) -> Result { - // Use shared ML strategy (stub - will be implemented with real ensemble) + warn!( + "Paper trading using stub ML signal — ensemble coordinator not wired. \ + Results are not representative of live trading." + ); Ok(TradingSignal { action: Some(Action::Hold), - confidence: 0.5, + confidence: 0.0, source: SignalSource::ML, model_votes: None, }) diff --git a/services/trading_service/src/services/enhanced_ml.rs b/services/trading_service/src/services/enhanced_ml.rs index 2aa397f13..3fb63805d 100644 --- a/services/trading_service/src/services/enhanced_ml.rs +++ b/services/trading_service/src/services/enhanced_ml.rs @@ -983,18 +983,105 @@ impl MlService for EnhancedMLServiceImpl { ) -> Result, Status> { let req = request.into_inner(); - // For this example, we'll use dummy features. In production, these would come from market data - let features = vec![0.1, 0.2, -0.05, 0.8]; // price_momentum, volume, spread, volatility + // Use the ensemble coordinator to fetch real market-data features and + // produce a genuine ensemble prediction. If the coordinator is not + // configured the handler must not fall back to hardcoded data. + let coordinator = self + .state + .ensemble_coordinator() + .ok_or_else(|| { + Status::unavailable( + "Ensemble coordinator is not configured; \ + cannot produce predictions without live market data", + ) + })? + .clone(); - let ensemble_vote = self.get_ensemble_prediction(&features, &req.symbol).await?; + let features = coordinator + .fetch_features_for_symbol(&req.symbol) + .await + .map_err(|e| { + Status::failed_precondition(format!( + "Failed to fetch market features for {}: {}. \ + Ensure market data is being fed via update_market_data()", + req.symbol, e, + )) + })?; - // Get individual votes (already calculated in ensemble prediction) - let individual_votes = Vec::new(); // Would be populated from ensemble_prediction + let decision = coordinator.predict(&features).await.map_err(|e| { + Status::internal(format!("Ensemble prediction failed: {}", e)) + })?; + + // Convert ml::ensemble::ModelVote -> proto ModelVote + let individual_votes: Vec = decision + .model_votes + .iter() + .map(|(model_id, vote)| { + let prediction_type = if vote.signal > 0.3 { + PredictionType::Buy + } else if vote.signal < -0.3 { + PredictionType::Sell + } else { + PredictionType::Hold + }; + ModelVote { + model_name: model_id.clone(), + prediction: prediction_type as i32, + confidence: vote.confidence, + weight: vote.weight, + } + }) + .collect(); + + // Map TradingAction to proto PredictionType for the consensus vote + let consensus_prediction = match decision.action { + ml::ensemble::TradingAction::Buy => PredictionType::Buy, + ml::ensemble::TradingAction::Sell => PredictionType::Sell, + ml::ensemble::TradingAction::Hold => PredictionType::Hold, + }; + + // Determine signal strength from model agreement + let total_models = decision.model_votes.len() as i32; + let buy_votes = decision + .model_votes + .values() + .filter(|v| v.signal > 0.3) + .count() as i32; + let sell_votes = decision + .model_votes + .values() + .filter(|v| v.signal < -0.3) + .count() as i32; + let hold_votes = total_models - buy_votes - sell_votes; + + let max_votes = buy_votes.max(sell_votes).max(hold_votes); + let signal_strength = if total_models > 0 { + match max_votes as f64 / total_models as f64 { + ratio if ratio >= 0.8 => crate::proto::ml::SignalStrength::VeryStrong, + ratio if ratio >= 0.6 => crate::proto::ml::SignalStrength::Strong, + ratio if ratio >= 0.4 => crate::proto::ml::SignalStrength::Moderate, + ratio if ratio >= 0.3 => crate::proto::ml::SignalStrength::Weak, + _ => crate::proto::ml::SignalStrength::VeryWeak, + } + } else { + crate::proto::ml::SignalStrength::VeryWeak + }; + + let ensemble_vote = EnsembleVote { + symbol: req.symbol, + consensus_prediction: consensus_prediction as i32, + consensus_confidence: decision.confidence, + votes_buy: buy_votes, + votes_sell: sell_votes, + votes_hold: hold_votes, + total_models, + signal_strength: signal_strength as i32, + }; Ok(Response::new(GetEnsembleVoteResponse { ensemble_vote: Some(ensemble_vote), individual_votes, - overall_confidence: 0.85, + overall_confidence: decision.confidence, timestamp: SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap_or_default()