From 7744becdd5516f1541643b2d49fd9b0d6b52f2ef Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sun, 1 Mar 2026 15:56:43 +0100 Subject: [PATCH] feat(training): move metrics to common::metrics, fix all clippy errors Move Prometheus training metrics from example-local baseline_common/ to common::metrics::{server,training_metrics} following the existing grpc_metrics.rs pattern. Fix 29 let_underscore_must_use clippy errors in push_metrics.rs, 3 shadow lint errors in training binaries, and demote gradient clipping log from warn to debug. Co-Authored-By: Claude Opus 4.6 --- crates/common/src/metrics/mod.rs | 2 + crates/common/src/metrics/server.rs | 62 +++++ crates/common/src/metrics/training_metrics.rs | 235 ++++++++++++++++ crates/ml/examples/baseline_common/metrics.rs | 258 ------------------ crates/ml/examples/baseline_common/mod.rs | 2 - crates/ml/examples/hyperopt_baseline_rl.rs | 16 +- .../examples/hyperopt_baseline_supervised.rs | 12 +- crates/ml/examples/train_baseline_rl.rs | 16 +- .../ml/examples/train_baseline_supervised.rs | 16 +- crates/ml/src/optimizers/adam.rs | 5 +- crates/ml/src/training/push_metrics.rs | 60 ++-- 11 files changed, 360 insertions(+), 324 deletions(-) create mode 100644 crates/common/src/metrics/server.rs create mode 100644 crates/common/src/metrics/training_metrics.rs delete mode 100644 crates/ml/examples/baseline_common/metrics.rs diff --git a/crates/common/src/metrics/mod.rs b/crates/common/src/metrics/mod.rs index 9297aceed..4037efe28 100644 --- a/crates/common/src/metrics/mod.rs +++ b/crates/common/src/metrics/mod.rs @@ -5,6 +5,8 @@ pub mod grpc_metrics; pub mod registry; +pub mod server; +pub mod training_metrics; // Re-export commonly used functions pub use registry::{ diff --git a/crates/common/src/metrics/server.rs b/crates/common/src/metrics/server.rs new file mode 100644 index 000000000..f9d79bc4b --- /dev/null +++ b/crates/common/src/metrics/server.rs @@ -0,0 +1,62 @@ +//! Lightweight Prometheus HTTP metrics server for standalone binaries. +//! +//! Services that already embed Axum/tonic can expose `/metrics` through their +//! existing HTTP stack. Training binaries and CLI tools that don't have an +//! HTTP server use this instead. + +use std::io::{BufRead, BufReader, Write as IoWrite}; +use std::net::TcpListener; + +use super::gather_metrics; + +/// Spawn a background daemon thread serving Prometheus metrics over HTTP. +/// +/// Responds to `GET /metrics` with the global registry output in Prometheus +/// text exposition format. Any other request gets a 404. The thread is +/// detached and dies with the process. +pub fn start_metrics_server(port: u16) { + std::thread::spawn(move || { + let addr = format!("0.0.0.0:{port}"); + let listener = match TcpListener::bind(&addr) { + Ok(l) => { + tracing::info!("Prometheus metrics server listening on {addr}"); + l + } + Err(e) => { + tracing::warn!("Failed to bind metrics server on {addr}: {e}"); + return; + } + }; + + for stream in listener.incoming() { + let Ok(mut stream) = stream else { + continue; + }; + + // Read the HTTP request line (e.g. "GET /metrics HTTP/1.1") + let mut reader = BufReader::new(&stream); + let mut request_line = String::new(); + if reader.read_line(&mut request_line).is_err() { + continue; + } + + let is_metrics = request_line.starts_with("GET /metrics"); + drop(reader); + + if is_metrics { + let body = gather_metrics(); + let response = format!( + "HTTP/1.1 200 OK\r\n\ + Content-Type: text/plain; version=0.0.4\r\n\ + Content-Length: {}\r\n\r\n\ + {}", + body.len(), + body, + ); + _ = stream.write_all(response.as_bytes()); + } else { + _ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n"); + } + } + }); +} diff --git a/crates/common/src/metrics/training_metrics.rs b/crates/common/src/metrics/training_metrics.rs new file mode 100644 index 000000000..fbcd18488 --- /dev/null +++ b/crates/common/src/metrics/training_metrics.rs @@ -0,0 +1,235 @@ +//! Prometheus metrics for ML training binaries. +//! +//! Registers the 18 metrics expected by the Training Cockpit Grafana dashboard +//! and provides typed helper functions so callers never mis-spell metric names. +//! +//! # Usage +//! +//! ```rust,ignore +//! use common::metrics::training_metrics; +//! use common::metrics::server; +//! +//! training_metrics::init(); +//! server::start_metrics_server(9094); +//! training_metrics::set_active_workers(1.0); +//! // … training loop … +//! training_metrics::set_epoch("kan", "0", 1.0); +//! training_metrics::set_epoch_loss("kan", "0", 0.0042); +//! ``` + +use super::{ + increment_counter_vec, observe_histogram_vec, register_counter_vec, register_gauge, + register_gauge_vec, register_histogram_vec, set_gauge, set_gauge_vec, +}; + +// --------------------------------------------------------------------------- +// Registration +// --------------------------------------------------------------------------- + +/// Register all 18 training metrics with the global Prometheus registry. +/// +/// Safe to call multiple times — the registry silently ignores duplicates. +pub fn init() { + let mf: &[&str] = &["model", "fold"]; + let m: &[&str] = &["model"]; + + // Gauges (model + fold) + _ = register_gauge_vec("foxhunt_training_current_epoch", "Current epoch number", mf); + _ = register_gauge_vec("foxhunt_training_epoch_loss", "Training loss at end of epoch", mf); + _ = register_gauge_vec( + "foxhunt_training_validation_loss", + "Validation loss at end of epoch", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_batches_per_second", + "Training throughput in batches per second", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_batches_processed", + "Total batches processed this run", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_iteration_seconds", + "Last epoch wall time in seconds", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_eval_accuracy", + "Eval accuracy (supervised models)", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_eval_precision", + "Eval precision (supervised models)", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_eval_recall", + "Eval recall (supervised models)", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_eval_f1", + "Eval F1 score (supervised models)", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_checkpoint_size_bytes", + "Size of last checkpoint file in bytes", + mf, + ); + + // Counters (model + fold) + _ = register_counter_vec( + "foxhunt_training_checkpoint_saves_total", + "Successful checkpoint saves", + mf, + ); + _ = register_counter_vec( + "foxhunt_training_checkpoint_failures_total", + "Failed checkpoint saves", + mf, + ); + _ = register_counter_vec( + "foxhunt_training_nan_detected_total", + "NaN loss events detected", + mf, + ); + _ = register_counter_vec( + "foxhunt_training_gradient_explosion_total", + "Gradient explosion events", + mf, + ); + _ = register_counter_vec( + "foxhunt_training_feature_errors_total", + "Feature extraction failures", + mf, + ); + + // Histograms (model only) + _ = register_histogram_vec( + "foxhunt_training_checkpoint_duration_seconds", + "Checkpoint save latency", + m, + vec![0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0], + ); + _ = register_histogram_vec( + "foxhunt_training_data_load_seconds", + "Data loading latency per fold", + m, + vec![0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0], + ); + + // Special gauge (no labels) + _ = register_gauge( + "foxhunt_training_active_workers", + "Number of active training workers", + ); +} + +// --------------------------------------------------------------------------- +// Typed helpers +// --------------------------------------------------------------------------- + +pub fn set_epoch(model: &str, fold: &str, epoch: f64) { + set_gauge_vec("foxhunt_training_current_epoch", &[model, fold], epoch); +} + +pub fn set_epoch_loss(model: &str, fold: &str, loss: f64) { + set_gauge_vec("foxhunt_training_epoch_loss", &[model, fold], loss); +} + +pub fn set_validation_loss(model: &str, fold: &str, loss: f64) { + set_gauge_vec("foxhunt_training_validation_loss", &[model, fold], loss); +} + +pub fn set_batches_per_second(model: &str, fold: &str, bps: f64) { + set_gauge_vec("foxhunt_training_batches_per_second", &[model, fold], bps); +} + +pub fn set_batches_processed(model: &str, fold: &str, total: f64) { + set_gauge_vec("foxhunt_training_batches_processed", &[model, fold], total); +} + +pub fn set_iteration_seconds(model: &str, fold: &str, secs: f64) { + set_gauge_vec("foxhunt_training_iteration_seconds", &[model, fold], secs); +} + +pub fn set_eval_metrics( + model: &str, + fold: &str, + accuracy: f64, + precision: f64, + recall: f64, + f1: f64, +) { + set_gauge_vec("foxhunt_training_eval_accuracy", &[model, fold], accuracy); + set_gauge_vec("foxhunt_training_eval_precision", &[model, fold], precision); + set_gauge_vec("foxhunt_training_eval_recall", &[model, fold], recall); + set_gauge_vec("foxhunt_training_eval_f1", &[model, fold], f1); +} + +pub fn record_checkpoint_save(model: &str, fold: &str, duration_secs: f64, size_bytes: f64) { + increment_counter_vec( + "foxhunt_training_checkpoint_saves_total", + &[model, fold], + 1.0, + ); + set_gauge_vec( + "foxhunt_training_checkpoint_size_bytes", + &[model, fold], + size_bytes, + ); + observe_histogram_vec( + "foxhunt_training_checkpoint_duration_seconds", + &[model], + duration_secs, + ); +} + +pub fn record_checkpoint_failure(model: &str, fold: &str) { + increment_counter_vec( + "foxhunt_training_checkpoint_failures_total", + &[model, fold], + 1.0, + ); +} + +pub fn record_nan_detected(model: &str, fold: &str) { + increment_counter_vec( + "foxhunt_training_nan_detected_total", + &[model, fold], + 1.0, + ); +} + +pub fn record_gradient_explosion(model: &str, fold: &str) { + increment_counter_vec( + "foxhunt_training_gradient_explosion_total", + &[model, fold], + 1.0, + ); +} + +pub fn record_feature_error(model: &str, fold: &str) { + increment_counter_vec( + "foxhunt_training_feature_errors_total", + &[model, fold], + 1.0, + ); +} + +pub fn record_data_load(model: &str, duration_secs: f64) { + observe_histogram_vec( + "foxhunt_training_data_load_seconds", + &[model], + duration_secs, + ); +} + +pub fn set_active_workers(value: f64) { + set_gauge("foxhunt_training_active_workers", value); +} diff --git a/crates/ml/examples/baseline_common/metrics.rs b/crates/ml/examples/baseline_common/metrics.rs deleted file mode 100644 index c4a3ecad3..000000000 --- a/crates/ml/examples/baseline_common/metrics.rs +++ /dev/null @@ -1,258 +0,0 @@ -//! Prometheus metrics for training binaries. -//! -//! Registers 18 dashboard-expected metrics and exposes them via a lightweight -//! HTTP endpoint on `/metrics` (default port 9094, matching K8s annotations). - -use std::io::{BufRead, BufReader, Write as IoWrite}; -use std::net::TcpListener; - -/// Register all 18 training metrics with the global Prometheus registry. -/// -/// Safe to call multiple times — `common::metrics` silently ignores duplicate -/// registrations. -pub fn init_training_metrics() { - use common::metrics::{ - register_counter_vec, register_gauge, register_gauge_vec, register_histogram_vec, - }; - - let mf = &["model", "fold"]; - let m = &["model"]; - - // Gauges (model + fold) - let _ = register_gauge_vec("foxhunt_training_current_epoch", "Current epoch number", mf); - let _ = register_gauge_vec("foxhunt_training_epoch_loss", "Training loss at end of epoch", mf); - let _ = register_gauge_vec( - "foxhunt_training_validation_loss", - "Validation loss at end of epoch", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_batches_per_second", - "Training throughput in batches per second", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_batches_processed", - "Total batches processed this run", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_iteration_seconds", - "Last epoch wall time in seconds", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_eval_accuracy", - "Eval accuracy (supervised models)", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_eval_precision", - "Eval precision (supervised models)", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_eval_recall", - "Eval recall (supervised models)", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_eval_f1", - "Eval F1 score (supervised models)", - mf, - ); - let _ = register_gauge_vec( - "foxhunt_training_checkpoint_size_bytes", - "Size of last checkpoint file in bytes", - mf, - ); - - // Counters (model + fold) - let _ = register_counter_vec( - "foxhunt_training_checkpoint_saves_total", - "Successful checkpoint saves", - mf, - ); - let _ = register_counter_vec( - "foxhunt_training_checkpoint_failures_total", - "Failed checkpoint saves", - mf, - ); - let _ = register_counter_vec( - "foxhunt_training_nan_detected_total", - "NaN loss events detected", - mf, - ); - let _ = register_counter_vec( - "foxhunt_training_gradient_explosion_total", - "Gradient explosion events", - mf, - ); - let _ = register_counter_vec( - "foxhunt_training_feature_errors_total", - "Feature extraction failures", - mf, - ); - - // Histograms (model only) - let _ = register_histogram_vec( - "foxhunt_training_checkpoint_duration_seconds", - "Checkpoint save latency", - m, - vec![0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0], - ); - let _ = register_histogram_vec( - "foxhunt_training_data_load_seconds", - "Data loading latency per fold", - m, - vec![0.5, 1.0, 5.0, 10.0, 30.0, 60.0, 120.0, 300.0], - ); - - // Special gauge (no labels) - let _ = register_gauge("foxhunt_training_active_workers", "Number of active training workers"); -} - -/// Spawn a background daemon thread serving Prometheus metrics over HTTP. -/// -/// Responds to `GET /metrics` with the global registry output. Any other -/// request gets a 404. The thread is detached and dies with the process. -pub fn start_metrics_server(port: u16) { - std::thread::spawn(move || { - let addr = format!("0.0.0.0:{port}"); - let listener = match TcpListener::bind(&addr) { - Ok(l) => { - tracing::info!("Prometheus metrics server listening on {addr}"); - l - } - Err(e) => { - tracing::warn!("Failed to bind metrics server on {addr}: {e}"); - return; - } - }; - - for stream in listener.incoming() { - let Ok(mut stream) = stream else { - continue; - }; - - // Read request line (e.g. "GET /metrics HTTP/1.1") - let mut reader = BufReader::new(&stream); - let mut request_line = String::new(); - if reader.read_line(&mut request_line).is_err() { - continue; - } - - let is_metrics = request_line.starts_with("GET /metrics"); - drop(reader); - - if is_metrics { - let body = common::metrics::gather_metrics(); - let response = format!( - "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4\r\nContent-Length: {}\r\n\r\n{}", - body.len(), - body - ); - let _ = stream.write_all(response.as_bytes()); - } else { - let _ = stream.write_all(b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\n\r\n"); - } - } - }); -} - -// --------------------------------------------------------------------------- -// Helper functions wrapping common::metrics -// --------------------------------------------------------------------------- - -pub fn set_epoch(model: &str, fold: &str, epoch: f64) { - common::metrics::set_gauge_vec("foxhunt_training_current_epoch", &[model, fold], epoch); -} - -pub fn set_epoch_loss(model: &str, fold: &str, loss: f64) { - common::metrics::set_gauge_vec("foxhunt_training_epoch_loss", &[model, fold], loss); -} - -pub fn set_validation_loss(model: &str, fold: &str, loss: f64) { - common::metrics::set_gauge_vec("foxhunt_training_validation_loss", &[model, fold], loss); -} - -pub fn set_batches_per_second(model: &str, fold: &str, bps: f64) { - common::metrics::set_gauge_vec("foxhunt_training_batches_per_second", &[model, fold], bps); -} - -pub fn set_batches_processed(model: &str, fold: &str, total: f64) { - common::metrics::set_gauge_vec("foxhunt_training_batches_processed", &[model, fold], total); -} - -pub fn set_iteration_seconds(model: &str, fold: &str, secs: f64) { - common::metrics::set_gauge_vec("foxhunt_training_iteration_seconds", &[model, fold], secs); -} - -pub fn set_eval_metrics(model: &str, fold: &str, accuracy: f64, precision: f64, recall: f64, f1: f64) { - common::metrics::set_gauge_vec("foxhunt_training_eval_accuracy", &[model, fold], accuracy); - common::metrics::set_gauge_vec("foxhunt_training_eval_precision", &[model, fold], precision); - common::metrics::set_gauge_vec("foxhunt_training_eval_recall", &[model, fold], recall); - common::metrics::set_gauge_vec("foxhunt_training_eval_f1", &[model, fold], f1); -} - -pub fn record_checkpoint_save(model: &str, fold: &str, duration_secs: f64, size_bytes: f64) { - common::metrics::increment_counter_vec( - "foxhunt_training_checkpoint_saves_total", - &[model, fold], - 1.0, - ); - common::metrics::set_gauge_vec( - "foxhunt_training_checkpoint_size_bytes", - &[model, fold], - size_bytes, - ); - common::metrics::observe_histogram_vec( - "foxhunt_training_checkpoint_duration_seconds", - &[model], - duration_secs, - ); -} - -pub fn record_checkpoint_failure(model: &str, fold: &str) { - common::metrics::increment_counter_vec( - "foxhunt_training_checkpoint_failures_total", - &[model, fold], - 1.0, - ); -} - -pub fn record_nan_detected(model: &str, fold: &str) { - common::metrics::increment_counter_vec( - "foxhunt_training_nan_detected_total", - &[model, fold], - 1.0, - ); -} - -pub fn record_gradient_explosion(model: &str, fold: &str) { - common::metrics::increment_counter_vec( - "foxhunt_training_gradient_explosion_total", - &[model, fold], - 1.0, - ); -} - -pub fn record_feature_error(model: &str, fold: &str) { - common::metrics::increment_counter_vec( - "foxhunt_training_feature_errors_total", - &[model, fold], - 1.0, - ); -} - -pub fn record_data_load(model: &str, duration_secs: f64) { - common::metrics::observe_histogram_vec( - "foxhunt_training_data_load_seconds", - &[model], - duration_secs, - ); -} - -pub fn set_active_workers(value: f64) { - common::metrics::set_gauge("foxhunt_training_active_workers", value); -} diff --git a/crates/ml/examples/baseline_common/mod.rs b/crates/ml/examples/baseline_common/mod.rs index 66b063a58..5b60d647c 100644 --- a/crates/ml/examples/baseline_common/mod.rs +++ b/crates/ml/examples/baseline_common/mod.rs @@ -6,8 +6,6 @@ #[allow(dead_code)] pub mod completion; -#[allow(dead_code)] -pub mod metrics; use std::path::{Path, PathBuf}; diff --git a/crates/ml/examples/hyperopt_baseline_rl.rs b/crates/ml/examples/hyperopt_baseline_rl.rs index 85de2bbb2..53edf2a03 100644 --- a/crates/ml/examples/hyperopt_baseline_rl.rs +++ b/crates/ml/examples/hyperopt_baseline_rl.rs @@ -47,9 +47,7 @@ use ml::hyperopt::adapters::ppo::PPOTrainer; use ml::hyperopt::paths::{generate_run_id, TrainingPaths}; use ml::hyperopt::ArgminOptimizer; -#[allow(unreachable_pub)] -mod baseline_common; -use baseline_common::metrics; +use common::metrics::{server as metrics_server, training_metrics}; /// Hyperparameter optimization runner for DQN/PPO on Databento market data #[derive(Parser, Debug)] @@ -256,9 +254,9 @@ fn main() -> Result<()> { .with_target(false) .init(); - metrics::init_training_metrics(); - metrics::start_metrics_server(9094); - metrics::set_active_workers(1.0); + training_metrics::init(); + metrics_server::start_metrics_server(9094); + training_metrics::set_active_workers(1.0); let args = Args::parse(); @@ -288,7 +286,7 @@ fn main() -> Result<()> { info!("Device: CUDA GPU"); // Resolve parallel: 0 = auto-detect. Reserve 2 cores for CUDA driver overhead. - let parallel = if args.parallel == 0 { + let mut parallel = if args.parallel == 0 { cpus.saturating_sub(2).max(1) } else { args.parallel @@ -297,7 +295,7 @@ fn main() -> Result<()> { // Cap by VRAM budget — no point having more threads than concurrent GPU trials let budget = ml::hyperopt::HardwareBudget::detect(); let vram_cap = budget.plan_hyperopt(300.0, 0.0002, 64.0, 4096.0).max_concurrent_trials; - let parallel = parallel.min(vram_cap); + parallel = parallel.min(vram_cap); info!("Parallel: {} threads (CPU: {}, VRAM cap: {})", parallel, cpus, vram_cap); // Configure rayon thread pool for parallel trial evaluation @@ -406,6 +404,6 @@ fn main() -> Result<()> { info!("========================================"); info!("{}", output_str); - metrics::set_active_workers(0.0); + training_metrics::set_active_workers(0.0); Ok(()) } diff --git a/crates/ml/examples/hyperopt_baseline_supervised.rs b/crates/ml/examples/hyperopt_baseline_supervised.rs index 98835107c..f40469d6d 100644 --- a/crates/ml/examples/hyperopt_baseline_supervised.rs +++ b/crates/ml/examples/hyperopt_baseline_supervised.rs @@ -37,9 +37,7 @@ use std::path::PathBuf; use std::time::Instant; use tracing::{error, info, warn, Level}; -#[allow(unreachable_pub)] -mod baseline_common; -use baseline_common::metrics; +use common::metrics::{server as metrics_server, training_metrics}; use ml::hyperopt::adapters::diffusion::DiffusionTrainer; use ml::hyperopt::adapters::kan::KANTrainer; @@ -503,9 +501,9 @@ fn main() -> Result<()> { .with_target(false) .init(); - metrics::init_training_metrics(); - metrics::start_metrics_server(9094); - metrics::set_active_workers(1.0); + training_metrics::init(); + metrics_server::start_metrics_server(9094); + training_metrics::set_active_workers(1.0); let args = Args::parse(); @@ -602,6 +600,6 @@ fn main() -> Result<()> { info!("========================================"); info!("{}", output_str); - metrics::set_active_workers(0.0); + training_metrics::set_active_workers(0.0); Ok(()) } diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 55440d1c5..d2061afc2 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -34,8 +34,8 @@ use ml::dqn::{DQNConfig, Experience, DQN}; #[allow(unreachable_pub)] mod baseline_common; use baseline_common::completion::{write_failure_marker, write_success_marker, CompletionMetrics}; -use baseline_common::metrics; use baseline_common::{load_all_bars, spread_cost_bps}; +use common::metrics::{server as metrics_server, training_metrics as metrics}; use ml::features::extraction::extract_ml_features; use ml::ppo::ppo::{PPOConfig, PPO}; use ml::ppo::trajectories::{Trajectory, TrajectoryBatch, TrajectoryStep}; @@ -1038,8 +1038,8 @@ fn main() -> Result<()> { ) .init(); - metrics::init_training_metrics(); - metrics::start_metrics_server(9094); + metrics::init(); + metrics_server::start_metrics_server(9094); metrics::set_active_workers(1.0); let args = Args::parse(); @@ -1054,20 +1054,20 @@ fn main() -> Result<()> { match result { Ok(results) => { - for result in &results { - let best_val = result + for training_result in &results { + let best_val = training_result .fold_results .iter() .map(|(_, loss)| *loss) .fold(f64::MAX, f64::min); let metrics = CompletionMetrics { - model: result.model_name.clone(), + model: training_result.model_name.clone(), symbol: args.symbol.clone(), best_val_loss: (best_val < f64::MAX).then_some(best_val), sharpe_ratio: None, - epochs_completed: result.total_epochs, - folds_completed: result.fold_results.len(), + epochs_completed: training_result.total_epochs, + folds_completed: training_result.fold_results.len(), }; write_success_marker(&args.output_dir, &metrics); } diff --git a/crates/ml/examples/train_baseline_supervised.rs b/crates/ml/examples/train_baseline_supervised.rs index 50d99eb97..411eef143 100644 --- a/crates/ml/examples/train_baseline_supervised.rs +++ b/crates/ml/examples/train_baseline_supervised.rs @@ -42,8 +42,8 @@ use ml::walk_forward::{generate_walk_forward_windows, NormStats, WalkForwardConf #[allow(unreachable_pub)] mod baseline_common; use baseline_common::completion::{write_failure_marker, write_success_marker, CompletionMetrics}; -use baseline_common::metrics; use baseline_common::{load_all_bars, spread_cost_bps}; +use common::metrics::{server as metrics_server, training_metrics as metrics}; // Model adapter imports -- using verified paths from existing examples use ml::diffusion::{DiffusionConfig, DiffusionTrainableAdapter}; @@ -804,8 +804,8 @@ fn main() -> Result<()> { ) .init(); - metrics::init_training_metrics(); - metrics::start_metrics_server(9094); + metrics::init(); + metrics_server::start_metrics_server(9094); metrics::set_active_workers(1.0); let args = Args::parse(); @@ -820,20 +820,20 @@ fn main() -> Result<()> { match result { Ok(results) => { - for result in &results { - let best_val = result + for training_result in &results { + let best_val = training_result .fold_results .iter() .map(|(_, loss)| *loss) .fold(f64::MAX, f64::min); let metrics = CompletionMetrics { - model: result.model_name.clone(), + model: training_result.model_name.clone(), symbol: args.symbol.clone(), best_val_loss: (best_val < f64::MAX).then_some(best_val), sharpe_ratio: None, - epochs_completed: result.total_epochs, - folds_completed: result.fold_results.len(), + epochs_completed: training_result.total_epochs, + folds_completed: training_result.fold_results.len(), }; write_success_marker(&args.output_dir, &metrics); } diff --git a/crates/ml/src/optimizers/adam.rs b/crates/ml/src/optimizers/adam.rs index 3e9bf6785..7b82dc823 100644 --- a/crates/ml/src/optimizers/adam.rs +++ b/crates/ml/src/optimizers/adam.rs @@ -158,9 +158,8 @@ impl Adam { .map_err(|e| MLError::TrainingError(format!("Optimizer step failed: {}", e)))?; if actual_grad_norm > max_norm { - tracing::warn!( - "Gradient clipping: {:.4} -> {:.4} - this is expected occasionally but should be rare. \ - If frequent, consider reducing learning rate.", + tracing::debug!( + "Gradient clipping: {:.4} -> {:.4}", actual_grad_norm, clipped_grad_norm ); diff --git a/crates/ml/src/training/push_metrics.rs b/crates/ml/src/training/push_metrics.rs index 01d8d166c..0d90af6a1 100644 --- a/crates/ml/src/training/push_metrics.rs +++ b/crates/ml/src/training/push_metrics.rs @@ -33,23 +33,23 @@ impl TrainingMetricsPusher { let mut body = String::new(); // Epoch progress - let _ = writeln!(body, "# HELP foxhunt_training_current_epoch Current training epoch"); - let _ = writeln!(body, "# TYPE foxhunt_training_current_epoch gauge"); - let _ = writeln!(body, "foxhunt_training_current_epoch {}", state.epoch); + _ = writeln!(body, "# HELP foxhunt_training_current_epoch Current training epoch"); + _ = writeln!(body, "# TYPE foxhunt_training_current_epoch gauge"); + _ = writeln!(body, "foxhunt_training_current_epoch {}", state.epoch); // Loss - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_epoch_loss Training loss value" ); - let _ = writeln!(body, "# TYPE foxhunt_training_epoch_loss gauge"); - let _ = writeln!( + _ = writeln!(body, "# TYPE foxhunt_training_epoch_loss gauge"); + _ = writeln!( body, "foxhunt_training_epoch_loss{{split=\"train\"}} {}", state.train_loss ); if let Some(val_loss) = state.val_loss { - let _ = writeln!( + _ = writeln!( body, "foxhunt_training_epoch_loss{{split=\"val\"}} {}", val_loss @@ -58,21 +58,21 @@ impl TrainingMetricsPusher { // Accuracy if let Some(accuracy) = state.accuracy { - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_eval_accuracy Model evaluation accuracy" ); - let _ = writeln!(body, "# TYPE foxhunt_training_eval_accuracy gauge"); - let _ = writeln!(body, "foxhunt_training_eval_accuracy {}", accuracy); + _ = writeln!(body, "# TYPE foxhunt_training_eval_accuracy gauge"); + _ = writeln!(body, "foxhunt_training_eval_accuracy {}", accuracy); } // Batches processed - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_batches_processed Total batches processed" ); - let _ = writeln!(body, "# TYPE foxhunt_training_batches_processed counter"); - let _ = writeln!( + _ = writeln!(body, "# TYPE foxhunt_training_batches_processed counter"); + _ = writeln!( body, "foxhunt_training_batches_processed {}", state.batches_processed @@ -80,12 +80,12 @@ impl TrainingMetricsPusher { // Training speed if state.batches_per_second > 0.0 { - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_batches_per_second Training throughput" ); - let _ = writeln!(body, "# TYPE foxhunt_training_batches_per_second gauge"); - let _ = writeln!( + _ = writeln!(body, "# TYPE foxhunt_training_batches_per_second gauge"); + _ = writeln!( body, "foxhunt_training_batches_per_second {}", state.batches_per_second @@ -93,12 +93,12 @@ impl TrainingMetricsPusher { } // Learning rate - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_learning_rate Current learning rate" ); - let _ = writeln!(body, "# TYPE foxhunt_training_learning_rate gauge"); - let _ = writeln!( + _ = writeln!(body, "# TYPE foxhunt_training_learning_rate gauge"); + _ = writeln!( body, "foxhunt_training_learning_rate {}", state.learning_rate @@ -106,27 +106,27 @@ impl TrainingMetricsPusher { // NaN / gradient events if state.nan_count > 0 { - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_nan_detected_total NaN gradient events" ); - let _ = writeln!(body, "# TYPE foxhunt_training_nan_detected_total counter"); - let _ = writeln!( + _ = writeln!(body, "# TYPE foxhunt_training_nan_detected_total counter"); + _ = writeln!( body, "foxhunt_training_nan_detected_total {}", state.nan_count ); } if state.gradient_explosion_count > 0 { - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_gradient_explosion_total Gradient clipping events" ); - let _ = writeln!( + _ = writeln!( body, "# TYPE foxhunt_training_gradient_explosion_total counter" ); - let _ = writeln!( + _ = writeln!( body, "foxhunt_training_gradient_explosion_total {}", state.gradient_explosion_count @@ -134,15 +134,15 @@ impl TrainingMetricsPusher { } // Checkpoint saves - let _ = writeln!( + _ = writeln!( body, "# HELP foxhunt_training_checkpoint_saves_total Checkpoints saved" ); - let _ = writeln!( + _ = writeln!( body, "# TYPE foxhunt_training_checkpoint_saves_total counter" ); - let _ = writeln!( + _ = writeln!( body, "foxhunt_training_checkpoint_saves_total {}", state.checkpoint_saves @@ -174,7 +174,9 @@ impl TrainingMetricsPusher { "{}/metrics/job/{}/model/{}", self.pushgateway_url, self.job_id, self.model_name ); - let _ = self.client.delete(&url).send().await; + if let Err(e) = self.client.delete(&url).send().await { + eprintln!("Failed to delete metrics from Pushgateway: {e}"); + } } }