From 5842859d78aea15bd8cd95bf9174ec4a53970ecd Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 3 Mar 2026 16:01:04 +0100 Subject: [PATCH] feat(monitoring): wire epoch financial metrics mapper + epoch history store - Fix action distribution metric names (add epoch_ prefix) - Implement epoch history ring buffer (50 epochs per session) - Wire GetEpochHistory RPC with real data - Add test for financial metric mapping Co-Authored-By: Claude Opus 4.6 --- services/monitoring_service/src/service.rs | 130 ++++++++++++++++++--- 1 file changed, 116 insertions(+), 14 deletions(-) diff --git a/services/monitoring_service/src/service.rs b/services/monitoring_service/src/service.rs index 656d22e43..510f342c6 100644 --- a/services/monitoring_service/src/service.rs +++ b/services/monitoring_service/src/service.rs @@ -1,22 +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, GetEpochHistoryRequest, - GetEpochHistoryResponse, GetLiveTrainingMetricsRequest, GetLiveTrainingMetricsResponse, - GpuSnapshot, StreamTrainingMetricsRequest, TrainingSession, + 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 { @@ -24,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(), @@ -41,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), @@ -57,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)) } @@ -76,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); @@ -96,12 +146,27 @@ impl MonitoringService for MonitoringServiceImpl { async fn get_epoch_history( &self, - _request: Request, + request: Request, ) -> Result, Status> { - // TODO(task-7): wire to epoch ring buffer storage - Err(Status::unimplemented( - "GetEpochHistory not yet wired — see task 7", - )) + 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, + })) } } @@ -209,9 +274,9 @@ fn group_into_sessions(samples: &[MetricSample], model_filter: &str) -> Vec session.action_buy_pct = s.value as f32, - "foxhunt_training_action_sell_pct" => session.action_sell_pct = s.value as f32, - "foxhunt_training_action_hold_pct" => session.action_hold_pct = s.value as f32, + "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, _ => {} } } @@ -409,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![