gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(alpha): Phase E.3 composition backtest — reveals slot 543 needs consumption

Browse Source 
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.
This commit is contained in:
jgrusewski
2026-05-15 17:59:28 +02:00
parent 91383507fc
commit 2af8e02fd8
5 changed files with 844 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files

581
crates/ml/examples/alpha_compose_backtest.rs Normal file
View File

@@ -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<f32>,
#[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<u8> = 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<f32> = (0..N_WEIGHTS)
.map(|_| xavier_scale * 2.0 * (rng.next_f32() - 0.5))
.collect();
let b_init: Vec<f32> = 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::<f32>(N_WEIGHTS)?;
let mut db_dev = stream.alloc_zeros::<f32>(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::<f32>(cli.horizon * STATE_DIM)?;
let mut next_states_dev = stream.alloc_zeros::<f32>(cli.horizon * STATE_DIM)?;
let mut actions_dev = stream.alloc_zeros::<i32>(cli.horizon)?;
let mut rewards_dev = stream.alloc_zeros::<f32>(cli.horizon)?;
let mut dones_dev = stream.alloc_zeros::<f32>(cli.horizon)?;
let mut q_current_dev = stream.alloc_zeros::<f32>(cli.horizon * N_ACTIONS)?;
let mut q_next_dev = stream.alloc_zeros::<f32>(cli.horizon * N_ACTIONS)?;
let mut target_dev = stream.alloc_zeros::<f32>(cli.horizon)?;
let mut single_state_dev = stream.alloc_zeros::<f32>(STATE_DIM)?;
let mut single_q_dev = stream.alloc_zeros::<f32>(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<f32> = Vec::with_capacity(cli.horizon * STATE_DIM);
let mut next_states_host: Vec<f32> = Vec::with_capacity(cli.horizon * STATE_DIM);
let mut actions_host: Vec<i32> = Vec::with_capacity(cli.horizon);
let mut rewards_host: Vec<f32> = Vec::with_capacity(cli.horizon);
let mut dones_host: Vec<f32> = 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<f32> = 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<CostBin> = Vec::with_capacity(cli.cost_grid.len());
for &cost in &cli.cost_grid {
env.config.cost_per_contract = cost;
let mut rewards: Vec<f32> = 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::<f64>() / n;
let var = rewards
.iter()
.map(|r| (*r as f64 - mean).powi(2))
.sum::<f64>()
/ 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<ReplayRng> = 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(())
}

159
crates/ml/examples/alpha_dqn_h600_smoke.rs
View File

@@ -184,159 +184,6 @@ struct Cli {
out_path: PathBuf,
}
fn load_fill_model(path: &std::path::Path) -> Result<FillModel> {
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<Vec<f32>> {
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<Vec<SnapshotRow>> {
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<SnapshotRow> = 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::<f32>();
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<Vec<f32>> = 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<SnapshotRow> = 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(),

182
crates/ml/src/env/loaders.rs vendored Normal file
View File

@@ -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<SnapshotRow>` 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<Vec<f32>> {
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<FillModel> {
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<SnapshotRow>` 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<Vec<SnapshotRow>> {
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<SnapshotRow> = 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::<f32>();
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)
}

1
crates/ml/src/env/mod.rs vendored
View File

@@ -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};