abcbf84509
Replace empty MLEngine struct with actual EnsembleCoordinator from ml crate. Registers DQN, PPO, TFT, Mamba2 models with configurable weights loaded from config repository. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1399 lines
52 KiB
Rust
1399 lines
52 KiB
Rust
//! Service state management and business logic coordination
|
|
//!
|
|
//! This module provides clean repository-based dependency injection,
|
|
//! eliminating direct database coupling from business logic.
|
|
|
|
use crate::error::TradingServiceResult;
|
|
use crate::event_persistence::EventPersistence;
|
|
use crate::event_streaming::publisher::EventPublisher;
|
|
use crate::proto::monitoring::SystemMetrics;
|
|
use crate::repositories::*;
|
|
use crate::repository_impls::PostgresConfigRepository;
|
|
use trading_engine::trading::account_manager::AccountManager;
|
|
use trading_engine::trading::order_manager::OrderManager;
|
|
use trading_engine::trading::position_manager::PositionManager;
|
|
|
|
// Import provider traits for data connection methods
|
|
use data::providers::{MarketDataProvider, RealTimeProvider};
|
|
|
|
use futures::StreamExt;
|
|
use std::sync::Arc;
|
|
use tokio::sync::broadcast;
|
|
use tokio::sync::RwLock;
|
|
|
|
/// Central state manager for the trading service with repository pattern
|
|
///
|
|
/// This struct coordinates all business logic using repository dependency injection:
|
|
/// - Trading operations through TradingRepository
|
|
///
|
|
/// - Market data access through MarketDataRepository
|
|
/// - Risk management through RiskRepository
|
|
///
|
|
/// - Configuration through ConfigRepository
|
|
/// - NO DIRECT DATABASE COUPLING
|
|
#[derive(Clone)]
|
|
pub struct TradingServiceState {
|
|
/// Trading repository for orders, executions, positions
|
|
pub trading_repository: Arc<dyn TradingRepository>,
|
|
|
|
/// Market data repository for prices, order books
|
|
pub market_data_repository: Arc<dyn MarketDataRepository>,
|
|
|
|
/// Risk repository for limits, calculations, alerts
|
|
pub risk_repository: Arc<dyn RiskRepository>,
|
|
|
|
/// Configuration repository for settings and secrets
|
|
pub config_repository: Arc<crate::repository_impls::PostgresConfigRepository>,
|
|
|
|
/// Database connection pool for direct SQL queries
|
|
pub db_pool: sqlx::PgPool,
|
|
|
|
/// Risk management engine (business logic only)
|
|
pub risk_engine: Arc<RwLock<RiskEngine>>,
|
|
|
|
/// ML model registry and predictor (business logic only)
|
|
pub ml_engine: Arc<RwLock<MLEngine>>,
|
|
|
|
/// Market data manager (business logic only)
|
|
pub market_data: Arc<RwLock<MarketDataManager>>,
|
|
|
|
/// `Order` management system (business logic only)
|
|
pub order_manager: Arc<RwLock<OrderManager>>,
|
|
|
|
/// `Position` tracking (business logic only)
|
|
pub position_manager: Arc<RwLock<PositionManager>>,
|
|
|
|
/// Account management (business logic only)
|
|
pub account_manager: Arc<RwLock<AccountManager>>,
|
|
|
|
/// Event publisher for real-time streaming
|
|
pub event_publisher: Arc<EventPublisher>,
|
|
|
|
/// Event persistence for compliance and audit trail
|
|
pub event_persistence: Arc<EventPersistence>,
|
|
|
|
/// System metrics and monitoring
|
|
pub metrics: Arc<RwLock<SystemMetrics>>,
|
|
|
|
/// Kill switch system for emergency shutdown
|
|
pub kill_switch_system: Option<Arc<crate::kill_switch_integration::TradingServiceKillSwitch>>,
|
|
|
|
/// Ensemble coordinator for ML predictions (DQN, PPO, TFT)
|
|
pub ensemble_coordinator: Option<Arc<crate::ensemble_coordinator::EnsembleCoordinator>>,
|
|
}
|
|
|
|
impl std::fmt::Debug for TradingServiceState {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("TradingServiceState")
|
|
.field("trading_repository", &"<dyn TradingRepository>")
|
|
.field("market_data_repository", &"<dyn MarketDataRepository>")
|
|
.field("risk_repository", &"<dyn RiskRepository>")
|
|
.field("config_repository", &self.config_repository)
|
|
.field("db_pool", &"<PgPool>")
|
|
.field("risk_engine", &self.risk_engine)
|
|
.field("ml_engine", &self.ml_engine)
|
|
.field("market_data", &self.market_data)
|
|
.field("order_manager", &self.order_manager)
|
|
.field("position_manager", &self.position_manager)
|
|
.field("account_manager", &self.account_manager)
|
|
.field("event_publisher", &self.event_publisher)
|
|
.field("event_persistence", &self.event_persistence)
|
|
.field("metrics", &self.metrics)
|
|
.field("kill_switch_system", &self.kill_switch_system)
|
|
.field("ensemble_coordinator", &self.ensemble_coordinator)
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl TradingServiceState {
|
|
/// Create new trading service state with repository dependency injection
|
|
pub async fn new_with_repositories(
|
|
trading_repository: Arc<dyn TradingRepository>,
|
|
market_data_repository: Arc<dyn MarketDataRepository>,
|
|
risk_repository: Arc<dyn RiskRepository>,
|
|
config_repository: Arc<PostgresConfigRepository>,
|
|
db_pool: sqlx::PgPool,
|
|
event_persistence: Arc<EventPersistence>,
|
|
kill_switch_system: Option<Arc<crate::kill_switch_integration::TradingServiceKillSwitch>>,
|
|
ensemble_coordinator: Option<Arc<crate::ensemble_coordinator::EnsembleCoordinator>>,
|
|
) -> TradingServiceResult<Self> {
|
|
// Initialize business logic components (no database coupling)
|
|
let risk_engine = Arc::new(RwLock::new(RiskEngine::new()));
|
|
let ml_engine = Arc::new(RwLock::new(MLEngine::new()));
|
|
let market_data = Arc::new(RwLock::new(MarketDataManager::new()));
|
|
let order_manager = Arc::new(RwLock::new(OrderManager::new()));
|
|
let position_manager = Arc::new(RwLock::new(PositionManager::new()));
|
|
let account_manager = Arc::new(RwLock::new(AccountManager::new()));
|
|
// Create broadcast channel for event publishing
|
|
let (event_sender, _) = broadcast::channel(10000);
|
|
let event_publisher = Arc::new(EventPublisher::new(event_sender));
|
|
let metrics = Arc::new(RwLock::new(SystemMetrics::default()));
|
|
|
|
Ok(Self {
|
|
trading_repository,
|
|
market_data_repository,
|
|
risk_repository,
|
|
config_repository,
|
|
db_pool,
|
|
event_persistence,
|
|
risk_engine,
|
|
ml_engine,
|
|
market_data,
|
|
order_manager,
|
|
position_manager,
|
|
account_manager,
|
|
event_publisher,
|
|
metrics,
|
|
kill_switch_system,
|
|
ensemble_coordinator,
|
|
})
|
|
}
|
|
|
|
/// Initialize service state with repository-based configuration
|
|
pub async fn initialize(&self) -> TradingServiceResult<()> {
|
|
// Initialize risk engine with repository-based configuration
|
|
let mut risk_engine = self.risk_engine.write().await;
|
|
risk_engine
|
|
.initialize_with_config_repository(&self.config_repository)
|
|
.await?;
|
|
|
|
// Initialize ML engine with repository-based configuration
|
|
let mut ml_engine = self.ml_engine.write().await;
|
|
ml_engine
|
|
.initialize_with_config_repository(&self.config_repository)
|
|
.await?;
|
|
|
|
// Initialize market data connections with repository-based configuration
|
|
let mut market_data = self.market_data.write().await;
|
|
market_data
|
|
.initialize_with_config_repository(&self.config_repository)
|
|
.await?;
|
|
|
|
// Start event processing for market data providers
|
|
market_data.start_event_processing().await?;
|
|
|
|
Ok(())
|
|
}
|
|
/// Create new trading service state for testing purposes
|
|
///
|
|
/// This creates a minimal state with mock repositories suitable for integration tests.
|
|
///
|
|
/// Note: This function is only available when building tests.
|
|
pub async fn new_for_testing() -> TradingServiceResult<Self> {
|
|
// For testing, create a minimal repository setup using PostgreSQL implementations
|
|
use crate::event_persistence::EventPersistence;
|
|
use crate::repository_impls::{
|
|
PostgresConfigRepository, PostgresMarketDataRepository, PostgresRiskRepository,
|
|
PostgresTradingRepository,
|
|
};
|
|
use sqlx::PgPool;
|
|
|
|
// Get database URL for testing
|
|
let db_url = std::env::var("DATABASE_URL").unwrap_or_else(|_| {
|
|
"postgresql://foxhunt:foxhunt_dev_password@localhost:5432/foxhunt".to_string()
|
|
});
|
|
|
|
let pool = PgPool::connect(&db_url).await.map_err(|e| {
|
|
crate::error::TradingServiceError::Internal {
|
|
message: format!("Failed to create test database pool: {}", e),
|
|
}
|
|
})?;
|
|
|
|
// Create PostgreSQL repositories for testing
|
|
let trading_repository =
|
|
Arc::new(PostgresTradingRepository::new(pool.clone())) as Arc<dyn TradingRepository>;
|
|
let market_data_repository = Arc::new(PostgresMarketDataRepository::new(pool.clone()))
|
|
as Arc<dyn MarketDataRepository>;
|
|
let risk_repository =
|
|
Arc::new(PostgresRiskRepository::new(pool.clone())) as Arc<dyn RiskRepository>;
|
|
|
|
// Create config repository using the same pool
|
|
let config_repository = Arc::new(PostgresConfigRepository::new(pool.clone()));
|
|
|
|
// Create event persistence with a test directory
|
|
let event_persistence =
|
|
Arc::new(EventPersistence::new_for_testing().await.map_err(|e| {
|
|
crate::error::TradingServiceError::Internal {
|
|
message: format!("Failed to create event persistence: {}", e),
|
|
}
|
|
})?);
|
|
|
|
// Create state without kill switch, model cache, or ensemble for simplicity
|
|
Self::new_with_repositories(
|
|
trading_repository,
|
|
market_data_repository,
|
|
risk_repository,
|
|
config_repository,
|
|
pool,
|
|
event_persistence,
|
|
None, // kill_switch_system
|
|
None, // ensemble_coordinator
|
|
)
|
|
.await
|
|
}
|
|
|
|
/// Get health status of all components
|
|
pub async fn get_health_status(&self) -> HealthStatus {
|
|
// Check all components and return overall health
|
|
let market_data_health = self.market_data.read().await.get_provider_health().await;
|
|
|
|
// Check if any providers are unhealthy
|
|
let has_unhealthy_providers = market_data_health
|
|
.iter()
|
|
.any(|(_, status)| !status.connected);
|
|
|
|
// Check ensemble coordinator health
|
|
let ensemble_health = self.check_ensemble_health().await;
|
|
|
|
if has_unhealthy_providers {
|
|
HealthStatus::Degraded
|
|
} else if market_data_health.is_empty() {
|
|
HealthStatus::Critical // No providers available
|
|
} else if matches!(ensemble_health, EnsembleHealth::Unhealthy) {
|
|
HealthStatus::Degraded // Ensemble issues but trading can continue
|
|
} else {
|
|
HealthStatus::Healthy
|
|
}
|
|
}
|
|
|
|
/// Check ensemble coordinator health
|
|
async fn check_ensemble_health(&self) -> EnsembleHealth {
|
|
use tracing::{info, warn};
|
|
|
|
let ensemble = match &self.ensemble_coordinator {
|
|
Some(coordinator) => coordinator,
|
|
None => {
|
|
info!("Ensemble coordinator not configured");
|
|
return EnsembleHealth::NotConfigured;
|
|
},
|
|
};
|
|
|
|
// Check if models are loaded
|
|
let model_count = ensemble.model_count().await;
|
|
if model_count == 0 {
|
|
warn!("Ensemble coordinator has no models loaded");
|
|
return EnsembleHealth::Unhealthy;
|
|
}
|
|
|
|
// Mock health check - in production, this would:
|
|
// 1. Verify all 6 models are loaded (DQN, PPO, TFT x 2 each)
|
|
// 2. Check inference latency (<50μs P99)
|
|
// 3. Check model staleness (last updated)
|
|
// 4. Verify checkpoint integrity
|
|
|
|
info!("Ensemble coordinator healthy with {} models", model_count);
|
|
EnsembleHealth::Healthy
|
|
}
|
|
|
|
/// Get detailed ensemble health report
|
|
pub async fn get_ensemble_health_report(&self) -> EnsembleHealthReport {
|
|
let ensemble = match &self.ensemble_coordinator {
|
|
Some(coordinator) => coordinator,
|
|
None => {
|
|
return EnsembleHealthReport {
|
|
status: EnsembleHealth::NotConfigured,
|
|
models_loaded: 0,
|
|
expected_models: 6, // DQN, PPO, TFT (2 checkpoints each)
|
|
inference_latency_us: None,
|
|
last_prediction: None,
|
|
model_details: vec![],
|
|
};
|
|
},
|
|
};
|
|
|
|
let model_count = ensemble.model_count().await;
|
|
let status = if model_count == 0 {
|
|
EnsembleHealth::Unhealthy
|
|
} else if model_count < 6 {
|
|
EnsembleHealth::Degraded
|
|
} else {
|
|
EnsembleHealth::Healthy
|
|
};
|
|
|
|
EnsembleHealthReport {
|
|
status,
|
|
models_loaded: model_count,
|
|
expected_models: 6,
|
|
inference_latency_us: Some(42.0), // Mock latency
|
|
last_prediction: Some(std::time::SystemTime::now()),
|
|
model_details: vec![
|
|
ModelHealthDetail {
|
|
model_id: "DQN".to_string(),
|
|
checkpoint: "epoch_30".to_string(),
|
|
loaded: true,
|
|
last_inference: std::time::SystemTime::now(),
|
|
},
|
|
ModelHealthDetail {
|
|
model_id: "PPO".to_string(),
|
|
checkpoint: "epoch_30".to_string(),
|
|
loaded: true,
|
|
last_inference: std::time::SystemTime::now(),
|
|
},
|
|
ModelHealthDetail {
|
|
model_id: "TFT".to_string(),
|
|
checkpoint: "epoch_30".to_string(),
|
|
loaded: true,
|
|
last_inference: std::time::SystemTime::now(),
|
|
},
|
|
],
|
|
}
|
|
}
|
|
|
|
/// Subscribe to market data for trading symbols
|
|
pub async fn subscribe_to_market_data(
|
|
&self,
|
|
symbols: Vec<common::types::Symbol>,
|
|
) -> TradingServiceResult<()> {
|
|
let mut market_data = self.market_data.write().await;
|
|
market_data.subscribe_to_symbols(symbols).await
|
|
}
|
|
|
|
/// Get market data event stream
|
|
pub async fn get_market_data_stream(
|
|
&self,
|
|
) -> tokio::sync::broadcast::Receiver<trading_engine::trading::data_interface::MarketDataEvent>
|
|
{
|
|
let market_data = self.market_data.read().await;
|
|
market_data.get_event_receiver()
|
|
}
|
|
|
|
/// Get ensemble coordinator reference (if initialized)
|
|
pub fn ensemble_coordinator(
|
|
&self,
|
|
) -> Option<&Arc<crate::ensemble_coordinator::EnsembleCoordinator>> {
|
|
self.ensemble_coordinator.as_ref()
|
|
}
|
|
|
|
/// Get trading signal from ensemble coordinator
|
|
///
|
|
/// This method orchestrates the full prediction flow:
|
|
/// 1. Extract features from market data
|
|
/// 2. Call ensemble coordinator for prediction
|
|
/// 3. Convert ensemble decision to trading signal
|
|
/// 4. Apply risk adjustments (position sizing)
|
|
/// 5. Return signal with ensemble attribution
|
|
///
|
|
/// Fallback: If ensemble fails, falls back to single model (DQN epoch 30)
|
|
pub async fn get_ensemble_trading_signal(
|
|
&self,
|
|
symbol: &str,
|
|
) -> TradingServiceResult<EnsembleTradingSignal> {
|
|
use tracing::{info, warn};
|
|
|
|
// Check if ensemble coordinator is available
|
|
let ensemble = match &self.ensemble_coordinator {
|
|
Some(coordinator) => coordinator,
|
|
None => {
|
|
warn!("Ensemble coordinator not initialized, using fallback strategy");
|
|
return self.get_fallback_trading_signal(symbol).await;
|
|
},
|
|
};
|
|
|
|
// Extract features from market data
|
|
let features = match self.extract_features_for_symbol(symbol).await {
|
|
Ok(f) => f,
|
|
Err(e) => {
|
|
warn!(
|
|
"Feature extraction failed for {}: {}, using fallback",
|
|
symbol, e
|
|
);
|
|
return self.get_fallback_trading_signal(symbol).await;
|
|
},
|
|
};
|
|
|
|
// Get ensemble prediction
|
|
let ensemble_decision = match ensemble.predict(&features).await {
|
|
Ok(decision) => decision,
|
|
Err(e) => {
|
|
warn!(
|
|
"Ensemble prediction failed for {}: {}, using fallback",
|
|
symbol, e
|
|
);
|
|
return self.get_fallback_trading_signal(symbol).await;
|
|
},
|
|
};
|
|
|
|
// Convert ensemble decision to trading signal
|
|
let action = match ensemble_decision.action {
|
|
ml::ensemble::TradingAction::Buy => TradingActionType::Buy,
|
|
ml::ensemble::TradingAction::Sell => TradingActionType::Sell,
|
|
ml::ensemble::TradingAction::Hold => TradingActionType::Hold,
|
|
};
|
|
|
|
// Apply risk-based position sizing
|
|
let position_size = self
|
|
.calculate_position_size(
|
|
symbol,
|
|
ensemble_decision.confidence,
|
|
ensemble_decision.disagreement_rate,
|
|
)
|
|
.await?;
|
|
|
|
info!(
|
|
"Ensemble signal for {}: action={:?}, confidence={:.3}, size={}",
|
|
symbol, action, ensemble_decision.confidence, position_size
|
|
);
|
|
|
|
Ok(EnsembleTradingSignal {
|
|
symbol: symbol.to_string(),
|
|
action,
|
|
confidence: ensemble_decision.confidence,
|
|
position_size,
|
|
disagreement_rate: ensemble_decision.disagreement_rate,
|
|
model_votes: ensemble_decision.model_votes.clone(),
|
|
timestamp: std::time::SystemTime::now(),
|
|
})
|
|
}
|
|
|
|
/// Extract features from market data for a symbol
|
|
async fn extract_features_for_symbol(
|
|
&self,
|
|
_symbol: &str,
|
|
) -> TradingServiceResult<ml::Features> {
|
|
// Mock feature extraction for now
|
|
// In production, this would:
|
|
// 1. Get latest OHLCV data from market_data_repository
|
|
// 2. Calculate technical indicators (RSI, MACD, Bollinger, etc.)
|
|
// 3. Extract order book features if available
|
|
// 4. Return Features struct
|
|
|
|
// For now, return mock features
|
|
Ok(ml::Features::new(
|
|
vec![0.5, 0.6, 0.7, 0.8, 0.9], // Mock feature values
|
|
vec![
|
|
"f1".to_string(),
|
|
"f2".to_string(),
|
|
"f3".to_string(),
|
|
"f4".to_string(),
|
|
"f5".to_string(),
|
|
],
|
|
))
|
|
}
|
|
|
|
/// Calculate position size based on ensemble confidence and disagreement
|
|
async fn calculate_position_size(
|
|
&self,
|
|
_symbol: &str,
|
|
confidence: f64,
|
|
disagreement_rate: f64,
|
|
) -> TradingServiceResult<u64> {
|
|
// Base position size (in shares/contracts)
|
|
let base_size: u64 = 100;
|
|
|
|
// Scale by confidence (0.5-1.0 confidence → 0-1.0 multiplier)
|
|
let confidence_multiplier = ((confidence - 0.5) * 2.0).max(0.0).min(1.0);
|
|
|
|
// Reduce size based on disagreement (high disagreement = lower size)
|
|
let disagreement_penalty = 1.0 - disagreement_rate;
|
|
|
|
// Final position size
|
|
let position_size =
|
|
(base_size as f64 * confidence_multiplier * disagreement_penalty) as u64;
|
|
|
|
Ok(position_size.max(10)) // Minimum 10 shares/contracts
|
|
}
|
|
|
|
/// Fallback trading signal using single model (DQN epoch 30)
|
|
async fn get_fallback_trading_signal(
|
|
&self,
|
|
symbol: &str,
|
|
) -> TradingServiceResult<EnsembleTradingSignal> {
|
|
use tracing::info;
|
|
|
|
info!("Using fallback DQN model for {}", symbol);
|
|
|
|
// Mock fallback signal
|
|
// In production, this would load DQN epoch 30 checkpoint and make prediction
|
|
Ok(EnsembleTradingSignal {
|
|
symbol: symbol.to_string(),
|
|
action: TradingActionType::Hold,
|
|
confidence: 0.60,
|
|
position_size: 50, // Conservative size for fallback
|
|
disagreement_rate: 0.0,
|
|
model_votes: std::collections::HashMap::new(),
|
|
timestamp: std::time::SystemTime::now(),
|
|
})
|
|
}
|
|
}
|
|
|
|
/// Trading signal with ensemble attribution
|
|
#[derive(Debug, Clone)]
|
|
pub struct EnsembleTradingSignal {
|
|
pub symbol: String,
|
|
pub action: TradingActionType,
|
|
pub confidence: f64,
|
|
pub position_size: u64,
|
|
pub disagreement_rate: f64,
|
|
pub model_votes: std::collections::HashMap<String, ml::ensemble::ModelVote>,
|
|
pub timestamp: std::time::SystemTime,
|
|
}
|
|
|
|
/// Trading action type
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum TradingActionType {
|
|
Buy,
|
|
Sell,
|
|
Hold,
|
|
}
|
|
|
|
/// Risk management engine wired to the real `risk` crate VaR calculators.
|
|
///
|
|
/// Provides two levels of VaR calculation:
|
|
/// - **Marginal VaR** via `risk::risk_engine::VarEngine` -- fast, parametric,
|
|
/// suitable for pre-trade per-order risk checks.
|
|
/// - **Comprehensive portfolio VaR** via `risk::RealVaREngine` -- multi-methodology
|
|
/// (historical simulation, parametric, Monte Carlo, hybrid) with stress testing
|
|
/// and risk decomposition.
|
|
pub struct RiskEngine {
|
|
/// Fast parametric VaR engine for marginal/per-order risk checks
|
|
var_engine: risk::risk_engine::VarEngine,
|
|
/// Multi-methodology portfolio VaR engine
|
|
portfolio_var_engine: risk::RealVaREngine,
|
|
/// VaR confidence level loaded from config repository (default 0.95)
|
|
var_confidence: f64,
|
|
/// Maximum VaR limit loaded from config repository
|
|
max_var_limit: f64,
|
|
}
|
|
|
|
impl std::fmt::Debug for RiskEngine {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("RiskEngine")
|
|
.field("var_confidence", &self.var_confidence)
|
|
.field("max_var_limit", &self.max_var_limit)
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl Default for RiskEngine {
|
|
fn default() -> Self {
|
|
Self::new()
|
|
}
|
|
}
|
|
|
|
impl RiskEngine {
|
|
/// Create a new `RiskEngine` backed by real VaR calculators from the `risk` crate.
|
|
pub fn new() -> Self {
|
|
let var_config = config::structures::VarConfig::default();
|
|
let var_engine = risk::risk_engine::VarEngine::with_defaults(var_config);
|
|
let portfolio_var_engine = risk::RealVaREngine::new();
|
|
Self {
|
|
var_engine,
|
|
portfolio_var_engine,
|
|
var_confidence: 0.95,
|
|
max_var_limit: 100_000.0,
|
|
}
|
|
}
|
|
|
|
/// Initialize the risk engine with parameters from the config repository.
|
|
///
|
|
/// Loads `var_confidence` and `max_var_limit` from the "Risk" config category.
|
|
/// If a value is missing the default is retained.
|
|
pub async fn initialize_with_config_repository(
|
|
&mut self,
|
|
config_repository: &Arc<PostgresConfigRepository>,
|
|
) -> TradingServiceResult<()> {
|
|
// Load VaR confidence level from config repository
|
|
if let Ok(Some(var_confidence)) = config_repository
|
|
.get_config_f64("Risk", "var_confidence")
|
|
.await
|
|
{
|
|
if (0.0..=1.0).contains(&var_confidence) {
|
|
self.var_confidence = var_confidence;
|
|
tracing::info!(
|
|
"Risk engine VaR confidence set to: {}",
|
|
var_confidence
|
|
);
|
|
} else {
|
|
tracing::warn!(
|
|
"Invalid VaR confidence {} from config, keeping default {}",
|
|
var_confidence,
|
|
self.var_confidence
|
|
);
|
|
}
|
|
}
|
|
|
|
// Load max VaR limit from config repository
|
|
if let Ok(Some(max_var_limit)) = config_repository
|
|
.get_config_f64("Risk", "max_var_limit")
|
|
.await
|
|
{
|
|
if max_var_limit > 0.0 {
|
|
self.max_var_limit = max_var_limit;
|
|
tracing::info!(
|
|
"Risk engine max VaR limit set to: {}",
|
|
max_var_limit
|
|
);
|
|
}
|
|
}
|
|
|
|
// Rebuild var_engine with updated configuration
|
|
let var_config = config::structures::VarConfig {
|
|
confidence_level: self.var_confidence,
|
|
max_var_limit: self.max_var_limit,
|
|
..config::structures::VarConfig::default()
|
|
};
|
|
self.var_engine = risk::risk_engine::VarEngine::with_defaults(var_config);
|
|
|
|
tracing::info!(
|
|
"RiskEngine initialized with real VaR calculators (confidence={}, max_limit={})",
|
|
self.var_confidence,
|
|
self.max_var_limit
|
|
);
|
|
Ok(())
|
|
}
|
|
|
|
/// Calculate the marginal Value at Risk for a proposed order.
|
|
///
|
|
/// Delegates to `risk::risk_engine::VarEngine::calculate_marginal_var` which
|
|
/// computes VaR = position_value * daily_volatility * z_score(confidence).
|
|
///
|
|
/// Returns the marginal VaR as `f64` for easy use in the trading service.
|
|
pub async fn calculate_marginal_var(
|
|
&self,
|
|
account_id: &str,
|
|
instrument_id: &str,
|
|
quantity: f64,
|
|
price: f64,
|
|
) -> Result<f64, String> {
|
|
let quantity_decimal = rust_decimal::Decimal::try_from(quantity)
|
|
.map_err(|e| format!("Invalid quantity for VaR: {e}"))?;
|
|
let price_decimal = rust_decimal::Decimal::try_from(price)
|
|
.map_err(|e| format!("Invalid price for VaR: {e}"))?;
|
|
|
|
let marginal_var = self
|
|
.var_engine
|
|
.calculate_marginal_var(account_id, instrument_id, quantity_decimal, price_decimal)
|
|
.await
|
|
.map_err(|e| format!("VaR calculation failed: {e}"))?;
|
|
|
|
use num_traits::ToPrimitive;
|
|
marginal_var
|
|
.to_f64()
|
|
.ok_or_else(|| "Failed to convert VaR Decimal to f64".to_string())
|
|
}
|
|
|
|
/// Check whether a proposed order's VaR impact exceeds the configured limit.
|
|
///
|
|
/// Returns `Ok(())` if within limits, or `Err(message)` describing the breach.
|
|
pub async fn check_var_limit(
|
|
&self,
|
|
account_id: &str,
|
|
instrument_id: &str,
|
|
quantity: f64,
|
|
price: f64,
|
|
) -> Result<(), String> {
|
|
let marginal_var = self
|
|
.calculate_marginal_var(account_id, instrument_id, quantity, price)
|
|
.await?;
|
|
|
|
if marginal_var > self.max_var_limit {
|
|
Err(format!(
|
|
"VaR limit exceeded: marginal VaR {:.2} > limit {:.2}",
|
|
marginal_var, self.max_var_limit
|
|
))
|
|
} else {
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// Get a reference to the comprehensive portfolio VaR engine.
|
|
///
|
|
/// Callers can use this to run `calculate_comprehensive_var` and
|
|
/// `check_circuit_breaker_conditions` from `risk::RealVaREngine`.
|
|
pub fn portfolio_var_engine(&self) -> &risk::RealVaREngine {
|
|
&self.portfolio_var_engine
|
|
}
|
|
|
|
/// Get the current VaR confidence level.
|
|
pub fn var_confidence(&self) -> f64 {
|
|
self.var_confidence
|
|
}
|
|
|
|
/// Get the current maximum VaR limit.
|
|
pub fn max_var_limit(&self) -> f64 {
|
|
self.max_var_limit
|
|
}
|
|
}
|
|
|
|
/// `Position` state validator
|
|
#[derive(Debug, Default)]
|
|
pub struct PositionStateValidator {
|
|
// Position validation logic
|
|
}
|
|
|
|
impl PositionStateValidator {
|
|
pub fn new() -> Self {
|
|
Self::default()
|
|
}
|
|
|
|
pub async fn initialize_with_config_repository(
|
|
&mut self,
|
|
config_repository: &Arc<PostgresConfigRepository>,
|
|
) -> TradingServiceResult<()> {
|
|
// Initialize risk parameters from repository (no direct database access)
|
|
// Load VaR confidence from config repository
|
|
if let Ok(Some(var_confidence)) = config_repository
|
|
.get_config_f64("Risk", "var_confidence")
|
|
.await
|
|
{
|
|
tracing::info!(
|
|
"Risk engine initialized with VaR confidence: {}",
|
|
var_confidence
|
|
);
|
|
}
|
|
|
|
// Load max drawdown limit from config repository
|
|
if let Ok(Some(max_drawdown)) = config_repository
|
|
.get_config_f64("Risk", "max_drawdown_limit")
|
|
.await
|
|
{
|
|
tracing::info!(
|
|
"Risk engine initialized with max drawdown limit: {}",
|
|
max_drawdown
|
|
);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
}
|
|
|
|
/// ML engine backed by the real `ml::ensemble::EnsembleCoordinator`.
|
|
///
|
|
/// The coordinator is lazily initialized during
|
|
/// [`initialize_with_config_repository`](MLEngine::initialize_with_config_repository)
|
|
/// because it needs async model registration. Before that call,
|
|
/// [`coordinator()`](MLEngine::coordinator) returns `None`.
|
|
pub struct MLEngine {
|
|
/// The real ensemble coordinator from the `ml` crate.
|
|
/// `None` until [`initialize_with_config_repository`] completes.
|
|
coordinator: Option<ml::ensemble::EnsembleCoordinator>,
|
|
}
|
|
|
|
impl std::fmt::Debug for MLEngine {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("MLEngine")
|
|
.field("coordinator", &self.coordinator)
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl Default for MLEngine {
|
|
fn default() -> Self {
|
|
Self { coordinator: None }
|
|
}
|
|
}
|
|
|
|
impl MLEngine {
|
|
pub fn new() -> Self {
|
|
Self::default()
|
|
}
|
|
|
|
/// Initialize the ML engine by creating an `EnsembleCoordinator` and
|
|
/// registering the default production models (DQN, PPO, TFT, MAMBA-2).
|
|
///
|
|
/// Model weights and the inference timeout are loaded from the config
|
|
/// repository. If a config key is missing the default value is used.
|
|
pub async fn initialize_with_config_repository(
|
|
&mut self,
|
|
config_repository: &Arc<PostgresConfigRepository>,
|
|
) -> TradingServiceResult<()> {
|
|
// Load ML inference timeout from config repository
|
|
if let Ok(Some(inference_timeout)) = config_repository
|
|
.get_config_u64("MachineLearning", "inference_timeout_ms")
|
|
.await
|
|
{
|
|
tracing::info!(
|
|
"ML engine initialized with inference timeout: {}ms",
|
|
inference_timeout
|
|
);
|
|
}
|
|
|
|
// Create the real ensemble coordinator
|
|
let coordinator = ml::ensemble::EnsembleCoordinator::new();
|
|
|
|
// Load per-model weights from config, falling back to equal weights
|
|
let default_weight = 0.25_f64;
|
|
|
|
let dqn_weight = config_repository
|
|
.get_config_f64("MachineLearning", "dqn_weight")
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.unwrap_or(default_weight);
|
|
|
|
let ppo_weight = config_repository
|
|
.get_config_f64("MachineLearning", "ppo_weight")
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.unwrap_or(default_weight);
|
|
|
|
let tft_weight = config_repository
|
|
.get_config_f64("MachineLearning", "tft_weight")
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.unwrap_or(default_weight);
|
|
|
|
let mamba2_weight = config_repository
|
|
.get_config_f64("MachineLearning", "mamba2_weight")
|
|
.await
|
|
.ok()
|
|
.flatten()
|
|
.unwrap_or(default_weight);
|
|
|
|
// Register the four production models
|
|
if let Err(e) = coordinator
|
|
.register_model("DQN".to_string(), dqn_weight)
|
|
.await
|
|
{
|
|
tracing::warn!("Failed to register DQN model: {}", e);
|
|
}
|
|
if let Err(e) = coordinator
|
|
.register_model("PPO".to_string(), ppo_weight)
|
|
.await
|
|
{
|
|
tracing::warn!("Failed to register PPO model: {}", e);
|
|
}
|
|
if let Err(e) = coordinator
|
|
.register_model("TFT".to_string(), tft_weight)
|
|
.await
|
|
{
|
|
tracing::warn!("Failed to register TFT model: {}", e);
|
|
}
|
|
if let Err(e) = coordinator
|
|
.register_model("MAMBA-2".to_string(), mamba2_weight)
|
|
.await
|
|
{
|
|
tracing::warn!("Failed to register MAMBA-2 model: {}", e);
|
|
}
|
|
|
|
let model_count = coordinator.model_count().await;
|
|
tracing::info!(
|
|
"ML engine EnsembleCoordinator initialized with {} models \
|
|
(DQN={:.2}, PPO={:.2}, TFT={:.2}, MAMBA-2={:.2})",
|
|
model_count,
|
|
dqn_weight,
|
|
ppo_weight,
|
|
tft_weight,
|
|
mamba2_weight,
|
|
);
|
|
|
|
self.coordinator = Some(coordinator);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Get a shared reference to the underlying `EnsembleCoordinator`.
|
|
///
|
|
/// Returns `None` if the engine has not been initialized yet.
|
|
pub fn coordinator(&self) -> Option<&ml::ensemble::EnsembleCoordinator> {
|
|
self.coordinator.as_ref()
|
|
}
|
|
|
|
/// Get a mutable reference to the underlying `EnsembleCoordinator`.
|
|
///
|
|
/// Useful for adding inference adapters via
|
|
/// [`EnsembleCoordinator::add_adapter`].
|
|
/// Returns `None` if the engine has not been initialized yet.
|
|
pub fn coordinator_mut(&mut self) -> Option<&mut ml::ensemble::EnsembleCoordinator> {
|
|
self.coordinator.as_mut()
|
|
}
|
|
}
|
|
|
|
/// Market data manager with multiple providers
|
|
pub struct MarketDataManager {
|
|
/// Databento provider for market data
|
|
databento_provider:
|
|
Option<Arc<RwLock<data::providers::databento_streaming::DatabentoStreamingProvider>>>,
|
|
/// Benzinga provider for news data
|
|
benzinga_provider: Option<
|
|
Arc<RwLock<data::providers::benzinga::production_streaming::ProductionBenzingaProvider>>,
|
|
>,
|
|
/// Unified feature extractor
|
|
feature_extractor: Option<Arc<ml::features::UnifiedFeatureExtractor>>,
|
|
/// Event broadcast sender
|
|
_event_sender: Arc<
|
|
tokio::sync::broadcast::Sender<trading_engine::trading::data_interface::MarketDataEvent>,
|
|
>,
|
|
}
|
|
|
|
impl std::fmt::Debug for MarketDataManager {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("MarketDataManager")
|
|
.field("databento_provider", &self.databento_provider)
|
|
.field("benzinga_provider", &self.benzinga_provider)
|
|
.field("feature_extractor", &"<UnifiedFeatureExtractor>")
|
|
.field("_event_sender", &self._event_sender)
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
impl Default for MarketDataManager {
|
|
fn default() -> Self {
|
|
let (_event_sender, _) = tokio::sync::broadcast::channel(10000);
|
|
Self {
|
|
databento_provider: None,
|
|
benzinga_provider: None,
|
|
feature_extractor: None,
|
|
_event_sender: Arc::new(_event_sender),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl MarketDataManager {
|
|
pub fn new() -> Self {
|
|
Self::default()
|
|
}
|
|
|
|
pub async fn initialize(&mut self) -> TradingServiceResult<()> {
|
|
// Initialize Databento provider if API key is available
|
|
if let Ok(api_key) = std::env::var("DATABENTO_API_KEY") {
|
|
match data::providers::databento_streaming::DatabentoStreamingProvider::new(api_key) {
|
|
Ok(mut provider) => {
|
|
if let Err(e) = provider.connect().await {
|
|
tracing::warn!("Failed to connect to Databento: {}", e);
|
|
} else {
|
|
tracing::info!("Connected to Databento successfully");
|
|
self.databento_provider = Some(Arc::new(RwLock::new(provider)));
|
|
}
|
|
},
|
|
Err(e) => {
|
|
tracing::error!("Failed to create Databento provider: {}", e);
|
|
},
|
|
}
|
|
} else {
|
|
tracing::warn!("DATABENTO_API_KEY not found, skipping Databento provider");
|
|
}
|
|
|
|
// Initialize Benzinga provider if API key is available
|
|
if let Ok(api_key) = std::env::var("BENZINGA_API_KEY") {
|
|
let benzinga_config =
|
|
data::providers::benzinga::production_streaming::ProductionBenzingaConfig {
|
|
api_key,
|
|
websocket_url: "wss://api.benzinga.com/api/v1/news/stream".to_string(),
|
|
connect_timeout_secs: 30,
|
|
ping_interval_secs: 30,
|
|
max_reconnect_attempts: 3,
|
|
initial_reconnect_delay_ms: 1000,
|
|
max_reconnect_delay_ms: 60000,
|
|
reconnect_backoff_multiplier: 2.0,
|
|
enable_news: true,
|
|
enable_sentiment: true,
|
|
enable_ratings: true,
|
|
enable_options: false,
|
|
event_buffer_size: 1000,
|
|
heartbeat_timeout_secs: 300,
|
|
rate_limit_per_second: 10,
|
|
dedup_window_secs: 60,
|
|
max_dedup_cache_size: 10000,
|
|
enable_compression: true,
|
|
batch_processing_size: 100,
|
|
enable_smart_categorization: true,
|
|
enable_ml_integration: true,
|
|
circuit_breaker_threshold: 5,
|
|
circuit_breaker_timeout_secs: 300,
|
|
max_concurrent_processing: 4,
|
|
};
|
|
|
|
match data::providers::benzinga::production_streaming::ProductionBenzingaProvider::new(
|
|
benzinga_config,
|
|
) {
|
|
Ok(mut provider) => {
|
|
if let Err(e) = provider.connect().await {
|
|
tracing::warn!("Failed to connect to Benzinga: {}", e);
|
|
} else {
|
|
tracing::info!("Connected to Benzinga successfully");
|
|
self.benzinga_provider = Some(Arc::new(RwLock::new(provider)));
|
|
}
|
|
},
|
|
Err(e) => {
|
|
tracing::error!("Failed to create Benzinga provider: {}", e);
|
|
},
|
|
}
|
|
} else {
|
|
tracing::warn!("BENZINGA_API_KEY not found, skipping Benzinga provider");
|
|
}
|
|
|
|
// Initialize UnifiedFeatureExtractor
|
|
let config = ml::features::FeatureExtractionConfig::default();
|
|
let safety_manager = Arc::new(ml::safety::MLSafetyManager::new(
|
|
ml::safety::MLSafetyConfig::default(),
|
|
));
|
|
self.feature_extractor = Some(Arc::new(ml::features::UnifiedFeatureExtractor::new(
|
|
config,
|
|
safety_manager,
|
|
)));
|
|
|
|
tracing::info!("MarketDataManager initialized with available providers");
|
|
Ok(())
|
|
}
|
|
|
|
pub async fn initialize_with_config_repository(
|
|
&mut self,
|
|
config_repository: &Arc<PostgresConfigRepository>,
|
|
) -> TradingServiceResult<()> {
|
|
// Initialize market data providers using repository-based configuration (no direct database access)
|
|
|
|
// Get API keys from config repository
|
|
if let Ok(Some(databento_key)) = config_repository.get_secret("databento_api_key").await {
|
|
match data::providers::databento_streaming::DatabentoStreamingProvider::new(
|
|
databento_key,
|
|
) {
|
|
Ok(mut provider) => {
|
|
if let Err(e) = provider.connect().await {
|
|
tracing::warn!("Failed to connect to Databento: {}", e);
|
|
} else {
|
|
tracing::info!("Connected to Databento successfully via config repository");
|
|
self.databento_provider = Some(Arc::new(RwLock::new(provider)));
|
|
}
|
|
},
|
|
Err(e) => {
|
|
tracing::error!("Failed to create Databento provider: {}", e);
|
|
},
|
|
}
|
|
} else {
|
|
tracing::warn!("Databento API key not found in config repository, skipping provider");
|
|
}
|
|
|
|
if let Ok(Some(benzinga_key)) = config_repository.get_secret("benzinga_api_key").await {
|
|
let benzinga_config =
|
|
data::providers::benzinga::production_streaming::ProductionBenzingaConfig {
|
|
api_key: benzinga_key,
|
|
websocket_url: "wss://api.benzinga.com/api/v1/news/stream".to_string(),
|
|
connect_timeout_secs: 30,
|
|
ping_interval_secs: 30,
|
|
max_reconnect_attempts: 3,
|
|
initial_reconnect_delay_ms: 1000,
|
|
max_reconnect_delay_ms: 60000,
|
|
reconnect_backoff_multiplier: 2.0,
|
|
enable_news: true,
|
|
enable_sentiment: true,
|
|
enable_ratings: true,
|
|
enable_options: false,
|
|
event_buffer_size: 1000,
|
|
heartbeat_timeout_secs: 300,
|
|
rate_limit_per_second: 10,
|
|
dedup_window_secs: 60,
|
|
max_dedup_cache_size: 10000,
|
|
enable_compression: true,
|
|
batch_processing_size: 100,
|
|
enable_smart_categorization: true,
|
|
enable_ml_integration: true,
|
|
circuit_breaker_threshold: 5,
|
|
circuit_breaker_timeout_secs: 300,
|
|
max_concurrent_processing: 4,
|
|
};
|
|
match data::providers::benzinga::production_streaming::ProductionBenzingaProvider::new(
|
|
benzinga_config,
|
|
) {
|
|
Ok(mut provider) => {
|
|
if let Err(e) = provider.connect().await {
|
|
tracing::warn!("Failed to connect to Benzinga: {}", e);
|
|
} else {
|
|
tracing::info!("Connected to Benzinga successfully via config repository");
|
|
self.benzinga_provider = Some(Arc::new(RwLock::new(provider)));
|
|
}
|
|
},
|
|
Err(e) => {
|
|
tracing::error!("Failed to create Benzinga provider: {}", e);
|
|
},
|
|
}
|
|
} else {
|
|
tracing::warn!("Benzinga API key not found in config repository, skipping provider");
|
|
}
|
|
|
|
// Initialize UnifiedFeatureExtractor with configuration
|
|
let config = ml::features::FeatureExtractionConfig::default();
|
|
let safety_manager = Arc::new(ml::safety::MLSafetyManager::new(
|
|
ml::safety::MLSafetyConfig::default(),
|
|
));
|
|
self.feature_extractor = Some(Arc::new(ml::features::UnifiedFeatureExtractor::new(
|
|
config,
|
|
safety_manager,
|
|
)));
|
|
|
|
tracing::info!("MarketDataManager initialized with repository-based configuration");
|
|
Ok(())
|
|
}
|
|
|
|
/// Subscribe to market data for given symbols
|
|
pub async fn subscribe_to_symbols(
|
|
&mut self,
|
|
symbols: Vec<common::types::Symbol>,
|
|
) -> TradingServiceResult<()> {
|
|
// Subscribe via Databento provider
|
|
if let Some(databento) = &self.databento_provider {
|
|
let mut provider = databento.write().await;
|
|
let symbol_strings: Vec<String> = symbols.iter().map(|s| s.to_string()).collect();
|
|
if let Err(e) = provider.subscribe(symbol_strings).await {
|
|
tracing::error!("Failed to subscribe to Databento: {}", e);
|
|
} else {
|
|
tracing::info!("Subscribed to {} symbols on Databento", symbols.len());
|
|
}
|
|
}
|
|
|
|
// Subscribe via Benzinga provider
|
|
if let Some(benzinga) = &self.benzinga_provider {
|
|
let mut provider = benzinga.write().await;
|
|
if let Err(e) = provider.subscribe(symbols.clone()).await {
|
|
tracing::error!("Failed to subscribe to Benzinga news: {}", e);
|
|
} else {
|
|
tracing::info!(
|
|
"Subscribed to news for {} symbols on Benzinga",
|
|
symbols.len()
|
|
);
|
|
}
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// Get market data event receiver
|
|
pub fn get_event_receiver(
|
|
&self,
|
|
) -> tokio::sync::broadcast::Receiver<trading_engine::trading::data_interface::MarketDataEvent>
|
|
{
|
|
self._event_sender.subscribe()
|
|
}
|
|
|
|
/// Start event processing from all providers
|
|
pub async fn start_event_processing(&self) -> TradingServiceResult<()> {
|
|
// Start processing events from Databento
|
|
if let Some(databento) = &self.databento_provider {
|
|
let provider = Arc::clone(databento);
|
|
let _event_sender = Arc::clone(&self._event_sender);
|
|
let feature_extractor = self.feature_extractor.clone();
|
|
|
|
tokio::spawn(async move {
|
|
let databento_provider = provider.read().await;
|
|
let mut event_receiver = databento_provider.subscribe_market_events();
|
|
drop(databento_provider); // Release the read lock
|
|
|
|
while let Ok(event) = event_receiver.recv().await {
|
|
// Process event through feature extractor if available
|
|
if let Some(_extractor) = &feature_extractor {
|
|
// TODO: Implement feature extraction pipeline
|
|
tracing::debug!("Processing market event through feature extractor");
|
|
}
|
|
|
|
// Forward event to subscribers
|
|
let _ = _event_sender.send(event);
|
|
}
|
|
});
|
|
}
|
|
|
|
// Start processing events from Benzinga
|
|
if let Some(benzinga) = &self.benzinga_provider {
|
|
let provider = Arc::clone(benzinga);
|
|
let _event_sender = Arc::clone(&self._event_sender);
|
|
|
|
tokio::spawn(async move {
|
|
let mut benzinga_provider = provider.write().await;
|
|
match benzinga_provider.stream().await {
|
|
Ok(mut stream) => {
|
|
drop(benzinga_provider); // Release the write lock
|
|
|
|
// Process events from the stream
|
|
while let Some(event) = stream.next().await {
|
|
// Forward news-derived market events to subscribers
|
|
let _ = _event_sender.send(event);
|
|
}
|
|
},
|
|
Err(e) => {
|
|
tracing::error!("Failed to get Benzinga stream: {}", e);
|
|
},
|
|
}
|
|
});
|
|
}
|
|
|
|
tracing::info!("Started event processing for all connected providers");
|
|
Ok(())
|
|
}
|
|
|
|
/// Get health status of all providers
|
|
pub async fn get_provider_health(
|
|
&self,
|
|
) -> Vec<(String, data::providers::ProviderHealthStatus)> {
|
|
let mut health_status = Vec::new();
|
|
|
|
if let Some(databento) = &self.databento_provider {
|
|
let provider = databento.read().await;
|
|
health_status.push(("databento".to_string(), provider.get_health_status()));
|
|
}
|
|
|
|
if let Some(benzinga) = &self.benzinga_provider {
|
|
let provider = benzinga.read().await;
|
|
// Create ProviderHealthStatus from ConnectionStatus
|
|
let connection_status = provider.get_connection_status();
|
|
let health = data::providers::ProviderHealthStatus {
|
|
connected: matches!(
|
|
connection_status.state,
|
|
data::providers::ConnectionState::Connected
|
|
),
|
|
last_connected: connection_status.last_connection_attempt,
|
|
active_subscriptions: connection_status.active_subscriptions,
|
|
messages_per_second: connection_status.events_per_second,
|
|
latency_micros: connection_status.latency_micros,
|
|
error_count: 0, // Default value
|
|
};
|
|
health_status.push(("benzinga".to_string(), health));
|
|
}
|
|
|
|
health_status
|
|
}
|
|
}
|
|
/// Health status enumeration
|
|
#[derive(Debug, Clone)]
|
|
pub enum HealthStatus {
|
|
/// All systems operational
|
|
Healthy,
|
|
/// Some degraded performance
|
|
Degraded,
|
|
/// Critical issues present
|
|
Unhealthy,
|
|
/// Service offline
|
|
Critical,
|
|
}
|
|
|
|
/// Ensemble coordinator health status
|
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
|
pub enum EnsembleHealth {
|
|
/// All models loaded and operational
|
|
Healthy,
|
|
/// Some models missing or degraded performance
|
|
Degraded,
|
|
/// Critical issues (no models loaded, inference failing)
|
|
Unhealthy,
|
|
/// Ensemble coordinator not configured (fallback mode)
|
|
NotConfigured,
|
|
}
|
|
|
|
/// Detailed ensemble health report
|
|
#[derive(Debug, Clone)]
|
|
pub struct EnsembleHealthReport {
|
|
pub status: EnsembleHealth,
|
|
pub models_loaded: usize,
|
|
pub expected_models: usize,
|
|
pub inference_latency_us: Option<f64>,
|
|
pub last_prediction: Option<std::time::SystemTime>,
|
|
pub model_details: Vec<ModelHealthDetail>,
|
|
}
|
|
|
|
/// Per-model health detail
|
|
#[derive(Debug, Clone)]
|
|
pub struct ModelHealthDetail {
|
|
pub model_id: String,
|
|
pub checkpoint: String,
|
|
pub loaded: bool,
|
|
pub last_inference: std::time::SystemTime,
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn test_risk_engine_creation() {
|
|
let engine = RiskEngine::new();
|
|
// Default confidence level is 0.95
|
|
assert!((engine.var_confidence() - 0.95).abs() < f64::EPSILON);
|
|
// Default max VaR limit is 100_000.0
|
|
assert!((engine.max_var_limit() - 100_000.0).abs() < f64::EPSILON);
|
|
}
|
|
|
|
#[test]
|
|
fn test_risk_engine_default() {
|
|
let engine = RiskEngine::default();
|
|
assert!((engine.var_confidence() - 0.95).abs() < f64::EPSILON);
|
|
assert!((engine.max_var_limit() - 100_000.0).abs() < f64::EPSILON);
|
|
}
|
|
|
|
#[test]
|
|
fn test_risk_engine_debug() {
|
|
let engine = RiskEngine::new();
|
|
let debug_str = format!("{engine:?}");
|
|
assert!(debug_str.contains("RiskEngine"));
|
|
assert!(debug_str.contains("var_confidence"));
|
|
assert!(debug_str.contains("max_var_limit"));
|
|
}
|
|
|
|
#[test]
|
|
fn test_portfolio_var_engine_accessible() {
|
|
let engine = RiskEngine::new();
|
|
// Verify the portfolio VaR engine is accessible and functional
|
|
let _portfolio_engine = engine.portfolio_var_engine();
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_marginal_var_calculation() {
|
|
let engine = RiskEngine::new();
|
|
// AAPL at $175, 100 shares -- should produce a positive VaR
|
|
let result = engine
|
|
.calculate_marginal_var("test_account", "AAPL", 100.0, 175.0)
|
|
.await;
|
|
assert!(result.is_ok(), "Marginal VaR calculation failed: {result:?}");
|
|
let var_value = result.unwrap_or(0.0);
|
|
assert!(var_value > 0.0, "Marginal VaR should be positive, got {var_value}");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_marginal_var_crypto_higher_than_equity() {
|
|
let engine = RiskEngine::new();
|
|
// Crypto (BTC) should have higher volatility than equity (AAPL)
|
|
// Same notional value for comparison
|
|
let btc_var = engine
|
|
.calculate_marginal_var("test_account", "BTC", 1.0, 50_000.0)
|
|
.await
|
|
.unwrap_or(0.0);
|
|
let aapl_var = engine
|
|
.calculate_marginal_var("test_account", "AAPL", 285.7, 175.0)
|
|
.await
|
|
.unwrap_or(0.0);
|
|
// Both should be non-zero and BTC should have higher VaR
|
|
assert!(btc_var > 0.0, "BTC VaR should be positive");
|
|
assert!(aapl_var > 0.0, "AAPL VaR should be positive");
|
|
assert!(
|
|
btc_var > aapl_var,
|
|
"BTC VaR ({btc_var:.2}) should exceed AAPL VaR ({aapl_var:.2})"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_check_var_limit_within_limit() {
|
|
let engine = RiskEngine::new();
|
|
// Small order should be within the 100k default limit
|
|
let result = engine
|
|
.check_var_limit("test_account", "AAPL", 10.0, 175.0)
|
|
.await;
|
|
assert!(result.is_ok(), "Small order should pass VaR limit check");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_check_var_limit_exceeds_limit() {
|
|
let mut engine = RiskEngine::new();
|
|
// Set a very low VaR limit
|
|
engine.max_var_limit = 1.0;
|
|
let result = engine
|
|
.check_var_limit("test_account", "AAPL", 1000.0, 175.0)
|
|
.await;
|
|
assert!(result.is_err(), "Large order should breach tiny VaR limit");
|
|
let err_msg = result.unwrap_err();
|
|
assert!(
|
|
err_msg.contains("VaR limit exceeded"),
|
|
"Error message should describe VaR breach: {err_msg}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_marginal_var_invalid_inputs() {
|
|
let engine = RiskEngine::new();
|
|
// Zero quantity should still work (VaR = 0 -> non-positive -> error from risk crate)
|
|
let result = engine
|
|
.calculate_marginal_var("test_account", "AAPL", 0.0, 175.0)
|
|
.await;
|
|
// The risk crate rejects non-positive VaR results, so this should fail
|
|
assert!(result.is_err(), "Zero quantity should produce an error");
|
|
}
|
|
}
|