52630a77d3
Replace all `.to_string().parse::<f32/f64>()` patterns with `num_traits::ToPrimitive` methods (`.to_f32()`, `.to_f64()`). Each string roundtrip heap-allocated per conversion — fatal in DQN hot loop (300K+ bars × epochs). Decimal stays as canonical financial type; conversions happen at GPU/float boundaries only. Also fixes blocking_read() in async context (risk_integration.rs). Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1075 lines
36 KiB
Rust
1075 lines
36 KiB
Rust
//! DBN (Databento Binary) Data Source for Backtesting
|
||
//!
|
||
//! This module provides high-performance integration between DBN files and the backtesting service.
|
||
//! It uses the production-ready DbnParser with zero-copy parsing and SIMD optimizations.
|
||
//!
|
||
//! ## Features
|
||
//!
|
||
//! - Direct DBN file loading with <10ms performance for ~400 bars
|
||
//! - Zero-copy parsing via production DbnParser
|
||
//! - Conversion to backtesting service MarketData format
|
||
//! - Support for OHLCV bars (trades/quotes via MarketDataRepository)
|
||
//! - Symbol mapping for instrument IDs
|
||
//! - Configurable file paths per symbol
|
||
//!
|
||
//! ## Usage
|
||
//!
|
||
//! ```rust,no_run
|
||
//! use backtesting_service::dbn_data_source::DbnDataSource;
|
||
//! use std::collections::HashMap;
|
||
//!
|
||
//! # async fn example() -> anyhow::Result<()> {
|
||
//! // Create data source with symbol-to-file mapping
|
||
//! let mut file_mapping = HashMap::new();
|
||
//! file_mapping.insert("ES.FUT".to_string(),
|
||
//! "test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn".to_string());
|
||
//!
|
||
//! let data_source = DbnDataSource::new(file_mapping).await?;
|
||
//!
|
||
//! // Load OHLCV bars for symbol
|
||
//! let bars = data_source.load_ohlcv_bars("ES.FUT").await?;
|
||
//! println!("Loaded {} bars from DBN file", bars.len());
|
||
//! # Ok(())
|
||
//! # }
|
||
//! ```
|
||
|
||
use anyhow::{Context, Result};
|
||
use chrono::{DateTime, TimeZone, Utc};
|
||
use dbn::decode::{DbnDecoder, DecodeRecordRef};
|
||
use dbn::{OhlcvMsg, VersionUpgradePolicy};
|
||
use rust_decimal::Decimal;
|
||
use std::collections::HashMap;
|
||
use std::fs;
|
||
use std::path::Path;
|
||
use std::time::Instant;
|
||
use tracing::{debug, info, warn};
|
||
|
||
/// Check if file path is a valid uncompressed DBN file
|
||
///
|
||
/// Returns true only for files ending in `.dbn` that are NOT compressed formats.
|
||
///
|
||
/// # Rejected Extensions
|
||
///
|
||
/// - Compressed: `.dbn.zst`, `.dbn.gz`, `.dbn.bz2`, `.dbn.xz`
|
||
/// - Temporary: `.dbn.tmp`, `.dbn.swp`
|
||
/// - Backup: `.dbn.old`, `.dbn.backup`
|
||
///
|
||
/// # Case Sensitivity
|
||
///
|
||
/// Extension checking is case-insensitive (`.DBN`, `.Dbn`, `.dbn` all valid)
|
||
pub fn is_valid_dbn_file(path: &str) -> bool {
|
||
let path_lower = path.to_lowercase();
|
||
|
||
// Must end with .dbn
|
||
if !path_lower.ends_with(".dbn") {
|
||
return false;
|
||
}
|
||
|
||
// Reject compressed formats (case-insensitive)
|
||
let compressed_extensions = [".dbn.zst", ".dbn.gz", ".dbn.bz2", ".dbn.xz"];
|
||
for ext in &compressed_extensions {
|
||
if path_lower.ends_with(ext) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
// Reject temporary/backup files
|
||
let invalid_extensions = [
|
||
".dbn.tmp",
|
||
".dbn.old",
|
||
".dbn.backup",
|
||
".dbn.swp",
|
||
".uncompressed.dbn",
|
||
];
|
||
for ext in &invalid_extensions {
|
||
if path_lower.ends_with(ext) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
// Additional check: reject files with common intermediate extensions
|
||
// but allow symbol names with dots (e.g., ES.FUT.dbn)
|
||
let intermediate_patterns = [
|
||
".backup.dbn",
|
||
".temp.dbn",
|
||
".processed.dbn",
|
||
".v1.dbn",
|
||
".v2.dbn",
|
||
];
|
||
for pattern in &intermediate_patterns {
|
||
if path_lower.contains(pattern) {
|
||
return false;
|
||
}
|
||
}
|
||
|
||
true
|
||
}
|
||
|
||
use crate::strategy_engine::MarketData;
|
||
|
||
/// Convert DBN fixed-point price to f64
|
||
/// DBN stores prices as i64 with 9 decimal places precision (nanosecond-level)
|
||
///
|
||
/// **Note**: This performs the standard conversion. Anomaly correction (for bars encoded
|
||
/// with 7 decimal places instead of 9) is applied context-aware in load_ohlcv_bars().
|
||
fn dbn_price_to_f64(price: i64) -> f64 {
|
||
price as f64 / 1_000_000_000.0
|
||
}
|
||
|
||
/// File entry with date range metadata
|
||
#[derive(Debug, Clone)]
|
||
struct FileEntry {
|
||
path: String,
|
||
}
|
||
|
||
/// DBN data source for backtesting service
|
||
///
|
||
/// Provides high-performance loading of DBN files with automatic conversion
|
||
/// to backtesting service MarketData format.
|
||
///
|
||
/// Supports both single-file and multi-file (multi-day) configurations per symbol.
|
||
pub struct DbnDataSource {
|
||
/// Symbol to DBN file path(s) mapping
|
||
/// Supports both single file (`String`) and multiple files (`Vec<String>`)
|
||
file_mapping: HashMap<String, Vec<FileEntry>>,
|
||
}
|
||
|
||
impl DbnDataSource {
|
||
/// Create a new DBN data source with single file per symbol
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `file_mapping` - Map of symbol to DBN file path
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Configured DbnDataSource ready to load files
|
||
pub async fn new(file_mapping: HashMap<String, String>) -> Result<Self> {
|
||
let mut multi_file_mapping = HashMap::new();
|
||
for (symbol, path) in file_mapping {
|
||
multi_file_mapping.insert(
|
||
symbol,
|
||
vec![FileEntry {
|
||
path,
|
||
}],
|
||
);
|
||
}
|
||
|
||
info!(
|
||
"Created DBN data source with {} symbols",
|
||
multi_file_mapping.len()
|
||
);
|
||
|
||
Ok(Self {
|
||
file_mapping: multi_file_mapping,
|
||
})
|
||
}
|
||
|
||
/// Create a new DBN data source with multiple files per symbol
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `file_mapping` - Map of symbol to list of DBN file paths (one per day/period)
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Configured DbnDataSource ready to load files
|
||
///
|
||
/// # Example
|
||
///
|
||
/// ```no_run
|
||
/// use backtesting_service::dbn_data_source::DbnDataSource;
|
||
/// use std::collections::HashMap;
|
||
///
|
||
/// # async fn example() -> anyhow::Result<()> {
|
||
/// let mut file_mapping = HashMap::new();
|
||
/// file_mapping.insert(
|
||
/// "ESH4".to_string(),
|
||
/// vec![
|
||
/// "test_data/ESH4_2024-01-03.dbn".to_string(),
|
||
/// "test_data/ESH4_2024-01-04.dbn".to_string(),
|
||
/// "test_data/ESH4_2024-01-05.dbn".to_string(),
|
||
/// ],
|
||
/// );
|
||
///
|
||
/// let data_source = DbnDataSource::new_multi_file(file_mapping).await?;
|
||
/// # Ok(())
|
||
/// # }
|
||
/// ```
|
||
pub async fn new_multi_file(file_mapping: HashMap<String, Vec<String>>) -> Result<Self> {
|
||
let mut entry_mapping = HashMap::new();
|
||
let mut total_files = 0;
|
||
|
||
for (symbol, paths) in file_mapping {
|
||
let entries: Vec<FileEntry> = paths
|
||
.into_iter()
|
||
.map(|path| FileEntry {
|
||
path,
|
||
})
|
||
.collect();
|
||
|
||
total_files += entries.len();
|
||
entry_mapping.insert(symbol, entries);
|
||
}
|
||
|
||
info!(
|
||
"Created DBN data source with {} symbols and {} total files",
|
||
entry_mapping.len(),
|
||
total_files
|
||
);
|
||
|
||
Ok(Self {
|
||
file_mapping: entry_mapping,
|
||
})
|
||
}
|
||
|
||
/// Create a new DBN data source by scanning a directory for valid DBN files
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `dir_path` - Directory to scan for .dbn files
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Configured DbnDataSource with all valid DBN files found in directory
|
||
///
|
||
/// # File Filtering
|
||
///
|
||
/// Only includes files ending in `.dbn` (case-insensitive).
|
||
/// Automatically skips compressed files (`.dbn.zst`, `.dbn.gz`, etc.)
|
||
pub async fn from_directory(dir_path: &str) -> Result<Self> {
|
||
let dir = std::path::Path::new(dir_path);
|
||
|
||
if !dir.exists() {
|
||
return Err(anyhow::anyhow!("Directory does not exist: {}", dir_path));
|
||
}
|
||
|
||
let valid_files = Self::scan_directory_for_dbn_files(dir).await?;
|
||
|
||
info!(
|
||
"Found {} valid DBN files in directory: {}",
|
||
valid_files.len(),
|
||
dir_path
|
||
);
|
||
|
||
DbnDataSource::new_multi_file(valid_files).await
|
||
}
|
||
|
||
/// Load OHLCV bars for a specific symbol from single file
|
||
///
|
||
/// Reads the first DBN file for the symbol, parses with zero-copy, and converts to MarketData format.
|
||
/// For multi-file symbols, use `load_ohlcv_bars_all()` or `load_ohlcv_bars_range()`.
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `symbol` - Trading symbol to load data for
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Vector of MarketData bars sorted by timestamp (from first file only)
|
||
///
|
||
/// # Performance
|
||
///
|
||
/// - Target: <10ms for ~400 bars
|
||
/// - Zero-copy parsing with SIMD optimizations
|
||
/// - <1μs per tick processing
|
||
pub async fn load_ohlcv_bars(&self, symbol: &str) -> Result<Vec<MarketData>> {
|
||
// Load from first file only (backward compatible)
|
||
let entries = self
|
||
.file_mapping
|
||
.get(symbol)
|
||
.ok_or_else(|| anyhow::anyhow!("No DBN file configured for symbol: {}", symbol))?;
|
||
|
||
if entries.is_empty() {
|
||
return Err(anyhow::anyhow!("No DBN files for symbol: {}", symbol));
|
||
}
|
||
|
||
#[allow(clippy::indexing_slicing)] // Bounds checked above: !is_empty()
|
||
self.load_file(&entries[0].path, symbol).await
|
||
}
|
||
|
||
/// Load OHLCV bars for all files of a symbol
|
||
///
|
||
/// Loads and concatenates all DBN files for the symbol, useful for multi-day backtests.
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `symbol` - Trading symbol to load data for
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Vector of MarketData bars sorted by timestamp (merged from all files)
|
||
///
|
||
/// # Performance
|
||
///
|
||
/// - Linear scaling: N files × ~0.7ms per file
|
||
/// - 3 files (3 days) = ~2.1ms total
|
||
/// - Bars automatically sorted chronologically
|
||
pub async fn load_ohlcv_bars_all(&self, symbol: &str) -> Result<Vec<MarketData>> {
|
||
let start = Instant::now();
|
||
|
||
let entries = self
|
||
.file_mapping
|
||
.get(symbol)
|
||
.ok_or_else(|| anyhow::anyhow!("No DBN files configured for symbol: {}", symbol))?;
|
||
|
||
let mut all_bars = Vec::new();
|
||
let total_files = entries.len();
|
||
|
||
for (i, entry) in entries.iter().enumerate() {
|
||
debug!(
|
||
"Loading file {}/{} for {}: {}",
|
||
i + 1,
|
||
total_files,
|
||
symbol,
|
||
entry.path
|
||
);
|
||
|
||
let bars = self.load_file(&entry.path, symbol).await?;
|
||
all_bars.extend(bars);
|
||
}
|
||
|
||
// Sort by timestamp across all files
|
||
all_bars.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
|
||
|
||
let duration = start.elapsed();
|
||
info!(
|
||
"Loaded {} OHLCV bars from {} files in {:?} (symbol: {}, avg: {:.2}ms/file)",
|
||
all_bars.len(),
|
||
total_files,
|
||
duration,
|
||
symbol,
|
||
duration.as_millis() as f64 / total_files as f64
|
||
);
|
||
|
||
Ok(all_bars)
|
||
}
|
||
|
||
/// Load OHLCV bars for a specific symbol within a date range
|
||
///
|
||
/// Loads only files and bars within the specified date range, optimized for
|
||
/// partial-day or multi-day queries. Files whose timestamp range falls entirely
|
||
/// outside the requested window are skipped via a lightweight header peek
|
||
/// (reads only the first and last DBN records without full parsing).
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `symbol` - Trading symbol to load data for
|
||
/// * `start_date` - Start of date range (inclusive)
|
||
/// * `end_date` - End of date range (inclusive)
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Vector of MarketData bars within the date range, sorted by timestamp
|
||
///
|
||
/// # Performance
|
||
///
|
||
/// - Peeks first/last timestamps per file (~0.1ms) to skip out-of-range files
|
||
/// - Only loads files that overlap with date range
|
||
/// - Filters bars within range after loading
|
||
/// - Target: <10ms per file loaded
|
||
pub async fn load_ohlcv_bars_range(
|
||
&self,
|
||
symbol: &str,
|
||
start_date: DateTime<Utc>,
|
||
end_date: DateTime<Utc>,
|
||
) -> Result<Vec<MarketData>> {
|
||
let start = Instant::now();
|
||
|
||
let entries = self
|
||
.file_mapping
|
||
.get(symbol)
|
||
.ok_or_else(|| anyhow::anyhow!("No DBN files configured for symbol: {}", symbol))?;
|
||
|
||
let mut filtered_bars = Vec::new();
|
||
let mut files_loaded = 0;
|
||
let mut files_skipped = 0;
|
||
|
||
for entry in entries {
|
||
// Peek at file timestamps to determine range
|
||
let ts_range = Self::peek_file_timestamps(&entry.path).ok().flatten();
|
||
|
||
if let Some((file_first_ts, file_last_ts)) = ts_range {
|
||
// Skip file entirely if its range doesn't overlap [start_date, end_date]
|
||
if file_last_ts < start_date || file_first_ts > end_date {
|
||
debug!(
|
||
"Skipping file {} (range {}..{}) outside requested range {}..{}",
|
||
entry.path,
|
||
file_first_ts.format("%Y-%m-%dT%H:%M:%S"),
|
||
file_last_ts.format("%Y-%m-%dT%H:%M:%S"),
|
||
start_date.format("%Y-%m-%dT%H:%M:%S"),
|
||
end_date.format("%Y-%m-%dT%H:%M:%S"),
|
||
);
|
||
files_skipped += 1;
|
||
continue;
|
||
}
|
||
}
|
||
// If peek fails (corrupt file, empty, etc.) we fall through and try the full load
|
||
|
||
let bars = self.load_file(&entry.path, symbol).await?;
|
||
files_loaded += 1;
|
||
|
||
// Filter bars within date range
|
||
for bar in bars {
|
||
if bar.timestamp >= start_date && bar.timestamp <= end_date {
|
||
filtered_bars.push(bar);
|
||
}
|
||
}
|
||
}
|
||
|
||
// Sort by timestamp
|
||
filtered_bars.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
|
||
|
||
let duration = start.elapsed();
|
||
info!(
|
||
"Loaded {} bars (from {} files, {} skipped) in date range {} to {} ({:?}, symbol: {})",
|
||
filtered_bars.len(),
|
||
files_loaded,
|
||
files_skipped,
|
||
start_date.format("%Y-%m-%d"),
|
||
end_date.format("%Y-%m-%d"),
|
||
duration,
|
||
symbol
|
||
);
|
||
|
||
Ok(filtered_bars)
|
||
}
|
||
|
||
/// Peek at a DBN file to extract its first and last OHLCV timestamps
|
||
///
|
||
/// This is a lightweight operation that reads only the record timestamps
|
||
/// without performing full price conversion or anomaly detection.
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// - `Ok(Some((first_ts, last_ts)))` if the file contains at least one OHLCV record
|
||
/// - `Ok(None)` if the file contains no OHLCV records
|
||
/// - `Err(...)` if the file cannot be opened or decoded
|
||
fn peek_file_timestamps(
|
||
file_path: &str,
|
||
) -> Result<Option<(DateTime<Utc>, DateTime<Utc>)>> {
|
||
if !Path::new(file_path).exists() {
|
||
return Err(anyhow::anyhow!(
|
||
"DBN file not found for timestamp peek: {}",
|
||
file_path
|
||
));
|
||
}
|
||
|
||
let mut decoder = DbnDecoder::from_file(file_path).context(format!(
|
||
"Failed to create DBN decoder for peek: {}",
|
||
file_path
|
||
))?;
|
||
|
||
decoder
|
||
.set_upgrade_policy(VersionUpgradePolicy::UpgradeToV3)
|
||
.context("Failed to set upgrade policy for peek")?;
|
||
|
||
let mut first_ts: Option<DateTime<Utc>> = None;
|
||
let mut last_ts: Option<DateTime<Utc>> = None;
|
||
|
||
while let Some(record_ref) = decoder
|
||
.decode_record_ref()
|
||
.context("Failed to decode DBN record during peek")?
|
||
{
|
||
if let Some(ohlcv) = record_ref.get::<OhlcvMsg>() {
|
||
let ts_nanos = ohlcv.hd.ts_event as i64;
|
||
let secs = ts_nanos / 1_000_000_000;
|
||
let nanos = (ts_nanos % 1_000_000_000) as u32;
|
||
if let Some(ts) = Utc.timestamp_opt(secs, nanos).single() {
|
||
if first_ts.is_none() {
|
||
first_ts = Some(ts);
|
||
}
|
||
last_ts = Some(ts);
|
||
}
|
||
}
|
||
}
|
||
|
||
match (first_ts, last_ts) {
|
||
(Some(first), Some(last)) => Ok(Some((first, last))),
|
||
_ => Ok(None),
|
||
}
|
||
}
|
||
|
||
/// Internal method to load a single DBN file
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `file_path` - Path to DBN file
|
||
/// * `symbol` - Symbol name for the bars
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Vector of MarketData bars from the file
|
||
async fn load_file(&self, file_path: &str, symbol: &str) -> Result<Vec<MarketData>> {
|
||
let start = Instant::now();
|
||
|
||
// Validate file extension
|
||
if !is_valid_dbn_file(file_path) {
|
||
warn!(
|
||
"Attempting to load non-standard DBN file: {} (compressed or invalid extension)",
|
||
file_path
|
||
);
|
||
}
|
||
|
||
// Check file exists
|
||
if !Path::new(file_path).exists() {
|
||
return Err(anyhow::anyhow!("DBN file not found: {}", file_path));
|
||
}
|
||
|
||
// Use official dbn crate decoder (handles headers, metadata, and messages)
|
||
let mut decoder = DbnDecoder::from_file(file_path).context(format!(
|
||
"Failed to create DBN decoder for file: {}",
|
||
file_path
|
||
))?;
|
||
|
||
// Enable version upgrades for compatibility with different DBN versions
|
||
decoder
|
||
.set_upgrade_policy(VersionUpgradePolicy::UpgradeToV3)
|
||
.context("Failed to set upgrade policy")?;
|
||
|
||
let mut bars = Vec::new();
|
||
let mut prev_close: Option<f64> = None;
|
||
let mut corrections_applied = 0;
|
||
|
||
// Decode all records from the DBN file
|
||
while let Some(record_ref) = decoder
|
||
.decode_record_ref()
|
||
.context("Failed to decode DBN record")?
|
||
{
|
||
// Try to extract OHLCV message from the record (returns Option)
|
||
if let Some(ohlcv) = record_ref.get::<OhlcvMsg>() {
|
||
// Convert timestamp from nanoseconds to DateTime<Utc>
|
||
let ts_nanos = ohlcv.hd.ts_event as i64;
|
||
let secs = ts_nanos / 1_000_000_000;
|
||
let nanos = (ts_nanos % 1_000_000_000) as u32;
|
||
let timestamp = Utc
|
||
.timestamp_opt(secs, nanos)
|
||
.single()
|
||
.ok_or_else(|| anyhow::anyhow!("Invalid timestamp: {}", ts_nanos))?;
|
||
|
||
// Convert DBN fixed-point prices to f64 with context-aware correction
|
||
let mut open_f64 = dbn_price_to_f64(ohlcv.open);
|
||
let mut high_f64 = dbn_price_to_f64(ohlcv.high);
|
||
let mut low_f64 = dbn_price_to_f64(ohlcv.low);
|
||
let mut close_f64 = dbn_price_to_f64(ohlcv.close);
|
||
|
||
// Price anomaly detection and correction
|
||
// GLBX.MDP3 ES.FUT data has some bars encoded with 7 decimal places instead of 9
|
||
if let Some(prev) = prev_close {
|
||
let pct_change = ((close_f64 - prev) / prev).abs();
|
||
|
||
// If >50% drop AND close < $1000, likely 100x encoding issue
|
||
if pct_change > 0.5 && close_f64 < 1000.0 {
|
||
// Try 100x correction
|
||
let corrected_close = close_f64 * 100.0;
|
||
|
||
// Verify corrected price is reasonable for ES.FUT ($3,000-$6,000 range)
|
||
if (3000.0..=6000.0).contains(&corrected_close) {
|
||
// Apply correction to all OHLCV prices
|
||
open_f64 *= 100.0;
|
||
high_f64 *= 100.0;
|
||
low_f64 *= 100.0;
|
||
close_f64 = corrected_close;
|
||
corrections_applied += 1;
|
||
|
||
if corrections_applied <= 5 {
|
||
debug!(
|
||
"Applied 100x price correction at bar {} ({}% change, ${:.2} -> ${:.2})",
|
||
bars.len() + 1,
|
||
pct_change * 100.0,
|
||
close_f64 / 100.0,
|
||
close_f64
|
||
);
|
||
}
|
||
} else {
|
||
// Correction results in unrealistic price - skip this bar entirely
|
||
warn!(
|
||
"Skipping corrupted bar at index {} (timestamp: {}): price ${:.2} (corrected: ${:.2}) outside valid ES.FUT range",
|
||
bars.len() + 1,
|
||
timestamp,
|
||
close_f64,
|
||
corrected_close
|
||
);
|
||
prev_close = Some(prev); // Keep previous close unchanged
|
||
continue; // Skip this bar
|
||
}
|
||
}
|
||
}
|
||
|
||
prev_close = Some(close_f64);
|
||
|
||
// Convert to Decimal
|
||
let open_decimal = Decimal::from_f64_retain(open_f64)
|
||
.ok_or_else(|| anyhow::anyhow!("Invalid open price: {}", ohlcv.open))?;
|
||
let high_decimal = Decimal::from_f64_retain(high_f64)
|
||
.ok_or_else(|| anyhow::anyhow!("Invalid high price: {}", ohlcv.high))?;
|
||
let low_decimal = Decimal::from_f64_retain(low_f64)
|
||
.ok_or_else(|| anyhow::anyhow!("Invalid low price: {}", ohlcv.low))?;
|
||
let close_decimal = Decimal::from_f64_retain(close_f64)
|
||
.ok_or_else(|| anyhow::anyhow!("Invalid close price: {}", ohlcv.close))?;
|
||
|
||
// Create MarketData bar
|
||
let market_data = MarketData {
|
||
symbol: symbol.to_string(),
|
||
timestamp,
|
||
open: open_decimal,
|
||
high: high_decimal,
|
||
low: low_decimal,
|
||
close: close_decimal,
|
||
volume: Decimal::from(ohlcv.volume),
|
||
};
|
||
|
||
bars.push(market_data);
|
||
}
|
||
}
|
||
|
||
if corrections_applied > 0 {
|
||
debug!(
|
||
"Applied {} automatic price corrections in file {}",
|
||
corrections_applied, file_path
|
||
);
|
||
}
|
||
|
||
let duration = start.elapsed();
|
||
debug!(
|
||
"Loaded {} bars from {} in {:.2}ms",
|
||
bars.len(),
|
||
Path::new(file_path)
|
||
.file_name()
|
||
.and_then(|n| n.to_str())
|
||
.unwrap_or(file_path),
|
||
duration.as_secs_f64() * 1000.0
|
||
);
|
||
|
||
// Performance validation
|
||
if duration.as_millis() > 10 && bars.len() > 100 {
|
||
warn!(
|
||
"DBN loading took {}ms for {} bars (target: <10ms) - file: {}",
|
||
duration.as_millis(),
|
||
bars.len(),
|
||
file_path
|
||
);
|
||
}
|
||
|
||
Ok(bars)
|
||
}
|
||
|
||
/// Scan directory for valid DBN files
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `dir` - Directory to scan
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Map of symbol -> list of file paths
|
||
///
|
||
/// # File Filtering
|
||
///
|
||
/// - Only includes files ending in `.dbn` (case-insensitive)
|
||
/// - Skips compressed files (`.dbn.zst`, `.dbn.gz`, `.dbn.bz2`, `.dbn.xz`)
|
||
/// - Skips temporary files (`.dbn.tmp`, `.dbn.swp`)
|
||
/// - Skips backup files (`.dbn.old`, `.dbn.backup`)
|
||
async fn scan_directory_for_dbn_files(
|
||
dir: &std::path::Path,
|
||
) -> Result<HashMap<String, Vec<String>>> {
|
||
let mut file_mapping: HashMap<String, Vec<String>> = HashMap::new();
|
||
let mut skipped_count = 0;
|
||
|
||
for entry in
|
||
fs::read_dir(dir).context(format!("Failed to read directory: {}", dir.display()))?
|
||
{
|
||
let entry = entry.context("Failed to read directory entry")?;
|
||
let path = entry.path();
|
||
|
||
// Only process files (not directories)
|
||
if !path.is_file() {
|
||
continue;
|
||
}
|
||
|
||
// Get path as string
|
||
let path_str = match path.to_str() {
|
||
Some(s) => s,
|
||
None => continue,
|
||
};
|
||
|
||
// Check if valid DBN file
|
||
if !is_valid_dbn_file(path_str) {
|
||
skipped_count += 1;
|
||
continue;
|
||
}
|
||
|
||
// Extract symbol from filename (e.g., "ES.FUT_2024-01-02.dbn" -> "ES.FUT")
|
||
if let Some(file_name) = path.file_name().and_then(|n| n.to_str()) {
|
||
let symbol = file_name
|
||
.split('_')
|
||
.next()
|
||
.unwrap_or(file_name)
|
||
.trim_end_matches(".dbn");
|
||
file_mapping
|
||
.entry(symbol.to_string())
|
||
.or_default()
|
||
.push(path_str.to_string());
|
||
}
|
||
}
|
||
|
||
debug!(
|
||
"Scanned directory: {} valid DBN files, {} skipped",
|
||
file_mapping.values().map(|v| v.len()).sum::<usize>(),
|
||
skipped_count
|
||
);
|
||
|
||
Ok(file_mapping)
|
||
}
|
||
|
||
/// Load OHLCV bars for multiple symbols (first file only per symbol)
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `symbols` - List of symbols to load
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Combined vector of MarketData bars from all symbols, sorted by timestamp
|
||
pub async fn load_multi_symbol_bars(&self, symbols: &[String]) -> Result<Vec<MarketData>> {
|
||
let mut all_bars = Vec::new();
|
||
|
||
for symbol in symbols {
|
||
let bars = self.load_ohlcv_bars(symbol).await?;
|
||
all_bars.extend(bars);
|
||
}
|
||
|
||
// Sort by timestamp across all symbols
|
||
all_bars.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
|
||
|
||
Ok(all_bars)
|
||
}
|
||
|
||
/// Load OHLCV bars for multiple symbols with all files
|
||
///
|
||
/// Loads all files for each symbol and merges chronologically.
|
||
///
|
||
/// # Arguments
|
||
///
|
||
/// * `symbols` - List of symbols to load
|
||
///
|
||
/// # Returns
|
||
///
|
||
/// Combined vector of MarketData bars from all symbols and files, sorted by timestamp
|
||
pub async fn load_multi_symbol_bars_all(&self, symbols: &[String]) -> Result<Vec<MarketData>> {
|
||
let mut all_bars = Vec::new();
|
||
|
||
for symbol in symbols {
|
||
let bars = self.load_ohlcv_bars_all(symbol).await?;
|
||
all_bars.extend(bars);
|
||
}
|
||
|
||
// Sort by timestamp across all symbols
|
||
all_bars.sort_by(|a, b| a.timestamp.cmp(&b.timestamp));
|
||
|
||
info!(
|
||
"Loaded {} total bars from {} symbols",
|
||
all_bars.len(),
|
||
symbols.len()
|
||
);
|
||
|
||
Ok(all_bars)
|
||
}
|
||
|
||
/// Check if data is available for symbol and time range
|
||
///
|
||
/// This is a lightweight check that doesn't load the full file.
|
||
pub async fn check_data_availability(
|
||
&self,
|
||
symbol: &str,
|
||
_start_time: DateTime<Utc>,
|
||
_end_time: DateTime<Utc>,
|
||
) -> Result<bool> {
|
||
// Check if files exist
|
||
let entries = match self.file_mapping.get(symbol) {
|
||
Some(e) => e,
|
||
None => return Ok(false),
|
||
};
|
||
|
||
// Check at least one file exists
|
||
for entry in entries {
|
||
if Path::new(&entry.path).exists() {
|
||
return Ok(true);
|
||
}
|
||
}
|
||
|
||
Ok(false)
|
||
}
|
||
|
||
/// Get list of available symbols
|
||
pub fn available_symbols(&self) -> Vec<String> {
|
||
self.file_mapping.keys().cloned().collect()
|
||
}
|
||
|
||
/// Get file paths for symbol
|
||
pub fn get_file_paths(&self, symbol: &str) -> Option<Vec<String>> {
|
||
self.file_mapping
|
||
.get(symbol)
|
||
.map(|entries| entries.iter().map(|e| e.path.clone()).collect())
|
||
}
|
||
|
||
/// Get file path for symbol (first file, backward compatible)
|
||
pub fn get_file_path(&self, symbol: &str) -> Option<String> {
|
||
self.file_mapping
|
||
.get(symbol)
|
||
.and_then(|entries| entries.first().map(|e| e.path.clone()))
|
||
}
|
||
|
||
/// Add or update file mapping for a symbol (single file)
|
||
pub fn add_symbol_mapping(&mut self, symbol: String, file_path: String) {
|
||
self.file_mapping.insert(
|
||
symbol,
|
||
vec![FileEntry {
|
||
path: file_path,
|
||
}],
|
||
);
|
||
}
|
||
|
||
/// Add multiple files for a symbol
|
||
pub fn add_symbol_mapping_multi(&mut self, symbol: String, file_paths: Vec<String>) {
|
||
let entries = file_paths
|
||
.into_iter()
|
||
.map(|path| FileEntry {
|
||
path,
|
||
})
|
||
.collect();
|
||
self.file_mapping.insert(symbol, entries);
|
||
}
|
||
|
||
/// Get number of files configured for a symbol
|
||
pub fn get_file_count(&self, symbol: &str) -> usize {
|
||
self.file_mapping
|
||
.get(symbol)
|
||
.map(|entries| entries.len())
|
||
.unwrap_or(0)
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
|
||
#[tokio::test]
|
||
async fn test_dbn_data_source_creation() {
|
||
let mut file_mapping = HashMap::new();
|
||
file_mapping.insert(
|
||
"ES.FUT".to_string(),
|
||
"test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn".to_string(),
|
||
);
|
||
|
||
let data_source = DbnDataSource::new(file_mapping).await;
|
||
assert!(data_source.is_ok());
|
||
|
||
let source = data_source.unwrap();
|
||
assert_eq!(source.available_symbols().len(), 1);
|
||
assert!(source.available_symbols().contains(&"ES.FUT".to_string()));
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_load_nonexistent_symbol() {
|
||
let file_mapping = HashMap::new();
|
||
let data_source = DbnDataSource::new(file_mapping).await.unwrap();
|
||
|
||
let result = data_source.load_ohlcv_bars("NONEXISTENT").await;
|
||
assert!(result.is_err());
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_symbol_mapping() {
|
||
let mut file_mapping = HashMap::new();
|
||
file_mapping.insert("TEST1".to_string(), "/path/to/test1.dbn".to_string());
|
||
file_mapping.insert("TEST2".to_string(), "/path/to/test2.dbn".to_string());
|
||
|
||
let data_source = DbnDataSource::new(file_mapping).await.unwrap();
|
||
|
||
let symbols = data_source.available_symbols();
|
||
assert_eq!(symbols.len(), 2);
|
||
assert!(symbols.contains(&"TEST1".to_string()));
|
||
assert!(symbols.contains(&"TEST2".to_string()));
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_load_real_dbn_file() {
|
||
let mut file_mapping = HashMap::new();
|
||
|
||
// Get absolute path to test file (workspace root + relative path)
|
||
let current_dir = std::env::current_dir().expect("INVARIANT: Current directory should be accessible");
|
||
let workspace_root = current_dir
|
||
.ancestors()
|
||
.find(|p| p.join("Cargo.toml").exists() && p.join("test_data").exists())
|
||
.expect("Could not find workspace root");
|
||
let test_file =
|
||
workspace_root.join("test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn");
|
||
|
||
if !test_file.exists() {
|
||
eprintln!(
|
||
"Test file not found, skipping test: {}",
|
||
test_file.display()
|
||
);
|
||
return;
|
||
}
|
||
|
||
file_mapping.insert(
|
||
"ES.FUT".to_string(),
|
||
test_file.to_string_lossy().to_string(),
|
||
);
|
||
|
||
let data_source = DbnDataSource::new(file_mapping).await.unwrap();
|
||
let bars = data_source.load_ohlcv_bars("ES.FUT").await;
|
||
|
||
println!("Result: {:?}", bars.as_ref().map(|b| b.len()));
|
||
|
||
// Check that we got bars
|
||
assert!(bars.is_ok(), "Failed to load bars: {:?}", bars.err());
|
||
let bars = bars.unwrap();
|
||
|
||
println!("Loaded {} bars", bars.len());
|
||
|
||
// ES.FUT should have a reasonable number of bars (1-minute data)
|
||
assert!(bars.len() > 0, "No bars loaded");
|
||
assert!(bars.len() > 100, "Too few bars loaded: {}", bars.len());
|
||
assert!(bars.len() < 10000, "Too many bars loaded: {}", bars.len());
|
||
|
||
// Check first bar
|
||
if let Some(first_bar) = bars.first() {
|
||
println!(
|
||
"First bar: symbol={}, timestamp={}, open={}, high={}, low={}, close={}, volume={}",
|
||
first_bar.symbol,
|
||
first_bar.timestamp,
|
||
first_bar.open,
|
||
first_bar.high,
|
||
first_bar.low,
|
||
first_bar.close,
|
||
first_bar.volume
|
||
);
|
||
|
||
assert_eq!(first_bar.symbol, "ES.FUT");
|
||
|
||
// ES.FUT prices should be in reasonable range (4000-5000)
|
||
let close_f64 = {
|
||
use num_traits::ToPrimitive;
|
||
first_bar.close.to_f64().unwrap_or(0.0)
|
||
};
|
||
assert!(
|
||
close_f64 > 4000.0 && close_f64 < 5000.0,
|
||
"Unexpected ES.FUT price: {}",
|
||
close_f64
|
||
);
|
||
|
||
// OHLCV relationship check
|
||
assert!(first_bar.low <= first_bar.open, "low > open");
|
||
assert!(first_bar.low <= first_bar.close, "low > close");
|
||
assert!(first_bar.high >= first_bar.open, "high < open");
|
||
assert!(first_bar.high >= first_bar.close, "high < close");
|
||
|
||
// Volume should be positive
|
||
assert!(first_bar.volume > Decimal::ZERO, "volume <= 0");
|
||
}
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_peek_file_timestamps_nonexistent_file() {
|
||
let result = DbnDataSource::peek_file_timestamps("/tmp/nonexistent_file.dbn");
|
||
assert!(result.is_err());
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_peek_file_timestamps_real_file() {
|
||
let current_dir =
|
||
std::env::current_dir().expect("INVARIANT: Current directory should be accessible");
|
||
let workspace_root = current_dir
|
||
.ancestors()
|
||
.find(|p| p.join("Cargo.toml").exists() && p.join("test_data").exists());
|
||
|
||
let workspace_root = match workspace_root {
|
||
Some(r) => r,
|
||
None => {
|
||
eprintln!("Could not find workspace root, skipping test");
|
||
return;
|
||
}
|
||
};
|
||
|
||
let test_file =
|
||
workspace_root.join("test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn");
|
||
|
||
if !test_file.exists() {
|
||
eprintln!("Test file not found, skipping test");
|
||
return;
|
||
}
|
||
|
||
let result =
|
||
DbnDataSource::peek_file_timestamps(&test_file.to_string_lossy());
|
||
assert!(result.is_ok(), "peek failed: {:?}", result.err());
|
||
|
||
let timestamps = result.unwrap();
|
||
assert!(timestamps.is_some(), "No OHLCV records found");
|
||
|
||
let (first_ts, last_ts) = timestamps.unwrap();
|
||
assert!(first_ts <= last_ts, "first_ts > last_ts");
|
||
|
||
// For a 2024-01-02 file, timestamps should be in 2024
|
||
assert_eq!(first_ts.format("%Y").to_string(), "2024");
|
||
}
|
||
|
||
#[tokio::test]
|
||
async fn test_range_filtering_skips_out_of_range_files() {
|
||
// Create a data source with a nonexistent file path.
|
||
// If range filtering correctly peeks and sees the file doesn't match,
|
||
// it should skip it (but since the file doesn't exist, peek will fail
|
||
// and fall through to load_file which will error).
|
||
// Instead, test with a real file and a range that doesn't overlap.
|
||
let current_dir =
|
||
std::env::current_dir().expect("INVARIANT: Current directory should be accessible");
|
||
let workspace_root = current_dir
|
||
.ancestors()
|
||
.find(|p| p.join("Cargo.toml").exists() && p.join("test_data").exists());
|
||
|
||
let workspace_root = match workspace_root {
|
||
Some(r) => r,
|
||
None => {
|
||
eprintln!("Could not find workspace root, skipping test");
|
||
return;
|
||
}
|
||
};
|
||
|
||
let test_file =
|
||
workspace_root.join("test_data/real/databento/ES.FUT_ohlcv-1m_2024-01-02.dbn");
|
||
|
||
if !test_file.exists() {
|
||
eprintln!("Test file not found, skipping test");
|
||
return;
|
||
}
|
||
|
||
let mut file_mapping = HashMap::new();
|
||
file_mapping.insert(
|
||
"ES.FUT".to_string(),
|
||
test_file.to_string_lossy().to_string(),
|
||
);
|
||
|
||
let data_source = DbnDataSource::new(file_mapping).await.unwrap();
|
||
|
||
// Query a range far in the future -- the file should be skipped
|
||
let bars = data_source
|
||
.load_ohlcv_bars_range(
|
||
"ES.FUT",
|
||
Utc.with_ymd_and_hms(2025, 6, 1, 0, 0, 0)
|
||
.single()
|
||
.unwrap_or_default(),
|
||
Utc.with_ymd_and_hms(2025, 6, 30, 0, 0, 0)
|
||
.single()
|
||
.unwrap_or_default(),
|
||
)
|
||
.await;
|
||
assert!(bars.is_ok());
|
||
assert!(
|
||
bars.unwrap().is_empty(),
|
||
"Expected no bars for out-of-range query"
|
||
);
|
||
}
|
||
}
|