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

feat(monitoring): wire epoch financial metrics mapper + epoch history store

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-03 16:01:04 +01:00
parent a471c913d7
commit 5842859d78
1 changed files with 116 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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<PrometheusClient>,
default_interval: u32,
epoch_histories: Arc<RwLock<HashMap<String, VecDeque<EpochFinancialSnapshot>>>>,
last_epochs: Arc<RwLock<HashMap<String, f32>>>,
}
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<HashMap<String, VecDeque<EpochFinancialSnapshot>>>,
last_epochs: &RwLock<HashMap<String, f32>>,
) -> Result<GetLiveTrainingMetricsResponse, Status> {
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<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))
}
@@ -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<GetEpochHistoryRequest>,
request: Request<GetEpochHistoryRequest>,
) -> Result<Response<GetEpochHistoryResponse>, 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<Trai
session.epoch_total_trades = s.value as u32;
}
// Action distribution
"foxhunt_training_action_buy_pct" => 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![