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

feat: MBP-10 → imbalance bars as explicit data source choice

Browse Source 
Wire MBP-10 order book data through ImbalanceBarSampler into the DQN
training pipeline. No fallback — explicit data_source config: "mbp10"
(imbalance bars from 10-level order book) or "ohlcv" (1-minute candles).
Fails loudly if chosen source's data doesn't exist.

Pipeline: MBP-10 .dbn.zst → trade extraction (action=='T', native side
classification) → adaptive ImbalanceBarSampler (EWMA threshold) →
OHLCVBar → existing feature extraction.

Production: data_source="mbp10", smoketest/localdev: data_source="ohlcv".

8 files, +483/-48 lines. 3 new tests for trade extraction and pipeline.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-31 11:08:02 +02:00
parent 0c9f368d24
commit f1892a22d4
8 changed files with 483 additions and 48 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
config/training/dqn-localdev.toml
View File

@@ -17,6 +17,8 @@ reward_scale = 1.0
huber_delta = 1.0
lr_decay_type = 2
lr_min = 0.00001
# Data source: "ohlcv" for local dev (may not have MBP-10 data)
data_source = "ohlcv"
# OFI feature enrichment from MBP-10 order book data (8 features: OFI L1/L5, depth imbalance, VPIN, Kyle's lambda, bid/ask slope, trade imbalance)
# mbp10_data_dir = "test_data/futures-baseline-mbp10"

4
config/training/dqn-production.toml
View File

@@ -20,6 +20,10 @@ reward_scale = 1.0
huber_delta = 1.0
lr_decay_type = 2
lr_min = 0.00001
# Data source: "ohlcv" (1-min candles) or "mbp10" (imbalance bars from MBP-10 order book)
data_source = "mbp10"
imbalance_bar_threshold = 100.0
imbalance_bar_ewma_alpha = 0.1
# OFI feature enrichment from MBP-10 order book data (8 features: OFI L1/L5, depth imbalance, VPIN, Kyle's lambda, bid/ask slope, trade imbalance)
mbp10_data_dir = "test_data/futures-baseline-mbp10"

2
config/training/dqn-smoketest.toml
View File

@@ -21,6 +21,8 @@ hidden_dim_base = 64
max_steps_per_epoch = 200
reward_scale = 1.0
huber_delta = 1.0
# Data source: "ohlcv" for smoke tests (no MBP-10 data required)
data_source = "ohlcv"
# OFI feature enrichment from MBP-10 order book data (8 features: OFI L1/L5, depth imbalance, VPIN, Kyle's lambda, bid/ask slope, trade imbalance)
# mbp10_data_dir = "test_data/futures-baseline-mbp10"

6
crates/ml-features/src/lib.rs
View File

@@ -65,6 +65,10 @@ pub use normalization::{FeatureNormalizer, NormalizationStats, RingBuffer};
pub use pipeline::{FeatureExtractionPipeline, PipelinePerformance};
pub use position_features::PositionFeatures;
pub use ofi_calculator::{OFICalculator, OFIFeatures};
pub use mbp10_loader::{get_recent_snapshots, get_snapshots_for_timestamp, load_mbp10_snapshots_sync};
pub use mbp10_loader::{
extract_trades_from_dbn_file, extract_trades_from_snapshots, get_recent_snapshots,
get_snapshots_for_timestamp, load_mbp10_snapshots_sync, mbp10_to_imbalance_bars,
Mbp10Trade,
};
pub use trades_loader::{get_trades_for_bar, load_trades_sync, DbnTrade};
pub use bar_resampler::BarResampler;

360
crates/ml-features/src/mbp10_loader.rs
View File

