- 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
325 lines
9.4 KiB
Rust
325 lines
9.4 KiB
Rust
//! Performance tests for DBN data loading
|
|
//!
|
|
//! These tests measure throughput, memory usage, and compare against targets
|
|
//! to ensure the DBN loading infrastructure meets HFT requirements.
|
|
|
|
use anyhow::Result;
|
|
use backtesting_service::dbn_repository::DbnMarketDataRepository;
|
|
use backtesting_service::repositories::MarketDataRepository;
|
|
use rust_decimal::Decimal;
|
|
use std::collections::HashMap;
|
|
use std::time::{Duration, Instant};
|
|
|
|
/// Helper to create a DBN repository with ES.FUT test data
|
|
async fn create_dbn_repository() -> Result<DbnMarketDataRepository> {
|
|
// Find workspace root to get absolute path to test file
|
|
let current_dir = std::env::current_dir()?;
|
|
let workspace_root = current_dir
|
|
.ancestors()
|
|
.find(|p| p.join("Cargo.toml").exists() && p.join("test_data").exists())
|
|
.ok_or_else(|| anyhow::anyhow!("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() {
|
|
anyhow::bail!("Test file not found: {}", test_file.display());
|
|
}
|
|
|
|
let mut file_mapping = HashMap::new();
|
|
file_mapping.insert(
|
|
"ES.FUT".to_string(),
|
|
test_file.to_string_lossy().to_string(),
|
|
);
|
|
|
|
DbnMarketDataRepository::new(file_mapping).await
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_throughput() -> Result<()> {
|
|
let repo = create_dbn_repository().await?;
|
|
|
|
// Load multiple times to get average
|
|
let iterations = 10;
|
|
let mut durations = Vec::new();
|
|
|
|
for _ in 0..iterations {
|
|
let start = Instant::now();
|
|
|
|
let symbols = vec!["ES.FUT".to_string()];
|
|
let data = repo
|
|
.load_historical_data(
|
|
&symbols,
|
|
1704153600_000_000_000i64, // 2024-01-02 00:00:00
|
|
1704240000_000_000_000i64, // 2024-01-03 00:00:00
|
|
)
|
|
.await?;
|
|
|
|
let duration = start.elapsed();
|
|
durations.push(duration);
|
|
|
|
assert!(!data.is_empty(), "Should load data");
|
|
}
|
|
|
|
// Calculate statistics
|
|
let avg_duration = durations.iter().sum::<Duration>() / iterations as u32;
|
|
let min_duration = durations.iter().min().unwrap();
|
|
let max_duration = durations.iter().max().unwrap();
|
|
|
|
let bars_loaded = 390; // ES.FUT has ~390 bars on 2024-01-02
|
|
let bars_per_sec = (bars_loaded as f64 / avg_duration.as_secs_f64()) as u64;
|
|
|
|
println!("\n=== Throughput Analysis ({} iterations) ===", iterations);
|
|
println!(" Min: {:?}", min_duration);
|
|
println!(" Avg: {:?}", avg_duration);
|
|
println!(" Max: {:?}", max_duration);
|
|
println!(" Throughput: {} bars/sec", bars_per_sec);
|
|
|
|
// Targets:
|
|
// - Average: <10ms
|
|
// - Throughput: >10,000 bars/sec
|
|
assert!(
|
|
avg_duration.as_millis() < 10,
|
|
"Average load time should be <10ms, got {:?}",
|
|
avg_duration
|
|
);
|
|
assert!(
|
|
bars_per_sec > 10_000,
|
|
"Throughput should be >10K bars/sec, got {}",
|
|
bars_per_sec
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_load_time_consistency() -> Result<()> {
|
|
let repo = create_dbn_repository().await?;
|
|
|
|
let iterations = 20;
|
|
let mut durations = Vec::new();
|
|
|
|
// Warm up
|
|
for _ in 0..5 {
|
|
let symbols = vec!["ES.FUT".to_string()];
|
|
let _ = repo
|
|
.load_historical_data(
|
|
&symbols,
|
|
1704153600_000_000_000i64,
|
|
1704240000_000_000_000i64,
|
|
)
|
|
.await?;
|
|
}
|
|
|
|
// Measure
|
|
for _ in 0..iterations {
|
|
let start = Instant::now();
|
|
|
|
let symbols = vec!["ES.FUT".to_string()];
|
|
let _ = repo
|
|
.load_historical_data(
|
|
&symbols,
|
|
1704153600_000_000_000i64,
|
|
1704240000_000_000_000i64,
|
|
)
|
|
.await?;
|
|
|
|
durations.push(start.elapsed().as_micros());
|
|
}
|
|
|
|
// Calculate statistics
|
|
let avg: u128 = durations.iter().sum::<u128>() / iterations;
|
|
let variance: u128 = durations
|
|
.iter()
|
|
.map(|d| {
|
|
let diff = (*d as i128) - (avg as i128);
|
|
(diff * diff) as u128
|
|
})
|
|
.sum::<u128>()
|
|
/ iterations;
|
|
let stddev = (variance as f64).sqrt();
|
|
let cv = stddev / (avg as f64); // Coefficient of variation
|
|
|
|
println!("\n=== Load Time Consistency ({} iterations) ===", iterations);
|
|
println!(" Avg: {} μs", avg);
|
|
println!(" StdDev: {:.2} μs", stddev);
|
|
println!(" CV: {:.2}% (lower is better)", cv * 100.0);
|
|
|
|
// Target: CV < 20% (consistent performance)
|
|
assert!(
|
|
cv < 0.20,
|
|
"Coefficient of variation should be <20%, got {:.2}%",
|
|
cv * 100.0
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_partial_day_load() -> Result<()> {
|
|
let repo = create_dbn_repository().await?;
|
|
|
|
// Test loading different time windows
|
|
let test_cases = vec![
|
|
(
|
|
"First hour",
|
|
1704203400_000_000_000i64, // 2024-01-02 09:30:00 ET
|
|
1704207000_000_000_000i64, // 2024-01-02 10:30:00 ET
|
|
),
|
|
(
|
|
"Morning session",
|
|
1704203400_000_000_000i64, // 2024-01-02 09:30:00 ET
|
|
1704218400_000_000_000i64, // 2024-01-02 14:00:00 ET
|
|
),
|
|
(
|
|
"Full day",
|
|
1704153600_000_000_000i64, // 2024-01-02 00:00:00
|
|
1704240000_000_000_000i64, // 2024-01-03 00:00:00
|
|
),
|
|
];
|
|
|
|
println!("\n=== Partial Day Load Performance ===");
|
|
|
|
for (name, start_time, end_time) in test_cases {
|
|
let start = Instant::now();
|
|
|
|
let symbols = vec!["ES.FUT".to_string()];
|
|
let data = repo.load_historical_data(&symbols, start_time, end_time).await?;
|
|
|
|
let duration = start.elapsed();
|
|
let bars_per_sec = (data.len() as f64 / duration.as_secs_f64()) as u64;
|
|
|
|
println!(
|
|
" {:<16} | Bars: {:>4} | Time: {:>7.2}ms | Throughput: {:>8} bars/s",
|
|
name,
|
|
data.len(),
|
|
duration.as_secs_f64() * 1000.0,
|
|
bars_per_sec
|
|
);
|
|
|
|
// All loads should be fast
|
|
assert!(
|
|
duration.as_millis() < 10,
|
|
"{} load took too long: {:?}",
|
|
name,
|
|
duration
|
|
);
|
|
}
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_cold_start_vs_warm() -> Result<()> {
|
|
// Cold start
|
|
let cold_start = Instant::now();
|
|
let repo = create_dbn_repository().await?;
|
|
let cold_init_duration = cold_start.elapsed();
|
|
|
|
let symbols = vec!["ES.FUT".to_string()];
|
|
let cold_load_start = Instant::now();
|
|
let data = repo
|
|
.load_historical_data(
|
|
&symbols,
|
|
1704153600_000_000_000i64,
|
|
1704240000_000_000_000i64,
|
|
)
|
|
.await?;
|
|
let cold_load_duration = cold_load_start.elapsed();
|
|
let bars_loaded = data.len();
|
|
|
|
// Warm loads (reuse repository)
|
|
let mut warm_durations = Vec::new();
|
|
for _ in 0..5 {
|
|
let warm_start = Instant::now();
|
|
let _ = repo
|
|
.load_historical_data(
|
|
&symbols,
|
|
1704153600_000_000_000i64,
|
|
1704240000_000_000_000i64,
|
|
)
|
|
.await?;
|
|
warm_durations.push(warm_start.elapsed());
|
|
}
|
|
|
|
let avg_warm = warm_durations.iter().sum::<Duration>() / warm_durations.len() as u32;
|
|
|
|
println!("\n=== Cold Start vs Warm Performance ===");
|
|
println!(" Repository init: {:?}", cold_init_duration);
|
|
println!(" Cold load: {:?}", cold_load_duration);
|
|
println!(" Warm load (avg): {:?}", avg_warm);
|
|
println!(
|
|
" Warm speedup: {:.2}x",
|
|
cold_load_duration.as_secs_f64() / avg_warm.as_secs_f64()
|
|
);
|
|
println!(" Bars loaded: {}", bars_loaded);
|
|
|
|
// Warm loads should be faster or similar
|
|
assert!(
|
|
avg_warm <= cold_load_duration,
|
|
"Warm loads should not be slower than cold"
|
|
);
|
|
|
|
Ok(())
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn test_data_correctness_at_speed() -> Result<()> {
|
|
let repo = create_dbn_repository().await?;
|
|
|
|
let start = Instant::now();
|
|
let symbols = vec!["ES.FUT".to_string()];
|
|
let data = repo
|
|
.load_historical_data(
|
|
&symbols,
|
|
1704153600_000_000_000i64,
|
|
1704240000_000_000_000i64,
|
|
)
|
|
.await?;
|
|
let duration = start.elapsed();
|
|
|
|
println!("\n=== Data Correctness at Speed ===");
|
|
println!(" Load time: {:?}", duration);
|
|
println!(" Bars loaded: {}", data.len());
|
|
|
|
// Verify data quality
|
|
assert!(!data.is_empty(), "Should load data");
|
|
assert!(data.len() > 100, "Should load substantial data");
|
|
|
|
// Check that bars are properly sorted
|
|
for i in 1..data.len() {
|
|
assert!(
|
|
data[i].timestamp >= data[i - 1].timestamp,
|
|
"Bars should be sorted by timestamp"
|
|
);
|
|
}
|
|
|
|
// Verify all bars have valid data
|
|
for (i, bar) in data.iter().enumerate() {
|
|
assert!(
|
|
bar.open > Decimal::ZERO,
|
|
"Bar {} should have positive open price",
|
|
i
|
|
);
|
|
assert!(
|
|
bar.high >= bar.low,
|
|
"Bar {} high should be >= low",
|
|
i
|
|
);
|
|
assert!(
|
|
bar.high >= bar.open && bar.high >= bar.close,
|
|
"Bar {} high should be highest price",
|
|
i
|
|
);
|
|
assert!(
|
|
bar.low <= bar.open && bar.low <= bar.close,
|
|
"Bar {} low should be lowest price",
|
|
i
|
|
);
|
|
assert!(bar.volume >= Decimal::ZERO, "Bar {} should have non-negative volume", i);
|
|
}
|
|
|
|
println!(" ✓ All {} bars passed validation", data.len());
|
|
|
|
Ok(())
|
|
}
|