diff --git a/bin/fxt/proto/monitoring.proto b/bin/fxt/proto/monitoring.proto index d7233faf6..157a89453 100644 --- a/bin/fxt/proto/monitoring.proto +++ b/bin/fxt/proto/monitoring.proto @@ -9,6 +9,10 @@ service MonitoringService { // Server-streaming: pushes updates every N seconds rpc StreamTrainingMetrics(StreamTrainingMetricsRequest) returns (stream GetLiveTrainingMetricsResponse); + + // Epoch history for a specific session (ring buffer, max 50 epochs) + rpc GetEpochHistory(GetEpochHistoryRequest) + returns (GetEpochHistoryResponse); } message GetLiveTrainingMetricsRequest { @@ -81,6 +85,20 @@ message TrainingSession { uint32 hyperopt_trial_epoch = 33; float hyperopt_trial_best_loss = 34; float hyperopt_elapsed_seconds = 35; + + // Epoch-level financial metrics + float epoch_sharpe = 36; + float epoch_sortino = 37; + float epoch_win_rate = 38; + float epoch_max_drawdown = 39; + float epoch_profit_factor = 40; + float epoch_total_return = 41; + float epoch_avg_return = 42; + uint32 epoch_total_trades = 43; + // Action distribution + float action_buy_pct = 44; + float action_sell_pct = 45; + float action_hold_pct = 46; } message GpuSnapshot { @@ -90,3 +108,33 @@ message GpuSnapshot { float temperature_celsius = 4; float power_watts = 5; } + +message GetEpochHistoryRequest { + string model = 1; + string fold = 2; + uint32 max_epochs = 3; // 0 = all (up to 50) +} + +message EpochFinancialSnapshot { + uint32 epoch = 1; + float sharpe = 2; + float sortino = 3; + float win_rate = 4; + float max_drawdown = 5; + float profit_factor = 6; + float total_return = 7; + float avg_return = 8; + uint32 total_trades = 9; + float loss = 10; + float val_loss = 11; + float learning_rate = 12; + float action_buy_pct = 13; + float action_sell_pct = 14; + float action_hold_pct = 15; +} + +message GetEpochHistoryResponse { + string model = 1; + string fold = 2; + repeated EpochFinancialSnapshot epochs = 3; +} diff --git a/bin/fxt/src/commands/train/monitor.rs b/bin/fxt/src/commands/train/monitor.rs index b8a13e00f..56c32d899 100644 --- a/bin/fxt/src/commands/train/monitor.rs +++ b/bin/fxt/src/commands/train/monitor.rs @@ -101,6 +101,7 @@ fn render_snapshot(resp: &GetLiveTrainingMetricsResponse) { print_rl_diagnostics(&resp.sessions); print_hyperopt_summary(&resp.sessions); print_health_summary(&resp.sessions); + print_financial_metrics(&resp.sessions); } fn render_tui(resp: &GetLiveTrainingMetricsResponse) { @@ -232,3 +233,42 @@ fn print_health_summary(sessions: &[TrainingSession]) { total_nan, total_grad, total_ckpt ); } + +fn print_financial_metrics(sessions: &[TrainingSession]) { + let financial: Vec<_> = sessions + .iter() + .filter(|s| s.epoch_sharpe != 0.0 || s.epoch_win_rate > 0.0) + .collect(); + if financial.is_empty() { + return; + } + println!(); + println!("{}", "Epoch Financial Metrics:".bright_cyan()); + for s in &financial { + let sharpe_colored = if s.epoch_sharpe >= 2.0 { + format!("{:.2}", s.epoch_sharpe).green() + } else if s.epoch_sharpe >= 1.0 { + format!("{:.2}", s.epoch_sharpe).yellow() + } else { + format!("{:.2}", s.epoch_sharpe).red() + }; + println!( + " {}: Sharpe={} WinRate={:.1}% MaxDD={:.1}% PF={:.2} Return={:+.2}% Trades={}", + s.model.bright_white(), + sharpe_colored, + s.epoch_win_rate * 100.0, + s.epoch_max_drawdown * 100.0, + s.epoch_profit_factor, + s.epoch_total_return * 100.0, + s.epoch_total_trades, + ); + if s.action_buy_pct > 0.0 || s.action_sell_pct > 0.0 { + println!( + " Actions: BUY {:.0}% | SELL {:.0}% | HOLD {:.0}%", + s.action_buy_pct * 100.0, + s.action_sell_pct * 100.0, + s.action_hold_pct * 100.0, + ); + } + } +} diff --git a/bin/fxt/src/commands/watch/render.rs b/bin/fxt/src/commands/watch/render.rs index a96b7347c..6653716b6 100644 --- a/bin/fxt/src/commands/watch/render.rs +++ b/bin/fxt/src/commands/watch/render.rs @@ -149,6 +149,8 @@ fn render_training_list(frame: &mut Frame, area: Rect, tab: &TrainingTabState) { Cell::from("LR"), Cell::from("Batch/s"), Cell::from("Grad"), + Cell::from("Sharpe"), + Cell::from("Win%"), ]) .style( Style::default() @@ -171,6 +173,8 @@ fn render_training_list(frame: &mut Frame, area: Rect, tab: &TrainingTabState) { Cell::from(format!("{:.2e}", s.learning_rate)), Cell::from(format!("{:.1}", s.batches_per_second)), Cell::from(format!("{:.3}", s.gradient_norm)), + Cell::from(if s.epoch_sharpe != 0.0 { format!("{:.2}", s.epoch_sharpe) } else { "-".to_owned() }), + Cell::from(if s.epoch_win_rate > 0.0 { format!("{:.0}%", s.epoch_win_rate * 100.0) } else { "-".to_owned() }), ]) }) .collect(); @@ -188,6 +192,8 @@ fn render_training_list(frame: &mut Frame, area: Rect, tab: &TrainingTabState) { Constraint::Length(10), Constraint::Length(8), Constraint::Length(8), + Constraint::Length(8), + Constraint::Length(8), ], ) .header(header) @@ -351,8 +357,8 @@ fn render_detail_overview( let chunks = Layout::default() .direction(Direction::Vertical) .constraints([ - Constraint::Length(8), // key stats - Constraint::Min(0), // loss sparkline + Constraint::Length(12), // key stats (incl. financial metrics) + Constraint::Min(0), // loss sparkline ]) .split(area); @@ -377,6 +383,17 @@ fn render_detail_overview( " NaN: {} Grad Explosions: {} Feature Errors: {}", session.nan_detected, session.gradient_explosions, session.feature_errors, )), + Line::from(""), + Line::from(format!( + " Sharpe: {:.2} Sortino: {:.2} Win Rate: {:.1}% Max DD: {:.1}%", + session.epoch_sharpe, session.epoch_sortino, + session.epoch_win_rate * 100.0, session.epoch_max_drawdown * 100.0, + )), + Line::from(format!( + " PF: {:.2} Return: {:+.2}% Avg: {:+.4} Trades: {}", + session.epoch_profit_factor, session.epoch_total_return * 100.0, + session.epoch_avg_return, session.epoch_total_trades, + )), ]) .block( Block::default() @@ -454,7 +471,7 @@ fn render_detail_metrics( let chunks = Layout::default() .direction(Direction::Vertical) .constraints([ - Constraint::Length(5), // current values + Constraint::Length(7), // current values (incl. action distribution) Constraint::Min(0), // sparklines ]) .split(area); @@ -468,6 +485,12 @@ fn render_detail_metrics( " Precision: {:.4} Recall: {:.4}", session.eval_precision, session.eval_recall, )), + Line::from(format!( + " Action: BUY {:.0}% SELL {:.0}% HOLD {:.0}%", + session.action_buy_pct * 100.0, + session.action_sell_pct * 100.0, + session.action_hold_pct * 100.0, + )), ]) .block( Block::default() @@ -479,10 +502,14 @@ fn render_detail_metrics( let spark_area = Layout::default() .direction(Direction::Vertical) .constraints([ - Constraint::Percentage(25), - Constraint::Percentage(25), - Constraint::Percentage(25), - Constraint::Percentage(25), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), + Constraint::Ratio(1, 8), ]) .split(chunks[1]); @@ -490,6 +517,10 @@ fn render_detail_metrics( render_sparkline_row(frame, spark_area[1], "Precision", &history.precision, 1.0, Color::Cyan); render_sparkline_row(frame, spark_area[2], "Recall", &history.recall, 1.0, Color::Magenta); render_sparkline_row(frame, spark_area[3], "F1", &history.f1, 1.0, Color::Yellow); + render_sparkline_row(frame, spark_area[4], "Sharpe", &history.sharpe, 20.0, Color::Cyan); + render_sparkline_row(frame, spark_area[5], "Win Rate", &history.win_rate, 1.0, Color::Green); + render_sparkline_row(frame, spark_area[6], "Max DD", &history.max_drawdown, 1.0, Color::Red); + render_sparkline_row(frame, spark_area[7], "Total Return", &history.total_return, 1.0, Color::Magenta); } fn render_detail_hyperopt( diff --git a/crates/common/src/metrics/training_metrics.rs b/crates/common/src/metrics/training_metrics.rs index eae70ea29..9618f0f60 100644 --- a/crates/common/src/metrics/training_metrics.rs +++ b/crates/common/src/metrics/training_metrics.rs @@ -1,6 +1,6 @@ //! Prometheus metrics for ML training binaries. //! -//! Registers the 41 metrics expected by the Training Cockpit Grafana dashboard +//! Registers the 52 metrics expected by the Training Cockpit Grafana dashboard //! and provides typed helper functions so callers never mis-spell metric names. //! //! # Usage @@ -35,7 +35,7 @@ pub fn verbose_enabled() -> bool { // Registration // --------------------------------------------------------------------------- -/// Register all 41 training metrics with the global Prometheus registry. +/// Register all 52 training metrics with the global Prometheus registry. /// /// Safe to call multiple times — the registry silently ignores duplicates. pub fn init() { @@ -183,6 +183,63 @@ pub fn init() { mf, ); + // Epoch-level financial metrics (model + fold) + _ = register_gauge_vec( + "foxhunt_training_epoch_sharpe", + "Epoch Sharpe ratio from validation backtest", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_sortino", + "Epoch Sortino ratio from validation backtest", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_win_rate", + "Epoch win rate 0-1", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_max_drawdown", + "Epoch max drawdown 0-1", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_profit_factor", + "Epoch profit factor (gross profit / gross loss)", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_total_return", + "Epoch total return (fractional)", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_avg_return", + "Epoch average return per trade", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_total_trades", + "Epoch total trade count", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_action_buy_pct", + "BUY action percentage 0-1", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_action_sell_pct", + "SELL action percentage 0-1", + mf, + ); + _ = register_gauge_vec( + "foxhunt_training_epoch_action_hold_pct", + "HOLD action percentage 0-1", + mf, + ); + // Hyperopt gauges (model only) _ = register_gauge_vec( "foxhunt_hyperopt_trial_current", @@ -448,6 +505,47 @@ pub fn set_action_diversity(model: &str, fold: &str, diversity: f64) { set_gauge_vec("foxhunt_training_action_diversity", &[model, fold], diversity); } +// --------------------------------------------------------------------------- +// Tier 1: Epoch financial metrics +// --------------------------------------------------------------------------- + +/// Push a full set of epoch-level financial metrics from a backtest evaluation. +#[allow(clippy::too_many_arguments)] +pub fn set_epoch_financial_metrics( + model: &str, + fold: &str, + sharpe: f64, + sortino: f64, + win_rate: f64, + max_drawdown: f64, + profit_factor: f64, + total_return: f64, + avg_return: f64, + total_trades: f64, +) { + set_gauge_vec("foxhunt_training_epoch_sharpe", &[model, fold], sharpe); + set_gauge_vec("foxhunt_training_epoch_sortino", &[model, fold], sortino); + set_gauge_vec("foxhunt_training_epoch_win_rate", &[model, fold], win_rate); + set_gauge_vec("foxhunt_training_epoch_max_drawdown", &[model, fold], max_drawdown); + set_gauge_vec("foxhunt_training_epoch_profit_factor", &[model, fold], profit_factor); + set_gauge_vec("foxhunt_training_epoch_total_return", &[model, fold], total_return); + set_gauge_vec("foxhunt_training_epoch_avg_return", &[model, fold], avg_return); + set_gauge_vec("foxhunt_training_epoch_total_trades", &[model, fold], total_trades); +} + +/// Push epoch action distribution percentages (all 0-1). +pub fn set_epoch_action_distribution( + model: &str, + fold: &str, + buy_pct: f64, + sell_pct: f64, + hold_pct: f64, +) { + set_gauge_vec("foxhunt_training_epoch_action_buy_pct", &[model, fold], buy_pct); + set_gauge_vec("foxhunt_training_epoch_action_sell_pct", &[model, fold], sell_pct); + set_gauge_vec("foxhunt_training_epoch_action_hold_pct", &[model, fold], hold_pct); +} + // --------------------------------------------------------------------------- // Tier 1: Hyperopt intra-trial // --------------------------------------------------------------------------- @@ -573,4 +671,11 @@ mod tests { set_hyperopt_trial_best_loss("dqn", 0.042); set_hyperopt_elapsed("dqn", 123.4); } + + #[test] + fn test_epoch_financial_metrics_no_panic() { + init(); + set_epoch_financial_metrics("dqn", "0", 2.31, 3.12, 0.55, 0.08, 1.84, 0.124, 0.003, 142.0); + set_epoch_action_distribution("dqn", "0", 0.35, 0.25, 0.40); + } } diff --git a/crates/ml/src/trainers/dqn/financials.rs b/crates/ml/src/trainers/dqn/financials.rs new file mode 100644 index 000000000..ea17f9b12 --- /dev/null +++ b/crates/ml/src/trainers/dqn/financials.rs @@ -0,0 +1,189 @@ +//! Epoch-level financial metrics computed from the DQN trainer's PnL history +//! and action counts. These are pushed to Prometheus at the end of each epoch. + +use std::collections::VecDeque; + +/// Financial metrics summary for a single training epoch. +#[derive(Debug, Clone, Default)] +pub(crate) struct EpochFinancials { + pub sharpe: f64, + pub sortino: f64, + pub win_rate: f64, + pub max_drawdown: f64, + pub profit_factor: f64, + pub total_return: f64, + pub avg_return: f64, + pub total_trades: usize, + pub buy_pct: f64, + pub sell_pct: f64, + pub hold_pct: f64, +} + +/// Compute financial metrics from the trainer's PnL history and action counts. +/// +/// - `pnl_history`: per-step PnL values accumulated this epoch +/// - `action_counts`: 45-element array (FactoredAction), grouped into BUY/SELL/HOLD +/// - `initial_capital`: starting equity for return calculation (default 100_000) +pub(crate) fn compute_epoch_financials( + pnl_history: &VecDeque, + action_counts: &[usize; 45], + initial_capital: f64, +) -> EpochFinancials { + if pnl_history.is_empty() { + return EpochFinancials::default(); + } + + let returns: Vec = pnl_history.iter().copied().collect(); + let n = returns.len(); + + // Total return + let total_pnl: f64 = returns.iter().sum(); + let total_return = total_pnl / initial_capital; + + // Win rate + let winning = returns.iter().filter(|&&r| r > 0.0).count(); + let win_rate = winning as f64 / n as f64; + + // Average return per trade + let avg_return = total_pnl / n as f64; + + // Sharpe ratio (annualized, 252 trading days) + let mean = total_pnl / n as f64; + let variance: f64 = returns.iter().map(|r| (r - mean).powi(2)).sum::() / n as f64; + let std = variance.sqrt(); + let sharpe = if std > 1e-10 { + (mean / std) * (252.0_f64).sqrt() + } else { + 0.0 + }; + + // Sortino ratio (only downside deviation) + let downside_returns: Vec = returns.iter().filter(|&&r| r < 0.0).copied().collect(); + let sortino = if downside_returns.len() > 1 { + let down_var: f64 = downside_returns.iter().map(|r| r.powi(2)).sum::() + / downside_returns.len() as f64; + let down_std = down_var.sqrt(); + if down_std > 1e-10 { + (mean / down_std) * (252.0_f64).sqrt() + } else { + 0.0 + } + } else { + 0.0 + }; + + // Max drawdown + let mut equity = initial_capital; + let mut peak = equity; + let mut max_dd = 0.0_f64; + for &pnl in &returns { + equity += pnl; + if equity > peak { + peak = equity; + } + let dd = (peak - equity) / peak; + if dd > max_dd { + max_dd = dd; + } + } + + // Profit factor + let gross_profit: f64 = returns.iter().filter(|&&r| r > 0.0).sum(); + let gross_loss: f64 = returns.iter().filter(|&&r| r < 0.0).map(|r| r.abs()).sum(); + let profit_factor = if gross_loss > 1e-10 { + gross_profit / gross_loss + } else if gross_profit > 0.0 { + f64::INFINITY + } else { + 0.0 + }; + + // Action distribution: group 45 actions into BUY/SELL/HOLD + // FactoredAction: exposure(5) x order(3) x urgency(3) + // Exposure 0,1 = reduce/close (SELL-like), 2 = hold, 3,4 = add/aggressive (BUY-like) + let total_actions: usize = action_counts.iter().sum(); + let (buy_pct, sell_pct, hold_pct) = if total_actions > 0 { + let mut buy = 0_usize; + let mut sell = 0_usize; + let mut hold = 0_usize; + for (i, &count) in action_counts.iter().enumerate() { + let exposure = i / 9; // 0-4 + match exposure { + 0 | 1 => sell += count, // reduce/close + 2 => hold += count, // neutral + 3 | 4 => buy += count, // add/aggressive + _ => {} + } + } + let t = total_actions as f64; + (buy as f64 / t, sell as f64 / t, hold as f64 / t) + } else { + (0.0, 0.0, 0.0) + }; + + EpochFinancials { + sharpe, + sortino, + win_rate, + max_drawdown: max_dd, + profit_factor, + total_return, + avg_return, + total_trades: n, + buy_pct, + sell_pct, + hold_pct, + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used)] +mod tests { + use super::*; + + #[test] + fn test_empty_history() { + let f = compute_epoch_financials(&VecDeque::new(), &[0; 45], 100_000.0); + assert_eq!(f.total_trades, 0); + assert_eq!(f.sharpe, 0.0); + } + + #[test] + fn test_all_winning() { + let pnl: VecDeque = vec![10.0, 20.0, 30.0, 15.0, 25.0].into(); + let f = compute_epoch_financials(&pnl, &[0; 45], 100_000.0); + assert_eq!(f.win_rate, 1.0); + assert_eq!(f.total_trades, 5); + assert!(f.sharpe > 0.0); + assert!(f.max_drawdown < 1e-10); // No drawdown with all wins + assert!(f.profit_factor.is_infinite()); // No losses + } + + #[test] + fn test_mixed_pnl() { + let pnl: VecDeque = vec![100.0, -50.0, 75.0, -25.0, 50.0].into(); + let f = compute_epoch_financials(&pnl, &[0; 45], 100_000.0); + assert_eq!(f.total_trades, 5); + assert!((f.win_rate - 0.6).abs() < 1e-10); + assert!(f.total_return > 0.0); + assert!(f.profit_factor > 1.0); + assert!(f.max_drawdown > 0.0); + assert!(f.sortino > 0.0); + } + + #[test] + fn test_action_distribution() { + let mut actions = [0usize; 45]; + // Exposure 3 (add), order 0, urgency 0 → index 27 + actions[27] = 100; // BUY + // Exposure 0 (reduce), order 0, urgency 0 → index 0 + actions[0] = 50; // SELL + // Exposure 2 (neutral), order 0, urgency 0 → index 18 + actions[18] = 50; // HOLD + let pnl: VecDeque = vec![1.0].into(); + let f = compute_epoch_financials(&pnl, &actions, 100_000.0); + assert!((f.buy_pct - 0.5).abs() < 1e-10); + assert!((f.sell_pct - 0.25).abs() < 1e-10); + assert!((f.hold_pct - 0.25).abs() < 1e-10); + } +} diff --git a/crates/ml/src/trainers/dqn/mod.rs b/crates/ml/src/trainers/dqn/mod.rs index 0b7828343..d42722b81 100644 --- a/crates/ml/src/trainers/dqn/mod.rs +++ b/crates/ml/src/trainers/dqn/mod.rs @@ -21,6 +21,7 @@ mod config; mod data_loading; mod early_stopping; +pub(crate) mod financials; mod features; pub mod lr_scheduler; mod monitoring; diff --git a/crates/ml/src/trainers/dqn/trainer.rs b/crates/ml/src/trainers/dqn/trainer.rs index 2f2676cbf..5086263cb 100644 --- a/crates/ml/src/trainers/dqn/trainer.rs +++ b/crates/ml/src/trainers/dqn/trainer.rs @@ -41,6 +41,7 @@ use crate::memory_optimization::auto_batch_size::{AutoBatchSizer, BatchSizeConfi // Import from sibling modules use super::config::{DQNAgentType, DQNHyperparameters}; +use super::financials::compute_epoch_financials; use super::monitoring::TrainingMonitor; use super::statistics::{FeatureStatistics, QValueStats}; use super::{EPISODE_LENGTH, FeatureVector51}; @@ -2788,6 +2789,39 @@ impl DQNTrainer { ); } + // Epoch financial metrics for monitoring service + { + let financials = compute_epoch_financials( + &self.pnl_history, + &monitor.action_counts, + 100_000.0, + ); + training_metrics::set_epoch_financial_metrics( + "dqn", "current", + financials.sharpe, + financials.sortino, + financials.win_rate, + financials.max_drawdown, + financials.profit_factor, + financials.total_return, + financials.avg_return, + financials.total_trades as f64, + ); + training_metrics::set_epoch_action_distribution( + "dqn", "current", + financials.buy_pct, + financials.sell_pct, + financials.hold_pct, + ); + info!( + "Epoch {}/{}: Sharpe={:.2} WinRate={:.1}% MaxDD={:.1}% PF={:.2} Return={:+.2}% Trades={}", + epoch + 1, self.hyperparams.epochs, + financials.sharpe, financials.win_rate * 100.0, + financials.max_drawdown * 100.0, financials.profit_factor, + financials.total_return * 100.0, financials.total_trades, + ); + } + // Track metrics for early stopping self.loss_history.push(avg_loss); self.q_value_history.push(avg_q_value); diff --git a/crates/ml/src/trainers/ppo.rs b/crates/ml/src/trainers/ppo.rs index b260af645..a618e2c05 100644 --- a/crates/ml/src/trainers/ppo.rs +++ b/crates/ml/src/trainers/ppo.rs @@ -678,6 +678,28 @@ impl PpoTrainer { training_metrics::set_advantage_std("ppo", "current", std_reward as f64); training_metrics::set_value_explained_variance("ppo", "current", explained_variance as f64); + // Epoch financial metrics (simplified for PPO — derived from reward stats) + // PPO doesn't run a backtest per epoch; use reward mean/std as proxy + { + let epoch_sharpe = if std_reward > 1e-10 { + (mean_reward / std_reward) as f64 * (252.0_f64).sqrt() + } else { + 0.0 + }; + training_metrics::set_epoch_financial_metrics( + "ppo", + "current", + epoch_sharpe, + 0.0, // sortino: not available without per-step returns + 0.0, // win_rate: not tracked per epoch in PPO + 0.0, // max_drawdown: not tracked per epoch in PPO + 0.0, // profit_factor: not tracked + mean_reward as f64, // total_return proxy + mean_reward as f64, // avg_return proxy + 0.0, // total_trades: not applicable for PPO + ); + } + // Track metrics for early stopping (bounded ring buffer) { let mut loss_history = self.value_loss_history.lock().await; diff --git a/services/api_gateway/src/grpc/monitoring_proxy.rs b/services/api_gateway/src/grpc/monitoring_proxy.rs index b4182df28..5af4b321f 100644 --- a/services/api_gateway/src/grpc/monitoring_proxy.rs +++ b/services/api_gateway/src/grpc/monitoring_proxy.rs @@ -11,7 +11,8 @@ use tracing::{error, info, instrument}; use crate::monitoring::monitoring_service_client::MonitoringServiceClient; use crate::monitoring::monitoring_service_server::{MonitoringService, MonitoringServiceServer}; use crate::monitoring::{ - GetLiveTrainingMetricsRequest, GetLiveTrainingMetricsResponse, StreamTrainingMetricsRequest, + GetEpochHistoryRequest, GetEpochHistoryResponse, GetLiveTrainingMetricsRequest, + GetLiveTrainingMetricsResponse, StreamTrainingMetricsRequest, }; /// Monitoring Service Proxy @@ -79,6 +80,23 @@ impl MonitoringService for MonitoringServiceProxy { info!("StreamTrainingMetrics forwarded successfully"); Ok(Response::new(boxed_stream)) } + + #[instrument(skip(self, request), fields(request_id = %uuid::Uuid::new_v4()), err)] + async fn get_epoch_history( + &self, + request: Request, + ) -> Result, Status> { + info!("Proxying GetEpochHistory request"); + + let mut client = self.client.clone(); + let response = client.get_epoch_history(request).await.map_err(|e| { + error!("Backend GetEpochHistory failed: {}", e); + e + })?; + + info!("GetEpochHistory forwarded successfully"); + Ok(response) + } } #[cfg(test)] diff --git a/services/monitoring_service/proto/monitoring.proto b/services/monitoring_service/proto/monitoring.proto index d7233faf6..157a89453 100644 --- a/services/monitoring_service/proto/monitoring.proto +++ b/services/monitoring_service/proto/monitoring.proto @@ -9,6 +9,10 @@ service MonitoringService { // Server-streaming: pushes updates every N seconds rpc StreamTrainingMetrics(StreamTrainingMetricsRequest) returns (stream GetLiveTrainingMetricsResponse); + + // Epoch history for a specific session (ring buffer, max 50 epochs) + rpc GetEpochHistory(GetEpochHistoryRequest) + returns (GetEpochHistoryResponse); } message GetLiveTrainingMetricsRequest { @@ -81,6 +85,20 @@ message TrainingSession { uint32 hyperopt_trial_epoch = 33; float hyperopt_trial_best_loss = 34; float hyperopt_elapsed_seconds = 35; + + // Epoch-level financial metrics + float epoch_sharpe = 36; + float epoch_sortino = 37; + float epoch_win_rate = 38; + float epoch_max_drawdown = 39; + float epoch_profit_factor = 40; + float epoch_total_return = 41; + float epoch_avg_return = 42; + uint32 epoch_total_trades = 43; + // Action distribution + float action_buy_pct = 44; + float action_sell_pct = 45; + float action_hold_pct = 46; } message GpuSnapshot { @@ -90,3 +108,33 @@ message GpuSnapshot { float temperature_celsius = 4; float power_watts = 5; } + +message GetEpochHistoryRequest { + string model = 1; + string fold = 2; + uint32 max_epochs = 3; // 0 = all (up to 50) +} + +message EpochFinancialSnapshot { + uint32 epoch = 1; + float sharpe = 2; + float sortino = 3; + float win_rate = 4; + float max_drawdown = 5; + float profit_factor = 6; + float total_return = 7; + float avg_return = 8; + uint32 total_trades = 9; + float loss = 10; + float val_loss = 11; + float learning_rate = 12; + float action_buy_pct = 13; + float action_sell_pct = 14; + float action_hold_pct = 15; +} + +message GetEpochHistoryResponse { + string model = 1; + string fold = 2; + repeated EpochFinancialSnapshot epochs = 3; +} diff --git a/services/monitoring_service/src/service.rs b/services/monitoring_service/src/service.rs index 85404dd49..510f342c6 100644 --- a/services/monitoring_service/src/service.rs +++ b/services/monitoring_service/src/service.rs @@ -1,21 +1,27 @@ -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::pin::Pin; use std::sync::Arc; use std::time::Duration; +use tokio::sync::RwLock; use tokio_stream::Stream; use tonic::{Request, Response, Status}; use tracing::error; use crate::monitoring::{ - monitoring_service_server::MonitoringService, GetLiveTrainingMetricsRequest, + monitoring_service_server::MonitoringService, EpochFinancialSnapshot, + GetEpochHistoryRequest, GetEpochHistoryResponse, GetLiveTrainingMetricsRequest, GetLiveTrainingMetricsResponse, GpuSnapshot, StreamTrainingMetricsRequest, TrainingSession, }; use crate::prometheus_client::{MetricSample, PrometheusClient}; +const MAX_EPOCH_HISTORY: usize = 50; + pub struct MonitoringServiceImpl { prom: Arc, default_interval: u32, + epoch_histories: Arc>>>, + last_epochs: Arc>>, } impl MonitoringServiceImpl { @@ -23,12 +29,16 @@ impl MonitoringServiceImpl { Self { prom: Arc::new(prom), default_interval, + epoch_histories: Arc::new(RwLock::new(HashMap::new())), + last_epochs: Arc::new(RwLock::new(HashMap::new())), } } async fn build_response( prom: &PrometheusClient, model_filter: &str, + epoch_histories: &RwLock>>, + last_epochs: &RwLock>, ) -> Result { let (training, gpu, jobs) = tokio::try_join!( prom.fetch_training_metrics(), @@ -40,6 +50,43 @@ impl MonitoringServiceImpl { let sessions = group_into_sessions(&training, model_filter); let gpu_snapshot = build_gpu_snapshot(&gpu); + // Record epoch history snapshots for sessions with new epoch data + { + let mut last = last_epochs.write().await; + let mut histories = epoch_histories.write().await; + for session in &sessions { + let key = format!("{}/{}", session.model, session.fold); + let prev_epoch = last.get(&key).copied().unwrap_or_default(); + if session.current_epoch > prev_epoch && session.epoch_sharpe != 0.0 { + last.insert(key.clone(), session.current_epoch); + let snapshot = EpochFinancialSnapshot { + epoch: session.current_epoch as u32, + sharpe: session.epoch_sharpe, + sortino: session.epoch_sortino, + win_rate: session.epoch_win_rate, + max_drawdown: session.epoch_max_drawdown, + profit_factor: session.epoch_profit_factor, + total_return: session.epoch_total_return, + avg_return: session.epoch_avg_return, + total_trades: session.epoch_total_trades, + loss: session.epoch_loss, + val_loss: session.validation_loss, + learning_rate: session.learning_rate, + action_buy_pct: session.action_buy_pct, + action_sell_pct: session.action_sell_pct, + action_hold_pct: session.action_hold_pct, + }; + let history = histories + .entry(key) + .or_insert_with(|| VecDeque::with_capacity(MAX_EPOCH_HISTORY)); + if history.len() >= MAX_EPOCH_HISTORY { + history.pop_front(); + } + history.push_back(snapshot); + } + } + } + Ok(GetLiveTrainingMetricsResponse { sessions, gpu: Some(gpu_snapshot), @@ -56,7 +103,9 @@ impl MonitoringService for MonitoringServiceImpl { request: Request, ) -> Result, Status> { let filter = &request.into_inner().model_filter; - let resp = Self::build_response(&self.prom, filter).await?; + let resp = + Self::build_response(&self.prom, filter, &self.epoch_histories, &self.last_epochs) + .await?; Ok(Response::new(resp)) } @@ -75,12 +124,14 @@ impl MonitoringService for MonitoringServiceImpl { }; let filter = req.model_filter; let prom = self.prom.clone(); + let epoch_histories = self.epoch_histories.clone(); + let last_epochs = self.last_epochs.clone(); let stream = async_stream::stream! { let mut interval = tokio::time::interval(Duration::from_secs(u64::from(interval_secs))); loop { interval.tick().await; - match Self::build_response(&prom, &filter).await { + match Self::build_response(&prom, &filter, &epoch_histories, &last_epochs).await { Ok(resp) => yield Ok(resp), Err(e) => { error!("Stream tick failed: {}", e); @@ -92,6 +143,31 @@ impl MonitoringService for MonitoringServiceImpl { Ok(Response::new(Box::pin(stream))) } + + async fn get_epoch_history( + &self, + request: Request, + ) -> Result, Status> { + let req = request.into_inner(); + let key = format!("{}/{}", req.model, req.fold); + let histories = self.epoch_histories.read().await; + let epochs = match histories.get(&key) { + Some(deque) => { + let max = if req.max_epochs == 0 { + MAX_EPOCH_HISTORY + } else { + req.max_epochs as usize + }; + deque.iter().rev().take(max).rev().cloned().collect() + } + None => vec![], + }; + Ok(Response::new(GetEpochHistoryResponse { + model: req.model, + fold: req.fold, + epochs, + })) + } } /// Group flat metric samples into TrainingSession structs keyed by (model, fold) @@ -178,6 +254,29 @@ fn group_into_sessions(samples: &[MetricSample], model_filter: &str) -> Vec { session.hyperopt_elapsed_seconds = s.value as f32; } + // Epoch-level financial metrics + "foxhunt_training_epoch_sharpe" => session.epoch_sharpe = s.value as f32, + "foxhunt_training_epoch_sortino" => session.epoch_sortino = s.value as f32, + "foxhunt_training_epoch_win_rate" => session.epoch_win_rate = s.value as f32, + "foxhunt_training_epoch_max_drawdown" => { + session.epoch_max_drawdown = s.value as f32; + } + "foxhunt_training_epoch_profit_factor" => { + session.epoch_profit_factor = s.value as f32; + } + "foxhunt_training_epoch_total_return" => { + session.epoch_total_return = s.value as f32; + } + "foxhunt_training_epoch_avg_return" => { + session.epoch_avg_return = s.value as f32; + } + "foxhunt_training_epoch_total_trades" => { + session.epoch_total_trades = s.value as u32; + } + // Action distribution + "foxhunt_training_epoch_action_buy_pct" => session.action_buy_pct = s.value as f32, + "foxhunt_training_epoch_action_sell_pct" => session.action_sell_pct = s.value as f32, + "foxhunt_training_epoch_action_hold_pct" => session.action_hold_pct = s.value as f32, _ => {} } } @@ -375,6 +474,43 @@ mod tests { assert!((s.hyperopt_elapsed_seconds - 123.4).abs() < 0.1); } + #[test] + fn test_group_financial_metrics() { + let samples = vec![ + MetricSample { + name: "foxhunt_training_epoch_sharpe".to_owned(), + model: "dqn".to_owned(), + fold: "0".to_owned(), + value: 2.31, + }, + MetricSample { + name: "foxhunt_training_epoch_win_rate".to_owned(), + model: "dqn".to_owned(), + fold: "0".to_owned(), + value: 0.552, + }, + MetricSample { + name: "foxhunt_training_epoch_max_drawdown".to_owned(), + model: "dqn".to_owned(), + fold: "0".to_owned(), + value: 0.081, + }, + MetricSample { + name: "foxhunt_training_epoch_action_buy_pct".to_owned(), + model: "dqn".to_owned(), + fold: "0".to_owned(), + value: 0.35, + }, + ]; + let sessions = group_into_sessions(&samples, ""); + assert_eq!(sessions.len(), 1); + let s = &sessions[0]; + assert!((s.epoch_sharpe - 2.31).abs() < 0.01); + assert!((s.epoch_win_rate - 0.552).abs() < 0.001); + assert!((s.epoch_max_drawdown - 0.081).abs() < 0.001); + assert!((s.action_buy_pct - 0.35).abs() < 0.01); + } + #[test] fn test_build_gpu_snapshot() { let samples = vec![