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

Merge branch 'worktree-fxt-broker-check'

Browse Source
This commit is contained in:
jgrusewski
2026-03-02 14:02:50 +01:00
parent eefb191a62 ea130bdb52
commit e0a9dcfb60
8 changed files with 859 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
bin/fxt/Cargo.toml
View File

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

2
bin/fxt/build.rs
View File

@@ -21,6 +21,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
"proto/config.proto",
"proto/ml_training.proto",
"proto/trading_agent.proto",
"proto/broker_gateway.proto",
],
&["proto"],
)?;
@@ -32,6 +33,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
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(())
}

212
bin/fxt/proto/broker_gateway.proto Normal file
View File

@@ -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<string, string> 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<string, string> 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<string, string> 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;
}

484
bin/fxt/src/commands/broker.rs Normal file
View File

@@ -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<String>,
/// IB Gateway port (4002 = paper, 4001 = live)
#[arg(long, env = "IBKR_PORT")]
pub port: Option<u16>,
/// TWS client ID (must be unique per connection)
#[arg(long, env = "IBKR_CLIENT_ID")]
pub client_id: Option<i32>,
/// Broker Gateway gRPC URL
#[arg(long, env = "BROKER_GATEWAY_URL")]
pub gateway_url: Option<String>,
/// 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<bool> {
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<bool> {
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");
}
}

2
bin/fxt/src/commands/mod.rs
View File

@@ -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};

94
bin/fxt/src/config.rs
View File

@@ -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);
}
}

6
bin/fxt/src/lib.rs
View File

@@ -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");
}
}

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

@@ -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},
@@ -46,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 _;
@@ -212,6 +215,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 +399,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 +450,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 +562,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"),
}
}
}