From ba2850b448159b2e38a47fe064f1e9bc337c686a Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 19 May 2026 18:15:22 +0200 Subject: [PATCH] feat(argo): run-sweep template + batched-mode in argo-lob-sweep.sh MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the operational gap from P6: when a sweep grid YAML carries sim_variants, argo-lob-sweep.sh now emits ONE run-sweep-batched task instead of N fan-out run-cell tasks. The new run-sweep template invokes `fxt-backtest sweep` against the full grid (base64-encoded inline as Argo parameter to avoid YAML special-char encoding traps). The binary's P6 sweep() function handles cell × variant fan-out internally via BatchedSimConfig::from_grid, so one pod processes all 4 windows × 140 variants sequentially. Trade-off: no inter-window parallelism in this rev (4 quarters sequential in one pod ≈ firm-bound 2h wall per spec §9). 3-pod scale-out is P7 future work — needs the script to split the YAML into per-window sub-grids and emit one run-sweep task per sub-grid. Backward-compat: legacy (no sim_variants) flow unchanged. The dry-run of sweep_smoke.yaml continues to emit run-cell-* fan-out tasks; the dry-run of sweep_deployability.yaml emits one run-sweep-batched task. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../k8s/argo/lob-backtest-sweep-template.yaml | 58 +++++++++++++++ scripts/argo-lob-sweep.sh | 74 ++++++++++++++----- 2 files changed, 114 insertions(+), 18 deletions(-) diff --git a/infra/k8s/argo/lob-backtest-sweep-template.yaml b/infra/k8s/argo/lob-backtest-sweep-template.yaml index 6e096c7e0..b571b2cb0 100644 --- a/infra/k8s/argo/lob-backtest-sweep-template.yaml +++ b/infra/k8s/argo/lob-backtest-sweep-template.yaml @@ -307,6 +307,64 @@ spec: echo "=== cell $CELL done ===" ls -lh "$OUT" + # ── run-sweep: P6 batched flow. Invokes `fxt-backtest sweep` ─── + # against the FULL grid YAML inside a single GPU pod. When the YAML + # carries sim_variants, each cell expands into n_parallel=variants.len() + # backtests sharing one forward pass. argo-lob-sweep.sh emits ONE + # run-sweep task (instead of N fan-out run-cell tasks) when --batched + # is set or sim_variants is detected in the YAML. + - name: run-sweep + inputs: + parameters: + - name: sha + - name: grid-yaml-b64 # base64-encoded grid YAML, written to a file in-pod + 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/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 + - name: feature-cache + mountPath: /feature-cache + args: + - | + set -e + SHA="{{inputs.parameters.sha}}" + BIN="/mnt/training-data/bin/$SHA/fxt-backtest" + OUT="{{workflow.parameters.sweep-root}}/{{workflow.parameters.sweep-tag}}" + mkdir -p "$OUT" + # Write the grid YAML to a known path. base64 in/out keeps Argo + # parameter encoding stable across YAML special characters + # (curly braces from data_template + sim_variants). + GRID_YAML=/tmp/sweep-grid.yaml + echo "{{inputs.parameters.grid-yaml-b64}}" | base64 -d > "$GRID_YAML" + echo "=== running fxt-backtest sweep on $(hostname) → $OUT ===" + echo "=== grid yaml: ===" + head -20 "$GRID_YAML" + echo "=== ... ===" + + "$BIN" sweep --grid "$GRID_YAML" --out "$OUT" + + echo "=== sweep complete ===" + ls -lh "$OUT" + # ── aggregate: runs fxt-backtest aggregate on the GPU pool. ─── # The binary is dynamically linked against libcuda.so.1, which the # ci-compile-cpu pool's host doesn't expose; the aggregate workload diff --git a/scripts/argo-lob-sweep.sh b/scripts/argo-lob-sweep.sh index 2e000e5ae..cb3d621b3 100755 --- a/scripts/argo-lob-sweep.sh +++ b/scripts/argo-lob-sweep.sh @@ -104,10 +104,11 @@ if [[ ! -f "$TPL" ]]; then fi RENDER_PY=$(python3 - "$GRID" <<'PYEOF' -import json, os, sys, yaml +import base64, json, os, sys, yaml with open(sys.argv[1], "r") as f: - grid = yaml.safe_load(f) + raw_yaml = f.read() + grid = yaml.safe_load(raw_yaml) base = grid.get("base") or {} cells = grid.get("cells") or [] @@ -118,11 +119,23 @@ if not cells: def base_or(key, default): return base.get(key, default) +# P6 batched-flow detection: when base.sim_variants is non-empty, the +# binary handles cell+variant fan-out internally via BatchedSimConfig::from_grid. +# We emit ONE Argo `run-sweep` task that invokes `fxt-backtest sweep` +# against the full grid. For inter-window parallelism (3-pod scale, +# P7-future), split the YAML into per-window sub-grids and emit one +# run-sweep task per sub-grid. +sim_variants = base.get("sim_variants") or [] +batched = len(sim_variants) > 0 + out = { "n_cells": len(cells), + "batched": batched, "base": { - "data": base_or("data", ""), - "predecoded_dir": base.get("predecoded_dir") or base.get("data", ""), + # `data` may be absent under the batched flow (replaced by + # `data_template`); legacy path uses scalar `data`. + "data": base_or("data", "") or "", + "predecoded_dir": base.get("predecoded_dir") or base.get("data") or "", "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), @@ -137,20 +150,38 @@ out = { # 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"]) - +if batched: + # Single run-sweep task: binary processes all cells × variants in one pod. + # base64-encode the full grid YAML so Argo parameter encoding doesn't + # trip on YAML-special characters (data_template's `{window}`, + # sim_variants' braces, comments, etc.). + grid_b64 = base64.b64encode(raw_yaml.encode("utf-8")).decode("ascii") tasks.append(f''' + - name: run-sweep-batched + template: run-sweep + dependencies: [ensure-binary] + arguments: + parameters: + - name: sha + value: "{{{{tasks.ensure-binary.outputs.parameters.sha}}}}" + - name: grid-yaml-b64 + value: "{grid_b64}"''') + deps.append("run-sweep-batched") +else: + 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] @@ -174,7 +205,7 @@ for c in cells: value: "{max_events}" - name: seed value: "{seed}"''') - deps.append(f"run-cell-{name}") + deps.append(f"run-cell-{name}") out["cell_tasks"] = "".join(tasks) out["aggregate_deps"] = ", ".join(["ensure-binary", *deps]) @@ -188,6 +219,7 @@ 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"])') +BATCHED=$(echo "$RENDER_PY" | python3 -c 'import json,sys;print(json.load(sys.stdin)["batched"])') 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"])') @@ -249,6 +281,12 @@ if [[ "$DRY_RUN" == "true" ]]; then exit 0 fi +if [[ "$BATCHED" == "True" ]]; then + echo " mode: batched (single run-sweep task, sim_variants packed inside the pod)" +else + echo " mode: legacy fan-out ($N_CELLS run-cell tasks)" +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[@]}"