From 0a15058990cee73544479fca88bdca29cbbb0d08 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Tue, 3 Mar 2026 01:39:06 +0100 Subject: [PATCH] feat(fxt): implement fxt watch streaming dashboard - streams.rs: gRPC stream manager with reconnect + exponential backoff (training metrics, order updates, risk alerts, system status) - render.rs: ratatui renderer for 4 tabs (Training/Trading/Risk/System) - event_loop.rs: tokio::select! loop over gRPC streams, crossterm keys, 200ms tick - Wired Watch command into CLI (main.rs Commands enum + dispatch) - 35 tests pass, 0 clippy warnings Co-Authored-By: Claude Opus 4.6 --- bin/fxt/src/commands/watch/event_loop.rs | 610 +++++++++++++++++- bin/fxt/src/commands/watch/render.rs | 537 +++++++++++++++- bin/fxt/src/commands/watch/streams.rs | 766 ++++++++++++++++++++++- bin/fxt/src/main.rs | 7 + 4 files changed, 1915 insertions(+), 5 deletions(-) diff --git a/bin/fxt/src/commands/watch/event_loop.rs b/bin/fxt/src/commands/watch/event_loop.rs index 306a241bb..c94df8518 100644 --- a/bin/fxt/src/commands/watch/event_loop.rs +++ b/bin/fxt/src/commands/watch/event_loop.rs @@ -1,5 +1,609 @@ -use anyhow::Result; +//! Event loop for the `fxt watch` streaming TUI dashboard. +//! +//! [`run_dashboard`] is the public entry point that sets up the terminal, +//! runs the select-loop, and guarantees clean-up on exit or error. -pub(super) async fn run_dashboard(_api_gateway_url: &str, _jwt_token: &str) -> Result<()> { - anyhow::bail!("Dashboard event loop not yet implemented") +use anyhow::{Context, Result}; +use crossterm::{ + event::{Event, EventStream, KeyCode, KeyEvent, KeyModifiers}, + execute, + terminal::{disable_raw_mode, enable_raw_mode, EnterAlternateScreen, LeaveAlternateScreen}, +}; +use ratatui::prelude::*; +use std::io; +use tokio::time::{interval, Duration}; +use tokio_stream::StreamExt; + +use super::render; +use super::state::{ + CircuitBreakerRow, DashboardState, FillRow, ServiceRow, +}; +use super::streams::{self, StreamEvent}; + +// --------------------------------------------------------------------------- +// Public entry point +// --------------------------------------------------------------------------- + +/// Initialise the terminal, run the event loop, and restore the terminal on +/// exit (even if an error occurred). +pub(super) async fn run_dashboard(api_gateway_url: &str, jwt_token: &str) -> Result<()> { + enable_raw_mode().context("failed to enable raw mode")?; + + let mut stdout = io::stdout(); + execute!(stdout, EnterAlternateScreen).context("failed to enter alternate screen")?; + + let backend = CrosstermBackend::new(stdout); + let mut terminal = + Terminal::new(backend).context("failed to create ratatui terminal")?; + + let result = run_event_loop(&mut terminal, api_gateway_url, jwt_token).await; + + // Always restore the terminal, regardless of whether the loop succeeded. + disable_raw_mode().ok(); + execute!(terminal.backend_mut(), LeaveAlternateScreen).ok(); + terminal.show_cursor().ok(); + + result +} + +// --------------------------------------------------------------------------- +// Core event loop +// --------------------------------------------------------------------------- + +async fn run_event_loop( + terminal: &mut Terminal>, + api_gateway_url: &str, + jwt_token: &str, +) -> Result<()> { + let mut state = DashboardState { + dirty: true, + ..DashboardState::default() + }; + + let mut stream_rx = streams::spawn_all_streams(api_gateway_url, jwt_token); + let mut tick = interval(Duration::from_millis(200)); + let mut event_stream = EventStream::new(); + + loop { + tokio::select! { + maybe_event = stream_rx.recv() => { + if let Some(event) = maybe_event { + apply_stream_event(&mut state, event); + } + // When None, all stream producers dropped — keep running so + // the user can still view stale data and quit manually. + } + + maybe_crossterm = event_stream.next() => { + if let Some(Ok(Event::Key(key))) = maybe_crossterm { + if handle_key(&mut state, key) { + return Ok(()); + } + } + } + + _ = tick.tick() => { + if state.dirty { + terminal + .draw(|f| render::render(f, &state)) + .context("failed to draw frame")?; + state.dirty = false; + } + } + } + } +} + +// --------------------------------------------------------------------------- +// Stream event application +// --------------------------------------------------------------------------- + +fn apply_stream_event(state: &mut DashboardState, event: StreamEvent) { + match event { + StreamEvent::TrainingUpdate { sessions, gpu } => { + // Push first session's epoch_loss to loss sparkline. + if let Some(first) = sessions.first() { + state.training.push_loss(f64::from(first.epoch_loss)); + } + state.training.sessions = sessions; + state.training.gpu = gpu.clone(); + state.system.gpu = gpu; + state.training.connected = true; + } + + StreamEvent::OrderUpdate { + order_id: _, + symbol, + status, + filled_qty, + last_fill_price, + timestamp_nanos, + } => { + let fill = FillRow { + time: format_nanos(timestamp_nanos), + side: if filled_qty >= 0.0 { + "BUY".to_owned() + } else { + "SELL".to_owned() + }, + symbol, + quantity: filled_qty.abs(), + price: last_fill_price, + status, + }; + state.trading.push_fill(fill); + state.trading.connected = true; + } + + StreamEvent::PositionsSnapshot { positions } => { + let unrealized: f64 = positions.iter().map(|p| p.unrealized_pnl).sum(); + state.trading.positions = positions; + state.trading.unrealized_pnl = unrealized; + state.trading.connected = true; + } + + StreamEvent::RiskMetrics { + var, + max_drawdown, + current_drawdown, + sharpe, + } => { + state.risk.portfolio_var = var; + state.risk.max_drawdown = max_drawdown; + state.risk.current_drawdown = current_drawdown; + state.risk.sharpe_ratio = sharpe; + state.risk.connected = true; + } + + StreamEvent::RiskAlert { + severity: _, + symbol: _, + message, + threshold, + current, + } => { + let status = if current < threshold { + "OK".to_owned() + } else { + "TRIPPED".to_owned() + }; + + // Upsert: find existing breaker by name or insert new. + if let Some(cb) = state + .risk + .circuit_breakers + .iter_mut() + .find(|cb| cb.name == message) + { + cb.status = status; + cb.current = format!("{current:.4}"); + cb.limit = format!("{threshold:.4}"); + } else { + state.risk.circuit_breakers.push(CircuitBreakerRow { + name: message, + status, + current: format!("{current:.4}"), + limit: format!("{threshold:.4}"), + }); + } + state.risk.connected = true; + } + + StreamEvent::SystemStatus { + service, + status, + message, + } => { + // Upsert: find existing service by name or insert new. + if let Some(svc) = state + .system + .services + .iter_mut() + .find(|s| s.name == service) + { + svc.status = status; + svc.message = message; + } else { + state.system.services.push(ServiceRow { + name: service, + status, + message, + latency_ms: 0.0, + }); + } + state.system.connected = true; + } + + StreamEvent::StreamDisconnected { stream_name } => { + match stream_name.as_str() { + "training" | "training_metrics" => state.training.connected = false, + "trading" | "orders" | "order_updates" => state.trading.connected = false, + "positions" => state.trading.connected = false, + "risk" | "risk_alerts" => state.risk.connected = false, + "system" | "system_status" => state.system.connected = false, + _ => {} + } + } + } + + state.dirty = true; +} + +// --------------------------------------------------------------------------- +// Key handling +// --------------------------------------------------------------------------- + +/// Returns `true` if the user wants to quit. +#[allow(clippy::wildcard_enum_match_arm)] // KeyCode has 25+ variants; exhaustive match is noise +fn handle_key(state: &mut DashboardState, key: KeyEvent) -> bool { + match key.code { + KeyCode::Char('q') => return true, + KeyCode::Char('c') if key.modifiers.contains(KeyModifiers::CONTROL) => return true, + KeyCode::Char('1') => state.select_tab(0), + KeyCode::Char('2') => state.select_tab(1), + KeyCode::Char('3') => state.select_tab(2), + KeyCode::Char('4') => state.select_tab(3), + KeyCode::Char('j') | KeyCode::Down => state.scroll_down(), + KeyCode::Char('k') | KeyCode::Up => state.scroll_up(), + _ => {} + } + false +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Format a nanosecond Unix timestamp as `HH:MM:SS`. +#[allow(clippy::modulo_arithmetic)] // Intentional clock arithmetic on non-negative quotients +fn format_nanos(nanos: i64) -> String { + let secs = nanos / 1_000_000_000; + let hour = (secs / 3600).rem_euclid(24); + let min = (secs / 60).rem_euclid(60); + let sec = secs.rem_euclid(60); + format!("{hour:02}:{min:02}:{sec:02}") +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + use super::super::state::GpuInfo; + use super::super::state::PositionRow; + use super::super::state::TrainingSession; + use crossterm::event::KeyEvent; + + // -- Key handling -------------------------------------------------------- + + #[test] + fn test_handle_key_quit() { + let mut state = DashboardState::default(); + let key = KeyEvent::new(KeyCode::Char('q'), KeyModifiers::NONE); + assert!(handle_key(&mut state, key)); + } + + #[test] + fn test_handle_key_ctrl_c_quit() { + let mut state = DashboardState::default(); + let key = KeyEvent::new(KeyCode::Char('c'), KeyModifiers::CONTROL); + assert!(handle_key(&mut state, key)); + } + + #[test] + fn test_handle_key_tab_switch() { + let mut state = DashboardState::default(); + let key = KeyEvent::new(KeyCode::Char('3'), KeyModifiers::NONE); + assert!(!handle_key(&mut state, key)); + assert_eq!(state.active_tab, 2); + } + + #[test] + fn test_handle_key_scroll() { + let mut state = DashboardState::default(); + // Scroll down with 'j'. + let j = KeyEvent::new(KeyCode::Char('j'), KeyModifiers::NONE); + handle_key(&mut state, j); + assert_eq!(state.training.scroll_offset, 1); + + // Scroll up with 'k'. + let k = KeyEvent::new(KeyCode::Char('k'), KeyModifiers::NONE); + handle_key(&mut state, k); + assert_eq!(state.training.scroll_offset, 0); + } + + #[test] + fn test_handle_key_arrow_scroll() { + let mut state = DashboardState::default(); + let down = KeyEvent::new(KeyCode::Down, KeyModifiers::NONE); + handle_key(&mut state, down); + assert_eq!(state.training.scroll_offset, 1); + + let up = KeyEvent::new(KeyCode::Up, KeyModifiers::NONE); + handle_key(&mut state, up); + assert_eq!(state.training.scroll_offset, 0); + } + + #[test] + fn test_handle_key_non_quit_returns_false() { + let mut state = DashboardState::default(); + let key = KeyEvent::new(KeyCode::Char('x'), KeyModifiers::NONE); + assert!(!handle_key(&mut state, key)); + } + + // -- format_nanos -------------------------------------------------------- + + #[test] + fn test_format_nanos() { + assert_eq!(format_nanos(52_321_000_000_000), "14:32:01"); + } + + #[test] + fn test_format_nanos_zero() { + assert_eq!(format_nanos(0), "00:00:00"); + } + + #[test] + fn test_format_nanos_midnight() { + // Exactly 24 hours in nanos wraps to 00:00:00. + let nanos = 24 * 3600 * 1_000_000_000_i64; + assert_eq!(format_nanos(nanos), "00:00:00"); + } + + // -- apply_stream_event -------------------------------------------------- + + #[test] + fn test_apply_training_update() { + let mut state = DashboardState::default(); + let session = TrainingSession { + model: "DQN".into(), + fold: "fold0".into(), + is_hyperopt: false, + epoch: 5.0, + epoch_loss: 0.123, + validation_loss: 0.456, + learning_rate: 1e-3, + gpu_percent: 85.0, + batches_per_second: 42.0, + gradient_norm: 1.5, + epoch_duration_seconds: 12.3, + }; + let gpu = GpuInfo { + utilization_percent: 85.0, + memory_used_mb: 3000.0, + memory_total_mb: 48000.0, + temperature_celsius: 62.0, + }; + + apply_stream_event( + &mut state, + StreamEvent::TrainingUpdate { + sessions: vec![session], + gpu: gpu.clone(), + }, + ); + + assert!(state.training.connected); + assert_eq!(state.training.sessions.len(), 1); + assert_eq!(state.training.loss_history.len(), 1); + // Loss should be epoch_loss as f64. + assert!((state.training.loss_history.back().unwrap() - 0.123_f64).abs() < 1e-3); + // GPU should be cloned to both training and system. + assert!((state.system.gpu.utilization_percent - 85.0).abs() < f32::EPSILON); + assert!(state.dirty); + } + + #[test] + fn test_apply_order_update() { + let mut state = DashboardState::default(); + apply_stream_event( + &mut state, + StreamEvent::OrderUpdate { + order_id: "ord-1".into(), + symbol: "ES".into(), + status: "Filled".into(), + filled_qty: 2.0, + last_fill_price: 5100.25, + timestamp_nanos: 52_321_000_000_000, + }, + ); + + assert!(state.trading.connected); + assert_eq!(state.trading.fills.len(), 1); + let fill = state.trading.fills.front().unwrap(); + assert_eq!(fill.time, "14:32:01"); + assert_eq!(fill.side, "BUY"); + assert_eq!(fill.symbol, "ES"); + assert!((fill.quantity - 2.0).abs() < f64::EPSILON); + } + + #[test] + fn test_apply_order_update_sell() { + let mut state = DashboardState::default(); + apply_stream_event( + &mut state, + StreamEvent::OrderUpdate { + order_id: "ord-2".into(), + symbol: "NQ".into(), + status: "Filled".into(), + filled_qty: -1.0, + last_fill_price: 18000.0, + timestamp_nanos: 0, + }, + ); + + let fill = state.trading.fills.front().unwrap(); + assert_eq!(fill.side, "SELL"); + assert!((fill.quantity - 1.0).abs() < f64::EPSILON); + } + + #[test] + fn test_apply_positions_snapshot() { + let mut state = DashboardState::default(); + let positions = vec![ + PositionRow { + symbol: "ES".into(), + side: "LONG".into(), + quantity: 2.0, + entry_price: 5000.0, + unrealized_pnl: 150.0, + status: "Open".into(), + }, + PositionRow { + symbol: "NQ".into(), + side: "SHORT".into(), + quantity: 1.0, + entry_price: 18000.0, + unrealized_pnl: -50.0, + status: "Open".into(), + }, + ]; + + apply_stream_event( + &mut state, + StreamEvent::PositionsSnapshot { + positions: positions.clone(), + }, + ); + + assert!(state.trading.connected); + assert_eq!(state.trading.positions.len(), 2); + assert!((state.trading.unrealized_pnl - 100.0).abs() < f64::EPSILON); + } + + #[test] + fn test_apply_risk_metrics() { + let mut state = DashboardState::default(); + apply_stream_event( + &mut state, + StreamEvent::RiskMetrics { + var: 0.02, + max_drawdown: 0.05, + current_drawdown: 0.01, + sharpe: 2.5, + }, + ); + + assert!(state.risk.connected); + assert!((state.risk.portfolio_var - 0.02).abs() < f64::EPSILON); + assert!((state.risk.max_drawdown - 0.05).abs() < f64::EPSILON); + assert!((state.risk.current_drawdown - 0.01).abs() < f64::EPSILON); + assert!((state.risk.sharpe_ratio - 2.5).abs() < f64::EPSILON); + } + + #[test] + fn test_apply_risk_alert_insert() { + let mut state = DashboardState::default(); + apply_stream_event( + &mut state, + StreamEvent::RiskAlert { + severity: "WARNING".into(), + symbol: "ES".into(), + message: "max_drawdown".into(), + threshold: 0.05, + current: 0.03, + }, + ); + + assert!(state.risk.connected); + assert_eq!(state.risk.circuit_breakers.len(), 1); + assert_eq!(state.risk.circuit_breakers[0].name, "max_drawdown"); + assert_eq!(state.risk.circuit_breakers[0].status, "OK"); + } + + #[test] + fn test_apply_risk_alert_tripped() { + let mut state = DashboardState::default(); + apply_stream_event( + &mut state, + StreamEvent::RiskAlert { + severity: "CRITICAL".into(), + symbol: "ES".into(), + message: "max_drawdown".into(), + threshold: 0.05, + current: 0.06, + }, + ); + + assert_eq!(state.risk.circuit_breakers[0].status, "TRIPPED"); + } + + #[test] + fn test_apply_system_status_upsert() { + let mut state = DashboardState::default(); + + // Insert first time. + apply_stream_event( + &mut state, + StreamEvent::SystemStatus { + service: "gateway".into(), + status: "HEALTHY".into(), + message: "ok".into(), + }, + ); + assert_eq!(state.system.services.len(), 1); + + // Update same service. + apply_stream_event( + &mut state, + StreamEvent::SystemStatus { + service: "gateway".into(), + status: "DEGRADED".into(), + message: "high latency".into(), + }, + ); + + // Should still be 1 entry, not 2. + assert_eq!(state.system.services.len(), 1); + assert_eq!(state.system.services[0].status, "DEGRADED"); + assert_eq!(state.system.services[0].message, "high latency"); + assert!(state.system.connected); + } + + #[test] + fn test_apply_stream_disconnected() { + let mut state = DashboardState::default(); + state.training.connected = true; + state.trading.connected = true; + state.risk.connected = true; + state.system.connected = true; + + apply_stream_event( + &mut state, + StreamEvent::StreamDisconnected { + stream_name: "training".into(), + }, + ); + assert!(!state.training.connected); + + apply_stream_event( + &mut state, + StreamEvent::StreamDisconnected { + stream_name: "risk".into(), + }, + ); + assert!(!state.risk.connected); + + apply_stream_event( + &mut state, + StreamEvent::StreamDisconnected { + stream_name: "system".into(), + }, + ); + assert!(!state.system.connected); + + apply_stream_event( + &mut state, + StreamEvent::StreamDisconnected { + stream_name: "orders".into(), + }, + ); + assert!(!state.trading.connected); + + // Dirty flag should be set after each event. + assert!(state.dirty); + } } diff --git a/bin/fxt/src/commands/watch/render.rs b/bin/fxt/src/commands/watch/render.rs index 1039d4c50..7b1056058 100644 --- a/bin/fxt/src/commands/watch/render.rs +++ b/bin/fxt/src/commands/watch/render.rs @@ -1 +1,536 @@ -// TODO: implement +//! Ratatui renderer for the `fxt watch` dashboard. +//! +//! One public entry point ([`render`]) dispatches to per-tab helpers. +//! No tests — visual rendering tests are brittle and not worth maintaining. + +use ratatui::{ + layout::{Constraint, Direction, Layout, Rect}, + style::{Color, Modifier, Style}, + text::{Line, Span}, + widgets::{Block, Borders, Cell, Paragraph, Row, Sparkline, Table, Tabs}, + Frame, +}; + +use super::state::{ + DashboardState, RiskTabState, SystemTabState, Tab, TradingTabState, TrainingTabState, +}; + +// --------------------------------------------------------------------------- +// Public entry point +// --------------------------------------------------------------------------- + +/// Render the full dashboard into the given frame. +pub(super) fn render(frame: &mut Frame, state: &DashboardState) { + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(3), // tab bar + Constraint::Min(0), // content + Constraint::Length(1), // status bar + ]) + .split(frame.area()); + + render_tab_bar(frame, chunks[0], state); + + match state.current_tab() { + Tab::Training => render_training_tab(frame, chunks[1], &state.training), + Tab::Trading => render_trading_tab(frame, chunks[1], &state.trading), + Tab::Risk => render_risk_tab(frame, chunks[1], &state.risk), + Tab::System => render_system_tab(frame, chunks[1], &state.system), + } + + render_status_bar(frame, chunks[2]); +} + +// --------------------------------------------------------------------------- +// Tab bar +// --------------------------------------------------------------------------- + +fn render_tab_bar(frame: &mut Frame, area: Rect, state: &DashboardState) { + let titles: Vec> = Tab::ALL + .iter() + .map(|t| Line::from(Span::raw(t.title()))) + .collect(); + + let tabs = Tabs::new(titles) + .block( + Block::default() + .borders(Borders::ALL) + .title(" fxt watch "), + ) + .select(state.active_tab) + .highlight_style( + Style::default() + .fg(Color::Cyan) + .add_modifier(Modifier::BOLD), + ); + + frame.render_widget(tabs, area); +} + +// --------------------------------------------------------------------------- +// Status bar +// --------------------------------------------------------------------------- + +fn render_status_bar(frame: &mut Frame, area: Rect) { + let bar = Paragraph::new(Line::from(vec![ + Span::styled("1-4", Style::default().fg(Color::Cyan)), + Span::raw(" tabs "), + Span::styled("j/k", Style::default().fg(Color::Cyan)), + Span::raw(" scroll "), + Span::styled("q", Style::default().fg(Color::Cyan)), + Span::raw(" quit"), + ])); + frame.render_widget(bar, area); +} + +// --------------------------------------------------------------------------- +// Training tab +// --------------------------------------------------------------------------- + +fn render_training_tab(frame: &mut Frame, area: Rect, tab: &TrainingTabState) { + let conn_label = if tab.connected { + "connected" + } else { + "disconnected" + }; + + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Min(0), // sessions table + Constraint::Length(6), // GPU + sparkline + ]) + .split(area); + + // -- Sessions table --------------------------------------------------- + let header = Row::new(vec![ + Cell::from("Model"), + Cell::from("Fold"), + Cell::from("Epoch"), + Cell::from("Loss"), + Cell::from("Val Loss"), + Cell::from("LR"), + Cell::from("Batch/s"), + Cell::from("Grad"), + ]) + .style( + Style::default() + .fg(Color::Yellow) + .add_modifier(Modifier::BOLD), + ); + + let rows: Vec> = tab + .sessions + .iter() + .skip(tab.scroll_offset) + .map(|s| { + Row::new(vec![ + Cell::from(s.model.as_str()), + Cell::from(s.fold.as_str()), + Cell::from(format!("{:.1}", s.epoch)), + Cell::from(format!("{:.4}", s.epoch_loss)), + Cell::from(format!("{:.4}", s.validation_loss)), + Cell::from(format!("{:.2e}", s.learning_rate)), + Cell::from(format!("{:.1}", s.batches_per_second)), + Cell::from(format!("{:.3}", s.gradient_norm)), + ]) + }) + .collect(); + + let table = Table::new( + rows, + [ + Constraint::Length(10), + Constraint::Length(8), + Constraint::Length(8), + Constraint::Length(10), + Constraint::Length(10), + Constraint::Length(10), + Constraint::Length(8), + Constraint::Length(8), + ], + ) + .header(header) + .block( + Block::default() + .borders(Borders::ALL) + .title(format!(" Training [{conn_label}] ")), + ); + + frame.render_widget(table, chunks[0]); + + // -- Bottom row: GPU info (left) + loss sparkline (right) ------------- + let bottom = Layout::default() + .direction(Direction::Horizontal) + .constraints([Constraint::Percentage(50), Constraint::Percentage(50)]) + .split(chunks[1]); + + let gpu = &tab.gpu; + let gpu_text = format!( + "GPU: {:.0}% VRAM: {:.0}/{:.0} MB Temp: {:.0}C", + gpu.utilization_percent, gpu.memory_used_mb, gpu.memory_total_mb, gpu.temperature_celsius, + ); + let gpu_para = Paragraph::new(gpu_text).block( + Block::default() + .borders(Borders::ALL) + .title(" GPU "), + ); + frame.render_widget(gpu_para, bottom[0]); + + // Sparkline: clamp to 0..10, multiply by 100 for u64 resolution. + let spark_data: Vec = tab + .loss_history + .iter() + .map(|&v| { + let clamped = v.clamp(0.0, 10.0); + (clamped * 100.0) as u64 + }) + .collect(); + + let sparkline = Sparkline::default() + .block( + Block::default() + .borders(Borders::ALL) + .title(" Loss "), + ) + .data(&spark_data) + .style(Style::default().fg(Color::Cyan)); + + frame.render_widget(sparkline, bottom[1]); +} + +// --------------------------------------------------------------------------- +// Trading tab +// --------------------------------------------------------------------------- + +fn render_trading_tab(frame: &mut Frame, area: Rect, tab: &TradingTabState) { + let conn_label = if tab.connected { + "connected" + } else { + "disconnected" + }; + + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Percentage(40), // positions + Constraint::Percentage(40), // fills + Constraint::Length(3), // PnL bar + ]) + .split(area); + + // -- Positions table -------------------------------------------------- + let pos_header = Row::new(vec![ + Cell::from("Symbol"), + Cell::from("Side"), + Cell::from("Qty"), + Cell::from("Entry"), + Cell::from("PnL"), + Cell::from("Status"), + ]) + .style( + Style::default() + .fg(Color::Yellow) + .add_modifier(Modifier::BOLD), + ); + + let pos_rows: Vec> = tab + .positions + .iter() + .map(|p| { + let pnl_color = if p.unrealized_pnl >= 0.0 { + Color::Green + } else { + Color::Red + }; + Row::new(vec![ + Cell::from(p.symbol.as_str()), + Cell::from(p.side.as_str()), + Cell::from(format!("{:.0}", p.quantity)), + Cell::from(format!("{:.2}", p.entry_price)), + Cell::from(Span::styled( + format!("{:+.2}", p.unrealized_pnl), + Style::default().fg(pnl_color), + )), + Cell::from(p.status.as_str()), + ]) + }) + .collect(); + + let pos_table = Table::new( + pos_rows, + [ + Constraint::Length(10), + Constraint::Length(6), + Constraint::Length(6), + Constraint::Length(10), + Constraint::Length(12), + Constraint::Length(8), + ], + ) + .header(pos_header) + .block( + Block::default() + .borders(Borders::ALL) + .title(format!(" Positions [{conn_label}] ")), + ); + + frame.render_widget(pos_table, chunks[0]); + + // -- Fills table ------------------------------------------------------ + let fill_header = Row::new(vec![ + Cell::from("Time"), + Cell::from("Side"), + Cell::from("Symbol"), + Cell::from("Qty"), + Cell::from("Price"), + Cell::from("Status"), + ]) + .style( + Style::default() + .fg(Color::Yellow) + .add_modifier(Modifier::BOLD), + ); + + let fill_rows: Vec> = tab + .fills + .iter() + .skip(tab.scroll_offset) + .take(10) + .map(|f| { + Row::new(vec![ + Cell::from(f.time.as_str()), + Cell::from(f.side.as_str()), + Cell::from(f.symbol.as_str()), + Cell::from(format!("{:.0}", f.quantity)), + Cell::from(format!("{:.2}", f.price)), + Cell::from(f.status.as_str()), + ]) + }) + .collect(); + + let fill_table = Table::new( + fill_rows, + [ + Constraint::Length(10), + Constraint::Length(6), + Constraint::Length(10), + Constraint::Length(6), + Constraint::Length(10), + Constraint::Length(8), + ], + ) + .header(fill_header) + .block( + Block::default() + .borders(Borders::ALL) + .title(" Fills "), + ); + + frame.render_widget(fill_table, chunks[1]); + + // -- PnL bar ---------------------------------------------------------- + let pnl_color = if tab.day_pnl >= 0.0 { + Color::Green + } else { + Color::Red + }; + let pnl_text = Line::from(vec![Span::styled( + format!( + "Day PnL: {:+.2} | Realized: {:+.2} Unrealized: {:+.2}", + tab.day_pnl, tab.realized_pnl, tab.unrealized_pnl, + ), + Style::default().fg(pnl_color), + )]); + let pnl_bar = Paragraph::new(pnl_text).block( + Block::default() + .borders(Borders::ALL) + .title(" PnL "), + ); + frame.render_widget(pnl_bar, chunks[2]); +} + +// --------------------------------------------------------------------------- +// Risk tab +// --------------------------------------------------------------------------- + +fn render_risk_tab(frame: &mut Frame, area: Rect, tab: &RiskTabState) { + let conn_label = if tab.connected { + "connected" + } else { + "disconnected" + }; + + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Length(5), // summary + Constraint::Min(0), // circuit breakers + ]) + .split(area); + + // -- Metrics summary -------------------------------------------------- + let summary = Paragraph::new(vec![ + Line::from(format!( + " VaR: {:.2} Max Drawdown: {:.2}%", + tab.portfolio_var, + tab.max_drawdown * 100.0, + )), + Line::from(format!( + " Sharpe: {:.2} Current Drawdown: {:.2}%", + tab.sharpe_ratio, + tab.current_drawdown * 100.0, + )), + ]) + .block( + Block::default() + .borders(Borders::ALL) + .title(format!(" Risk [{conn_label}] ")), + ); + frame.render_widget(summary, chunks[0]); + + // -- Circuit breakers table ------------------------------------------- + let cb_header = Row::new(vec![ + Cell::from("Breaker"), + Cell::from("Status"), + Cell::from("Current"), + Cell::from("Limit"), + ]) + .style( + Style::default() + .fg(Color::Yellow) + .add_modifier(Modifier::BOLD), + ); + + let cb_rows: Vec> = tab + .circuit_breakers + .iter() + .skip(tab.scroll_offset) + .map(|cb| { + let status_color = if cb.status == "OK" { + Color::Green + } else { + Color::Red + }; + Row::new(vec![ + Cell::from(cb.name.as_str()), + Cell::from(Span::styled( + cb.status.as_str(), + Style::default().fg(status_color), + )), + Cell::from(cb.current.as_str()), + Cell::from(cb.limit.as_str()), + ]) + }) + .collect(); + + let cb_table = Table::new( + cb_rows, + [ + Constraint::Length(18), + Constraint::Length(10), + Constraint::Length(15), + Constraint::Length(15), + ], + ) + .header(cb_header) + .block( + Block::default() + .borders(Borders::ALL) + .title(" Circuit Breakers "), + ); + + frame.render_widget(cb_table, chunks[1]); +} + +// --------------------------------------------------------------------------- +// System tab +// --------------------------------------------------------------------------- + +fn render_system_tab(frame: &mut Frame, area: Rect, tab: &SystemTabState) { + let conn_label = if tab.connected { + "connected" + } else { + "disconnected" + }; + + let chunks = Layout::default() + .direction(Direction::Vertical) + .constraints([ + Constraint::Min(0), // services table + Constraint::Length(5), // resources bar + ]) + .split(area); + + // -- Services table --------------------------------------------------- + let svc_header = Row::new(vec![ + Cell::from("Service"), + Cell::from("Status"), + Cell::from("Latency"), + Cell::from("Message"), + ]) + .style( + Style::default() + .fg(Color::Yellow) + .add_modifier(Modifier::BOLD), + ); + + let svc_rows: Vec> = tab + .services + .iter() + .skip(tab.scroll_offset) + .map(|s| { + let status_color = match s.status.as_str() { + "HEALTHY" | "UP" => Color::Green, + "DEGRADED" => Color::Yellow, + _ => Color::Red, + }; + Row::new(vec![ + Cell::from(s.name.as_str()), + Cell::from(Span::styled( + s.status.as_str(), + Style::default().fg(status_color), + )), + Cell::from(format!("{:.1}ms", s.latency_ms)), + Cell::from(s.message.as_str()), + ]) + }) + .collect(); + + let svc_table = Table::new( + svc_rows, + [ + Constraint::Length(22), + Constraint::Length(12), + Constraint::Length(10), + Constraint::Min(0), + ], + ) + .header(svc_header) + .block( + Block::default() + .borders(Borders::ALL) + .title(format!(" Services [{conn_label}] ")), + ); + + frame.render_widget(svc_table, chunks[0]); + + // -- Resources bar ---------------------------------------------------- + let gpu = &tab.gpu; + let resources = Paragraph::new(format!( + " GPU: {:.0}% VRAM: {:.0}/{:.0} MB CPU: {:.1}% RAM: {:.1}/{:.1} GB", + gpu.utilization_percent, + gpu.memory_used_mb, + gpu.memory_total_mb, + tab.cpu_percent, + tab.ram_used_gb, + tab.ram_total_gb, + )) + .block( + Block::default() + .borders(Borders::ALL) + .title(" Resources "), + ); + frame.render_widget(resources, chunks[1]); +} diff --git a/bin/fxt/src/commands/watch/streams.rs b/bin/fxt/src/commands/watch/streams.rs index 1039d4c50..409627cf2 100644 --- a/bin/fxt/src/commands/watch/streams.rs +++ b/bin/fxt/src/commands/watch/streams.rs @@ -1 +1,765 @@ -// TODO: implement +//! gRPC stream manager for `fxt watch`. +//! +//! Spawns one tokio task per server-streaming RPC plus one one-shot task for +//! initial state (positions + risk metrics). All tasks funnel [`StreamEvent`]s +//! into a single `mpsc` channel that the event loop consumes. +//! +//! Every streaming task runs an infinite retry loop with exponential backoff +//! (1 s initial, 30 s cap) so the dashboard stays alive across transient +//! network failures. + +use anyhow::{Context, Result}; +use tokio::sync::mpsc; +use tokio_stream::StreamExt; +use tonic::metadata::{Ascii, MetadataValue}; +use tonic::transport::Channel; +use tracing::{debug, warn}; + +use crate::proto::monitoring::{ + self as mon, monitoring_service_client::MonitoringServiceClient, +}; +use crate::proto::trading::{self as trd, trading_service_client::TradingServiceClient}; + +use super::state::{GpuInfo, PositionRow, TrainingSession}; + +// --------------------------------------------------------------------------- +// StreamEvent — the single enum consumed by the event loop +// --------------------------------------------------------------------------- + +/// Typed events produced by the individual gRPC stream tasks. +#[derive(Debug, Clone)] +#[allow(dead_code)] // Fields like order_id, severity, symbol reserved for future UI use +pub(super) enum StreamEvent { + /// Training telemetry snapshot (sessions + GPU). + TrainingUpdate { + sessions: Vec, + gpu: GpuInfo, + }, + + /// A single order/fill update. + OrderUpdate { + order_id: String, + symbol: String, + status: String, + filled_qty: f64, + last_fill_price: f64, + timestamp_nanos: i64, + }, + + /// Full positions snapshot (replaces the previous set). + PositionsSnapshot { positions: Vec }, + + /// Periodic risk-metric broadcast. + RiskMetrics { + var: f64, + max_drawdown: f64, + current_drawdown: f64, + sharpe: f64, + }, + + /// Risk circuit-breaker alert. + RiskAlert { + severity: String, + symbol: String, + message: String, + threshold: f64, + current: f64, + }, + + /// Service health heartbeat. + SystemStatus { + service: String, + status: String, + message: String, + }, + + /// Indicates that a named stream has disconnected. + StreamDisconnected { stream_name: String }, +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Create a lazily-connected gRPC channel to the given URL. +fn connect_channel(url: &str) -> Result { + Ok(Channel::from_shared(url.to_owned()) + .context("invalid gRPC endpoint URL")? + .connect_lazy()) +} + +/// Parse a JWT string into a tonic `MetadataValue` suitable for the +/// `authorization` header. +fn auth_metadata(jwt: &str) -> Result> { + let bearer = format!("Bearer {jwt}"); + bearer + .parse::>() + .context("JWT token contains invalid header characters") +} + +// --------------------------------------------------------------------------- +// Backoff helper +// --------------------------------------------------------------------------- + +/// Capped exponential backoff: doubles each call, capped at 30 s. +fn next_backoff(current: std::time::Duration) -> std::time::Duration { + let doubled = current.saturating_mul(2); + let cap = std::time::Duration::from_secs(30); + if doubled > cap { + cap + } else { + doubled + } +} + +const INITIAL_BACKOFF: std::time::Duration = std::time::Duration::from_secs(1); + +// --------------------------------------------------------------------------- +// Proto -> state conversions +// --------------------------------------------------------------------------- + +fn convert_training_session(s: &mon::TrainingSession) -> TrainingSession { + TrainingSession { + model: s.model.clone(), + fold: s.fold.clone(), + is_hyperopt: s.is_hyperopt, + epoch: s.current_epoch, + epoch_loss: s.epoch_loss, + validation_loss: s.validation_loss, + learning_rate: s.learning_rate, + gpu_percent: 0.0, // per-session GPU not in proto; filled from GpuSnapshot + batches_per_second: s.batches_per_second, + gradient_norm: s.gradient_norm, + epoch_duration_seconds: s.epoch_duration_seconds, + } +} + +fn convert_gpu(g: &mon::GpuSnapshot) -> GpuInfo { + GpuInfo { + utilization_percent: g.utilization_percent, + memory_used_mb: g.memory_used_mb, + memory_total_mb: g.memory_total_mb, + temperature_celsius: g.temperature_celsius, + } +} + +fn convert_position(p: &trd::Position) -> PositionRow { + let side = if p.quantity >= 0.0 { "Long" } else { "Short" }; + PositionRow { + symbol: p.symbol.clone(), + side: side.to_owned(), + quantity: p.quantity.abs(), + entry_price: p.average_cost, + unrealized_pnl: p.unrealized_pnl, + status: "Open".to_owned(), + } +} + +// --------------------------------------------------------------------------- +// Public entry point +// --------------------------------------------------------------------------- + +/// Spawn background tasks for every dashboard stream and return the unified +/// receiver. +/// +/// Five tasks are spawned: +/// 1. Training metrics stream (monitoring service) +/// 2. Order updates stream (trading service) +/// 3. Risk alerts stream (trading service) +/// 4. System status stream (trading service) +/// 5. Initial state fetch -- one-shot unary RPCs for positions + risk metrics +/// +/// Each streaming task will reconnect automatically on transient failures and +/// emit [`StreamEvent::StreamDisconnected`] when the connection drops. +pub(super) fn spawn_all_streams( + api_gateway_url: &str, + jwt_token: &str, +) -> mpsc::Receiver { + let (tx, rx) = mpsc::channel::(256); + + // 1. Training metrics stream + { + let url = api_gateway_url.to_owned(); + let jwt = jwt_token.to_owned(); + let sender = tx.clone(); + tokio::spawn(async move { + stream_training_loop(&url, &jwt, &sender).await; + }); + } + + // 2. Order updates stream + { + let url = api_gateway_url.to_owned(); + let jwt = jwt_token.to_owned(); + let sender = tx.clone(); + tokio::spawn(async move { + stream_orders_loop(&url, &jwt, &sender).await; + }); + } + + // 3. Risk alerts stream + { + let url = api_gateway_url.to_owned(); + let jwt = jwt_token.to_owned(); + let sender = tx.clone(); + tokio::spawn(async move { + stream_risk_alerts_loop(&url, &jwt, &sender).await; + }); + } + + // 4. System status stream + { + let url = api_gateway_url.to_owned(); + let jwt = jwt_token.to_owned(); + let sender = tx.clone(); + tokio::spawn(async move { + stream_system_status_loop(&url, &jwt, &sender).await; + }); + } + + // 5. One-shot initial state + { + let url = api_gateway_url.to_owned(); + let jwt = jwt_token.to_owned(); + // Last clone — `tx` itself is dropped after the block, leaving only + // the spawned tasks holding senders. + tokio::spawn(async move { + if let Err(e) = fetch_initial_state(&url, &jwt, &tx).await { + warn!(error = %e, "failed to fetch initial state"); + } + }); + } + + rx +} + +// --------------------------------------------------------------------------- +// Training metrics stream +// --------------------------------------------------------------------------- + +async fn stream_training_loop(url: &str, jwt: &str, tx: &mpsc::Sender) -> ! { + let mut backoff = INITIAL_BACKOFF; + loop { + match try_stream_training(url, jwt, tx).await { + Ok(()) => { + // Stream ended cleanly (server closed) -- reset backoff. + backoff = INITIAL_BACKOFF; + } + Err(e) => { + warn!(error = %e, "training metrics stream error"); + } + } + drop( + tx.send(StreamEvent::StreamDisconnected { + stream_name: "training_metrics".to_owned(), + }) + .await, + ); + debug!(backoff_ms = backoff.as_millis(), "reconnecting training stream"); + tokio::time::sleep(backoff).await; + backoff = next_backoff(backoff); + } +} + +async fn try_stream_training( + url: &str, + jwt: &str, + tx: &mpsc::Sender, +) -> Result<()> { + let channel = connect_channel(url)?; + let token = auth_metadata(jwt)?; + + let mut client = MonitoringServiceClient::with_interceptor( + channel, + move |mut req: tonic::Request<()>| { + req.metadata_mut() + .insert("authorization", token.clone()); + Ok(req) + }, + ); + + let request = mon::StreamTrainingMetricsRequest { + model_filter: String::new(), + interval_seconds: 3, + }; + + let response = client + .stream_training_metrics(request) + .await + .context("StreamTrainingMetrics RPC failed")?; + + let mut stream = response.into_inner(); + + while let Some(frame) = stream.next().await { + let snapshot = frame.context("training metrics stream message error")?; + let sessions: Vec = + snapshot.sessions.iter().map(convert_training_session).collect(); + let gpu = snapshot.gpu.as_ref().map(convert_gpu).unwrap_or_default(); + + tx.send(StreamEvent::TrainingUpdate { sessions, gpu }) + .await + .context("event channel closed")?; + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Order updates stream +// --------------------------------------------------------------------------- + +async fn stream_orders_loop(url: &str, jwt: &str, tx: &mpsc::Sender) -> ! { + let mut backoff = INITIAL_BACKOFF; + loop { + match try_stream_orders(url, jwt, tx).await { + Ok(()) => { + backoff = INITIAL_BACKOFF; + } + Err(e) => { + warn!(error = %e, "order updates stream error"); + } + } + drop( + tx.send(StreamEvent::StreamDisconnected { + stream_name: "order_updates".to_owned(), + }) + .await, + ); + debug!(backoff_ms = backoff.as_millis(), "reconnecting orders stream"); + tokio::time::sleep(backoff).await; + backoff = next_backoff(backoff); + } +} + +async fn try_stream_orders( + url: &str, + jwt: &str, + tx: &mpsc::Sender, +) -> Result<()> { + let channel = connect_channel(url)?; + let token = auth_metadata(jwt)?; + + let mut client = TradingServiceClient::with_interceptor( + channel, + move |mut req: tonic::Request<()>| { + req.metadata_mut() + .insert("authorization", token.clone()); + Ok(req) + }, + ); + + let request = trd::SubscribeOrderUpdatesRequest { account_id: None }; + + let response = client + .subscribe_order_updates(request) + .await + .context("SubscribeOrderUpdates RPC failed")?; + + let mut stream = response.into_inner(); + + while let Some(frame) = stream.next().await { + let update = frame.context("order updates stream message error")?; + tx.send(StreamEvent::OrderUpdate { + order_id: update.order_id, + symbol: update.symbol, + status: format!("{}", update.status), + filled_qty: update.filled_quantity, + last_fill_price: update.last_fill_price, + timestamp_nanos: update.timestamp_unix_nanos, + }) + .await + .context("event channel closed")?; + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Risk alerts stream +// --------------------------------------------------------------------------- + +async fn stream_risk_alerts_loop(url: &str, jwt: &str, tx: &mpsc::Sender) -> ! { + let mut backoff = INITIAL_BACKOFF; + loop { + match try_stream_risk_alerts(url, jwt, tx).await { + Ok(()) => { + backoff = INITIAL_BACKOFF; + } + Err(e) => { + warn!(error = %e, "risk alerts stream error"); + } + } + drop( + tx.send(StreamEvent::StreamDisconnected { + stream_name: "risk_alerts".to_owned(), + }) + .await, + ); + debug!(backoff_ms = backoff.as_millis(), "reconnecting risk alerts stream"); + tokio::time::sleep(backoff).await; + backoff = next_backoff(backoff); + } +} + +async fn try_stream_risk_alerts( + url: &str, + jwt: &str, + tx: &mpsc::Sender, +) -> Result<()> { + let channel = connect_channel(url)?; + let token = auth_metadata(jwt)?; + + let mut client = TradingServiceClient::with_interceptor( + channel, + move |mut req: tonic::Request<()>| { + req.metadata_mut() + .insert("authorization", token.clone()); + Ok(req) + }, + ); + + let request = trd::SubscribeRiskAlertsRequest { + min_severity: Vec::new(), + symbols: Vec::new(), + }; + + let response = client + .subscribe_risk_alerts(request) + .await + .context("SubscribeRiskAlerts RPC failed")?; + + let mut stream = response.into_inner(); + + while let Some(frame) = stream.next().await { + let alert = frame.context("risk alerts stream message error")?; + tx.send(StreamEvent::RiskAlert { + severity: format!("{}", alert.severity), + symbol: alert.symbol, + message: alert.message, + threshold: alert.threshold_value, + current: alert.current_value, + }) + .await + .context("event channel closed")?; + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// System status stream +// --------------------------------------------------------------------------- + +async fn stream_system_status_loop(url: &str, jwt: &str, tx: &mpsc::Sender) -> ! { + let mut backoff = INITIAL_BACKOFF; + loop { + match try_stream_system_status(url, jwt, tx).await { + Ok(()) => { + backoff = INITIAL_BACKOFF; + } + Err(e) => { + warn!(error = %e, "system status stream error"); + } + } + drop( + tx.send(StreamEvent::StreamDisconnected { + stream_name: "system_status".to_owned(), + }) + .await, + ); + debug!(backoff_ms = backoff.as_millis(), "reconnecting system status stream"); + tokio::time::sleep(backoff).await; + backoff = next_backoff(backoff); + } +} + +async fn try_stream_system_status( + url: &str, + jwt: &str, + tx: &mpsc::Sender, +) -> Result<()> { + let channel = connect_channel(url)?; + let token = auth_metadata(jwt)?; + + let mut client = TradingServiceClient::with_interceptor( + channel, + move |mut req: tonic::Request<()>| { + req.metadata_mut() + .insert("authorization", token.clone()); + Ok(req) + }, + ); + + let request = trd::SubscribeSystemStatusRequest { + service_names: Vec::new(), + }; + + let response = client + .subscribe_system_status(request) + .await + .context("SubscribeSystemStatus RPC failed")?; + + let mut stream = response.into_inner(); + + while let Some(frame) = stream.next().await { + let heartbeat = frame.context("system status stream message error")?; + tx.send(StreamEvent::SystemStatus { + service: heartbeat.service_name, + status: format!("{}", heartbeat.status), + message: heartbeat.message, + }) + .await + .context("event channel closed")?; + } + + Ok(()) +} + +// --------------------------------------------------------------------------- +// Initial state (one-shot unary RPCs) +// --------------------------------------------------------------------------- + +/// Fetches positions and risk metrics via unary RPCs to seed the dashboard +/// with initial state before any streaming updates arrive. +async fn fetch_initial_state( + url: &str, + jwt: &str, + tx: &mpsc::Sender, +) -> Result<()> { + let channel = connect_channel(url)?; + let token = auth_metadata(jwt)?; + + let mut client = TradingServiceClient::with_interceptor( + channel, + move |mut req: tonic::Request<()>| { + req.metadata_mut() + .insert("authorization", token.clone()); + Ok(req) + }, + ); + + // Fetch positions. + let positions_resp = client + .get_positions(trd::GetPositionsRequest { symbol: None }) + .await + .context("GetPositions RPC failed")?; + + let positions: Vec = positions_resp + .into_inner() + .positions + .iter() + .map(convert_position) + .collect(); + + tx.send(StreamEvent::PositionsSnapshot { positions }) + .await + .context("event channel closed")?; + + // Fetch risk metrics. + let risk_resp = client + .get_risk_metrics(trd::GetRiskMetricsRequest { + portfolio_id: None, + start_time_unix_nanos: None, + end_time_unix_nanos: None, + }) + .await + .context("GetRiskMetrics RPC failed")?; + + let rm = risk_resp.into_inner(); + tx.send(StreamEvent::RiskMetrics { + var: rm.value_at_risk, + max_drawdown: rm.max_drawdown, + current_drawdown: rm.current_drawdown, + sharpe: rm.sharpe_ratio, + }) + .await + .context("event channel closed")?; + + debug!("initial state fetched successfully"); + Ok(()) +} + +// --------------------------------------------------------------------------- +// Tests +// --------------------------------------------------------------------------- + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + + #[test] + fn test_backoff_doubles() { + let b = next_backoff(std::time::Duration::from_secs(1)); + assert_eq!(b, std::time::Duration::from_secs(2)); + } + + #[test] + fn test_backoff_caps_at_30s() { + let b = next_backoff(std::time::Duration::from_secs(20)); + assert_eq!(b, std::time::Duration::from_secs(30)); + let b2 = next_backoff(std::time::Duration::from_secs(30)); + assert_eq!(b2, std::time::Duration::from_secs(30)); + } + + #[test] + fn test_auth_metadata_valid() { + let meta = auth_metadata("eyJhbGciOiJIUzI1NiJ9.test.sig"); + assert!(meta.is_ok()); + } + + #[tokio::test] + async fn test_connect_channel_valid_url() { + let ch = connect_channel("http://localhost:50051"); + assert!(ch.is_ok()); + } + + #[test] + fn test_connect_channel_invalid_url() { + // Invalid URLs fail at the `from_shared` step, before any runtime is + // needed, so a sync test is fine. + let ch = connect_channel("not a url at all"); + assert!(ch.is_err()); + } + + #[test] + fn test_convert_gpu() { + let gpu = convert_gpu(&mon::GpuSnapshot { + utilization_percent: 85.0, + memory_used_mb: 4096.0, + memory_total_mb: 8192.0, + temperature_celsius: 72.0, + power_watts: 200.0, + }); + assert!((gpu.utilization_percent - 85.0).abs() < f32::EPSILON); + assert!((gpu.memory_used_mb - 4096.0).abs() < f32::EPSILON); + assert!((gpu.memory_total_mb - 8192.0).abs() < f32::EPSILON); + assert!((gpu.temperature_celsius - 72.0).abs() < f32::EPSILON); + } + + #[test] + fn test_convert_position_long() { + let pos = convert_position(&trd::Position { + symbol: "ES".to_owned(), + quantity: 2.0, + market_price: 5100.0, + market_value: 10200.0, + average_cost: 5000.0, + unrealized_pnl: 200.0, + realized_pnl: 0.0, + }); + assert_eq!(pos.side, "Long"); + assert!((pos.quantity - 2.0).abs() < f64::EPSILON); + assert!((pos.entry_price - 5000.0).abs() < f64::EPSILON); + assert_eq!(pos.status, "Open"); + } + + #[test] + fn test_convert_position_short() { + let pos = convert_position(&trd::Position { + symbol: "NQ".to_owned(), + quantity: -1.0, + market_price: 18000.0, + market_value: -18000.0, + average_cost: 18100.0, + unrealized_pnl: 100.0, + realized_pnl: 0.0, + }); + assert_eq!(pos.side, "Short"); + assert!((pos.quantity - 1.0).abs() < f64::EPSILON); + } + + #[test] + fn test_convert_position_zero_quantity() { + let pos = convert_position(&trd::Position { + symbol: "ZN".to_owned(), + quantity: 0.0, + market_price: 110.0, + market_value: 0.0, + average_cost: 109.0, + unrealized_pnl: 0.0, + realized_pnl: 50.0, + }); + // Zero quantity is technically "Long" (>= 0). + assert_eq!(pos.side, "Long"); + assert!((pos.quantity - 0.0).abs() < f64::EPSILON); + } + + #[test] + fn test_convert_training_session() { + let sess = convert_training_session(&mon::TrainingSession { + model: "DQN".to_owned(), + fold: "fold_0".to_owned(), + is_hyperopt: false, + current_epoch: 5.0, + epoch_loss: 0.123, + validation_loss: 0.456, + learning_rate: 0.001, + batches_per_second: 42.0, + gradient_norm: 1.5, + epoch_duration_seconds: 12.0, + ..Default::default() + }); + assert_eq!(sess.model, "DQN"); + assert_eq!(sess.fold, "fold_0"); + assert!(!sess.is_hyperopt); + assert!((sess.epoch - 5.0).abs() < f32::EPSILON); + assert!((sess.epoch_loss - 0.123).abs() < f32::EPSILON); + assert!((sess.gradient_norm - 1.5).abs() < f32::EPSILON); + } + + #[test] + fn test_stream_event_variants_are_clone() { + let event = StreamEvent::SystemStatus { + service: "gateway".into(), + status: "UP".into(), + message: "ok".into(), + }; + let _cloned = event.clone(); + } + + #[test] + fn test_stream_event_all_variants() { + // Verify all variants can be constructed and cloned. + let events = vec![ + StreamEvent::TrainingUpdate { + sessions: vec![], + gpu: GpuInfo::default(), + }, + StreamEvent::OrderUpdate { + order_id: "o1".into(), + symbol: "ES".into(), + status: "3".into(), + filled_qty: 1.0, + last_fill_price: 5000.0, + timestamp_nanos: 123_456_789, + }, + StreamEvent::PositionsSnapshot { positions: vec![] }, + StreamEvent::RiskMetrics { + var: 0.05, + max_drawdown: 0.10, + current_drawdown: 0.02, + sharpe: 1.5, + }, + StreamEvent::RiskAlert { + severity: "3".into(), + symbol: "NQ".into(), + message: "VaR breach".into(), + threshold: 0.05, + current: 0.07, + }, + StreamEvent::SystemStatus { + service: "trading".into(), + status: "1".into(), + message: "healthy".into(), + }, + StreamEvent::StreamDisconnected { + stream_name: "training_metrics".into(), + }, + ]; + for e in &events { + let _ = e.clone(); + } + assert_eq!(events.len(), 7); + } +} diff --git a/bin/fxt/src/main.rs b/bin/fxt/src/main.rs index 5d28dcb16..5b863f852 100644 --- a/bin/fxt/src/main.rs +++ b/bin/fxt/src/main.rs @@ -230,6 +230,9 @@ enum Commands { #[command(flatten)] broker_args: BrokerArgs, }, + + /// Live streaming dashboard + Watch, } /// JWT token claims structure for validation @@ -457,6 +460,10 @@ async fn main() -> Result<()> { } Ok(()) } + Commands::Watch => { + let jwt_token = load_jwt_token(&cli.api_gateway_url).await?; + fxt::commands::watch::run(&cli.api_gateway_url, &jwt_token).await + } } } #[cfg(test)]