- Created data/examples/download_ml_training_data.rs using reqwest + Databento HTTP API - Downloaded 90 days × 4 symbols (ES.FUT, NQ.FUT, ZN.FUT, 6E.FUT) - Files saved to test_data/real/databento/ml_training/ - Total: 360 files, 15 MB compressed DBN format - Used existing Rust pattern from download_nq_fut.rs - API key loaded from .env file - 100% success rate (360/360 files) - Ready for ML training benchmarks Next: Create simplified training benchmark for RTX 3050 Ti GPU measurements
9.9 KiB
9.9 KiB
DbnMarketDataRepository - Advanced Query Usage Examples
Overview
The DbnMarketDataRepository provides advanced querying capabilities for DBN (Databento Binary) market data, optimized for complex test scenarios and backtesting operations.
Performance Targets
- <10ms for typical queries (~400 bars)
- Zero-copy parsing with SIMD optimizations
- Efficient filtering and aggregation
Basic Setup
use backtesting_service::dbn_repository::DbnMarketDataRepository;
use std::collections::HashMap;
// Create repository with 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 repo = DbnMarketDataRepository::new(file_mapping).await?;
Advanced Query Methods
1. Time Range Queries
Load data with precise DateTime filtering:
use chrono::{TimeZone, Utc};
let start = Utc.with_ymd_and_hms(2024, 1, 2, 9, 30, 0).unwrap(); // Market open
let end = Utc.with_ymd_and_hms(2024, 1, 2, 16, 0, 0).unwrap(); // Market close
let symbols = vec!["ES.FUT".to_string()];
let bars = repo.load_by_time_range(&symbols, start, end).await?;
println!("Loaded {} bars for market hours", bars.len());
2. Volume Filtering
Filter for high-liquidity bars:
use rust_decimal::Decimal;
// Load only bars with volume >= 100
let min_volume = Decimal::new(100, 0);
let start_time = 1704153600_000_000_000i64;
let end_time = 1704240000_000_000_000i64;
let symbols = vec!["ES.FUT".to_string()];
let high_liquidity_bars = repo.load_with_volume_filter(
&symbols,
min_volume,
start_time,
end_time
).await?;
println!("High liquidity bars: {}", high_liquidity_bars.len());
3. Regime-Specific Sampling
Load data matching specific market regimes:
// Load trending regime samples (>0.5% intrabar range)
let trending_samples = repo.load_regime_samples(
"trending",
20, // limit to 20 samples
&vec!["ES.FUT".to_string()]
).await?;
// Load ranging/sideways regime samples (<0.2% range)
let ranging_samples = repo.load_regime_samples(
"ranging",
20,
&vec!["ES.FUT".to_string()]
).await?;
// Load volatile regime samples (>0.8% range + high volume)
let volatile_samples = repo.load_regime_samples(
"volatile",
10,
&vec!["ES.FUT".to_string()]
).await?;
// Load stable regime samples (<0.15% range)
let stable_samples = repo.load_regime_samples(
"stable",
15,
&vec!["ES.FUT".to_string()]
).await?;
Supported Regime Types:
"trending"- High price movement (>0.5% range)"ranging"/"sideways"- Low volatility (<0.2% range)"volatile"- High volatility + high volume (>0.8% range)"stable"- Very low volatility (<0.15% range)
4. Date Range Discovery
Get available date ranges for symbols:
let (first, last) = repo.get_date_range("ES.FUT").await?;
println!("Data available from {} to {}", first, last);
println!("Days of data: {}", (last - first).num_days());
5. Timeframe Resampling
Aggregate minute bars into larger timeframes:
// Load 1-minute bars
let bars = repo.load_historical_data(&symbols, start_time, end_time).await?;
// Resample to 5-minute bars
let bars_5m = repo.resample_bars(&bars, 5)?;
// Resample to 15-minute bars
let bars_15m = repo.resample_bars(&bars, 15)?;
// Resample to 1-hour bars
let bars_1h = repo.resample_bars(&bars, 60)?;
println!("1m: {} bars", bars.len());
println!("5m: {} bars", bars_5m.len());
println!("15m: {} bars", bars_15m.len());
println!("1h: {} bars", bars_1h.len());
6. Rolling Statistics
Calculate moving window statistics:
let bars = repo.load_historical_data(&symbols, start_time, end_time).await?;
// 20-bar rolling window
let window_size = 20;
let stats = repo.calculate_rolling_stats(&bars, window_size);
for (i, (mean, std_dev, min, max)) in stats.iter().enumerate() {
println!(
"Window {}: mean={:.2}, std={:.2}, range=[{:.2}, {:.2}]",
i, mean, std_dev, min, max
);
}
Returns: Vec<(mean, std_dev, min, max)> for each window
7. Summary Statistics
Generate comprehensive statistics:
let bars = repo.load_historical_data(&symbols, start_time, end_time).await?;
let stats = repo.generate_summary_stats(&bars);
println!("Summary Statistics:");
println!(" Count: {}", stats.get("count").unwrap());
println!(" Mean Close: {:.2}", stats.get("mean_close").unwrap());
println!(" Std Close: {:.2}", stats.get("std_close").unwrap());
println!(" Min Close: {:.2}", stats.get("min_close").unwrap());
println!(" Max Close: {:.2}", stats.get("max_close").unwrap());
println!(" Mean Volume: {:.0}", stats.get("mean_volume").unwrap());
println!(" Total Volume: {:.0}", stats.get("total_volume").unwrap());
Available Statistics:
count- Number of barsmean_close- Average close pricestd_close- Standard deviation of close pricesmin_close- Minimum close pricemax_close- Maximum close pricemean_volume- Average volume per bartotal_volume- Cumulative volume
Complex Test Scenarios
Example 1: Multi-Timeframe Analysis
// Load base data
let symbols = vec!["ES.FUT".to_string()];
let bars_1m = repo.load_historical_data(&symbols, start_time, end_time).await?;
// Create multiple timeframes
let bars_5m = repo.resample_bars(&bars_1m, 5)?;
let bars_15m = repo.resample_bars(&bars_1m, 15)?;
let bars_1h = repo.resample_bars(&bars_1m, 60)?;
// Analyze each timeframe
for (tf_name, bars) in [
("1m", &bars_1m),
("5m", &bars_5m),
("15m", &bars_15m),
("1h", &bars_1h),
] {
let stats = repo.generate_summary_stats(bars);
println!("{}: {} bars, volatility={:.2}%",
tf_name,
bars.len(),
stats.get("std_close").unwrap() / stats.get("mean_close").unwrap() * 100.0
);
}
Example 2: Regime Detection Testing
// Test adaptive strategy across different regimes
for regime in ["trending", "ranging", "volatile", "stable"] {
let samples = repo.load_regime_samples(regime, 50, &symbols).await?;
println!("\nTesting {} regime with {} samples", regime, samples.len());
// Run strategy on regime-specific data
let trades = strategy.backtest(&samples).await?;
println!(" Trades: {}", trades.len());
println!(" Win rate: {:.1}%", calculate_win_rate(&trades));
}
Example 3: Liquidity Analysis
// Compare high vs low liquidity performance
let all_bars = repo.load_historical_data(&symbols, start_time, end_time).await?;
let high_liq = repo.load_with_volume_filter(
&symbols,
Decimal::new(200, 0),
start_time,
end_time
).await?;
let low_liq = all_bars
.into_iter()
.filter(|b| b.volume < Decimal::new(50, 0))
.collect::<Vec<_>>();
println!("High liquidity bars: {} ({:.1}%)",
high_liq.len(),
100.0 * high_liq.len() as f64 / (high_liq.len() + low_liq.len()) as f64
);
// Test strategy on both conditions
let high_liq_pnl = strategy.backtest(&high_liq).await?.total_pnl();
let low_liq_pnl = strategy.backtest(&low_liq).await?.total_pnl();
println!("High liquidity PnL: ${:.2}", high_liq_pnl);
println!("Low liquidity PnL: ${:.2}", low_liq_pnl);
Performance Benchmarks
use std::time::Instant;
let start = Instant::now();
let bars = repo.load_historical_data(&symbols, start_time, end_time).await?;
let duration = start.elapsed();
println!("Performance:");
println!(" Bars loaded: {}", bars.len());
println!(" Time: {:.2}ms", duration.as_secs_f64() * 1000.0);
println!(" Rate: {:.0} bars/ms", bars.len() as f64 / duration.as_millis() as f64);
// Target: <10ms for ~400 bars
assert!(duration.as_millis() < 10, "Performance target missed");
Error Handling
// Invalid regime type
match repo.load_regime_samples("invalid", 10, &symbols).await {
Ok(_) => panic!("Should have failed"),
Err(e) => assert!(e.to_string().contains("Unknown regime type")),
}
// Symbol not found
let result = repo.get_date_range("INVALID").await;
assert!(result.is_err());
// Empty data
let empty_bars = Vec::new();
assert!(repo.resample_bars(&empty_bars, 5)?.is_empty());
assert!(repo.generate_summary_stats(&empty_bars).is_empty());
Best Practices
-
Use Time Range Queries for precise filtering:
// Better: DateTime-based let bars = repo.load_by_time_range(&symbols, start, end).await?; // Avoid: Manual nanosecond conversion let start_nanos = start.timestamp_nanos_opt().unwrap(); -
Cache Resampled Data to avoid redundant computation:
let bars_1m = repo.load_historical_data(&symbols, start, end).await?; let bars_5m = repo.resample_bars(&bars_1m, 5)?; // Cache this result -
Use Volume Filtering for realistic trading scenarios:
// Focus on tradeable liquidity let min_volume = Decimal::new(100, 0); let bars = repo.load_with_volume_filter(&symbols, min_volume, start, end).await?; -
Validate Regime Samples before testing:
let samples = repo.load_regime_samples("trending", 100, &symbols).await?; if samples.len() < 50 { println!("Warning: Insufficient {} regime samples", "trending"); }
Integration with Backtesting Service
use backtesting_service::{
dbn_repository::DbnMarketDataRepository,
strategy_engine::StrategyEngine,
};
// Create repository
let repo = DbnMarketDataRepository::new(file_mapping).await?;
// Load regime-specific data
let volatile_data = repo.load_regime_samples("volatile", 100, &symbols).await?;
// Run backtesting
let engine = StrategyEngine::new(strategy_config);
let results = engine.backtest(&volatile_data).await?;
// Analyze results
println!("Volatile regime performance:");
println!(" Sharpe Ratio: {:.2}", results.sharpe_ratio);
println!(" Max Drawdown: {:.2}%", results.max_drawdown_pct);
See Also
DbnDataSource- Underlying DBN file loaderMarketDataRepositorytrait - Repository interfaceStrategyEngine- Backtesting execution engineTESTING_PLAN.md- Overall ML testing strategy