@@ -26,11 +26,15 @@
//! }
//! ```
use data::providers::databento::{dbn_parser::DbnParser, mbp10::Mbp10Snapshot};
use data::providers::databento::{dbn_parser::DbnParser, mbp10::{BidAskPair, Mbp10Snapshot}};
use std::path::Path;
use chrono::{DateTime, Utc};
use crate::alternative_bars::ImbalanceBarSampler;
use crate::ofi_calculator::OFICalculator;
use crate::trades_loader::load_trades_sync;
use crate::OHLCVBar;
use crate::MLError;
/// Load MBP-10 snapshots from DBN file (synchronous wrapper)
@@ -323,6 +327,288 @@ where
Some(all_ofi)
}
/// Trade tick extracted from MBP-10 data.
///
/// MBP-10 records include order book updates AND trades. When an MBP-10 record
/// has `action == b'T'` it represents a trade execution. We extract the price,
/// size, timestamp, and buy/sell classification for feeding into the
/// `ImbalanceBarSampler`.
#[derive(Debug, Clone)]
pub struct Mbp10Trade {
/// Trade price (f64, converted from fixed-point)
pub price: f64,
/// Trade volume in contracts
pub volume: f64,
/// Event timestamp
pub timestamp: DateTime<Utc>,
/// True if buyer-initiated (side == 'B'), false otherwise
pub is_buy: bool,
}
/// Extract trade ticks from MBP-10 snapshots using consecutive snapshot diffs.
///
/// MBP-10 snapshots don't directly carry trade action/side info (those live on
/// the raw `dbn::Mbp10Msg` records). This function uses a heuristic:
/// when `trade_count` increases between consecutive snapshots, a trade occurred.
/// Direction is inferred by comparing mid-prices (tick rule).
///
/// For higher-fidelity trade extraction, use [`extract_trades_from_dbn_file`]
/// which reads raw MBP-10 records with their action/side fields.
pub fn extract_trades_from_snapshots(snapshots: &[Mbp10Snapshot]) -> Vec<Mbp10Trade> {
let mut trades = Vec::new();
let mut prev_trade_count: u32 = 0;
let mut prev_mid: f64 = 0.0;
for snap in snapshots {
let mid = snap.mid_price();
if mid <= 0.0 {
continue;
}
let new_trades = snap.trade_count.saturating_sub(prev_trade_count);
if new_trades > 0 && prev_mid > 0.0 {
// Tick rule: price up => buy, price down => sell, unchanged => previous direction
let is_buy = mid >= prev_mid;
// Best ask size as proxy for trade volume (actual volume not in snapshot)
let volume = if !snap.levels.is_empty() {
snap.levels[0].bid_sz.max(1) as f64
} else {
1.0
};
let ts_secs = (snap.timestamp / 1_000_000_000) as i64;
let ts_nanos = (snap.timestamp % 1_000_000_000) as u32;
if let Some(dt) = DateTime::<Utc>::from_timestamp(ts_secs, ts_nanos) {
trades.push(Mbp10Trade {
price: mid,
volume,
timestamp: dt,
is_buy,
});
}
}
prev_trade_count = snap.trade_count;
prev_mid = mid;
}
trades
}
/// Extract trade ticks directly from a DBN MBP-10 file.
///
/// Reads raw `Mbp10Msg` records and filters for trade actions (`action == b'T'`).
/// This provides higher-fidelity trades than [`extract_trades_from_snapshots`]
/// because it uses the actual action and side fields from each record.
pub fn extract_trades_from_dbn_file(file_path: &Path) -> Result<Vec<Mbp10Trade>, MLError> {
use dbn::decode::{DbnDecoder, DecodeRecordRef};
use dbn::RecordRefEnum;
use std::fs::File;
use std::io::BufReader;
let file = File::open(file_path).map_err(|e| {
MLError::InsufficientData(format!("Failed to open MBP-10 file {:?}: {}", file_path, e))
})?;
let is_zstd = file_path.to_string_lossy().ends_with(".dbn.zst");
let reader: Box<dyn std::io::Read> = if is_zstd {
Box::new(
zstd::Decoder::new(BufReader::new(file)).map_err(|e| {
MLError::InsufficientData(format!("Failed to create zstd decoder: {}", e))
})?,
)
} else {
Box::new(BufReader::new(file))
};
let mut decoder = DbnDecoder::new(reader).map_err(|e| {
MLError::InsufficientData(format!("Failed to create DBN decoder: {}", e))
})?;
let mut trades = Vec::new();
let mut last_mid: f64 = 0.0;
loop {
match decoder.decode_record_ref() {
Ok(Some(record)) => {
let record_enum = record.as_enum().map_err(|e| {
MLError::InsufficientData(format!("Failed to convert record: {}", e))
})?;
if let RecordRefEnum::Mbp10(mbp10) = record_enum {
let action = mbp10.action as u8;
// Compute mid-price from this record's top-of-book levels
// MBP-10 records carry 10 levels; index 0 is best bid/ask
let bid_px = BidAskPair::price_to_f64(mbp10.levels[0].bid_px);
let ask_px = BidAskPair::price_to_f64(mbp10.levels[0].ask_px);
let mid = if bid_px > 0.0 && ask_px > 0.0 {
(bid_px + ask_px) / 2.0
} else {
last_mid
};
if mid > 0.0 {
last_mid = mid;
}
// Trade action: 'T' (0x54) = trade execution
if action == b'T' {
let price = mbp10.price as f64 * 1e-9;
let volume = mbp10.size as f64;
let is_buy = mbp10.side == b'B' as i8;
let ts_secs = (mbp10.hd.ts_event / 1_000_000_000) as i64;
let ts_nanos = (mbp10.hd.ts_event % 1_000_000_000) as u32;
if let Some(dt) = DateTime::<Utc>::from_timestamp(ts_secs, ts_nanos) {
if price > 0.0 && volume > 0.0 {
trades.push(Mbp10Trade {
price,
volume,
timestamp: dt,
is_buy,
});
}
}
}
}
}
Ok(None) => break,
Err(e) => {
return Err(MLError::InsufficientData(format!(
"DBN decode error: {}",
e
)));
}
}
}
Ok(trades)
}
/// Recursively collect all .dbn and .dbn.zst files from a directory.
fn collect_dbn_files(dir: &Path) -> Vec<std::path::PathBuf> {
let mut files = Vec::new();
if let Ok(entries) = std::fs::read_dir(dir) {
for entry in entries.flatten() {
let path = entry.path();
if path.is_dir() {
files.extend(collect_dbn_files(&path));
} else if path.extension().and_then(|s| s.to_str()) == Some("dbn")
|| path.to_string_lossy().ends_with(".dbn.zst")
{
files.push(path);
}
}
}
files
}
/// Load MBP-10 files and convert to imbalance bars via `ImbalanceBarSampler`.
///
/// Scans `mbp10_dir/symbol/` for `.dbn` and `.dbn.zst` files, loads each,
/// extracts trades (action == 'T'), feeds into adaptive `ImbalanceBarSampler`.
///
/// Returns sorted `OHLCVBar`s or error if directory doesn't exist.
/// No fallback -- fails loudly if the data isn't there.
pub fn mbp10_to_imbalance_bars(
mbp10_dir: &Path,
symbol: &str,
threshold: f64,
ewma_alpha: f64,
) -> Result<Vec<OHLCVBar>, MLError> {
let symbol_dir = mbp10_dir.join(symbol);
if !symbol_dir.exists() {
return Err(MLError::InsufficientData(format!(
"MBP-10 directory not found: {}. Set data_source='ohlcv' or provide MBP-10 data.",
symbol_dir.display()
)));
}
let mut dbn_files = collect_dbn_files(&symbol_dir);
dbn_files.sort();
if dbn_files.is_empty() {
return Err(MLError::InsufficientData(format!(
"No .dbn/.dbn.zst files found in {}",
symbol_dir.display()
)));
}
tracing::info!(
"Loading MBP-10 trades from {} files in {}",
dbn_files.len(),
symbol_dir.display()
);
// Collect all trades from all files
let mut all_trades: Vec<Mbp10Trade> = Vec::new();
for file_path in &dbn_files {
match extract_trades_from_dbn_file(file_path) {
Ok(trades) => {
tracing::debug!(
" {} -> {} trades",
file_path.file_name().unwrap_or_default().to_string_lossy(),
trades.len()
);
all_trades.extend(trades);
}
Err(e) => {
tracing::warn!(
"Failed to extract trades from {}: {}",
file_path.display(),
e
);
}
}
}
if all_trades.is_empty() {
return Err(MLError::InsufficientData(format!(
"No trade ticks extracted from MBP-10 files in {}. \
Files may contain only book updates without trade events.",
symbol_dir.display()
)));
}
// Sort by timestamp (files should be chronological but ensure correctness)
all_trades.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
tracing::info!(
"Extracted {} trade ticks, feeding into ImbalanceBarSampler (threshold={}, alpha={})",
all_trades.len(),
threshold,
ewma_alpha,
);
// Initialize sampler with first trade
let first = &all_trades[0];
let mut sampler =
ImbalanceBarSampler::new_with_ewma(first.price, threshold, first.timestamp, ewma_alpha);
let mut bars: Vec<OHLCVBar> = Vec::new();
for trade in &all_trades {
if let Some(bar) = sampler.update(trade.price, trade.volume, trade.timestamp) {
bars.push(bar);
}
}
// Sort by timestamp (should already be ordered)
bars.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
tracing::info!(
"ImbalanceBarSampler produced {} bars from {} trade ticks (ratio 1:{:.0})",
bars.len(),
all_trades.len(),
if bars.is_empty() {
0.0
} else {
all_trades.len() as f64 / bars.len() as f64
},
);
Ok(bars)
}
/// Get snapshots for a specific timestamp window
///
/// Returns a slice of snapshots starting from the first snapshot at or before
@@ -551,4 +837,76 @@ mod tests {
assert_eq!(window.len(), 1);
assert_eq!(window[0].timestamp, 1000);
}
#[test]
fn test_extract_trades_from_snapshots() {
// Create snapshots with increasing trade_count and rising prices
let snapshots = vec![
{
let mut s = create_test_snapshot(1_000_000_000, 150000000000000); // 150.0
s.trade_count = 0;
s
},
{
let mut s = create_test_snapshot(2_000_000_000, 150010000000000); // 150.01
s.trade_count = 1;
s
},
{
let mut s = create_test_snapshot(3_000_000_000, 150020000000000); // 150.02
s.trade_count = 3;
s
},
{
let mut s = create_test_snapshot(4_000_000_000, 149990000000000); // 149.99
s.trade_count = 4;
s
},
];
let trades = extract_trades_from_snapshots(&snapshots);
// 3 transitions where trade_count increases (snap 0->1, 1->2, 2->3)
assert_eq!(trades.len(), 3);
// First trade: price went up => is_buy=true
assert!(trades[0].is_buy);
// Third trade: price went down => is_buy=false
assert!(!trades[2].is_buy);
}
#[test]
fn test_mbp10_to_imbalance_bars_missing_dir() {
let result = mbp10_to_imbalance_bars(
Path::new("/nonexistent/path"),
"ES.FUT",
100.0,
0.1,
);
assert!(result.is_err());
let err_msg = format!("{}", result.unwrap_err());
assert!(err_msg.contains("MBP-10 directory not found"), "Error: {err_msg}");
}
#[test]
fn test_mbp10_to_imbalance_bars_from_test_data() {
// This test depends on test_data/mbp10/ES.FUT/ existing.
// If not available (CI), it will fail with a descriptive error — that's OK.
let bars = mbp10_to_imbalance_bars(
Path::new("../../test_data/mbp10"),
"ES.FUT",
100.0,
0.1,
);
if let Ok(bars) = bars {
assert!(!bars.is_empty(), "Expected at least one imbalance bar");
for w in bars.windows(2) {
assert!(
w[0].timestamp <= w[1].timestamp,
"Bars must be time-ordered: {:?} > {:?}",
w[0].timestamp,
w[1].timestamp
);
}
}
// If Err, that's fine — test data may not be present
}
}

