From 5829d6378e8df6102d741a10a11e148ee2d8542f Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 7 Mar 2026 12:19:46 +0100 Subject: [PATCH] feat(data): schema-configurable Databento download with MinIO upload - download_baseline now reads schema from universe TOML config (was hardcoded to ohlcv-1m, now supports mbp-10 and other schemas) - Add --rclone-dest flag for direct upload to MinIO after each download - Add config/universe-es-mbp10.toml: ES.FUT MBP-10, same date range - Add infra/k8s/training/download-mbp10-job.yaml: K8s Job to download ES.FUT MBP-10 data in-cluster and upload to MinIO Usage: kubectl apply -f infra/k8s/training/download-mbp10-job.yaml Prereqs: databento-credentials secret, download_baseline binary in MinIO Co-Authored-By: Claude Opus 4.6 --- config/universe-es-mbp10.toml | 20 +++ crates/ml/examples/download_baseline.rs | 98 +++++++++++-- infra/k8s/training/download-mbp10-job.yaml | 162 +++++++++++++++++++++ 3 files changed, 270 insertions(+), 10 deletions(-) create mode 100644 config/universe-es-mbp10.toml create mode 100644 infra/k8s/training/download-mbp10-job.yaml diff --git a/config/universe-es-mbp10.toml b/config/universe-es-mbp10.toml new file mode 100644 index 000000000..870ce1cd0 --- /dev/null +++ b/config/universe-es-mbp10.toml @@ -0,0 +1,20 @@ +# ES.FUT MBP-10 (Level 2 Order Book) Data +# +# 10-level order book snapshots for OFI feature extraction. +# Same date range as OHLCV baseline, ES.FUT only. +# +# Schema: mbp-10 via Databento (GLBX.MDP3 dataset) + +[universe] +name = "es-mbp10" +description = "ES.FUT MBP-10 order book data for OFI features" +date_range_start = "2024-03-01" +date_range_end = "2026-02-22" +bar_size = "tick" +databento_dataset = "GLBX.MDP3" +databento_schema = "mbp-10" + +[[symbols]] +symbol = "ES.FUT" +exchange = "GLBX.MDP3" +asset_class = "Future" diff --git a/crates/ml/examples/download_baseline.rs b/crates/ml/examples/download_baseline.rs index a2db16dc4..8a1efca6b 100644 --- a/crates/ml/examples/download_baseline.rs +++ b/crates/ml/examples/download_baseline.rs @@ -1,20 +1,25 @@ -//! Download 730 days of Databento OHLCV-1m data in quarterly chunks +//! Download quarterly-chunked Databento data (any schema) with optional MinIO upload //! -//! Downloads futures baseline data for ML model training, split into -//! ~90-day quarterly files for efficient caching and resume support. +//! Downloads futures data for ML model training, split into ~90-day quarterly +//! files for efficient caching and resume support. Schema (ohlcv-1m, mbp-10, +//! etc.) is read from the universe TOML config. //! //! Configuration is read from a universe TOML file (default: //! `config/universe-futures-baseline.toml`). //! //! Usage: //! # Dry run (preview config and cost estimate) -//! cargo run -p ml --example `download_baseline` --release -- --dry-run +//! cargo run -p ml --example download_baseline --release -- --dry-run //! -//! # Download with confirmation prompt +//! # Download OHLCV with confirmation prompt //! cargo run -p ml --example download_baseline --release //! -//! # Skip confirmation prompt -//! cargo run -p ml --example download_baseline --release -- --yes +//! # Download MBP-10 for ES.FUT and upload to MinIO +//! cargo run -p ml --example download_baseline --release -- \ +//! --universe-config config/universe-es-mbp10.toml \ +//! --output-dir /tmp/mbp10 \ +//! --rclone-dest "s3:foxhunt-training-data/futures-baseline-mbp10" \ +//! --yes use anyhow::{Context, Result}; use chrono::{Datelike, NaiveDate}; @@ -27,6 +32,7 @@ use serde::Deserialize; use std::env; use std::fs; use std::path::PathBuf; +use std::process::Command; use std::time::Instant; // --------------------------------------------------------------------------- @@ -36,7 +42,7 @@ use std::time::Instant; #[derive(Debug, Parser)] #[command( name = "download_baseline", - about = "Download quarterly OHLCV-1m futures data from Databento" + about = "Download quarterly Databento data (any schema) with optional MinIO upload" )] struct Opts { /// Output directory for downloaded files @@ -54,6 +60,11 @@ struct Opts { /// Skip confirmation prompt #[arg(long)] yes: bool, + + /// Upload each file to MinIO/S3 via rclone after download. + /// Format: "s3:bucket/prefix" (e.g. "s3:foxhunt-training-data/futures-baseline-mbp10") + #[arg(long)] + rclone_dest: Option, } // --------------------------------------------------------------------------- @@ -72,6 +83,28 @@ struct UniverseMeta { date_range_start: String, date_range_end: String, databento_dataset: String, + #[serde(default = "default_schema")] + databento_schema: String, +} + +fn default_schema() -> String { + "ohlcv-1m".to_owned() +} + +/// Map config schema string to dbn::Schema enum. +fn parse_schema(s: &str) -> Result { + match s { + "ohlcv-1s" => Ok(Schema::Ohlcv1S), + "ohlcv-1m" => Ok(Schema::Ohlcv1M), + "ohlcv-1h" => Ok(Schema::Ohlcv1H), + "ohlcv-1d" => Ok(Schema::Ohlcv1D), + "trades" => Ok(Schema::Trades), + "mbp-1" => Ok(Schema::Mbp1), + "mbp-10" => Ok(Schema::Mbp10), + "bbo-1s" => Ok(Schema::Bbo1S), + "bbo-1m" => Ok(Schema::Bbo1M), + other => Err(anyhow::anyhow!("Unknown Databento schema: {}", other)), + } } #[derive(Debug, Deserialize)] @@ -176,6 +209,7 @@ async fn download_quarter( symbol: &str, quarter: &Quarter, dataset: &str, + schema: Schema, output_dir: &std::path::Path, ) -> Result { let filename = format!("{}_{}.dbn.zst", symbol, quarter.label); @@ -196,7 +230,7 @@ async fn download_quarter( let params = GetRangeToFileParams::builder() .dataset(dataset.to_owned()) .symbols(vec![symbol.to_owned()]) - .schema(Schema::Ohlcv1M) + .schema(schema) .stype_in(SType::Parent) .date_time_range(dt_range) .path(file_path.clone()) @@ -217,6 +251,31 @@ async fn download_quarter( Ok(meta.len()) } +/// Upload a single file to MinIO/S3 via rclone. +/// +/// Runs `rclone copyto ://`. +/// Rclone flags for MinIO are sourced from RCLONE_* env vars (set in K8s job). +fn rclone_upload(local_path: &std::path::Path, dest: &str, symbol: &str) -> Result<()> { + let filename = local_path + .file_name() + .and_then(|n| n.to_str()) + .ok_or_else(|| anyhow::anyhow!("Invalid filename: {:?}", local_path))?; + + let remote_path = format!(":{}/{}/{}", dest, symbol, filename); + + let status = Command::new("rclone") + .args(["copyto", &local_path.to_string_lossy(), &remote_path]) + .status() + .context("Failed to execute rclone (is it installed?)")?; + + if !status.success() { + anyhow::bail!("rclone upload failed for {}", remote_path); + } + + println!(" [MINIO] uploaded → {}", remote_path); + Ok(()) +} + // --------------------------------------------------------------------------- // Main // --------------------------------------------------------------------------- @@ -243,6 +302,7 @@ async fn main() -> Result<()> { let end_date = NaiveDate::parse_from_str(&config.universe.date_range_end, "%Y-%m-%d") .context("Failed to parse date_range_end")?; + let schema = parse_schema(&config.universe.databento_schema)?; let symbols: Vec<&str> = config.symbols.iter().map(|s| s.symbol.as_str()).collect(); let quarters = generate_quarters(start_date, end_date); let total_days = (end_date - start_date).num_days(); @@ -256,7 +316,7 @@ async fn main() -> Result<()> { println!(); println!("Universe: {}", config.universe.name); println!("Dataset: {}", config.universe.databento_dataset); - println!("Schema: ohlcv-1m"); + println!("Schema: {}", config.universe.databento_schema); println!("Date range: {} to {}", start_date, end_date); println!("Total days: {}", total_days); println!( @@ -266,6 +326,9 @@ async fn main() -> Result<()> { ); println!("Quarters: {}", quarters.len()); println!("Output: {}", opts.output_dir); + if let Some(ref dest) = opts.rclone_dest { + println!("MinIO dest: :{}", dest); + } println!(); // Print quarter breakdown @@ -355,6 +418,13 @@ async fn main() -> Result<()> { ); stats.skipped += 1; stats.total_bytes += meta.len(); + + // Still upload skipped files if --rclone-dest is set + if let Some(ref dest) = opts.rclone_dest { + if let Err(e) = rclone_upload(&file_path, dest, symbol) { + println!(" [WARN] rclone upload failed: {}", e); + } + } continue; } _ => {} @@ -367,6 +437,7 @@ async fn main() -> Result<()> { symbol, quarter, &config.universe.databento_dataset, + schema, &output_dir, ) .await @@ -379,6 +450,13 @@ async fn main() -> Result<()> { ); stats.successful += 1; stats.total_bytes += size; + + // Upload to MinIO if --rclone-dest is set + if let Some(ref dest) = opts.rclone_dest { + if let Err(e) = rclone_upload(&file_path, dest, symbol) { + println!(" [WARN] rclone upload failed: {}", e); + } + } } Err(e) => { println!(" [FAIL] {} -- {}", filename, e); diff --git a/infra/k8s/training/download-mbp10-job.yaml b/infra/k8s/training/download-mbp10-job.yaml new file mode 100644 index 000000000..8de732d71 --- /dev/null +++ b/infra/k8s/training/download-mbp10-job.yaml @@ -0,0 +1,162 @@ +# Download MBP-10 order book data for ES.FUT from Databento → MinIO. +# +# Prerequisites: +# 1. Compile + upload the binary: +# cargo build -p ml --example download_baseline --release +# rclone copyto target/release/examples/download_baseline \ +# :s3:foxhunt-binaries/tools/download_baseline \ +# --s3-provider=Minio --s3-endpoint= \ +# --s3-access-key-id= --s3-secret-access-key= \ +# --s3-no-check-bucket +# +# 2. Create Databento API key secret: +# kubectl -n foxhunt create secret generic databento-credentials \ +# --from-literal=api-key=$DATABENTO_API_KEY +# +# 3. Apply: +# kubectl apply -f infra/k8s/training/download-mbp10-job.yaml +# +# The job downloads ES.FUT MBP-10 data (730 days, quarterly chunks) and +# uploads each file to s3:foxhunt-training-data/futures-baseline-mbp10/. +apiVersion: batch/v1 +kind: Job +metadata: + name: download-mbp10 + namespace: foxhunt + labels: + app.kubernetes.io/name: download-mbp10 + app.kubernetes.io/part-of: foxhunt + foxhunt/job-type: data-download +spec: + backoffLimit: 2 + activeDeadlineSeconds: 14400 # 4 hours + ttlSecondsAfterFinished: 3600 + template: + metadata: + labels: + app.kubernetes.io/name: download-mbp10 + foxhunt/job-type: data-download + spec: + nodeSelector: + k8s.scaleway.com/pool-name: default + imagePullSecrets: + - name: gitlab-registry + restartPolicy: Never + initContainers: + - name: fetch-binary + image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest + command: ["/bin/sh", "-c"] + args: + - | + set -e + rclone copyto \ + :s3:foxhunt-binaries/tools/download_baseline \ + /binaries/download_baseline \ + --s3-provider=Minio \ + --s3-endpoint=http://minio.foxhunt.svc.cluster.local:9000 \ + --s3-access-key-id="${MINIO_ACCESS_KEY}" \ + --s3-secret-access-key="${MINIO_SECRET_KEY}" \ + --s3-no-check-bucket + chmod +x /binaries/download_baseline + echo "Fetched download_baseline ($(stat -c%s /binaries/download_baseline) bytes)" + env: + - name: MINIO_ACCESS_KEY + valueFrom: + secretKeyRef: + name: minio-credentials + key: access-key + - name: MINIO_SECRET_KEY + valueFrom: + secretKeyRef: + name: minio-credentials + key: secret-key + volumeMounts: + - name: binaries + mountPath: /binaries + resources: + requests: + cpu: 100m + memory: 64Mi + limits: + cpu: 500m + memory: 128Mi + containers: + - name: downloader + image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest + command: ["/bin/sh", "-c"] + args: + - | + set -e + + # Write universe config inline (avoids ConfigMap for a one-off job) + mkdir -p /tmp/config + cat > /tmp/config/universe-es-mbp10.toml <<'TOML' + [universe] + name = "es-mbp10" + description = "ES.FUT MBP-10 order book data for OFI features" + date_range_start = "2024-03-01" + date_range_end = "2026-02-22" + bar_size = "tick" + databento_dataset = "GLBX.MDP3" + databento_schema = "mbp-10" + + [[symbols]] + symbol = "ES.FUT" + exchange = "GLBX.MDP3" + asset_class = "Future" + TOML + + echo "=== ES.FUT MBP-10 Download Job ===" + echo "Downloading from Databento and uploading to MinIO..." + + /binaries/download_baseline \ + --universe-config /tmp/config/universe-es-mbp10.toml \ + --output-dir /tmp/mbp10-download \ + --rclone-dest "s3:foxhunt-training-data/futures-baseline-mbp10" \ + --yes + + echo "=== Download complete ===" + echo "Files in MinIO: foxhunt-training-data/futures-baseline-mbp10/ES.FUT/" + env: + - name: DATABENTO_API_KEY + valueFrom: + secretKeyRef: + name: databento-credentials + key: api-key + # rclone env-var config (used by download_baseline --rclone-dest) + - name: RCLONE_S3_PROVIDER + value: Minio + - name: RCLONE_S3_ENDPOINT + value: http://minio.foxhunt.svc.cluster.local:9000 + - name: RCLONE_S3_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: minio-credentials + key: access-key + - name: RCLONE_S3_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: minio-credentials + key: secret-key + - name: RCLONE_S3_NO_CHECK_BUCKET + value: "true" + volumeMounts: + - name: binaries + mountPath: /binaries + readOnly: true + - name: scratch + mountPath: /tmp + resources: + requests: + cpu: "1" + memory: 1Gi + limits: + cpu: "2" + memory: 4Gi + volumes: + - name: binaries + emptyDir: + sizeLimit: 100Mi + - name: scratch + emptyDir: + sizeLimit: 50Gi # MBP-10 data can be large