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

feat(fxt): implement fxt watch streaming dashboard

Browse Source 
- 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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-03 01:39:06 +01:00
parent d4d070b70b
commit 0a15058990
4 changed files with 1915 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

610
bin/fxt/src/commands/watch/event_loop.rs
View File

@@ -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<CrosstermBackend<io::Stdout>>,
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);
}
}

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

@@ -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<Line<'_>> = 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<Row<'_>> = 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<u64> = 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<Row<'_>> = 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<Row<'_>> = 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<Row<'_>> = 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<Row<'_>> = 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]);
}

766
bin/fxt/src/commands/watch/streams.rs
View File

@@ -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<TrainingSession>,
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<PositionRow> },
/// 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<Channel> {
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<MetadataValue<Ascii>> {
let bearer = format!("Bearer {jwt}");
bearer
.parse::<MetadataValue<Ascii>>()
.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<StreamEvent> {
let (tx, rx) = mpsc::channel::<StreamEvent>(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<StreamEvent>) -> ! {
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<StreamEvent>,
) -> 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<TrainingSession> =
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<StreamEvent>) -> ! {
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<StreamEvent>,
) -> 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<StreamEvent>) -> ! {
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<StreamEvent>,
) -> 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<StreamEvent>) -> ! {
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<StreamEvent>,
) -> 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<StreamEvent>,
) -> 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<PositionRow> = 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);
}
}

7
bin/fxt/src/main.rs
View File

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