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

fix(gpu): wire PPO GPU experience collection, fix debug log levels, increase hyperopt deadline

Browse Source 
- Wire GPU experience collection into PPO hyperopt adapter with CPU fallback
  (ensure_gpu_data, gpu_collect_trajectories, batch-to-trajectory converter)
- Downgrade 6 ERROR-level debug logs to debug!() in DQN hyperopt adapter
- Increase hyperopt activeDeadlineSeconds from 1h to 6h (job-template + train.sh)
- Mark 3 data-dependent real_data_loader tests as #[ignore]
- Fix 4 clippy map_or → is_none_or warnings in trading_service

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-02-28 18:28:15 +01:00
parent 64d57e7454
commit cd8a45d8a8
7 changed files with 327 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
crates/ml/src/hyperopt/adapters/dqn.rs
View File

@@ -2083,7 +2083,7 @@ impl HyperparameterOptimizable for DQNTrainer {
type Metrics = DQNMetrics;
fn train_with_params(&mut self, params: Self::Params) -> Result<Self::Metrics, MLError> {
tracing::error!("DEBUG: train_with_params called");
tracing::debug!("train_with_params called");
// START: Add trial timing
let trial_start = std::time::Instant::now();
@@ -2490,8 +2490,8 @@ impl HyperparameterOptimizable for DQNTrainer {
},
};
tracing::error!("DEBUG: Training completed successfully, extracting metrics");
tracing::error!("DEBUG: Checking validation data - internal_trainer.get_val_data().len() = {}", internal_trainer.get_val_data().len());
tracing::debug!("Training completed successfully, extracting metrics");
tracing::debug!("Checking validation data - internal_trainer.get_val_data().len() = {}", internal_trainer.get_val_data().len());
// Extract metrics from TrainingMetrics struct
// Note: TrainingMetrics.loss is a single f64, not a Vec
@@ -2576,12 +2576,12 @@ impl HyperparameterOptimizable for DQNTrainer {
.copied()
.unwrap_or(0.0);
tracing::error!("DEBUG: Reached backtest decision point");
tracing::error!("DEBUG: self.enable_backtest = {}", self.enable_backtest);
tracing::debug!("Reached backtest decision point");
tracing::debug!("enable_backtest = {}", self.enable_backtest);
// Run backtest if enabled
let backtest_metrics = if self.enable_backtest {
tracing::error!("DEBUG: Backtest is enabled, starting backtest...");
tracing::debug!("Backtest is enabled, starting backtest...");
tracing::info!("Running backtest on validation data...");
// Get validation data from trainer

315
crates/ml/src/hyperopt/adapters/ppo.rs
View File

@@ -207,7 +207,6 @@ pub struct PPOMetrics {
/// - Clip epsilon
/// - Value loss coefficient
/// - Entropy coefficient
#[derive(Debug, Clone)]
pub struct PPOTrainer {
dbn_data_dir: std::path::PathBuf,
episodes: usize,
@@ -226,6 +225,55 @@ pub struct PPOTrainer {
tick_size: f64,
/// Spread width in ticks (default 1.0)
spread_ticks: f64,
/// GPU market features [num_bars * 51] — uploaded once, reused across trials
#[cfg(feature = "cuda")]
features_cuda: Option<candle_core::cuda_backend::cudarc::driver::CudaSlice<f32>>,
/// GPU market targets [num_bars * 4] — uploaded once, reused across trials
#[cfg(feature = "cuda")]
targets_cuda: Option<candle_core::cuda_backend::cudarc::driver::CudaSlice<f32>>,
/// GPU PPO experience collector — initialized on first trial, weights synced per trial
#[cfg(feature = "cuda")]
gpu_collector: Option<crate::cuda_pipeline::gpu_ppo_collector::GpuPpoExperienceCollector>,
/// Number of bars in the uploaded market data
#[cfg(feature = "cuda")]
gpu_num_bars: usize,
}
impl std::fmt::Debug for PPOTrainer {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("PPOTrainer")
.field("dbn_data_dir", &self.dbn_data_dir)
.field("episodes", &self.episodes)
.field("device", &self.device)
.field("early_stopping_patience", &self.early_stopping_patience)
.finish_non_exhaustive()
}
}
impl Clone for PPOTrainer {
fn clone(&self) -> Self {
Self {
dbn_data_dir: self.dbn_data_dir.clone(),
episodes: self.episodes,
device: self.device.clone(),
training_paths: self.training_paths.clone(),
early_stopping_patience: self.early_stopping_patience,
early_stopping_min_epochs: self.early_stopping_min_epochs,
trial_counter: self.trial_counter.clone(),
tx_cost_bps: self.tx_cost_bps,
tick_size: self.tick_size,
spread_ticks: self.spread_ticks,
// GPU resources are not cloned — each clone re-initialises on first use
#[cfg(feature = "cuda")]
features_cuda: None,
#[cfg(feature = "cuda")]
targets_cuda: None,
#[cfg(feature = "cuda")]
gpu_collector: None,
#[cfg(feature = "cuda")]
gpu_num_bars: 0,
}
}
}
impl PPOTrainer {
@@ -282,6 +330,14 @@ impl PPOTrainer {
tx_cost_bps: 1.0, // 1 bps commission (ES futures typical)
tick_size: 0.25, // ES futures tick size
spread_ticks: 1.0, // 1 tick spread (ES typical)
#[cfg(feature = "cuda")]
features_cuda: None,
#[cfg(feature = "cuda")]
targets_cuda: None,
#[cfg(feature = "cuda")]
gpu_collector: None,
#[cfg(feature = "cuda")]
gpu_num_bars: 0,
})
}
@@ -336,6 +392,235 @@ impl PPOTrainer {
self.training_paths = paths;
self
}
/// Upload training data to GPU and initialize the experience collector.
///
/// Called once at the start of the first trial. Subsequent trials reuse the
/// uploaded data and call `sync_weights()` on the collector.
#[cfg(feature = "cuda")]
fn ensure_gpu_data(
&mut self,
training_data: &[([f32; 51], f64)],
ppo_agent: &PPO,
) {
use tracing::debug;
// Data already uploaded — just sync weights for the new trial's model
if self.features_cuda.is_some() && self.targets_cuda.is_some() {
if let Some(ref mut collector) = self.gpu_collector {
if let Err(e) = collector.sync_weights(
ppo_agent.actor.vars(),
ppo_agent.critic.vars(),
) {
warn!("PPO GPU weight sync failed: {e}");
} else {
debug!("PPO GPU weights synced for new trial");
}
}
return;
}
let cuda_dev = match &self.device {
candle_core::Device::Cuda(d) => d,
candle_core::Device::Cpu | candle_core::Device::Metal(_) => return,
};
let stream = cuda_dev.cuda_stream();
let num_bars = training_data.len();
// Upload features [num_bars * 51]
let mut flat_features = Vec::with_capacity(num_bars * 51);
for (features, _) in training_data {
flat_features.extend_from_slice(features);
}
match stream.memcpy_stod(&flat_features) {
Ok(buf) => {
info!("PPO CUDA features uploaded: {} bars × 51 ({:.1} MB)",
num_bars, (num_bars * 51 * 4) as f64 / 1_048_576.0);
self.features_cuda = Some(buf);
}
Err(e) => {
warn!("PPO CUDA features upload failed: {e}");
return;
}
}
// Upload targets [num_bars * 4]: kernel expects [close, next_close, close_raw, next_close_raw]
let mut flat_targets = Vec::with_capacity(num_bars * 4);
for (i, (_, price)) in training_data.iter().enumerate() {
let close = *price as f32;
let next_close = training_data
.get(i + 1)
.map(|(_, p)| *p as f32)
.unwrap_or(close);
flat_targets.push(close);
flat_targets.push(next_close);
flat_targets.push(close);
flat_targets.push(next_close);
}
match stream.memcpy_stod(&flat_targets) {
Ok(buf) => {
info!("PPO CUDA targets uploaded: {} bars × 4 ({:.1} MB)",
num_bars, (num_bars * 4 * 4) as f64 / 1_048_576.0);
self.targets_cuda = Some(buf);
}
Err(e) => {
warn!("PPO CUDA targets upload failed: {e}");
self.features_cuda = None; // Both must succeed
return;
}
}
self.gpu_num_bars = num_bars;
// Initialize GPU collector with current model's weights
if self.gpu_collector.is_none() {
let actor_vars = ppo_agent.actor.vars();
let critic_vars = ppo_agent.critic.vars();
match crate::cuda_pipeline::gpu_ppo_collector::GpuPpoExperienceCollector::new(
stream,
actor_vars,
critic_vars,
&candle_nn::VarMap::new(), // curiosity placeholder
35_000.0, // initial_capital
self.tick_size as f32 * self.spread_ticks as f32, // avg_spread
0.20, // cash_reserve_pct
) {
Ok(collector) => {
info!("PPO GPU experience collector initialized (zero-roundtrip CUDA kernel)");
self.gpu_collector = Some(collector);
}
Err(e) => {
warn!("PPO GPU collector init failed, using CPU fallback: {e}");
}
}
} else {
// Sync weights from new PPO model
if let Some(ref mut collector) = self.gpu_collector {
if let Err(e) = collector.sync_weights(
ppo_agent.actor.vars(),
ppo_agent.critic.vars(),
) {
warn!("PPO GPU weight sync failed: {e}");
}
debug!("PPO GPU weights synced for new trial");
}
}
}
/// Collect experiences using GPU kernel, returning a TrajectoryBatch.
///
/// Returns None if GPU collection is unavailable or fails (caller should
/// fall back to CPU trajectory generation).
#[cfg(feature = "cuda")]
fn gpu_collect_trajectories(
&mut self,
num_episodes: usize,
) -> Option<TrajectoryBatch> {
use crate::cuda_pipeline::gpu_ppo_collector::PpoCollectorConfig;
let collector = self.gpu_collector.as_mut()?;
let features_buf = self.features_cuda.as_ref()?;
let targets_buf = self.targets_cuda.as_ref()?;
let total_bars = self.gpu_num_bars;
let n_episodes = (num_episodes as i32).min(128);
let timesteps = 100_i32; // Match CPU episode length
let usable_bars = (total_bars as i32 - timesteps).max(1);
let stride = (usable_bars / n_episodes).max(1);
let episode_starts: Vec<i32> = (0..n_episodes)
.map(|i| (i * stride).rem_euclid(usable_bars))
.collect();
let config = PpoCollectorConfig {
n_episodes,
timesteps_per_episode: timesteps,
total_bars: total_bars as i32,
gamma: 0.99,
gae_lambda: 0.95,
..Default::default()
};
if let Err(e) = collector.reset_episodes(35_000.0, self.tick_size as f32 * self.spread_ticks as f32, 0.20) {
warn!("PPO GPU reset_episodes failed: {e}");
return None;
}
match collector.collect_experiences(features_buf, targets_buf, &episode_starts, &config) {
Ok(batch) => {
info!("GPU PPO collected {} experiences ({} episodes × {} timesteps)",
batch.n_episodes * batch.timesteps, batch.n_episodes, batch.timesteps);
Some(gpu_ppo_batch_to_trajectory_batch(&batch))
}
Err(e) => {
warn!("PPO GPU experience collection failed, falling back to CPU: {e}");
None
}
}
}
}
/// Convert a GPU PPO experience batch to a TrajectoryBatch for `PPO::update()`.
///
/// The GPU kernel produces 54-dim states (51 features + 3 portfolio state) but
/// the PPO model in hyperopt uses state_dim=51. We truncate to match.
#[cfg(feature = "cuda")]
fn gpu_ppo_batch_to_trajectory_batch(
batch: &crate::cuda_pipeline::gpu_ppo_collector::PpoExperienceBatch,
) -> TrajectoryBatch {
let kernel_state_dim = 54; // GPU kernel output: 51 features + 3 portfolio state
let model_state_dim = 51; // PPO hyperopt model state_dim
let total = batch.n_episodes * batch.timesteps;
// Truncate states from 54 → 51 dims (strip portfolio state appended by kernel)
let states: Vec<Vec<f32>> = batch
.states
.chunks(kernel_state_dim)
.take(total)
.map(|c| c.get(..model_state_dim).unwrap_or(c).to_vec())
.collect();
let actions: Vec<TradingAction> = batch
.actions
.iter()
.take(total)
.map(|&a| match a.clamp(0, 2) {
0 => TradingAction::Buy,
1 => TradingAction::Sell,
_ => TradingAction::Hold,
})
.collect();
let log_probs: Vec<f32> = batch.log_probs.iter().take(total).copied().collect();
// V(s) = R(s) - A(s)
let values: Vec<f32> = batch
.returns
.iter()
.zip(&batch.advantages)
.take(total)
.map(|(r, a)| r - a)
.collect();
let dones: Vec<bool> = batch
.done_flags
.iter()
.take(total)
.map(|&d| d != 0)
.collect();
let rewards = vec![0.0_f32; states.len()];
TrajectoryBatch {
trajectories: vec![],
states,
actions,
log_probs,
values,
rewards,
dones,
advantages: batch.advantages.iter().take(total).copied().collect(),
returns: batch.returns.iter().take(total).copied().collect(),
}
}
/// Write a log entry to the training log file
@@ -515,18 +800,34 @@ impl HyperparameterOptimizable for PPOTrainer {
num_val
);
// Upload raw market data to GPU and init/sync collector
// Uses full dataset (kernel indexes by bar offset). First trial compiles
// the CUDA kernel + uploads data; subsequent trials just sync weights.
#[cfg(feature = "cuda")]
self.ensure_gpu_data(&training_data, &ppo_agent);
// Batch episodes: min(64, num_train) to avoid zero batches with small episode counts
let batch_episodes = 64.min(num_train);
let num_batches = (num_train + batch_episodes - 1) / batch_episodes; // ceil division
for _batch_idx in 0..num_batches {
let episodes_this_batch = batch_episodes.min(num_train - _batch_idx * batch_episodes);
// Generate trajectories from real market data using current policy
let mut trajectory_batch = self
.generate_trajectories_from_data(&ppo_agent, train_data, episodes_this_batch)
.map_err(|e| {
MLError::TrainingError(format!("Failed to generate trajectories: {}", e))
})?;
// GPU-first experience collection with CPU fallback
#[cfg(feature = "cuda")]
let gpu_batch = self.gpu_collect_trajectories(episodes_this_batch);
#[cfg(not(feature = "cuda"))]
let gpu_batch: Option<TrajectoryBatch> = None;
let mut trajectory_batch = if let Some(batch) = gpu_batch {
batch
} else {
// CPU fallback: generate trajectories from real market data
self.generate_trajectories_from_data(&ppo_agent, train_data, episodes_this_batch)
.map_err(|e| {
MLError::TrainingError(format!("Failed to generate trajectories: {}", e))
})?
};
// Update PPO with trajectory batch
let (policy_loss, value_loss) = ppo_agent

3
crates/ml/src/real_data_loader.rs
View File

@@ -616,6 +616,7 @@ mod tests {
use super::*;
#[tokio::test]
#[ignore] // Requires local DBN data files
async fn test_load_symbol_data() -> Result<()> {
let mut loader = RealDataLoader::new_from_workspace()?;
@@ -638,6 +639,7 @@ mod tests {
}
#[tokio::test]
#[ignore] // Requires local DBN data files
async fn test_extract_features() -> Result<()> {
let mut loader = RealDataLoader::new_from_workspace()?;
let bars = loader.load_symbol_data("ZN.FUT").await?;
@@ -663,6 +665,7 @@ mod tests {
}
#[tokio::test]
#[ignore] // Requires local DBN data files
async fn test_calculate_indicators() -> Result<()> {
let mut loader = RealDataLoader::new_from_workspace()?;
let bars = loader.load_symbol_data("ZN.FUT").await?;

2
infra/k8s/training/job-template.yaml
View File

@@ -9,7 +9,7 @@ metadata:
foxhunt/job-type: training
spec:
backoffLimit: 1
activeDeadlineSeconds: 3600
activeDeadlineSeconds: 21600 # 6 hours — hyperopt runs 20 trials × 8 epochs
ttlSecondsAfterFinished: 600
template:
metadata:

6
infra/scripts/train.sh
View File

@@ -19,7 +19,7 @@ set -euo pipefail
SYMBOL="ES.FUT"
TRIALS=20
MAX_STEPS=2000
TIMEOUT=3600
TIMEOUT=3600 # default; hyperopt preset overrides to 6h below
DATA_DIR="/data/futures-baseline"
NAMESPACE="foxhunt"
IMAGE="rg.fr-par.scw.cloud/foxhunt/training:latest"
@@ -114,6 +114,10 @@ case "$PRESET" in
hyperopt)
[[ -z "$MODEL" ]] && die "hyperopt preset requires --model"
validate_model "$MODEL"
# Hyperopt runs 20 trials × 8 epochs — needs 6h unless user specified --timeout
if [[ "$TIMEOUT" -eq 3600 ]]; then
TIMEOUT=21600
fi
;;
evaluate)
[[ -z "$RUN_ID" ]] && die "evaluate preset requires --run-id (from a previous training run)"

4
services/trading_service/src/core/order_manager.rs
View File

@@ -497,7 +497,7 @@ impl OrderManager {
if is_match {
let match_quality = self.calculate_match_quality(price, entry_price, side);
if best_match.map_or(true, |(_, qual)| match_quality > qual) {
if best_match.is_none_or(|(_, qual)| match_quality > qual) {
best_match = Some((i, match_quality));
}
}
@@ -516,7 +516,7 @@ impl OrderManager {
if is_match {
let match_quality = self.calculate_match_quality(price, entry.price, side);
if best_match.map_or(true, |(_, qual)| match_quality > qual) {
if best_match.is_none_or(|(_, qual)| match_quality > qual) {
best_match = Some((i, match_quality));
}
}

4
services/trading_service/src/services/risk.rs
View File

@@ -594,12 +594,12 @@ impl RiskService for RiskServiceImpl {
.filter(|p| {
req.symbol
.as_ref()
.map_or(true, |s| s.is_empty() || p.symbol == *s)
.is_none_or(|s| s.is_empty() || p.symbol == *s)
})
.filter(|p| {
req.account_id
.as_ref()
.map_or(true, |a| a.is_empty() || p.account_id == *a)
.is_none_or(|a| a.is_empty() || p.account_id == *a)
})
.collect();