From 9e42c9264a55e1a52d13d1b891d16e6ae0a43bcb Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 13:33:43 +0100 Subject: [PATCH 1/6] feat(fxt): add ibapi optional dep behind broker-check feature Co-Authored-By: Claude Opus 4.6 --- bin/fxt/Cargo.toml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/bin/fxt/Cargo.toml b/bin/fxt/Cargo.toml index 3a6d05478..96edbfcd3 100644 --- a/bin/fxt/Cargo.toml +++ b/bin/fxt/Cargo.toml @@ -43,6 +43,9 @@ tracing-subscriber.workspace = true # Common types for communication common.workspace = true + +# Broker connectivity check (optional, for direct IB Gateway validation) +ibapi = { workspace = true, optional = true } # REMOVED trading_engine dependency - violates pure client architecture # Time and financial types @@ -98,8 +101,11 @@ base64 = "0.22" # Base64 encoding for encrypted token storage # Logging setup will be added when needed [features] +default = ["broker-check"] # Test utilities feature for integration tests test-utils = [] +# Direct IBKR connectivity check via ibapi crate +broker-check = ["dep:ibapi"] [build-dependencies] # Build dependencies - USE WORKSPACE From ec23919c97c8ae77a6fca79a0acd5e14cad49881 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 13:36:35 +0100 Subject: [PATCH 2/6] feat(fxt): add BrokerConfig section to TliConfig Adds IbkrConfig (host/port/client_id with defaults for paper trading) and BrokerConfig (gateway_url + ibkr nested) to the TLI config file. Both structs derive serde defaults so existing config files remain backward-compatible. Co-Authored-By: Claude Opus 4.6 --- bin/fxt/src/config.rs | 94 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 94 insertions(+) diff --git a/bin/fxt/src/config.rs b/bin/fxt/src/config.rs index 2a562d8ce..a31e7794f 100644 --- a/bin/fxt/src/config.rs +++ b/bin/fxt/src/config.rs @@ -19,6 +19,64 @@ use anyhow::{Context, Result}; use serde::{Deserialize, Serialize}; use std::path::PathBuf; +/// IBKR-specific connection settings +#[derive(Debug, Serialize, Deserialize)] +pub struct IbkrConfig { + /// IB Gateway host + #[serde(default = "default_ibkr_host")] + pub host: String, + /// IB Gateway port (4002 = paper, 4001 = live) + #[serde(default = "default_ibkr_port")] + pub port: u16, + /// TWS client ID (unique per connection) + #[serde(default = "default_ibkr_client_id")] + pub client_id: i32, +} + +fn default_ibkr_host() -> String { + "127.0.0.1".to_owned() +} +fn default_ibkr_port() -> u16 { + 4002 +} +fn default_ibkr_client_id() -> i32 { + 99 +} + +impl Default for IbkrConfig { + fn default() -> Self { + Self { + host: default_ibkr_host(), + port: default_ibkr_port(), + client_id: default_ibkr_client_id(), + } + } +} + +/// Broker connectivity settings +#[derive(Debug, Serialize, Deserialize)] +pub struct BrokerConfig { + /// Broker Gateway gRPC URL + #[serde(default = "default_broker_gateway_url")] + pub gateway_url: String, + /// IBKR-specific settings + #[serde(default)] + pub ibkr: IbkrConfig, +} + +fn default_broker_gateway_url() -> String { + "http://localhost:50060".to_owned() +} + +impl Default for BrokerConfig { + fn default() -> Self { + Self { + gateway_url: default_broker_gateway_url(), + ibkr: IbkrConfig::default(), + } + } +} + /// TLI configuration structure /// /// Contains all configurable settings for the TLI client application. @@ -37,6 +95,10 @@ pub struct TliConfig { /// Token storage backend (default: keyring) #[serde(default = "default_token_storage")] pub token_storage: String, + + /// Broker connectivity settings + #[serde(default)] + pub broker: BrokerConfig, } fn default_api_gateway_url() -> String { @@ -57,6 +119,7 @@ impl Default for TliConfig { api_gateway_url: default_api_gateway_url(), log_level: default_log_level(), token_storage: default_token_storage(), + broker: BrokerConfig::default(), } } } @@ -102,6 +165,7 @@ impl TliConfig { /// api_gateway_url: "http://localhost:50051".to_string(), /// log_level: "debug".to_string(), /// token_storage: "keyring".to_string(), + /// broker: BrokerConfig::default(), /// }; /// config.save().unwrap(); /// ``` @@ -150,6 +214,7 @@ mod tests { api_gateway_url: "http://example.com:50051".to_owned(), log_level: "debug".to_owned(), token_storage: "file".to_owned(), + broker: BrokerConfig::default(), }; let toml_str = toml::to_string(&config).unwrap(); @@ -183,4 +248,33 @@ mod tests { assert_eq!(config.log_level, "info"); assert_eq!(config.token_storage, "keyring"); } + + #[test] + fn test_config_with_broker_section() { + let toml_str = r#" + api_gateway_url = "http://localhost:50051" + + [broker] + gateway_url = "http://broker-gw:50060" + + [broker.ibkr] + host = "10.0.0.5" + port = 4002 + client_id = 42 + "#; + let config: TliConfig = toml::from_str(toml_str).unwrap(); + assert_eq!(config.broker.gateway_url, "http://broker-gw:50060"); + assert_eq!(config.broker.ibkr.host, "10.0.0.5"); + assert_eq!(config.broker.ibkr.port, 4002); + assert_eq!(config.broker.ibkr.client_id, 42); + } + + #[test] + fn test_config_broker_defaults() { + let config: TliConfig = toml::from_str("").unwrap(); + assert_eq!(config.broker.gateway_url, "http://localhost:50060"); + assert_eq!(config.broker.ibkr.host, "127.0.0.1"); + assert_eq!(config.broker.ibkr.port, 4002); + assert_eq!(config.broker.ibkr.client_id, 99); + } } From 9afd985a34a72c759004982513851e66b6a3ae3b Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 13:37:49 +0100 Subject: [PATCH 3/6] feat(fxt): compile broker_gateway.proto for gRPC client stubs Copy broker_gateway.proto from broker_gateway_service into fxt/proto/ and wire it into build.rs compile_protos + lib.rs proto module so the CLI can call BrokerGatewayService RPCs (health check, account state, session status). Co-Authored-By: Claude Opus 4.6 --- bin/fxt/build.rs | 2 + bin/fxt/proto/broker_gateway.proto | 212 +++++++++++++++++++++++++++++ bin/fxt/src/lib.rs | 6 + 3 files changed, 220 insertions(+) create mode 100644 bin/fxt/proto/broker_gateway.proto diff --git a/bin/fxt/build.rs b/bin/fxt/build.rs index 12cb07603..c79fb51af 100644 --- a/bin/fxt/build.rs +++ b/bin/fxt/build.rs @@ -21,6 +21,7 @@ fn main() -> Result<(), Box> { "proto/config.proto", "proto/ml_training.proto", "proto/trading_agent.proto", + "proto/broker_gateway.proto", ], &["proto"], )?; @@ -32,6 +33,7 @@ fn main() -> Result<(), Box> { 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/broker_gateway.proto"); Ok(()) } diff --git a/bin/fxt/proto/broker_gateway.proto b/bin/fxt/proto/broker_gateway.proto new file mode 100644 index 000000000..5677dfa76 --- /dev/null +++ b/bin/fxt/proto/broker_gateway.proto @@ -0,0 +1,212 @@ +// Broker Gateway Service - FIX Order Routing Protocol +// +// This service handles all broker communication via FIX 4.2/4.4 protocol +// for order routing, execution management, and account state synchronization. + +syntax = "proto3"; + +package broker_gateway; + +// ============================================================================ +// Broker Gateway Service +// ============================================================================ + +service BrokerGatewayService { + // Route order to broker via FIX protocol + rpc RouteOrder(RouteOrderRequest) returns (RouteOrderResponse); + + // Cancel existing order + rpc CancelOrder(CancelOrderRequest) returns (CancelOrderResponse); + + // Get current account state (balance, margin, positions) + rpc GetAccountState(GetAccountStateRequest) returns (GetAccountStateResponse); + + // Get all positions for account + rpc GetPositions(GetPositionsRequest) returns (GetPositionsResponse); + + // Get FIX session status + rpc GetSessionStatus(GetSessionStatusRequest) returns (GetSessionStatusResponse); + + // Stream real-time executions from broker + rpc StreamExecutions(StreamExecutionsRequest) returns (stream ExecutionEvent); + + // Health check + rpc HealthCheck(HealthCheckRequest) returns (HealthCheckResponse); +} + +// ============================================================================ +// Order Routing +// ============================================================================ + +message RouteOrderRequest { + string symbol = 1; // ES, NQ, etc. + OrderSide side = 2; // BUY, SELL + double quantity = 3; // Number of contracts + OrderType order_type = 4; // MARKET, LIMIT, STOP, STOP_LIMIT + optional double price = 5; // Limit price (required for LIMIT orders) + optional double stop_price = 6; // Stop price (required for STOP orders) + string account_id = 7; // AMP account identifier + map metadata = 8; // Strategy, model_name, etc. +} + +message RouteOrderResponse { + string broker_order_id = 1; // Broker-assigned OrderID (Tag 37, filled after ack) + string client_order_id = 2; // Our ClOrdID (Tag 11) + OrderStatus status = 3; // PENDING_SUBMIT, SUBMITTED, etc. + int64 submitted_at = 4; // Timestamp (nanoseconds) + string message = 5; // Success/error message +} + +message CancelOrderRequest { + string client_order_id = 1; // Order to cancel + string account_id = 2; // Account verification +} + +message CancelOrderResponse { + bool success = 1; + string message = 2; + OrderStatus new_status = 3; // CANCEL_PENDING, CANCELLED, etc. +} + +// ============================================================================ +// Account & Position Management +// ============================================================================ + +message GetAccountStateRequest { + string account_id = 1; +} + +message GetAccountStateResponse { + string account_id = 1; + double cash_balance = 2; + double equity = 3; + double margin_used = 4; + double margin_available = 5; + double buying_power = 6; + double unrealized_pnl = 7; + double realized_pnl = 8; + int64 last_updated = 9; // Timestamp (nanoseconds) +} + +message GetPositionsRequest { + string account_id = 1; + optional string symbol = 2; // Filter by symbol (optional) +} + +message GetPositionsResponse { + repeated Position positions = 1; + double total_equity = 2; + double total_exposure = 3; + double leverage_ratio = 4; + int64 timestamp = 5; +} + +message Position { + string symbol = 1; + double quantity = 2; // Positive = long, negative = short + double average_price = 3; + double market_value = 4; + double unrealized_pnl = 5; +} + +// ============================================================================ +// Session Management +// ============================================================================ + +message GetSessionStatusRequest { + optional string session_id = 1; // Optional: default to active session +} + +message GetSessionStatusResponse { + string session_id = 1; + SessionState state = 2; + int64 sender_seq_num = 3; // Current outgoing sequence + int64 target_seq_num = 4; // Expected incoming sequence + int64 last_heartbeat_sent = 5; // Timestamp (nanoseconds) + int64 last_heartbeat_received = 6; // Timestamp (nanoseconds) + double heartbeat_rtt_ms = 7; // Round-trip time in milliseconds + int64 connected_at = 8; // Timestamp (nanoseconds) + map details = 9; // Additional session info +} + +// ============================================================================ +// Execution Streaming +// ============================================================================ + +message StreamExecutionsRequest { + optional string account_id = 1; // Filter by account + optional string symbol = 2; // Filter by symbol +} + +message ExecutionEvent { + string execution_id = 1; // ExecID (Tag 17) + string broker_order_id = 2; // OrderID (Tag 37) + string client_order_id = 3; // ClOrdID (Tag 11) + string symbol = 4; + OrderSide side = 5; + ExecutionType exec_type = 6; // NEW, TRADE, CANCELED, REJECTED + OrderStatus order_status = 7; // Order status after this execution + double last_qty = 8; // Quantity filled (Tag 32) + double last_price = 9; // Fill price (Tag 31) + double cum_qty = 10; // Total filled (Tag 14) + double avg_price = 11; // Average fill price (Tag 6) + int64 transact_time = 12; // Execution timestamp + optional string text = 13; // Reject reason (if applicable) +} + +// ============================================================================ +// Health Check +// ============================================================================ + +message HealthCheckRequest {} + +message HealthCheckResponse { + bool healthy = 1; + string message = 2; + map details = 3; +} + +// ============================================================================ +// Enums +// ============================================================================ + +enum OrderSide { + ORDER_SIDE_UNSPECIFIED = 0; + ORDER_SIDE_BUY = 1; + ORDER_SIDE_SELL = 2; +} + +enum OrderType { + ORDER_TYPE_UNSPECIFIED = 0; + ORDER_TYPE_MARKET = 1; + ORDER_TYPE_LIMIT = 2; + ORDER_TYPE_STOP = 3; + ORDER_TYPE_STOP_LIMIT = 4; +} + +enum OrderStatus { + ORDER_STATUS_UNSPECIFIED = 0; + ORDER_STATUS_PENDING_SUBMIT = 1; + ORDER_STATUS_SUBMITTED = 2; + ORDER_STATUS_PARTIALLY_FILLED = 3; + ORDER_STATUS_FILLED = 4; + ORDER_STATUS_CANCEL_PENDING = 5; + ORDER_STATUS_CANCELLED = 6; + ORDER_STATUS_REJECTED = 7; +} + +enum ExecutionType { + EXECUTION_TYPE_UNSPECIFIED = 0; + EXECUTION_TYPE_NEW = 1; // Order accepted + EXECUTION_TYPE_TRADE = 2; // Partial or full fill + EXECUTION_TYPE_CANCELED = 3; // Order canceled + EXECUTION_TYPE_REJECTED = 4; // Order rejected +} + +enum SessionState { + SESSION_STATE_DISCONNECTED = 0; + SESSION_STATE_CONNECTED = 1; + SESSION_STATE_LOGGING_IN = 2; + SESSION_STATE_ACTIVE = 3; + SESSION_STATE_LOGGING_OUT = 4; +} diff --git a/bin/fxt/src/lib.rs b/bin/fxt/src/lib.rs index 5dda8fa65..aa427a1eb 100644 --- a/bin/fxt/src/lib.rs +++ b/bin/fxt/src/lib.rs @@ -241,4 +241,10 @@ pub mod proto { pub mod trading_agent { tonic::include_proto!("trading_agent"); } + + /// Broker Gateway service protobuf definitions + #[allow(clippy::shadow_reuse, clippy::impl_trait_in_params, clippy::empty_structs_with_brackets)] + pub mod broker_gateway { + tonic::include_proto!("broker_gateway"); + } } From 33208c928c066668f474716a894f7409b908435c Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 13:44:20 +0100 Subject: [PATCH 4/6] feat(fxt): add broker check command with config resolution and connectivity checks Co-Authored-By: Claude Opus 4.6 --- bin/fxt/src/commands/broker.rs | 484 +++++++++++++++++++++++++++++++++ bin/fxt/src/commands/mod.rs | 2 + 2 files changed, 486 insertions(+) create mode 100644 bin/fxt/src/commands/broker.rs diff --git a/bin/fxt/src/commands/broker.rs b/bin/fxt/src/commands/broker.rs new file mode 100644 index 000000000..5da1be4b3 --- /dev/null +++ b/bin/fxt/src/commands/broker.rs @@ -0,0 +1,484 @@ +//! Broker connectivity check command +//! +//! Validates IBKR TWS/Gateway connectivity via: +//! 1. Direct ibapi handshake (feature-gated behind `broker-check`) +//! 2. Broker Gateway gRPC health check + +use anyhow::Result; +use clap::{Args, Subcommand}; +use colored::Colorize; +use std::time::Instant; + +use crate::config::TliConfig; + +/// Resolved broker check settings (CLI > env > config > defaults). +#[derive(Debug)] +pub struct ResolvedBrokerConfig { + pub ibkr_host: String, + pub ibkr_port: u16, + pub ibkr_client_id: i32, + pub gateway_url: String, + pub skip_ibkr: bool, + pub skip_gateway: bool, +} + +/// Broker command arguments +#[derive(Args, Debug)] +pub struct BrokerArgs { + #[command(subcommand)] + command: BrokerCommand, +} + +/// Broker subcommands +#[derive(Subcommand, Debug, Clone)] +pub enum BrokerCommand { + /// Check broker connectivity (IB Gateway + Broker Gateway gRPC) + #[clap( + long_about = "Validate IBKR TWS/Gateway connectivity.\n\n\ + Runs two checks:\n\ + 1. Direct IB Gateway: TCP connect + ibapi handshake\n\ + 2. Broker Gateway gRPC: HealthCheck + GetSessionStatus RPCs\n\n\ + Config precedence: CLI flags > env vars > ~/.foxhunt/config.toml > defaults\n\n\ + Examples:\n\ + fxt broker check\n\ + fxt broker check --host 10.0.0.5 --port 4002\n\ + fxt broker check --skip-ibkr\n\ + fxt broker check --skip-gateway" + )] + Check(CheckArgs), +} + +/// Arguments for `fxt broker check` +#[derive(Debug, Args, Clone)] +pub struct CheckArgs { + /// IB Gateway host + #[arg(long, env = "IBKR_HOST")] + pub host: Option, + + /// IB Gateway port (4002 = paper, 4001 = live) + #[arg(long, env = "IBKR_PORT")] + pub port: Option, + + /// TWS client ID (must be unique per connection) + #[arg(long, env = "IBKR_CLIENT_ID")] + pub client_id: Option, + + /// Broker Gateway gRPC URL + #[arg(long, env = "BROKER_GATEWAY_URL")] + pub gateway_url: Option, + + /// Skip direct IB Gateway check + #[arg(long, default_value_t = false)] + pub skip_ibkr: bool, + + /// Skip Broker Gateway gRPC check + #[arg(long, default_value_t = false)] + pub skip_gateway: bool, +} + +/// Resolve config: CLI flags > env vars (handled by clap) > config file > defaults. +pub fn resolve_config(args: &CheckArgs, config: &TliConfig) -> ResolvedBrokerConfig { + ResolvedBrokerConfig { + ibkr_host: args + .host + .clone() + .unwrap_or_else(|| config.broker.ibkr.host.clone()), + ibkr_port: args.port.unwrap_or(config.broker.ibkr.port), + ibkr_client_id: args.client_id.unwrap_or(config.broker.ibkr.client_id), + gateway_url: args + .gateway_url + .clone() + .unwrap_or_else(|| config.broker.gateway_url.clone()), + skip_ibkr: args.skip_ibkr, + skip_gateway: args.skip_gateway, + } +} + +/// Single check result +#[derive(Debug)] +pub struct CheckResult { + pub name: String, + pub passed: bool, + pub detail: String, + pub duration_ms: u128, +} + +/// Print a check result line +fn print_check(result: &CheckResult) { + let status = if result.passed { + "\u{2713}".green().bold() + } else { + "\u{2717}".red().bold() + }; + let dots = ".".repeat(24_usize.saturating_sub(result.name.len())); + println!( + " {} {} {} {}", + result.name, dots, status, result.detail + ); +} + +// ── IB Gateway direct check ────────────────────────────────────────── + +/// TCP connect probe (fail-fast before ibapi handshake). +async fn check_tcp_connect(host: &str, port: u16) -> CheckResult { + let addr = format!("{host}:{port}"); + let start = Instant::now(); + match tokio::time::timeout( + std::time::Duration::from_secs(5), + tokio::net::TcpStream::connect(&addr), + ) + .await + { + Ok(Ok(_)) => CheckResult { + name: "TCP connect".to_owned(), + passed: true, + detail: format!("{addr} ({}ms)", start.elapsed().as_millis()), + duration_ms: start.elapsed().as_millis(), + }, + Ok(Err(e)) => CheckResult { + name: "TCP connect".to_owned(), + passed: false, + detail: format!("{e} ({addr})"), + duration_ms: start.elapsed().as_millis(), + }, + Err(_) => CheckResult { + name: "TCP connect".to_owned(), + passed: false, + detail: format!("Timeout after 5s ({addr})"), + duration_ms: start.elapsed().as_millis(), + }, + } +} + +/// Full ibapi handshake (blocking, runs on spawn_blocking). +#[cfg(feature = "broker-check")] +async fn check_ibapi_handshake(host: &str, port: u16, client_id: i32) -> CheckResult { + let addr = format!("{host}:{port}"); + let start = Instant::now(); + match tokio::task::spawn_blocking(move || ibapi::Client::connect(&addr, client_id)).await { + Ok(Ok(client)) => { + let server_version = client.server_version(); + // client is dropped here which disconnects + CheckResult { + name: "ibapi handshake".to_owned(), + passed: true, + detail: format!( + "client_id={client_id}, server v{server_version} ({}ms)", + start.elapsed().as_millis() + ), + duration_ms: start.elapsed().as_millis(), + } + } + Ok(Err(e)) => CheckResult { + name: "ibapi handshake".to_owned(), + passed: false, + detail: format!("{e}"), + duration_ms: start.elapsed().as_millis(), + }, + Err(e) => CheckResult { + name: "ibapi handshake".to_owned(), + passed: false, + detail: format!("Task join error: {e}"), + duration_ms: start.elapsed().as_millis(), + }, + } +} + +#[cfg(not(feature = "broker-check"))] +async fn check_ibapi_handshake(_host: &str, _port: u16, _client_id: i32) -> CheckResult { + CheckResult { + name: "ibapi handshake".to_owned(), + passed: false, + detail: "Skipped (compile with --features broker-check)".to_owned(), + duration_ms: 0, + } +} + +// ── Broker Gateway gRPC check ──────────────────────────────────────── + +/// gRPC HealthCheck against broker_gateway_service. +async fn check_grpc_health(gateway_url: &str) -> CheckResult { + use crate::proto::broker_gateway::broker_gateway_service_client::BrokerGatewayServiceClient; + use crate::proto::broker_gateway::HealthCheckRequest; + + let start = Instant::now(); + let channel = match tonic::transport::Channel::from_shared(gateway_url.to_owned()) { + Ok(endpoint) => { + match tokio::time::timeout(std::time::Duration::from_secs(5), endpoint.connect()).await + { + Ok(Ok(ch)) => ch, + Ok(Err(e)) => { + return CheckResult { + name: "Health check".to_owned(), + passed: false, + detail: format!("Connection failed: {e}"), + duration_ms: start.elapsed().as_millis(), + }; + } + Err(_) => { + return CheckResult { + name: "Health check".to_owned(), + passed: false, + detail: format!("Timeout after 5s ({gateway_url})"), + duration_ms: start.elapsed().as_millis(), + }; + } + } + } + Err(e) => { + return CheckResult { + name: "Health check".to_owned(), + passed: false, + detail: format!("Invalid URL: {e}"), + duration_ms: start.elapsed().as_millis(), + }; + } + }; + + let mut client = BrokerGatewayServiceClient::new(channel); + match client.health_check(HealthCheckRequest {}).await { + Ok(resp) => { + let inner = resp.into_inner(); + CheckResult { + name: "Health check".to_owned(), + passed: inner.healthy, + detail: format!( + "{} ({}ms)", + inner.message, + start.elapsed().as_millis() + ), + duration_ms: start.elapsed().as_millis(), + } + } + Err(e) => CheckResult { + name: "Health check".to_owned(), + passed: false, + detail: format!("{} ({})", e.code(), e.message()), + duration_ms: start.elapsed().as_millis(), + }, + } +} + +/// gRPC GetSessionStatus against broker_gateway_service. +async fn check_grpc_session(gateway_url: &str) -> CheckResult { + use crate::proto::broker_gateway::broker_gateway_service_client::BrokerGatewayServiceClient; + use crate::proto::broker_gateway::GetSessionStatusRequest; + + let start = Instant::now(); + let channel = match tonic::transport::Channel::from_shared(gateway_url.to_owned()) { + Ok(endpoint) => match endpoint.connect().await { + Ok(ch) => ch, + Err(e) => { + return CheckResult { + name: "Session status".to_owned(), + passed: false, + detail: format!("Connection failed: {e}"), + duration_ms: start.elapsed().as_millis(), + }; + } + }, + Err(e) => { + return CheckResult { + name: "Session status".to_owned(), + passed: false, + detail: format!("Invalid URL: {e}"), + duration_ms: start.elapsed().as_millis(), + }; + } + }; + + let mut client = BrokerGatewayServiceClient::new(channel); + match client + .get_session_status(GetSessionStatusRequest { session_id: None }) + .await + { + Ok(resp) => { + let inner = resp.into_inner(); + let state_name = match inner.state { + 0 => "DISCONNECTED", + 1 => "CONNECTED", + 2 => "LOGGING_IN", + 3 => "ACTIVE", + 4 => "LOGGING_OUT", + _ => "UNKNOWN", + }; + CheckResult { + name: "Session status".to_owned(), + passed: inner.state >= 1, // CONNECTED or better + detail: format!( + "{state_name} (seq: {}/{}, RTT: {:.1}ms)", + inner.sender_seq_num, inner.target_seq_num, inner.heartbeat_rtt_ms + ), + duration_ms: start.elapsed().as_millis(), + } + } + Err(e) => CheckResult { + name: "Session status".to_owned(), + passed: false, + detail: format!("{} ({})", e.code(), e.message()), + duration_ms: start.elapsed().as_millis(), + }, + } +} + +// ── Orchestrator ───────────────────────────────────────────────────── + +/// Run all broker checks and return overall pass/fail. +pub async fn run_broker_check(resolved: &ResolvedBrokerConfig) -> Result { + println!("{}", "Broker Connectivity Check".bold()); + println!("{}", "\u{2550}".repeat(40)); + + let mut all_passed = true; + + // ── IB Gateway (direct) ── + if !resolved.skip_ibkr { + println!("{}", "IB Gateway (direct)".bold()); + let tcp = check_tcp_connect(&resolved.ibkr_host, resolved.ibkr_port).await; + print_check(&tcp); + + if tcp.passed { + let handshake = check_ibapi_handshake( + &resolved.ibkr_host, + resolved.ibkr_port, + resolved.ibkr_client_id, + ) + .await; + print_check(&handshake); + if !handshake.passed { + all_passed = false; + } + } else { + all_passed = false; + } + println!(); + } + + // ── Broker Gateway (gRPC) ── + if !resolved.skip_gateway { + println!("{}", "Broker Gateway (gRPC)".bold()); + let health = check_grpc_health(&resolved.gateway_url).await; + print_check(&health); + + if health.passed { + let session = check_grpc_session(&resolved.gateway_url).await; + print_check(&session); + if !session.passed { + all_passed = false; + } + } else { + all_passed = false; + } + println!(); + } + + // ── Summary ── + if all_passed { + println!("{}", "Result: All checks passed \u{2713}".green().bold()); + } else { + println!("{}", "Result: One or more checks failed \u{2717}".red().bold()); + } + + Ok(all_passed) +} + +/// Execute broker command (public interface for main.rs). +pub async fn execute_broker_command(args: BrokerArgs, config: &TliConfig) -> Result { + match args.command { + BrokerCommand::Check(check_args) => { + let resolved = resolve_config(&check_args, config); + run_broker_check(&resolved).await + } + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + use crate::config::TliConfig; + + #[test] + fn test_resolve_config_uses_defaults() { + let args = CheckArgs { + host: None, + port: None, + client_id: None, + gateway_url: None, + skip_ibkr: false, + skip_gateway: false, + }; + let config = TliConfig::default(); + let resolved = resolve_config(&args, &config); + + assert_eq!(resolved.ibkr_host, "127.0.0.1"); + assert_eq!(resolved.ibkr_port, 4002); + assert_eq!(resolved.ibkr_client_id, 99); + assert_eq!(resolved.gateway_url, "http://localhost:50060"); + assert!(!resolved.skip_ibkr); + assert!(!resolved.skip_gateway); + } + + #[test] + fn test_resolve_config_cli_overrides() { + let args = CheckArgs { + host: Some("10.0.0.5".to_owned()), + port: Some(4001), + client_id: Some(7), + gateway_url: Some("http://custom:9090".to_owned()), + skip_ibkr: true, + skip_gateway: false, + }; + let config = TliConfig::default(); + let resolved = resolve_config(&args, &config); + + assert_eq!(resolved.ibkr_host, "10.0.0.5"); + assert_eq!(resolved.ibkr_port, 4001); + assert_eq!(resolved.ibkr_client_id, 7); + assert_eq!(resolved.gateway_url, "http://custom:9090"); + assert!(resolved.skip_ibkr); + } + + #[test] + fn test_resolve_config_file_overrides_defaults() { + let args = CheckArgs { + host: None, + port: None, + client_id: None, + gateway_url: None, + skip_ibkr: false, + skip_gateway: false, + }; + let config: TliConfig = toml::from_str( + r#" + [broker] + gateway_url = "http://broker-gw:50060" + [broker.ibkr] + host = "192.168.1.100" + port = 4001 + client_id = 55 + "#, + ) + .unwrap(); + let resolved = resolve_config(&args, &config); + + assert_eq!(resolved.ibkr_host, "192.168.1.100"); + assert_eq!(resolved.ibkr_port, 4001); + assert_eq!(resolved.ibkr_client_id, 55); + assert_eq!(resolved.gateway_url, "http://broker-gw:50060"); + } + + #[tokio::test] + async fn test_tcp_connect_refused() { + // Connect to a port that's almost certainly not listening + let result = check_tcp_connect("127.0.0.1", 19999).await; + assert!(!result.passed); + assert_eq!(result.name, "TCP connect"); + } + + #[tokio::test] + async fn test_grpc_health_unreachable() { + let result = check_grpc_health("http://127.0.0.1:19998").await; + assert!(!result.passed); + assert_eq!(result.name, "Health check"); + } +} diff --git a/bin/fxt/src/commands/mod.rs b/bin/fxt/src/commands/mod.rs index e4e5401cf..c9fa8c839 100644 --- a/bin/fxt/src/commands/mod.rs +++ b/bin/fxt/src/commands/mod.rs @@ -18,6 +18,7 @@ pub mod agent; pub mod auth; pub mod backtest_ml; +pub mod broker; pub mod model; pub mod trade; pub mod trade_ml; @@ -28,6 +29,7 @@ pub mod tune; pub use agent::{execute_agent_command, AgentArgs}; pub use auth::{execute_auth_command, AuthCommand}; +pub use broker::{execute_broker_command, BrokerArgs}; pub use backtest_ml::{execute_backtest_ml_command, BacktestMlArgs, BacktestMlCommand}; pub use model::{execute_model_command, ModelCommand}; pub use trade::{execute_trade_command, TradeArgs}; From 7dbecf2b64355a499d059918b52baa5f19db31d9 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 13:49:43 +0100 Subject: [PATCH 5/6] feat(fxt): wire broker check subcommand into CLI Add Broker variant to Commands enum with BrokerArgs (flatten), route to execute_broker_command in match block (no JWT required), exit(1) on check failure. Clone config.api_gateway_url to avoid partial move. Add two CLI parsing tests for broker check. Co-Authored-By: Claude Opus 4.6 --- bin/fxt/src/main.rs | 52 ++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 51 insertions(+), 1 deletion(-) diff --git a/bin/fxt/src/main.rs b/bin/fxt/src/main.rs index 75ef8beab..5897267cd 100644 --- a/bin/fxt/src/main.rs +++ b/bin/fxt/src/main.rs @@ -21,6 +21,7 @@ use fxt::{ agent::{execute_agent_command, AgentArgs}, auth::{execute_auth_command, AuthCommand}, backtest_ml::{execute_backtest_ml_command, BacktestMlArgs}, + broker::{execute_broker_command, BrokerArgs}, model::{execute_model_command, ModelCommand}, trade::{execute_trade_command, TradeArgs}, train::{execute_train_command, TrainCommand}, @@ -212,6 +213,21 @@ enum Commands { trade_args: TradeArgs, }, + /// Broker connectivity check + #[clap( + long_about = "Validate broker connectivity.\n\n\ + Checks:\n\ + 1. Direct IB Gateway: TCP + ibapi handshake\n\ + 2. Broker Gateway gRPC: HealthCheck + SessionStatus\n\n\ + Examples:\n\ + fxt broker check\n\ + fxt broker check --host 10.0.0.5 --port 4002\n\ + fxt broker check --skip-ibkr" + )] + Broker { + #[command(flatten)] + broker_args: BrokerArgs, + }, } /// JWT token claims structure for validation @@ -381,7 +397,7 @@ async fn main() -> Result<()> { // Merge: CLI args override config file (precedence: CLI > Config > Default) if cli.api_gateway_url == "http://localhost:50051" { // Using default, check if config has override - cli.api_gateway_url = config.api_gateway_url; + cli.api_gateway_url = config.api_gateway_url.clone(); } if cli.log_level == "info" { @@ -432,6 +448,13 @@ async fn main() -> Result<()> { let jwt_token = load_jwt_token(&cli.api_gateway_url).await?; execute_trade_command(trade_args, &cli.api_gateway_url, &jwt_token).await } + Commands::Broker { broker_args } => { + let passed = execute_broker_command(broker_args, &config).await?; + if !passed { + std::process::exit(1); + } + Ok(()) + } } } #[cfg(test)] @@ -537,4 +560,31 @@ mod tests { _ => panic!("Expected Auth command"), } } + + #[test] + fn test_cli_parsing_broker_check() { + let cli = Cli::parse_from(&["fxt", "broker", "check"]); + match cli.command { + Commands::Broker { .. } => {}, + _ => panic!("Expected Broker command"), + } + } + + #[test] + fn test_cli_parsing_broker_check_with_flags() { + let cli = Cli::parse_from(&[ + "fxt", + "broker", + "check", + "--host", + "10.0.0.5", + "--port", + "4001", + "--skip-gateway", + ]); + match cli.command { + Commands::Broker { .. } => {}, + _ => panic!("Expected Broker command"), + } + } } From ea130bdb5289b88aca6d8e85baa9f7953d144527 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 2 Mar 2026 13:53:02 +0100 Subject: [PATCH 6/6] chore(fxt): suppress unused_crate_dependencies for ibapi Co-Authored-By: Claude Opus 4.6 --- bin/fxt/src/main.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/bin/fxt/src/main.rs b/bin/fxt/src/main.rs index 5897267cd..3ad363d4a 100644 --- a/bin/fxt/src/main.rs +++ b/bin/fxt/src/main.rs @@ -47,6 +47,8 @@ use dirs as _; use futures_util as _; use getrandom as _; use hex as _; +#[cfg(feature = "broker-check")] +use ibapi as _; use indicatif as _; use keyring as _; use owo_colors as _;