feat(argo): run-sweep template + batched-mode in argo-lob-sweep.sh
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) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -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
|
||||
|
||||
@@ -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[@]}"
|
||||
|
||||
Reference in New Issue
Block a user