9c3d741a08
Move 17 library crates into crates/, CLI binary into bin/fxt, consolidate 10 test crates into testing/, split config crate from deployment config files. Root directory reduced from 38+ to ~17 directories. All Cargo.toml paths and build.rs proto refs updated. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
1178 lines
44 KiB
Rust
1178 lines
44 KiB
Rust
//! Interactive Brokers TWS/Gateway Integration
|
|
//!
|
|
//! Production client backed by the `ibapi` crate (TWS API v10.19+).
|
|
//! Enabled via the `interactive-brokers` feature flag which pulls in `ibapi`.
|
|
//!
|
|
//! When compiled without `interactive-brokers`, every method returns
|
|
//! `BrokerError::BrokerNotAvailable` so the rest of the codebase can
|
|
//! reference this module unconditionally.
|
|
|
|
use crate::brokers::config::InteractiveBrokersConfig;
|
|
use crate::trading::data_interface::{BrokerConnectionStatus, BrokerError, BrokerInterface};
|
|
use crate::trading_operations::TradingOrder;
|
|
use async_trait::async_trait;
|
|
use common::{Execution as ExecutionReport, OrderStatus, Position};
|
|
use std::collections::HashMap;
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use common::{OrderSide, OrderType};
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use tracing::{error, info, warn};
|
|
|
|
// ── ibapi imports (only when the feature is active) ──────────────────
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use ibapi::Client as IbClient;
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use ibapi::contracts::Contract as IbContract;
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use ibapi::orders::{Action as IbAction, order_builder, PlaceOrder};
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use std::sync::atomic::{AtomicI32, Ordering};
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use std::sync::{Arc, Mutex as StdMutex};
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use std::time::Duration;
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use rust_decimal::prelude::ToPrimitive;
|
|
#[cfg(feature = "interactive-brokers")]
|
|
use uuid::Uuid;
|
|
|
|
// =====================================================================
|
|
// REAL IMPLEMENTATION — interactive-brokers feature enabled
|
|
// =====================================================================
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
/// Interactive Brokers client backed by the `ibapi` crate.
|
|
///
|
|
/// All ibapi calls are synchronous (blocking I/O), so every operation is
|
|
/// dispatched onto `tokio::task::spawn_blocking` to avoid starving the
|
|
/// async runtime. The underlying `IbClient` is stored inside an
|
|
/// `Arc<StdMutex<Option<IbClient>>>` because ibapi is not `Send`-safe
|
|
/// through a tokio Mutex.
|
|
pub struct InteractiveBrokersClient {
|
|
config: InteractiveBrokersConfig,
|
|
/// The ibapi `Client` handle, wrapped for thread-safe shared access.
|
|
/// `None` when disconnected, `Some(client)` when connected.
|
|
client: Arc<StdMutex<Option<IbClient>>>,
|
|
/// Monotonically increasing order-id counter, seeded from TWS on connect.
|
|
next_order_id: Arc<AtomicI32>,
|
|
}
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
impl std::fmt::Debug for InteractiveBrokersClient {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("InteractiveBrokersClient")
|
|
.field("host", &self.config.host)
|
|
.field("port", &self.config.port)
|
|
.field("client_id", &self.config.client_id)
|
|
.field("account_id", &self.config.account_id)
|
|
.field(
|
|
"connected",
|
|
&self
|
|
.client
|
|
.lock()
|
|
.map(|g| g.is_some())
|
|
.unwrap_or(false),
|
|
)
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
impl InteractiveBrokersClient {
|
|
/// Creates a new Interactive Brokers TWS client in disconnected state.
|
|
pub fn new(config: InteractiveBrokersConfig) -> Self {
|
|
Self {
|
|
config,
|
|
client: Arc::new(StdMutex::new(None)),
|
|
next_order_id: Arc::new(AtomicI32::new(0)),
|
|
}
|
|
}
|
|
|
|
/// Allocate the next order id (atomic increment).
|
|
fn allocate_order_id(&self) -> i32 {
|
|
self.next_order_id.fetch_add(1, Ordering::SeqCst)
|
|
}
|
|
}
|
|
|
|
// ── Type conversion helpers ──────────────────────────────────────────
|
|
|
|
/// Classify a symbol string into an ibapi `Contract`.
|
|
///
|
|
/// Convention:
|
|
/// - `"ES.FUT"`, `"NQ.FUT"`, `"6E.FUT"` etc. -> futures contract
|
|
/// - Anything else -> US equity (stock)
|
|
#[cfg(feature = "interactive-brokers")]
|
|
fn to_ib_contract(order: &TradingOrder) -> IbContract {
|
|
if order.symbol.ends_with(".FUT") {
|
|
let base = order
|
|
.symbol
|
|
.strip_suffix(".FUT")
|
|
.unwrap_or(&order.symbol);
|
|
let mut contract = IbContract::futures(base);
|
|
// Allow override through metadata
|
|
if let Some(exchange) = order.metadata.get("exchange") {
|
|
contract.exchange = exchange.clone();
|
|
}
|
|
if let Some(currency) = order.metadata.get("currency") {
|
|
contract.currency = currency.clone();
|
|
}
|
|
if let Some(expiry) = order.metadata.get("expiry") {
|
|
contract.last_trade_date_or_contract_month = expiry.clone();
|
|
}
|
|
contract
|
|
} else {
|
|
let mut contract = IbContract::stock(&order.symbol);
|
|
if let Some(exchange) = order.metadata.get("exchange") {
|
|
contract.exchange = exchange.clone();
|
|
}
|
|
if let Some(currency) = order.metadata.get("currency") {
|
|
contract.currency = currency.clone();
|
|
}
|
|
if let Some(primary_exchange) = order.metadata.get("primary_exchange") {
|
|
contract.primary_exchange = primary_exchange.clone();
|
|
}
|
|
contract
|
|
}
|
|
}
|
|
|
|
/// Convert a `TradingOrder` to an ibapi `Order`.
|
|
#[cfg(feature = "interactive-brokers")]
|
|
fn to_ib_order(order: &TradingOrder, account: &Option<String>) -> ibapi::orders::Order {
|
|
let action = match order.side {
|
|
OrderSide::Buy => IbAction::Buy,
|
|
OrderSide::Sell => IbAction::Sell,
|
|
};
|
|
|
|
let qty = order.quantity.to_f64().unwrap_or(0.0);
|
|
let price = order.price.to_f64().unwrap_or(0.0);
|
|
|
|
let mut ib_order = match order.order_type {
|
|
OrderType::Market => order_builder::market_order(action, qty),
|
|
OrderType::Limit => order_builder::limit_order(action, qty, price),
|
|
OrderType::Stop => order_builder::stop(action, qty, price),
|
|
OrderType::StopLimit => {
|
|
// ibapi stop_limit(action, qty, stop_price, limit_price)
|
|
let stop_price = order
|
|
.metadata
|
|
.get("stop_price")
|
|
.and_then(|s| s.parse::<f64>().ok())
|
|
.unwrap_or(price);
|
|
order_builder::stop_limit(action, qty, stop_price, price)
|
|
}
|
|
// Unsupported order types fall back to market
|
|
_ => order_builder::market_order(action, qty),
|
|
};
|
|
|
|
// Time-in-force
|
|
ib_order.tif = match order.time_in_force {
|
|
common::TimeInForce::Day => "DAY".to_owned(),
|
|
common::TimeInForce::GoodTillCancel => "GTC".to_owned(),
|
|
common::TimeInForce::ImmediateOrCancel => "IOC".to_owned(),
|
|
common::TimeInForce::FillOrKill => "FOK".to_owned(),
|
|
};
|
|
|
|
// Account
|
|
if let Some(acct) = account {
|
|
ib_order.account = acct.clone();
|
|
}
|
|
|
|
ib_order
|
|
}
|
|
|
|
/// Map ibapi OrderStatus.status string to our canonical `OrderStatus`.
|
|
#[cfg(feature = "interactive-brokers")]
|
|
fn map_ib_order_status(status: &str) -> OrderStatus {
|
|
match status {
|
|
"Submitted" | "PreSubmitted" => OrderStatus::Submitted,
|
|
"Filled" => OrderStatus::Filled,
|
|
"Cancelled" | "ApiCancelled" => OrderStatus::Cancelled,
|
|
"Inactive" => OrderStatus::Rejected,
|
|
"PendingSubmit" | "PendingCancel" => OrderStatus::Pending,
|
|
_ => OrderStatus::New,
|
|
}
|
|
}
|
|
|
|
// ── BrokerInterface: real ibapi implementation ───────────────────────
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
#[async_trait]
|
|
impl BrokerInterface for InteractiveBrokersClient {
|
|
async fn connect(&mut self) -> Result<(), BrokerError> {
|
|
let addr = format!("{}:{}", self.config.host, self.config.port);
|
|
let client_id = self.config.client_id;
|
|
let client_arc = Arc::clone(&self.client);
|
|
let next_oid = Arc::clone(&self.next_order_id);
|
|
|
|
info!(
|
|
host = %self.config.host,
|
|
port = self.config.port,
|
|
client_id = client_id,
|
|
account_id = ?self.config.account_id,
|
|
"connecting to Interactive Brokers TWS/Gateway"
|
|
);
|
|
|
|
// ibapi::Client::connect is blocking — run on the blocking pool.
|
|
let result = tokio::task::spawn_blocking(move || {
|
|
IbClient::connect(&addr, client_id)
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?;
|
|
|
|
match result {
|
|
Ok(ib_client) => {
|
|
// Seed the order-id counter from TWS
|
|
let seed_id = ib_client.next_order_id();
|
|
next_oid.store(seed_id, Ordering::SeqCst);
|
|
info!(
|
|
server_version = ib_client.server_version(),
|
|
next_order_id = seed_id,
|
|
"Interactive Brokers connection established"
|
|
);
|
|
|
|
let mut guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
*guard = Some(ib_client);
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
error!(error = %e, "Interactive Brokers connection failed");
|
|
Err(BrokerError::ConnectionFailed(format!(
|
|
"TWS connect to {}:{} failed: {e}",
|
|
self.config.host, self.config.port
|
|
)))
|
|
}
|
|
}
|
|
}
|
|
|
|
async fn disconnect(&mut self) -> Result<(), BrokerError> {
|
|
info!("disconnecting from Interactive Brokers");
|
|
let mut guard = self
|
|
.client
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
// Drop the IbClient which triggers its Drop impl (disconnect).
|
|
*guard = None;
|
|
info!("Interactive Brokers disconnected");
|
|
Ok(())
|
|
}
|
|
|
|
fn is_connected(&self) -> bool {
|
|
self.client
|
|
.lock()
|
|
.map(|g| g.is_some())
|
|
.unwrap_or(false)
|
|
}
|
|
|
|
fn connection_status(&self) -> BrokerConnectionStatus {
|
|
if self.is_connected() {
|
|
BrokerConnectionStatus::Connected
|
|
} else {
|
|
BrokerConnectionStatus::Disconnected
|
|
}
|
|
}
|
|
|
|
async fn submit_order(&self, order: &TradingOrder) -> Result<String, BrokerError> {
|
|
let client_arc = Arc::clone(&self.client);
|
|
let account = self.config.account_id.clone();
|
|
let order_id = self.allocate_order_id();
|
|
let contract = to_ib_contract(order);
|
|
let ib_order = to_ib_order(order, &account);
|
|
let symbol = order.symbol.clone();
|
|
let side = order.side;
|
|
let qty = order.quantity;
|
|
|
|
info!(
|
|
order_id,
|
|
symbol = %symbol,
|
|
side = ?side,
|
|
quantity = %qty,
|
|
order_type = ?order.order_type,
|
|
"submitting order to Interactive Brokers"
|
|
);
|
|
|
|
// place_order is blocking — dispatch to blocking pool
|
|
let result = tokio::task::spawn_blocking(move || {
|
|
let guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
let ib = guard
|
|
.as_ref()
|
|
.ok_or_else(|| BrokerError::ConnectionFailed("not connected to TWS".into()))?;
|
|
|
|
// place_order returns a Subscription<PlaceOrder> which streams
|
|
// order updates. We iterate until we get the first OrderStatus
|
|
// or Execution to confirm placement, with a timeout.
|
|
let sub = ib
|
|
.place_order(order_id, &contract, &ib_order)
|
|
.map_err(|e| BrokerError::OrderSubmissionFailed(format!("place_order: {e}")))?;
|
|
|
|
// Wait up to 10 seconds for initial confirmation
|
|
let mut confirmed = false;
|
|
for msg in sub.timeout_iter(Duration::from_secs(10)) {
|
|
match msg {
|
|
PlaceOrder::OrderStatus(status) => {
|
|
info!(
|
|
order_id,
|
|
ib_status = %status.status,
|
|
filled = status.filled,
|
|
remaining = status.remaining,
|
|
"IB order status"
|
|
);
|
|
confirmed = true;
|
|
break;
|
|
}
|
|
PlaceOrder::OpenOrder(_) => {
|
|
confirmed = true;
|
|
break;
|
|
}
|
|
PlaceOrder::ExecutionData(exec) => {
|
|
info!(
|
|
order_id,
|
|
exec_id = %exec.execution.execution_id,
|
|
shares = exec.execution.shares,
|
|
price = exec.execution.price,
|
|
"IB execution received during placement"
|
|
);
|
|
confirmed = true;
|
|
break;
|
|
}
|
|
PlaceOrder::CommissionReport(cr) => {
|
|
info!(
|
|
order_id,
|
|
commission = cr.commission,
|
|
"IB commission report"
|
|
);
|
|
}
|
|
PlaceOrder::Message(notice) => {
|
|
warn!(
|
|
order_id,
|
|
notice = %notice,
|
|
"IB notice during order placement"
|
|
);
|
|
}
|
|
}
|
|
}
|
|
|
|
if !confirmed {
|
|
warn!(
|
|
order_id,
|
|
"No immediate confirmation from TWS within timeout -- order may still be pending"
|
|
);
|
|
}
|
|
|
|
Ok::<String, BrokerError>(order_id.to_string())
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?;
|
|
|
|
match &result {
|
|
Ok(id) => info!(broker_order_id = %id, symbol = %symbol, "order submitted to IB"),
|
|
Err(e) => error!(error = %e, symbol = %symbol, "order submission to IB failed"),
|
|
}
|
|
|
|
result
|
|
}
|
|
|
|
async fn cancel_order(&self, broker_order_id: &str) -> Result<(), BrokerError> {
|
|
let order_id: i32 = broker_order_id
|
|
.parse()
|
|
.map_err(|_| BrokerError::OrderNotFound(format!("invalid IB order ID: {broker_order_id}")))?;
|
|
|
|
let client_arc = Arc::clone(&self.client);
|
|
info!(order_id, "cancelling order at Interactive Brokers");
|
|
|
|
let result = tokio::task::spawn_blocking(move || {
|
|
let guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
let ib = guard
|
|
.as_ref()
|
|
.ok_or_else(|| BrokerError::ConnectionFailed("not connected to TWS".into()))?;
|
|
|
|
// cancel_order requires a manual_order_cancel_time; empty string means "now"
|
|
let sub = ib
|
|
.cancel_order(order_id, "")
|
|
.map_err(|e| BrokerError::OrderSubmissionFailed(format!("cancel_order: {e}")))?;
|
|
|
|
// Drain the subscription for confirmation
|
|
for msg in sub.timeout_iter(Duration::from_secs(5)) {
|
|
match msg {
|
|
ibapi::orders::CancelOrder::OrderStatus(status) => {
|
|
info!(
|
|
order_id,
|
|
ib_status = %status.status,
|
|
"IB cancel status"
|
|
);
|
|
}
|
|
ibapi::orders::CancelOrder::Notice(notice) => {
|
|
warn!(order_id, notice = %notice, "IB notice during cancel");
|
|
}
|
|
}
|
|
}
|
|
|
|
Ok::<(), BrokerError>(())
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?;
|
|
|
|
result?;
|
|
info!(broker_order_id, "order cancelled at IB");
|
|
Ok(())
|
|
}
|
|
|
|
async fn modify_order(
|
|
&self,
|
|
broker_order_id: &str,
|
|
new_order: &TradingOrder,
|
|
) -> Result<(), BrokerError> {
|
|
let order_id: i32 = broker_order_id
|
|
.parse()
|
|
.map_err(|_| BrokerError::OrderNotFound(format!("invalid IB order ID: {broker_order_id}")))?;
|
|
|
|
let client_arc = Arc::clone(&self.client);
|
|
let account = self.config.account_id.clone();
|
|
let contract = to_ib_contract(new_order);
|
|
let ib_order = to_ib_order(new_order, &account);
|
|
|
|
info!(
|
|
order_id,
|
|
symbol = %new_order.symbol,
|
|
"modifying order at Interactive Brokers"
|
|
);
|
|
|
|
let result = tokio::task::spawn_blocking(move || {
|
|
let guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
let ib = guard
|
|
.as_ref()
|
|
.ok_or_else(|| BrokerError::ConnectionFailed("not connected to TWS".into()))?;
|
|
|
|
// In IB, order modification = re-submit with the same order_id
|
|
let sub = ib
|
|
.place_order(order_id, &contract, &ib_order)
|
|
.map_err(|e| BrokerError::OrderSubmissionFailed(format!("modify (place_order): {e}")))?;
|
|
|
|
// Wait for confirmation
|
|
for msg in sub.timeout_iter(Duration::from_secs(10)) {
|
|
match msg {
|
|
PlaceOrder::OrderStatus(status) => {
|
|
info!(
|
|
order_id,
|
|
ib_status = %status.status,
|
|
"IB modify status"
|
|
);
|
|
break;
|
|
}
|
|
PlaceOrder::OpenOrder(_) => break,
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
Ok::<(), BrokerError>(())
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?;
|
|
|
|
result?;
|
|
info!(broker_order_id, "order modified at IB");
|
|
Ok(())
|
|
}
|
|
|
|
async fn get_order_status(&self, broker_order_id: &str) -> Result<OrderStatus, BrokerError> {
|
|
let _order_id: i32 = broker_order_id
|
|
.parse()
|
|
.map_err(|_| BrokerError::OrderNotFound(format!("invalid IB order ID: {broker_order_id}")))?;
|
|
|
|
let client_arc = Arc::clone(&self.client);
|
|
let target_id = _order_id;
|
|
|
|
tokio::task::spawn_blocking(move || {
|
|
let guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
let ib = guard
|
|
.as_ref()
|
|
.ok_or_else(|| BrokerError::ConnectionFailed("not connected to TWS".into()))?;
|
|
|
|
// Query open orders and find the one matching our order_id
|
|
let sub = ib
|
|
.all_open_orders()
|
|
.map_err(|e| BrokerError::ProtocolError(format!("open_orders: {e}")))?;
|
|
|
|
for msg in sub.timeout_iter(Duration::from_secs(5)) {
|
|
match msg {
|
|
ibapi::orders::Orders::OrderData(data) => {
|
|
if data.order.order_id == target_id {
|
|
return Ok(map_ib_order_status(&data.order_state.status));
|
|
}
|
|
}
|
|
ibapi::orders::Orders::OrderStatus(status) => {
|
|
if status.order_id == target_id {
|
|
return Ok(map_ib_order_status(&status.status));
|
|
}
|
|
}
|
|
_ => {}
|
|
}
|
|
}
|
|
|
|
// Also check completed orders
|
|
let completed = ib
|
|
.completed_orders(true)
|
|
.map_err(|e| BrokerError::ProtocolError(format!("completed_orders: {e}")))?;
|
|
|
|
for msg in completed.timeout_iter(Duration::from_secs(5)) {
|
|
if let ibapi::orders::Orders::OrderData(data) = msg {
|
|
if data.order.order_id == target_id {
|
|
return Ok(map_ib_order_status(&data.order_state.status));
|
|
}
|
|
}
|
|
}
|
|
|
|
// Not found in either open or completed orders
|
|
Err(BrokerError::OrderNotFound(format!(
|
|
"order {target_id} not found in open or completed orders"
|
|
)))
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?
|
|
}
|
|
|
|
async fn get_account_info(&self) -> Result<HashMap<String, String>, BrokerError> {
|
|
let client_arc = Arc::clone(&self.client);
|
|
let account_id = self.config.account_id.clone();
|
|
|
|
tokio::task::spawn_blocking(move || {
|
|
let guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
let ib = guard
|
|
.as_ref()
|
|
.ok_or_else(|| BrokerError::ConnectionFailed("not connected to TWS".into()))?;
|
|
|
|
let tags: &[&str] = &[
|
|
"NetLiquidation",
|
|
"TotalCashValue",
|
|
"GrossPositionValue",
|
|
"BuyingPower",
|
|
"EquityWithLoanValue",
|
|
"MaintMarginReq",
|
|
"AvailableFunds",
|
|
"ExcessLiquidity",
|
|
"Cushion",
|
|
"FullInitMarginReq",
|
|
"FullMaintMarginReq",
|
|
"AccountType",
|
|
];
|
|
|
|
let sub = ib
|
|
.account_summary("All", tags)
|
|
.map_err(|e| BrokerError::ProtocolError(format!("account_summary: {e}")))?;
|
|
|
|
let mut info = HashMap::new();
|
|
info.insert("broker".to_owned(), "Interactive Brokers".to_owned());
|
|
if let Some(ref acct) = account_id {
|
|
info.insert("account_id".to_owned(), acct.clone());
|
|
}
|
|
|
|
for msg in sub.timeout_iter(Duration::from_secs(10)) {
|
|
match msg {
|
|
ibapi::accounts::AccountSummaries::Summary(summary) => {
|
|
// Filter to our account if specified
|
|
if let Some(ref target) = account_id {
|
|
if !summary.account.is_empty() && summary.account != *target {
|
|
continue;
|
|
}
|
|
}
|
|
let key = if summary.currency.is_empty() {
|
|
summary.tag.clone()
|
|
} else {
|
|
format!("{}_{}", summary.tag, summary.currency)
|
|
};
|
|
info.insert(key, summary.value);
|
|
}
|
|
ibapi::accounts::AccountSummaries::End => break,
|
|
}
|
|
}
|
|
|
|
Ok(info)
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?
|
|
}
|
|
|
|
async fn get_positions(&self) -> Result<Vec<Position>, BrokerError> {
|
|
let client_arc = Arc::clone(&self.client);
|
|
|
|
tokio::task::spawn_blocking(move || {
|
|
let guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
let ib = guard
|
|
.as_ref()
|
|
.ok_or_else(|| BrokerError::ConnectionFailed("not connected to TWS".into()))?;
|
|
|
|
let sub = ib
|
|
.positions()
|
|
.map_err(|e| BrokerError::ProtocolError(format!("positions: {e}")))?;
|
|
|
|
let mut positions = Vec::new();
|
|
|
|
for msg in sub.timeout_iter(Duration::from_secs(10)) {
|
|
match msg {
|
|
ibapi::accounts::PositionUpdate::Position(ib_pos) => {
|
|
let symbol = if ib_pos.contract.local_symbol.is_empty() {
|
|
ib_pos.contract.symbol.clone()
|
|
} else {
|
|
ib_pos.contract.local_symbol.clone()
|
|
};
|
|
|
|
let quantity =
|
|
rust_decimal::Decimal::from_f64_retain(ib_pos.position)
|
|
.unwrap_or(rust_decimal::Decimal::ZERO);
|
|
let avg_cost =
|
|
rust_decimal::Decimal::from_f64_retain(ib_pos.average_cost)
|
|
.unwrap_or(rust_decimal::Decimal::ZERO);
|
|
|
|
let pos = Position::new(symbol, quantity, avg_cost);
|
|
positions.push(pos);
|
|
}
|
|
ibapi::accounts::PositionUpdate::PositionEnd => break,
|
|
}
|
|
}
|
|
|
|
info!(count = positions.len(), "retrieved positions from IB");
|
|
Ok(positions)
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?
|
|
}
|
|
|
|
async fn subscribe_executions(
|
|
&self,
|
|
) -> Result<tokio::sync::mpsc::Receiver<ExecutionReport>, BrokerError> {
|
|
let client_arc = Arc::clone(&self.client);
|
|
let (tx, rx) = tokio::sync::mpsc::channel(1000);
|
|
|
|
// Spawn a blocking task that continuously polls the executions subscription
|
|
tokio::task::spawn_blocking(move || {
|
|
let guard = match client_arc.lock() {
|
|
Ok(g) => g,
|
|
Err(e) => {
|
|
error!(error = %e, "mutex poisoned in subscribe_executions");
|
|
return;
|
|
}
|
|
};
|
|
let ib = match guard.as_ref() {
|
|
Some(c) => c,
|
|
None => {
|
|
error!("not connected in subscribe_executions");
|
|
return;
|
|
}
|
|
};
|
|
|
|
let filter = ibapi::orders::ExecutionFilter::default();
|
|
let sub = match ib.executions(filter) {
|
|
Ok(s) => s,
|
|
Err(e) => {
|
|
error!(error = %e, "failed to subscribe to IB executions");
|
|
return;
|
|
}
|
|
};
|
|
|
|
for msg in sub {
|
|
match msg {
|
|
ibapi::orders::Executions::ExecutionData(data) => {
|
|
let exec_side = match data.execution.side.as_str() {
|
|
"BOT" | "BUY" => OrderSide::Buy,
|
|
_ => OrderSide::Sell,
|
|
};
|
|
|
|
let quantity =
|
|
rust_decimal::Decimal::from_f64_retain(data.execution.shares)
|
|
.unwrap_or(rust_decimal::Decimal::ZERO);
|
|
let price =
|
|
rust_decimal::Decimal::from_f64_retain(data.execution.price)
|
|
.unwrap_or(rust_decimal::Decimal::ZERO);
|
|
|
|
let now = chrono::Utc::now();
|
|
let gross = quantity
|
|
.checked_mul(price)
|
|
.unwrap_or(rust_decimal::Decimal::ZERO);
|
|
|
|
let symbol = if data.contract.local_symbol.is_empty() {
|
|
data.contract.symbol.clone()
|
|
} else {
|
|
data.contract.local_symbol.clone()
|
|
};
|
|
|
|
let exec = ExecutionReport {
|
|
id: Uuid::new_v4(),
|
|
order_id: Uuid::nil(), // IB uses i32 order IDs, not UUIDs
|
|
symbol,
|
|
quantity,
|
|
price,
|
|
side: exec_side,
|
|
fees: rust_decimal::Decimal::ZERO, // Fees come via CommissionReport
|
|
fee_currency: "USD".to_owned(),
|
|
executed_at: now,
|
|
timestamp: now,
|
|
symbol_hash: 0,
|
|
broker_execution_id: Some(data.execution.execution_id.clone()),
|
|
counterparty: None,
|
|
venue: Some(data.execution.exchange.clone()),
|
|
gross_value: gross,
|
|
net_value: gross, // Updated when CommissionReport arrives
|
|
};
|
|
|
|
info!(
|
|
exec_id = %data.execution.execution_id,
|
|
symbol = %exec.symbol,
|
|
shares = data.execution.shares,
|
|
price = data.execution.price,
|
|
side = %data.execution.side,
|
|
"IB execution received"
|
|
);
|
|
|
|
if tx.blocking_send(exec).is_err() {
|
|
// Receiver dropped, stop listening
|
|
break;
|
|
}
|
|
}
|
|
ibapi::orders::Executions::CommissionReport(cr) => {
|
|
info!(
|
|
commission = cr.commission,
|
|
currency = %cr.currency,
|
|
"IB commission report received"
|
|
);
|
|
}
|
|
ibapi::orders::Executions::Notice(notice) => {
|
|
warn!(notice = %notice, "IB executions notice");
|
|
}
|
|
}
|
|
}
|
|
});
|
|
|
|
Ok(rx)
|
|
}
|
|
|
|
fn broker_name(&self) -> &str {
|
|
"Interactive Brokers"
|
|
}
|
|
|
|
async fn send_heartbeat(&self) -> Result<(), BrokerError> {
|
|
// ibapi handles heartbeat/keepalive internally via its connection loop.
|
|
// We verify the connection is alive by checking if the client exists.
|
|
let connected = self
|
|
.client
|
|
.lock()
|
|
.map(|g| g.is_some())
|
|
.unwrap_or(false);
|
|
|
|
if connected {
|
|
Ok(())
|
|
} else {
|
|
Err(BrokerError::ConnectionFailed(
|
|
"TWS connection lost".to_owned(),
|
|
))
|
|
}
|
|
}
|
|
|
|
async fn reconnect(&self) -> Result<(), BrokerError> {
|
|
warn!("reconnect requested -- dropping existing connection and re-establishing");
|
|
|
|
// Drop existing client
|
|
{
|
|
let mut guard = self
|
|
.client
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
*guard = None;
|
|
}
|
|
|
|
// Re-connect
|
|
let addr = format!("{}:{}", self.config.host, self.config.port);
|
|
let client_id = self.config.client_id;
|
|
let client_arc = Arc::clone(&self.client);
|
|
let next_oid = Arc::clone(&self.next_order_id);
|
|
|
|
let result = tokio::task::spawn_blocking(move || {
|
|
IbClient::connect(&addr, client_id)
|
|
})
|
|
.await
|
|
.map_err(|e| BrokerError::InternalError(format!("spawn_blocking join error: {e}")))?;
|
|
|
|
match result {
|
|
Ok(ib_client) => {
|
|
let seed_id = ib_client.next_order_id();
|
|
next_oid.store(seed_id, Ordering::SeqCst);
|
|
info!(
|
|
server_version = ib_client.server_version(),
|
|
next_order_id = seed_id,
|
|
"Interactive Brokers reconnected"
|
|
);
|
|
let mut guard = client_arc
|
|
.lock()
|
|
.map_err(|e| BrokerError::InternalError(format!("mutex poisoned: {e}")))?;
|
|
*guard = Some(ib_client);
|
|
Ok(())
|
|
}
|
|
Err(e) => {
|
|
error!(error = %e, "Interactive Brokers reconnection failed");
|
|
Err(BrokerError::ConnectionFailed(format!(
|
|
"TWS reconnect failed: {e}"
|
|
)))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// =====================================================================
|
|
// STUB IMPLEMENTATION — interactive-brokers feature NOT enabled
|
|
// =====================================================================
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
/// Stub Interactive Brokers client when `interactive-brokers` feature is disabled.
|
|
///
|
|
/// Every method returns `BrokerError::BrokerNotAvailable` with a helpful
|
|
/// message explaining how to enable the feature.
|
|
pub struct InteractiveBrokersClient {
|
|
#[allow(dead_code)]
|
|
config: InteractiveBrokersConfig,
|
|
}
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
impl std::fmt::Debug for InteractiveBrokersClient {
|
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
f.debug_struct("InteractiveBrokersClient")
|
|
.field("enabled", &false)
|
|
.field("feature", &"interactive-brokers (disabled)")
|
|
.finish()
|
|
}
|
|
}
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
impl InteractiveBrokersClient {
|
|
/// Creates a new stub client (feature disabled).
|
|
pub const fn new(config: InteractiveBrokersConfig) -> Self {
|
|
Self { config }
|
|
}
|
|
}
|
|
|
|
/// Error message returned when the ibkr feature is not compiled in.
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
const FEATURE_DISABLED_MSG: &str =
|
|
"IBKR feature not enabled — compile with --features interactive-brokers";
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
#[async_trait]
|
|
impl BrokerInterface for InteractiveBrokersClient {
|
|
async fn connect(&mut self) -> Result<(), BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn disconnect(&mut self) -> Result<(), BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
fn is_connected(&self) -> bool {
|
|
false
|
|
}
|
|
|
|
fn connection_status(&self) -> BrokerConnectionStatus {
|
|
BrokerConnectionStatus::Disconnected
|
|
}
|
|
|
|
async fn submit_order(&self, _order: &TradingOrder) -> Result<String, BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn cancel_order(&self, _order_id: &str) -> Result<(), BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn modify_order(
|
|
&self,
|
|
_broker_order_id: &str,
|
|
_new_order: &TradingOrder,
|
|
) -> Result<(), BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn get_order_status(&self, _order_id: &str) -> Result<OrderStatus, BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn get_account_info(&self) -> Result<HashMap<String, String>, BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn get_positions(&self) -> Result<Vec<Position>, BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn subscribe_executions(
|
|
&self,
|
|
) -> Result<tokio::sync::mpsc::Receiver<ExecutionReport>, BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
fn broker_name(&self) -> &str {
|
|
"Interactive Brokers"
|
|
}
|
|
|
|
async fn send_heartbeat(&self) -> Result<(), BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
|
|
async fn reconnect(&self) -> Result<(), BrokerError> {
|
|
Err(BrokerError::BrokerNotAvailable(
|
|
FEATURE_DISABLED_MSG.to_owned(),
|
|
))
|
|
}
|
|
}
|
|
|
|
// ── Tests ────────────────────────────────────────────────────────────
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use crate::brokers::config::InteractiveBrokersConfig;
|
|
|
|
#[test]
|
|
fn new_client_is_disconnected() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
assert!(!client.is_connected());
|
|
assert_eq!(
|
|
client.connection_status(),
|
|
BrokerConnectionStatus::Disconnected
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn broker_name_is_interactive_brokers() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
assert_eq!(client.broker_name(), "Interactive Brokers");
|
|
}
|
|
|
|
#[test]
|
|
fn debug_impl_does_not_panic() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
let debug_str = format!("{:?}", client);
|
|
assert!(!debug_str.is_empty());
|
|
}
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
#[tokio::test]
|
|
async fn stub_submit_returns_not_available() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
let order = TradingOrder {
|
|
id: "test-stub".to_owned().into(),
|
|
symbol: "AAPL".to_owned(),
|
|
side: common::OrderSide::Buy,
|
|
order_type: common::OrderType::Market,
|
|
quantity: rust_decimal::Decimal::from(1),
|
|
price: rust_decimal::Decimal::from(150),
|
|
time_in_force: common::TimeInForce::Day,
|
|
account_id: None,
|
|
metadata: HashMap::new(),
|
|
created_at: chrono::Utc::now(),
|
|
submitted_at: None,
|
|
executed_at: None,
|
|
status: common::OrderStatus::Created,
|
|
fill_quantity: rust_decimal::Decimal::ZERO,
|
|
average_fill_price: None,
|
|
};
|
|
let result = client.submit_order(&order).await;
|
|
assert!(result.is_err());
|
|
let err = result.err().map(|e| e.to_string()).unwrap_or_default();
|
|
assert!(
|
|
err.contains("IBKR feature not enabled"),
|
|
"Expected feature-not-enabled error, got: {err}"
|
|
);
|
|
}
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
#[tokio::test]
|
|
async fn stub_get_positions_returns_not_available() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
let result = client.get_positions().await;
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[cfg(not(feature = "interactive-brokers"))]
|
|
#[tokio::test]
|
|
async fn stub_heartbeat_returns_not_available() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
let result = client.send_heartbeat().await;
|
|
assert!(result.is_err());
|
|
}
|
|
|
|
#[cfg(feature = "interactive-brokers")]
|
|
mod conversion_tests {
|
|
use super::*;
|
|
|
|
fn make_test_order(symbol: &str, side: OrderSide, order_type: OrderType) -> TradingOrder {
|
|
TradingOrder {
|
|
id: "conv-test".to_owned().into(),
|
|
symbol: symbol.to_owned(),
|
|
side,
|
|
order_type,
|
|
quantity: rust_decimal::Decimal::from(10),
|
|
price: rust_decimal::Decimal::from(100),
|
|
time_in_force: common::TimeInForce::Day,
|
|
account_id: None,
|
|
metadata: HashMap::new(),
|
|
created_at: chrono::Utc::now(),
|
|
submitted_at: None,
|
|
executed_at: None,
|
|
status: common::OrderStatus::Created,
|
|
fill_quantity: rust_decimal::Decimal::ZERO,
|
|
average_fill_price: None,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn stock_contract_from_symbol() {
|
|
let order = make_test_order("AAPL", OrderSide::Buy, OrderType::Market);
|
|
let contract = to_ib_contract(&order);
|
|
assert_eq!(contract.symbol, "AAPL");
|
|
assert_eq!(contract.security_type, ibapi::contracts::SecurityType::Stock);
|
|
}
|
|
|
|
#[test]
|
|
fn futures_contract_from_symbol() {
|
|
let order = make_test_order("ES.FUT", OrderSide::Buy, OrderType::Market);
|
|
let contract = to_ib_contract(&order);
|
|
assert_eq!(contract.symbol, "ES");
|
|
assert_eq!(
|
|
contract.security_type,
|
|
ibapi::contracts::SecurityType::Future
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn market_order_conversion() {
|
|
let order = make_test_order("AAPL", OrderSide::Buy, OrderType::Market);
|
|
let ib_order = to_ib_order(&order, &None);
|
|
assert_eq!(ib_order.action, IbAction::Buy);
|
|
assert_eq!(ib_order.order_type, "MKT");
|
|
assert_eq!(ib_order.total_quantity, 10.0);
|
|
}
|
|
|
|
#[test]
|
|
fn limit_order_conversion() {
|
|
let order = make_test_order("AAPL", OrderSide::Sell, OrderType::Limit);
|
|
let ib_order = to_ib_order(&order, &Some("DU123456".to_owned()));
|
|
assert_eq!(ib_order.action, IbAction::Sell);
|
|
assert_eq!(ib_order.order_type, "LMT");
|
|
assert_eq!(ib_order.limit_price, Some(100.0));
|
|
assert_eq!(ib_order.account, "DU123456");
|
|
assert_eq!(ib_order.tif, "DAY");
|
|
}
|
|
|
|
#[test]
|
|
fn stop_order_conversion() {
|
|
let order = make_test_order("MSFT", OrderSide::Sell, OrderType::Stop);
|
|
let ib_order = to_ib_order(&order, &None);
|
|
assert_eq!(ib_order.order_type, "STP");
|
|
assert_eq!(ib_order.aux_price, Some(100.0));
|
|
}
|
|
|
|
#[test]
|
|
fn order_status_mapping() {
|
|
assert_eq!(map_ib_order_status("Submitted"), OrderStatus::Submitted);
|
|
assert_eq!(map_ib_order_status("PreSubmitted"), OrderStatus::Submitted);
|
|
assert_eq!(map_ib_order_status("Filled"), OrderStatus::Filled);
|
|
assert_eq!(map_ib_order_status("Cancelled"), OrderStatus::Cancelled);
|
|
assert_eq!(map_ib_order_status("ApiCancelled"), OrderStatus::Cancelled);
|
|
assert_eq!(map_ib_order_status("Inactive"), OrderStatus::Rejected);
|
|
assert_eq!(map_ib_order_status("PendingSubmit"), OrderStatus::Pending);
|
|
assert_eq!(map_ib_order_status("PendingCancel"), OrderStatus::Pending);
|
|
assert_eq!(map_ib_order_status("SomethingElse"), OrderStatus::New);
|
|
}
|
|
|
|
#[test]
|
|
fn metadata_overrides_contract_fields() {
|
|
let mut order = make_test_order("AAPL", OrderSide::Buy, OrderType::Market);
|
|
order.metadata.insert("exchange".to_owned(), "NYSE".to_owned());
|
|
order.metadata.insert("currency".to_owned(), "EUR".to_owned());
|
|
order
|
|
.metadata
|
|
.insert("primary_exchange".to_owned(), "ISLAND".to_owned());
|
|
let contract = to_ib_contract(&order);
|
|
assert_eq!(contract.exchange, "NYSE");
|
|
assert_eq!(contract.currency, "EUR");
|
|
assert_eq!(contract.primary_exchange, "ISLAND");
|
|
}
|
|
|
|
#[test]
|
|
fn futures_metadata_overrides() {
|
|
let mut order = make_test_order("ES.FUT", OrderSide::Buy, OrderType::Market);
|
|
order
|
|
.metadata
|
|
.insert("exchange".to_owned(), "CME".to_owned());
|
|
order
|
|
.metadata
|
|
.insert("expiry".to_owned(), "202603".to_owned());
|
|
let contract = to_ib_contract(&order);
|
|
assert_eq!(contract.symbol, "ES");
|
|
assert_eq!(contract.exchange, "CME");
|
|
assert_eq!(
|
|
contract.last_trade_date_or_contract_month, "202603"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn tif_mapping() {
|
|
let mut order = make_test_order("AAPL", OrderSide::Buy, OrderType::Market);
|
|
|
|
order.time_in_force = common::TimeInForce::Day;
|
|
assert_eq!(to_ib_order(&order, &None).tif, "DAY");
|
|
|
|
order.time_in_force = common::TimeInForce::GoodTillCancel;
|
|
assert_eq!(to_ib_order(&order, &None).tif, "GTC");
|
|
|
|
order.time_in_force = common::TimeInForce::ImmediateOrCancel;
|
|
assert_eq!(to_ib_order(&order, &None).tif, "IOC");
|
|
|
|
order.time_in_force = common::TimeInForce::FillOrKill;
|
|
assert_eq!(to_ib_order(&order, &None).tif, "FOK");
|
|
}
|
|
|
|
#[test]
|
|
fn order_id_allocation_is_atomic() {
|
|
let client = InteractiveBrokersClient::new(InteractiveBrokersConfig::default());
|
|
client.next_order_id.store(100, Ordering::SeqCst);
|
|
assert_eq!(client.allocate_order_id(), 100);
|
|
assert_eq!(client.allocate_order_id(), 101);
|
|
assert_eq!(client.allocate_order_id(), 102);
|
|
}
|
|
}
|
|
}
|