diff --git a/bin/fxt/src/commands/agent.rs b/bin/fxt/src/commands/agent.rs index 0696648e7..14e9228d2 100644 --- a/bin/fxt/src/commands/agent.rs +++ b/bin/fxt/src/commands/agent.rs @@ -2,9 +2,12 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::trading_agent; #[derive(Parser, Debug)] pub struct AgentCommand { @@ -14,23 +17,441 @@ pub struct AgentCommand { #[derive(Subcommand, Debug)] enum AgentAction { - /// Start the trading agent - Start, - /// Stop the trading agent - Stop, + /// Start the trading agent (enable all strategies) + Start { + /// Strategy ID to enable (omit to enable all) + #[arg(long)] + strategy: Option, + }, + /// Stop the trading agent (disable all strategies) + Stop { + /// Strategy ID to disable (omit to disable all) + #[arg(long)] + strategy: Option, + }, /// Show agent status (strategy, positions, signals) Status, - /// Show/update agent configuration - Config { - /// Configuration key to get/set - key: Option, - /// Value to set (omit to read current value) - value: Option, - }, + /// Show strategy configuration + Config, + /// Show agent performance metrics + Performance, } -impl AgentCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("agent command not yet implemented") +// ── Result types ──────────────────────────────────────────────────── + +#[derive(Serialize)] +struct AgentStatusResult { + state: String, + active_strategies: u32, + selected_assets: u32, + portfolio_utilization: f64, + current_universe_id: String, + performance: Option, +} + +#[derive(Serialize)] +struct PerformanceInfo { + total_pnl: f64, + sharpe_ratio: f64, + max_drawdown: f64, + win_rate: f64, + total_trades: u32, + avg_trade_pnl: f64, +} + +impl HumanReadable for AgentStatusResult { + fn print_human(&self) { + let state_colored = match self.state.as_str() { + "ACTIVE" => self.state.green().bold(), + "INITIALIZING" | "PAUSED" => self.state.yellow().bold(), + _ => self.state.red().bold(), + }; + println!("{}", "Trading Agent".bold().underline()); + println!(" State: {state_colored}"); + println!(" Active Strategies: {}", self.active_strategies); + println!(" Selected Assets: {}", self.selected_assets); + println!( + " Portfolio Util: {:.1}%", + self.portfolio_utilization * 100.0 + ); + if !self.current_universe_id.is_empty() { + println!(" Universe: {}", self.current_universe_id.dimmed()); + } + + if let Some(ref perf) = self.performance { + println!(); + println!("{}", "Performance".bold().underline()); + let pnl_str = if perf.total_pnl >= 0.0 { + format!("${:.2}", perf.total_pnl).green().to_string() + } else { + format!("${:.2}", perf.total_pnl).red().to_string() + }; + println!(" Total P&L: {pnl_str}"); + println!(" Sharpe Ratio: {:.3}", perf.sharpe_ratio); + println!(" Max Drawdown: {:.2}%", perf.max_drawdown * 100.0); + println!(" Win Rate: {:.1}%", perf.win_rate * 100.0); + println!(" Total Trades: {}", perf.total_trades); + println!(" Avg Trade P&L: ${:.2}", perf.avg_trade_pnl); + } + } +} + +#[derive(Serialize)] +struct StrategyUpdateResult { + success: bool, + message: String, + strategy_id: String, + new_status: String, +} + +impl HumanReadable for StrategyUpdateResult { + fn print_human(&self) { + if self.success { + println!( + "{} Strategy {} -> {}", + "OK".green().bold(), + self.strategy_id.cyan(), + self.new_status.bold() + ); + } else { + println!( + "{} {}: {}", + "FAIL".red().bold(), + self.strategy_id, + self.message + ); + } + } +} + +#[derive(Serialize)] +struct StrategyInfo { + strategy_id: String, + name: String, + strategy_type: String, + status: String, + target_symbols: Vec, + max_capital_pct: f64, + total_pnl: f64, + sharpe_ratio: f64, + win_rate: f64, + total_trades: u32, +} + +#[derive(Serialize)] +struct StrategiesResult { + strategies: Vec, +} + +impl HumanReadable for StrategiesResult { + fn print_human(&self) { + if self.strategies.is_empty() { + println!("{}", "No strategies registered.".dimmed()); + return; + } + println!("{}", "Strategies".bold().underline()); + for s in &self.strategies { + let status_colored = match s.status.as_str() { + "ENABLED" => s.status.green(), + "PAUSED" => s.status.yellow(), + _ => s.status.red(), + }; + println!( + " {} [{}] {:<20} {}", + s.strategy_id.dimmed(), + status_colored, + s.name.bold(), + s.strategy_type.dimmed() + ); + if !s.target_symbols.is_empty() { + println!(" Symbols: {}", s.target_symbols.join(", ")); + } + println!( + " Cap: {:.0}% P&L: ${:.2} Sharpe: {:.3} WR: {:.1}% Trades: {}", + s.max_capital_pct * 100.0, + s.total_pnl, + s.sharpe_ratio, + s.win_rate * 100.0, + s.total_trades, + ); + } + } +} + +#[derive(Serialize)] +struct AgentPerformanceResult { + total_pnl: f64, + sharpe_ratio: f64, + max_drawdown: f64, + win_rate: f64, + total_trades: u32, + avg_trade_pnl: f64, + portfolio_turnover: f64, + strategy_breakdown: Vec, +} + +#[derive(Serialize)] +struct StrategyPerfEntry { + strategy_id: String, + total_pnl: f64, + sharpe_ratio: f64, + win_rate: f64, + total_trades: u32, +} + +impl HumanReadable for AgentPerformanceResult { + fn print_human(&self) { + println!("{}", "Agent Performance".bold().underline()); + let pnl_str = if self.total_pnl >= 0.0 { + format!("${:.2}", self.total_pnl).green().to_string() + } else { + format!("${:.2}", self.total_pnl).red().to_string() + }; + println!(" Total P&L: {pnl_str}"); + println!(" Sharpe Ratio: {:.3}", self.sharpe_ratio); + println!(" Max Drawdown: {:.2}%", self.max_drawdown * 100.0); + println!(" Win Rate: {:.1}%", self.win_rate * 100.0); + println!(" Total Trades: {}", self.total_trades); + println!(" Avg Trade P&L: ${:.2}", self.avg_trade_pnl); + println!( + " Turnover (ann.): {:.1}%", + self.portfolio_turnover * 100.0 + ); + + if !self.strategy_breakdown.is_empty() { + println!(); + println!(" {}", "Per-Strategy Breakdown".bold()); + println!( + " {:<20} {:>12} {:>10} {:>10} {:>8}", + "Strategy", "P&L", "Sharpe", "Win Rate", "Trades" + ); + println!(" {}", "-".repeat(64)); + for s in &self.strategy_breakdown { + println!( + " {:<20} {:>12.2} {:>10.3} {:>9.1}% {:>8}", + s.strategy_id, + s.total_pnl, + s.sharpe_ratio, + s.win_rate * 100.0, + s.total_trades, + ); + } + } + } +} + +// ── Helpers ───────────────────────────────────────────────────────── + +fn agent_state_name(v: i32) -> &'static str { + match trading_agent::AgentState::try_from(v) { + Ok(trading_agent::AgentState::Initializing) => "INITIALIZING", + Ok(trading_agent::AgentState::Active) => "ACTIVE", + Ok(trading_agent::AgentState::Paused) => "PAUSED", + Ok(trading_agent::AgentState::Error) => "ERROR", + Ok(trading_agent::AgentState::Shutdown) => "SHUTDOWN", + _ => "UNKNOWN", + } +} + +fn strategy_status_name(v: i32) -> &'static str { + match trading_agent::StrategyStatus::try_from(v) { + Ok(trading_agent::StrategyStatus::Enabled) => "ENABLED", + Ok(trading_agent::StrategyStatus::Disabled) => "DISABLED", + Ok(trading_agent::StrategyStatus::Paused) => "PAUSED", + Ok(trading_agent::StrategyStatus::Error) => "ERROR", + _ => "UNKNOWN", + } +} + +fn strategy_type_name(v: i32) -> &'static str { + match trading_agent::StrategyType::try_from(v) { + Ok(trading_agent::StrategyType::MlEnsemble) => "ML_ENSEMBLE", + Ok(trading_agent::StrategyType::MeanReversion) => "MEAN_REVERSION", + Ok(trading_agent::StrategyType::Momentum) => "MOMENTUM", + Ok(trading_agent::StrategyType::Arbitrage) => "ARBITRAGE", + Ok(trading_agent::StrategyType::MarketMaking) => "MARKET_MAKING", + _ => "UNKNOWN", + } +} + +// ── Execute ───────────────────────────────────────────────────────── + +impl AgentCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + AgentAction::Start { strategy } => { + self.update_strategy(client, format, strategy, true) + .await?; + } + + AgentAction::Stop { strategy } => { + self.update_strategy(client, format, strategy, false) + .await?; + } + + AgentAction::Status => { + let resp = client + .trading_agent() + .get_agent_status(trading_agent::GetAgentStatusRequest { + include_performance: true, + include_positions: false, + }) + .await? + .into_inner(); + + let status = resp.status.unwrap_or_default(); + let performance = resp.performance.map(|p| PerformanceInfo { + total_pnl: p.total_pnl, + sharpe_ratio: p.sharpe_ratio, + max_drawdown: p.max_drawdown, + win_rate: p.win_rate, + total_trades: p.total_trades, + avg_trade_pnl: p.avg_trade_pnl, + }); + + let result = AgentStatusResult { + state: agent_state_name(status.state).to_owned(), + active_strategies: status.active_strategies, + selected_assets: status.selected_assets, + portfolio_utilization: status.portfolio_utilization, + current_universe_id: status.current_universe_id, + performance, + }; + output::render(&result, format)?; + } + + AgentAction::Config => { + let resp = client + .trading_agent() + .list_strategies(trading_agent::ListStrategiesRequest { + status_filter: None, + }) + .await? + .into_inner(); + + let strategies = resp + .strategies + .into_iter() + .map(|s| { + let config = s.config.unwrap_or_default(); + let perf = s.performance.unwrap_or_default(); + StrategyInfo { + strategy_id: s.strategy_id, + name: s.strategy_name, + strategy_type: strategy_type_name(s.strategy_type).to_owned(), + status: strategy_status_name(s.status).to_owned(), + target_symbols: config.target_symbols, + max_capital_pct: config.max_capital_pct, + total_pnl: perf.total_pnl, + sharpe_ratio: perf.sharpe_ratio, + win_rate: perf.win_rate, + total_trades: perf.total_trades, + } + }) + .collect(); + + output::render(&StrategiesResult { strategies }, format)?; + } + + AgentAction::Performance => { + let resp = client + .trading_agent() + .get_agent_performance(trading_agent::GetAgentPerformanceRequest { + start_time: None, + end_time: None, + include_strategy_breakdown: true, + }) + .await? + .into_inner(); + + let metrics = resp.metrics.unwrap_or_default(); + let strategy_breakdown = resp + .strategy_performance + .into_iter() + .map(|s| StrategyPerfEntry { + strategy_id: s.strategy_id, + total_pnl: s.total_pnl, + sharpe_ratio: s.sharpe_ratio, + win_rate: s.win_rate, + total_trades: s.total_trades, + }) + .collect(); + + let result = AgentPerformanceResult { + total_pnl: metrics.total_pnl, + sharpe_ratio: metrics.sharpe_ratio, + max_drawdown: metrics.max_drawdown, + win_rate: metrics.win_rate, + total_trades: metrics.total_trades, + avg_trade_pnl: metrics.avg_trade_pnl, + portfolio_turnover: metrics.portfolio_turnover, + strategy_breakdown, + }; + output::render(&result, format)?; + } + } + + Ok(()) + } + + /// Enable or disable a strategy (or all strategies if no ID given). + async fn update_strategy( + &self, + client: &FoxhuntClient, + format: OutputFormat, + strategy_id: &Option, + enable: bool, + ) -> Result<()> { + let new_status = if enable { + trading_agent::StrategyStatus::Enabled + } else { + trading_agent::StrategyStatus::Disabled + }; + + // If a specific strategy ID is provided, update that one. + // Otherwise, list all strategies and update each. + let ids: Vec = if let Some(id) = strategy_id { + vec![id.clone()] + } else { + let resp = client + .trading_agent() + .list_strategies(trading_agent::ListStrategiesRequest { + status_filter: None, + }) + .await? + .into_inner(); + resp.strategies + .into_iter() + .map(|s| s.strategy_id) + .collect() + }; + + if ids.is_empty() { + println!("{}", "No strategies registered.".dimmed()); + return Ok(()); + } + + for id in &ids { + let resp = client + .trading_agent() + .update_strategy_status(trading_agent::UpdateStrategyStatusRequest { + strategy_id: id.clone(), + new_status: new_status.into(), + reason: None, + }) + .await? + .into_inner(); + + let updated = resp.updated_strategy.unwrap_or_default(); + let result = StrategyUpdateResult { + success: resp.success, + message: resp.message, + strategy_id: id.clone(), + new_status: strategy_status_name(updated.status).to_owned(), + }; + output::render(&result, format)?; + } + + Ok(()) } } diff --git a/bin/fxt/src/commands/auth.rs b/bin/fxt/src/commands/auth.rs index 80b4a39d3..fc7bbb454 100644 --- a/bin/fxt/src/commands/auth.rs +++ b/bin/fxt/src/commands/auth.rs @@ -2,9 +2,13 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; +use crate::auth::token_manager::{AuthTokenManager, FileTokenStorage}; +use crate::auth::LoginClient; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; #[derive(Parser, Debug)] pub struct AuthCommand { @@ -26,8 +30,145 @@ enum AuthAction { Status, } -impl AuthCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("auth command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct LoginResult { + success: bool, + message: String, +} + +impl HumanReadable for LoginResult { + fn print_human(&self) { + if self.success { + println!("{} {}", "Authenticated.".green().bold(), self.message); + } else { + println!("{} {}", "Login failed.".red().bold(), self.message); + } + } +} + +#[derive(Serialize)] +struct LogoutResult { + success: bool, + message: String, +} + +impl HumanReadable for LogoutResult { + fn print_human(&self) { + if self.success { + println!("{} {}", "Logged out.".green().bold(), self.message); + } else { + println!("{} {}", "Logout failed.".red().bold(), self.message); + } + } +} + +#[derive(Serialize)] +struct AuthStatusResult { + authenticated: bool, + has_refresh_token: bool, + expires_in: Option, +} + +impl HumanReadable for AuthStatusResult { + fn print_human(&self) { + println!("{}", "Authentication Status".cyan().bold()); + if self.authenticated { + println!(" Status: {}", "authenticated".green()); + if let Some(ref exp) = self.expires_in { + println!(" Token expires in: {exp}"); + } + } else { + println!(" Status: {}", "not authenticated".red()); + if self.has_refresh_token { + println!( + " {}", + "Refresh token found -- run `fxt auth login` to re-authenticate.".yellow() + ); + } else { + println!(" Run `fxt auth login` to authenticate."); + } + } + } +} + +// ── Execute ─────────────────────────────────────────────────────────── + +impl AuthCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + AuthAction::Login { username } => { + let storage = FileTokenStorage::new()?; + let manager = AuthTokenManager::new(storage); + let login_client = LoginClient::new(client.channel()); + + if let Some(user) = username { + // Non-interactive: read password from stdin + print!("Password: "); + std::io::Write::flush(&mut std::io::stdout())?; + let password = + rpassword::read_password().map_err(|e| anyhow::anyhow!("{e}"))?; + + login_client + .login_with_credentials(user, &password, &manager) + .await?; + + let result = LoginResult { + success: true, + message: format!("Logged in as {user}"), + }; + output::render(&result, format)?; + } else { + // Interactive login flow + login_client.interactive_login(&manager).await?; + + let result = LoginResult { + success: true, + message: "Interactive login complete.".to_owned(), + }; + output::render(&result, format)?; + } + } + + AuthAction::Logout => { + let storage = FileTokenStorage::new()?; + let manager = AuthTokenManager::new(storage); + manager.clear_tokens().await?; + + let result = LogoutResult { + success: true, + message: "All tokens cleared.".to_owned(), + }; + output::render(&result, format)?; + } + + AuthAction::Status => { + let storage = FileTokenStorage::new()?; + let manager = AuthTokenManager::new(storage); + + let has_token = manager.has_valid_token().await; + let has_refresh = manager.get_refresh_token().await?.is_some(); + #[allow(clippy::integer_division)] + let expires_in = manager.time_until_expiry().await.map(|d| { + let secs = d.as_secs(); + if secs >= 3600 { + format!("{}h {}m", secs / 3600, (secs % 3600) / 60) + } else if secs >= 60 { + format!("{}m {}s", secs / 60, secs % 60) + } else { + format!("{secs}s") + } + }); + + let result = AuthStatusResult { + authenticated: has_token, + has_refresh_token: has_refresh, + expires_in, + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/backtest.rs b/bin/fxt/src/commands/backtest.rs index 8ec3f8674..0eab648f6 100644 --- a/bin/fxt/src/commands/backtest.rs +++ b/bin/fxt/src/commands/backtest.rs @@ -2,9 +2,11 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; #[derive(Parser, Debug)] pub struct BacktestCommand { @@ -41,8 +43,121 @@ enum BacktestAction { }, } -impl BacktestCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("backtest command not yet implemented") +// ── Result types ──────────────────────────────────────────────────── + +#[derive(Serialize)] +struct BacktestRunResult { + message: String, + strategy: String, + symbol: String, + from: String, + to: String, + note: String, +} + +impl HumanReadable for BacktestRunResult { + fn print_human(&self) { + println!("{}", "Backtest".bold().underline()); + println!(" Strategy: {}", self.strategy.cyan()); + println!(" Symbol: {}", self.symbol); + println!(" Period: {} to {}", self.from, self.to); + println!(); + println!(" {}", self.message.yellow()); + if !self.note.is_empty() { + println!(" {}", self.note.dimmed()); + } + } +} + +#[derive(Serialize)] +struct BacktestStatusResult { + job_id: String, + message: String, + note: String, +} + +impl HumanReadable for BacktestStatusResult { + fn print_human(&self) { + println!("{}", "Backtest Status".bold().underline()); + println!(" Job ID: {}", self.job_id.cyan()); + println!(" {}", self.message.yellow()); + if !self.note.is_empty() { + println!(" {}", self.note.dimmed()); + } + } +} + +#[derive(Serialize)] +struct BacktestResultsResult { + job_id: String, + message: String, + note: String, +} + +impl HumanReadable for BacktestResultsResult { + fn print_human(&self) { + println!("{}", "Backtest Results".bold().underline()); + println!(" Job ID: {}", self.job_id.cyan()); + println!(" {}", self.message.yellow()); + if !self.note.is_empty() { + println!(" {}", self.note.dimmed()); + } + } +} + +// ── Execute ───────────────────────────────────────────────────────── + +impl BacktestCommand { + pub async fn execute(&self, _client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + // Backtesting RPCs are not yet defined in trading.proto. + // These stubs describe what each subcommand will do once the + // backtesting service exposes gRPC endpoints. + + match &self.action { + BacktestAction::Run { + strategy, + symbol, + from, + to, + } => { + let result = BacktestRunResult { + message: "Backtest submission requires the backtesting gRPC service. \ + Use the training pipeline directly for now." + .to_owned(), + strategy: strategy.clone(), + symbol: symbol.clone(), + from: from.clone(), + to: to.clone(), + note: "Pipeline: fxt train run --model --symbol \ + (walk-forward with backtest windows)" + .to_owned(), + }; + output::render(&result, format)?; + } + + BacktestAction::Status { job_id } => { + let result = BacktestStatusResult { + job_id: job_id.clone(), + message: "Backtest status tracking requires the backtesting gRPC service." + .to_owned(), + note: "Monitor training jobs: kubectl get jobs -n foxhunt -l app=training" + .to_owned(), + }; + output::render(&result, format)?; + } + + BacktestAction::Results { job_id } => { + let result = BacktestResultsResult { + job_id: job_id.clone(), + message: "Backtest results require the backtesting gRPC service." + .to_owned(), + note: "View evaluation results: fxt model status " + .to_owned(), + }; + output::render(&result, format)?; + } + } + + Ok(()) } } diff --git a/bin/fxt/src/commands/broker.rs b/bin/fxt/src/commands/broker.rs index 5a03178fc..659650311 100644 --- a/bin/fxt/src/commands/broker.rs +++ b/bin/fxt/src/commands/broker.rs @@ -2,9 +2,13 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; +use tokio_stream::StreamExt; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::broker_gateway; #[derive(Parser, Debug)] pub struct BrokerCommand { @@ -27,14 +31,213 @@ enum BrokerAction { }, /// List recent executions Executions { - /// Maximum number of executions to show + /// Follow execution stream in real-time + #[arg(long, short)] + follow: bool, + /// Maximum number of executions to show (non-follow mode) #[arg(long, default_value = "20")] limit: u32, }, } -impl BrokerCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("broker command not yet implemented") +// ── Result types ──────────────────────────────────────────────────── + +#[derive(Serialize)] +struct SessionStatusResult { + session_id: String, + state: String, + sender_seq_num: i64, + target_seq_num: i64, + heartbeat_rtt_ms: f64, + details: std::collections::HashMap, +} + +impl HumanReadable for SessionStatusResult { + fn print_human(&self) { + let state_colored = match self.state.as_str() { + "ACTIVE" => self.state.green().bold(), + "CONNECTED" | "LOGGING_IN" => self.state.yellow().bold(), + _ => self.state.red().bold(), + }; + println!("{}", "Broker Session".bold().underline()); + println!(" Session ID: {}", self.session_id.cyan()); + println!(" State: {state_colored}"); + println!(" Sender Seq: {}", self.sender_seq_num); + println!(" Target Seq: {}", self.target_seq_num); + if self.heartbeat_rtt_ms > 0.0 { + println!(" Heartbeat: {:.1} ms RTT", self.heartbeat_rtt_ms); + } + if !self.details.is_empty() { + println!(" Details:"); + for (k, v) in &self.details { + println!(" {k}: {v}"); + } + } + } +} + +#[derive(Serialize)] +struct HealthResult { + healthy: bool, + message: String, + details: std::collections::HashMap, +} + +impl HumanReadable for HealthResult { + fn print_human(&self) { + if self.healthy { + println!("{} {}", "HEALTHY".green().bold(), self.message); + } else { + println!("{} {}", "UNHEALTHY".red().bold(), self.message); + } + } +} + +// ── Helpers ───────────────────────────────────────────────────────── + +fn session_state_name(v: i32) -> &'static str { + match broker_gateway::SessionState::try_from(v) { + Ok(broker_gateway::SessionState::Disconnected) => "DISCONNECTED", + Ok(broker_gateway::SessionState::Connected) => "CONNECTED", + Ok(broker_gateway::SessionState::LoggingIn) => "LOGGING_IN", + Ok(broker_gateway::SessionState::Active) => "ACTIVE", + Ok(broker_gateway::SessionState::LoggingOut) => "LOGGING_OUT", + _ => "UNKNOWN", + } +} + +fn exec_side_name(v: i32) -> &'static str { + match broker_gateway::OrderSide::try_from(v) { + Ok(broker_gateway::OrderSide::Buy) => "BUY", + Ok(broker_gateway::OrderSide::Sell) => "SELL", + _ => "?", + } +} + +fn exec_type_name(v: i32) -> &'static str { + match broker_gateway::ExecutionType::try_from(v) { + Ok(broker_gateway::ExecutionType::New) => "NEW", + Ok(broker_gateway::ExecutionType::Trade) => "TRADE", + Ok(broker_gateway::ExecutionType::Canceled) => "CANCELED", + Ok(broker_gateway::ExecutionType::Rejected) => "REJECTED", + _ => "?", + } +} + +// ── Execute ───────────────────────────────────────────────────────── + +impl BrokerCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + BrokerAction::Status => { + let resp = client + .broker_gateway() + .get_session_status(broker_gateway::GetSessionStatusRequest { + session_id: None, + }) + .await? + .into_inner(); + + let result = SessionStatusResult { + session_id: resp.session_id, + state: session_state_name(resp.state).to_owned(), + sender_seq_num: resp.sender_seq_num, + target_seq_num: resp.target_seq_num, + heartbeat_rtt_ms: resp.heartbeat_rtt_ms, + details: resp.details, + }; + output::render(&result, format)?; + } + + BrokerAction::Connect { .. } => { + // First do a health check to verify connectivity. + let resp = client + .broker_gateway() + .health_check(broker_gateway::HealthCheckRequest {}) + .await? + .into_inner(); + + let result = HealthResult { + healthy: resp.healthy, + message: if resp.healthy { + "Broker connection is managed by the broker-gateway service. \ + Gateway is healthy." + .to_owned() + } else { + format!( + "Broker gateway reports unhealthy: {}", + resp.message + ) + }, + details: resp.details, + }; + output::render(&result, format)?; + } + + BrokerAction::Executions { follow, limit } => { + if *follow { + // Stream executions in real-time. + let mut stream = client + .broker_gateway() + .stream_executions(broker_gateway::StreamExecutionsRequest { + account_id: None, + symbol: None, + }) + .await? + .into_inner(); + + println!( + "{} Streaming executions (Ctrl-C to stop)...", + "LIVE".green().bold() + ); + println!( + " {:<14} {:<10} {:<6} {:>10} {:>12} {:<8}", + "ExecID", "Symbol", "Side", "Qty", "Price", "Type" + ); + println!(" {}", "-".repeat(64)); + + while let Some(msg) = stream.next().await { + let exec = msg?; + println!( + " {:<14} {:<10} {:<6} {:>10.2} {:>12.4} {:<8}", + truncate_id(&exec.execution_id, 14), + exec.symbol, + exec_side_name(exec.side), + exec.last_qty, + exec.last_price, + exec_type_name(exec.exec_type), + ); + } + } else { + // Non-streaming: inform user about the limitation. + println!( + "{} The broker gateway streams executions in real-time.", + "INFO".blue().bold() + ); + println!( + " Use {} to stream live executions.", + "fxt broker executions --follow".cyan() + ); + println!( + " Use {} to view execution history.", + "fxt trade orders ".cyan() + ); + let _ = limit; // acknowledged but not used without a history RPC + } + } + } + + Ok(()) + } +} + +/// Truncate a string to at most `max` characters for display. +fn truncate_id(s: &str, max: usize) -> String { + if s.len() <= max { + s.to_owned() + } else { + let visible = max.saturating_sub(2); + let prefix: String = s.chars().take(visible).collect(); + format!("{prefix}..") } } diff --git a/bin/fxt/src/commands/cluster.rs b/bin/fxt/src/commands/cluster.rs index 9cf1ed323..6e148fbdf 100644 --- a/bin/fxt/src/commands/cluster.rs +++ b/bin/fxt/src/commands/cluster.rs @@ -2,9 +2,11 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; #[derive(Parser, Debug)] pub struct ClusterCommand { @@ -26,8 +28,250 @@ enum ClusterAction { }, } -impl ClusterCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("cluster command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct ServiceInfo { + name: String, + health: String, + state: String, + version: String, + uptime_seconds: i64, + error_message: String, +} + +#[derive(Serialize)] +struct ClusterStatusResult { + overall_health: String, + healthy_services: i32, + total_services: i32, + critical_issues: Vec, + system_uptime_seconds: i64, + cpu_usage_pct: f64, + memory_usage_pct: f64, + disk_usage_pct: f64, + error_rate_pct: f64, + services: Vec, +} + +impl HumanReadable for ClusterStatusResult { + fn print_human(&self) { + let health_colored = match self.overall_health.as_str() { + "HEALTHY" => self.overall_health.green().bold().to_string(), + "DEGRADED" => self.overall_health.yellow().bold().to_string(), + "UNHEALTHY" | "CRITICAL" => self.overall_health.red().bold().to_string(), + _ => self.overall_health.clone(), + }; + println!("{} {health_colored}", "Cluster Status:".cyan().bold()); + println!( + " Services: {}/{} healthy", + self.healthy_services, self.total_services, + ); + #[allow(clippy::integer_division)] + let uptime_hours = self.system_uptime_seconds / 3600; + println!(" Uptime: {uptime_hours} hours"); + println!(" CPU: {:.1}%", self.cpu_usage_pct); + println!(" Memory: {:.1}%", self.memory_usage_pct); + println!(" Disk: {:.1}%", self.disk_usage_pct); + println!(" Errors: {:.2}%", self.error_rate_pct); + + if !self.critical_issues.is_empty() { + println!(); + println!("{}", "Critical Issues:".red().bold()); + for issue in &self.critical_issues { + println!(" - {}", issue.red()); + } + } + + if !self.services.is_empty() { + println!(); + println!("{}", "Services:".cyan().bold()); + for svc in &self.services { + let health_str = match svc.health.as_str() { + "HEALTHY" => svc.health.green().to_string(), + "DEGRADED" => svc.health.yellow().to_string(), + "UNHEALTHY" | "CRITICAL" => svc.health.red().to_string(), + _ => svc.health.clone(), + }; + let version_str = if svc.version.is_empty() { + String::new() + } else { + format!(" v{}", svc.version) + }; + println!( + " [{}] {}{} ({})", + health_str, svc.name, version_str, svc.state, + ); + if !svc.error_message.is_empty() { + println!(" {}", svc.error_message.red()); + } + } + } + } +} + +#[derive(Serialize)] +struct MetricInfo { + name: String, + value: f64, + unit: String, +} + +#[derive(Serialize)] +struct ResourcesResult { + metrics: Vec, +} + +impl HumanReadable for ResourcesResult { + fn print_human(&self) { + println!("{}", "Cluster Resources".cyan().bold()); + if self.metrics.is_empty() { + println!(" No resource metrics available."); + return; + } + for m in &self.metrics { + let unit_suffix = if m.unit.is_empty() { + String::new() + } else { + format!(" {}", m.unit) + }; + println!(" {:30} {:>10.2}{}", m.name, m.value, unit_suffix); + } + } +} + +#[derive(Serialize)] +struct EventsResult { + message: String, +} + +impl HumanReadable for EventsResult { + fn print_human(&self) { + println!("{}", "Cluster Events".cyan().bold()); + println!(" {}", self.message); + } +} + +// ── Helpers ─────────────────────────────────────────────────────────── + +fn system_health_name(value: i32) -> &'static str { + match value { + 1 => "HEALTHY", + 2 => "DEGRADED", + 3 => "UNHEALTHY", + 4 => "CRITICAL", + _ => "UNKNOWN", + } +} + +fn service_health_name(value: i32) -> &'static str { + match value { + 1 => "HEALTHY", + 2 => "DEGRADED", + 3 => "UNHEALTHY", + 4 => "CRITICAL", + _ => "UNKNOWN", + } +} + +fn service_state_name(value: i32) -> &'static str { + match value { + 1 => "starting", + 2 => "running", + 3 => "stopping", + 4 => "stopped", + 5 => "error", + _ => "unknown", + } +} + +// ── Execute ─────────────────────────────────────────────────────────── + +impl ClusterCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + ClusterAction::Status => { + let resp = client + .monitoring() + .get_system_status(crate::proto::monitoring::GetSystemStatusRequest { + service_names: vec![], + }) + .await? + .into_inner(); + + let overall = resp.overall_status.unwrap_or_default(); + let sys_metrics = overall.system_metrics.unwrap_or_default(); + + let services: Vec = resp + .service_statuses + .iter() + .map(|s| ServiceInfo { + name: s.service_name.clone(), + health: service_health_name(s.health).to_owned(), + state: service_state_name(s.state).to_owned(), + version: s.version.clone().unwrap_or_default(), + uptime_seconds: s.uptime_seconds, + error_message: s.error_message.clone().unwrap_or_default(), + }) + .collect(); + + let result = ClusterStatusResult { + overall_health: system_health_name(overall.overall_health).to_owned(), + healthy_services: overall.healthy_services, + total_services: overall.total_services, + critical_issues: overall.critical_issues, + system_uptime_seconds: overall.system_uptime_seconds, + cpu_usage_pct: sys_metrics.cpu_usage_percent, + memory_usage_pct: sys_metrics.memory_usage_percent, + disk_usage_pct: sys_metrics.disk_usage_percent, + error_rate_pct: sys_metrics.error_rate_percent, + services, + }; + output::render(&result, format)?; + } + + ClusterAction::Resources => { + let resource_names = vec![ + "cpu.usage".to_owned(), + "memory.usage".to_owned(), + "disk.usage".to_owned(), + "gpu.utilization".to_owned(), + "gpu.memory.used".to_owned(), + "network.io".to_owned(), + ]; + + let resp = client + .monitoring() + .get_metrics(crate::proto::monitoring::GetMetricsRequest { + metric_names: resource_names, + start_time: None, + end_time: None, + aggregation: None, + }) + .await? + .into_inner(); + + let metrics: Vec = resp + .metrics + .iter() + .map(|m| MetricInfo { + name: m.name.clone(), + value: m.value, + unit: m.unit.clone(), + }) + .collect(); + + let result = ResourcesResult { metrics }; + output::render(&result, format)?; + } + + ClusterAction::Events { limit: _ } => { + let result = EventsResult { + message: "Cluster events require kubectl access. Use `kubectl get events --sort-by=.metadata.creationTimestamp` for direct access.".to_owned(), + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/config_cmd.rs b/bin/fxt/src/commands/config_cmd.rs index 12b704a2b..61d78e472 100644 --- a/bin/fxt/src/commands/config_cmd.rs +++ b/bin/fxt/src/commands/config_cmd.rs @@ -2,9 +2,11 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; #[derive(Parser, Debug)] pub struct ConfigCommand { @@ -18,6 +20,9 @@ enum ConfigAction { Get { /// Configuration key key: String, + /// Configuration category (optional filter) + #[arg(long)] + category: Option, }, /// Set a configuration value Set { @@ -25,15 +30,237 @@ enum ConfigAction { key: String, /// Configuration value value: String, + /// Configuration category + #[arg(long)] + category: Option, }, /// Export full configuration as TOML - Export, + Export { + /// Category to export (omit for all) + #[arg(long)] + category: Option, + }, /// Show resolved environment (merged CLI + env + config file) Env, } -impl ConfigCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("config command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct ConfigSettingInfo { + category: String, + key: String, + value: String, + data_type: String, + hot_reload: bool, + description: String, +} + +#[derive(Serialize)] +struct GetConfigResult { + settings: Vec, +} + +impl HumanReadable for GetConfigResult { + fn print_human(&self) { + if self.settings.is_empty() { + println!("{}", "No matching configuration settings found.".yellow()); + return; + } + for s in &self.settings { + println!( + "{}.{} = {}", + s.category.dimmed(), + s.key.bold(), + s.value.green(), + ); + if !s.description.is_empty() { + println!(" # {}", s.description.dimmed()); + } + if s.hot_reload { + println!(" {} hot-reload enabled", "~".cyan()); + } + } + } +} + +#[derive(Serialize)] +struct SetConfigResult { + success: bool, + message: String, +} + +impl HumanReadable for SetConfigResult { + fn print_human(&self) { + if self.success { + println!("{} {}", "OK".green().bold(), self.message); + } else { + println!("{} {}", "FAILED".red().bold(), self.message); + } + } +} + +#[derive(Serialize)] +struct ExportConfigResult { + data: String, + format: String, + settings_count: i32, +} + +impl HumanReadable for ExportConfigResult { + fn print_human(&self) { + println!( + "{} ({} settings, format={})", + "Configuration Export".cyan().bold(), + self.settings_count, + self.format, + ); + println!("{}", self.data); + } +} + +#[derive(Serialize)] +struct EnvResult { + api_url: String, + config_file: String, + log_level: String, + token_storage: String, +} + +impl HumanReadable for EnvResult { + fn print_human(&self) { + println!("{}", "Resolved Environment".cyan().bold()); + println!(" API URL: {}", self.api_url); + println!(" Config File: {}", self.config_file); + println!(" Log Level: {}", self.log_level); + println!(" Token Storage: {}", self.token_storage); + } +} + +// ── Helpers ─────────────────────────────────────────────────────────── + +fn config_data_type_name(value: i32) -> &'static str { + match value { + 1 => "string", + 2 => "number", + 3 => "boolean", + 4 => "json", + 5 => "encrypted", + _ => "unspecified", + } +} + +// ── Execute ─────────────────────────────────────────────────────────── + +impl ConfigCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + ConfigAction::Get { key, category } => { + let resp = client + .config() + .get_configuration(crate::proto::config::GetConfigurationRequest { + category: category.clone(), + key: Some(key.clone()), + environment: None, + }) + .await? + .into_inner(); + + let settings: Vec = resp + .settings + .iter() + .map(|s| ConfigSettingInfo { + category: s.category.clone(), + key: s.key.clone(), + value: if s.sensitive { + "********".to_owned() + } else { + s.value.clone() + }, + data_type: config_data_type_name(s.data_type).to_owned(), + hot_reload: s.hot_reload, + description: s.description.clone(), + }) + .collect(); + + let result = GetConfigResult { settings }; + output::render(&result, format)?; + } + + ConfigAction::Set { + key, + value, + category, + } => { + let cat = category.clone().unwrap_or_else(|| "default".to_owned()); + let resp = client + .config() + .update_configuration(crate::proto::config::UpdateConfigurationRequest { + category: cat, + key: key.clone(), + value: value.clone(), + changed_by: "fxt-cli".to_owned(), + change_reason: None, + environment: None, + }) + .await? + .into_inner(); + + let result = SetConfigResult { + success: resp.success, + message: resp.message, + }; + output::render(&result, format)?; + } + + ConfigAction::Export { category } => { + let categories = match category { + Some(c) => vec![c.clone()], + None => vec![], + }; + + let resp = client + .config() + .export_configuration(crate::proto::config::ExportConfigurationRequest { + categories, + environment: None, + format: 3, // TOML + }) + .await? + .into_inner(); + + let format_name = match resp.format { + 1 => "json", + 2 => "yaml", + 3 => "toml", + 4 => "env", + _ => "unknown", + }; + + let result = ExportConfigResult { + data: resp.exported_data, + format: format_name.to_owned(), + settings_count: resp.settings_count, + }; + output::render(&result, format)?; + } + + ConfigAction::Env => { + let config = crate::config::TliConfig::load().unwrap_or_default(); + let config_path = dirs::home_dir() + .map(|h| h.join(".foxhunt").join("config.toml")) + .map(|p| p.display().to_string()) + .unwrap_or_else(|| "~/.foxhunt/config.toml".to_owned()); + + let result = EnvResult { + api_url: config.api_gateway_url, + config_file: config_path, + log_level: config.log_level, + token_storage: config.token_storage, + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/data.rs b/bin/fxt/src/commands/data.rs index e9f513a15..eb867172e 100644 --- a/bin/fxt/src/commands/data.rs +++ b/bin/fxt/src/commands/data.rs @@ -2,9 +2,11 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; #[derive(Parser, Debug)] pub struct DataCommand { @@ -37,8 +39,220 @@ enum DataAction { Feeds, } -impl DataCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("data command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct DownloadResult { + job_id: String, + status: String, + estimated_cost_usd: f64, + message: String, +} + +impl HumanReadable for DownloadResult { + fn print_human(&self) { + println!("{}", "Download Scheduled".green().bold()); + println!(" Job ID: {}", self.job_id); + println!(" Status: {}", self.status); + println!( + " Est. Cost: ${:.2}", + self.estimated_cost_usd + ); + println!(" Message: {}", self.message); + } +} + +#[derive(Serialize)] +struct DownloadJobInfo { + job_id: String, + status: String, + dataset: String, + symbols: Vec, + start_date: String, + end_date: String, + progress_pct: f32, + bytes_downloaded: u64, + total_bytes: u64, + records_count: u64, + data_quality_score: f64, + error_message: String, +} + +#[derive(Serialize)] +struct StatusResult { + jobs: Vec, + total_count: u32, +} + +impl HumanReadable for StatusResult { + fn print_human(&self) { + println!( + "{} ({} total)", + "Download Jobs".cyan().bold(), + self.total_count, + ); + if self.jobs.is_empty() { + println!(" No download jobs found."); + return; + } + for job in &self.jobs { + let status_colored = match job.status.as_str() { + "COMPLETED" => job.status.green().to_string(), + "DOWNLOADING" | "VALIDATING" | "UPLOADING" => job.status.yellow().to_string(), + "FAILED" => job.status.red().to_string(), + "CANCELLED" => job.status.dimmed().to_string(), + _ => job.status.clone(), + }; + println!( + " {} [{}] {:.0}% {} {} -> {}", + job.job_id.dimmed(), + status_colored, + job.progress_pct, + job.symbols.join(","), + job.start_date, + job.end_date, + ); + if !job.error_message.is_empty() { + println!(" Error: {}", job.error_message.red()); + } + } + } +} + +#[derive(Serialize)] +struct CacheResult { + message: String, +} + +impl HumanReadable for CacheResult { + fn print_human(&self) { + println!("{}", "Data Cache".cyan().bold()); + println!(" {}", self.message); + } +} + +#[derive(Serialize)] +struct FeedsResult { + message: String, +} + +impl HumanReadable for FeedsResult { + fn print_human(&self) { + println!("{}", "Live Data Feeds".cyan().bold()); + println!(" {}", self.message); + } +} + +// ── Helpers ─────────────────────────────────────────────────────────── + +fn download_status_name(value: i32) -> &'static str { + match value { + 0 => "UNKNOWN", + 1 => "PENDING", + 2 => "DOWNLOADING", + 3 => "VALIDATING", + 4 => "UPLOADING", + 5 => "COMPLETED", + 6 => "FAILED", + 7 => "CANCELLED", + _ => "UNKNOWN", + } +} + +// ── Execute ─────────────────────────────────────────────────────────── + +impl DataCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + DataAction::Download { + symbol, + from, + to, + schema, + } => { + let response = client + .data_acquisition() + .schedule_download(crate::proto::data_acquisition::ScheduleDownloadRequest { + dataset: "GLBX.MDP3".to_owned(), + symbols: vec![symbol.clone()], + start_date: from.clone(), + end_date: to.clone(), + schema: schema.clone(), + description: String::new(), + tags: Default::default(), + priority: 3, + }) + .await? + .into_inner(); + + let result = DownloadResult { + job_id: response.job_id, + status: download_status_name(response.status).to_owned(), + estimated_cost_usd: response.estimated_cost_usd, + message: response.message, + }; + output::render(&result, format)?; + } + + DataAction::Status => { + let response = client + .data_acquisition() + .list_download_jobs( + crate::proto::data_acquisition::ListDownloadJobsRequest { + page: 0, + page_size: 50, + status_filter: 0, + start_time: 0, + end_time: 0, + }, + ) + .await? + .into_inner(); + + let jobs: Vec = response + .jobs + .into_iter() + .map(|j| DownloadJobInfo { + job_id: j.job_id, + status: download_status_name(j.status).to_owned(), + dataset: j.dataset, + symbols: j.symbols, + start_date: j.start_date, + end_date: j.end_date, + progress_pct: j.progress_percentage, + bytes_downloaded: 0, + total_bytes: 0, + records_count: 0, + data_quality_score: 0.0, + error_message: String::new(), + }) + .collect(); + + let result = StatusResult { + total_count: response.total_count, + jobs, + }; + output::render(&result, format)?; + } + + DataAction::Cache => { + let result = CacheResult { + message: + "Data cache managed by MinIO. Use `fxt data download` to fetch new data." + .to_owned(), + }; + output::render(&result, format)?; + } + + DataAction::Feeds => { + let result = FeedsResult { + message: + "Live feed status requires a running data acquisition service. Use `fxt service status data-acquisition` for details." + .to_owned(), + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/model.rs b/bin/fxt/src/commands/model.rs index 7a2fbe23c..6670d93b9 100644 --- a/bin/fxt/src/commands/model.rs +++ b/bin/fxt/src/commands/model.rs @@ -2,9 +2,12 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::ml; #[derive(Parser, Debug)] pub struct ModelCommand { @@ -37,7 +40,11 @@ enum ModelAction { symbol: String, }, /// Show ensemble composition and weights - Ensemble, + Ensemble { + /// Symbol for ensemble vote + #[arg(long, default_value = "ES.FUT")] + symbol: String, + }, /// Promote a model to production Promote { /// Model ID @@ -48,8 +55,418 @@ enum ModelAction { }, } -impl ModelCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("model command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct ModelListResult { + count: usize, + models: Vec, +} + +#[derive(Serialize)] +struct ModelRow { + name: String, + model_type: String, + description: String, + symbols: Vec, +} + +#[derive(Serialize)] +struct ModelStatusResult { + name: String, + state: String, + health: String, + error: String, + last_updated: i64, + last_prediction: i64, + metadata: Vec<(String, String)>, +} + +#[derive(Serialize)] +struct PredictResult { + model: String, + symbol: String, + prediction_type: String, + value: f64, + confidence: f64, + horizon_minutes: i32, +} + +#[derive(Serialize)] +struct EnsembleResult { + symbol: String, + consensus: String, + confidence: f64, + signal_strength: String, + votes_buy: i32, + votes_sell: i32, + votes_hold: i32, + total_models: i32, + individual_votes: Vec, +} + +#[derive(Serialize)] +struct VoteRow { + model: String, + prediction: String, + confidence: f64, + weight: f64, +} + +#[derive(Serialize)] +struct StubMessage { + message: String, +} + +// ── Display helpers ────────────────────────────────────────────────── + +fn model_state_str(s: i32) -> &'static str { + match s { + 1 => "loading", + 2 => "ready", + 3 => "predicting", + 4 => "training", + 5 => "error", + 6 => "offline", + _ => "unknown", + } +} + +fn model_health_str(h: i32) -> &'static str { + match h { + 1 => "healthy", + 2 => "degraded", + 3 => "unhealthy", + 4 => "critical", + _ => "unknown", + } +} + +fn prediction_type_str(p: i32) -> &'static str { + match p { + 1 => "BUY", + 2 => "SELL", + 3 => "HOLD", + 4 => "PRICE_UP", + 5 => "PRICE_DOWN", + 6 => "VOL_HIGH", + 7 => "VOL_LOW", + _ => "UNKNOWN", + } +} + +fn signal_strength_str(s: i32) -> &'static str { + match s { + 1 => "very_weak", + 2 => "weak", + 3 => "moderate", + 4 => "strong", + 5 => "very_strong", + _ => "unknown", + } +} + +fn colorize_state(s: &str) -> String { + match s { + "ready" | "predicting" => s.green().to_string(), + "loading" | "training" => s.cyan().to_string(), + "error" => s.red().bold().to_string(), + "offline" => s.dimmed().to_string(), + _ => s.dimmed().to_string(), + } +} + +fn colorize_health(s: &str) -> String { + match s { + "healthy" => s.green().bold().to_string(), + "degraded" => s.yellow().bold().to_string(), + "unhealthy" => s.red().to_string(), + "critical" => s.red().bold().to_string(), + _ => s.dimmed().to_string(), + } +} + +fn colorize_prediction(s: &str) -> String { + match s { + "BUY" | "PRICE_UP" => s.green().bold().to_string(), + "SELL" | "PRICE_DOWN" => s.red().bold().to_string(), + "HOLD" => s.yellow().to_string(), + _ => s.dimmed().to_string(), + } +} + +fn format_timestamp(ts: i64) -> String { + if ts == 0 { + return "-".to_owned(); + } + chrono::DateTime::from_timestamp(ts, 0) + .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()) + .unwrap_or_else(|| ts.to_string()) +} + +// ── HumanReadable impls ────────────────────────────────────────────── + +impl HumanReadable for ModelListResult { + fn print_human(&self) { + println!("{} model(s)\n", self.count); + println!( + " {:<24} {:<14} {:<20} {}", + "NAME".bold(), + "TYPE".bold(), + "SYMBOLS".bold(), + "DESCRIPTION".bold(), + ); + println!(" {}", "-".repeat(76)); + + for m in &self.models { + let syms = if m.symbols.is_empty() { + "-".to_owned() + } else if m.symbols.len() > 3 { + let first_three = m.symbols.get(..3).map_or_else( + || m.symbols.join(", "), + |s| s.join(", "), + ); + format!("{first_three}, +{}", m.symbols.len() - 3) + } else { + m.symbols.join(", ") + }; + println!( + " {:<24} {:<14} {:<20} {}", + m.name, + m.model_type, + syms, + m.description.dimmed(), + ); + } + } +} + +impl HumanReadable for ModelStatusResult { + fn print_human(&self) { + println!("{} {}", "Model:".bold(), self.name); + println!(" State: {}", colorize_state(&self.state)); + println!(" Health: {}", colorize_health(&self.health)); + if !self.error.is_empty() { + println!(" Error: {}", self.error.red()); + } + println!( + " Last Updated: {}", + format_timestamp(self.last_updated) + ); + println!( + " Last Prediction: {}", + format_timestamp(self.last_prediction) + ); + + if !self.metadata.is_empty() { + println!("\n {}:", "Metadata".bold()); + for (k, v) in &self.metadata { + println!(" {k}: {v}"); + } + } + } +} + +impl HumanReadable for PredictResult { + fn print_human(&self) { + println!( + "{} {} on {} -> {} (confidence {:.1}%)", + "Prediction:".bold(), + self.model, + self.symbol, + colorize_prediction(&self.prediction_type), + self.confidence * 100.0, + ); + println!(" Value: {:.6}", self.value); + println!(" Horizon: {} min", self.horizon_minutes); + } +} + +impl HumanReadable for EnsembleResult { + fn print_human(&self) { + println!( + "{} {} -- {} (confidence {:.1}%, signal {})", + "Ensemble:".bold(), + self.symbol, + colorize_prediction(&self.consensus), + self.confidence * 100.0, + self.signal_strength, + ); + println!( + " Votes: {} buy / {} sell / {} hold ({} models)", + self.votes_buy.to_string().green(), + self.votes_sell.to_string().red(), + self.votes_hold.to_string().yellow(), + self.total_models, + ); + + if !self.individual_votes.is_empty() { + println!( + "\n {:<24} {:<12} {:>12} {:>8}", + "MODEL".bold(), + "VOTE".bold(), + "CONFIDENCE".bold(), + "WEIGHT".bold(), + ); + println!(" {}", "-".repeat(60)); + + for v in &self.individual_votes { + println!( + " {:<24} {:<12} {:>11.1}% {:>7.3}", + v.model, + colorize_prediction(&v.prediction), + v.confidence * 100.0, + v.weight, + ); + } + } + } +} + +impl HumanReadable for StubMessage { + fn print_human(&self) { + println!("{}", self.message); + } +} + +// ── Execute ────────────────────────────────────────────────────────── + +impl ModelCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + ModelAction::List { model, status: _ } => { + let resp = client + .ml() + .get_available_models(ml::GetAvailableModelsRequest {}) + .await? + .into_inner(); + + let mut models: Vec = resp + .available_models + .iter() + .map(|m| ModelRow { + name: m.model_name.clone(), + model_type: m.model_type.clone(), + description: m.description.clone(), + symbols: m.supported_symbols.clone(), + }) + .collect(); + + // Apply client-side model type filter if provided. + if let Some(filter) = model { + let filter_upper = filter.to_uppercase(); + models.retain(|m| m.model_type.to_uppercase().contains(&filter_upper)); + } + + let result = ModelListResult { + count: models.len(), + models, + }; + output::render(&result, format)?; + } + ModelAction::Status { model_id } => { + let resp = client + .ml() + .get_model_status(ml::GetModelStatusRequest { + model_name: Some(model_id.clone()), + }) + .await? + .into_inner(); + + let status = resp + .model_statuses + .into_iter() + .find(|s| s.model_name == *model_id); + + let Some(s) = status else { + anyhow::bail!("model '{}' not found", model_id); + }; + + let metadata: Vec<(String, String)> = + s.metadata.into_iter().collect(); + + let result = ModelStatusResult { + name: s.model_name, + state: model_state_str(s.state).to_owned(), + health: model_health_str(s.health).to_owned(), + error: s.error_message.unwrap_or_default(), + last_updated: s.last_updated, + last_prediction: s.last_prediction, + metadata, + }; + output::render(&result, format)?; + } + ModelAction::Predict { model_id, symbol } => { + let resp = client + .ml() + .get_prediction(ml::GetPredictionRequest { + model_name: model_id.clone(), + symbol: symbol.clone(), + horizon_minutes: None, + features: Default::default(), + }) + .await? + .into_inner(); + + let pred = resp.prediction.unwrap_or_default(); + let result = PredictResult { + model: pred.model_name, + symbol: pred.symbol, + prediction_type: prediction_type_str(pred.prediction_type).to_owned(), + value: pred.value, + confidence: resp.confidence, + horizon_minutes: pred.horizon_minutes, + }; + output::render(&result, format)?; + } + ModelAction::Ensemble { symbol } => { + let resp = client + .ml() + .get_ensemble_vote(ml::GetEnsembleVoteRequest { + symbol: symbol.clone(), + horizon_minutes: None, + model_names: vec![], + }) + .await? + .into_inner(); + + let vote = resp.ensemble_vote.unwrap_or_default(); + let individual_votes: Vec = resp + .individual_votes + .iter() + .map(|v| VoteRow { + model: v.model_name.clone(), + prediction: prediction_type_str(v.prediction).to_owned(), + confidence: v.confidence, + weight: v.weight, + }) + .collect(); + + let result = EnsembleResult { + symbol: vote.symbol, + consensus: prediction_type_str(vote.consensus_prediction).to_owned(), + confidence: resp.overall_confidence, + signal_strength: signal_strength_str(vote.signal_strength).to_owned(), + votes_buy: vote.votes_buy, + votes_sell: vote.votes_sell, + votes_hold: vote.votes_hold, + total_models: vote.total_models, + individual_votes, + }; + output::render(&result, format)?; + } + ModelAction::Promote { + model_id: _, + reason: _, + } => { + let result = StubMessage { + message: "Model promotion is managed through the tuning pipeline.\n\ + Use `fxt tune approve ` to promote a tuned model." + .to_owned(), + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/risk.rs b/bin/fxt/src/commands/risk.rs index 8a7af1ef8..5b6ccb288 100644 --- a/bin/fxt/src/commands/risk.rs +++ b/bin/fxt/src/commands/risk.rs @@ -2,9 +2,11 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; #[derive(Parser, Debug)] pub struct RiskCommand { @@ -39,8 +41,373 @@ enum RiskAction { }, } -impl RiskCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("risk command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct CircuitBreakerInfo { + name: String, + is_triggered: bool, + trigger_reason: String, + breaker_type: String, +} + +#[derive(Serialize)] +struct RiskStatusResult { + var_1d: f64, + var_5d: f64, + current_drawdown: f64, + max_drawdown: f64, + sharpe_ratio: f64, + sortino_ratio: f64, + volatility: f64, + circuit_breakers: Vec, +} + +impl HumanReadable for RiskStatusResult { + fn print_human(&self) { + println!("{}", "Risk Status".cyan().bold()); + println!(" VaR (1-day): {:.4}", self.var_1d); + println!(" VaR (5-day): {:.4}", self.var_5d); + let dd_str = format!("{:.2}%", self.current_drawdown * 100.0); + let dd_colored = if self.current_drawdown > 0.05 { + dd_str.red().to_string() + } else if self.current_drawdown > 0.02 { + dd_str.yellow().to_string() + } else { + dd_str.green().to_string() + }; + println!(" Current Drawdown: {dd_colored}"); + println!(" Max Drawdown: {:.2}%", self.max_drawdown * 100.0); + println!(" Sharpe Ratio: {:.2}", self.sharpe_ratio); + println!(" Sortino Ratio: {:.2}", self.sortino_ratio); + println!(" Volatility: {:.4}", self.volatility); + + println!(); + println!("{}", "Circuit Breakers".cyan().bold()); + if self.circuit_breakers.is_empty() { + println!(" No circuit breakers configured."); + } + for cb in &self.circuit_breakers { + let status = if cb.is_triggered { + "TRIGGERED".red().bold().to_string() + } else { + "OK".green().to_string() + }; + println!(" [{}] {} ({})", status, cb.name, cb.breaker_type); + if cb.is_triggered && !cb.trigger_reason.is_empty() { + println!(" Reason: {}", cb.trigger_reason); + } + } + } +} + +#[derive(Serialize)] +struct PositionRiskInfo { + symbol: String, + position_size: f64, + market_value: f64, + var_contribution: f64, + concentration_risk: f64, + liquidity_risk: f64, +} + +#[derive(Serialize)] +struct LimitsResult { + var_1d: f64, + var_5d: f64, + var_30d: f64, + positions: Vec, +} + +impl HumanReadable for LimitsResult { + fn print_human(&self) { + println!("{}", "Position Limits".cyan().bold()); + println!(" Portfolio VaR (1d): {:.4}", self.var_1d); + println!(" Portfolio VaR (5d): {:.4}", self.var_5d); + println!(" Portfolio VaR (30d): {:.4}", self.var_30d); + if !self.positions.is_empty() { + println!(); + println!( + " {:>10} {:>12} {:>12} {:>10} {:>10}", + "Symbol", "Size", "Market Val", "VaR Contr", "Concent." + ); + for p in &self.positions { + println!( + " {:>10} {:>12.2} {:>12.2} {:>10.4} {:>9.1}%", + p.symbol, + p.position_size, + p.market_value, + p.var_contribution, + p.concentration_risk, + ); + } + } + } +} + +#[derive(Serialize)] +struct DrawdownResult { + current_drawdown: f64, + max_drawdown: f64, + sharpe_ratio: f64, + sortino_ratio: f64, + beta: f64, + alpha: f64, +} + +impl HumanReadable for DrawdownResult { + fn print_human(&self) { + println!("{}", "Drawdown Analysis".cyan().bold()); + println!(" Current Drawdown: {:.2}%", self.current_drawdown * 100.0); + println!(" Max Drawdown: {:.2}%", self.max_drawdown * 100.0); + println!(" Sharpe Ratio: {:.2}", self.sharpe_ratio); + println!(" Sortino Ratio: {:.2}", self.sortino_ratio); + println!(" Beta: {:.4}", self.beta); + println!(" Alpha: {:.4}", self.alpha); + } +} + +#[derive(Serialize)] +struct StressResult { + symbol: String, + portfolio_risk_score: f64, + positions: Vec, +} + +impl HumanReadable for StressResult { + fn print_human(&self) { + println!("{}", "Stress Test Results".cyan().bold()); + if !self.symbol.is_empty() { + println!(" Symbol filter: {}", self.symbol); + } + let score_str = format!("{:.1}", self.portfolio_risk_score); + let score_colored = if self.portfolio_risk_score > 75.0 { + score_str.red().to_string() + } else if self.portfolio_risk_score > 50.0 { + score_str.yellow().to_string() + } else { + score_str.green().to_string() + }; + println!(" Portfolio Risk Score: {score_colored}/100"); + for p in &self.positions { + println!(); + println!(" {} :", p.symbol.bold()); + println!(" Position Size: {:.2}", p.position_size); + println!(" Market Value: {:.2}", p.market_value); + println!(" VaR Contribution: {:.4}", p.var_contribution); + println!(" Concentration Risk: {:.1}", p.concentration_risk); + println!(" Liquidity Risk: {:.1}", p.liquidity_risk); + } + } +} + +#[derive(Serialize)] +struct EmergencyResult { + success: bool, + message: String, + affected_orders: Vec, +} + +impl HumanReadable for EmergencyResult { + fn print_human(&self) { + if self.success { + println!( + "{} {}", + "EMERGENCY STOP ACTIVATED".red().bold(), + self.message, + ); + if !self.affected_orders.is_empty() { + println!( + " Affected orders ({}): {}", + self.affected_orders.len(), + self.affected_orders.join(", "), + ); + } + } else { + println!("{}: {}", "Emergency stop failed".red(), self.message); + } + } +} + +// ── Helpers ─────────────────────────────────────────────────────────── + +fn breaker_type_name(value: i32) -> &'static str { + match value { + 1 => "portfolio-loss", + 2 => "symbol-volatility", + 3 => "position-size", + 4 => "drawdown", + _ => "unknown", + } +} + +fn position_risk_to_info(pr: &crate::proto::risk::PositionRisk) -> PositionRiskInfo { + PositionRiskInfo { + symbol: pr.symbol.clone(), + position_size: pr.position_size, + market_value: pr.market_value, + var_contribution: pr.var_contribution, + concentration_risk: pr.concentration_risk, + liquidity_risk: pr.liquidity_risk, + } +} + +// ── Execute ─────────────────────────────────────────────────────────── + +impl RiskCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + RiskAction::Status => { + let metrics_resp = client + .risk() + .get_risk_metrics(crate::proto::risk::GetRiskMetricsRequest { + portfolio_id: None, + }) + .await? + .into_inner(); + + let cb_resp = client + .risk() + .get_circuit_breaker_status( + crate::proto::risk::GetCircuitBreakerStatusRequest { symbol: None }, + ) + .await? + .into_inner(); + + let m = metrics_resp.metrics.unwrap_or_default(); + + let circuit_breakers: Vec = cb_resp + .circuit_breakers + .iter() + .map(|cb| CircuitBreakerInfo { + name: cb.name.clone(), + is_triggered: cb.is_triggered, + trigger_reason: cb.trigger_reason.clone().unwrap_or_default(), + breaker_type: breaker_type_name(cb.breaker_type).to_owned(), + }) + .collect(); + + let result = RiskStatusResult { + var_1d: m.portfolio_var_1d, + var_5d: m.portfolio_var_5d, + current_drawdown: m.current_drawdown, + max_drawdown: m.max_drawdown, + sharpe_ratio: m.sharpe_ratio, + sortino_ratio: m.sortino_ratio, + volatility: m.volatility, + circuit_breakers, + }; + output::render(&result, format)?; + } + + RiskAction::Limits { key: _, value: _ } => { + let metrics_resp = client + .risk() + .get_risk_metrics(crate::proto::risk::GetRiskMetricsRequest { + portfolio_id: None, + }) + .await? + .into_inner(); + + let m = metrics_resp.metrics.unwrap_or_default(); + + let positions: Vec = + m.position_risks.iter().map(position_risk_to_info).collect(); + + let result = LimitsResult { + var_1d: m.portfolio_var_1d, + var_5d: m.portfolio_var_5d, + var_30d: m.portfolio_var_30d, + positions, + }; + output::render(&result, format)?; + } + + RiskAction::Drawdown => { + let metrics_resp = client + .risk() + .get_risk_metrics(crate::proto::risk::GetRiskMetricsRequest { + portfolio_id: None, + }) + .await? + .into_inner(); + + let m = metrics_resp.metrics.unwrap_or_default(); + + let result = DrawdownResult { + current_drawdown: m.current_drawdown, + max_drawdown: m.max_drawdown, + sharpe_ratio: m.sharpe_ratio, + sortino_ratio: m.sortino_ratio, + beta: m.beta, + alpha: m.alpha, + }; + output::render(&result, format)?; + } + + RiskAction::Stress { scenario } => { + let symbol_filter = scenario.clone().unwrap_or_default(); + + let resp = client + .risk() + .get_position_risk(crate::proto::risk::GetPositionRiskRequest { + symbol: if symbol_filter.is_empty() { + None + } else { + Some(symbol_filter.clone()) + }, + account_id: None, + }) + .await? + .into_inner(); + + let positions: Vec = resp + .position_risks + .iter() + .map(position_risk_to_info) + .collect(); + + let result = StressResult { + symbol: symbol_filter, + portfolio_risk_score: resp.portfolio_risk_score, + positions, + }; + output::render(&result, format)?; + } + + RiskAction::Emergency { confirm } => { + if !confirm { + println!( + "{}", + "Emergency stop requires --confirm flag. This will halt ALL trading." + .yellow() + ); + println!( + " Usage: {} risk emergency --confirm", + "fxt".bold() + ); + return Ok(()); + } + + let resp = client + .risk() + .emergency_stop(crate::proto::risk::EmergencyStopRequest { + stop_type: 1, // ALL_TRADING + reason: "Manual emergency halt via fxt CLI".to_owned(), + symbol: None, + account_id: None, + }) + .await? + .into_inner(); + + let result = EmergencyResult { + success: resp.success, + message: resp.message, + affected_orders: resp.affected_orders, + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/service.rs b/bin/fxt/src/commands/service.rs index e5d49264b..44434067c 100644 --- a/bin/fxt/src/commands/service.rs +++ b/bin/fxt/src/commands/service.rs @@ -2,9 +2,12 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::monitoring; #[derive(Parser, Debug)] pub struct ServiceCommand { @@ -43,8 +46,413 @@ enum ServiceAction { Health, } -impl ServiceCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("service command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct ServiceListResult { + overall_health: String, + healthy: i32, + total: i32, + uptime_seconds: i64, + services: Vec, +} + +#[derive(Serialize)] +struct ServiceEntry { + name: String, + health: String, + state: String, + version: String, + uptime_seconds: i64, + error: String, +} + +#[derive(Serialize)] +struct ServiceStatusResult { + name: String, + health: String, + state: String, + version: String, + uptime_seconds: i64, + error: String, + metadata: Vec<(String, String)>, + dependencies: Vec, +} + +#[derive(Serialize)] +struct DependencyEntry { + name: String, + dep_type: String, + status: String, + endpoint: String, + response_time_ms: f64, +} + +#[derive(Serialize)] +struct HealthCheckResult { + overall: String, + checks: Vec, +} + +#[derive(Serialize)] +struct HealthEntry { + name: String, + status: String, + message: String, + response_time_ms: f64, +} + +// ── Display helpers ────────────────────────────────────────────────── + +fn service_health_str(h: i32) -> &'static str { + match h { + 1 => "healthy", + 2 => "degraded", + 3 => "unhealthy", + 4 => "critical", + _ => "unknown", + } +} + +fn service_state_str(s: i32) -> &'static str { + match s { + 1 => "starting", + 2 => "running", + 3 => "stopping", + 4 => "stopped", + 5 => "error", + _ => "unknown", + } +} + +fn system_health_str(h: i32) -> &'static str { + match h { + 1 => "healthy", + 2 => "degraded", + 3 => "unhealthy", + 4 => "critical", + _ => "unknown", + } +} + +fn health_status_str(h: i32) -> &'static str { + match h { + 1 => "healthy", + 2 => "degraded", + 3 => "unhealthy", + 4 => "critical", + _ => "unknown", + } +} + +fn dependency_type_str(t: i32) -> &'static str { + match t { + 1 => "database", + 2 => "message_queue", + 3 => "cache", + 4 => "external_api", + 5 => "file_system", + 6 => "network", + _ => "unknown", + } +} + +fn colorize_health(s: &str) -> String { + match s { + "healthy" => s.green().bold().to_string(), + "degraded" => s.yellow().bold().to_string(), + "unhealthy" => s.red().to_string(), + "critical" => s.red().bold().to_string(), + _ => s.dimmed().to_string(), + } +} + +fn colorize_state(s: &str) -> String { + match s { + "running" => s.green().to_string(), + "starting" | "stopping" => s.yellow().to_string(), + "stopped" => s.dimmed().to_string(), + "error" => s.red().bold().to_string(), + _ => s.dimmed().to_string(), + } +} + +#[allow(clippy::integer_division, clippy::modulo_arithmetic)] +fn format_uptime(secs: i64) -> String { + if secs < 60 { + return format!("{secs}s"); + } + let days = secs / 86400; + let hours = (secs % 86400) / 3600; + let mins = (secs % 3600) / 60; + if days > 0 { + format!("{days}d {hours}h {mins}m") + } else if hours > 0 { + format!("{hours}h {mins}m") + } else { + format!("{mins}m") + } +} + +// ── HumanReadable impls ────────────────────────────────────────────── + +impl HumanReadable for ServiceListResult { + fn print_human(&self) { + println!( + "{} {} ({}/{} healthy, uptime {})\n", + "System:".bold(), + colorize_health(&self.overall_health), + self.healthy, + self.total, + format_uptime(self.uptime_seconds).dimmed(), + ); + + println!( + " {:<28} {:<12} {:<12} {:<10} {}", + "SERVICE".bold(), + "HEALTH".bold(), + "STATE".bold(), + "UPTIME".bold(), + "VERSION".bold(), + ); + println!(" {}", "-".repeat(76)); + + for svc in &self.services { + let err_suffix = if svc.error.is_empty() { + String::new() + } else { + format!(" {}", svc.error.red()) + }; + println!( + " {:<28} {:<12} {:<12} {:<10} {}{}", + svc.name, + colorize_health(&svc.health), + colorize_state(&svc.state), + format_uptime(svc.uptime_seconds), + svc.version.dimmed(), + err_suffix, + ); + } + } +} + +impl HumanReadable for ServiceStatusResult { + fn print_human(&self) { + println!("{} {}", "Service:".bold(), self.name); + println!(" Health: {}", colorize_health(&self.health)); + println!(" State: {}", colorize_state(&self.state)); + println!(" Version: {}", self.version); + println!(" Uptime: {}", format_uptime(self.uptime_seconds)); + + if !self.error.is_empty() { + println!(" Error: {}", self.error.red()); + } + + if !self.metadata.is_empty() { + println!("\n {}:", "Metadata".bold()); + for (k, v) in &self.metadata { + println!(" {k}: {v}"); + } + } + + if !self.dependencies.is_empty() { + println!("\n {}:", "Dependencies".bold()); + for dep in &self.dependencies { + let status_col = colorize_health(&dep.status); + let ep = if dep.endpoint.is_empty() { + String::new() + } else { + format!(" ({})", dep.endpoint.dimmed()) + }; + println!( + " {} [{}] {} {:.1}ms{}", + dep.name, dep.dep_type, status_col, dep.response_time_ms, ep, + ); + } + } + } +} + +impl HumanReadable for HealthCheckResult { + fn print_human(&self) { + println!("{} {}\n", "Overall:".bold(), colorize_health(&self.overall)); + + println!( + " {:<32} {:<12} {:>10} {}", + "CHECK".bold(), + "STATUS".bold(), + "LATENCY".bold(), + "MESSAGE".bold(), + ); + println!(" {}", "-".repeat(72)); + + for c in &self.checks { + let latency_str = if c.response_time_ms > 0.0 { + format!("{:.1}ms", c.response_time_ms) + } else { + "-".to_owned() + }; + println!( + " {:<32} {:<12} {:>10} {}", + c.name, + colorize_health(&c.status), + latency_str, + c.message.dimmed(), + ); + } + } +} + +// ── Stub result for non-gRPC commands ──────────────────────────────── + +#[derive(Serialize)] +struct StubMessage { + message: String, +} + +impl HumanReadable for StubMessage { + fn print_human(&self) { + println!("{}", self.message); + } +} + +// ── Execute ────────────────────────────────────────────────────────── + +impl ServiceCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + ServiceAction::List => { + let resp = client + .monitoring() + .get_system_status(monitoring::GetSystemStatusRequest { + service_names: vec![], + }) + .await? + .into_inner(); + + let overall = resp.overall_status.unwrap_or_default(); + let services: Vec = resp + .service_statuses + .iter() + .map(|s| ServiceEntry { + name: s.service_name.clone(), + health: service_health_str(s.health).to_owned(), + state: service_state_str(s.state).to_owned(), + version: s.version.clone().unwrap_or_default(), + uptime_seconds: s.uptime_seconds, + error: s.error_message.clone().unwrap_or_default(), + }) + .collect(); + + let result = ServiceListResult { + overall_health: system_health_str(overall.overall_health).to_owned(), + healthy: overall.healthy_services, + total: overall.total_services, + uptime_seconds: overall.system_uptime_seconds, + services, + }; + output::render(&result, format)?; + } + ServiceAction::Status { service } => { + let resp = client + .monitoring() + .get_system_status(monitoring::GetSystemStatusRequest { + service_names: vec![service.clone()], + }) + .await? + .into_inner(); + + let maybe_svc = resp + .service_statuses + .into_iter() + .find(|s| s.service_name == *service); + + let Some(svc) = maybe_svc else { + anyhow::bail!("service '{}' not found in system status", service); + }; + + let metadata: Vec<(String, String)> = + svc.metadata.into_iter().collect(); + + let dependencies: Vec = svc + .dependencies + .iter() + .map(|d| DependencyEntry { + name: d.name.clone(), + dep_type: dependency_type_str(d.dependency_type).to_owned(), + status: health_status_str(d.status).to_owned(), + endpoint: d.endpoint.clone().unwrap_or_default(), + response_time_ms: d.response_time_ms.unwrap_or_default(), + }) + .collect(); + + let result = ServiceStatusResult { + name: svc.service_name, + health: service_health_str(svc.health).to_owned(), + state: service_state_str(svc.state).to_owned(), + version: svc.version.unwrap_or_default(), + uptime_seconds: svc.uptime_seconds, + error: svc.error_message.unwrap_or_default(), + metadata, + dependencies, + }; + output::render(&result, format)?; + } + ServiceAction::Health => { + let resp = client + .monitoring() + .get_health_check(monitoring::GetHealthCheckRequest { + service_name: None, + }) + .await? + .into_inner(); + + let checks: Vec = resp + .health_checks + .iter() + .map(|c| HealthEntry { + name: c.check_name.clone(), + status: health_status_str(c.status).to_owned(), + message: c.message.clone().unwrap_or_default(), + response_time_ms: c.response_time_ms.unwrap_or_default(), + }) + .collect(); + + let result = HealthCheckResult { + overall: health_status_str(resp.health_status).to_owned(), + checks, + }; + output::render(&result, format)?; + } + ServiceAction::Logs { service, follow: _ } => { + let result = StubMessage { + message: format!( + "Service logs are available via kubectl:\n \ + kubectl logs -f deploy/{service} -n foxhunt" + ), + }; + output::render(&result, format)?; + } + ServiceAction::Restart { service } => { + let result = StubMessage { + message: format!( + "Service restart requires kubectl access:\n \ + kubectl rollout restart deploy/{service} -n foxhunt" + ), + }; + output::render(&result, format)?; + } + ServiceAction::Deploy { service } => { + let result = StubMessage { + message: format!( + "Use pod-writer-deploy for deployment:\n \ + kubectl apply -f infra/k8s/services/{service}.yaml && \ + kubectl rollout restart deploy/{service} -n foxhunt" + ), + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/trade.rs b/bin/fxt/src/commands/trade.rs index 5c6cfb319..e45cdeed9 100644 --- a/bin/fxt/src/commands/trade.rs +++ b/bin/fxt/src/commands/trade.rs @@ -2,9 +2,12 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::trading; #[derive(Parser, Debug)] pub struct TradeCommand { @@ -36,14 +39,342 @@ enum TradeAction { }, /// List open positions Positions, - /// List open orders - Orders, + /// Query order status or list orders + Orders { + /// Order ID to query (omit to list recent orders) + order_id: Option, + }, /// Account summary (balance, margin, P&L) Account, } -impl TradeCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("trade command not yet implemented") +// ── Result types ──────────────────────────────────────────────────── + +#[derive(Serialize)] +struct SubmitResult { + order_id: String, + status: String, + message: String, +} + +impl HumanReadable for SubmitResult { + fn print_human(&self) { + println!( + "{} Order submitted: {}", + "OK".green().bold(), + self.order_id.cyan() + ); + println!(" Status: {}", self.status); + if !self.message.is_empty() { + println!(" Message: {}", self.message); + } + } +} + +#[derive(Serialize)] +struct CancelResult { + success: bool, + message: String, +} + +impl HumanReadable for CancelResult { + fn print_human(&self) { + if self.success { + println!("{} {}", "OK".green().bold(), self.message); + } else { + println!("{} {}", "FAIL".red().bold(), self.message); + } + } +} + +#[derive(Serialize)] +struct PositionInfo { + symbol: String, + quantity: f64, + avg_price: f64, + market_value: f64, + unrealized_pnl: f64, + realized_pnl: f64, +} + +#[derive(Serialize)] +struct PositionsResult { + positions: Vec, +} + +impl HumanReadable for PositionsResult { + fn print_human(&self) { + if self.positions.is_empty() { + println!("{}", "No open positions.".dimmed()); + return; + } + println!("{}", "Open Positions".bold().underline()); + println!( + " {:<12} {:>10} {:>12} {:>14} {:>14}", + "Symbol", "Qty", "Avg Price", "Mkt Value", "Unreal P&L" + ); + println!(" {}", "-".repeat(66)); + for p in &self.positions { + let pnl_colored = if p.unrealized_pnl >= 0.0 { + format!("{:>14.2}", p.unrealized_pnl).green() + } else { + format!("{:>14.2}", p.unrealized_pnl).red() + }; + println!( + " {:<12} {:>10.2} {:>12.2} {:>14.2} {}", + p.symbol, p.quantity, p.avg_price, p.market_value, pnl_colored + ); + } + } +} + +#[derive(Serialize)] +struct OrderStatusResult { + order_id: String, + symbol: String, + side: String, + quantity: f64, + filled_quantity: f64, + order_type: String, + price: Option, + status: String, +} + +impl HumanReadable for OrderStatusResult { + fn print_human(&self) { + println!("{}", "Order Details".bold().underline()); + println!(" Order ID: {}", self.order_id.cyan()); + println!(" Symbol: {}", self.symbol); + println!(" Side: {}", self.side); + println!(" Type: {}", self.order_type); + println!(" Qty: {} / {} filled", self.filled_quantity, self.quantity); + if let Some(px) = self.price { + println!(" Price: ${:.2}", px); + } + println!(" Status: {}", self.status); + } +} + +#[derive(Serialize)] +struct AccountResult { + account_id: String, + cash_balance: f64, + equity: f64, + margin_used: f64, + margin_available: f64, + buying_power: f64, + unrealized_pnl: f64, + realized_pnl: f64, +} + +impl HumanReadable for AccountResult { + fn print_human(&self) { + println!("{}", "Account Summary".bold().underline()); + println!(" Account: {}", self.account_id.cyan()); + println!(" Cash Balance: ${:>14.2}", self.cash_balance); + println!(" Equity: ${:>14.2}", self.equity); + println!(" Margin Used: ${:>14.2}", self.margin_used); + println!(" Margin Avail: ${:>14.2}", self.margin_available); + println!(" Buying Power: ${:>14.2}", self.buying_power); + let upnl = if self.unrealized_pnl >= 0.0 { + format!("${:>14.2}", self.unrealized_pnl).green().to_string() + } else { + format!("${:>14.2}", self.unrealized_pnl).red().to_string() + }; + let rpnl = if self.realized_pnl >= 0.0 { + format!("${:>14.2}", self.realized_pnl).green().to_string() + } else { + format!("${:>14.2}", self.realized_pnl).red().to_string() + }; + println!(" Unrealized P&L: {upnl}"); + println!(" Realized P&L: {rpnl}"); + } +} + +// ── Helpers ───────────────────────────────────────────────────────── + +fn parse_side(s: &str) -> Result { + match s.to_lowercase().as_str() { + "buy" | "long" => Ok(trading::OrderSide::Buy.into()), + "sell" | "short" => Ok(trading::OrderSide::Sell.into()), + _ => anyhow::bail!("invalid side '{s}': expected 'buy' or 'sell'"), + } +} + +fn side_name(v: i32) -> &'static str { + match trading::OrderSide::try_from(v) { + Ok(trading::OrderSide::Buy) => "BUY", + Ok(trading::OrderSide::Sell) => "SELL", + _ => "UNKNOWN", + } +} + +fn order_type_name(v: i32) -> &'static str { + match trading::OrderType::try_from(v) { + Ok(trading::OrderType::Market) => "MARKET", + Ok(trading::OrderType::Limit) => "LIMIT", + Ok(trading::OrderType::Stop) => "STOP", + Ok(trading::OrderType::StopLimit) => "STOP_LIMIT", + _ => "UNKNOWN", + } +} + +fn order_status_name(v: i32) -> &'static str { + match trading::OrderStatus::try_from(v) { + Ok(trading::OrderStatus::Pending) => "PENDING", + Ok(trading::OrderStatus::Submitted) => "SUBMITTED", + Ok(trading::OrderStatus::PartiallyFilled) => "PARTIALLY_FILLED", + Ok(trading::OrderStatus::Filled) => "FILLED", + Ok(trading::OrderStatus::Cancelled) => "CANCELLED", + Ok(trading::OrderStatus::Rejected) => "REJECTED", + _ => "UNKNOWN", + } +} + +// ── Execute ───────────────────────────────────────────────────────── + +impl TradeCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + TradeAction::Submit { + symbol, + side, + qty, + price, + } => { + let order_type = if price.is_some() { + trading::OrderType::Limit.into() + } else { + trading::OrderType::Market.into() + }; + + let resp = client + .trading() + .submit_order(trading::SubmitOrderRequest { + symbol: symbol.clone(), + side: parse_side(side)?, + quantity: *qty, + order_type, + price: *price, + stop_price: None, + account_id: String::new(), + metadata: std::collections::HashMap::new(), + }) + .await? + .into_inner(); + + let result = SubmitResult { + order_id: resp.order_id, + status: order_status_name(resp.status).to_owned(), + message: resp.message, + }; + output::render(&result, format)?; + } + + TradeAction::Cancel { order_id } => { + let resp = client + .trading() + .cancel_order(trading::CancelOrderRequest { + order_id: order_id.clone(), + account_id: String::new(), + }) + .await? + .into_inner(); + + let result = CancelResult { + success: resp.success, + message: resp.message, + }; + output::render(&result, format)?; + } + + TradeAction::Positions => { + let resp = client + .trading() + .get_positions(trading::GetPositionsRequest { + account_id: None, + symbol: None, + }) + .await? + .into_inner(); + + let positions = resp + .positions + .into_iter() + .map(|p| PositionInfo { + symbol: p.symbol, + quantity: p.quantity, + avg_price: p.average_price, + market_value: p.market_value, + unrealized_pnl: p.unrealized_pnl, + realized_pnl: p.realized_pnl, + }) + .collect(); + + output::render(&PositionsResult { positions }, format)?; + } + + TradeAction::Orders { order_id } => { + let Some(oid) = order_id else { + println!( + "{} Use {} to query a specific order, or {} for streaming.", + "INFO".blue().bold(), + "fxt trade orders ".cyan(), + "fxt watch".cyan(), + ); + return Ok(()); + }; + + let resp = client + .trading() + .get_order_status(trading::GetOrderStatusRequest { + order_id: oid.clone(), + }) + .await? + .into_inner(); + + let Some(order) = resp.order else { + anyhow::bail!("order {oid} not found"); + }; + + let result = OrderStatusResult { + order_id: order.order_id, + symbol: order.symbol, + side: side_name(order.side).to_owned(), + quantity: order.quantity, + filled_quantity: order.filled_quantity, + order_type: order_type_name(order.order_type).to_owned(), + price: order.price, + status: order_status_name(order.status).to_owned(), + }; + output::render(&result, format)?; + } + + TradeAction::Account => { + let resp = client + .broker_gateway() + .get_account_state( + crate::proto::broker_gateway::GetAccountStateRequest { + account_id: String::new(), + }, + ) + .await? + .into_inner(); + + let result = AccountResult { + account_id: resp.account_id, + cash_balance: resp.cash_balance, + equity: resp.equity, + margin_used: resp.margin_used, + margin_available: resp.margin_available, + buying_power: resp.buying_power, + unrealized_pnl: resp.unrealized_pnl, + realized_pnl: resp.realized_pnl, + }; + output::render(&result, format)?; + } + } + + Ok(()) } } diff --git a/bin/fxt/src/commands/train.rs b/bin/fxt/src/commands/train.rs index c3f9ec06c..d8af0a734 100644 --- a/bin/fxt/src/commands/train.rs +++ b/bin/fxt/src/commands/train.rs @@ -2,9 +2,13 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; +use tokio_stream::StreamExt; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::ml_training; #[derive(Parser, Debug)] pub struct TrainCommand { @@ -55,8 +59,424 @@ enum TrainAction { }, } -impl TrainCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("train command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct TrainStartResult { + job_id: String, + status: String, + message: String, +} + +#[derive(Serialize)] +struct TrainStopResult { + success: bool, + message: String, +} + +#[derive(Serialize)] +struct TrainStatusResult { + job_id: String, + model_type: String, + status: String, + description: String, + created_at: i64, + started_at: i64, + completed_at: i64, + error: String, + artifact_path: String, + financial_metrics: Option, +} + +#[derive(Serialize)] +struct FinancialMetricsSummary { + sharpe_ratio: f32, + max_drawdown: f32, + hit_rate: f32, + simulated_return: f32, +} + +#[derive(Serialize)] +struct TrainListResult { + total: u32, + jobs: Vec, +} + +#[derive(Serialize)] +struct TrainJobRow { + job_id: String, + model_type: String, + status: String, + description: String, + final_loss: f32, + best_val: f32, + created_at: i64, +} + +// ── Display helpers ────────────────────────────────────────────────── + +fn training_status_str(s: i32) -> &'static str { + match s { + 1 => "pending", + 2 => "running", + 3 => "completed", + 4 => "failed", + 5 => "stopped", + 6 => "paused", + _ => "unknown", + } +} + +fn colorize_status(s: &str) -> String { + match s { + "running" => s.cyan().bold().to_string(), + "completed" => s.green().bold().to_string(), + "failed" => s.red().bold().to_string(), + "stopped" => s.yellow().to_string(), + "pending" => s.dimmed().to_string(), + "paused" => s.yellow().dimmed().to_string(), + _ => s.dimmed().to_string(), + } +} + +fn status_filter_to_i32(s: &str) -> i32 { + match s.to_lowercase().as_str() { + "pending" => 1, + "running" => 2, + "completed" => 3, + "failed" => 4, + "stopped" => 5, + "paused" => 6, + _ => 0, + } +} + +fn format_timestamp(ts: i64) -> String { + if ts == 0 { + return "-".to_owned(); + } + chrono::DateTime::from_timestamp(ts, 0) + .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()) + .unwrap_or_else(|| ts.to_string()) +} + +// ── HumanReadable impls ────────────────────────────────────────────── + +impl HumanReadable for TrainStartResult { + fn print_human(&self) { + println!( + "{} Training job {} started (status: {})", + "OK".green().bold(), + self.job_id.bold(), + colorize_status(&self.status), + ); + if !self.message.is_empty() { + println!(" {}", self.message.dimmed()); + } + } +} + +impl HumanReadable for TrainStopResult { + fn print_human(&self) { + if self.success { + println!("{} {}", "OK".green().bold(), self.message); + } else { + println!("{} {}", "FAILED".red().bold(), self.message); + } + } +} + +impl HumanReadable for TrainStatusResult { + fn print_human(&self) { + println!("{} {}", "Job:".bold(), self.job_id); + println!(" Model: {}", self.model_type); + println!(" Status: {}", colorize_status(&self.status)); + if !self.description.is_empty() { + println!(" Description: {}", self.description); + } + println!(" Created: {}", format_timestamp(self.created_at)); + if self.started_at > 0 { + println!(" Started: {}", format_timestamp(self.started_at)); + } + if self.completed_at > 0 { + println!(" Completed: {}", format_timestamp(self.completed_at)); + } + if !self.error.is_empty() { + println!(" Error: {}", self.error.red()); + } + if !self.artifact_path.is_empty() { + println!(" Artifact: {}", self.artifact_path.dimmed()); + } + if let Some(fm) = &self.financial_metrics { + println!("\n {}:", "Financial Metrics".bold()); + println!(" Sharpe: {:.3}", fm.sharpe_ratio); + println!(" MaxDD: {:.2}%", fm.max_drawdown * 100.0); + println!(" Hit Rate: {:.1}%", fm.hit_rate * 100.0); + println!(" Return: {:.2}%", fm.simulated_return * 100.0); + } + } +} + +impl HumanReadable for TrainListResult { + fn print_human(&self) { + println!("{} training job(s)\n", self.total); + println!( + " {:<20} {:<12} {:<12} {:>10} {:>10} {}", + "JOB ID".bold(), + "MODEL".bold(), + "STATUS".bold(), + "LOSS".bold(), + "BEST VAL".bold(), + "CREATED".bold(), + ); + println!(" {}", "-".repeat(86)); + + for j in &self.jobs { + let loss_str = if j.final_loss > 0.0 { + format!("{:.4}", j.final_loss) + } else { + "-".to_owned() + }; + let val_str = if j.best_val > 0.0 { + format!("{:.4}", j.best_val) + } else { + "-".to_owned() + }; + println!( + " {:<20} {:<12} {:<12} {:>10} {:>10} {}", + truncate_id(&j.job_id, 18), + j.model_type, + colorize_status(&j.status), + loss_str, + val_str, + format_timestamp(j.created_at).dimmed(), + ); + } + } +} + +fn truncate_id(id: &str, max: usize) -> String { + if id.len() <= max { + id.to_owned() + } else { + // Safe: only truncate at ASCII boundaries (UUIDs and job IDs are ASCII) + let suffix_len = max.saturating_sub(2); + let chars: Vec = id.chars().collect(); + let start = chars.len().saturating_sub(suffix_len); + let tail: String = chars.get(start..).unwrap_or_default().iter().collect(); + format!("..{tail}") + } +} + +// ── Execute ────────────────────────────────────────────────────────── + +impl TrainCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + TrainAction::Start { model, config, gpu } => { + let data_source = config.as_ref().map(|path| ml_training::DataSource { + source: Some(ml_training::data_source::Source::FilePath(path.clone())), + start_time: 0, + end_time: 0, + }); + + let resp = client + .ml_training() + .start_training(ml_training::StartTrainingRequest { + model_type: model.to_uppercase(), + data_source, + hyperparameters: None, + use_gpu: *gpu, + description: String::new(), + tags: Default::default(), + mode: 0, // FULL + resume_checkpoint_path: String::new(), + max_epochs: 0, + }) + .await? + .into_inner(); + + let result = TrainStartResult { + job_id: resp.job_id, + status: training_status_str(resp.status).to_owned(), + message: resp.message, + }; + output::render(&result, format)?; + } + TrainAction::Stop { job_id } => { + let resp = client + .ml_training() + .stop_training(ml_training::StopTrainingRequest { + job_id: job_id.clone(), + reason: String::new(), + }) + .await? + .into_inner(); + + let result = TrainStopResult { + success: resp.success, + message: resp.message, + }; + output::render(&result, format)?; + } + TrainAction::Status { job_id } => { + let resp = client + .ml_training() + .get_training_job_details( + ml_training::GetTrainingJobDetailsRequest { + job_id: job_id.clone(), + }, + ) + .await? + .into_inner(); + + let details = resp.job_details.unwrap_or_default(); + let fm = details.final_financial_metrics.map(|m| { + FinancialMetricsSummary { + sharpe_ratio: m.sharpe_ratio, + max_drawdown: m.max_drawdown, + hit_rate: m.hit_rate, + simulated_return: m.simulated_return, + } + }); + + let result = TrainStatusResult { + job_id: details.job_id, + model_type: details.model_type, + status: training_status_str(details.status).to_owned(), + description: details.description, + created_at: details.created_at, + started_at: details.started_at, + completed_at: details.completed_at, + error: details.error_message, + artifact_path: details.model_artifact_path, + financial_metrics: fm, + }; + output::render(&result, format)?; + } + TrainAction::List { status, model } => { + let status_filter = status + .as_deref() + .map(status_filter_to_i32) + .unwrap_or(0); + + let resp = client + .ml_training() + .list_training_jobs(ml_training::ListTrainingJobsRequest { + page: 0, + page_size: 50, + status_filter, + model_type_filter: model.clone().unwrap_or_default(), + start_time: 0, + end_time: 0, + }) + .await? + .into_inner(); + + let jobs: Vec = resp + .jobs + .iter() + .map(|j| TrainJobRow { + job_id: j.job_id.clone(), + model_type: j.model_type.clone(), + status: training_status_str(j.status).to_owned(), + description: j.description.clone(), + final_loss: j.final_loss, + best_val: j.best_validation_score, + created_at: j.created_at, + }) + .collect(); + + let result = TrainListResult { + total: resp.total_count, + jobs, + }; + output::render(&result, format)?; + } + TrainAction::Logs { job_id, follow } => { + if *follow { + // Streaming mode: subscribe to live training status updates + let mut stream = client + .ml_training() + .subscribe_to_training_status( + ml_training::SubscribeToTrainingStatusRequest { + job_id: job_id.clone(), + }, + ) + .await? + .into_inner(); + + println!( + "{} Streaming training updates for {} (Ctrl+C to stop)\n", + "LIVE".cyan().bold(), + job_id.bold(), + ); + + while let Some(msg) = stream.next().await { + let upd = msg?; + let status = colorize_status(training_status_str(upd.status)); + let ts = format_timestamp(upd.timestamp); + + println!( + "[{}] {} epoch {}/{} progress {:.1}% | {}", + ts.dimmed(), + status, + upd.current_epoch, + upd.total_epochs, + upd.progress_percentage, + upd.message, + ); + + if !upd.metrics.is_empty() { + let parts: Vec = upd + .metrics + .iter() + .map(|(k, v)| format!("{k}={v:.4}")) + .collect(); + println!(" metrics: {}", parts.join(", ").dimmed()); + } + } + } else { + // Non-follow mode: show job details with status history + let resp = client + .ml_training() + .get_training_job_details( + ml_training::GetTrainingJobDetailsRequest { + job_id: job_id.clone(), + }, + ) + .await? + .into_inner(); + + let details = resp.job_details.unwrap_or_default(); + + println!("{} {} ({})\n", "Job:".bold(), details.job_id, details.model_type); + + if details.status_history.is_empty() { + println!(" No status updates recorded."); + } else { + for upd in &details.status_history { + let status = colorize_status(training_status_str(upd.status)); + let ts = format_timestamp(upd.timestamp); + println!( + " [{}] {} epoch {}/{} {:.1}% - {}", + ts.dimmed(), + status, + upd.current_epoch, + upd.total_epochs, + upd.progress_percentage, + upd.message, + ); + } + } + + println!( + "\n Tip: use {} to stream live updates.", + "--follow".bold() + ); + } + } + } + Ok(()) } } diff --git a/bin/fxt/src/commands/tune.rs b/bin/fxt/src/commands/tune.rs index 4fbadc2db..c1f9579df 100644 --- a/bin/fxt/src/commands/tune.rs +++ b/bin/fxt/src/commands/tune.rs @@ -2,9 +2,12 @@ use anyhow::Result; use clap::{Parser, Subcommand}; +use colored::Colorize; +use serde::Serialize; use crate::grpc::FoxhuntClient; -use crate::output::OutputFormat; +use crate::output::{self, HumanReadable, OutputFormat}; +use crate::proto::ml_training; #[derive(Parser, Debug)] pub struct TuneCommand { @@ -46,8 +49,299 @@ enum TuneAction { }, } -impl TuneCommand { - pub async fn execute(&self, _client: &FoxhuntClient, _format: OutputFormat) -> Result<()> { - anyhow::bail!("tune command not yet implemented") +// ── Result types ────────────────────────────────────────────────────── + +#[derive(Serialize)] +struct TuneStartResult { + job_id: String, + status: String, + message: String, +} + +#[derive(Serialize)] +struct TuneStopResult { + success: bool, + message: String, + final_status: String, +} + +#[derive(Serialize)] +struct TuneStatusResult { + job_id: String, + status: String, + current_trial: u32, + total_trials: u32, + best_params: Vec<(String, f32)>, + best_metrics: Vec<(String, f32)>, + message: String, + started_at: i64, + updated_at: i64, + trials: Vec, +} + +#[derive(Serialize)] +struct TrialRow { + trial: u32, + objective: f32, + state: String, + started_at: i64, + completed_at: i64, +} + +#[derive(Serialize)] +struct TuneApproveResult { + success: bool, + message: String, +} + +// ── Display helpers ────────────────────────────────────────────────── + +fn tuning_status_str(s: i32) -> &'static str { + match s { + 1 => "pending", + 2 => "running", + 3 => "completed", + 4 => "failed", + 5 => "stopped", + _ => "unknown", + } +} + +fn trial_state_str(s: i32) -> &'static str { + match s { + 1 => "running", + 2 => "complete", + 3 => "pruned", + 4 => "failed", + _ => "unknown", + } +} + +fn colorize_tuning(s: &str) -> String { + match s { + "running" => s.cyan().bold().to_string(), + "completed" | "complete" => s.green().bold().to_string(), + "failed" => s.red().bold().to_string(), + "stopped" => s.yellow().to_string(), + "pending" => s.dimmed().to_string(), + "pruned" => s.yellow().dimmed().to_string(), + _ => s.dimmed().to_string(), + } +} + +fn format_timestamp(ts: i64) -> String { + if ts == 0 { + return "-".to_owned(); + } + chrono::DateTime::from_timestamp(ts, 0) + .map(|dt| dt.format("%Y-%m-%d %H:%M:%S").to_string()) + .unwrap_or_else(|| ts.to_string()) +} + +// ── HumanReadable impls ────────────────────────────────────────────── + +impl HumanReadable for TuneStartResult { + fn print_human(&self) { + println!( + "{} Tuning job {} started (status: {})", + "OK".green().bold(), + self.job_id.bold(), + colorize_tuning(&self.status), + ); + if !self.message.is_empty() { + println!(" {}", self.message.dimmed()); + } + } +} + +impl HumanReadable for TuneStopResult { + fn print_human(&self) { + if self.success { + println!( + "{} {} (final: {})", + "OK".green().bold(), + self.message, + colorize_tuning(&self.final_status), + ); + } else { + println!("{} {}", "FAILED".red().bold(), self.message); + } + } +} + +impl HumanReadable for TuneStatusResult { + fn print_human(&self) { + println!("{} {}", "Job:".bold(), self.job_id); + println!( + " Status: {} (trial {}/{})", + colorize_tuning(&self.status), + self.current_trial, + self.total_trials, + ); + println!(" Started: {}", format_timestamp(self.started_at)); + println!(" Updated: {}", format_timestamp(self.updated_at)); + if !self.message.is_empty() { + println!(" Message: {}", self.message.dimmed()); + } + + if !self.best_params.is_empty() { + println!("\n {}:", "Best Parameters".bold()); + for (k, v) in &self.best_params { + println!(" {k:<30} {v:.6}"); + } + } + + if !self.best_metrics.is_empty() { + println!("\n {}:", "Best Metrics".bold()); + for (k, v) in &self.best_metrics { + println!(" {k:<30} {v:.6}"); + } + } + + if !self.trials.is_empty() { + println!( + "\n {}:\n {:<8} {:>12} {:<12} {}", + "Trial History".bold(), + "TRIAL".bold(), + "OBJECTIVE".bold(), + "STATE".bold(), + "COMPLETED".bold(), + ); + println!(" {}", "-".repeat(56)); + + for t in &self.trials { + let obj_str = if t.objective != 0.0 { + format!("{:.6}", t.objective) + } else { + "-".to_owned() + }; + println!( + " {:<8} {:>12} {:<12} {}", + t.trial, + obj_str, + colorize_tuning(&t.state), + format_timestamp(t.completed_at).dimmed(), + ); + } + } + } +} + +impl HumanReadable for TuneApproveResult { + fn print_human(&self) { + if self.success { + println!("{} {}", "OK".green().bold(), self.message); + } else { + println!("{} {}", "FAILED".red().bold(), self.message); + } + } +} + +// ── Execute ────────────────────────────────────────────────────────── + +impl TuneCommand { + pub async fn execute(&self, client: &FoxhuntClient, format: OutputFormat) -> Result<()> { + match &self.action { + TuneAction::Start { + model, + trials, + config, + gpu, + } => { + let resp = client + .ml_training() + .start_tuning_job(ml_training::StartTuningJobRequest { + model_type: model.to_uppercase(), + num_trials: *trials, + config_path: config.clone().unwrap_or_default(), + data_source: None, + use_gpu: *gpu, + description: String::new(), + tags: Default::default(), + }) + .await? + .into_inner(); + + let result = TuneStartResult { + job_id: resp.job_id, + status: tuning_status_str(resp.status).to_owned(), + message: resp.message, + }; + output::render(&result, format)?; + } + TuneAction::Stop { job_id } => { + let resp = client + .ml_training() + .stop_tuning_job(ml_training::StopTuningJobRequest { + job_id: job_id.clone(), + reason: String::new(), + }) + .await? + .into_inner(); + + let result = TuneStopResult { + success: resp.success, + message: resp.message, + final_status: tuning_status_str(resp.final_status).to_owned(), + }; + output::render(&result, format)?; + } + TuneAction::Status { job_id } => { + let resp = client + .ml_training() + .get_tuning_job_status(ml_training::GetTuningJobStatusRequest { + job_id: job_id.clone(), + }) + .await? + .into_inner(); + + let best_params: Vec<(String, f32)> = + resp.best_params.into_iter().collect(); + let best_metrics: Vec<(String, f32)> = + resp.best_metrics.into_iter().collect(); + + let trials: Vec = resp + .trial_history + .iter() + .map(|t| TrialRow { + trial: t.trial_number, + objective: t.objective_value, + state: trial_state_str(t.state).to_owned(), + started_at: t.started_at, + completed_at: t.completed_at, + }) + .collect(); + + let result = TuneStatusResult { + job_id: resp.job_id, + status: tuning_status_str(resp.status).to_owned(), + current_trial: resp.current_trial, + total_trials: resp.total_trials, + best_params, + best_metrics, + message: resp.message, + started_at: resp.started_at, + updated_at: resp.updated_at, + trials, + }; + output::render(&result, format)?; + } + TuneAction::Approve { job_id } => { + let resp = client + .ml_training() + .approve_promotion(ml_training::ApprovePromotionRequest { + model_id: job_id.clone(), + }) + .await? + .into_inner(); + + let result = TuneApproveResult { + success: resp.success, + message: resp.message, + }; + output::render(&result, format)?; + } + } + Ok(()) } } diff --git a/crates/training_uploader/build.rs b/crates/training_uploader/build.rs index 430a038c8..a5d6ca3b2 100644 --- a/crates/training_uploader/build.rs +++ b/crates/training_uploader/build.rs @@ -6,11 +6,11 @@ fn main() -> Result<(), Box> { .extern_path(".google.protobuf", "::prost_types") .client_mod_attribute(".", "#[allow(unused_qualifications)]") .compile_protos( - &["../../services/ml_training_service/proto/ml_training.proto"], - &["../../services/ml_training_service/proto"], + &["../../proto/ml_training.proto"], + &["../../proto"], )?; - println!("cargo:rerun-if-changed=../../services/ml_training_service/proto/ml_training.proto"); + println!("cargo:rerun-if-changed=../../proto/ml_training.proto"); Ok(()) } diff --git a/crates/web-gateway/build.rs b/crates/web-gateway/build.rs index 29ae527db..3fa3371ce 100644 --- a/crates/web-gateway/build.rs +++ b/crates/web-gateway/build.rs @@ -7,24 +7,26 @@ fn main() -> Result<(), Box> { .protoc_arg("--experimental_allow_proto3_optional") .compile_protos( &[ - "../../bin/fxt/proto/trading.proto", - "../../bin/fxt/proto/health.proto", - "../../bin/fxt/proto/ml.proto", - "../../bin/fxt/proto/config.proto", - "../../bin/fxt/proto/ml_training.proto", - "../../bin/fxt/proto/trading_agent.proto", - "../../bin/fxt/proto/monitoring.proto", + "../../proto/fxt_trading.proto", + "../../proto/trading.proto", + "../../proto/health.proto", + "../../proto/ml.proto", + "../../proto/config.proto", + "../../proto/ml_training.proto", + "../../proto/trading_agent.proto", + "../../proto/monitoring.proto", ], - &["../../bin/fxt/proto"], + &["../../proto"], )?; - println!("cargo:rerun-if-changed=../../bin/fxt/proto/trading.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/health.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/ml.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/config.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/ml_training.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/trading_agent.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/monitoring.proto"); + println!("cargo:rerun-if-changed=../../proto/fxt_trading.proto"); + println!("cargo:rerun-if-changed=../../proto/trading.proto"); + println!("cargo:rerun-if-changed=../../proto/health.proto"); + println!("cargo:rerun-if-changed=../../proto/ml.proto"); + println!("cargo:rerun-if-changed=../../proto/config.proto"); + println!("cargo:rerun-if-changed=../../proto/ml_training.proto"); + println!("cargo:rerun-if-changed=../../proto/trading_agent.proto"); + println!("cargo:rerun-if-changed=../../proto/monitoring.proto"); Ok(()) } diff --git a/crates/web-gateway/src/config.rs b/crates/web-gateway/src/config.rs index 67a4f8822..18bf66c1e 100644 --- a/crates/web-gateway/src/config.rs +++ b/crates/web-gateway/src/config.rs @@ -10,7 +10,7 @@ pub struct GatewayConfig { pub backtesting_service_url: String, /// gRPC ML training service endpoint pub ml_training_service_url: String, - /// gRPC monitoring service endpoint + /// gRPC monitoring endpoint (now served by API Gateway) pub monitoring_service_url: String, /// CORS allowed origins (comma-separated) pub cors_origins: Vec, @@ -25,7 +25,7 @@ impl Default for GatewayConfig { trading_service_url: "https://localhost:50051".to_owned(), backtesting_service_url: "https://localhost:50052".to_owned(), ml_training_service_url: "https://localhost:50053".to_owned(), - monitoring_service_url: "https://localhost:50057".to_owned(), + monitoring_service_url: "https://localhost:50051".to_owned(), cors_origins: vec!["http://localhost:5173".to_owned()], jwt_secret: String::new(), } @@ -44,7 +44,7 @@ impl GatewayConfig { ml_training_service_url: std::env::var("ML_TRAINING_SERVICE_URL") .unwrap_or_else(|_| "https://localhost:50053".to_owned()), monitoring_service_url: std::env::var("MONITORING_SERVICE_URL") - .unwrap_or_else(|_| "https://localhost:50057".to_owned()), + .unwrap_or_else(|_| "https://localhost:50051".to_owned()), cors_origins: std::env::var("CORS_ORIGINS") .unwrap_or_else(|_| "http://localhost:5173".to_owned()) .split(',') @@ -86,7 +86,7 @@ mod tests { #[test] fn test_default_monitoring_service_url() { let cfg = GatewayConfig::default(); - assert_eq!(cfg.monitoring_service_url, "https://localhost:50057"); + assert_eq!(cfg.monitoring_service_url, "https://localhost:50051"); } #[test] diff --git a/crates/web-gateway/src/main.rs b/crates/web-gateway/src/main.rs index 529523c82..20ba4de39 100644 --- a/crates/web-gateway/src/main.rs +++ b/crates/web-gateway/src/main.rs @@ -44,7 +44,7 @@ async fn main() -> Result<()> { info!(" Trading service: {}", config.trading_service_url); info!(" Backtesting service: {}", config.backtesting_service_url); info!(" ML Training service: {}", config.ml_training_service_url); - info!(" Monitoring service: {}", config.monitoring_service_url); + info!(" Monitoring (via API Gateway): {}", config.monitoring_service_url); info!(" CORS origins: {:?}", config.cors_origins); let state = AppState::new(config.clone()).await?; diff --git a/crates/web-gateway/src/routes/training.rs b/crates/web-gateway/src/routes/training.rs index 0fa33188f..5dbc60498 100644 --- a/crates/web-gateway/src/routes/training.rs +++ b/crates/web-gateway/src/routes/training.rs @@ -93,6 +93,9 @@ async fn start_job( use_gpu: body.use_gpu, description: body.description, tags: Default::default(), + mode: 0, // FULL training + resume_checkpoint_path: String::new(), // no checkpoint + max_epochs: 0, // use service default })) .await?; Ok(Json( diff --git a/services/backtesting_service/build.rs b/services/backtesting_service/build.rs index 06897a667..f52dbfa18 100644 --- a/services/backtesting_service/build.rs +++ b/services/backtesting_service/build.rs @@ -6,6 +6,6 @@ fn main() -> Result<(), Box> { // Suppress warnings in generated code .server_mod_attribute(".", "#[allow(unused_qualifications, missing_docs)]") .client_mod_attribute(".", "#[allow(unused_qualifications, missing_docs)]") - .compile_protos(&["../../bin/fxt/proto/trading.proto"], &["../../bin/fxt/proto"])?; + .compile_protos(&["../../proto/fxt_trading.proto"], &["../../proto"])?; Ok(()) } diff --git a/testing/e2e/build.rs b/testing/e2e/build.rs index 245e56cdf..bc61db340 100644 --- a/testing/e2e/build.rs +++ b/testing/e2e/build.rs @@ -2,9 +2,6 @@ use std::io::Result; fn main() -> Result<()> { // Build gRPC service definitions for E2E test clients (Tonic 0.14+) - // - // NOTE: Building TLI trading.proto separately to avoid naming conflicts - // with trading_service trading.proto (both define trading.proto but with different packages) tonic_prost_build::configure() .build_server(false) // We only need clients for E2E tests .build_client(true) @@ -14,15 +11,12 @@ fn main() -> Result<()> { .client_mod_attribute(".", "#[allow(unused_qualifications)]") .compile_protos( &[ - "../../services/trading_service/proto/trading.proto", - "../../services/trading_service/proto/config.proto", - "../../services/trading_service/proto/risk.proto", - "../../services/ml_training_service/proto/ml_training.proto", - ], - &[ - "../../services/trading_service/proto", - "../../services/ml_training_service/proto", + "../../proto/trading.proto", + "../../proto/config.proto", + "../../proto/risk.proto", + "../../proto/ml_training.proto", ], + &["../../proto"], )?; // Build TLI protos separately (backtesting service uses TLI proto) @@ -32,8 +26,8 @@ fn main() -> Result<()> { .out_dir("src/proto") .server_mod_attribute(".", "#[allow(unused_qualifications)]") .client_mod_attribute(".", "#[allow(unused_qualifications)]") - .compile_protos(&["../../bin/fxt/proto/trading.proto"], &["../../bin/fxt/proto"])?; + .compile_protos(&["../../proto/fxt_trading.proto"], &["../../proto"])?; - println!("cargo:rerun-if-changed=../../services/"); + println!("cargo:rerun-if-changed=../../proto/"); Ok(()) } diff --git a/testing/load/build.rs b/testing/load/build.rs index 71c96f0ce..c484e0f3f 100644 --- a/testing/load/build.rs +++ b/testing/load/build.rs @@ -1,6 +1,6 @@ fn main() -> Result<(), Box> { - // Find proto files in the services directory - let proto_file = "../../services/trading_service/proto/trading.proto"; + // Find proto files in the consolidated proto directory + let proto_file = "../../proto/trading.proto"; tonic_prost_build::compile_protos(proto_file)?; diff --git a/testing/service-integration/build.rs b/testing/service-integration/build.rs index b71849a6b..a060a4d86 100644 --- a/testing/service-integration/build.rs +++ b/testing/service-integration/build.rs @@ -1,23 +1,23 @@ fn main() -> Result<(), Box> { - // Compile TLI proto files for client testing (Tonic 0.14+) + // Compile proto files for client testing (Tonic 0.14+) tonic_prost_build::configure() .build_server(false) .build_client(true) .compile_protos( &[ - "../../bin/fxt/proto/trading.proto", - "../../bin/fxt/proto/ml.proto", - "../../bin/fxt/proto/config.proto", - "../../bin/fxt/proto/health.proto", + "../../proto/trading.proto", + "../../proto/ml.proto", + "../../proto/config.proto", + "../../proto/health.proto", ], - &["../../bin/fxt/proto"], + &["../../proto"], )?; // Tell cargo to recompile if proto files change - println!("cargo:rerun-if-changed=../../bin/fxt/proto/trading.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/ml.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/config.proto"); - println!("cargo:rerun-if-changed=../../bin/fxt/proto/health.proto"); + println!("cargo:rerun-if-changed=../../proto/trading.proto"); + println!("cargo:rerun-if-changed=../../proto/ml.proto"); + println!("cargo:rerun-if-changed=../../proto/config.proto"); + println!("cargo:rerun-if-changed=../../proto/health.proto"); Ok(()) } diff --git a/testing/service-load/build.rs b/testing/service-load/build.rs index 7db4ac1c8..aac52c0de 100644 --- a/testing/service-load/build.rs +++ b/testing/service-load/build.rs @@ -1,7 +1,7 @@ fn main() -> Result<(), Box> { - // Build proto files from trading service - tonic_prost_build::compile_protos("../../services/trading_service/proto/trading.proto")?; + // Build proto files from consolidated proto directory + tonic_prost_build::compile_protos("../../proto/trading.proto")?; - println!("cargo:rerun-if-changed=../../services/trading_service/proto/trading.proto"); + println!("cargo:rerun-if-changed=../../proto/trading.proto"); Ok(()) }