6e339316cf
Adds a parent/child GitLab CI pipeline for ML model training: - Generator script produces per-model hyperopt/train/evaluate jobs - Parent pipeline (.gitlab-ci-training.yml) with manual trigger - NFS-backed ReadWriteMany PVC for shared training outputs - Hyperopt params wired into training binaries (DQN, PPO, TFT, Mamba2) - Shared DBN loader eliminates duplicate code across hyperopt adapters - Supervised hyperopt unified to DBN data (was parquet-only) Pipeline: hyperopt (4 models) → train (10 models) → evaluate ensemble Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
98 lines
3.4 KiB
Rust
98 lines
3.4 KiB
Rust
//! Shared DBN file loading utilities for hyperopt adapters.
|
|
//!
|
|
//! Provides recursive .dbn/.dbn.zst file discovery and OHLCV bar decoding.
|
|
//! Used by TFT, Mamba2, and DQN hyperopt adapters to load training data
|
|
//! from Databento files.
|
|
|
|
use std::path::{Path, PathBuf};
|
|
|
|
use tracing::info;
|
|
|
|
use crate::features::extraction::OHLCVBar;
|
|
|
|
/// Recursively collect all .dbn and .dbn.zst files from a directory.
|
|
pub fn collect_dbn_files(dir: &Path) -> Vec<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);
|
|
} else {
|
|
// Skip non-DBN files
|
|
}
|
|
}
|
|
}
|
|
files.sort();
|
|
files
|
|
}
|
|
|
|
/// Decode OHLCV bars from a DBN decoder stream.
|
|
pub fn decode_ohlcv_bars<R: std::io::Read>(
|
|
decoder: &mut dbn::decode::dbn::Decoder<R>,
|
|
) -> anyhow::Result<Vec<OHLCVBar>> {
|
|
use dbn::decode::DecodeRecordRef;
|
|
use dbn::OhlcvMsg;
|
|
|
|
let mut bars = Vec::new();
|
|
while let Some(record_ref) = decoder
|
|
.decode_record_ref()
|
|
.map_err(|e| anyhow::anyhow!("Failed to decode record: {}", e))?
|
|
{
|
|
if let Some(ohlcv) = record_ref.get::<OhlcvMsg>() {
|
|
bars.push(OHLCVBar {
|
|
timestamp: chrono::DateTime::from_timestamp_nanos(ohlcv.hd.ts_event as i64),
|
|
open: ohlcv.open as f64 / 1e9,
|
|
high: ohlcv.high as f64 / 1e9,
|
|
low: ohlcv.low as f64 / 1e9,
|
|
close: ohlcv.close as f64 / 1e9,
|
|
volume: ohlcv.volume as f64,
|
|
});
|
|
}
|
|
}
|
|
Ok(bars)
|
|
}
|
|
|
|
/// Load all OHLCV bars from DBN files in a directory, sorted chronologically.
|
|
pub fn load_bars_from_dbn_dir(dir: &Path) -> anyhow::Result<Vec<OHLCVBar>> {
|
|
use dbn::decode::dbn::Decoder as DbnDecoder;
|
|
|
|
let dbn_files = collect_dbn_files(dir);
|
|
if dbn_files.is_empty() {
|
|
anyhow::bail!("No .dbn or .dbn.zst files found in {}", dir.display());
|
|
}
|
|
|
|
info!("Found {} DBN files in {}", dbn_files.len(), dir.display());
|
|
|
|
let mut all_bars = Vec::new();
|
|
for dbn_file in &dbn_files {
|
|
let is_zstd = dbn_file.to_string_lossy().ends_with(".dbn.zst");
|
|
let bars = if is_zstd {
|
|
let mut decoder = DbnDecoder::from_zstd_file(dbn_file)
|
|
.map_err(|e| anyhow::anyhow!("Failed to open {}: {}", dbn_file.display(), e))?;
|
|
decode_ohlcv_bars(&mut decoder)?
|
|
} else {
|
|
let file = std::fs::File::open(dbn_file)
|
|
.map_err(|e| anyhow::anyhow!("Failed to open {}: {}", dbn_file.display(), e))?;
|
|
let buf = std::io::BufReader::new(file);
|
|
let mut decoder = DbnDecoder::new(buf)
|
|
.map_err(|e| anyhow::anyhow!("Failed to decode {}: {}", dbn_file.display(), e))?;
|
|
decode_ohlcv_bars(&mut decoder)?
|
|
};
|
|
info!(" {} -> {} bars", dbn_file.display(), bars.len());
|
|
all_bars.extend(bars);
|
|
}
|
|
|
|
all_bars.sort_by_key(|b| b.timestamp);
|
|
info!(
|
|
"Loaded {} bars total from {} DBN files",
|
|
all_bars.len(),
|
|
dbn_files.len()
|
|
);
|
|
Ok(all_bars)
|
|
}
|