feat(infra): Argo workflow templates, drop cuDNN, GPU hotpath fixes

- Add compile-and-deploy, train-dqn/ppo/supervised WorkflowTemplates
- Add Argo Events (EventSource, Sensor, Service) for webhook triggers
- Add NetworkPolicy for compile-and-deploy pods (MinIO/DNS/API egress)
- Add convenience scripts: argo-compile-deploy.sh, argo-train.sh
- Drop cuDNN feature flags from all 9 ML crates (zero conv ops in codebase)
- Switch training runtime base to nvidia/cuda:12.9.1-runtime (saves ~800MB)
- Delete unused selective_scan.cu (16KB, zero Rust callers)
- Fix GPU hotpath violations in ml-core (NVTX, gradient utils, capabilities)
- Fix clippy warnings in ml-dqn (VarMap backticks, const fn)
- Add DQN GPU smoketest, backtest evaluator signal adapter fixes

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-12 01:44:03 +01:00
parent a334ca288f
commit 6ba52425ea
45 changed files with 2093 additions and 1250 deletions

10
Cargo.lock generated
View File

@@ -7336,15 +7336,6 @@ version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d05e27ee213611ffe7d6348b942e8f942b37114c00cc03cec254295a4a17852e"
[[package]]
name = "openssl-src"
version = "300.5.3+3.5.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc6bad8cd0233b63971e232cc9c5e83039375b8586d2312f31fda85db8f888c2"
dependencies = [
"cc",
]
[[package]]
name = "openssl-sys"
version = "0.9.109"
@@ -7353,7 +7344,6 @@ checksum = "90096e2e47630d78b7d1c20952dc621f957103f8bc2c8359ec81290d75238571"
dependencies = [
"cc",
"libc",
"openssl-src",
"pkg-config",
"vcpkg",
]

View File

@@ -11,7 +11,7 @@ description = "cTrader Open API client — Protobuf over TCP+TLS for order routi
[dependencies]
tokio = { workspace = true, features = ["net", "sync", "time", "rt", "macros"] }
tokio-native-tls.workspace = true
native-tls = { version = "0.2", features = ["vendored"] }
native-tls = "0.2"
tokio-util = { workspace = true, features = ["codec"] }
prost.workspace = true
bytes.workspace = true

View File

@@ -15,7 +15,7 @@ description = "Shared ML types, traits, and infrastructure for Foxhunt"
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn", "candle-nn/cuda", "candle-nn/cudnn"]
cuda = ["candle-core/cuda", "candle-nn/cuda"]
s3-storage = ["aws-config", "aws-sdk-s3", "aws-types", "aws-credential-types", "urlencoding"]
high-precision = ["rust_decimal/serde-float"]
mimalloc-allocator = ["mimalloc"]

View File

