From 2af8e02fd80732394cd0d12f187807284260733a Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 15 May 2026 17:59:28 +0200 Subject: [PATCH] =?UTF-8?q?feat(alpha):=20Phase=20E.3=20composition=20back?= =?UTF-8?q?test=20=E2=80=94=20reveals=20slot=20543=20needs=20consumption?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Phase E.3 Task 23. Trains the Phase E execution-policy DQN on the first 80% of fxcache snapshots, then evaluates the frozen policy (ε=0) on the held-out 20% across a transaction-cost sweep. Compares absolute Sharpe vs the Phase 1d.4 always-market-when-confident baseline. Pipeline pieces: - Shared loaders extracted into crates/ml/src/env/loaders.rs (used by both alpha_dqn_h600_smoke and alpha_compose_backtest) - alpha_compose_backtest.rs: train DQN on first n_train bars, then frozen-eval n_eval episodes per cost level - cost grid: [0.0, 0.0625, 0.125, 0.25, 0.5] (price units per contract round-turn) - Annualised Sharpe via per-episode Sharpe × sqrt(episodes/year) where episodes/year ≈ 252 · 6.5h · 3600s / (horizon · 12s) Run (horizon=600, 1000 train ep, 500 eval ep/cost, 1.5M snapshots): cost n_ep mean_R std_R Sharpe/ep Sharpe_ann win_rate 0.0000 500 -11.09 8.03 -1.380 -39.50 0.090 0.0625 500 -20.68 9.88 -2.093 -59.89 0.012 0.1250 500 -29.38 9.49 -3.095 -88.58 0.000 0.2500 500 -48.23 11.90 -4.052 -115.98 0.000 0.5000 500 -84.59 17.43 -4.854 -138.92 0.000 Phase 1d.4 baseline for comparison: +4.4 ann. at cost=0, -4.0 at half-tick. The Phase E policy LOSES MONEY across the whole cost grid — even at frictionless cost=0. This is not a contradiction with the H=600 PASS verdict (rvr=+1.04σ): the smoke's rvr is RELATIVE TO RANDOM, while backtest Sharpe is ABSOLUTE. "Better than random by 1 std" is still losing if random loses big. The diagnostic that the E.2 controller already surfaced: ISV[543] STACKER_THRESHOLD saturated at upper clamp (0.5) — policy trades 85% of the time vs the 8% target. Over-trading pays spread on every bar regardless of alpha confidence. Even with perfect alpha (Phase 1d.3 AUC=0.673), trading 85% × spread cost > alpha edge. The Phase 1d.4 baseline beats us at cost=0 because it WAITS unless |stacker_logit| > threshold — the threshold gate filters bars with weak alpha signal. The Phase E controller PRODUCES slot 543 but the DQN's action selection doesn't CONSUME it. This is exactly what the E.3 backtest is FOR: revealing that the Phase E.1/E.2 producer-side architecture without consumer-side gating is incomplete. The composition backtest validates the architecture's weak link. NEXT (E.3 task 24-28 or a side fix): wire slot 543 consumption into the action selection. At each step: if |ISV[543] − 0.5| > |stacker_logit − 0.5|: action = Wait // confidence below threshold, sit out else: action = argmax(Q) Or equivalently: action = if confidence_high(alpha_logit, ISV[543]) { argmax(Q) over Buy/Sell actions } else { Wait }. Once slot 543 is consumed, re-run alpha_compose_backtest and expect Sharpe to move toward / past the Phase 1d.4 baseline. Loader refactor: extracted load_fill_model_from_json, load_alpha_cache, load_snapshots_from_fxcache from alpha_dqn_h600_smoke.rs into crates/ml/src/env/loaders.rs. The smoke now calls the shared module via ml::env::loaders::*. ~150 lines of duplicated code removed. Build + run verified: smoke still builds clean. Backtest runs in ~30s (train 8s + eval 20s + setup). Branch: sp20-aux-h-fixed, pushed. --- config/ml/alpha_compose_backtest.json | 77 +++ crates/ml/examples/alpha_compose_backtest.rs | 581 +++++++++++++++++++ crates/ml/examples/alpha_dqn_h600_smoke.rs | 159 +---- crates/ml/src/env/loaders.rs | 182 ++++++ crates/ml/src/env/mod.rs | 1 + 5 files changed, 844 insertions(+), 156 deletions(-) create mode 100644 config/ml/alpha_compose_backtest.json create mode 100644 crates/ml/examples/alpha_compose_backtest.rs create mode 100644 crates/ml/src/env/loaders.rs diff --git a/config/ml/alpha_compose_backtest.json b/config/ml/alpha_compose_backtest.json new file mode 100644 index 000000000..2faef85ea --- /dev/null +++ b/config/ml/alpha_compose_backtest.json @@ -0,0 +1,77 @@ +{ + "bins": [ + { + "cost": 0.0, + "mean_reward": -11.089629173278809, + "n_episodes": 500, + "p05": -22.522815704345703, + "p50": -11.164673805236816, + "p95": 3.1911258697509766, + "sharpe_annualised": -39.49838638305664, + "sharpe_per_episode": -1.3801853656768799, + "std_reward": 8.034883499145508, + "win_rate": 0.09000000357627869 + }, + { + "cost": 0.0625, + "mean_reward": -20.684438705444336, + "n_episodes": 500, + "p05": -36.075843811035156, + "p50": -19.98797607421875, + "p95": -7.718697547912598, + "sharpe_annualised": -59.89125061035156, + "sharpe_per_episode": -2.0927696228027344, + "std_reward": 9.88376235961914, + "win_rate": 0.012000000104308128 + }, + { + "cost": 0.125, + "mean_reward": -29.383296966552734, + "n_episodes": 500, + "p05": -44.815711975097656, + "p50": -28.835330963134766, + "p95": -15.50295352935791, + "sharpe_annualised": -88.58140563964844, + "sharpe_per_episode": -3.095284938812256, + "std_reward": 9.492920875549316, + "win_rate": 0.0 + }, + { + "cost": 0.25, + "mean_reward": -48.22906494140625, + "n_episodes": 500, + "p05": -67.93758392333984, + "p50": -48.01160430908203, + "p95": -29.78984832763672, + "sharpe_annualised": -115.97515106201172, + "sharpe_per_episode": -4.052499771118164, + "std_reward": 11.9010648727417, + "win_rate": 0.0 + }, + { + "cost": 0.5, + "mean_reward": -84.58942413330078, + "n_episodes": 500, + "p05": -110.6795883178711, + "p50": -84.19437408447266, + "p95": -58.62353515625, + "sharpe_annualised": -138.9206085205078, + "sharpe_per_episode": -4.854279041290283, + "std_reward": 17.425745010375977, + "win_rate": 0.0 + } + ], + "cost_grid": [ + 0.0, + 0.0625, + 0.125, + 0.25, + 0.5 + ], + "horizon": 600, + "n_eval_episodes": 500, + "n_train_episodes": 1000, + "phase": "E.3 Task 23", + "train_cost": 0.0625, + "train_frac": 0.800000011920929 +} \ No newline at end of file diff --git a/crates/ml/examples/alpha_compose_backtest.rs b/crates/ml/examples/alpha_compose_backtest.rs new file mode 100644 index 000000000..3de2b1fc3 --- /dev/null +++ b/crates/ml/examples/alpha_compose_backtest.rs @@ -0,0 +1,581 @@ +//! Phase E.3 Task 23 — Composition backtest with cost sweep. +//! +//! Trains the Phase E execution-policy DQN (linear Q + Phase 1d.3 +//! alpha-cache + stabilizers) on the first `--train-frac` of the +//! fxcache, then evaluates the FROZEN policy (no SGD, ε=0 greedy) over +//! the held-out remainder at multiple transaction costs. Compares +//! per-cost annualized Sharpe vs the Phase 1d.4 "always-market-when- +//! confident" baseline (+4.4 frictionless, -4.0 at half-tick). +//! +//! Goal per the plan: lift the half-tick Sharpe above 0 — i.e., let +//! the execution-policy intelligence offset the cost the +//! threshold-only baseline can't. +//! +//! ## What's swept and what's frozen +//! +//! Frozen across costs: trained Q-network weights, fill-model +//! coefficients (from `alpha_fill_coeffs.json`), alpha-logit cache +//! (from `alpha_logits_cache.bin`). One policy evaluated at multiple +//! costs. +//! +//! Swept: only `ExecutionEnvConfig.cost_per_contract`. The env's +//! mutable config is updated between cost levels (snapshots stay in +//! place, cursor re-seeded each episode). +//! +//! ## Run +//! +//! ```bash +//! cargo run -p ml --release --example alpha_compose_backtest -- \ +//! --fxcache-path /home/jgrusewski/Work/foxhunt/test_data/feature-cache/9297....fxcache \ +//! --alpha-cache config/ml/alpha_logits_cache.bin \ +//! --fill-coeffs config/ml/alpha_fill_coeffs.json +//! ``` + +use std::fs::File; +use std::io::Write; +use std::path::PathBuf; + +use anyhow::{Context, Result}; +use clap::Parser; +use cudarc::driver::{CudaContext, DevicePtr, DevicePtrMut}; +use tracing::info; + +use ml::cuda_pipeline::alpha_isv_slots::{ + RANDOM_BASELINE_MEAN_INDEX, RANDOM_BASELINE_STD_INDEX, +}; +use ml::env::action_space::N_ACTIONS; +use ml::env::execution_env::{ + EpisodeState, ExecutionEnv, ExecutionEnvConfig, ReplayRng, SnapshotRow, +}; + +const STATE_DIM: usize = 10; +const N_WEIGHTS: usize = N_ACTIONS * STATE_DIM; +const N_BIASES: usize = N_ACTIONS; + +#[derive(Debug, Parser)] +#[command( + name = "alpha_compose_backtest", + about = "Phase E.3 Task 23 — composition backtest with cost sweep" +)] +struct Cli { + #[arg(long)] + fxcache_path: PathBuf, + #[arg(long, default_value = "config/ml/alpha_fill_coeffs.json")] + fill_coeffs: PathBuf, + #[arg(long)] + alpha_cache: PathBuf, + /// Train segment fraction. First N% of snapshots used for DQN training, + /// the rest for evaluation. + #[arg(long, default_value_t = 0.8)] + train_frac: f32, + /// Snapshots to load from the fxcache. + #[arg(long, default_value_t = 1_500_000)] + max_snapshots: usize, + /// Episode horizon in snapshots. + #[arg(long, default_value_t = 600)] + horizon: usize, + /// DQN training episodes (on train segment). + #[arg(long, default_value_t = 1_000)] + n_train_episodes: usize, + /// Frozen-policy evaluation episodes per cost level. + #[arg(long, default_value_t = 500)] + n_eval_episodes: usize, + /// Comma-separated cost grid (price units per contract round-turn). + /// Phase 1d.4 used [0.0, 0.0625, 0.125, 0.25, 0.50]. + #[arg(long, value_delimiter = ',', default_value = "0.0,0.0625,0.125,0.25,0.5")] + cost_grid: Vec, + #[arg(long, default_value_t = 1)] + trade_size: i32, + #[arg(long, default_value_t = 0xCAFEBABE_u64)] + seed: u64, + /// Training-time cost (the policy LEARNED against this cost). + #[arg(long, default_value_t = 0.0625)] + train_cost: f32, + /// SGD learning rate during training. + #[arg(long, default_value_t = 1.0e-4)] + lr: f32, + #[arg(long, default_value_t = 0.50)] + eps_start: f32, + #[arg(long, default_value_t = 0.05)] + eps_end: f32, + #[arg(long, default_value_t = 0.99)] + gamma: f32, + #[arg(long, default_value_t = 0.9)] + alpha_m: f32, + #[arg(long, default_value_t = 0.03)] + tau: f32, + #[arg(long, default_value_t = -1.0)] + log_clip_min: f32, + #[arg(long, default_value_t = 1000.0)] + reward_scale: f32, + #[arg(long, default_value_t = 10)] + target_update_every: usize, + #[arg(long, default_value_t = 1.0)] + grad_clip: f32, + #[arg(long, default_value = "config/ml/alpha_compose_backtest.json")] + out_path: PathBuf, +} + +struct SmokeRng { + state: u64, +} + +impl SmokeRng { + fn new(seed: u64) -> Self { + Self { state: seed } + } + fn next_u64(&mut self) -> u64 { + self.state = self.state.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.state; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + fn next_f32(&mut self) -> f32 { + ((self.next_u64() >> 40) as f32) / ((1u64 << 24) as f32) + } +} + +fn epsilon_greedy(q: &[f32], eps: f32, rng: &mut SmokeRng) -> u8 { + if rng.next_f32() < eps { + (rng.next_u64() % N_ACTIONS as u64) as u8 + } else { + let mut best_i: usize = 0; + let mut best_v: f32 = q[0]; + for i in 1..N_ACTIONS { + if q[i] > best_v { + best_v = q[i]; + best_i = i; + } + } + best_i as u8 + } +} + +#[derive(Debug, Clone, serde::Serialize)] +struct CostBin { + cost: f32, + n_episodes: usize, + mean_reward: f32, + std_reward: f32, + sharpe_per_episode: f32, + /// Sharpe scaled by sqrt(episodes_per_year). Time span derived from + /// the eval window's mid_price index span × bar_seconds. + sharpe_annualised: f32, + win_rate: f32, + p05: f32, + p50: f32, + p95: f32, +} + +fn main() -> Result<()> { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), + ) + .init(); + let cli = Cli::parse(); + info!("Phase E.3 Task 23 — composition backtest starting"); + + let ctx = CudaContext::new(0).context("CUDA init")?; + let stream = ctx.default_stream(); + + // --- Load cubins --- + let lq_module = ctx + .load_cubin(ml::cuda_pipeline::alpha_kernels::ALPHA_LINEAR_Q_CUBIN.to_vec()) + .context("alpha_linear_q cubin")?; + let lq_fwd = lq_module.load_function("alpha_linear_q_forward_kernel")?; + let lq_grad = lq_module.load_function("alpha_linear_q_grad_kernel")?; + let lq_sgd = lq_module.load_function("alpha_linear_q_sgd_step_kernel")?; + let lq_clip = lq_module.load_function("alpha_clip_inplace_kernel")?; + let munch_cubin: Vec = std::fs::read(concat!( + env!("OUT_DIR"), + "/alpha_munchausen_target.cubin" + ))?; + let munch_module = ctx.load_cubin(munch_cubin)?; + let munch_kernel = + munch_module.load_function("alpha_munchausen_target_kernel")?; + + // --- Load env data --- + let fill_model = ml::env::loaders::load_fill_model_from_json(&cli.fill_coeffs)?; + info!("Loaded fill model"); + let alpha_cache = ml::env::loaders::load_alpha_cache(&cli.alpha_cache)?; + info!("Loaded alpha cache: {} entries", alpha_cache.len()); + let rows = ml::env::loaders::load_snapshots_from_fxcache( + &cli.fxcache_path, + cli.max_snapshots, + Some(&alpha_cache), + )?; + let n_total = rows.len(); + info!("Loaded {} snapshots", n_total); + + let n_train = ((n_total as f32) * cli.train_frac) as usize; + let n_eval = n_total - n_train; + if n_train <= cli.horizon || n_eval <= cli.horizon { + anyhow::bail!( + "insufficient snapshots: train={}, eval={}, horizon={}", + n_train, + n_eval, + cli.horizon + ); + } + info!( + "Split: train={} bars (cursor 0..{}), eval={} bars (cursor {}..{})", + n_train, n_train, n_eval, n_train, n_total + ); + + let mut env = ExecutionEnv::new( + ExecutionEnvConfig { + horizon_snapshots: cli.horizon, + trade_size_contracts: cli.trade_size, + cost_per_contract: cli.train_cost, + }, + fill_model, + rows, + cli.seed, + ); + + // --- Initialize Q-network --- + let mut rng = SmokeRng::new(cli.seed.wrapping_add(0xDEAD_BEEF)); + let xavier_scale = (2.0_f32 / STATE_DIM as f32).sqrt(); + let w_init: Vec = (0..N_WEIGHTS) + .map(|_| xavier_scale * 2.0 * (rng.next_f32() - 0.5)) + .collect(); + let b_init: Vec = vec![0.0; N_BIASES]; + let mut w_dev = stream.clone_htod(&w_init)?; + let mut b_dev = stream.clone_htod(&b_init)?; + let mut w_target_dev = stream.clone_htod(&w_init)?; + let mut b_target_dev = stream.clone_htod(&b_init)?; + let mut dw_dev = stream.alloc_zeros::(N_WEIGHTS)?; + let mut db_dev = stream.alloc_zeros::(N_BIASES)?; + + let state_dim_i = STATE_DIM as i32; + let n_act_i = N_ACTIONS as i32; + let mut states_dev = stream.alloc_zeros::(cli.horizon * STATE_DIM)?; + let mut next_states_dev = stream.alloc_zeros::(cli.horizon * STATE_DIM)?; + let mut actions_dev = stream.alloc_zeros::(cli.horizon)?; + let mut rewards_dev = stream.alloc_zeros::(cli.horizon)?; + let mut dones_dev = stream.alloc_zeros::(cli.horizon)?; + let mut q_current_dev = stream.alloc_zeros::(cli.horizon * N_ACTIONS)?; + let mut q_next_dev = stream.alloc_zeros::(cli.horizon * N_ACTIONS)?; + let mut target_dev = stream.alloc_zeros::(cli.horizon)?; + let mut single_state_dev = stream.alloc_zeros::(STATE_DIM)?; + let mut single_q_dev = stream.alloc_zeros::(N_ACTIONS)?; + + // --------------------------------------------------------------- + // Phase 1: TRAIN DQN on train segment. + // --------------------------------------------------------------- + info!("=== Training phase: {} episodes on train segment ===", cli.n_train_episodes); + let mut episode_rng = SmokeRng::new(cli.seed.wrapping_add(0xFEED)); + let train_max_start = (n_train.saturating_sub(cli.horizon + 1)).max(1); + for ep in 0..cli.n_train_episodes { + let eps = cli.eps_start + + (cli.eps_end - cli.eps_start) + * (ep as f32 / cli.n_train_episodes.max(1) as f32); + let start_cursor = (episode_rng.next_u64() as usize) % train_max_start; + let env_seed = episode_rng.next_u64(); + env.reset_at(env_seed, start_cursor); + let mut state = EpisodeState::new(); + + let mut states_host: Vec = Vec::with_capacity(cli.horizon * STATE_DIM); + let mut next_states_host: Vec = Vec::with_capacity(cli.horizon * STATE_DIM); + 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); + loop { + let s_vec = env.state(&state).to_vec(); + stream.memcpy_htod(&s_vec, &mut single_state_dev)?; + { + let (w_ptr, _g0) = w_dev.device_ptr(&stream); + let (b_ptr, _g1) = b_dev.device_ptr(&stream); + let (s_ptr, _g2) = single_state_dev.device_ptr(&stream); + let (q_ptr, _g3) = single_q_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_forward( + &stream, &lq_fwd, w_ptr, b_ptr, s_ptr, q_ptr, + 1, state_dim_i, n_act_i, + )?; + } + } + stream.synchronize()?; + let q_host = stream.clone_dtoh(&single_q_dev)?; + let action = epsilon_greedy(&q_host, eps, &mut episode_rng); + let (_s_next, reward, done) = env + .step(action, &mut state) + .ok_or_else(|| anyhow::anyhow!("step returned None"))?; + let s_next_vec = env.state(&state).to_vec(); + states_host.extend_from_slice(&s_vec); + next_states_host.extend_from_slice(&s_next_vec); + actions_host.push(action as i32); + rewards_host.push(reward); + dones_host.push(if done { 1.0 } else { 0.0 }); + if done { + break; + } + } + let ep_len = actions_host.len() as i32; + if ep_len < 2 { + continue; + } + // Batched train update (same logic as alpha_dqn_h600_smoke). + let rewards_norm: Vec = rewards_host + .iter() + .map(|r| r / cli.reward_scale) + .collect(); + stream.memcpy_htod(&states_host, &mut states_dev)?; + stream.memcpy_htod(&next_states_host, &mut next_states_dev)?; + stream.memcpy_htod(&actions_host, &mut actions_dev)?; + stream.memcpy_htod(&rewards_norm, &mut rewards_dev)?; + stream.memcpy_htod(&dones_host, &mut dones_dev)?; + { + 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 (q_ptr, _g3) = q_current_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_forward( + &stream, &lq_fwd, w_ptr, b_ptr, s_ptr, q_ptr, + ep_len, state_dim_i, n_act_i, + )?; + } + } + { + 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 (q_ptr, _g3) = q_next_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_forward( + &stream, &lq_fwd, w_ptr, b_ptr, s_ptr, q_ptr, + ep_len, state_dim_i, n_act_i, + )?; + } + } + { + let (qn_ptr, _g0) = q_next_dev.device_ptr(&stream); + let (qc_ptr, _g1) = q_current_dev.device_ptr(&stream); + let (a_ptr, _g2) = actions_dev.device_ptr(&stream); + let (r_ptr, _g3) = rewards_dev.device_ptr(&stream); + let (d_ptr, _g4) = dones_dev.device_ptr(&stream); + let (t_ptr, _g5) = target_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_munchausen_target( + &stream, &munch_kernel, + qn_ptr, qc_ptr, a_ptr, r_ptr, d_ptr, + cli.gamma, cli.alpha_m, cli.tau, cli.log_clip_min, + t_ptr, ep_len, n_act_i, + )?; + } + } + { + let (qc_ptr, _g0) = q_current_dev.device_ptr(&stream); + let (t_ptr, _g1) = target_dev.device_ptr(&stream); + let (a_ptr, _g2) = actions_dev.device_ptr(&stream); + 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); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_grad( + &stream, &lq_grad, + qc_ptr, t_ptr, a_ptr, s_ptr, dw_ptr, db_ptr, + ep_len, state_dim_i, n_act_i, + 1.0 / ep_len as f32, + )?; + } + } + // Clip + SGD + { + let (dw_ptr, _g0) = dw_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_clip_inplace( + &stream, &lq_clip, dw_ptr, cli.grad_clip, N_WEIGHTS as i32, + )?; + } + } + { + let (db_ptr, _g0) = db_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_clip_inplace( + &stream, &lq_clip, db_ptr, cli.grad_clip, N_BIASES as i32, + )?; + } + } + { + let (w_ptr, _g0) = w_dev.device_ptr_mut(&stream); + let (dw_ptr, _g1) = dw_dev.device_ptr(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_sgd_step( + &stream, &lq_sgd, w_ptr, dw_ptr, cli.lr, N_WEIGHTS as i32, + )?; + } + } + { + let (b_ptr, _g0) = b_dev.device_ptr_mut(&stream); + let (db_ptr, _g1) = db_dev.device_ptr(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_sgd_step( + &stream, &lq_sgd, b_ptr, db_ptr, cli.lr, N_BIASES as i32, + )?; + } + } + // Target net hard update + if (ep + 1) % cli.target_update_every == 0 { + stream.synchronize()?; + let w_now = stream.clone_dtoh(&w_dev)?; + let b_now = stream.clone_dtoh(&b_dev)?; + stream.memcpy_htod(&w_now, &mut w_target_dev)?; + stream.memcpy_htod(&b_now, &mut b_target_dev)?; + } + if (ep + 1) % 200 == 0 { + info!(" train ep {}/{}", ep + 1, cli.n_train_episodes); + } + } + info!("Training complete. Frozen policy ready for eval."); + + // --------------------------------------------------------------- + // Phase 2: COST SWEEP — frozen-policy greedy eval per cost. + // --------------------------------------------------------------- + info!("=== Eval phase: {} episodes per cost × {} costs ===", + cli.n_eval_episodes, cli.cost_grid.len()); + let eval_max_start = (n_total - n_train).saturating_sub(cli.horizon + 1).max(1); + let mut bins: Vec = Vec::with_capacity(cli.cost_grid.len()); + for &cost in &cli.cost_grid { + env.config.cost_per_contract = cost; + let mut rewards: Vec = Vec::with_capacity(cli.n_eval_episodes); + let mut win_count = 0_usize; + for _ in 0..cli.n_eval_episodes { + let start_cursor = + n_train + (episode_rng.next_u64() as usize) % eval_max_start; + let env_seed = episode_rng.next_u64(); + env.reset_at(env_seed, start_cursor); + let mut state = EpisodeState::new(); + let mut terminal_r = 0.0_f32; + loop { + let s_vec = env.state(&state).to_vec(); + stream.memcpy_htod(&s_vec, &mut single_state_dev)?; + { + let (w_ptr, _g0) = w_dev.device_ptr(&stream); + let (b_ptr, _g1) = b_dev.device_ptr(&stream); + let (s_ptr, _g2) = single_state_dev.device_ptr(&stream); + let (q_ptr, _g3) = single_q_dev.device_ptr_mut(&stream); + unsafe { + ml::cuda_pipeline::alpha_kernels::launch_alpha_linear_q_forward( + &stream, &lq_fwd, w_ptr, b_ptr, s_ptr, q_ptr, + 1, state_dim_i, n_act_i, + )?; + } + } + stream.synchronize()?; + let q_host = stream.clone_dtoh(&single_q_dev)?; + let mut greedy_rng = SmokeRng::new(0); // unused — eps=0 + let action = epsilon_greedy(&q_host, 0.0, &mut greedy_rng); + match env.step(action, &mut state) { + Some((_, reward, done)) => { + if done { + terminal_r = reward; + break; + } + } + None => break, + } + } + rewards.push(terminal_r); + if terminal_r > 0.0 { + win_count += 1; + } + } + let n = rewards.len() as f64; + let mean = rewards.iter().map(|r| *r as f64).sum::() / n; + let var = rewards + .iter() + .map(|r| (*r as f64 - mean).powi(2)) + .sum::() + / n; + let std = var.sqrt().max(1e-6); + let sharpe_per = (mean / std) as f32; + // Annualised Sharpe via per-episode Sharpe × sqrt(episodes/year). + // Episode duration ≈ horizon × bar_seconds. ES.FUT MBP-10 snapshots + // at snapshot_interval=50 ≈ 12 sec/bar on average; 600-bar horizon + // ≈ 7200 s ≈ 2 hours. Episodes per year ≈ 252 × 6.5 × 3600 / 7200 + // ≈ 819. Sharpe_ann ≈ Sharpe_per × sqrt(819) ≈ 28.6. + let episodes_per_year = 252.0 * 6.5 * 3600.0 / (cli.horizon as f64 * 12.0); + let sharpe_ann = (sharpe_per as f64 * episodes_per_year.sqrt()) as f32; + let win_rate = win_count as f32 / cli.n_eval_episodes as f32; + let mut sorted = rewards.clone(); + sorted.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal)); + let pick = |q: f64| -> f32 { + let idx = (q * (n - 1.0)).round() as usize; + sorted[idx.min(sorted.len() - 1)] + }; + let bin = CostBin { + cost, + n_episodes: cli.n_eval_episodes, + mean_reward: mean as f32, + std_reward: std as f32, + sharpe_per_episode: sharpe_per, + sharpe_annualised: sharpe_ann, + win_rate, + p05: pick(0.05), + p50: pick(0.50), + p95: pick(0.95), + }; + info!( + " cost={:>6.4} mean={:>+9.2} std={:>8.2} Sharpe/ep={:+.3} Sharpe_ann={:+.3} win_rate={:.3}", + cost, mean, std, sharpe_per, sharpe_ann, win_rate + ); + bins.push(bin); + } + + // --- Print table --- + info!(""); + info!("=== Phase E.3 cost-sweep table (vs Phase 1d.4 baseline) ==="); + info!( + " {:>6} {:>8} {:>9} {:>9} {:>10} {:>11} {:>8}", + "cost", "n_ep", "mean_R", "std_R", "Sharpe/ep", "Sharpe_ann", "win_rate" + ); + for b in &bins { + info!( + " {:>6.4} {:>8} {:>9.2} {:>9.2} {:>10.4} {:>11.4} {:>8.3}", + b.cost, b.n_episodes, b.mean_reward, b.std_reward, + b.sharpe_per_episode, b.sharpe_annualised, b.win_rate + ); + } + info!(""); + info!("Phase 1d.4 baseline for comparison: +4.4 annualised at cost=0,"); + info!(" -4.0 annualised at cost=0.125."); + + // --- Save JSON --- + let json = serde_json::json!({ + "phase": "E.3 Task 23", + "horizon": cli.horizon, + "train_frac": cli.train_frac, + "n_train_episodes": cli.n_train_episodes, + "n_eval_episodes": cli.n_eval_episodes, + "train_cost": cli.train_cost, + "cost_grid": cli.cost_grid, + "bins": bins, + }); + let mut f = File::create(&cli.out_path)?; + write!(f, "{}", serde_json::to_string_pretty(&json)?)?; + info!("Wrote table to {}", cli.out_path.display()); + + // Touch unused ISV slot constants so they're imported for future expansion. + let _ = (RANDOM_BASELINE_MEAN_INDEX, RANDOM_BASELINE_STD_INDEX); + let _: Option = None; + let _ = SnapshotRow { + mid_price: 0.0, + bid_l: [0.0; 3], + ask_l: [0.0; 3], + alpha_logit: 0.0, + alpha_confidence: 0.0, + spread_bps: 0.0, + l1_imbalance: 0.0, + ofi_sum_5: 0.0, + mid_drift_5: 0.0, + time_since_trade_s: 0.0, + book_event_rate: 0.0, + }; + + Ok(()) +} diff --git a/crates/ml/examples/alpha_dqn_h600_smoke.rs b/crates/ml/examples/alpha_dqn_h600_smoke.rs index f2b3404dc..f721cb66d 100644 --- a/crates/ml/examples/alpha_dqn_h600_smoke.rs +++ b/crates/ml/examples/alpha_dqn_h600_smoke.rs @@ -184,159 +184,6 @@ struct Cli { out_path: PathBuf, } -fn load_fill_model(path: &std::path::Path) -> Result { - let s = std::fs::read_to_string(path) - .with_context(|| format!("read fill coeffs JSON at {}", path.display()))?; - let v: serde_json::Value = serde_json::from_str(&s)?; - let parse_levels = |key: &str| -> Result<[FillCoeffs; 3]> { - let arr = v[key] - .as_array() - .ok_or_else(|| anyhow::anyhow!("missing/non-array `{}`", key))?; - if arr.len() != 3 { - anyhow::bail!("{} must have 3 entries", key); - } - let mut out: [FillCoeffs; 3] = [FillCoeffs { beta: [0.0; 5] }; 3]; - for (i, lvl) in arr.iter().enumerate() { - let vv = lvl - .as_array() - .ok_or_else(|| anyhow::anyhow!("{}[{}] not array", key, i))?; - for k in 0..5 { - out[i].beta[k] = vv[k] - .as_f64() - .ok_or_else(|| anyhow::anyhow!("{}[{}][{}] not number", key, i, k))? - as f32; - } - } - Ok(out) - }; - Ok(FillModel { - bid_coeffs: parse_levels("bid_coeffs")?, - ask_coeffs: parse_levels("ask_coeffs")?, - }) -} - -/// Load the alpha-logit cache produced by `alpha_train_stacker -/// --alpha-cache-out`. Binary file: 4 bytes little-endian u32 length -/// prefix, then `n` f32 values. Each index aligns to the fxcache bar -/// at the same index. -fn load_alpha_cache(path: &std::path::Path) -> Result> { - use std::io::Read; - let mut f = std::fs::File::open(path) - .with_context(|| format!("open alpha cache {}", path.display()))?; - let mut len_bytes = [0u8; 4]; - f.read_exact(&mut len_bytes).context("read alpha-cache header")?; - let n = u32::from_le_bytes(len_bytes) as usize; - let mut buf = vec![0u8; n * 4]; - f.read_exact(&mut buf).context("read alpha-cache body")?; - let mut out = Vec::with_capacity(n); - for i in 0..n { - let off = i * 4; - let v = f32::from_le_bytes([buf[off], buf[off + 1], buf[off + 2], buf[off + 3]]); - out.push(v); - } - Ok(out) -} - -/// Build the env::SnapshotRow stream from a precomputed fxcache. Used when -/// `--alpha-cache` is set (cache indices align to fxcache bars). Synthesizes -/// L1/L2/L3 bid/ask from mid at fixed half-tick offsets — the env's -/// execution simulation gets a stable spread of 0.25 (one ES tick) instead -/// of real MBP-10 LOB variation. Acceptable for the smoke; production env -/// would use real bid/ask. -/// -/// Feature mapping (Block-S region of the 81-dim alpha_features row): -/// features[75] → time_since_trade_s -/// features[77] → book_event_rate -/// features[78] → spread_bps -/// features[79] → l1_imbalance -/// features[80] → mid_drift_5 (micro_mid_drift) -/// -/// `ofi_sum_5` is computed as the sum of features[0..5] (the multi-level -/// OFI L1-L5 block from snapshot_pipeline.rs). -fn load_snapshots_from_fxcache( - fxcache_path: &std::path::Path, - max_snapshots: usize, - alpha_cache: Option<&[f32]>, -) -> Result> { - use ml_alpha::fxcache_reader::FxCacheReader; - - let reader = FxCacheReader::open(fxcache_path) - .with_context(|| format!("open fxcache {}", fxcache_path.display()))?; - let alpha_dim = reader.alpha_feature_dim() - .ok_or_else(|| anyhow::anyhow!("fxcache lacks alpha column"))?; - if alpha_dim < 81 { - anyhow::bail!( - "fxcache alpha_dim={} but smoke expects ≥81 (snapshot_pipeline layout)", - alpha_dim - ); - } - let n_bars_total = reader.bar_count(); - let n = n_bars_total.min(max_snapshots); - info!("fxcache: {} total bars, taking {} for the env", n_bars_total, n); - - if let Some(cache) = alpha_cache { - if cache.len() < n { - anyhow::bail!( - "alpha cache has {} entries but env wants {} bars", - cache.len(), - n - ); - } - } - - let mut rows: Vec = Vec::with_capacity(n); - let mut n_degenerate = 0_usize; - for i in 0..n { - let rec = reader.record(i); - let mid = rec.targets[COL_RAW_CLOSE - FEAT_DIM]; - if !mid.is_finite() || mid <= 0.0 { - n_degenerate += 1; - continue; - } - let features = reader.alpha_features(i) - .ok_or_else(|| anyhow::anyhow!("missing alpha row at bar {}", i))?; - - let bid_l1 = mid - 0.125; - let ask_l1 = mid + 0.125; - let bid_l = [bid_l1, bid_l1 - TICK, bid_l1 - 2.0 * TICK]; - let ask_l = [ask_l1, ask_l1 + TICK, ask_l1 + 2.0 * TICK]; - - let spread_bps = features[78]; - let l1_imbalance = features[79]; - let ofi_sum_5 = features[0..5].iter().sum::(); - let mid_drift_5 = features[80]; - let time_since_trade_s = features[75]; - let book_event_rate = features[77]; - - // alpha_logit: real stacker output if cache provided, else 0 (sentinel). - let alpha_logit = alpha_cache.map(|c| c[i]).unwrap_or(0.0); - // alpha_confidence: distance from 0.5 in probability space. - let alpha_confidence = { - let p = 1.0_f32 / (1.0 + (-alpha_logit.clamp(-50.0, 50.0)).exp()); - (p - 0.5).abs() - }; - - rows.push(SnapshotRow { - mid_price: mid, - bid_l, - ask_l, - alpha_logit, - alpha_confidence, - spread_bps, - l1_imbalance, - ofi_sum_5, - mid_drift_5, - time_since_trade_s, - book_event_rate, - }); - } - if n_degenerate > 0 { - warn!("fxcache loader: skipped {} degenerate bars (non-finite or zero mid)", - n_degenerate); - } - Ok(rows) -} - /// Build the env::SnapshotRow stream from MBP-10 files. Same loader pattern /// as `alpha_random_baseline.rs` — L2/L3 prices synthesized at ±tick offsets /// because `parse_mbp10_streaming` doesn't populate levels[1..10]. @@ -542,12 +389,12 @@ fn main() -> Result<()> { .context("controller kernel load")?; // --- Load env data --- - let fill_model = load_fill_model(&cli.fill_coeffs)?; + let fill_model = ml::env::loaders::load_fill_model_from_json(&cli.fill_coeffs)?; info!("Loaded FillModel from {}", cli.fill_coeffs.display()); // Load alpha cache first (if any) so we can pass it to the fxcache loader. let alpha_cache_vec: Option> = if let Some(p) = cli.alpha_cache.as_ref() { - let cache = load_alpha_cache(p)?; + let cache = ml::env::loaders::load_alpha_cache(p)?; info!("Loaded alpha cache: {} entries from {}", cache.len(), p.display()); Some(cache) } else { @@ -560,7 +407,7 @@ fn main() -> Result<()> { let rows: Vec = match (cli.fxcache_path.as_ref(), cli.mbp10_dir.as_ref()) { (Some(fxc), _) => { info!("Loading snapshots from fxcache: {}", fxc.display()); - load_snapshots_from_fxcache( + ml::env::loaders::load_snapshots_from_fxcache( fxc, cli.max_snapshots, alpha_cache_vec.as_deref(), diff --git a/crates/ml/src/env/loaders.rs b/crates/ml/src/env/loaders.rs new file mode 100644 index 000000000..7e03aca58 --- /dev/null +++ b/crates/ml/src/env/loaders.rs @@ -0,0 +1,182 @@ +//! Phase E loader helpers shared by `alpha_dqn_h600_smoke` and +//! `alpha_compose_backtest` (and any future Phase E binary). +//! +//! Three loaders: +//! +//! - [`load_alpha_cache`] — binary cache produced by +//! `alpha_train_stacker --alpha-cache-out`. Format: `[u32 n] [f32; n]` +//! little-endian. Each f32 is the per-bar stacker logit; bars +//! before `seq_len - 1` are 0.0 (Pearl A sentinel). +//! +//! - [`load_fill_model_from_json`] — parses the FillCoeffs JSON +//! emitted by `alpha_fit_fill_model`. Expects `bid_coeffs` and +//! `ask_coeffs` as 3-row arrays of 5-element f32 vectors. +//! +//! - [`load_snapshots_from_fxcache`] — builds the env's +//! `Vec` from a precomputed snapshot fxcache. Mid +//! comes from `raw_close`; bid/ask synthesized at fixed half-tick +//! offsets (the parser bug for L2-L9 is fixed but the fxcache +//! itself doesn't store the LOB so we synthesize). 81-dim Block-S +//! features feed `spread_bps`, `l1_imbalance`, `ofi_sum_5`, +//! `time_since_trade_s`, `book_event_rate`, `mid_drift_5`. +//! `alpha_logit` is filled from the optional alpha cache (else 0.0). + +use std::io::Read; +use std::path::Path; + +use anyhow::{Context, Result}; +use ml_alpha::fxcache_reader::{COL_RAW_CLOSE, FEAT_DIM, FxCacheReader}; +use serde_json::Value; +use tracing::{info, warn}; + +use crate::env::execution_env::SnapshotRow; +use crate::env::fill_model::{FillCoeffs, FillModel}; + +/// ES futures minimum price increment. +const TICK: f32 = 0.25; + +/// Load the alpha-logit cache (`[u32 n] [f32; n]` little-endian binary). +/// Each index aligns to the fxcache bar at the same index. +pub fn load_alpha_cache(path: &Path) -> Result> { + let mut f = std::fs::File::open(path) + .with_context(|| format!("open alpha cache {}", path.display()))?; + let mut len_bytes = [0u8; 4]; + f.read_exact(&mut len_bytes).context("read alpha-cache header")?; + let n = u32::from_le_bytes(len_bytes) as usize; + let mut buf = vec![0u8; n * 4]; + f.read_exact(&mut buf).context("read alpha-cache body")?; + let mut out = Vec::with_capacity(n); + for i in 0..n { + let off = i * 4; + let v = f32::from_le_bytes([buf[off], buf[off + 1], buf[off + 2], buf[off + 3]]); + out.push(v); + } + Ok(out) +} + +/// Parse FillModel from JSON (the artifact produced by +/// `alpha_fit_fill_model`). Expects `bid_coeffs` and `ask_coeffs` keys +/// each holding a 3-row × 5-element f32 array. +pub fn load_fill_model_from_json(path: &Path) -> Result { + let s = std::fs::read_to_string(path) + .with_context(|| format!("read fill coeffs JSON at {}", path.display()))?; + let v: Value = serde_json::from_str(&s) + .with_context(|| format!("parse fill coeffs JSON at {}", path.display()))?; + let parse_levels = |key: &str| -> Result<[FillCoeffs; 3]> { + let arr = v[key] + .as_array() + .ok_or_else(|| anyhow::anyhow!("missing/non-array `{}`", key))?; + if arr.len() != 3 { + anyhow::bail!("{} must have 3 entries, got {}", key, arr.len()); + } + let mut out: [FillCoeffs; 3] = [FillCoeffs { beta: [0.0; 5] }; 3]; + for (i, lvl) in arr.iter().enumerate() { + let vv = lvl + .as_array() + .ok_or_else(|| anyhow::anyhow!("{}[{}] not array", key, i))?; + if vv.len() != 5 { + anyhow::bail!("{}[{}] must have 5 floats, got {}", key, i, vv.len()); + } + for k in 0..5 { + out[i].beta[k] = vv[k] + .as_f64() + .ok_or_else(|| anyhow::anyhow!("{}[{}][{}] not number", key, i, k))? + as f32; + } + } + Ok(out) + }; + Ok(FillModel { + bid_coeffs: parse_levels("bid_coeffs")?, + ask_coeffs: parse_levels("ask_coeffs")?, + }) +} + +/// Build `Vec` from a precomputed fxcache. Mid from +/// `raw_close`; bid/ask synthesized at fixed half-tick offsets; 81-dim +/// Block-S features feed the runtime feature fields. If `alpha_cache` +/// is `Some`, each `SnapshotRow.alpha_logit` is populated from the +/// cache (and `alpha_confidence = |sigmoid(z) − 0.5|`); otherwise 0.0. +pub fn load_snapshots_from_fxcache( + fxcache_path: &Path, + max_snapshots: usize, + alpha_cache: Option<&[f32]>, +) -> Result> { + let reader = FxCacheReader::open(fxcache_path) + .with_context(|| format!("open fxcache {}", fxcache_path.display()))?; + let alpha_dim = reader + .alpha_feature_dim() + .ok_or_else(|| anyhow::anyhow!("fxcache lacks alpha column"))?; + if alpha_dim < 81 { + anyhow::bail!( + "fxcache alpha_dim={} but expected ≥81 (snapshot_pipeline layout)", + alpha_dim + ); + } + let n_bars_total = reader.bar_count(); + let n = n_bars_total.min(max_snapshots); + info!("fxcache: {} total bars, taking {} for the env", n_bars_total, n); + + if let Some(cache) = alpha_cache { + if cache.len() < n { + anyhow::bail!( + "alpha cache has {} entries but env wants {} bars", + cache.len(), + n + ); + } + } + + let mut rows: Vec = Vec::with_capacity(n); + let mut n_degenerate = 0_usize; + for i in 0..n { + let rec = reader.record(i); + let mid = rec.targets[COL_RAW_CLOSE - FEAT_DIM]; + if !mid.is_finite() || mid <= 0.0 { + n_degenerate += 1; + continue; + } + let features = reader + .alpha_features(i) + .ok_or_else(|| anyhow::anyhow!("missing alpha row at bar {}", i))?; + + let bid_l1 = mid - 0.125; + let ask_l1 = mid + 0.125; + let bid_l = [bid_l1, bid_l1 - TICK, bid_l1 - 2.0 * TICK]; + let ask_l = [ask_l1, ask_l1 + TICK, ask_l1 + 2.0 * TICK]; + + let spread_bps = features[78]; + let l1_imbalance = features[79]; + let ofi_sum_5 = features[0..5].iter().sum::(); + let mid_drift_5 = features[80]; + let time_since_trade_s = features[75]; + let book_event_rate = features[77]; + + let alpha_logit = alpha_cache.map(|c| c[i]).unwrap_or(0.0); + let alpha_confidence = { + let p = 1.0_f32 / (1.0 + (-alpha_logit.clamp(-50.0, 50.0)).exp()); + (p - 0.5).abs() + }; + + rows.push(SnapshotRow { + mid_price: mid, + bid_l, + ask_l, + alpha_logit, + alpha_confidence, + spread_bps, + l1_imbalance, + ofi_sum_5, + mid_drift_5, + time_since_trade_s, + book_event_rate, + }); + } + if n_degenerate > 0 { + warn!( + "fxcache loader: skipped {} degenerate bars (non-finite or zero mid)", + n_degenerate + ); + } + Ok(rows) +} diff --git a/crates/ml/src/env/mod.rs b/crates/ml/src/env/mod.rs index 1e77e2bf9..64611ec3d 100644 --- a/crates/ml/src/env/mod.rs +++ b/crates/ml/src/env/mod.rs @@ -12,6 +12,7 @@ pub mod action_space; pub mod execution_env; pub mod fill_model; +pub mod loaders; pub use action_space::{DecodedAction, ExecAction, N_ACTIONS, Placement, Side}; pub use execution_env::{EpisodeState, ExecutionEnv, ExecutionEnvConfig, ReplayRng, SnapshotRow, STATE_DIM};