gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

merge: pull liquid CfC v2 and codebase deduplication from main into production-hardening

Browse Source
This commit is contained in:
jgrusewski
2026-02-23 10:09:25 +01:00
parent 2e4b0c9455 5a90eaba8c
commit 15c0fb5397
44 changed files with 3125 additions and 1638 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
Cargo.lock generated
View File

@@ -5826,8 +5826,8 @@ dependencies = [
"anyhow",
"async-trait",
"chrono",
"common",
"lru",
"ml",
"parking_lot 0.12.5",
"semver 1.0.27",
"serde",

22
adaptive-strategy/src/config.rs
View File

@@ -66,9 +66,6 @@ pub struct GeneralConfig {
impl Default for GeneralConfig {
fn default() -> Self {
tracing::warn!(
"Using hardcoded GeneralConfig::default() - migrate to database configuration!"
);
Self {
execution_interval: Duration::from_millis(100_u64),
error_backoff_duration: Duration::from_secs(1_u64),
@@ -80,9 +77,6 @@ impl Default for GeneralConfig {
impl Default for AdaptiveStrategyConfig {
fn default() -> Self {
tracing::warn!("Using hardcoded AdaptiveStrategyConfig::default() - migrate to database configuration!");
tracing::warn!(" Load configuration from database using DatabaseConfigLoader instead.");
tracing::warn!(" Available strategies: 'default-production', 'development', 'aggressive'");
Self {
general: GeneralConfig::default(),
ensemble: EnsembleConfig::default(),
@@ -112,9 +106,6 @@ pub struct EnsembleConfig {
impl Default for EnsembleConfig {
fn default() -> Self {
tracing::warn!(
"Using hardcoded EnsembleConfig::default() - migrate to database configuration!"
);
Self {
max_parallel_models: 4_usize,
rebalancing_interval: Duration::from_secs(300_u64),
@@ -162,9 +153,6 @@ pub struct ModelConfig {
impl Default for ModelConfig {
fn default() -> Self {
tracing::warn!(
"Using hardcoded ModelConfig::default() - migrate to database configuration!"
);
Self {
id: "default_model".to_owned(),
name: "default_model".to_owned(),
@@ -198,9 +186,6 @@ pub struct RiskConfig {
impl Default for RiskConfig {
fn default() -> Self {
tracing::warn!(
"Using hardcoded RiskConfig::default() - migrate to database configuration!"
);
Self {
max_position_size: 0.1_f64,
max_leverage: 2.0_f64,
@@ -252,9 +237,6 @@ pub struct MicrostructureConfig {
impl Default for MicrostructureConfig {
fn default() -> Self {
tracing::warn!(
"Using hardcoded MicrostructureConfig::default() - migrate to database configuration!"
);
Self {
book_depth: 10_usize,
vpin_window: 50_usize,
@@ -284,9 +266,6 @@ pub struct RegimeConfig {
impl Default for RegimeConfig {
fn default() -> Self {
eprintln!(
"WARNING: Using hardcoded RegimeConfig::default() - migrate to database configuration!"
);
Self {
detection_method: RegimeDetectionMethod::HMM,
lookback_window: 252_usize,
@@ -338,7 +317,6 @@ pub struct ExecutionConfig {
impl Default for ExecutionConfig {
fn default() -> Self {
eprintln!("WARNING: Using hardcoded ExecutionConfig::default() - migrate to database configuration!");
Self {
algorithm: ExecutionAlgorithm::TWAP,
max_order_size: 10000.0_f64,

91
adaptive-strategy/src/execution/mod.rs
View File

@@ -111,11 +111,9 @@ pub struct ExecutionPerformanceTracker {
/// Performance metrics by algorithm
algorithm_performance: HashMap<String, AlgorithmPerformance>,
/// Slippage measurements
#[allow(dead_code)]
slippage_tracker: SlippageTracker,
_slippage_tracker: SlippageTracker,
/// Implementation shortfall tracker
#[allow(dead_code)]
shortfall_tracker: ShortfallTracker,
_shortfall_tracker: ShortfallTracker,
}
/// Algorithm performance metrics
@@ -143,11 +141,9 @@ pub struct AlgorithmPerformance {
#[derive(Debug)]
pub struct SlippageTracker {
/// Slippage measurements
#[allow(dead_code)]
measurements: VecDeque<SlippageMeasurement>,
_measurements: VecDeque<SlippageMeasurement>,
/// Slippage statistics by symbol
#[allow(dead_code)]
stats_by_symbol: HashMap<String, SlippageStatistics>,
_stats_by_symbol: HashMap<String, SlippageStatistics>,
}
/// Slippage measurement
@@ -215,11 +211,9 @@ pub struct SmartOrderRouter {
/// Available venues
venues: Vec<TradingVenue>,
/// Routing rules
#[allow(dead_code)]
routing_rules: HashMap<String, RoutingRule>,
_routing_rules: HashMap<String, RoutingRule>,
/// Venue performance tracker
#[allow(dead_code)]
venue_performance: HashMap<String, VenuePerformance>,
_venue_performance: HashMap<String, VenuePerformance>,
}
/// Trading venue information
@@ -406,11 +400,9 @@ pub struct TWAPAlgorithm {
/// Number of slices
slice_count: u32,
/// Current slice
#[allow(dead_code)]
current_slice: u32,
_current_slice: u32,
/// Slice orders
#[allow(dead_code)]
slice_orders: Vec<Order>,
_slice_orders: Vec<Order>,
}
/// Volume-Weighted Average Price (VWAP) algorithm
@@ -419,24 +411,20 @@ pub struct VWAPAlgorithm {
/// Algorithm name
name: String,
/// Historical volume profile
#[allow(dead_code)]
volume_profile: HashMap<String, VolumeProfile>,
_volume_profile: HashMap<String, VolumeProfile>,
/// Participation rate
participation_rate: f64,
/// Current volume tracking
#[allow(dead_code)]
volume_tracker: VolumeTracker,
_volume_tracker: VolumeTracker,
}
/// Volume profile for VWAP calculation
#[derive(Debug, Clone)]
pub struct VolumeProfile {
/// Time buckets
#[allow(dead_code)]
buckets: Vec<VolumeBucket>,
_buckets: Vec<VolumeBucket>,
/// Profile date
#[allow(dead_code)]
date: NaiveDate,
_date: NaiveDate,
}
/// Volume bucket
@@ -454,11 +442,9 @@ pub struct VolumeBucket {
#[derive(Debug)]
pub struct VolumeTracker {
/// Current period volumes
#[allow(dead_code)]
period_volumes: HashMap<String, f64>,
_period_volumes: HashMap<String, f64>,
/// Target volumes
#[allow(dead_code)]
target_volumes: HashMap<String, f64>,
_target_volumes: HashMap<String, f64>,
}
/// Implementation Shortfall algorithm
@@ -469,25 +455,20 @@ pub struct ImplementationShortfallAlgorithm {
/// Risk aversion parameter
risk_aversion: f64,
/// Market impact model
#[allow(dead_code)]
impact_model: MarketImpactModel,
_impact_model: MarketImpactModel,
/// Optimal schedule
#[allow(dead_code)]
execution_schedule: Vec<ScheduleSlice>,
_execution_schedule: Vec<ScheduleSlice>,
}
/// Market impact model
#[derive(Debug)]
pub struct MarketImpactModel {
/// Temporary impact coefficient
#[allow(dead_code)]
temp_impact_coeff: f64,
_temp_impact_coeff: f64,
/// Permanent impact coefficient
#[allow(dead_code)]
perm_impact_coeff: f64,
_perm_impact_coeff: f64,
/// Volatility estimate
#[allow(dead_code)]
volatility: f64,
_volatility: f64,
}
/// Execution schedule slice
@@ -939,8 +920,8 @@ impl ExecutionPerformanceTracker {
pub fn new() -> Self {
Self {
algorithm_performance: HashMap::new(),
slippage_tracker: SlippageTracker::new(),
shortfall_tracker: ShortfallTracker::new(),
_slippage_tracker: SlippageTracker::new(),
_shortfall_tracker: ShortfallTracker::new(),
}
}
@@ -985,8 +966,8 @@ impl SlippageTracker {
/// Create a new slippage tracker
pub fn new() -> Self {
Self {
measurements: VecDeque::new(),
stats_by_symbol: HashMap::new(),
_measurements: VecDeque::new(),
_stats_by_symbol: HashMap::new(),
}
}
}
@@ -1033,8 +1014,8 @@ impl SmartOrderRouter {
Ok(Self {
venues,
routing_rules: HashMap::new(),
venue_performance: HashMap::new(),
_routing_rules: HashMap::new(),
_venue_performance: HashMap::new(),
})
}
@@ -1065,8 +1046,8 @@ impl TWAPAlgorithm {
name: "TWAP".to_owned(),
window_duration: Duration::from_secs(300_u64), // 5 minutes
slice_count: 10_u32,
current_slice: 0_u32,
slice_orders: Vec::new(),
_current_slice: 0_u32,
_slice_orders: Vec::new(),
})
}
}
@@ -1138,9 +1119,9 @@ impl VWAPAlgorithm {
pub fn new() -> Result<Self> {
Ok(Self {
name: "VWAP".to_owned(),
volume_profile: HashMap::new(),
_volume_profile: HashMap::new(),
participation_rate: 0.1_f64, // 10% participation
volume_tracker: VolumeTracker::new(),
_volume_tracker: VolumeTracker::new(),
})
}
}
@@ -1205,8 +1186,8 @@ impl VolumeTracker {
/// Create a new volume tracker
pub fn new() -> Self {
Self {
period_volumes: HashMap::new(),
target_volumes: HashMap::new(),
_period_volumes: HashMap::new(),
_target_volumes: HashMap::new(),
}
}
}
@@ -1217,8 +1198,8 @@ impl ImplementationShortfallAlgorithm {
Ok(Self {
name: "ImplementationShortfall".to_owned(),
risk_aversion: 1e-6_f64,
impact_model: MarketImpactModel::new(),
execution_schedule: Vec::new(),
_impact_model: MarketImpactModel::new(),
_execution_schedule: Vec::new(),
})
}
}
@@ -1288,9 +1269,9 @@ impl MarketImpactModel {
/// Create a new market impact model
pub fn new() -> Self {
Self {
temp_impact_coeff: 0.01_f64,
perm_impact_coeff: 0.001_f64,
volatility: 0.02_f64,
_temp_impact_coeff: 0.01_f64,
_perm_impact_coeff: 0.001_f64,
_volatility: 0.02_f64,
}
}
}

32
adaptive-strategy/src/models/deep_learning.rs
View File

@@ -293,12 +293,10 @@ impl ModelTrait for LSTMModel {
#[derive(Debug)]
pub struct GRUModel {
name: String,
/// Model `configuration` (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
/// Model configuration (stub for future ML integration)
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl GRUModel {
@@ -320,8 +318,8 @@ impl GRUModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}
@@ -377,11 +375,9 @@ impl ModelTrait for GRUModel {
pub struct TransformerModel {
name: String,
/// Model configuration (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl TransformerModel {
@@ -389,8 +385,8 @@ impl TransformerModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}
@@ -446,11 +442,9 @@ impl ModelTrait for TransformerModel {
pub struct CNNModel {
name: String,
/// Model configuration (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl CNNModel {
@@ -458,8 +452,8 @@ impl CNNModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}

42
adaptive-strategy/src/models/traditional.rs
View File

@@ -11,12 +11,10 @@ use async_trait::async_trait;
#[derive(Debug)]
pub struct RandomForestModel {
name: String,
/// Model `configuration` (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
/// Model configuration (stub for future ML integration)
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl RandomForestModel {
@@ -24,8 +22,8 @@ impl RandomForestModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}
@@ -81,11 +79,9 @@ impl ModelTrait for RandomForestModel {
pub struct XGBoostModel {
name: String,
/// Model configuration (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl XGBoostModel {
@@ -93,8 +89,8 @@ impl XGBoostModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}
@@ -152,11 +148,9 @@ impl ModelTrait for XGBoostModel {
pub struct SVMModel {
name: String,
/// Model configuration (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl SVMModel {
@@ -164,8 +158,8 @@ impl SVMModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}
@@ -223,11 +217,9 @@ impl ModelTrait for SVMModel {
pub struct LinearRegressionModel {
name: String,
/// Model configuration (stub for future ML integration)
#[allow(dead_code)]
config: ModelConfig,
_config: ModelConfig,
/// Model readiness flag (stub for future ML integration)
#[allow(dead_code)]
ready: bool,
_ready: bool,
}
impl LinearRegressionModel {
@@ -235,8 +227,8 @@ impl LinearRegressionModel {
pub async fn new(name: String, config: ModelConfig) -> Result<Self> {
Ok(Self {
name,
config,
ready: false,
_config: config,
_ready: false,
})
}
}

269
adaptive-strategy/src/risk/kelly_position_sizer.rs
View File

@@ -201,8 +201,7 @@ pub struct DynamicRiskAdjuster {
/// Portfolio drawdown tracker
drawdown_tracker: DrawdownTracker,
/// Volatility environment (reserved for volatility-adjusted sizing)
#[allow(dead_code)]
volatility_regime: VolatilityRegime,
_volatility_regime: VolatilityRegime,
}
// MarketRegime is already imported from ml::prelude at the top
@@ -225,225 +224,162 @@ pub enum VolatilityRegime {
/// Portfolio concentration monitoring
#[derive(Debug)]
#[allow(dead_code)]
pub(super) struct ConcentrationMonitor {
/// Current position concentrations by symbol
concentrations: HashMap<String, f64>,
/// Sector concentrations
#[allow(dead_code)]
sector_concentrations: HashMap<String, f64>,
_sector_concentrations: HashMap<String, f64>,
/// Geographic concentrations
#[allow(dead_code)]
geographic_concentrations: HashMap<String, f64>,
_geographic_concentrations: HashMap<String, f64>,
/// Asset class concentrations
#[allow(dead_code)]
asset_class_concentrations: HashMap<String, f64>,
_asset_class_concentrations: HashMap<String, f64>,
/// Correlation matrix
#[allow(dead_code)]
correlation_matrix: CorrelationMatrix,
_correlation_matrix: CorrelationMatrix,
}
/// Correlation matrix for position sizing adjustments
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub(super) struct CorrelationMatrix {
/// Symbols included in matrix
#[allow(dead_code)]
symbols: Vec<String>,
_symbols: Vec<String>,
/// Correlation coefficients (symmetric matrix)
#[allow(dead_code)]
correlations: Vec<Vec<f64>>,
_correlations: Vec<Vec<f64>>,
/// Last update timestamp
#[allow(dead_code)]
last_update: DateTime<Utc>,
_last_update: DateTime<Utc>,
/// Average correlation
#[allow(dead_code)]
avg_correlation: f64,
_avg_correlation: f64,
}
/// Volatility-based position optimization
#[derive(Debug)]
#[allow(dead_code)]
pub(super) struct VolatilityOptimizer {
/// Volatility estimates by symbol
volatility_estimates: HashMap<String, VolatilityEstimate>,
/// Target portfolio volatility
target_volatility: f64,
/// Current portfolio volatility
#[allow(dead_code)]
current_volatility: f64,
_current_volatility: f64,
/// Volatility forecasting model
#[allow(dead_code)]
volatility_model: VolatilityModel,
_volatility_model: VolatilityModel,
}
/// Volatility estimate with confidence intervals
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct VolatilityEstimate {
/// Current volatility estimate (annualized)
pub(super) current: f64,
/// 1-day ahead forecast
#[allow(dead_code)]
forecast_1d: f64,
pub(super) _forecast_1d: f64,
/// 5-day ahead forecast
#[allow(dead_code)]
forecast_5d: f64,
pub(super) _forecast_5d: f64,
/// Confidence interval (95%)
#[allow(dead_code)]
confidence_interval: (f64, f64),
pub(super) _confidence_interval: (f64, f64),
/// Model used for estimation
#[allow(dead_code)]
model_type: VolatilityModelType,
pub(super) _model_type: VolatilityModelType,
/// Last update timestamp
#[allow(dead_code)]
last_update: DateTime<Utc>,
pub(super) _last_update: DateTime<Utc>,
}
/// Volatility forecasting models
#[derive(Debug, Clone)]
pub(super) enum VolatilityModelType {
/// GARCH(1,1) model
#[allow(dead_code)]
Garch,
/// Exponentially weighted moving average
Ewma,
/// Range-based volatility
#[allow(dead_code)]
RangeBased,
/// Realized volatility
#[allow(dead_code)]
Realized,
}
/// Volatility forecasting model
#[derive(Debug)]
#[allow(dead_code)]
pub(super) struct VolatilityModel {
/// Model parameters
#[allow(dead_code)]
parameters: HashMap<String, f64>,
_parameters: HashMap<String, f64>,
/// Model type
#[allow(dead_code)]
model_type: VolatilityModelType,
_model_type: VolatilityModelType,
/// Calibration history
#[allow(dead_code)]
calibration_history: Vec<CalibrationRecord>,
_calibration_history: Vec<CalibrationRecord>,
}
/// Volatility model calibration record
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub(super) struct CalibrationRecord {
/// Calibration timestamp
#[allow(dead_code)]
timestamp: DateTime<Utc>,
_timestamp: DateTime<Utc>,
/// Model parameters at calibration
#[allow(dead_code)]
parameters: HashMap<String, f64>,
_parameters: HashMap<String, f64>,
/// In-sample error metrics
#[allow(dead_code)]
in_sample_error: f64,
_in_sample_error: f64,
/// Out-of-sample error metrics
#[allow(dead_code)]
out_of_sample_error: Option<f64>,
_out_of_sample_error: Option<f64>,
}
/// Portfolio drawdown tracking
#[derive(Debug)]
#[allow(dead_code)]
pub struct DrawdownTracker {
/// High water mark
#[allow(dead_code)]
high_water_mark: f64,
_high_water_mark: f64,
/// Current drawdown
#[allow(dead_code)]
current_drawdown: f64,
_current_drawdown: f64,
/// Maximum drawdown
#[allow(dead_code)]
max_drawdown: f64,
_max_drawdown: f64,
/// Drawdown start time
#[allow(dead_code)]
drawdown_start: Option<DateTime<Utc>>,
_drawdown_start: Option<DateTime<Utc>>,
/// Recovery factor (how much to reduce risk during drawdowns)
pub(super) recovery_factor: f64,
}
/// Performance tracking for Kelly optimization
#[derive(Debug)]
#[allow(dead_code)]
pub(super) struct PerformanceTracker {
/// Daily returns history
#[allow(dead_code)]
returns_history: Vec<DailyReturn>,
_returns_history: Vec<DailyReturn>,
/// Kelly sizing performance
pub(super) kelly_performance: KellyPerformanceMetrics,
/// Model accuracy tracking
#[allow(dead_code)]
accuracy_tracker: AccuracyTracker,
_accuracy_tracker: AccuracyTracker,
}
/// Daily return record
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub(super) struct DailyReturn {
/// Date
#[allow(dead_code)]
date: NaiveDate,
_date: NaiveDate,
/// Portfolio return
#[allow(dead_code)]
portfolio_return: f64,
_portfolio_return: f64,
/// Kelly-sized positions return
#[allow(dead_code)]
kelly_return: f64,
_kelly_return: f64,
/// Attribution by position
#[allow(dead_code)]
position_attribution: HashMap<String, f64>,
_position_attribution: HashMap<String, f64>,
}
/// Kelly performance metrics
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub struct KellyPerformanceMetrics {
/// Sharpe ratio
#[allow(dead_code)]
sharpe_ratio: f64,
_sharpe_ratio: f64,
/// Sortino ratio
#[allow(dead_code)]
sortino_ratio: f64,
_sortino_ratio: f64,
/// Maximum drawdown
#[allow(dead_code)]
max_drawdown: f64,
_max_drawdown: f64,
/// Calmar ratio
#[allow(dead_code)]
calmar_ratio: f64,
_calmar_ratio: f64,
/// Win rate
#[allow(dead_code)]
win_rate: f64,
_win_rate: f64,
/// Average win/loss ratio
#[allow(dead_code)]
win_loss_ratio: f64,
_win_loss_ratio: f64,
/// Kelly criterion effectiveness
#[allow(dead_code)]
kelly_effectiveness: f64,
_kelly_effectiveness: f64,
}
/// Model accuracy tracking
#[derive(Debug)]
#[allow(dead_code)]
pub(super) struct AccuracyTracker {
/// Prediction accuracy by horizon
#[allow(dead_code)]
accuracy_by_horizon: HashMap<String, f64>,
_accuracy_by_horizon: HashMap<String, f64>,
/// Calibration score
#[allow(dead_code)]
calibration_score: f64,
_calibration_score: f64,
/// Information coefficient
#[allow(dead_code)]
information_coefficient: f64,
_information_coefficient: f64,
/// Hit rate
#[allow(dead_code)]
hit_rate: f64,
_hit_rate: f64,
}
/// Enhanced Kelly position recommendation
@@ -639,61 +575,8 @@ impl KellyPositionSizer {
Ok(recommendation)
}
/// Calculate base Kelly fraction using multiple methods
#[allow(dead_code)]
fn calculate_base_kelly(
&self,
_symbol: &str,
expected_return: f64,
historical_returns: &[f64],
) -> Result<f64> {
if historical_returns.is_empty() {
return Ok(0.0);
}
// Method 1: Classic Kelly formula
let variance = self.calculate_variance(historical_returns);
let classic_kelly = if variance > 0.0 {
expected_return / variance
} else {
0.0
};
// Method 2: Win/loss statistics Kelly
let (win_rate, avg_win, avg_loss) = self.calculate_win_loss_stats(historical_returns);
let empirical_kelly = if avg_loss > 0.0 {
let odds = avg_win / avg_loss;
(win_rate * odds - (1.0 - win_rate)) / odds
} else {
0.0
};
// Method 3: Fractional Kelly for safety
let fractional_kelly = classic_kelly * 0.5; // Half Kelly for safety
// Combine methods with weighting
let combined_kelly = 0.4 * classic_kelly + 0.4 * empirical_kelly + 0.2 * fractional_kelly;
Ok(combined_kelly.clamp(0.0, self.config.max_fraction))
}
/// Calculate variance of historical returns
#[allow(dead_code)]
fn calculate_variance(&self, returns: &[f64]) -> f64 {
if returns.len() < 2 {
return 0.0;
}
let mean = returns.iter().sum::<f64>() / returns.len() as f64;
let variance =
returns.iter().map(|r| (r - mean).powi(2)).sum::<f64>() / returns.len() as f64;
variance
}
/// Calculate win/loss statistics
#[allow(dead_code)]
fn calculate_win_loss_stats(&self, returns: &[f64]) -> (f64, f64, f64) {
pub fn calculate_win_loss_stats(&self, returns: &[f64]) -> (f64, f64, f64) {
let wins: Vec<f64> = returns.iter().filter(|&&r| r > 0.0).copied().collect();
let losses: Vec<f64> = returns.iter().filter(|&&r| r < 0.0).map(|r| -r).collect();
@@ -814,7 +697,7 @@ impl DynamicRiskAdjuster {
current_regime: MarketRegime::Unknown,
regime_scalers,
drawdown_tracker: DrawdownTracker::new(config),
volatility_regime: VolatilityRegime::Normal,
_volatility_regime: VolatilityRegime::Normal,
})
}
@@ -903,10 +786,10 @@ impl ConcentrationMonitor {
pub(super) fn new(_config: &KellyConfig) -> Result<Self> {
Ok(Self {
concentrations: HashMap::new(),
sector_concentrations: HashMap::new(),
geographic_concentrations: HashMap::new(),
asset_class_concentrations: HashMap::new(),
correlation_matrix: CorrelationMatrix::new(),
_sector_concentrations: HashMap::new(),
_geographic_concentrations: HashMap::new(),
_asset_class_concentrations: HashMap::new(),
_correlation_matrix: CorrelationMatrix::new(),
})
}
@@ -977,8 +860,8 @@ impl VolatilityOptimizer {
Ok(Self {
volatility_estimates: HashMap::new(),
target_volatility: 0.15, // 15% target volatility
current_volatility: 0.0,
volatility_model: VolatilityModel::new()?,
_current_volatility: 0.0,
_volatility_model: VolatilityModel::new()?,
})
}
@@ -1014,9 +897,9 @@ impl VolatilityOptimizer {
impl VolatilityModel {
pub(super) fn new() -> Result<Self> {
Ok(Self {
parameters: HashMap::new(),
model_type: VolatilityModelType::Ewma,
calibration_history: Vec::new(),
_parameters: HashMap::new(),
_model_type: VolatilityModelType::Ewma,
_calibration_history: Vec::new(),
})
}
}
@@ -1036,10 +919,10 @@ impl DrawdownTracker {
/// A new `DrawdownTracker` instance ready for monitoring
pub fn new(_config: &KellyConfig) -> Self {
Self {
high_water_mark: 100000.0, // Initial portfolio value
current_drawdown: 0.0,
max_drawdown: 0.0,
drawdown_start: None,
_high_water_mark: 100000.0, // Initial portfolio value
_current_drawdown: 0.0,
_max_drawdown: 0.0,
_drawdown_start: None,
recovery_factor: 1.0,
}
}
@@ -1048,9 +931,9 @@ impl DrawdownTracker {
impl PerformanceTracker {
pub(super) fn new() -> Result<Self> {
Ok(Self {
returns_history: Vec::new(),
_returns_history: Vec::new(),
kelly_performance: KellyPerformanceMetrics::default(),
accuracy_tracker: AccuracyTracker::new(),
_accuracy_tracker: AccuracyTracker::new(),
})
}
@@ -1066,13 +949,13 @@ impl PerformanceTracker {
impl Default for KellyPerformanceMetrics {
fn default() -> Self {
Self {
sharpe_ratio: 0.0,
sortino_ratio: 0.0,
max_drawdown: 0.0,
calmar_ratio: 0.0,
win_rate: 0.0,
win_loss_ratio: 0.0,
kelly_effectiveness: 0.0,
_sharpe_ratio: 0.0,
_sortino_ratio: 0.0,
_max_drawdown: 0.0,
_calmar_ratio: 0.0,
_win_rate: 0.0,
_win_loss_ratio: 0.0,
_kelly_effectiveness: 0.0,
}
}
}
@@ -1080,10 +963,10 @@ impl Default for KellyPerformanceMetrics {
impl AccuracyTracker {
pub(super) fn new() -> Self {
Self {
accuracy_by_horizon: HashMap::new(),
calibration_score: 0.0,
information_coefficient: 0.0,
hit_rate: 0.0,
_accuracy_by_horizon: HashMap::new(),
_calibration_score: 0.0,
_information_coefficient: 0.0,
_hit_rate: 0.0,
}
}
}
@@ -1091,10 +974,10 @@ impl AccuracyTracker {
impl CorrelationMatrix {
pub(super) fn new() -> Self {
Self {
symbols: Vec::new(),
correlations: Vec::new(),
last_update: Utc::now(),
avg_correlation: 0.0,
_symbols: Vec::new(),
_correlations: Vec::new(),
_last_update: Utc::now(),
_avg_correlation: 0.0,
}
}
}

43
adaptive-strategy/src/risk/mod.rs
View File

@@ -120,16 +120,13 @@ pub struct RiskLimits {
#[derive(Debug, Clone)]
pub struct PnLTracker {
/// Daily P&L history
#[allow(dead_code)]
daily_pnl: Vec<DailyPnL>,
_daily_pnl: Vec<DailyPnL>,
/// Current session P&L
#[allow(dead_code)]
session_pnl: f64,
_session_pnl: f64,
/// Total portfolio value
portfolio_value: f64,
/// High-water mark for drawdown calculation
#[allow(dead_code)]
high_water_mark: f64,
_high_water_mark: f64,
}
/// Daily P&L record
@@ -153,16 +150,12 @@ pub struct DrawdownCalculator {
/// Portfolio value history
value_history: Vec<(chrono::DateTime<chrono::Utc>, f64)>,
/// Current drawdown
#[allow(dead_code)]
current_drawdown: f64,
/// Maximum drawdown
#[allow(dead_code)]
max_drawdown: f64,
/// High-water mark
#[allow(dead_code)]
high_water_mark: f64,
/// Drawdown start time
#[allow(dead_code)]
drawdown_start: Option<chrono::DateTime<chrono::Utc>>,
}
@@ -172,11 +165,9 @@ pub struct RiskMetricsCalculator {
/// Historical price data
price_history: HashMap<String, Vec<PricePoint>>,
/// Portfolio returns history
#[allow(dead_code)]
portfolio_returns: Vec<f64>,
_portfolio_returns: Vec<f64>,
/// Confidence levels for VaR calculation
#[allow(dead_code)]
confidence_levels: Vec<f64>,
_confidence_levels: Vec<f64>,
}
/// Price point for historical data
@@ -190,20 +181,6 @@ pub struct PricePoint {
pub volume: f64,
}
/// Correlation matrix for risk calculations
#[derive(Debug, Clone)]
pub struct CorrelationMatrix {
/// Asset symbols
#[allow(dead_code)]
symbols: Vec<String>,
/// Correlation coefficients
#[allow(dead_code)]
correlations: Vec<Vec<f64>>,
/// Last update timestamp
#[allow(dead_code)]
last_update: chrono::DateTime<chrono::Utc>,
}
/// Risk adjustment record
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RiskAdjustment {
@@ -1277,10 +1254,10 @@ impl PnLTracker {
/// Create a new P&L tracker
pub fn new(initial_value: f64) -> Self {
Self {
daily_pnl: Vec::new(),
session_pnl: 0.0,
_daily_pnl: Vec::new(),
_session_pnl: 0.0,
portfolio_value: initial_value,
high_water_mark: initial_value,
_high_water_mark: initial_value,
}
}
}
@@ -1340,8 +1317,8 @@ impl RiskMetricsCalculator {
pub fn new() -> Result<Self> {
Ok(Self {
price_history: HashMap::new(),
portfolio_returns: Vec::new(),
confidence_levels: vec![0.95_f64, 0.99_f64], // 95% and 99% confidence levels
_portfolio_returns: Vec::new(),
_confidence_levels: vec![0.95_f64, 0.99_f64], // 95% and 99% confidence levels
})
}

10
adaptive-strategy/src/risk/tests.rs
View File

@@ -339,11 +339,11 @@ async fn test_volatility_estimates_update() {
let mut estimates = HashMap::new();
estimates.insert(TEST_SYMBOL_1.to_string(), kelly_position_sizer::VolatilityEstimate {
current: 0.18,
forecast_1d: 0.19,
forecast_5d: 0.20,
confidence_interval: (0.15, 0.22),
model_type: kelly_position_sizer::VolatilityModelType::Garch,
last_update: chrono::Utc::now(),
_forecast_1d: 0.19,
_forecast_5d: 0.20,
_confidence_interval: (0.15, 0.22),
_model_type: kelly_position_sizer::VolatilityModelType::Ewma,
_last_update: chrono::Utc::now(),
});
let result = sizer.update_volatility_estimates(estimates).await;

17
common/src/lib.rs
View File

@@ -27,6 +27,7 @@
pub mod constants;
pub mod database;
pub mod error;
pub mod model_types;
pub mod features; // Wave D: Shared feature extraction (225 features)
pub mod market_data;
pub mod metrics; // Wave 5: Prometheus metrics infrastructure (W5-3)
@@ -35,6 +36,7 @@ pub mod observability; // Wave 5: Structured logging and observability
pub mod regime_persistence;
pub mod resilience; // Wave 5: Circuit breaker and retry patterns
pub mod thresholds;
pub mod tls;
pub mod traits;
pub mod types;
@@ -55,21 +57,17 @@ pub use types::{
// Re-export error types
pub use error::{CommonError, CommonResult};
// Re-export model types
pub use model_types::ModelType;
// Re-export common traits for convenience
pub use traits::{
CircuitBreaker, Configurable, DetailedHealth, GracefulShutdown, HealthCheck, HealthStatus,
Metrics, RateLimitStatus, RateLimited, Reloadable, Service,
};
pub use market_data::{
BarEvent as BarEventFromMarketData, BarInterval,
MarketDataEvent as MarketDataEventFromMarketData, NewsEvent,
OrderBookEvent as OrderBookEventFromMarketData, QuoteEvent as QuoteEventFromMarketData,
TradeEvent as TradeEventFromMarketData,
};
// Import market data types for canonical use
// Use common::market_data::{MarketDataEvent, TradeEvent, QuoteEvent, BarEvent} etc.
// Market data types: use common::market_data::{MarketDataEvent, TradeEvent, QuoteEvent, BarEvent, etc.}
// (Not re-exported at crate root to avoid name collisions with types:: re-exports)
pub mod trading;
// Re-export shared ML strategy types
@@ -93,7 +91,6 @@ pub use features::{
pub use resilience::{
BoundedExecutor,
BoundedExecutorError,
CircuitBreaker as ResilienceCircuitBreaker,
CircuitBreakerConfig,
CircuitBreakerState,
retry_with_backoff,

220
common/src/model_types.rs Normal file
View File

@@ -0,0 +1,220 @@
//! Canonical model type definitions for the Foxhunt trading system.
//!
//! This module provides the unified `ModelType` enum used across all crates
//! (ml, model_loader, services) to identify ML model architectures.
use serde::{Deserialize, Serialize};
use std::fmt;
/// Canonical model type enum used throughout the Foxhunt system.
///
/// This is the single source of truth for model type identification.
/// All crates should import this type rather than defining their own.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum ModelType {
/// Compact Deep Q-Network
CompactDQN,
/// Distilled micro network for ultra-low latency
DistilledMicroNet,
/// Standard Deep Q-Network
DQN,
/// Rainbow DQN with all enhancements
RainbowDQN,
/// MAMBA/Mamba2 state space model
MAMBA,
/// Temporal Fusion Transformer
TFT,
/// Temporal Graph Neural Network
TGGN,
/// Liquid Neural Network
LNN,
/// Temporal Limit Order Book transformer
TLOB,
/// Proximal Policy Optimization
PPO,
/// Transformer for sequence modeling
Transformer,
/// Mamba state space model (alias for MAMBA)
Mamba,
/// Liquid time constant networks (alias for LNN)
LiquidNet,
/// Temporal Graph Neural Network (alias for TGGN)
TGNN,
/// Ensemble of multiple models
Ensemble,
}
impl ModelType {
/// Get file extension for model type (used for checkpoint filenames).
pub fn file_extension(&self) -> &'static str {
match self {
Self::DQN => "dqn",
Self::MAMBA | Self::Mamba => "mamba",
Self::TFT => "tft",
Self::TGGN | Self::TGNN => "tggn",
Self::LNN | Self::LiquidNet => "lnn",
Self::CompactDQN => "compact_dqn",
Self::DistilledMicroNet => "distilled",
Self::RainbowDQN => "rainbow_dqn",
Self::TLOB => "tlob",
Self::PPO => "ppo",
Self::Transformer => "transformer",
Self::Ensemble => "ensemble",
}
}
/// String representation of the model type (lowercase, unique per variant family).
pub const fn as_str(&self) -> &'static str {
match self {
Self::DQN | Self::CompactDQN | Self::RainbowDQN => "dqn",
Self::MAMBA | Self::Mamba => "mamba",
Self::TFT => "tft",
Self::TGGN | Self::TGNN => "tggn",
Self::LNN | Self::LiquidNet => "liquid",
Self::TLOB => "tlob",
Self::PPO => "ppo",
Self::DistilledMicroNet => "distilled",
Self::Transformer => "transformer",
Self::Ensemble => "ensemble",
}
}
/// S3 storage prefix for model files (used by model_loader).
pub const fn s3_prefix(&self) -> &'static str {
match self {
Self::TLOB => "tlob_transformer",
Self::DQN | Self::CompactDQN | Self::RainbowDQN => "dqn",
Self::MAMBA | Self::Mamba => "mamba2",
Self::TFT => "tft",
Self::PPO => "ppo",
Self::LNN | Self::LiquidNet => "liquid",
Self::Ensemble => "ensemble",
Self::TGGN | Self::TGNN => "tggn",
Self::DistilledMicroNet => "distilled",
Self::Transformer => "transformer",
}
}
/// Convert to database string representation (for PostgreSQL storage).
pub fn to_db_string(&self) -> &'static str {
match self {
Self::DQN => "DQN",
Self::PPO => "PPO",
Self::MAMBA | Self::Mamba => "MAMBA-2",
Self::TFT => "TFT",
Self::CompactDQN => "COMPACT_DQN",
Self::DistilledMicroNet => "DISTILLED_MICRO_NET",
Self::RainbowDQN => "RAINBOW_DQN",
Self::TGGN | Self::TGNN => "TGGN",
Self::LNN | Self::LiquidNet => "LNN",
Self::TLOB => "TLOB",
Self::Transformer => "TRANSFORMER",
Self::Ensemble => "ENSEMBLE",
}
}
/// Get model weight for progress aggregation.
pub fn weight(&self) -> f64 {
0.25 // Equal weight for all models
}
/// Parse model type from string (case-insensitive).
#[allow(clippy::should_implement_trait)]
pub fn from_str(s: &str) -> Option<Self> {
match s.to_lowercase().as_str() {
"dqn" => Some(Self::DQN),
"mamba" | "mamba2" | "mamba-2" => Some(Self::MAMBA),
"tft" => Some(Self::TFT),
"tggn" | "tgnn" => Some(Self::TGGN),
"lnn" | "liquidnet" | "liquid" => Some(Self::LNN),
"compact_dqn" | "compactdqn" => Some(Self::CompactDQN),
"distilled" | "distilledmicronet" => Some(Self::DistilledMicroNet),
"rainbow_dqn" | "rainbowdqn" => Some(Self::RainbowDQN),
"tlob" | "tlob_transformer" => Some(Self::TLOB),
"ppo" => Some(Self::PPO),
"transformer" => Some(Self::Transformer),
"ensemble" => Some(Self::Ensemble),
_ => None,
}
}
}
impl fmt::Display for ModelType {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(f, "{}", self.as_str())
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_file_extension() {
assert_eq!(ModelType::DQN.file_extension(), "dqn");
assert_eq!(ModelType::PPO.file_extension(), "ppo");
assert_eq!(ModelType::TFT.file_extension(), "tft");
assert_eq!(ModelType::MAMBA.file_extension(), "mamba");
assert_eq!(ModelType::Mamba.file_extension(), "mamba");
assert_eq!(ModelType::Ensemble.file_extension(), "ensemble");
}
#[test]
fn test_as_str() {
assert_eq!(ModelType::TLOB.as_str(), "tlob");
assert_eq!(ModelType::DQN.as_str(), "dqn");
assert_eq!(ModelType::MAMBA.as_str(), "mamba");
assert_eq!(ModelType::TFT.as_str(), "tft");
assert_eq!(ModelType::PPO.as_str(), "ppo");
assert_eq!(ModelType::LNN.as_str(), "liquid");
assert_eq!(ModelType::Ensemble.as_str(), "ensemble");
}
#[test]
fn test_s3_prefix() {
assert_eq!(ModelType::TLOB.s3_prefix(), "tlob_transformer");
assert_eq!(ModelType::DQN.s3_prefix(), "dqn");
assert_eq!(ModelType::MAMBA.s3_prefix(), "mamba2");
assert_eq!(ModelType::TFT.s3_prefix(), "tft");
}
#[test]
fn test_to_db_string() {
assert_eq!(ModelType::DQN.to_db_string(), "DQN");
assert_eq!(ModelType::PPO.to_db_string(), "PPO");
assert_eq!(ModelType::MAMBA.to_db_string(), "MAMBA-2");
assert_eq!(ModelType::TFT.to_db_string(), "TFT");
}
#[test]
fn test_weight() {
assert_eq!(ModelType::DQN.weight(), 0.25);
assert_eq!(ModelType::PPO.weight(), 0.25);
}
#[test]
fn test_display() {
assert_eq!(format!("{}", ModelType::DQN), "dqn");
assert_eq!(format!("{}", ModelType::PPO), "ppo");
assert_eq!(format!("{}", ModelType::MAMBA), "mamba");
assert_eq!(format!("{}", ModelType::TLOB), "tlob");
}
#[test]
fn test_from_str() {
assert_eq!(ModelType::from_str("dqn"), Some(ModelType::DQN));
assert_eq!(ModelType::from_str("DQN"), Some(ModelType::DQN));
assert_eq!(ModelType::from_str("mamba2"), Some(ModelType::MAMBA));
assert_eq!(ModelType::from_str("ppo"), Some(ModelType::PPO));
assert_eq!(ModelType::from_str("unknown"), None);
}
#[test]
fn test_serde_roundtrip() {
let model = ModelType::DQN;
let json = serde_json::to_string(&model).unwrap_or_default();
let deserialized: ModelType =
serde_json::from_str(&json).unwrap_or(ModelType::DQN);
assert_eq!(model, deserialized);
}
}

47
common/src/resilience/circuit_breaker.rs
View File

@@ -47,6 +47,30 @@ use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::Mutex;
/// Core circuit breaker trait for the state machine pattern.
///
/// Implementors manage the Closed -> Open -> HalfOpen state transitions.
/// This trait provides a common interface so different circuit breaker
/// implementations (e.g., tokio-based, parking_lot-based) can be used
/// polymorphically.
#[async_trait::async_trait]
pub trait CircuitBreakerTrait: Send + Sync {
/// Check if a request is allowed through.
async fn can_execute(&self) -> bool;
/// Record a successful operation.
async fn record_success(&self);
/// Record a failed operation.
async fn record_failure(&self);
/// Get current state.
async fn state(&self) -> CircuitBreakerState;
/// Reset to closed state.
async fn reset(&self);
}
/// Circuit breaker states
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum CircuitBreakerState {
@@ -397,3 +421,26 @@ impl std::fmt::Debug for CircuitBreaker {
.finish()
}
}
#[async_trait::async_trait]
impl CircuitBreakerTrait for CircuitBreaker {
async fn can_execute(&self) -> bool {
CircuitBreaker::can_execute(self).await
}
async fn record_success(&self) {
CircuitBreaker::record_success(self).await;
}
async fn record_failure(&self) {
CircuitBreaker::record_failure(self).await;
}
async fn state(&self) -> CircuitBreakerState {
CircuitBreaker::state(self).await
}
async fn reset(&self) {
CircuitBreaker::reset(self).await;
}
}

4
common/src/resilience/mod.rs
View File

@@ -61,5 +61,7 @@ mod tests;
// Re-export commonly used types
pub use bounded_concurrency::{BoundedExecutor, BoundedExecutorError};
pub use circuit_breaker::{CircuitBreaker, CircuitBreakerConfig, CircuitBreakerState};
pub use circuit_breaker::{
CircuitBreaker, CircuitBreakerConfig, CircuitBreakerState, CircuitBreakerTrait,
};
pub use retry::{retry_with_backoff, RetryConfig, RetryContext, RetryError};

178
common/src/tls.rs Normal file
View File

@@ -0,0 +1,178 @@
//! Shared TLS types for Foxhunt services.
//!
//! This module contains type definitions shared across all service TLS configurations:
//! - [`TlsProtocolVersion`] - TLS 1.2 / 1.3 selection
//! - [`UserRole`] - RBAC roles extracted from client certificates
//! - [`ClientIdentity`] - Client identity extracted from mTLS certificates
//!
//! Each service retains its own `TlsConfig` struct and validation logic because
//! the validation approaches differ (async vs sync, delegated vs monolithic).
/// TLS protocol version options.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(dead_code)]
pub enum TlsProtocolVersion {
/// TLS version 1.2
Tls12,
/// TLS version 1.3
Tls13,
}
/// User roles based on certificate attributes (RBAC).
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(dead_code)]
pub enum UserRole {
/// Administrator with full access
Admin,
/// Trader with trading permissions
Trader,
/// Analyst with read/analysis permissions
Analyst,
/// Risk manager with risk oversight
RiskManager,
/// Compliance officer with audit access
ComplianceOfficer,
/// Read-only access
ReadOnly,
}
#[allow(dead_code)]
impl UserRole {
/// Get permissions for this role.
pub fn get_permissions(&self) -> Vec<&'static str> {
match self {
UserRole::Admin => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"analytics.run_backtest",
"compliance.view_reports",
"system.configure",
],
UserRole::Trader => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"analytics.view_data",
],
UserRole::Analyst => vec![
"analytics.view_data",
"analytics.run_backtest",
"risk.view_positions",
],
UserRole::RiskManager => vec![
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"compliance.view_reports",
],
UserRole::ComplianceOfficer => vec![
"compliance.view_reports",
"analytics.view_data",
"risk.view_positions",
],
UserRole::ReadOnly => vec!["analytics.view_data"],
}
}
}
/// Client identity extracted from mTLS certificate.
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(dead_code)]
pub struct ClientIdentity {
/// Common Name (CN) from certificate
pub common_name: String,
/// Organizational Unit (OU) from certificate
pub organizational_unit: String,
/// Certificate serial number
pub serial_number: String,
/// Certificate issuer
pub issuer: String,
}
#[allow(dead_code)]
impl ClientIdentity {
/// Check if client is authorized for trading operations.
pub fn is_authorized_for_trading(&self) -> bool {
matches!(self.organizational_unit.as_str(), "trading" | "admin")
}
/// Check if client is authorized for read-only operations.
pub fn is_authorized_for_readonly(&self) -> bool {
matches!(
self.organizational_unit.as_str(),
"trading" | "admin" | "analytics" | "risk" | "compliance"
)
}
/// Get user role based on certificate.
pub fn get_role(&self) -> UserRole {
match self.organizational_unit.as_str() {
"admin" => UserRole::Admin,
"trading" => UserRole::Trader,
"analytics" => UserRole::Analyst,
"risk" => UserRole::RiskManager,
"compliance" => UserRole::ComplianceOfficer,
_ => UserRole::ReadOnly,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_client_identity_authorization() {
let trading_identity = ClientIdentity {
common_name: "trader1.trading.foxhunt.internal".to_string(),
organizational_unit: "trading".to_string(),
serial_number: "12345".to_string(),
issuer: "Foxhunt Trading CA".to_string(),
};
assert!(trading_identity.is_authorized_for_trading());
assert!(trading_identity.is_authorized_for_readonly());
assert_eq!(trading_identity.get_role(), UserRole::Trader);
let readonly_identity = ClientIdentity {
common_name: "analyst1.analytics.foxhunt.internal".to_string(),
organizational_unit: "analytics".to_string(),
serial_number: "12346".to_string(),
issuer: "Foxhunt Trading CA".to_string(),
};
assert!(!readonly_identity.is_authorized_for_trading());
assert!(readonly_identity.is_authorized_for_readonly());
assert_eq!(readonly_identity.get_role(), UserRole::Analyst);
}
#[test]
fn test_user_role_permissions() {
let trader = UserRole::Trader;
let permissions = trader.get_permissions();
assert!(permissions.contains(&"trading.submit_order"));
assert!(permissions.contains(&"trading.cancel_order"));
assert!(!permissions.contains(&"system.configure"));
let readonly = UserRole::ReadOnly;
let readonly_permissions = readonly.get_permissions();
assert!(!readonly_permissions.contains(&"trading.submit_order"));
assert!(readonly_permissions.contains(&"analytics.view_data"));
}
#[test]
fn test_tls_protocol_version_equality() {
let tls13 = TlsProtocolVersion::Tls13;
let tls12 = TlsProtocolVersion::Tls12;
assert_ne!(tls13, tls12);
assert_eq!(tls13, TlsProtocolVersion::Tls13);
}
}

106
data/src/providers/benzinga/production_streaming.rs
View File

@@ -21,6 +21,7 @@ use crate::types::ExtendedMarketDataEvent;
use async_trait::async_trait;
use chrono::{DateTime, NaiveDate, NaiveDateTime, Utc};
use common::error::ErrorCategory;
use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
use common::{MarketDataEvent, Price, Quantity, Symbol};
use futures_core::Stream;
use futures_util::{SinkExt, StreamExt};
@@ -187,84 +188,6 @@ pub struct ProductionMetrics {
pub last_processed_at: RwLock<Option<DateTime<Utc>>>,
}
/// Circuit breaker states
#[derive(Debug, Clone, Copy)]
pub enum CircuitBreakerState {
/// Normal operation - requests flowing
Closed,
/// Blocking requests - failure threshold exceeded
Open,
/// Testing recovery - allowing limited requests
HalfOpen,
}
/// Circuit breaker for fault tolerance
#[derive(Debug)]
pub struct CircuitBreaker {
state: Arc<RwLock<CircuitBreakerState>>,
failure_count: Arc<AtomicU64>,
threshold: u32,
timeout: Duration,
last_failure: Arc<RwLock<Option<Instant>>>,
}
impl CircuitBreaker {
/// Create new circuit breaker with failure threshold and timeout
pub fn new(threshold: u32, timeout: Duration) -> Self {
Self {
state: Arc::new(RwLock::new(CircuitBreakerState::Closed)),
failure_count: Arc::new(AtomicU64::new(0)),
threshold,
timeout,
last_failure: Arc::new(RwLock::new(None)),
}
}
/// Check if call is permitted based on circuit breaker state
pub async fn is_call_permitted(&self) -> bool {
let state = *self.state.read().await;
match state {
CircuitBreakerState::Closed => true,
CircuitBreakerState::Open => {
let last_failure = *self.last_failure.read().await;
if let Some(failure_time) = last_failure {
if failure_time.elapsed() >= self.timeout {
// Try half-open
let mut state_guard = self.state.write().await;
*state_guard = CircuitBreakerState::HalfOpen;
true
} else {
false
}
} else {
false
}
},
CircuitBreakerState::HalfOpen => true,
}
}
/// Record successful call - resets circuit breaker to closed state
pub async fn record_success(&self) {
let mut state_guard = self.state.write().await;
*state_guard = CircuitBreakerState::Closed;
self.failure_count.store(0, Ordering::Relaxed);
}
/// Record failed call - may open circuit breaker if threshold exceeded
pub async fn record_failure(&self) {
let failures = self.failure_count.fetch_add(1, Ordering::Relaxed) + 1;
let mut last_failure_guard = self.last_failure.write().await;
*last_failure_guard = Some(Instant::now());
if failures >= self.threshold as u64 {
let mut state_guard = self.state.write().await;
*state_guard = CircuitBreakerState::Open;
}
}
}
/// Production Benzinga WebSocket streaming provider
#[derive(Debug)]
pub struct ProductionBenzingaProvider {
@@ -456,10 +379,12 @@ impl ProductionBenzingaProvider {
let rate_limiter = Arc::new(RateLimiter::direct(quota));
// Create circuit breaker
let circuit_breaker = Arc::new(CircuitBreaker::new(
config.circuit_breaker_threshold,
Duration::from_secs(config.circuit_breaker_timeout_secs),
));
let cb_config = CircuitBreakerConfig {
failure_threshold: config.circuit_breaker_threshold,
timeout: Duration::from_secs(config.circuit_breaker_timeout_secs),
success_threshold: 1,
};
let circuit_breaker = Arc::new(CircuitBreaker::new("benzinga", cb_config));
// Create processing semaphore
let processing_semaphore = Arc::new(Semaphore::new(config.max_concurrent_processing));
@@ -631,7 +556,7 @@ impl ProductionBenzingaProvider {
for message in messages {
// Check circuit breaker
if !self.circuit_breaker.is_call_permitted().await {
if !self.circuit_breaker.can_execute().await {
warn!("Circuit breaker open, skipping message processing");
continue;
}
@@ -1302,10 +1227,15 @@ mod tests {
#[tokio::test]
async fn test_circuit_breaker() {
let cb = CircuitBreaker::new(3, Duration::from_millis(100));
let config = CircuitBreakerConfig {
failure_threshold: 3,
timeout: Duration::from_millis(100),
success_threshold: 1,
};
let cb = CircuitBreaker::new("test_benzinga", config);
// Initially closed
assert!(cb.is_call_permitted().await);
assert!(cb.can_execute().await);
// Record failures to trip breaker
cb.record_failure().await;
@@ -1313,14 +1243,14 @@ mod tests {
cb.record_failure().await;
// Now should be open
assert!(!cb.is_call_permitted().await);
assert!(!cb.can_execute().await);
// Wait for timeout and check half-open
tokio::time::sleep(Duration::from_millis(150)).await;
assert!(cb.is_call_permitted().await);
assert!(cb.can_execute().await);
// Record success to close
cb.record_success().await;
assert!(cb.is_call_permitted().await);
assert!(cb.can_execute().await);
}
}

8
data/src/providers/common.rs
View File

@@ -47,14 +47,14 @@ use serde::{Deserialize, Serialize};
/// # Examples
///
/// ```rust
/// use data::providers::common::ErrorCategory;
/// use data::providers::common::ProviderErrorCategory;
///
/// match error_category {
/// ErrorCategory::RateLimit => {
/// ProviderErrorCategory::RateLimit => {
/// // Implement exponential backoff
/// tokio::time::sleep(Duration::from_millis(1000)).await;
/// },
/// ErrorCategory::Authentication => {
/// ProviderErrorCategory::Authentication => {
/// // Refresh credentials and reconnect
/// provider.refresh_auth().await?;
/// },
@@ -64,7 +64,7 @@ use serde::{Deserialize, Serialize};
/// }
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ErrorCategory {
pub enum ProviderErrorCategory {
/// Network connectivity issues, WebSocket disconnections
///
/// Indicates problems with the underlying network connection to the

176
docs/plans/2026-02-22-ml-ensemble-expansion-design.md Normal file
View File

@@ -0,0 +1,176 @@
# ML Ensemble Expansion Design: 7 → 10 Models
**Date**: 2026-02-22
**Status**: Approved
**Scope**: Expand ensemble from 4 fully-integrated + 3 partial models to 10 fully-integrated models
## Goal
Bring TGGN and TLOB to full integration (hyperopt + UnifiedTrainable), then add three new architectures (KAN, xLSTM, Diffusion) — each filling a distinct signal gap in the ensemble.
## Complementarity Analysis
| Capability Gap | Current Coverage | New Algorithm | Signal Type |
|---|---|---|---|
| Non-linear feature discovery | TFT attention (fixed activations) | **KAN** | Learnable activation functions discover non-obvious price/volume relationships |
| Long-range temporal memory | Mamba2 SSM, PPO LSTM (~1K steps) | **xLSTM** | Exponential gating + matrix memory handles 10K+ timestep dependencies |
| Probabilistic scenarios | TFT quantile regression (point-ish) | **Diffusion** | Generates distributions of future price paths for tail risk |
## Build Order (Sequential, one worktree per model)
1. TGGN full stack (gap-fill — code exists)
2. TLOB full stack (gap-fill — code exists)
3. KAN (new architecture)
4. xLSTM (new architecture)
5. Diffusion (new architecture)
Each model gets its own worktree branch → PR → merge before the next starts.
## Constraint: No God Classes
No file exceeds 500 lines. Each model decomposes into focused modules:
- `config.rs` — parameter structs
- `network.rs` — forward pass
- `trainable.rs` — UnifiedTrainable impl
- `trainer.rs` — training loop only
- Component files for model-specific pieces (e.g., `spline.rs` for KAN)
## Per-Model File Structure
```
ml/src/<model>/
├── mod.rs — public API, re-exports
├── config.rs — Config struct
├── <components>.rs — Model-specific pieces
├── network.rs — Neural network (forward pass)
├── trainable.rs — UnifiedTrainable impl
└── tests.rs — Unit tests (50+ per model)
ml/src/trainers/<model>.rs — Training loop + checkpointing
ml/src/hyperopt/adapters/<model>.rs — PSO hyperopt adapter (ParameterSpace trait)
```
Shared files touched per model:
- `ml/src/common/model_type.rs` — add ModelType variant
- `ml/src/inference.rs` — add inference branch
- `ml/src/integration/coordinator.rs` — register in ensemble
- `ml/src/lib.rs` — module declaration
- `ml/src/hyperopt/adapters/mod.rs` — adapter registration
## Model Designs
### 1. TGGN Full Stack (Gap-Fill)
**Existing**: `ml/src/tgnn/` — 8 files, graph convolution + message passing + inference + checkpoints.
**Add**:
- `ml/src/tgnn/trainable_adapter.rs` (~150 lines) — impl UnifiedTrainable
- `ml/src/hyperopt/adapters/tggn.rs` (~300 lines) — ParameterSpace: num_layers, hidden_dim, attention_heads, message_passing_steps, learning_rate, dropout
### 2. TLOB Full Stack (Gap-Fill)
**Existing**: `ml/src/tlob/` — 8 files, transformer + LOB features + data loader + trainer.
**Add**:
- `ml/src/tlob/trainable_adapter.rs` (~150 lines) — impl UnifiedTrainable
- `ml/src/hyperopt/adapters/tlob.rs` (~300 lines) — ParameterSpace: num_heads, d_model, num_encoder_layers, feature_window, learning_rate, dropout
### 3. KAN (Kolmogorov-Arnold Network) — New
**Purpose**: Non-linear relationship discovery via learnable B-spline activation functions.
**Modules**:
- `kan/config.rs` (~80 lines) — KANConfig
- `kan/spline.rs` (~200 lines) — B-spline basis functions (core innovation)
- `kan/layer.rs` (~150 lines) — KANLayer (replaces MLP layer with spline-based activations)
- `kan/network.rs` (~200 lines) — KANNetwork (stack of KANLayers + output head)
- `kan/trainable.rs` (~150 lines) — UnifiedTrainable impl
- `trainers/kan.rs` (~300 lines) — training loop
- `hyperopt/adapters/kan.rs` (~300 lines) — ParameterSpace: grid_size, spline_order, num_layers, width, learning_rate
**Key design**: B-spline control points are learned during training. Each edge in the network has its own learnable activation function, replacing fixed ReLU/GELU.
### 4. xLSTM — New
**Purpose**: Long-range temporal memory (10K+ timesteps) for regime persistence and macro cycles.
**Modules**:
- `xlstm/config.rs` (~80 lines) — xLSTMConfig
- `xlstm/slstm.rs` (~250 lines) — sLSTM cell (scalar memory, exponential gating)
- `xlstm/mlstm.rs` (~250 lines) — mLSTM cell (matrix memory, covariance-based updates)
- `xlstm/block.rs` (~150 lines) — xLSTMBlock (residual + pre-LayerNorm wrapper)
- `xlstm/network.rs` (~200 lines) — xLSTMNetwork (stack of blocks + prediction head)
- `xlstm/trainable.rs` (~150 lines) — UnifiedTrainable impl
- `trainers/xlstm.rs` (~350 lines) — training loop
- `hyperopt/adapters/xlstm.rs` (~300 lines) — ParameterSpace: num_blocks, head_dim, num_heads, slstm_ratio, learning_rate
**Key design**: Two cell types mixed in configurable ratio — sLSTM for precision (exponential gating), mLSTM for capacity (matrix memory). Extends existing PPO LSTM patterns but with fundamentally better long-range memory.
### 5. Diffusion Model — New
**Purpose**: Probabilistic scenario generation — distributions of future price paths for tail risk.
**Modules**:
- `diffusion/config.rs` (~80 lines) — DiffusionConfig
- `diffusion/noise.rs` (~150 lines) — noise scheduler (cosine/linear schedules)
- `diffusion/unet.rs` (~300 lines) — U-Net denoiser (time-conditioned)
- `diffusion/sampler.rs` (~200 lines) — DDPM/DDIM sampling for fast inference
- `diffusion/network.rs` (~150 lines) — DiffusionModel (wraps U-Net + scheduler)
- `diffusion/trainable.rs` (~150 lines) — UnifiedTrainable impl
- `trainers/diffusion.rs` (~350 lines) — training loop (denoising score matching)
- `hyperopt/adapters/diffusion.rs` (~300 lines) — ParameterSpace: num_timesteps, unet_channels, num_res_blocks, learning_rate, schedule_type
**Key design**: Learns to denoise price path sequences. At inference, generates multiple paths from noise via DDIM (10 steps for speed). Ensemble extracts: (a) median path for signal direction, (b) distribution width for confidence/volatility scaling.
## Ensemble Expansion
**Current**: 4 models at equal 0.25 weight with DynamicWeighting/AdaptiveEnsemble strategies.
**Target**: 10 models with initial equal 0.10 weight, dynamic weighting adjusts based on recent accuracy.
**ModelType enum additions**:
```rust
KAN,
XLSTM,
Diffusion,
```
**Diffusion special handling**: Output is a distribution, not a point prediction. Ensemble extracts median for signal direction and distribution width as confidence scaling factor.
**EnsembleCoordinator changes**: Register all 10 models. No structural changes needed — existing DynamicWeighting and AdaptiveEnsemble strategies scale to N models.
## Test Targets
| Model | New Tests | Total Target |
|-------|-----------|--------------|
| TGGN | +50 | 50+ (trainable + hyperopt) |
| TLOB | +50 | 50+ (trainable + hyperopt) |
| KAN | +80 | 80 (full stack) |
| xLSTM | +80 | 80 (full stack) |
| Diffusion | +80 | 80 (full stack) |
| Ensemble | +20 | 20 (10-model integration) |
| **Total** | **+360** | |
## Estimated Lines of Code
| Model | New Code | Category |
|-------|----------|----------|
| TGGN | ~450 | Gap-fill |
| TLOB | ~450 | Gap-fill |
| KAN | ~1,380 | New architecture |
| xLSTM | ~1,730 | New architecture |
| Diffusion | ~1,680 | New architecture |
| Shared (ModelType, inference, ensemble) | ~200 | Integration |
| **Total** | **~5,890** | |
## Dependencies
- Candle v0.9.1 (existing) — all models use candle tensors
- No new external crates required
- B-spline implementation (KAN) is self-contained
- U-Net (Diffusion) built from candle primitives
## Risks
1. **GPU memory (RTX 3050 Ti 4GB)**: 10-model ensemble inference may not fit simultaneously. Mitigation: sequential inference with model swapping, or batch by priority.
2. **Training time**: 5 new trainers × hyperopt tuning. Mitigation: sequential build means each model is tuned independently.
3. **Diffusion inference latency**: DDIM 10-step sampling is slower than single forward pass. Mitigation: async inference, configurable step count, skip in ultra-low-latency mode.

1843
docs/plans/2026-02-23-ml-ensemble-expansion-implementation.md Normal file
View File

File diff suppressed because it is too large Load Diff

63
ml/src/common/action.rs Normal file
View File

@@ -0,0 +1,63 @@
use crate::MLError;
use serde::{Deserialize, Serialize};
/// Order type for execution strategy.
///
/// Three execution modes with calibrated transaction costs (Wave 2.5).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum OrderType {
/// Immediate execution, taker fee (15 bps)
Market = 0,
/// Passive order, maker rebate (5 bps)
LimitMaker = 1,
/// Immediate-or-cancel (10 bps)
IoC = 2,
}
impl OrderType {
/// Transaction cost as decimal (e.g., 0.0015 for 15 bps).
pub fn transaction_cost(&self) -> f64 {
match self {
OrderType::Market => 0.0015,
OrderType::LimitMaker => 0.0005,
OrderType::IoC => 0.0010,
}
}
/// Transaction cost in basis points (e.g., 15.0 for Market).
pub fn cost_bps(&self) -> f32 {
match self {
OrderType::Market => 15.0,
OrderType::LimitMaker => 5.0,
OrderType::IoC => 10.0,
}
}
/// Cost as decimal from bps (convenience wrapper).
pub fn cost_decimal(&self) -> f64 {
(self.cost_bps() as f64) / 10000.0
}
/// Convert from index (0-2).
pub fn from_index(idx: usize) -> Result<Self, MLError> {
match idx {
0 => Ok(OrderType::Market),
1 => Ok(OrderType::LimitMaker),
2 => Ok(OrderType::IoC),
_ => Err(MLError::InvalidInput(format!(
"Invalid order type index: {}",
idx
))),
}
}
}
impl std::fmt::Display for OrderType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
OrderType::Market => write!(f, "Market"),
OrderType::LimitMaker => write!(f, "LimitMaker"),
OrderType::IoC => write!(f, "IoC"),
}
}
}

30
ml/src/common/circuit_breaker.rs
View File

@@ -8,6 +8,9 @@ use std::sync::atomic::{AtomicU32, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use ::common::resilience::circuit_breaker::{
CircuitBreakerState as CommonCBState, CircuitBreakerTrait,
};
use parking_lot::RwLock;
use tracing::{debug, info, warn};
@@ -253,6 +256,33 @@ impl CircuitBreaker {
}
}
#[async_trait::async_trait]
impl CircuitBreakerTrait for CircuitBreaker {
async fn can_execute(&self) -> bool {
self.allow_request()
}
async fn record_success(&self) {
CircuitBreaker::record_success(self);
}
async fn record_failure(&self) {
CircuitBreaker::record_failure(self);
}
async fn state(&self) -> CommonCBState {
match self.current_state() {
CircuitState::Closed => CommonCBState::Closed,
CircuitState::Open => CommonCBState::Open,
CircuitState::HalfOpen => CommonCBState::HalfOpen,
}
}
async fn reset(&self) {
CircuitBreaker::reset(self);
}
}
/// Circuit breaker statistics
#[derive(Debug, Clone)]
pub struct CircuitBreakerStats {

1
ml/src/common/mod.rs
View File

@@ -10,6 +10,7 @@ use uuid::Uuid;
use common::types::{Price, Quantity, Symbol, Volume};
use rust_decimal::Decimal;
pub mod action;
pub mod circuit_breaker;
pub mod config;
pub mod metrics;

44
ml/src/dqn/action_space.rs
View File

@@ -1,3 +1,4 @@
pub use crate::common::action::OrderType;
use crate::MLError;
use serde::{Deserialize, Serialize};
@@ -39,39 +40,6 @@ impl ExposureLevel {
}
}
/// Order type for execution strategy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum OrderType {
Market = 0, // Immediate execution, high cost (0.20%)
LimitMaker = 1, // Passive order, maker rebate (0.10%)
IoC = 2, // Immediate-or-cancel (0.15%)
}
impl OrderType {
/// Get transaction cost multiplier
/// Wave 2.5 Calibration: Reduced fees (Market 20→15 bps, LimitMaker 10→5 bps, IoC 15→10 bps)
pub fn transaction_cost(&self) -> f64 {
match self {
OrderType::Market => 0.0015, // 0.15% (was 0.20%)
OrderType::LimitMaker => 0.0005, // 0.05% (was 0.10%)
OrderType::IoC => 0.0010, // 0.10% (was 0.15%)
}
}
/// Convert from index (0-2)
pub fn from_index(idx: usize) -> Result<Self, MLError> {
match idx {
0 => Ok(OrderType::Market),
1 => Ok(OrderType::LimitMaker),
2 => Ok(OrderType::IoC),
_ => Err(MLError::InvalidInput(format!(
"Invalid order type index: {}",
idx
))),
}
}
}
/// Urgency level for execution timing
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum Urgency {
@@ -303,16 +271,6 @@ impl std::fmt::Display for ExposureLevel {
}
}
impl std::fmt::Display for OrderType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
OrderType::Market => write!(f, "Market"),
OrderType::LimitMaker => write!(f, "LimitMaker"),
OrderType::IoC => write!(f, "IoC"),
}
}
}
impl std::fmt::Display for Urgency {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {

2
ml/src/hyperopt/campaign.rs
View File

@@ -8,7 +8,7 @@ use std::path::PathBuf;
use serde::{Deserialize, Serialize};
// Re-export the canonical ModelType from the crate root
// Use canonical ModelType from crate root (re-exported from common)
pub use crate::ModelType;
/// Campaign configuration for multi-trial hyperparameter optimization.

419
ml/src/lib.rs
View File

@@ -49,11 +49,7 @@
)]
// Import common types properly - NO ALIASES THAT CONFLICT!
use candle_core::backprop::GradStore;
use candle_core::Tensor;
use candle_core::Var;
use candle_nn::Optimizer; // For Adam optimizer support
use serde::{Deserialize, Serialize}; // For tensor variables
use serde::{Deserialize, Serialize};
// Silence unused crate warnings for dependencies used in tests or feature-gated code
use approx as _;
@@ -66,264 +62,6 @@ use semver as _;
use tempfile as _;
use trading_engine as _;
/// Wrapper for Adam optimizer to provide required methods
///
/// This wrapper provides a unified interface around the candle_optimisers Adam optimizer,
/// ensuring consistent behavior across the ML crate and providing additional convenience methods.
/// Adam is an adaptive learning rate optimization algorithm that computes individual learning
/// rates for different parameters from estimates of first and second moments of the gradients.
///
/// # Examples
///
/// ```rust,no_run
/// use ml::Adam;
/// use candle_core::Var;
/// use candle_optimisers::adam::ParamsAdam;
///
/// let vars = vec![]; // Your model variables
/// let params = ParamsAdam::default();
/// let optimizer = Adam::new(vars, params)?;
/// # Ok::<(), ml::MLError>(())
/// ```
#[derive(Debug)]
pub struct Adam {
optimizer: candle_optimisers::adam::Adam,
learning_rate: f64,
vars: Vec<Var>,
}
impl Adam {
/// Create a new Adam optimizer with the given variables and parameters
///
/// # Arguments
///
/// * `vars` - Vector of model variables to optimize
/// * `params` - Adam optimizer parameters including learning rate, betas, and epsilon
///
/// # Returns
///
/// Returns `Ok(Adam)` on success, or `Err(MLError::TrainingError)` if optimizer creation fails
///
/// # Errors
///
/// This function will return an error if the underlying candle Adam optimizer fails to initialize
pub fn new(
vars: Vec<Var>,
params: candle_optimisers::adam::ParamsAdam,
) -> Result<Self, MLError> {
let learning_rate = params.lr;
let optimizer = candle_optimisers::adam::Adam::new(vars.clone(), params).map_err(|e| {
MLError::TrainingError(format!("Failed to create Adam optimizer: {}", e))
})?;
Ok(Self {
optimizer,
learning_rate,
vars,
})
}
/// Perform a backward pass and optimizer step
///
/// This method computes gradients via backpropagation and then applies the Adam
/// optimization update to all registered variables.
///
/// # Arguments
///
/// * `loss` - The loss tensor to compute gradients from
///
/// # Returns
///
/// Returns `Ok(())` on successful optimization step, or `Err(MLError::TrainingError)` on failure
///
/// # Errors
///
/// This function will return an error if:
/// - The backward pass fails to compute gradients
/// - The optimizer step fails to apply updates
pub fn backward_step(&mut self, loss: &Tensor) -> Result<(), MLError> {
// Calculate gradients
let grads = loss
.backward()
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// Apply optimizer step using trait method
Optimizer::step(&mut self.optimizer, &grads)
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
Ok(())
}
/// Get the learning rate used by this optimizer
///
/// # Returns
///
/// Returns the learning rate as a 64-bit floating point number
pub fn learning_rate(&self) -> f64 {
self.learning_rate
}
/// Get a reference to the variables tracked by this optimizer
///
/// # Returns
///
/// Returns a slice reference to the vector of variables
pub fn vars(&self) -> &[Var] {
&self.vars
}
/// Perform backward pass with gradient clipping
///
/// Implements proper gradient clipping by norm to prevent gradient explosions.
/// Uses a two-pass approach: first pass computes gradient norm, second pass
/// (if needed) computes clipped gradients by scaling the loss.
///
/// # Arguments
///
/// * `loss` - The loss tensor to compute gradients from
/// * `max_norm` - Maximum allowed gradient norm (gradients will be clipped to this value)
///
/// # Returns
///
/// Returns `Ok(gradient_norm)` on success with the actual gradient norm (before clipping),
/// or `Err(MLError::TrainingError)` on failure
pub fn backward_step_with_monitoring(
&mut self,
loss: &Tensor,
max_norm: f64,
) -> Result<f64, MLError> {
// Bug fix: Single backward pass to prevent gradient accumulation
// Root cause: Two backward passes caused 1.5x gradient amplification
// Solution: Compute gradients once, then scale loss before optimizer step if needed
// 1. Compute gradients via backward pass
let mut grads = loss
.backward()
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// 2. Apply gradient clipping IN-PLACE (Bug #32 fix - gradient explosion)
// This modifies the grads directly before optimizer step
let (actual_grad_norm, clipped_grad_norm) = crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm)
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
// BUG #14 FIX: Log actual gradient norms to detect if threshold is too low
// Temporary diagnostic logging to investigate gradient clipping issue
static GRAD_LOG_COUNTER: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
let count = GRAD_LOG_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if count < 20 || count % 100 == 0 {
tracing::info!(
"BUG #14 DIAGNOSTIC: Gradient norm BEFORE clipping: {:.4}, AFTER clipping: {:.4}, max_norm: {:.4}",
actual_grad_norm,
clipped_grad_norm,
max_norm
);
}
// 3. Apply optimizer step with clipped gradients
Optimizer::step(&mut self.optimizer, &grads)
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
if actual_grad_norm > max_norm {
tracing::warn!(
"Gradient clipping: {:.4} -> {:.4} - this is expected occasionally but should be rare. \
If frequent, consider reducing learning rate.",
actual_grad_norm,
clipped_grad_norm
);
}
Ok(clipped_grad_norm)
}
/// Perform backward pass with gradient clipping but WITHOUT an optimizer step.
///
/// This is useful for gradient accumulation workflows where you want to
/// accumulate clipped gradients across multiple micro-batches before
/// applying a single optimizer step.
///
/// # Arguments
///
/// * `loss` - The loss tensor to compute gradients from
/// * `max_norm` - Maximum allowed gradient norm (gradients will be clipped to this value)
///
/// # Returns
///
/// Returns `Ok((grads, clipped_norm))` on success, where `grads` is the clipped
/// `GradStore` and `clipped_norm` is the gradient norm after clipping.
///
/// # Errors
///
/// This function will return an error if:
/// - The backward pass fails to compute gradients
/// - Gradient clipping fails
pub fn backward_and_clip(
&self,
loss: &Tensor,
max_norm: f64,
) -> Result<(GradStore, f64), MLError> {
// Compute gradients via backward pass
let mut grads = loss
.backward()
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// Apply gradient clipping in-place
let (_actual_norm, clipped_norm) =
crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm)
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
Ok((grads, clipped_norm))
}
/// Apply pre-computed gradients to the optimizer.
///
/// This is the companion to `backward_and_clip` for gradient accumulation
/// workflows. After accumulating and scaling gradients, call this method
/// to perform the optimizer step.
///
/// # Arguments
///
/// * `grads` - Pre-computed (and possibly accumulated/scaled) gradients
///
/// # Errors
///
/// This function will return an error if the optimizer step fails
pub fn apply_grads(&mut self, grads: &GradStore) -> Result<(), MLError> {
Optimizer::step(&mut self.optimizer, grads)
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
Ok(())
}
/// Compute the L2 norm of all gradients
fn compute_gradient_norm(
&self,
grads: &candle_core::backprop::GradStore,
) -> Result<f64, MLError> {
let mut total_norm_sq = 0.0f64;
// Get all variables from the optimizer
for var in &self.vars {
if let Some(grad) = grads.get(var) {
// Compute L2 norm squared for this gradient
let grad_norm_sq = grad
.sqr()
.map_err(|e| {
MLError::TrainingError(format!("Failed to square gradient: {}", e))
})?
.sum_all()
.map_err(|e| MLError::TrainingError(format!("Failed to sum gradient: {}", e)))?
.to_vec0::<f32>()
.map_err(|e| {
MLError::TrainingError(format!("Failed to extract gradient norm: {}", e))
})? as f64;
total_norm_sq += grad_norm_sq;
}
}
Ok(total_norm_sq.sqrt())
}
}
// Direct type imports - no compatibility aliases
use rust_decimal::Decimal;
@@ -457,15 +195,11 @@ impl CommonError {
}
}
/// Error categories for system-wide error classification
///
/// This enum provides a way to categorize errors across the entire system,
/// enabling better error handling, logging, and monitoring strategies.
#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub enum ErrorCategory {
/// System-level errors including hardware, network, and infrastructure issues
System,
}
// Re-export canonical ErrorCategory from common crate (24 variants)
pub use ::common::error::ErrorCategory;
// Re-export Adam optimizer (moved from inline to optimizers/adam.rs)
pub use optimizers::Adam;
// Now using real types from common crate
@@ -963,6 +697,7 @@ pub mod liquid;
pub mod mamba;
pub mod memory_optimization; // Memory optimization utilities (lazy loading, quantization, precision)
pub mod microstructure;
pub mod optimizers;
pub mod paper_trading;
pub mod ppo;
pub mod preprocessing; // Data preprocessing (log returns, normalization, outlier clipping)
@@ -2175,144 +1910,8 @@ impl ModelMetadata {
}
}
/// Canonical model type enum used throughout ML module
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Hash)]
pub enum ModelType {
/// Compact Deep Q-Network
CompactDQN,
/// Distilled micro network for ultra-low latency
DistilledMicroNet,
/// Standard Deep Q-Network
DQN,
/// Rainbow `DQN` with all enhancements
RainbowDQN,
/// `MAMBA` model (SSM)
MAMBA,
/// Temporal Fusion Transformer
TFT,
/// Temporal Graph Neural Network
TGGN,
/// Liquid Neural Network
LNN,
/// Temporal Limit Order Book transformer
TLOB,
/// Proximal Policy Optimization
PPO,
/// Transformer for sequence modeling
Transformer,
/// Mamba state space model (alias for `MAMBA`)
Mamba,
/// Liquid time constant networks (alias for LNN)
LiquidNet,
/// Temporal Graph Neural Network (alias for TGGN)
TGNN,
/// Ensemble methods
Ensemble,
}
impl ModelType {
/// Get file extension for model type
pub fn file_extension(&self) -> &'static str {
match self {
ModelType::DQN => "dqn",
ModelType::MAMBA | ModelType::Mamba => "mamba",
ModelType::TFT => "tft",
ModelType::TGGN | ModelType::TGNN => "tggn",
ModelType::LNN | ModelType::LiquidNet => "lnn",
ModelType::CompactDQN => "compact_dqn",
ModelType::DistilledMicroNet => "distilled",
ModelType::RainbowDQN => "rainbow_dqn",
ModelType::TLOB => "tlob",
ModelType::PPO => "ppo",
ModelType::Transformer => "transformer",
ModelType::Ensemble => "ensemble",
}
}
/// Get model type as a lowercase string identifier (for display, metrics, etc.)
pub const fn as_str(&self) -> &'static str {
match self {
ModelType::CompactDQN => "compact_dqn",
ModelType::DistilledMicroNet => "distilled",
ModelType::DQN => "dqn",
ModelType::RainbowDQN => "rainbow_dqn",
ModelType::MAMBA | ModelType::Mamba => "mamba",
ModelType::TFT => "tft",
ModelType::TGGN | ModelType::TGNN => "tggn",
ModelType::LNN | ModelType::LiquidNet => "liquid",
ModelType::TLOB => "tlob",
ModelType::PPO => "ppo",
ModelType::Transformer => "transformer",
ModelType::Ensemble => "ensemble",
}
}
/// Get S3 storage prefix for model type (for S3 model loading)
pub const fn s3_prefix(&self) -> &'static str {
match self {
ModelType::CompactDQN => "compact_dqn",
ModelType::DistilledMicroNet => "distilled_micro_net",
ModelType::DQN => "dqn",
ModelType::RainbowDQN => "rainbow_dqn",
ModelType::MAMBA | ModelType::Mamba => "mamba2",
ModelType::TFT => "tft",
ModelType::TGGN | ModelType::TGNN => "tggn",
ModelType::LNN | ModelType::LiquidNet => "liquid",
ModelType::TLOB => "tlob_transformer",
ModelType::PPO => "ppo",
ModelType::Transformer => "transformer",
ModelType::Ensemble => "ensemble",
}
}
/// Convert to database string representation (for PostgreSQL storage)
pub fn to_db_string(&self) -> &'static str {
match self {
ModelType::DQN => "DQN",
ModelType::PPO => "PPO",
ModelType::MAMBA | ModelType::Mamba => "MAMBA-2",
ModelType::TFT => "TFT",
ModelType::CompactDQN => "COMPACT_DQN",
ModelType::DistilledMicroNet => "DISTILLED_MICRO_NET",
ModelType::RainbowDQN => "RAINBOW_DQN",
ModelType::TGGN | ModelType::TGNN => "TGGN",
ModelType::LNN | ModelType::LiquidNet => "LNN",
ModelType::TLOB => "TLOB",
ModelType::Transformer => "TRANSFORMER",
ModelType::Ensemble => "ENSEMBLE",
}
}
/// Get model weight for progress aggregation in batch training
pub fn weight(&self) -> f64 {
0.25 // Equal weight for all models
}
/// Get model type from string
pub fn from_str(s: &str) -> Option<Self> {
match s.to_lowercase().as_str() {
"dqn" => Some(ModelType::DQN),
"mamba" | "mamba2" => Some(ModelType::MAMBA),
"tft" => Some(ModelType::TFT),
"tggn" | "tgnn" => Some(ModelType::TGGN),
"lnn" | "liquidnet" | "liquid" => Some(ModelType::LNN),
"compact_dqn" | "compactdqn" => Some(ModelType::CompactDQN),
"distilled" | "distilledmicronet" => Some(ModelType::DistilledMicroNet),
"rainbow_dqn" | "rainbowdqn" => Some(ModelType::RainbowDQN),
"tlob" | "tlob_transformer" => Some(ModelType::TLOB),
"ppo" => Some(ModelType::PPO),
"transformer" => Some(ModelType::Transformer),
"ensemble" => Some(ModelType::Ensemble),
_ => None,
}
}
}
impl std::fmt::Display for ModelType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.as_str())
}
}
// Re-export canonical ModelType from common crate
pub use ::common::model_types::ModelType;
/// Prelude module for convenient imports of commonly used ML types
///

4
ml/src/observability/metrics.rs
View File

@@ -561,8 +561,8 @@ pub struct MLMetricsReport {
pub health_score: f64,
}
// Model type to string conversion for metrics is now provided by
// the Display impl on ModelType (via as_str()), which auto-generates ToString.
// ModelType already implements Display (via common::model_types),
// which automatically provides ToString.
/// Global metrics collector instance
static GLOBAL_METRICS: once_cell::sync::Lazy<Arc<RwLock<Option<MLMetricsCollector>>>> =

264
ml/src/optimizers/adam.rs Normal file
View File

@@ -0,0 +1,264 @@
use candle_core::backprop::GradStore;
use candle_core::Tensor;
use candle_core::Var;
use candle_nn::Optimizer;
use crate::MLError;
/// Wrapper for Adam optimizer to provide required methods
///
/// This wrapper provides a unified interface around the candle_optimisers Adam optimizer,
/// ensuring consistent behavior across the ML crate and providing additional convenience methods.
/// Adam is an adaptive learning rate optimization algorithm that computes individual learning
/// rates for different parameters from estimates of first and second moments of the gradients.
///
/// # Examples
///
/// ```rust,no_run
/// use ml::Adam;
/// use candle_core::Var;
/// use candle_optimisers::adam::ParamsAdam;
///
/// let vars = vec![]; // Your model variables
/// let params = ParamsAdam::default();
/// let optimizer = Adam::new(vars, params)?;
/// # Ok::<(), ml::MLError>(())
/// ```
#[derive(Debug)]
pub struct Adam {
optimizer: candle_optimisers::adam::Adam,
learning_rate: f64,
vars: Vec<Var>,
}
impl Adam {
/// Create a new Adam optimizer with the given variables and parameters
///
/// # Arguments
///
/// * `vars` - Vector of model variables to optimize
/// * `params` - Adam optimizer parameters including learning rate, betas, and epsilon
///
/// # Returns
///
/// Returns `Ok(Adam)` on success, or `Err(MLError::TrainingError)` if optimizer creation fails
///
/// # Errors
///
/// This function will return an error if the underlying candle Adam optimizer fails to initialize
pub fn new(
vars: Vec<Var>,
params: candle_optimisers::adam::ParamsAdam,
) -> Result<Self, MLError> {
let learning_rate = params.lr;
let optimizer = candle_optimisers::adam::Adam::new(vars.clone(), params).map_err(|e| {
MLError::TrainingError(format!("Failed to create Adam optimizer: {}", e))
})?;
Ok(Self {
optimizer,
learning_rate,
vars,
})
}
/// Perform a backward pass and optimizer step
///
/// This method computes gradients via backpropagation and then applies the Adam
/// optimization update to all registered variables.
///
/// # Arguments
///
/// * `loss` - The loss tensor to compute gradients from
///
/// # Returns
///
/// Returns `Ok(())` on successful optimization step, or `Err(MLError::TrainingError)` on failure
///
/// # Errors
///
/// This function will return an error if:
/// - The backward pass fails to compute gradients
/// - The optimizer step fails to apply updates
pub fn backward_step(&mut self, loss: &Tensor) -> Result<(), MLError> {
// Calculate gradients
let grads = loss
.backward()
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// Apply optimizer step using trait method
Optimizer::step(&mut self.optimizer, &grads)
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
Ok(())
}
/// Get the learning rate used by this optimizer
///
/// # Returns
///
/// Returns the learning rate as a 64-bit floating point number
pub fn learning_rate(&self) -> f64 {
self.learning_rate
}
/// Get a reference to the variables tracked by this optimizer
///
/// # Returns
///
/// Returns a slice reference to the vector of variables
pub fn vars(&self) -> &[Var] {
&self.vars
}
/// Perform backward pass with gradient clipping
///
/// Implements proper gradient clipping by norm to prevent gradient explosions.
/// Uses a two-pass approach: first pass computes gradient norm, second pass
/// (if needed) computes clipped gradients by scaling the loss.
///
/// # Arguments
///
/// * `loss` - The loss tensor to compute gradients from
/// * `max_norm` - Maximum allowed gradient norm (gradients will be clipped to this value)
///
/// # Returns
///
/// Returns `Ok(gradient_norm)` on success with the actual gradient norm (before clipping),
/// or `Err(MLError::TrainingError)` on failure
pub fn backward_step_with_monitoring(
&mut self,
loss: &Tensor,
max_norm: f64,
) -> Result<f64, MLError> {
// Bug fix: Single backward pass to prevent gradient accumulation
// Root cause: Two backward passes caused 1.5x gradient amplification
// Solution: Compute gradients once, then scale loss before optimizer step if needed
// 1. Compute gradients via backward pass
let mut grads = loss
.backward()
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// 2. Apply gradient clipping IN-PLACE (Bug #32 fix - gradient explosion)
// This modifies the grads directly before optimizer step
let (actual_grad_norm, clipped_grad_norm) = crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm)
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
// BUG #14 FIX: Log actual gradient norms to detect if threshold is too low
// Temporary diagnostic logging to investigate gradient clipping issue
static GRAD_LOG_COUNTER: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
let count = GRAD_LOG_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
if count < 20 || count % 100 == 0 {
tracing::info!(
"BUG #14 DIAGNOSTIC: Gradient norm BEFORE clipping: {:.4}, AFTER clipping: {:.4}, max_norm: {:.4}",
actual_grad_norm,
clipped_grad_norm,
max_norm
);
}
// 3. Apply optimizer step with clipped gradients
Optimizer::step(&mut self.optimizer, &grads)
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
if actual_grad_norm > max_norm {
tracing::warn!(
"Gradient clipping: {:.4} -> {:.4} - this is expected occasionally but should be rare. \
If frequent, consider reducing learning rate.",
actual_grad_norm,
clipped_grad_norm
);
}
Ok(clipped_grad_norm)
}
/// Perform backward pass with gradient clipping but WITHOUT an optimizer step.
///
/// This is useful for gradient accumulation workflows where you want to
/// accumulate clipped gradients across multiple micro-batches before
/// applying a single optimizer step.
///
/// # Arguments
///
/// * `loss` - The loss tensor to compute gradients from
/// * `max_norm` - Maximum allowed gradient norm (gradients will be clipped to this value)
///
/// # Returns
///
/// Returns `Ok((grads, clipped_norm))` on success, where `grads` is the clipped
/// `GradStore` and `clipped_norm` is the gradient norm after clipping.
///
/// # Errors
///
/// This function will return an error if:
/// - The backward pass fails to compute gradients
/// - Gradient clipping fails
pub fn backward_and_clip(
&self,
loss: &Tensor,
max_norm: f64,
) -> Result<(GradStore, f64), MLError> {
// Compute gradients via backward pass
let mut grads = loss
.backward()
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// Apply gradient clipping in-place
let (_actual_norm, clipped_norm) =
crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm)
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
Ok((grads, clipped_norm))
}
/// Apply pre-computed gradients to the optimizer.
///
/// This is the companion to `backward_and_clip` for gradient accumulation
/// workflows. After accumulating and scaling gradients, call this method
/// to perform the optimizer step.
///
/// # Arguments
///
/// * `grads` - Pre-computed (and possibly accumulated/scaled) gradients
///
/// # Errors
///
/// This function will return an error if the optimizer step fails
pub fn apply_grads(&mut self, grads: &GradStore) -> Result<(), MLError> {
Optimizer::step(&mut self.optimizer, grads)
.map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?;
Ok(())
}
/// Compute the L2 norm of all gradients
fn compute_gradient_norm(
&self,
grads: &candle_core::backprop::GradStore,
) -> Result<f64, MLError> {
let mut total_norm_sq = 0.0f64;
// Get all variables from the optimizer
for var in &self.vars {
if let Some(grad) = grads.get(var) {
// Compute L2 norm squared for this gradient
let grad_norm_sq = grad
.sqr()
.map_err(|e| {
MLError::TrainingError(format!("Failed to square gradient: {}", e))
})?
.sum_all()
.map_err(|e| MLError::TrainingError(format!("Failed to sum gradient: {}", e)))?
.to_vec0::<f32>()
.map_err(|e| {
MLError::TrainingError(format!("Failed to extract gradient norm: {}", e))
})? as f64;
total_norm_sq += grad_norm_sq;
}
}
Ok(total_norm_sq.sqrt())
}
}

3
ml/src/optimizers/mod.rs Normal file
View File

@@ -0,0 +1,3 @@
pub mod adam;
pub use adam::Adam;

34
ml/src/ppo/continuous_transaction_costs.rs
View File

@@ -16,6 +16,7 @@
//! - LimitMaker: 5 bps (0.05%) - Passive order, maker rebate
//! - IoC: 10 bps (0.10%) - Immediate-or-cancel, medium cost
pub use crate::common::action::OrderType;
use serde::{Deserialize, Serialize};
/// Transaction cost model for continuous position sizing
@@ -74,39 +75,6 @@ pub enum SlippageModel {
Quadratic { coefficient: f32 },
}
/// Order type for execution strategy
///
/// Matches DQN's OrderType enum with identical fee structure.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum OrderType {
/// Immediate execution, high cost (15 bps)
Market,
/// Passive order, maker rebate (5 bps)
LimitMaker,
/// Immediate-or-cancel (10 bps)
IoC,
}
impl OrderType {
/// Get base transaction cost in bps
///
/// Aligned with DQN transaction costs (Wave 2.5 calibration)
pub fn cost_bps(&self) -> f32 {
match self {
OrderType::Market => 15.0, // 0.15%
OrderType::LimitMaker => 5.0, // 0.05%
OrderType::IoC => 10.0, // 0.10%
}
}
/// Get cost as decimal (for convenience)
pub fn cost_decimal(&self) -> f64 {
(self.cost_bps() as f64) / 10000.0
}
}
impl ContinuousTransactionCosts {
/// Create new transaction cost model
pub fn new(

44
ml/src/ppo/factored_action.rs
View File

@@ -10,6 +10,7 @@
///
/// Index mapping: index = exposure*9 + order*3 + urgency (0-44)
pub use crate::common::action::OrderType;
use crate::MLError;
use serde::{Deserialize, Serialize};
@@ -51,39 +52,6 @@ impl ExposureLevel {
}
}
/// Order type for execution strategy
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum OrderType {
Market = 0, // Immediate execution, high cost (0.15%)
LimitMaker = 1, // Passive order, maker rebate (0.05%)
IoC = 2, // Immediate-or-cancel (0.10%)
}
impl OrderType {
/// Get transaction cost multiplier
/// Calibrated fees: Market 15 bps, LimitMaker 5 bps, IoC 10 bps
pub fn transaction_cost(&self) -> f64 {
match self {
OrderType::Market => 0.0015, // 0.15%
OrderType::LimitMaker => 0.0005, // 0.05%
OrderType::IoC => 0.0010, // 0.10%
}
}
/// Convert from index (0-2)
pub fn from_index(idx: usize) -> Result<Self, MLError> {
match idx {
0 => Ok(OrderType::Market),
1 => Ok(OrderType::LimitMaker),
2 => Ok(OrderType::IoC),
_ => Err(MLError::InvalidInput(format!(
"Invalid order type index: {}",
idx
))),
}
}
}
/// Urgency level for execution timing
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum Urgency {
@@ -239,16 +207,6 @@ impl std::fmt::Display for ExposureLevel {
}
}
impl std::fmt::Display for OrderType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
OrderType::Market => write!(f, "Market"),
OrderType::LimitMaker => write!(f, "LimitMaker"),
OrderType::IoC => write!(f, "IoC"),
}
}
}
impl std::fmt::Display for Urgency {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {

31
ml/src/ppo/transaction_costs.rs
View File

@@ -8,36 +8,7 @@
//! - **LimitMaker**: 0.05% (0.0005) - Passive order, maker rebate
//! - **IoC**: 0.10% (0.0010) - Immediate-or-cancel, medium cost
use serde::{Deserialize, Serialize};
/// Order type for execution strategy
///
/// Maps to DQN's OrderType enum with identical fee structure.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum OrderType {
/// Immediate execution, high cost (0.15%)
Market = 0,
/// Passive order, maker rebate (0.05%)
LimitMaker = 1,
/// Immediate-or-cancel (0.10%)
IoC = 2,
}
impl OrderType {
/// Get transaction cost multiplier
///
/// Wave 2.5 Calibration: Reduced fees
/// - Market: 20 bps → 15 bps (0.0015)
/// - LimitMaker: 10 bps → 5 bps (0.0005)
/// - IoC: 15 bps → 10 bps (0.0010)
pub fn transaction_cost(&self) -> f64 {
match self {
OrderType::Market => 0.0015, // 0.15% (was 0.20%)
OrderType::LimitMaker => 0.0005, // 0.05% (was 0.10%)
OrderType::IoC => 0.0010, // 0.10% (was 0.15%)
}
}
}
pub use crate::common::action::OrderType;
/// Calculate total transaction cost for a trade
///

2
model_loader/Cargo.toml
View File

@@ -14,7 +14,7 @@ description = "ML model loading and caching infrastructure for Foxhunt HFT syste
[dependencies]
# Internal workspace crates
storage = { path = "../storage" }
ml = { workspace = true } # Canonical ModelType re-exported from here
common.workspace = true
# Core workspace dependencies
anyhow.workspace = true

11
model_loader/src/lib.rs
View File

@@ -17,8 +17,8 @@ use std::time::SystemTime;
use storage::{ObjectStoreBackend, Storage};
use tracing::{debug, info, warn};
// Re-export canonical ModelType from ml crate (single source of truth)
pub use ml::ModelType;
// Re-export canonical ModelType from common crate
pub use common::model_types::ModelType;
/// Model metadata for version tracking
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -379,6 +379,13 @@ mod tests {
assert_eq!(ModelType::MAMBA.s3_prefix(), "mamba2");
}
#[test]
fn test_model_type_as_str() {
assert_eq!(ModelType::TLOB.as_str(), "tlob");
assert_eq!(ModelType::DQN.as_str(), "dqn");
assert_eq!(ModelType::MAMBA.as_str(), "mamba");
}
#[test]
fn test_cache_key_equality() {
let key1 = CacheKey {

3
services/api_gateway/src/auth/mtls/mod.rs
View File

@@ -73,4 +73,5 @@ pub use validator::{ClientIdentity, UserRole, X509CertificateValidator};
pub use revocation::RevocationChecker;
pub use tls_config::{ApiGatewayTlsConfig, TlsInterceptor, TlsProtocolVersion};
pub use common::tls::TlsProtocolVersion;
pub use tls_config::{ApiGatewayTlsConfig, TlsInterceptor};

8
services/api_gateway/src/auth/mtls/tls_config.rs
View File

@@ -8,6 +8,7 @@
//! - Performance optimized for HFT requirements
use anyhow::{Context, Result};
use common::tls::TlsProtocolVersion;
use config::manager::ConfigManager;
use config::structures::TlsConfig;
use std::sync::Arc;
@@ -16,13 +17,6 @@ use tracing::info;
use super::validator::{ClientIdentity, X509CertificateValidator};
/// TLS protocol version
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TlsProtocolVersion {
Tls12,
Tls13,
}
/// TLS configuration for the API Gateway
#[derive(Debug, Clone)]
pub struct ApiGatewayTlsConfig {

93
services/api_gateway/src/auth/mtls/validator.rs
View File

@@ -9,6 +9,7 @@
//! 6. Hostname verification
use anyhow::Result;
pub use common::tls::{ClientIdentity, UserRole};
use tracing::{debug, info, warn};
use x509_parser::certificate::X509Certificate;
use x509_parser::extensions::{GeneralName, ParsedExtension};
@@ -491,98 +492,6 @@ impl X509CertificateValidator {
}
}
/// Client identity extracted from certificate
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientIdentity {
pub common_name: String,
pub organizational_unit: String,
pub serial_number: String,
pub issuer: String,
}
impl ClientIdentity {
/// Check if client is authorized for trading operations
pub fn is_authorized_for_trading(&self) -> bool {
// Implement authorization logic based on certificate attributes
matches!(self.organizational_unit.as_str(), "trading" | "admin")
}
/// Check if client is authorized for read-only operations
pub fn is_authorized_for_readonly(&self) -> bool {
// Allow broader access for read-only operations
matches!(
self.organizational_unit.as_str(),
"trading" | "admin" | "analytics" | "risk" | "compliance"
)
}
/// Get user role based on certificate
pub fn get_role(&self) -> UserRole {
match self.organizational_unit.as_str() {
"admin" => UserRole::Admin,
"trading" => UserRole::Trader,
"analytics" => UserRole::Analyst,
"risk" => UserRole::RiskManager,
"compliance" => UserRole::ComplianceOfficer,
_ => UserRole::ReadOnly,
}
}
}
/// User roles based on certificate attributes
#[derive(Debug, Clone, PartialEq)]
pub enum UserRole {
Admin,
Trader,
Analyst,
RiskManager,
ComplianceOfficer,
ReadOnly,
}
impl UserRole {
/// Get permissions for this role
pub fn get_permissions(&self) -> Vec<&'static str> {
match self {
UserRole::Admin => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"analytics.run_backtest",
"compliance.view_reports",
"system.configure",
],
UserRole::Trader => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"analytics.view_data",
],
UserRole::Analyst => vec![
"analytics.view_data",
"analytics.run_backtest",
"risk.view_positions",
],
UserRole::RiskManager => vec![
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"compliance.view_reports",
],
UserRole::ComplianceOfficer => vec![
"compliance.view_reports",
"analytics.view_data",
"risk.view_positions",
],
UserRole::ReadOnly => vec!["analytics.view_data"],
}
}
}
#[cfg(test)]
mod tests {
use super::*;

8
services/api_gateway/src/config/endpoints.rs
View File

@@ -1,7 +1,7 @@
//! gRPC endpoints for configuration management
use crate::config::ConfigurationManager;
use crate::error::ConfigError;
use crate::error::GatewayConfigError;
use tonic::{Request, Response, Status};
use tracing::{debug, info};
@@ -46,7 +46,7 @@ impl ConfigurationService for ConfigurationServiceImpl {
.get_config(&req.service_scope, &req.config_key)
.await
.map_err(|e| match e {
ConfigError::NotFound { .. } => Status::not_found(format!("{}", e)),
GatewayConfigError::NotFound { .. } => Status::not_found(format!("{}", e)),
_ => Status::internal(format!("{}", e)),
})?;
@@ -84,8 +84,8 @@ impl ConfigurationService for ConfigurationServiceImpl {
)
.await
.map_err(|e| match e {
ConfigError::Validation(msg) => Status::invalid_argument(msg),
ConfigError::NotFound { .. } => Status::not_found(format!("{}", e)),
GatewayConfigError::Validation(msg) => Status::invalid_argument(msg),
GatewayConfigError::NotFound { .. } => Status::not_found(format!("{}", e)),
_ => Status::internal(format!("{}", e)),
})?;

30
services/api_gateway/src/config/manager.rs
View File

@@ -1,7 +1,7 @@
//! Configuration management with PostgreSQL NOTIFY/LISTEN hot-reload and Redis caching
use crate::config::validator::ConfigValidator;
use crate::error::{ConfigError, ConfigResult};
use crate::error::{GatewayConfigError, GatewayConfigResult};
use chrono::{DateTime, Utc};
use redis::aio::ConnectionManager;
use serde_json::Value;
@@ -46,7 +46,7 @@ impl ConfigurationManager {
/// * `db_pool` - Postgre`SQL` connection pool
///
/// * `redis` - Redis connection manager
pub async fn new(db_pool: PgPool, redis: ConnectionManager) -> ConfigResult<Self> {
pub async fn new(db_pool: PgPool, redis: ConnectionManager) -> GatewayConfigResult<Self> {
Ok(Self {
db_pool: Arc::new(db_pool),
redis: Arc::new(RwLock::new(redis)),
@@ -56,7 +56,7 @@ impl ConfigurationManager {
}
/// Starts listening for configuration changes via Postgre`SQL` NOTIFY
pub async fn start_listening(&mut self) -> ConfigResult<()> {
pub async fn start_listening(&mut self) -> GatewayConfigResult<()> {
let mut listener = sqlx::postgres::PgListener::connect_with(&self.db_pool).await?;
// Listen to global config updates channel
@@ -69,7 +69,7 @@ impl ConfigurationManager {
}
/// Processes configuration change notifications
pub async fn handle_notifications(&mut self) -> ConfigResult<()> {
pub async fn handle_notifications(&mut self) -> GatewayConfigResult<()> {
// Collect notifications first to avoid borrow checker issues
let mut invalidations = Vec::new();
@@ -113,7 +113,7 @@ impl ConfigurationManager {
&self,
service_scope: &str,
config_key: &str,
) -> ConfigResult<ConfigItem> {
) -> GatewayConfigResult<ConfigItem> {
// Try Redis cache first
if let Some(cached) = self.get_from_cache(service_scope, config_key).await? {
debug!("Cache hit for {}/{}", service_scope, config_key);
@@ -145,7 +145,7 @@ impl ConfigurationManager {
config_key: &str,
new_value: Value,
updated_by: &str,
) -> ConfigResult<()> {
) -> GatewayConfigResult<()> {
// Load current configuration for validation rules and audit
let current = self.load_from_db(service_scope, config_key).await?;
@@ -207,7 +207,7 @@ impl ConfigurationManager {
///
/// # Arguments
/// * `service_scope` - Service scope (None for all scopes)
pub async fn list_configs(&self, service_scope: Option<&str>) -> ConfigResult<Vec<ConfigItem>> {
pub async fn list_configs(&self, service_scope: Option<&str>) -> GatewayConfigResult<Vec<ConfigItem>> {
let configs = if let Some(scope) = service_scope {
sqlx::query_as::<_, ConfigItem>(
r#"
@@ -239,7 +239,7 @@ impl ConfigurationManager {
&self,
service_scope: &str,
config_key: &str,
) -> ConfigResult<ConfigItem> {
) -> GatewayConfigResult<ConfigItem> {
let config = sqlx::query_as::<_, ConfigItem>(
r#"
SELECT * FROM config_settings
@@ -250,7 +250,7 @@ impl ConfigurationManager {
.bind(config_key)
.fetch_optional(&*self.db_pool)
.await?
.ok_or_else(|| ConfigError::NotFound {
.ok_or_else(|| GatewayConfigError::NotFound {
service_scope: service_scope.to_string(),
key: config_key.to_string(),
})?;
@@ -263,7 +263,7 @@ impl ConfigurationManager {
&self,
service_scope: &str,
config_key: &str,
) -> ConfigResult<Option<ConfigItem>> {
) -> GatewayConfigResult<Option<ConfigItem>> {
let redis_key = format!("config:{}:{}", service_scope, config_key);
let mut redis = self.redis.write().await;
@@ -271,7 +271,7 @@ impl ConfigurationManager {
.arg(&redis_key)
.query_async(&mut *redis)
.await
.map_err(ConfigError::Redis)?;
.map_err(GatewayConfigError::Redis)?;
if let Some(cached_json) = cached {
let config: ConfigItem = serde_json::from_str(&cached_json)?;
@@ -282,7 +282,7 @@ impl ConfigurationManager {
}
/// Stores configuration in Redis cache
async fn set_in_cache(&self, config: &ConfigItem) -> ConfigResult<()> {
async fn set_in_cache(&self, config: &ConfigItem) -> GatewayConfigResult<()> {
let redis_key = format!("config:{}:{}", config.service_scope, config.config_key);
let config_json = serde_json::to_string(config)?;
let mut redis = self.redis.write().await;
@@ -294,13 +294,13 @@ impl ConfigurationManager {
.arg(&config_json)
.query_async::<()>(&mut *redis)
.await
.map_err(ConfigError::Redis)?;
.map_err(GatewayConfigError::Redis)?;
Ok(())
}
/// Invalidates Redis cache for a configuration
async fn invalidate_cache(&self, service_scope: &str, config_key: &str) -> ConfigResult<()> {
async fn invalidate_cache(&self, service_scope: &str, config_key: &str) -> GatewayConfigResult<()> {
let redis_key = format!("config:{}:{}", service_scope, config_key);
let mut redis = self.redis.write().await;
@@ -308,7 +308,7 @@ impl ConfigurationManager {
.arg(&redis_key)
.query_async::<()>(&mut *redis)
.await
.map_err(ConfigError::Redis)?;
.map_err(GatewayConfigError::Redis)?;
Ok(())
}

42
services/api_gateway/src/config/validator.rs
View File

@@ -1,6 +1,6 @@
//! Configuration validation with type checking, range validation, and regex matching
use crate::error::{ConfigError, ConfigResult};
use crate::error::{GatewayConfigError, GatewayConfigResult};
use regex::Regex;
use serde_json::Value;
use std::collections::HashMap;
@@ -53,7 +53,7 @@ impl ConfigValidator {
value: &Value,
data_type: &str,
rules: Option<&Value>,
) -> ConfigResult<()> {
) -> GatewayConfigResult<()> {
// Parse validation rules if provided
let validation_rules: Option<ValidationRules> = match rules {
Some(r) => serde_json::from_value(r.clone()).ok(),
@@ -77,7 +77,7 @@ impl ConfigValidator {
}
/// Validates that the value matches the expected type
fn validate_type(&self, value: &Value, data_type: &str) -> ConfigResult<()> {
fn validate_type(&self, value: &Value, data_type: &str) -> GatewayConfigResult<()> {
let matches = match data_type {
"string" => value.is_string(),
"integer" => value.is_i64() || value.is_u64(),
@@ -86,7 +86,7 @@ impl ConfigValidator {
"json" => value.is_object(),
"array" => value.is_array(),
_ => {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Unknown data type: {}",
data_type
)))
@@ -94,7 +94,7 @@ impl ConfigValidator {
};
if !matches {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Type mismatch: expected {}, got {}",
data_type,
value_type_name(value)
@@ -105,14 +105,14 @@ impl ConfigValidator {
}
/// Validates numeric values against min/max rules
fn validate_numeric(&self, value: &Value, rules: &ValidationRules) -> ConfigResult<()> {
fn validate_numeric(&self, value: &Value, rules: &ValidationRules) -> GatewayConfigResult<()> {
let num = value
.as_f64()
.ok_or_else(|| ConfigError::Validation("Not a number".to_string()))?;
.ok_or_else(|| GatewayConfigError::Validation("Not a number".to_string()))?;
if let Some(min) = rules.min {
if num < min {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Value {} is below minimum {}",
num, min
)));
@@ -121,7 +121,7 @@ impl ConfigValidator {
if let Some(max) = rules.max {
if num > max {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Value {} exceeds maximum {}",
num, max
)));
@@ -132,15 +132,15 @@ impl ConfigValidator {
}
/// Validates string values against length and regex rules
fn validate_string(&mut self, value: &Value, rules: &ValidationRules) -> ConfigResult<()> {
fn validate_string(&mut self, value: &Value, rules: &ValidationRules) -> GatewayConfigResult<()> {
let s = value
.as_str()
.ok_or_else(|| ConfigError::Validation("Not a string".to_string()))?;
.ok_or_else(|| GatewayConfigError::Validation("Not a string".to_string()))?;
// Length validation
if let Some(min_len) = rules.min_len {
if s.len() < min_len {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"String length {} is below minimum {}",
s.len(),
min_len
@@ -150,7 +150,7 @@ impl ConfigValidator {
if let Some(max_len) = rules.max_len {
if s.len() > max_len {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"String length {} exceeds maximum {}",
s.len(),
max_len
@@ -162,17 +162,17 @@ impl ConfigValidator {
if let Some(pattern) = &rules.regex {
if !self.regex_cache.contains_key(pattern) {
let compiled = Regex::new(pattern).map_err(|e| {
ConfigError::Validation(format!("Invalid regex pattern '{}': {}", pattern, e))
GatewayConfigError::Validation(format!("Invalid regex pattern '{}': {}", pattern, e))
})?;
self.regex_cache.insert(pattern.clone(), compiled);
}
let regex = self
.regex_cache
.get(pattern)
.ok_or_else(|| ConfigError::Validation("Regex cache miss".to_owned()))?;
.ok_or_else(|| GatewayConfigError::Validation("Regex cache miss".to_owned()))?;
if !regex.is_match(s) {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"String '{}' does not match pattern '{}'",
s, pattern
)));
@@ -182,7 +182,7 @@ impl ConfigValidator {
// Enum validation
if let Some(enum_values) = &rules.enum_values {
if !enum_values.contains(&s.to_string()) {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Value '{}' is not in allowed values: {:?}",
s, enum_values
)));
@@ -193,14 +193,14 @@ impl ConfigValidator {
}
/// Validates array values against length rules
fn validate_array(&self, value: &Value, rules: &ValidationRules) -> ConfigResult<()> {
fn validate_array(&self, value: &Value, rules: &ValidationRules) -> GatewayConfigResult<()> {
let arr = value
.as_array()
.ok_or_else(|| ConfigError::Validation("Not an array".to_string()))?;
.ok_or_else(|| GatewayConfigError::Validation("Not an array".to_string()))?;
if let Some(min_len) = rules.min_len {
if arr.len() < min_len {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Array length {} is below minimum {}",
arr.len(),
min_len
@@ -210,7 +210,7 @@ impl ConfigValidator {
if let Some(max_len) = rules.max_len {
if arr.len() > max_len {
return Err(ConfigError::Validation(format!(
return Err(GatewayConfigError::Validation(format!(
"Array length {} exceeds maximum {}",
arr.len(),
max_len

4
services/api_gateway/src/error.rs
View File

@@ -4,7 +4,7 @@ use thiserror::Error;
/// Configuration management errors
#[derive(Debug, Error)]
pub enum ConfigError {
pub enum GatewayConfigError {
#[error("Database error: {0}")]
Database(#[from] sqlx::Error),
@@ -28,4 +28,4 @@ pub enum ConfigError {
}
/// Result type for configuration operations
pub type ConfigResult<T> = Result<T, ConfigError>;
pub type GatewayConfigResult<T> = Result<T, GatewayConfigError>;

2
services/api_gateway/src/lib.rs
View File

@@ -64,7 +64,7 @@ pub mod metrics;
pub mod routing;
// Re-export error types
pub use error::{ConfigError, ConfigResult};
pub use error::{GatewayConfigError, GatewayConfigResult};
// Re-export configuration management types
pub use config::{

120
services/backtesting_service/src/tls_config.rs
View File

@@ -8,6 +8,10 @@
use anyhow::{Context, Result};
use config::manager::ConfigManager;
// Re-export shared TLS types from common for backward compatibility
#[allow(unused_imports)]
pub use common::tls::{ClientIdentity, TlsProtocolVersion, UserRole};
use config::structures::TlsConfig;
use std::sync::Arc;
// TLS imports - TLS feature should be enabled in Cargo.toml
@@ -38,16 +42,6 @@ pub struct BacktestingServiceTlsConfig {
pub crl_url: Option<String>,
}
/// TLS protocol version options
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub enum TlsProtocolVersion {
/// TLS version 1.2
Tls12,
/// TLS version 1.3
Tls13,
}
#[allow(dead_code)]
impl BacktestingServiceTlsConfig {
/// Create TLS configuration from certificate files
@@ -647,112 +641,6 @@ impl BacktestingServiceTlsConfig {
}
}
/// Client identity extracted from certificate
#[allow(dead_code)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientIdentity {
/// Common Name (CN) from certificate
pub common_name: String,
/// Organizational Unit (OU) from certificate
pub organizational_unit: String,
/// Certificate serial number
pub serial_number: String,
/// Certificate issuer
pub issuer: String,
}
#[allow(dead_code)]
impl ClientIdentity {
/// Check if client is authorized for trading operations
pub fn is_authorized_for_trading(&self) -> bool {
// Implement authorization logic based on certificate attributes
matches!(self.organizational_unit.as_str(), "trading" | "admin")
}
/// Check if client is authorized for read-only operations
pub fn is_authorized_for_readonly(&self) -> bool {
// Allow broader access for read-only operations
matches!(
self.organizational_unit.as_str(),
"trading" | "admin" | "analytics" | "risk" | "compliance"
)
}
/// Get user role based on certificate
pub fn get_role(&self) -> UserRole {
match self.organizational_unit.as_str() {
"admin" => UserRole::Admin,
"trading" => UserRole::Trader,
"analytics" => UserRole::Analyst,
"risk" => UserRole::RiskManager,
"compliance" => UserRole::ComplianceOfficer,
_ => UserRole::ReadOnly,
}
}
}
/// User roles based on certificate attributes
#[allow(dead_code)]
#[derive(Debug, Clone, PartialEq)]
pub enum UserRole {
/// Administrator with full access
Admin,
/// Trader with trading permissions
Trader,
/// Analyst with read/analysis permissions
Analyst,
/// Risk manager with risk oversight
RiskManager,
/// Compliance officer with audit access
ComplianceOfficer,
/// Read-only access
ReadOnly,
}
#[allow(dead_code)]
impl UserRole {
/// Get permissions for this role
pub fn get_permissions(&self) -> Vec<&'static str> {
match self {
UserRole::Admin => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"analytics.run_backtest",
"compliance.view_reports",
"system.configure",
],
UserRole::Trader => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"analytics.view_data",
],
UserRole::Analyst => vec![
"analytics.view_data",
"analytics.run_backtest",
"risk.view_positions",
],
UserRole::RiskManager => vec![
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"compliance.view_reports",
],
UserRole::ComplianceOfficer => vec![
"compliance.view_reports",
"analytics.view_data",
"risk.view_positions",
],
UserRole::ReadOnly => vec!["analytics.view_data"],
}
}
}
/// TLS interceptor for gRPC requests
#[allow(dead_code)]
#[derive(Clone)]

161
services/broker_gateway_service/src/error_handler.rs
View File

@@ -6,6 +6,9 @@
//! - Dead letter queue for unrecoverable orders
//! - Error classification and routing
use common::resilience::circuit_breaker::{CircuitBreaker, CircuitBreakerConfig};
#[cfg(test)]
use common::resilience::circuit_breaker::CircuitBreakerState;
use std::collections::VecDeque;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -43,17 +46,6 @@ pub enum ErrorRecoveryStrategy {
Fallback,
}
/// Circuit breaker state for fault isolation
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CircuitBreakerState {
/// Circuit is closed, requests pass through normally
Closed,
/// Circuit is open, requests fail fast to prevent cascading failures
Open,
/// Circuit is half-open, testing if service has recovered
HalfOpen,
}
/// Dead letter queue entry for unrecoverable orders
#[derive(Debug, Clone)]
pub struct DeadLetterEntry {
@@ -67,124 +59,6 @@ pub struct DeadLetterEntry {
pub retry_attempts: u32,
}
/// Circuit breaker for fault isolation and cascading failure prevention
pub struct CircuitBreaker {
/// Current circuit breaker state
state: Arc<RwLock<CircuitBreakerState>>,
/// Consecutive failure count
failure_count: Arc<RwLock<usize>>,
/// Timestamp when circuit breaker opened
opened_at: Arc<RwLock<Option<Instant>>>,
}
impl CircuitBreaker {
/// Create a new circuit breaker in CLOSED state
pub fn new() -> Self {
Self {
state: Arc::new(RwLock::new(CircuitBreakerState::Closed)),
failure_count: Arc::new(RwLock::new(0)),
opened_at: Arc::new(RwLock::new(None)),
}
}
/// Get current circuit breaker state
pub async fn state(&self) -> CircuitBreakerState {
*self.state.read().await
}
/// Record a successful operation (resets failure count)
pub async fn record_success(&self) {
let mut failure_count = self.failure_count.write().await;
*failure_count = 0;
let mut state = self.state.write().await;
if *state == CircuitBreakerState::HalfOpen {
info!("Circuit breaker transitioning: HALF_OPEN → CLOSED");
*state = CircuitBreakerState::Closed;
}
}
/// Record a failed operation (increments failure count)
pub async fn record_failure(&self) {
let mut failure_count = self.failure_count.write().await;
*failure_count += 1;
let current_state = *self.state.read().await;
if current_state == CircuitBreakerState::Closed
&& *failure_count >= CIRCUIT_BREAKER_THRESHOLD
{
warn!(
"Circuit breaker OPEN after {} failures (threshold: {})",
*failure_count, CIRCUIT_BREAKER_THRESHOLD
);
let mut state = self.state.write().await;
*state = CircuitBreakerState::Open;
let mut opened_at = self.opened_at.write().await;
*opened_at = Some(Instant::now());
} else if current_state == CircuitBreakerState::HalfOpen {
warn!("Circuit breaker transitioning: HALF_OPEN → OPEN (failure during test)");
let mut state = self.state.write().await;
*state = CircuitBreakerState::Open;
let mut opened_at = self.opened_at.write().await;
*opened_at = Some(Instant::now());
}
}
/// Check if circuit breaker allows requests (handles state transitions)
pub async fn allow_request(&self) -> bool {
let current_state = *self.state.read().await;
match current_state {
CircuitBreakerState::Closed => true,
CircuitBreakerState::Open => {
// Check if timeout has elapsed
let opened_at = self.opened_at.read().await;
if let Some(opened_time) = *opened_at {
if opened_time.elapsed() >= CIRCUIT_BREAKER_TIMEOUT {
info!("Circuit breaker transitioning: OPEN → HALF_OPEN (timeout elapsed)");
drop(opened_at);
let mut state = self.state.write().await;
*state = CircuitBreakerState::HalfOpen;
let mut failure_count = self.failure_count.write().await;
*failure_count = 0;
return true;
}
}
false
}
CircuitBreakerState::HalfOpen => true,
}
}
/// Reset circuit breaker to CLOSED state (manual recovery)
pub async fn reset(&self) {
let mut state = self.state.write().await;
*state = CircuitBreakerState::Closed;
let mut failure_count = self.failure_count.write().await;
*failure_count = 0;
let mut opened_at = self.opened_at.write().await;
*opened_at = None;
info!("Circuit breaker manually reset to CLOSED state");
}
}
impl Default for CircuitBreaker {
fn default() -> Self {
Self::new()
}
}
/// Dead letter queue for unrecoverable orders
pub struct DeadLetterQueue {
/// Queue of failed orders
@@ -275,8 +149,13 @@ pub struct ErrorHandler {
impl ErrorHandler {
/// Create a new error handler
pub fn new() -> Self {
let cb_config = CircuitBreakerConfig {
failure_threshold: CIRCUIT_BREAKER_THRESHOLD as u32,
timeout: CIRCUIT_BREAKER_TIMEOUT,
success_threshold: 1,
};
Self {
circuit_breaker: CircuitBreaker::new(),
circuit_breaker: CircuitBreaker::new("broker_gateway", cb_config),
dead_letter_queue: DeadLetterQueue::new(),
}
}
@@ -422,13 +301,21 @@ impl Default for ErrorHandler {
mod tests {
use super::*;
fn test_cb_config() -> CircuitBreakerConfig {
CircuitBreakerConfig {
failure_threshold: CIRCUIT_BREAKER_THRESHOLD as u32,
timeout: CIRCUIT_BREAKER_TIMEOUT,
success_threshold: 1,
}
}
#[tokio::test]
async fn test_circuit_breaker_closed_to_open() {
let cb = CircuitBreaker::new();
let cb = CircuitBreaker::new("test_broker", test_cb_config());
// Initially CLOSED
assert_eq!(cb.state().await, CircuitBreakerState::Closed);
assert!(cb.allow_request().await);
assert!(cb.can_execute().await);
// Record failures up to threshold
for _ in 0..CIRCUIT_BREAKER_THRESHOLD {
@@ -437,12 +324,12 @@ mod tests {
// Should now be OPEN
assert_eq!(cb.state().await, CircuitBreakerState::Open);
assert!(!cb.allow_request().await);
assert!(!cb.can_execute().await);
}
#[tokio::test]
async fn test_circuit_breaker_half_open_success() {
let cb = CircuitBreaker::new();
let cb = CircuitBreaker::new("test_broker", test_cb_config());
// Force HALF_OPEN state
for _ in 0..CIRCUIT_BREAKER_THRESHOLD {
@@ -451,7 +338,7 @@ mod tests {
// Wait for timeout to transition to HALF_OPEN
tokio::time::sleep(CIRCUIT_BREAKER_TIMEOUT + Duration::from_millis(10)).await;
assert!(cb.allow_request().await);
assert!(cb.can_execute().await);
assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);
// Success should close circuit
@@ -461,7 +348,7 @@ mod tests {
#[tokio::test]
async fn test_circuit_breaker_half_open_failure() {
let cb = CircuitBreaker::new();
let cb = CircuitBreaker::new("test_broker", test_cb_config());
// Force HALF_OPEN state
for _ in 0..CIRCUIT_BREAKER_THRESHOLD {
@@ -469,7 +356,7 @@ mod tests {
}
tokio::time::sleep(CIRCUIT_BREAKER_TIMEOUT + Duration::from_millis(10)).await;
assert!(cb.allow_request().await); // Trigger transition to HALF_OPEN
assert!(cb.can_execute().await); // Trigger transition to HALF_OPEN
assert_eq!(cb.state().await, CircuitBreakerState::HalfOpen);
// Failure should reopen circuit

4
services/ml_training_service/src/job_spawner.rs
View File

@@ -51,8 +51,8 @@ use std::path::PathBuf;
use tracing::{debug, info, warn};
use uuid::Uuid;
// Re-export canonical ModelType from ml crate
pub use ml::ModelType;
// Use canonical ModelType from common crate
pub use common::model_types::ModelType;
/// Trading asset with data file
#[derive(Debug, Clone, Serialize, Deserialize)]

107
services/ml_training_service/src/tls_config.rs
View File

@@ -8,6 +8,10 @@
use anyhow::{Context, Result};
use config::manager::ConfigManager;
// Re-export shared TLS types from common for backward compatibility
#[allow(unused_imports)]
pub use common::tls::{ClientIdentity, TlsProtocolVersion, UserRole};
use config::structures::TlsConfig;
use std::sync::Arc;
// TLS imports - TLS feature should be enabled in Cargo.toml
@@ -37,13 +41,6 @@ pub struct MLTrainingServiceTlsConfig {
pub crl_url: Option<String>,
}
#[derive(Debug, Clone)]
#[allow(dead_code)] // Variants used in future implementations
pub enum TlsProtocolVersion {
Tls12,
Tls13,
}
#[allow(dead_code)] // Methods used in future implementations
impl MLTrainingServiceTlsConfig {
/// Create TLS configuration from certificate files
@@ -642,102 +639,6 @@ impl MLTrainingServiceTlsConfig {
}
}
/// Client identity extracted from certificate
#[derive(Debug, Clone, PartialEq, Eq)]
#[allow(dead_code)] // Used in future implementations
pub struct ClientIdentity {
pub common_name: String,
pub organizational_unit: String,
pub serial_number: String,
pub issuer: String,
}
#[allow(dead_code)] // Methods used in future implementations
impl ClientIdentity {
/// Check if client is authorized for trading operations
pub fn is_authorized_for_trading(&self) -> bool {
// Implement authorization logic based on certificate attributes
matches!(self.organizational_unit.as_str(), "trading" | "admin")
}
/// Check if client is authorized for read-only operations
pub fn is_authorized_for_readonly(&self) -> bool {
// Allow broader access for read-only operations
matches!(
self.organizational_unit.as_str(),
"trading" | "admin" | "analytics" | "risk" | "compliance"
)
}
/// Get user role based on certificate
pub fn get_role(&self) -> UserRole {
match self.organizational_unit.as_str() {
"admin" => UserRole::Admin,
"trading" => UserRole::Trader,
"analytics" => UserRole::Analyst,
"risk" => UserRole::RiskManager,
"compliance" => UserRole::ComplianceOfficer,
_ => UserRole::ReadOnly,
}
}
}
/// User roles based on certificate attributes
#[derive(Debug, Clone, PartialEq)]
#[allow(dead_code)] // Used in future implementations
pub enum UserRole {
Admin,
Trader,
Analyst,
RiskManager,
ComplianceOfficer,
ReadOnly,
}
#[allow(dead_code)] // Methods used in future implementations
impl UserRole {
/// Get permissions for this role
pub fn get_permissions(&self) -> Vec<&'static str> {
match self {
UserRole::Admin => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"analytics.run_backtest",
"compliance.view_reports",
"system.configure",
],
UserRole::Trader => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"analytics.view_data",
],
UserRole::Analyst => vec![
"analytics.view_data",
"analytics.run_backtest",
"risk.view_positions",
],
UserRole::RiskManager => vec![
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"compliance.view_reports",
],
UserRole::ComplianceOfficer => vec![
"compliance.view_reports",
"analytics.view_data",
"risk.view_positions",
],
UserRole::ReadOnly => vec!["analytics.view_data"],
}
}
}
/// TLS interceptor for gRPC requests
#[derive(Clone)]
#[allow(dead_code)] // Used in future implementations

119
services/trading_service/src/tls_config.rs
View File

@@ -8,6 +8,9 @@
use anyhow::{Context, Result};
use config::manager::ConfigManager;
// Re-export shared TLS types from common for backward compatibility
pub use common::tls::{ClientIdentity, TlsProtocolVersion, UserRole};
use config::structures::TlsConfig;
use std::sync::Arc;
// TLS imports - TLS feature should be enabled in Cargo.toml
@@ -38,16 +41,6 @@ pub struct TradingServiceTlsConfig {
pub crl_url: Option<String>,
}
/// TLS protocol version options
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub enum TlsProtocolVersion {
/// TLS version 1.2
Tls12,
/// TLS version 1.3
Tls13,
}
#[allow(dead_code)]
impl TradingServiceTlsConfig {
/// Create TLS configuration from certificate files
@@ -594,112 +587,6 @@ impl TradingServiceTlsConfig {
}
}
/// Client identity extracted from certificate
#[allow(dead_code)]
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ClientIdentity {
/// Common Name (CN) from certificate
pub common_name: String,
/// Organizational Unit (OU) from certificate
pub organizational_unit: String,
/// Certificate serial number
pub serial_number: String,
/// Certificate issuer
pub issuer: String,
}
#[allow(dead_code)]
impl ClientIdentity {
/// Check if client is authorized for trading operations
pub fn is_authorized_for_trading(&self) -> bool {
// Implement authorization logic based on certificate attributes
matches!(self.organizational_unit.as_str(), "trading" | "admin")
}
/// Check if client is authorized for read-only operations
pub fn is_authorized_for_readonly(&self) -> bool {
// Allow broader access for read-only operations
matches!(
self.organizational_unit.as_str(),
"trading" | "admin" | "analytics" | "risk" | "compliance"
)
}
/// Get user role based on certificate
pub fn get_role(&self) -> UserRole {
match self.organizational_unit.as_str() {
"admin" => UserRole::Admin,
"trading" => UserRole::Trader,
"analytics" => UserRole::Analyst,
"risk" => UserRole::RiskManager,
"compliance" => UserRole::ComplianceOfficer,
_ => UserRole::ReadOnly,
}
}
}
/// User roles based on certificate attributes
#[allow(dead_code)]
#[derive(Debug, Clone, PartialEq)]
pub enum UserRole {
/// Administrator with full access
Admin,
/// Trader with trading permissions
Trader,
/// Analyst with read/analysis permissions
Analyst,
/// Risk manager with risk oversight
RiskManager,
/// Compliance officer with audit access
ComplianceOfficer,
/// Read-only access
ReadOnly,
}
#[allow(dead_code)]
impl UserRole {
/// Get permissions for this role
pub fn get_permissions(&self) -> Vec<&'static str> {
match self {
UserRole::Admin => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"analytics.run_backtest",
"compliance.view_reports",
"system.configure",
],
UserRole::Trader => vec![
"trading.submit_order",
"trading.cancel_order",
"trading.modify_order",
"risk.view_positions",
"analytics.view_data",
],
UserRole::Analyst => vec![
"analytics.view_data",
"analytics.run_backtest",
"risk.view_positions",
],
UserRole::RiskManager => vec![
"risk.view_positions",
"risk.modify_limits",
"analytics.view_data",
"compliance.view_reports",
],
UserRole::ComplianceOfficer => vec![
"compliance.view_reports",
"analytics.view_data",
"risk.view_positions",
],
UserRole::ReadOnly => vec!["analytics.view_data"],
}
}
}
/// TLS interceptor for gRPC requests
#[allow(dead_code)]
#[derive(Clone)]