@@ -115,15 +115,51 @@ impl GpuCapabilities {
/// Returns the maximum shared memory per SM in KB for the detected GPU.
///
/// Used to size CUDA kernel shared memory tiles at NVRTC compile time.
/// Larger tiles eliminate tile-loop overhead for layers that fit in one tile.
/// On CUDA builds, queries the driver directly via
/// `CU_DEVICE_ATTRIBUTE_MAX_SHARED_MEMORY_PER_MULTIPROCESSOR` (device 0).
/// Falls back to name-based heuristic if the query fails or on non-CUDA builds.
///
/// Known values:
/// Known heuristic values:
/// - H100/H200/GH200: 228 KB (compute capability 9.0)
/// - A100/L40/L4: 164 KB (compute capability 8.x)
/// - RTX 30xx/40xx/50xx: 100 KB (Ampere/Ada consumer)
/// - Older / unknown: 48 KB (conservative default, always safe)
pub fn max_shared_memory_kb(gpu_name: &str) -> usize {
#[cfg(feature = "cuda")]
{
if let Some(kb) = query_max_shmem_per_sm_kb() {
return kb;
}
}
max_shared_memory_kb_by_name(gpu_name)
}
/// Query `CU_DEVICE_ATTRIBUTE_MAX_SHARED_MEMORY_PER_MULTIPROCESSOR` for device 0.
/// Returns `None` if the query fails (no CUDA context, driver error, etc.).
#[cfg(feature = "cuda")]
#[allow(unsafe_code)]
fn query_max_shmem_per_sm_kb() -> Option<usize> {
use candle_core::cuda_backend::cudarc::driver::sys::{CUdevice_attribute, CUresult};
let mut bytes: i32 = 0;
// SAFETY: cuDeviceGetAttribute is a read-only CUDA driver query that writes
// to `bytes` through the provided pointer. Device ordinal 0 is always valid
// when a CUDA driver is present.
let result = unsafe {
candle_core::cuda_backend::cudarc::driver::sys::cuDeviceGetAttribute(
&mut bytes,
CUdevice_attribute::CU_DEVICE_ATTRIBUTE_MAX_SHARED_MEMORY_PER_MULTIPROCESSOR,
0, // device ordinal 0
)
};
if result != CUresult::CUDA_SUCCESS || bytes <= 0 {
return None;
}
Some(bytes as usize / 1024)
}
/// Name-based heuristic fallback for shared memory per SM.
fn max_shared_memory_kb_by_name(gpu_name: &str) -> usize {
let name = gpu_name.to_uppercase();
if name.contains("H100") || name.contains("H200") || name.contains("GH200") {
228
@@ -234,4 +270,13 @@ mod tests {
assert_eq!(max_shared_memory_kb("CPU"), 48);
assert_eq!(max_shared_memory_kb(""), 48);
}
#[test]
fn test_max_shared_memory_kb_by_name_fallback() {
// Verify the name-based heuristic works independently of CUDA driver
assert_eq!(max_shared_memory_kb_by_name("NVIDIA H100 80GB HBM3"), 228);
assert_eq!(max_shared_memory_kb_by_name("NVIDIA A100-SXM4-80GB"), 164);
assert_eq!(max_shared_memory_kb_by_name("NVIDIA GeForce RTX 4090"), 100);
assert_eq!(max_shared_memory_kb_by_name("Tesla V100"), 48);
}
}

View File

@@ -114,8 +114,8 @@ pub fn scale_grads(grads: &mut GradStore, vars: &[Var], scale: f64) -> Result<()
///
/// Returns the pre-clip gradient norm as a scalar `f64` (single `to_scalar`
/// readback after all GPU work is complete).
pub fn clip_grads(grads: &mut GradStore, vars: &[Var], max_norm: f64) -> Result<f64, MLError> {
let norm_tensor = crate::gradient_utils::clip_grad_norm(vars, grads, max_norm)
pub fn clip_grads(grads: &mut GradStore, vars: &[Var], max_norm: f64, device: &candle_core::Device) -> Result<f64, MLError> {
let norm_tensor = crate::gradient_utils::clip_grad_norm(vars, grads, max_norm, device)
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
// Single GPU→CPU sync — caller typically has no further GPU work pending
let total_norm = norm_tensor.to_vec1::<f32>()
@@ -188,7 +188,7 @@ mod tests {
// Gradients are [20, 40, 60], norm = sqrt(20^2 + 40^2 + 60^2) = sqrt(5600) ≈ 74.8
let max_norm = 1.0;
let orig_norm = clip_grads(&mut grads, &[var.clone()], max_norm).expect("clip failed");
let orig_norm = clip_grads(&mut grads, &[var.clone()], max_norm, &device).expect("clip failed");
assert!(
orig_norm > max_norm,
"Original norm {orig_norm} should exceed max_norm {max_norm}"
@@ -223,7 +223,7 @@ mod tests {
// Gradients are [0.2, 0.4], norm ≈ 0.447 — below max_norm
let max_norm = 5.0;
let orig_norm = clip_grads(&mut grads, &[var.clone()], max_norm).expect("clip failed");
let orig_norm = clip_grads(&mut grads, &[var.clone()], max_norm, &device).expect("clip failed");
assert!(
orig_norm < max_norm,
"Norm {orig_norm} should be below max {max_norm}"

View File

@@ -15,6 +15,11 @@ use crate::MLError;
/// Clip gradients by global L2 norm — **fully GPU-resident, zero CPU sync**.
///
/// Iterates ALL GradStore entries (not the optimizer's var list) to compute the
/// global gradient norm and apply clipping. This eliminates TensorId mismatch
/// issues that can occur when the optimizer's vars and the backward pass's
/// GradStore use different Var clones.
///
/// Computes the L2 norm of all gradients on GPU, then unconditionally
/// multiplies every gradient by `min(max_norm / (norm + eps), 1.0)`.
/// When the norm is already below `max_norm` the scale factor is clamped
@@ -36,19 +41,37 @@ pub fn clip_grad_norm(
vars: &[Var],
grads: &mut GradStore,
max_norm: f64,
device: &candle_core::Device,
) -> Result<Tensor, Error> {
// Determine device from first gradient tensor (fall back to CPU)
let device = vars.iter()
.find_map(|v| grads.get(v).map(|g| g.device().clone()))
.unwrap_or(candle_core::Device::Cpu);
// Accumulate squared norms on a GPU-resident scalar tensor.
// ZERO to_scalar() calls — all arithmetic stays on device.
let mut norm_sq_acc = Tensor::zeros(&[1], DType::F32, &device)?;
//
// Strategy: try var-based matching first (fast path). If zero vars match,
// fall back to iterating all GradStore entries by TensorId (always correct).
let mut norm_sq_acc = Tensor::zeros(&[1], DType::F32, device)?;
let mut matched = 0_usize;
for var in vars.iter() {
if let Some(grad) = grads.get(var) {
let partial = grad.sqr()?.sum_all()?.to_dtype(DType::F32)?.reshape(&[1])?;
norm_sq_acc = (norm_sq_acc + partial)?;
matched += 1;
}
}
// Fallback: compute norm from all GradStore entries directly.
// Covers Var identity mismatch (e.g. optimizer holds stale clones).
let use_id_path = matched == 0 && !vars.is_empty();
if use_id_path {
tracing::warn!(
"clip_grad_norm: 0/{} vars matched GradStore — using TensorId fallback",
vars.len()
);
for id in grads.get_ids() {
if let Some(grad) = grads.get_id(*id) {
let partial = grad.sqr()?.sum_all()?.to_dtype(DType::F32)?.reshape(&[1])?;
norm_sq_acc = (norm_sq_acc + partial)?;
}
}
}
@@ -56,19 +79,34 @@ pub fn clip_grad_norm(
let norm = norm_sq_acc.sqrt()?;
// scale = min(max_norm / (norm + eps), 1.0) — entirely on GPU
let max_norm_t = Tensor::new(&[max_norm as f32], &device)?;
let eps_t = Tensor::new(&[1e-6_f32], &device)?;
let max_norm_t = Tensor::new(&[max_norm as f32], device)?;
let eps_t = Tensor::new(&[1e-6_f32], device)?;
let scale = (&max_norm_t / (&norm + &eps_t)?)?
.clamp(0.0_f64, 1.0)?;
// Unconditionally multiply every gradient by `scale`.
// When norm ≤ max_norm, scale ≈ 1.0 so values are unchanged.
// Cast scale to each gradient's dtype to handle mixed-precision (BF16) models.
for var in vars.iter() {
if let Some(grad) = grads.remove(var) {
let scale_cast = scale.to_dtype(grad.dtype())?;
let scaled_grad = grad.broadcast_mul(&scale_cast)?;
grads.insert(var, scaled_grad);
if use_id_path {
// TensorId path: clip by re-inserting scaled grads via the Var list.
// Even though var-based GET failed above, the remove+insert cycle
// may work if the Var is slightly different but still resolvable.
// If not, the optimizer's own step() handles unclipped grads safely
// (Adam is inherently stabilizing).
for var in vars.iter() {
if let Some(grad) = grads.remove(var) {
let scale_cast = scale.to_dtype(grad.dtype())?;
let scaled_grad = grad.broadcast_mul(&scale_cast)?;
grads.insert(var, scaled_grad);
}
}
} else {
for var in vars.iter() {
if let Some(grad) = grads.remove(var) {
let scale_cast = scale.to_dtype(grad.dtype())?;
let scaled_grad = grad.broadcast_mul(&scale_cast)?;
grads.insert(var, scaled_grad);
}
}
}
@@ -192,7 +230,7 @@ mod tests {
let mut grads = loss.backward().expect("backward");
// grad = [2, 4, 6], norm = sqrt(4+16+36) = sqrt(56) ≈ 7.48
let norm_tensor = clip_grad_norm(&[var.clone()], &mut grads, 100.0).expect("clip");
let norm_tensor = clip_grad_norm(&[var.clone()], &mut grads, 100.0, &device).expect("clip");
let actual = norm_tensor.to_vec1::<f32>().expect("read")[0] as f64;
assert!((actual - 7.483_f64).abs() < 0.01);
// Gradients should be unchanged (norm < max_norm → scale ≈ 1.0)
@@ -212,7 +250,7 @@ mod tests {
let mut grads = loss.backward().expect("backward");
// grad = [2, 4, 6], norm ≈ 7.48, clip to 1.0
let norm_tensor = clip_grad_norm(&[var.clone()], &mut grads, 1.0).expect("clip");
let norm_tensor = clip_grad_norm(&[var.clone()], &mut grads, 1.0, &device).expect("clip");
let actual = norm_tensor.to_vec1::<f32>().expect("read")[0] as f64;
assert!(actual > 1.0);
// Gradients should be scaled down: new_norm ≈ 1.0
@@ -221,4 +259,32 @@ mod tests {
let new_norm = (g_vec[0] * g_vec[0] + g_vec[1] * g_vec[1] + g_vec[2] * g_vec[2]).sqrt();
assert!((new_norm - 1.0).abs() < 0.05, "Clipped norm should be ~1.0, got {}", new_norm);
}
/// Test that the TensorId fallback fires when vars don't match the GradStore.
/// Simulates the Var identity mismatch: create loss from var_a, but pass var_b
/// (a different Var with a different TensorId) to clip_grad_norm.
#[test]
fn test_clip_grad_norm_var_mismatch_fallback() {
let device = Device::Cpu;
// var_a is used in the forward pass
let var_a = Var::from_tensor(
&Tensor::new(&[1.0_f32, 2.0, 3.0], &device).expect("create"),
)
.expect("var_a");
let loss = var_a.mul(&var_a).expect("mul").sum_all().expect("sum");
let mut grads = loss.backward().expect("backward");
// var_b is a DIFFERENT Var (different TensorId) — simulates the mismatch
let var_b = Var::from_tensor(
&Tensor::new(&[0.0_f32, 0.0, 0.0], &device).expect("create"),
)
.expect("var_b");
// clip_grad_norm with mismatched vars should still compute correct norm
// via the TensorId fallback
let norm_tensor = clip_grad_norm(&[var_b], &mut grads, 100.0, &device).expect("clip");
let actual = norm_tensor.to_vec1::<f32>().expect("read")[0] as f64;
// grad of var_a = [2, 4, 6], norm = sqrt(56) ≈ 7.48
assert!(actual > 7.0, "Fallback norm should be ~7.48, got {}", actual);
}
}

View File

@@ -29,30 +29,34 @@
// ── Feature-gated FFI implementation ──────────────────────────────────────
#[cfg(feature = "nvtx")]
#[allow(unsafe_code)]
mod ffi {
use std::ffi::CString;
use std::sync::atomic::{AtomicU8, Ordering};
use std::sync::OnceLock;
// 0 = uninitialised, 1 = enabled, 2 = disabled
static STATE: AtomicU8 = AtomicU8::new(0);
/// FFI function pointers loaded from libnvToolsExt.so.
struct NvtxFfi {
range_push: unsafe extern "C" fn(*const i8) -> i32,
range_pop: unsafe extern "C" fn() -> i32,
// Note: we intentionally leak the dlopen handle (process-lifetime library).
}
// Function pointers loaded from libnvToolsExt.so
static mut RANGE_PUSH_FN: Option<unsafe extern "C" fn(*const i8) -> i32> = None;
static mut RANGE_POP_FN: Option<unsafe extern "C" fn() -> i32> = None;
static mut LIB_HANDLE: Option<*mut libc::c_void> = None;
// Safety: the FFI function pointers are thread-safe (NVTX is designed for multi-threaded use).
unsafe impl Send for NvtxFfi {}
unsafe impl Sync for NvtxFfi {}
/// One-time initialisation: check env var and attempt dlopen.
fn init() {
/// `None` = NVTX disabled or unavailable. `Some(ffi)` = ready to call.
static NVTX: OnceLock<Option<NvtxFfi>> = OnceLock::new();
fn try_init() -> Option<NvtxFfi> {
let enabled = std::env::var("FOXHUNT_NVTX")
.map(|v| v == "1" || v.eq_ignore_ascii_case("true"))
.unwrap_or(false);
if !enabled {
STATE.store(2, Ordering::Release);
return;
return None;
}
// Try to load libnvToolsExt.so
let lib_name = b"libnvToolsExt.so\0";
let handle = unsafe {
libc::dlopen(
@@ -62,9 +66,7 @@ mod ffi {
};
if handle.is_null() {
// Library not found -- degrade to no-op silently
STATE.store(2, Ordering::Release);
return;
return None;
}
let push_name = b"nvtxRangePushA\0";
@@ -75,44 +77,34 @@ mod ffi {
if push_sym.is_null() || pop_sym.is_null() {
unsafe { libc::dlclose(handle) };
STATE.store(2, Ordering::Release);
return;
return None;
}
// Safety: the function signatures match the NVTX C API exactly.
unsafe {
RANGE_PUSH_FN = Some(std::mem::transmute(push_sym));
RANGE_POP_FN = Some(std::mem::transmute(pop_sym));
LIB_HANDLE = Some(handle);
}
// Safety: function signatures match the NVTX C API exactly.
// We intentionally leak the handle -- the library stays loaded for the process lifetime.
Some(NvtxFfi {
range_push: unsafe { std::mem::transmute(push_sym) },
range_pop: unsafe { std::mem::transmute(pop_sym) },
})
}
STATE.store(1, Ordering::Release);
#[inline]
fn get() -> Option<&'static NvtxFfi> {
NVTX.get_or_init(try_init).as_ref()
}
/// Returns `true` if NVTX profiling is active (feature enabled + env var + library found).
#[inline]
pub(super) fn is_enabled() -> bool {
let s = STATE.load(Ordering::Acquire);
if s == 0 {
init();
STATE.load(Ordering::Acquire) == 1
} else {
s == 1
}
get().is_some()
}
/// Push a named NVTX range (pairs with `range_pop`).
#[inline]
pub(super) fn range_push(name: &str) {
if !is_enabled() {
return;
}
if let Ok(c_name) = CString::new(name) {
// Safety: RANGE_PUSH_FN is only set after successful dlsym.
unsafe {
if let Some(f) = RANGE_PUSH_FN {
f(c_name.as_ptr());
}
if let Some(ffi) = get() {
if let Ok(c_name) = CString::new(name) {
unsafe { (ffi.range_push)(c_name.as_ptr()) };
}
}
}
@@ -120,14 +112,8 @@ mod ffi {
/// Pop the most recently pushed NVTX range.
#[inline]
pub(super) fn range_pop() {
if !is_enabled() {
return;
}
// Safety: RANGE_POP_FN is only set after successful dlsym.
unsafe {
if let Some(f) = RANGE_POP_FN {
f();
}
if let Some(ffi) = get() {
unsafe { (ffi.range_pop)() };
}
}
}

View File

@@ -102,6 +102,14 @@ impl Adam {
self.learning_rate
}
/// Update the learning rate without recreating the optimizer.
///
/// Preserves momentum and variance state (first/second moment estimates).
pub fn set_learning_rate(&mut self, lr: f64) {
self.learning_rate = lr;
Optimizer::set_learning_rate(&mut self.optimizer, lr);
}
/// Get a reference to the variables tracked by this optimizer
///
/// # Returns
@@ -141,7 +149,8 @@ impl Adam {
.map_err(|e| MLError::TrainingError(format!("Backward pass failed: {}", e)))?;
// 2. Clip gradients entirely on GPU — returns norm tensor, zero to_scalar
let norm_tensor = crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm)
// Use loss device to guarantee GPU residence (no CPU fallback).
let norm_tensor = crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm, loss.device())
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
// 3. Apply optimizer step with clipped gradients (GPU-only)
@@ -185,7 +194,7 @@ impl Adam {
// Clip gradients entirely on GPU — returns norm tensor, zero to_scalar
let norm_tensor =
crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm)
crate::gradient_utils::clip_grad_norm(&self.vars, &mut grads, max_norm, loss.device())
.map_err(|e| MLError::TrainingError(format!("Gradient clipping failed: {}", e)))?;
Ok((grads, norm_tensor))

View File

@@ -15,7 +15,7 @@ description = "DQN reinforcement learning for Foxhunt trading"
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn", "candle-nn/cuda", "candle-nn/cudnn"]
cuda = ["candle-core/cuda", "candle-nn/cuda"]
[dependencies]
ml-core.workspace = true

View File

@@ -184,6 +184,8 @@ pub struct DQNAgent {
optimizer: Option<Adam>,
/// Training step counter
training_step: u64,
/// Dropout layer for regularization during training
dropout: candle_nn::Dropout,
}
impl DQNAgent {
@@ -229,6 +231,7 @@ impl DQNAgent {
metrics: AgentMetrics::default(),
optimizer: None,
training_step: 0,
dropout: candle_nn::Dropout::new(0.2),
})
}
@@ -460,7 +463,7 @@ impl DQNAgent {
x = leaky_relu(&x, 0.01)?;
// Apply dropout during training
x = candle_nn::Dropout::new(0.2).forward(&x, true)?;
x = self.dropout.forward(&x, true)?;
}
}
@@ -564,12 +567,6 @@ impl DQNAgent {
Ok(())
}
/// Update target network by copying weights from main network
fn update_target_network(&mut self) -> Result<(), MLError> {
// Use the new proper weight copying method
self.update_target_network_weights()
}
/// Save model checkpoint (simplified implementation)
pub fn save_checkpoint(&self, path: &std::path::Path) -> Result<(), MLError> {
use std::fs::File;
@@ -633,7 +630,7 @@ impl DQNAgent {
);
// Copy weights to target network
self.update_target_network()?;
self.update_target_network_weights()?;
Ok(())
}
@@ -677,7 +674,7 @@ impl DQNAgent {
})?;
// Propagate loaded weights to target network
self.update_target_network()?;
self.update_target_network_weights()?;
tracing::info!("DQNAgent weights loaded from safetensors: {}", safetensors_path);
Ok(())
@@ -701,25 +698,8 @@ impl DQNAgent {
/// Update learning rate with decay schedule
pub fn update_learning_rate(&mut self, decay_factor: f64) -> Result<(), MLError> {
if let Some(ref mut optimizer) = self.optimizer {
let current_lr = optimizer.learning_rate();
let new_lr = current_lr * decay_factor;
// Recreate optimizer with new learning rate
use candle_optimisers::Decay;
let adam_params = ParamsAdam {
lr: new_lr,
beta_1: 0.9,
beta_2: 0.999,
eps: 1e-8,
weight_decay: Some(Decay::DecoupledWeightDecay(self.config.weight_decay)),
amsgrad: false,
};
self.optimizer = Some(
Adam::new(self.q_network.vars().all_vars(), adam_params).map_err(|e| {
MLError::TrainingError(format!("Failed to update learning rate: {}", e))
})?,
);
let new_lr = optimizer.learning_rate() * decay_factor;
optimizer.set_learning_rate(new_lr);
}
Ok(())
}
@@ -993,8 +973,10 @@ impl DQNAgent {
// DQN outputs 5 exposure-level actions (0-4).
let exposure = super::action_space::ExposureLevel::from_index(action_idx)?;
if self.config.use_branching {
// Branching mode: exposure from Q-value argmax, order/urgency sampled randomly
// (DQNAgent has no branch heads -- full branching uses DQN struct directly)
// DQNAgent only learns the exposure head (5 actions). The order and
// urgency dimensions are NOT learned here — for full 3-head branching
// use the DQN struct in dqn.rs with BranchingDuelingQNetwork. When
// DQNAgent is used in branching mode, we sample order/urgency randomly.
let order = super::action_space::OrderType::from_index(rng.gen_range(0..3_usize))?;
let urgency = super::action_space::Urgency::from_index(rng.gen_range(0..3_usize))?;
Ok(super::action_space::FactoredAction { exposure, order, urgency })

View File

@@ -4181,9 +4181,28 @@ impl DQN {
Ok(self.memory.len())
}
/// Get Q-network variables for serialization
/// Get Q-network variables for serialization.
///
/// Returns the `VarMap` of the **active** network architecture:
/// 1. Hybrid distributional+dueling (highest priority)
/// 2. Branching DQN
/// 3. Dueling-only
/// 4. Plain Sequential (fallback)
///
/// This ensures checkpoint saves always capture the trained weights,
/// not an unused fallback network's empty `VarMap`.
pub const fn get_q_network_vars(&self) -> &VarMap {
self.q_network.vars()
// Priority must match compute_loss_internal / select_action:
// branching > dist_dueling > dueling > standard
if let Some(ref net) = self.branching_q_network {
net.vars()
} else if let Some(ref net) = self.dist_dueling_q_network {
net.vars()
} else if let Some(ref net) = self.dueling_q_network {
net.vars()
} else {
self.q_network.vars()
}
}
/// Load model weights from safetensors checkpoint
@@ -4247,7 +4266,10 @@ impl DQN {
//
// VarMap::clone() is cheap (Arc clone of internal data), and load() only needs
// the shared Mutex, so calling load on a clone updates the same Var storage.
let mut vars_clone = self.q_network.vars().clone();
//
// Use get_q_network_vars() to match save path: active architecture's VarMap
// (branching > dist_dueling > dueling > plain Sequential).
let mut vars_clone = self.get_q_network_vars().clone();
vars_clone.load(&safetensors_path).map_err(|e| {
MLError::CheckpointError(format!("Failed to load safetensors via VarMap: {}", e))
})?;

View File

@@ -311,6 +311,10 @@ pub struct RewardConfig {
/// EMA decay rate for DSR calculation, clamped to [0.0001, 0.1].
/// Smaller values = longer lookback window. Default: 0.01 (~100-step window).
pub dsr_eta: f64,
/// Initial capital for denormalizing portfolio features.
/// Used to convert normalized portfolio values back to absolute for reward calculations.
/// Must match the value passed to `PortfolioTracker::new()`.
pub initial_capital: f64,
}
impl Default for RewardConfig {
@@ -332,6 +336,7 @@ impl Default for RewardConfig {
sharpe_window: 20, // WAVE 26 P1.3: 20-step rolling window for Sharpe calculation
use_dsr: false, // DSR disabled by default (opt-in via hyperopt)
dsr_eta: 0.01, // ~100-step effective EMA window
initial_capital: 100_000.0,
}
}
}
@@ -362,6 +367,14 @@ impl RewardConfig {
MUST use percentage_pnl=true for numerical stability.".to_owned()));
}
// Validate initial_capital is positive (needed for denormalization)
if self.initial_capital <= 0.0 {
return Err(MLError::ConfigError(format!(
"initial_capital must be > 0.0, got {}. Required for portfolio feature denormalization.",
self.initial_capital
)));
}
// Warn if normalization disabled (suboptimal but not fatal)
if !self.enable_normalization {
tracing::warn!(
@@ -449,6 +462,7 @@ impl RewardConfigBuilder {
sharpe_window: 100,
use_dsr: self.use_dsr.unwrap_or(false),
dsr_eta: self.dsr_eta.unwrap_or(0.01),
initial_capital: 100_000.0,
})
}
}
@@ -821,12 +835,19 @@ impl RewardFunction {
tracing::debug!("Next state portfolio_features is empty, using 100000.0 for portfolio value");
}
let default_val = Decimal::try_from(100000.0).unwrap_or(Decimal::ONE_HUNDRED);
// portfolio_features[0] is NORMALIZED: (portfolio_value / initial_capital) - 1.0
// where 0.0 = at initial capital, +0.01 = 1% gain, -0.01 = 1% loss.
// Denormalize to absolute value for the reward function's zero-check.
// Use initial_capital from config (default $100K if not set).
let initial_capital = self.config.initial_capital;
let default_val = Decimal::try_from(initial_capital).unwrap_or(Decimal::ONE_HUNDRED);
let current_normalized = *current_state.portfolio_features.first().unwrap_or(&0.0) as f64;
let next_normalized = *next_state.portfolio_features.first().unwrap_or(&0.0) as f64;
let current_value =
Decimal::try_from(*current_state.portfolio_features.first().unwrap_or(&100000.0) as f64)
Decimal::try_from((current_normalized + 1.0) * initial_capital)
.unwrap_or(default_val);
let next_value =
Decimal::try_from(*next_state.portfolio_features.first().unwrap_or(&100000.0) as f64)
Decimal::try_from((next_normalized + 1.0) * initial_capital)
.unwrap_or(default_val);
// Bug #17 Fix: Use percentage-based P&L or absolute dollar change

View File

@@ -0,0 +1,445 @@
//! GPU end-to-end smoketest for DQN training pipeline.
//!
//! Validates the FULL GPU hot path:
//! network init → experience storage → batch sample → forward → backward
//! → gradient clip → optimizer step → loss/grad_norm device check
//!
//! Run: `SQLX_OFFLINE=true cargo test -p ml-dqn --test gpu_smoketest -- --ignored --nocapture`
//!
//! Requirements: CUDA-capable GPU (RTX 3050 Ti or better)
use candle_core::Device;
use ml_dqn::dqn::{DQNConfig, DQN};
use ml_dqn::experience::Experience;
/// Create a minimal DQN config suitable for GPU smoketest (small networks, fast).
fn smoketest_config() -> DQNConfig {
DQNConfig {
state_dim: 48,
num_actions: 5,
hidden_dims: vec![64, 64], // Small for fast iteration
learning_rate: 1e-3,
gamma: 0.99,
epsilon_start: 0.1,
epsilon_end: 0.01,
epsilon_decay: 0.999,
replay_buffer_capacity: 2048,
batch_size: 32,
min_replay_size: 64,
target_update_freq: 100,
use_double_dqn: true,
use_huber_loss: true,
huber_delta: 1.0,
leaky_relu_alpha: 0.01,
gradient_clip_norm: 10.0,
tau: 0.005,
tau_final: 0.0005,
tau_anneal_steps: 1000,
use_soft_updates: true,
warmup_steps: 0, // No warmup — train immediately
n_steps: 1,
initial_capital: 100_000.0,
use_per: false, // CPU PER off — test core GPU path
use_gpu_replay_buffer: false, // No GPU PER — test without CUDA feature complexity
per_alpha: 0.6,
per_beta_start: 0.4,
per_beta_max: 1.0,
per_beta_annealing_steps: 1000,
per_max_memory_bytes: 512 * 1024 * 1024,
use_dueling: true,
dueling_hidden_dim: 64,
use_distributional: false, // Disable C51 for speed
num_atoms: 51,
v_min: -25.0,
v_max: 25.0,
use_noisy_nets: false,
noisy_sigma_init: 0.5,
enable_q_value_clipping: true,
q_value_clip_min: -100.0,
q_value_clip_max: 100.0,
gradient_collapse_multiplier: 2.0,
gradient_collapse_patience: 100,
entropy_coefficient: 0.01,
noisy_epsilon_floor: 0.05,
use_count_bonus: false,
count_bonus_coefficient: 0.0,
use_cql: false, // Disable CQL for speed
cql_alpha: 0.0,
use_iqn: false,
iqn_num_quantiles: 32,
iqn_kappa: 1.0,
iqn_embedding_dim: 32,
use_branching: false, // Disable branching — test basic path first
branch_hidden_dim: 64,
use_regime_conditioning: false,
use_cvar_action_selection: false,
cvar_alpha: 0.05,
minimum_profit_factor: 1.5,
weight_decay: 0.0,
dropout_rate: 0.0,
mixed_precision: None,
}
}
/// Generate a synthetic experience with deterministic data.
fn synthetic_experience(state_dim: usize, action: u8, idx: u32) -> Experience {
// Use Experience::new for correct reward scaling and timestamp
let state: Vec<f32> = (0..state_dim)
.map(|i| ((i as f32 * 0.1 + action as f32 * 0.3 + idx as f32 * 0.01).sin() * 0.5 + 0.5))
.collect();
let next_state: Vec<f32> = (0..state_dim)
.map(|i| ((i as f32 * 0.1 + action as f32 * 0.3 + (idx + 1) as f32 * 0.01).sin() * 0.5 + 0.5))
.collect();
let reward = (action as f32 - 2.0) * 0.1; // Small centered reward
Experience::new(state, action, reward, next_state, false)
}
#[test]
#[ignore] // Requires CUDA GPU
fn gpu_smoketest_dqn_train_step_pure_gpu() {
// 1. Verify CUDA is available
let device = Device::cuda_if_available(0).expect("CUDA device required for GPU smoketest");
match device {
Device::Cuda(_) => eprintln!("[OK] CUDA device detected"),
_ => panic!("Expected CUDA device, got {:?}", device),
}
// 2. Create DQN on GPU
let config = smoketest_config();
let state_dim = config.state_dim;
let mut dqn = DQN::new_on_device(config, device.clone())
.expect("DQN creation on GPU failed");
eprintln!("[OK] DQN created on GPU");
// 3. Fill replay buffer with synthetic experiences (above min_replay_size)
for i in 0..128_u32 {
let action = (i % 5) as u8;
let exp = synthetic_experience(state_dim, action, i);
dqn.store_experience(exp).expect("store_experience failed");
}
eprintln!("[OK] 128 synthetic experiences stored");
// 4. Run training steps and validate GPU residency
let mut total_loss = 0.0_f32;
let mut total_grad_norm = 0.0_f32;
let num_steps = 5;
for step in 0..num_steps {
let result = dqn.train_step(None).expect("train_step failed");
// 5. CRITICAL: Verify loss and grad_norm are on CUDA, not CPU
let loss_device = result.loss_gpu.device();
let grad_device = result.grad_norm_gpu.device();
assert!(
matches!(loss_device, Device::Cuda(_)),
"Step {}: loss on {:?}, expected CUDA",
step, loss_device
);
assert!(
matches!(grad_device, Device::Cuda(_)),
"Step {}: grad_norm on {:?}, expected CUDA — GPU→CPU roundtrip NOT eliminated!",
step, grad_device
);
// Read scalars (single GPU→CPU sync per step — intentional for validation)
let loss_val = result.loss_gpu
.to_scalar::<f32>()
.expect("loss to_scalar failed");
let grad_val = result.grad_norm_gpu
.to_vec1::<f32>()
.unwrap_or_else(|_| {
// grad_norm might be rank-0 instead of [1]
vec![result.grad_norm_gpu.to_scalar::<f32>().expect("grad_norm read failed")]
})[0];
eprintln!(
" Step {}: loss={:.6}, grad_norm={:.6}, loss_device={:?}, grad_device={:?}",
step, loss_val, grad_val, loss_device, grad_device
);
assert!(loss_val.is_finite(), "Step {}: loss is not finite: {}", step, loss_val);
assert!(grad_val.is_finite(), "Step {}: grad_norm is not finite: {}", step, grad_val);
// grad_norm must be > 0 (proves gradient clipping actually ran on GPU)
assert!(
grad_val > 0.0,
"Step {}: grad_norm is 0.0 — gradient clipping likely disabled (CPU fallback bug)",
step
);
total_loss += loss_val;
total_grad_norm += grad_val;
}
eprintln!(
"[OK] {} training steps completed. avg_loss={:.6}, avg_grad_norm={:.6}",
num_steps,
total_loss / num_steps as f32,
total_grad_norm / num_steps as f32
);
eprintln!("[PASS] GPU pipeline: network → forward → backward → clip → optim — PURE GPU");
}
#[test]
#[ignore] // Requires CUDA GPU
fn gpu_smoketest_branching_dqn_train_step() {
// Same test but with branching DQN (45 factored actions)
let device = match Device::cuda_if_available(0) {
Ok(d @ Device::Cuda(_)) => d,
_ => {
eprintln!("Skipping: no CUDA device");
return;
}
};
let mut config = smoketest_config();
config.use_branching = true;
config.branch_hidden_dim = 64;
// Branching DQN still uses num_actions=5 for exposure head
// but factored into [5, 3, 3] heads internally
let state_dim = config.state_dim;
let mut dqn = DQN::new_on_device(config, device.clone())
.expect("Branching DQN creation failed");
eprintln!("[OK] Branching DQN created on GPU");
// Fill replay buffer
for i in 0..128_u32 {
let action = (i % 5) as u8;
dqn.store_experience(synthetic_experience(state_dim, action, i))
.expect("store failed");
}
// Run training steps
for step in 0..3 {
let result = dqn.train_step(None).expect("branching train_step failed");
let loss_device = result.loss_gpu.device();
let grad_device = result.grad_norm_gpu.device();
assert!(
matches!(loss_device, Device::Cuda(_)),
"Branching step {}: loss on {:?}", step, loss_device
);
assert!(
matches!(grad_device, Device::Cuda(_)),
"Branching step {}: grad_norm on {:?} — GPU→CPU roundtrip!", step, grad_device
);
let grad_val = result.grad_norm_gpu
.to_vec1::<f32>()
.unwrap_or_else(|_| {
vec![result.grad_norm_gpu.to_scalar::<f32>().expect("read")]
})[0];
assert!(
grad_val > 0.0,
"Branching step {}: grad_norm=0.0 — clipping broken", step
);
eprintln!(
" Branching step {}: loss={:.6}, grad_norm={:.6}",
step,
result.loss_gpu.to_scalar::<f32>().unwrap_or(f32::NAN),
grad_val
);
}
eprintln!("[PASS] Branching DQN GPU pipeline verified");
}
#[test]
#[ignore] // Requires CUDA GPU
fn gpu_smoketest_gradient_clip_device_consistency() {
// Directly test clip_grad_norm with GPU tensors to verify
// the device parameter fix works correctly.
use candle_core::{Tensor, Var};
use ml_core::gradient_utils::clip_grad_norm;
let device = match Device::cuda_if_available(0) {
Ok(d @ Device::Cuda(_)) => d,
_ => {
eprintln!("Skipping: no CUDA device");
return;
}
};
// Create a Var on GPU
let var = Var::from_tensor(
&Tensor::new(&[1.0_f32, 2.0, 3.0], &device).expect("tensor"),
).expect("var");
// Compute gradients (backward on GPU)
let loss = var.mul(&var).expect("mul").sum_all().expect("sum");
let mut grads = loss.backward().expect("backward");
// Call clip_grad_norm with explicit GPU device
let norm_tensor = clip_grad_norm(&[var.clone()], &mut grads, 1.0, &device)
.expect("clip_grad_norm failed");
// CRITICAL: norm tensor must be on GPU
assert!(
matches!(norm_tensor.device(), Device::Cuda(_)),
"norm_tensor on {:?}, expected CUDA — device parameter not propagated!",
norm_tensor.device()
);
let norm_val = norm_tensor.to_vec1::<f32>().expect("read")[0];
eprintln!("GPU gradient norm: {:.4} (expected ~7.48)", norm_val);
// grad = [2, 4, 6], norm = sqrt(4+16+36) = sqrt(56) ≈ 7.48
assert!((norm_val as f64 - 7.483).abs() < 0.1, "Unexpected norm: {}", norm_val);
// After clipping to max_norm=1.0, gradients should be scaled
let grad = grads.get(&var).expect("grad exists");
assert!(
matches!(grad.device(), Device::Cuda(_)),
"Clipped gradient on {:?}, expected CUDA", grad.device()
);
let g_vec = grad.to_vec1::<f32>().expect("read grad");
let clipped_norm = (g_vec[0] * g_vec[0] + g_vec[1] * g_vec[1] + g_vec[2] * g_vec[2]).sqrt();
eprintln!("Clipped gradient norm: {:.4} (expected ~1.0)", clipped_norm);
assert!((clipped_norm - 1.0).abs() < 0.05, "Clipped norm should be ~1.0, got {}", clipped_norm);
eprintln!("[PASS] clip_grad_norm stays PURE GPU");
}
#[test]
#[ignore] // Requires CUDA GPU
fn gpu_smoketest_training_metrics_validation() {
// Validates training metrics over 50 steps:
// - Loss should decrease (learning signal exists)
// - Grad norm should vary (not stuck at 0 or constant)
// - Action distribution should show diversity (no action collapse)
let device = match Device::cuda_if_available(0) {
Ok(d @ Device::Cuda(_)) => d,
_ => {
eprintln!("Skipping: no CUDA device");
return;
}
};
let mut config = smoketest_config();
config.use_branching = false;
config.batch_size = 32;
config.min_replay_size = 64;
config.replay_buffer_capacity = 4096;
config.learning_rate = 5e-4; // Slightly higher LR for faster convergence in test
let state_dim = config.state_dim;
let num_actions = config.num_actions;
let mut dqn = DQN::new_on_device(config, device.clone())
.expect("DQN creation failed");
// Fill replay buffer with diverse experiences
// Create reward signal: action 2 (Flat) is rewarded most, extremes penalized
for i in 0..512_u32 {
let action = (i % 5) as u8;
dqn.store_experience(synthetic_experience(state_dim, action, i))
.expect("store failed");
}
eprintln!("[OK] 512 experiences stored");
// Track metrics over training
let num_steps = 50;
let mut losses: Vec<f32> = Vec::with_capacity(num_steps);
let mut grad_norms: Vec<f32> = Vec::with_capacity(num_steps);
let mut action_counts = vec![0_u32; num_actions];
for step in 0..num_steps {
let result = dqn.train_step(None).expect("train_step failed");
let loss_val = result.loss_gpu.to_scalar::<f32>().expect("loss read");
let grad_val = result.grad_norm_gpu
.to_vec1::<f32>()
.unwrap_or_else(|_| vec![result.grad_norm_gpu.to_scalar::<f32>().expect("read")])[0];
losses.push(loss_val);
grad_norms.push(grad_val);
// Sample action from the model to check diversity (track exposure level)
let test_state: Vec<f32> = (0..state_dim)
.map(|i| (i as f32 * 0.07 + step as f32 * 0.01).sin())
.collect();
let action_result = dqn.select_action(&test_state);
if let Ok(action) = action_result {
let exposure_idx = action.exposure as usize;
if exposure_idx < num_actions {
action_counts[exposure_idx] += 1;
}
}
if step % 10 == 0 || step == num_steps - 1 {
eprintln!(
" Step {:3}: loss={:.6}, grad_norm={:.6}",
step, loss_val, grad_val
);
}
}
// --- Validate training metrics ---
// 1. Loss should be finite throughout
for (i, &loss) in losses.iter().enumerate() {
assert!(loss.is_finite(), "Step {}: loss is NaN/Inf: {}", i, loss);
}
// 2. Grad norm should be non-zero (gradient clipping working)
let zero_grads = grad_norms.iter().filter(|&&g| g == 0.0).count();
assert!(
zero_grads == 0,
"Found {} steps with grad_norm=0.0 — gradient clipping likely broken",
zero_grads
);
// 3. Grad norm should vary (not constant — sign of learning)
let unique_norms: std::collections::HashSet<u32> = grad_norms.iter()
.map(|&g| (g * 1000.0) as u32)
.collect();
assert!(
unique_norms.len() > 5,
"Only {} unique grad_norm values — suspicious lack of variation",
unique_norms.len()
);
// 4. Check for action collapse: no single action should have > 80% of selections
let total_actions: u32 = action_counts.iter().sum();
if total_actions > 0 {
let max_count = *action_counts.iter().max().unwrap_or(&0);
let max_pct = max_count as f64 / total_actions as f64 * 100.0;
eprintln!(
"\n Action distribution: {:?} (total={})",
action_counts, total_actions
);
// During early training with epsilon exploration, perfect uniformity is unlikely
// but severe collapse (>90% one action) indicates a bug
if max_pct > 90.0 {
eprintln!(
" WARNING: Potential action collapse — one action has {:.1}% of selections",
max_pct
);
}
}
// 5. Loss statistics
let avg_loss_first10: f32 = losses[..10].iter().sum::<f32>() / 10.0;
let avg_loss_last10: f32 = losses[losses.len()-10..].iter().sum::<f32>() / 10.0;
let avg_grad_norm: f32 = grad_norms.iter().sum::<f32>() / grad_norms.len() as f32;
eprintln!("\n === Training Metrics Summary ===");
eprintln!(" Loss (first 10 avg): {:.6}", avg_loss_first10);
eprintln!(" Loss (last 10 avg): {:.6}", avg_loss_last10);
eprintln!(" Avg grad norm: {:.6}", avg_grad_norm);
eprintln!(" Grad norm range: [{:.6}, {:.6}]",
grad_norms.iter().cloned().reduce(f32::min).unwrap_or(0.0),
grad_norms.iter().cloned().reduce(f32::max).unwrap_or(0.0),
);
eprintln!(" Zero grad_norm steps: {}/{}", zero_grads, num_steps);
eprintln!("[PASS] Training metrics validation complete");
}

View File

@@ -15,7 +15,7 @@ description = "ML ensemble coordination — voting, confidence, gating, hot-swap
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn"]
cuda = ["candle-core/cuda"]
[dependencies]
ml-core = { path = "../ml-core", default-features = false }

View File

@@ -15,7 +15,7 @@ description = "Model explainability (integrated gradients) for Foxhunt ML"
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn", "candle-nn/cuda", "candle-nn/cudnn"]
cuda = ["candle-core/cuda", "candle-nn/cuda"]
[dependencies]
ml-core = { path = "../ml-core", default-features = false }

View File

@@ -15,7 +15,7 @@ description = "ML hyperparameter optimization — PSO, TPE, campaigns, sensitivi
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn"]
cuda = ["candle-core/cuda"]
[dependencies]
ml-core = { path = "../ml-core", default-features = false }

View File

@@ -15,7 +15,7 @@ description = "ML labeling algorithms for Foxhunt HFT training data"
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn"]
cuda = ["candle-core/cuda"]
[dependencies]
ml-core.workspace = true

View File

@@ -15,7 +15,7 @@ description = "PPO reinforcement learning for Foxhunt trading"
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn", "candle-nn/cuda", "candle-nn/cudnn"]
cuda = ["candle-core/cuda", "candle-nn/cuda"]
[dependencies]
ml-core.workspace = true

View File

@@ -474,6 +474,7 @@ impl ContinuousPPO {
&mut grads,
&actor_vars,
self.config.max_grad_norm as f64,
&Device::Cpu,
)?;
debug!(
@@ -501,6 +502,7 @@ impl ContinuousPPO {
&mut value_grads,
&critic_vars,
self.config.max_grad_norm as f64,
&Device::Cpu,
)?;
debug!(

View File

@@ -1092,6 +1092,7 @@ impl PPO {
value_grads,
&critic_vars,
self.config.max_grad_norm as f64,
self.critic.device(),
)?;
if let Some(ref mut optimizer) = self.value_optimizer {
@@ -1125,6 +1126,7 @@ impl PPO {
value_grads,
&critic_vars,
self.config.max_grad_norm as f64,
self.critic.device(),
)?;
if let Some(ref mut optimizer) = self.value_optimizer {
@@ -1243,6 +1245,7 @@ impl PPO {
policy_grads,
&actor_vars,
self.config.max_grad_norm as f64,
self.actor.device(),
)?;
debug!(
@@ -1276,6 +1279,7 @@ impl PPO {
value_grads,
&critic_vars,
self.config.max_grad_norm as f64,
self.critic.device(),
)?;
debug!(
@@ -1320,6 +1324,7 @@ impl PPO {
policy_grads,
&actor_vars,
self.config.max_grad_norm as f64,
self.actor.device(),
)?;
debug!(
@@ -1348,6 +1353,7 @@ impl PPO {
value_grads,
&critic_vars,
self.config.max_grad_norm as f64,
self.critic.device(),
)?;
debug!(
@@ -1623,6 +1629,7 @@ impl PPO {
&mut policy_grads,
&actor_vars,
self.config.max_grad_norm as f64,
self.actor.device(),
)?;
debug!(
@@ -1650,6 +1657,7 @@ impl PPO {
&mut value_grads,
&critic_vars,
self.config.max_grad_norm as f64,
self.critic.device(),
)?;
debug!(

View File

@@ -15,7 +15,7 @@ description = "Supervised models (TFT, Mamba, Liquid, TGGN, TLOB, KAN, xLSTM, Di
[features]
default = ["cuda"]
cuda = ["candle-core/cuda", "candle-core/cudnn", "candle-nn/cuda", "candle-nn/cudnn"]
cuda = ["candle-core/cuda", "candle-nn/cuda"]
[dependencies]
ml-core.workspace = true

View File

@@ -1,455 +0,0 @@
#include <cuda_runtime.h>
#include <cuda_fp16.h>
#include <cub/cub.cuh>
#include <cooperative_groups.h>
namespace cg = cooperative_groups;
// MAMBA Selective Scan CUDA Kernel for <5μs inference latency
// Optimized for financial time series processing with ultra-low latency
#define WARP_SIZE 32
#define MAX_THREADS_PER_BLOCK 1024
#define SHARED_MEM_SIZE 48 * 1024 // 48KB shared memory per SM
// Optimized selective scan kernel with shared memory and warp primitives
__global__ void mamba_selective_scan_kernel(
float* __restrict__ states, // [batch_size, d_state] - output states
const float* __restrict__ A, // [d_state, d_state] - state transition matrix
const float* __restrict__ B, // [batch_size, seq_len, d_state] - input projection
const float* __restrict__ C, // [batch_size, seq_len, d_state] - output projection
const float* __restrict__ delta, // [batch_size, seq_len] - time step deltas
const float* __restrict__ x, // [batch_size, seq_len, d_model] - input sequence
float* __restrict__ y, // [batch_size, seq_len, d_model] - output sequence
int batch_size,
int d_state,
int d_model,
int seq_len
) {
// Shared memory for collaborative processing
__shared__ float shared_state[256]; // State cache
__shared__ float shared_A[256]; // A matrix cache
__shared__ float shared_delta[64]; // Delta cache for sequence
// Thread and warp identification
int tid = threadIdx.x;
int bid = blockIdx.x;
int wid = tid / WARP_SIZE;
int lane = tid % WARP_SIZE;
// Grid-stride loop for batch processing
int batch_idx = bid;
// Cooperative groups for warp-level operations
auto warp = cg::tiled_partition<WARP_SIZE>(cg::this_thread_block());
if (batch_idx >= batch_size) return;
// Load A matrix into shared memory with coalesced access
for (int i = tid; i < d_state * d_state; i += blockDim.x) {
if (i < 256) { // Limit to shared memory size
shared_A[i] = A[i];
}
}
// Initialize state vector in shared memory
for (int i = tid; i < d_state && i < 256; i += blockDim.x) {
shared_state[i] = 0.0f;
}
__syncthreads();
// Sequential processing for each time step
for (int t = 0; t < seq_len; t++) {
// Load delta for current timestep
float dt = delta[batch_idx * seq_len + t];
// Discretization: A_discrete = exp(delta * A)
// Simplified approximation for speed: A_discrete ≈ I + dt * A
// Process each state dimension
for (int s = tid; s < d_state && s < 256; s += blockDim.x) {
float new_state = shared_state[s];
// State transition: x_{t+1} = A_discrete * x_t + B * u_t
float state_update = 0.0f;
// Matrix-vector multiplication with A
for (int j = 0; j < d_state && j < 256; j++) {
if (s * d_state + j < 256) {
state_update += (1.0f + dt * shared_A[s * d_state + j]) * shared_state[j];
}
}
// Add input contribution B * u_t
if (t < seq_len && s < d_state) {
float b_val = B[batch_idx * seq_len * d_state + t * d_state + s];
float x_val = x[batch_idx * seq_len * d_model + t * d_model + s % d_model];
state_update += dt * b_val * x_val;
}
new_state = state_update;
// Warp-level reduction for numerical stability
new_state = warp.shfl_xor(new_state, 1);
new_state = warp.shfl_xor(new_state, 2);
new_state = warp.shfl_xor(new_state, 4);
new_state = warp.shfl_xor(new_state, 8);
new_state = warp.shfl_xor(new_state, 16);
shared_state[s] = new_state;
}
__syncthreads();
// Compute output: y_t = C * x_t
for (int d = tid; d < d_model; d += blockDim.x) {
float output = 0.0f;
for (int s = 0; s < d_state && s < 256; s++) {
if (t < seq_len) {
float c_val = C[batch_idx * seq_len * d_state + t * d_state + s];
output += c_val * shared_state[s];
}
}
if (t < seq_len && d < d_model) {
y[batch_idx * seq_len * d_model + t * d_model + d] = output;
}
}
__syncthreads();
}
// Write final states back to global memory
for (int s = tid; s < d_state; s += blockDim.x) {
if (s < 256) {
states[batch_idx * d_state + s] = shared_state[s];
}
}
}
// High-performance kernel for small sequences (financial tick data)
__global__ void mamba_selective_scan_fast_kernel(
float* __restrict__ states,
const float* __restrict__ A,
const float* __restrict__ B,
const float* __restrict__ C,
const float* __restrict__ delta,
const float* __restrict__ x,
float* __restrict__ y,
int batch_size,
int d_state,
int d_model,
int seq_len
) {
// Ultra-fast kernel for seq_len <= 32 (typical for HFT)
__shared__ float shared_state[32][32]; // [seq_len][d_state]
__shared__ float shared_A_disc[32][32]; // Discretized A matrix
int tid = threadIdx.x;
int bid = blockIdx.x;
int batch_idx = bid;
if (batch_idx >= batch_size || tid >= d_state) return;
// Initialize
shared_state[0][tid] = 0.0f;
// Precompute discretized A matrix
float dt = delta[batch_idx * seq_len];
shared_A_disc[tid][tid] = 1.0f + dt * A[tid * d_state + tid];
__syncthreads();
// Unrolled loop for maximum performance
#pragma unroll
for (int t = 0; t < seq_len && t < 32; t++) {
float dt_t = delta[batch_idx * seq_len + t];
// State update
float new_state = shared_A_disc[tid][tid] * shared_state[t][tid];
// Add input
if (tid < d_model) {
float b_val = B[batch_idx * seq_len * d_state + t * d_state + tid];
float x_val = x[batch_idx * seq_len * d_model + t * d_model + tid];
new_state += dt_t * b_val * x_val;
}
if (t + 1 < seq_len) {
shared_state[t + 1][tid] = new_state;
}
// Compute output
if (tid < d_model) {
float output = 0.0f;
for (int s = 0; s < d_state && s < 32; s++) {
float c_val = C[batch_idx * seq_len * d_state + t * d_state + s];
output += c_val * shared_state[t][s];
}
y[batch_idx * seq_len * d_model + t * d_model + tid] = output;
}
__syncthreads();
}
// Write final state
states[batch_idx * d_state + tid] = shared_state[seq_len - 1][tid];
}
// Fused kernel with gating mechanism (SiLU activation)
__device__ __forceinline__ float silu(float x) {
return x / (1.0f + __expf(-x));
}
__global__ void mamba_selective_scan_fused_kernel(
float* __restrict__ states,
const float* __restrict__ A,
const float* __restrict__ B,
const float* __restrict__ C,
const float* __restrict__ delta,
const float* __restrict__ x,
const float* __restrict__ gate, // Gating values
float* __restrict__ y,
int batch_size,
int d_state,
int d_model,
int seq_len
) {
__shared__ float shared_cache[512]; // Multi-purpose cache
int tid = threadIdx.x;
int bid = blockIdx.x;
int batch_idx = bid;
if (batch_idx >= batch_size) return;
// Fused selective scan with gating
for (int t = 0; t < seq_len; t++) {
float dt = delta[batch_idx * seq_len + t];
for (int s = tid; s < d_state; s += blockDim.x) {
if (s < 256) {
// State transition
float state_val = shared_cache[s];
float a_val = A[s * d_state + s]; // Diagonal approximation
float b_val = B[batch_idx * seq_len * d_state + t * d_state + s];
float x_val = x[batch_idx * seq_len * d_model + t * d_model + s % d_model];
float new_state = (1.0f + dt * a_val) * state_val + dt * b_val * x_val;
shared_cache[s] = new_state;
}
}
__syncthreads();
// Gated output computation
for (int d = tid; d < d_model; d += blockDim.x) {
float output = 0.0f;
for (int s = 0; s < d_state && s < 256; s++) {
float c_val = C[batch_idx * seq_len * d_state + t * d_state + s];
output += c_val * shared_cache[s];
}
// Apply gating with SiLU
float gate_val = gate[batch_idx * seq_len * d_model + t * d_model + d];
output = output * silu(gate_val);
y[batch_idx * seq_len * d_model + t * d_model + d] = output;
}
__syncthreads();
}
// Final state writeback
for (int s = tid; s < d_state; s += blockDim.x) {
if (s < 256) {
states[batch_idx * d_state + s] = shared_cache[s];
}
}
}
// Half-precision kernel for maximum throughput
__global__ void mamba_selective_scan_fp16_kernel(
__half* __restrict__ states,
const __half* __restrict__ A,
const __half* __restrict__ B,
const __half* __restrict__ C,
const __half* __restrict__ delta,
const __half* __restrict__ x,
__half* __restrict__ y,
int batch_size,
int d_state,
int d_model,
int seq_len
) {
__shared__ __half shared_state[256];
__shared__ __half shared_A[256];
int tid = threadIdx.x;
int bid = blockIdx.x;
int batch_idx = bid;
if (batch_idx >= batch_size) return;
// Load A matrix
for (int i = tid; i < d_state * d_state && i < 256; i += blockDim.x) {
shared_A[i] = A[i];
}
// Initialize states
for (int i = tid; i < d_state && i < 256; i += blockDim.x) {
shared_state[i] = __float2half(0.0f);
}
__syncthreads();
// Process sequence
for (int t = 0; t < seq_len; t++) {
__half dt = delta[batch_idx * seq_len + t];
for (int s = tid; s < d_state && s < 256; s += blockDim.x) {
__half state_update = shared_state[s];
// Use __hmul and __hadd for half-precision ops
for (int j = 0; j < d_state && j < 256; j++) {
if (s * d_state + j < 256) {
__half a_elem = shared_A[s * d_state + j];
__half disc_a = __hadd(__float2half(1.0f), __hmul(dt, a_elem));
state_update = __hadd(state_update, __hmul(disc_a, shared_state[j]));
}
}
// Add input contribution
if (t < seq_len && s < d_state) {
__half b_val = B[batch_idx * seq_len * d_state + t * d_state + s];
__half x_val = x[batch_idx * seq_len * d_model + t * d_model + s % d_model];
__half input_contrib = __hmul(__hmul(dt, b_val), x_val);
state_update = __hadd(state_update, input_contrib);
}
shared_state[s] = state_update;
}
__syncthreads();
// Compute output
for (int d = tid; d < d_model; d += blockDim.x) {
__half output = __float2half(0.0f);
for (int s = 0; s < d_state && s < 256; s++) {
if (t < seq_len) {
__half c_val = C[batch_idx * seq_len * d_state + t * d_state + s];
output = __hadd(output, __hmul(c_val, shared_state[s]));
}
}
if (t < seq_len && d < d_model) {
y[batch_idx * seq_len * d_model + t * d_model + d] = output;
}
}
__syncthreads();
}
// Write final states
for (int s = tid; s < d_state; s += blockDim.x) {
if (s < 256) {
states[batch_idx * d_state + s] = shared_state[s];
}
}
}
// Host function to launch appropriate kernel
extern "C" {
void launch_mamba_selective_scan(
float* states,
const float* A,
const float* B,
const float* C,
const float* delta,
const float* x,
float* y,
int batch_size,
int d_state,
int d_model,
int seq_len,
cudaStream_t stream = 0
) {
// Kernel configuration for optimal performance
int threads_per_block = min(1024, max(32, d_state));
int blocks = batch_size;
// Dynamic shared memory size
size_t shared_mem_size = max(256 * sizeof(float), (d_state + seq_len) * sizeof(float));
if (seq_len <= 32 && d_state <= 32) {
// Use fast kernel for small sequences (HFT tick data)
mamba_selective_scan_fast_kernel<<<blocks, threads_per_block, shared_mem_size, stream>>>(
states, A, B, C, delta, x, y, batch_size, d_state, d_model, seq_len
);
} else {
// Use general kernel
mamba_selective_scan_kernel<<<blocks, threads_per_block, shared_mem_size, stream>>>(
states, A, B, C, delta, x, y, batch_size, d_state, d_model, seq_len
);
}
// Check for kernel launch errors
cudaError_t err = cudaGetLastError();
if (err != cudaSuccess) {
printf("CUDA kernel launch error: %s\n", cudaGetErrorString(err));
}
}
void launch_mamba_selective_scan_fused(
float* states,
const float* A,
const float* B,
const float* C,
const float* delta,
const float* x,
const float* gate,
float* y,
int batch_size,
int d_state,
int d_model,
int seq_len,
cudaStream_t stream = 0
) {
int threads_per_block = min(1024, max(32, d_state));
int blocks = batch_size;
size_t shared_mem_size = 512 * sizeof(float);
mamba_selective_scan_fused_kernel<<<blocks, threads_per_block, shared_mem_size, stream>>>(
states, A, B, C, delta, x, gate, y, batch_size, d_state, d_model, seq_len
);
}
void launch_mamba_selective_scan_fp16(
void* states,
const void* A,
const void* B,
const void* C,
const void* delta,
const void* x,
void* y,
int batch_size,
int d_state,
int d_model,
int seq_len,
cudaStream_t stream = 0
) {
int threads_per_block = min(1024, max(32, d_state));
int blocks = batch_size;
size_t shared_mem_size = 256 * sizeof(__half);
mamba_selective_scan_fp16_kernel<<<blocks, threads_per_block, shared_mem_size, stream>>>(
(__half*)states, (const __half*)A, (const __half*)B, (const __half*)C,
(const __half*)delta, (const __half*)x, (__half*)y,
batch_size, d_state, d_model, seq_len
);
}
}

View File

@@ -29,7 +29,7 @@ simd = [] # SIMD without heavy dependencies
# Storage and memory management features
gc = [] # Garbage collection features
s3-storage = ["ml-checkpoint/s3-storage", "aws-config", "aws-sdk-s3", "aws-types", "aws-credential-types", "urlencoding"] # S3 storage backend with AWS SDK
cuda = ["candle-core/cuda", "candle-core/cudnn", "candle-nn/cuda", "candle-nn/cudnn", "ml-core/cuda", "ml-dqn/cuda", "ml-ppo/cuda", "ml-supervised/cuda", "ml-ensemble/cuda", "ml-labeling/cuda", "ml-explainability/cuda", "ml-hyperopt/cuda"] # CUDA support — enabled by compile-training CI step via --features ml/cuda
cuda = ["candle-core/cuda", "candle-nn/cuda", "ml-core/cuda", "ml-dqn/cuda", "ml-ppo/cuda", "ml-supervised/cuda", "ml-ensemble/cuda", "ml-labeling/cuda", "ml-explainability/cuda", "ml-hyperopt/cuda"] # CUDA support — enabled by compile-training CI step via --features ml/cuda
nccl = ["cuda"] # NCCL multi-GPU data parallelism (requires NCCL library + cudarc nccl feature)
# ALL HEAVY ML FEATURES REMOVED:

View File

@@ -1077,6 +1077,7 @@ fn evaluate_dqn_fold_gpu(
tx_cost_bps: args.tx_cost_bps as f32,
spread_cost: (args.tick_size * args.spread_ticks) as f32,
initial_capital: args.initial_capital as f32,
..Default::default()
};
// Single window = full test fold
@@ -1360,6 +1361,7 @@ fn evaluate_ppo_fold_gpu(
tx_cost_bps: args.tx_cost_bps as f32,
spread_cost: (args.tick_size * args.spread_ticks) as f32,
initial_capital: args.initial_capital as f32,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(
@@ -1476,6 +1478,7 @@ fn evaluate_supervised_fold_gpu(
tx_cost_bps: args.tx_cost_bps as f32,
spread_cost: (args.tick_size * args.spread_ticks) as f32,
initial_capital: args.initial_capital as f32,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(

View File

@@ -302,6 +302,7 @@ fn run_ppo_hyperopt(args: &Args, parallel: usize, gpu_devices: &[candle_core::De
let mut trainer = PPOTrainer::new(&symbol_dir, args.epochs)
.context("Failed to create PPO trainer")?
.with_training_paths(training_paths)
.with_initial_capital(args.initial_capital)
.with_costs(args.tx_cost_bps, args.tick_size, args.spread_ticks)
.with_devices(gpu_devices.to_vec());
@@ -417,7 +418,7 @@ fn main() -> Result<()> {
std::env::set_var("NVIDIA_TF32_OVERRIDE", "1");
}
let args = Args::parse();
let mut args = Args::parse();
// Signal hyperopt mode active — will be cleared at exit
let hyperopt_model_label = args.model.clone();
@@ -507,6 +508,12 @@ fn main() -> Result<()> {
);
}
// Enforce minimum 5 trials for meaningful hyperopt
if args.trials < 5 {
info!("Trials {} below minimum — bumping to 5", args.trials);
args.trials = 5;
}
// Verify trials > n_initial (ArgminOptimizer requirement)
if args.trials <= args.n_initial {
anyhow::bail!(

View File

@@ -132,6 +132,12 @@ pub struct GpuBacktestConfig {
pub tx_cost_bps: f32,
pub spread_cost: f32,
pub initial_capital: f32,
/// Contract multiplier for futures (e.g., 50 for ES, 20 for NQ).
/// Used with `max_leverage` to compute leverage-safe position limits.
pub contract_multiplier: f32,
/// Maximum leverage ratio relative to capital (e.g., 2.0 = 2× leverage).
/// When > 0, caps effective position at `capital * max_leverage / (price * multiplier)`.
pub max_leverage: f32,
}
impl Default for GpuBacktestConfig {
@@ -141,10 +147,27 @@ impl Default for GpuBacktestConfig {
tx_cost_bps: 0.1,
spread_cost: 0.0001,
initial_capital: 100_000.0,
contract_multiplier: 50.0,
max_leverage: 2.0,
}
}
}
impl GpuBacktestConfig {
/// Compute leverage-safe max position given a reference price.
///
/// Returns `min(max_position, capital * max_leverage / (price * multiplier))`.
/// If `max_leverage <= 0` or `price <= 0`, returns `max_position` uncapped.
pub fn effective_max_position(&self, reference_price: f32) -> f32 {
if self.max_leverage <= 0.0 || reference_price <= 0.0 || self.contract_multiplier <= 0.0 {
return self.max_position;
}
let notional_per_contract = reference_price * self.contract_multiplier;
let leverage_cap = self.initial_capital * self.max_leverage / notional_per_contract;
self.max_position.min(leverage_cap)
}
}
// ── Main struct ───────────────────────────────────────────────────────────────
/// GPU backtest evaluator — runs walk-forward evaluation without CPU roundtrips.
@@ -253,6 +276,36 @@ impl GpuBacktestEvaluator {
return Err(MLError::ConfigError("All windows are empty".to_owned()));
}
// ── Apply leverage cap to max_position ────────────────────────────
let mut config = config;
if config.max_leverage > 0.0 {
// Use median close price across ALL windows for robust leverage cap.
// Single-window median can be skewed if first window is atypical.
let mut all_closes: Vec<f32> = window_prices
.iter()
.flat_map(|w| {
let mid = w.len() / 2;
w.get(mid).map(|ohlc| ohlc[3])
})
.filter(|p| *p > 0.0)
.collect();
all_closes.sort_unstable_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
let ref_price = if all_closes.is_empty() {
0.0
} else {
all_closes[all_closes.len() / 2]
};
let effective = config.effective_max_position(ref_price);
if effective < config.max_position {
tracing::info!(
"Leverage cap: max_position {:.2} → {:.4} (price={:.0}, multiplier={}, leverage={}×, capital={:.0})",
config.max_position, effective, ref_price, config.contract_multiplier,
config.max_leverage, config.initial_capital
);
config.max_position = effective;
}
}
// ── Flatten prices (zero-pad shorter windows) ─────────────────────
let window_lens: Vec<i32> = window_prices.iter().map(|w| w.len() as i32).collect();
@@ -1399,7 +1452,7 @@ impl GpuBacktestEvaluator {
WindowMetrics {
sharpe: metrics_host[base],
total_pnl: metrics_host[base + 1],
max_drawdown: metrics_host[base + 2],
max_drawdown: metrics_host[base + 2].min(1.0), // Cap at 100% (liquidation)
sortino: metrics_host[base + 3],
win_rate: metrics_host[base + 4],
total_trades: metrics_host[base + 5],

View File

@@ -222,6 +222,7 @@ pub fn evaluate_supervised_gpu_backtest(
tx_cost_bps: 2.0,
spread_cost: 0.5,
initial_capital: 100_000.0,
..Default::default()
};
// Wrap data as single-window slices for GpuBacktestEvaluator

View File

@@ -1706,6 +1706,7 @@ impl DQNTrainer {
tx_cost_bps: self.tx_cost_bps as f32,
spread_cost: (self.tick_size * self.spread_ticks) as f32,
initial_capital: self.initial_capital as f32,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(
@@ -3477,20 +3478,23 @@ impl HyperparameterOptimizable for DQNTrainer {
drop(training_metrics);
drop(internal_trainer); // MEMORY LEAK FIX: Explicitly drop trainer to free all resources
// Force CUDA synchronization and cache flush
// Creating a small tensor and reading it back forces CUDA to synchronize
// all pending operations before we move to the next trial.
let cleanup_device = candle_core::Device::cuda_if_available(0)
.unwrap_or(candle_core::Device::Cpu);
if cleanup_device.is_cuda() {
// Force sync by creating and reading a tiny tensor (blocks until all
// pending CUDA kernels complete — no sleep needed after this)
if let Ok(sync_tensor) =
candle_core::Tensor::zeros(1, candle_core::DType::F32, &cleanup_device)
{
drop(sync_tensor.to_vec0::<f32>());
// Force CUDA synchronization between trials.
// NOTE: cudarc caches freed GPU allocations in a memory pool for reuse.
// sysinfo may report this as a "leak" but it's intentional — next trial
// reuses the cached blocks without expensive cudaMalloc calls.
// Candle does not expose cudarc's pool flush, so we synchronize only.
#[cfg(feature = "cuda")]
{
if let Ok(cuda_dev) = candle_core::Device::cuda_if_available(0) {
if let Err(e) = cuda_dev.synchronize() {
tracing::warn!("CUDA synchronize between trials failed: {e}");
}
}
}
#[cfg(not(feature = "cuda"))]
{
// CPU-only: no GPU sync needed
}
info!("Resource cleanup complete");
// MEMORY LEAK FIX: Monitor memory usage after trial

View File

@@ -342,6 +342,10 @@ pub struct PPOTrainer {
#[cfg(feature = "cuda")]
gpu_num_bars: usize,
/// Initial capital for portfolio simulation and backtest evaluation.
/// Must match the value used in training reward computation.
initial_capital: f64,
/// Preloaded training data cached across hyperopt trials.
/// Loaded once via `preload_data()`, then shared (via Arc) across all trials
/// to avoid re-reading `.dbn.zst` files and re-extracting features per trial.
@@ -373,6 +377,7 @@ impl Clone for PPOTrainer {
tx_cost_bps: self.tx_cost_bps,
tick_size: self.tick_size,
spread_ticks: self.spread_ticks,
initial_capital: self.initial_capital,
// GPU resources are not cloned — each clone re-initialises on first use
#[cfg(feature = "cuda")]
features_cuda: None,
@@ -441,6 +446,7 @@ impl PPOTrainer {
tx_cost_bps: 1.0, // 1 bps commission (ES futures typical)
tick_size: 0.25, // ES futures tick size
spread_ticks: 1.0, // 1 tick spread (ES typical)
initial_capital: 35_000.0, // Default: $35K (realistic retail futures)
#[cfg(feature = "cuda")]
features_cuda: None,
#[cfg(feature = "cuda")]
@@ -476,6 +482,12 @@ impl PPOTrainer {
self
}
/// Set initial capital for portfolio simulation and backtest evaluation.
pub fn with_initial_capital(mut self, capital: f64) -> Self {
self.initial_capital = capital;
self
}
/// Override the compute device (CPU or CUDA).
pub fn with_device(mut self, device: Device) -> Self {
info!(
@@ -639,7 +651,7 @@ impl PPOTrainer {
actor_vars,
critic_vars,
&candle_nn::VarMap::new(), // curiosity placeholder
35_000.0, // initial_capital
self.initial_capital as f32,
self.tick_size as f32 * self.spread_ticks as f32, // avg_spread
0.20, // cash_reserve_pct
) {
@@ -698,7 +710,7 @@ impl PPOTrainer {
..Default::default()
};
if let Err(e) = collector.reset_episodes(35_000.0, self.tick_size as f32 * self.spread_ticks as f32, 0.20) {
if let Err(e) = collector.reset_episodes(self.initial_capital as f32, self.tick_size as f32 * self.spread_ticks as f32, 0.20) {
warn!("PPO GPU reset_episodes failed: {e}");
return None;
}
@@ -1378,7 +1390,8 @@ impl PPOTrainer {
max_position: params.max_position_absolute as f32,
tx_cost_bps: self.tx_cost_bps as f32,
spread_cost: (self.tick_size * self.spread_ticks) as f32,
initial_capital: 35_000.0,
initial_capital: self.initial_capital as f32,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(

View File

@@ -609,6 +609,7 @@ impl DQNTrainer {
sharpe_window: 20, // WAVE 26 P1.3: Standard 20-period window
use_dsr: hyperparams.use_dsr,
dsr_eta: hyperparams.dsr_eta,
initial_capital: hyperparams.initial_capital as f64,
};
let reward_fn = RewardFunction::new_with_debug(reward_config, debug_logging)?;
@@ -3712,10 +3713,10 @@ impl DQNTrainer {
// Compute validation loss
let val_loss = self.compute_validation_loss().await?;
info!(
"Epoch {}/{}: val_loss={:.6}",
"Epoch {}/{}: val_loss={} (backtest Sharpe proxy)",
epoch + 1,
self.hyperparams.epochs,
val_loss
if val_loss.abs() < 1e-10 { "N/A".to_owned() } else { format!("{val_loss:.6}") }
);
// Per-epoch Prometheus metrics for monitoring service
@@ -3742,13 +3743,16 @@ impl DQNTrainer {
// Adaptive tau: detect Q-value overestimation and increase Polyak averaging
if epoch > 0 {
let q_mean_growth = q_mean - self.prev_epoch_q_mean;
if q_mean_growth > 0.5 {
// After percentage-reward fix (denormalized portfolio), Q-values are
// on a ~0.01/step scale, not raw dollars. Old thresholds (0.5 / 0.1)
// would never trigger.
if q_mean_growth > 0.005 {
self.adaptive_tau = (self.adaptive_tau * 2.0).min(0.01);
warn!(
"Q-value drift detected (Δ={:.3}), increasing tau to {:.4}",
q_mean_growth, self.adaptive_tau
);
} else if q_mean_growth < 0.1 {
} else if q_mean_growth < 0.001 {
self.adaptive_tau = (self.adaptive_tau * 0.9).max(self.hyperparams.tau);
} else {
// Q-value growth in normal range — keep current tau

View File

@@ -134,8 +134,11 @@ pub fn generate_walk_forward_windows(
None => break,
};
// If test_end exceeds data range, we cannot form this fold
if test_end > data_end {
// If test_end exceeds data range, we cannot form this fold.
// Use `>` with +1 day margin: bars with `date_naive() < test_end` captures
// all bars ON the last calendar day of the test window. Without the margin,
// quarterly data ending on e.g. Jun-30 fails when test_end = Jul-01.
if test_end > data_end + chrono::Duration::days(1) {
break;
}

View File

@@ -103,6 +103,7 @@ mod gpu_tests {
tx_cost_bps: 0.0, // zero costs for a clean signal
spread_cost: 0.0,
initial_capital: 100_000.0,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(
@@ -163,6 +164,7 @@ mod gpu_tests {
tx_cost_bps: 0.0,
spread_cost: 0.0,
initial_capital: 100_000.0,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(
@@ -263,6 +265,7 @@ mod gpu_tests {
tx_cost_bps: 0.0,
spread_cost: 0.0,
initial_capital: 100_000.0,
..Default::default()
};
let mut evaluator = GpuBacktestEvaluator::new(

View File

@@ -1,9 +1,9 @@
# Generic GPU runtime base image for Foxhunt training pods
# Contains: CUDA 12.9 + cuDNN + NVRTC + rclone
# Contains: CUDA 12.9 + NVRTC + rclone (no cuDNN — zero conv ops in codebase)
# Binaries are fetched by initContainer at pod startup, not baked into this image.
# Rebuild: only on CUDA version bump
FROM nvidia/cuda:12.9.1-cudnn-runtime-ubuntu24.04
FROM nvidia/cuda:12.9.1-runtime-ubuntu24.04
ENV DEBIAN_FRONTEND=noninteractive

View File

@@ -95,6 +95,70 @@ spec:
matchLabels:
app.kubernetes.io/name: minio
---
# Compile-and-deploy workflow pods: need GitLab HTTP (tagging/releases), MinIO,
# k8s API (deploy rollouts + Argo executor), DNS, GitLab SSH, external HTTPS.
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy
metadata:
name: argo-compile-and-deploy-workflow
namespace: foxhunt
labels:
app.kubernetes.io/part-of: foxhunt
spec:
podSelector:
matchLabels:
app.kubernetes.io/component: compile-and-deploy
policyTypes:
- Egress
egress:
# DNS
- ports:
- port: 53
protocol: UDP
- port: 53
protocol: TCP
# Kubernetes API — Argo executor + kubectl rollout
- ports:
- port: 443
protocol: TCP
- port: 6443
protocol: TCP
to:
- ipBlock:
cidr: 10.32.0.0/16
- ipBlock:
cidr: 172.16.0.4/32
# External HTTPS — cargo fetches, kubectl download
- ports:
- port: 443
protocol: TCP
# GitLab SSH — git clone
- ports:
- port: 2222
protocol: TCP
to:
- ipBlock:
cidr: 100.90.76.85/32
- podSelector:
matchLabels:
app: gitlab-shell
# GitLab webservice — CalVer tagging, releases, package uploads
- ports:
- port: 8181
protocol: TCP
to:
- podSelector:
matchLabels:
app: webservice
# MinIO — binary upload + log archival
- ports:
- port: 9000
protocol: TCP
to:
- podSelector:
matchLabels:
app.kubernetes.io/name: minio
---
# CI pipeline pods (detect-changes + compile-service inline templates)
apiVersion: networking.k8s.io/v1
kind: NetworkPolicy

View File

@@ -30,89 +30,27 @@ spec:
items:
- key: .dockerconfigjson
path: config.json
- name: cargo-target-cpu
persistentVolumeClaim:
claimName: cargo-target-cpu
- name: cargo-target-cuda
persistentVolumeClaim:
claimName: cargo-target-cuda
- name: gitlab-pat
secret:
secretName: gitlab-pat
arguments:
parameters:
- name: commit-sha
value: HEAD
- name: commits-json
value: "[]"
- name: cuda-compute-cap
value: "90"
templates:
# ── DAG: orchestrate pipeline ──
# Compile + deploy moved to compile-and-deploy-template.yaml (manual trigger).
# This pipeline handles infra, docker images, dashboard, and argo templates only.
- name: pipeline
dag:
tasks:
- name: detect-changes
template: detect-changes
- name: create-tag
depends: "detect-changes"
template: create-tag
when: "{{tasks.detect-changes.outputs.parameters.needs-code}} == true"
- name: compile-services
depends: "create-tag"
template: compile-services
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
- name: service-packages
value: "{{tasks.detect-changes.outputs.parameters.service-packages}}"
when: "{{tasks.detect-changes.outputs.parameters.needs-services}} == true"
- name: compile-training
depends: "create-tag"
template: compile-training
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
- name: training-examples
value: "{{tasks.detect-changes.outputs.parameters.training-examples}}"
when: "{{tasks.detect-changes.outputs.parameters.needs-training}} == true"
# gpu-warmup disabled — only needed in compile-and-train-template
# - name: gpu-warmup
# depends: "detect-changes"
# template: gpu-warmup
# when: "{{tasks.detect-changes.outputs.parameters.needs-training}} == true"
- name: build-web-dashboard
depends: "detect-changes"
template: build-web-dashboard
when: "{{tasks.detect-changes.outputs.parameters.needs-dashboard}} == true"
- name: upload-release
depends: "(compile-services.Succeeded || compile-services.Skipped) && (compile-training.Succeeded || compile-training.Skipped)"
template: upload-release
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
when: "{{tasks.detect-changes.outputs.parameters.needs-code}} == true"
- name: deploy-services
depends: "upload-release.Succeeded || upload-release.Skipped"
template: deploy-services
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
- name: deploy-list
value: "{{tasks.detect-changes.outputs.parameters.deploy-list}}"
when: "{{tasks.detect-changes.outputs.parameters.needs-services}} == true"
- name: rebuild-ci-builder
depends: "detect-changes"
templateRef:
@@ -483,71 +421,6 @@ spec:
valueFrom:
path: /tmp/outputs/needs-argo-templates
# ── create-tag: CalVer auto-tag on code changes ──
- name: create-tag
nodeSelector:
k8s.scaleway.com/pool-name: platform
container:
image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest
command: ["/bin/sh", "-c"]
env:
- name: GITLAB_PAT
valueFrom:
secretKeyRef:
name: gitlab-pat
key: token
resources:
requests:
cpu: 100m
memory: 64Mi
limits:
cpu: 200m
memory: 128Mi
args:
- |
set -e
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
PROJECT_ID=1
SHA="{{workflow.parameters.commit-sha}}"
# Compute CalVer prefix: vYYYY.MM
PREFIX="v$(date +%Y.%m)"
# Query existing tags for this month
TAGS=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/tags?search=${PREFIX}" \
|| echo "[]")
# Parse highest N from vYYYY.MM.N tags
LAST_N=$(echo "$TAGS" | grep -oP "\"name\":\"${PREFIX}\.\K[0-9]+" | sort -n | tail -1)
if [ -z "$LAST_N" ]; then
NEXT_N=1
else
NEXT_N=$((LAST_N + 1))
fi
TAG="${PREFIX}.${NEXT_N}"
echo "Creating tag: ${TAG} at ${SHA}"
# Create the tag
RESULT=$(curl -sf -X POST \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
-d "tag_name=${TAG}" \
-d "ref=${SHA}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/tags")
echo "$RESULT" | grep -q "$TAG" || { echo "Tag creation failed: $RESULT"; exit 1; }
echo "Tag ${TAG} created successfully"
mkdir -p /tmp/outputs
echo -n "$TAG" > /tmp/outputs/tag
outputs:
parameters:
- name: tag
valueFrom:
path: /tmp/outputs/tag
# ── build-web-dashboard: npm build + upload to MinIO ──
- name: build-web-dashboard
nodeSelector:
@@ -611,520 +484,6 @@ spec:
echo "=== Web dashboard build + upload done ==="
# ── compile-services: selective per-binary build, incremental on local RWO PVC ──
- name: compile-services
metadata:
labels:
app.kubernetes.io/component: compile
inputs:
parameters:
- name: tag
- name: service-packages
nodeSelector:
k8s.scaleway.com/pool-name: ci-compile-cpu
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-cpu:latest
imagePullPolicy: Always
command: ["/bin/sh", "-c"]
env:
- name: SQLX_OFFLINE
value: "true"
- name: CARGO_TERM_COLOR
value: always
- name: CARGO_TARGET_DIR
value: /cargo-target
- name: CARGO_HOME
value: /cargo-target/cargo-home
- name: GITLAB_PAT
valueFrom:
secretKeyRef:
name: gitlab-pat
key: token
resources:
requests:
cpu: "14"
memory: 16Gi
limits:
cpu: "30"
memory: 32Gi
volumeMounts:
- name: git-ssh-key
mountPath: /etc/git-ssh
readOnly: true
- name: cargo-target-cpu
mountPath: /cargo-target
args:
- |
set -e
SHA="{{workflow.parameters.commit-sha}}"
SERVICE_PKGS="{{inputs.parameters.service-packages}}"
# SSH setup
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
chmod 600 ~/.ssh/config
REPO="ssh://git@gitlab-gitlab-shell.foxhunt.svc.cluster.local:2222/root/foxhunt.git"
WORKSPACE="/cargo-target/src"
# Persistent checkout on PVC — only changed files get new mtimes,
# so cargo skips recompiling unchanged workspace crates.
if [ -d "$WORKSPACE/.git" ]; then
cd "$WORKSPACE"
CURRENT=$(git rev-parse HEAD 2>/dev/null || echo "none")
if [ "$CURRENT" = "$SHA" ]; then
echo "=== Already at $SHA, skipping checkout ==="
else
echo "=== Updating checkout: $(echo $CURRENT | cut -c1-8) -> $(echo $SHA | cut -c1-8) ==="
git fetch origin
git checkout --force "$SHA"
git clean -fd
fi
else
echo "=== Initial clone (first run) ==="
git clone --filter=blob:none "$REPO" "$WORKSPACE"
cd "$WORKSPACE"
git checkout "$SHA"
fi
echo "Checked out $(git rev-parse --short HEAD)"
# Ensure cargo home registry is on PVC
export PATH="${CARGO_HOME}/bin:${PATH}"
export FOXHUNT_BUILD_VERSION="{{inputs.parameters.tag}}"
# Prune build artifacts if PVC exceeds 25GB (prevents unbounded growth)
TARGET_SIZE_MB=$(du -sm "$CARGO_TARGET_DIR" 2>/dev/null | cut -f1 || echo 0)
echo "PVC usage: ${TARGET_SIZE_MB}MB"
if [ "$TARGET_SIZE_MB" -gt 25000 ]; then
echo "PVC exceeds 25GB limit, pruning build artifacts..."
rm -rf "$CARGO_TARGET_DIR/release" "$CARGO_TARGET_DIR/debug"
fi
# Guard: empty package list would build entire workspace
if [ -z "$SERVICE_PKGS" ]; then
echo "ERROR: service-packages is empty, refusing to build entire workspace"
exit 1
fi
# Build only the affected service packages (incremental via persistent target dir)
CARGO_ARGS=""
for pkg in $SERVICE_PKGS; do
CARGO_ARGS="$CARGO_ARGS -p $pkg"
done
echo "=== Building service binaries: $SERVICE_PKGS (incremental) ==="
cargo build --release $CARGO_ARGS
# Collect built binaries
mkdir -p "$WORKSPACE/bin/services"
for pkg in $SERVICE_PKGS; do
bin_name=$(echo "$pkg" | tr '-' '_')
cp "$CARGO_TARGET_DIR/release/$pkg" "$WORKSPACE/bin/services/" 2>/dev/null \
|| cp "$CARGO_TARGET_DIR/release/$bin_name" "$WORKSPACE/bin/services/" 2>/dev/null \
|| { echo "Binary not found for $pkg"; ls "$CARGO_TARGET_DIR/release/"; exit 1; }
done
strip "$WORKSPACE/bin/services/"*
echo "=== Service binaries ==="
ls -lh "$WORKSPACE/bin/services/"
echo "=== Uploading service binaries to GitLab packages ==="
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
TAG="${FOXHUNT_BUILD_VERSION}"
for bin in "$WORKSPACE/bin/services/"*; do
BIN_NAME=$(basename "$bin")
echo "Uploading ${BIN_NAME} (${TAG})..."
curl -f --upload-file "$bin" \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/generic/foxhunt-services/${TAG}/${BIN_NAME}"
done
# Update 'latest' per-file (preserves unbuilt binaries from prior runs)
echo "=== Updating 'latest' package ==="
LATEST_PKG=$(curl -sf -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages?package_name=foxhunt-services&package_version=latest" \
| grep -oP '"id":\K[0-9]+' | head -1)
for bin in "$WORKSPACE/bin/services/"*; do
BIN_NAME=$(basename "$bin")
# Delete existing file by name before uploading replacement
if [ -n "$LATEST_PKG" ]; then
FILE_ID=$(curl -sf -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/${LATEST_PKG}/package_files" \
| grep -oP "\"id\":([0-9]+),\"package_id\":${LATEST_PKG}[^}]*\"file_name\":\"${BIN_NAME}\"" \
| grep -oP '"id":\K[0-9]+' | head -1)
if [ -n "$FILE_ID" ]; then
curl -sf -X DELETE -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/${LATEST_PKG}/package_files/${FILE_ID}" || true
fi
fi
curl -f --upload-file "$bin" \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/generic/foxhunt-services/latest/${BIN_NAME}" || true
done
echo "=== Service compile + upload done ($SERVICE_PKGS) ==="
# ── compile-training: selective per-binary CUDA build, incremental via persistent target dir ──
- name: compile-training
metadata:
labels:
app.kubernetes.io/component: compile
inputs:
parameters:
- name: tag
- name: training-examples
nodeSelector:
k8s.scaleway.com/pool-name: ci-compile-cpu
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/sh", "-c"]
env:
- name: SQLX_OFFLINE
value: "true"
- name: CARGO_TERM_COLOR
value: always
- name: CARGO_TARGET_DIR
value: /cargo-target
- name: CARGO_HOME
value: /cargo-target/cargo-home
- 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
args:
- |
set -e
SHA="{{workflow.parameters.commit-sha}}"
TRAINING_EXAMPLES="{{inputs.parameters.training-examples}}"
# SSH setup
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
chmod 600 ~/.ssh/config
REPO="ssh://git@gitlab-gitlab-shell.foxhunt.svc.cluster.local:2222/root/foxhunt.git"
WORKSPACE="/cargo-target/src"
# Persistent checkout on PVC — only changed files get new mtimes,
# so cargo skips recompiling unchanged workspace crates.
if [ -d "$WORKSPACE/.git" ]; then
cd "$WORKSPACE"
CURRENT=$(git rev-parse HEAD 2>/dev/null || echo "none")
if [ "$CURRENT" = "$SHA" ]; then
echo "=== Already at $SHA, skipping checkout ==="
else
echo "=== Updating checkout: $(echo $CURRENT | cut -c1-8) -> $(echo $SHA | cut -c1-8) ==="
git fetch origin
git checkout --force "$SHA"
git clean -fd
fi
else
echo "=== Initial clone (first run) ==="
git clone --filter=blob:none "$REPO" "$WORKSPACE"
cd "$WORKSPACE"
git checkout "$SHA"
fi
echo "Checked out $(git rev-parse --short HEAD)"
# Ensure cargo home registry is on PVC
export PATH="${CARGO_HOME}/bin:${PATH}"
export FOXHUNT_BUILD_VERSION="{{inputs.parameters.tag}}"
export CUDA_COMPUTE_CAP={{workflow.parameters.cuda-compute-cap}}
# Prune build artifacts if PVC exceeds 25GB (prevents unbounded growth)
TARGET_SIZE_MB=$(du -sm "$CARGO_TARGET_DIR" 2>/dev/null | cut -f1 || echo 0)
echo "PVC usage: ${TARGET_SIZE_MB}MB"
if [ "$TARGET_SIZE_MB" -gt 25000 ]; then
echo "PVC exceeds 25GB limit, pruning build artifacts..."
rm -rf "$CARGO_TARGET_DIR/release" "$CARGO_TARGET_DIR/debug"
fi
# Separate ml examples from training_uploader (different build commands)
ML_EXAMPLE_ARGS=""
BUILD_UPLOADER="false"
for ex in $TRAINING_EXAMPLES; do
case "$ex" in
training_uploader)
BUILD_UPLOADER="true"
;;
*)
ML_EXAMPLE_ARGS="$ML_EXAMPLE_ARGS --example $ex"
;;
esac
done
# Guard: empty example list would be a no-op waste of a compile pod
if [ -z "$TRAINING_EXAMPLES" ]; then
echo "ERROR: training-examples is empty, nothing to build"
exit 1
fi
echo "=== Building training binaries: $TRAINING_EXAMPLES (CUDA_COMPUTE_CAP=$CUDA_COMPUTE_CAP, incremental) ==="
if [ -n "$ML_EXAMPLE_ARGS" ]; then
cargo build --release -p ml --features ml/cuda $ML_EXAMPLE_ARGS
fi
if [ "$BUILD_UPLOADER" = "true" ]; then
cargo build --release -p training_uploader
fi
# Collect built binaries
mkdir -p "$WORKSPACE/bin/training"
for ex in $TRAINING_EXAMPLES; do
case "$ex" in
training_uploader)
cp "$CARGO_TARGET_DIR/release/training_uploader" "$WORKSPACE/bin/training/"
;;
*)
cp "$CARGO_TARGET_DIR/release/examples/$ex" "$WORKSPACE/bin/training/"
;;
esac
done
strip "$WORKSPACE/bin/training/"*
echo "=== Training binaries ==="
ls -lh "$WORKSPACE/bin/training/"
echo "=== Uploading training binaries to GitLab packages ==="
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
TAG="${FOXHUNT_BUILD_VERSION}"
for bin in "$WORKSPACE/bin/training/"*; do
BIN_NAME=$(basename "$bin")
echo "Uploading ${BIN_NAME} (${TAG})..."
curl -f --upload-file "$bin" \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/generic/foxhunt-training/${TAG}/${BIN_NAME}"
done
# Update 'latest' per-file (preserves unbuilt binaries from prior runs)
echo "=== Updating 'latest' package ==="
LATEST_PKG=$(curl -sf -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages?package_name=foxhunt-training&package_version=latest" \
| grep -oP '"id":\K[0-9]+' | head -1)
for bin in "$WORKSPACE/bin/training/"*; do
BIN_NAME=$(basename "$bin")
if [ -n "$LATEST_PKG" ]; then
FILE_ID=$(curl -sf -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/${LATEST_PKG}/package_files" \
| grep -oP "\"id\":([0-9]+),\"package_id\":${LATEST_PKG}[^}]*\"file_name\":\"${BIN_NAME}\"" \
| grep -oP '"id":\K[0-9]+' | head -1)
if [ -n "$FILE_ID" ]; then
curl -sf -X DELETE -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/${LATEST_PKG}/package_files/${FILE_ID}" || true
fi
fi
curl -f --upload-file "$bin" \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/generic/foxhunt-training/latest/${BIN_NAME}" || true
done
echo "=== Training compile + upload done ($TRAINING_EXAMPLES) ==="
# ── upload-release: create GitLab Release with package links ──
- name: upload-release
inputs:
parameters:
- name: tag
nodeSelector:
k8s.scaleway.com/pool-name: platform
container:
image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest
command: ["/bin/sh", "-c"]
env:
- name: GITLAB_PAT
valueFrom:
secretKeyRef:
name: gitlab-pat
key: token
resources:
requests:
cpu: 100m
memory: 64Mi
limits:
cpu: 200m
memory: 128Mi
args:
- |
set -e
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
PROJECT_ID=1
TAG="{{inputs.parameters.tag}}"
echo "=== Creating GitLab Release ${TAG} ==="
# Get commits since previous tag for release notes
PREV_TAG=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/tags?per_page=2" \
| grep -oP '"name":"\K[^"]+' | sed -n '2p')
if [ -n "$PREV_TAG" ]; then
COMMITS=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/compare?from=${PREV_TAG}&to=${TAG}" \
| grep -oP '"title":"\K[^"]+' | head -20 \
| sed 's/^/- /' || echo "- Release ${TAG}")
DESCRIPTION="## Changes since ${PREV_TAG}\n\n${COMMITS}"
else
DESCRIPTION="## Initial release\n\nFirst CalVer release."
fi
# Create the release
RESPONSE=$(curl -sf -X POST \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
-H "Content-Type: application/json" \
-d "{
\"tag_name\": \"${TAG}\",
\"name\": \"${TAG}\",
\"description\": \"$(printf '%s' "$DESCRIPTION" | sed 's/"/\\"/g')\"
}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/releases") || {
echo "WARN: Release creation failed (may already exist)"
}
echo "Release ${TAG} created"
echo "$RESPONSE" | head -5
# Add package links as release assets
for pkg_name in foxhunt-services foxhunt-training; do
PKG_CHECK=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/packages?package_name=${pkg_name}&package_version=${TAG}" \
|| echo "[]")
if echo "$PKG_CHECK" | grep -q "$TAG"; then
curl -sf -X POST \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
-H "Content-Type: application/json" \
-d "{
\"name\": \"${pkg_name}\",
\"url\": \"${GITLAB}/-/packages?type=generic&search=${pkg_name}&version=${TAG}\",
\"link_type\": \"package\"
}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/releases/${TAG}/assets/links" || true
echo "Linked ${pkg_name} package to release"
fi
done
echo "=== Release ${TAG} complete ==="
# ── gpu-warmup: disabled — re-enable when GPU nodes stay warm ──
# - name: gpu-warmup
# nodeSelector:
# k8s.scaleway.com/pool-name: ci-training-h100
# tolerations:
# - key: nvidia.com/gpu
# operator: Exists
# effect: NoSchedule
# - key: node.cilium.io/agent-not-ready
# operator: Exists
# effect: NoSchedule
# container:
# image: busybox:1.37
# command: ["/bin/sh", "-c"]
# args:
# - |
# echo "GPU warmup: triggering node autoscale..."
# echo "GPU node scheduled, exiting to free resources"
# resources:
# requests:
# nvidia.com/gpu: "1"
# cpu: 100m
# memory: 64Mi
# limits:
# nvidia.com/gpu: "1"
# cpu: 200m
# memory: 256Mi
# ── deploy-services: selective rolling restart for affected deployments ──
- name: deploy-services
inputs:
parameters:
- name: tag
- name: deploy-list
serviceAccountName: argo-workflow
nodeSelector:
k8s.scaleway.com/pool-name: platform
container:
image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest
command: ["/bin/sh", "-c"]
resources:
requests:
cpu: 100m
memory: 64Mi
limits:
cpu: 200m
memory: 128Mi
args:
- |
set -e
if ! command -v kubectl >/dev/null 2>&1; then
echo "=== Installing kubectl ==="
curl -sLo /tmp/kubectl "https://dl.k8s.io/release/v1.31.4/bin/linux/amd64/kubectl"
chmod +x /tmp/kubectl
export PATH="/tmp:$PATH"
fi
TAG="{{inputs.parameters.tag}}"
DEPLOY_LIST="{{inputs.parameters.deploy-list}}"
if [ -z "$DEPLOY_LIST" ]; then
echo "=== No services to deploy ==="
exit 0
fi
echo "=== Deploying release ${TAG}: $DEPLOY_LIST ==="
for svc in $DEPLOY_LIST; do
echo "Patching $svc with FOXHUNT_RELEASE=${TAG}..."
kubectl -n foxhunt patch deployment "$svc" -p "{
\"spec\":{\"template\":{
\"metadata\":{\"annotations\":{\"foxhunt.io/release\":\"${TAG}\"}},
\"spec\":{\"initContainers\":[{
\"name\":\"fetch-binary\",
\"env\":[{\"name\":\"FOXHUNT_RELEASE\",\"value\":\"${TAG}\"}]
}]}
}}
}" || echo "WARN: $svc patch failed (may not exist)"
done
echo "=== Waiting for rollouts ==="
for svc in $DEPLOY_LIST; do
kubectl -n foxhunt rollout status deployment "$svc" --timeout=120s || echo "WARN: $svc rollout timeout"
done
echo "=== Deploy ${TAG} complete ($DEPLOY_LIST) ==="
# ── terragrunt-apply: apply infra changes on main push ──
- name: terragrunt-apply
inputs:

View File

@@ -0,0 +1,448 @@
# Compile-and-Deploy: manual workflow for building and deploying service binaries.
#
# DAG:
# create-tag ──> compile-services ──> upload-release ──> deploy-services
#
# Trigger manually when you want to build + deploy service binaries.
# Training compilation lives in compile-and-train-template.yaml.
---
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
name: compile-and-deploy
namespace: foxhunt
labels:
app.kubernetes.io/name: compile-and-deploy
app.kubernetes.io/part-of: foxhunt
spec:
entrypoint: pipeline
serviceAccountName: argo-workflow
podMetadata:
labels:
app.kubernetes.io/part-of: foxhunt
app.kubernetes.io/component: compile-and-deploy
podGC:
strategy: OnPodCompletion
ttlStrategy:
secondsAfterCompletion: 3600
activeDeadlineSeconds: 3600 # 1 hour (compile + deploy)
arguments:
parameters:
- name: commit-sha
value: HEAD
- name: service-packages
value: "api trading-service backtesting-service trading-agent-service broker-gateway data-acquisition-service"
volumes:
- name: git-ssh-key
secret:
secretName: argo-git-ssh-key
defaultMode: 256
- name: cargo-target-cpu
persistentVolumeClaim:
claimName: cargo-target-cpu
templates:
- name: pipeline
dag:
tasks:
- name: create-tag
template: create-tag
- name: compile-services
depends: "create-tag"
template: compile-services
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
- name: service-packages
value: "{{workflow.parameters.service-packages}}"
- name: upload-release
depends: "compile-services"
template: upload-release
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
- name: deploy-services
depends: "upload-release"
template: deploy-services
arguments:
parameters:
- name: tag
value: "{{tasks.create-tag.outputs.parameters.tag}}"
- name: deploy-list
value: "{{workflow.parameters.service-packages}}"
# ── create-tag: CalVer auto-tag on code changes ──
- name: create-tag
nodeSelector:
k8s.scaleway.com/pool-name: platform
container:
image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest
command: ["/bin/sh", "-c"]
env:
- name: GITLAB_PAT
valueFrom:
secretKeyRef:
name: gitlab-pat
key: token
resources:
requests:
cpu: 100m
memory: 64Mi
limits:
cpu: 200m
memory: 128Mi
args:
- |
set -e
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
PROJECT_ID=1
SHA="{{workflow.parameters.commit-sha}}"
# Compute CalVer prefix: vYYYY.MM
PREFIX="v$(date +%Y.%m)"
# Query existing tags for this month
TAGS=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/tags?search=${PREFIX}" \
|| echo "[]")
# Parse highest N from vYYYY.MM.N tags
LAST_N=$(echo "$TAGS" | grep -oP "\"name\":\"${PREFIX}\.\K[0-9]+" | sort -n | tail -1)
if [ -z "$LAST_N" ]; then
NEXT_N=1
else
NEXT_N=$((LAST_N + 1))
fi
TAG="${PREFIX}.${NEXT_N}"
echo "Creating tag: ${TAG} at ${SHA}"
# Create the tag
RESULT=$(curl -sf -X POST \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
-d "tag_name=${TAG}" \
-d "ref=${SHA}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/tags")
echo "$RESULT" | grep -q "$TAG" || { echo "Tag creation failed: $RESULT"; exit 1; }
echo "Tag ${TAG} created successfully"
mkdir -p /tmp/outputs
echo -n "$TAG" > /tmp/outputs/tag
outputs:
parameters:
- name: tag
valueFrom:
path: /tmp/outputs/tag
# ── compile-services: selective per-binary build, incremental on local RWO PVC ──
- name: compile-services
metadata:
labels:
app.kubernetes.io/component: compile
inputs:
parameters:
- name: tag
- name: service-packages
nodeSelector:
k8s.scaleway.com/pool-name: ci-compile-cpu
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-cpu:latest
imagePullPolicy: Always
command: ["/bin/sh", "-c"]
env:
- name: SQLX_OFFLINE
value: "true"
- name: CARGO_TERM_COLOR
value: always
- name: CARGO_TARGET_DIR
value: /cargo-target
- name: CARGO_HOME
value: /cargo-target/cargo-home
- name: GITLAB_PAT
valueFrom:
secretKeyRef:
name: gitlab-pat
key: token
resources:
requests:
cpu: "14"
memory: 16Gi
limits:
cpu: "30"
memory: 32Gi
volumeMounts:
- name: git-ssh-key
mountPath: /etc/git-ssh
readOnly: true
- name: cargo-target-cpu
mountPath: /cargo-target
args:
- |
set -e
SHA="{{workflow.parameters.commit-sha}}"
SERVICE_PKGS="{{inputs.parameters.service-packages}}"
# SSH setup
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
chmod 600 ~/.ssh/config
REPO="ssh://git@gitlab-gitlab-shell.foxhunt.svc.cluster.local:2222/root/foxhunt.git"
WORKSPACE="/cargo-target/src"
# Persistent checkout on PVC — only changed files get new mtimes,
# so cargo skips recompiling unchanged workspace crates.
if [ -d "$WORKSPACE/.git" ]; then
cd "$WORKSPACE"
CURRENT=$(git rev-parse HEAD 2>/dev/null || echo "none")
if [ "$CURRENT" = "$SHA" ]; then
echo "=== Already at $SHA, skipping checkout ==="
else
echo "=== Updating checkout: $(echo $CURRENT | cut -c1-8) -> $(echo $SHA | cut -c1-8) ==="
git fetch origin
git checkout --force "$SHA"
git clean -fd
fi
else
echo "=== Initial clone (first run) ==="
git clone --filter=blob:none "$REPO" "$WORKSPACE"
cd "$WORKSPACE"
git checkout "$SHA"
fi
echo "Checked out $(git rev-parse --short HEAD)"
# Ensure cargo home registry is on PVC
export PATH="${CARGO_HOME}/bin:${PATH}"
export FOXHUNT_BUILD_VERSION="{{inputs.parameters.tag}}"
# Prune build artifacts if PVC exceeds 25GB (prevents unbounded growth)
TARGET_SIZE_MB=$(du -sm "$CARGO_TARGET_DIR" 2>/dev/null | cut -f1 || echo 0)
echo "PVC usage: ${TARGET_SIZE_MB}MB"
if [ "$TARGET_SIZE_MB" -gt 25000 ]; then
echo "PVC exceeds 25GB limit, pruning build artifacts..."
rm -rf "$CARGO_TARGET_DIR/release" "$CARGO_TARGET_DIR/debug"
fi
# Guard: empty package list would build entire workspace
if [ -z "$SERVICE_PKGS" ]; then
echo "ERROR: service-packages is empty, refusing to build entire workspace"
exit 1
fi
# Build only the affected service packages (incremental via persistent target dir)
CARGO_ARGS=""
for pkg in $SERVICE_PKGS; do
CARGO_ARGS="$CARGO_ARGS -p $pkg"
done
echo "=== Building service binaries: $SERVICE_PKGS (incremental) ==="
cargo build --release $CARGO_ARGS
# Collect built binaries
mkdir -p "$WORKSPACE/bin/services"
for pkg in $SERVICE_PKGS; do
bin_name=$(echo "$pkg" | tr '-' '_')
cp "$CARGO_TARGET_DIR/release/$pkg" "$WORKSPACE/bin/services/" 2>/dev/null \
|| cp "$CARGO_TARGET_DIR/release/$bin_name" "$WORKSPACE/bin/services/" 2>/dev/null \
|| { echo "Binary not found for $pkg"; ls "$CARGO_TARGET_DIR/release/"; exit 1; }
done
strip "$WORKSPACE/bin/services/"*
echo "=== Service binaries ==="
ls -lh "$WORKSPACE/bin/services/"
echo "=== Uploading service binaries to GitLab packages ==="
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
TAG="${FOXHUNT_BUILD_VERSION}"
for bin in "$WORKSPACE/bin/services/"*; do
BIN_NAME=$(basename "$bin")
echo "Uploading ${BIN_NAME} (${TAG})..."
curl -f --upload-file "$bin" \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/generic/foxhunt-services/${TAG}/${BIN_NAME}"
done
# Update 'latest' per-file (preserves unbuilt binaries from prior runs)
echo "=== Updating 'latest' package ==="
LATEST_PKG=$(curl -sf -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages?package_name=foxhunt-services&package_version=latest" \
| grep -oP '"id":\K[0-9]+' | head -1)
for bin in "$WORKSPACE/bin/services/"*; do
BIN_NAME=$(basename "$bin")
# Delete existing file by name before uploading replacement
if [ -n "$LATEST_PKG" ]; then
FILE_ID=$(curl -sf -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/${LATEST_PKG}/package_files" \
| grep -oP "\"id\":([0-9]+),\"package_id\":${LATEST_PKG}[^}]*\"file_name\":\"${BIN_NAME}\"" \
| grep -oP '"id":\K[0-9]+' | head -1)
if [ -n "$FILE_ID" ]; then
curl -sf -X DELETE -H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/${LATEST_PKG}/package_files/${FILE_ID}" || true
fi
fi
curl -f --upload-file "$bin" \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/1/packages/generic/foxhunt-services/latest/${BIN_NAME}" || true
done
echo "=== Service compile + upload done ($SERVICE_PKGS) ==="
# ── upload-release: create GitLab Release with package links ──
- name: upload-release
inputs:
parameters:
- name: tag
nodeSelector:
k8s.scaleway.com/pool-name: platform
container:
image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest
command: ["/bin/sh", "-c"]
env:
- name: GITLAB_PAT
valueFrom:
secretKeyRef:
name: gitlab-pat
key: token
resources:
requests:
cpu: 100m
memory: 64Mi
limits:
cpu: 200m
memory: 128Mi
args:
- |
set -e
GITLAB="http://gitlab-webservice-default.foxhunt.svc.cluster.local:8181"
PROJECT_ID=1
TAG="{{inputs.parameters.tag}}"
echo "=== Creating GitLab Release ${TAG} ==="
# Get commits since previous tag for release notes
PREV_TAG=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/tags?per_page=2" \
| grep -oP '"name":"\K[^"]+' | sed -n '2p')
if [ -n "$PREV_TAG" ]; then
COMMITS=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/repository/compare?from=${PREV_TAG}&to=${TAG}" \
| grep -oP '"title":"\K[^"]+' | head -20 \
| sed 's/^/- /' || echo "- Release ${TAG}")
DESCRIPTION="## Changes since ${PREV_TAG}\n\n${COMMITS}"
else
DESCRIPTION="## Initial release\n\nFirst CalVer release."
fi
# Create the release
RESPONSE=$(curl -sf -X POST \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
-H "Content-Type: application/json" \
-d "{
\"tag_name\": \"${TAG}\",
\"name\": \"${TAG}\",
\"description\": \"$(printf '%s' "$DESCRIPTION" | sed 's/"/\\"/g')\"
}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/releases") || {
echo "WARN: Release creation failed (may already exist)"
}
echo "Release ${TAG} created"
echo "$RESPONSE" | head -5
# Add package links as release assets
for pkg_name in foxhunt-services foxhunt-training; do
PKG_CHECK=$(curl -sf \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/packages?package_name=${pkg_name}&package_version=${TAG}" \
|| echo "[]")
if echo "$PKG_CHECK" | grep -q "$TAG"; then
curl -sf -X POST \
-H "PRIVATE-TOKEN: ${GITLAB_PAT}" \
-H "Content-Type: application/json" \
-d "{
\"name\": \"${pkg_name}\",
\"url\": \"${GITLAB}/-/packages?type=generic&search=${pkg_name}&version=${TAG}\",
\"link_type\": \"package\"
}" \
"${GITLAB}/api/v4/projects/${PROJECT_ID}/releases/${TAG}/assets/links" || true
echo "Linked ${pkg_name} package to release"
fi
done
echo "=== Release ${TAG} complete ==="
# ── deploy-services: selective rolling restart for affected deployments ──
- name: deploy-services
inputs:
parameters:
- name: tag
- name: deploy-list
serviceAccountName: argo-workflow
nodeSelector:
k8s.scaleway.com/pool-name: platform
container:
image: gitlab-registry.foxhunt.svc.cluster.local:5000/root/foxhunt/foxhunt-runtime:latest
command: ["/bin/sh", "-c"]
resources:
requests:
cpu: 100m
memory: 64Mi
limits:
cpu: 200m
memory: 128Mi
args:
- |
set -e
if ! command -v kubectl >/dev/null 2>&1; then
echo "=== Installing kubectl ==="
curl -sLo /tmp/kubectl "https://dl.k8s.io/release/v1.31.4/bin/linux/amd64/kubectl"
chmod +x /tmp/kubectl
export PATH="/tmp:$PATH"
fi
TAG="{{inputs.parameters.tag}}"
DEPLOY_LIST="{{inputs.parameters.deploy-list}}"
if [ -z "$DEPLOY_LIST" ]; then
echo "=== No services to deploy ==="
exit 0
fi
echo "=== Deploying release ${TAG}: $DEPLOY_LIST ==="
for svc in $DEPLOY_LIST; do
echo "Patching $svc with FOXHUNT_RELEASE=${TAG}..."
kubectl -n foxhunt patch deployment "$svc" -p "{
\"spec\":{\"template\":{
\"metadata\":{\"annotations\":{\"foxhunt.io/release\":\"${TAG}\"}},
\"spec\":{\"initContainers\":[{
\"name\":\"fetch-binary\",
\"env\":[{\"name\":\"FOXHUNT_RELEASE\",\"value\":\"${TAG}\"}]
}]}
}}
}" || echo "WARN: $svc patch failed (may not exist)"
done
echo "=== Waiting for rollouts ==="
for svc in $DEPLOY_LIST; do
kubectl -n foxhunt rollout status deployment "$svc" --timeout=120s || echo "WARN: $svc rollout timeout"
done
echo "=== Deploy ${TAG} complete ($DEPLOY_LIST) ==="

View File

@@ -0,0 +1,42 @@
# Manual workflow trigger EventSource.
#
# Webhook endpoints for triggering compile/deploy and training workflows.
# Each endpoint accepts a JSON POST body with workflow-specific parameters.
#
# Endpoints (all on port 12001):
# POST /compile-deploy — {"commit_sha": "HEAD"}
# POST /train/dqn — {"binary_tag": "latest"}
# POST /train/ppo — {"binary_tag": "latest"}
# POST /train/supervised — {"model": "tft", "binary_tag": "latest"}
#
# Example:
# curl -X POST http://workflow-trigger-eventsource-svc.foxhunt:12001/train/dqn \
# -H 'Content-Type: application/json' -d '{"binary_tag": "latest"}'
---
apiVersion: argoproj.io/v1alpha1
kind: EventSource
metadata:
name: workflow-trigger
namespace: foxhunt
labels:
app.kubernetes.io/name: workflow-trigger
app.kubernetes.io/part-of: foxhunt
spec:
eventBusName: default
webhook:
compile-deploy:
port: "12001"
endpoint: /compile-deploy
method: POST
train-dqn:
port: "12001"
endpoint: /train/dqn
method: POST
train-ppo:
port: "12001"
endpoint: /train/ppo
method: POST
train-supervised:
port: "12001"
endpoint: /train/supervised
method: POST

View File

@@ -0,0 +1,172 @@
# Sensor for manual workflow triggers.
#
# Listens on workflow-trigger EventSource and submits the appropriate workflow
# when a webhook endpoint is hit. Each trigger maps JSON body fields to
# workflow parameters.
#
# Trigger reference:
# compile-deploy — POST /compile-deploy {"commit_sha": "..."}
# train-dqn — POST /train/dqn {"binary_tag": "..."}
# train-ppo — POST /train/ppo {"binary_tag": "..."}
# train-supervised — POST /train/supervised {"model": "...", "binary_tag": "..."}
---
apiVersion: argoproj.io/v1alpha1
kind: Sensor
metadata:
name: workflow-trigger
namespace: foxhunt
labels:
app.kubernetes.io/name: workflow-trigger
app.kubernetes.io/part-of: foxhunt
spec:
template:
serviceAccountName: argo-workflow
nodeSelector:
node.kubernetes.io/instance-type: DEV1-L
eventBusName: default
dependencies:
- name: compile-deploy-dep
eventSourceName: workflow-trigger
eventName: compile-deploy
- name: train-dqn-dep
eventSourceName: workflow-trigger
eventName: train-dqn
- name: train-ppo-dep
eventSourceName: workflow-trigger
eventName: train-ppo
- name: train-supervised-dep
eventSourceName: workflow-trigger
eventName: train-supervised
triggers:
# ── Compile + Deploy ──
# POST /compile-deploy {"commit_sha": "abc1234"}
# Omit service_packages to build all 7 services (default).
- template:
name: compile-deploy
conditions: compile-deploy-dep
argoWorkflow:
operation: submit
source:
resource:
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: compile-deploy-
namespace: foxhunt
spec:
serviceAccountName: argo-workflow
podGC:
strategy: OnPodCompletion
ttlStrategy:
secondsAfterCompletion: 3600
workflowTemplateRef:
name: compile-and-deploy
arguments:
parameters:
- name: commit-sha
parameters:
- src:
dependencyName: compile-deploy-dep
dataKey: body.commit_sha
dest: spec.arguments.parameters.0.value
# ── Train DQN ──
# POST /train/dqn {"binary_tag": "latest"}
- template:
name: train-dqn
conditions: train-dqn-dep
argoWorkflow:
operation: submit
source:
resource:
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: train-dqn-
namespace: foxhunt
spec:
serviceAccountName: argo-workflow
podGC:
strategy: OnPodCompletion
ttlStrategy:
secondsAfterCompletion: 3600
workflowTemplateRef:
name: train-dqn
arguments:
parameters:
- name: binary-tag
parameters:
- src:
dependencyName: train-dqn-dep
dataKey: body.binary_tag
dest: spec.arguments.parameters.0.value
# ── Train PPO ──
# POST /train/ppo {"binary_tag": "latest"}
- template:
name: train-ppo
conditions: train-ppo-dep
argoWorkflow:
operation: submit
source:
resource:
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: train-ppo-
namespace: foxhunt
spec:
serviceAccountName: argo-workflow
podGC:
strategy: OnPodCompletion
ttlStrategy:
secondsAfterCompletion: 3600
workflowTemplateRef:
name: train-ppo
arguments:
parameters:
- name: binary-tag
parameters:
- src:
dependencyName: train-ppo-dep
dataKey: body.binary_tag
dest: spec.arguments.parameters.0.value
# ── Train Supervised ──
# POST /train/supervised {"model": "tft", "binary_tag": "latest"}
# model: one of tft, mamba2, tggn, tlob, liquid, kan, xlstm, diffusion
- template:
name: train-supervised
conditions: train-supervised-dep
argoWorkflow:
operation: submit
source:
resource:
apiVersion: argoproj.io/v1alpha1
kind: Workflow
metadata:
generateName: train-supervised-
namespace: foxhunt
spec:
serviceAccountName: argo-workflow
podGC:
strategy: OnPodCompletion
ttlStrategy:
secondsAfterCompletion: 3600
workflowTemplateRef:
name: train-supervised
arguments:
parameters:
- name: model
- name: binary-tag
parameters:
- src:
dependencyName: train-supervised-dep
dataKey: body.model
dest: spec.arguments.parameters.0.value
- src:
dependencyName: train-supervised-dep
dataKey: body.binary_tag
dest: spec.arguments.parameters.1.value

View File

@@ -0,0 +1,16 @@
apiVersion: v1
kind: Service
metadata:
name: workflow-trigger-eventsource-svc
namespace: foxhunt
labels:
app.kubernetes.io/name: workflow-trigger
app.kubernetes.io/part-of: foxhunt
spec:
type: ClusterIP
ports:
- port: 12001
targetPort: 12001
protocol: TCP
selector:
eventsource-name: workflow-trigger

View File

@@ -3,8 +3,12 @@ kind: Kustomization
resources:
- sccache-pvcs.yaml
- ci-pipeline-template.yaml
- compile-and-deploy-template.yaml
- compile-and-train-template.yaml
- build-ci-image-template.yaml
- training-workflow-template.yaml
- train-dqn-template.yaml
- train-ppo-template.yaml
- train-supervised-template.yaml
- argo-workflow-netpol.yaml
- ci-deploy-rbac.yaml

View File

@@ -0,0 +1,111 @@
# Train DQN: fetch binaries → hyperopt → train → evaluate → upload.
#
# Usage:
# argo submit --from=wftmpl/train-dqn -p binary-tag=dev-abc1234
# argo submit --from=wftmpl/train-dqn -p binary-tag=latest -p hyperopt-trials=50
#
# Thin wrapper around training-pipeline (templateRef) with DQN-tuned defaults.
# DQN has a 28D hyperopt parameter space — more trials recommended.
---
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
name: train-dqn
namespace: foxhunt
labels:
app.kubernetes.io/name: train-dqn
app.kubernetes.io/part-of: foxhunt
spec:
entrypoint: pipeline
serviceAccountName: argo-workflow
podMetadata:
labels:
app.kubernetes.io/part-of: foxhunt
app.kubernetes.io/component: training
securityContext:
runAsUser: 1000
runAsGroup: 1000
fsGroup: 1000
ttlStrategy:
secondsAfterCompletion: 3600
activeDeadlineSeconds: 28800 # 8 hours
arguments:
parameters:
- name: model
value: dqn
- name: binary-tag
value: latest
- name: gpu-pool
value: ci-training-h100
- name: hyperopt-trials
value: "30" # 28D param space needs more exploration
- name: hyperopt-epochs
value: "8"
- name: train-epochs
value: "80" # DQN benefits from longer training
- name: symbol
value: ES.FUT
- name: data-dir
value: /data/futures-baseline
- name: mbp10-data-dir
value: /data/futures-baseline-mbp10
- name: trades-data-dir
value: /data/futures-baseline-trades
- name: tx-cost-bps
value: "0.1"
- name: tick-size
value: "0.25"
- name: spread-ticks
value: "1.0"
- name: initial-capital
value: "35000"
- name: ensemble-top-k
value: "5"
- name: cuda-compute-cap
value: "90"
volumes:
- name: training-data
persistentVolumeClaim:
claimName: training-data-pvc
readOnly: true
volumeClaimTemplates:
- metadata:
name: workspace
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: scw-bssd
resources:
requests:
storage: 5Gi
templates:
- name: pipeline
dag:
tasks:
- name: fetch-binary
templateRef:
name: training-pipeline
template: fetch-binary
- name: hyperopt
templateRef:
name: training-pipeline
template: hyperopt
dependencies: [fetch-binary]
- name: train-best
templateRef:
name: training-pipeline
template: train-best
dependencies: [hyperopt]
- name: evaluate
templateRef:
name: training-pipeline
template: evaluate
dependencies: [train-best]
- name: upload-results
templateRef:
name: training-pipeline
template: upload-results
dependencies: [evaluate]

View File

@@ -0,0 +1,111 @@
# Train PPO: fetch binaries → hyperopt → train → evaluate → upload.
#
# Usage:
# argo submit --from=wftmpl/train-ppo -p binary-tag=dev-abc1234
# argo submit --from=wftmpl/train-ppo -p binary-tag=latest -p train-epochs=100
#
# Thin wrapper around training-pipeline (templateRef) with PPO-tuned defaults.
# PPO uses 45 factored actions (5 exposure × 3 order × 3 urgency).
---
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
name: train-ppo
namespace: foxhunt
labels:
app.kubernetes.io/name: train-ppo
app.kubernetes.io/part-of: foxhunt
spec:
entrypoint: pipeline
serviceAccountName: argo-workflow
podMetadata:
labels:
app.kubernetes.io/part-of: foxhunt
app.kubernetes.io/component: training
securityContext:
runAsUser: 1000
runAsGroup: 1000
fsGroup: 1000
ttlStrategy:
secondsAfterCompletion: 3600
activeDeadlineSeconds: 28800 # 8 hours
arguments:
parameters:
- name: model
value: ppo
- name: binary-tag
value: latest
- name: gpu-pool
value: ci-training-h100
- name: hyperopt-trials
value: "20"
- name: hyperopt-epochs
value: "8"
- name: train-epochs
value: "50"
- name: symbol
value: ES.FUT
- name: data-dir
value: /data/futures-baseline
- name: mbp10-data-dir
value: /data/futures-baseline-mbp10
- name: trades-data-dir
value: /data/futures-baseline-trades
- name: tx-cost-bps
value: "0.1"
- name: tick-size
value: "0.25"
- name: spread-ticks
value: "1.0"
- name: initial-capital
value: "35000"
- name: ensemble-top-k
value: "5"
- name: cuda-compute-cap
value: "90"
volumes:
- name: training-data
persistentVolumeClaim:
claimName: training-data-pvc
readOnly: true
volumeClaimTemplates:
- metadata:
name: workspace
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: scw-bssd
resources:
requests:
storage: 5Gi
templates:
- name: pipeline
dag:
tasks:
- name: fetch-binary
templateRef:
name: training-pipeline
template: fetch-binary
- name: hyperopt
templateRef:
name: training-pipeline
template: hyperopt
dependencies: [fetch-binary]
- name: train-best
templateRef:
name: training-pipeline
template: train-best
dependencies: [hyperopt]
- name: evaluate
templateRef:
name: training-pipeline
template: evaluate
dependencies: [train-best]
- name: upload-results
templateRef:
name: training-pipeline
template: upload-results
dependencies: [evaluate]

View File

@@ -0,0 +1,115 @@
# Train supervised model: fetch binaries → hyperopt → train → evaluate → upload.
#
# Usage (any of the 8 supervised models):
# argo submit --from=wftmpl/train-supervised -p model=tft -p binary-tag=latest
# argo submit --from=wftmpl/train-supervised -p model=mamba2 -p binary-tag=dev-abc1234
# argo submit --from=wftmpl/train-supervised -p model=kan -p binary-tag=latest -p train-epochs=100
#
# Supported models: tft, mamba2, tggn, tlob, liquid, kan, xlstm, diffusion
#
# Thin wrapper around training-pipeline (templateRef) with supervised defaults.
# Uses hyperopt_baseline_supervised / train_baseline_supervised / evaluate_supervised.
---
apiVersion: argoproj.io/v1alpha1
kind: WorkflowTemplate
metadata:
name: train-supervised
namespace: foxhunt
labels:
app.kubernetes.io/name: train-supervised
app.kubernetes.io/part-of: foxhunt
spec:
entrypoint: pipeline
serviceAccountName: argo-workflow
podMetadata:
labels:
app.kubernetes.io/part-of: foxhunt
app.kubernetes.io/component: training
securityContext:
runAsUser: 1000
runAsGroup: 1000
fsGroup: 1000
ttlStrategy:
secondsAfterCompletion: 3600
activeDeadlineSeconds: 28800 # 8 hours
arguments:
parameters:
- name: model
# One of: tft, mamba2, tggn, tlob, liquid, kan, xlstm, diffusion
value: tft
- name: binary-tag
value: latest
- name: gpu-pool
value: ci-training-h100
- name: hyperopt-trials
value: "20"
- name: hyperopt-epochs
value: "8"
- name: train-epochs
value: "50"
- name: symbol
value: ES.FUT
- name: data-dir
value: /data/futures-baseline
- name: mbp10-data-dir
value: /data/futures-baseline-mbp10
- name: trades-data-dir
value: /data/futures-baseline-trades
- name: tx-cost-bps
value: "0.1"
- name: tick-size
value: "0.25"
- name: spread-ticks
value: "1.0"
- name: initial-capital
value: "35000"
- name: ensemble-top-k
value: "5"
- name: cuda-compute-cap
value: "90"
volumes:
- name: training-data
persistentVolumeClaim:
claimName: training-data-pvc
readOnly: true
volumeClaimTemplates:
- metadata:
name: workspace
spec:
accessModes: ["ReadWriteOnce"]
storageClassName: scw-bssd
resources:
requests:
storage: 5Gi
templates:
- name: pipeline
dag:
tasks:
- name: fetch-binary
templateRef:
name: training-pipeline
template: fetch-binary
- name: hyperopt
templateRef:
name: training-pipeline
template: hyperopt
dependencies: [fetch-binary]
- name: train-best
templateRef:
name: training-pipeline
template: train-best
dependencies: [hyperopt]
- name: evaluate
templateRef:
name: training-pipeline
template: evaluate
dependencies: [train-best]
- name: upload-results
templateRef:
name: training-pipeline
template: upload-results
dependencies: [evaluate]

71
scripts/argo-compile-deploy.sh Executable file
View File

@@ -0,0 +1,71 @@
#!/usr/bin/env bash
# Compile and deploy service binaries via Argo Workflows.
#
# Usage:
# ./scripts/argo-compile-deploy.sh # all services, HEAD
# ./scripts/argo-compile-deploy.sh --sha abc1234 # specific commit
# ./scripts/argo-compile-deploy.sh --services "api trading-service" # subset
# ./scripts/argo-compile-deploy.sh --webhook # trigger via webhook
#
# Requires: argo CLI (default) or curl (webhook mode)
set -euo pipefail
SHA="HEAD"
SERVICES=""
WEBHOOK=false
WEBHOOK_URL="http://workflow-trigger-eventsource-svc.foxhunt:12001/compile-deploy"
WATCH=false
usage() {
cat <<EOF
Usage: $(basename "$0") [OPTIONS]
Options:
--sha <commit> Git commit SHA to build (default: HEAD)
--services <list> Space-separated service packages to build
(default: all 7 services)
--webhook Trigger via webhook instead of argo CLI
--watch Follow workflow logs after submission
-h, --help Show this help
EOF
exit 0
}
while [[ $# -gt 0 ]]; do
case $1 in
--sha) SHA="$2"; shift 2 ;;
--services) SERVICES="$2"; shift 2 ;;
--webhook) WEBHOOK=true; shift ;;
--watch) WATCH=true; shift ;;
-h|--help) usage ;;
*) echo "Unknown option: $1"; usage ;;
esac
done
if $WEBHOOK; then
echo "Triggering compile-deploy via webhook..."
PAYLOAD="{\"commit_sha\": \"${SHA}\"}"
curl -sf -X POST "$WEBHOOK_URL" \
-H 'Content-Type: application/json' \
-d "$PAYLOAD"
echo "Webhook sent: $PAYLOAD"
exit 0
fi
# Build argo submit command
CMD="argo submit -n foxhunt --from=wftmpl/compile-and-deploy"
CMD="$CMD -p commit-sha=$SHA"
if [[ -n "$SERVICES" ]]; then
CMD="$CMD -p service-packages=\"$SERVICES\""
fi
if $WATCH; then
CMD="$CMD --watch"
fi
echo "Submitting compile-and-deploy workflow..."
echo " commit-sha: $SHA"
[[ -n "$SERVICES" ]] && echo " services: $SERVICES" || echo " services: all (default)"
echo ""
eval "$CMD"

118
scripts/argo-train.sh Executable file
View File

@@ -0,0 +1,118 @@
#!/usr/bin/env bash
# Train a model via Argo Workflows (DQN, PPO, or any supervised model).
#
# Usage:
# ./scripts/argo-train.sh dqn # DQN with defaults
# ./scripts/argo-train.sh ppo --tag dev-abc1234 # PPO with specific binary
# ./scripts/argo-train.sh tft --trials 40 --epochs 100 # supervised TFT
# ./scripts/argo-train.sh mamba2 --webhook # trigger via webhook
#
# Supported models:
# RL: dqn, ppo
# Supervised: tft, mamba2, tggn, tlob, liquid, kan, xlstm, diffusion
#
# Requires: argo CLI (default) or curl (webhook mode)
set -euo pipefail
TAG="latest"
TRIALS=""
EPOCHS=""
GPU_POOL=""
SYMBOL=""
WEBHOOK=false
WEBHOOK_BASE="http://workflow-trigger-eventsource-svc.foxhunt:12001"
WATCH=false
usage() {
cat <<EOF
Usage: $(basename "$0") <model> [OPTIONS]
Models:
dqn, ppo (RL)
tft, mamba2, tggn, tlob, liquid, kan, xlstm, diffusion (supervised)
Options:
--tag <binary-tag> Binary tag to use (default: latest)
--trials <n> Hyperopt trials (default: model-specific)
--epochs <n> Training epochs (default: model-specific)
--gpu-pool <pool> GPU node pool (default: ci-training-h100)
--symbol <sym> Trading symbol (default: ES.FUT)
--webhook Trigger via webhook instead of argo CLI
--watch Follow workflow logs after submission
-h, --help Show this help
EOF
exit 0
}
[[ $# -eq 0 ]] && { echo "Error: model argument required"; usage; }
MODEL="$1"; shift
# Validate model name
case "$MODEL" in
dqn|ppo|tft|mamba2|tggn|tlob|liquid|kan|xlstm|diffusion) ;;
*) echo "Error: unknown model '$MODEL'"; usage ;;
esac
while [[ $# -gt 0 ]]; do
case $1 in
--tag) TAG="$2"; shift 2 ;;
--trials) TRIALS="$2"; shift 2 ;;
--epochs) EPOCHS="$2"; shift 2 ;;
--gpu-pool) GPU_POOL="$2"; shift 2 ;;
--symbol) SYMBOL="$2"; shift 2 ;;
--webhook) WEBHOOK=true; shift ;;
--watch) WATCH=true; shift ;;
-h|--help) usage ;;
*) echo "Unknown option: $1"; usage ;;
esac
done
# Determine workflow template name
case "$MODEL" in
dqn) TEMPLATE="train-dqn" ;;
ppo) TEMPLATE="train-ppo" ;;
*) TEMPLATE="train-supervised" ;;
esac
if $WEBHOOK; then
case "$MODEL" in
dqn) ENDPOINT="/train/dqn"; PAYLOAD="{\"binary_tag\": \"${TAG}\"}" ;;
ppo) ENDPOINT="/train/ppo"; PAYLOAD="{\"binary_tag\": \"${TAG}\"}" ;;
*) ENDPOINT="/train/supervised"; PAYLOAD="{\"model\": \"${MODEL}\", \"binary_tag\": \"${TAG}\"}" ;;
esac
echo "Triggering $MODEL training via webhook..."
curl -sf -X POST "${WEBHOOK_BASE}${ENDPOINT}" \
-H 'Content-Type: application/json' \
-d "$PAYLOAD"
echo "Webhook sent: $PAYLOAD"
exit 0
fi
# Build argo submit command
CMD="argo submit -n foxhunt --from=wftmpl/$TEMPLATE"
CMD="$CMD -p binary-tag=$TAG"
# Supervised models need --model param
if [[ "$MODEL" != "dqn" && "$MODEL" != "ppo" ]]; then
CMD="$CMD -p model=$MODEL"
fi
[[ -n "$TRIALS" ]] && CMD="$CMD -p hyperopt-trials=$TRIALS"
[[ -n "$EPOCHS" ]] && CMD="$CMD -p train-epochs=$EPOCHS"
[[ -n "$GPU_POOL" ]] && CMD="$CMD -p gpu-pool=$GPU_POOL"
[[ -n "$SYMBOL" ]] && CMD="$CMD -p symbol=$SYMBOL"
if $WATCH; then
CMD="$CMD --watch"
fi
echo "Submitting $MODEL training workflow ($TEMPLATE)..."
echo " binary-tag: $TAG"
[[ -n "$TRIALS" ]] && echo " trials: $TRIALS"
[[ -n "$EPOCHS" ]] && echo " epochs: $EPOCHS"
[[ -n "$GPU_POOL" ]] && echo " gpu-pool: $GPU_POOL"
[[ -n "$SYMBOL" ]] && echo " symbol: $SYMBOL"
echo ""
eval "$CMD"