feat: parameterized Argo WorkflowTemplate for Databento downloads
Replace hardcoded K8s Job with reusable Argo WorkflowTemplate (databento-download) parameterized by schema, output-dir, parallel, and node-pool. Add argo-download.sh CLI wrapper matching the argo-train.sh pattern. Remove old streaming job file. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
20
config/universe-es-trades.toml
Normal file
20
config/universe-es-trades.toml
Normal file
@@ -0,0 +1,20 @@
|
||||
# ES.FUT Trades Data
|
||||
#
|
||||
# Individual trade executions for VPIN, Kyle's Lambda, and trade imbalance.
|
||||
# Same date range as MBP-10/OHLCV baseline, ES.FUT only.
|
||||
#
|
||||
# Schema: trades via Databento (GLBX.MDP3 dataset)
|
||||
|
||||
[universe]
|
||||
name = "es-trades"
|
||||
description = "ES.FUT trade data for VPIN/Kyle's Lambda features"
|
||||
date_range_start = "2024-03-01"
|
||||
date_range_end = "2026-02-22"
|
||||
bar_size = "tick"
|
||||
databento_dataset = "GLBX.MDP3"
|
||||
databento_schema = "trades"
|
||||
|
||||
[[symbols]]
|
||||
symbol = "ES.FUT"
|
||||
exchange = "GLBX.MDP3"
|
||||
asset_class = "Future"
|
||||
134
infra/k8s/argo/databento-download-template.yaml
Normal file
134
infra/k8s/argo/databento-download-template.yaml
Normal file
@@ -0,0 +1,134 @@
|
||||
# Databento data download via streaming API (get_range).
|
||||
#
|
||||
# Parameterized Argo WorkflowTemplate — replaces the hardcoded K8s Job.
|
||||
# Binary pre-uploaded to /data/bin/download_baseline on training-data-pvc.
|
||||
# Universe configs loaded from configmap databento-universe-all.
|
||||
#
|
||||
# Usage:
|
||||
# argo submit -n foxhunt --from=wftmpl/databento-download
|
||||
# argo submit -n foxhunt --from=wftmpl/databento-download -p schema=trades -p output-dir=/data/futures-baseline-trades
|
||||
# ./scripts/argo-download.sh mbp-10 --parallel 9 --pool ci-compile-cpu
|
||||
#
|
||||
# Databento rate limits: 100 concurrent connections per IP, 100 req/s.
|
||||
---
|
||||
apiVersion: argoproj.io/v1alpha1
|
||||
kind: WorkflowTemplate
|
||||
metadata:
|
||||
name: databento-download
|
||||
namespace: foxhunt
|
||||
labels:
|
||||
app.kubernetes.io/name: databento-download
|
||||
app.kubernetes.io/part-of: foxhunt
|
||||
spec:
|
||||
entrypoint: download
|
||||
serviceAccountName: argo-workflow
|
||||
podMetadata:
|
||||
labels:
|
||||
app.kubernetes.io/part-of: foxhunt
|
||||
app.kubernetes.io/component: databento-download
|
||||
securityContext:
|
||||
fsGroup: 1000
|
||||
ttlStrategy:
|
||||
secondsAfterCompletion: 3600
|
||||
activeDeadlineSeconds: 86400 # 24 hours
|
||||
|
||||
arguments:
|
||||
parameters:
|
||||
- name: schema
|
||||
value: mbp-10
|
||||
description: "Databento schema: ohlcv-1m, ohlcv-1s, trades, mbp-10"
|
||||
- name: output-dir
|
||||
value: /data/futures-baseline-mbp10
|
||||
description: "Output directory on training-data PVC"
|
||||
- name: parallel
|
||||
value: "9"
|
||||
description: "Number of parallel quarterly downloads (max ~100 per Databento rate limit)"
|
||||
- name: node-pool
|
||||
value: ci-compile-cpu
|
||||
description: "K8s node pool: ci-compile-cpu (32c/64G) or platform (4c/8G)"
|
||||
|
||||
volumes:
|
||||
- name: training-data
|
||||
persistentVolumeClaim:
|
||||
claimName: training-data-pvc
|
||||
- name: universe-configs
|
||||
configMap:
|
||||
name: databento-universe-all
|
||||
|
||||
templates:
|
||||
- name: download
|
||||
inputs:
|
||||
parameters:
|
||||
- name: schema
|
||||
- name: output-dir
|
||||
- name: parallel
|
||||
- name: node-pool
|
||||
nodeSelector:
|
||||
k8s.scaleway.com/pool-name: "{{inputs.parameters.node-pool}}"
|
||||
container:
|
||||
image: ubuntu:24.04
|
||||
command: ["/bin/bash", "-c"]
|
||||
env:
|
||||
- name: RUST_LOG
|
||||
value: info
|
||||
- name: DATABENTO_API_KEY
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: databento-credentials
|
||||
key: api-key
|
||||
resources:
|
||||
requests:
|
||||
cpu: "2"
|
||||
memory: 4Gi
|
||||
limits:
|
||||
cpu: "8"
|
||||
memory: 16Gi
|
||||
volumeMounts:
|
||||
- name: training-data
|
||||
mountPath: /data
|
||||
- name: universe-configs
|
||||
mountPath: /configs
|
||||
args:
|
||||
- |
|
||||
set -e
|
||||
apt-get update -qq > /dev/null 2>&1 && apt-get install -y -qq ca-certificates > /dev/null 2>&1
|
||||
|
||||
BINARY=/data/bin/download_baseline
|
||||
chmod +x "$BINARY"
|
||||
|
||||
SCHEMA="{{inputs.parameters.schema}}"
|
||||
OUTPUT_DIR="{{inputs.parameters.output-dir}}"
|
||||
PARALLEL="{{inputs.parameters.parallel}}"
|
||||
|
||||
# Map schema name to configmap key
|
||||
CONFIG="/configs/${SCHEMA}.toml"
|
||||
if [ ! -f "$CONFIG" ]; then
|
||||
echo "ERROR: Config not found: $CONFIG"
|
||||
echo "Available configs:"
|
||||
ls /configs/
|
||||
exit 1
|
||||
fi
|
||||
|
||||
mkdir -p "$OUTPUT_DIR"
|
||||
|
||||
echo "=== Databento Download ==="
|
||||
echo "Schema: $SCHEMA"
|
||||
echo "Config: $CONFIG"
|
||||
echo "Output: $OUTPUT_DIR"
|
||||
echo "Parallel: $PARALLEL"
|
||||
echo "Node: $(hostname), CPUs: $(nproc), RAM: $(free -g | awk '/Mem:/{print $2}')G"
|
||||
echo ""
|
||||
|
||||
$BINARY \
|
||||
--universe-config "$CONFIG" \
|
||||
--output-dir "$OUTPUT_DIR" \
|
||||
--parallel "$PARALLEL" \
|
||||
--yes
|
||||
|
||||
echo ""
|
||||
echo "=== Download complete ==="
|
||||
echo "Files:"
|
||||
find "$OUTPUT_DIR" -name '*.dbn.zst' -exec ls -lh {} \;
|
||||
echo ""
|
||||
du -sh "$OUTPUT_DIR"
|
||||
df -h /data
|
||||
@@ -1,127 +0,0 @@
|
||||
# Databento data download via streaming API (get_range).
|
||||
#
|
||||
# Binary pre-uploaded to /data/bin/download_baseline on training-data-pvc.
|
||||
# Universe configs loaded from configmap.
|
||||
#
|
||||
# Test: kubectl apply -f infra/k8s/jobs/databento-download-streaming.yaml
|
||||
---
|
||||
apiVersion: networking.k8s.io/v1
|
||||
kind: NetworkPolicy
|
||||
metadata:
|
||||
name: databento-download
|
||||
namespace: foxhunt
|
||||
labels:
|
||||
app.kubernetes.io/part-of: foxhunt
|
||||
spec:
|
||||
podSelector:
|
||||
matchLabels:
|
||||
app.kubernetes.io/component: databento-download
|
||||
policyTypes:
|
||||
- Egress
|
||||
egress:
|
||||
- ports:
|
||||
- port: 53
|
||||
protocol: UDP
|
||||
- port: 53
|
||||
protocol: TCP
|
||||
- ports:
|
||||
- port: 80
|
||||
protocol: TCP
|
||||
- port: 443
|
||||
protocol: TCP
|
||||
- ports:
|
||||
- port: 443
|
||||
protocol: TCP
|
||||
- port: 6443
|
||||
protocol: TCP
|
||||
to:
|
||||
- ipBlock:
|
||||
cidr: 10.32.0.0/16
|
||||
- ipBlock:
|
||||
cidr: 172.16.0.4/32
|
||||
---
|
||||
apiVersion: batch/v1
|
||||
kind: Job
|
||||
metadata:
|
||||
name: databento-download-all
|
||||
namespace: foxhunt
|
||||
labels:
|
||||
app.kubernetes.io/name: databento-download
|
||||
app.kubernetes.io/part-of: foxhunt
|
||||
spec:
|
||||
backoffLimit: 3
|
||||
activeDeadlineSeconds: 86400
|
||||
ttlSecondsAfterFinished: 3600
|
||||
template:
|
||||
metadata:
|
||||
labels:
|
||||
app.kubernetes.io/name: databento-download
|
||||
app.kubernetes.io/part-of: foxhunt
|
||||
app.kubernetes.io/component: databento-download
|
||||
spec:
|
||||
restartPolicy: OnFailure
|
||||
securityContext:
|
||||
fsGroup: 1000
|
||||
nodeSelector:
|
||||
k8s.scaleway.com/pool-name: platform
|
||||
containers:
|
||||
- name: downloader
|
||||
image: ubuntu:24.04
|
||||
command: ["/bin/bash", "-c"]
|
||||
env:
|
||||
- name: RUST_LOG
|
||||
value: debug
|
||||
- name: DATABENTO_API_KEY
|
||||
valueFrom:
|
||||
secretKeyRef:
|
||||
name: databento-credentials
|
||||
key: api-key
|
||||
resources:
|
||||
requests:
|
||||
cpu: 500m
|
||||
memory: 512Mi
|
||||
limits:
|
||||
cpu: "2"
|
||||
memory: 2Gi
|
||||
volumeMounts:
|
||||
- name: training-data
|
||||
mountPath: /data
|
||||
- name: universe-configs
|
||||
mountPath: /configs
|
||||
args:
|
||||
- |
|
||||
set -e
|
||||
apt-get update -qq > /dev/null 2>&1 && apt-get install -y -qq ca-certificates > /dev/null 2>&1
|
||||
BINARY=/data/bin/download_baseline
|
||||
chmod +x "$BINARY"
|
||||
|
||||
# Download 4 schemas from configmap-mounted TOML configs
|
||||
for SCHEMA in ohlcv-1m trades ohlcv-1s mbp-10; do
|
||||
case $SCHEMA in
|
||||
ohlcv-1m) DIR=/data/futures-baseline ;;
|
||||
trades) DIR=/data/futures-baseline-trades ;;
|
||||
ohlcv-1s) DIR=/data/futures-baseline-1s ;;
|
||||
mbp-10) DIR=/data/futures-baseline-mbp10 ;;
|
||||
esac
|
||||
|
||||
echo ""
|
||||
echo "=== Downloading ${SCHEMA} ==="
|
||||
mkdir -p "${DIR}"
|
||||
$BINARY \
|
||||
--universe-config /configs/${SCHEMA}.toml \
|
||||
--output-dir "${DIR}" \
|
||||
--yes || echo "WARN: ${SCHEMA} had failures"
|
||||
du -sh "${DIR}" 2>/dev/null
|
||||
done
|
||||
|
||||
echo ""
|
||||
echo "=== Final PVC summary ==="
|
||||
du -sh /data/*/
|
||||
df -h /data
|
||||
volumes:
|
||||
- name: training-data
|
||||
persistentVolumeClaim:
|
||||
claimName: training-data-pvc
|
||||
- name: universe-configs
|
||||
configMap:
|
||||
name: databento-universe-all
|
||||
91
scripts/argo-download.sh
Executable file
91
scripts/argo-download.sh
Executable file
@@ -0,0 +1,91 @@
|
||||
#!/usr/bin/env bash
|
||||
# Download Databento data via Argo Workflows.
|
||||
#
|
||||
# Usage:
|
||||
# ./scripts/argo-download.sh mbp-10 # MBP-10 with defaults
|
||||
# ./scripts/argo-download.sh trades --output /data/futures-baseline-trades
|
||||
# ./scripts/argo-download.sh ohlcv-1s --parallel 4 --pool platform
|
||||
# ./scripts/argo-download.sh mbp-10 --watch # follow logs
|
||||
#
|
||||
# Schemas: ohlcv-1m, ohlcv-1s, trades, mbp-10
|
||||
#
|
||||
# Requires: argo CLI
|
||||
set -euo pipefail
|
||||
|
||||
TEMPLATE="databento-download"
|
||||
PARALLEL=""
|
||||
OUTPUT=""
|
||||
POOL=""
|
||||
WATCH=false
|
||||
|
||||
usage() {
|
||||
cat <<EOF
|
||||
Usage: $(basename "$0") <schema> [OPTIONS]
|
||||
|
||||
Schemas:
|
||||
ohlcv-1m OHLCV 1-minute bars
|
||||
ohlcv-1s OHLCV 1-second bars
|
||||
trades Individual trade executions
|
||||
mbp-10 10-level order book snapshots
|
||||
|
||||
Options:
|
||||
--output <dir> Output dir on PVC (default: schema-based)
|
||||
--parallel <n> Parallel quarterly downloads (default: 9)
|
||||
--pool <name> Node pool (default: ci-compile-cpu)
|
||||
--watch Follow workflow logs after submission
|
||||
-h, --help Show this help
|
||||
EOF
|
||||
exit 0
|
||||
}
|
||||
|
||||
[[ $# -eq 0 ]] && { echo "Error: schema argument required"; usage; }
|
||||
[[ "$1" == "-h" || "$1" == "--help" ]] && usage
|
||||
|
||||
SCHEMA="$1"; shift
|
||||
|
||||
# Validate schema
|
||||
case "$SCHEMA" in
|
||||
ohlcv-1m|ohlcv-1s|trades|mbp-10) ;;
|
||||
*) echo "Error: unknown schema '$SCHEMA'"; usage ;;
|
||||
esac
|
||||
|
||||
while [[ $# -gt 0 ]]; do
|
||||
case $1 in
|
||||
--output) OUTPUT="$2"; shift 2 ;;
|
||||
--parallel) PARALLEL="$2"; shift 2 ;;
|
||||
--pool) POOL="$2"; shift 2 ;;
|
||||
--watch) WATCH=true; shift ;;
|
||||
-h|--help) usage ;;
|
||||
*) echo "Unknown option: $1"; usage ;;
|
||||
esac
|
||||
done
|
||||
|
||||
# Default output dir based on schema
|
||||
if [[ -z "$OUTPUT" ]]; then
|
||||
case "$SCHEMA" in
|
||||
ohlcv-1m) OUTPUT="/data/futures-baseline" ;;
|
||||
ohlcv-1s) OUTPUT="/data/futures-baseline-1s" ;;
|
||||
trades) OUTPUT="/data/futures-baseline-trades" ;;
|
||||
mbp-10) OUTPUT="/data/futures-baseline-mbp10" ;;
|
||||
esac
|
||||
fi
|
||||
|
||||
# Build argo submit command
|
||||
CMD="argo submit -n foxhunt --from=wftmpl/$TEMPLATE"
|
||||
CMD="$CMD -p schema=$SCHEMA"
|
||||
CMD="$CMD -p output-dir=$OUTPUT"
|
||||
|
||||
[[ -n "$PARALLEL" ]] && CMD="$CMD -p parallel=$PARALLEL"
|
||||
[[ -n "$POOL" ]] && CMD="$CMD -p node-pool=$POOL"
|
||||
|
||||
if $WATCH; then
|
||||
CMD="$CMD --watch"
|
||||
fi
|
||||
|
||||
echo "Submitting Databento download workflow..."
|
||||
echo " schema: $SCHEMA"
|
||||
echo " output: $OUTPUT"
|
||||
[[ -n "$PARALLEL" ]] && echo " parallel: $PARALLEL"
|
||||
[[ -n "$POOL" ]] && echo " pool: $POOL"
|
||||
echo ""
|
||||
eval "$CMD"
|
||||
Reference in New Issue
Block a user