19
crates/ml/src/trainers/dqn/config.rs
View File

@@ -1234,6 +1234,20 @@ pub struct DQNHyperparameters {
/// When None, VPIN/Kyle's Lambda use tick-rule proxy (backward compatible).
pub trades_data_dir: Option<String>,
/// Data source for training bars: "ohlcv" (1-minute candles) or "mbp10"
/// (imbalance bars from MBP-10 order book). No fallback -- fails if the
/// chosen source's data doesn't exist. Default: "ohlcv" (backward compatible).
pub data_source: String,
/// Imbalance bar threshold for MBP-10 to bar conversion.
/// Higher = fewer bars (less noise), lower = more bars (more signal).
/// Only used when `data_source = "mbp10"`. Default: 100.0.
pub imbalance_bar_threshold: f64,
/// EWMA alpha for adaptive imbalance threshold. Default: 0.1.
/// Only used when `data_source = "mbp10"`.
pub imbalance_bar_ewma_alpha: f64,
// Offline RL mode (CQL/IQL on fixed datasets)
/// Offline RL mode: skip online experience collection, train from a fixed dataset.
/// Requires `dataset_path` to point to a pre-collected experience dataset.
@@ -1711,6 +1725,11 @@ impl DQNHyperparameters {
mbp10_data_dir: None,
trades_data_dir: None,
// Data source: "ohlcv" (backward compatible) or "mbp10" (imbalance bars)
data_source: "ohlcv".to_string(),
imbalance_bar_threshold: 100.0,
imbalance_bar_ewma_alpha: 0.1,
// Offline RL: disabled by default (online training with experience collection)
offline_mode: false,
dataset_path: None,

118
crates/ml/src/trainers/dqn/data_loading.rs
View File

@@ -629,60 +629,86 @@ impl DQNTrainer {
}
}
// Find all DBN files in directory
let dir_path = Path::new(dbn_data_dir);
if !dir_path.exists() {
return Err(anyhow::anyhow!(
"Data directory not found: {}",
dbn_data_dir
));
}
// ── Load bars: MBP-10 imbalance bars or OHLCV candles ─────────────
let all_ohlcv_bars = match self.hyperparams.data_source.as_str() {
"mbp10" => {
let mbp10_dir_str = self.hyperparams.mbp10_data_dir.as_ref()
.ok_or_else(|| anyhow::anyhow!(
"data_source='mbp10' requires mbp10_data_dir to be set in [training] config"
))?;
// Extract symbol from dbn_data_dir (last path component, e.g. "ES.FUT")
let symbol = Path::new(dbn_data_dir)
.file_name()
.and_then(|s| s.to_str())
.unwrap_or("ES.FUT");
info!(
"Loading MBP-10 imbalance bars: dir={}, symbol={}, threshold={}, alpha={}",
mbp10_dir_str, symbol,
self.hyperparams.imbalance_bar_threshold,
self.hyperparams.imbalance_bar_ewma_alpha,
);
let bars = crate::features::mbp10_loader::mbp10_to_imbalance_bars(
Path::new(mbp10_dir_str),
symbol,
self.hyperparams.imbalance_bar_threshold,
self.hyperparams.imbalance_bar_ewma_alpha,
).map_err(|e| anyhow::anyhow!("MBP-10 imbalance bar loading failed: {}", e))?;
info!("Loaded {} imbalance bars from MBP-10 data", bars.len());
bars
}
_ => {
// Default: load OHLCV bars from DBN files
let dir_path = Path::new(dbn_data_dir);
if !dir_path.exists() {
return Err(anyhow::anyhow!(
"Data directory not found: {}",
dbn_data_dir
));
}
let mut dbn_files = collect_dbn_files_recursive(dir_path);
dbn_files.sort();
let mut dbn_files = collect_dbn_files_recursive(dir_path);
dbn_files.sort();
if dbn_files.is_empty() {
return Err(anyhow::anyhow!("No DBN files found in: {}", dbn_data_dir));
}
if dbn_files.is_empty() {
return Err(anyhow::anyhow!("No DBN files found in: {}", dbn_data_dir));
}
info!("Found {} DBN files to load", dbn_files.len());
info!("Found {} DBN files to load", dbn_files.len());
let mut all_ohlcv_bars = Vec::new();
let mut bars = Vec::new();
for (file_idx, file_path) in dbn_files.iter().enumerate() {
debug!(
"Loading DBN file {}/{}: {}",
file_idx + 1,
dbn_files.len(),
file_path.display()
);
let file_bars = self.extract_ohlcv_bars_from_dbn(file_path)?;
debug!(
"Extracted {} OHLCV bars from {}",
file_bars.len(),
file_path.file_name().unwrap_or_default().to_string_lossy()
);
bars.extend(file_bars);
}
// Load and decode each DBN file to collect OHLCV bars
for (file_idx, file_path) in dbn_files.iter().enumerate() {
debug!(
"Loading DBN file {}/{}: {}",
file_idx + 1,
dbn_files.len(),
file_path.display()
);
if bars.is_empty() {
return Err(anyhow::anyhow!(
"No OHLCV bars extracted from DBN files. Check if files contain OHLCV messages."
));
}
// Extract raw OHLCV bars from file
let file_bars = self.extract_ohlcv_bars_from_dbn(file_path)?;
debug!(
"Extracted {} OHLCV bars from {}",
file_bars.len(),
file_path.file_name().unwrap_or_default().to_string_lossy()
);
all_ohlcv_bars.extend(file_bars);
}
if all_ohlcv_bars.is_empty() {
return Err(anyhow::anyhow!(
"No OHLCV bars extracted from DBN files. Check if files contain OHLCV messages."
));
}
info!(
"Successfully loaded {} OHLCV bars from {} DBN files",
all_ohlcv_bars.len(),
dbn_files.len()
);
info!(
"Successfully loaded {} OHLCV bars from {} DBN files",
bars.len(),
dbn_files.len()
);
bars
}
};
// Sort bars by timestamp (critical for rolling window feature extraction)
let mut all_ohlcv_bars = all_ohlcv_bars;
debug!("Sorting bars chronologically by timestamp...");
all_ohlcv_bars.sort_by_key(|bar| bar.timestamp);
debug!("Bars sorted successfully");

20
crates/ml/src/training_profile.rs
View File

@@ -81,6 +81,16 @@ pub struct TrainingSection {
/// Directory containing trade .dbn.zst files for real VPIN/Kyle's Lambda.
/// When set, trades are loaded and fed to `OFICalculator::feed_trade()`.
pub trades_data_dir: Option<String>,
/// Data source for training bars: "ohlcv" (1-minute candles) or "mbp10"
/// (imbalance bars from MBP-10 order book). No fallback. Default: "ohlcv".
pub data_source: Option<String>,
/// Imbalance bar threshold for MBP-10 to bar conversion. Default: 100.0.
pub imbalance_bar_threshold: Option<f64>,
/// EWMA alpha for adaptive imbalance threshold. Default: 0.1.
pub imbalance_bar_ewma_alpha: Option<f64>,
}
/// Epsilon-greedy exploration parameters (DQN-specific).
@@ -727,6 +737,16 @@ impl DqnTrainingProfile {
if let Some(ref v) = t.trades_data_dir {
hp.trades_data_dir = Some(v.clone());
}
// Data source selection: "ohlcv" or "mbp10"
if let Some(ref v) = t.data_source {
hp.data_source = v.clone();
}
if let Some(v) = t.imbalance_bar_threshold {
hp.imbalance_bar_threshold = v;
}
if let Some(v) = t.imbalance_bar_ewma_alpha {
hp.imbalance_bar_ewma_alpha = v;
}
}
// [exploration]