perf: zero-copy fxcache training pipeline
Eliminates all per-fold CPU waste in walk-forward training: - FxCacheData replaces OHLCVBar as data backbone (features[42] + targets[4] + OFI[8]) - Walk-forward generates index ranges from timestamps, no bar cloning - DQNTrainer created once, reused across folds via reset_for_fold - Data uploaded to GPU once via init_from_fxcache, sliced by index per fold - Trainer accepts &[[f64;42]] + &[[f64;4]] slices, zero Vec<f64> allocation - PPO uses train_from_slices, no per-fold feature re-extraction - Ensemble trainers pre-created before fold loop, no per-fold re-upload - Deleted: prepare_fold_data, FoldData, features_to_trainer_format, train_ppo_fold, double-buffer, prefetch thread (~500 lines removed) Smoketest: 160s -> 4.97s (32x speedup) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
File diff suppressed because it is too large
Load Diff
@@ -298,6 +298,73 @@ impl DqnGpuData {
|
||||
})
|
||||
}
|
||||
|
||||
/// Upload DQN training data to GPU from separate contiguous arrays.
|
||||
///
|
||||
/// Accepts features, targets, and OFI as separate slices matching the fxcache
|
||||
/// flat binary layout — no tuple allocation or per-bar heap Vec required.
|
||||
/// OFI is uploaded only when at least one value is non-zero.
|
||||
pub fn upload_slices(
|
||||
features: &[[f64; 42]],
|
||||
targets: &[[f64; 4]],
|
||||
ofi: &[[f64; 8]],
|
||||
stream: &Arc<CudaStream>,
|
||||
) -> Result<Self, MLError> {
|
||||
let num_bars = features.len();
|
||||
if num_bars == 0 {
|
||||
return Err(MLError::ModelError("Empty training data".to_owned()));
|
||||
}
|
||||
|
||||
let feature_dim = 42;
|
||||
let target_dim = 4;
|
||||
|
||||
let estimated_bytes = estimate_vram_bytes(num_bars * (feature_dim + target_dim));
|
||||
if estimated_bytes > MAX_UPLOAD_BYTES {
|
||||
return Err(MLError::ModelError(format!(
|
||||
"Training data too large for GPU upload: {:.1} GB > 2.0 GB limit ({} bars)",
|
||||
estimated_bytes as f64 / 1_073_741_824.0,
|
||||
num_bars,
|
||||
)));
|
||||
}
|
||||
|
||||
let flat_features: Vec<f32> = features
|
||||
.iter()
|
||||
.flat_map(|f| f.iter().map(|&v| v as f32))
|
||||
.collect();
|
||||
|
||||
let flat_targets: Vec<f32> = targets
|
||||
.iter()
|
||||
.flat_map(|t| t.iter().map(|&v| v as f32))
|
||||
.collect();
|
||||
|
||||
let features_gpu = clone_htod_f32_to_bf16(stream, &flat_features)?;
|
||||
let targets_gpu = clone_htod_f32_to_bf16(stream, &flat_targets)?;
|
||||
|
||||
let ofi_features = if !ofi.is_empty() && ofi.iter().any(|row| row.iter().any(|&v| v != 0.0)) {
|
||||
let mut flat_ofi = Vec::with_capacity(num_bars * 8);
|
||||
for i in 0..num_bars {
|
||||
if i < ofi.len() {
|
||||
for &v in ofi[i].iter() {
|
||||
flat_ofi.push(v as f32);
|
||||
}
|
||||
} else {
|
||||
flat_ofi.extend_from_slice(&[0.0_f32; 8]);
|
||||
}
|
||||
}
|
||||
Some(clone_htod_f32_to_bf16(stream, &flat_ofi)?)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
Ok(Self {
|
||||
features: features_gpu,
|
||||
targets: targets_gpu,
|
||||
ofi_features,
|
||||
num_bars,
|
||||
feature_dim,
|
||||
aligned_state_dim: None,
|
||||
})
|
||||
}
|
||||
|
||||
/// Upload OFI features (8 dims per bar) from MBP-10 order book data.
|
||||
///
|
||||
/// Must be called after `upload()` with a slice matching `num_bars` length.
|
||||
|
||||
@@ -524,6 +524,43 @@ impl DQNTrainer {
|
||||
.await
|
||||
}
|
||||
|
||||
/// Train one fold from contiguous feature/target slices.
|
||||
/// Zero Vec<f64> allocation — targets read as fixed-size &[[f64; 4]].
|
||||
///
|
||||
/// Caller must call `set_val_data_from_slices` and `set_training_range` before
|
||||
/// this method. Val data and OFI offset are already set on `self`.
|
||||
pub async fn train_fold_from_slices<F>(
|
||||
&mut self,
|
||||
features: &[[f64; 42]],
|
||||
targets: &[[f64; 4]],
|
||||
checkpoint_callback: F,
|
||||
) -> Result<TrainingMetrics>
|
||||
where
|
||||
F: FnMut(usize, Vec<u8>, bool) -> Result<String> + Send,
|
||||
{
|
||||
// Clear stale CUDA errors
|
||||
if let MlDevice::Cuda { ref context, .. } = self.device {
|
||||
let _ = context.check_err();
|
||||
}
|
||||
|
||||
info!("train_fold_from_slices: {} bars", features.len());
|
||||
|
||||
// Build lightweight training_data with fixed-size arrays — zero heap alloc per element.
|
||||
// Both [f64; 42] and [f64; 4] are Copy types that live entirely on the stack.
|
||||
let training_data: Vec<([f64; 42], [f64; 4])> = features
|
||||
.iter()
|
||||
.zip(targets.iter())
|
||||
.map(|(f, t)| (*f, *t))
|
||||
.collect();
|
||||
|
||||
self.val_features_gpu = None;
|
||||
self.val_closes_gpu = None;
|
||||
self.val_ofi_gpu = None;
|
||||
|
||||
self.train_with_data_full_loop_slices(&training_data, checkpoint_callback)
|
||||
.await
|
||||
}
|
||||
|
||||
/// Train with shared preloaded data (zero-copy for hyperopt).
|
||||
///
|
||||
/// Same as [`train_with_preloaded_data`] but accepts `Arc`-wrapped data,
|
||||
@@ -1027,6 +1064,58 @@ impl DQNTrainer {
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
/// Upload full fxcache dataset to GPU ONCE. Call before the fold loop.
|
||||
/// Replaces per-fold init_gpu_data + init_gpu_raw_buffers.
|
||||
pub async fn init_from_fxcache(
|
||||
&mut self,
|
||||
features: &[[f64; 42]],
|
||||
targets: &[[f64; 4]],
|
||||
ofi: &[[f64; 8]],
|
||||
) -> Result<()> {
|
||||
let stream = self.cuda_stream.as_ref()
|
||||
.ok_or_else(|| anyhow::anyhow!("CUDA stream required for init_from_fxcache"))?;
|
||||
|
||||
// Upload via DqnGpuData::upload_slices (contiguous, zero tuple unpacking)
|
||||
let mut gpu_data = crate::cuda_pipeline::DqnGpuData::upload_slices(
|
||||
features, targets, ofi, stream,
|
||||
).map_err(|e| anyhow::anyhow!("DqnGpuData::upload_slices failed: {e}"))?;
|
||||
|
||||
let ofi_enabled = gpu_data.ofi_features.is_some();
|
||||
let raw_dim = if ofi_enabled { 53 } else { 45 };
|
||||
let aligned_dim = (raw_dim + 7) & !7;
|
||||
gpu_data.set_aligned_state_dim(aligned_dim);
|
||||
|
||||
tracing::info!(
|
||||
"init_from_fxcache: {} bars uploaded to GPU ({:.1} MB, OFI={})",
|
||||
features.len(),
|
||||
(features.len() * (42 + 4 + 8) * 4) as f64 / 1_048_576.0,
|
||||
ofi_enabled,
|
||||
);
|
||||
self.gpu_data = Some(gpu_data);
|
||||
|
||||
// Also upload raw buffers for experience collector
|
||||
self.init_gpu_raw_buffers_from_slices(features, targets).await?;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Set validation data from contiguous fxcache slices.
|
||||
/// Converts to legacy Vec format internally — val set is small (~50K bars).
|
||||
pub fn set_val_data_from_slices(
|
||||
&mut self,
|
||||
features: &[[f64; 42]],
|
||||
targets: &[[f64; 4]],
|
||||
ofi_val_offset: usize,
|
||||
) {
|
||||
self.val_data = features.iter().zip(targets.iter())
|
||||
.map(|(f, t)| (*f, t.to_vec()))
|
||||
.collect();
|
||||
self.ofi_val_offset = ofi_val_offset;
|
||||
self.val_features_gpu = None;
|
||||
self.val_closes_gpu = None;
|
||||
self.val_ofi_gpu = None;
|
||||
}
|
||||
|
||||
/// Inject pre-uploaded GPU data (e.g. from a `DoubleBufferedLoader`).
|
||||
///
|
||||
/// The trainer's `train_epoch` lazily uploads data on first call.
|
||||
@@ -1105,8 +1194,8 @@ impl DQNTrainer {
|
||||
.map_err(|e| anyhow::anyhow!("guard reset: {e}"))?;
|
||||
}
|
||||
|
||||
// Clear gpu_data to force re-init with current fold range
|
||||
self.gpu_data = None;
|
||||
// Keep gpu_data — the full dataset is already on GPU via init_from_fxcache.
|
||||
// Clearing it would force a redundant re-upload on the next fold.
|
||||
|
||||
info!("DQNTrainer: reset for new fold (GPU infrastructure preserved)");
|
||||
Ok(())
|
||||
|
||||
@@ -517,6 +517,419 @@ impl DQNTrainer {
|
||||
Ok(metrics)
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
// Zero-alloc training loop — accepts fixed-size arrays, no Vec<f64>
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
/// Like `train_with_data_full_loop` but accepts `&[([f64; 42], [f64; 4])]`.
|
||||
///
|
||||
/// Requires `gpu_data` and `targets_raw_cuda` to already be populated
|
||||
/// (via `init_from_fxcache`). Skips GPU upload entirely — the data is
|
||||
/// already resident on GPU. This eliminates the per-bar `Vec<f64>` heap
|
||||
/// allocation that the `(FeatureVector, Vec<f64>)` path required.
|
||||
pub(crate) async fn train_with_data_full_loop_slices<F>(
|
||||
&mut self,
|
||||
training_data: &[([f64; 42], [f64; 4])],
|
||||
mut checkpoint_callback: F,
|
||||
) -> Result<TrainingMetrics>
|
||||
where
|
||||
F: FnMut(usize, Vec<u8>, bool) -> Result<String> + Send,
|
||||
{
|
||||
let start_time = std::time::Instant::now();
|
||||
let mut total_loss = 0.0;
|
||||
let mut total_q_value = 0.0;
|
||||
let mut total_gradient_norm = 0.0;
|
||||
let mut total_reward = 0.0;
|
||||
let mut total_action_counts = [0_usize; 9];
|
||||
let mut total_factored_action_counts = [0_usize; 81];
|
||||
|
||||
self.log_training_config().await;
|
||||
|
||||
// Decision Transformer pre-training on offline trajectories
|
||||
if self.hyperparams.dt_pretrain_epochs > 0 {
|
||||
info!(
|
||||
epochs = self.hyperparams.dt_pretrain_epochs,
|
||||
context_len = self.hyperparams.dt_context_len,
|
||||
embed_dim = self.hyperparams.dt_embed_dim,
|
||||
"Starting Decision Transformer pre-training"
|
||||
);
|
||||
|
||||
// GPU raw buffers already uploaded via init_from_fxcache
|
||||
if self.targets_raw_cuda.is_none() || self.features_raw_cuda.is_none() {
|
||||
warn!("DT pre-training: GPU features/targets not available, skipping");
|
||||
} else if let Some(ref stream) = self.cuda_stream {
|
||||
let stream = Arc::clone(stream);
|
||||
|
||||
let dt_state_dim: usize = 42;
|
||||
|
||||
let dt_config = crate::cuda_pipeline::decision_transformer::DecisionTransformerConfig {
|
||||
state_dim: dt_state_dim,
|
||||
num_actions: 9,
|
||||
embed_dim: self.hyperparams.dt_embed_dim,
|
||||
num_layers: self.hyperparams.dt_num_layers,
|
||||
num_heads: 4,
|
||||
context_len: self.hyperparams.dt_context_len,
|
||||
dropout: 0.1,
|
||||
batch_size: self.hyperparams.batch_size.min(256),
|
||||
};
|
||||
|
||||
let mut dt = crate::cuda_pipeline::decision_transformer::DecisionTransformer::new(
|
||||
stream, dt_config,
|
||||
).map_err(|e| anyhow::anyhow!("DT init: {e}"))?;
|
||||
|
||||
let num_bars = training_data.len();
|
||||
if let (Some(ref features_gpu), Some(ref targets_gpu)) =
|
||||
(&self.features_raw_cuda, &self.targets_raw_cuda)
|
||||
{
|
||||
let (trajectories, target_actions, num_batches) = dt
|
||||
.build_dt_trajectories(
|
||||
features_gpu,
|
||||
targets_gpu,
|
||||
num_bars,
|
||||
self.hyperparams.gamma,
|
||||
)
|
||||
.map_err(|e| anyhow::anyhow!("DT trajectory build: {e}"))?;
|
||||
|
||||
if num_batches == 0 {
|
||||
warn!(
|
||||
num_bars,
|
||||
context_len = self.hyperparams.dt_context_len,
|
||||
batch_size = dt.config().batch_size,
|
||||
"DT pre-training: not enough data for a full batch, skipping"
|
||||
);
|
||||
} else {
|
||||
let batch_size = dt.config().batch_size;
|
||||
let context_len = dt.config().context_len;
|
||||
let input_dim = dt.config().state_dim + 2;
|
||||
|
||||
for epoch in 0..self.hyperparams.dt_pretrain_epochs {
|
||||
let mut epoch_loss = 0.0_f32;
|
||||
let mut batch_count = 0_usize;
|
||||
|
||||
for batch_idx in 0..num_batches {
|
||||
let ep_offset = batch_idx * batch_size;
|
||||
let traj_elem_offset = ep_offset * context_len * input_dim;
|
||||
let act_elem_offset = ep_offset * context_len;
|
||||
|
||||
let loss = dt.pretrain_step(
|
||||
&trajectories,
|
||||
&target_actions,
|
||||
batch_size,
|
||||
traj_elem_offset,
|
||||
act_elem_offset,
|
||||
).map_err(|e| anyhow::anyhow!("DT pretrain_step: {e}"))?;
|
||||
|
||||
epoch_loss += loss;
|
||||
batch_count += 1;
|
||||
}
|
||||
|
||||
let avg_loss = if batch_count > 0 {
|
||||
epoch_loss / batch_count as f32
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
|
||||
info!(
|
||||
epoch = epoch + 1,
|
||||
total_epochs = self.hyperparams.dt_pretrain_epochs,
|
||||
avg_loss = format!("{avg_loss:.4}"),
|
||||
batches = batch_count,
|
||||
"DT pre-training"
|
||||
);
|
||||
|
||||
training_metrics::set_epoch(
|
||||
"dt_pretrain",
|
||||
"loss",
|
||||
avg_loss as f64,
|
||||
);
|
||||
}
|
||||
|
||||
info!(
|
||||
epochs = self.hyperparams.dt_pretrain_epochs,
|
||||
"DT pre-training complete"
|
||||
);
|
||||
}
|
||||
} else {
|
||||
warn!("DT pre-training: GPU features/targets not available, skipping");
|
||||
}
|
||||
} else {
|
||||
warn!("DT pre-training: no CUDA stream available, skipping");
|
||||
}
|
||||
}
|
||||
|
||||
// Training loop
|
||||
for epoch in 0..self.hyperparams.epochs {
|
||||
self.current_epoch = epoch;
|
||||
self.reset_epoch_state(epoch);
|
||||
|
||||
// C51 warmup ramp
|
||||
{
|
||||
let alpha = if self.hyperparams.c51_warmup_epochs > 0 {
|
||||
(epoch as f32 / self.hyperparams.c51_warmup_epochs as f32).min(1.0)
|
||||
} else {
|
||||
1.0
|
||||
};
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
fused.set_c51_alpha(alpha);
|
||||
}
|
||||
}
|
||||
|
||||
// Q-gap warmup
|
||||
if self.hyperparams.q_gap_threshold > 0.0 {
|
||||
let warmup_epochs = 5.0_f32;
|
||||
let ramp = (epoch as f32 / warmup_epochs).min(1.0);
|
||||
let ramped_threshold = self.hyperparams.q_gap_threshold as f32 * ramp;
|
||||
if let Some(ref mut selector) = self.gpu_action_selector {
|
||||
selector.set_q_gap_threshold(ramped_threshold);
|
||||
}
|
||||
}
|
||||
|
||||
log_epoch_start(epoch + 1, self.hyperparams.epochs, self.hyperparams.learning_rate);
|
||||
training_metrics::set_epoch("dqn", "current", (epoch + 1) as f64);
|
||||
|
||||
let mut monitor = TrainingMonitor::new(epoch + 1);
|
||||
|
||||
self.portfolio_tracker.reset_drawdown_tracking();
|
||||
|
||||
let epoch_start = std::time::Instant::now();
|
||||
|
||||
// ── Phase 1: GPU data already uploaded via init_from_fxcache ──
|
||||
let phase1_start = std::time::Instant::now();
|
||||
// Assert gpu_data + raw buffers are pre-populated (no re-upload)
|
||||
if self.gpu_data.is_none() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"train_with_data_full_loop_slices requires gpu_data (call init_from_fxcache first)"
|
||||
));
|
||||
}
|
||||
if self.targets_raw_cuda.is_none() {
|
||||
return Err(anyhow::anyhow!(
|
||||
"train_with_data_full_loop_slices requires targets_raw_cuda (call init_from_fxcache first)"
|
||||
));
|
||||
}
|
||||
self.init_gpu_experience_collector().await?;
|
||||
let phase1_ms = phase1_start.elapsed().as_secs_f64() * 1000.0;
|
||||
|
||||
// Causal feature masking
|
||||
{
|
||||
let frac = self.hyperparams.feature_mask_fraction;
|
||||
self.epoch_feature_mask = None;
|
||||
let _ = frac; // suppress unused warning
|
||||
}
|
||||
|
||||
// Vol normalization
|
||||
self.epoch_vol_normalizer = if training_data.len() > 20 {
|
||||
let returns: Vec<f64> = training_data.iter()
|
||||
.map(|(fv, _)| fv[3])
|
||||
.collect();
|
||||
let n = returns.len() as f64;
|
||||
let mean = returns.iter().sum::<f64>() / n;
|
||||
let var = returns.iter().map(|r| (r - mean).powi(2)).sum::<f64>() / (n - 1.0);
|
||||
var.sqrt().max(1e-8) as f32
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
|
||||
// ── Phase 2: GPU experience collection ──
|
||||
eprintln!("[DEBUG] Phase 2: starting GPU experience collection (epoch {})", epoch);
|
||||
let phase2_start = std::time::Instant::now();
|
||||
let gpu_experiences_collected = self.collect_gpu_experiences_slices(
|
||||
training_data,
|
||||
).await?;
|
||||
let phase2_ms = phase2_start.elapsed().as_secs_f64() * 1000.0;
|
||||
eprintln!("[DEBUG] Phase 2: done in {:.1}ms, collected={}", phase2_ms, gpu_experiences_collected);
|
||||
|
||||
if !gpu_experiences_collected {
|
||||
return Err(anyhow::anyhow!(
|
||||
"GPU experience collector MUST be active for CUDA training. \
|
||||
No CPU fallback path exists in CUDA builds."
|
||||
));
|
||||
}
|
||||
|
||||
// Periodic shrink-and-perturb
|
||||
let sp_interval = self.hyperparams.shrink_perturb_interval;
|
||||
if sp_interval > 0 && epoch > 0 && epoch % sp_interval == 0 {
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
let alpha = self.hyperparams.shrink_perturb_alpha as f32;
|
||||
let sigma = self.hyperparams.shrink_perturb_sigma as f32;
|
||||
match fused.shrink_and_perturb(alpha, sigma) {
|
||||
Ok(()) => info!(epoch, alpha, sigma, "Periodic shrink-and-perturb applied"),
|
||||
Err(e) => warn!(epoch, "Shrink-and-perturb failed (non-fatal): {e}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Phase 3: Batched training from replay buffer ──
|
||||
eprintln!("[DEBUG] Phase 3: starting training steps (epoch {})", epoch);
|
||||
let phase3_start = std::time::Instant::now();
|
||||
let train_step_count = self.run_training_steps_slices(training_data.len()).await?;
|
||||
let phase3_ms = phase3_start.elapsed().as_secs_f64() * 1000.0;
|
||||
eprintln!("[DEBUG] Phase 3: done in {:.1}ms, steps={}", phase3_ms, train_step_count);
|
||||
|
||||
// ── Phase 4: Epoch boundary + validation ──
|
||||
let phase4_start = std::time::Instant::now();
|
||||
let boundary = if train_step_count > 0 {
|
||||
let b = self.process_epoch_boundary(epoch, train_step_count, &mut monitor).await?;
|
||||
Some(b)
|
||||
} else {
|
||||
None
|
||||
};
|
||||
|
||||
self.sync_gpu_weights().await?;
|
||||
|
||||
// Three-Phase Training Pipeline
|
||||
let total_epochs = self.hyperparams.epochs;
|
||||
let phase3_start_epoch = total_epochs * 80 / 100;
|
||||
let in_phase3 = epoch >= phase3_start_epoch && total_epochs > 4;
|
||||
|
||||
if in_phase3 {
|
||||
if let Some(ref mut c) = self.gpu_experience_collector {
|
||||
c.set_expert_ratio(0.0);
|
||||
}
|
||||
if epoch == phase3_start_epoch {
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
let alpha = 0.9_f32;
|
||||
let sigma = 0.01_f32;
|
||||
if let Err(e) = fused.shrink_and_perturb(alpha, sigma) {
|
||||
tracing::warn!("Shrink-and-Perturb failed at Phase 3 start (non-fatal): {e}");
|
||||
} else {
|
||||
tracing::info!(
|
||||
epoch,
|
||||
phase3_start = phase3_start_epoch,
|
||||
alpha,
|
||||
sigma,
|
||||
"Phase 3 (Refinement): Shrink-and-Perturb applied at phase boundary"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if !in_phase3 && self.hyperparams.epochs > 4 {
|
||||
let interval = (self.hyperparams.epochs / 4).max(1);
|
||||
if epoch > 0 && epoch % interval == 0 {
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
let alpha = 0.9_f32;
|
||||
let sigma = 0.01_f32;
|
||||
if let Err(e) = fused.shrink_and_perturb(alpha, sigma) {
|
||||
tracing::warn!("Shrink-and-Perturb failed (non-fatal): {e}");
|
||||
} else {
|
||||
tracing::info!(
|
||||
epoch,
|
||||
alpha,
|
||||
sigma,
|
||||
"Shrink-and-Perturb applied (plasticity maintenance)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// CVaR scales from IQN head
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
let bs = fused.batch_size();
|
||||
let actions_clone = {
|
||||
let actions = fused.actions_buf();
|
||||
actions.clone()
|
||||
};
|
||||
let cvar_ptr = fused.compute_cvar_device_ptr(&actions_clone, bs, 0.05)
|
||||
.unwrap_or(0);
|
||||
if let Some(ref mut collector) = self.gpu_experience_collector {
|
||||
collector.set_cvar_scales(cvar_ptr);
|
||||
}
|
||||
}
|
||||
|
||||
// Flush GPU-accumulated max priority
|
||||
if train_step_count > 0 {
|
||||
let agent = self.agent.read().await;
|
||||
if let Err(e) = agent.flush_max_priority() {
|
||||
debug!("GPU max_priority flush failed (non-fatal): {}", e);
|
||||
}
|
||||
}
|
||||
let phase4_ms = phase4_start.elapsed().as_secs_f64() * 1000.0;
|
||||
|
||||
let epoch_duration = epoch_start.elapsed();
|
||||
|
||||
info!(
|
||||
"Epoch {}/{} phase breakdown: init={:.0}ms experience={:.0}ms training={:.0}ms validation={:.0}ms total={:.0}ms",
|
||||
epoch + 1, self.hyperparams.epochs,
|
||||
phase1_ms, phase2_ms, phase3_ms, phase4_ms,
|
||||
epoch_duration.as_secs_f64() * 1000.0,
|
||||
);
|
||||
|
||||
let log_output = self.log_epoch_metrics_and_financials(
|
||||
epoch,
|
||||
train_step_count,
|
||||
&boundary,
|
||||
&mut monitor,
|
||||
epoch_duration,
|
||||
&mut total_action_counts,
|
||||
&mut total_factored_action_counts,
|
||||
).await?;
|
||||
|
||||
total_loss += log_output.avg_loss;
|
||||
total_q_value += log_output.avg_q_value;
|
||||
total_gradient_norm += log_output.avg_grad_norm;
|
||||
|
||||
let epoch_avg_reward = if !monitor.reward_history.is_empty() {
|
||||
monitor.reward_history.iter().sum::<f32>() / monitor.reward_history.len() as f32
|
||||
} else {
|
||||
0.0
|
||||
};
|
||||
total_reward += epoch_avg_reward as f64;
|
||||
|
||||
self.handle_epoch_checkpoints_and_early_stopping(
|
||||
epoch,
|
||||
train_step_count,
|
||||
&log_output,
|
||||
&mut checkpoint_callback,
|
||||
).await?;
|
||||
|
||||
if (epoch + 1) % self.hyperparams.checkpoint_frequency == 0 {
|
||||
info!(
|
||||
"Saving periodic checkpoint at epoch {}/{}",
|
||||
epoch + 1, self.hyperparams.epochs
|
||||
);
|
||||
let checkpoint_data = self.serialize_model().await?;
|
||||
let checkpoint_size = checkpoint_data.len();
|
||||
let checkpoint_path = checkpoint_callback(epoch + 1, checkpoint_data, false)
|
||||
.context("Failed to save periodic checkpoint")?;
|
||||
info!(
|
||||
"Periodic checkpoint saved: {} ({} bytes)",
|
||||
checkpoint_path, checkpoint_size
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let training_duration = start_time.elapsed();
|
||||
|
||||
let metrics = self.create_final_metrics(
|
||||
total_loss, total_q_value, total_gradient_norm, total_reward,
|
||||
self.hyperparams.epochs, training_duration, false,
|
||||
total_action_counts, total_factored_action_counts,
|
||||
).await?;
|
||||
|
||||
{
|
||||
let mut stored_metrics = self.metrics.write().await;
|
||||
*stored_metrics = metrics.clone();
|
||||
}
|
||||
|
||||
info!(
|
||||
"Training completed in {:.2}s: final_loss={:.6}, avg_q_value={:.4}",
|
||||
training_duration.as_secs_f64(),
|
||||
metrics.loss,
|
||||
metrics.additional_metrics.get("avg_q_value").unwrap_or(&0.0)
|
||||
);
|
||||
|
||||
info!("Best model summary:");
|
||||
info!(
|
||||
" Best Sharpe: {:.4} at epoch {} (val_loss={:.6})",
|
||||
self.best_sharpe, self.best_epoch, self.best_val_loss
|
||||
);
|
||||
info!(" Best model checkpoint: best_model.safetensors");
|
||||
|
||||
Ok(metrics)
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
// Helper: one-time training configuration logging
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
@@ -718,6 +1131,42 @@ impl DQNTrainer {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Upload raw features + targets from contiguous fxcache slices.
|
||||
/// No tuple unpacking — flat_map directly from &[[f64; N]] arrays.
|
||||
pub(crate) async fn init_gpu_raw_buffers_from_slices(
|
||||
&mut self,
|
||||
features: &[[f64; 42]],
|
||||
targets: &[[f64; 4]],
|
||||
) -> Result<()> {
|
||||
if self.targets_raw_cuda.is_some() {
|
||||
return Ok(());
|
||||
}
|
||||
let stream = match self.cuda_stream {
|
||||
Some(ref s) => std::sync::Arc::clone(s),
|
||||
None => return Ok(()),
|
||||
};
|
||||
|
||||
let num_bars = features.len();
|
||||
let flat_targets: Vec<f32> = targets.iter()
|
||||
.flat_map(|t| t.iter().map(|&v| v as f32))
|
||||
.collect();
|
||||
let flat_features: Vec<f32> = features.iter()
|
||||
.flat_map(|f| f.iter().map(|&v| v as f32))
|
||||
.collect();
|
||||
|
||||
self.targets_raw_cuda = Some(
|
||||
crate::cuda_pipeline::clone_htod_f32_to_bf16(&stream, &flat_targets)
|
||||
.map_err(|e| anyhow::anyhow!("targets_raw upload: {e}"))?
|
||||
);
|
||||
self.features_raw_cuda = Some(
|
||||
crate::cuda_pipeline::clone_htod_f32_to_bf16(&stream, &flat_features)
|
||||
.map_err(|e| anyhow::anyhow!("features_raw upload: {e}"))?
|
||||
);
|
||||
|
||||
tracing::info!("CUDA raw buffers uploaded from slices: {} bars x 42+4", num_bars);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
// Helper: Phase 1c — GPU experience collector init (once)
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
@@ -1242,6 +1691,329 @@ impl DQNTrainer {
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
// Helper: Phase 3 — GPU experience collection (zero-alloc slice variant)
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
/// Like `collect_gpu_experiences` but accepts `&[([f64; 42], [f64; 4])]`.
|
||||
/// Only `.len()` and `.0` (features) are accessed — the `[f64; 4]` targets
|
||||
/// are never read here (GPU raw buffers are already uploaded).
|
||||
pub(crate) async fn collect_gpu_experiences_slices(
|
||||
&mut self,
|
||||
training_data: &[([f64; 42], [f64; 4])],
|
||||
) -> Result<bool> {
|
||||
// Feature normalization stats calculation epoch (kept for API compat)
|
||||
let _stats_collection_epochs = {
|
||||
let ratio_based = (self.hyperparams.epochs as f32 * self.hyperparams.feature_stats_collection_ratio) as usize;
|
||||
let capped = match self.hyperparams.max_feature_stats_epochs {
|
||||
Some(max_epochs) => ratio_based.min(max_epochs),
|
||||
None => ratio_based,
|
||||
};
|
||||
capped.max(1)
|
||||
};
|
||||
|
||||
// Set expert demo ratio BEFORE borrowing collector (avoids borrow conflict)
|
||||
if self.hyperparams.expert_demo_ratio > 0.0 {
|
||||
use crate::trainers::dqn::expert_demos::ExpertDemoGenerator;
|
||||
let effective = ExpertDemoGenerator::effective_ratio(
|
||||
self.hyperparams.expert_demo_ratio,
|
||||
self.hyperparams.expert_demo_decay_epochs,
|
||||
self.current_epoch,
|
||||
) as f32;
|
||||
if let Some(ref mut c) = self.gpu_experience_collector {
|
||||
c.set_expert_ratio(effective);
|
||||
}
|
||||
}
|
||||
|
||||
let (
|
||||
Some(ref mut collector),
|
||||
Some(ref features_buf),
|
||||
Some(ref targets_buf),
|
||||
) = (
|
||||
&mut self.gpu_experience_collector,
|
||||
&self.features_raw_cuda,
|
||||
&self.targets_raw_cuda,
|
||||
) else {
|
||||
return Ok(false);
|
||||
};
|
||||
|
||||
use crate::cuda_pipeline::gpu_experience_collector::ExperienceCollectorConfig;
|
||||
|
||||
let raw_sd = if !self.hyperparams.mbp10_data_dir.is_empty() { 53 } else { 45 };
|
||||
let aligned_sd = (raw_sd + 7) & !7;
|
||||
|
||||
// Cache n_episodes on first epoch — always auto-scaled from VRAM.
|
||||
let n_episodes = if let Some(cached) = self.cached_n_episodes {
|
||||
cached
|
||||
} else {
|
||||
use ml_core::memory_optimization::detect_gpu_hardware;
|
||||
let computed = match detect_gpu_hardware() {
|
||||
Ok(hw) => {
|
||||
let optimal = hw.optimal_n_episodes(
|
||||
aligned_sd,
|
||||
self.hyperparams.gpu_timesteps_per_episode,
|
||||
);
|
||||
let chosen = optimal.max(32).min(16384) as i32;
|
||||
info!(
|
||||
"GPU auto-scaled n_episodes: {} (SMs={}, VRAM={:.0}MB)",
|
||||
chosen, hw.sm_count, hw.free_memory_mb
|
||||
);
|
||||
chosen
|
||||
}
|
||||
Err(_) => 256_i32,
|
||||
};
|
||||
self.cached_n_episodes = Some(computed);
|
||||
computed
|
||||
};
|
||||
|
||||
let timesteps = self.hyperparams.gpu_timesteps_per_episode.min(1000) as i32;
|
||||
let total_bars = training_data.len() as i32;
|
||||
let usable_bars = (total_bars - timesteps).max(1);
|
||||
let stride = (usable_bars / n_episodes).max(1);
|
||||
|
||||
collector.generate_episode_starts_gpu(
|
||||
n_episodes as usize, stride, usable_bars,
|
||||
).map_err(|e| anyhow::anyhow!("GPU episode starts: {e}"))?;
|
||||
collector.generate_sim_params_gpu(
|
||||
n_episodes as usize,
|
||||
self.hyperparams.transaction_cost_multiplier as f32,
|
||||
self.hyperparams.avg_spread as f32,
|
||||
self.hyperparams.fill_ioc_fill_prob as f32,
|
||||
self.hyperparams.fill_limit_fill_min as f32,
|
||||
self.hyperparams.fill_limit_fill_max as f32,
|
||||
).map_err(|e| anyhow::anyhow!("GPU sim params: {e}"))?;
|
||||
|
||||
let mut episode_starts: Vec<i32> = (0..n_episodes)
|
||||
.map(|i| {
|
||||
let base = (i * stride).rem_euclid(usable_bars);
|
||||
base
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Curriculum learning: filter episode starts by difficulty
|
||||
let curriculum = self.hyperparams.curriculum_phase;
|
||||
if curriculum < 2 && !training_data.is_empty() {
|
||||
let adx_idx = 40;
|
||||
let adx_threshold = 30.0_f64;
|
||||
let original = episode_starts.clone();
|
||||
episode_starts.clear();
|
||||
|
||||
for &start in &original {
|
||||
let bar = (start as usize).min(training_data.len().saturating_sub(1));
|
||||
let features = &training_data[bar].0;
|
||||
let adx = if features.len() > adx_idx { features[adx_idx] } else { 0.0 };
|
||||
|
||||
match curriculum {
|
||||
0 => {
|
||||
if adx > adx_threshold {
|
||||
episode_starts.push(start);
|
||||
}
|
||||
}
|
||||
1 => {
|
||||
episode_starts.push(start);
|
||||
if adx > adx_threshold {
|
||||
episode_starts.push(start);
|
||||
}
|
||||
}
|
||||
_ => episode_starts.push(start),
|
||||
}
|
||||
}
|
||||
|
||||
if episode_starts.len() < 16 {
|
||||
tracing::warn!(
|
||||
curriculum,
|
||||
filtered = episode_starts.len(),
|
||||
original = original.len(),
|
||||
"Curriculum phase filtered too aggressively, falling back to Full"
|
||||
);
|
||||
episode_starts = original;
|
||||
} else {
|
||||
tracing::debug!(
|
||||
curriculum,
|
||||
filtered = episode_starts.len(),
|
||||
original = original.len(),
|
||||
"Curriculum learning: episode starts filtered by ADX difficulty"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Reset per-episode state before each epoch
|
||||
if let Err(e) = collector.reset_episodes(
|
||||
self.hyperparams.initial_capital as f32,
|
||||
self.hyperparams.avg_spread as f32,
|
||||
self.hyperparams.cash_reserve_percent as f32,
|
||||
) {
|
||||
return Err(anyhow::anyhow!("GPU episode reset FAILED (no CPU fallback): {e}"));
|
||||
}
|
||||
|
||||
let agent = self.agent.read().await;
|
||||
let epsilon = agent.get_epsilon();
|
||||
drop(agent);
|
||||
|
||||
let adversarial = self.adversarial_active;
|
||||
if adversarial {
|
||||
info!("Adversarial regime ACTIVE this epoch: 3x spread, 2x tx_cost, 0.5x fill");
|
||||
}
|
||||
|
||||
let config = ExperienceCollectorConfig {
|
||||
n_episodes,
|
||||
timesteps_per_episode: timesteps,
|
||||
total_bars,
|
||||
epsilon,
|
||||
gamma: self.hyperparams.gamma as f32,
|
||||
max_position: self.max_position as f32,
|
||||
enable_action_masking: self.enable_action_masking,
|
||||
curiosity_scale: 1.0,
|
||||
loss_aversion: self.hyperparams.loss_aversion as f32,
|
||||
tx_cost_multiplier: {
|
||||
let base = self.hyperparams.transaction_cost_multiplier as f32;
|
||||
if adversarial { base * 2.0 } else { base }
|
||||
},
|
||||
count_bonus_coefficient: self.hyperparams.count_bonus_coefficient
|
||||
.unwrap_or(0.0) as f32,
|
||||
q_clip_min: self.hyperparams.q_clip_min as f32,
|
||||
q_clip_max: self.hyperparams.q_clip_max as f32,
|
||||
huber_kappa: if true {
|
||||
self.hyperparams.huber_delta as f32
|
||||
} else {
|
||||
0.0
|
||||
},
|
||||
use_noisy_nets: true,
|
||||
noisy_sigma_init: self.hyperparams.noisy_sigma_init as f32,
|
||||
use_distributional: true,
|
||||
num_atoms: self.hyperparams.num_atoms as i32,
|
||||
v_min: self.hyperparams.v_min as f32,
|
||||
v_max: self.hyperparams.v_max as f32,
|
||||
fill_median_spread: {
|
||||
let base = self.hyperparams.avg_spread as f32;
|
||||
if adversarial { base * 3.0 } else { base }
|
||||
},
|
||||
fill_median_vol: self.median_vol as f32,
|
||||
fill_ioc_fill_prob: {
|
||||
let base = self.hyperparams.fill_ioc_fill_prob as f32;
|
||||
if adversarial { base * 0.5 } else { base }
|
||||
},
|
||||
fill_limit_fill_min: self.hyperparams.fill_limit_fill_min as f32,
|
||||
fill_limit_fill_max: self.hyperparams.fill_limit_fill_max as f32,
|
||||
fill_spread_cost_frac: self.hyperparams.fill_spread_cost_frac as f32,
|
||||
fill_spread_capture_frac: self.hyperparams.fill_spread_capture_frac as f32,
|
||||
fill_simulation_enabled: self.median_vol > 0.0,
|
||||
q_gap_threshold: {
|
||||
let target = self.hyperparams.q_gap_threshold as f32;
|
||||
let epoch = self.current_epoch.min(10) as f32;
|
||||
target * (epoch / 10.0)
|
||||
},
|
||||
dsr_eta: self.hyperparams.dsr_eta as f32,
|
||||
n_steps: self.hyperparams.n_steps as i32,
|
||||
min_hold_bars: self.hyperparams.min_hold_bars as i32,
|
||||
spread_cost: {
|
||||
let base = (self.hyperparams.tick_size * self.hyperparams.contract_multiplier
|
||||
* self.hyperparams.fill_spread_cost_frac) as f32;
|
||||
if adversarial { base * 3.0 } else { base }
|
||||
},
|
||||
contract_multiplier: self.hyperparams.contract_multiplier as f32,
|
||||
margin_pct: self.hyperparams.margin_pct as f32,
|
||||
dd_threshold: self.hyperparams.dd_threshold as f32,
|
||||
w_dd: self.hyperparams.w_dd as f32,
|
||||
beta_penalty: self.hyperparams.beta_penalty_strength as f32,
|
||||
time_reversal_mod: self.hyperparams.time_reversal_mod as i32,
|
||||
regret_blend: self.hyperparams.regret_blend as f32,
|
||||
mirror_active: self.current_epoch % 2 == 1,
|
||||
position_entropy_weight: self.hyperparams.position_entropy_weight as f32,
|
||||
trade_clustering_penalty: self.hyperparams.trade_clustering_penalty as f32,
|
||||
feature_noise_scale: self.hyperparams.feature_noise_scale as f32,
|
||||
vol_normalizer: self.epoch_vol_normalizer,
|
||||
feature_mask: self.epoch_feature_mask.clone(),
|
||||
..Default::default()
|
||||
};
|
||||
|
||||
let gpu_batch = collector.collect_experiences_gpu(
|
||||
features_buf, targets_buf, &episode_starts, &config,
|
||||
).map_err(|e| anyhow::anyhow!(
|
||||
"GPU zero-roundtrip collection FAILED (no CPU fallback): {e}"
|
||||
))?;
|
||||
|
||||
let count = gpu_batch.n_episodes * gpu_batch.timesteps;
|
||||
|
||||
if let Some(ref stream) = self.cuda_stream {
|
||||
self.experience_done_event = Some(stream.record_event(None)
|
||||
.map_err(|e| { eprintln!("!!! EVENT RECORD FAILED: {e}"); anyhow::anyhow!("experience event record: {e}") })?);
|
||||
}
|
||||
|
||||
if let Some(ref mut mon) = self.gpu_monitoring {
|
||||
if let Err(e) = mon.reduce(collector.rewards_gpu(), collector.actions_gpu(), count) {
|
||||
debug!("GPU monitoring reduce failed: {e}");
|
||||
}
|
||||
if let Some(ref stream) = self.cuda_stream {
|
||||
if stream.synchronize().is_err() {
|
||||
warn!("GPU monitoring kernel crashed — disabling monitoring");
|
||||
self.gpu_monitoring = None;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if count > 0 {
|
||||
if let Err(e) = collector.train_curiosity_gpu(
|
||||
gpu_batch.n_episodes, gpu_batch.timesteps,
|
||||
) {
|
||||
debug!("GPU curiosity training failed (non-fatal): {e}");
|
||||
}
|
||||
if let Some(ref stream) = self.cuda_stream {
|
||||
match stream.record_event(None) {
|
||||
Ok(event) => {
|
||||
if !event.is_complete() {
|
||||
if let Err(e) = event.synchronize() {
|
||||
warn!("Curiosity kernel async error detected: {e} — disabling curiosity for this run");
|
||||
self.curiosity_module = None;
|
||||
collector.disable_curiosity();
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Curiosity event record failed: {e} — disabling curiosity for this run");
|
||||
self.curiosity_module = None;
|
||||
collector.disable_curiosity();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if count > 0 {
|
||||
let total = gpu_batch.n_episodes * gpu_batch.timesteps;
|
||||
|
||||
let actions_u32: &CudaSlice<u32> = unsafe {
|
||||
&*(&gpu_batch.actions as *const CudaSlice<i32> as *const CudaSlice<u32>)
|
||||
};
|
||||
|
||||
let agent = self.agent.read().await;
|
||||
let mut gpu_buf = agent.memory().as_gpu_buffer()
|
||||
.ok_or_else(|| anyhow::anyhow!("GPU PER buffer required"))?;
|
||||
gpu_buf.gpu.insert_batch_bf16(
|
||||
&gpu_batch.states,
|
||||
&gpu_batch.next_states,
|
||||
actions_u32,
|
||||
&gpu_batch.rewards,
|
||||
&gpu_batch.dones,
|
||||
total,
|
||||
).map_err(|e| anyhow::anyhow!("GPU PER insert_batch: {e}"))?;
|
||||
}
|
||||
|
||||
match collector.read_epoch_dsr_state() {
|
||||
Ok((dsr_a, dsr_b)) => {
|
||||
self.reward_fn.sync_dsr_from_gpu(dsr_a as f64, dsr_b as f64);
|
||||
debug!(
|
||||
"DSR GPU->CPU sync: ema_return={:.6}, ema_return_sq={:.6}",
|
||||
dsr_a, dsr_b
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
debug!("DSR epoch_state readback failed (non-fatal): {e}");
|
||||
}
|
||||
}
|
||||
|
||||
Ok(true)
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
// Helper: Batched training steps with guard kernel
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
@@ -1435,6 +2207,186 @@ impl DQNTrainer {
|
||||
Ok(train_step_count)
|
||||
}
|
||||
|
||||
/// Like `run_training_steps` but accepts `num_bars` directly instead of
|
||||
/// extracting `.len()` from a `&[(FeatureVector, Vec<f64>)]` slice.
|
||||
/// This avoids requiring a `Vec<f64>` allocation per bar.
|
||||
pub(crate) async fn run_training_steps_slices(
|
||||
&mut self,
|
||||
num_bars: usize,
|
||||
) -> Result<usize> {
|
||||
let batch_size = self.hyperparams.batch_size;
|
||||
let num_training_steps = if self.can_train().await? {
|
||||
(num_bars / batch_size).max(1)
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
let mut train_step_count = 0;
|
||||
|
||||
// Lazy-init fused CUDA training context.
|
||||
if num_training_steps > 0 {
|
||||
let needs_init = match &self.fused_ctx {
|
||||
None => self.device.is_cuda(),
|
||||
Some(ctx) => ctx.batch_size() != self.current_batch_size,
|
||||
};
|
||||
if needs_init {
|
||||
if let Some(ref stream) = self.cuda_stream {
|
||||
if self.fused_ctx.is_some() {
|
||||
tracing::info!("Fused CUDA context: batch_size changed, recreating");
|
||||
self.fused_ctx = None;
|
||||
}
|
||||
let agent = self.agent.read().await;
|
||||
match super::super::fused_training::FusedTrainingCtx::new(
|
||||
&self.device, &*agent, &self.hyperparams, self.current_batch_size, std::sync::Arc::clone(stream),
|
||||
) {
|
||||
Ok(ctx) => { self.fused_ctx = Some(ctx); }
|
||||
Err(e) => { tracing::error!("Fused CUDA context init failed: {e}"); }
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Lazy-init + reset guard accumulators for this epoch
|
||||
{
|
||||
if self.training_guard.is_none() && self.device.is_cuda() {
|
||||
match crate::cuda_pipeline::gpu_training_guard::GpuTrainingGuard::from_stream(Arc::clone(self.cuda_stream.as_ref().ok_or_else(|| anyhow::anyhow!("CUDA stream for training guard"))?)) {
|
||||
Ok(g) => {
|
||||
info!("GPU training guard initialized (epoch loop)");
|
||||
self.training_guard = Some(g);
|
||||
}
|
||||
Err(e) => return Err(anyhow::anyhow!("GPU training guard init: {e}")),
|
||||
}
|
||||
}
|
||||
if let Some(ref mut guard) = self.training_guard {
|
||||
guard.reset_accumulators()
|
||||
.map_err(|e| anyhow::anyhow!("guard reset: {e}"))?;
|
||||
}
|
||||
}
|
||||
|
||||
let guard_collapse_thresh =
|
||||
self.hyperparams.learning_rate as f32
|
||||
* self.hyperparams.gradient_collapse_multiplier as f32;
|
||||
let guard_past_warmup = {
|
||||
let ws = (self.collapse_warmup_buffer_size as f64 * 0.2) as u64;
|
||||
self.gradient_logging_step as u64 > ws
|
||||
};
|
||||
|
||||
let mut sample_total_us = 0_u64;
|
||||
let mut fused_total_us = 0_u64;
|
||||
let mut guard_total_us = 0_u64;
|
||||
|
||||
// Wait for experience collection GPU kernels to complete before sampling.
|
||||
if let Some(event) = self.experience_done_event.take() {
|
||||
if !event.is_complete() {
|
||||
event.synchronize()
|
||||
.map_err(|e| anyhow::anyhow!("experience event wait: {e}"))?;
|
||||
}
|
||||
}
|
||||
|
||||
for _step in 0..num_training_steps {
|
||||
let sample_start = std::time::Instant::now();
|
||||
let (batch, vaccine_batch) = {
|
||||
let agent = self.agent.read().await;
|
||||
let buffer = agent.memory();
|
||||
if !buffer.can_sample(self.current_batch_size) {
|
||||
break;
|
||||
}
|
||||
let b = buffer.sample(self.current_batch_size)
|
||||
.map_err(|e| anyhow::anyhow!("PER sample: {e}"))?;
|
||||
let vb = buffer.sample(self.current_batch_size).ok();
|
||||
(b, vb)
|
||||
};
|
||||
sample_total_us += sample_start.elapsed().as_micros() as u64;
|
||||
|
||||
{
|
||||
let mut agent = self.agent.write().await;
|
||||
|
||||
let fused_start = std::time::Instant::now();
|
||||
let _gpu_result = if let Some(ref mut fused) = self.fused_ctx {
|
||||
if let Some(vb) = vaccine_batch {
|
||||
fused.pending_vaccine_batch = vb.gpu_batch;
|
||||
}
|
||||
let result = fused.run_full_step(&batch, &mut *agent, &self.device)
|
||||
.map_err(|e| { eprintln!("!!! FUSED STEP ERROR: {:#}", e); e })
|
||||
.context("Fused CUDA training step failed")?;
|
||||
result
|
||||
} else {
|
||||
unreachable!("Fused CUDA training is the only production path")
|
||||
};
|
||||
|
||||
fused_total_us += fused_start.elapsed().as_micros() as u64;
|
||||
|
||||
let guard_start = std::time::Instant::now();
|
||||
if let Some(ref mut guard) = self.training_guard {
|
||||
let (loss_raw, grad_raw) = if let Some(ref fused) = self.fused_ctx {
|
||||
(fused.loss_gpu_buf().raw_ptr(), fused.grad_norm_gpu_buf().raw_ptr())
|
||||
} else {
|
||||
let ls = _gpu_result.loss_cuda_slice()
|
||||
.map_err(|e| anyhow::anyhow!("guard loss: {e}"))?;
|
||||
let gs = _gpu_result.grad_norm_cuda_slice()
|
||||
.map_err(|e| anyhow::anyhow!("guard grad: {e}"))?;
|
||||
(ls.raw_ptr(), gs.raw_ptr())
|
||||
};
|
||||
let gr = guard.check_and_accumulate(
|
||||
loss_raw,
|
||||
grad_raw,
|
||||
1e6_f32,
|
||||
guard_collapse_thresh,
|
||||
!guard_past_warmup,
|
||||
).map_err(|e| anyhow::anyhow!("guard check: {e}"))?;
|
||||
if gr.halt_nan {
|
||||
return Err(anyhow::anyhow!(
|
||||
"NaN/Inf at step {}: loss={}, grad={}",
|
||||
train_step_count, gr.raw_loss, gr.raw_grad_norm
|
||||
));
|
||||
}
|
||||
if gr.halt_grad_collapse {
|
||||
agent.check_gradient_collapse(gr.raw_grad_norm).map_err(|e| {
|
||||
tracing::info!("Early stopping (gradient collapse): {}", e);
|
||||
anyhow::anyhow!("Early stopping: {}", e)
|
||||
})?;
|
||||
}
|
||||
}
|
||||
guard_total_us += guard_start.elapsed().as_micros() as u64;
|
||||
|
||||
if train_step_count % 50 == 0 {
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
if let Ok(stats) = fused.reduce_current_q_stats() {
|
||||
self.cached_avg_q = stats.q_mean as f64;
|
||||
if stats.q_min < self.epoch_q_min { self.epoch_q_min = stats.q_min; }
|
||||
if stats.q_max > self.epoch_q_max { self.epoch_q_max = stats.q_max; }
|
||||
}
|
||||
}
|
||||
}
|
||||
train_step_count += 1;
|
||||
self.gradient_logging_step += 1;
|
||||
}
|
||||
}
|
||||
|
||||
if let Some(ref mut fused) = self.fused_ctx {
|
||||
let _ = fused.flush_readback();
|
||||
if let Ok(stats) = fused.flush_q_stats_readback() {
|
||||
self.cached_avg_q = stats.q_mean as f64;
|
||||
if stats.q_min < self.epoch_q_min { self.epoch_q_min = stats.q_min; }
|
||||
if stats.q_max > self.epoch_q_max { self.epoch_q_max = stats.q_max; }
|
||||
}
|
||||
}
|
||||
|
||||
if train_step_count > 0 {
|
||||
info!(
|
||||
"Training step breakdown ({} steps): sample={:.0}ms fused={:.0}ms guard={:.0}ms (per-step: sample={:.1}ms fused={:.1}ms guard={:.1}ms)",
|
||||
train_step_count,
|
||||
sample_total_us as f64 / 1000.0,
|
||||
fused_total_us as f64 / 1000.0,
|
||||
guard_total_us as f64 / 1000.0,
|
||||
sample_total_us as f64 / 1000.0 / train_step_count as f64,
|
||||
fused_total_us as f64 / 1000.0 / train_step_count as f64,
|
||||
guard_total_us as f64 / 1000.0 / train_step_count as f64,
|
||||
);
|
||||
}
|
||||
Ok(train_step_count)
|
||||
}
|
||||
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
// Helper: Epoch boundary readback + safety checks
|
||||
// ═══════════════════════════════════════════════════════════════════════
|
||||
|
||||
@@ -438,6 +438,22 @@ impl PpoTrainer {
|
||||
self.train_gpu(market_data, progress_callback).await
|
||||
}
|
||||
|
||||
/// Train from contiguous feature slices (fxcache format).
|
||||
/// Converts to Vec<Vec<f32>> internally — PPO's train() API requires it.
|
||||
pub async fn train_from_slices<F>(
|
||||
&self,
|
||||
features: &[[f64; 42]],
|
||||
progress_callback: F,
|
||||
) -> Result<PpoTrainingMetrics, MLError>
|
||||
where
|
||||
F: FnMut(PpoTrainingMetrics) + Send,
|
||||
{
|
||||
let market_data: Vec<Vec<f32>> = features.iter()
|
||||
.map(|f| f.iter().map(|&v| v as f32).collect())
|
||||
.collect();
|
||||
self.train(market_data, progress_callback).await
|
||||
}
|
||||
|
||||
async fn train_gpu<F>(
|
||||
&self,
|
||||
market_data: Vec<Vec<f32>>,
|
||||
|
||||
@@ -525,6 +525,94 @@ pub fn generate_walk_forward_indices(
|
||||
ranges
|
||||
}
|
||||
|
||||
/// Generate walk-forward folds as index ranges from nanosecond timestamps.
|
||||
///
|
||||
/// Same date-boundary logic as [`generate_walk_forward_indices`] but operates
|
||||
/// on raw `i64` timestamps (nanoseconds since Unix epoch) from fxcache data
|
||||
/// instead of `OHLCVBar`. Difficulty score is always 0.0 because we don't
|
||||
/// have OHLC data to compute ADX.
|
||||
pub fn generate_walk_forward_indices_from_timestamps(
|
||||
timestamps: &[i64],
|
||||
config: &WalkForwardConfig,
|
||||
) -> Vec<FoldRange> {
|
||||
if timestamps.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
|
||||
let first_ts = timestamps[0];
|
||||
let last_ts = timestamps[timestamps.len() - 1];
|
||||
|
||||
let data_start = chrono::DateTime::from_timestamp_nanos(first_ts).date_naive();
|
||||
let data_end = chrono::DateTime::from_timestamp_nanos(last_ts).date_naive();
|
||||
|
||||
let mut ranges = Vec::new();
|
||||
let mut fold = 0_usize;
|
||||
|
||||
loop {
|
||||
let total_train_months = config
|
||||
.initial_train_months
|
||||
.saturating_add(fold as u32 * config.step_months);
|
||||
let train_end = match data_start.checked_add_months(Months::new(total_train_months)) {
|
||||
Some(d) => d,
|
||||
None => break,
|
||||
};
|
||||
let val_end = match train_end.checked_add_months(Months::new(config.val_months)) {
|
||||
Some(d) => d,
|
||||
None => break,
|
||||
};
|
||||
let test_end = match val_end.checked_add_months(Months::new(config.test_months)) {
|
||||
Some(d) => d,
|
||||
None => break,
|
||||
};
|
||||
|
||||
if test_end > data_end + chrono::Duration::days(1) {
|
||||
break;
|
||||
}
|
||||
|
||||
// O(log n) index lookups via partition_point (timestamps are sorted)
|
||||
let train_end_idx = timestamps.partition_point(|&ts| {
|
||||
chrono::DateTime::from_timestamp_nanos(ts).date_naive() < train_end
|
||||
});
|
||||
let val_end_idx = timestamps.partition_point(|&ts| {
|
||||
chrono::DateTime::from_timestamp_nanos(ts).date_naive() < val_end
|
||||
});
|
||||
let test_end_idx = timestamps.partition_point(|&ts| {
|
||||
chrono::DateTime::from_timestamp_nanos(ts).date_naive() < test_end
|
||||
});
|
||||
|
||||
let train_start = 0; // expanding window: always starts at beginning
|
||||
let val_start = train_end_idx;
|
||||
let val_end_bound = val_end_idx;
|
||||
|
||||
// Skip folds where any split is too small
|
||||
let train_len = train_end_idx - train_start;
|
||||
let val_len = val_end_bound - val_start;
|
||||
let test_len = test_end_idx - val_end_bound;
|
||||
|
||||
if train_len < MIN_BARS_PER_SPLIT || val_len < MIN_BARS_PER_SPLIT || test_len < MIN_BARS_PER_SPLIT {
|
||||
if val_len < MIN_BARS_PER_SPLIT || test_len < MIN_BARS_PER_SPLIT {
|
||||
if fold > 0 { break; }
|
||||
}
|
||||
fold = fold.saturating_add(1);
|
||||
continue;
|
||||
}
|
||||
|
||||
// No ADX computation — fxcache doesn't have OHLC bars for difficulty
|
||||
ranges.push(FoldRange {
|
||||
fold,
|
||||
train_start,
|
||||
train_end: train_end_idx,
|
||||
val_start,
|
||||
val_end: val_end_bound,
|
||||
difficulty_score: 0.0,
|
||||
});
|
||||
|
||||
fold = fold.saturating_add(1);
|
||||
}
|
||||
|
||||
ranges
|
||||
}
|
||||
|
||||
/// Per-feature normalization statistics (mean and standard deviation).
|
||||
///
|
||||
/// Computed from training data and applied to val/test data to prevent
|
||||
|
||||
Reference in New Issue
Block a user