diff --git a/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs b/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs index 0168bcbbe..29d889364 100644 --- a/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs +++ b/crates/ml/src/cuda_pipeline/gpu_dqn_trainer.rs @@ -383,6 +383,14 @@ impl GpuDqnTrainer { let b = config.batch_size; let total_params = compute_total_params(&config); + // Disable cudarc's automatic event tracking on this stream permanently. + // The forked stream is single-owner (GpuDqnTrainer), we control all + // ordering via explicit synchronize() calls, and cudarc's sync manager + // causes CUDA_ERROR_INVALID_VALUE from stale Drop-recorded events when + // mixed with CUDA Graph capture/replay. Safe: single stream, no cross- + // stream dependencies, all sync points are explicit. + unsafe { stream.context().disable_event_tracking(); } + // ── Compile all 5 training kernels from same module ────────── let (forward_loss_kernel, backward_kernel, grad_norm_kernel, adam_update_kernel, f32_to_bf16_kernel) = compile_training_kernels(&stream, &config)?; @@ -1013,16 +1021,7 @@ impl GpuDqnTrainer { self.stream.synchronize() .map_err(|e| MLError::ModelError(format!("stream sync before capture: {e}")))?; - // Disable cudarc's automatic event tracking during graph capture. - // After fork(), cudarc enters multi-stream mode and injects - // cuStreamWaitEvent/cuEventRecord into every buffer access (memset_zeros, - // launch_builder.arg, launch). These cross-stream event references - // invalidate CUDA graph capture. Safe to disable because: - // 1. stream.synchronize() above ensures all pending work is complete - // 2. Only one stream is active during capture (no cross-stream hazard) - // 3. No new CudaSlice allocations during capture (all buffers pre-allocated) - // SAFETY: single-stream capture, all work synchronized, re-enabled below. - unsafe { self.stream.context().disable_event_tracking(); } + // Event tracking already disabled globally in GpuDqnTrainer::new(). // Begin stream capture — only work submitted from this thread on this // stream is captured (THREAD_LOCAL mode, safe for single-stream use). @@ -1030,8 +1029,6 @@ impl GpuDqnTrainer { cudarc::driver::sys::CUstreamCaptureMode::CU_STREAM_CAPTURE_MODE_THREAD_LOCAL, ); if let Err(e) = begin_result { - // Re-enable event tracking before propagating error - unsafe { self.stream.context().enable_event_tracking(); } return Err(MLError::ModelError(format!("CUDA graph begin_capture: {e}"))); } @@ -1045,11 +1042,6 @@ impl GpuDqnTrainer { cudarc::driver::sys::CUgraphInstantiate_flags::CUDA_GRAPH_INSTANTIATE_FLAG_AUTO_FREE_ON_LAUNCH, ); - // Re-enable event tracking now that capture is complete. - // SAFETY: restores normal multi-stream synchronization for all - // subsequent buffer accesses (readback, next train_step, etc.). - unsafe { self.stream.context().enable_event_tracking(); } - // Propagate submission error first submit_result?; @@ -1668,12 +1660,17 @@ impl GpuDqnTrainer { &mut self, online_d: &DuelingWeightSet, online_b: &BranchingWeightSet, - target_d: &DuelingWeightSet, - target_b: &BranchingWeightSet, + target_d: &mut DuelingWeightSet, + target_b: &mut BranchingWeightSet, tau: f32, ) -> Result<(), MLError> { let sizes = compute_param_sizes(&self.config); + // Synchronize stream: ensures CUDA Graph execution is fully complete + // before EMA kernels modify target weights. + self.stream.synchronize() + .map_err(|e| MLError::ModelError(format!("EMA pre-sync: {e}")))?; + // Paired (target, online) slices in GOFF_* order (20 pairs) let pairs: [(&CudaSlice, &CudaSlice); 20] = [ (&target_d.w_s1, &online_d.w_s1), @@ -1684,15 +1681,15 @@ impl GpuDqnTrainer { (&target_d.b_v1, &online_d.b_v1), (&target_d.w_v2, &online_d.w_v2), (&target_d.b_v2, &online_d.b_v2), - (&target_d.w_a1, &online_d.w_a1), // branch 0 (exposure) + (&target_d.w_a1, &online_d.w_a1), (&target_d.b_a1, &online_d.b_a1), (&target_d.w_a2, &online_d.w_a2), (&target_d.b_a2, &online_d.b_a2), - (&target_b.w_bo1, &online_b.w_bo1), // branch 1 (order) + (&target_b.w_bo1, &online_b.w_bo1), (&target_b.b_bo1, &online_b.b_bo1), (&target_b.w_bo2, &online_b.w_bo2), (&target_b.b_bo2, &online_b.b_bo2), - (&target_b.w_bu1, &online_b.w_bu1), // branch 2 (urgency) + (&target_b.w_bu1, &online_b.w_bu1), (&target_b.b_bu1, &online_b.b_bu1), (&target_b.w_bu2, &online_b.w_bu2), (&target_b.b_bu2, &online_b.b_bu2), @@ -1700,6 +1697,9 @@ impl GpuDqnTrainer { for (i, (target_slice, online_slice)) in pairs.iter().enumerate() { let n = sizes[i] as i32; + if sizes[i] == 0 { + continue; + } let blocks = ((sizes[i] + 255) / 256) as u32; let launch_cfg = LaunchConfig { grid_dim: (blocks, 1, 1), @@ -1707,9 +1707,10 @@ impl GpuDqnTrainer { shared_mem_bytes: 0, }; - // Safety: argument order matches the extern "C" dqn_ema_kernel signature. - // target and online CudaSlice buffers have size >= sizes[i]. - // Both belong to the same CUDA context as self.stream. + // Safety: argument order matches dqn_ema_kernel(float* target, const float* online, float tau, int n). + // target and online CudaSlice buffers have size >= sizes[i]. + // Both allocated on the same forked stream. Stream synchronized above + // to ensure CUDA Graph execution is complete. unsafe { self.stream .launch_builder(&self.ema_kernel) diff --git a/crates/ml/src/trainers/dqn/data_loading.rs b/crates/ml/src/trainers/dqn/data_loading.rs index 6327bf170..ae8c802ad 100644 --- a/crates/ml/src/trainers/dqn/data_loading.rs +++ b/crates/ml/src/trainers/dqn/data_loading.rs @@ -886,13 +886,26 @@ impl DQNTrainer { let mut ohlcv_bars = Vec::new(); - // Read metadata (for logging) + // Read metadata — reject non-OHLCV schemas at file level let metadata = decoder.metadata(); debug!( "DBN file metadata: dataset={:?}, schema={:?}, symbols={:?}", metadata.dataset, metadata.schema, metadata.symbols ); + // Reject files with non-OHLCV schema upfront (MBP-10, trades, etc.) + if let Some(schema) = metadata.schema { + let schema_str = format!("{schema:?}"); + if !schema_str.contains("Ohlcv") { + tracing::warn!( + schema = %schema_str, + file = %file_path.display(), + "Skipping non-OHLCV DBN file" + ); + return Ok(Vec::new()); + } + } + // Decode all OHLCV records let mut ohlcv_count = 0; let mut other_count = 0; @@ -903,13 +916,13 @@ impl DQNTrainer { Ok(Some(record)) => { idx += 1; - // Convert RecordRef to RecordRefEnum for pattern matching. - // Skip records with unknown RType (e.g. MBP-10 files mixed - // into an OHLCV directory) instead of hard-erroring. - let record_enum = match record.as_enum() { - Ok(e) => e, - Err(_) => continue, // non-OHLCV record type — skip - }; + // Convert RecordRef to RecordRefEnum for pattern matching + let record_enum = record + .as_enum() + .map_err(|e| anyhow::anyhow!( + "Corrupt record at idx {} in {}: {}", + idx, file_path.display(), e + ))?; match record_enum { dbn::RecordRefEnum::Ohlcv(ohlcv) => { @@ -997,7 +1010,10 @@ impl DQNTrainer { break; }, Err(e) => { - return Err(anyhow::anyhow!("Failed to decode record {}: {}", idx, e)); + return Err(anyhow::anyhow!( + "Failed to decode record {} in {}: {}", + idx, file_path.display(), e + )); }, } } diff --git a/crates/ml/src/trainers/dqn/fused_training.rs b/crates/ml/src/trainers/dqn/fused_training.rs index c51ae4289..49d6ad204 100644 --- a/crates/ml/src/trainers/dqn/fused_training.rs +++ b/crates/ml/src/trainers/dqn/fused_training.rs @@ -348,7 +348,7 @@ impl FusedTrainingCtx { self.trainer.target_ema_update( &self.online_dueling, &self.online_branching, - &self.target_dueling, &self.target_branching, + &mut self.target_dueling, &mut self.target_branching, tau as f32, ).map_err(|e| anyhow::anyhow!("GPU EMA target update: {e}"))?; } diff --git a/crates/ml/src/trainers/dqn/smoke_tests/helpers.rs b/crates/ml/src/trainers/dqn/smoke_tests/helpers.rs index 8da74b395..5cb39cb11 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/helpers.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/helpers.rs @@ -88,23 +88,26 @@ pub(super) fn assert_finite_f32(val: f32, name: &str) { assert!(val.is_finite(), "{name} is not finite: {val}"); } -/// Resolve the test data directory from `FOXHUNT_TEST_DATA` env var, -/// falling back to the repo-relative `test_data/` path. +/// Resolve single-symbol test data directory. /// -/// Returns `None` if the directory doesn't exist (CI builders without data). +/// Uses `FOXHUNT_TEST_DATA` env var if set, otherwise auto-detects +/// `test_data/ES.FUT` relative to the workspace root. +/// Returns `None` if the directory doesn't exist. pub(super) fn test_data_dir() -> Option { - let dir = std::env::var("FOXHUNT_TEST_DATA") - .unwrap_or_else(|_| { - // Workspace root is 4 levels up from smoke_tests/ - let manifest = std::env::var("CARGO_MANIFEST_DIR").unwrap_or_default(); - let workspace = std::path::Path::new(&manifest) - .parent() // crates/ - .and_then(|p| p.parent()) // repo root - .unwrap_or(std::path::Path::new("../..")); - workspace.join("test_data").to_string_lossy().into_owned() - }); - if std::path::Path::new(&dir).exists() { - Some(dir) + if let Ok(dir) = std::env::var("FOXHUNT_TEST_DATA") { + if std::path::Path::new(&dir).exists() { + return Some(dir); + } + } + // Auto-detect: workspace_root/test_data/ES.FUT (single symbol, no cross-symbol mixing) + let manifest = std::env::var("CARGO_MANIFEST_DIR").unwrap_or_default(); + let workspace = std::path::Path::new(&manifest) + .parent() // crates/ + .and_then(|p| p.parent()) // repo root + .unwrap_or(std::path::Path::new("../..")); + let es_dir = workspace.join("test_data").join("ES.FUT"); + if es_dir.exists() { + Some(es_dir.to_string_lossy().into_owned()) } else { None } diff --git a/test_data/ES.FUT b/test_data/ES.FUT new file mode 120000 index 000000000..6179c0ebb --- /dev/null +++ b/test_data/ES.FUT @@ -0,0 +1 @@ +/home/jgrusewski/Work/foxhunt/test_data/ES.FUT \ No newline at end of file diff --git a/test_data/test_data b/test_data/test_data new file mode 120000 index 000000000..4c186e8af --- /dev/null +++ b/test_data/test_data @@ -0,0 +1 @@ +/home/jgrusewski/Work/foxhunt/test_data \ No newline at end of file