use std::{
future::Future,
sync::Arc,
time::{Duration, Instant},
};
use ahash::AHashMap;
use anyhow::Context;
use async_trait::async_trait;
use futures_util::{StreamExt, pin_mut};
use nautilus_common::{
clients::ExecutionClient,
live::{get_runtime, runner::get_exec_event_sender, task::TaskHandles},
messages::execution::{
BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
SubmitOrderList,
},
};
use nautilus_core::{
UnixNanos,
params::Params,
time::{AtomicTime, get_atomic_clock_realtime},
};
use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter};
use nautilus_model::{
accounts::AccountAny,
enums::{AccountType, OmsType, OrderSide, OrderType, TimeInForce, TrailingOffsetType},
events::OrderDeniedReason,
identifiers::{
AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, TraderId, Venue, VenueOrderId,
},
instruments::InstrumentAny,
orders::Order,
reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
types::{AccountBalance, MarginBalance, Money, Quantity},
};
use rust_decimal::Decimal;
use tokio::task::JoinHandle;
use ustr::Ustr;
use crate::{
common::{
consts::{
OKX_CONDITIONAL_ORDER_TYPES, OKX_SUCCESS_CODE, OKX_VENUE, OKX_WS_HEARTBEAT_SECS,
resolve_instrument_families, should_retry_error_code, validate_okx_client_order_id,
},
enums::{OKXInstrumentType, OKXMarginMode, OKXTradeMode, is_advance_algo_order},
parse::{is_okx_spread_symbol, nanos_to_datetime, okx_instrument_type_from_symbol},
},
config::OKXExecClientConfig,
http::{client::OKXHttpClient, error::OKXHttpError, models::OKXCancelAlgoOrderRequest},
websocket::{
client::OKXWebSocketClient,
dispatch::{
AlgoCancelContext, OrderIdentity, WsDispatchState, dispatch_ws_message,
emit_algo_cancel_rejections, emit_batch_cancel_failure,
},
error::OKXWsError,
parse::OrderStateSnapshot,
},
};
#[derive(Debug)]
pub struct OKXExecutionClient {
core: ExecutionClientCore,
clock: &'static AtomicTime,
config: OKXExecClientConfig,
emitter: ExecutionEventEmitter,
http_client: OKXHttpClient,
ws_private: OKXWebSocketClient,
ws_business: OKXWebSocketClient,
trade_mode: OKXTradeMode,
ws_stream_handle: Option<JoinHandle<()>>,
ws_business_stream_handle: Option<JoinHandle<()>>,
ws_dispatch_state: Arc<WsDispatchState>,
pending_tasks: TaskHandles,
}
impl OKXExecutionClient {
pub fn new(core: ExecutionClientCore, config: OKXExecClientConfig) -> anyhow::Result<Self> {
let http_client = OKXHttpClient::with_credentials(
config.api_key.clone(),
config.api_secret.clone(),
config.api_passphrase.clone(),
Some(config.http_base_url()),
config.http_timeout_secs,
config.max_retries,
config.retry_delay_initial_ms,
config.retry_delay_max_ms,
config.environment,
config.proxy_url.clone(),
)?;
let account_id = core.account_id;
let ws_private = OKXWebSocketClient::with_credentials(
Some(config.ws_private_url()),
config.api_key.clone(),
config.api_secret.clone(),
config.api_passphrase.clone(),
Some(account_id),
Some(OKX_WS_HEARTBEAT_SECS),
config.auth_timeout_secs,
config.transport_backend,
config.proxy_url.clone(),
)
.context("failed to construct OKX private websocket client")?;
let ws_business = OKXWebSocketClient::with_credentials(
Some(config.ws_business_url()),
config.api_key.clone(),
config.api_secret.clone(),
config.api_passphrase.clone(),
Some(account_id),
Some(OKX_WS_HEARTBEAT_SECS),
config.auth_timeout_secs,
config.transport_backend,
config.proxy_url.clone(),
)
.context("failed to construct OKX business websocket client")?;
let trade_mode = Self::derive_default_trade_mode(core.account_type, &config);
let clock = get_atomic_clock_realtime();
let emitter = ExecutionEventEmitter::new(
clock,
core.trader_id,
core.account_id,
core.account_type,
None,
);
let ws_dispatch_state = Arc::new(WsDispatchState::with_pending_maps(
ws_private.pending_orders.clone(),
ws_private.pending_cancels.clone(),
ws_private.pending_amends.clone(),
));
Ok(Self {
core,
clock,
config,
emitter,
http_client,
ws_private,
ws_business,
trade_mode,
ws_stream_handle: None,
ws_business_stream_handle: None,
ws_dispatch_state,
pending_tasks: TaskHandles::default(),
})
}
fn derive_default_trade_mode(
account_type: AccountType,
config: &OKXExecClientConfig,
) -> OKXTradeMode {
let is_cross_margin = config.margin_mode == Some(OKXMarginMode::Cross);
if account_type == AccountType::Cash {
if !config.use_spot_margin {
return OKXTradeMode::Cash;
}
return if is_cross_margin {
OKXTradeMode::Cross
} else {
OKXTradeMode::Isolated
};
}
if is_cross_margin {
OKXTradeMode::Cross
} else {
OKXTradeMode::Isolated
}
}
fn trade_mode_for_order(
&self,
instrument_id: InstrumentId,
params: &Option<Params>,
) -> OKXTradeMode {
if let Some(td_mode_str) = get_param_as_string(params, "td_mode") {
match td_mode_str.parse::<OKXTradeMode>() {
Ok(mode) => return mode,
Err(_) => {
log::warn!("Invalid td_mode '{td_mode_str}', using derived trade mode");
}
}
}
derive_trade_mode_for_instrument(
instrument_id,
self.config.margin_mode,
self.config.use_spot_margin,
)
}
fn instrument_types(&self) -> Vec<OKXInstrumentType> {
if self.config.instrument_types.is_empty() {
vec![OKXInstrumentType::Spot]
} else {
self.config.instrument_types.clone()
}
}
fn update_account_state(&self) {
let http_client = self.http_client.clone();
let account_id = self.core.account_id;
let emitter = self.emitter.clone();
self.spawn_task("query_account", async move {
let account_state = http_client
.request_account_state(account_id)
.await
.context("failed to request OKX account state")?;
emitter.send_account_state(account_state);
Ok(())
});
}
fn is_conditional_order(&self, order_type: OrderType) -> bool {
OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
}
fn submit_order_route(
&self,
instrument_id: InstrumentId,
order_type: OrderType,
) -> anyhow::Result<OrderCommandRoute> {
if self.is_conditional_order(order_type) {
if is_spread_instrument(instrument_id) {
anyhow::bail!(
"Trigger/conditional orders ({order_type:?}) are not supported for OKX spreads"
);
}
let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
if inst_type == OKXInstrumentType::Option {
anyhow::bail!(
"Trigger/conditional orders ({order_type:?}) are not supported for OKX options"
);
}
return Ok(OrderCommandRoute::AlgoHttp);
}
if is_spread_instrument(instrument_id) {
Ok(OrderCommandRoute::SpreadHttp)
} else {
Ok(OrderCommandRoute::RegularWs)
}
}
fn cancel_order_route(
&self,
instrument_id: InstrumentId,
order_state: Option<(OrderType, Option<bool>)>,
) -> OrderCommandRoute {
if is_spread_instrument(instrument_id) {
return OrderCommandRoute::SpreadHttp;
}
if order_state.is_some_and(|(order_type, is_triggered)| {
self.is_conditional_order(order_type) && is_triggered != Some(true)
}) {
OrderCommandRoute::AlgoHttp
} else {
OrderCommandRoute::RegularWs
}
}
fn cancel_all_orders_route(&self, instrument_id: InstrumentId) -> CancelAllOrdersRoute {
if is_spread_instrument(instrument_id) {
CancelAllOrdersRoute::SpreadHttp
} else if self.config.use_mm_mass_cancel {
CancelAllOrdersRoute::MassCancelHttp
} else {
CancelAllOrdersRoute::BatchWs
}
}
fn submit_regular_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
let order = {
let cache = self.core.cache();
cache.try_order_owned(&cmd.client_order_id)?
};
let ws_private = self.ws_private.clone();
let trade_mode = self.trade_mode_for_order(cmd.instrument_id, &cmd.params);
let emitter = self.emitter.clone();
let clock = self.clock;
let trader_id = self.core.trader_id;
let client_order_id = order.client_order_id();
let strategy_id = order.strategy_id();
let instrument_id = order.instrument_id();
self.ws_dispatch_state.order_identities.insert(
client_order_id,
OrderIdentity {
instrument_id,
strategy_id,
order_side: order.order_side(),
order_type: order.order_type(),
},
);
let order_side = order.order_side();
let order_type = order.order_type();
let quantity = order.quantity();
let time_in_force = order.time_in_force();
let price = order.price();
let trigger_price = order.trigger_price();
let is_post_only = order.is_post_only();
let is_reduce_only = order.is_reduce_only();
let is_quote_quantity = order.is_quote_quantity();
let px_usd = get_param_as_string(&cmd.params, "px_usd");
let px_vol = get_param_as_string(&cmd.params, "px_vol");
let speed_bump = get_param_as_string(&cmd.params, "speed_bump");
let outcome = get_param_as_string(&cmd.params, "outcome");
let slippage_pct = get_param_as_string(&cmd.params, "slippage_pct");
let rpi = get_param_as_bool(&cmd.params, "rpi");
let rpi_taker_access = get_param_as_bool(&cmd.params, "rpi_taker_access");
let rpi_px_round = get_param_as_bool(&cmd.params, "rpi_px_round");
self.spawn_task("submit_order", async move {
let result = ws_private
.submit_order(
trader_id,
strategy_id,
instrument_id,
trade_mode,
client_order_id,
order_side,
order_type,
quantity,
Some(time_in_force),
price,
trigger_price,
Some(is_post_only),
Some(is_reduce_only),
Some(is_quote_quantity),
None,
None,
px_usd,
px_vol,
speed_bump,
outcome,
slippage_pct,
rpi,
rpi_taker_access,
rpi_px_round,
)
.await;
if let Err(e) = result {
if is_okx_ws_local_command_failure(&e) {
let ts_event = clock.get_time_ns();
emitter.emit_order_rejected_event(
strategy_id,
instrument_id,
client_order_id,
&format!("submit-order-error: {e}"),
ts_event,
false,
);
} else {
log::warn!(
"Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {e}"
);
}
return Err(anyhow::Error::new(e).context("submit order failed"));
}
Ok(())
});
Ok(())
}
fn submit_order_http(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
let order = {
let cache = self.core.cache();
cache.try_order_owned(&cmd.client_order_id)?
};
let http_client = self.http_client.clone();
let trade_mode = self.trade_mode_for_order(cmd.instrument_id, &cmd.params);
let emitter = self.emitter.clone();
let clock = self.clock;
let client_order_id = order.client_order_id();
let strategy_id = order.strategy_id();
let instrument_id = order.instrument_id();
self.ws_dispatch_state.order_identities.insert(
client_order_id,
OrderIdentity {
instrument_id,
strategy_id,
order_side: order.order_side(),
order_type: order.order_type(),
},
);
let order_side = order.order_side();
let order_type = order.order_type();
let quantity = order.quantity();
let time_in_force = order.time_in_force();
let price = order.price();
let is_post_only = order.is_post_only();
let rpi = get_param_as_bool(&cmd.params, "rpi");
let rpi_taker_access = get_param_as_bool(&cmd.params, "rpi_taker_access");
let rpi_px_round = get_param_as_bool(&cmd.params, "rpi_px_round");
self.spawn_task("submit_order_http", async move {
let result = http_client
.place_order_with_domain_types(
instrument_id,
trade_mode,
client_order_id,
order_side,
order_type,
quantity,
Some(time_in_force),
price,
Some(is_post_only),
None,
None,
None,
None,
None,
None,
None,
None,
None,
rpi,
rpi_taker_access,
rpi_px_round,
)
.await;
if let Err(e) = result {
if is_okx_http_submit_rejection(&e) || is_okx_http_local_command_failure(&e) {
let ts_event = clock.get_time_ns();
emitter.emit_order_rejected_event(
strategy_id,
instrument_id,
client_order_id,
&format!("submit-order-error: {e}"),
ts_event,
false,
);
} else {
log::warn!(
"Ambiguous HTTP submit failure for {client_order_id}, awaiting reconciliation: {e}"
);
}
return Err(anyhow::Error::new(e).context("submit order failed"));
}
Ok(())
});
Ok(())
}
fn submit_conditional_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
let order = {
let cache = self.core.cache();
cache.try_order_owned(&cmd.client_order_id)?
};
let http_client = self.http_client.clone();
let trade_mode = self.trade_mode_for_order(cmd.instrument_id, &cmd.params);
let emitter = self.emitter.clone();
let clock = self.clock;
let client_order_id = order.client_order_id();
let strategy_id = order.strategy_id();
let instrument_id = order.instrument_id();
let order_side = order.order_side();
let order_type = order.order_type();
self.ws_dispatch_state.order_identities.insert(
client_order_id,
OrderIdentity {
instrument_id,
strategy_id,
order_side,
order_type,
},
);
let quantity = order.quantity();
let trigger_type = order.trigger_type();
let trigger_price = order.trigger_price();
let price = order.price();
let is_reduce_only = order.is_reduce_only();
let trailing_offset = order.trailing_offset();
let trailing_offset_type = order.trailing_offset_type();
let activation_price = order.activation_price();
let close_fraction = get_param_as_string(&cmd.params, "close_fraction");
let reduce_only = if close_fraction.is_some() {
Some(true)
} else {
Some(is_reduce_only)
};
let (callback_ratio, callback_spread) = if order_type == OrderType::TrailingStopMarket {
let offset = trailing_offset
.ok_or_else(|| anyhow::anyhow!("TrailingStopMarket requires trailing_offset"))?;
let offset_type = trailing_offset_type.ok_or_else(|| {
anyhow::anyhow!("TrailingStopMarket requires trailing_offset_type")
})?;
match offset_type {
TrailingOffsetType::BasisPoints => {
let ratio = offset / Decimal::from(10000);
(Some(ratio.to_string()), None)
}
TrailingOffsetType::Price => (None, Some(offset.to_string())),
_ => {
anyhow::bail!("Unsupported trailing_offset_type for OKX: {offset_type:?}");
}
}
} else {
(None, None)
};
self.spawn_task("submit_algo_order", async move {
let result = http_client
.place_algo_order_with_domain_types(
instrument_id,
trade_mode,
client_order_id,
order_side,
order_type,
quantity,
trigger_price,
trigger_type,
price,
reduce_only,
close_fraction,
callback_ratio,
callback_spread,
activation_price,
)
.await;
if let Err(e) = result {
if is_okx_http_submit_rejection(&e) || is_okx_http_local_command_failure(&e) {
let ts_event = clock.get_time_ns();
emitter.emit_order_rejected_event(
strategy_id,
instrument_id,
client_order_id,
&format!("submit-order-error: {e}"),
ts_event,
false,
);
} else {
log::warn!(
"Ambiguous algo submit failure for {client_order_id}, awaiting reconciliation: {e}"
);
}
return Err(anyhow::Error::new(e).context("submit algo order failed"));
}
Ok(())
});
Ok(())
}
fn cancel_ws_order(&self, cmd: &CancelOrder) {
self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
let ws_private = self.ws_private.clone();
let command = cmd.clone();
self.spawn_task("cancel_order", async move {
let result = ws_private
.cancel_order(
command.trader_id,
command.strategy_id,
command.instrument_id,
Some(command.client_order_id),
command.venue_order_id,
)
.await;
if let Err(e) = result {
if is_okx_ws_local_command_failure(&e) {
log::warn!(
"Cancel command failed local validation for {}: {e}",
command.client_order_id
);
} else {
log::warn!(
"Ambiguous cancel failure for {}, awaiting reconciliation: {e}",
command.client_order_id
);
}
return Err(anyhow::Error::new(e).context("cancel order failed"));
}
Ok(())
});
}
fn cancel_order_http(&self, cmd: &CancelOrder) {
self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
let http_client = self.http_client.clone();
let command = cmd.clone();
let emitter = self.emitter.clone();
let clock = self.clock;
self.spawn_task("cancel_order_http", async move {
let result = http_client
.cancel_order(
command.instrument_id,
Some(command.client_order_id),
command.venue_order_id,
)
.await;
if let Err(e) = result {
if is_okx_http_structured_venue_rejection(&e) {
let ts_event = clock.get_time_ns();
emitter.emit_order_cancel_rejected_event(
command.strategy_id,
command.instrument_id,
command.client_order_id,
command.venue_order_id,
&format!("cancel-order-error: {e}"),
ts_event,
);
} else if is_okx_http_local_command_failure(&e) {
log::warn!(
"HTTP cancel command failed local validation for {}: {e}",
command.client_order_id
);
} else {
log::warn!(
"Ambiguous HTTP cancel failure for {}, awaiting reconciliation: {e}",
command.client_order_id
);
}
return Err(anyhow::Error::new(e).context("cancel order failed"));
}
Ok(())
});
}
fn cancel_algo_order(&self, cmd: &CancelOrder) {
let http_client = self.http_client.clone();
let command = cmd.clone();
let emitter = self.emitter.clone();
let clock = self.clock;
let cache = self.core.cache();
let is_advance = cache
.order(&cmd.client_order_id)
.is_some_and(|o| is_advance_algo_order(o.order_type()));
drop(cache);
let request = OKXCancelAlgoOrderRequest {
inst_id: cmd.instrument_id.symbol.to_string(),
inst_id_code: None,
algo_id: cmd.venue_order_id.map(|id| id.to_string()),
algo_cl_ord_id: if cmd.venue_order_id.is_none() {
Some(cmd.client_order_id.to_string())
} else {
None
},
};
self.spawn_task("cancel_algo_order", async move {
let responses = if is_advance {
http_client.cancel_advance_algo_orders(vec![request]).await
} else {
http_client.cancel_algo_orders(vec![request]).await
};
let reject_reason = match &responses {
Err(e) if is_okx_http_structured_venue_rejection(e) => {
Some(format!("cancel-algo-order-error: {e}"))
}
Err(e) if is_okx_http_local_command_failure(e) => {
log::warn!(
"Algo cancel command failed local validation for {}: {e}",
command.client_order_id
);
None
}
Err(e) => {
log::warn!(
"Ambiguous algo cancel failure for {}, awaiting reconciliation: {e}",
command.client_order_id
);
None
}
Ok(resps) => {
resps.first().and_then(|r| {
r.s_code.as_deref().and_then(|code| {
if code == OKX_SUCCESS_CODE {
None
} else {
let msg = r.s_msg.as_deref().unwrap_or("unknown");
Some(format!(
"cancel-algo-order-rejected: s_code={code}, s_msg={msg}"
))
}
})
})
}
};
if let Some(reason) = reject_reason {
let ts_event = clock.get_time_ns();
emitter.emit_order_cancel_rejected_event(
command.strategy_id,
command.instrument_id,
command.client_order_id,
command.venue_order_id,
&reason,
ts_event,
);
anyhow::bail!("{reason}");
}
if let Err(e) = responses {
return Err(anyhow::Error::new(e).context("cancel algo order failed"));
}
Ok(())
});
}
fn mass_cancel_instrument(&self, instrument_id: InstrumentId) {
if is_spread_instrument(instrument_id) {
let http_client = self.http_client.clone();
self.spawn_task("mass_cancel_orders_http", async move {
http_client
.cancel_all_orders(instrument_id)
.await
.map_err(|e| anyhow::anyhow!("Mass cancel orders failed: {e}"))?;
Ok(())
});
return;
}
let ws_private = self.ws_private.clone();
self.spawn_task("mass_cancel_orders", async move {
ws_private.mass_cancel_orders(instrument_id).await?;
Ok(())
});
}
fn ensure_order_identity(
&self,
client_order_id: ClientOrderId,
strategy_id: StrategyId,
instrument_id: InstrumentId,
) {
self.ws_dispatch_state
.order_identities
.entry(client_order_id)
.or_insert_with(|| {
let cache = self.core.cache();
let (order_side, order_type) = cache
.order(&client_order_id)
.map_or((OrderSide::NoOrderSide, OrderType::Market), |o| {
(o.order_side(), o.order_type())
});
drop(cache);
OrderIdentity {
instrument_id,
strategy_id,
order_side,
order_type,
}
});
}
fn spawn_task<F>(&self, description: &'static str, fut: F)
where
F: Future<Output = anyhow::Result<()>> + Send + 'static,
{
let runtime = get_runtime();
let handle = runtime.spawn(async move {
if let Err(e) = fut.await {
log::warn!("{description} failed: {e:?}");
}
});
self.pending_tasks.push(handle);
}
fn dispatch_algo_cancels(&self, items: Vec<(OKXCancelAlgoOrderRequest, AlgoCancelContext)>) {
let mut regular_requests = Vec::new();
let mut regular_contexts = Vec::new();
let mut advance_requests = Vec::new();
let mut advance_contexts = Vec::new();
let cache = self.core.cache();
for (request, ctx) in items {
let is_advance = cache
.order(&ctx.client_order_id)
.is_some_and(|o| is_advance_algo_order(o.order_type()));
if is_advance {
advance_requests.push(request);
advance_contexts.push(ctx);
} else {
regular_requests.push(request);
regular_contexts.push(ctx);
}
}
drop(cache);
if !regular_requests.is_empty() {
let client = self.http_client.clone();
let emitter = self.emitter.clone();
let clock = self.clock;
self.spawn_task("cancel_algo_orders", async move {
match client.cancel_algo_orders(regular_requests).await {
Ok(responses) => {
emit_algo_cancel_rejections(&responses, ®ular_contexts, &emitter, clock);
}
Err(e) => {
if is_okx_http_local_command_failure(&e) {
for ctx in ®ular_contexts {
log::warn!(
"Algo batch cancel command failed local validation for {}: {e}",
ctx.client_order_id
);
}
} else {
let msg = format!("{e}");
emit_batch_cancel_failure(®ular_contexts, &msg, &emitter, clock);
}
return Err(anyhow::Error::new(e).context("cancel algo orders failed"));
}
}
Ok(())
});
}
if !advance_requests.is_empty() {
let client = self.http_client.clone();
let emitter = self.emitter.clone();
let clock = self.clock;
self.spawn_task("cancel_advance_algo_orders", async move {
match client.cancel_advance_algo_orders(advance_requests).await {
Ok(responses) => {
emit_algo_cancel_rejections(&responses, &advance_contexts, &emitter, clock);
}
Err(e) => {
if is_okx_http_local_command_failure(&e) {
for ctx in &advance_contexts {
log::warn!(
"Advance algo batch cancel command failed local validation for {}: {e}",
ctx.client_order_id
);
}
} else {
let msg = format!("{e}");
emit_batch_cancel_failure(&advance_contexts, &msg, &emitter, clock);
}
return Err(
anyhow::Error::new(e).context("cancel advance algo orders failed")
);
}
}
Ok(())
});
}
}
fn abort_pending_tasks(&self) {
self.pending_tasks.abort_all();
}
async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
let account_id = self.core.account_id;
if self.core.cache().account(&account_id).is_some() {
log::info!("Account {account_id} registered");
return Ok(());
}
let start = Instant::now();
let timeout = Duration::from_secs_f64(timeout_secs);
let interval = Duration::from_millis(10);
loop {
tokio::time::sleep(interval).await;
if self.core.cache().account(&account_id).is_some() {
log::info!("Account {account_id} registered");
return Ok(());
}
if start.elapsed() >= timeout {
anyhow::bail!(
"Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
);
}
}
}
}
fn derive_trade_mode_for_instrument(
instrument_id: InstrumentId,
margin_mode: Option<OKXMarginMode>,
use_spot_margin: bool,
) -> OKXTradeMode {
let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
let is_cross_margin = margin_mode == Some(OKXMarginMode::Cross);
match inst_type {
OKXInstrumentType::Spot => {
if use_spot_margin {
if is_cross_margin {
OKXTradeMode::Cross
} else {
OKXTradeMode::Isolated
}
} else {
OKXTradeMode::Cash
}
}
_ => {
if is_cross_margin {
OKXTradeMode::Cross
} else {
OKXTradeMode::Isolated
}
}
}
}
#[async_trait(?Send)]
impl ExecutionClient for OKXExecutionClient {
fn is_connected(&self) -> bool {
self.core.is_connected()
}
fn client_id(&self) -> ClientId {
self.core.client_id
}
fn account_id(&self) -> AccountId {
self.core.account_id
}
fn venue(&self) -> Venue {
*OKX_VENUE
}
fn oms_type(&self) -> OmsType {
self.core.oms_type
}
fn get_account(&self) -> Option<AccountAny> {
self.core.cache().account_owned(&self.core.account_id)
}
async fn connect(&mut self) -> anyhow::Result<()> {
if self.core.is_connected() {
return Ok(());
}
let instrument_types = self.instrument_types();
if !self.core.instruments_initialized() {
let mut all_instruments = Vec::new();
let mut all_inst_id_codes = Vec::new();
for instrument_type in &instrument_types {
let Some(families) =
resolve_instrument_families(&self.config.instrument_families, *instrument_type)
else {
continue;
};
if families.is_empty() {
let (instruments, inst_id_codes) = self
.http_client
.request_instruments(*instrument_type, None)
.await
.with_context(|| {
format!("failed to request OKX instruments for {instrument_type:?}")
})?;
if instruments.is_empty() {
log::warn!("No instruments returned for {instrument_type:?}");
continue;
}
log::debug!(
"Loaded {} {instrument_type:?} instruments",
instruments.len()
);
self.http_client.cache_instruments(&instruments);
all_instruments.extend(instruments);
all_inst_id_codes.extend(inst_id_codes);
} else {
for family in &families {
let (instruments, inst_id_codes) = self
.http_client
.request_instruments(*instrument_type, Some(family.clone()))
.await
.with_context(|| {
format!(
"failed to request OKX instruments for {instrument_type:?} family {family}"
)
})?;
if instruments.is_empty() {
log::warn!(
"No instruments returned for {instrument_type:?} family {family}"
);
continue;
}
log::debug!(
"Loaded {} {instrument_type:?} instruments for family {family}",
instruments.len()
);
self.http_client.cache_instruments(&instruments);
all_instruments.extend(instruments);
all_inst_id_codes.extend(inst_id_codes);
}
}
}
if all_instruments.is_empty() {
anyhow::bail!(
"No instruments loaded for configured types {instrument_types:?}, \
cannot initialize execution client"
);
}
self.ws_private.cache_instruments(&all_instruments);
self.ws_private
.cache_inst_id_codes(all_inst_id_codes.clone());
self.ws_business.cache_instruments(&all_instruments);
self.ws_business.cache_inst_id_codes(all_inst_id_codes);
self.core.set_instruments_initialized();
}
self.ws_private.connect().await?;
self.ws_private.wait_until_active(10.0).await?;
log::info!("Connected to private WebSocket");
if self.ws_stream_handle.is_none() {
let stream = self.ws_private.stream();
let emitter = self.emitter.clone();
let state = Arc::clone(&self.ws_dispatch_state);
let account_id = self.core.account_id;
let instruments = self.ws_private.instruments_cache_arc();
let clock = self.clock;
let handle = get_runtime().spawn(async move {
let mut fee_cache: AHashMap<Ustr, Money> = AHashMap::new();
let mut filled_qty_cache: AHashMap<Ustr, Quantity> = AHashMap::new();
let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
AHashMap::new();
pin_mut!(stream);
while let Some(message) = stream.next().await {
dispatch_ws_message(
message,
&emitter,
&state,
account_id,
&instruments,
&mut fee_cache,
&mut filled_qty_cache,
&mut order_state_cache,
clock,
);
}
});
self.ws_stream_handle = Some(handle);
}
self.ws_business.connect().await?;
self.ws_business.wait_until_active(10.0).await?;
log::info!("Connected to business WebSocket");
if self.ws_business_stream_handle.is_none() {
let stream = self.ws_business.stream();
let emitter = self.emitter.clone();
let state = Arc::clone(&self.ws_dispatch_state);
let account_id = self.core.account_id;
let instruments = self.ws_business.instruments_cache_arc();
let clock = self.clock;
let handle = get_runtime().spawn(async move {
let mut fee_cache: AHashMap<Ustr, Money> = AHashMap::new();
let mut filled_qty_cache: AHashMap<Ustr, Quantity> = AHashMap::new();
let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
AHashMap::new();
pin_mut!(stream);
while let Some(message) = stream.next().await {
dispatch_ws_message(
message,
&emitter,
&state,
account_id,
&instruments,
&mut fee_cache,
&mut filled_qty_cache,
&mut order_state_cache,
clock,
);
}
});
self.ws_business_stream_handle = Some(handle);
}
for inst_type in &instrument_types {
log::debug!("Subscribing to orders channel for {inst_type:?}");
self.ws_private.subscribe_orders(*inst_type).await?;
if self.config.use_fills_channel {
log::debug!("Subscribing to fills channel for {inst_type:?}");
if let Err(e) = self.ws_private.subscribe_fills(*inst_type).await {
log::warn!("Failed to subscribe to fills channel ({inst_type:?}): {e}");
}
}
}
self.ws_private.subscribe_account().await?;
if self.config.load_spreads {
log::debug!("Subscribing to Nitro spread orders channel");
self.ws_business.subscribe_spread_orders().await?;
}
for inst_type in &instrument_types {
if supports_algo_orders(*inst_type) {
self.ws_business.subscribe_orders_algo(*inst_type).await?;
self.ws_business.subscribe_algo_advance(*inst_type).await?;
}
}
let account_state = self
.http_client
.request_account_state(self.core.account_id)
.await
.context("failed to request OKX account state")?;
if !account_state.balances.is_empty() {
log::debug!(
"Received account state with {} balance(s)",
account_state.balances.len()
);
}
self.emitter.send_account_state(account_state);
self.await_account_registered(30.0).await?;
self.core.set_connected();
log::info!("Connected: client_id={}", self.core.client_id);
Ok(())
}
async fn disconnect(&mut self) -> anyhow::Result<()> {
if self.core.is_disconnected() {
return Ok(());
}
self.abort_pending_tasks();
self.http_client.cancel_all_requests();
if let Err(e) = self.ws_private.close().await {
log::warn!("Error closing private websocket: {e:?}");
}
if let Err(e) = self.ws_business.close().await {
log::warn!("Error closing business websocket: {e:?}");
}
if let Some(handle) = self.ws_stream_handle.take() {
handle.abort();
}
if let Some(handle) = self.ws_business_stream_handle.take() {
handle.abort();
}
self.core.set_disconnected();
log::info!("Disconnected: client_id={}", self.core.client_id);
Ok(())
}
fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
self.update_account_state();
Ok(())
}
fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
let http_client = self.http_client.clone();
let account_id = self.core.account_id;
let emitter = self.emitter.clone();
let instrument_id = cmd.instrument_id;
let client_order_id = cmd.client_order_id;
let venue_order_id = cmd.venue_order_id;
let should_query_algo = !is_spread_instrument(instrument_id)
&& supports_algo_orders(okx_instrument_type_from_symbol(
instrument_id.symbol.as_str(),
));
self.spawn_task("query_order", async move {
let mut reports = match http_client
.request_order_status_reports(
account_id,
None,
Some(instrument_id),
None,
None,
false,
None,
)
.await
{
Ok(r) => r,
Err(e) => {
log::error!("OKX query_order failed to fetch orders: {e}");
Vec::new()
}
};
if should_query_algo {
match http_client
.request_algo_order_status_reports(
account_id,
None,
Some(instrument_id),
None,
Some(client_order_id),
None,
None,
)
.await
{
Ok(mut algo) => reports.append(&mut algo),
Err(e) => {
log::warn!("OKX query_order algo lookup failed for {instrument_id}: {e}");
}
}
}
let Some(report) = select_query_order_report(reports, client_order_id, venue_order_id)
else {
log::warn!(
"OKX query_order found no order for client_order_id={client_order_id}, venue_order_id={venue_order_id:?}",
);
return Ok(());
};
emitter.send_order_status_report(report);
Ok(())
});
Ok(())
}
fn generate_account_state(
&self,
balances: Vec<AccountBalance>,
margins: Vec<MarginBalance>,
reported: bool,
ts_event: UnixNanos,
) -> anyhow::Result<()> {
self.emitter
.emit_account_state(balances, margins, reported, ts_event);
Ok(())
}
fn on_instrument(&mut self, instrument: InstrumentAny) {
self.http_client.cache_instrument(instrument.clone());
self.ws_private.cache_instrument(instrument.clone());
self.ws_business.cache_instrument(instrument);
}
fn start(&mut self) -> anyhow::Result<()> {
if self.core.is_started() {
return Ok(());
}
let sender = get_exec_event_sender();
self.emitter.set_sender(sender);
self.core.set_started();
let http_client = self.http_client.clone();
let ws_private = self.ws_private.clone();
let ws_business = self.ws_business.clone();
let instrument_types = self.config.instrument_types.clone();
let instrument_families = self.config.instrument_families.clone();
get_runtime().spawn(async move {
let mut all_instruments = Vec::new();
let mut all_inst_id_codes = Vec::new();
for instrument_type in instrument_types {
let Some(families) =
resolve_instrument_families(&instrument_families, instrument_type)
else {
continue;
};
if families.is_empty() {
match http_client.request_instruments(instrument_type, None).await {
Ok((instruments, inst_id_codes)) => {
if instruments.is_empty() {
log::warn!("No instruments returned for {instrument_type:?}");
continue;
}
http_client.cache_instruments(&instruments);
all_instruments.extend(instruments);
all_inst_id_codes.extend(inst_id_codes);
}
Err(e) => {
log::error!(
"Failed to request instruments for {instrument_type:?}: {e}"
);
}
}
} else {
for family in &families {
match http_client
.request_instruments(instrument_type, Some(family.clone()))
.await
{
Ok((instruments, inst_id_codes)) => {
if instruments.is_empty() {
log::warn!(
"No instruments returned for {instrument_type:?} family {family}"
);
continue;
}
http_client.cache_instruments(&instruments);
all_instruments.extend(instruments);
all_inst_id_codes.extend(inst_id_codes);
}
Err(e) => {
log::error!(
"Failed to request instruments for {instrument_type:?} family {family}: {e}"
);
}
}
}
}
}
if all_instruments.is_empty() {
log::error!(
"Instrument bootstrap yielded no instruments, order submissions will fail"
);
} else {
ws_private.cache_instruments(&all_instruments);
ws_private.cache_inst_id_codes(all_inst_id_codes.clone());
ws_business.cache_instruments(&all_instruments);
ws_business.cache_inst_id_codes(all_inst_id_codes);
log::debug!("Instruments initialized");
}
});
log::info!(
"Started: client_id={}, account_id={}, account_type={:?}, trade_mode={:?}, instrument_types={:?}, use_fills_channel={}, environment={}, proxy_url={:?}",
self.core.client_id,
self.core.account_id,
self.core.account_type,
self.trade_mode,
self.config.instrument_types,
self.config.use_fills_channel,
self.config.environment,
self.config.proxy_url,
);
Ok(())
}
fn stop(&mut self) -> anyhow::Result<()> {
if self.core.is_stopped() {
return Ok(());
}
self.core.set_stopped();
self.core.set_disconnected();
if let Some(handle) = self.ws_stream_handle.take() {
handle.abort();
}
if let Some(handle) = self.ws_business_stream_handle.take() {
handle.abort();
}
self.abort_pending_tasks();
log::info!("Stopped: client_id={}", self.core.client_id);
Ok(())
}
async fn generate_order_status_report(
&self,
cmd: &GenerateOrderStatusReport,
) -> anyhow::Result<Option<OrderStatusReport>> {
let Some(instrument_id) = cmd.instrument_id else {
log::warn!("generate_order_status_report requires instrument_id: {cmd:?}");
return Ok(None);
};
let mut reports = self
.http_client
.request_order_status_reports(
self.core.account_id,
None,
Some(instrument_id),
None,
None,
false,
None,
)
.await?;
if !is_spread_instrument(instrument_id)
&& supports_algo_orders(okx_instrument_type_from_symbol(
instrument_id.symbol.as_str(),
))
{
match self
.http_client
.request_algo_order_status_reports(
self.core.account_id,
None,
Some(instrument_id),
None,
cmd.client_order_id,
None,
None,
)
.await
{
Ok(mut algo_reports) => reports.append(&mut algo_reports),
Err(e) => {
log::warn!(
"Failed to fetch algo order status reports for {instrument_id}: {e}"
);
}
}
}
if let Some(client_order_id) = cmd.client_order_id {
reports.retain(|report| report.client_order_id == Some(client_order_id));
}
if let Some(venue_order_id) = cmd.venue_order_id {
reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
}
Ok(reports.into_iter().next())
}
async fn generate_order_status_reports(
&self,
cmd: &GenerateOrderStatusReports,
) -> anyhow::Result<Vec<OrderStatusReport>> {
let mut reports = Vec::new();
if let Some(instrument_id) = cmd.instrument_id {
let mut fetched = self
.http_client
.request_order_status_reports(
self.core.account_id,
None,
Some(instrument_id),
None,
None,
false,
None,
)
.await?;
reports.append(&mut fetched);
if !is_spread_instrument(instrument_id)
&& supports_algo_orders(okx_instrument_type_from_symbol(
instrument_id.symbol.as_str(),
))
{
match self
.http_client
.request_algo_order_status_reports(
self.core.account_id,
None,
Some(instrument_id),
None,
None,
None,
None,
)
.await
{
Ok(mut algo) => reports.append(&mut algo),
Err(e) => {
log::warn!(
"Failed to fetch algo order status reports for {instrument_id}: {e}"
);
}
}
}
} else {
for inst_type in self.instrument_types() {
let mut fetched = self
.http_client
.request_order_status_reports(
self.core.account_id,
Some(inst_type),
None,
None,
None,
false,
None,
)
.await?;
reports.append(&mut fetched);
if supports_algo_orders(inst_type) {
match self
.http_client
.request_algo_order_status_reports(
self.core.account_id,
Some(inst_type),
None,
None,
None,
None,
None,
)
.await
{
Ok(mut algo) => reports.append(&mut algo),
Err(e) => log::warn!(
"Failed to fetch algo order status reports for {inst_type:?}: {e}"
),
}
}
}
if self.config.load_spreads {
match self
.http_client
.request_order_status_reports(
self.core.account_id,
None,
None,
None,
None,
false,
None,
)
.await
{
Ok(mut spreads) => reports.append(&mut spreads),
Err(e) => log::warn!("Failed to fetch spread order status reports: {e}"),
}
}
}
if cmd.open_only {
reports.retain(|r| r.order_status.is_open());
}
if let Some(start) = cmd.start {
reports.retain(|r| r.ts_last >= start);
}
if let Some(end) = cmd.end {
reports.retain(|r| r.ts_last <= end);
}
Ok(reports)
}
async fn generate_fill_reports(
&self,
cmd: GenerateFillReports,
) -> anyhow::Result<Vec<FillReport>> {
let start_dt = nanos_to_datetime(cmd.start);
let end_dt = nanos_to_datetime(cmd.end);
let mut reports = Vec::new();
if let Some(instrument_id) = cmd.instrument_id {
let mut fetched = self
.http_client
.request_fill_reports(
self.core.account_id,
None,
Some(instrument_id),
start_dt,
end_dt,
None,
)
.await?;
reports.append(&mut fetched);
} else {
for inst_type in self.instrument_types() {
let mut fetched = self
.http_client
.request_fill_reports(
self.core.account_id,
Some(inst_type),
None,
start_dt,
end_dt,
None,
)
.await?;
reports.append(&mut fetched);
}
if self.config.load_spreads {
match self
.http_client
.request_fill_reports(self.core.account_id, None, None, start_dt, end_dt, None)
.await
{
Ok(mut spreads) => reports.append(&mut spreads),
Err(e) => log::warn!("Failed to fetch spread fill reports: {e}"),
}
}
}
if let Some(venue_order_id) = cmd.venue_order_id {
reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
}
Ok(reports)
}
async fn generate_position_status_reports(
&self,
cmd: &GeneratePositionStatusReports,
) -> anyhow::Result<Vec<PositionStatusReport>> {
let mut reports = Vec::new();
if let Some(instrument_id) = cmd.instrument_id {
if is_spread_instrument(instrument_id) {
return Ok(reports);
}
let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
if inst_type != OKXInstrumentType::Spot && inst_type != OKXInstrumentType::Margin {
let mut fetched = self
.http_client
.request_position_status_reports(
self.core.account_id,
None,
Some(instrument_id),
)
.await?;
reports.append(&mut fetched);
}
} else {
for inst_type in self.instrument_types() {
if inst_type == OKXInstrumentType::Spot || inst_type == OKXInstrumentType::Margin {
continue;
}
let mut fetched = self
.http_client
.request_position_status_reports(self.core.account_id, Some(inst_type), None)
.await?;
reports.append(&mut fetched);
}
}
let mut margin_reports = self
.http_client
.request_spot_margin_position_reports(self.core.account_id)
.await?;
if let Some(instrument_id) = cmd.instrument_id {
margin_reports.retain(|report| report.instrument_id == instrument_id);
}
reports.append(&mut margin_reports);
Ok(reports)
}
async fn generate_mass_status(
&self,
lookback_mins: Option<u64>,
) -> anyhow::Result<Option<ExecutionMassStatus>> {
log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
let ts_now = self.clock.get_time_ns();
let start = lookback_mins.map(|mins| {
let lookback_ns = mins * 60 * 1_000_000_000;
UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns))
});
let order_cmd = GenerateOrderStatusReportsBuilder::default()
.ts_init(ts_now)
.open_only(false) .start(start)
.build()
.map_err(|e| anyhow::anyhow!("{e}"))?;
let fill_cmd = GenerateFillReportsBuilder::default()
.ts_init(ts_now)
.start(start)
.build()
.map_err(|e| anyhow::anyhow!("{e}"))?;
let position_cmd = GeneratePositionStatusReportsBuilder::default()
.ts_init(ts_now)
.start(start)
.build()
.map_err(|e| anyhow::anyhow!("{e}"))?;
let (order_reports, fill_reports, position_reports) = tokio::try_join!(
self.generate_order_status_reports(&order_cmd),
self.generate_fill_reports(fill_cmd),
self.generate_position_status_reports(&position_cmd),
)?;
log::info!("Received {} OrderStatusReports", order_reports.len());
log::info!("Received {} FillReports", fill_reports.len());
log::info!("Received {} PositionReports", position_reports.len());
let mut mass_status = ExecutionMassStatus::new(
self.core.client_id,
self.core.account_id,
*OKX_VENUE,
ts_now,
None,
);
mass_status.add_order_reports(order_reports);
mass_status.add_fill_reports(fill_reports);
mass_status.add_position_reports(position_reports);
Ok(Some(mass_status))
}
fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
let route = {
let cache = self.core.cache();
let order = cache.try_order(&cmd.client_order_id)?;
if order.is_closed() {
log::warn!("Cannot submit closed order {}", order.client_order_id());
return Ok(());
}
if let Err(reason) = validate_okx_client_order_id(cmd.client_order_id.as_str()) {
let denied = OrderDeniedReason::InvalidClientOrderId { detail: reason };
self.emitter.emit_order_denied(&order, &denied.to_string());
return Ok(());
}
let order_type = order.order_type();
let route = self.submit_order_route(cmd.instrument_id, order_type)?;
log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
self.emitter.emit_order_submitted(&order);
route
};
match route {
OrderCommandRoute::RegularWs => self.submit_regular_order(&cmd),
OrderCommandRoute::AlgoHttp => self.submit_conditional_order(&cmd),
OrderCommandRoute::SpreadHttp => self.submit_order_http(&cmd),
}
}
fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
if is_spread_instrument(cmd.instrument_id) {
let cache = self.core.cache();
let denied = OrderDeniedReason::UnsupportedOrderList {
detail: "spread instruments are not supported in order lists".to_string(),
}
.to_string();
for client_order_id in &cmd.order_list.client_order_ids {
let order = cache.try_order(client_order_id)?;
self.emitter.emit_order_denied(&order, &denied);
}
return Ok(());
}
let inst_type = okx_instrument_type_from_symbol(cmd.instrument_id.symbol.as_str());
let cache = self.core.cache();
let invalid: Vec<(ClientOrderId, String)> = cmd
.order_list
.client_order_ids
.iter()
.filter_map(|cid| {
validate_okx_client_order_id(cid.as_str())
.err()
.map(|r| (*cid, r))
})
.collect();
if !invalid.is_empty() {
let order_list_id = cmd.order_list.id;
for client_order_id in &cmd.order_list.client_order_ids {
let order = cache.try_order(client_order_id)?;
let denied = invalid
.iter()
.find(|(cid, _)| cid == client_order_id)
.map_or_else(
|| OrderDeniedReason::OrderListDenied { order_list_id },
|(_, r)| OrderDeniedReason::InvalidClientOrderId { detail: r.clone() },
);
self.emitter.emit_order_denied(&order, &denied.to_string());
}
return Ok(());
}
for client_order_id in &cmd.order_list.client_order_ids {
let order = cache.try_order(client_order_id)?;
if self.is_conditional_order(order.order_type()) {
anyhow::bail!("Conditional orders not supported in order lists: {client_order_id}");
}
if order.time_in_force() != TimeInForce::Gtc {
anyhow::bail!(
"Only GTC orders supported in order lists: {client_order_id} has {:?}",
order.time_in_force()
);
}
}
let mut batch_orders = Vec::new();
let speed_bump = get_param_as_string(&cmd.params, "speed_bump");
let outcome = get_param_as_string(&cmd.params, "outcome");
let rpi = get_param_as_bool(&cmd.params, "rpi");
let rpi_taker_access = get_param_as_bool(&cmd.params, "rpi_taker_access");
let rpi_px_round = get_param_as_bool(&cmd.params, "rpi_px_round");
for client_order_id in &cmd.order_list.client_order_ids {
let order = cache.order(client_order_id).expect("validated above");
batch_orders.push((
inst_type,
cmd.instrument_id,
self.trade_mode_for_order(cmd.instrument_id, &cmd.params),
order.client_order_id(),
order.order_side(),
None, order.order_type(),
order.quantity(),
order.price(),
order.trigger_price(),
Some(order.is_post_only()),
Some(order.is_reduce_only()),
speed_bump.clone(),
outcome.clone(),
rpi,
rpi_taker_access,
rpi_px_round,
));
self.ws_dispatch_state.order_identities.insert(
order.client_order_id(),
OrderIdentity {
instrument_id: cmd.instrument_id,
strategy_id: order.strategy_id(),
order_side: order.order_side(),
order_type: order.order_type(),
},
);
log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
self.emitter.emit_order_submitted(&order);
}
drop(cache);
let ws_private = self.ws_private.clone();
let emitter = self.emitter.clone();
let clock = self.clock;
let instrument_id = cmd.instrument_id;
let strategy_id = cmd.strategy_id;
let client_order_ids: Vec<_> = cmd.order_list.client_order_ids;
let dispatch_state = Arc::clone(&self.ws_dispatch_state);
self.spawn_task("batch_submit_orders", async move {
let result = ws_private
.batch_submit_orders(batch_orders)
.await;
if let Err(e) = result {
if is_okx_ws_local_command_failure(&e) {
let ts_event = clock.get_time_ns();
for cid in &client_order_ids {
dispatch_state.order_identities.remove(cid);
emitter.emit_order_rejected_event(
strategy_id,
instrument_id,
*cid,
&format!("batch-submit-error: {e}"),
ts_event,
false,
);
}
} else {
log::warn!(
"Ambiguous batch submit failure for {} orders on {instrument_id}, awaiting reconciliation: {e}",
client_order_ids.len()
);
}
return Err(anyhow::Error::new(e).context("batch submit orders failed"));
}
Ok(())
});
Ok(())
}
fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
if is_spread_instrument(cmd.instrument_id) {
self.emitter.emit_order_modify_rejected_event(
cmd.strategy_id,
cmd.instrument_id,
cmd.client_order_id,
cmd.venue_order_id,
"OKX spread orders do not support modify requests",
self.clock.get_time_ns(),
);
return Ok(());
}
self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
let ws_private = self.ws_private.clone();
let command = cmd.clone();
let new_px_usd = get_param_as_string(&cmd.params, "px_usd");
let new_px_vol = get_param_as_string(&cmd.params, "px_vol");
let speed_bump = get_param_as_string(&cmd.params, "speed_bump");
let rpi_taker_access = get_param_as_bool(&cmd.params, "rpi_taker_access");
let rpi_px_round = get_param_as_bool(&cmd.params, "rpi_px_round");
let emitter = self.emitter.clone();
let clock = self.clock;
self.spawn_task("modify_order", async move {
let result = ws_private
.modify_order(
command.trader_id,
command.strategy_id,
command.instrument_id,
Some(command.client_order_id),
command.price,
command.quantity,
command.venue_order_id,
new_px_usd,
new_px_vol,
speed_bump,
rpi_taker_access,
rpi_px_round,
)
.await;
if let Err(e) = result {
if is_okx_ws_local_command_failure(&e) {
let ts_event = clock.get_time_ns();
emitter.emit_order_modify_rejected_event(
command.strategy_id,
command.instrument_id,
command.client_order_id,
command.venue_order_id,
&format!("modify-order-error: {e}"),
ts_event,
);
} else {
log::warn!(
"Ambiguous modify failure for {}, awaiting reconciliation: {e}",
command.client_order_id
);
}
return Err(anyhow::Error::new(e).context("modify order failed"));
}
Ok(())
});
Ok(())
}
fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
let route = {
let cache = self.core.cache();
let order_state = cache
.order(&cmd.client_order_id)
.map(|order| (order.order_type(), order.is_triggered()));
self.cancel_order_route(cmd.instrument_id, order_state)
};
match route {
OrderCommandRoute::RegularWs => self.cancel_ws_order(&cmd),
OrderCommandRoute::AlgoHttp => self.cancel_algo_order(&cmd),
OrderCommandRoute::SpreadHttp => self.cancel_order_http(&cmd),
}
Ok(())
}
fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
match self.cancel_all_orders_route(cmd.instrument_id) {
CancelAllOrdersRoute::SpreadHttp | CancelAllOrdersRoute::MassCancelHttp => {
self.mass_cancel_instrument(cmd.instrument_id);
Ok(())
}
CancelAllOrdersRoute::BatchWs => {
let cache = self.core.cache();
let open_orders =
cache.orders_open(None, Some(&cmd.instrument_id), None, None, None);
if open_orders.is_empty() {
log::debug!("No open orders to cancel for {}", cmd.instrument_id);
return Ok(());
}
let mut regular_payload = Vec::new();
let mut regular_cancel_contexts = Vec::new();
let mut algo_orders: Vec<(
InstrumentId,
ClientOrderId,
Option<VenueOrderId>,
TraderId,
StrategyId,
)> = Vec::new();
for order in &open_orders {
let order_state = Some((order.order_type(), order.is_triggered()));
match self.cancel_order_route(order.instrument_id(), order_state) {
OrderCommandRoute::RegularWs => {
self.ensure_order_identity(
order.client_order_id(),
order.strategy_id(),
order.instrument_id(),
);
regular_payload.push((
order.instrument_id(),
Some(order.client_order_id()),
order.venue_order_id(),
));
regular_cancel_contexts.push((
order.client_order_id(),
order.instrument_id(),
order.strategy_id(),
));
}
OrderCommandRoute::AlgoHttp => {
algo_orders.push((
order.instrument_id(),
order.client_order_id(),
order.venue_order_id(),
order.trader_id(),
order.strategy_id(),
));
}
OrderCommandRoute::SpreadHttp => {}
}
}
drop(open_orders);
drop(cache);
log::debug!(
"Canceling {} regular orders and {} algo orders for {}",
regular_payload.len(),
algo_orders.len(),
cmd.instrument_id
);
if !regular_payload.is_empty() {
let ws_private = self.ws_private.clone();
self.spawn_task("batch_cancel_orders", async move {
if let Err(e) = ws_private.batch_cancel_orders(regular_payload).await {
if is_okx_ws_local_command_failure(&e) {
log::warn!(
"Batch cancel command failed local validation for {} orders: {e}",
regular_cancel_contexts.len()
);
} else {
log::warn!(
"Ambiguous batch cancel failure for {} orders, awaiting reconciliation: {e}",
regular_cancel_contexts.len()
);
}
return Err(anyhow::Error::new(e).context("batch cancel orders failed"));
}
Ok(())
});
}
if !algo_orders.is_empty() {
let items: Vec<_> = algo_orders
.into_iter()
.map(
|(
instrument_id,
client_order_id,
venue_order_id,
_trader_id,
strategy_id,
)| {
let request = OKXCancelAlgoOrderRequest {
inst_id: instrument_id.symbol.to_string(),
inst_id_code: None,
algo_id: venue_order_id.map(|id| id.to_string()),
algo_cl_ord_id: if venue_order_id.is_none() {
Some(client_order_id.to_string())
} else {
None
},
};
let ctx = AlgoCancelContext {
client_order_id,
instrument_id,
strategy_id,
venue_order_id,
};
(request, ctx)
},
)
.collect();
self.dispatch_algo_cancels(items);
}
Ok(())
}
}
}
fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
let cache = self.core.cache();
let mut regular_payload = Vec::new();
let mut algo_orders = Vec::new();
let mut http_orders = Vec::new();
for cancel in &cmd.cancels {
let order_state = cache
.order(&cancel.client_order_id)
.map(|order| (order.order_type(), order.is_triggered()));
match self.cancel_order_route(cancel.instrument_id, order_state) {
OrderCommandRoute::RegularWs => {
self.ensure_order_identity(
cancel.client_order_id,
cancel.strategy_id,
cancel.instrument_id,
);
regular_payload.push((
cancel.instrument_id,
Some(cancel.client_order_id),
cancel.venue_order_id,
));
}
OrderCommandRoute::AlgoHttp => algo_orders.push(cancel.clone()),
OrderCommandRoute::SpreadHttp => {
self.ensure_order_identity(
cancel.client_order_id,
cancel.strategy_id,
cancel.instrument_id,
);
http_orders.push((
cancel.client_order_id,
cancel.instrument_id,
cancel.strategy_id,
cancel.venue_order_id,
));
}
}
}
drop(cache);
if !regular_payload.is_empty() {
let ws_private = self.ws_private.clone();
let cancel_contexts: Vec<_> = cmd
.cancels
.iter()
.filter(|c| {
regular_payload
.iter()
.any(|(_, cid, _)| *cid == Some(c.client_order_id))
})
.map(|c| (c.client_order_id, c.instrument_id, c.strategy_id))
.collect();
self.spawn_task("batch_cancel_orders", async move {
if let Err(e) = ws_private.batch_cancel_orders(regular_payload).await {
if is_okx_ws_local_command_failure(&e) {
log::warn!(
"Batch cancel command failed local validation for {} orders: {e}",
cancel_contexts.len()
);
} else {
log::warn!(
"Ambiguous batch cancel failure for {} orders, awaiting reconciliation: {e}",
cancel_contexts.len()
);
}
return Err(anyhow::Error::new(e).context("batch cancel orders failed"));
}
Ok(())
});
}
if !algo_orders.is_empty() {
let items: Vec<_> = algo_orders
.into_iter()
.map(|cancel| {
let request = OKXCancelAlgoOrderRequest {
inst_id: cancel.instrument_id.symbol.to_string(),
inst_id_code: None,
algo_id: cancel.venue_order_id.map(|id| id.to_string()),
algo_cl_ord_id: if cancel.venue_order_id.is_none() {
Some(cancel.client_order_id.to_string())
} else {
None
},
};
let ctx = AlgoCancelContext {
client_order_id: cancel.client_order_id,
instrument_id: cancel.instrument_id,
strategy_id: cancel.strategy_id,
venue_order_id: cancel.venue_order_id,
};
(request, ctx)
})
.collect();
self.dispatch_algo_cancels(items);
}
if !http_orders.is_empty() {
let client = self.http_client.clone();
let emitter = self.emitter.clone();
let clock = self.clock;
self.spawn_task("cancel_http_orders", async move {
for (client_order_id, instrument_id, strategy_id, venue_order_id) in http_orders {
if let Err(e) = client
.cancel_order(instrument_id, Some(client_order_id), venue_order_id)
.await
{
if is_okx_http_structured_venue_rejection(&e) {
let ts_event = clock.get_time_ns();
emitter.emit_order_cancel_rejected_event(
strategy_id,
instrument_id,
client_order_id,
venue_order_id,
&format!("cancel-http-order-error: {e}"),
ts_event,
);
} else if is_okx_http_local_command_failure(&e) {
log::warn!(
"HTTP cancel command failed local validation for {client_order_id}: {e}"
);
} else {
log::warn!(
"Ambiguous HTTP cancel failure for {client_order_id}, awaiting reconciliation: {e}"
);
}
}
}
Ok(())
});
}
Ok(())
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum OrderCommandRoute {
RegularWs,
AlgoHttp,
SpreadHttp,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum CancelAllOrdersRoute {
BatchWs,
MassCancelHttp,
SpreadHttp,
}
fn is_okx_http_structured_venue_rejection(error: &OKXHttpError) -> bool {
matches!(error, OKXHttpError::OkxError { .. })
}
fn is_okx_http_submit_rejection(error: &OKXHttpError) -> bool {
match error {
OKXHttpError::OkxError {
error_code,
message,
} => {
!matches!(error_code.as_str(), "50004" | "51149")
&& (should_retry_error_code(error_code) || !is_ambiguous_okx_http_failure(message))
}
_ => false,
}
}
fn is_okx_http_local_command_failure(error: &OKXHttpError) -> bool {
match error {
OKXHttpError::MissingCredentials => true,
OKXHttpError::ValidationError(message) => !is_ambiguous_okx_http_failure(message),
_ => false,
}
}
fn is_okx_ws_local_command_failure(error: &OKXWsError) -> bool {
match error {
OKXWsError::ClientError(message) => !is_ambiguous_okx_ws_client_failure(message),
OKXWsError::JsonError(_) => true,
_ => false,
}
}
fn is_ambiguous_okx_http_failure(message: &str) -> bool {
contains_any_ignore_ascii_case(
message,
&[
"empty response",
"timeout",
"timed out",
"retry",
"request canceled",
"network",
"unexpected http status",
"status code",
"failed to deserialize",
"failed to parse",
],
)
}
fn is_ambiguous_okx_ws_client_failure(message: &str) -> bool {
contains_any_ignore_ascii_case(
message,
&[
"handler not available",
"no active websocket client",
"send failed",
"timeout",
"timed out",
"retry",
"connection",
"network",
],
)
}
fn contains_any_ignore_ascii_case(value: &str, needles: &[&str]) -> bool {
needles.iter().any(|needle| {
value
.as_bytes()
.windows(needle.len())
.any(|window| window.eq_ignore_ascii_case(needle.as_bytes()))
})
}
fn get_param_as_string(params: &Option<Params>, key: &str) -> Option<String> {
params.as_ref().and_then(|p| {
p.get(key).and_then(|v| {
v.as_str()
.map(ToString::to_string)
.or_else(|| v.as_f64().map(|n| n.to_string()))
})
})
}
fn get_param_as_bool(params: &Option<Params>, key: &str) -> Option<bool> {
params.as_ref().and_then(|params| params.get_bool(key))
}
fn supports_algo_orders(instrument_type: OKXInstrumentType) -> bool {
!matches!(
instrument_type,
OKXInstrumentType::Option | OKXInstrumentType::Events
)
}
fn is_spread_instrument(instrument_id: InstrumentId) -> bool {
is_okx_spread_symbol(instrument_id.symbol.as_str())
}
fn select_query_order_report(
reports: Vec<OrderStatusReport>,
client_order_id: ClientOrderId,
venue_order_id: Option<VenueOrderId>,
) -> Option<OrderStatusReport> {
let mut by_vid: Option<OrderStatusReport> = None;
for report in reports {
if report.client_order_id == Some(client_order_id) {
return Some(report);
}
if by_vid.is_none()
&& venue_order_id
.as_ref()
.is_some_and(|vid| report.venue_order_id.as_str() == vid.as_str())
{
by_vid = Some(report);
}
}
by_vid
}
#[cfg(test)]
mod tests {
use std::{cell::RefCell, rc::Rc};
use nautilus_common::cache::Cache;
use nautilus_model::{enums::OrderStatus, instruments::Instrument};
use rstest::rstest;
use serde_json::Value;
use super::*;
fn build_config(
margin_mode: Option<OKXMarginMode>,
use_spot_margin: bool,
) -> OKXExecClientConfig {
OKXExecClientConfig {
margin_mode,
use_spot_margin,
..OKXExecClientConfig::default()
}
}
#[rstest]
#[case::spot(OKXInstrumentType::Spot, true)]
#[case::margin(OKXInstrumentType::Margin, true)]
#[case::swap(OKXInstrumentType::Swap, true)]
#[case::futures(OKXInstrumentType::Futures, true)]
#[case::option(OKXInstrumentType::Option, false)]
#[case::events(OKXInstrumentType::Events, false)]
fn test_supports_algo_orders(
#[case] instrument_type: OKXInstrumentType,
#[case] expected: bool,
) {
assert_eq!(supports_algo_orders(instrument_type), expected);
}
#[rstest]
#[case::accepted_despite_timeout("51149", "Order timed out. Please try again.", false)]
#[case::documented_unknown_outcome(
"50004",
"API endpoint request timeout; please check the request result",
false
)]
#[case::unknown_timeout_code("59999", "Order request timed out", false)]
#[case::temporary_system_rejection("50013", "System busy, please retry later", true)]
#[case::parameter_rejection("51000", "Parameter state error", true)]
fn test_is_okx_http_submit_rejection(
#[case] error_code: &str,
#[case] message: &str,
#[case] expected: bool,
) {
let error = OKXHttpError::OkxError {
error_code: error_code.to_string(),
message: message.to_string(),
};
assert_eq!(is_okx_http_submit_rejection(&error), expected);
}
#[rstest]
fn test_is_okx_http_submit_rejection_ignores_local_error() {
let error = OKXHttpError::ValidationError("invalid quantity".to_string());
assert!(!is_okx_http_submit_rejection(&error));
}
#[rstest]
#[case::cash_no_spot_margin(AccountType::Cash, None, false, OKXTradeMode::Cash)]
#[case::cash_spot_margin_cross(
AccountType::Cash,
Some(OKXMarginMode::Cross),
true,
OKXTradeMode::Cross
)]
#[case::cash_spot_margin_isolated(
AccountType::Cash,
Some(OKXMarginMode::Isolated),
true,
OKXTradeMode::Isolated
)]
#[case::cash_spot_margin_none(AccountType::Cash, None, true, OKXTradeMode::Isolated)]
#[case::margin_cross(
AccountType::Margin,
Some(OKXMarginMode::Cross),
false,
OKXTradeMode::Cross
)]
#[case::margin_isolated(
AccountType::Margin,
Some(OKXMarginMode::Isolated),
false,
OKXTradeMode::Isolated
)]
#[case::margin_none(AccountType::Margin, None, false, OKXTradeMode::Isolated)]
fn test_derive_default_trade_mode(
#[case] account_type: AccountType,
#[case] margin_mode: Option<OKXMarginMode>,
#[case] use_spot_margin: bool,
#[case] expected: OKXTradeMode,
) {
let config = build_config(margin_mode, use_spot_margin);
let result = OKXExecutionClient::derive_default_trade_mode(account_type, &config);
assert_eq!(result, expected);
}
#[rstest]
#[case::spot_no_margin("BTC-USDT", None, false, OKXTradeMode::Cash)]
#[case::spot_cross_margin("BTC-USDT", Some(OKXMarginMode::Cross), true, OKXTradeMode::Cross)]
#[case::spot_isolated_margin(
"ETH-USDT",
Some(OKXMarginMode::Isolated),
true,
OKXTradeMode::Isolated
)]
#[case::spot_margin_no_mode("BTC-USDT", None, true, OKXTradeMode::Isolated)]
#[case::swap_cross(
"BTC-USDT-SWAP",
Some(OKXMarginMode::Cross),
false,
OKXTradeMode::Cross
)]
#[case::swap_isolated(
"BTC-USDT-SWAP",
Some(OKXMarginMode::Isolated),
false,
OKXTradeMode::Isolated
)]
#[case::swap_no_mode("ETH-USDT-SWAP", None, false, OKXTradeMode::Isolated)]
#[case::futures_cross(
"BTC-USDT-250328",
Some(OKXMarginMode::Cross),
false,
OKXTradeMode::Cross
)]
#[case::futures_isolated("BTC-USDT-250328", None, false, OKXTradeMode::Isolated)]
#[case::option_cross(
"BTC-USD-250328-50000-C",
Some(OKXMarginMode::Cross),
false,
OKXTradeMode::Cross
)]
#[case::option_isolated("BTC-USD-250328-50000-C", None, false, OKXTradeMode::Isolated)]
fn test_derive_trade_mode_for_instrument(
#[case] symbol: &str,
#[case] margin_mode: Option<OKXMarginMode>,
#[case] use_spot_margin: bool,
#[case] expected: OKXTradeMode,
) {
let instrument_id = InstrumentId::from(format!("{symbol}.OKX").as_str());
let result = derive_trade_mode_for_instrument(instrument_id, margin_mode, use_spot_margin);
assert_eq!(result, expected);
}
#[rstest]
#[case::override_to_cross("cross", OKXTradeMode::Cross)]
#[case::override_to_cash("cash", OKXTradeMode::Cash)]
#[case::override_to_isolated("isolated", OKXTradeMode::Isolated)]
#[case::override_to_spot_isolated("spot_isolated", OKXTradeMode::SpotIsolated)]
#[case::case_insensitive("CROSS", OKXTradeMode::Cross)]
fn test_td_mode_param_override(#[case] td_mode_value: &str, #[case] expected: OKXTradeMode) {
let mut params = Params::new();
params.insert(
"td_mode".to_string(),
Value::String(td_mode_value.to_string()),
);
let result = get_param_as_string(&Some(params), "td_mode")
.and_then(|s| s.parse::<OKXTradeMode>().ok());
assert_eq!(result, Some(expected));
}
#[rstest]
fn test_td_mode_param_invalid_falls_through() {
let mut params = Params::new();
params.insert("td_mode".to_string(), Value::String("invalid".to_string()));
let result = get_param_as_string(&Some(params), "td_mode")
.and_then(|s| s.parse::<OKXTradeMode>().ok());
assert_eq!(result, None);
}
#[rstest]
fn test_td_mode_param_absent_falls_through() {
let result = get_param_as_string(&None, "td_mode");
assert_eq!(result, None);
}
#[rstest]
fn test_close_fraction_present_sets_reduce_only_true() {
let mut params = Params::new();
params.insert("close_fraction".to_string(), Value::String("1".to_string()));
let params = Some(params);
let close_fraction = get_param_as_string(¶ms, "close_fraction");
let is_reduce_only = false;
let reduce_only = if close_fraction.is_some() {
Some(true)
} else {
Some(is_reduce_only)
};
assert_eq!(close_fraction, Some("1".to_string()));
assert_eq!(reduce_only, Some(true));
}
#[rstest]
fn test_close_fraction_absent_preserves_reduce_only() {
let params: Option<Params> = None;
let close_fraction = get_param_as_string(¶ms, "close_fraction");
let is_reduce_only = false;
let reduce_only = if close_fraction.is_some() {
Some(true)
} else {
Some(is_reduce_only)
};
assert_eq!(close_fraction, None);
assert_eq!(reduce_only, Some(false));
}
#[rstest]
fn test_close_fraction_absent_with_reduce_only_true() {
let params: Option<Params> = None;
let close_fraction = get_param_as_string(¶ms, "close_fraction");
let is_reduce_only = true;
let reduce_only = if close_fraction.is_some() {
Some(true)
} else {
Some(is_reduce_only)
};
assert_eq!(close_fraction, None);
assert_eq!(reduce_only, Some(true));
}
fn make_query_order_report(cid: Option<&str>, vid: &str) -> OrderStatusReport {
OrderStatusReport::new(
AccountId::from("OKX-001"),
InstrumentId::from("BTC-USDT.OKX"),
cid.map(ClientOrderId::from),
VenueOrderId::from(vid),
OrderSide::Buy,
OrderType::Limit,
TimeInForce::Gtc,
OrderStatus::Accepted,
Quantity::new(1.0, 0),
Quantity::zero(0),
UnixNanos::default(),
UnixNanos::default(),
UnixNanos::default(),
None,
)
}
fn with_linked(mut report: OrderStatusReport, linked: &[&str]) -> OrderStatusReport {
report.linked_order_ids = Some(linked.iter().map(|s| ClientOrderId::from(*s)).collect());
report
}
#[rstest]
fn test_select_query_order_report_matches_client_order_id() {
let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
assert_eq!(
selected.and_then(|r| r.client_order_id),
Some(ClientOrderId::from("O-001"))
);
}
#[rstest]
fn test_select_query_order_report_client_wins_over_venue_mismatch() {
let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
let selected = select_query_order_report(
reports,
ClientOrderId::from("O-001"),
Some(VenueOrderId::from("V-OTHER")),
);
assert_eq!(
selected.and_then(|r| r.client_order_id),
Some(ClientOrderId::from("O-001"))
);
}
#[rstest]
fn test_select_query_order_report_falls_back_to_venue_order_id() {
let reports = vec![make_query_order_report(Some("O-CHILD"), "V-1")];
let selected = select_query_order_report(
reports,
ClientOrderId::from("O-PARENT"),
Some(VenueOrderId::from("V-1")),
);
assert_eq!(
selected.map(|r| r.venue_order_id.as_str().to_string()),
Some("V-1".to_string()),
);
}
#[rstest]
fn test_select_query_order_report_rejects_when_nothing_matches() {
let reports = vec![make_query_order_report(Some("O-OTHER"), "V-OTHER")];
let selected = select_query_order_report(
reports,
ClientOrderId::from("O-001"),
Some(VenueOrderId::from("V-1")),
);
assert!(selected.is_none());
}
#[rstest]
fn test_select_query_order_report_rejects_when_client_differs_and_no_vid_provided() {
let reports = vec![make_query_order_report(Some("O-OTHER"), "V-1")];
let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
assert!(selected.is_none());
}
#[rstest]
fn test_select_query_order_report_ignores_linked_order_ids_for_parent_with_attached_tp() {
let child_cid = "O-CHILD-TP";
let reports = vec![with_linked(
make_query_order_report(Some("O-PARENT"), "V-PARENT"),
&[child_cid, "O-CHILD-SL"],
)];
let selected = select_query_order_report(reports, ClientOrderId::from(child_cid), None);
assert!(selected.is_none());
}
#[rstest]
fn test_select_query_order_report_client_match_wins_over_vid_match_elsewhere() {
let reports = vec![
make_query_order_report(Some("O-OTHER"), "V-1"),
make_query_order_report(Some("O-001"), "V-2"),
];
let selected = select_query_order_report(
reports,
ClientOrderId::from("O-001"),
Some(VenueOrderId::from("V-1")),
);
assert_eq!(
selected.and_then(|r| r.client_order_id),
Some(ClientOrderId::from("O-001")),
);
}
fn build_test_exec_client() -> OKXExecutionClient {
let config = OKXExecClientConfig {
api_key: Some("test_key".to_string()),
api_secret: Some("test_secret".to_string()),
api_passphrase: Some("test_pass".to_string()),
..OKXExecClientConfig::default()
};
let cache = Rc::new(RefCell::new(Cache::default()));
let core = ExecutionClientCore::new(
config.trader_id,
ClientId::from("OKX-TEST"),
*OKX_VENUE,
OmsType::Hedging,
config.account_id,
AccountType::Cash,
None,
cache,
);
OKXExecutionClient::new(core, config).expect("failed to build test client")
}
#[rstest]
fn test_on_instrument_writes_through_to_client_caches() {
use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
let mut client = build_test_exec_client();
let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
let symbol = instrument.symbol().inner();
let raw_symbol = instrument.raw_symbol().inner();
client.on_instrument(instrument.clone());
let private_cache = client.ws_private.instruments_cache_arc();
let business_cache = client.ws_business.instruments_cache_arc();
assert_eq!(
client
.http_client
.get_instrument(&raw_symbol)
.map(|i| i.id()),
Some(instrument.id()),
);
assert_eq!(
private_cache.load().get(&symbol).map(|i| i.id()),
Some(instrument.id()),
);
assert_eq!(
business_cache.load().get(&symbol).map(|i| i.id()),
Some(instrument.id()),
);
}
}