9c3d741a08
Move 17 library crates into crates/, CLI binary into bin/fxt, consolidate 10 test crates into testing/, split config crate from deployment config files. Root directory reduced from 38+ to ~17 directories. All Cargo.toml paths and build.rs proto refs updated. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
483 lines
17 KiB
Rust
483 lines
17 KiB
Rust
//! Databento DBN File Validation Tool
|
|
//!
|
|
//! Validates integrity of downloaded Databento datasets for Wave 12 ML training.
|
|
//! Performs comprehensive checks on file integrity, bar counts, timestamps, and price/volume sanity.
|
|
//!
|
|
//! ## Usage
|
|
//! ```bash
|
|
//! cargo run -p ml --example validate_databento_files --release
|
|
//! ```
|
|
|
|
use anyhow::{Context, Result};
|
|
use dbn::decode::{DbnDecoder, DbnMetadata, DecodeRecordRef};
|
|
use dbn::OhlcvMsg;
|
|
use serde::{Deserialize, Serialize};
|
|
use std::path::{Path, PathBuf};
|
|
use std::time::Instant;
|
|
|
|
/// Expected bar counts for 180-day datasets (from download plan)
|
|
const EXPECTED_BARS_ES_180D: u64 = 1_330_000;
|
|
const EXPECTED_BARS_NQ_180D: u64 = 1_330_000;
|
|
const EXPECTED_BARS_6E_180D: u64 = 1_100_000;
|
|
const EXPECTED_BARS_ZN_180D: u64 = 900_000;
|
|
|
|
/// Acceptable tolerance for bar counts (±10%)
|
|
const BAR_COUNT_TOLERANCE: f64 = 0.10;
|
|
|
|
/// Maximum acceptable gap rate
|
|
const MAX_GAP_RATE: f64 = 0.001; // 0.1%
|
|
|
|
/// Expected time between 1-minute bars (in nanoseconds)
|
|
const ONE_MINUTE_NS: i64 = 60_000_000_000;
|
|
|
|
/// Maximum acceptable gap between bars (5 minutes in nanoseconds)
|
|
const MAX_GAP_NS: i64 = 5 * ONE_MINUTE_NS;
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
struct ValidationResult {
|
|
symbol: String,
|
|
file_path: String,
|
|
file_exists: bool,
|
|
file_size_bytes: u64,
|
|
bar_count: u64,
|
|
expected_bar_count: u64,
|
|
bar_count_within_tolerance: bool,
|
|
bar_count_deviation_pct: f64,
|
|
timestamp_errors: u64,
|
|
price_sanity_errors: u64,
|
|
volume_sanity_errors: u64,
|
|
gap_count: u64,
|
|
gap_rate_pct: f64,
|
|
gap_rate_acceptable: bool,
|
|
validation_passed: bool,
|
|
validation_time_ms: u64,
|
|
error_message: Option<String>,
|
|
}
|
|
|
|
#[derive(Debug, Clone, Serialize, Deserialize)]
|
|
struct DataValidationReport {
|
|
total_symbols: usize,
|
|
passed: usize,
|
|
failed: usize,
|
|
total_bars: u64,
|
|
total_validation_time_ms: u64,
|
|
results: Vec<ValidationResult>,
|
|
}
|
|
|
|
/// Validate a single DBN file
|
|
fn validate_dbn_file(
|
|
file_path: &Path,
|
|
symbol: &str,
|
|
expected_bars: u64,
|
|
) -> Result<ValidationResult> {
|
|
let start_time = Instant::now();
|
|
|
|
let mut result = ValidationResult {
|
|
symbol: symbol.to_string(),
|
|
file_path: file_path.display().to_string(),
|
|
file_exists: file_path.exists(),
|
|
file_size_bytes: 0,
|
|
bar_count: 0,
|
|
expected_bar_count: expected_bars,
|
|
bar_count_within_tolerance: false,
|
|
bar_count_deviation_pct: 0.0,
|
|
timestamp_errors: 0,
|
|
price_sanity_errors: 0,
|
|
volume_sanity_errors: 0,
|
|
gap_count: 0,
|
|
gap_rate_pct: 0.0,
|
|
gap_rate_acceptable: false,
|
|
validation_passed: false,
|
|
validation_time_ms: 0,
|
|
error_message: None,
|
|
};
|
|
|
|
if !file_path.exists() {
|
|
result.error_message = Some(format!("File does not exist: {}", file_path.display()));
|
|
result.validation_time_ms = start_time.elapsed().as_millis() as u64;
|
|
return Ok(result);
|
|
}
|
|
|
|
// Get file size
|
|
let file_metadata = std::fs::metadata(file_path)?;
|
|
result.file_size_bytes = file_metadata.len();
|
|
|
|
// Open and decode DBN file
|
|
let mut decoder = DbnDecoder::from_file(file_path)
|
|
.with_context(|| format!("Failed to create DBN decoder for: {}", file_path.display()))?;
|
|
|
|
// Read metadata
|
|
let metadata = decoder.metadata().clone();
|
|
|
|
println!(" Schema: {:?}", metadata.schema);
|
|
println!(" Dataset: {}", metadata.dataset);
|
|
println!(" Start: {}", metadata.start);
|
|
if let Some(end) = metadata.end {
|
|
println!(" End: {}", end);
|
|
}
|
|
|
|
// Validate messages
|
|
let mut prev_timestamp: Option<i64> = None;
|
|
let mut bar_count: u64 = 0;
|
|
|
|
// Decode records
|
|
while let Some(record_ref) = decoder
|
|
.decode_record_ref()
|
|
.with_context(|| format!("Failed to decode DBN record at bar {}", bar_count))?
|
|
{
|
|
if let Some(ohlcv) = record_ref.get::<OhlcvMsg>() {
|
|
bar_count += 1;
|
|
|
|
let ts_event = ohlcv.hd.ts_event as i64;
|
|
|
|
// Check timestamp ordering
|
|
if let Some(prev_ts) = prev_timestamp {
|
|
if ts_event <= prev_ts {
|
|
result.timestamp_errors += 1;
|
|
}
|
|
|
|
// Check for gaps
|
|
let gap = ts_event - prev_ts;
|
|
if gap > MAX_GAP_NS {
|
|
result.gap_count += 1;
|
|
}
|
|
}
|
|
prev_timestamp = Some(ts_event);
|
|
|
|
// Price sanity checks (DBN uses 9 decimal places)
|
|
let open = ohlcv.open as f64 / 1_000_000_000.0;
|
|
let high = ohlcv.high as f64 / 1_000_000_000.0;
|
|
let low = ohlcv.low as f64 / 1_000_000_000.0;
|
|
let close = ohlcv.close as f64 / 1_000_000_000.0;
|
|
|
|
if open <= 0.0 || high <= 0.0 || low <= 0.0 || close <= 0.0 {
|
|
result.price_sanity_errors += 1;
|
|
}
|
|
|
|
if high < low {
|
|
result.price_sanity_errors += 1;
|
|
}
|
|
|
|
if open < low || open > high || close < low || close > high {
|
|
result.price_sanity_errors += 1;
|
|
}
|
|
|
|
// Volume sanity check
|
|
if ohlcv.volume < 0 {
|
|
result.volume_sanity_errors += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
result.bar_count = bar_count;
|
|
|
|
// Calculate bar count deviation
|
|
if expected_bars > 0 {
|
|
let deviation = (bar_count as f64 - expected_bars as f64).abs() / expected_bars as f64;
|
|
result.bar_count_deviation_pct = deviation * 100.0;
|
|
result.bar_count_within_tolerance = deviation <= BAR_COUNT_TOLERANCE;
|
|
} else {
|
|
// No expected count provided, assume tolerance passed
|
|
result.bar_count_within_tolerance = true;
|
|
}
|
|
|
|
// Calculate gap rate
|
|
if bar_count > 0 {
|
|
result.gap_rate_pct = (result.gap_count as f64 / bar_count as f64) * 100.0;
|
|
result.gap_rate_acceptable = result.gap_rate_pct <= (MAX_GAP_RATE * 100.0);
|
|
}
|
|
|
|
// Overall validation pass/fail
|
|
result.validation_passed = result.file_exists
|
|
&& result.bar_count > 0
|
|
&& result.bar_count_within_tolerance
|
|
&& result.timestamp_errors == 0
|
|
&& result.price_sanity_errors == 0
|
|
&& result.volume_sanity_errors == 0
|
|
&& result.gap_rate_acceptable;
|
|
|
|
result.validation_time_ms = start_time.elapsed().as_millis() as u64;
|
|
|
|
Ok(result)
|
|
}
|
|
|
|
#[tokio::main]
|
|
async fn main() -> Result<()> {
|
|
tracing_subscriber::fmt::init();
|
|
|
|
println!("═══════════════════════════════════════════════════════════════");
|
|
println!(" Databento Data Validation - Wave 12 Agent W12-06");
|
|
println!("═══════════════════════════════════════════════════════════════\n");
|
|
|
|
let start_time = Instant::now();
|
|
|
|
// Define expected files (180-day datasets)
|
|
let expected_files = vec![
|
|
("ES.FUT", "test_data/ES_FUT_180d.dbn", EXPECTED_BARS_ES_180D),
|
|
("NQ.FUT", "test_data/NQ_FUT_180d.dbn", EXPECTED_BARS_NQ_180D),
|
|
("6E.FUT", "test_data/6E_FUT_180d.dbn", EXPECTED_BARS_6E_180D),
|
|
("ZN.FUT", "test_data/ZN_FUT_180d.dbn", EXPECTED_BARS_ZN_180D),
|
|
];
|
|
|
|
let mut results = Vec::new();
|
|
let mut total_bars = 0u64;
|
|
|
|
for (symbol, file_path, expected_bars) in expected_files {
|
|
println!("Validating {} ({})...", symbol, file_path);
|
|
|
|
let path = PathBuf::from(file_path);
|
|
match validate_dbn_file(&path, symbol, expected_bars) {
|
|
Ok(result) => {
|
|
if result.validation_passed {
|
|
println!(
|
|
" ✅ PASS - {} bars ({} MB)",
|
|
result.bar_count,
|
|
result.file_size_bytes / 1_000_000
|
|
);
|
|
} else {
|
|
println!(
|
|
" ❌ FAIL - {}",
|
|
result
|
|
.error_message
|
|
.as_ref()
|
|
.unwrap_or(&"Validation failed".to_string())
|
|
);
|
|
if !result.file_exists {
|
|
println!(" File does not exist (may not be downloaded yet)");
|
|
}
|
|
if !result.bar_count_within_tolerance && result.bar_count > 0 {
|
|
println!(
|
|
" Bar count deviation: {:.2}%",
|
|
result.bar_count_deviation_pct
|
|
);
|
|
}
|
|
if result.timestamp_errors > 0 {
|
|
println!(" Timestamp errors: {}", result.timestamp_errors);
|
|
}
|
|
if result.price_sanity_errors > 0 {
|
|
println!(" Price sanity errors: {}", result.price_sanity_errors);
|
|
}
|
|
if result.volume_sanity_errors > 0 {
|
|
println!(" Volume sanity errors: {}", result.volume_sanity_errors);
|
|
}
|
|
if !result.gap_rate_acceptable && result.bar_count > 0 {
|
|
println!(
|
|
" Gap rate: {:.3}% ({} gaps)",
|
|
result.gap_rate_pct, result.gap_count
|
|
);
|
|
}
|
|
}
|
|
|
|
total_bars += result.bar_count;
|
|
results.push(result);
|
|
},
|
|
Err(e) => {
|
|
println!(" ❌ ERROR: {}", e);
|
|
results.push(ValidationResult {
|
|
symbol: symbol.to_string(),
|
|
file_path: file_path.to_string(),
|
|
file_exists: false,
|
|
file_size_bytes: 0,
|
|
bar_count: 0,
|
|
expected_bar_count: expected_bars,
|
|
bar_count_within_tolerance: false,
|
|
bar_count_deviation_pct: 0.0,
|
|
timestamp_errors: 0,
|
|
price_sanity_errors: 0,
|
|
volume_sanity_errors: 0,
|
|
gap_count: 0,
|
|
gap_rate_pct: 0.0,
|
|
gap_rate_acceptable: false,
|
|
validation_passed: false,
|
|
validation_time_ms: 0,
|
|
error_message: Some(e.to_string()),
|
|
});
|
|
},
|
|
}
|
|
println!();
|
|
}
|
|
|
|
let total_validation_time_ms = start_time.elapsed().as_millis() as u64;
|
|
|
|
// Generate summary report
|
|
let passed = results.iter().filter(|r| r.validation_passed).count();
|
|
let failed = results.len() - passed;
|
|
|
|
let report = DataValidationReport {
|
|
total_symbols: results.len(),
|
|
passed,
|
|
failed,
|
|
total_bars,
|
|
total_validation_time_ms,
|
|
results,
|
|
};
|
|
|
|
println!("═══════════════════════════════════════════════════════════════");
|
|
println!(" VALIDATION SUMMARY");
|
|
println!("═══════════════════════════════════════════════════════════════");
|
|
println!(" Total Symbols: {}", report.total_symbols);
|
|
println!(" Passed: {} ✅", report.passed);
|
|
println!(" Failed: {} ❌", report.failed);
|
|
println!(" Total Bars: {}", report.total_bars);
|
|
println!(" Validation Time: {} ms", report.total_validation_time_ms);
|
|
println!("═══════════════════════════════════════════════════════════════\n");
|
|
|
|
// Save validation report
|
|
let report_json = serde_json::to_string_pretty(&report)?;
|
|
std::fs::write("/tmp/databento_validation_report.json", &report_json)?;
|
|
println!("✅ Validation report saved to: /tmp/databento_validation_report.json");
|
|
|
|
// Save bar counts
|
|
let bar_counts: serde_json::Value = serde_json::json!({
|
|
"symbols": report.results.iter().map(|r| {
|
|
serde_json::json!({
|
|
"symbol": r.symbol,
|
|
"bar_count": r.bar_count,
|
|
"expected": r.expected_bar_count,
|
|
"deviation_pct": r.bar_count_deviation_pct,
|
|
})
|
|
}).collect::<Vec<_>>(),
|
|
"total_bars": report.total_bars,
|
|
});
|
|
let bar_counts_json = serde_json::to_string_pretty(&bar_counts)?;
|
|
std::fs::write("/tmp/databento_bar_counts.json", &bar_counts_json)?;
|
|
println!("✅ Bar counts saved to: /tmp/databento_bar_counts.json");
|
|
|
|
// Generate markdown report
|
|
let mut markdown = String::new();
|
|
markdown.push_str("# Databento Data Validation Report - Wave 12 Agent W12-06\n\n");
|
|
markdown.push_str(&format!(
|
|
"**Date**: {}\n",
|
|
chrono::Utc::now().format("%Y-%m-%d %H:%M:%S UTC")
|
|
));
|
|
markdown.push_str(&format!(
|
|
"**Validation Time**: {} ms\n\n",
|
|
report.total_validation_time_ms
|
|
));
|
|
markdown.push_str("---\n\n");
|
|
markdown.push_str("## Executive Summary\n\n");
|
|
markdown.push_str(&format!("- **Total Symbols**: {}\n", report.total_symbols));
|
|
markdown.push_str(&format!("- **Passed**: {} ✅\n", report.passed));
|
|
markdown.push_str(&format!("- **Failed**: {} ❌\n", report.failed));
|
|
markdown.push_str(&format!("- **Total Bars**: {}\n", report.total_bars));
|
|
markdown.push_str(&format!(
|
|
"- **Overall Status**: {}\n\n",
|
|
if report.failed == 0 {
|
|
"✅ ALL PASSED"
|
|
} else {
|
|
"❌ VALIDATION FAILURES"
|
|
}
|
|
));
|
|
markdown.push_str("---\n\n");
|
|
markdown.push_str("## Validation Results by Symbol\n\n");
|
|
|
|
for result in &report.results {
|
|
markdown.push_str(&format!(
|
|
"### {} - {}\n\n",
|
|
result.symbol,
|
|
if result.validation_passed {
|
|
"✅ PASS"
|
|
} else {
|
|
"❌ FAIL"
|
|
}
|
|
));
|
|
markdown.push_str(&format!("- **File**: `{}`\n", result.file_path));
|
|
markdown.push_str(&format!(
|
|
"- **File Exists**: {}\n",
|
|
if result.file_exists {
|
|
"✅ Yes"
|
|
} else {
|
|
"❌ No"
|
|
}
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **File Size**: {} MB\n",
|
|
result.file_size_bytes / 1_000_000
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Bar Count**: {} (expected: {})\n",
|
|
result.bar_count, result.expected_bar_count
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Bar Count Deviation**: {:.2}%\n",
|
|
result.bar_count_deviation_pct
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Bar Count Within Tolerance**: {}\n",
|
|
if result.bar_count_within_tolerance {
|
|
"✅ Yes"
|
|
} else {
|
|
"❌ No"
|
|
}
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Timestamp Errors**: {}\n",
|
|
result.timestamp_errors
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Price Sanity Errors**: {}\n",
|
|
result.price_sanity_errors
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Volume Sanity Errors**: {}\n",
|
|
result.volume_sanity_errors
|
|
));
|
|
markdown.push_str(&format!("- **Gap Count**: {}\n", result.gap_count));
|
|
markdown.push_str(&format!("- **Gap Rate**: {:.3}%\n", result.gap_rate_pct));
|
|
markdown.push_str(&format!(
|
|
"- **Gap Rate Acceptable**: {}\n",
|
|
if result.gap_rate_acceptable {
|
|
"✅ Yes"
|
|
} else {
|
|
"❌ No"
|
|
}
|
|
));
|
|
markdown.push_str(&format!(
|
|
"- **Validation Time**: {} ms\n",
|
|
result.validation_time_ms
|
|
));
|
|
if let Some(error) = &result.error_message {
|
|
markdown.push_str(&format!("- **Error**: {}\n", error));
|
|
}
|
|
markdown.push_str("\n");
|
|
}
|
|
|
|
markdown.push_str("---\n\n");
|
|
markdown.push_str("## Next Steps\n\n");
|
|
if report.failed > 0 {
|
|
markdown.push_str("❌ **Action Required**: Some validations failed. Please:\n\n");
|
|
for result in &report.results {
|
|
if !result.validation_passed {
|
|
if !result.file_exists {
|
|
markdown.push_str(&format!(
|
|
"- Download {} from Databento (Agent W12-02 to W12-05)\n",
|
|
result.symbol
|
|
));
|
|
} else {
|
|
markdown.push_str(&format!(
|
|
"- Investigate {} validation errors\n",
|
|
result.symbol
|
|
));
|
|
}
|
|
}
|
|
}
|
|
} else {
|
|
markdown.push_str("✅ **All validations passed!** Ready to proceed with:\n\n");
|
|
markdown.push_str("- Wave 13: ML model retraining with 180-day datasets\n");
|
|
markdown.push_str("- Wave 14: Wave Comparison Backtest (Wave C vs Wave D)\n");
|
|
markdown.push_str("- Wave 15: Production deployment\n");
|
|
}
|
|
|
|
std::fs::write("/tmp/databento_validation_report.md", &markdown)?;
|
|
println!("✅ Markdown report saved to: /tmp/databento_validation_report.md\n");
|
|
|
|
// Exit with appropriate status code
|
|
if report.failed > 0 {
|
|
eprintln!("❌ Validation failed for {} symbols", report.failed);
|
|
std::process::exit(1);
|
|
} else {
|
|
println!("✅ All validations passed!");
|
|
Ok(())
|
|
}
|
|
}
|