From 4d65ace6253877ab3f2dc75761fb1242848d063e Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 15 May 2026 21:28:40 +0200 Subject: [PATCH] feat(phase-e-4-a): mirror --temporal in backtest binary + T11 ISV-continual MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase E.4.A Tasks 11+12: backtest binary gains the same --temporal Mamba2 forward chain as the smoke binary, plus the --isv-continual flag that fires the stacker-threshold controller at the end of each eval episode (Pillar B). Changes: - Imports: ml_alpha::mamba2_block::{Mamba2Block, Mamba2BlockConfig}, ml_core::cuda_autograd::gpu_tensor::GpuTensor - CLI flags: --temporal, --window-k, --mamba2-hidden-dim, --mamba2-state-dim, --isv-continual - Cubin loading: alpha_window_push + stacker_threshold_controller - Q-net sizing: c51_input_dim = mamba2_hidden_dim when --c51 --temporal - Buffers: window_tensor GpuTensor, h_enriched_buf_dev, isv_dev, ctl_wiener_dev - Training inference path: push + Mamba2 forward + h_enriched → C51 forward (mirrors smoke binary) - Training batched compute: terminal Mamba2 forward, h_enriched_buf for current/next, c51_input_dim threading - Eval inference path: push + Mamba2 forward + h_enriched → C51 forward + Thompson select (with scoped borrow guard) - T11 ISV-continual: stacker-threshold controller fires at end of each eval episode; ISV slot 543 (threshold), 545 (observed-rate), 546 (Kelly atten) update with realized rollout stats. Co-exists with the τ-grid sweep (τ-grid still gates; ISV updates parallel observation of "live deployment" behaviour). - JSON output: new fields temporal, window_k, mamba2_hidden_dim, mamba2_state_dim, isv_continual. Co-Authored-By: Claude Opus 4.7 --- crates/ml/examples/alpha_compose_backtest.rs | 240 +++++++++++++++++-- 1 file changed, 225 insertions(+), 15 deletions(-) diff --git a/crates/ml/examples/alpha_compose_backtest.rs b/crates/ml/examples/alpha_compose_backtest.rs index 0c83b6552..4ae7ac829 100644 --- a/crates/ml/examples/alpha_compose_backtest.rs +++ b/crates/ml/examples/alpha_compose_backtest.rs @@ -41,6 +41,10 @@ use clap::Parser; use cudarc::driver::{CudaContext, DevicePtr, DevicePtrMut}; use tracing::info; +// Phase E.4.A.8/T12: Mamba2 temporal encoder (ml-alpha Phase 1d.1). +use ml_alpha::mamba2_block::{Mamba2Block, Mamba2BlockConfig}; +use ml_core::cuda_autograd::gpu_tensor::GpuTensor; + // ── Mapped-pinned helpers (mirror gpu_training_guard.rs MappedBuffer) ── struct MappedI32 { @@ -234,6 +238,22 @@ struct Cli { /// ±0.125-tick. Phase E.3 Path 3 follow-up. #[arg(long, default_value_t = false)] real_spread: bool, + /// Phase E.4.A: enable the temporal-encoder stack (sliding window + + /// Mamba2 + C51). + #[arg(long, default_value_t = false)] + temporal: bool, + #[arg(long, default_value_t = 16)] + window_k: usize, + #[arg(long, default_value_t = 32)] + mamba2_hidden_dim: usize, + #[arg(long, default_value_t = 16)] + mamba2_state_dim: usize, + /// Phase E.4.A Pillar B: fire stacker-threshold controller at the + /// end of EVERY eval episode (not just training). Slot 543 + /// (threshold) adapts during the 2D sweep based on observed trade + /// rate / Sharpe. + #[arg(long, default_value_t = false)] + isv_continual: bool, #[arg(long, default_value = "config/ml/alpha_compose_backtest.json")] out_path: PathBuf, } @@ -353,6 +373,17 @@ fn main() -> Result<()> { let c51_project_kernel = c51_module.load_function("alpha_c51_project_kernel")?; let c51_grad_kernel = c51_module.load_function("alpha_c51_grad_kernel")?; let c51_thompson_kernel = c51_module.load_function("alpha_c51_thompson_select_kernel")?; + // Phase E.4.A.6: window push kernel (used in --temporal path). + let push_module = ctx + .load_cubin(ml::cuda_pipeline::alpha_kernels::ALPHA_WINDOW_PUSH_CUBIN.to_vec()) + .context("alpha_window_push cubin")?; + let push_kernel = push_module.load_function("alpha_window_push_kernel")?; + // Phase E.4.A.7 controller (used in --isv-continual eval path). + let ctl_module = ctx + .load_cubin(ml::cuda_pipeline::alpha_kernels::STACKER_THRESHOLD_CONTROLLER_CUBIN.to_vec()) + .context("stacker controller cubin")?; + let ctl_kernel = ctl_module + .load_function("stacker_threshold_controller_update")?; let c51_n_atoms = cli.c51_n_atoms; let c51_v_min = cli.c51_vmin; let c51_v_max = cli.c51_vmax; @@ -414,14 +445,19 @@ fn main() -> Result<()> { ); // --- Initialize Q-network --- + let c51_input_dim: usize = if cli.c51 && cli.temporal { + cli.mamba2_hidden_dim + } else { + STATE_DIM + }; let n_weights_eff: usize = if cli.c51 { - N_ACTIONS * c51_n_atoms * STATE_DIM + N_ACTIONS * c51_n_atoms * c51_input_dim } else { N_WEIGHTS }; let n_biases_eff: usize = if cli.c51 { N_ACTIONS * c51_n_atoms } else { N_BIASES }; let mut rng = SmokeRng::new(cli.seed.wrapping_add(0xDEAD_BEEF)); - let xavier_scale = (2.0_f32 / STATE_DIM as f32).sqrt(); + let xavier_scale = (2.0_f32 / c51_input_dim as f32).sqrt(); let w_init: Vec = (0..n_weights_eff) .map(|_| xavier_scale * 2.0 * (rng.next_f32() - 0.5)) .collect(); @@ -455,6 +491,37 @@ fn main() -> Result<()> { let state_pinned = unsafe { MappedF32::new(STATE_DIM)? }; let action_pinned = unsafe { MappedI32::new()? }; + // Phase E.4.A.8 (T12): temporal buffers, allocated unconditionally. + let mut window_tensor = GpuTensor::zeros( + &[1, cli.window_k, STATE_DIM], &stream + ).map_err(|e| anyhow::anyhow!("alloc window: {e}"))?; + let h_enriched_buf_capacity = (cli.horizon + 1) * cli.mamba2_hidden_dim; + let mut h_enriched_buf_dev = stream + .alloc_zeros::(h_enriched_buf_capacity) + .context("alloc h_enriched buffer")?; + let mamba2_block: Option = if cli.temporal { + let cfg = Mamba2BlockConfig { + in_dim: STATE_DIM, + hidden_dim: cli.mamba2_hidden_dim, + state_dim: cli.mamba2_state_dim, + seq_len: cli.window_k, + }; + info!("TEMPORAL Mamba2: in={} hidden={} state={} K={}", + cfg.in_dim, cfg.hidden_dim, cfg.state_dim, cfg.seq_len); + Some(Mamba2Block::new(cfg, stream.clone()) + .map_err(|e| anyhow::anyhow!("Mamba2Block init: {e}"))?) + } else { + None + }; + let n_atoms_i = c51_n_atoms as i32; + // ISV buffer + Wiener state for the eval-time controller path. + let mut isv_host: Vec = vec![0.0; 552]; + isv_host[ml::cuda_pipeline::alpha_isv_slots::RANDOM_BASELINE_MEAN_INDEX] = -5191.53; + isv_host[ml::cuda_pipeline::alpha_isv_slots::RANDOM_BASELINE_STD_INDEX] = 4963.62; + isv_host[ml::cuda_pipeline::alpha_isv_slots::TRADE_RATE_TARGET_INDEX] = 0.08; + let mut isv_dev = stream.clone_htod(&isv_host).context("upload isv")?; + let mut ctl_wiener_dev = stream.alloc_zeros::(3).context("alloc ctl wiener")?; + // --------------------------------------------------------------- // Phase 1: TRAIN DQN on train segment. // --------------------------------------------------------------- @@ -475,11 +542,51 @@ fn main() -> Result<()> { let mut actions_host: Vec = Vec::with_capacity(cli.horizon); let mut rewards_host: Vec = Vec::with_capacity(cli.horizon); let mut dones_host: Vec = Vec::with_capacity(cli.horizon); + // Phase E.4.A.7 (T12): zero window at episode start. + if cli.temporal { + stream.memset_zeros(window_tensor.data_mut()) + .context("zero window on episode reset (train)")?; + } loop { let s_vec = env.state(&state).to_vec(); let action: u8 = if cli.c51 { state_pinned.write(&s_vec); - { + // Phase E.4.A.7 (T12): push and run Mamba2. + if cli.temporal { + let (w_ptr, _g1) = window_tensor.data_mut().device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_window_push( + &stream, &push_kernel, + state_pinned.dev_u64(), w_ptr, + cli.window_k as i32, state_dim_i, + )?; + } + } + if cli.temporal { + let block = mamba2_block.as_ref().expect("Mamba2Block missing"); + let (_logit, cache) = block.forward_train(&window_tensor) + .map_err(|e| anyhow::anyhow!("mamba2 forward: {e}"))?; + // Store h_enriched for batched training. + let h_host = stream.clone_dtoh(cache.h_enriched.cuda_data())?; + let slot_offset = state.step * cli.mamba2_hidden_dim; + let mut buf_host = stream.clone_dtoh(&h_enriched_buf_dev)?; + for j in 0..cli.mamba2_hidden_dim { + buf_host[slot_offset + j] = h_host[j]; + } + stream.memcpy_htod(&buf_host, &mut h_enriched_buf_dev)?; + // C51 forward on h_enriched. + let (w_ptr, _g0) = w_dev.device_ptr(&stream); + let (b_ptr, _g1) = b_dev.device_ptr(&stream); + let (p_ptr, _g2) = single_probs_dev.device_ptr_mut(&stream); + let (h_ptr, _g3) = cache.h_enriched.cuda_data().device_ptr(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_forward( + &stream, &c51_fwd_kernel, + w_ptr, b_ptr, h_ptr, p_ptr, + 1, c51_input_dim as i32, n_act_i, n_atoms_i, + )?; + } + } else { let (w_ptr, _g0) = w_dev.device_ptr(&stream); let (b_ptr, _g1) = b_dev.device_ptr(&stream); let (p_ptr, _g2) = single_probs_dev.device_ptr_mut(&stream); @@ -487,7 +594,7 @@ fn main() -> Result<()> { ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_forward( &stream, &c51_fwd_kernel, w_ptr, b_ptr, state_pinned.dev_u64(), p_ptr, - 1, state_dim_i, n_act_i, n_atoms_i, + 1, c51_input_dim as i32, n_act_i, n_atoms_i, )?; } } @@ -564,30 +671,64 @@ fn main() -> Result<()> { stream.memcpy_htod(&rewards_norm, &mut rewards_dev)?; stream.memcpy_htod(&dones_host, &mut dones_dev)?; if cli.c51 { - // C51 forward/project/grad + // Phase E.4.A.8 (T12): in temporal mode, the batched C51 + // forwards read h_enriched_buf_dev populated during the + // per-step inference. Run one extra Mamba2 forward on the + // terminal window to fill slot ep_len. + if cli.temporal { + let block = mamba2_block.as_ref().expect("Mamba2Block missing"); + let (_logit, cache) = block.forward_train(&window_tensor) + .map_err(|e| anyhow::anyhow!("mamba2 terminal forward: {e}"))?; + let h_host_term = stream.clone_dtoh(cache.h_enriched.cuda_data())?; + let term_offset = (ep_len as usize) * cli.mamba2_hidden_dim; + let mut buf_host = stream.clone_dtoh(&h_enriched_buf_dev)?; + for j in 0..cli.mamba2_hidden_dim { + buf_host[term_offset + j] = h_host_term[j]; + } + stream.memcpy_htod(&buf_host, &mut h_enriched_buf_dev)?; + } + let (curr_input_ptr_guard, next_input_ptr_guard); + let (curr_input_ptr, next_input_ptr): (u64, u64) = if cli.temporal { + let (h_ptr_curr, g_curr) = h_enriched_buf_dev.device_ptr(&stream); + let (h_ptr_next_base, g_next) = h_enriched_buf_dev.device_ptr(&stream); + curr_input_ptr_guard = g_curr; + next_input_ptr_guard = g_next; + let _ = (&curr_input_ptr_guard, &next_input_ptr_guard); + ( + h_ptr_curr, + h_ptr_next_base + (cli.mamba2_hidden_dim as u64) * 4u64, + ) + } else { + let (s_ptr, g_curr) = states_dev.device_ptr(&stream); + let (n_ptr, g_next) = next_states_dev.device_ptr(&stream); + curr_input_ptr_guard = g_curr; + next_input_ptr_guard = g_next; + let _ = (&curr_input_ptr_guard, &next_input_ptr_guard); + (s_ptr, n_ptr) + }; + // C51 forward online → probs_current { let (w_ptr, _g0) = w_dev.device_ptr(&stream); let (b_ptr, _g1) = b_dev.device_ptr(&stream); - let (s_ptr, _g2) = states_dev.device_ptr(&stream); let (p_ptr, _g3) = probs_current_dev.device_ptr_mut(&stream); unsafe { ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_forward( &stream, &c51_fwd_kernel, - w_ptr, b_ptr, s_ptr, p_ptr, - ep_len, state_dim_i, n_act_i, n_atoms_i, + w_ptr, b_ptr, curr_input_ptr, p_ptr, + ep_len, c51_input_dim as i32, n_act_i, n_atoms_i, )?; } } + // C51 forward target → probs_next { let (w_ptr, _g0) = w_target_dev.device_ptr(&stream); let (b_ptr, _g1) = b_target_dev.device_ptr(&stream); - let (s_ptr, _g2) = next_states_dev.device_ptr(&stream); let (p_ptr, _g3) = probs_next_dev.device_ptr_mut(&stream); unsafe { ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_forward( &stream, &c51_fwd_kernel, - w_ptr, b_ptr, s_ptr, p_ptr, - ep_len, state_dim_i, n_act_i, n_atoms_i, + w_ptr, b_ptr, next_input_ptr, p_ptr, + ep_len, c51_input_dim as i32, n_act_i, n_atoms_i, )?; } } @@ -612,11 +753,12 @@ fn main() -> Result<()> { let (s_ptr, _g3) = states_dev.device_ptr(&stream); let (dw_ptr, _g4) = dw_dev.device_ptr_mut(&stream); let (db_ptr, _g5) = db_dev.device_ptr_mut(&stream); + let grad_input_ptr = if cli.temporal { curr_input_ptr } else { s_ptr }; unsafe { ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_grad( &stream, &c51_grad_kernel, - p_ptr, m_ptr, a_ptr, s_ptr, dw_ptr, db_ptr, - ep_len, state_dim_i, n_act_i, n_atoms_i, + p_ptr, m_ptr, a_ptr, grad_input_ptr, dw_ptr, db_ptr, + ep_len, c51_input_dim as i32, n_act_i, n_atoms_i, 1.0 / ep_len as f32, )?; } @@ -752,11 +894,44 @@ fn main() -> Result<()> { let mut state = EpisodeState::new(); let mut terminal_r = 0.0_f32; let mut ep_n_trades = 0_u32; + // Phase E.4.A.7 (T12): zero window at eval episode start. + if cli.temporal { + stream.memset_zeros(window_tensor.data_mut()) + .context("zero window on eval episode reset")?; + } loop { let s_vec = env.state(&state).to_vec(); let action: u8 = if cli.c51 { state_pinned.write(&s_vec); - { + // Phase E.4.A (T12): push + Mamba2 + C51. The push + // is scoped in its own block so the mutable borrow + // guard drops before the &window_tensor read. + if cli.temporal { + { + let (w_ptr, _g1) = window_tensor.data_mut().device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_window_push( + &stream, &push_kernel, + state_pinned.dev_u64(), w_ptr, + cli.window_k as i32, state_dim_i, + )?; + } + } + let block = mamba2_block.as_ref().expect("Mamba2Block missing"); + let (_logit, cache) = block.forward_train(&window_tensor) + .map_err(|e| anyhow::anyhow!("mamba2 forward (eval): {e}"))?; + let (w_ptr, _g0) = w_dev.device_ptr(&stream); + let (b_ptr, _g1) = b_dev.device_ptr(&stream); + let (p_ptr, _g2) = single_probs_dev.device_ptr_mut(&stream); + let (h_ptr, _g3) = cache.h_enriched.cuda_data().device_ptr(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_forward( + &stream, &c51_fwd_kernel, + w_ptr, b_ptr, h_ptr, p_ptr, + 1, c51_input_dim as i32, n_act_i, n_atoms_i, + )?; + } + } else { let (w_ptr, _g0) = w_dev.device_ptr(&stream); let (b_ptr, _g1) = b_dev.device_ptr(&stream); let (p_ptr, _g2) = single_probs_dev.device_ptr_mut(&stream); @@ -764,7 +939,7 @@ fn main() -> Result<()> { ml::cuda_pipeline::alpha_kernels::launch_alpha_c51_forward( &stream, &c51_fwd_kernel, w_ptr, b_ptr, state_pinned.dev_u64(), p_ptr, - 1, state_dim_i, n_act_i, n_atoms_i, + 1, c51_input_dim as i32, n_act_i, n_atoms_i, )?; } } @@ -832,6 +1007,36 @@ fn main() -> Result<()> { if terminal_r > 0.0 { win_count += 1; } + // Phase E.4.A Pillar B (T11): fire the stacker-threshold + // controller at end of each eval episode when + // --isv-continual. Adapts ISV slot 543/545/546 based on + // observed trade-rate and Sharpe; the τ-grid still + // overrides via `threshold` for the actual gate, but the + // ISV evolution is the "live deployment" simulation. + if cli.isv_continual { + let trade_count_ep = ep_n_trades as f32; + let decisions_ep = (state.step as f32).max(1.0); + let baseline_std = isv_host[ml::cuda_pipeline::alpha_isv_slots::RANDOM_BASELINE_STD_INDEX].max(1e-6); + let rollout_sharpe = terminal_r / baseline_std; + unsafe { + let (isv_ptr, _gi) = isv_dev.device_ptr_mut(&stream); + let (wv_ptr, _gw) = ctl_wiener_dev.device_ptr_mut(&stream); + ml::cuda_pipeline::alpha_kernels::launch_stacker_threshold_controller( + &stream, &ctl_kernel, + trade_count_ep, decisions_ep, rollout_sharpe, + /* target_sharpe */ 0.5, + /* k_threshold */ 0.01, + /* k_atten */ 0.005, + /* wiener_alpha_floor */ 0.4, + /* ctl_alpha_meta */ 0.1, + ml::cuda_pipeline::alpha_isv_slots::STACKER_THRESHOLD_INDEX as i32, + ml::cuda_pipeline::alpha_isv_slots::TRADE_RATE_TARGET_INDEX as i32, + ml::cuda_pipeline::alpha_isv_slots::TRADE_RATE_OBSERVED_EMA_INDEX as i32, + ml::cuda_pipeline::alpha_isv_slots::STACKER_KELLY_ATTENUATION_INDEX as i32, + isv_ptr, wv_ptr, + )?; + } + } } let n = rewards.len() as f64; let mean = rewards.iter().map(|r| *r as f64).sum::() / n; @@ -919,6 +1124,11 @@ fn main() -> Result<()> { "c51_n_atoms": if cli.c51 { c51_n_atoms } else { 0 }, "c51_vmin": if cli.c51 { c51_v_min } else { 0.0 }, "c51_vmax": if cli.c51 { c51_v_max } else { 0.0 }, + "temporal": cli.temporal, + "window_k": if cli.temporal { cli.window_k } else { 0 }, + "mamba2_hidden_dim": if cli.temporal { cli.mamba2_hidden_dim } else { 0 }, + "mamba2_state_dim": if cli.temporal { cli.mamba2_state_dim } else { 0 }, + "isv_continual": cli.isv_continual, "horizon": cli.horizon, "train_frac": cli.train_frac, "n_train_episodes": cli.n_train_episodes,