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

feat: epoch-level financial metrics in fxt CLI

Browse Source 
Add 11 Prometheus gauges for epoch financial metrics (Sharpe, Sortino,
win rate, max DD, profit factor, total return, avg return, total trades,
action distribution) pushed from DQN/PPO trainers at epoch end.

- common: 11 new gauges + set_epoch_financial_metrics/set_epoch_action_distribution helpers
- ml: compute_epoch_financials helper from PnL history + action counts
- ml: push from DQN trainer (full metrics) and PPO trainer (Sharpe proxy)
- proto: 11 new TrainingSession fields (36-46) + GetEpochHistory RPC
- monitoring_service: metric mapper + epoch history ring buffer (50 epochs)
- api_gateway: GetEpochHistory proxy forwarding
- fxt watch: Sharpe/Win% columns, financial summary, sparklines, action distribution
- fxt train monitor: colored financial metrics section

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-03 16:21:41 +01:00
parent e9e6c431de 7cbcfce781
commit 4beef7069c
11 changed files with 686 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
bin/fxt/proto/monitoring.proto
View File

@@ -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;
}

40
bin/fxt/src/commands/train/monitor.rs
View File

@@ -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,
);
}
}
}

45
bin/fxt/src/commands/watch/render.rs
View File

@@ -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(

109
crates/common/src/metrics/training_metrics.rs
View File

@@ -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);
}
}

189
crates/ml/src/trainers/dqn/financials.rs Normal file
View File

@@ -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<f64>,
action_counts: &[usize; 45],
initial_capital: f64,
) -> EpochFinancials {
if pnl_history.is_empty() {
return EpochFinancials::default();
}
let returns: Vec<f64> = 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::<f64>() / 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<f64> = 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::<f64>()
/ 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<f64> = 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<f64> = 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<f64> = 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);
}
}

1
crates/ml/src/trainers/dqn/mod.rs
View File

@@ -21,6 +21,7 @@
mod config;
mod data_loading;
mod early_stopping;
pub(crate) mod financials;
mod features;
pub mod lr_scheduler;
mod monitoring;

34
crates/ml/src/trainers/dqn/trainer.rs
View File

@@ -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);

22
crates/ml/src/trainers/ppo.rs
View File

@@ -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;

20
services/api_gateway/src/grpc/monitoring_proxy.rs
View File

@@ -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<GetEpochHistoryRequest>,
) -> Result<Response<GetEpochHistoryResponse>, 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)]

48
services/monitoring_service/proto/monitoring.proto
View File

@@ -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;
}

144
services/monitoring_service/src/service.rs
View File

@@ -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<PrometheusClient>,
default_interval: u32,
epoch_histories: Arc<RwLock<HashMap<String, VecDeque<EpochFinancialSnapshot>>>>,
last_epochs: Arc<RwLock<HashMap<String, f32>>>,
}
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<HashMap<String, VecDeque<EpochFinancialSnapshot>>>,
last_epochs: &RwLock<HashMap<String, f32>>,
) -> Result<GetLiveTrainingMetricsResponse, Status> {
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<GetLiveTrainingMetricsRequest>,
) -> Result<Response<GetLiveTrainingMetricsResponse>, 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<GetEpochHistoryRequest>,
) -> Result<Response<GetEpochHistoryResponse>, 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<Trai
"foxhunt_hyperopt_elapsed_seconds" => {
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![