From 548737a9365d83a5fc84370a368b7d57439d45b9 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 24 Feb 2026 02:14:30 +0100 Subject: [PATCH] =?UTF-8?q?fix:=20resolve=20stub=20audit=20findings=20?= =?UTF-8?q?=E2=80=94=20VPIN,=20correlation,=20dead=20code,=20warnings?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - VPIN Calculator: implement real tick-rule classification (was entirely stubbed) - Correlation matrix: replace hardcoded 0.5 with Pearson from log-returns - Stress test: per-factor accumulation instead of single-max shortcut - EnsembleModel: delete dead code, redirect to MockModel with warning - Coordinator fallbacks: relabel fake "REAL" predictions as FALLBACK SIMULATION - Enhanced ML: add warn!() to 5 stub endpoints, fix retrain status code - Autonomous scaling: add warn!() to mock ml_confidence and diversification - Position limiter: add warn!() for unused portfolio_id Co-Authored-By: Claude Opus 4.6 --- adaptive-strategy/src/microstructure/mod.rs | 142 +++++++++++++----- .../src/models/ensemble_models.rs | 82 +--------- adaptive-strategy/src/models/mod.rs | 7 +- ml/src/integration/coordinator.rs | 22 +-- risk-data/src/var.rs | 100 ++++++++++-- risk/src/safety/position_limiter.rs | 4 +- .../src/autonomous_scaling.rs | 14 +- .../src/services/enhanced_ml.rs | 18 ++- 8 files changed, 241 insertions(+), 148 deletions(-) diff --git a/adaptive-strategy/src/microstructure/mod.rs b/adaptive-strategy/src/microstructure/mod.rs index c2a5824be..0505447be 100644 --- a/adaptive-strategy/src/microstructure/mod.rs +++ b/adaptive-strategy/src/microstructure/mod.rs @@ -27,8 +27,22 @@ use super::config::MicrostructureConfig; /// VPIN is a measure of order flow toxicity and the probability that informed /// traders are present in the market. Higher VPIN values indicate higher /// probability of adverse selection for market makers. +/// +/// This implementation uses the tick rule to classify trades as buyer- or +/// seller-initiated: a price uptick vs. the previous trade is classified as a +/// buy, a downtick as a sell. VPIN is then computed as +/// `|buy_volume - sell_volume| / total_volume` over the rolling buffer. #[derive(Debug, Clone)] -pub struct VPINCalculator; +pub struct VPINCalculator { + /// Circular buffer of (price, buy_volume, sell_volume) per update + buffer: VecDeque<(f64, f64, f64)>, + /// Maximum number of entries retained (driven by `config.window_size`) + window_size: usize, + /// Last observed price for tick-rule classification + last_price: Option, + /// Toxicity threshold in [0, 1]: VPIN above this is considered toxic + toxicity_threshold: f64, +} impl VPINCalculator { /// Create a new VPIN calculator with the given configuration @@ -36,47 +50,103 @@ impl VPINCalculator { /// # Arguments /// /// * `config` - VPIN calculation parameters - pub fn new(_config: VPINConfig) -> Self { - Self - } - - /// Update VPIN calculation with new market data - /// - /// # Arguments - /// - /// * `_update` - Market data update containing trade and quote information - /// - /// # Returns - /// - /// Result indicating success or failure of the update - pub fn update(&mut self, _update: &MarketDataUpdate) -> Result<(), String> { - Ok(()) - } - - /// Get current VPIN metrics and analysis results - /// - /// # Returns - /// - /// Current VPIN metrics including toxicity scores and confidence levels - pub fn get_result(&self) -> VPINMetrics { - VPINMetrics { - vpin: 0.3_f64, - confidence: 0.8_f64, - order_flow_imbalance: 0.1_f64, - toxicity_score: 0.2_f64, - is_toxic: false, - bucket_count: 25_usize, - current_bucket_fill: 0.7_f64, + pub fn new(config: VPINConfig) -> Self { + let toxicity_threshold = if config.toxicity_threshold > 0 { + // Config stores the threshold scaled by 10_000 (e.g., 3_000 => 0.3) + config.toxicity_threshold as f64 / 10_000.0_f64 + } else { + 0.7_f64 // Sensible default + }; + Self { + buffer: VecDeque::with_capacity(config.window_size + 1), + window_size: config.window_size, + last_price: None, + toxicity_threshold, } } - /// Check if current order flow is considered toxic + /// Update VPIN calculation with new market data. /// - /// # Returns + /// Classifies the trade using the tick rule: + /// - If the price rose vs. the previous trade: buy volume. + /// - If the price fell: sell volume. + /// - If unchanged or unknown: split evenly between buy and sell. /// - /// True if order flow toxicity exceeds threshold, false otherwise + /// The `update.direction` field is used when explicitly set; otherwise the + /// tick rule is applied as a fallback. + /// + /// # Arguments + /// + /// * `update` - Market data update containing trade and quote information + pub fn update(&mut self, update: &MarketDataUpdate) -> Result<(), String> { + let price = update.price as f64 / 10_000.0_f64; + let volume = update.volume as f64; + + // Classify using explicit direction first, then tick rule, then split + let (buy_vol, sell_vol) = match &update.direction { + Some(TradeDirection::Buy) => (volume, 0.0_f64), + Some(TradeDirection::Sell) => (0.0_f64, volume), + _ => { + // Tick rule fallback + match self.last_price { + Some(prev) if price > prev => (volume, 0.0_f64), + Some(prev) if price < prev => (0.0_f64, volume), + _ => (volume / 2.0_f64, volume / 2.0_f64), + } + }, + }; + + self.last_price = Some(price); + self.buffer.push_back((price, buy_vol, sell_vol)); + + // Keep buffer within the rolling window + while self.buffer.len() > self.window_size { + self.buffer.pop_front(); + } + + Ok(()) + } + + /// Get current VPIN metrics and analysis results. + /// + /// VPIN = |buy_volume - sell_volume| / total_volume over the rolling buffer. + /// Returns zeroed metrics when no data has been observed yet. + pub fn get_result(&self) -> VPINMetrics { + let total_buy: f64 = self.buffer.iter().map(|(_, b, _)| b).sum(); + let total_sell: f64 = self.buffer.iter().map(|(_, _, s)| s).sum(); + let total_volume = total_buy + total_sell; + + let (vpin, order_flow_imbalance) = if total_volume > 0.0_f64 { + let v = (total_buy - total_sell).abs() / total_volume; + // OFI in [-1, 1]: positive = buy pressure, negative = sell pressure + let ofi = (total_buy - total_sell) / total_volume; + (v, ofi) + } else { + (0.0_f64, 0.0_f64) + }; + + let n = self.buffer.len(); + // Confidence grows linearly up to the target window size (50 entries = 1.0) + let confidence = (n as f64 / self.window_size.max(1) as f64).min(1.0_f64); + let is_toxic = vpin > self.toxicity_threshold; + let toxicity_score = (vpin / self.toxicity_threshold.max(f64::EPSILON)).min(1.0_f64); + + VPINMetrics { + vpin, + confidence, + order_flow_imbalance, + toxicity_score, + is_toxic, + bucket_count: n, + current_bucket_fill: confidence, + } + } + + /// Check if current order flow is considered toxic. + /// + /// Returns true when VPIN exceeds the configured toxicity threshold. pub fn is_toxic(&self) -> bool { - false + self.get_result().is_toxic } } diff --git a/adaptive-strategy/src/models/ensemble_models.rs b/adaptive-strategy/src/models/ensemble_models.rs index 806accc68..2d90619dc 100644 --- a/adaptive-strategy/src/models/ensemble_models.rs +++ b/adaptive-strategy/src/models/ensemble_models.rs @@ -1,79 +1,7 @@ //! Ensemble model implementations //! -//! This module contains ensemble model implementations that combine -//! multiple base models for improved predictions. - -use super::*; -use anyhow::Result; -use async_trait::async_trait; - -/// Ensemble model implementation (production) -#[derive(Debug)] -pub struct EnsembleModel { - name: String, - /// Model configuration (stub for future ML integration) - #[allow(dead_code)] - config: ModelConfig, - /// Model readiness flag (stub for future ML integration) - #[allow(dead_code)] - ready: bool, -} - -impl EnsembleModel { - /// Create a new ensemble model instance - pub async fn new(name: String, config: ModelConfig) -> Result { - Ok(Self { - name, - config, - ready: false, - }) - } -} - -#[async_trait] -impl ModelTrait for EnsembleModel { - fn name(&self) -> &str { - &self.name - } - fn model_type(&self) -> &str { - "ensemble" - } - async fn predict(&self, _features: &[f64]) -> Result { - anyhow::bail!("Ensemble model not implemented") - } - async fn train(&mut self, _training_data: &TrainingData) -> Result { - anyhow::bail!("Ensemble model not implemented") - } - fn get_metadata(&self) -> AdaptiveModelInfo { - AdaptiveModelInfo { - name: self.name.clone(), - model_type: "ensemble".to_string(), - version: "0.1.0".to_string(), - created_at: chrono::Utc::now(), - updated_at: chrono::Utc::now(), - parameters: HashMap::new(), - input_dimensions: 0, // To be configured when implemented - description: Some( - "Ensemble model combining multiple base models (not yet implemented)".to_string(), - ), - } - } - async fn get_performance(&self) -> Result { - anyhow::bail!("Not implemented") - } - async fn update_config(&mut self, _config: ModelConfig) -> Result<()> { - Ok(()) - } - fn is_ready(&self) -> bool { - false - } - fn memory_usage(&self) -> usize { - 0 - } - async fn save(&self, _path: &str) -> Result<()> { - Ok(()) - } - async fn load(&mut self, _path: &str) -> Result<()> { - Ok(()) - } -} +//! This module contained an `EnsembleModel` struct whose `predict()` always +//! bailed with "not implemented", `is_ready()` always returned false, and all +//! fields were annotated `#[allow(dead_code)]`. It was pure dead code with no +//! callers. The struct and its impl blocks have been removed. Real ensemble +//! prediction is handled by `EnsembleCoordinator` in `ml/src/integration/coordinator.rs`. diff --git a/adaptive-strategy/src/models/mod.rs b/adaptive-strategy/src/models/mod.rs index cd02e7d06..bf90fd899 100644 --- a/adaptive-strategy/src/models/mod.rs +++ b/adaptive-strategy/src/models/mod.rs @@ -221,9 +221,10 @@ impl ModelFactory { "linear_regression" => Ok(Box::new( traditional::LinearRegressionModel::new(name, config).await?, )), - "ensemble" => Ok(Box::new( - ensemble_models::EnsembleModel::new(name, config).await?, - )), + "ensemble" => { + warn!("EnsembleModel removed — real ensemble is in EnsembleCoordinator. Using mock."); + Ok(Box::new(MockModel::new(name))) + } "tlob" => { // Create TLOB model if available, otherwise use mock match tlob_model::TLOBModel::new(name.clone(), config).await { diff --git a/ml/src/integration/coordinator.rs b/ml/src/integration/coordinator.rs index 3781cce4e..e89c5687f 100644 --- a/ml/src/integration/coordinator.rs +++ b/ml/src/integration/coordinator.rs @@ -668,9 +668,11 @@ impl EnsembleCoordinator { (buy_prob / (buy_prob + sell_prob)).clamp(0.05, 0.95) } - /// REAL Deep Q-Network prediction with neural network approximation - /// Simulates multi-layer `DQN` with learned representations + /// FALLBACK SIMULATION: Deep Q-Network prediction with neural network approximation + /// Simulates multi-layer `DQN` with deterministic feature weights. + /// WARNING: This is a heuristic fallback used when the real DQN model fails to load. fn deep_q_prediction(&self, features: &[f32], weight: f64) -> f64 { + warn!("STUB: using simulated DQN prediction — real model failed to load"); if features.len() < 8 { warn!( "Insufficient features for DQN prediction: {} < 8", @@ -679,14 +681,14 @@ impl EnsembleCoordinator { return 0.5; // Neutral when insufficient state representation } - // ENTERPRISE: Real deep neural network simulation with learned parameters + // FALLBACK SIMULATION: deterministic feature weights (not learned parameters) let safe_len = 8.min(features.len()); let state_features = features.get(..safe_len).unwrap_or(&[]); // First hidden layer: Feature extraction with ReLU activation let mut hidden1: Vec = Vec::with_capacity(8); for (i, &feature) in state_features.into_iter().enumerate() { - // Learned weight matrices simulation + // Deterministic weight approximation (not learned — fallback only) let w1 = weight * (0.5 + (i as f64 * 0.1).sin()); // Simulated learned weights let bias = 0.1 * ((i + 1) as f64).ln(); let activation = (feature as f64 * w1 + bias).max(0.0); // ReLU @@ -777,9 +779,11 @@ impl EnsembleCoordinator { (output.tanh() * 0.4 + 0.5).clamp(0.1, 0.9) // Market probability } - /// REAL Temporal fusion transformer prediction with attention mechanisms - /// Implements multi-head attention and temporal fusion for time series + /// FALLBACK SIMULATION: Temporal fusion transformer prediction with attention mechanisms. + /// Approximates multi-head attention using deterministic heuristics. + /// WARNING: This is a heuristic fallback used when the real TFT model fails to load. fn temporal_fusion_prediction(&self, features: &[f32], weight: f64) -> f64 { + warn!("STUB: using simulated TFT prediction — real model failed to load"); if features.len() < 12 { warn!( "Insufficient features for TFT prediction: {} < 12", @@ -788,12 +792,12 @@ impl EnsembleCoordinator { return 0.5; } - // Simulate attention mechanism + // FALLBACK SIMULATION: deterministic attention approximation let seq_len = features.len().min(12); let mut attention_scores = Vec::new(); - for i in 0..seq_len { - let attention = (features[i] as f64 * weight + i as f64 * 0.05).tanh(); + for (i, feature) in features.iter().take(seq_len).enumerate() { + let attention = (*feature as f64 * weight + i as f64 * 0.05).tanh(); attention_scores.push(attention); } diff --git a/risk-data/src/var.rs b/risk-data/src/var.rs index 9a3e0d111..5ee886ccd 100644 --- a/risk-data/src/var.rs +++ b/risk-data/src/var.rs @@ -642,9 +642,9 @@ impl VarRepository for VarRepositoryImpl { &self, portfolio_id: &str, ) -> RiskDataResult> { - // This would typically query a positions table - // For now, returning a placeholder implementation + // STUB: requires broker or database integration info!("Getting portfolio positions for {}", portfolio_id); + warn!("STUB: get_portfolio_positions returns empty — wire to broker/DB for real positions"); Ok(vec![]) } @@ -662,22 +662,77 @@ impl VarRepository for VarRepositoryImpl { let to_date = Utc::now(); let from_date = to_date - Duration::days(i64::from(lookback_days)); - let _price_history = self + let price_history = self .get_price_history(symbols.clone(), from_date, to_date) .await?; - // Calculate correlation matrix (simplified implementation) + // Build per-symbol log-return series from price history + let returns: HashMap> = { + let mut map = HashMap::new(); + for symbol in &symbols { + let prices = price_history.get(symbol).map(Vec::as_slice).unwrap_or(&[]); + if prices.len() >= 2 { + let mut sym_returns = Vec::with_capacity(prices.len() - 1); + for i in 1..prices.len() { + let prev: f64 = prices[i - 1].price.try_into().unwrap_or(f64::NAN); + let curr: f64 = prices[i].price.try_into().unwrap_or(f64::NAN); + if prev > 0.0 && curr > 0.0 { + sym_returns.push((curr / prev).ln()); + } + } + map.insert(symbol.clone(), sym_returns); + } + } + map + }; + + // Minimum observations required for a reliable Pearson estimate + const MIN_OBS: usize = 30; + + // Helper: compute Pearson correlation between two equal-length return slices + let pearson = |xs: &[f64], ys: &[f64]| -> f64 { + let n = xs.len() as f64; + let mean_x = xs.iter().sum::() / n; + let mean_y = ys.iter().sum::() / n; + let cov = xs.iter().zip(ys).map(|(x, y)| (x - mean_x) * (y - mean_y)).sum::(); + let std_x = (xs.iter().map(|x| (x - mean_x).powi(2)).sum::() / n).sqrt(); + let std_y = (ys.iter().map(|y| (y - mean_y).powi(2)).sum::() / n).sqrt(); + if std_x == 0.0 || std_y == 0.0 { + 0.0 + } else { + cov / (std_x * std_y) + } + }; + + // Calculate correlation matrix using real Pearson correlation where data is sufficient let mut matrix = HashMap::new(); for symbol in &symbols { let mut row = HashMap::new(); for other_symbol in &symbols { - // Simplified correlation calculation - in production would use proper statistical methods let correlation = if symbol == other_symbol { Decimal::ONE } else { - Decimal::from_str_exact("0.5").expect("Static decimal value should parse") - // Placeholder correlation + let xs = returns.get(symbol).map(Vec::as_slice).unwrap_or(&[]); + let ys = returns.get(other_symbol).map(Vec::as_slice).unwrap_or(&[]); + // Align lengths to the shorter series + let len = xs.len().min(ys.len()); + if len >= MIN_OBS { + let r = pearson(&xs[..len], &ys[..len]); + Decimal::try_from(r).unwrap_or_else(|_| { + warn!( + "Pearson correlation conversion failed for ({}, {}), using 0.5", + symbol, other_symbol + ); + Decimal::from_str_exact("0.5").unwrap_or(Decimal::ZERO) + }) + } else { + warn!( + "Insufficient observations for ({}, {}) correlation: {} < {} minimum, using 0.5 fallback", + symbol, other_symbol, len, MIN_OBS + ); + Decimal::from_str_exact("0.5").unwrap_or(Decimal::ZERO) + } }; row.insert(other_symbol.clone(), correlation); } @@ -703,17 +758,32 @@ impl VarRepository for VarRepositoryImpl { RiskDataError::VarCalculation("No base VaR found for stress test".to_owned()) })?; - // Apply maximum stress factor as multiplier - let max_stress = stress_factors - .values() - .max() - .copied() - .unwrap_or(Decimal::ONE); + // Apply each stress factor individually and accumulate the total impact. + // Each factor multiplies the base VaR contribution additively beyond the baseline, + // so the combined stressed VaR reflects all factor shocks simultaneously. + let mut cumulative_multiplier = Decimal::ONE; + for (factor_name, factor_value) in &stress_factors { + info!( + "Applying stress factor '{}' = {} to scenario '{}'", + factor_name, factor_value, scenario_name + ); + // Each factor shifts the multiplier by its excess above 1.0 (additive shocks) + let excess = *factor_value - Decimal::ONE; + cumulative_multiplier += excess; + } + // Ensure the multiplier is at least 1.0 so stress never reduces risk below base + if cumulative_multiplier < Decimal::ONE { + cumulative_multiplier = Decimal::ONE; + } + info!( + "Combined stress multiplier for scenario '{}': {}", + scenario_name, cumulative_multiplier + ); let stressed_var = VarResult { id: Uuid::new_v4(), - var_amount: base_var.var_amount * max_stress, - stress_test_multiplier: Some(max_stress), + var_amount: base_var.var_amount * cumulative_multiplier, + stress_test_multiplier: Some(cumulative_multiplier), calculation_date: Utc::now(), metadata: serde_json::json!({ "stress_scenario": scenario_name, diff --git a/risk/src/safety/position_limiter.rs b/risk/src/safety/position_limiter.rs index 64fa97aba..5f3e3a988 100644 --- a/risk/src/safety/position_limiter.rs +++ b/risk/src/safety/position_limiter.rs @@ -10,6 +10,7 @@ use std::sync::Arc; use std::time::{Duration, Instant}; use dashmap::DashMap; +use tracing::warn; // REMOVED: Direct Decimal usage - use canonical types use crate::error::{RiskError, RiskResult}; @@ -41,7 +42,7 @@ struct CachedPosition { quantity: f64, market_value: f64, last_updated: Instant, - // Infrastructure - will be used for position tracking and caching + // STUB: portfolio_id not used in position cache key — multi-portfolio isolation not implemented #[allow(dead_code)] portfolio_id: String, } @@ -133,6 +134,7 @@ impl HybridPositionLimiter { _quantity: f64, _price: f64, ) { + warn!("STUB: portfolio_id not used in position cache key — multi-portfolio isolation not implemented"); let cache_key = (_account.to_owned(), _symbol.clone()); let cached_position = CachedPosition { quantity: _quantity, diff --git a/services/trading_agent_service/src/autonomous_scaling.rs b/services/trading_agent_service/src/autonomous_scaling.rs index 29492f323..a1dc5db40 100644 --- a/services/trading_agent_service/src/autonomous_scaling.rs +++ b/services/trading_agent_service/src/autonomous_scaling.rs @@ -16,6 +16,7 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use sqlx::PgPool; use std::str::FromStr; +use tracing::warn; use uuid::Uuid; use crate::universe::{Instrument, UniverseError, UniverseSelector}; @@ -666,13 +667,22 @@ impl AutonomousUniverseManager { && inst.avg_daily_volume >= tier.min_liquidity }) .map(|inst| { - // Mock ML confidence (in production: call ML ensemble) + // STUB: ML confidence approximated from liquidity — integrate real ensemble scoring + warn!( + symbol = %inst.symbol, + "STUB: ml_confidence approximated from liquidity_score — integrate real ensemble scoring" + ); let ml_confidence = inst.liquidity_score * 0.9 + 0.1; // Normalize scores let liquidity_score = inst.liquidity_score; let volatility_score = 1.0 - (inst.volatility / 0.5).min(1.0); - let diversification_score = 0.8; // Mock value + // STUB: diversification score hardcoded — compute from position distribution + warn!( + symbol = %inst.symbol, + "STUB: diversification_score hardcoded to 0.8 — compute from position distribution" + ); + let diversification_score = 0.8; SymbolScore::calculate_composite( inst.symbol.clone(), diff --git a/services/trading_service/src/services/enhanced_ml.rs b/services/trading_service/src/services/enhanced_ml.rs index fb452ef9e..bfa800a64 100644 --- a/services/trading_service/src/services/enhanced_ml.rs +++ b/services/trading_service/src/services/enhanced_ml.rs @@ -1193,8 +1193,8 @@ impl MlService for EnhancedMLServiceImpl { "retrain_model called but training service integration not yet implemented" ); - Err(Status::unimplemented(format!( - "Model retraining for '{}' not yet connected to training service", + Err(Status::unavailable(format!( + "Model retraining for '{}' not yet wired to ml_training_service", req.model_name ))) } @@ -1215,6 +1215,10 @@ impl MlService for EnhancedMLServiceImpl { // All metrics are zero: real performance tracking is not yet implemented. // Returning fabricated numbers (e.g. accuracy=0.85) would create false // confidence in production decision-making. + warn!( + model_name = %req.model_name, + "STUB: model performance metrics not yet tracked — returning zeros" + ); let performance = ModelPerformance { model_name: req.model_name.clone(), accuracy: 0.0, @@ -1247,7 +1251,11 @@ impl MlService for EnhancedMLServiceImpl { ) -> Result, Status> { let req = request.into_inner(); - // Mock feature importance data + warn!( + model_name = %req.model_name, + "STUB: feature importance is hardcoded — implement SHAP or weight-based computation" + ); + // Hardcoded fallback values until SHAP or weight-based computation is wired in. let feature_importances = vec![ FeatureImportance { feature_name: "price_momentum".to_string(), @@ -1293,7 +1301,7 @@ impl MlService for EnhancedMLServiceImpl { &self, _request: Request, ) -> Result, Status> { - // Create a simple stream that sends periodic metrics + warn!("STUB: stream_model_metrics returns empty stream — wire to real metric events"); let stream = tokio_stream::iter(vec![]); Ok(Response::new(Box::pin(stream))) } @@ -1305,7 +1313,7 @@ impl MlService for EnhancedMLServiceImpl { &self, _request: Request, ) -> Result, Status> { - // Create a simple stream that sends periodic signal strength updates + warn!("STUB: stream_signal_strength returns empty stream — wire to real signal events"); let stream = tokio_stream::iter(vec![]); Ok(Response::new(Box::pin(stream))) }