foxhunt/services/backtesting_service/DBN_REPOSITORY_USAGE.md

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 bars
• mean_close - Average close price
• std_close - Standard deviation of close prices
• min_close - Minimum close price
• max_close - Maximum close price
• mean_volume - Average volume per bar
• total_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

1. 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();
   

2. 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
   

3. 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?;
   

4. 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 loader
• MarketDataRepository trait - Repository interface
• StrategyEngine - Backtesting execution engine
• TESTING_PLAN.md - Overall ML testing strategy