diff --git a/infra/k8s/argo/lob-backtest-sweep-template.yaml b/infra/k8s/argo/lob-backtest-sweep-template.yaml new file mode 100644 index 000000000..99e7299b0 --- /dev/null +++ b/infra/k8s/argo/lob-backtest-sweep-template.yaml @@ -0,0 +1,348 @@ +# Real-LOB backtest sweep workflow. +# +# Fans out a parameter grid across N parallel GPU pods, each running +# `fxt-backtest run` against the same MBP-10 data + checkpoint with one +# cell's worth of overrides. A final CPU pod runs `fxt-backtest aggregate` +# to produce aggregate.parquet + pareto_frontier.json at the sweep root. +# +# The `# __SWEEP_CELLS__` marker on the dag.tasks line is replaced by +# scripts/argo-lob-sweep.sh with N generated WorkflowTask stanzas before +# submission. Same convention as train-multi-seed-template.yaml's +# # __MATRIX_TASKS__ marker. +# +# DAG: +# ensure-binary ──> [N parallel run-cell- tasks on ci-training-l40s, +# each with its own outputDir on the shared PVC] +# │ +# └──> aggregate (CPU node, runs fxt-backtest aggregate +# against the sweep root) +# +# Per `feedback_default_to_l40s_pool.md` (2026-05-09): defaults to +# ci-training-l40s + sm_89. Override via --gpu-pool ci-training-h100 +# for sm_90 / 80 GB. +# +# Usage: +# ./scripts/argo-lob-sweep.sh --grid config/ml/sweep_decision_stride_example.yaml +# ./scripts/argo-lob-sweep.sh --grid --sha abc1234 --gpu-pool ci-training-h100 +# ./scripts/argo-lob-sweep.sh --grid --watch +# ./scripts/argo-lob-sweep.sh --grid --dry-run > /tmp/wf.yaml + +apiVersion: argoproj.io/v1alpha1 +kind: WorkflowTemplate +metadata: + name: lob-backtest-sweep + namespace: foxhunt + labels: + app.kubernetes.io/name: lob-backtest-sweep + app.kubernetes.io/part-of: foxhunt + app.kubernetes.io/component: backtest +spec: + entrypoint: sweep-matrix + onExit: notify-result + serviceAccountName: argo-workflow + archiveLogs: true + podMetadata: + labels: + app.kubernetes.io/part-of: foxhunt + app.kubernetes.io/component: backtest + securityContext: + fsGroup: 0 + ttlStrategy: + secondsAfterCompletion: 3600 + # LOB sweep cells are much faster than training (~minutes vs hours); + # 2h walltime is generous even for N=128 cells. + activeDeadlineSeconds: 7200 + + arguments: + parameters: + - name: commit-sha + value: HEAD + - name: git-branch + value: main + - name: cuda-compute-cap + value: "89" + - name: gpu-pool + value: ci-training-l40s + - name: data-root + value: /mnt/training-data/futures-baseline/ES.FUT + - name: predecoded-dir + value: /mnt/training-data/futures-baseline/ES.FUT + - name: checkpoint + value: "" # empty = trunk runs with --seed (random init); + # set to a path on the data PVC for trained weights + - name: sweep-root + value: /mnt/training-data/sweeps/lob-backtest + - name: sweep-tag + value: default # subdirectory: // + # Operational knobs. argo-lob-sweep.sh sets these per-grid. + - name: n-parallel + value: "1" + - name: latency-ns + value: "100000000" + - name: target-annual-vol-units + value: "50.0" + - name: annualisation-factor + value: "825.0" + - name: max-lots + value: "5" + - name: max-events + value: "0" + + volumes: + - name: training-data + persistentVolumeClaim: + claimName: training-data + - name: git-ssh-key + secret: + secretName: git-ssh-key + defaultMode: 0400 + - name: cargo-target-cuda + persistentVolumeClaim: + claimName: cargo-target-cuda + + templates: + # ── DAG entrypoint ────────────────────────────────────────────── + - name: sweep-matrix + dag: + tasks: + - name: ensure-binary + template: ensure-binary + # __SWEEP_CELLS__ — replaced by argo-lob-sweep.sh + - name: aggregate + template: aggregate + dependencies: [ensure-binary] # plus all run-cell-* via shell-injected list + arguments: + parameters: + - name: sha + value: "{{tasks.ensure-binary.outputs.parameters.sha}}" + + # ── ensure-binary: cache-or-compile fxt-backtest by commit SHA ─ + # Same shape as train-multi-seed-template.yaml's ensure-binary but + # for the fxt-backtest binary only. + - name: ensure-binary + outputs: + parameters: + - name: sha + valueFrom: + path: /tmp/sha + nodeSelector: + k8s.scaleway.com/pool-name: ci-compile-cpu + topology.kubernetes.io/zone: fr-par-2 + tolerations: + - key: node.cilium.io/agent-not-ready + operator: Exists + effect: NoSchedule + container: + image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/ci-builder:latest + imagePullPolicy: Always + command: ["/bin/bash", "-c"] + env: + - name: SQLX_OFFLINE + value: "true" + - name: CARGO_TARGET_DIR + value: /cargo-target + - name: CARGO_HOME + value: /cargo-target/cargo-home + - name: RUSTC_WRAPPER + value: sccache + - name: SCCACHE_DIR + value: /cargo-target/sccache + - name: SCCACHE_CACHE_SIZE + value: "40G" + - name: CARGO_INCREMENTAL + value: "0" + - name: FOXHUNT_CUDA_ARCH + value: "sm_{{workflow.parameters.cuda-compute-cap}}" + - name: GITLAB_PAT + valueFrom: + secretKeyRef: + name: gitlab-pat + key: token + resources: + requests: + cpu: "14" + memory: 32Gi + limits: + cpu: "30" + memory: 64Gi + volumeMounts: + - name: git-ssh-key + mountPath: /etc/git-ssh + readOnly: true + - name: cargo-target-cuda + mountPath: /cargo-target + - name: training-data + mountPath: /mnt/training-data + args: + - | + set -e + SHA="{{workflow.parameters.commit-sha}}" + BRANCH="{{workflow.parameters.git-branch}}" + + mkdir -p ~/.ssh + cp /etc/git-ssh/ssh-privatekey ~/.ssh/id_ed25519 + chmod 600 ~/.ssh/id_ed25519 + printf 'Host *\n StrictHostKeyChecking no\n UserKnownHostsFile /dev/null\n' > ~/.ssh/config + + REPO="ssh://git@gitlab-gitlab-shell.foxhunt.svc.cluster.local:2222/root/foxhunt.git" + BUILD="/cargo-target/src" + git config --global --add safe.directory "$BUILD" + + if [ "$SHA" = "HEAD" ]; then + if [ -d "$BUILD/.git" ]; then + cd "$BUILD"; git fetch origin + SHA=$(git rev-parse "origin/$BRANCH"); cd / + else + git clone --filter=blob:none "$REPO" "$BUILD" + cd "$BUILD"; git checkout "origin/$BRANCH" + SHA=$(git rev-parse HEAD); cd / + fi + fi + SHORT_SHA=$(echo "$SHA" | cut -c1-9) + echo "Resolved SHA: $SHA (short: $SHORT_SHA)" + + BIN_DIR="/mnt/training-data/bin/$SHORT_SHA" + mkdir -p "$BIN_DIR" + if [ -x "$BIN_DIR/fxt-backtest" ]; then + echo "=== Cache HIT: fxt-backtest present in $BIN_DIR ===" + ls -lh "$BIN_DIR/fxt-backtest" + echo "$SHORT_SHA" > /tmp/sha + exit 0 + fi + + echo "=== Cache MISS: compiling fxt-backtest for $SHORT_SHA ===" + if [ -d "$BUILD/.git" ]; then + cd "$BUILD"; git fetch origin; git checkout "$SHA" + else + git clone --filter=blob:none "$REPO" "$BUILD" + cd "$BUILD"; git checkout "$SHA" + fi + + cargo build -p fxt-backtest --release + cp "$BUILD/target/release/fxt-backtest" "$BIN_DIR/fxt-backtest" + echo "$SHORT_SHA" > /tmp/sha + ls -lh "$BIN_DIR/fxt-backtest" + + # ── run-cell: one sweep cell against a GPU pod ───────────────── + # Inputs: cell name + every Run arg that varies across the grid. + # The shell-rendered DAG tasks supply these per-cell. + - name: run-cell + inputs: + parameters: + - name: sha + - name: cell-name + - name: decision-stride + value: "4" + - name: latency-ns + value: "{{workflow.parameters.latency-ns}}" + - name: target-annual-vol-units + value: "{{workflow.parameters.target-annual-vol-units}}" + - name: annualisation-factor + value: "{{workflow.parameters.annualisation-factor}}" + - name: max-lots + value: "{{workflow.parameters.max-lots}}" + - name: max-events + value: "{{workflow.parameters.max-events}}" + - name: seed + value: "0xC0FFEE" + nodeSelector: + k8s.scaleway.com/pool-name: "{{workflow.parameters.gpu-pool}}" + topology.kubernetes.io/zone: fr-par-2 + tolerations: + - key: nvidia.com/gpu + operator: Exists + effect: NoSchedule + container: + image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/training-runtime:latest + imagePullPolicy: Always + command: ["/bin/bash", "-c"] + resources: + requests: + cpu: "4" + memory: 16Gi + nvidia.com/gpu: "1" + limits: + cpu: "8" + memory: 32Gi + nvidia.com/gpu: "1" + volumeMounts: + - name: training-data + mountPath: /mnt/training-data + args: + - | + set -e + SHA="{{inputs.parameters.sha}}" + CELL="{{inputs.parameters.cell-name}}" + BIN="/mnt/training-data/bin/$SHA/fxt-backtest" + CKPT="{{workflow.parameters.checkpoint}}" + OUT="{{workflow.parameters.sweep-root}}/{{workflow.parameters.sweep-tag}}/$CELL" + mkdir -p "$OUT" + echo "=== sweep cell $CELL on $(hostname) → $OUT ===" + + CKPT_FLAG="" + if [ -n "$CKPT" ]; then + CKPT_FLAG="--checkpoint $CKPT" + fi + + "$BIN" run \ + --data "{{workflow.parameters.data-root}}" \ + --predecoded-dir "{{workflow.parameters.predecoded-dir}}" \ + --n-parallel "{{workflow.parameters.n-parallel}}" \ + --decision-stride "{{inputs.parameters.decision-stride}}" \ + --latency-ns "{{inputs.parameters.latency-ns}}" \ + --target-annual-vol-units "{{inputs.parameters.target-annual-vol-units}}" \ + --annualisation-factor "{{inputs.parameters.annualisation-factor}}" \ + --max-lots "{{inputs.parameters.max-lots}}" \ + --max-events "{{inputs.parameters.max-events}}" \ + --seed "{{inputs.parameters.seed}}" \ + $CKPT_FLAG \ + --out "$OUT" + + echo "=== cell $CELL done ===" + ls -lh "$OUT" + + # ── aggregate: final CPU pod runs fxt-backtest aggregate ─────── + - name: aggregate + inputs: + parameters: + - name: sha + nodeSelector: + k8s.scaleway.com/pool-name: ci-compile-cpu + topology.kubernetes.io/zone: fr-par-2 + tolerations: + - key: node.cilium.io/agent-not-ready + operator: Exists + effect: NoSchedule + container: + image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/training-runtime:latest + imagePullPolicy: Always + command: ["/bin/bash", "-c"] + resources: + requests: + cpu: "2" + memory: 4Gi + limits: + cpu: "4" + memory: 8Gi + volumeMounts: + - name: training-data + mountPath: /mnt/training-data + args: + - | + set -e + SHA="{{inputs.parameters.sha}}" + BIN="/mnt/training-data/bin/$SHA/fxt-backtest" + SWEEP_DIR="{{workflow.parameters.sweep-root}}/{{workflow.parameters.sweep-tag}}" + echo "=== aggregate sweep at $SWEEP_DIR ===" + "$BIN" aggregate "$SWEEP_DIR" + ls -lh "$SWEEP_DIR" + + # ── notify-result: exit hook (placeholder; real impl emits to Slack/MinIO) ─ + - name: notify-result + container: + image: alpine:3.20 + command: ["sh", "-c"] + args: + - | + echo "lob-backtest-sweep workflow {{workflow.name}} finished with status {{workflow.status}}" + echo "results under {{workflow.parameters.sweep-root}}/{{workflow.parameters.sweep-tag}}" diff --git a/scripts/argo-lob-sweep.sh b/scripts/argo-lob-sweep.sh new file mode 100755 index 000000000..0998fb3e5 --- /dev/null +++ b/scripts/argo-lob-sweep.sh @@ -0,0 +1,251 @@ +#!/usr/bin/env bash +# Run a real-LOB backtest sweep via Argo Workflows. +# +# Reads the same `fxt-backtest sweep` grid YAML as the single-machine +# binary and fans each cell out to its own GPU pod. Mirrors the +# argo-train.sh + train-multi-seed-template.yaml conventions: +# `# __SWEEP_CELLS__` marker in the template is replaced with one +# WorkflowTask stanza per cell before submission. +# +# Usage: +# ./scripts/argo-lob-sweep.sh --grid config/ml/sweep_decision_stride_example.yaml +# ./scripts/argo-lob-sweep.sh --grid --sha abc1234 +# ./scripts/argo-lob-sweep.sh --grid --gpu-pool ci-training-h100 +# ./scripts/argo-lob-sweep.sh --grid --watch +# ./scripts/argo-lob-sweep.sh --grid --dry-run > /tmp/wf.yaml +# +# Requires: argo CLI, kubectl, python3 (with PyYAML — universally +# available in foxhunt CI images). + +set -euo pipefail + +GRID="" +SHA="HEAD" +BRANCH="main" +GPU_POOL="" +SWEEP_TAG="" +WATCH=false +DRY_RUN=false +TAG="" + +usage() { + cat < [OPTIONS] + +Options: + --grid REQUIRED. Sweep grid YAML (same format as + \`fxt-backtest sweep --grid\`). + --sha Git commit SHA (default: HEAD). + --branch Git branch (default: main). + --gpu-pool GPU node pool (default: ci-training-l40s; opt in to + ci-training-h100 for sm_90 / 80 GB). + --sweep-tag Subdirectory name under / (default: + -). + --watch Follow workflow logs after submission. + --tag Label workflow with foxhunt-tag=. + --dry-run Print rendered workflow YAML to stdout, do not submit. + -h, --help Show this help. +EOF + exit 0 +} + +while [[ $# -gt 0 ]]; do + case "$1" in + --grid) GRID="$2"; shift 2 ;; + --sha) SHA="$2"; shift 2 ;; + --branch) BRANCH="$2"; shift 2 ;; + --gpu-pool) GPU_POOL="$2"; shift 2 ;; + --sweep-tag) SWEEP_TAG="$2"; shift 2 ;; + --watch) WATCH=true; shift ;; + --tag) TAG="$2"; shift 2 ;; + --dry-run) DRY_RUN=true; shift ;; + -h|--help) usage ;; + *) echo "unknown arg: $1" >&2; usage ;; + esac +done + +if [[ -z "$GRID" ]]; then + echo "ERROR: --grid is required" >&2 + usage +fi +if [[ ! -f "$GRID" ]]; then + echo "ERROR: grid file not found: $GRID" >&2 + exit 1 +fi + +# Default GPU pool follows feedback_default_to_l40s_pool.md (2026-05-09). +if [[ -z "$GPU_POOL" ]]; then + GPU_POOL="ci-training-l40s" +fi +if [[ "$GPU_POOL" == "ci-training-h100" ]]; then + CUDA_COMPUTE_CAP="90" +else + CUDA_COMPUTE_CAP="89" +fi + +# Short SHA for sweep-tag default. +if [[ "$SHA" == "HEAD" ]]; then + RESOLVED_SHA=$(git rev-parse HEAD 2>/dev/null || echo "00000000") +else + RESOLVED_SHA="$SHA" +fi +SHORT_SHA="${RESOLVED_SHA:0:9}" +if [[ -z "$SWEEP_TAG" ]]; then + SWEEP_TAG="$(basename "$GRID" .yaml)-$SHORT_SHA" +fi + +# Single Python invocation parses the grid + emits a JSON object with +# parsed base + cell records + the rendered Argo WorkflowTask stanzas +# for the __SWEEP_CELLS__ marker. Keeps the shell side parser-free. +TPL="$(dirname "$0")/../infra/k8s/argo/lob-backtest-sweep-template.yaml" +if [[ ! -f "$TPL" ]]; then + echo "ERROR: template not found: $TPL" >&2 + exit 1 +fi + +RENDER_PY=$(python3 - "$GRID" <<'PYEOF' +import json, os, sys, yaml + +with open(sys.argv[1], "r") as f: + grid = yaml.safe_load(f) + +base = grid.get("base") or {} +cells = grid.get("cells") or [] +if not cells: + print("ERROR: grid has no cells", file=sys.stderr) + sys.exit(1) + +def base_or(key, default): + return base.get(key, default) + +out = { + "n_cells": len(cells), + "base": { + "data": base_or("data", ""), + "predecoded_dir": base.get("predecoded_dir") or base.get("data", ""), + "n_parallel": base_or("n_parallel", 1), + "latency_ns": base_or("latency_ns", 100_000_000), + "target_annual_vol_units": base_or("target_annual_vol_units", 50.0), + "annualisation_factor": base_or("annualisation_factor", 825.0), + "max_lots": base_or("max_lots", 5), + "max_events": base_or("max_events", 0), + "seed": base_or("seed", 0xC0FFEE), + "checkpoint": base.get("checkpoint") or "", + }, +} + +# Render per-cell WorkflowTask stanzas + collect dependency names. +tasks = [] +deps = [] +for c in cells: + name = c.get("name") + if not name: + print("ERROR: cell missing 'name'", file=sys.stderr) + sys.exit(1) + stride = c.get("decision_stride", base_or("decision_stride", 4)) + latency = c.get("latency_ns", out["base"]["latency_ns"]) + target_vol = c.get("target_annual_vol_units", out["base"]["target_annual_vol_units"]) + ann_factor = c.get("annualisation_factor", out["base"]["annualisation_factor"]) + max_lots = c.get("max_lots", out["base"]["max_lots"]) + max_events = c.get("max_events", out["base"]["max_events"]) + seed = c.get("seed", out["base"]["seed"]) + + tasks.append(f''' + - name: run-cell-{name} + template: run-cell + dependencies: [ensure-binary] + arguments: + parameters: + - name: sha + value: "{{{{tasks.ensure-binary.outputs.parameters.sha}}}}" + - name: cell-name + value: "{name}" + - name: decision-stride + value: "{stride}" + - name: latency-ns + value: "{latency}" + - name: target-annual-vol-units + value: "{target_vol}" + - name: annualisation-factor + value: "{ann_factor}" + - name: max-lots + value: "{max_lots}" + - name: max-events + value: "{max_events}" + - name: seed + value: "{seed}"''') + deps.append(f"run-cell-{name}") + +out["cell_tasks"] = "".join(tasks) +out["aggregate_deps"] = ", ".join(["ensure-binary", *deps]) +out["short_sha"] = os.environ.get("SHORT_SHA", "") +out["sweep_tag"] = os.environ.get("SWEEP_TAG", "") +print(json.dumps(out)) +PYEOF +) + +export SHORT_SHA SWEEP_TAG + +# Extract render outputs via python -c (no jq dependency). +N_CELLS=$(echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["n_cells"])') +BASE_DATA=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["data"])') +BASE_PREDECODED=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["predecoded_dir"])') +BASE_N_PARALLEL=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["n_parallel"])') +BASE_LATENCY_NS=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["latency_ns"])') +BASE_TARGET_VOL=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["target_annual_vol_units"])') +BASE_ANN_FACTOR=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["annualisation_factor"])') +BASE_MAX_LOTS=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["max_lots"])') +BASE_MAX_EVENTS=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["max_events"])') +BASE_CHECKPOINT=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["base"]["checkpoint"])') +CELL_TASKS=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["cell_tasks"])') +AGG_DEPS=$( echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["aggregate_deps"])') + +# Render template: replace # __SWEEP_CELLS__ marker AND inject all +# per-cell tasks into the aggregate's `dependencies: [...]` list. +RENDERED=$(awk -v cells="$CELL_TASKS" -v deps="$AGG_DEPS" ' + /# __SWEEP_CELLS__/ { print cells; next } + /dependencies: \[ensure-binary\] # plus all run-cell-/ { + sub(/\[ensure-binary\]/, "[" deps "]") + } + { print } +' "$TPL") + +# Build the submission command. +SUBMIT_ARGS=( + --from "workflowtemplate/lob-backtest-sweep" + -p "commit-sha=$SHA" + -p "git-branch=$BRANCH" + -p "cuda-compute-cap=$CUDA_COMPUTE_CAP" + -p "gpu-pool=$GPU_POOL" + -p "data-root=$BASE_DATA" + -p "predecoded-dir=$BASE_PREDECODED" + -p "n-parallel=$BASE_N_PARALLEL" + -p "latency-ns=$BASE_LATENCY_NS" + -p "target-annual-vol-units=$BASE_TARGET_VOL" + -p "annualisation-factor=$BASE_ANN_FACTOR" + -p "max-lots=$BASE_MAX_LOTS" + -p "max-events=$BASE_MAX_EVENTS" + -p "sweep-tag=$SWEEP_TAG" +) +if [[ -n "$BASE_CHECKPOINT" ]]; then + SUBMIT_ARGS+=(-p "checkpoint=$BASE_CHECKPOINT") +fi +if [[ -n "$TAG" ]]; then + SUBMIT_ARGS+=(-l "foxhunt-tag=$TAG") +fi +if [[ "$WATCH" == "true" ]]; then + SUBMIT_ARGS+=(--watch) +fi + +if [[ "$DRY_RUN" == "true" ]]; then + echo "# === Rendered workflow (cells: $N_CELLS, sweep-tag: $SWEEP_TAG) ===" + echo "$RENDERED" + echo "" + echo "# === Would submit with: ===" + echo "# argo submit ${SUBMIT_ARGS[*]}" + exit 0 +fi + +# Apply the rendered template (covers any in-flight template edits) then submit. +echo "$RENDERED" | kubectl apply -f - -n foxhunt +argo submit -n foxhunt "${SUBMIT_ARGS[@]}"