7f4cd86421
CLI flags: --resume-from <path> and --checkpoint-every <n> (default 5000). Resume loads checkpoint and continues from saved step. Save writes rolling checkpoints (keeps last 2, deletes older). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1308 lines
63 KiB
Rust
1308 lines
63 KiB
Rust
// The per-step diag `json!{...}` block now spans ≥ 130 fields after the
|
||
// adaptive-clamp additions, exceeding serde_json's default macro recursion
|
||
// budget of 128. Bumping at the crate root keeps the expansion within limits.
|
||
#![recursion_limit = "256"]
|
||
|
||
//! Integrated RL trainer CLI (DQN + PPO on shared Mamba2 -> CfC encoder).
|
||
//!
|
||
//! Drives `IntegratedTrainer::step_with_lobsim` against a real `LobSimCuda`
|
||
//! backed by MBP-10 windows loaded via `MultiHorizonLoader::next_sequence_pair`
|
||
//! (R2) for true `(s_t, s_{t+1})` adjacency. The trainer combines:
|
||
//! * R3: GPU-resident EMA + advantage/return kernels.
|
||
//! * R4: GPU Thompson + Double-DQN argmax + log π_old kernels.
|
||
//! * R5: 7 RL controllers + target-net soft update.
|
||
//! * R6: GPU-pure env step (extract_realized_pnl_delta, apply_reward_scale,
|
||
//! actions_to_market_targets).
|
||
//! * R7a/R7b: host-lift of every per-step orchestration loop.
|
||
//! * R7c-data: true `h_{t+1}` / `V(s_{t+1})` / Bellman target on `h_{t+1}`.
|
||
//! * R7d: PER push/sample + off-policy DQN with stop-grad on encoder.
|
||
//!
|
||
//! Emits `alpha_rl_train_summary.json` on exit. Gate G8 (NaN abort): if any
|
||
//! per-head loss is non-finite at any step, the binary exits with code 2 so
|
||
//! the cluster smoke harness can terminate the workflow on first divergence.
|
||
//!
|
||
//! Usage:
|
||
//! alpha_rl_train \
|
||
//! --mbp10-data-dir /data/futures-baseline-mbp10/ES.FUT \
|
||
//! --predecoded-dir /feature-cache/predecoded \
|
||
//! --out /feature-cache/alpha-rl-runs/<sha> \
|
||
//! --n-steps 1000 \
|
||
//! --seed 16962 \
|
||
//! --instrument-mode front-month \
|
||
//! --gpu-idx 0
|
||
//!
|
||
//! Plan-doc reference: docs/superpowers/plans/2026-05-23-integrated-rl-trainer-rebuild.md
|
||
//! Section R8 (CLI binary + Argo template + dispatcher).
|
||
|
||
use anyhow::{Context, Result};
|
||
use clap::Parser;
|
||
use data::providers::databento::dbn_parser::InstrumentFilter;
|
||
// Use the canonical Action enum size, NOT a literal. Caught 2026-05-24
|
||
// during SP20 P4 dogfood: a `0..9` range survived the N_ACTIONS=9→11
|
||
// bump and silently dropped HalfFlat samples (a9, a10 showed 0% in
|
||
// diag despite P_MIN=0.02 floor guaranteeing 2% each). Per the
|
||
// emerging meta-pattern: any constant that mirrors a kernel-side
|
||
// structural dimension MUST reference the Rust const, not duplicate
|
||
// the literal value.
|
||
use ml_alpha::rl::common::{N_ACTIONS, FRD_N_ATOMS, FRD_N_HORIZONS};
|
||
use ml_alpha::rl::frd::FRD_OUT_DIM;
|
||
use ml_alpha::data::gpu_dataset::{GpuDataLoader, GpuDataset};
|
||
use ml_alpha::data::loader::{
|
||
discover_mbp10_files_sorted, MultiHorizonLoader, MultiHorizonLoaderConfig,
|
||
DEFAULT_OUTCOME_LABEL_COST_ES,
|
||
};
|
||
use ml_alpha::heads::HORIZONS;
|
||
use ml_alpha::rl::isv_slots::{
|
||
RL_ADVANTAGE_VAR_RATIO_EMA_INDEX, RL_ENTROPY_COEF_INDEX, RL_ENTROPY_OBSERVED_EMA_INDEX,
|
||
RL_GAMMA_INDEX, RL_KL_PI_EMA_INDEX, RL_LR_AUX_INDEX, RL_LR_BCE_INDEX, RL_LR_PI_BEST_LOSS_INDEX,
|
||
RL_LR_PI_INDEX, RL_LR_PI_LOSS_EMA_INDEX, RL_LR_PI_STEPS_SINCE_BEST_INDEX,
|
||
RL_LR_PI_WARMUP_COUNTER_INDEX, RL_LR_Q_BEST_LOSS_INDEX, RL_LR_Q_INDEX,
|
||
RL_LR_Q_LOSS_EMA_INDEX, RL_LR_Q_STEPS_SINCE_BEST_INDEX, RL_LR_Q_WARMUP_COUNTER_INDEX,
|
||
RL_LR_V_BEST_LOSS_INDEX, RL_LR_V_INDEX, RL_LR_V_LOSS_EMA_INDEX,
|
||
RL_LR_V_STEPS_SINCE_BEST_INDEX, RL_LR_V_WARMUP_COUNTER_INDEX,
|
||
RL_ADV_VAR_RATIO_CLAMP_INDEX, RL_ADV_VAR_RATIO_TARGET_INDEX, RL_ADV_VAR_STREAM_M2_INDEX,
|
||
RL_ADV_VAR_STREAM_MEAN_INDEX, RL_DIV_TARGET_INDEX, RL_ENTROPY_TARGET_FRAC_INDEX,
|
||
RL_EPS_BOOTSTRAP_INDEX, RL_IMPROVEMENT_THRESHOLD_INDEX, RL_KL_TARGET_INDEX,
|
||
RL_KURT_GAUSSIAN_INDEX, RL_KURT_LIFT_SCALE_INDEX, RL_KURT_NOISE_FLOOR_INDEX,
|
||
RL_K_LOOP_DIVISOR_INDEX, RL_K_LOOP_MAX_INDEX, RL_LOSS_LAMBDA_AUX_INDEX,
|
||
RL_LR_BOOTSTRAP_INDEX, RL_LR_DECAY_FACTOR_INDEX, RL_LR_LOSS_EMA_ALPHA_INDEX,
|
||
RL_LR_MAX_INDEX, RL_LR_MIN_INDEX, RL_LR_WARMUP_STEPS_INDEX,
|
||
RL_PLATEAU_PATIENCE_INDEX, RL_PPO_CLAMP_MARGIN_INDEX,
|
||
RL_PPO_RATIO_CLAMP_BOOTSTRAP_INDEX, RL_Q_ARG_VS_PI_AGREE_INDEX,
|
||
RL_REWARD_CLAMP_LOSS_INDEX, RL_REWARD_CLAMP_WIN_INDEX, RL_REWARD_SCALE_BOOTSTRAP_INDEX,
|
||
RL_ROLLOUT_BOOTSTRAP_INDEX, RL_SCHULMAN_ADJUST_RATE_INDEX, RL_SCHULMAN_TOLERANCE_INDEX,
|
||
RL_STREAM_ALPHA_INDEX, RL_TAU_BOOTSTRAP_INDEX,
|
||
RL_MAX_ABS_SCALED_REWARD_PRE_CLAMP_INDEX, RL_MEAN_ABS_PNL_EMA_INDEX,
|
||
RL_POS_SCALED_REWARD_MAX_INDEX, RL_POS_SCALED_REWARD_MAX_EMA_INDEX,
|
||
RL_REWARD_CLAMP_MARGIN_INDEX, RL_REWARD_CLAMP_RATIO_INDEX,
|
||
RL_REWARD_CLAMP_CLIP_RATE_EMA_INDEX, RL_REWARD_CLAMP_CLIP_RATE_TARGET_INDEX,
|
||
RL_C51_V_MAX_INDEX, RL_C51_V_MIN_INDEX,
|
||
RL_Q_DISTILL_LAMBDA_INDEX, RL_Q_DISTILL_TEMPERATURE_INDEX, RL_Q_DISTILL_KL_EMA_INDEX,
|
||
RL_NEG_SCALED_REWARD_MAX_INDEX, RL_NEG_SCALED_REWARD_MAX_EMA_INDEX,
|
||
RL_Q_DISTILL_KL_TARGET_INDEX, RL_REWARD_SCALE_MIN_INDEX,
|
||
RL_MEAN_TRADE_DURATION_EMA_INDEX, RL_N_ROLLOUT_STEPS_INDEX, RL_PER_ALPHA_INDEX,
|
||
RL_PI_GRAD_NORM_EMA_INDEX, RL_PPO_CLIP_INDEX, RL_PPO_LOG_RATIO_ABS_MAX_INDEX,
|
||
RL_PPO_RATIO_CLAMP_MAX_INDEX, RL_Q_DIVERGENCE_EMA_INDEX, RL_Q_GRAD_NORM_EMA_INDEX,
|
||
RL_REWARD_SCALE_INDEX, RL_TARGET_TAU_INDEX, RL_TD_KURTOSIS_CLAMP_INDEX,
|
||
RL_TD_KURTOSIS_EMA_INDEX, RL_TD_KURT_STREAM_M2_INDEX, RL_TD_KURT_STREAM_M4_INDEX,
|
||
RL_TD_KURT_STREAM_MEAN_INDEX, RL_V_GRAD_NORM_EMA_INDEX,
|
||
// P1+P2 feature metrics
|
||
RL_POPART_MEAN_INDEX, RL_POPART_SIGMA_INDEX, RL_POPART_VAR_INDEX,
|
||
RL_SPECTRAL_NORM_MAX_INDEX, RL_SPECTRAL_DECOUPLE_LAMBDA_INDEX,
|
||
RL_Q_BIAS_EMA_INDEX, RL_Q_BIAS_CORRECTION_INDEX,
|
||
RL_LR_SCALE_Q_INDEX, RL_LR_SCALE_PI_INDEX, RL_LR_SCALE_V_INDEX, RL_LR_SCALE_IQN_INDEX,
|
||
RL_OUTCOME_AUX_LAMBDA_INDEX,
|
||
RL_SAC_ALPHA_INDEX, RL_SAC_ENTROPY_TARGET_INDEX,
|
||
RL_ACTION_ENTROPY_EMA_INDEX,
|
||
};
|
||
use ml_alpha::trainer::diag_staging::DiagStaging;
|
||
use ml_alpha::trainer::integrated::{
|
||
IntegratedStepStats, IntegratedTrainer, IntegratedTrainerConfig,
|
||
};
|
||
use ml_alpha::trainer::perception::PerceptionTrainerConfig;
|
||
use ml_backtesting::sim::LobSimCuda;
|
||
use ml_core::device::MlDevice;
|
||
use serde::Serialize;
|
||
use serde_json::json;
|
||
use std::io::{BufWriter, Write};
|
||
use std::path::PathBuf;
|
||
use std::process;
|
||
|
||
#[derive(Parser)]
|
||
#[command(name = "alpha_rl_train")]
|
||
struct Cli {
|
||
/// Directory containing MBP-10 .dbn.zst files. Per
|
||
/// `feedback_mbp10_mandatory`, this argument is required — no
|
||
/// synthetic-data fallback in the production training path.
|
||
#[arg(long)]
|
||
mbp10_data_dir: PathBuf,
|
||
|
||
/// Directory holding predecoded sidecar caches (created on miss).
|
||
#[arg(long)]
|
||
predecoded_dir: PathBuf,
|
||
|
||
/// Output directory for summary JSON + per-step log artefacts.
|
||
#[arg(long)]
|
||
out: PathBuf,
|
||
|
||
/// Total `step_with_lobsim` calls to run. The validation smoke
|
||
/// (audit prerequisite) ships with 1000; production scale-up runs
|
||
/// 50k+. Each call processes one `(s_t, s_{t+1})` sequence pair
|
||
/// across `n_backtests` parallel LobSim instances.
|
||
#[arg(long, default_value_t = 1000)]
|
||
n_steps: usize,
|
||
|
||
/// Encoder sequence length (snapshots per step). Must match
|
||
/// `PerceptionTrainerConfig::seq_len`. Default 32 matches the
|
||
/// validated single-scale `1:32` window from alpha_train.
|
||
#[arg(long, default_value_t = 32)]
|
||
seq_len: usize,
|
||
|
||
/// Batch dimension — number of parallel LobSim backtests run in
|
||
/// lockstep per step. Larger batches amortise the encoder forward
|
||
/// cost across more trajectories but push GPU memory pressure
|
||
/// linearly. Default 16 — fixes the b_size=1 signal starvation
|
||
/// per `pearl_b_size_1_signal_starvation_blocks_q_learning`
|
||
/// (Q stayed at uniform baseline ln(21)=3.04 across all b_size=1
|
||
/// smokes regardless of controller tuning; 16 parallel
|
||
/// transitions per Adam step give 16× the gradient signal). Was
|
||
/// 1 (validation-smoke default); production sweep can push to
|
||
/// 32-64 on L40S 48GB.
|
||
#[arg(long, default_value_t = 16)]
|
||
n_backtests: usize,
|
||
|
||
/// PER buffer capacity. At b_size=16, capacity/16 = unique steps
|
||
/// retained. 32768/16 = 2048 steps of replay depth.
|
||
#[arg(long, default_value_t = 32768)]
|
||
per_capacity: usize,
|
||
|
||
/// Random seed pinning RNG + Xavier init + ISV controller bootstrap
|
||
/// state. Per `pearl_scoped_init_seed_for_reproducibility`.
|
||
#[arg(long, default_value_t = 16962)]
|
||
seed: u64,
|
||
|
||
/// MBP-10 instrument filter — `all`, `front-month`, or `id=<N>`.
|
||
/// `front-month` is the canonical setting for ES (auto-detects the
|
||
/// dominant contract per file, handles quarterly roll).
|
||
#[arg(long, default_value = "front-month", value_parser = parse_instrument_mode)]
|
||
instrument_mode: InstrumentFilter,
|
||
|
||
/// CUDA device index. Single-GPU dev (RTX 3050 Ti) uses 0; cluster
|
||
/// L40S/H100 also 0 (one GPU per pod per the alpha-rl template).
|
||
#[arg(long, default_value_t = 0)]
|
||
gpu_idx: usize,
|
||
|
||
/// Round-trip trade cost in price units for the D-style outcome
|
||
/// labels (consumed by the perception aux head, not by the RL
|
||
/// path directly — but the loader needs it). Default 0.5 = 2 ticks
|
||
/// × $0.25/tick for ES.
|
||
#[arg(long, default_value_t = DEFAULT_OUTCOME_LABEL_COST_ES)]
|
||
outcome_label_cost: f32,
|
||
|
||
/// How often (in steps) to flush a progress line to stderr. The
|
||
/// summary JSON is written ONCE at the end; this controls in-flight
|
||
/// visibility only.
|
||
#[arg(long, default_value_t = 100)]
|
||
log_every: usize,
|
||
|
||
/// Walk-forward fold index for the multi-fold G8 gate (per
|
||
/// `pearl_single_window_oos_is_not_oos` — a single window is NOT
|
||
/// out-of-sample). Slices the MBP-10 file list into K equal-sized
|
||
/// blocks; fold k trains on blocks [0..k+1) and evaluates on block
|
||
/// [k+1] (purged walk-forward). At `fold_idx=0` the first block
|
||
/// is the train window, second is eval. At `fold_idx=K-2` the
|
||
/// trainer sees almost all the data and evaluates on the final
|
||
/// block. The dispatcher (`scripts/argo-alpha-rl.sh --n-folds K`)
|
||
/// fans out K-1 workflows, one per fold; the aggregator computes
|
||
/// mean ± SD of the per-fold profit_factor.
|
||
///
|
||
/// Default 0 + n-folds=1 = single-window mode (no eval split).
|
||
#[arg(long, default_value_t = 0)]
|
||
fold_idx: usize,
|
||
|
||
/// Total walk-forward fold count. Set with `--fold-idx` for the
|
||
/// multi-fold G8 gate. n_folds=1 disables the train/eval split
|
||
/// (single-window mode — only meaningful for smoke validation).
|
||
/// n_folds≥3 enables a real evaluation: train on first
|
||
/// `fold_idx + 1` of `n_folds` data blocks, evaluate on the next.
|
||
#[arg(long, default_value_t = 1)]
|
||
n_folds: usize,
|
||
|
||
/// Eval-phase step count. After the train phase completes its
|
||
/// `--n-steps`, the eval phase runs `--n-eval-steps` additional
|
||
/// steps on the held-out fold using the same trainer
|
||
/// (acknowledged minor in-eval learning contamination at b_size=1
|
||
/// — pure eval mode is a follow-up). LobSim trade records are
|
||
/// drained at the END so the summary reflects eval-phase trades
|
||
/// only (train-phase trades are flushed when the eval phase
|
||
/// starts).
|
||
///
|
||
/// 0 disables the eval phase entirely (single-window training
|
||
/// only — the summary's profit_factor will reflect the full run
|
||
/// including train).
|
||
#[arg(long, default_value_t = 0)]
|
||
n_eval_steps: usize,
|
||
|
||
/// Per-step diagnostic JSONL path. One record per step capturing:
|
||
/// * step number, wall time
|
||
/// * loss components (l_bce/q/pi/v/aux/total) + λs (5)
|
||
/// * all 7 RL controller outputs (γ τ ε coef n_roll per_α scale)
|
||
/// * all 5 per-head learning rates (lr_bce/q/pi/v/aux)
|
||
/// * all 7 EMA inputs the controllers consume
|
||
/// * replay buffer length
|
||
/// * per-step reward + done aggregates (DtoH b_size floats per stat)
|
||
/// * per-step action histogram (counts per action class)
|
||
///
|
||
/// Default: `<out>/diag.jsonl`. Pass `--diag-jsonl /dev/null` (or
|
||
/// any unwritable path that errors at open) to disable. The dump
|
||
/// happens AFTER each `step_with_lobsim` so it sees the freshly-
|
||
/// adapted ISV values. Writes are line-buffered + flushed every
|
||
/// `log_every` steps so a `tail -f` shows in-flight progress.
|
||
///
|
||
/// Critically: this provides the per-step diagnostic visibility
|
||
/// the cluster smoke needs to detect anomalies early. Without it,
|
||
/// the smoke is "blind" — only the final summary is observable,
|
||
/// and a controller drifting off-anchor mid-run is invisible until
|
||
/// it produces a non-finite loss (G8 NaN abort).
|
||
#[arg(long)]
|
||
diag_jsonl: Option<PathBuf>,
|
||
|
||
/// Resume from checkpoint file path.
|
||
#[arg(long)]
|
||
resume_from: Option<PathBuf>,
|
||
|
||
/// Checkpoint interval in steps (default: 5000). Set to 0 to disable.
|
||
#[arg(long, default_value = "5000")]
|
||
checkpoint_every: usize,
|
||
}
|
||
|
||
fn parse_instrument_mode(s: &str) -> Result<InstrumentFilter, String> {
|
||
let trimmed = s.trim();
|
||
if trimmed.eq_ignore_ascii_case("all") {
|
||
return Ok(InstrumentFilter::All);
|
||
}
|
||
if trimmed.eq_ignore_ascii_case("front-month")
|
||
|| trimmed.eq_ignore_ascii_case("front_month")
|
||
{
|
||
return Ok(InstrumentFilter::FrontMonth);
|
||
}
|
||
if let Some(num) = trimmed.strip_prefix("id=") {
|
||
let id: u32 = num.parse().map_err(|e| {
|
||
format!("instrument-mode: expected `id=<u32>` but failed to parse '{num}': {e}")
|
||
})?;
|
||
return Ok(InstrumentFilter::Id(id));
|
||
}
|
||
Err(format!(
|
||
"instrument-mode: expected `all`, `front-month`, or `id=<N>`; got '{s}'"
|
||
))
|
||
}
|
||
|
||
#[derive(Serialize, Default)]
|
||
struct AlphaRlTrainSummary {
|
||
n_steps_planned: usize,
|
||
n_steps_completed: usize,
|
||
seq_len: usize,
|
||
n_backtests: usize,
|
||
per_capacity: usize,
|
||
seed: u64,
|
||
|
||
/// True iff `n_steps_completed == n_steps_planned` AND no NaN
|
||
/// abort fired. False if early-terminated for any reason (NaN,
|
||
/// loader EOF, panic — though panics also exit nonzero).
|
||
completed_clean: bool,
|
||
|
||
/// Final-step loss components (raw, not λ-weighted). All NaN if
|
||
/// trainer never returned a finite step.
|
||
final_l_bce: f32,
|
||
final_l_q: f32,
|
||
final_l_pi: f32,
|
||
final_l_v: f32,
|
||
final_l_aux: f32,
|
||
final_l_total: f32,
|
||
|
||
/// Replay buffer state at exit.
|
||
final_replay_len: usize,
|
||
|
||
/// Step at which the first NaN was observed, if any. -1 means
|
||
/// no NaN observed.
|
||
nan_abort_step: i64,
|
||
}
|
||
|
||
fn main() -> Result<()> {
|
||
tracing_subscriber::fmt()
|
||
.with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
|
||
.init();
|
||
|
||
let cli = Cli::parse();
|
||
|
||
std::fs::create_dir_all(&cli.out)
|
||
.with_context(|| format!("mkdir -p {}", cli.out.display()))?;
|
||
std::fs::create_dir_all(&cli.predecoded_dir)
|
||
.with_context(|| format!("mkdir -p {}", cli.predecoded_dir.display()))?;
|
||
|
||
let dev = MlDevice::cuda(cli.gpu_idx)
|
||
.with_context(|| format!("MlDevice::cuda({})", cli.gpu_idx))?;
|
||
eprintln!("CUDA device {} initialised", cli.gpu_idx);
|
||
|
||
// Loader: paired (s_t, s_{t+1}) sequences. inference_only=false so
|
||
// next_sequence_pair is available (the labels it preloads are
|
||
// unused by step_with_lobsim, but the API requires the label
|
||
// precomputation pass per `feedback_no_partial_refactor`).
|
||
let all_files = discover_mbp10_files_sorted(&cli.mbp10_data_dir)
|
||
.with_context(|| format!("discover_mbp10_files_sorted({})", cli.mbp10_data_dir.display()))?;
|
||
anyhow::ensure!(
|
||
!all_files.is_empty(),
|
||
"no MBP-10 .dbn.zst files under {}",
|
||
cli.mbp10_data_dir.display()
|
||
);
|
||
eprintln!("loader: {} MBP-10 files total", all_files.len());
|
||
|
||
// Walk-forward fold split per `pearl_single_window_oos_is_not_oos`.
|
||
// Slice the file list into `n_folds` equal blocks. fold k uses
|
||
// blocks [0..=k] for training and block [k+1] for evaluation. With
|
||
// n_folds=1 (default) train_files = all_files and eval_files is
|
||
// empty (single-window mode, no eval phase).
|
||
let (train_files, eval_files) = if cli.n_folds <= 1 || cli.n_eval_steps == 0 {
|
||
eprintln!("walk-forward: disabled (n_folds={}, n_eval_steps={})", cli.n_folds, cli.n_eval_steps);
|
||
(all_files, Vec::new())
|
||
} else {
|
||
anyhow::ensure!(
|
||
cli.fold_idx + 1 < cli.n_folds,
|
||
"fold_idx {} requires fold_idx + 1 < n_folds ({}); the last block must be the eval window",
|
||
cli.fold_idx,
|
||
cli.n_folds
|
||
);
|
||
let n_files = all_files.len();
|
||
let block_size = n_files / cli.n_folds;
|
||
anyhow::ensure!(
|
||
block_size >= 1,
|
||
"not enough MBP-10 files ({}) to split into n_folds={} (need ≥ n_folds files)",
|
||
n_files,
|
||
cli.n_folds
|
||
);
|
||
// Train: blocks 0..=fold_idx, Eval: block fold_idx+1.
|
||
let train_end = (cli.fold_idx + 1) * block_size;
|
||
let eval_end = (cli.fold_idx + 2) * block_size;
|
||
let train: Vec<_> = all_files[..train_end].to_vec();
|
||
let eval: Vec<_> = all_files[train_end..eval_end.min(n_files)].to_vec();
|
||
eprintln!(
|
||
"walk-forward fold {}/{}: train={} files ({}..{}), eval={} files ({}..{})",
|
||
cli.fold_idx, cli.n_folds, train.len(), 0, train_end,
|
||
eval.len(), train_end, eval_end.min(n_files)
|
||
);
|
||
(train, eval)
|
||
};
|
||
|
||
let multi_resolution: ml_alpha::data::aggregation::MultiResolutionConfig =
|
||
format!("1:{}", cli.seq_len)
|
||
.parse()
|
||
.with_context(|| format!("multi-resolution `1:{}`", cli.seq_len))?;
|
||
anyhow::ensure!(
|
||
multi_resolution.total_positions() == cli.seq_len,
|
||
"internal: single-scale 1:{} should produce total_positions={} but got {}",
|
||
cli.seq_len,
|
||
cli.seq_len,
|
||
multi_resolution.total_positions()
|
||
);
|
||
|
||
let loader_cfg = MultiHorizonLoaderConfig {
|
||
files: train_files,
|
||
predecoded_dir: cli.predecoded_dir.clone(),
|
||
multi_resolution: multi_resolution.clone(),
|
||
horizons: HORIZONS,
|
||
// Soft cap above n_steps × n_backtests × 2 so next_sequence_pair
|
||
// never returns Ok(None) mid-run. Each step samples n_backtests
|
||
// pairs (one per batch slot) to build the B×K snapshot tensor
|
||
// that forward_encoder consumes — per-batch market diversity is
|
||
// what gives the gradient-variance reduction promised by
|
||
// `pearl_b_size_1_signal_starvation_blocks_q_learning`.
|
||
n_max_sequences: cli.n_steps
|
||
.saturating_mul(cli.n_backtests.max(1))
|
||
.saturating_mul(2)
|
||
.max(64),
|
||
seed: cli.seed,
|
||
inference_only: false,
|
||
outcome_label_cost: cli.outcome_label_cost,
|
||
instrument_filter: cli.instrument_mode,
|
||
frd_horizon_ticks: ml_alpha::rl::common::FRD_HORIZON_TICKS,
|
||
frd_bucket_range_sigma: ml_alpha::rl::common::FRD_BUCKET_RANGE_SIGMA,
|
||
};
|
||
let loader = MultiHorizonLoader::new(&loader_cfg).context("MultiHorizonLoader::new")?;
|
||
|
||
// ── GPU-resident dataset upload. ────────────────────────────────
|
||
// Upload ALL pre-converted snapshots + FRD labels to GPU memory as
|
||
// a single contiguous device buffer. Per-step data loading becomes
|
||
// pure GPU kernel work (sample + gather) — zero CPU data work.
|
||
let gpu_dataset: GpuDataset = loader
|
||
.upload_to_gpu(&dev, cli.n_backtests.max(1), cli.seed)
|
||
.context("loader.upload_to_gpu")?;
|
||
let mut gpu_data_loader = GpuDataLoader::new(&dev, cli.n_backtests.max(1))
|
||
.context("GpuDataLoader::new")?;
|
||
eprintln!(
|
||
"GPU dataset uploaded: {} snapshots, {} files, {:.1} GB",
|
||
gpu_dataset.total_snapshots,
|
||
gpu_dataset.n_files,
|
||
(gpu_dataset.total_snapshots * 216) as f64 / 1e9,
|
||
);
|
||
|
||
// Trainer + LobSim.
|
||
let perception_cfg = PerceptionTrainerConfig {
|
||
seq_len: cli.seq_len,
|
||
n_batch: cli.n_backtests,
|
||
..PerceptionTrainerConfig::default()
|
||
};
|
||
let trainer_cfg = IntegratedTrainerConfig {
|
||
perception: perception_cfg,
|
||
dqn_seed: cli.seed.wrapping_add(0xDA),
|
||
ppo_seed: cli.seed.wrapping_add(0xBE),
|
||
per_capacity: cli.per_capacity,
|
||
per_seed: cli.seed.wrapping_add(0x9E37_79B9),
|
||
};
|
||
let mut trainer =
|
||
IntegratedTrainer::new(&dev, trainer_cfg).context("IntegratedTrainer::new")?;
|
||
let mut sim = LobSimCuda::new(cli.n_backtests, &dev).context("LobSimCuda::new")?;
|
||
eprintln!(
|
||
"trainer + LobSimCuda initialised: seq_len={} n_backtests={} per_capacity={}",
|
||
cli.seq_len, cli.n_backtests, cli.per_capacity
|
||
);
|
||
|
||
let mut summary = AlphaRlTrainSummary {
|
||
n_steps_planned: cli.n_steps,
|
||
seq_len: cli.seq_len,
|
||
n_backtests: cli.n_backtests,
|
||
per_capacity: cli.per_capacity,
|
||
seed: cli.seed,
|
||
nan_abort_step: -1,
|
||
..Default::default()
|
||
};
|
||
let mut last_stats: Option<IntegratedStepStats> = None;
|
||
|
||
// ── Per-step JSONL diag writer. ──────────────────────────────────
|
||
// Default path: `<out>/diag.jsonl`. The smoke run's failure-mode
|
||
// catalogue (NaN abort, controller drift, replay stagnation, reward
|
||
// explosion) is detected via post-hoc inspection of this file —
|
||
// without it, the run is blind beyond the eprintln every-N-step
|
||
// ticks.
|
||
let diag_path = cli
|
||
.diag_jsonl
|
||
.clone()
|
||
.unwrap_or_else(|| cli.out.join("diag.jsonl"));
|
||
let diag_file = std::fs::File::create(&diag_path)
|
||
.with_context(|| format!("create diag jsonl {}", diag_path.display()))?;
|
||
let mut diag = BufWriter::new(diag_file);
|
||
eprintln!("per-step diag JSONL: {}", diag_path.display());
|
||
|
||
// Async non-blocking diagnostic staging: separate CUDA stream +
|
||
// double-buffered mapped-pinned memory. ALL per-step device reads
|
||
// (ISV, rewards, dones, actions, raw_rewards, trade_duration,
|
||
// outcome_ema, position_lots, pyramid_count, unit state x4,
|
||
// close_unit_index, frd_logits) use async DtoD copies — zero
|
||
// stream syncs on the training stream. The host reads the PREVIOUS
|
||
// step's buffer while current step copies run on the diag stream.
|
||
let mut diag_staging = DiagStaging::new(&dev, cli.n_backtests)
|
||
.context("DiagStaging::new")?;
|
||
|
||
// Per-step scratch (replaced each step via mapped-pinned helpers).
|
||
|
||
// ── Training loop. ───────────────────────────────────────────────
|
||
// Windowed action distribution — EMA-smoothed across recent steps
|
||
// so action_entropy is meaningful at b_size=1 (per-step act_hist
|
||
// is one-hot, H=0; windowed dist captures actual policy variety).
|
||
// α = 1/1000 → half-life ≈ 690 steps. Tracks the recent ~1k steps'
|
||
// action distribution; entropy of normalised windowed_act_hist is
|
||
// a real exploration signal.
|
||
// SP20 P4: action histogram width MUST track the Action enum
|
||
// size. Always reference the const (NOT a literal) — see
|
||
// import-site comment for the dogfood-caught literal-drift bug.
|
||
let mut windowed_act_hist: [f32; N_ACTIONS] = [0.0; N_ACTIONS];
|
||
const WINDOWED_ACT_ALPHA: f32 = 1.0 / 1000.0;
|
||
|
||
let mut pnl_cum_usd: f64 = 0.0;
|
||
let mut win_count: u64 = 0;
|
||
let mut total_trades: u64 = 0;
|
||
let mut hold_time_sum: f64 = 0.0;
|
||
let mut trail_fired_total: u64 = 0;
|
||
let mut trail_tighten_total: u64 = 0;
|
||
let mut trail_loosen_total: u64 = 0;
|
||
let mut pyramid_added_total: u64 = 0;
|
||
let mut partial_flat_total: u64 = 0;
|
||
let mut partial_flat_long_total: u64 = 0;
|
||
let mut partial_flat_short_total: u64 = 0;
|
||
let mut conf_gate_total: u64 = 0;
|
||
let mut frd_gate_total: u64 = 0;
|
||
let mut heat_cap_total: u64 = 0;
|
||
|
||
let start_step = if let Some(ref ckpt_path) = cli.resume_from {
|
||
let resumed_step = trainer.load_checkpoint(&dev, ckpt_path)
|
||
.with_context(|| format!("resume from {}", ckpt_path.display()))?;
|
||
eprintln!("resumed from checkpoint at step {resumed_step}");
|
||
resumed_step as usize
|
||
} else {
|
||
0
|
||
};
|
||
|
||
let t_start = std::time::Instant::now();
|
||
for step in start_step..cli.n_steps {
|
||
// GPU-resident data loading: sample_and_gather + gather_next +
|
||
// gather_current + gather_frd_labels all run as GPU kernels
|
||
// inside step_with_lobsim_gpu. Zero CPU data work per step.
|
||
let stats = trainer
|
||
.step_with_lobsim_gpu(&mut gpu_data_loader, &gpu_dataset, &mut sim)
|
||
.with_context(|| format!("step_with_lobsim_gpu at step {step}"))?;
|
||
|
||
// Gate G8: NaN abort. Per `feedback_stop_on_anomaly` +
|
||
// `feedback_kill_runs_on_anomaly_quickly` the cluster smoke
|
||
// tail-watcher kills the workflow on non-zero exit; here we
|
||
// signal that with exit code 2 (distinct from clean exit 0
|
||
// and panic-derived exit codes).
|
||
let non_finite = !stats.l_total.is_finite()
|
||
|| !stats.l_bce.is_finite()
|
||
|| !stats.l_q.is_finite()
|
||
|| !stats.l_pi.is_finite()
|
||
|| !stats.l_v.is_finite()
|
||
|| !stats.l_aux.is_finite();
|
||
if non_finite {
|
||
eprintln!(
|
||
"G8 NaN ABORT at step {step}: l_bce={} l_q={} l_pi={} l_v={} l_aux={} l_total={}",
|
||
stats.l_bce, stats.l_q, stats.l_pi, stats.l_v, stats.l_aux, stats.l_total
|
||
);
|
||
summary.n_steps_completed = step;
|
||
summary.nan_abort_step = step as i64;
|
||
summary.completed_clean = false;
|
||
write_summary(&cli.out, &summary)?;
|
||
process::exit(2);
|
||
}
|
||
|
||
// Event-sync the diag snapshot and swap buffers. Previous step's
|
||
// async DtoD copies complete here. Placed AFTER step_with_lobsim
|
||
// so the diag copies have had the entire training step to finish —
|
||
// the event sync is instant (~0 us). On step 0 this is a no-op
|
||
// (event sync on a freshly-created event returns immediately).
|
||
diag_staging.sync_and_swap().context("diag sync_and_swap")?;
|
||
|
||
// Launch async DtoD copies of ALL diag buffers into the current
|
||
// staging buffer. Non-blocking on the training stream — the copies
|
||
// run on diag_staging's separate stream.
|
||
diag_staging
|
||
.snapshot_async(
|
||
trainer.isv_dev_ptr,
|
||
trainer.rewards_d.raw_ptr(),
|
||
trainer.dones_d.raw_ptr(),
|
||
trainer.actions_d.raw_ptr(),
|
||
trainer.raw_rewards_d.raw_ptr(),
|
||
trainer.trade_duration_emit_d.raw_ptr(),
|
||
trainer.outcome_ema_d.raw_ptr(),
|
||
trainer.prev_position_lots_d.raw_ptr(),
|
||
trainer.pyramid_units_count_d.raw_ptr(),
|
||
trainer.unit_entry_price_d.raw_ptr(),
|
||
trainer.unit_entry_step_d.raw_ptr(),
|
||
trainer.unit_lots_d.raw_ptr(),
|
||
trainer.unit_trail_distance_d.raw_ptr(),
|
||
trainer.close_unit_index_d.raw_ptr(),
|
||
trainer.frd_logits_d.raw_ptr(),
|
||
)
|
||
.context("diag snapshot_async")?;
|
||
|
||
// ── Per-step diag dump (async staging reads). ────────────────
|
||
// ALL diag reads come from the DiagStaging double-buffer — zero
|
||
// GPU stalls on the training stream. On step 0 these report
|
||
// zeros (no prior snapshot); from step 1+ they report step N-1's
|
||
// data (one-step delay, acceptable for diagnostics).
|
||
let rewards_host = diag_staging.read_rewards();
|
||
let dones_host = diag_staging.read_dones();
|
||
let actions_raw = diag_staging.read_actions_raw();
|
||
let actions_host: Vec<i32> = actions_raw
|
||
.iter()
|
||
.map(|f| f.to_bits() as i32)
|
||
.collect();
|
||
let raw_rewards_host = diag_staging.read_raw_rewards();
|
||
let trade_duration_host = diag_staging.read_trade_duration();
|
||
let outcome_ema_host = diag_staging.read_outcome_ema();
|
||
let position_lots_host = diag_staging.read_position_lots();
|
||
let pyramid_count_host = diag_staging.read_pyramid_count();
|
||
let unit_entry_price_host = diag_staging.read_unit_entry_price();
|
||
let unit_entry_step_host = diag_staging.read_unit_entry_step();
|
||
let unit_lots_host = diag_staging.read_unit_lots();
|
||
let unit_trail_host = diag_staging.read_unit_trail();
|
||
// Derive active mask from unit_lots (lots != 0 → active).
|
||
// Avoids a u8-to-f32 type mismatch in DtoD staging.
|
||
let unit_active_host: Vec<u8> = unit_lots_host
|
||
.iter()
|
||
.map(|&l| if l != 0 { 1u8 } else { 0u8 })
|
||
.collect();
|
||
let close_unit_index_host = diag_staging.read_close_unit_index();
|
||
|
||
let mut act_hist = [0u32; N_ACTIONS];
|
||
for &a in &actions_host {
|
||
if (a as usize) < N_ACTIONS {
|
||
act_hist[a as usize] += 1;
|
||
}
|
||
}
|
||
// audit — action entropy H(action_dist) windowed.
|
||
//
|
||
// Per-batch act_hist is one-hot at b_size=1 so per-step entropy
|
||
// is structurally 0. Maintain an EMA-smoothed action histogram
|
||
// across recent steps (α=1/1000, half-life ≈ 690 steps) and
|
||
// compute entropy on the normalised window. This captures the
|
||
// actual policy exploration variety over the recent run.
|
||
let total_actions: u32 = act_hist.iter().sum();
|
||
if total_actions > 0 {
|
||
for i in 0..N_ACTIONS {
|
||
let p_step = (act_hist[i] as f32) / (total_actions as f32);
|
||
windowed_act_hist[i] = (1.0 - WINDOWED_ACT_ALPHA)
|
||
* windowed_act_hist[i]
|
||
+ WINDOWED_ACT_ALPHA * p_step;
|
||
}
|
||
}
|
||
let win_sum: f32 = windowed_act_hist.iter().sum();
|
||
let action_entropy: f32 = if win_sum > 1e-9 {
|
||
windowed_act_hist
|
||
.iter()
|
||
.filter(|&&p| p > 1e-9)
|
||
.map(|&p_raw| {
|
||
let p = p_raw / win_sum;
|
||
-p * p.ln()
|
||
})
|
||
.sum()
|
||
} else {
|
||
0.0
|
||
};
|
||
let reward_sum: f32 = rewards_host.iter().sum();
|
||
let reward_max = rewards_host.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
|
||
let reward_min = rewards_host.iter().cloned().fold(f32::INFINITY, f32::min);
|
||
let reward_abs_max = rewards_host.iter().map(|r| r.abs()).fold(0.0f32, f32::max);
|
||
let done_count: u32 = dones_host.iter().map(|&d| if d > 0.5 { 1 } else { 0 }).sum();
|
||
|
||
// Per-trade cumulative stats (surfer validation).
|
||
// Use post-scale reward / current_scale to recover approximate USD.
|
||
let current_scale = trainer.read_isv_host(ml_alpha::rl::isv_slots::RL_REWARD_SCALE_INDEX);
|
||
for b in 0..cli.n_backtests {
|
||
if dones_host[b] > 0.5 {
|
||
let pnl_usd = if current_scale > 1e-9 {
|
||
rewards_host[b] as f64 / current_scale as f64
|
||
} else {
|
||
rewards_host[b] as f64
|
||
};
|
||
pnl_cum_usd += pnl_usd;
|
||
total_trades += 1;
|
||
if pnl_usd > 0.0 { win_count += 1; }
|
||
hold_time_sum += trade_duration_host[b] as f64;
|
||
}
|
||
}
|
||
let win_rate = if total_trades > 0 { win_count as f64 / total_trades as f64 } else { 0.0 };
|
||
let avg_hold = if total_trades > 0 { hold_time_sum / total_trades as f64 } else { 0.0 };
|
||
|
||
let isv = trainer.isv_host_slice();
|
||
|
||
// Per-step counters derived from action histogram + ISV diag slots.
|
||
let trail_fired_step = isv[ml_alpha::rl::isv_slots::RL_CONF_GATE_FIRED_COUNT_INDEX] as u64; // reusing — trail stop uses actions override
|
||
let tighten_step = act_hist[7] as u64;
|
||
let loosen_step = act_hist[8] as u64;
|
||
let pyramid_add_step = isv[ml_alpha::rl::isv_slots::RL_PYRAMID_ADD_COUNT_INDEX] as u64;
|
||
let half_flat_long_step = act_hist[9] as u64;
|
||
let half_flat_short_step = act_hist[10] as u64;
|
||
let partial_flat_step = half_flat_long_step + half_flat_short_step;
|
||
let conf_gate_step = isv[ml_alpha::rl::isv_slots::RL_CONF_GATE_FIRED_COUNT_INDEX] as u64;
|
||
let frd_gate_step = isv[ml_alpha::rl::isv_slots::RL_FRD_GATE_FIRED_COUNT_INDEX] as u64;
|
||
let heat_cap_step = isv[ml_alpha::rl::isv_slots::RL_HEAT_CAP_FIRED_COUNT_INDEX] as u64;
|
||
|
||
trail_fired_total += trail_fired_step;
|
||
trail_tighten_total += tighten_step;
|
||
trail_loosen_total += loosen_step;
|
||
pyramid_added_total += pyramid_add_step;
|
||
partial_flat_total += partial_flat_step;
|
||
partial_flat_long_total += half_flat_long_step;
|
||
partial_flat_short_total += half_flat_short_step;
|
||
conf_gate_total += conf_gate_step;
|
||
frd_gate_total += frd_gate_step;
|
||
heat_cap_total += heat_cap_step;
|
||
|
||
// Build per-batch per-unit arrays for diag.
|
||
let units_diag = {
|
||
let b = cli.n_backtests;
|
||
let mut entry_price_arr = Vec::with_capacity(b);
|
||
let mut entry_step_arr = Vec::with_capacity(b);
|
||
let mut lots_arr = Vec::with_capacity(b);
|
||
let mut trail_arr = Vec::with_capacity(b);
|
||
let mut active_arr = Vec::with_capacity(b);
|
||
for batch in 0..b {
|
||
let off = batch * 4;
|
||
entry_price_arr.push(&unit_entry_price_host[off..off + 4]);
|
||
entry_step_arr.push(&unit_entry_step_host[off..off + 4]);
|
||
lots_arr.push(&unit_lots_host[off..off + 4]);
|
||
trail_arr.push(&unit_trail_host[off..off + 4]);
|
||
active_arr.push(&unit_active_host[off..off + 4]);
|
||
}
|
||
json!({
|
||
"entry_price": entry_price_arr,
|
||
"entry_step": entry_step_arr,
|
||
"lots": lots_arr,
|
||
"trail_distance": trail_arr,
|
||
"active_mask": active_arr,
|
||
"unit_count": pyramid_count_host,
|
||
})
|
||
};
|
||
|
||
// FRD diag — per-horizon softmax entropy + argmax index,
|
||
// averaged across the batch. Reads from DiagStaging (async,
|
||
// zero GPU stalls). Computes softmax host-side (small: B x 63).
|
||
let frd_diag = {
|
||
let frd_logits = diag_staging.read_frd_logits();
|
||
let mut per_h_entropy = [0.0_f32; FRD_N_HORIZONS];
|
||
let mut per_h_argmax_sum = [0.0_f32; FRD_N_HORIZONS];
|
||
for b in 0..cli.n_backtests {
|
||
for h in 0..FRD_N_HORIZONS {
|
||
let off = b * FRD_OUT_DIM + h * FRD_N_ATOMS;
|
||
let row = &frd_logits[off..off + FRD_N_ATOMS];
|
||
let max_l = row.iter().cloned().fold(f32::NEG_INFINITY, f32::max);
|
||
let denom: f32 = row.iter().map(|x| (x - max_l).exp()).sum();
|
||
let mut entropy = 0.0_f32;
|
||
let mut argmax_idx = 0;
|
||
let mut argmax_v = f32::NEG_INFINITY;
|
||
for (a, x) in row.iter().enumerate() {
|
||
let p = (x - max_l).exp() / denom;
|
||
if p > 1e-9 {
|
||
entropy -= p * p.ln();
|
||
}
|
||
if *x > argmax_v {
|
||
argmax_v = *x;
|
||
argmax_idx = a;
|
||
}
|
||
}
|
||
per_h_entropy[h] += entropy;
|
||
per_h_argmax_sum[h] += argmax_idx as f32;
|
||
}
|
||
}
|
||
let b_f = cli.n_backtests.max(1) as f32;
|
||
json!({
|
||
"h1": { "entropy_mean": per_h_entropy[0] / b_f, "argmax_mean": per_h_argmax_sum[0] / b_f },
|
||
"h2": { "entropy_mean": per_h_entropy[1] / b_f, "argmax_mean": per_h_argmax_sum[1] / b_f },
|
||
"h3": { "entropy_mean": per_h_entropy[2] / b_f, "argmax_mean": per_h_argmax_sum[2] / b_f },
|
||
})
|
||
};
|
||
|
||
let record = json!({
|
||
"step": step,
|
||
"elapsed_s": t_start.elapsed().as_secs_f32(),
|
||
"loss": {
|
||
"bce": stats.l_bce,
|
||
"q": stats.l_q,
|
||
"pi": stats.l_pi,
|
||
"v": stats.l_v,
|
||
"aux": stats.l_aux,
|
||
"frd": stats.l_frd,
|
||
"total": stats.l_total,
|
||
},
|
||
"lambdas": {
|
||
"bce": stats.lambdas.bce,
|
||
"q": stats.lambdas.q,
|
||
"pi": stats.lambdas.pi,
|
||
"v": stats.lambdas.v,
|
||
"aux": stats.lambdas.aux,
|
||
"frd": stats.lambdas.frd,
|
||
},
|
||
// 7 R5 controller outputs (the adaptive knobs).
|
||
"isv_out": {
|
||
"gamma": isv[RL_GAMMA_INDEX],
|
||
"target_tau": isv[RL_TARGET_TAU_INDEX],
|
||
"ppo_clip_eps": isv[RL_PPO_CLIP_INDEX],
|
||
"entropy_coef": isv[RL_ENTROPY_COEF_INDEX],
|
||
"n_rollout_steps": isv[RL_N_ROLLOUT_STEPS_INDEX],
|
||
"per_alpha": isv[RL_PER_ALPHA_INDEX],
|
||
"reward_scale": isv[RL_REWARD_SCALE_INDEX],
|
||
},
|
||
// 5 per-head learning rates (rl_lr_controller emits).
|
||
"isv_lr": {
|
||
"bce": isv[RL_LR_BCE_INDEX],
|
||
"q": isv[RL_LR_Q_INDEX],
|
||
"pi": isv[RL_LR_PI_INDEX],
|
||
"v": isv[RL_LR_V_INDEX],
|
||
"aux": isv[RL_LR_AUX_INDEX],
|
||
},
|
||
// 7 EMA inputs the controllers consume — knowing these
|
||
// makes controller behaviour debuggable (was target wrong?
|
||
// was input wrong? was alpha wrong?).
|
||
"isv_ema_in": {
|
||
"mean_trade_duration": isv[RL_MEAN_TRADE_DURATION_EMA_INDEX],
|
||
"q_divergence": isv[RL_Q_DIVERGENCE_EMA_INDEX],
|
||
"kl_pi": isv[RL_KL_PI_EMA_INDEX],
|
||
"entropy_observed": isv[RL_ENTROPY_OBSERVED_EMA_INDEX],
|
||
"advantage_var_ratio": isv[RL_ADVANTAGE_VAR_RATIO_EMA_INDEX],
|
||
"td_kurtosis": isv[RL_TD_KURTOSIS_EMA_INDEX],
|
||
"mean_abs_pnl": isv[RL_MEAN_ABS_PNL_EMA_INDEX],
|
||
},
|
||
// Grad-norm EMAs (diagnostic — no longer drive the LR
|
||
// controller as of commit 042de99e6 but still useful to
|
||
// correlate with plateau-decay events).
|
||
"grad_norm_ema": {
|
||
"q": isv[RL_Q_GRAD_NORM_EMA_INDEX],
|
||
"pi": isv[RL_PI_GRAD_NORM_EMA_INDEX],
|
||
"v": isv[RL_V_GRAD_NORM_EMA_INDEX],
|
||
},
|
||
// Plateau-decay LR controller state (per head). loss_ema is
|
||
// the controller's internal slow EMA at α=0.05; best is the
|
||
// lowest loss_ema observed so far; steps_since_best is the
|
||
// staleness counter that fires LR halving at PLATEAU_PATIENCE
|
||
// (1000). Confirms whether the controller is seeing plateaus
|
||
// and decaying for the right reason.
|
||
"lr_plateau": {
|
||
"q": { "loss_ema": isv[RL_LR_Q_LOSS_EMA_INDEX],
|
||
"best": isv[RL_LR_Q_BEST_LOSS_INDEX],
|
||
"stale": isv[RL_LR_Q_STEPS_SINCE_BEST_INDEX],
|
||
"warmup": isv[RL_LR_Q_WARMUP_COUNTER_INDEX] },
|
||
"pi": { "loss_ema": isv[RL_LR_PI_LOSS_EMA_INDEX],
|
||
"best": isv[RL_LR_PI_BEST_LOSS_INDEX],
|
||
"stale": isv[RL_LR_PI_STEPS_SINCE_BEST_INDEX],
|
||
"warmup": isv[RL_LR_PI_WARMUP_COUNTER_INDEX] },
|
||
"v": { "loss_ema": isv[RL_LR_V_LOSS_EMA_INDEX],
|
||
"best": isv[RL_LR_V_BEST_LOSS_INDEX],
|
||
"stale": isv[RL_LR_V_STEPS_SINCE_BEST_INDEX],
|
||
"warmup": isv[RL_LR_V_WARMUP_COUNTER_INDEX] },
|
||
},
|
||
"replay_len": trainer.gpu_replay.capacity.min(step + 1),
|
||
// Post-scale, POST-clamp reward stats — what V regression
|
||
// and Q distributional projection actually saw this step.
|
||
// The `scaled_pre_clamp_max` field (below) is the SAME
|
||
// batch's max BEFORE the [-3, +1] clamp fired. If
|
||
// scaled_pre_clamp_max > 1.0 (clamp threshold on the win
|
||
// side) repeatedly, the clamp is doing load-bearing work
|
||
// — i.e. the reward_scale controller is failing to keep
|
||
// typical trade magnitudes within the C51 atom support.
|
||
"rewards": {
|
||
"sum": reward_sum,
|
||
"max": reward_max,
|
||
"min": reward_min,
|
||
"abs_max": reward_abs_max,
|
||
"scaled_pre_clamp_max":
|
||
isv[RL_MAX_ABS_SCALED_REWARD_PRE_CLAMP_INDEX],
|
||
// Adaptive clamp signals — surface the controller's
|
||
// POINT input + EMA + emitted bounds so post-hoc
|
||
// analysis can verify the adaptation chain is moving.
|
||
"pos_scaled_max":
|
||
isv[RL_POS_SCALED_REWARD_MAX_INDEX],
|
||
"pos_scaled_max_ema":
|
||
isv[RL_POS_SCALED_REWARD_MAX_EMA_INDEX],
|
||
// MARGIN controller state — clip rate vs target tells
|
||
// us whether the controller is happy (rate ≈ target)
|
||
// or fighting the tail (rate > target × tolerance).
|
||
"clip_rate_ema":
|
||
isv[RL_REWARD_CLAMP_CLIP_RATE_EMA_INDEX],
|
||
// C51 atom span — ratcheted from WIN/LOSS clamp by the
|
||
// same controller. Watching these expand confirms Q's
|
||
// distributional learning ceiling is being lifted.
|
||
"c51_v_max":
|
||
isv[RL_C51_V_MAX_INDEX],
|
||
"c51_v_min":
|
||
isv[RL_C51_V_MIN_INDEX],
|
||
// Q→π distillation KL signal — drops toward 0 if π is
|
||
// successfully chasing Q's preferences.
|
||
"q_distill_kl_ema":
|
||
isv[RL_Q_DISTILL_KL_EMA_INDEX],
|
||
// Negative-tail tracking — input to adaptive RATIO.
|
||
"neg_scaled_max":
|
||
isv[RL_NEG_SCALED_REWARD_MAX_INDEX],
|
||
"neg_scaled_max_ema":
|
||
isv[RL_NEG_SCALED_REWARD_MAX_EMA_INDEX],
|
||
},
|
||
// audit — PPO importance-ratio clamp diagnostics.
|
||
//
|
||
// ratio_clamp_max : the adaptive ceiling at ISV[440],
|
||
// emitted by rl_ppo_ratio_clamp_controller
|
||
// from (1+ε) × PPO_CLAMP_MARGIN.
|
||
// log_ratio_abs_max : per-step max(|log π_new − log π_old|)
|
||
// over the batch at ISV[441], written
|
||
// by ppo_log_ratio_abs_max_b.
|
||
//
|
||
// The clamp fires when log_ratio_abs_max > ln(ratio_clamp_max).
|
||
// For ratio_clamp_max = 10, ln = 2.30. Healthy training has
|
||
// log_ratio_abs_max well below this most steps; outliers
|
||
// touch or exceed it on rare excursions which the clamp
|
||
// bounds before they pollute l_pi.
|
||
"ppo": {
|
||
"ratio_clamp_max": isv[RL_PPO_RATIO_CLAMP_MAX_INDEX],
|
||
"log_ratio_abs_max": isv[RL_PPO_LOG_RATIO_ABS_MAX_INDEX],
|
||
},
|
||
// audit — streaming-kernel state for the variance-ratio +
|
||
// kurtosis estimators that feed rollout_steps and per_α.
|
||
// * `*_stream.{mean, m2, m4}` are the Welford-EMA
|
||
// internal state slots (442-446) — confirms the
|
||
// kernel is folding observations across STEPS.
|
||
// * `*_clamp` is the ISV-resident output ceiling
|
||
// (447, 448) seeded once at trainer init by
|
||
// rl_streaming_clamp_init — visible here so the diag
|
||
// can spot when the clamp is firing
|
||
// (post-clamp EMA at the consumer-input slot equals
|
||
// `clamp` ↔ clamp active).
|
||
"streaming": {
|
||
"adv_var": {
|
||
"mean": isv[RL_ADV_VAR_STREAM_MEAN_INDEX],
|
||
"m2": isv[RL_ADV_VAR_STREAM_M2_INDEX],
|
||
"clamp": isv[RL_ADV_VAR_RATIO_CLAMP_INDEX],
|
||
},
|
||
"td_kurt": {
|
||
"mean": isv[RL_TD_KURT_STREAM_MEAN_INDEX],
|
||
"m2": isv[RL_TD_KURT_STREAM_M2_INDEX],
|
||
"m4": isv[RL_TD_KURT_STREAM_M4_INDEX],
|
||
"clamp": isv[RL_TD_KURTOSIS_CLAMP_INDEX],
|
||
},
|
||
},
|
||
// audit — controller decision diagnostics (which adaptation
|
||
// branch the multiplicative controllers fired this step).
|
||
//
|
||
// ratio = input_ema / TARGET (per controller's design)
|
||
// branch = "WIDEN" if ratio > TOLERANCE = 1.5
|
||
// "HOLD" if ratio ∈ [1/1.5, 1.5]
|
||
// "SHRINK" if ratio < 1/1.5
|
||
// "NOISE" if input_ema is below the noise floor
|
||
//
|
||
// For rl_rollout_steps_controller (TARGET = 0.1) and
|
||
// rl_ppo_clip_controller (TARGET = 0.01), this reveals
|
||
// whether the controller is being driven by real divergence
|
||
// signal or sitting in the in-band hold zone. Use to spot
|
||
// controllers stuck saturating at MAX/MIN (which would
|
||
// show 100% WIDEN or 100% SHRINK).
|
||
//
|
||
// Targets and tolerances are reflected from the kernel
|
||
// #defines (kept synchronised by code review at the
|
||
// controller-cu file level — there's no ISV slot for these
|
||
// design constants because they're fundamental to the
|
||
// controller's behaviour, not adaptive).
|
||
// audit — K-loop training-intensification iterations performed
|
||
// this step. K = clamp(isv[404] / isv[450], 1, isv[451]).
|
||
// Both divisor + max are ISV-driven (per
|
||
// feedback_isv_for_adaptive_bounds) — surfaced here so
|
||
// post-hoc analysis can compute K from any sample without
|
||
// recompile. Defaults: divisor=2048 (K=1 at controller
|
||
// bootstrap), max=4 (prevents gradient overtraining).
|
||
"k_loop": {
|
||
"k_updates": trainer.last_k_updates,
|
||
"divisor": isv[RL_K_LOOP_DIVISOR_INDEX],
|
||
"max": isv[RL_K_LOOP_MAX_INDEX],
|
||
},
|
||
// audit — ISV-driven design constants (was kernel `#define`s).
|
||
// All tunable at runtime via re-launching rl_isv_write per
|
||
// `feedback_isv_for_adaptive_bounds`. Surfaced here so
|
||
// post-hoc analysis sees exactly what each controller was
|
||
// regressing against / clamping at, every step.
|
||
"isv_config": {
|
||
"reward_clamp_win": isv[RL_REWARD_CLAMP_WIN_INDEX],
|
||
"reward_clamp_loss": isv[RL_REWARD_CLAMP_LOSS_INDEX],
|
||
"kl_target": isv[RL_KL_TARGET_INDEX],
|
||
"improvement_threshold": isv[RL_IMPROVEMENT_THRESHOLD_INDEX],
|
||
"plateau_patience": isv[RL_PLATEAU_PATIENCE_INDEX],
|
||
"div_target": isv[RL_DIV_TARGET_INDEX],
|
||
"entropy_target_frac": isv[RL_ENTROPY_TARGET_FRAC_INDEX],
|
||
"kurt_lift_scale": isv[RL_KURT_LIFT_SCALE_INDEX],
|
||
"ppo_clamp_margin": isv[RL_PPO_CLAMP_MARGIN_INDEX],
|
||
"lr_warmup_steps": isv[RL_LR_WARMUP_STEPS_INDEX],
|
||
"lr_bootstrap": isv[RL_LR_BOOTSTRAP_INDEX],
|
||
"lr_min": isv[RL_LR_MIN_INDEX],
|
||
"lr_max": isv[RL_LR_MAX_INDEX],
|
||
"lr_loss_ema_alpha": isv[RL_LR_LOSS_EMA_ALPHA_INDEX],
|
||
"lr_decay_factor": isv[RL_LR_DECAY_FACTOR_INDEX],
|
||
"loss_lambda_aux": isv[RL_LOSS_LAMBDA_AUX_INDEX],
|
||
"schulman_tolerance": isv[RL_SCHULMAN_TOLERANCE_INDEX],
|
||
"schulman_adjust_rate": isv[RL_SCHULMAN_ADJUST_RATE_INDEX],
|
||
"stream_alpha": isv[RL_STREAM_ALPHA_INDEX],
|
||
"kurt_gaussian": isv[RL_KURT_GAUSSIAN_INDEX],
|
||
"kurt_noise_floor": isv[RL_KURT_NOISE_FLOOR_INDEX],
|
||
"tau_bootstrap": isv[RL_TAU_BOOTSTRAP_INDEX],
|
||
"eps_bootstrap": isv[RL_EPS_BOOTSTRAP_INDEX],
|
||
"rollout_bootstrap": isv[RL_ROLLOUT_BOOTSTRAP_INDEX],
|
||
"reward_scale_bootstrap":isv[RL_REWARD_SCALE_BOOTSTRAP_INDEX],
|
||
"ppo_ratio_clamp_bootstrap": isv[RL_PPO_RATIO_CLAMP_BOOTSTRAP_INDEX],
|
||
"reward_clamp_margin": isv[RL_REWARD_CLAMP_MARGIN_INDEX],
|
||
"reward_clamp_ratio": isv[RL_REWARD_CLAMP_RATIO_INDEX],
|
||
"reward_clamp_clip_rate_target": isv[RL_REWARD_CLAMP_CLIP_RATE_TARGET_INDEX],
|
||
"q_distill_lambda": isv[RL_Q_DISTILL_LAMBDA_INDEX],
|
||
"q_distill_temperature": isv[RL_Q_DISTILL_TEMPERATURE_INDEX],
|
||
"q_distill_kl_target": isv[RL_Q_DISTILL_KL_TARGET_INDEX],
|
||
"sac_alpha": isv[RL_SAC_ALPHA_INDEX],
|
||
"sac_entropy_target": isv[RL_SAC_ENTROPY_TARGET_INDEX],
|
||
"action_entropy_ema": isv[RL_ACTION_ENTROPY_EMA_INDEX],
|
||
"reward_scale_min": isv[RL_REWARD_SCALE_MIN_INDEX],
|
||
},
|
||
// audit — Q vs π action agreement EMA at slot 407
|
||
// (previously dead). 1.0 = perfect ranking consistency,
|
||
// 0.0 = total disagreement. Alarm if persistently < 0.5.
|
||
"q_pi_agree_ema": isv[RL_Q_ARG_VS_PI_AGREE_INDEX],
|
||
"controller_branch": {
|
||
"rollout_steps_input": isv[RL_ADVANTAGE_VAR_RATIO_EMA_INDEX],
|
||
// ISV-driven target — seeded by rl_streaming_clamp_init,
|
||
// visible here (no longer a hardcoded `#define`).
|
||
"rollout_steps_target": isv[RL_ADV_VAR_RATIO_TARGET_INDEX],
|
||
"ppo_clip_input": isv[RL_KL_PI_EMA_INDEX],
|
||
"ppo_clip_target": 0.01f32,
|
||
"target_tau_input": isv[RL_Q_DIVERGENCE_EMA_INDEX],
|
||
"target_tau_target": 0.01f32,
|
||
"per_alpha_input": isv[RL_TD_KURTOSIS_EMA_INDEX],
|
||
"per_alpha_target": 0.6f32, // kurt=10 fixed point
|
||
},
|
||
"done_count": done_count,
|
||
// Action histogram: indices match `Action` enum
|
||
// (0=ShortLarge, 1=ShortSmall, 2=Hold, 3=FlatFromLong,
|
||
// 4=FlatFromShort, 5=LongSmall, 6=LongLarge,
|
||
// 7=TrailTighten, 8=TrailLoosen).
|
||
"action_hist": act_hist.to_vec(),
|
||
// audit additions:
|
||
// * action_entropy — H(action_dist) computed from
|
||
// act_hist; uniform-9 = ln(9) ≈ 2.197. Lower = policy
|
||
// collapsed to few actions (exploration failed).
|
||
// * position.lots — signed lot count per batch (long/short
|
||
// /flat). Reveals whether the agent is actually moving
|
||
// positions or stuck.
|
||
"action_entropy": action_entropy,
|
||
"position": {
|
||
"lots": position_lots_host,
|
||
},
|
||
"units": units_diag,
|
||
"trail": {
|
||
"fired_count_step": trail_fired_step,
|
||
"fired_count_total": trail_fired_total,
|
||
"tightened_count_step": tighten_step,
|
||
"loosened_count_step": loosen_step,
|
||
"tightened_count_total": trail_tighten_total,
|
||
"loosened_count_total": trail_loosen_total,
|
||
},
|
||
"pyramid": {
|
||
"added_count_step": pyramid_add_step,
|
||
"added_count_total": pyramid_added_total,
|
||
"units_distribution": pyramid_count_host,
|
||
"max_units_reached": pyramid_count_host.iter().any(|&c| c >= 4),
|
||
},
|
||
"partial_flat": {
|
||
"fired_count_step": partial_flat_step,
|
||
"fired_count_total": partial_flat_total,
|
||
"long_count_total": partial_flat_long_total,
|
||
"short_count_total": partial_flat_short_total,
|
||
"close_unit_index": close_unit_index_host,
|
||
},
|
||
"confidence_gate": {
|
||
"gated_count_step": conf_gate_step,
|
||
"gated_count_total": conf_gate_total,
|
||
},
|
||
"position_heat": {
|
||
"capped_count_step": heat_cap_step,
|
||
"capped_count_total": heat_cap_total,
|
||
"heat_max_lots": isv[ml_alpha::rl::isv_slots::RL_HEAT_CAP_MAX_LOTS_INDEX],
|
||
},
|
||
"anti_martingale": {
|
||
"outcome_ema": outcome_ema_host,
|
||
"kappa": isv[ml_alpha::rl::isv_slots::RL_ANTIMARTINGALE_KAPPA_INDEX],
|
||
},
|
||
"frd_gate": {
|
||
"gated_count_step": frd_gate_step,
|
||
"gated_count_total": frd_gate_total,
|
||
},
|
||
"trading": {
|
||
"pnl_cum_usd": pnl_cum_usd,
|
||
"total_trades": total_trades,
|
||
"win_rate": win_rate,
|
||
"avg_hold_steps": avg_hold,
|
||
"raw_reward_sum": raw_rewards_host.iter().sum::<f32>(),
|
||
},
|
||
// SP20 P3 FRD head diag — per-horizon softmax entropy + argmax-mode
|
||
// bucket index (averaged across the batch). At init, Xavier × 0.1
|
||
// weights → logits ≈ 0 → entropy ≈ ln(21) = 3.044 and the argmax
|
||
// is dominated by tiny initial-weight noise. Watching these drift
|
||
// away from the uniform baseline is the supervised-learning signal
|
||
// for the FRD head (CE loss + bwd land in F.3).
|
||
"frd": frd_diag,
|
||
// P1+P2 feature metrics — PopArt reward normalization,
|
||
// spectral norm/decoupling, Q-bias correction, per-branch
|
||
// LR scaling, and outcome aux head lambda.
|
||
"popart": {
|
||
"mean": isv[RL_POPART_MEAN_INDEX],
|
||
"sigma": isv[RL_POPART_SIGMA_INDEX],
|
||
"var": isv[RL_POPART_VAR_INDEX],
|
||
},
|
||
"spectral": {
|
||
"norm_max_config": isv[RL_SPECTRAL_NORM_MAX_INDEX],
|
||
"decouple_lambda": isv[RL_SPECTRAL_DECOUPLE_LAMBDA_INDEX],
|
||
},
|
||
"q_bias": {
|
||
"ema": isv[RL_Q_BIAS_EMA_INDEX],
|
||
"correction": isv[RL_Q_BIAS_CORRECTION_INDEX],
|
||
},
|
||
"per_branch_lr": {
|
||
"scale_q": isv[RL_LR_SCALE_Q_INDEX],
|
||
"scale_pi": isv[RL_LR_SCALE_PI_INDEX],
|
||
"scale_v": isv[RL_LR_SCALE_V_INDEX],
|
||
"scale_iqn": isv[RL_LR_SCALE_IQN_INDEX],
|
||
},
|
||
"outcome_aux": {
|
||
"lambda": isv[RL_OUTCOME_AUX_LAMBDA_INDEX],
|
||
},
|
||
});
|
||
writeln!(diag, "{}", record).context("diag: writeln jsonl")?;
|
||
|
||
if cli.checkpoint_every > 0 && step > 0 && step % cli.checkpoint_every == 0 {
|
||
let ckpt_path = cli.out.join(format!("checkpoint-{step}.bin"));
|
||
trainer.save_checkpoint(&ckpt_path, step as u64)
|
||
.with_context(|| format!("checkpoint at step {step}"))?;
|
||
eprintln!("checkpoint saved: {}", ckpt_path.display());
|
||
let old_step = step.saturating_sub(cli.checkpoint_every * 2);
|
||
if old_step > 0 {
|
||
let old_path = cli.out.join(format!("checkpoint-{old_step}.bin"));
|
||
let _ = std::fs::remove_file(&old_path);
|
||
}
|
||
}
|
||
|
||
if step % cli.log_every == 0 || step == cli.n_steps - 1 {
|
||
diag.flush().context("diag: flush")?;
|
||
let elapsed = t_start.elapsed().as_secs_f32();
|
||
let sps = if elapsed > 0.0 { (step + 1) as f32 / elapsed } else { 0.0 };
|
||
eprintln!(
|
||
"step {:>6}/{}: l_q={:.4} l_pi={:.4} l_v={:.4} l_total={:.4} \
|
||
γ={:.4} ε={:.4} per_α={:.4} scale={:.4} \
|
||
replay={} dones={} rew_sum={:.3} sps={:.0} pnl=${:.0} wr={:.2} elapsed={:.1}s",
|
||
step,
|
||
cli.n_steps,
|
||
stats.l_q,
|
||
stats.l_pi,
|
||
stats.l_v,
|
||
stats.l_total,
|
||
isv[RL_GAMMA_INDEX],
|
||
isv[RL_PPO_CLIP_INDEX],
|
||
isv[RL_PER_ALPHA_INDEX],
|
||
isv[RL_REWARD_SCALE_INDEX],
|
||
trainer.gpu_replay.capacity.min(step + 1),
|
||
done_count,
|
||
reward_sum,
|
||
sps,
|
||
pnl_cum_usd,
|
||
win_rate,
|
||
elapsed,
|
||
);
|
||
}
|
||
last_stats = Some(stats);
|
||
summary.n_steps_completed = step + 1;
|
||
}
|
||
diag.flush().context("diag: final flush")?;
|
||
|
||
// ── Eval phase (walk-forward G8). ────────────────────────────────
|
||
// After the train phase, run n_eval_steps additional step_with_lobsim
|
||
// calls on the held-out eval files. Trade records accumulated by
|
||
// LobSim during eval are isolated from train-phase records via the
|
||
// `head_before_eval` checkpoint — eval records are sliced as
|
||
// `all_records[head_before_eval..]` post-eval.
|
||
//
|
||
// Note: this is NOT pure-eval mode — step_with_lobsim still runs
|
||
// its learning machinery (Adam steps, PER updates). At b_size=1
|
||
// the per-step learning effect is small relative to the policy
|
||
// already accumulated in train; the eval profit_factor approximates
|
||
// the train-end policy's OOS performance. True pure-eval (forward
|
||
// only, no backward) is a follow-up architectural change.
|
||
if cli.n_eval_steps > 0 && !eval_files.is_empty() {
|
||
let head_before_eval = sim
|
||
.read_total_trade_count()
|
||
.context("read trade count pre-eval")?;
|
||
eprintln!(
|
||
"── eval phase: {} steps on {} held-out files (trade-record checkpoint head={}) ──",
|
||
cli.n_eval_steps, eval_files.len(), head_before_eval
|
||
);
|
||
|
||
let eval_loader_cfg = MultiHorizonLoaderConfig {
|
||
files: eval_files.clone(),
|
||
predecoded_dir: cli.predecoded_dir.clone(),
|
||
multi_resolution,
|
||
horizons: HORIZONS,
|
||
n_max_sequences: cli.n_eval_steps
|
||
.saturating_mul(cli.n_backtests.max(1))
|
||
.saturating_mul(2)
|
||
.max(64),
|
||
// Different seed for eval so sequence sampling is independent
|
||
// of the train seed (deterministic given the eval seed).
|
||
seed: cli.seed.wrapping_add(0xE7AE),
|
||
inference_only: false,
|
||
outcome_label_cost: cli.outcome_label_cost,
|
||
instrument_filter: cli.instrument_mode,
|
||
frd_horizon_ticks: ml_alpha::rl::common::FRD_HORIZON_TICKS,
|
||
frd_bucket_range_sigma: ml_alpha::rl::common::FRD_BUCKET_RANGE_SIGMA,
|
||
};
|
||
let eval_loader = MultiHorizonLoader::new(&eval_loader_cfg)
|
||
.context("MultiHorizonLoader::new (eval)")?;
|
||
|
||
// Upload eval dataset to GPU (separate from train dataset).
|
||
let eval_gpu_dataset = eval_loader
|
||
.upload_to_gpu(&dev, cli.n_backtests.max(1), cli.seed.wrapping_add(0xE7AE))
|
||
.context("eval loader.upload_to_gpu")?;
|
||
let mut eval_gpu_loader = GpuDataLoader::new(&dev, cli.n_backtests.max(1))
|
||
.context("GpuDataLoader::new (eval)")?;
|
||
|
||
for eval_step in 0..cli.n_eval_steps {
|
||
let stats = trainer
|
||
.step_with_lobsim_gpu(&mut eval_gpu_loader, &eval_gpu_dataset, &mut sim)
|
||
.with_context(|| format!("step_with_lobsim_gpu eval step {eval_step}"))?;
|
||
if !stats.l_total.is_finite() {
|
||
eprintln!("G8 NaN ABORT during eval at step {eval_step}");
|
||
process::exit(2);
|
||
}
|
||
if eval_step % cli.log_every == 0 || eval_step == cli.n_eval_steps - 1 {
|
||
eprintln!(
|
||
"eval {:>5}/{}: l_total={:.4} elapsed={:.1}s",
|
||
eval_step, cli.n_eval_steps, stats.l_total,
|
||
t_start.elapsed().as_secs_f32()
|
||
);
|
||
}
|
||
}
|
||
|
||
// Drain trade records, slice to eval-only, compute summary.
|
||
let all_records = sim
|
||
.read_trade_records(0)
|
||
.context("read trade records post-eval")?;
|
||
let head_before_usize = head_before_eval as usize;
|
||
let eval_records: Vec<_> = if all_records.len() > head_before_usize {
|
||
all_records[head_before_usize..].to_vec()
|
||
} else {
|
||
// Trade log wrapped past TRADE_LOG_CAP — use what we have.
|
||
// For smoke (b_size=1, ≤200 trades) this branch never fires.
|
||
eprintln!(
|
||
"warning: trade log wrapped — head_before={} but only {} records readable",
|
||
head_before_usize, all_records.len()
|
||
);
|
||
all_records
|
||
};
|
||
|
||
// Synthesise a pnl_curve from per-trade cumulative PnL for
|
||
// compute_summary's max_drawdown calc.
|
||
let mut pnl_curve = Vec::with_capacity(eval_records.len());
|
||
let mut cum = 0.0f32;
|
||
for r in &eval_records {
|
||
cum += (r.realised_pnl_usd_fp as f32) / 100.0;
|
||
pnl_curve.push(cum);
|
||
}
|
||
let eval_summary =
|
||
ml_backtesting::artifacts::compute_summary(&eval_records, &pnl_curve);
|
||
|
||
eprintln!(
|
||
"eval summary: n_trades={} pnl_usd={:.2} pf={:.3} sharpe_ann={:.3} max_dd_usd={:.2} win_rate={:.3}",
|
||
eval_summary.n_trades,
|
||
eval_summary.total_pnl_usd,
|
||
eval_summary.profit_factor,
|
||
eval_summary.sharpe_ann,
|
||
eval_summary.max_drawdown_usd,
|
||
eval_summary.win_rate,
|
||
);
|
||
|
||
let eval_summary_path = cli.out.join("eval_summary.json");
|
||
let f = std::fs::File::create(&eval_summary_path)
|
||
.with_context(|| format!("create {}", eval_summary_path.display()))?;
|
||
serde_json::to_writer_pretty(f, &eval_summary)
|
||
.with_context(|| format!("write {}", eval_summary_path.display()))?;
|
||
eprintln!("eval summary written: {}", eval_summary_path.display());
|
||
}
|
||
|
||
// ── Compose + emit summary. ──────────────────────────────────────
|
||
if let Some(s) = last_stats {
|
||
summary.final_l_bce = s.l_bce;
|
||
summary.final_l_q = s.l_q;
|
||
summary.final_l_pi = s.l_pi;
|
||
summary.final_l_v = s.l_v;
|
||
summary.final_l_aux = s.l_aux;
|
||
summary.final_l_total = s.l_total;
|
||
}
|
||
summary.final_replay_len = trainer.gpu_replay.capacity.min(cli.n_steps);
|
||
summary.completed_clean = summary.n_steps_completed == summary.n_steps_planned;
|
||
write_summary(&cli.out, &summary)?;
|
||
|
||
eprintln!(
|
||
"alpha_rl_train complete: {} / {} steps in {:.1}s (replay len = {})",
|
||
summary.n_steps_completed,
|
||
summary.n_steps_planned,
|
||
t_start.elapsed().as_secs_f32(),
|
||
summary.final_replay_len
|
||
);
|
||
Ok(())
|
||
}
|
||
|
||
fn write_summary(out_dir: &std::path::Path, summary: &AlphaRlTrainSummary) -> Result<()> {
|
||
let path = out_dir.join("alpha_rl_train_summary.json");
|
||
let f = std::fs::File::create(&path)
|
||
.with_context(|| format!("create {}", path.display()))?;
|
||
serde_json::to_writer_pretty(f, summary)
|
||
.with_context(|| format!("write {}", path.display()))?;
|
||
eprintln!("summary written: {}", path.display());
|
||
Ok(())
|
||
}
|