From d4b707bfa84b03cdc4778087c6fc4e6aad675fce Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 24 Feb 2026 20:57:00 +0100 Subject: [PATCH] =?UTF-8?q?fix(ml):=20repair=20download=5Fl2=5Fdata=20exam?= =?UTF-8?q?ple=20(time=E2=86=92chrono,=20GetRangeToFileParams)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.6 --- ml/examples/download_l2_data.rs | 90 +++++++++++++++++---------------- 1 file changed, 47 insertions(+), 43 deletions(-) diff --git a/ml/examples/download_l2_data.rs b/ml/examples/download_l2_data.rs index cb709d575..3969f9b77 100644 --- a/ml/examples/download_l2_data.rs +++ b/ml/examples/download_l2_data.rs @@ -32,14 +32,12 @@ use anyhow::{Context, Result}; use chrono::Datelike; use chrono::NaiveDate; use clap::Parser; -use databento::historical::timeseries::GetRangeParams; +use databento::historical::timeseries::GetRangeToFileParams; use databento::{historical::DateTimeRange, HistoricalClient}; -use dbn::{Compression, Schema}; +use dbn::{Schema, SType}; use std::env; use std::fs; use std::path::{Path, PathBuf}; -use std::str::FromStr; -use tokio::io::AsyncReadExt; #[derive(Debug, Parser)] #[command( @@ -111,40 +109,53 @@ fn generate_trading_dates(start_date_str: &str, num_days: i64) -> Result u64 { + let ts = date + .and_hms_opt(0, 0, 0) + .map(|dt| dt.and_utc().timestamp()) + .unwrap_or(0); + (ts as u64).saturating_mul(1_000_000_000) +} + +/// Build a `DateTimeRange` from two `NaiveDate`s. +fn date_range(start: NaiveDate, end: NaiveDate) -> Result { + let start_ns = naive_date_to_unix_nanos(start); + let end_ns = naive_date_to_unix_nanos(end); + DateTimeRange::try_from((start_ns, end_ns)).map_err(|e| anyhow::anyhow!("{}", e)) +} + async fn download_symbol_day( client: &mut HistoricalClient, symbol: &str, date: &str, output_dir: &Path, ) -> Result> { - let output_file = output_dir.join(format!("{}_mbp-10_{}.dbn", symbol, date)); + let output_file = output_dir.join(format!("{}_mbp-10_{}.dbn.zst", symbol, date)); // Skip if file already exists if output_file.exists() { let size = fs::metadata(&output_file)?.len(); - // Estimate record count (avg 480 bytes per MBP-10 record) - let estimated_records = size / 480; + // Estimate record count (avg 480 bytes per MBP-10 record, ~70% compression) + let estimated_records = (size as f64 / (480.0 * 0.3)) as u64; return Ok(Some((size, estimated_records))); } - // Parse date range (full trading day UTC) - use time::{Date, PrimitiveDateTime, Time, UtcOffset}; - let date_obj = Date::parse( - date, - &time::format_description::parse("[year]-[month]-[day]")?, - )?; - let start_dt = PrimitiveDateTime::new(date_obj, Time::MIDNIGHT).assume_offset(UtcOffset::UTC); - let end_dt = start_dt + time::Duration::days(1); - let date_time_range: DateTimeRange = (start_dt, end_dt).into(); + // Parse start date and compute next day for the range + let start_date = NaiveDate::parse_from_str(date, "%Y-%m-%d") + .context("Failed to parse date for download")?; + let end_date = start_date.succ_opt().context("Date overflow computing end date")?; + let dt_range = date_range(start_date, end_date)?; - let schema_enum = Schema::from_str("mbp-10").context("Failed to parse schema")?; - - // Build download parameters - let params = GetRangeParams::builder() - .dataset("GLBX.MDP3".to_string()) // CME Globex + // Build download parameters — writes zstd-compressed DBN directly to file + let params = GetRangeToFileParams::builder() + .dataset("GLBX.MDP3") // CME Globex .symbols(vec![symbol.to_string()]) - .schema(schema_enum) // Level 2: 10 bid/ask levels - .date_time_range(date_time_range) + .schema(Schema::Mbp10) // Level 2: 10 bid/ask levels + .stype_in(SType::Parent) // .FUT parent symbols + .date_time_range(dt_range) + .path(&output_file) .build(); // Download data with retry logic @@ -152,41 +163,34 @@ async fn download_symbol_day( let max_retries = 3; loop { - match client.timeseries().get_range(¶ms).await { - Ok(mut decoder) => { - // Read all data into buffer - let mut buffer = Vec::new(); - let mut temp_buf = vec![0u8; 8192]; - loop { - let n = decoder.get_mut().read(&mut temp_buf).await?; - if n == 0 { - break; - } - buffer.extend_from_slice(&temp_buf[..n]); - } - - let size = buffer.len() as u64; + match client.timeseries().get_range_to_file(¶ms).await { + Ok(_decoder) => { + let size = fs::metadata(&output_file) + .map(|m| m.len()) + .unwrap_or(0); // Validate minimum size (should be >1 KB for a trading day) if size < 1024 { + // Remove the tiny/empty file and report as no data + let _ = fs::remove_file(&output_file); return Ok(None); // Likely no data (holiday/no trading) } - // Write to file - fs::write(&output_file, &buffer).context("Failed to write data file")?; - - // Estimate record count - let estimated_records = size / 480; + // Estimate record count (compressed, ~70% compression ratio) + let estimated_records = (size as f64 / (480.0 * 0.3)) as u64; return Ok(Some((size, estimated_records))); }, Err(e) => { + // Clean up partial file on failure + let _ = fs::remove_file(&output_file); + retries += 1; if retries >= max_retries { return Err(anyhow::anyhow!("Max retries exceeded: {}", e)); } - eprintln!(" ⚠️ Retry {}/{}: {}", retries, max_retries, e); + eprintln!(" Warning: Retry {}/{}: {}", retries, max_retries, e); tokio::time::sleep(tokio::time::Duration::from_secs(2_u64.pow(retries))).await; }, }