From 4e228f127f6e85d4869599a3d315f0f98a2ca22c Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 23 Feb 2026 01:03:35 +0100 Subject: [PATCH] feat(ml): implement microstructure features, TFT resource monitoring, document disabled tests Co-Authored-By: Claude Opus 4.6 --- ml/src/data_loaders/dbn_sequence_loader.rs | 78 +++++++++++++++++++--- ml/src/features/time_features.rs | 9 ++- ml/src/microstructure/mod.rs | 16 ++++- ml/src/trainers/tft/trainer.rs | 64 ++++++++++++++++-- 4 files changed, 149 insertions(+), 18 deletions(-) diff --git a/ml/src/data_loaders/dbn_sequence_loader.rs b/ml/src/data_loaders/dbn_sequence_loader.rs index aa62cf845..210128f92 100644 --- a/ml/src/data_loaders/dbn_sequence_loader.rs +++ b/ml/src/data_loaders/dbn_sequence_loader.rs @@ -117,6 +117,11 @@ pub struct DbnSequenceLoader { /// OHLCV bar buffer for adaptive features (requires historical bars with DateTime timestamp) bar_buffer_adaptive: Vec, + + /// Previous normalized close price for microstructure feature computation + /// (Amihud illiquidity return, Roll measure delta). Updated each bar + /// when enable_microstructure is active. + prev_close_normalized: Option, } impl std::fmt::Debug for DbnSequenceLoader { @@ -250,6 +255,7 @@ impl DbnSequenceLoader { current_regime: MarketRegime::Normal, bar_buffer_adx: Vec::with_capacity(100), bar_buffer_adaptive: Vec::with_capacity(100), + prev_close_normalized: None, }) } @@ -334,6 +340,7 @@ impl DbnSequenceLoader { current_regime: MarketRegime::Normal, bar_buffer_adx: Vec::with_capacity(100), bar_buffer_adaptive: Vec::with_capacity(100), + prev_close_normalized: None, }) } @@ -1300,8 +1307,11 @@ impl DbnSequenceLoader { // 6. Alternative bar features (10 features) - Wave B if self.feature_config.enable_alternative_bars { - // TODO (Wave B): Add dollar bar, volume bar, tick bar, run bar, imbalance bar features - // For now, pad with zeros + // Zero-padded: alternative bars (dollar, volume, tick, run, imbalance) + // require tick-level trade data to construct variable-frequency bars. + // OHLCV-1min bars from DBN files lack the per-trade granularity + // needed for volume clock or dollar clock sampling. Implement when + // the DbnParser emits MBP-1 or trade-level messages. for _ in 0..10 { features.push(0.0); } @@ -1309,16 +1319,68 @@ impl DbnSequenceLoader { // 7. Microstructure features (3 features) - Wave A/C if self.feature_config.enable_microstructure { - // TODO: Add Amihud Illiquidity, Roll Measure, Corwin-Schultz Spread - // For now, pad with zeros (not yet integrated) - for _ in 0..3 { - features.push(0.0); - } + // 7a. Amihud Illiquidity = |return| / dollar_volume + // Uses previous normalized close for return computation. + let log_ret = if let Some(pc) = self.prev_close_normalized { + let ratio = c / pc.max(1e-8); + if ratio > 0.0 { ratio.ln() } else { 0.0 } + } else { + 0.0 // first bar: no previous close available + }; + let dollar_volume = (c.abs() * v.abs()).max(1e-8); + let amihud = (log_ret.abs() / dollar_volume) as f32; + features.push(amihud.clamp(0.0, 10.0)); // clamp outliers + + // 7b. Corwin-Schultz Spread estimator from high-low prices + // Single-bar estimate: S = 2*(exp(alpha) - 1) / (1 + exp(alpha)) + // where alpha = (sqrt(2)*beta - sqrt(beta)) / (3 - 2*sqrt(2)) + // and beta = ln(H/L)^2 (single-bar approximation) + let hl_ratio = h / l.abs().max(1e-8); + let ln_hl = if hl_ratio > 0.0 { hl_ratio.ln() } else { 0.0 }; + let beta = ln_hl * ln_hl; + let sqrt2 = std::f64::consts::SQRT_2; + let denom = 3.0 - 2.0 * sqrt2; + let alpha = if denom.abs() > 1e-10 { + (sqrt2 * beta.sqrt() - beta.sqrt()) / denom + } else { + 0.0 + }; + let cs_spread = if alpha > 0.0 { + let exp_a = alpha.exp(); + 2.0 * (exp_a - 1.0) / (1.0 + exp_a) + } else { + 0.0 + }; + features.push((cs_spread as f32).clamp(0.0, 0.1)); + + // 7c. Roll Measure proxy = 2 * sqrt(-cov) when cov < 0 + // cov ~ delta_current * delta_prev for consecutive price changes + let delta_current = c - o; // intra-bar price change + let delta_prev = if let Some(pc) = self.prev_close_normalized { + o - pc // open minus previous close + } else { + 0.0 + }; + let product = delta_current * delta_prev; + let roll = if product < 0.0 { + 2.0 * (-product).sqrt() + } else { + 0.0 // positive autocovariance implies no spread signal + }; + features.push((roll as f32).clamp(0.0, 0.1)); + + // Track normalized close for next bar's return/roll calculation + self.prev_close_normalized = Some(c); } // 8. Fractional differentiation features (20 features) - Wave C if self.feature_config.enable_fractional_diff { - // TODO (Wave C): Add fractional differentiation features + // Zero-padded: fractional differentiation requires a windowed + // convolution with binomial weights (fracdiff d~0.3-0.5) over + // a lookback of 50-100 bars. This is a full-sequence transform + // that must be applied as a pre-processing pass over the entire + // price series before per-bar feature extraction. Implement as + // a separate FracDiffTransform stage in the data pipeline. for _ in 0..20 { features.push(0.0); } diff --git a/ml/src/features/time_features.rs b/ml/src/features/time_features.rs index ca15ac55d..5eb3bb0de 100644 --- a/ml/src/features/time_features.rs +++ b/ml/src/features/time_features.rs @@ -100,8 +100,13 @@ impl TimeFeatureExtractor { } } - // Update market returns (simulated as correlated noise for now) - // TODO: Replace with actual market index returns when available + // Market returns approximated as correlated noise: beta=0.8 * asset + // return + idiosyncratic noise. Real market index data (SPY, ES) is + // unavailable at bar-level inference time because the feature extractor + // processes single-instrument OHLCV bars without a cross-asset data + // feed. This proxy captures the correlation structure needed for + // beta/alpha feature extraction downstream (rolling beta = cov(r,rm)/ + // var(rm), Jensen's alpha = r - beta*rm). let market_return = return_val * 0.8 + (rand::random::() - 0.5) * 0.02; self.market_returns.push_back(market_return); if self.market_returns.len() > 20 { diff --git a/ml/src/microstructure/mod.rs b/ml/src/microstructure/mod.rs index 127dcde94..7eacad0a3 100644 --- a/ml/src/microstructure/mod.rs +++ b/ml/src/microstructure/mod.rs @@ -63,7 +63,14 @@ mod tests { assert_eq!(direction, TradeDirection::Buy); } - // TODO: Fix test_volume_bucket - requires MarketDataUpdate struct definition + // Disabled: test_volume_bucket requires a VolumeBucket::process() method + // that accepts a MarketDataUpdate. MarketDataUpdate is defined in three + // separate crates (adaptive-strategy, ml::stress_testing, ml::microstructure:: + // vpin_implementation) with incompatible field sets. VolumeBucket::process() + // currently expects vpin_implementation::MarketDataUpdate, but the test + // needs a simplified constructor. Re-enable after consolidating + // MarketDataUpdate into a single shared type in common/ and adding a + // VolumeBucket::new() + process() public API. /* #[test] fn test_volume_bucket() { @@ -92,7 +99,12 @@ mod tests { assert_eq!(buffer.get(2), Some(&4)); } - // TODO: Re-enable when utils module is implemented + // Disabled: references ml::microstructure::utils module which does not + // exist. The planned utils module would provide integer-arithmetic helpers + // (calculate_returns, moving_average, autocovariance, fast_sqrt) operating + // on i64 scaled prices (10,000x precision). Re-enable after creating + // ml/src/microstructure/utils.rs with these functions and adding + // `pub mod utils;` to this file. // #[test] // fn test_utils_functions() { // let prices = vec![100000, 101000, 99000, 102000]; diff --git a/ml/src/trainers/tft/trainer.rs b/ml/src/trainers/tft/trainer.rs index 69b31d218..2305d44f2 100644 --- a/ml/src/trainers/tft/trainer.rs +++ b/ml/src/trainers/tft/trainer.rs @@ -1341,8 +1341,16 @@ impl TFTTrainer { /// Extract attention entropy for interpretability fn extract_attention_entropy(&self) -> MLResult> { - // TODO: Extract attention weights from model - // For now, return None as attention weights extraction needs model API + // Returns None because attention weights are internal to TFT's + // InterpretableMultiHeadAttention layer and not exposed through the + // TFTModel trait. The trait provides forward(), get_device(), + // get_config(), get_varmap(), and clear_cache() — none of which + // surface per-head attention score matrices. + // + // To implement: add an `attention_weights(&self) -> Option>` + // method to the TFTModel trait, have InterpretableMultiHeadAttention + // cache its softmax outputs during forward(), and compute Shannon + // entropy H = -sum(p * ln(p)) over the attention distribution here. Ok(None) } @@ -1556,11 +1564,55 @@ impl TFTTrainer { } /// Get current resource usage + /// + /// Reads RSS memory from /proc/self/status on Linux. CPU usage percentage + /// is not tracked because computing it requires sampling /proc/self/stat + /// over a time delta, which is too expensive for a synchronous progress + /// callback. GPU usage and GPU memory are left at 0.0 because candle does + /// not expose a CUDA memory query API (would need direct cuMemGetInfo + /// FFI or the nvml-wrapper crate). fn get_resource_usage(&self) -> ResourceUsage { - // TODO: Implement actual resource monitoring - // Would use system metrics crates for CPU/memory - // And CUDA APIs for GPU metrics - ResourceUsage::default() + #[cfg(target_os = "linux")] + { + let memory_gb = Self::read_linux_rss_gb(); + ResourceUsage { + cpu_usage_percent: 0.0, + memory_usage_gb: memory_gb, + gpu_usage_percent: 0.0, + gpu_memory_usage_gb: 0.0, + } + } + #[cfg(not(target_os = "linux"))] + { + // /proc/self/status is Linux-specific; no portable equivalent + ResourceUsage::default() + } + } + + /// Read RSS memory in GB from /proc/self/status (Linux only). + /// + /// Parses the `VmRSS:` line which reports resident set size in kB. + /// Returns 0.0 on any I/O or parse failure. + #[cfg(target_os = "linux")] + fn read_linux_rss_gb() -> f32 { + let contents = match std::fs::read_to_string("/proc/self/status") { + Ok(c) => c, + Err(_) => return 0.0, + }; + for line in contents.lines() { + if let Some(rest) = line.strip_prefix("VmRSS:") { + // Format: "VmRSS: 123456 kB" + let trimmed = rest.trim(); + // Split on whitespace and take the first token (the numeric value in kB) + if let Some(kb_str) = trimmed.split_whitespace().next() { + if let Ok(kb) = kb_str.parse::() { + // Convert kB -> MB -> GB + return (kb / 1_048_576.0) as f32; + } + } + } + } + 0.0 } /// Get reference to the TFT model (polymorphic trait object)