fix(ml): repair download_l2_data example (time→chrono, GetRangeToFileParams)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

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<Vec<Str
     Ok(dates)
 }
 
+/// Convert a `chrono::NaiveDate` (midnight UTC) to UNIX nanoseconds for the
+/// Databento `DateTimeRange` API.
+fn naive_date_to_unix_nanos(date: NaiveDate) -> 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<DateTimeRange> {
+    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<Option<(u64, u64)>> {
-    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(&params).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(&params).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;
             },
         }