From 4d08472d95f73ee818b65c4c69f37ac0245e9acf Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 17 Apr 2026 00:06:32 +0200 Subject: [PATCH] =?UTF-8?q?feat(tick):=20MicrostructureState=20=E2=80=94?= =?UTF-8?q?=2012=20new=20tick-level=20features,=20O(1)/tick?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OFI trajectory (online LinReg), realized variance, Hawkes intensity, book pressure gradient, spread dynamics, aggression ratio, queue depletion, order count flux, intra-bar momentum (Welford), regime score, OFI acceleration, toxicity gradient. ExtendedOFIFeatures [20]. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/ml-features/src/ofi_calculator.rs | 487 ++++++++++++++++++++++- 1 file changed, 486 insertions(+), 1 deletion(-) diff --git a/crates/ml-features/src/ofi_calculator.rs b/crates/ml-features/src/ofi_calculator.rs index 6b6ec390a..25f87e12e 100644 --- a/crates/ml-features/src/ofi_calculator.rs +++ b/crates/ml-features/src/ofi_calculator.rs @@ -36,7 +36,7 @@ //! assert!(features.depth_imbalance >= -1.0 && features.depth_imbalance <= 1.0); //! ``` -use data::providers::databento::mbp10::Mbp10Snapshot; +use data::providers::databento::mbp10::{BidAskPair, Mbp10Snapshot}; use std::collections::VecDeque; use crate::MLError; @@ -661,6 +661,491 @@ const fn safe_clip(value: f64, min: f64, max: f64) -> f64 { value.max(min).min(max) } +// --------------------------------------------------------------------------- +// Tick Microstructure Intelligence — 12 new features (indices 8-19) +// --------------------------------------------------------------------------- + +/// Running microstructure statistics computed O(1) per tick. +/// +/// Produces 12 features that extend the base 8 OFI features: +/// +/// | Idx | Name | +/// |-----|---------------------------| +/// | 8 | OFI Trajectory | +/// | 9 | Realized Variance | +/// | 10 | Hawkes Trade Intensity | +/// | 11 | Book Pressure Gradient | +/// | 12 | Spread Dynamics | +/// | 13 | Aggression Ratio | +/// | 14 | Queue Depletion Asymmetry | +/// | 15 | Order Count Flux | +/// | 16 | Intra-Bar Momentum | +/// | 17 | Microstructure Regime | +/// | 18 | OFI Acceleration | +/// | 19 | Toxicity Gradient | +#[derive(Debug, Clone)] +#[allow(dead_code)] +pub struct MicrostructureState { + // --- bar timing --- + bar_start_ns: u64, + bar_duration_ns: u64, + bar_mid_ns: u64, + + // --- [8] OFI Trajectory: online linear regression of OFI_L1 over time --- + ofi_linreg_n: f64, + ofi_linreg_sum_x: f64, + ofi_linreg_sum_y: f64, + ofi_linreg_sum_xy: f64, + ofi_linreg_sum_xx: f64, + + // --- [9] Realized Variance: sum of squared log-returns of mid price --- + prev_mid: f64, + realized_var: f64, + + // --- [10] Hawkes Trade Intensity: exponential kernel EMA --- + hawkes_intensity: f64, + last_trade_ns: u64, + + // --- [11] Book Pressure Gradient: computed from latest snapshot (stateless) --- + book_pressure: f64, + + // --- [12] Spread Dynamics: max/min/sum/count of spread within bar --- + spread_max: f64, + spread_min: f64, + spread_sum: f64, + spread_count: u64, + + // --- [13] Aggression Ratio: trades at ask vs total --- + trades_at_ask: u64, + total_trades: u64, + + // --- [14] Queue Depletion Asymmetry: EMA of bid/ask depletion --- + prev_bid_sz_l1: u32, + prev_ask_sz_l1: u32, + bid_depletion_ema: f64, + ask_depletion_ema: f64, + + // --- [15] Order Count Flux: ct_increase / (ct_increase + ct_decrease) --- + prev_total_ct: u32, + ct_increase_count: u64, + ct_decrease_count: u64, + + // --- [16] Intra-Bar Momentum: Welford online mean for half1 and half2 --- + half1_mean: f64, + half1_n: u64, + half2_mean: f64, + half2_n: u64, + prev_mid_momentum: f64, + + // --- [17] Microstructure Regime Score: computed from aggression/spread/thin_book --- + // (derived from other fields in snapshot(), no extra state needed) + + // --- [18] OFI Acceleration: EMA of OFI_L1 deltas --- + prev_ofi_l1: f64, + ofi_accel_ema: f64, + has_prev_ofi: bool, + + // --- [19] Toxicity Gradient: EMA of VPIN deltas --- + prev_vpin: f64, + toxicity_grad_ema: f64, + has_prev_vpin: bool, + + // --- snapshot count (for first-snapshot guards) --- + snapshot_count: u64, +} + +impl MicrostructureState { + /// Create a new state for a bar starting at `bar_start_ns` with given duration. + pub fn new(bar_start_ns: u64, bar_duration_ns: u64) -> Self { + Self { + bar_start_ns, + bar_duration_ns, + bar_mid_ns: bar_start_ns + bar_duration_ns / 2, + + ofi_linreg_n: 0.0, + ofi_linreg_sum_x: 0.0, + ofi_linreg_sum_y: 0.0, + ofi_linreg_sum_xy: 0.0, + ofi_linreg_sum_xx: 0.0, + + prev_mid: 0.0, + realized_var: 0.0, + + hawkes_intensity: 0.0, + last_trade_ns: 0, + + book_pressure: 0.0, + + spread_max: f64::NEG_INFINITY, + spread_min: f64::INFINITY, + spread_sum: 0.0, + spread_count: 0, + + trades_at_ask: 0, + total_trades: 0, + + prev_bid_sz_l1: 0, + prev_ask_sz_l1: 0, + bid_depletion_ema: 0.0, + ask_depletion_ema: 0.0, + + prev_total_ct: 0, + ct_increase_count: 0, + ct_decrease_count: 0, + + half1_mean: 0.0, + half1_n: 0, + half2_mean: 0.0, + half2_n: 0, + prev_mid_momentum: 0.0, + + prev_ofi_l1: 0.0, + ofi_accel_ema: 0.0, + has_prev_ofi: false, + + prev_vpin: 0.0, + toxicity_grad_ema: 0.0, + has_prev_vpin: false, + + snapshot_count: 0, + } + } + + /// Update from an MBP-10 snapshot. Called per snapshot event. + pub fn update_snapshot(&mut self, snapshot: &Mbp10Snapshot) { + const EMA_ALPHA: f64 = 0.05; + + if snapshot.levels.is_empty() { + return; + } + + let mid = snapshot.mid_price(); + let spread = snapshot.spread(); + + // --- [9] Realized Variance --- + if self.snapshot_count > 0 && self.prev_mid > 0.0 && mid > 0.0 { + let log_return = (mid / self.prev_mid).ln(); + if log_return.is_finite() { + self.realized_var += log_return * log_return; + } + } + + // --- [11] Book Pressure Gradient --- + self.book_pressure = Self::calc_book_pressure(&snapshot.levels); + + // --- [12] Spread Dynamics --- + if spread.is_finite() && spread >= 0.0 { + if spread > self.spread_max { + self.spread_max = spread; + } + if spread < self.spread_min { + self.spread_min = spread; + } + self.spread_sum += spread; + self.spread_count += 1; + } + + // --- [14] Queue Depletion Asymmetry --- + if self.snapshot_count > 0 { + let bid_sz = snapshot.levels[0].bid_sz; + let ask_sz = snapshot.levels[0].ask_sz; + + // Depletion = how much size decreased (positive means depletion) + let bid_depl = if bid_sz < self.prev_bid_sz_l1 { + (self.prev_bid_sz_l1 - bid_sz) as f64 + } else { + 0.0 + }; + let ask_depl = if ask_sz < self.prev_ask_sz_l1 { + (self.prev_ask_sz_l1 - ask_sz) as f64 + } else { + 0.0 + }; + + self.bid_depletion_ema = + EMA_ALPHA * bid_depl + (1.0 - EMA_ALPHA) * self.bid_depletion_ema; + self.ask_depletion_ema = + EMA_ALPHA * ask_depl + (1.0 - EMA_ALPHA) * self.ask_depletion_ema; + + self.prev_bid_sz_l1 = bid_sz; + self.prev_ask_sz_l1 = ask_sz; + } else { + self.prev_bid_sz_l1 = snapshot.levels[0].bid_sz; + self.prev_ask_sz_l1 = snapshot.levels[0].ask_sz; + } + + // --- [15] Order Count Flux --- + let total_ct: u32 = snapshot + .levels + .iter() + .map(|l| l.bid_ct + l.ask_ct) + .sum(); + if self.snapshot_count > 0 { + if total_ct > self.prev_total_ct { + self.ct_increase_count += 1; + } else if total_ct < self.prev_total_ct { + self.ct_decrease_count += 1; + } + } + self.prev_total_ct = total_ct; + + // --- [16] Intra-Bar Momentum (Welford online mean for each half) --- + if self.snapshot_count > 0 && self.prev_mid_momentum > 0.0 && mid > 0.0 { + let ret = (mid / self.prev_mid_momentum).ln(); + if ret.is_finite() { + // Determine which half based on timestamp + if snapshot.timestamp < self.bar_mid_ns { + self.half1_n += 1; + let delta = ret - self.half1_mean; + self.half1_mean += delta / self.half1_n as f64; + } else { + self.half2_n += 1; + let delta = ret - self.half2_mean; + self.half2_mean += delta / self.half2_n as f64; + } + } + } + self.prev_mid_momentum = mid; + + self.prev_mid = mid; + self.snapshot_count += 1; + } + + /// Update from a trade event. Called per trade. + pub fn update_trade(&mut self, _price: f64, _volume: u64, is_buy: bool, timestamp_ns: u64) { + // --- [10] Hawkes Trade Intensity --- + const MU: f64 = 1.0; + const ALPHA: f64 = 0.5; + const BETA: f64 = 1.0; + + if self.last_trade_ns > 0 && timestamp_ns > self.last_trade_ns { + let dt_sec = + (timestamp_ns - self.last_trade_ns) as f64 / 1_000_000_000.0; + let decay = (-BETA * dt_sec).exp(); + // Hawkes: intensity = mu + alpha * decay * (prev_intensity - mu) + alpha + self.hawkes_intensity = MU + (self.hawkes_intensity - MU) * decay + ALPHA; + } else { + // First trade + self.hawkes_intensity = MU + ALPHA; + } + // Clamp to [0, 100] + self.hawkes_intensity = self.hawkes_intensity.clamp(0.0, 100.0); + self.last_trade_ns = timestamp_ns; + + // --- [13] Aggression Ratio --- + self.total_trades += 1; + if is_buy { + self.trades_at_ask += 1; + } + } + + /// Update with derived OFI values. Called after OFI calculation. + pub fn update_ofi_derived(&mut self, ofi_l1: f64, vpin: f64) { + const EMA_ALPHA: f64 = 0.1; + + // --- [8] OFI Trajectory: online linear regression --- + // Use ofi update count as x (monotonic) + self.ofi_linreg_n += 1.0; + let x = self.ofi_linreg_n; + let y = ofi_l1; + self.ofi_linreg_sum_x += x; + self.ofi_linreg_sum_y += y; + self.ofi_linreg_sum_xy += x * y; + self.ofi_linreg_sum_xx += x * x; + + // --- [18] OFI Acceleration: EMA of OFI_L1 deltas --- + if self.has_prev_ofi { + let delta = ofi_l1 - self.prev_ofi_l1; + self.ofi_accel_ema = EMA_ALPHA * delta + (1.0 - EMA_ALPHA) * self.ofi_accel_ema; + } + self.prev_ofi_l1 = ofi_l1; + self.has_prev_ofi = true; + + // --- [19] Toxicity Gradient: EMA of VPIN deltas --- + if self.has_prev_vpin { + let delta = vpin - self.prev_vpin; + self.toxicity_grad_ema = + EMA_ALPHA * delta + (1.0 - EMA_ALPHA) * self.toxicity_grad_ema; + } + self.prev_vpin = vpin; + self.has_prev_vpin = true; + } + + /// Pure read — produce the 12-feature snapshot. Idempotent, non-mutating. + pub fn snapshot(&self) -> [f64; 12] { + // [8] OFI Trajectory — online linear regression slope + let ofi_trajectory = if self.ofi_linreg_n >= 2.0 { + let n = self.ofi_linreg_n; + let denom = n * self.ofi_linreg_sum_xx - self.ofi_linreg_sum_x * self.ofi_linreg_sum_x; + if denom.abs() > 1e-12 { + let slope = (n * self.ofi_linreg_sum_xy + - self.ofi_linreg_sum_x * self.ofi_linreg_sum_y) + / denom; + safe_clip(slope, -1e6, 1e6) + } else { + 0.0 + } + } else { + 0.0 + }; + + // [9] Realized Variance + let realized_variance = safe_clip(self.realized_var, 0.0, 1e6); + + // [10] Hawkes Trade Intensity + let hawkes_intensity = self.hawkes_intensity; + + // [11] Book Pressure Gradient + let book_pressure = safe_clip(self.book_pressure, -1.0, 1.0); + + // [12] Spread Dynamics: (max - min) / mean + let spread_dynamics = if self.spread_count >= 2 { + let mean_spread = self.spread_sum / self.spread_count as f64; + if mean_spread > 1e-15 { + safe_clip( + (self.spread_max - self.spread_min) / mean_spread, + 0.0, + 100.0, + ) + } else { + 0.0 + } + } else { + 0.0 + }; + + // [13] Aggression Ratio + let aggression_ratio = if self.total_trades > 0 { + self.trades_at_ask as f64 / self.total_trades as f64 + } else { + 0.5 // neutral default + }; + + // [14] Queue Depletion Asymmetry + let queue_depletion = { + let max_depl = self.bid_depletion_ema.abs().max(self.ask_depletion_ema.abs()); + if max_depl > 1e-12 { + safe_clip( + (self.bid_depletion_ema - self.ask_depletion_ema) / max_depl, + -1.0, + 1.0, + ) + } else { + 0.0 + } + }; + + // [15] Order Count Flux + let order_count_flux = { + let total = self.ct_increase_count + self.ct_decrease_count; + if total > 0 { + self.ct_increase_count as f64 / total as f64 + } else { + 0.5 + } + }; + + // [16] Intra-Bar Momentum: sign-weighted product of half means + let intra_bar_momentum = if self.half1_n > 0 && self.half2_n > 0 { + let sign1 = if self.half1_mean >= 0.0 { 1.0 } else { -1.0 }; + let sign2 = if self.half2_mean >= 0.0 { 1.0 } else { -1.0 }; + let product = sign1 * sign2 * (self.half1_mean.abs() * self.half2_mean.abs()).sqrt(); + safe_clip(product, -1.0, 1.0) + } else { + 0.0 + }; + + // [17] Microstructure Regime Score: sigmoid(aggression * spread_dyn * thin_book) + let regime_score = { + // thin_book: inverse of total depth, clamped + let aggr = (aggression_ratio - 0.5).abs() * 2.0; // how far from balanced + let thin_book = if book_pressure.abs() > 1e-12 { + 1.0 / (1.0 + book_pressure.abs() * 10.0) + } else { + 0.5 + }; + let raw = aggr * spread_dynamics * thin_book; + // sigmoid: 1 / (1 + exp(-x)) + 1.0 / (1.0 + (-raw).exp()) + }; + + // [18] OFI Acceleration + let ofi_acceleration = safe_clip(self.ofi_accel_ema, -1e6, 1e6); + + // [19] Toxicity Gradient + let toxicity_gradient = safe_clip(self.toxicity_grad_ema, -1.0, 1.0); + + [ + ofi_trajectory, // 8 + realized_variance, // 9 + hawkes_intensity, // 10 + book_pressure, // 11 + spread_dynamics, // 12 + aggression_ratio, // 13 + queue_depletion, // 14 + order_count_flux, // 15 + intra_bar_momentum, // 16 + regime_score, // 17 + ofi_acceleration, // 18 + toxicity_gradient, // 19 + ] + } + + /// Reset state for a new bar. + pub fn reset(&mut self, bar_start_ns: u64, bar_duration_ns: u64) { + *self = Self::new(bar_start_ns, bar_duration_ns); + } + + /// Weighted book pressure across all 10 levels. + /// + /// `sum(exp(-0.3*l) * (bid_sz - ask_sz)) / sum(bid_sz + ask_sz)` + fn calc_book_pressure(levels: &[BidAskPair]) -> f64 { + let max_levels = levels.len().min(10); + let mut weighted_sum = 0.0; + let mut total_volume = 0.0; + + for (i, level) in levels.iter().take(max_levels).enumerate() { + let weight = (-0.3 * i as f64).exp(); + weighted_sum += weight * (level.bid_sz as f64 - level.ask_sz as f64); + total_volume += level.bid_sz as f64 + level.ask_sz as f64; + } + + if total_volume < 1.0 { + return 0.0; + } + + safe_clip(weighted_sum / total_volume, -1.0, 1.0) + } +} + +/// Extended OFI features: base 8 + microstructure 12 = 20 features. +#[derive(Debug, Clone)] +pub struct ExtendedOFIFeatures { + /// Base 8 OFI features + pub base: OFIFeatures, + /// 12 microstructure features (indices 8-19) + pub micro: [f64; 12], +} + +impl ExtendedOFIFeatures { + /// Convert to flat 20-element array. + pub fn to_array(&self) -> [f64; 20] { + let base = self.base.to_array(); + let mut out = [0.0f64; 20]; + out[..8].copy_from_slice(&base); + out[8..20].copy_from_slice(&self.micro); + out + } + + /// Create zero-initialized extended features. + pub fn zeros() -> Self { + Self { + base: OFIFeatures::zeros(), + micro: [0.0; 12], + } + } +} + #[cfg(test)] mod tests { use super::*;