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 <noreply@anthropic.com>
2026-03-07 12:19:46 +01:00
parent a35a564f45
commit 5829d6378e
3 changed files with 270 additions and 10 deletions
--- a/config/universe-es-mbp10.toml
+++ 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"
--- 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<String>,
 }

 // ---------------------------------------------------------------------------
@@ -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<Schema> {
+    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<u64> {
    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 <local_path> :<dest>/<symbol>/<filename>`.
+/// 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);
--- a/infra/k8s/training/download-mbp10-job.yaml
+++ 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=<MINIO_ENDPOINT> \
+#          --s3-access-key-id=<KEY> --s3-secret-access-key=<SECRET> \
+#          --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