chore: remove old Python-based Databento batch download job

All downloads now use the Rust download_baseline binary via the
Argo WorkflowTemplate (databento-download).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-31 23:03:03 +02:00
parent 3ca1f92339
commit aa1accf5b7

View File

@@ -1,137 +0,0 @@
# Download Databento batch jobs to training-data-pvc.
#
# Prerequisites:
# 1. Submit batch jobs via databento Python API
# 2. Wait for all jobs to reach state=done
# 3. kubectl apply -f infra/k8s/jobs/databento-download.yaml
#
# The job downloads all completed batch files, organizes by schema/symbol,
# and cleans up old corrupt data.
---
apiVersion: batch/v1
kind: Job
metadata:
name: databento-download
namespace: foxhunt
labels:
app.kubernetes.io/name: databento-download
app.kubernetes.io/part-of: foxhunt
spec:
backoffLimit: 2
ttlSecondsAfterFinished: 3600
template:
metadata:
labels:
app.kubernetes.io/name: databento-download
app.kubernetes.io/part-of: foxhunt
app.kubernetes.io/component: compile-and-train
spec:
restartPolicy: OnFailure
securityContext:
runAsUser: 1000
runAsGroup: 1000
fsGroup: 1000
nodeSelector:
k8s.scaleway.com/pool-name: ci-compile-cpu
tolerations:
- key: node.cilium.io/agent-not-ready
operator: Exists
effect: NoSchedule
containers:
- name: downloader
image: python:3.12-slim
command: ["/bin/bash", "-c"]
env:
- name: DATABENTO_API_KEY
valueFrom:
secretKeyRef:
name: databento-credentials
key: api-key
resources:
requests:
cpu: "4"
memory: 8Gi
limits:
cpu: "8"
memory: 16Gi
volumeMounts:
- name: training-data
mountPath: /data
args:
- |
set -e
export HOME=/tmp
pip install --no-cache-dir --break-system-packages databento
python3 << 'PYEOF'
import databento as db
import os
import shutil
from pathlib import Path
client = db.Historical(os.environ["DATABENTO_API_KEY"])
DATA_ROOT = Path("/data")
SYMBOL = "ES.FUT"
# Schema → directory mapping
SCHEMA_DIRS = {
"mbp-10": "futures-baseline-mbp10",
"trades": "futures-baseline-trades",
"ohlcv-1m": "futures-baseline",
"ohlcv-1s": "futures-baseline-1s",
"mbp-1": "futures-baseline-mbp1",
}
jobs = client.batch.list_jobs()
print(f"Found {len(jobs)} batch jobs")
for j in jobs:
jd = j if isinstance(j, dict) else vars(j)
schema = jd["schema"]
state = jd["state"]
job_id = jd["id"]
if state != "done":
print(f" SKIP {schema} ({job_id}): state={state}")
continue
dir_name = SCHEMA_DIRS.get(schema, f"futures-{schema}")
out_dir = DATA_ROOT / dir_name / SYMBOL
out_dir.mkdir(parents=True, exist_ok=True)
print(f"\n=== Downloading {schema} ({job_id}) to {out_dir} ===")
# Clean old data for this schema
old_files = list(out_dir.glob("*.dbn*"))
if old_files:
print(f" Cleaning {len(old_files)} old files...")
for f in old_files:
f.unlink()
# Download all files for this job
try:
files = client.batch.download(job_id, output_dir=str(out_dir))
print(f" Downloaded {len(files)} files:")
for f in sorted(files):
size_mb = os.path.getsize(f) / 1024**2
print(f" {os.path.basename(f)}: {size_mb:.1f} MB")
except Exception as e:
print(f" ERROR downloading {schema}: {e}")
continue
# Summary
print("\n=== Download complete ===")
for schema, dir_name in SCHEMA_DIRS.items():
d = DATA_ROOT / dir_name / SYMBOL
if d.exists():
files = list(d.glob("*"))
total = sum(f.stat().st_size for f in files if f.is_file())
print(f" {schema:10s}: {len(files)} files, {total/1024**3:.1f}G")
else:
print(f" {schema:10s}: not downloaded")
PYEOF
volumes:
- name: training-data
persistentVolumeClaim:
claimName: training-data-pvc