pub mod account;
mod connect;
pub mod contract;
mod ended_orders;
pub mod execution;
pub mod order;
mod recovery;
use crate::{
AccountEventKind, AccountSnapshot, InstrumentAccountSnapshot, Snapshot, UnindexedAccountEvent,
UnindexedAccountSnapshot,
balance::AssetBalance,
client::{
BracketOrderClient, ClientInstrument, ExecutionClient, OrderStatusClient,
dedup::new_dedup_cache,
order_recovery::{KnownLiveOrders, SharedKnownLiveOrders},
},
error::{
ApiError, ConnectivityError, OrderError, StreamTerminationReason, UnindexedClientError,
},
order::{
Order, OrderKey, OrderKind, TimeInForce, UnindexedOrderSnapshot,
bracket::{
BracketOrderRequest as UnifiedBracketOrderRequest,
BracketOrderResult as UnifiedBracketOrderResult,
},
id::{ClientOrderId, OrderId, StrategyId, VenueOrderId},
request::{
OrderRequestCancel, OrderRequestOpen, OrderResponseCancel, UnindexedOrderResponseCancel,
},
state::{Cancelled, Expired, Filled, Open, OrderState, UnindexedOrderState},
},
position::PositionReport,
trade::{Trade, TradesRead},
};
use account::{BalanceAggregator, PositionAggregator};
use chrono::{DateTime, Utc};
pub use connect::{ConnectOutcome, ContractSkipReason, IbkrConnectError, SkippedContract};
use contract::ResolveContractError;
use execution::{ExecutionBuffer, ExecutionRevision, parse_decimal_or_warn, try_decimal_or_warn};
use futures::stream::BoxStream;
use ibapi::{
accounts::{
AccountSummaryResult,
types::{AccountGroup, AccountId},
},
client::blocking::Client,
};
pub use order::{BracketOrderRequest, BracketOrderResult};
use order::{
OrderContext, OrderIdMap, PendingCancels, build_ib_bracket_with_oca, build_ib_order,
side_to_action, time_in_force_to_ib,
};
use parking_lot::Mutex;
use rust_decimal::Decimal;
use rustrade_instrument::{
Side,
asset::name::AssetNameExchange,
exchange::ExchangeId,
ibkr::{ContractRegistry, ContractRegistryError},
instrument::{kind::InstrumentKindDiscriminant, name::InstrumentNameExchange},
};
use serde::{Deserialize, Serialize};
use smol_str::format_smolstr;
use std::{
collections::HashSet,
ops::ControlFlow,
panic::{AssertUnwindSafe, catch_unwind},
sync::Arc,
};
use thiserror::Error;
use tokio::sync::mpsc;
use tracing::{debug, error, info, trace, warn};
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct IbkrConfig {
pub host: String,
pub port: u16,
pub client_id: i32,
pub account: String,
#[serde(default)]
pub contracts: Vec<ContractConfig>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ContractConfig {
pub name: String,
pub symbol: String,
pub security_type: String,
pub exchange: String,
pub currency: String,
#[serde(default)]
pub last_trade_date: Option<String>,
#[serde(default)]
pub strike: Option<f64>,
#[serde(default)]
pub right: Option<String>,
}
impl ContractConfig {
fn to_contract(&self) -> Result<ibapi::contracts::Contract, contract::ContractConfigError> {
use contract::ContractConfigError as E;
Ok(match self.security_type.as_str() {
"STK" => contract::stock_contract(&self.symbol, &self.exchange, &self.currency),
"FUT" => contract::futures_contract(
&self.symbol,
self.last_trade_date
.as_deref()
.ok_or(E::MissingLastTradeDate)?,
&self.exchange,
&self.currency,
),
"OPT" => contract::option_contract(
&self.symbol,
self.last_trade_date
.as_deref()
.ok_or(E::MissingLastTradeDate)?,
self.strike.ok_or(E::MissingStrike)?,
self.right.as_deref().ok_or(E::MissingOptionRight)?,
&self.exchange,
&self.currency,
)?,
"CASH" => contract::forex_contract(&self.symbol, &self.currency),
other => {
return Err(E::UnrecognizedSecurityType {
security_type: other.to_string(),
});
}
})
}
}
fn security_types_of(kind: InstrumentKindDiscriminant) -> &'static [&'static str] {
match kind {
InstrumentKindDiscriminant::Spot => &["STK", "CASH"],
InstrumentKindDiscriminant::Future => &["FUT"],
InstrumentKindDiscriminant::Option => &["OPT"],
InstrumentKindDiscriminant::Perpetual | InstrumentKindDiscriminant::Cfd => &[],
}
}
static ACCOUNT_GROUP_ALL: std::sync::LazyLock<AccountGroup> =
std::sync::LazyLock::new(|| AccountGroup("All".to_string()));
const POSITION_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
const PLACEMENT_STATUS_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const INFORMATIONAL_ORDER_CODES: &[i32] = &[399];
#[must_use]
#[derive(Debug)]
enum PlacementOutcome {
Accepted { filled: f64 },
Rejected(String),
HeldPending(String),
NoStatus,
}
fn is_transient_transport_loss(e: &ibapi::Error) -> bool {
matches!(e, ibapi::Error::ConnectionReset | ibapi::Error::Io(_))
}
fn is_transport_loss(e: &ibapi::Error) -> bool {
matches!(
e,
ibapi::Error::ConnectionReset
| ibapi::Error::ConnectionFailed
| ibapi::Error::Shutdown
| ibapi::Error::Io(_)
)
}
fn resolve_execution(
execution: &ibapi::orders::ExecutionData,
api_client_id: i32,
contracts: &ContractRegistry,
order_ids: &OrderIdMap,
) -> Option<InstrumentNameExchange> {
let order_id = execution.execution.order_id;
let con_id = execution.contract.contract_id;
if execution.execution.client_id != api_client_id {
trace!(
ib_order_id = order_id,
api_client_id = execution.execution.client_id,
"ExecutionData for another API client's order, dropping"
);
return None;
}
if !order_ids.contains(order_id) {
debug!(
ib_order_id = order_id,
con_id, "ExecutionData for unknown order ID, dropping"
);
return None;
}
let Some(instrument) = contracts.get_name_by_con_id(con_id) else {
debug!(
ib_order_id = order_id,
con_id, "ExecutionData for unknown contract ID, dropping"
);
return None;
};
Some(instrument)
}
const LISTING_STALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
fn read_listing<T>(
what: &str,
stall_timeout: std::time::Duration,
connected: impl Fn() -> bool,
next: impl FnMut(
std::time::Duration,
) -> Option<Result<ibapi::subscriptions::SubscriptionItem<T>, ibapi::Error>>,
on_item: impl FnMut(T) -> ControlFlow<()>,
) -> Result<(), UnindexedClientError> {
read_listing_items(what, stall_timeout, connected, next, on_item)
.map_err(|e| UnindexedClientError::Internal(format!("{what}: {e}")))
}
#[derive(Debug, Error)]
enum ListingError {
#[error(transparent)]
Ibapi(ibapi::Error),
#[error("the connection dropped before the end of the listing")]
Dropped,
#[error("IB sent nothing for {0:?} before the end of the listing")]
Stalled(std::time::Duration),
}
fn read_listing_items<T>(
what: &str,
stall_timeout: std::time::Duration,
connected: impl Fn() -> bool,
mut next: impl FnMut(
std::time::Duration,
) -> Option<Result<ibapi::subscriptions::SubscriptionItem<T>, ibapi::Error>>,
mut on_item: impl FnMut(T) -> ControlFlow<()>,
) -> Result<(), ListingError> {
use ibapi::subscriptions::SubscriptionItem;
loop {
let waited = std::time::Instant::now();
match next(stall_timeout) {
Some(Ok(SubscriptionItem::Data(item))) => {
if on_item(item).is_break() {
return Ok(());
}
}
Some(Ok(SubscriptionItem::Notice(notice))) => {
debug!(%notice, what, "Notice during an IBKR listing read");
}
Some(Err(e)) => return Err(ListingError::Ibapi(e)),
None if recovery::returned_early(waited.elapsed(), stall_timeout) => {
if !connected() {
return Err(ListingError::Dropped);
}
return Ok(());
}
None => return Err(ListingError::Stalled(stall_timeout)),
}
}
}
const NO_SECURITY_DEFINITION_CODE: i32 = 200;
fn resolve_contract_blocking(
client: &Client,
contract: &ibapi::contracts::Contract,
) -> Result<ibapi::contracts::Contract, ResolveContractError> {
let subscription = client
.contract_details_stream(contract)
.subscribe()
.map_err(contract_details_error)?;
let mut matches = Vec::new();
let read = read_listing_items(
"contract_details",
LISTING_STALL_TIMEOUT,
|| client.is_connected(),
|timeout| subscription.next_timeout(timeout),
|details: ibapi::contracts::ContractDetails| {
matches.push(details.contract);
ControlFlow::Continue(())
},
);
classify_contract_details(read, matches)
}
fn classify_contract_details(
read: Result<(), ListingError>,
mut matches: Vec<ibapi::contracts::Contract>,
) -> Result<ibapi::contracts::Contract, ResolveContractError> {
match read {
Ok(()) => {}
Err(ListingError::Ibapi(e)) => return Err(contract_details_error(e)),
Err(e @ ListingError::Dropped) => {
return Err(ConnectivityError::Socket(e.to_string()).into());
}
Err(ListingError::Stalled(_)) => return Err(ConnectivityError::Timeout.into()),
}
if matches.len() > 1 {
return Err(ResolveContractError::Ambiguous { matches });
}
matches.pop().ok_or(ResolveContractError::NoMatch)
}
fn contract_details_error(e: ibapi::Error) -> ResolveContractError {
match e {
ibapi::Error::Notice(notice) if notice.code == NO_SECURITY_DEFINITION_CODE => {
ResolveContractError::NoMatch
}
ibapi::Error::Notice(notice) => ResolveContractError::Refused {
code: notice.code,
message: notice.message,
},
e if is_transient_transport_loss(&e) => ConnectivityError::Socket(e.to_string()).into(),
e => ResolveContractError::Failed(e.to_string()),
}
}
#[derive(Debug, Clone, Default)]
struct ListingLock(Arc<Mutex<()>>);
impl ListingLock {
fn open_orders(
&self,
client: &Client,
) -> Result<Vec<ibapi::orders::Orders>, UnindexedClientError> {
let _serial = self.0.lock();
let subscription = client
.all_open_orders()
.map_err(|e| UnindexedClientError::Internal(format!("open_orders: {e}")))?;
let mut listing = Vec::new();
read_listing(
"open_orders",
LISTING_STALL_TIMEOUT,
|| client.is_connected(),
|timeout| subscription.next_timeout(timeout),
|item| {
listing.push(item);
ControlFlow::Continue(())
},
)?;
Ok(listing)
}
fn completed_orders(
&self,
client: &Client,
mut on_item: impl FnMut(ibapi::orders::Orders),
) -> Result<(), UnindexedClientError> {
let _serial = self.0.lock();
let subscription = client
.completed_orders(false)
.map_err(|e| UnindexedClientError::Internal(format!("completed_orders: {e}")))?;
read_listing(
"completed_orders",
LISTING_STALL_TIMEOUT,
|| client.is_connected(),
|timeout| subscription.next_timeout(timeout),
|item| {
on_item(item);
ControlFlow::Continue(())
},
)
}
}
fn read_account_summary(
stall_timeout: std::time::Duration,
connected: impl Fn() -> bool,
next: impl FnMut(
std::time::Duration,
) -> Option<
Result<ibapi::subscriptions::SubscriptionItem<AccountSummaryResult>, ibapi::Error>,
>,
) -> Result<BalanceAggregator, UnindexedClientError> {
let mut aggregator = BalanceAggregator::new();
let mut ended = false;
read_listing(
"account_summary",
stall_timeout,
connected,
next,
|summary| match summary {
AccountSummaryResult::Summary(s) => {
aggregator.process(&s);
ControlFlow::Continue(())
}
AccountSummaryResult::End => {
ended = true;
ControlFlow::Break(())
}
},
)?;
if !ended {
return Err(UnindexedClientError::Internal(
"account_summary: the subscription ended before the end of the listing".to_string(),
));
}
Ok(aggregator)
}
fn read_account_positions(
client: &Client,
account: AccountId,
contracts: &ContractRegistry,
instruments: Option<&HashSet<InstrumentNameExchange>>,
positions: &mut PositionAggregator,
) -> Result<bool, UnindexedClientError> {
use ibapi::accounts::{Position, PositionUpdateMulti};
let subscription = client
.positions_multi(Some(&account), None)
.map_err(|e| UnindexedClientError::Internal(format!("positions of {account}: {e}")))?;
for update in subscription.timeout_iter_data(POSITION_STALL_TIMEOUT) {
let update = update.map_err(|e| {
UnindexedClientError::Internal(format!("positions subscription of {account}: {e}"))
})?;
let position = match update {
PositionUpdateMulti::Position(position) => Position {
account: position.account,
contract: position.contract,
position: position.position,
average_cost: position.average_cost,
},
PositionUpdateMulti::PositionEnd => return Ok(true),
};
let Some(instrument) = contracts.get_name_by_con_id(position.contract.contract_id) else {
continue;
};
if instruments.is_some_and(|filter| !filter.contains(&instrument)) {
continue;
}
positions.process(instrument, position);
}
Ok(false)
}
fn forward_order_updates(
updates: impl IntoIterator<Item = Result<ibapi::orders::OrderUpdate, ibapi::Error>>,
sink: &recovery::EventSink,
api_client_id: i32,
contracts: &ContractRegistry,
order_ids: &OrderIdMap,
pending_cancels: &PendingCancels,
exec_buffer: &ExecutionBuffer,
) {
use ibapi::orders::{OrderStatusKind, OrderUpdate};
for update in updates {
if !sink.is_open() {
return;
}
let update = match update {
Ok(u) => u,
Err(ibapi::Error::Shutdown) => {
warn!("IBKR client shut down; terminating the account stream");
sink.terminate(StreamTerminationReason::Error(
recovery::CLIENT_SHUT_DOWN.to_string(),
));
return;
}
Err(e) => {
error!(error = %e, "Order stream subscription error");
sink.terminate(StreamTerminationReason::Error(e.to_string()));
return;
}
};
let event = match update {
OrderUpdate::OrderStatus(status) if status.client_id != api_client_id => {
trace!(
ib_order_id = status.order_id,
api_client_id = status.client_id,
"OrderStatus for another API client's order, dropping"
);
None
}
OrderUpdate::OrderStatus(status) => {
let ib_id = status.order_id;
let lookup_result = match status.status {
OrderStatusKind::Cancelled | OrderStatusKind::Inactive => {
order_ids.remove_and_get_context(ib_id)
}
OrderStatusKind::Filled => order_ids.release_client_id(ib_id),
_ => order_ids.get_client_id_and_context(ib_id),
};
match lookup_result {
Some((client_id, _)) if order_ids.names_other_order(&client_id, ib_id) => {
debug!(
ib_order_id = ib_id,
cid = %client_id,
"OrderStatus for an ended order whose client order id names a later order, dropping"
);
None
}
Some((client_id, ctx)) => {
let order =
make_order_from_status(&status, client_id, &ctx, pending_cancels);
Some(UnindexedAccountEvent {
exchange: ExchangeId::Ibkr,
kind: AccountEventKind::OrderSnapshot(Snapshot::new(order)),
})
}
None => {
debug!(ib_order_id = ib_id, "OrderStatus for unknown order ID");
None
}
}
}
OrderUpdate::ExecutionData(exec) => {
if recovery::is_replayed_execution(&exec) {
trace!(
exec_id = %exec.execution.execution_id,
request_id = exec.request_id,
"ExecutionData answering an executions request, skipping"
);
continue;
}
let Some(instrument) =
resolve_execution(&exec, api_client_id, contracts, order_ids)
else {
continue;
};
if ExecutionRevision::parse(&exec.execution.execution_id)
.is_some_and(|revision| revision.is_correction())
{
warn!(
exec_id = %exec.execution.execution_id,
%instrument,
price = exec.execution.price,
shares = exec.execution.shares,
"IBKR corrected an execution; it will be reported as TradeAmended if its \
commission report arrives"
);
}
exec_buffer.add_execution(exec, instrument);
None
}
OrderUpdate::CommissionReport(report) => {
if let Some(trade) = exec_buffer.complete_with_commission(&report)
&& !sink.send_execution(trade)
{
return;
}
None
}
_ => None,
};
if let Some(e) = event
&& !sink.send(e)
{
return;
}
}
sink.terminate(StreamTerminationReason::Error(
"IBKR order-update stream ended".to_string(),
));
}
fn panic_message(panic_info: &(dyn std::any::Any + Send)) -> String {
panic_info
.downcast_ref::<&str>()
.map(|s| s.to_string())
.or_else(|| panic_info.downcast_ref::<String>().cloned())
.unwrap_or_else(|| "unknown panic".to_string())
}
fn send_error<AssetKey, InstrumentKey>(
e: &ibapi::Error,
message: String,
) -> OrderError<AssetKey, InstrumentKey> {
if is_transient_transport_loss(e) {
OrderError::Connectivity(ConnectivityError::Socket(message))
} else {
OrderError::Rejected(ApiError::OrderRejected(message))
}
}
fn await_order_placement<I>(events: I) -> PlacementOutcome
where
I: IntoIterator<Item = Result<ibapi::orders::PlaceOrder, ibapi::Error>>,
{
use ibapi::orders::PlaceOrder;
for event in events {
let event = match event {
Ok(event) => event,
Err(ibapi::Error::Notice(n)) if INFORMATIONAL_ORDER_CODES.contains(&n.code) => {
return PlacementOutcome::HeldPending(format!("[{}] {}", n.code, n.message));
}
Err(e) if is_transport_loss(&e) => {
warn!(error = %e, "transport lost while awaiting order placement; status unknown");
return PlacementOutcome::NoStatus;
}
Err(e) => return PlacementOutcome::Rejected(e.to_string()),
};
if let PlaceOrder::OrderStatus(status) = event {
use ibapi::orders::OrderStatusKind;
match status.status {
OrderStatusKind::Submitted
| OrderStatusKind::PreSubmitted
| OrderStatusKind::PendingSubmit
| OrderStatusKind::Filled => {
return PlacementOutcome::Accepted {
filled: status.filled,
};
}
OrderStatusKind::Cancelled | OrderStatusKind::Inactive => {
return PlacementOutcome::Rejected(status.status.to_string());
}
OrderStatusKind::ApiCancelled | OrderStatusKind::PendingCancel => {
return PlacementOutcome::Accepted {
filled: status.filled,
};
}
OrderStatusKind::ApiPending => {}
OrderStatusKind::Unknown(ref raw) => {
warn!(
ib_order_id = status.order_id,
status = raw.as_str(),
"unmodelled IBKR order status during placement; \
treating as live, account stream is authoritative"
);
return PlacementOutcome::HeldPending(format!(
"unmodelled order status: {raw}"
));
}
}
}
}
PlacementOutcome::NoStatus
}
#[derive(Clone)]
pub struct IbkrClient {
config: Arc<IbkrConfig>,
client: Arc<Client>,
contracts: ContractRegistry,
order_ids: OrderIdMap,
pending_cancels: PendingCancels,
execution_buffer: ExecutionBuffer,
next_order_id: Arc<Mutex<i32>>,
known_live: SharedKnownLiveOrders,
listings: ListingLock,
earlier_completions: ended_orders::EarlierCompletions,
}
impl std::fmt::Debug for IbkrClient {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("IbkrClient")
.field("config", &self.config)
.field("contracts_count", &self.contracts.len())
.field("pending_orders", &self.order_ids.len())
.finish()
}
}
impl IbkrClient {
pub fn connect_sync(config: IbkrConfig) -> Result<Self, IbkrConnectError> {
let ConnectOutcome { client, skipped } = Self::connect_inner(config)?;
if skipped.is_empty() {
Ok(client)
} else {
drop(client);
Err(IbkrConnectError::Contracts(skipped))
}
}
pub fn connect_sync_lenient(
config: IbkrConfig,
) -> Result<ConnectOutcome, UnindexedClientError> {
Self::connect_inner(config)
}
fn connect_inner(config: IbkrConfig) -> Result<ConnectOutcome, UnindexedClientError> {
let url = format!("{}:{}", config.host, config.port);
info!(url = %url, client_id = config.client_id, "Connecting to IB");
let client = Client::connect(&url, config.client_id).map_err(|e| {
UnindexedClientError::Connectivity(ConnectivityError::Socket(e.to_string()))
})?;
let next_id = client.next_order_id();
let contracts = ContractRegistry::new();
let skipped = connect::register_configured(&config.contracts, &contracts, |contract| {
resolve_contract_blocking(&client, contract)
});
info!(
contracts = contracts.len(),
skipped = skipped.len(),
next_order_id = next_id,
"Connected to IB"
);
let client = Self {
config: Arc::new(config),
client: Arc::new(client),
contracts,
order_ids: OrderIdMap::new(),
pending_cancels: PendingCancels::new(),
execution_buffer: ExecutionBuffer::new(),
next_order_id: Arc::new(Mutex::new(next_id)),
known_live: KnownLiveOrders::shared(ExchangeId::Ibkr),
listings: ListingLock::default(),
earlier_completions: ended_orders::EarlierCompletions::new(),
};
Ok(ConnectOutcome { client, skipped })
}
fn allocate_order_id(&self) -> i32 {
self.allocate_order_id_range(1)
}
#[allow(clippy::expect_used)] fn allocate_order_id_range(&self, count: u32) -> i32 {
let count_i32: i32 = count.try_into().expect("count exceeds i32::MAX");
let mut id = self.next_order_id.lock();
let base = *id;
*id = id
.checked_add(count_i32)
.expect("order ID overflow: i32::MAX exceeded");
base
}
pub async fn resolve_contract(
&self,
contract: &ibapi::contracts::Contract,
) -> Result<ibapi::contracts::Contract, ResolveContractError> {
let client = Arc::clone(&self.client);
let contract = contract.clone();
tokio::task::spawn_blocking(move || resolve_contract_blocking(&client, &contract))
.await
.map_err(|e| ResolveContractError::Failed(format!("task join: {e}")))?
}
pub fn register_contract(
&self,
name: InstrumentNameExchange,
contract: ibapi::contracts::Contract,
) -> Result<(), ContractRegistryError> {
self.contracts.register(name, contract)
}
pub fn contract_registry(&self) -> &ContractRegistry {
&self.contracts
}
pub fn pending_execution_count(&self) -> usize {
self.execution_buffer.pending_count()
}
pub fn clear_stale_executions(&self, max_age: std::time::Duration) -> usize {
self.execution_buffer.clear_stale(max_age)
}
pub fn clear_stale_order_ids(&self, max_age: std::time::Duration) -> usize {
self.order_ids.clear_stale(max_age)
}
pub fn clear_stale_pending_cancels(&self, max_age: std::time::Duration) -> usize {
self.pending_cancels.clear_stale(max_age)
}
pub fn disconnect(&self) {
debug!("Disconnecting IbkrClient");
self.client.disconnect();
}
pub async fn open_bracket_order(&self, request: BracketOrderRequest) -> BracketOrderResult {
let instrument = request.instrument.clone();
let contract = match self.contracts.get_contract(&instrument) {
Some(c) => c,
None => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::InstrumentInvalid(
instrument,
"contract not registered".to_string(),
)),
);
}
};
let quantity: f64 = match request.quantity.try_into() {
Ok(q) => q,
Err(_) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(format!(
"quantity {} exceeds f64 range",
request.quantity
))),
);
}
};
let entry_price: f64 = match request.entry_price.try_into() {
Ok(p) => p,
Err(_) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(format!(
"entry_price {} exceeds f64 range",
request.entry_price
))),
);
}
};
let tp_price: f64 = match request.take_profit_price.try_into() {
Ok(p) => p,
Err(_) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(format!(
"take_profit_price {} exceeds f64 range",
request.take_profit_price
))),
);
}
};
let sl_price: f64 = match request.stop_loss_price.try_into() {
Ok(p) => p,
Err(_) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(format!(
"stop_loss_price {} exceeds f64 range",
request.stop_loss_price
))),
);
}
};
let ib_tif = match time_in_force_to_ib(&request.time_in_force) {
Ok(tif) => tif,
Err(e) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(format!(
"TIF not supported by IB: {e}"
))),
);
}
};
let parent_ib_id = self.allocate_order_id_range(3);
let tp_ib_id = parent_ib_id + 1;
let sl_ib_id = parent_ib_id + 2;
let action = side_to_action(request.side);
let mut ib_orders = match build_ib_bracket_with_oca(
parent_ib_id,
action,
quantity,
entry_price,
tp_price,
sl_price,
ib_tif,
) {
Ok(orders) => orders,
Err(e) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(e.to_string())),
);
}
};
let parent_cid = request.parent_cid.clone();
let (tp_cid, sl_cid) = derive_child_cids(&parent_cid);
for (ib_order, cid) in ib_orders.iter_mut().zip([&parent_cid, &tp_cid, &sl_cid]) {
match order::order_ref(cid) {
Ok(order_ref) => ib_order.order_ref = order_ref,
Err(e) => {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(e.to_string())),
);
}
}
}
let exit_side = match request.side {
Side::Buy => Side::Sell,
Side::Sell => Side::Buy,
};
let parent_ctx = OrderContext {
instrument: instrument.clone(),
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
};
let tp_ctx = OrderContext {
instrument: instrument.clone(),
side: exit_side,
price: Some(request.take_profit_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
};
let sl_ctx = OrderContext {
instrument: instrument.clone(),
side: exit_side,
price: None, quantity: request.quantity,
kind: OrderKind::Stop {
trigger_price: request.stop_loss_price,
},
time_in_force: request.time_in_force,
};
if let Err(in_use) = self.order_ids.register_all([
(parent_cid.clone(), parent_ib_id, parent_ctx),
(tp_cid.clone(), tp_ib_id, tp_ctx),
(sl_cid.clone(), sl_ib_id, sl_ctx),
]) {
return make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::DuplicateClientOrderId(in_use.to_string())),
);
}
{
let mut known = self.known_live.lock();
for cid in [&parent_cid, &tp_cid, &sl_cid] {
known.placing(cid);
}
}
let client = self.client.clone();
let result = tokio::task::spawn_blocking(move || {
let rollback = |ids: &[i32]| -> (Vec<i32>, String) {
let failed: Vec<(i32, ibapi::Error)> = ids
.iter()
.filter_map(|&id| {
let e = client.cancel_order(id, "").err()?;
error!(order_id = id, error = %e, "bracket rollback cancel failed; leg may be live");
Some((id, e))
})
.collect();
if failed.is_empty() {
return (Vec::new(), String::new());
}
let named: Vec<String> = failed.iter().map(|(id, e)| format!("{id} ({e})")).collect();
let suffix = format!(
"; rollback cancel failed for order ids {}, which may be live",
named.join(", ")
);
(failed.into_iter().map(|(id, _)| id).collect(), suffix)
};
let leg_send_failure = |e: &ibapi::Error, message: String, sent: &[i32]| {
let (cancel_failed, rollback) = rollback(sent);
BracketFailure {
error: leg_send_error(e, message, &rollback),
unresolved: cancel_failed,
}
};
let parent_sub = match client.place_order(parent_ib_id, &contract, &ib_orders[0]) {
Ok(s) => s,
Err(e) => {
return Err(BracketFailure {
error: send_error(&e, format!("parent order failed: {e}")),
unresolved: Vec::new(),
});
}
};
let tp_sub = match client.place_order(tp_ib_id, &contract, &ib_orders[1]) {
Ok(s) => s,
Err(e) => {
return Err(leg_send_failure(
&e,
format!("take_profit order failed: {e}"),
&[parent_ib_id],
));
}
};
let sl_sub = match client.place_order(sl_ib_id, &contract, &ib_orders[2]) {
Ok(s) => s,
Err(e) => {
return Err(leg_send_failure(
&e,
format!("stop_loss order failed: {e}"),
&[parent_ib_id, tp_ib_id],
));
}
};
let leg_status = |outcome: PlacementOutcome, leg: &str| match outcome {
PlacementOutcome::Accepted { filled } => Some(Ok(filled)),
PlacementOutcome::HeldPending(notice) => {
warn!(leg, %notice, "bracket leg held/pending; tracking via stream");
Some(Ok(0.0))
}
PlacementOutcome::Rejected(reason) => Some(Err(reason)),
PlacementOutcome::NoStatus => None,
};
let parent_status = leg_status(
await_order_placement(parent_sub.timeout_iter_data(PLACEMENT_STATUS_TIMEOUT)),
"parent",
);
let tp_status = leg_status(
await_order_placement(tp_sub.timeout_iter_data(PLACEMENT_STATUS_TIMEOUT)),
"take_profit",
);
let sl_status = leg_status(
await_order_placement(sl_sub.timeout_iter_data(PLACEMENT_STATUS_TIMEOUT)),
"stop_loss",
);
match (parent_status, tp_status, sl_status) {
(Some(Ok(parent)), Some(Ok(tp)), Some(Ok(sl))) => Ok((parent, tp, sl)),
(parent, tp, sl) => {
let (cancel_failed, rollback) = rollback(&[parent_ib_id, tp_ib_id, sl_ib_id]);
let unresolved = unresolved_legs(
[(parent_ib_id, &parent), (tp_ib_id, &tp), (sl_ib_id, &sl)],
&cancel_failed,
);
let show = |status: Option<Result<f64, String>>| match status {
Some(status) => format!("{status:?}"),
None => "no status (timed out or transport lost)".to_string(),
};
Err(BracketFailure {
error: OrderError::Rejected(ApiError::OrderRejected(format!(
"bracket order failed: parent={}, tp={}, sl={}{rollback}",
show(parent),
show(tp),
show(sl)
))),
unresolved,
})
}
}
})
.await;
let result = match result {
Ok(Ok((parent_filled, tp_filled, sl_filled))) => {
let now = Utc::now();
let parent_filled_dec = parse_decimal_or_warn(parent_filled, "parent.filled");
let tp_filled_dec = parse_decimal_or_warn(tp_filled, "tp.filled");
let sl_filled_dec = parse_decimal_or_warn(sl_filled, "sl.filled");
BracketOrderResult {
parent: Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: instrument.clone(),
strategy: request.strategy.clone(),
cid: parent_cid,
},
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::active(Open::new(
VenueOrderId::Assigned(OrderId::new(format_smolstr!(
"{}",
parent_ib_id
))),
now,
parent_filled_dec,
)),
},
take_profit: Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: instrument.clone(),
strategy: request.strategy.clone(),
cid: tp_cid,
},
side: exit_side,
price: Some(request.take_profit_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::active(Open::new(
VenueOrderId::Assigned(OrderId::new(format_smolstr!("{}", tp_ib_id))),
now,
tp_filled_dec,
)),
},
stop_loss: Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: instrument.clone(),
strategy: request.strategy.clone(),
cid: sl_cid,
},
side: exit_side,
price: None, quantity: request.quantity,
kind: OrderKind::Stop {
trigger_price: request.stop_loss_price,
},
time_in_force: request.time_in_force,
state: OrderState::active(Open::new(
VenueOrderId::Assigned(OrderId::new(format_smolstr!("{}", sl_ib_id))),
now,
sl_filled_dec,
)),
},
}
}
Ok(Err(failure)) => failed_bracket(
&request,
[parent_ib_id, tp_ib_id, sl_ib_id],
failure,
&self.order_ids,
),
Err(join_err) => {
self.order_ids.remove_by_ib_id(parent_ib_id);
self.order_ids.remove_by_ib_id(tp_ib_id);
self.order_ids.remove_by_ib_id(sl_ib_id);
make_all_inactive_bracket(
&request,
OrderError::Rejected(ApiError::OrderRejected(format!(
"task join error: {join_err}"
))),
)
}
};
let mut known = self.known_live.lock();
for leg in [&result.parent, &result.take_profit, &result.stop_loss] {
known.placed(&leg.key, leg.quantity, leg.kind, &leg.state);
}
drop(known);
result
}
}
fn derive_child_cids(parent_cid: &ClientOrderId) -> (ClientOrderId, ClientOrderId) {
(
ClientOrderId::new(format_smolstr!("{}_tp", parent_cid.0)),
ClientOrderId::new(format_smolstr!("{}_sl", parent_cid.0)),
)
}
struct BracketFailure {
error: OrderError<AssetNameExchange, InstrumentNameExchange>,
unresolved: Vec<i32>,
}
fn unresolved_legs<T, E>(
legs: [(i32, &Option<Result<T, E>>); 3],
cancel_failed: &[i32],
) -> Vec<i32> {
legs.into_iter()
.filter(|(id, status)| match status {
None => true,
Some(Ok(_)) => cancel_failed.contains(id),
Some(Err(_)) => false,
})
.map(|(id, _)| id)
.collect()
}
fn leg_send_error(
e: &ibapi::Error,
message: String,
rollback: &str,
) -> OrderError<AssetNameExchange, InstrumentNameExchange> {
if rollback.is_empty() {
send_error(e, message)
} else {
OrderError::Rejected(ApiError::OrderRejected(format!("{message}{rollback}")))
}
}
fn failed_bracket(
request: &BracketOrderRequest,
[parent_ib_id, tp_ib_id, sl_ib_id]: [i32; 3],
failure: BracketFailure,
order_ids: &OrderIdMap,
) -> BracketOrderResult {
let BracketFailure { error, unresolved } = failure;
if !unresolved.is_empty() {
warn!(
%error,
?unresolved,
"bracket placement failed with legs of unknown fate; returning them Open, tracking via stream"
);
}
let mut result = make_all_inactive_bracket(request, error);
let now = Utc::now();
for (leg, ib_id) in [
(&mut result.parent, parent_ib_id),
(&mut result.take_profit, tp_ib_id),
(&mut result.stop_loss, sl_ib_id),
] {
if unresolved.contains(&ib_id) {
leg.state = OrderState::active(Open::new(
VenueOrderId::Assigned(OrderId::new(format_smolstr!("{ib_id}"))),
now,
Decimal::ZERO,
));
} else {
order_ids.remove_by_ib_id(ib_id);
}
}
result
}
fn make_all_inactive_bracket(
request: &BracketOrderRequest,
error: OrderError<AssetNameExchange, InstrumentNameExchange>,
) -> BracketOrderResult {
let exit_side = match request.side {
Side::Buy => Side::Sell,
Side::Sell => Side::Buy,
};
let parent_cid = request.parent_cid.clone();
let (tp_cid, sl_cid) = derive_child_cids(&parent_cid);
BracketOrderResult {
parent: Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: request.instrument.clone(),
strategy: request.strategy.clone(),
cid: parent_cid,
},
side: request.side,
price: Some(request.entry_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(error.clone()),
},
take_profit: Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: request.instrument.clone(),
strategy: request.strategy.clone(),
cid: tp_cid,
},
side: exit_side,
price: Some(request.take_profit_price),
quantity: request.quantity,
kind: OrderKind::Limit,
time_in_force: request.time_in_force,
state: OrderState::inactive(error.clone()),
},
stop_loss: Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: request.instrument.clone(),
strategy: request.strategy.clone(),
cid: sl_cid,
},
side: exit_side,
price: None, quantity: request.quantity,
kind: OrderKind::Stop {
trigger_price: request.stop_loss_price,
},
time_in_force: request.time_in_force,
state: OrderState::inactive(error),
},
}
}
impl ExecutionClient for IbkrClient {
const EXCHANGE: ExchangeId = ExchangeId::Ibkr;
const SUPPORTED_KINDS: &'static [InstrumentKindDiscriminant] = &[
InstrumentKindDiscriminant::Spot,
InstrumentKindDiscriminant::Future,
InstrumentKindDiscriminant::Option,
];
type Config = IbkrConfig;
type AccountStream = BoxStream<'static, UnindexedAccountEvent>;
fn validate_config(
config: &Self::Config,
instruments: &[ClientInstrument<'_>],
) -> Result<(), String> {
let problems = config
.contracts
.iter()
.filter_map(|contract| {
if let Err(error) = contract.to_contract() {
return Some(format!("contract {:?}: {error}", contract.name));
}
let instrument = instruments
.iter()
.find(|instrument| instrument.name_exchange.as_ref() == contract.name)?;
let expected = security_types_of(instrument.kind);
if expected.is_empty() {
return Some(format!(
"contract {:?} is for an instrument of kind {}, which this client cannot \
trade",
contract.name, instrument.kind,
));
}
(!expected.contains(&contract.security_type.as_str())).then(|| {
format!(
"contract {:?} names security type {:?}, but its instrument's kind is \
{}, which takes {}",
contract.name,
contract.security_type,
instrument.kind,
expected.join(" or "),
)
})
})
.collect::<Vec<_>>();
if problems.is_empty() {
Ok(())
} else {
Err(problems.join("; "))
}
}
#[track_caller]
fn new(config: Self::Config) -> Self {
#[allow(clippy::expect_used)] Self::connect_sync(config)
.expect("failed to connect to IB or to register its configured contracts")
}
async fn account_snapshot(
&self,
assets: &[AssetNameExchange],
instruments: &[InstrumentNameExchange],
) -> Result<UnindexedAccountSnapshot, UnindexedClientError> {
let client = self.client.clone();
let contracts = self.contracts.clone();
let instruments_filter: Option<HashSet<_>> = if instruments.is_empty() {
None
} else {
Some(instruments.iter().cloned().collect())
};
let balances_future = self.fetch_balances(assets);
let positions_filter = instruments_filter.clone();
let positions_future = tokio::task::spawn_blocking(move || {
let instruments_filter = positions_filter;
let accounts = client
.managed_accounts()
.map_err(|e| UnindexedClientError::Internal(format!("managed accounts: {e}")))?;
if accounts.is_empty() {
return Err(UnindexedClientError::Internal(
"no managed accounts: IB listed none, or the session ended".to_string(),
));
}
let mut positions = PositionAggregator::default();
let mut listed_all = true;
for account in accounts {
listed_all &= read_account_positions(
&client,
AccountId(account),
&contracts,
instruments_filter.as_ref(),
&mut positions,
)?;
}
if listed_all {
positions.listing_ended();
}
let requested: Vec<_> = instruments_filter
.iter()
.flatten()
.filter(|instrument| {
contracts.get_contract(instrument).is_some_and(|contract| {
contract.contract_id != 0
&& contracts.get_name_by_con_id(contract.contract_id).as_ref()
== Some(*instrument)
})
})
.cloned()
.collect();
Ok::<_, UnindexedClientError>((positions, requested))
});
let client = self.client.clone();
let contracts = self.contracts.clone();
let order_ids = self.order_ids.clone();
let listings = self.listings.clone();
let orders_future = tokio::task::spawn_blocking(move || {
let listing = listings.open_orders(&client)?;
Ok::<_, UnindexedClientError>(open_orders_from_listing(
listing,
client.client_id(),
&contracts,
&order_ids,
))
});
let (balances_result, positions_result, orders_result) =
tokio::join!(balances_future, positions_future, orders_future);
let balances = balances_result?;
let (positions, requested) = positions_result
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))??;
let listed = match orders_result
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))?
{
Ok(mut listed) => {
if let Some(filter) = &instruments_filter {
listed.retain(|listed| filter.contains(&listed.instrument));
}
Some(listed)
}
Err(error) => {
warn!(
%error,
"Could not read IBKR's open orders for the account snapshot; no instrument's \
orders are reported complete"
);
None
}
};
if let Some(listed) = &listed {
remember_listed(&self.known_live, listed);
}
let mut orders = SnapshotOrders::new(listed);
let reported = positions.into_reports(
Utc::now(),
requested
.iter()
.chain(orders.instruments())
.collect::<HashSet<_>>(),
)?;
let mut instrument_snapshots = Vec::with_capacity(reported.len());
for (instrument, position) in reported {
let (orders, orders_complete) = orders.take(&instrument);
instrument_snapshots.push(InstrumentAccountSnapshot {
instrument,
orders,
orders_complete,
position,
isolated: None,
});
}
instrument_snapshots.extend(orders.into_rest().map(|(instrument, (orders, complete))| {
InstrumentAccountSnapshot {
instrument,
orders,
orders_complete: complete,
position: PositionReport::Unreported,
isolated: None,
}
}));
Ok(AccountSnapshot {
exchange: ExchangeId::Ibkr,
balances,
instruments: instrument_snapshots,
})
}
async fn account_stream(
&self,
_assets: &[AssetNameExchange],
_instruments: &[InstrumentNameExchange],
) -> Result<Self::AccountStream, UnindexedClientError> {
let client = self.client.clone();
let (notices, order_sub) = tokio::task::spawn_blocking(move || {
let notices = client.notice_stream()?;
client.order_update_stream().map(|sub| (notices, sub))
})
.await
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))?
.map_err(|e| UnindexedClientError::Internal(format!("order updates: {e}")))?;
let api_client_id = self.client.client_id();
let contracts_clone = self.contracts.clone();
let order_ids_clone = self.order_ids.clone();
let pending_cancels_clone = self.pending_cancels.clone();
let exec_buffer_clone = self.execution_buffer.clone();
let (tx, rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(tx, new_dedup_cache(), self.known_live.clone());
let watcher = recovery::RecoveryWatcher {
client: self.client.clone(),
notices,
contracts: self.contracts.clone(),
order_ids: self.order_ids.clone(),
pending_cancels: self.pending_cancels.clone(),
pending: self.execution_buffer.clone(),
known: self.known_live.clone(),
listings: self.listings.clone(),
earlier_completions: self.earlier_completions.clone(),
sink: sink.clone(),
runtime: tokio::runtime::Handle::current(),
};
let watcher_sink = sink.clone();
std::thread::Builder::new()
.name("ibkr-recovery".to_string())
.spawn(move || {
if let Err(panic_info) = catch_unwind(AssertUnwindSafe(|| watcher.run())) {
let msg = panic_message(panic_info.as_ref());
error!("IBKR recovery worker panicked: {msg}");
watcher_sink.terminate(StreamTerminationReason::Error(format!(
"IBKR recovery worker panicked: {msg}"
)));
}
})
.map_err(|e| UnindexedClientError::TaskFailed(format!("thread spawn: {e}")))?;
std::thread::Builder::new()
.name("ibkr-order-stream".to_string())
.spawn(move || {
let result = catch_unwind(AssertUnwindSafe(|| {
forward_order_updates(
order_sub.iter_data(),
&sink,
api_client_id,
&contracts_clone,
&order_ids_clone,
&pending_cancels_clone,
&exec_buffer_clone,
);
}));
if let Err(panic_info) = result {
let msg = panic_message(panic_info.as_ref());
error!("Order stream worker panicked: {msg}");
sink.terminate(StreamTerminationReason::Error(format!(
"IBKR order-update worker panicked: {msg}"
)));
}
})
.map_err(|e| UnindexedClientError::TaskFailed(format!("thread spawn: {e}")))?;
Ok(Box::pin(
tokio_stream::wrappers::UnboundedReceiverStream::new(rx),
))
}
async fn cancel_order(
&self,
request: OrderRequestCancel<ExchangeId, &InstrumentNameExchange>,
) -> UnindexedOrderResponseCancel {
let key = OrderKey {
exchange: request.key.exchange,
instrument: request.key.instrument.clone(),
strategy: request.key.strategy.clone(),
cid: request.key.cid.clone(),
};
let ib_order_id = match self.order_ids.get_ib_id(&request.key.cid) {
Some(id) => id,
None => {
return OrderResponseCancel {
key,
state: Err(crate::error::OrderError::Rejected(ApiError::OrderRejected(
"order ID not found in map".to_string(),
))),
};
}
};
let client = self.client.clone();
let result =
tokio::task::spawn_blocking(move || client.cancel_order(ib_order_id, "")).await;
match result {
Ok(Ok(_sub)) => {
self.pending_cancels.insert(ib_order_id);
OrderResponseCancel {
key,
state: Ok(Cancelled::new(
OrderId::new(format_smolstr!("{}", ib_order_id)),
Utc::now(),
None,
)),
}
}
Ok(Err(e)) => {
error!(order_id = ib_order_id, error = %e, "Failed to cancel order");
OrderResponseCancel {
key,
state: Err(send_error(&e, e.to_string())),
}
}
Err(e) => {
error!(order_id = ib_order_id, error = %e, "Task join error");
OrderResponseCancel {
key,
state: Err(crate::error::OrderError::Rejected(ApiError::OrderRejected(
e.to_string(),
))),
}
}
}
}
async fn open_order(
&self,
request: OrderRequestOpen<ExchangeId, &InstrumentNameExchange>,
) -> Order<ExchangeId, InstrumentNameExchange, UnindexedOrderState> {
let key = OrderKey {
exchange: ExchangeId::Ibkr,
instrument: request.key.instrument.clone(),
strategy: request.key.strategy.clone(),
cid: request.key.cid.clone(),
};
let contract = match self.contracts.get_contract(request.key.instrument) {
Some(c) => c,
None => {
return Order {
key,
side: request.state.side,
price: request.state.price,
quantity: request.state.quantity,
kind: request.state.kind,
time_in_force: request.state.time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::InstrumentInvalid(
request.key.instrument.clone(),
"contract not registered".to_string(),
))),
};
}
};
let quantity: f64 = match request.state.quantity.try_into() {
Ok(q) => q,
Err(_) => {
return Order {
key,
side: request.state.side,
price: request.state.price,
quantity: request.state.quantity,
kind: request.state.kind,
time_in_force: request.state.time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
format!("quantity {} exceeds f64 range", request.state.quantity),
))),
};
}
};
let ib_order = match order::order_ref(&request.key.cid).and_then(|order_ref| {
build_ib_order(
request.state.side,
quantity,
&request.state.kind,
request.state.price,
&request.state.time_in_force,
)
.map(|order| ibapi::orders::Order { order_ref, ..order })
}) {
Ok(o) => o,
Err(e) => {
return Order {
key,
side: request.state.side,
price: request.state.price,
quantity: request.state.quantity,
kind: request.state.kind,
time_in_force: request.state.time_in_force,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
e.to_string(),
))),
};
}
};
let ib_order_id = self.allocate_order_id();
let context = OrderContext {
instrument: request.key.instrument.clone(),
side: request.state.side,
price: request.state.price,
quantity: request.state.quantity,
kind: request.state.kind,
time_in_force: request.state.time_in_force,
};
if let Err(in_use) = self
.order_ids
.register(request.key.cid.clone(), ib_order_id, context)
{
return Order {
key,
side: request.state.side,
price: request.state.price,
quantity: request.state.quantity,
kind: request.state.kind,
time_in_force: request.state.time_in_force,
state: OrderState::inactive(OrderError::Rejected(
ApiError::DuplicateClientOrderId(in_use.to_string()),
)),
};
}
self.known_live.lock().placing(&request.key.cid);
let client = self.client.clone();
let side = request.state.side;
let price = request.state.price;
let req_quantity = request.state.quantity;
let kind = request.state.kind;
let tif = request.state.time_in_force;
let result = tokio::task::spawn_blocking(move || {
let sub = match client.place_order(ib_order_id, &contract, &ib_order) {
Ok(s) => s,
Err(e) => return Err(send_error(&e, e.to_string())),
};
match await_order_placement(sub.timeout_iter_data(PLACEMENT_STATUS_TIMEOUT)) {
PlacementOutcome::Accepted { filled } => {
let filled = parse_decimal_or_warn(filled, "status.filled");
Ok(Some((ib_order_id, filled)))
}
PlacementOutcome::Rejected(reason) => {
Err(OrderError::Rejected(ApiError::OrderRejected(reason)))
}
PlacementOutcome::HeldPending(notice) => {
warn!(ib_order_id, %notice, "order held/pending; tracking via stream");
Ok(None)
}
PlacementOutcome::NoStatus => Ok(None),
}
})
.await;
let order = match result {
Ok(Ok(Some((order_id, filled)))) => {
Order {
key,
side,
price,
quantity: req_quantity,
kind,
time_in_force: tif,
state: OrderState::active(Open::new(
VenueOrderId::Assigned(OrderId::new(format_smolstr!("{}", order_id))),
Utc::now(),
filled,
)),
}
}
Ok(Ok(None)) => {
warn!(
ib_order_id,
"Order subscription ended without terminal status, returning Open"
);
Order {
key,
side,
price,
quantity: req_quantity,
kind,
time_in_force: tif,
state: OrderState::active(Open::new(
VenueOrderId::Assigned(OrderId::new(format_smolstr!("{}", ib_order_id))),
Utc::now(),
Decimal::ZERO,
)),
}
}
Ok(Err(error)) => {
self.order_ids.remove_by_ib_id(ib_order_id);
Order {
key,
side,
price,
quantity: req_quantity,
kind,
time_in_force: tif,
state: OrderState::inactive(error),
}
}
Err(e) => {
self.order_ids.remove_by_ib_id(ib_order_id);
Order {
key,
side,
price,
quantity: req_quantity,
kind,
time_in_force: tif,
state: OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
e.to_string(),
))),
}
}
};
self.known_live
.lock()
.placed(&order.key, order.quantity, order.kind, &order.state);
order
}
async fn fetch_balances(
&self,
assets: &[AssetNameExchange],
) -> Result<Vec<AssetBalance<AssetNameExchange>>, UnindexedClientError> {
let client = self.client.clone();
let assets_filter: Option<HashSet<AssetNameExchange>> = if assets.is_empty() {
None
} else {
Some(assets.iter().cloned().collect())
};
tokio::task::spawn_blocking(move || {
let sub = client
.account_summary(&ACCOUNT_GROUP_ALL, &["TotalCashValue", "AvailableFunds"])
.map_err(|e| UnindexedClientError::Internal(format!("account_summary: {e}")))?;
let mut balances = read_account_summary(
LISTING_STALL_TIMEOUT,
|| client.is_connected(),
|timeout| sub.next_timeout(timeout),
)?
.to_balances();
if let Some(ref filter) = assets_filter {
balances.retain(|b| filter.contains(&b.asset));
}
Ok(balances)
})
.await
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))?
}
async fn fetch_open_orders(
&self,
instruments: &[InstrumentNameExchange],
) -> Result<Vec<Order<ExchangeId, InstrumentNameExchange, Open>>, UnindexedClientError> {
let client = self.client.clone();
let contracts = self.contracts.clone();
let order_ids = self.order_ids.clone();
let listings = self.listings.clone();
let instruments_filter: Option<HashSet<_>> = if instruments.is_empty() {
None
} else {
Some(instruments.iter().cloned().collect())
};
let listed = tokio::task::spawn_blocking(move || {
let listing = listings.open_orders(&client)?;
let mut listed =
open_orders_from_listing(listing, client.client_id(), &contracts, &order_ids);
if let Some(filter) = &instruments_filter {
listed.retain(|listed| filter.contains(&listed.instrument));
}
Ok::<_, UnindexedClientError>(listed)
})
.await
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))??;
remember_listed(&self.known_live, &listed);
Ok(listed
.into_iter()
.filter_map(|listed| listed.order)
.collect())
}
async fn fetch_trades(
&self,
start: DateTime<Utc>,
end: DateTime<Utc>,
instruments: &[InstrumentNameExchange],
) -> Result<TradesRead<AssetNameExchange, InstrumentNameExchange>, UnindexedClientError> {
if start > end {
return Ok(TradesRead::complete(Vec::new()));
}
let client = self.client.clone();
let contracts = self.contracts.clone();
let instruments_filter: Option<HashSet<_>> = if instruments.is_empty() {
None
} else {
Some(instruments.iter().cloned().collect())
};
tokio::task::spawn_blocking(move || {
let api_client_id = client.client_id();
let exec_filter = ibapi::orders::ExecutionFilter {
client_id: Some(api_client_id),
..ibapi::orders::ExecutionFilter::default()
};
let sub = client
.executions(exec_filter)
.map_err(|e| UnindexedClientError::Internal(format!("executions: {e}")))?;
let mut listing = Vec::new();
read_listing(
"executions",
LISTING_STALL_TIMEOUT,
|| client.is_connected(),
|timeout| sub.next_timeout(timeout),
|item| {
listing.push(item);
ControlFlow::Continue(())
},
)?;
let trades = trades_from_executions(
listing,
api_client_id,
&contracts,
start..=end,
instruments_filter.as_ref(),
);
Ok(TradesRead::complete(execution::keep_latest_revisions(
trades,
)))
})
.await
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))?
}
}
fn trades_from_executions(
listing: impl IntoIterator<Item = ibapi::orders::Executions>,
api_client_id: i32,
contracts: &ContractRegistry,
span: std::ops::RangeInclusive<DateTime<Utc>>,
instruments: Option<&HashSet<InstrumentNameExchange>>,
) -> Vec<Trade<AssetNameExchange, InstrumentNameExchange>> {
use ibapi::orders::Executions;
let awaiting_commission = ExecutionBuffer::new();
let mut trades = Vec::new();
for item in listing {
match item {
Executions::ExecutionData(data) => {
let exec = &data.execution;
if exec.client_id != api_client_id {
debug!(
exec_id = %exec.execution_id,
api_client_id = exec.client_id,
"Execution of another API client's order, skipping"
);
continue;
}
let Some(instrument) = contracts.get_name_by_con_id(data.contract.contract_id)
else {
continue;
};
if instruments.is_some_and(|filter| !filter.contains(&instrument)) {
continue;
}
match execution::parse_ib_timestamp(&exec.time) {
Some(time) if span.contains(&time) => {
awaiting_commission.add_execution(data, instrument);
}
Some(_) => {}
None => warn!(
exec_id = %exec.execution_id,
time = %exec.time,
"Unparseable timestamp in execution, skipping"
),
}
}
Executions::CommissionReport(report) => {
if let Some(trade) = awaiting_commission.complete_with_commission(&report) {
trades.push(trade);
}
}
}
}
let without_commission = awaiting_commission.take_without_commission();
if !without_commission.is_empty() {
warn!(
count = without_commission.len(),
"IBKR executions arrived without a commission report; returned with an unknown fee"
);
trades.extend(without_commission);
}
trades.sort_by(|a, b| {
a.time_exchange
.cmp(&b.time_exchange)
.then_with(|| a.id.cmp(&b.id))
});
trades
}
#[derive(Debug)]
struct ListedOpenOrder {
instrument: InstrumentNameExchange,
order: Option<Order<ExchangeId, InstrumentNameExchange, Open>>,
named_as_placed: bool,
paired_with_status: bool,
}
impl ListedOpenOrder {
fn is_exact(&self) -> bool {
self.order.is_some() && self.named_as_placed && self.paired_with_status
}
}
fn open_orders_from_listing(
listing: impl IntoIterator<Item = ibapi::orders::Orders>,
api_client_id: i32,
contracts: &ContractRegistry,
order_ids: &OrderIdMap,
) -> Vec<ListedOpenOrder> {
use ibapi::orders::Orders;
let mut listed = std::collections::BTreeMap::new();
let mut statuses = fnv::FnvHashMap::default();
for item in listing {
match item {
Orders::OrderData(data) if data.order.client_id == api_client_id => {
listed.insert(data.order_id, data);
}
Orders::OrderStatus(status) if status.client_id == api_client_id => {
statuses.insert(status.order_id, status);
}
_ => {}
}
}
listed
.into_values()
.filter_map(|data| {
open_order_from_listing(&data, statuses.get(&data.order_id), contracts, order_ids)
})
.collect()
}
fn open_order_from_listing(
data: &ibapi::orders::OrderData,
status: Option<&ibapi::orders::OrderStatus>,
contracts: &ContractRegistry,
order_ids: &OrderIdMap,
) -> Option<ListedOpenOrder> {
let ib_id = data.order_id;
let instrument = contracts.get_name_by_con_id(data.contract.contract_id)?;
let state = status.map_or(&data.order_state.status, |status| &status.status);
if state.is_terminal() {
debug!(
ib_order_id = ib_id,
?state,
"Listed open order has ended, skipping"
);
return None;
}
let filled = status.map_or(data.order.filled_quantity, |status| status.filled);
let filled = try_decimal_or_warn(filled, format_args!("filled quantity of order {ib_id}"))
.unwrap_or(Decimal::ZERO);
let order_ref = (!data.order.order_ref.is_empty())
.then(|| ClientOrderId::new(data.order.order_ref.as_str()));
let tracked = order_ids
.get_client_id_and_context(ib_id)
.filter(|(cid, ctx)| {
ctx.instrument == instrument && order_ref.as_ref().is_none_or(|r| r == cid)
});
let (cid, ctx, named_as_placed) = match tracked {
Some((cid, ctx)) => (cid, ctx, true),
None => {
let shape = match order::order_shape_from_ib(&data.order) {
Ok(shape) => shape,
Err(reason) => {
warn!(
ib_order_id = ib_id,
%instrument,
%reason,
"Open IBKR order cannot be read back, leaving it out"
);
return Some(ListedOpenOrder {
instrument,
order: None,
named_as_placed: false,
paired_with_status: status.is_some(),
});
}
};
let ctx = OrderContext {
instrument: instrument.clone(),
side: order::action_to_side(&data.order.action),
price: shape.price,
quantity: parse_decimal_or_warn(data.order.total_quantity, "total_quantity"),
kind: shape.kind,
time_in_force: shape.time_in_force,
};
match order_ref {
Some(cid) if adopt_listed_order(order_ids, &cid, ib_id, &ctx) => (cid, ctx, true),
Some(_) | None => (ClientOrderId::new(format_smolstr!("{ib_id}")), ctx, false),
}
}
};
Some(ListedOpenOrder {
instrument,
order: Some(Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: ctx.instrument,
strategy: StrategyId::unknown(),
cid,
},
side: ctx.side,
price: ctx.price,
quantity: ctx.quantity,
kind: ctx.kind,
time_in_force: ctx.time_in_force,
state: Open::new(
VenueOrderId::Assigned(execution::ib_order_id(ib_id)),
Utc::now(),
filled,
),
}),
named_as_placed,
paired_with_status: status.is_some(),
})
}
#[derive(Debug)]
struct SnapshotOrders {
by_instrument: fnv::FnvHashMap<InstrumentNameExchange, (Vec<UnindexedOrderSnapshot>, bool)>,
read: bool,
}
impl SnapshotOrders {
fn new(listed: Option<Vec<ListedOpenOrder>>) -> Self {
let read = listed.is_some();
let mut by_instrument = fnv::FnvHashMap::default();
for listed in listed.into_iter().flatten() {
let exact = listed.is_exact();
let (orders, complete): &mut (Vec<_>, bool) = by_instrument
.entry(listed.instrument)
.or_insert_with(|| (Vec::new(), true));
*complete &= exact;
if let Some(order) = listed.order {
orders.push(order.into());
}
}
Self {
by_instrument,
read,
}
}
fn instruments(&self) -> impl Iterator<Item = &InstrumentNameExchange> {
self.by_instrument.keys()
}
fn take(&mut self, instrument: &InstrumentNameExchange) -> (Vec<UnindexedOrderSnapshot>, bool) {
self.by_instrument
.remove(instrument)
.unwrap_or_else(|| (Vec::new(), self.read))
}
fn into_rest(
self,
) -> impl Iterator<Item = (InstrumentNameExchange, (Vec<UnindexedOrderSnapshot>, bool))> {
self.by_instrument.into_iter()
}
}
fn listed_cids(
listing: impl IntoIterator<Item = ibapi::orders::Orders>,
api_client_id: i32,
order_ids: &OrderIdMap,
) -> fnv::FnvHashSet<ClientOrderId> {
listing
.into_iter()
.filter_map(|item| match item {
ibapi::orders::Orders::OrderData(data) if data.order.client_id == api_client_id => {
if data.order.order_ref.is_empty() {
order_ids.get_client_id(data.order_id)
} else {
Some(ClientOrderId::new(data.order.order_ref.as_str()))
}
}
_ => None,
})
.collect()
}
fn remember_listed(known: &SharedKnownLiveOrders, listed: &[ListedOpenOrder]) {
let mut known = known.lock();
for listed in listed.iter().filter(|listed| listed.named_as_placed) {
if let Some(order) = &listed.order {
known.live(&order.key, order.quantity, order.kind, &order.state);
}
}
}
fn adopt_listed_order(
order_ids: &OrderIdMap,
cid: &ClientOrderId,
ib_id: i32,
ctx: &OrderContext,
) -> bool {
match order_ids.adopt(cid.clone(), ib_id, ctx.clone()) {
Ok(()) => {
info!(
ib_order_id = ib_id,
%cid,
instrument = %ctx.instrument,
"Adopted an open IBKR order this client did not track"
);
true
}
Err(refused) => {
warn!(
ib_order_id = ib_id,
%cid,
reason = %refused,
"Listed IBKR order cannot be tracked under its client order id; listing it under \
its IB order id"
);
false
}
}
}
const GTD_EXPIRY_TOLERANCE: chrono::Duration = chrono::Duration::seconds(5);
fn cancelled_at_expiry(user_cancel: bool, time_in_force: &TimeInForce, at: DateTime<Utc>) -> bool {
!user_cancel
&& match time_in_force {
TimeInForce::GoodUntilEndOfDay => true,
TimeInForce::GoodTillDate { expiry } => at + GTD_EXPIRY_TOLERANCE >= *expiry,
_ => false,
}
}
fn make_order_from_status(
status: &ibapi::orders::OrderStatus,
client_id: ClientOrderId,
ctx: &OrderContext,
pending_cancels: &PendingCancels,
) -> Order<ExchangeId, InstrumentNameExchange, OrderState<AssetNameExchange, InstrumentNameExchange>>
{
use ibapi::orders::OrderStatusKind;
let ib_id = status.order_id;
let order_id = execution::ib_order_id(ib_id);
let reported_fill = try_decimal_or_warn(
status.filled,
format_args!("status.filled of order {ib_id}"),
);
let filled_qty = reported_fill.unwrap_or(Decimal::ZERO);
let state = match status.status {
OrderStatusKind::Inactive => {
OrderState::inactive(OrderError::Rejected(ApiError::OrderRejected(
"IB status: Inactive (order blocked by validation/margin/exchange)".into(),
)))
}
OrderStatusKind::Cancelled => {
let now = Utc::now();
if cancelled_at_expiry(pending_cancels.remove(ib_id), &ctx.time_in_force, now) {
OrderState::inactive(Expired::new(order_id, now, reported_fill))
} else {
OrderState::inactive(Cancelled::new(order_id, now, reported_fill))
}
}
OrderStatusKind::Filled => {
let filled_qty = reported_fill.unwrap_or(ctx.quantity);
OrderState::fully_filled(Filled::new(order_id, Utc::now(), filled_qty, None))
}
OrderStatusKind::Submitted
| OrderStatusKind::PreSubmitted
| OrderStatusKind::PendingSubmit
| OrderStatusKind::ApiPending
| OrderStatusKind::PendingCancel
| OrderStatusKind::ApiCancelled => OrderState::active(Open::new(
VenueOrderId::Assigned(order_id),
Utc::now(),
filled_qty,
)),
OrderStatusKind::Unknown(ref raw) => {
warn!(
ib_id,
status = raw.as_str(),
"unmodelled IBKR order status; treating as live, account \
stream is authoritative"
);
OrderState::active(Open::new(
VenueOrderId::Assigned(order_id),
Utc::now(),
filled_qty,
))
}
};
Order {
key: OrderKey {
exchange: ExchangeId::Ibkr,
instrument: ctx.instrument.clone(),
strategy: StrategyId::unknown(),
cid: client_id,
},
side: ctx.side,
price: ctx.price,
quantity: ctx.quantity,
kind: ctx.kind,
time_in_force: ctx.time_in_force,
state,
}
}
pub use execution::parse_ib_timestamp;
impl BracketOrderClient for IbkrClient {
async fn open_bracket_order(
&self,
request: UnifiedBracketOrderRequest<ExchangeId, &InstrumentNameExchange>,
) -> UnifiedBracketOrderResult {
let ibkr_request = BracketOrderRequest {
instrument: request.key.instrument.clone(),
strategy: request.key.strategy.clone(),
parent_cid: request.key.cid.clone(),
side: request.state.side,
quantity: request.state.quantity,
entry_price: request.state.entry_price,
take_profit_price: request.state.take_profit_price,
stop_loss_price: request.state.stop_loss_price,
time_in_force: request.state.time_in_force,
};
let result = self.open_bracket_order(ibkr_request).await;
UnifiedBracketOrderResult::with_all_legs(
result.parent,
result.take_profit,
result.stop_loss,
)
}
}
impl OrderStatusClient for IbkrClient {
async fn fetch_ended_orders(
&self,
orders: &[crate::order::UnindexedOrderKey],
) -> Result<Vec<crate::order::UnindexedInactiveOrder>, UnindexedClientError> {
if orders.is_empty() {
return Ok(Vec::new());
}
let orders = orders.to_vec();
let client = self.client.clone();
let listings = self.listings.clone();
let contracts = self.contracts.clone();
let order_ids = self.order_ids.clone();
let pending_cancels = self.pending_cancels.clone();
let earlier_completions = self.earlier_completions.clone();
tokio::task::spawn_blocking(move || {
let reader = ended_orders::EndedOrderReader::new(
&client,
&listings,
&contracts,
&order_ids,
&pending_cancels,
&earlier_completions,
);
futures::executor::block_on(crate::client::order_recovery::fetch_ended_by_key(
&orders,
|key| std::future::ready(reader.lookup(&key)),
))
})
.await
.map_err(|e| UnindexedClientError::TaskFailed(format!("task join: {e}")))?
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used)] mod contract_config_tests {
use super::*;
use contract::ContractConfigError as E;
fn full_config(security_type: &str) -> ContractConfig {
ContractConfig {
name: "TEST".to_string(),
symbol: "AAPL".to_string(),
security_type: security_type.to_string(),
exchange: "SMART".to_string(),
currency: "USD".to_string(),
last_trade_date: Some("20260116".to_string()),
strike: Some(150.0),
right: Some("C".to_string()),
}
}
#[test]
fn stk_and_cash_ignore_option_fields() {
let mut stk = full_config("STK");
stk.last_trade_date = None;
stk.strike = None;
stk.right = None;
assert!(stk.to_contract().is_ok());
let mut cash = full_config("CASH");
cash.last_trade_date = None;
cash.strike = None;
cash.right = None;
assert!(cash.to_contract().is_ok());
}
#[test]
fn valid_fut_and_opt_build() {
assert!(full_config("FUT").to_contract().is_ok());
assert!(full_config("OPT").to_contract().is_ok());
}
#[test]
fn opt_missing_right_is_error_not_silent_call() {
let mut cfg = full_config("OPT");
cfg.right = None;
assert_eq!(cfg.to_contract().unwrap_err(), E::MissingOptionRight);
}
#[test]
fn opt_unrecognized_right_is_error() {
let mut cfg = full_config("OPT");
cfg.right = Some("X".to_string());
assert_eq!(
cfg.to_contract().unwrap_err(),
E::UnrecognizedOptionRight {
right: "X".to_string()
}
);
}
#[test]
fn opt_missing_strike_is_error() {
let mut cfg = full_config("OPT");
cfg.strike = None;
assert_eq!(cfg.to_contract().unwrap_err(), E::MissingStrike);
}
#[test]
fn fut_and_opt_missing_last_trade_date_is_error() {
let mut fut = full_config("FUT");
fut.last_trade_date = None;
assert_eq!(fut.to_contract().unwrap_err(), E::MissingLastTradeDate);
let mut opt = full_config("OPT");
opt.last_trade_date = None;
assert_eq!(opt.to_contract().unwrap_err(), E::MissingLastTradeDate);
}
#[test]
fn unknown_security_type_is_error_not_silent_stk() {
let cfg = full_config("BOND");
assert_eq!(
cfg.to_contract().unwrap_err(),
E::UnrecognizedSecurityType {
security_type: "BOND".to_string()
}
);
}
fn ibkr_config(contracts: Vec<ContractConfig>) -> IbkrConfig {
IbkrConfig {
host: "127.0.0.1".to_string(),
port: 4002,
client_id: 1,
account: String::new(),
contracts,
}
}
fn named(name: &str, security_type: &str) -> ContractConfig {
ContractConfig {
name: name.to_string(),
..full_config(security_type)
}
}
#[test]
fn validate_config_accepts_each_security_type_its_kind_takes() {
let names = ["STOCK", "FX", "FUTURE", "OPTION"].map(InstrumentNameExchange::new);
let instruments = [
ClientInstrument::new(&names[0], InstrumentKindDiscriminant::Spot),
ClientInstrument::new(&names[1], InstrumentKindDiscriminant::Spot),
ClientInstrument::new(&names[2], InstrumentKindDiscriminant::Future),
ClientInstrument::new(&names[3], InstrumentKindDiscriminant::Option),
];
let config = ibkr_config(vec![
named("STOCK", "STK"),
named("FX", "CASH"),
named("FUTURE", "FUT"),
named("OPTION", "OPT"),
]);
assert_eq!(IbkrClient::validate_config(&config, &instruments), Ok(()));
}
#[test]
fn validate_config_rejects_a_security_type_its_instrument_kind_contradicts() {
let name = InstrumentNameExchange::new("AAPL-C150");
let instruments = [ClientInstrument::new(
&name,
InstrumentKindDiscriminant::Option,
)];
let config = ibkr_config(vec![named("AAPL-C150", "STK")]);
let error = IbkrClient::validate_config(&config, &instruments).unwrap_err();
assert!(error.contains("AAPL-C150"), "{error}");
assert!(error.contains("STK"), "{error}");
assert!(error.contains("OPT"), "{error}");
}
#[test]
fn validate_config_reports_every_invalid_or_contradicting_entry() {
let name = InstrumentNameExchange::new("ES");
let instruments = [ClientInstrument::new(
&name,
InstrumentKindDiscriminant::Future,
)];
let mut missing_date = named("NQ", "FUT");
missing_date.last_trade_date = None;
let config = ibkr_config(vec![named("ES", "STK"), missing_date]);
let error = IbkrClient::validate_config(&config, &instruments).unwrap_err();
assert!(error.contains("\"ES\""), "{error}");
assert!(error.contains("\"NQ\""), "{error}");
}
#[test]
fn validate_config_checks_every_kind_against_every_security_type() {
use InstrumentKindDiscriminant as K;
let accepted = [
(K::Spot, "STK"),
(K::Spot, "CASH"),
(K::Future, "FUT"),
(K::Option, "OPT"),
];
let name = InstrumentNameExchange::new("X");
for kind in [K::Spot, K::Perpetual, K::Future, K::Option, K::Cfd] {
for security_type in ["STK", "CASH", "FUT", "OPT"] {
let instruments = [ClientInstrument::new(&name, kind)];
let config = ibkr_config(vec![named("X", security_type)]);
assert_eq!(
IbkrClient::validate_config(&config, &instruments).is_ok(),
accepted.contains(&(kind, security_type)),
"{kind} with {security_type}"
);
}
}
}
#[test]
fn security_types_cover_every_type_to_contract_builds() {
use InstrumentKindDiscriminant as K;
let listed = [K::Spot, K::Perpetual, K::Future, K::Option, K::Cfd]
.into_iter()
.flat_map(security_types_of)
.copied()
.collect::<Vec<_>>();
for security_type in &listed {
assert!(
full_config(security_type).to_contract().is_ok(),
"{security_type}"
);
}
for security_type in ["STK", "CASH", "FUT", "OPT", "BOND"] {
assert_eq!(
full_config(security_type).to_contract().is_ok(),
listed.contains(&security_type),
"{security_type}"
);
}
}
#[test]
fn validate_config_names_a_kind_this_client_cannot_trade() {
let name = InstrumentNameExchange::new("X");
let instruments = [ClientInstrument::new(
&name,
InstrumentKindDiscriminant::Perpetual,
)];
let config = ibkr_config(vec![named("X", "STK")]);
let error = IbkrClient::validate_config(&config, &instruments).unwrap_err();
assert!(error.contains("perpetual"), "{error}");
assert!(error.contains("cannot trade"), "{error}");
}
#[test]
fn validate_config_accepts_entries_and_instruments_without_a_counterpart() {
let name = InstrumentNameExchange::new("MSFT");
let instruments = [ClientInstrument::new(
&name,
InstrumentKindDiscriminant::Spot,
)];
let config = ibkr_config(vec![named("SOMETHING-ELSE", "OPT")]);
assert_eq!(IbkrClient::validate_config(&config, &instruments), Ok(()));
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod order_status_tests {
use super::*;
use crate::order::state::ActiveOrderState;
use ibapi::orders::{OrderStatus, OrderStatusKind, PlaceOrder};
fn status(kind: OrderStatusKind, filled: f64) -> Result<PlaceOrder, ibapi::Error> {
Ok(PlaceOrder::OrderStatus(OrderStatus {
order_id: 42,
status: kind,
filled,
..OrderStatus::default()
}))
}
fn ctx() -> OrderContext {
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(Decimal::from(150)),
quantity: Decimal::from(10),
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
}
}
#[test]
fn unknown_status_is_held_pending_during_placement() {
let events = vec![status(OrderStatusKind::Unknown("Reclassified".into()), 0.0)];
match await_order_placement(events) {
PlacementOutcome::HeldPending(reason) => {
assert!(
reason.contains("Reclassified"),
"the raw status must reach the caller, got {reason:?}"
);
}
other => panic!("expected HeldPending, got {other:?}"),
}
}
#[test]
fn unknown_status_does_not_fall_through_to_no_status() {
let events = vec![status(OrderStatusKind::Unknown("Reclassified".into()), 0.0)];
assert!(!matches!(
await_order_placement(events),
PlacementOutcome::NoStatus
));
let events = vec![status(OrderStatusKind::ApiPending, 0.0)];
assert!(matches!(
await_order_placement(events),
PlacementOutcome::NoStatus
));
}
#[test]
fn transport_loss_while_awaiting_placement_is_no_status() {
for e in [
ibapi::Error::ConnectionReset,
ibapi::Error::ConnectionFailed,
ibapi::Error::Shutdown,
ibapi::Error::Io(std::io::Error::from(std::io::ErrorKind::BrokenPipe)),
] {
let label = e.to_string();
assert!(
matches!(
await_order_placement(vec![Err(e)]),
PlacementOutcome::NoStatus
),
"{label} must leave the placement unresolved, not rejected"
);
}
let events = vec![
status(OrderStatusKind::Submitted, 0.0),
Err(ibapi::Error::ConnectionReset),
];
assert!(matches!(
await_order_placement(events),
PlacementOutcome::Accepted { .. }
));
}
#[test]
fn non_transport_error_while_awaiting_placement_is_rejected() {
let events = vec![Err(ibapi::Error::Simple("boom".into()))];
assert!(matches!(
await_order_placement(events),
PlacementOutcome::Rejected(reason) if reason.contains("boom")
));
}
#[test]
fn send_error_separates_transport_loss_from_refusal() {
let refused: OrderError<AssetNameExchange, InstrumentNameExchange> =
send_error(&ibapi::Error::ConnectionReset, "down".into());
assert!(
matches!(&refused, OrderError::Connectivity(ConnectivityError::Socket(m)) if m == "down")
);
assert!(refused.is_transient());
let io: OrderError<AssetNameExchange, InstrumentNameExchange> = send_error(
&ibapi::Error::Io(std::io::Error::from(std::io::ErrorKind::BrokenPipe)),
"io".into(),
);
assert!(io.is_transient());
for e in [
ibapi::Error::Shutdown,
ibapi::Error::ConnectionFailed,
ibapi::Error::Simple("no".into()),
] {
let label = e.to_string();
let rejected: OrderError<AssetNameExchange, InstrumentNameExchange> =
send_error(&e, "no".into());
assert!(
matches!(&rejected, OrderError::Rejected(ApiError::OrderRejected(m)) if m == "no"),
"{label} must be a non-transient refusal"
);
assert!(!rejected.is_transient(), "{label}");
}
}
#[test]
fn known_statuses_keep_their_placement_outcomes() {
assert!(matches!(
await_order_placement(vec![status(OrderStatusKind::Submitted, 0.0)]),
PlacementOutcome::Accepted { .. }
));
assert!(matches!(
await_order_placement(vec![status(OrderStatusKind::Filled, 10.0)]),
PlacementOutcome::Accepted { filled } if filled == 10.0
));
assert!(matches!(
await_order_placement(vec![status(OrderStatusKind::Inactive, 0.0)]),
PlacementOutcome::Rejected(_)
));
assert!(matches!(
await_order_placement(vec![status(OrderStatusKind::Cancelled, 0.0)]),
PlacementOutcome::Rejected(_)
));
}
#[test]
fn unknown_status_maps_to_active_open() {
let raw = OrderStatus {
order_id: 42,
status: OrderStatusKind::Unknown("Reclassified".into()),
filled: 3.0,
..OrderStatus::default()
};
let order = make_order_from_status(
&raw,
ClientOrderId::new("cid-1"),
&ctx(),
&PendingCancels::new(),
);
assert!(
matches!(order.state, OrderState::Active(_)),
"unmodelled status must stay active, got {:?}",
order.state
);
}
#[test]
fn an_ended_status_carries_its_fill_or_none() {
let ended = |status: OrderStatusKind, filled: f64| {
let raw = OrderStatus {
order_id: 42,
status,
filled,
..OrderStatus::default()
};
make_order_from_status(
&raw,
ClientOrderId::new("cid-1"),
&ctx(),
&PendingCancels::new(),
)
.state
};
let cancelled_fill = |state: OrderState<_, _>| match state {
OrderState::Inactive(crate::order::state::InactiveOrderState::Cancelled(cancelled)) => {
cancelled.filled_quantity
}
other => panic!("expected Cancelled, got {other:?}"),
};
assert_eq!(
cancelled_fill(ended(OrderStatusKind::Cancelled, 3.0)),
Some(Decimal::from(3))
);
assert_eq!(
cancelled_fill(ended(OrderStatusKind::Cancelled, f64::NAN)),
None
);
assert!(matches!(
ended(OrderStatusKind::Filled, f64::NAN),
OrderState::Inactive(crate::order::state::InactiveOrderState::FullyFilled(Filled { filled_quantity, .. }))
if filled_quantity == ctx().quantity
));
}
#[test]
fn a_good_till_date_order_cancelled_at_its_expiry_expired() {
let cancelled_at = |expiry: DateTime<Utc>| {
let ctx = OrderContext {
time_in_force: TimeInForce::GoodTillDate { expiry },
..ctx()
};
let raw = OrderStatus {
order_id: 42,
status: OrderStatusKind::Cancelled,
..OrderStatus::default()
};
make_order_from_status(
&raw,
ClientOrderId::new("cid-1"),
&ctx,
&PendingCancels::new(),
)
.state
};
use crate::order::state::InactiveOrderState;
assert!(matches!(
cancelled_at(Utc::now() - chrono::Duration::minutes(1)),
OrderState::Inactive(InactiveOrderState::Expired(_))
));
assert!(matches!(
cancelled_at(Utc::now() + chrono::Duration::seconds(2)),
OrderState::Inactive(InactiveOrderState::Expired(_))
));
assert!(matches!(
cancelled_at(Utc::now() + chrono::Duration::hours(1)),
OrderState::Inactive(InactiveOrderState::Cancelled(_))
));
}
#[test]
fn unknown_status_leaves_pending_cancel_intact() {
let pending = PendingCancels::new();
pending.insert(42);
let raw = OrderStatus {
order_id: 42,
status: OrderStatusKind::Unknown("Reclassified".into()),
..OrderStatus::default()
};
let _ = make_order_from_status(&raw, ClientOrderId::new("cid-1"), &ctx(), &pending);
assert!(
pending.remove(42),
"the pending cancel must still be there for the confirmed Cancelled"
);
}
#[test]
fn rejected_leg_with_failed_cancel_is_settled() {
let accepted: Option<Result<f64, String>> = Some(Ok(0.0));
let rejected: Option<Result<f64, String>> = Some(Err("Cancelled".into()));
assert_eq!(
unresolved_legs([(1, &rejected), (2, &accepted), (3, &accepted)], &[1, 2, 3]),
vec![2, 3]
);
}
#[test]
fn leg_send_error_is_transient_only_after_a_clean_rollback() {
let clean = leg_send_error(&ibapi::Error::ConnectionReset, "sl failed".into(), "");
assert!(matches!(clean, OrderError::Connectivity(_)));
let dirty = leg_send_error(
&ibapi::Error::ConnectionReset,
"sl failed".into(),
"; rollback cancel failed for order ids 10 (connection reset), which may be live",
);
match dirty {
OrderError::Rejected(ApiError::OrderRejected(message)) => {
assert!(message.starts_with("sl failed; rollback cancel failed"));
}
other => panic!("expected a non-transient rejection, got {other:?}"),
}
}
#[test]
fn unresolved_legs_are_no_status_or_failed_cancel() {
let accepted: Option<Result<f64, String>> = Some(Ok(0.0));
let rejected: Option<Result<f64, String>> = Some(Err("201".into()));
let no_status: Option<Result<f64, String>> = None;
assert_eq!(
unresolved_legs([(1, &accepted), (2, &rejected), (3, &no_status)], &[]),
vec![3]
);
assert_eq!(
unresolved_legs([(1, &accepted), (2, &rejected), (3, &accepted)], &[1]),
vec![1]
);
assert_eq!(
unresolved_legs([(1, &no_status), (2, &no_status), (3, &accepted)], &[2]),
vec![1, 2]
);
assert!(unresolved_legs([(1, &accepted), (2, &rejected), (3, &accepted)], &[]).is_empty());
}
fn bracket_request() -> BracketOrderRequest {
BracketOrderRequest {
instrument: InstrumentNameExchange::new("AAPL"),
strategy: StrategyId::new("test"),
parent_cid: ClientOrderId::new("br"),
side: Side::Buy,
quantity: Decimal::from(10),
entry_price: Decimal::from(150),
take_profit_price: Decimal::from(160),
stop_loss_price: Decimal::from(145),
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
}
}
#[test]
fn failed_bracket_keeps_unresolved_legs_open_and_tracked() {
let request = bracket_request();
let (tp_cid, sl_cid) = derive_child_cids(&request.parent_cid);
let order_ids = OrderIdMap::new();
order_ids
.register(request.parent_cid.clone(), 10, ctx())
.unwrap();
order_ids.register(tp_cid, 11, ctx()).unwrap();
order_ids.register(sl_cid, 12, ctx()).unwrap();
let result = failed_bracket(
&request,
[10, 11, 12],
BracketFailure {
error: OrderError::Rejected(ApiError::OrderRejected("boom".into())),
unresolved: vec![11],
},
&order_ids,
);
assert!(matches!(result.parent.state, OrderState::Inactive(_)));
assert!(matches!(result.stop_loss.state, OrderState::Inactive(_)));
match &result.take_profit.state {
OrderState::Active(ActiveOrderState::Open(open)) => {
assert_eq!(open.id.assigned(), Some(&OrderId::new("11")));
assert_eq!(open.filled_quantity, Decimal::ZERO);
}
other => panic!("unresolved leg must be Open, got {other:?}"),
}
assert!(order_ids.get_client_id(10).is_none());
assert!(order_ids.get_client_id(11).is_some());
assert!(order_ids.get_client_id(12).is_none());
}
#[test]
fn failed_bracket_without_unresolved_legs_is_all_inactive() {
let request = bracket_request();
let (tp_cid, sl_cid) = derive_child_cids(&request.parent_cid);
let order_ids = OrderIdMap::new();
order_ids
.register(request.parent_cid.clone(), 10, ctx())
.unwrap();
order_ids.register(tp_cid, 11, ctx()).unwrap();
order_ids.register(sl_cid, 12, ctx()).unwrap();
let result = failed_bracket(
&request,
[10, 11, 12],
BracketFailure {
error: OrderError::Rejected(ApiError::OrderRejected("boom".into())),
unresolved: Vec::new(),
},
&order_ids,
);
for leg in [&result.parent, &result.take_profit, &result.stop_loss] {
assert!(matches!(leg.state, OrderState::Inactive(_)));
}
for ib_id in [10, 11, 12] {
assert!(order_ids.get_client_id(ib_id).is_none());
}
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used)] mod order_reader_tests {
use super::*;
use crate::client::dedup::new_dedup_cache;
use crate::trade::TradeId;
use ibapi::orders::{OrderStatus, OrderUpdate};
use std::cell::Cell;
fn unforwarded(n: usize) -> Vec<Result<OrderUpdate, ibapi::Error>> {
(0..n)
.map(|_| {
Ok(OrderUpdate::OrderStatus(OrderStatus {
order_id: 999,
..OrderStatus::default()
}))
})
.collect()
}
fn run(sink: &recovery::EventSink, updates: Vec<Result<OrderUpdate, ibapi::Error>>) -> usize {
let pulled = Cell::new(0);
forward_order_updates(
updates
.into_iter()
.inspect(|_| pulled.set(pulled.get() + 1)),
sink,
0,
&ContractRegistry::new(),
&OrderIdMap::new(),
&PendingCancels::new(),
&ExecutionBuffer::new(),
);
pulled.get()
}
#[test]
fn stops_on_the_first_update_after_the_stream_ends() {
let (tx, _rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
sink.terminate(StreamTerminationReason::Error("recovery failed".into()));
assert_eq!(run(&sink, unforwarded(3)), 1);
}
#[test]
fn stops_on_the_first_update_after_the_consumer_goes() {
let (tx, rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
drop(rx);
assert_eq!(run(&sink, unforwarded(3)), 1);
}
#[test]
fn a_terminal_status_frees_the_orders_client_id() {
use ibapi::orders::OrderStatusKind;
for (kind, entry_kept, id_held) in [
(OrderStatusKind::Filled, true, false),
(OrderStatusKind::Cancelled, false, false),
(OrderStatusKind::Inactive, false, false),
(OrderStatusKind::Submitted, true, true),
] {
let (tx, mut rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
let order_ids = OrderIdMap::new();
let cid = ClientOrderId::new("cid-7");
order_ids
.register(
cid.clone(),
7,
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(Decimal::from(100)),
quantity: Decimal::ONE,
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
},
)
.unwrap();
forward_order_updates(
[Ok(OrderUpdate::OrderStatus(OrderStatus {
order_id: 7,
status: kind.clone(),
..OrderStatus::default()
}))],
&sink,
0,
&ContractRegistry::new(),
&order_ids,
&PendingCancels::new(),
&ExecutionBuffer::new(),
);
assert!(rx.try_recv().is_ok(), "{kind:?}: the status is forwarded");
assert_eq!(
order_ids.get_client_id(7).is_some(),
entry_kept,
"{kind:?}: entry"
);
assert_eq!(order_ids.get_ib_id(&cid).is_some(), id_held, "{kind:?}: id");
}
}
#[test]
fn another_api_clients_status_does_not_reach_a_tracked_order() {
use ibapi::orders::OrderStatusKind;
let (tx, mut rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
let order_ids = OrderIdMap::new();
let cid = ClientOrderId::new("cid-7");
order_ids
.register(
cid.clone(),
7,
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(Decimal::from(100)),
quantity: Decimal::ONE,
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
},
)
.unwrap();
forward_order_updates(
[Ok(OrderUpdate::OrderStatus(OrderStatus {
order_id: 7,
client_id: 905,
status: OrderStatusKind::Cancelled,
..OrderStatus::default()
}))],
&sink,
903,
&ContractRegistry::new(),
&order_ids,
&PendingCancels::new(),
&ExecutionBuffer::new(),
);
assert!(
matches!(
rx.try_recv().map(|event| event.kind),
Ok(AccountEventKind::StreamTerminated(_))
),
"nothing forwarded before the stream ended"
);
assert_eq!(order_ids.get_ib_id(&cid), Some(7));
assert!(order_ids.contains(7));
}
#[test]
fn a_resent_status_for_an_ended_order_does_not_reach_a_reused_id() {
use ibapi::orders::OrderStatusKind;
let (tx, mut rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
let order_ids = OrderIdMap::new();
let cid = ClientOrderId::new("reused");
let ctx = || OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(Decimal::from(100)),
quantity: Decimal::ONE,
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
};
let status = |order_id, status| {
Ok(OrderUpdate::OrderStatus(OrderStatus {
order_id,
status,
..OrderStatus::default()
}))
};
order_ids.register(cid.clone(), 1, ctx()).unwrap();
let updates = [status(1, OrderStatusKind::Filled)]
.into_iter()
.chain(std::iter::once_with(|| {
order_ids.register(cid.clone(), 2, ctx()).unwrap();
status(1, OrderStatusKind::Filled)
}))
.chain([status(2, OrderStatusKind::Submitted)]);
forward_order_updates(
updates,
&sink,
0,
&ContractRegistry::new(),
&order_ids,
&PendingCancels::new(),
&ExecutionBuffer::new(),
);
let Ok(first) = rx.try_recv() else {
panic!("the first order's fill is forwarded");
};
assert!(
matches!(first.kind, AccountEventKind::OrderSnapshot(Snapshot(ref order)) if !matches!(order.state, OrderState::Active(_))),
"{first:?}"
);
let Ok(event) = rx.try_recv() else {
panic!("the later order's status is forwarded");
};
let AccountEventKind::OrderSnapshot(Snapshot(order)) = event.kind else {
panic!("an order snapshot, got {event:?}");
};
assert!(
matches!(order.state, OrderState::Active(_)),
"it is the later, working order's: {order:?}"
);
assert!(
matches!(
rx.try_recv().map(|event| event.kind),
Ok(AccountEventKind::StreamTerminated(_))
),
"and the re-sent fill was dropped: nothing else before the stream ended"
);
assert_eq!(order_ids.get_ib_id(&cid), Some(2));
}
#[test]
fn a_corrected_execution_is_reported_as_an_amendment() {
use crate::trade::TradeAmendmentKind;
use ibapi::{
contracts::Contract,
orders::{CommissionReport, Execution, ExecutionData},
};
let contracts = ContractRegistry::new();
contracts
.register(
InstrumentNameExchange::new("AAPL"),
Contract {
contract_id: 265598,
..Contract::default()
},
)
.unwrap();
let order_ids = OrderIdMap::new();
order_ids
.register(
ClientOrderId::new("cid-7"),
7,
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(Decimal::from(100)),
quantity: Decimal::ONE,
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
},
)
.unwrap();
let execution = |exec_id: &str, price: f64| {
Ok(OrderUpdate::ExecutionData(ExecutionData {
request_id: -1,
contract: Contract {
contract_id: 265598,
..Contract::default()
},
execution: Execution {
order_id: 7,
execution_id: exec_id.to_string(),
time: "20261005 14:00:00 UTC".to_string(),
shares: 1.0,
price,
cumulative_quantity: 1.0,
..Execution::default()
},
}))
};
let commission = |exec_id: &str| {
Ok(OrderUpdate::CommissionReport(CommissionReport {
execution_id: exec_id.to_string(),
commission: 1.0,
currency: "USD".to_string(),
..CommissionReport::default()
}))
};
let (tx, mut rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
forward_order_updates(
[
execution("0000e0d5.5f8b1c2a.01.01", 100.0),
commission("0000e0d5.5f8b1c2a.01.01"),
execution("0000e0d5.5f8b1c2a.01.02", 100.5),
commission("0000e0d5.5f8b1c2a.01.02"),
],
&sink,
0,
&contracts,
&order_ids,
&PendingCancels::new(),
&ExecutionBuffer::new(),
);
let AccountEventKind::Trade(original) = rx.try_recv().unwrap().kind else {
panic!("expected the original as a trade");
};
assert_eq!(original.id, TradeId::new("0000e0d5.5f8b1c2a.01.01"));
let AccountEventKind::TradeAmended(amendment) = rx.try_recv().unwrap().kind else {
panic!("expected the correction as an amendment");
};
assert_eq!(amendment.original, Some(original.id));
assert_eq!(amendment.order_id, OrderId::new("7"));
let TradeAmendmentKind::Corrected { replacement } = amendment.kind else {
panic!("expected a resolved correction, got {:?}", amendment.kind);
};
assert_eq!(replacement.id, TradeId::new("0000e0d5.5f8b1c2a.01.02"));
assert_eq!(replacement.price, Decimal::new(1005, 1));
assert_eq!(replacement.fees.fees, Decimal::ONE);
assert!(matches!(
rx.try_recv().unwrap().kind,
AccountEventKind::StreamTerminated(_)
));
}
#[test]
fn a_shutdown_terminates_the_stream_as_one() {
let (tx, mut rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
let mut updates = unforwarded(1);
updates.push(Err(ibapi::Error::Shutdown));
updates.extend(unforwarded(1));
assert_eq!(run(&sink, updates), 2, "nothing is read past the shutdown");
assert!(matches!(
rx.try_recv().unwrap().kind,
AccountEventKind::StreamTerminated(StreamTerminationReason::Error(ref reason))
if reason == recovery::CLIENT_SHUT_DOWN
));
assert!(
rx.try_recv().is_err(),
"the stream ends with its termination"
);
}
#[test]
fn open_stream_reads_past_unforwarded_updates() {
let (tx, mut rx) = mpsc::unbounded_channel();
let sink = recovery::EventSink::new(
tx,
new_dedup_cache(),
KnownLiveOrders::shared(ExchangeId::Ibkr),
);
assert_eq!(run(&sink, unforwarded(3)), 3);
assert!(matches!(
rx.try_recv().unwrap().kind,
AccountEventKind::StreamTerminated(StreamTerminationReason::Error(ref reason))
if reason == "IBKR order-update stream ended"
));
}
mod listings {
use super::*;
use ibapi::{
contracts::Contract,
orders::{
CommissionReport, Execution, ExecutionData, Executions, OrderData,
OrderState as IbOrderState, OrderStatusKind, Orders,
},
subscriptions::SubscriptionItem,
};
use rust_decimal_macros::dec;
use std::time::Duration;
const API_CLIENT: i32 = 903;
const CON_ID: i32 = 265598;
fn contracts() -> ContractRegistry {
let contracts = ContractRegistry::new();
contracts
.register(
InstrumentNameExchange::new("AAPL"),
Contract {
contract_id: CON_ID,
..Contract::default()
},
)
.unwrap();
contracts
}
fn subscription<T>(
items: Vec<Result<SubscriptionItem<T>, ibapi::Error>>,
stalls: bool,
) -> impl FnMut(Duration) -> Option<Result<SubscriptionItem<T>, ibapi::Error>> {
let mut items = items.into_iter();
move |timeout| {
items.next().or_else(|| {
if stalls {
std::thread::sleep(timeout);
}
None
})
}
}
fn read(
items: Vec<Result<SubscriptionItem<u32>, ibapi::Error>>,
stalls: bool,
stop_at: Option<u32>,
) -> Result<Vec<u32>, UnindexedClientError> {
read_while(items, stalls, stop_at, true)
}
fn read_while(
items: Vec<Result<SubscriptionItem<u32>, ibapi::Error>>,
stalls: bool,
stop_at: Option<u32>,
connected: bool,
) -> Result<Vec<u32>, UnindexedClientError> {
let mut read = Vec::new();
read_listing(
"test",
Duration::from_millis(40),
|| connected,
subscription(items, stalls),
|item| {
read.push(item);
if Some(item) == stop_at {
ControlFlow::Break(())
} else {
ControlFlow::Continue(())
}
},
)?;
Ok(read)
}
#[test]
fn a_listing_read_ends_at_the_end_of_the_listing() {
let items = || vec![Ok(SubscriptionItem::Data(1)), Ok(SubscriptionItem::Data(2))];
assert_eq!(read(items(), false, None).unwrap(), [1, 2]);
assert_eq!(
read(items(), false, Some(1)).unwrap(),
[1],
"or where it breaks"
);
}
#[test]
fn a_stalled_or_failed_listing_read_is_an_error() {
let stalled = read(vec![Ok(SubscriptionItem::Data(1))], true, None);
assert!(
matches!(&stalled, Err(UnindexedClientError::Internal(m)) if m.contains("sent nothing")),
"{stalled:?}"
);
let failed = read(
vec![
Ok(SubscriptionItem::Data(1)),
Err(ibapi::Error::ConnectionReset),
],
false,
None,
);
assert!(
matches!(failed, Err(UnindexedClientError::Internal(_))),
"{failed:?}"
);
}
#[test]
fn a_listing_that_stops_while_disconnected_is_an_error() {
let read = read_while(vec![Ok(SubscriptionItem::Data(1))], false, None, false);
assert!(
matches!(&read, Err(UnindexedClientError::Internal(m)) if m.contains("connection dropped")),
"{read:?}"
);
}
fn contract(contract_id: i32) -> Contract {
Contract {
contract_id,
..Contract::default()
}
}
fn ib_notice(code: i32) -> ListingError {
ListingError::Ibapi(ibapi::Error::Notice(ibapi::Notice {
request_id: Some(9),
code,
message: "notice".to_string(),
error_time: None,
advanced_order_reject_json: String::new(),
}))
}
#[test]
fn a_contract_description_resolves_only_to_a_single_match() {
assert_eq!(
classify_contract_details(Ok(()), vec![contract(1)]),
Ok(contract(1))
);
assert_eq!(
classify_contract_details(Ok(()), Vec::new()),
Err(ResolveContractError::NoMatch)
);
assert_eq!(
classify_contract_details(Ok(()), vec![contract(1), contract(2)]),
Err(ResolveContractError::Ambiguous {
matches: vec![contract(1), contract(2)]
})
);
}
#[test]
fn a_failed_contract_details_read_says_whether_a_retry_can_help() {
let classify = |failure| classify_contract_details(Err(failure), vec![contract(1)]);
assert_eq!(
classify(ib_notice(NO_SECURITY_DEFINITION_CODE)),
Err(ResolveContractError::NoMatch)
);
assert_eq!(
classify(ib_notice(321)),
Err(ResolveContractError::Refused {
code: 321,
message: "notice".to_string()
})
);
let transient = [
(
ListingError::Dropped,
ConnectivityError::Socket(ListingError::Dropped.to_string()),
),
(
ListingError::Stalled(Duration::from_secs(10)),
ConnectivityError::Timeout,
),
(
ListingError::Ibapi(ibapi::Error::ConnectionReset),
ConnectivityError::Socket(ibapi::Error::ConnectionReset.to_string()),
),
];
for (failure, connectivity) in transient {
let classified = classify(failure);
assert_eq!(
classified,
Err(ResolveContractError::Connectivity(connectivity))
);
assert!(classified.unwrap_err().is_transient());
}
for permanent in [ibapi::Error::ConnectionFailed, ibapi::Error::Shutdown] {
let classified = classify(ListingError::Ibapi(permanent));
assert!(
matches!(&classified, Err(ResolveContractError::Failed(_))),
"{classified:?}"
);
assert!(!classified.unwrap_err().is_transient());
}
assert!(
!ResolveContractError::Refused {
code: 321,
message: String::new()
}
.is_transient()
);
assert!(!ResolveContractError::NoMatch.is_transient());
}
#[test]
fn an_account_summary_read_needs_its_end_item() {
use ibapi::accounts::{AccountSummary, AccountSummaryResult};
let summary = || {
Ok(SubscriptionItem::Data(AccountSummaryResult::Summary(
AccountSummary {
account: "DU1".to_string(),
tag: "TotalCashValue".to_string(),
value: "100".to_string(),
currency: "USD".to_string(),
},
)))
};
let read = |items| {
read_account_summary(
Duration::from_millis(40),
|| true,
subscription(items, false),
)
};
let balances = read(vec![
summary(),
Ok(SubscriptionItem::Data(AccountSummaryResult::End)),
])
.unwrap()
.to_balances();
assert_eq!(balances.len(), 1, "{balances:?}");
let cut_short = read(vec![summary()]);
assert!(
matches!(cut_short, Err(UnindexedClientError::Internal(_))),
"{cut_short:?}"
);
}
fn listed(api_client_id: i32, order_id: i32, order: ibapi::orders::Order) -> Orders {
Orders::OrderData(OrderData {
order_id,
contract: Contract {
contract_id: CON_ID,
..Contract::default()
},
order: ibapi::orders::Order {
client_id: api_client_id,
order_id,
..order
},
order_state: IbOrderState {
status: OrderStatusKind::Submitted,
..IbOrderState::default()
},
})
}
fn status(
api_client_id: i32,
order_id: i32,
status: OrderStatusKind,
filled: f64,
) -> Orders {
Orders::OrderStatus(OrderStatus {
order_id,
client_id: api_client_id,
status,
filled,
..OrderStatus::default()
})
}
fn stop_order() -> ibapi::orders::Order {
ibapi::orders::Order {
action: ibapi::orders::Action::Sell,
order_type: "STP".to_owned(),
total_quantity: 3.0,
aux_price: Some(140.5),
tif: ibapi::orders::TimeInForce::Day,
..ibapi::orders::Order::default()
}
}
#[test]
fn open_orders_are_this_api_clients_with_their_fills() {
let order_ids = OrderIdMap::new();
order_ids
.register(
ClientOrderId::new("tracked"),
2,
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(dec!(150)),
quantity: dec!(10),
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
},
)
.unwrap();
let listed_orders = open_orders_from_listing(
[
listed(API_CLIENT, 5, stop_order()),
status(API_CLIENT, 5, OrderStatusKind::PreSubmitted, 0.0),
listed(API_CLIENT, 2, ibapi::orders::Order::default()),
status(API_CLIENT, 2, OrderStatusKind::Submitted, 4.0),
listed(904, 2, stop_order()),
status(904, 2, OrderStatusKind::Submitted, 9.0),
],
API_CLIENT,
&contracts(),
&order_ids,
);
assert_eq!(listed_orders.len(), 2, "{listed_orders:?}");
let exact: Vec<_> = listed_orders
.iter()
.map(ListedOpenOrder::is_exact)
.collect();
assert_eq!(exact, [true, false]);
let orders: Vec<_> = listed_orders
.into_iter()
.map(|listed| listed.order.unwrap())
.collect();
let (tracked, untracked) = (&orders[0], &orders[1]);
assert_eq!(tracked.key.cid, ClientOrderId::new("tracked"));
assert_eq!(tracked.kind, OrderKind::Limit);
assert_eq!(tracked.price, Some(dec!(150)));
assert_eq!(tracked.quantity, dec!(10));
assert_eq!(tracked.state.id, VenueOrderId::Assigned(OrderId::new("2")));
assert_eq!(tracked.state.filled_quantity, dec!(4));
assert_eq!(untracked.key.cid, ClientOrderId::new("5"));
assert_eq!(
untracked.key.instrument,
InstrumentNameExchange::new("AAPL")
);
assert_eq!(untracked.side, Side::Sell);
assert_eq!(untracked.quantity, dec!(3));
assert_eq!(
untracked.kind,
OrderKind::Stop {
trigger_price: dec!(140.5)
}
);
assert_eq!(untracked.time_in_force, TimeInForce::GoodUntilEndOfDay);
assert_eq!(
untracked.state.id,
VenueOrderId::Assigned(OrderId::new("5"))
);
assert_eq!(untracked.state.filled_quantity, Decimal::ZERO);
}
#[test]
fn open_orders_leave_out_what_cannot_be_reported_open() {
let unregistered = Orders::OrderData(OrderData {
order_id: 3,
contract: Contract {
contract_id: 1,
..Contract::default()
},
order: ibapi::orders::Order {
client_id: API_CLIENT,
..stop_order()
},
order_state: IbOrderState::default(),
});
let unreadable = ibapi::orders::Order {
order_type: "REL".to_owned(),
..stop_order()
};
let orders = open_orders_from_listing(
[
listed(API_CLIENT, 1, stop_order()),
status(API_CLIENT, 1, OrderStatusKind::Cancelled, 0.0),
unregistered,
listed(API_CLIENT, 4, unreadable),
],
API_CLIENT,
&contracts(),
&OrderIdMap::new(),
);
assert_eq!(orders.len(), 1, "{orders:?}");
assert!(orders[0].order.is_none());
assert!(!orders[0].is_exact());
}
fn referenced(order_ref: &str) -> ibapi::orders::Order {
ibapi::orders::Order {
order_ref: order_ref.to_owned(),
..stop_order()
}
}
#[test]
fn an_untracked_order_is_named_by_its_reference_and_adopted() {
let order_ids = OrderIdMap::new();
let listed = open_orders_from_listing(
[
listed(API_CLIENT, 7, referenced("before-restart")),
status(API_CLIENT, 7, OrderStatusKind::Submitted, 1.0),
],
API_CLIENT,
&contracts(),
&order_ids,
);
assert_eq!(listed.len(), 1, "{listed:?}");
assert!(listed[0].is_exact());
let order = listed[0].order.as_ref().unwrap();
assert_eq!(order.key.cid, ClientOrderId::new("before-restart"));
assert_eq!(order.state.filled_quantity, dec!(1));
assert_eq!(
order_ids.get_ib_id(&ClientOrderId::new("before-restart")),
Some(7)
);
let (_, ctx) = order_ids.get_client_id_and_context(7).unwrap();
assert_eq!(ctx.quantity, dec!(3));
}
#[test]
fn a_reused_ib_order_id_does_not_name_the_listed_order() {
let order_ids = OrderIdMap::new();
order_ids
.register(
ClientOrderId::new("earlier"),
7,
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(dec!(150)),
quantity: dec!(10),
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
},
)
.unwrap();
let listed = open_orders_from_listing(
[
listed(API_CLIENT, 7, referenced("later")),
status(API_CLIENT, 7, OrderStatusKind::Submitted, 0.0),
],
API_CLIENT,
&contracts(),
&order_ids,
);
let order = listed[0].order.as_ref().unwrap();
assert_eq!(order.key.cid, ClientOrderId::new("7"));
assert!(!listed[0].is_exact());
assert_eq!(
order.kind,
OrderKind::Stop {
trigger_price: dec!(140.5)
}
);
assert_eq!(order_ids.get_ib_id(&ClientOrderId::new("earlier")), Some(7));
assert_eq!(order_ids.get_ib_id(&ClientOrderId::new("later")), None);
}
#[test]
fn listed_cids_name_this_clients_orders_without_adopting_them() {
let order_ids = OrderIdMap::new();
order_ids
.register(
ClientOrderId::new("tracked"),
2,
OrderContext {
instrument: InstrumentNameExchange::new("AAPL"),
side: Side::Buy,
price: Some(dec!(150)),
quantity: dec!(10),
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
},
)
.unwrap();
let cids = listed_cids(
[
listed(API_CLIENT, 1, referenced("ended")),
status(API_CLIENT, 1, OrderStatusKind::Cancelled, 0.0),
listed(API_CLIENT, 2, stop_order()),
listed(API_CLIENT, 3, stop_order()),
listed(904, 4, referenced("other-client")),
],
API_CLIENT,
&order_ids,
);
assert_eq!(
cids,
fnv::FnvHashSet::from_iter([
ClientOrderId::new("ended"),
ClientOrderId::new("tracked")
])
);
assert_eq!(order_ids.get_ib_id(&ClientOrderId::new("ended")), None);
}
#[test]
fn snapshot_orders_are_complete_only_when_each_is_exact() {
let msft = InstrumentNameExchange::new("MSFT");
let contracts = contracts();
contracts
.register(
msft.clone(),
Contract {
contract_id: 272093,
..Contract::default()
},
)
.unwrap();
let in_msft = |order_id, order| {
let Orders::OrderData(mut data) = listed(API_CLIENT, order_id, order) else {
unreachable!()
};
data.contract.contract_id = 272093;
Orders::OrderData(data)
};
let listed_orders = open_orders_from_listing(
[
listed(API_CLIENT, 1, referenced("a")),
status(API_CLIENT, 1, OrderStatusKind::Submitted, 0.0),
in_msft(2, stop_order()),
status(API_CLIENT, 2, OrderStatusKind::Submitted, 0.0),
in_msft(3, referenced("b")),
status(API_CLIENT, 3, OrderStatusKind::Submitted, 0.0),
],
API_CLIENT,
&contracts,
&OrderIdMap::new(),
);
let mut orders = SnapshotOrders::new(Some(listed_orders));
let (aapl_orders, aapl_complete) = orders.take(&InstrumentNameExchange::new("AAPL"));
assert_eq!(aapl_orders.len(), 1);
assert!(aapl_complete);
let (msft_orders, msft_complete) = orders.take(&msft);
assert_eq!(msft_orders.len(), 2);
assert!(!msft_complete);
assert_eq!(
orders.take(&InstrumentNameExchange::new("TSLA")),
(vec![], true)
);
let mut unread = SnapshotOrders::new(None);
assert_eq!(unread.take(&msft), (vec![], false));
}
fn executed(api_client_id: i32, order_id: i32, exec_id: &str, time: &str) -> Executions {
Executions::ExecutionData(ExecutionData {
request_id: 9001,
contract: Contract {
contract_id: CON_ID,
..Contract::default()
},
execution: Execution {
order_id,
client_id: api_client_id,
execution_id: exec_id.to_string(),
time: time.to_string(),
shares: 1.0,
price: 100.0,
cumulative_quantity: 1.0,
..Execution::default()
},
})
}
fn commission(exec_id: &str) -> Executions {
Executions::CommissionReport(CommissionReport {
execution_id: exec_id.to_string(),
commission: 1.25,
currency: "USD".to_string(),
..CommissionReport::default()
})
}
#[test]
fn trades_carry_their_ib_order_id_and_commission() {
let at = |s: &str| s.parse::<DateTime<Utc>>().unwrap();
let trades = trades_from_executions(
[
executed(API_CLIENT, 7, "paired", "20261007 10:00:02 UTC"),
commission("paired"),
executed(API_CLIENT, 7, "unpaired", "20261007 10:00:01 UTC"),
executed(API_CLIENT, 7, "too-early", "20261007 09:59:59 UTC"),
commission("too-early"),
executed(905, 7, "other-client", "20261007 10:00:03 UTC"),
commission("other-client"),
],
API_CLIENT,
&contracts(),
at("2026-10-07T10:00:00Z")..=at("2026-10-07T11:00:00Z"),
None,
);
let read: Vec<_> = trades
.iter()
.map(|trade| {
(
trade.id.0.as_str(),
trade.order_id.0.as_str(),
trade.fees.asset.as_ref(),
trade.fees.fees,
)
})
.collect();
assert_eq!(
read,
[
("unpaired", "7", execution::UNKNOWN_FEE_ASSET, Decimal::ZERO),
("paired", "7", "USD", dec!(1.25)),
]
);
let filtered = trades_from_executions(
[
executed(API_CLIENT, 7, "paired", "20261007 10:00:00 UTC"),
commission("paired"),
],
API_CLIENT,
&contracts(),
at("2026-10-07T10:00:00Z")..=at("2026-10-07T11:00:00Z"),
Some(&HashSet::from([InstrumentNameExchange::new("MSFT")])),
);
assert!(filtered.is_empty(), "{filtered:?}");
}
}
}