1use std::{future::Future, sync::Arc};
19
20use ahash::{AHashMap, AHashSet};
21use anyhow::Context;
22use async_trait::async_trait;
23use futures_util::{StreamExt, pin_mut};
24use nautilus_common::{
25 clients::{ExecutionClient, ExecutionReportTask},
26 live::{
27 dst::time::{self, Duration, Instant},
28 runner::get_exec_event_sender,
29 },
30 messages::execution::{
31 BatchCancelOrders, CancelAllOrders, CancelOrder, GenerateFillReports,
32 GenerateFillReportsBuilder, GenerateOrderStatusReport, GenerateOrderStatusReports,
33 GenerateOrderStatusReportsBuilder, GeneratePositionStatusReports,
34 GeneratePositionStatusReportsBuilder, ModifyOrder, QueryAccount, QueryOrder, SubmitOrder,
35 SubmitOrderList,
36 },
37};
38use nautilus_core::{
39 DurationNanos, UnixNanos,
40 params::Params,
41 time::{AtomicTime, get_atomic_clock_realtime},
42};
43use nautilus_live::{
44 ExecutionClientCore, ExecutionEventEmitter, SocketControl,
45 execution::{
46 context::{OrderContext, OrderIdentity},
47 failure::CommandFailure,
48 reports::retain_order_status_reports,
49 },
50 task::{TaskGroup, TaskGroupGuard},
51};
52use nautilus_model::{
53 accounts::AccountAny,
54 enums::{
55 AccountType, OmsType, OrderSide, OrderStatus, OrderType, PositionSide, TimeInForce,
56 TrailingOffsetType,
57 },
58 events::OrderDeniedReason,
59 identifiers::{
60 AccountId, ClientId, ClientOrderId, InstrumentId, PositionId, StrategyId, Venue,
61 VenueOrderId,
62 },
63 instruments::InstrumentAny,
64 orders::Order,
65 reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
66 types::{AccountBalance, MarginBalance, Quantity},
67};
68use rust_decimal::Decimal;
69
70use crate::{
71 common::{
72 consts::{
73 OKX_CONDITIONAL_ORDER_TYPES, OKX_RECONCILIATION_LOOKBACK_DEFAULT_MINS,
74 OKX_RECONCILIATION_LOOKBACK_MAX_MINS, OKX_SUCCESS_CODE, OKX_VENUE,
75 OKX_WS_HEARTBEAT_SECS, okx_algo_target_currency, okx_reduce_only_wire_value,
76 resolve_instrument_families, validate_algo_trigger_type, validate_okx_client_order_id,
77 },
78 enums::{
79 OKXAlgoOrderType, OKXInstrumentType, OKXMarginMode, OKXPositionMode, OKXPositionSide,
80 OKXTradeMode, OKXTriggerType, conditional_order_to_algo_type, is_advance_algo_order,
81 },
82 failure::{classify_okx_http_failure, classify_okx_venue_code, classify_okx_ws_failure},
83 parse::{
84 is_order_status_report_more_advanced, nanos_to_datetime,
85 okx_instrument_type_from_symbol, venue_position_id,
86 },
87 task::{spawn_task, terminate_tasks},
88 },
89 config::OKXExecutionClientConfig,
90 http::{
91 client::{
92 AlgoOrderReportSweep, FillHistory, OKXHttpClient, OKXPendingAlgoOrderReportsError,
93 ReportInstrumentScope,
94 },
95 models::{OKXAmendAlgoOrderRequest, OKXCancelAlgoOrderRequest},
96 query::GetPositionsParams,
97 },
98 websocket::{
99 client::OKXWebSocketClient,
100 dispatch::{
101 AlgoCancelContext, AlgoOrderState, WsDispatchState, dispatch_ws_message,
102 emit_algo_cancel_rejections, emit_send_failed_cancel, format_order_response_reason,
103 },
104 error::OKXWsError,
105 messages::OKXWsMessage,
106 parse::{FeeCache, FilledQtyCache, OrderStateSnapshot},
107 },
108};
109
110#[derive(Debug)]
111pub struct OKXExecutionClient {
112 core: ExecutionClientCore,
113 clock: &'static AtomicTime,
114 config: OKXExecutionClientConfig,
115 emitter: ExecutionEventEmitter,
116 http_client: OKXHttpClient,
117 ws_private: OKXWebSocketClient,
118 ws_business: OKXWebSocketClient,
119 trade_mode: OKXTradeMode,
120 position_mode: OKXPositionMode,
121 ws_dispatch_state: Arc<WsDispatchState>,
122 session_tasks: TaskGroup,
123 pending_tasks: TaskGroup,
124}
125
126impl OKXExecutionClient {
127 pub fn new(
133 core: ExecutionClientCore,
134 config: OKXExecutionClientConfig,
135 ) -> anyhow::Result<Self> {
136 config.validate()?;
137
138 let api_key = config
139 .api_key
140 .as_ref()
141 .map(|value| value.expose_secret().to_owned());
142 let api_secret = config
143 .api_secret
144 .as_ref()
145 .map(|value| value.expose_secret().to_owned());
146 let api_passphrase = config
147 .api_passphrase
148 .as_ref()
149 .map(|value| value.expose_secret().to_owned());
150 let proxy_url = config
151 .proxy_url
152 .as_ref()
153 .map(|value| value.expose_secret().to_owned());
154 let http_client = OKXHttpClient::with_credentials(
155 api_key.clone(),
156 api_secret.clone(),
157 api_passphrase.clone(),
158 Some(config.http_base_url()),
159 config.http_timeout_secs,
160 config.max_retries,
161 config.retry_delay_initial_ms,
162 config.retry_delay_max_ms,
163 config.environment,
164 proxy_url.clone(),
165 )?;
166
167 let account_id = core.account_id;
168
169 let ws_private = OKXWebSocketClient::with_credentials(
170 Some(config.ws_private_url()),
171 api_key.clone(),
172 api_secret.clone(),
173 api_passphrase.clone(),
174 Some(account_id),
175 Some(OKX_WS_HEARTBEAT_SECS),
176 config.auth_timeout_secs,
177 config.transport_backend,
178 proxy_url.clone(),
179 )
180 .context("failed to construct OKX private websocket client")?
181 .with_socket_control(SocketControl::new(
182 core.client_id,
183 Some(*OKX_VENUE),
184 "okx-private-user-streams",
185 ));
186
187 let ws_business = OKXWebSocketClient::with_credentials(
188 Some(config.ws_business_url()),
189 api_key,
190 api_secret,
191 api_passphrase,
192 Some(account_id),
193 Some(OKX_WS_HEARTBEAT_SECS),
194 config.auth_timeout_secs,
195 config.transport_backend,
196 proxy_url,
197 )
198 .context("failed to construct OKX business websocket client")?
199 .with_socket_control(SocketControl::new(
200 core.client_id,
201 Some(*OKX_VENUE),
202 "okx-business-user-streams",
203 ));
204
205 http_client.set_spot_trade_quote_ccy(config.spot_trade_quote_ccy.clone());
206 ws_private.set_spot_trade_quote_ccy(config.spot_trade_quote_ccy.clone());
207 ws_business.set_spot_trade_quote_ccy(config.spot_trade_quote_ccy.clone());
208
209 let trade_mode = Self::derive_default_trade_mode(core.account_type, &config);
210 let clock = get_atomic_clock_realtime();
211 let emitter = ExecutionEventEmitter::new(
212 clock,
213 core.trader_id,
214 core.account_id,
215 core.account_type,
216 None,
217 );
218
219 let ws_dispatch_state = Arc::new(WsDispatchState::with_pending_maps(
220 ws_private.pending_orders.clone(),
221 ws_private.pending_cancels.clone(),
222 ws_private.pending_amends.clone(),
223 ));
224
225 Ok(Self {
226 core,
227 clock,
228 config,
229 emitter,
230 http_client,
231 ws_private,
232 ws_business,
233 trade_mode,
234 position_mode: OKXPositionMode::NetMode,
235 ws_dispatch_state,
236 session_tasks: TaskGroup::new(),
237 pending_tasks: TaskGroup::new(),
238 })
239 }
240
241 fn derive_default_trade_mode(
242 account_type: AccountType,
243 config: &OKXExecutionClientConfig,
244 ) -> OKXTradeMode {
245 let is_cross_margin = config.margin_mode == Some(OKXMarginMode::Cross);
246
247 if account_type == AccountType::Cash {
248 if !config.use_spot_margin {
249 return OKXTradeMode::Cash;
250 }
251 return if is_cross_margin {
252 OKXTradeMode::Cross
253 } else {
254 OKXTradeMode::Isolated
255 };
256 }
257
258 if is_cross_margin {
259 OKXTradeMode::Cross
260 } else {
261 OKXTradeMode::Isolated
262 }
263 }
264
265 fn trade_mode_for_order(
266 &self,
267 instrument_id: InstrumentId,
268 params: Option<&Params>,
269 ) -> OKXTradeMode {
270 if let Some(td_mode_str) = get_param_as_string(params, "td_mode") {
271 match td_mode_str.parse::<OKXTradeMode>() {
272 Ok(mode) => return mode,
273 Err(_) => {
274 log::warn!("Invalid td_mode '{td_mode_str}', using derived trade mode");
275 }
276 }
277 }
278
279 derive_trade_mode_for_instrument(
280 instrument_id,
281 self.config.margin_mode,
282 self.config.use_spot_margin,
283 )
284 }
285
286 fn report_client(&self) -> OKXReportClient {
287 OKXReportClient {
288 account_id: self.core.account_id,
289 http_client: self.http_client.clone(),
290 instrument_types: self.config.instrument_types.clone(),
291 load_spreads: self.config.load_spreads,
292 margin_mode: (self.position_mode == OKXPositionMode::LongShortMode).then_some(
293 self.config
294 .margin_mode
295 .filter(|mode| *mode != OKXMarginMode::None)
296 .unwrap_or(OKXMarginMode::Isolated),
297 ),
298 }
299 }
300
301 fn instrument_types(&self) -> Vec<OKXInstrumentType> {
302 OKXReportClient::instrument_types(&self.config.instrument_types)
303 }
304
305 async fn collect_order_status_reports(
306 &self,
307 cmd: &GenerateOrderStatusReports,
308 require_complete_active_coverage: bool,
309 max_history_pages: Option<u32>,
310 ) -> anyhow::Result<OrderReportSweep> {
311 let sweep = self
312 .report_client()
313 .collect_order_status_reports(cmd, require_complete_active_coverage, max_history_pages)
314 .await?;
315 Ok(finish_order_report_sweep(&self.core, sweep, cmd))
316 }
317
318 async fn collect_fill_reports(
319 &self,
320 cmd: GenerateFillReports,
321 history: FillHistory,
322 max_history_pages: Option<u32>,
323 ) -> anyhow::Result<(Vec<FillReport>, bool)> {
324 self.report_client()
325 .collect_fill_reports(cmd, history, max_history_pages)
326 .await
327 }
328
329 async fn collect_position_status_reports(
330 &self,
331 cmd: &GeneratePositionStatusReports,
332 ) -> anyhow::Result<(Vec<PositionStatusReport>, bool)> {
333 self.report_client()
334 .collect_position_status_reports(cmd)
335 .await
336 }
337
338 fn query_order_state(&self, cmd: &GenerateOrderStatusReport) -> Option<CachedQueryOrderState> {
339 let cache = self.core.cache();
340 cmd.client_order_id.and_then(|client_order_id| {
341 cache
342 .order(&client_order_id)
343 .map(|order| CachedQueryOrderState {
344 order_type: order.order_type(),
345 venue_order_id: order.venue_order_id(),
346 })
347 })
348 }
349
350 fn update_account_state(&self) {
351 let http_client = self.http_client.clone();
352 let account_id = self.core.account_id;
353 let account_type = self.core.account_type;
354 let emitter = self.emitter.clone();
355
356 self.spawn_task("query_account", async move {
357 let account_state = http_client
358 .request_account_state(account_id, account_type)
359 .await
360 .context("failed to request OKX account state")?;
361 emitter.send_account_state(account_state);
362 Ok(())
363 });
364 }
365
366 fn is_conditional_order(&self, order_type: OrderType) -> bool {
367 OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
368 }
369
370 fn submit_order_route(
371 &self,
372 instrument_id: InstrumentId,
373 order_type: OrderType,
374 ) -> anyhow::Result<OrderCommandRoute> {
375 if self.is_conditional_order(order_type) {
376 if self.http_client.is_spread(instrument_id) {
377 anyhow::bail!(
378 "Trigger/conditional orders ({order_type:?}) are not supported for OKX spreads"
379 );
380 }
381
382 let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
383 if inst_type == OKXInstrumentType::Option {
384 anyhow::bail!(
385 "Trigger/conditional orders ({order_type:?}) are not supported for OKX options"
386 );
387 }
388
389 return Ok(OrderCommandRoute::AlgoHttp);
390 }
391
392 if self.http_client.is_spread(instrument_id) {
393 Ok(OrderCommandRoute::SpreadHttp)
394 } else {
395 Ok(OrderCommandRoute::RegularWs)
396 }
397 }
398
399 fn order_command_route(
400 &self,
401 instrument_id: InstrumentId,
402 order_state: Option<(OrderType, Option<bool>)>,
403 has_bound_child: bool,
404 ) -> OrderCommandRoute {
405 if self.http_client.is_spread(instrument_id) {
406 return OrderCommandRoute::SpreadHttp;
407 }
408
409 if has_bound_child {
410 return OrderCommandRoute::RegularWs;
411 }
412
413 if order_state.is_some_and(|(order_type, is_triggered)| {
414 self.is_conditional_order(order_type) && is_triggered != Some(true)
415 }) {
416 OrderCommandRoute::AlgoHttp
417 } else {
418 OrderCommandRoute::RegularWs
419 }
420 }
421
422 fn cancel_all_orders_route(
423 &self,
424 instrument_id: InstrumentId,
425 order_side: Option<OrderSide>,
426 ) -> CancelAllOrdersRoute {
427 if order_side.is_some() {
428 CancelAllOrdersRoute::Batch
429 } else if self.http_client.is_spread(instrument_id) {
430 CancelAllOrdersRoute::SpreadHttp
431 } else if self.config.use_mm_mass_cancel {
432 CancelAllOrdersRoute::MassCancelHttp
433 } else {
434 CancelAllOrdersRoute::Batch
435 }
436 }
437
438 fn submit_regular_order(
439 &self,
440 cmd: &SubmitOrder,
441 position_side: Option<PositionSide>,
442 ) -> anyhow::Result<()> {
443 let order = {
444 let cache = self.core.cache();
445 cache.try_order_owned(&cmd.client_order_id)?
446 };
447 let ws_private = self.ws_private.clone();
448 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
449
450 let emitter = self.emitter.clone();
451 let clock = self.clock;
452 let trader_id = self.core.trader_id;
453 let context = OrderContext::from(&order);
454
455 self.ws_dispatch_state
456 .order_identities
457 .insert(context.identity.client_order_id, context.identity);
458 let client_order_id = context.identity.client_order_id;
459 let strategy_id = context.identity.strategy_id;
460 let instrument_id = context.identity.instrument_id;
461 let order_side = context.identity.order_side;
462 let order_type = context.identity.order_type;
463 let quantity = context.quantity;
464 let time_in_force = context.time_in_force;
465 let price = context.price;
466 let trigger_price = context.trigger_price;
467 let is_post_only = context.is_post_only;
468 let is_reduce_only = context.is_reduce_only;
469
470 let is_quote_quantity = context.is_quote_quantity;
471
472 let px_usd = get_param_as_string(cmd.params.as_ref(), "px_usd");
473 let px_vol = get_param_as_string(cmd.params.as_ref(), "px_vol");
474 let outcome = get_param_as_string(cmd.params.as_ref(), "outcome");
475 let slippage_pct = get_param_as_string(cmd.params.as_ref(), "slippage_pct");
476 let rpi = get_param_as_bool(cmd.params.as_ref(), "rpi");
477 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
478 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
479
480 self.spawn_task("submit_order", async move {
481 let result = ws_private
482 .submit_order(
483 trader_id,
484 strategy_id,
485 instrument_id,
486 trade_mode,
487 client_order_id,
488 order_side,
489 order_type,
490 quantity,
491 Some(time_in_force),
492 price,
493 trigger_price,
494 Some(is_post_only),
495 Some(is_reduce_only),
496 Some(is_quote_quantity),
497 position_side,
498 None,
499 px_usd,
500 px_vol,
501 outcome,
502 slippage_pct,
503 rpi,
504 rpi_taker_access,
505 rpi_px_round,
506 )
507 .await;
508
509 if let Err(e) = result {
510 emit_submit_failure(
511 classify_okx_ws_failure(&e),
512 &emitter,
513 clock,
514 strategy_id,
515 instrument_id,
516 client_order_id,
517 );
518 return Err(anyhow::Error::new(e).context("submit order failed"));
519 }
520
521 Ok(())
522 });
523
524 Ok(())
525 }
526
527 fn submit_order_http(
528 &self,
529 cmd: &SubmitOrder,
530 position_side: Option<PositionSide>,
531 ) -> anyhow::Result<()> {
532 let order = {
533 let cache = self.core.cache();
534 cache.try_order_owned(&cmd.client_order_id)?
535 };
536 let http_client = self.http_client.clone();
537 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
538
539 let emitter = self.emitter.clone();
540 let clock = self.clock;
541 let context = OrderContext::from(&order);
542
543 self.ws_dispatch_state
544 .order_identities
545 .insert(context.identity.client_order_id, context.identity);
546 let client_order_id = context.identity.client_order_id;
547 let strategy_id = context.identity.strategy_id;
548 let instrument_id = context.identity.instrument_id;
549 let order_side = context.identity.order_side;
550 let order_type = context.identity.order_type;
551 let quantity = context.quantity;
552 let time_in_force = context.time_in_force;
553 let price = context.price;
554 let is_post_only = context.is_post_only;
555 let is_reduce_only = context.is_reduce_only;
556 let rpi = get_param_as_bool(cmd.params.as_ref(), "rpi");
557 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
558 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
559
560 self.spawn_task("submit_order_http", async move {
561 let result = http_client
562 .place_order_with_domain_types(
563 instrument_id,
564 trade_mode,
565 client_order_id,
566 order_side,
567 order_type,
568 quantity,
569 Some(time_in_force),
570 price,
571 Some(is_post_only),
572 Some(is_reduce_only),
573 None,
574 position_side,
575 None,
576 None,
577 None,
578 None,
579 None,
580 rpi,
581 rpi_taker_access,
582 rpi_px_round,
583 )
584 .await;
585
586 if let Err(e) = result {
587 emit_submit_failure(
588 classify_okx_http_failure(&e),
589 &emitter,
590 clock,
591 strategy_id,
592 instrument_id,
593 client_order_id,
594 );
595 return Err(anyhow::Error::new(e).context("submit order failed"));
596 }
597
598 Ok(())
599 });
600
601 Ok(())
602 }
603
604 fn submit_conditional_order(
605 &self,
606 cmd: &SubmitOrder,
607 position_side: Option<PositionSide>,
608 ) -> anyhow::Result<()> {
609 let order = {
610 let cache = self.core.cache();
611 cache.try_order_owned(&cmd.client_order_id)?
612 };
613 let http_client = self.http_client.clone();
614 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
615
616 let emitter = self.emitter.clone();
617 let clock = self.clock;
618 let context = OrderContext::from(&order);
619
620 let client_order_id = context.identity.client_order_id;
621 let strategy_id = context.identity.strategy_id;
622 let instrument_id = context.identity.instrument_id;
623 let order_side = context.identity.order_side;
624 let order_type = context.identity.order_type;
625 let quantity = context.quantity;
626 let trigger_type = context.trigger_type;
627 let trigger_price = context.trigger_price;
628 let price = context.price;
629 let is_reduce_only = context.is_reduce_only;
630
631 let tgt_ccy = okx_algo_target_currency(
632 okx_instrument_type_from_symbol(instrument_id.symbol.as_str()),
633 trade_mode,
634 order_side,
635 order_type,
636 context.is_quote_quantity,
637 )?;
638
639 let trailing_offset = order.trailing_offset();
640 let trailing_offset_type = order.trailing_offset_type();
641 let activation_price = order.activation_price();
642
643 let close_fraction = get_param_as_string(cmd.params.as_ref(), "close_fraction");
644
645 let reduce_only = if close_fraction.is_some() {
646 Some(true)
647 } else {
648 Some(is_reduce_only)
649 };
650
651 let (callback_ratio, callback_spread) = if order_type == OrderType::TrailingStopMarket {
652 let offset = trailing_offset
653 .ok_or_else(|| anyhow::anyhow!("TrailingStopMarket requires trailing_offset"))?;
654 let offset_type = trailing_offset_type.ok_or_else(|| {
655 anyhow::anyhow!("TrailingStopMarket requires trailing_offset_type")
656 })?;
657
658 match offset_type {
659 TrailingOffsetType::BasisPoints => {
660 let ratio = offset / Decimal::from(10000);
662 (Some(ratio.to_string()), None)
663 }
664 TrailingOffsetType::Price => (None, Some(offset.to_string())),
665 _ => {
666 anyhow::bail!("Unsupported trailing_offset_type for OKX: {offset_type:?}");
667 }
668 }
669 } else {
670 (None, None)
671 };
672
673 let algo_order_type = if close_fraction.is_some() || tgt_ccy.is_some() {
674 OKXAlgoOrderType::Conditional
675 } else {
676 conditional_order_to_algo_type(order_type)?
677 };
678
679 self.ws_dispatch_state.algo_order_states.insert(
680 client_order_id,
681 AlgoOrderState {
682 order_type: algo_order_type,
683 ts_last: UnixNanos::default(),
684 },
685 );
686
687 self.ws_dispatch_state.track_order_context(context);
688 let dispatch_state = Arc::clone(&self.ws_dispatch_state);
689
690 self.spawn_task("submit_algo_order", async move {
691 let result = http_client
692 .place_algo_order_with_domain_types(
693 instrument_id,
694 trade_mode,
695 client_order_id,
696 order_side,
697 order_type,
698 quantity,
699 context.is_quote_quantity,
700 trigger_price,
701 trigger_type,
702 price,
703 reduce_only,
704 position_side,
705 close_fraction,
706 callback_ratio,
707 callback_spread,
708 activation_price,
709 )
710 .await;
711
712 match result {
713 Ok(response) => {
714 dispatch_state.bind_order(
715 client_order_id,
716 VenueOrderId::new(response.algo_id.as_str()),
717 );
718 }
719 Err(e) => {
720 let failure = classify_okx_http_failure(&e);
721 dispatch_state.resolve_algo_submit_failure(client_order_id, &failure);
722 emit_submit_failure(
723 failure,
724 &emitter,
725 clock,
726 strategy_id,
727 instrument_id,
728 client_order_id,
729 );
730 return Err(anyhow::Error::new(e).context("submit algo order failed"));
731 }
732 }
733
734 Ok(())
735 });
736
737 Ok(())
738 }
739
740 fn modify_algo_order(&self, cmd: &ModifyOrder) {
741 let algo_type = self
742 .ws_dispatch_state
743 .algo_order_states
744 .get(&cmd.client_order_id)
745 .map(|state| state.order_type);
746 let context = self.ws_dispatch_state.order_context(cmd.client_order_id);
747
748 let request = match algo_amend_request(cmd, algo_type, context) {
749 Ok(request) => request,
750 Err(reason) => {
751 emit_modify_failure(
752 CommandFailure::NotSent(reason.to_string()),
753 &self.emitter,
754 self.clock,
755 cmd.strategy_id,
756 cmd.instrument_id,
757 cmd.client_order_id,
758 cmd.venue_order_id,
759 );
760 return;
761 }
762 };
763
764 let http_client = self.http_client.clone();
765 let command = cmd.clone();
766 let emitter = self.emitter.clone();
767 let clock = self.clock;
768 self.spawn_task("modify_algo_order", async move {
769 let result = http_client.amend_algo_order(request).await;
770
771 let failure = match result {
772 Err(e) => Some(classify_okx_http_failure(&e)),
773 Ok(response) => response
774 .s_code
775 .as_deref()
776 .filter(|code| *code != OKX_SUCCESS_CODE)
777 .map(|code| {
778 classify_okx_venue_code(
779 code,
780 format_order_response_reason(
781 code,
782 response.s_msg.as_deref().unwrap_or(""),
783 "",
784 ),
785 )
786 }),
787 };
788
789 if let Some(failure) = failure {
790 emit_modify_failure(
791 failure,
792 &emitter,
793 clock,
794 command.strategy_id,
795 command.instrument_id,
796 command.client_order_id,
797 command.venue_order_id,
798 );
799 }
800
801 Ok(())
802 });
803 }
804
805 fn cancel_ws_order(&self, cmd: &CancelOrder) {
806 self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
807
808 let ws_private = self.ws_private.clone();
809 let state = Arc::clone(&self.ws_dispatch_state);
810 let emitter = self.emitter.clone();
811 let clock = self.clock;
812 let mut command = cmd.clone();
813 command.venue_order_id = self
814 .ws_dispatch_state
815 .order_venue_binding(cmd.client_order_id)
816 .map(|(venue_order_id, _)| venue_order_id)
817 .or(cmd.venue_order_id);
818
819 self.spawn_task("cancel_order", async move {
820 let result = ws_private
821 .cancel_order(
822 command.trader_id,
823 command.strategy_id,
824 command.instrument_id,
825 Some(command.client_order_id),
826 command.venue_order_id,
827 )
828 .await;
829
830 if let Err(e) = result {
831 let failure = classify_okx_ws_failure(&e);
832
833 if matches!(e, OKXWsError::HandlerUnavailable(_)) {
834 emit_send_failed_cancel(
835 &failure,
836 &state,
837 &emitter,
838 clock,
839 command.client_order_id,
840 command.venue_order_id,
841 );
842 } else {
843 emit_cancel_failure(
844 failure,
845 None,
846 command.client_order_id,
847 command.instrument_id,
848 command.strategy_id,
849 command.venue_order_id,
850 );
851 }
852
853 return Err(anyhow::Error::new(e).context("cancel order failed"));
854 }
855
856 Ok(())
857 });
858 }
859
860 fn cancel_order_http(&self, cmd: &CancelOrder) {
861 self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
862
863 let http_client = self.http_client.clone();
864 let command = cmd.clone();
865 let emitter = self.emitter.clone();
866 let clock = self.clock;
867
868 self.spawn_task("cancel_order_http", async move {
869 let result = http_client
870 .cancel_order(
871 command.instrument_id,
872 Some(command.client_order_id),
873 command.venue_order_id,
874 )
875 .await;
876
877 if let Err(e) = result {
878 emit_cancel_failure(
879 classify_okx_http_failure(&e),
880 Some((&emitter, clock)),
881 command.client_order_id,
882 command.instrument_id,
883 command.strategy_id,
884 command.venue_order_id,
885 );
886 return Err(anyhow::Error::new(e).context("cancel order failed"));
887 }
888
889 Ok(())
890 });
891 }
892
893 fn cancel_algo_order(&self, cmd: &CancelOrder) {
894 let http_client = self.http_client.clone();
895 let command = cmd.clone();
896 let emitter = self.emitter.clone();
897 let clock = self.clock;
898
899 let cache = self.core.cache();
900 let is_advance = cache
901 .order(&cmd.client_order_id)
902 .is_some_and(|o| is_advance_algo_order(o.order_type()));
903 drop(cache);
904
905 let request = OKXCancelAlgoOrderRequest {
906 inst_id: cmd.instrument_id.symbol.to_string(),
907 inst_id_code: None,
908 algo_id: cmd.venue_order_id.map(|id| id.to_string()),
909 algo_cl_ord_id: if cmd.venue_order_id.is_none() {
910 Some(cmd.client_order_id.to_string())
911 } else {
912 None
913 },
914 };
915
916 self.spawn_task("cancel_algo_order", async move {
917 let responses = if is_advance {
918 http_client.cancel_advance_algo_orders(vec![request]).await
919 } else {
920 http_client.cancel_algo_orders(vec![request]).await
921 };
922
923 match responses {
924 Err(e) => {
925 emit_cancel_failure(
926 classify_okx_http_failure(&e),
927 Some((&emitter, clock)),
928 command.client_order_id,
929 command.instrument_id,
930 command.strategy_id,
931 command.venue_order_id,
932 );
933 return Err(anyhow::Error::new(e).context("cancel algo order failed"));
934 }
935 Ok(resps) => {
936 if let Some((code, msg)) = resps.first().and_then(|r| {
937 r.s_code.as_deref().and_then(|code| {
938 (code != OKX_SUCCESS_CODE)
939 .then_some((code, r.s_msg.as_deref().unwrap_or("")))
940 })
941 }) {
942 let reason = format_order_response_reason(code, msg, "");
943 let failure = classify_okx_venue_code(code, reason.clone());
944 let is_rejected = matches!(failure, CommandFailure::VenueRejected(_));
945 emit_cancel_failure(
946 failure,
947 Some((&emitter, clock)),
948 command.client_order_id,
949 command.instrument_id,
950 command.strategy_id,
951 command.venue_order_id,
952 );
953
954 if is_rejected {
955 anyhow::bail!("{reason}");
956 }
957 }
958 }
959 }
960
961 Ok(())
962 });
963 }
964
965 fn mass_cancel_instrument(&self, instrument_id: InstrumentId) {
966 if self.http_client.is_spread(instrument_id) {
967 let http_client = self.http_client.clone();
968 self.spawn_task("mass_cancel_orders_http", async move {
969 if let Err(e) = http_client.cancel_all_orders(instrument_id).await {
970 log_mass_cancel_failure(classify_okx_http_failure(&e), instrument_id);
971 return Err(anyhow::Error::new(e).context("mass cancel orders failed"));
972 }
973 Ok(())
974 });
975 return;
976 }
977
978 let ws_private = self.ws_private.clone();
979
980 self.spawn_task("mass_cancel_orders", async move {
981 if let Err(e) = ws_private.mass_cancel_orders(instrument_id).await {
982 log_mass_cancel_failure(classify_okx_ws_failure(&e), instrument_id);
983 return Err(anyhow::Error::new(e).context("mass cancel orders failed"));
984 }
985 Ok(())
986 });
987 }
988
989 fn ensure_order_identity(
998 &self,
999 client_order_id: ClientOrderId,
1000 strategy_id: StrategyId,
1001 instrument_id: InstrumentId,
1002 ) {
1003 if self
1004 .ws_dispatch_state
1005 .order_identity(client_order_id)
1006 .is_some()
1007 {
1008 return;
1009 }
1010
1011 let cache = self.core.cache();
1012
1013 let order_identity = cache
1014 .order(&client_order_id)
1015 .filter(|order| !OKX_CONDITIONAL_ORDER_TYPES.contains(&order.order_type()))
1016 .map(|order| OrderIdentity {
1017 client_order_id,
1018 instrument_id,
1019 strategy_id,
1020 order_side: order.order_side(),
1021 order_type: order.order_type(),
1022 });
1023
1024 drop(cache);
1025
1026 if let Some(order_identity) = order_identity {
1027 self.ws_dispatch_state
1028 .order_identities
1029 .entry(client_order_id)
1030 .or_insert(order_identity);
1031 }
1032 }
1033
1034 fn spawn_task<F>(&self, description: &'static str, fut: F)
1035 where
1036 F: Future<Output = anyhow::Result<()>> + Send + 'static,
1037 {
1038 let fut = async move {
1039 if let Err(e) = fut.await {
1040 log::warn!("{description} failed: {e:?}");
1041 }
1042 };
1043
1044 match self.pending_tasks.spawner() {
1045 Ok(spawner) => spawn_task(&spawner, fut),
1046 Err(e) => log::debug!("Skipping {description} after OKX shutdown began: {e}"),
1047 }
1048 }
1049
1050 fn dispatch_algo_cancels(&self, items: Vec<(OKXCancelAlgoOrderRequest, AlgoCancelContext)>) {
1053 const BATCH_SIZE: usize = 10;
1054
1055 let mut regular_requests = Vec::new();
1056 let mut regular_contexts = Vec::new();
1057 let mut advance_requests = Vec::new();
1058 let mut advance_contexts = Vec::new();
1059
1060 let cache = self.core.cache();
1061
1062 for (request, ctx) in items {
1063 let is_advance = cache
1064 .order(&ctx.client_order_id)
1065 .is_some_and(|o| is_advance_algo_order(o.order_type()));
1066
1067 if is_advance {
1068 advance_requests.push(request);
1069 advance_contexts.push(ctx);
1070 } else {
1071 regular_requests.push(request);
1072 regular_contexts.push(ctx);
1073 }
1074 }
1075
1076 drop(cache);
1077
1078 for (requests, contexts) in regular_requests
1079 .chunks(BATCH_SIZE)
1080 .zip(regular_contexts.chunks(BATCH_SIZE))
1081 {
1082 let regular_requests = requests.to_vec();
1083 let regular_contexts = contexts.to_vec();
1084 let client = self.http_client.clone();
1085 let emitter = self.emitter.clone();
1086 let clock = self.clock;
1087
1088 self.spawn_task("cancel_algo_orders", async move {
1089 match client.cancel_algo_orders(regular_requests).await {
1090 Ok(responses) => {
1091 emit_algo_cancel_rejections(&responses, ®ular_contexts, &emitter, clock);
1092 }
1093 Err(e) => {
1094 log_algo_batch_cancel_failure(
1095 classify_okx_http_failure(&e),
1096 ®ular_contexts,
1097 );
1098 return Err(anyhow::Error::new(e).context("cancel algo orders failed"));
1099 }
1100 }
1101 Ok(())
1102 });
1103 }
1104
1105 for (requests, contexts) in advance_requests
1106 .chunks(BATCH_SIZE)
1107 .zip(advance_contexts.chunks(BATCH_SIZE))
1108 {
1109 let advance_requests = requests.to_vec();
1110 let advance_contexts = contexts.to_vec();
1111 let client = self.http_client.clone();
1112 let emitter = self.emitter.clone();
1113 let clock = self.clock;
1114
1115 self.spawn_task("cancel_advance_algo_orders", async move {
1116 match client.cancel_advance_algo_orders(advance_requests).await {
1117 Ok(responses) => {
1118 emit_algo_cancel_rejections(&responses, &advance_contexts, &emitter, clock);
1119 }
1120 Err(e) => {
1121 log_algo_batch_cancel_failure(
1122 classify_okx_http_failure(&e),
1123 &advance_contexts,
1124 );
1125 return Err(
1126 anyhow::Error::new(e).context("cancel advance algo orders failed")
1127 );
1128 }
1129 }
1130 Ok(())
1131 });
1132 }
1133 }
1134
1135 fn begin_generation_shutdown(&self) {
1136 self.pending_tasks.begin_shutdown();
1137 self.session_tasks.begin_shutdown();
1138 self.ws_private.begin_shutdown();
1139 self.ws_business.begin_shutdown();
1140 self.core.set_disconnected();
1141 }
1142
1143 async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
1145 let account_id = self.core.account_id;
1146
1147 if self.core.cache().account(&account_id).is_some() {
1148 log::info!("Account {account_id} registered");
1149 return Ok(());
1150 }
1151
1152 let start = Instant::now();
1153 let timeout = Duration::from_secs_f64(timeout_secs);
1154 let interval = Duration::from_millis(10);
1155
1156 loop {
1157 time::sleep(interval).await;
1158
1159 if self.core.cache().account(&account_id).is_some() {
1160 log::info!("Account {account_id} registered");
1161 return Ok(());
1162 }
1163
1164 if start.elapsed() >= timeout {
1165 anyhow::bail!(
1166 "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
1167 );
1168 }
1169 }
1170 }
1171
1172 async fn establish_session(&mut self) -> anyhow::Result<()> {
1178 if !self.pending_tasks.is_empty()
1180 || !self.session_tasks.is_empty()
1181 || !self.pending_tasks.is_open()
1182 || !self.session_tasks.is_open()
1183 || self.ws_private.is_active()
1184 || self.ws_business.is_active()
1185 || self.ws_private.has_task()
1186 || self.ws_business.has_task()
1187 {
1188 self.teardown_session().await?;
1189 }
1190
1191 if !self.pending_tasks.is_open() {
1192 self.pending_tasks
1193 .start_generation()
1194 .context("failed to start OKX execution request task generation")?;
1195 }
1196
1197 if !self.session_tasks.is_open() {
1198 self.session_tasks
1199 .start_generation()
1200 .context("failed to start OKX execution stream task generation")?;
1201 }
1202 let instrument_types = self.instrument_types();
1203
1204 self.update_position_mode().await?;
1205
1206 if !self.core.instruments_initialized() {
1207 let mut all_instruments = Vec::new();
1208 let mut all_inst_id_codes = Vec::new();
1209
1210 for instrument_type in &instrument_types {
1211 let Some(families) =
1212 resolve_instrument_families(&self.config.instrument_families, *instrument_type)
1213 else {
1214 continue;
1215 };
1216
1217 if families.is_empty() {
1218 let (instruments, inst_id_codes) = self
1219 .http_client
1220 .request_instruments(*instrument_type, None)
1221 .await
1222 .with_context(|| {
1223 format!("failed to request OKX instruments for {instrument_type:?}")
1224 })?;
1225
1226 if instruments.is_empty() {
1227 anyhow::bail!(
1228 "No usable instruments for {instrument_type:?}, \
1229 cannot initialize execution client"
1230 );
1231 }
1232
1233 log::debug!(
1234 "Loaded {} {instrument_type:?} instruments",
1235 instruments.len()
1236 );
1237
1238 self.http_client.cache_instruments(&instruments);
1239 all_instruments.extend(instruments);
1240 all_inst_id_codes.extend(inst_id_codes);
1241 } else {
1242 for family in &families {
1243 let (instruments, inst_id_codes) = self
1244 .http_client
1245 .request_instruments(*instrument_type, Some(family.clone()))
1246 .await
1247 .with_context(|| {
1248 format!(
1249 "failed to request OKX instruments for {instrument_type:?} family {family}"
1250 )
1251 })?;
1252
1253 if instruments.is_empty() {
1254 anyhow::bail!(
1255 "No usable instruments for {instrument_type:?} family {family}, \
1256 cannot initialize execution client"
1257 );
1258 }
1259
1260 log::debug!(
1261 "Loaded {} {instrument_type:?} instruments for family {family}",
1262 instruments.len()
1263 );
1264
1265 self.http_client.cache_instruments(&instruments);
1266 all_instruments.extend(instruments);
1267 all_inst_id_codes.extend(inst_id_codes);
1268 }
1269 }
1270 }
1271
1272 if all_instruments.is_empty() {
1273 anyhow::bail!(
1274 "No instruments loaded for configured types {instrument_types:?}, \
1275 cannot initialize execution client"
1276 );
1277 }
1278
1279 if instrument_types.contains(&OKXInstrumentType::Spot)
1280 && let Err(e) = self
1281 .http_client
1282 .refresh_account_trade_quote_ccy_lists(OKXInstrumentType::Spot, None)
1283 .await
1284 {
1285 log::warn!("Failed to refresh account tradeQuoteCcyList: {e}");
1286 }
1287
1288 let trade_quote_ccy_lists = self.http_client.trade_quote_ccy_lists_snapshot();
1289 self.ws_private.cache_instruments(&all_instruments);
1290 self.ws_private
1291 .cache_inst_id_codes(all_inst_id_codes.clone());
1292 self.ws_private
1293 .cache_trade_quote_ccy_lists(trade_quote_ccy_lists.clone());
1294 self.ws_business.cache_instruments(&all_instruments);
1295 self.ws_business.cache_inst_id_codes(all_inst_id_codes);
1296 self.ws_business
1297 .cache_trade_quote_ccy_lists(trade_quote_ccy_lists);
1298 self.core.set_instruments_initialized();
1299 }
1300
1301 self.ws_private.connect().await?;
1302 self.ws_private.wait_until_active(10.0).await?;
1303 log::info!("Connected to private WebSocket");
1304
1305 {
1306 let stream = self.ws_private.stream();
1307 let emitter = self.emitter.clone();
1308 let state = Arc::clone(&self.ws_dispatch_state);
1309 let account_id = self.core.account_id;
1310 let account_type = self.core.account_type;
1311 let instruments = self.ws_private.instruments_cache_arc();
1312 let tasks = self
1313 .session_tasks
1314 .spawner()
1315 .context("OKX execution stream task admission is closed")?;
1316 let cancel = tasks.cancellation_token();
1317 let clock = self.clock;
1318
1319 spawn_task(&tasks, async move {
1320 let mut fee_cache = FeeCache::new();
1321 let mut filled_qty_cache = FilledQtyCache::new();
1322 let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
1323 AHashMap::new();
1324
1325 pin_mut!(stream);
1326
1327 loop {
1328 tokio::select! {
1329 biased;
1330 () = cancel.cancelled() => break,
1331 message = stream.next() => {
1332 let Some(message) = message else {
1333 break;
1334 };
1335 dispatch_ws_message(
1336 message,
1337 &emitter,
1338 &state,
1339 account_id,
1340 account_type,
1341 &instruments,
1342 &mut fee_cache,
1343 &mut filled_qty_cache,
1344 &mut order_state_cache,
1345 clock,
1346 );
1347 }
1348 () = state.wait_for_linked_child_route() => {
1349 dispatch_ws_message(
1350 OKXWsMessage::Orders(Vec::new()),
1351 &emitter,
1352 &state,
1353 account_id,
1354 account_type,
1355 &instruments,
1356 &mut fee_cache,
1357 &mut filled_qty_cache,
1358 &mut order_state_cache,
1359 clock,
1360 );
1361 }
1362 }
1363 }
1364 });
1365 }
1366
1367 self.ws_business.connect().await?;
1368 self.ws_business.wait_until_active(10.0).await?;
1369 log::info!("Connected to business WebSocket");
1370
1371 {
1372 let stream = self.ws_business.stream();
1373 let emitter = self.emitter.clone();
1374 let state = Arc::clone(&self.ws_dispatch_state);
1375 let account_id = self.core.account_id;
1376 let account_type = self.core.account_type;
1377 let instruments = self.ws_business.instruments_cache_arc();
1378 let tasks = self
1379 .session_tasks
1380 .spawner()
1381 .context("OKX execution stream task admission is closed")?;
1382 let cancel = tasks.cancellation_token();
1383 let clock = self.clock;
1384
1385 spawn_task(&tasks, async move {
1386 let mut fee_cache = FeeCache::new();
1387 let mut filled_qty_cache = FilledQtyCache::new();
1388 let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
1389 AHashMap::new();
1390
1391 pin_mut!(stream);
1392
1393 loop {
1394 tokio::select! {
1395 biased;
1396 () = cancel.cancelled() => break,
1397 message = stream.next() => {
1398 let Some(message) = message else {
1399 break;
1400 };
1401 dispatch_ws_message(
1402 message,
1403 &emitter,
1404 &state,
1405 account_id,
1406 account_type,
1407 &instruments,
1408 &mut fee_cache,
1409 &mut filled_qty_cache,
1410 &mut order_state_cache,
1411 clock,
1412 );
1413 }
1414 }
1415 }
1416 });
1417 }
1418
1419 let order_routing_types = order_routing_instrument_types(&instrument_types);
1420
1421 for inst_type in &order_routing_types {
1422 log::debug!("Subscribing to orders channel for {inst_type:?}");
1423 self.ws_private.subscribe_orders(*inst_type).await?;
1424 }
1425
1426 self.ws_private.subscribe_account().await?;
1427
1428 if order_routing_types.iter().any(|t| {
1430 matches!(
1431 t,
1432 OKXInstrumentType::Margin
1433 | OKXInstrumentType::Swap
1434 | OKXInstrumentType::Futures
1435 | OKXInstrumentType::Option
1436 )
1437 }) {
1438 log::debug!("Subscribing to liquidation warning channel");
1439 self.ws_private
1440 .subscribe_liquidation_warning(OKXInstrumentType::Any)
1441 .await?;
1442 }
1443
1444 if self.config.load_spreads {
1445 log::debug!("Subscribing to Nitro spread orders channel");
1446 self.ws_business.subscribe_spread_orders().await?;
1447 }
1448
1449 for inst_type in &order_routing_types {
1451 if supports_algo_orders(*inst_type) {
1452 self.ws_business.subscribe_orders_algo(*inst_type).await?;
1453 self.ws_business.subscribe_algo_advance(*inst_type).await?;
1454 }
1455 }
1456
1457 let account_state = self
1458 .http_client
1459 .request_account_state(self.core.account_id, self.core.account_type)
1460 .await
1461 .context("failed to request OKX account state")?;
1462
1463 if !account_state.balances.is_empty() {
1464 log::debug!(
1465 "Received account state with {} balance(s)",
1466 account_state.balances.len()
1467 );
1468 }
1469 self.emitter.send_account_state(account_state);
1470
1471 self.await_account_registered(30.0).await?;
1473
1474 Ok(())
1475 }
1476
1477 async fn update_position_mode(&mut self) -> anyhow::Result<()> {
1478 let instrument_types = self.instrument_types();
1479 if instrument_types.iter().any(|instrument_type| {
1480 matches!(
1481 instrument_type,
1482 OKXInstrumentType::Swap | OKXInstrumentType::Futures
1483 )
1484 }) {
1485 let configurations = self
1486 .http_client
1487 .raw_client()
1488 .get_account_configuration()
1489 .await?;
1490 self.position_mode = configurations
1491 .first()
1492 .context("OKX account configuration response is empty")?
1493 .position_mode;
1494
1495 if self.position_mode == OKXPositionMode::LongShortMode {
1496 let report_client = self.report_client();
1497 let scope = report_client.report_scope(&instrument_types);
1498 for inst_type in &instrument_types {
1499 if !matches!(
1500 inst_type,
1501 OKXInstrumentType::Swap | OKXInstrumentType::Futures
1502 ) {
1503 continue;
1504 }
1505
1506 let positions = self
1507 .http_client
1508 .get_positions(GetPositionsParams {
1509 inst_type: Some(*inst_type),
1510 ..Default::default()
1511 })
1512 .await?;
1513
1514 for position in positions {
1515 scope.validate_position_margin(&position)?;
1516 }
1517 }
1518 }
1519
1520 let expected_oms = match self.position_mode {
1521 OKXPositionMode::NetMode => OmsType::Netting,
1522 OKXPositionMode::LongShortMode => OmsType::Hedging,
1523 };
1524
1525 if self.core.oms_type != expected_oms {
1526 log::warn!(
1527 "OKX position mode {:?} differs from OMS type {:?}, set oms_type to {expected_oms}",
1528 self.position_mode,
1529 self.core.oms_type,
1530 );
1531 }
1532 }
1533
1534 Ok(())
1535 }
1536
1537 async fn teardown_session(&mut self) -> anyhow::Result<()> {
1540 self.begin_generation_shutdown();
1541 self.ws_private.request_close().await;
1542 self.ws_business.request_close().await;
1543 let pending_result = terminate_tasks(&self.pending_tasks, "OKX execution request").await;
1544 let session_result = terminate_tasks(&self.session_tasks, "OKX execution stream").await;
1545
1546 let private_result = self
1547 .ws_private
1548 .close()
1549 .await
1550 .context("failed to close private websocket");
1551 let business_result = self
1552 .ws_business
1553 .close()
1554 .await
1555 .context("failed to close business websocket");
1556
1557 self.core.set_disconnected();
1558
1559 let mut errors = Vec::new();
1560 if let Err(e) = pending_result {
1561 errors.push(e.to_string());
1562 }
1563
1564 if let Err(e) = session_result {
1565 errors.push(e.to_string());
1566 }
1567
1568 if let Err(e) = private_result {
1569 errors.push(e.to_string());
1570 }
1571
1572 if let Err(e) = business_result {
1573 errors.push(e.to_string());
1574 }
1575
1576 if errors.is_empty() {
1577 Ok(())
1578 } else {
1579 anyhow::bail!(errors.join("; "))
1580 }
1581 }
1582}
1583
1584fn derive_trade_mode_for_instrument(
1585 instrument_id: InstrumentId,
1586 margin_mode: Option<OKXMarginMode>,
1587 use_spot_margin: bool,
1588) -> OKXTradeMode {
1589 let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
1590 let is_cross_margin = margin_mode == Some(OKXMarginMode::Cross);
1591
1592 match inst_type {
1593 OKXInstrumentType::Spot => {
1594 if use_spot_margin {
1595 if is_cross_margin {
1596 OKXTradeMode::Cross
1597 } else {
1598 OKXTradeMode::Isolated
1599 }
1600 } else {
1601 OKXTradeMode::Cash
1602 }
1603 }
1604 _ => {
1605 if is_cross_margin {
1606 OKXTradeMode::Cross
1607 } else {
1608 OKXTradeMode::Isolated
1609 }
1610 }
1611 }
1612}
1613
1614#[async_trait(?Send)]
1615impl ExecutionClient for OKXExecutionClient {
1616 fn is_connected(&self) -> bool {
1617 self.core.is_connected()
1618 }
1619
1620 fn client_id(&self) -> ClientId {
1621 self.core.client_id
1622 }
1623
1624 fn account_id(&self) -> AccountId {
1625 self.core.account_id
1626 }
1627
1628 fn venue(&self) -> Venue {
1629 *OKX_VENUE
1630 }
1631
1632 fn oms_type(&self) -> OmsType {
1633 self.core.oms_type
1634 }
1635
1636 fn get_account(&self) -> Option<AccountAny> {
1637 self.core.cache().account_owned(&self.core.account_id)
1638 }
1639
1640 fn register_external_order(
1641 &self,
1642 client_order_id: ClientOrderId,
1643 venue_order_id: VenueOrderId,
1644 _instrument_id: InstrumentId,
1645 _strategy_id: StrategyId,
1646 _ts_init: UnixNanos,
1647 ) {
1648 let cache = self.core.cache();
1649
1650 let Some(order) = cache.order_ref(&client_order_id) else {
1651 log::warn!("Cannot track imported OKX order {client_order_id}: order not cached");
1652 return;
1653 };
1654
1655 if order.order_type() != OrderType::Limit {
1656 return;
1657 }
1658
1659 let context = OrderContext::from(&*order);
1660 self.ws_dispatch_state
1661 .register_external_order(context, venue_order_id);
1662 }
1663
1664 async fn connect(&mut self) -> anyhow::Result<()> {
1665 if self.core.is_connected() && self.pending_tasks.is_open() && self.session_tasks.is_open()
1666 {
1667 return Ok(());
1668 }
1669 let ws_private = self.ws_private.clone();
1670 let ws_business = self.ws_business.clone();
1671 let setup_guard =
1672 TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
1673 ws_private.begin_shutdown();
1674 ws_business.begin_shutdown();
1675 });
1676
1677 if let Err(e) = self.establish_session().await {
1678 if let Err(teardown_error) = self.teardown_session().await {
1679 return Err(e.context(format!(
1680 "OKX execution startup teardown failed: {teardown_error}"
1681 )));
1682 }
1683 return Err(e);
1684 }
1685
1686 self.core.set_connected();
1687 setup_guard.disarm();
1688 log::info!("Connected: client_id={}", self.core.client_id);
1689 Ok(())
1690 }
1691
1692 async fn disconnect(&mut self) -> anyhow::Result<()> {
1693 if self.core.is_disconnected()
1694 && self.pending_tasks.is_empty()
1695 && self.session_tasks.is_empty()
1696 && !self.ws_private.has_task()
1697 && !self.ws_business.has_task()
1698 {
1699 return Ok(());
1700 }
1701
1702 self.teardown_session().await?;
1703 log::info!("Disconnected: client_id={}", self.core.client_id);
1704 Ok(())
1705 }
1706
1707 fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
1708 self.update_account_state();
1709 Ok(())
1710 }
1711
1712 fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
1713 let http_client = self.http_client.clone();
1714 let account_id = self.core.account_id;
1715 let emitter = self.emitter.clone();
1716 let instrument_id = cmd.instrument_id;
1717 let client_order_id = cmd.client_order_id;
1718 let venue_order_id = cmd.venue_order_id;
1719 let order_state = {
1720 let cache = self.core.cache();
1721 cache
1722 .order(&client_order_id)
1723 .map(|order| CachedQueryOrderState {
1724 order_type: order.order_type(),
1725 venue_order_id: order.venue_order_id(),
1726 })
1727 };
1728
1729 let venue_binding = self.ws_dispatch_state.order_venue_binding(client_order_id);
1730 let authoritative_venue_order_id = venue_binding.map(|(venue_order_id, _)| venue_order_id);
1731 let has_bound_child = venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child);
1732 let cached_venue_order_id = order_state.and_then(|state| state.venue_order_id);
1733 let regular_venue_order_id = if has_bound_child {
1734 authoritative_venue_order_id
1735 } else {
1736 order_state.and_then(|state| {
1737 if OKX_CONDITIONAL_ORDER_TYPES.contains(&state.order_type) {
1738 state.venue_order_id.or(venue_order_id)
1739 } else {
1740 state.venue_order_id
1741 }
1742 })
1743 };
1744
1745 let selection_venue_order_id = authoritative_venue_order_id
1746 .or(cached_venue_order_id)
1747 .or(venue_order_id);
1748 let route = query_order_route(
1749 instrument_id,
1750 order_state.map(|state| state.order_type),
1751 regular_venue_order_id.is_some(),
1752 self.http_client.is_spread(instrument_id),
1753 );
1754 self.spawn_task("query_order", async move {
1755 let mut reports = Vec::with_capacity(1);
1756 let mut query_algo = matches!(
1757 route,
1758 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo
1759 );
1760
1761 match route {
1762 QueryOrderRoute::Spread => {
1763 match http_client
1764 .request_order_status_reports(
1765 account_id,
1766 None,
1767 Some(instrument_id),
1768 None,
1769 None,
1770 false,
1771 None,
1772 )
1773 .await
1774 {
1775 Ok(spread_reports) => reports.extend(spread_reports),
1776 Err(e) => {
1777 log::warn!("OKX query_order failed to fetch spread order: {e}");
1778 }
1779 }
1780 }
1781 QueryOrderRoute::Regular | QueryOrderRoute::RegularThenAlgo => {
1782 let result = if let Some(venue_order_id) = regular_venue_order_id {
1783 http_client
1784 .request_order_status_report_by_venue_order_id(
1785 account_id,
1786 instrument_id,
1787 venue_order_id,
1788 )
1789 .await
1790 } else {
1791 http_client
1792 .request_order_status_report(
1793 account_id,
1794 instrument_id,
1795 client_order_id,
1796 )
1797 .await
1798 };
1799
1800 match result {
1801 Ok(Some(report)) => reports.push(report),
1802 Ok(None) => {
1803 query_algo |= route == QueryOrderRoute::RegularThenAlgo;
1804 }
1805 Err(e) => {
1806 log::warn!("OKX query_order failed to fetch regular order: {e}");
1807 }
1808 }
1809 }
1810 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo => {}
1811 }
1812
1813 if query_algo {
1818 let mut regular_child_venue_order_id = None;
1819
1820 match http_client
1821 .request_algo_order_status_reports(
1822 account_id,
1823 None,
1824 Some(instrument_id),
1825 None,
1826 Some(client_order_id),
1827 None,
1828 Some(1),
1829 )
1830 .await
1831 {
1832 Ok(algo_reports) => {
1833 if matches!(
1834 route,
1835 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo
1836 ) {
1837 regular_child_venue_order_id = algo_reports
1838 .iter()
1839 .find(|report| {
1840 matches!(
1841 report.order_status,
1842 OrderStatus::Triggered | OrderStatus::Filled
1843 )
1844 })
1845 .map(|report| report.venue_order_id)
1846 .or_else(|| {
1847 regular_venue_order_id.filter(|venue_order_id| {
1848 algo_reports.first().is_none_or(|report| {
1849 report.venue_order_id != *venue_order_id
1850 })
1851 })
1852 });
1853 }
1854
1855 merge_order_status_reports(&mut reports, algo_reports);
1856 }
1857 Err(e) => {
1858 if route == QueryOrderRoute::RegularAndAlgo {
1859 regular_child_venue_order_id = regular_venue_order_id;
1860 }
1861
1862 log::warn!("OKX query_order algo lookup failed for {instrument_id}: {e}");
1863 }
1864 }
1865
1866 if let Some(child_venue_order_id) = regular_child_venue_order_id {
1867 match http_client
1868 .request_order_status_report_by_venue_order_id(
1869 account_id,
1870 instrument_id,
1871 child_venue_order_id,
1872 )
1873 .await
1874 {
1875 Ok(Some(child_report)) => {
1876 merge_order_status_reports(&mut reports, vec![child_report]);
1877 }
1878 Ok(None) => {}
1879 Err(e) => {
1880 log::warn!(
1881 "OKX query_order failed to fetch regular child order: {e}"
1882 );
1883 }
1884 }
1885 }
1886 }
1887
1888 let Some(report) = select_query_order_report(
1889 reports,
1890 client_order_id,
1891 selection_venue_order_id,
1892 ) else {
1893 log::warn!(
1894 "OKX query_order found no order for client_order_id={client_order_id}, venue_order_id={selection_venue_order_id:?}",
1895 );
1896 return Ok(());
1897 };
1898
1899 emitter.send_order_status_report(report);
1900 Ok(())
1901 });
1902 Ok(())
1903 }
1904
1905 fn generate_account_state(
1906 &self,
1907 balances: Vec<AccountBalance>,
1908 margins: Vec<MarginBalance>,
1909 reported: bool,
1910 ts_event: UnixNanos,
1911 info: Option<Params>,
1912 ) -> anyhow::Result<()> {
1913 self.emitter
1914 .emit_account_state(balances, margins, reported, ts_event, info);
1915 Ok(())
1916 }
1917
1918 fn on_instrument(&mut self, instrument: InstrumentAny) {
1919 self.http_client.cache_instrument(instrument.clone());
1920 self.ws_private.cache_instrument(instrument.clone());
1921 self.ws_business.cache_instrument(instrument);
1922 }
1923
1924 fn start(&mut self) -> anyhow::Result<()> {
1925 if self.core.is_started() {
1926 return Ok(());
1927 }
1928
1929 let sender = get_exec_event_sender();
1930 self.emitter.set_sender(sender);
1931 self.core.set_started();
1932
1933 log::info!(
1934 "Started: client_id={}, account_id={}, account_type={:?}, trade_mode={:?}, instrument_types={:?}, environment={}, proxy_url={:?}",
1935 self.core.client_id,
1936 self.core.account_id,
1937 self.core.account_type,
1938 self.trade_mode,
1939 self.config.instrument_types,
1940 self.config.environment,
1941 self.config.proxy_url,
1942 );
1943 Ok(())
1944 }
1945
1946 fn stop(&mut self) -> anyhow::Result<()> {
1947 let was_started = self.core.is_started();
1948 self.core.set_stopped();
1949 self.begin_generation_shutdown();
1950
1951 if was_started {
1952 log::info!("Stopped: client_id={}", self.core.client_id);
1953 }
1954 Ok(())
1955 }
1956
1957 fn reset(&mut self) -> anyhow::Result<()> {
1958 self.begin_generation_shutdown();
1959 Ok(())
1960 }
1961
1962 fn dispose(&mut self) -> anyhow::Result<()> {
1963 self.begin_generation_shutdown();
1964 Ok(())
1965 }
1966
1967 fn generate_order_status_report_task(
1968 &self,
1969 cmd: &GenerateOrderStatusReport,
1970 ) -> Option<ExecutionReportTask<Option<OrderStatusReport>>> {
1971 let client = self.report_client();
1972 let order_state = self.query_order_state(cmd);
1973 let command = cmd.clone();
1974 Some(ExecutionReportTask::new(
1975 async move {
1976 client
1977 .generate_order_status_report(&command, order_state)
1978 .await
1979 },
1980 Ok,
1981 ))
1982 }
1983
1984 fn generate_order_status_reports_task(
1985 &self,
1986 cmd: &GenerateOrderStatusReports,
1987 ) -> Option<ExecutionReportTask<Vec<OrderStatusReport>>> {
1988 let client = self.report_client();
1989 let core = self.core.clone();
1990 let command = cmd.clone();
1991 let collection_command = command.clone();
1992 Some(ExecutionReportTask::new(
1993 async move {
1994 let mut sweep = client
1995 .collect_order_status_reports(&collection_command, false, None)
1996 .await?;
1997 sweep.regular_by_venue_order_id = AHashMap::new();
1998 sweep.ambiguous_triggered_child_ids = AHashSet::new();
1999 Ok(sweep)
2000 },
2001 move |sweep| Ok(finish_order_report_sweep(&core, sweep, &command).reports),
2002 ))
2003 }
2004
2005 fn generate_fill_reports_task(
2006 &self,
2007 cmd: &GenerateFillReports,
2008 ) -> Option<ExecutionReportTask<Vec<FillReport>>> {
2009 let client = self.report_client();
2010 let command = cmd.clone();
2011 Some(ExecutionReportTask::new(
2012 async move {
2013 client
2014 .collect_fill_reports(command, FillHistory::Recent, None)
2015 .await
2016 },
2017 |(reports, _)| Ok(reports),
2018 ))
2019 }
2020
2021 fn generate_position_status_reports_task(
2022 &self,
2023 cmd: &GeneratePositionStatusReports,
2024 ) -> Option<ExecutionReportTask<Vec<PositionStatusReport>>> {
2025 let client = self.report_client();
2026 let command = cmd.clone();
2027 Some(ExecutionReportTask::new(
2028 async move { client.collect_position_status_reports(&command).await },
2029 |(reports, _)| Ok(reports),
2030 ))
2031 }
2032
2033 async fn generate_order_status_report(
2034 &self,
2035 cmd: &GenerateOrderStatusReport,
2036 ) -> anyhow::Result<Option<OrderStatusReport>> {
2037 self.report_client()
2038 .generate_order_status_report(cmd, self.query_order_state(cmd))
2039 .await
2040 }
2041
2042 async fn generate_order_status_reports(
2043 &self,
2044 cmd: &GenerateOrderStatusReports,
2045 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2046 Ok(self
2047 .collect_order_status_reports(cmd, false, None)
2048 .await?
2049 .reports)
2050 }
2051
2052 async fn generate_fill_reports(
2053 &self,
2054 cmd: GenerateFillReports,
2055 ) -> anyhow::Result<Vec<FillReport>> {
2056 Ok(self
2057 .collect_fill_reports(cmd, FillHistory::Recent, None)
2058 .await?
2059 .0)
2060 }
2061
2062 async fn generate_position_status_reports(
2063 &self,
2064 cmd: &GeneratePositionStatusReports,
2065 ) -> anyhow::Result<Vec<PositionStatusReport>> {
2066 Ok(self.collect_position_status_reports(cmd).await?.0)
2067 }
2068
2069 async fn generate_mass_status(
2070 &self,
2071 lookback_mins: Option<u64>,
2072 ) -> anyhow::Result<Option<ExecutionMassStatus>> {
2073 log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
2074
2075 let ts_now = self.clock.get_time_ns();
2076
2077 let lookback_mins = lookback_mins
2078 .unwrap_or(OKX_RECONCILIATION_LOOKBACK_DEFAULT_MINS)
2079 .min(OKX_RECONCILIATION_LOOKBACK_MAX_MINS);
2080 let fill_history = if lookback_mins <= OKX_RECONCILIATION_LOOKBACK_DEFAULT_MINS {
2081 FillHistory::Recent
2082 } else {
2083 FillHistory::Extended
2084 };
2085 let lookback = DurationNanos::try_from_mins(lookback_mins)?;
2086 let start = Some(ts_now.saturating_sub(lookback));
2087
2088 let order_cmd = GenerateOrderStatusReportsBuilder::default()
2089 .ts_init(ts_now)
2090 .open_only(false) .start(start)
2092 .build()
2093 .map_err(|e| anyhow::anyhow!("{e}"))?;
2094
2095 let fill_cmd = GenerateFillReportsBuilder::default()
2096 .ts_init(ts_now)
2097 .start(start)
2098 .build()
2099 .map_err(|e| anyhow::anyhow!("{e}"))?;
2100
2101 let position_cmd = GeneratePositionStatusReportsBuilder::default()
2102 .ts_init(ts_now)
2103 .start(start)
2104 .build()
2105 .map_err(|e| anyhow::anyhow!("{e}"))?;
2106
2107 let (
2108 order_sweep,
2109 (fill_reports, fills_complete),
2110 (mut position_reports, positions_complete),
2111 ) = tokio::try_join!(
2112 self.collect_order_status_reports(
2113 &order_cmd,
2114 true,
2115 self.config.max_reconciliation_history_pages,
2116 ),
2117 self.collect_fill_reports(
2118 fill_cmd,
2119 fill_history,
2120 self.config.max_reconciliation_history_pages,
2121 ),
2122 self.collect_position_status_reports(&position_cmd),
2123 )?;
2124 let OrderReportSweep {
2125 fallback_parents: _,
2126 reports: mut order_reports,
2127 complete: mut orders_complete,
2128 ambiguous_triggered_child_ids,
2129 regular_by_venue_order_id,
2130 } = order_sweep;
2131 orders_complete &= self
2132 .recover_triggered_child_order_reports(
2133 &mut order_reports,
2134 &ambiguous_triggered_child_ids,
2135 ®ular_by_venue_order_id,
2136 )
2137 .await;
2138
2139 if positions_complete {
2140 self.add_flat_derivative_position_reports(
2141 &order_reports,
2142 &fill_reports,
2143 &mut position_reports,
2144 ts_now,
2145 );
2146 }
2147 let reports_complete = orders_complete && fills_complete && positions_complete;
2148
2149 log::info!("Received {} OrderStatusReports", order_reports.len());
2150 log::info!("Received {} FillReports", fill_reports.len());
2151 log::info!("Received {} PositionReports", position_reports.len());
2152
2153 let mut mass_status = ExecutionMassStatus::new(
2154 self.core.client_id,
2155 self.core.account_id,
2156 *OKX_VENUE,
2157 ts_now,
2158 None,
2159 );
2160 mass_status.set_report_window(start, reports_complete);
2161 mass_status.add_order_reports(order_reports);
2162 mass_status.add_fill_reports(fill_reports);
2163 mass_status.add_position_reports(position_reports);
2164
2165 Ok(Some(mass_status))
2166 }
2167
2168 fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
2169 let (route, position_side) = {
2170 let cache = self.core.cache();
2171 let order = cache.try_order(&cmd.client_order_id)?;
2172
2173 if order.is_closed() {
2174 log::warn!("Cannot submit closed order {}", order.client_order_id());
2175 return Ok(());
2176 }
2177
2178 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
2179
2180 if let Err(reason) = validate_order(
2181 &*order,
2182 trade_mode,
2183 OrderSubmission::Single,
2184 self.http_client.is_spread(order.instrument_id()),
2185 ) {
2186 self.emitter.emit_order_denied(&order, &reason.to_string());
2187 return Ok(());
2188 }
2189
2190 let position_side = self.order_position_side(&*order, cmd.params.as_ref());
2191 if let Err(reason) = self
2192 .validate_trade_mode(cmd.instrument_id, position_side, trade_mode)
2193 .and_then(|()| {
2194 validate_submit_position_id(cmd.instrument_id, position_side, cmd.position_id)
2195 })
2196 {
2197 self.emitter.emit_order_denied(&order, &reason.to_string());
2198 return Ok(());
2199 }
2200
2201 let order_type = order.order_type();
2202 let route = self.submit_order_route(cmd.instrument_id, order_type)?;
2203
2204 log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
2205 self.emitter.emit_order_submitted(&order);
2206
2207 (route, position_side)
2208 };
2209
2210 match route {
2211 OrderCommandRoute::RegularWs => self.submit_regular_order(&cmd, position_side),
2212 OrderCommandRoute::AlgoHttp => self.submit_conditional_order(&cmd, position_side),
2213 OrderCommandRoute::SpreadHttp => self.submit_order_http(&cmd, position_side),
2214 }
2215 }
2216
2217 fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
2218 if self.http_client.is_spread(cmd.instrument_id) {
2219 let cache = self.core.cache();
2220 let denied = OrderDeniedReason::UnsupportedOrderList {
2221 detail: "spread instruments are not supported in order lists".to_string(),
2222 }
2223 .to_string();
2224
2225 for client_order_id in &cmd.order_list.client_order_ids {
2226 let order = cache.try_order(client_order_id)?;
2227 self.emitter.emit_order_denied(&order, &denied);
2228 }
2229 return Ok(());
2230 }
2231
2232 let inst_type = okx_instrument_type_from_symbol(cmd.instrument_id.symbol.as_str());
2233 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
2234
2235 let orders = self.core.get_orders_for_list(&cmd.order_list)?;
2237
2238 let invalid: Vec<(ClientOrderId, OrderDeniedReason)> = orders
2241 .iter()
2242 .filter_map(|order| {
2243 validate_order(
2244 order,
2245 trade_mode,
2246 OrderSubmission::List,
2247 self.http_client.is_spread(order.instrument_id()),
2248 )
2249 .and_then(|()| {
2250 self.validate_trade_mode(
2251 cmd.instrument_id,
2252 self.order_position_side(order, cmd.params.as_ref()),
2253 trade_mode,
2254 )
2255 })
2256 .and_then(|()| {
2257 validate_submit_position_id(
2258 cmd.instrument_id,
2259 self.order_position_side(order, cmd.params.as_ref()),
2260 cmd.position_id,
2261 )
2262 })
2263 .err()
2264 .map(|reason| (order.client_order_id(), reason))
2265 })
2266 .collect();
2267
2268 if !invalid.is_empty() {
2269 let order_list_id = cmd.order_list.id;
2270
2271 for order in &orders {
2272 let denied = invalid
2273 .iter()
2274 .find(|(client_order_id, _)| client_order_id == &order.client_order_id())
2275 .map_or_else(
2276 || OrderDeniedReason::OrderListDenied { order_list_id },
2277 |(_, reason)| reason.clone(),
2278 );
2279 self.emitter.emit_order_denied(order, &denied.to_string());
2280 }
2281 return Ok(());
2282 }
2283
2284 let mut batch_orders = Vec::new();
2286 let outcome = get_param_as_string(cmd.params.as_ref(), "outcome");
2287 let rpi = get_param_as_bool(cmd.params.as_ref(), "rpi");
2288 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
2289 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
2290
2291 for order in &orders {
2292 let context = OrderContext::from(order);
2293
2294 batch_orders.push((
2295 inst_type,
2296 cmd.instrument_id,
2297 trade_mode,
2298 context.identity.client_order_id,
2299 context.identity.order_side,
2300 self.order_position_side(order, cmd.params.as_ref()),
2301 context.identity.order_type,
2302 context.quantity,
2303 context.price,
2304 context.trigger_price,
2305 Some(context.is_post_only),
2306 Some(context.is_reduce_only),
2307 outcome.clone(),
2308 rpi,
2309 rpi_taker_access,
2310 rpi_px_round,
2311 ));
2312
2313 self.ws_dispatch_state
2314 .order_identities
2315 .insert(context.identity.client_order_id, context.identity);
2316
2317 log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
2318 self.emitter.emit_order_submitted(order);
2319 }
2320
2321 let ws_private = self.ws_private.clone();
2322 let emitter = self.emitter.clone();
2323 let clock = self.clock;
2324 let instrument_id = cmd.instrument_id;
2325 let strategy_id = cmd.strategy_id;
2326 let client_order_ids: Vec<_> = cmd.order_list.client_order_ids;
2327 let dispatch_state = Arc::clone(&self.ws_dispatch_state);
2328
2329 self.spawn_task("batch_submit_orders", async move {
2330 let result = ws_private
2331 .batch_submit_orders(batch_orders)
2332 .await;
2333
2334 if let Err(e) = result {
2335 match classify_okx_ws_failure(&e) {
2336 CommandFailure::NotSent(reason) => {
2337 let ts_event = clock.get_time_ns();
2338
2339 for cid in &client_order_ids {
2340 dispatch_state.order_identities.remove(cid);
2341 emitter.emit_order_rejected_event(
2342 strategy_id,
2343 instrument_id,
2344 *cid,
2345 &reason,
2346 ts_event,
2347 false,
2348 );
2349 }
2350 }
2351 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
2352 log::warn!(
2353 "Ambiguous batch submit failure for {} orders on {instrument_id}, awaiting reconciliation: {reason}",
2354 client_order_ids.len()
2355 );
2356 }
2357 }
2358 return Err(anyhow::Error::new(e).context("batch submit orders failed"));
2359 }
2360
2361 Ok(())
2362 });
2363
2364 Ok(())
2365 }
2366
2367 fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
2368 if self.http_client.is_spread(cmd.instrument_id) {
2369 self.emitter.emit_order_modify_rejected_event(
2370 cmd.strategy_id,
2371 cmd.instrument_id,
2372 cmd.client_order_id,
2373 cmd.venue_order_id,
2374 "OKX spread orders do not support modify requests",
2375 self.clock.get_time_ns(),
2376 );
2377 return Ok(());
2378 }
2379
2380 let venue_binding = self
2381 .ws_dispatch_state
2382 .order_venue_binding(cmd.client_order_id);
2383
2384 let route = {
2385 let cache = self.core.cache();
2386 let order_state = cache
2387 .order(&cmd.client_order_id)
2388 .map(|order| (order.order_type(), order.is_triggered()));
2389 self.order_command_route(
2390 cmd.instrument_id,
2391 order_state,
2392 venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child),
2393 )
2394 };
2395
2396 let mut command = cmd;
2397 command.venue_order_id = venue_binding
2398 .map(|(venue_order_id, _)| venue_order_id)
2399 .or(command.venue_order_id);
2400 if route == OrderCommandRoute::AlgoHttp {
2401 self.modify_algo_order(&command);
2402 return Ok(());
2403 }
2404
2405 self.ensure_order_identity(
2406 command.client_order_id,
2407 command.strategy_id,
2408 command.instrument_id,
2409 );
2410 let ws_private = self.ws_private.clone();
2411
2412 let new_px_usd = get_param_as_string(command.params.as_ref(), "px_usd");
2413 let new_px_vol = get_param_as_string(command.params.as_ref(), "px_vol");
2414 let rpi_taker_access = get_param_as_bool(command.params.as_ref(), "rpi_taker_access");
2415 let rpi_px_round = get_param_as_bool(command.params.as_ref(), "rpi_px_round");
2416
2417 let emitter = self.emitter.clone();
2418 let clock = self.clock;
2419
2420 self.spawn_task("modify_order", async move {
2421 let result = ws_private
2422 .modify_order(
2423 command.trader_id,
2424 command.strategy_id,
2425 command.instrument_id,
2426 Some(command.client_order_id),
2427 command.price,
2428 command.quantity,
2429 command.venue_order_id,
2430 new_px_usd,
2431 new_px_vol,
2432 rpi_taker_access,
2433 rpi_px_round,
2434 )
2435 .await;
2436
2437 if let Err(e) = result {
2438 emit_modify_failure(
2439 classify_okx_ws_failure(&e),
2440 &emitter,
2441 clock,
2442 command.strategy_id,
2443 command.instrument_id,
2444 command.client_order_id,
2445 command.venue_order_id,
2446 );
2447 return Err(anyhow::Error::new(e).context("modify order failed"));
2448 }
2449
2450 Ok(())
2451 });
2452
2453 Ok(())
2454 }
2455
2456 fn cancel_order(&self, mut cmd: CancelOrder) -> anyhow::Result<()> {
2457 let venue_binding = self
2458 .ws_dispatch_state
2459 .order_venue_binding(cmd.client_order_id);
2460 let route = {
2461 let cache = self.core.cache();
2462 let order_state = cache
2463 .order(&cmd.client_order_id)
2464 .map(|order| (order.order_type(), order.is_triggered()));
2465 self.order_command_route(
2466 cmd.instrument_id,
2467 order_state,
2468 venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child),
2469 )
2470 };
2471
2472 cmd.venue_order_id = venue_binding
2473 .map(|(venue_order_id, _)| venue_order_id)
2474 .or(cmd.venue_order_id);
2475
2476 match route {
2477 OrderCommandRoute::RegularWs => self.cancel_ws_order(&cmd),
2478 OrderCommandRoute::AlgoHttp => self.cancel_algo_order(&cmd),
2479 OrderCommandRoute::SpreadHttp => self.cancel_order_http(&cmd),
2480 }
2481 Ok(())
2482 }
2483
2484 fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
2485 match self.cancel_all_orders_route(cmd.instrument_id, cmd.order_side) {
2486 CancelAllOrdersRoute::SpreadHttp | CancelAllOrdersRoute::MassCancelHttp => {
2487 self.mass_cancel_instrument(cmd.instrument_id);
2488 Ok(())
2489 }
2490 CancelAllOrdersRoute::Batch => {
2491 let cancels: Vec<_> = {
2492 let cache = self.core.cache();
2493 cache
2494 .orders_open(None, Some(&cmd.instrument_id), None, None, cmd.order_side)
2495 .iter()
2496 .map(|order| CancelOrder {
2497 trader_id: order.trader_id(),
2498 client_id: cmd.client_id,
2499 strategy_id: order.strategy_id(),
2500 instrument_id: order.instrument_id(),
2501 client_order_id: order.client_order_id(),
2502 venue_order_id: order.venue_order_id(),
2503 command_id: cmd.command_id,
2504 ts_init: cmd.ts_init,
2505 params: cmd.params.clone(),
2506 correlation_id: cmd.correlation_id,
2507 causation_id: cmd.causation_id,
2508 })
2509 .collect()
2510 };
2511
2512 if cancels.is_empty() {
2513 log::debug!("No open orders to cancel for {}", cmd.instrument_id);
2514 return Ok(());
2515 }
2516
2517 self.batch_cancel_orders(BatchCancelOrders {
2518 trader_id: cmd.trader_id,
2519 client_id: cmd.client_id,
2520 strategy_id: cmd.strategy_id,
2521 instrument_id: cmd.instrument_id,
2522 cancels,
2523 command_id: cmd.command_id,
2524 ts_init: cmd.ts_init,
2525 params: cmd.params,
2526 correlation_id: cmd.correlation_id,
2527 causation_id: cmd.causation_id,
2528 })
2529 }
2530 }
2531 }
2532
2533 fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
2534 let cache = self.core.cache();
2535
2536 let mut regular_payload = Vec::new();
2537 let mut algo_orders = Vec::new();
2538 let mut http_orders = Vec::new();
2539
2540 for cancel in &cmd.cancels {
2541 let order_state = cache
2542 .order(&cancel.client_order_id)
2543 .map(|order| (order.order_type(), order.is_triggered()));
2544
2545 let venue_binding = self
2546 .ws_dispatch_state
2547 .order_venue_binding(cancel.client_order_id);
2548 let authoritative_venue_order_id = venue_binding
2549 .map(|(venue_order_id, _)| venue_order_id)
2550 .or(cancel.venue_order_id);
2551
2552 match self.order_command_route(
2553 cancel.instrument_id,
2554 order_state,
2555 venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child),
2556 ) {
2557 OrderCommandRoute::RegularWs => {
2558 self.ensure_order_identity(
2559 cancel.client_order_id,
2560 cancel.strategy_id,
2561 cancel.instrument_id,
2562 );
2563 regular_payload.push((
2564 cancel.instrument_id,
2565 Some(cancel.client_order_id),
2566 authoritative_venue_order_id,
2567 ));
2568 }
2569 OrderCommandRoute::AlgoHttp => {
2570 let mut cancel = cancel.clone();
2571 cancel.venue_order_id = authoritative_venue_order_id;
2572 algo_orders.push(cancel);
2573 }
2574 OrderCommandRoute::SpreadHttp => {
2575 self.ensure_order_identity(
2576 cancel.client_order_id,
2577 cancel.strategy_id,
2578 cancel.instrument_id,
2579 );
2580 http_orders.push((
2581 cancel.client_order_id,
2582 cancel.instrument_id,
2583 cancel.strategy_id,
2584 authoritative_venue_order_id,
2585 ));
2586 }
2587 }
2588 }
2589 drop(cache);
2590
2591 if !regular_payload.is_empty() {
2592 let ws_private = self.ws_private.clone();
2593 let state = Arc::clone(&self.ws_dispatch_state);
2594 let emitter = self.emitter.clone();
2595 let clock = self.clock;
2596 self.spawn_task("batch_cancel_orders", async move {
2597 const BATCH_SIZE: usize = 20;
2598
2599 for (index, batch) in regular_payload.chunks(BATCH_SIZE).enumerate() {
2600 if let Err(e) = ws_private.batch_cancel_orders(batch.to_vec()).await {
2601 let failure = classify_okx_ws_failure(&e);
2602
2603 if matches!(e, OKXWsError::HandlerUnavailable(_)) {
2604 for (_, client_order_id, venue_order_id) in
2605 ®ular_payload[index * BATCH_SIZE..]
2606 {
2607 let Some(client_order_id) = client_order_id else {
2608 continue;
2609 };
2610
2611 emit_send_failed_cancel(
2612 &failure,
2613 &state,
2614 &emitter,
2615 clock,
2616 *client_order_id,
2617 *venue_order_id,
2618 );
2619 }
2620 } else {
2621 log_batch_cancel_failure(failure, batch.len());
2622 }
2623
2624 return Err(anyhow::Error::new(e).context("batch cancel orders failed"));
2625 }
2626 }
2627 Ok(())
2628 });
2629 }
2630
2631 if !algo_orders.is_empty() {
2633 let items: Vec<_> = algo_orders
2634 .into_iter()
2635 .map(|cancel| {
2636 let request = OKXCancelAlgoOrderRequest {
2637 inst_id: cancel.instrument_id.symbol.to_string(),
2638 inst_id_code: None,
2639 algo_id: cancel.venue_order_id.map(|id| id.to_string()),
2640 algo_cl_ord_id: if cancel.venue_order_id.is_none() {
2641 Some(cancel.client_order_id.to_string())
2642 } else {
2643 None
2644 },
2645 };
2646 let ctx = AlgoCancelContext {
2647 client_order_id: cancel.client_order_id,
2648 instrument_id: cancel.instrument_id,
2649 strategy_id: cancel.strategy_id,
2650 venue_order_id: cancel.venue_order_id,
2651 };
2652 (request, ctx)
2653 })
2654 .collect();
2655 self.dispatch_algo_cancels(items);
2656 }
2657
2658 if !http_orders.is_empty() {
2659 let client = self.http_client.clone();
2660 let emitter = self.emitter.clone();
2661 let clock = self.clock;
2662
2663 self.spawn_task("cancel_http_orders", async move {
2664 for (client_order_id, instrument_id, strategy_id, venue_order_id) in http_orders {
2665 if let Err(e) = client
2666 .cancel_order(instrument_id, Some(client_order_id), venue_order_id)
2667 .await
2668 {
2669 emit_cancel_failure(
2670 classify_okx_http_failure(&e),
2671 Some((&emitter, clock)),
2672 client_order_id,
2673 instrument_id,
2674 strategy_id,
2675 venue_order_id,
2676 );
2677 }
2678 }
2679
2680 Ok(())
2681 });
2682 }
2683
2684 Ok(())
2685 }
2686}
2687
2688const MAX_TRIGGERED_CHILD_RECOVERIES: usize = 100;
2689
2690struct OrderReportSweep {
2691 fallback_parents: AHashMap<usize, OrderStatusReport>,
2692 reports: Vec<OrderStatusReport>,
2693 complete: bool,
2694 ambiguous_triggered_child_ids: AHashSet<VenueOrderId>,
2695 regular_by_venue_order_id: AHashMap<VenueOrderId, OrderStatusReport>,
2696}
2697
2698impl OKXExecutionClient {
2699 fn order_position_side(
2700 &self,
2701 order: &impl Order,
2702 params: Option<&Params>,
2703 ) -> Option<PositionSide> {
2704 if self.position_mode != OKXPositionMode::LongShortMode
2705 || !matches!(
2706 okx_instrument_type_from_symbol(order.instrument_id().symbol.as_str()),
2707 OKXInstrumentType::Swap | OKXInstrumentType::Futures
2708 )
2709 {
2710 return None;
2711 }
2712
2713 let is_closing = order.is_reduce_only()
2714 || (self.is_conditional_order(order.order_type())
2715 && get_param_as_string(params, "close_fraction").is_some());
2716
2717 Some(match (order.order_side(), is_closing) {
2718 (OrderSide::Buy, false) | (OrderSide::Sell, true) => PositionSide::Long,
2719 (OrderSide::Sell, false) | (OrderSide::Buy, true) => PositionSide::Short,
2720 })
2721 }
2722
2723 fn validate_trade_mode(
2724 &self,
2725 instrument_id: InstrumentId,
2726 position_side: Option<PositionSide>,
2727 trade_mode: OKXTradeMode,
2728 ) -> Result<(), OrderDeniedReason> {
2729 if position_side.is_some() {
2730 let expected = derive_trade_mode_for_instrument(
2731 instrument_id,
2732 self.config.margin_mode,
2733 self.config.use_spot_margin,
2734 );
2735
2736 if trade_mode != expected {
2737 return Err(OrderDeniedReason::ValidationFailed {
2738 detail: format!(
2739 "OKX long/short mode requires configured trade mode {expected}, received {trade_mode}; mixed margin modes are not supported"
2740 ),
2741 });
2742 }
2743 }
2744
2745 Ok(())
2746 }
2747
2748 fn add_flat_derivative_position_reports(
2749 &self,
2750 order_reports: &[OrderStatusReport],
2751 fill_reports: &[FillReport],
2752 position_reports: &mut Vec<PositionStatusReport>,
2753 ts_init: UnixNanos,
2754 ) {
2755 let mut instrument_ids: AHashSet<InstrumentId> = order_reports
2756 .iter()
2757 .map(|report| report.instrument_id)
2758 .chain(fill_reports.iter().map(|report| report.instrument_id))
2759 .collect();
2760 let instrument_types = self.instrument_types();
2761 let mut hedging_instrument_ids = AHashSet::new();
2762 let mut hedging_positions = AHashMap::new();
2763
2764 if self.position_mode == OKXPositionMode::LongShortMode {
2765 for (instrument_id, position_id, precision) in order_reports
2766 .iter()
2767 .map(|report| {
2768 (
2769 report.instrument_id,
2770 report.venue_position_id,
2771 report.quantity.precision,
2772 )
2773 })
2774 .chain(fill_reports.iter().map(|report| {
2775 (
2776 report.instrument_id,
2777 report.venue_position_id,
2778 report.last_qty.precision,
2779 )
2780 }))
2781 {
2782 if let Some(position_id) = position_id {
2783 hedging_positions.insert((instrument_id, position_id), precision);
2784 }
2785 }
2786 }
2787
2788 {
2789 let cache = self.core.cache();
2790 for position in cache.positions_open(
2791 Some(&OKX_VENUE),
2792 None,
2793 None,
2794 Some(&self.core.account_id),
2795 None,
2796 ) {
2797 instrument_ids.insert(position.instrument_id);
2798 if cache.oms_type(&position.id) == Some(OmsType::Hedging) {
2799 hedging_instrument_ids.insert(position.instrument_id);
2800 }
2801
2802 if self.position_mode == OKXPositionMode::LongShortMode
2803 && instrument_types.contains(&okx_instrument_type_from_symbol(
2804 position.instrument_id.symbol.as_str(),
2805 ))
2806 && [OKXPositionSide::Long, OKXPositionSide::Short]
2807 .iter()
2808 .any(|side| {
2809 venue_position_id(position.instrument_id, *side) == Some(position.id)
2810 })
2811 {
2812 hedging_positions.insert(
2813 (position.instrument_id, position.id),
2814 position.quantity.precision,
2815 );
2816 }
2817 }
2818 }
2819
2820 for ((instrument_id, position_id), precision) in hedging_positions {
2821 if !position_reports.iter().any(|report| {
2822 report.instrument_id == instrument_id
2823 && report.venue_position_id == Some(position_id)
2824 }) {
2825 position_reports.push(PositionStatusReport::new(
2826 self.core.account_id,
2827 instrument_id,
2828 PositionSide::Flat,
2829 Quantity::zero(precision),
2830 ts_init,
2831 ts_init,
2832 None,
2833 Some(position_id),
2834 None,
2835 ));
2836 }
2837 }
2838
2839 instrument_ids.retain(|instrument_id| {
2840 if self.http_client.is_spread(*instrument_id)
2841 || hedging_instrument_ids.contains(instrument_id)
2842 || position_reports
2843 .iter()
2844 .any(|report| report.instrument_id == *instrument_id)
2845 {
2846 return false;
2847 }
2848
2849 match okx_instrument_type_from_symbol(instrument_id.symbol.as_str()) {
2850 OKXInstrumentType::Swap | OKXInstrumentType::Futures => {
2851 self.position_mode != OKXPositionMode::LongShortMode
2852 }
2853 OKXInstrumentType::Option => true,
2854 _ => false,
2855 }
2856 });
2857
2858 for instrument_id in instrument_ids {
2859 position_reports.push(PositionStatusReport::new(
2861 self.core.account_id,
2862 instrument_id,
2863 PositionSide::Flat,
2864 Quantity::zero(0),
2865 ts_init,
2866 ts_init,
2867 None,
2868 None,
2869 None,
2870 ));
2871 }
2872 }
2873
2874 async fn recover_triggered_child_order_reports(
2875 &self,
2876 reports: &mut Vec<OrderStatusReport>,
2877 ambiguous_triggered_child_ids: &AHashSet<VenueOrderId>,
2878 regular_by_venue_order_id: &AHashMap<VenueOrderId, OrderStatusReport>,
2879 ) -> bool {
2880 let mut fallback_parents = AHashMap::new();
2881 let complete = self
2882 .report_client()
2883 .recover_triggered_child_order_reports(
2884 reports,
2885 ambiguous_triggered_child_ids,
2886 regular_by_venue_order_id,
2887 &mut fallback_parents,
2888 )
2889 .await;
2890 retain_external_regular_order_fallbacks(&self.core, reports, &fallback_parents);
2891 complete
2892 }
2893}
2894
2895struct OKXReportClient {
2896 account_id: AccountId,
2897 http_client: OKXHttpClient,
2898 instrument_types: Vec<OKXInstrumentType>,
2899 load_spreads: bool,
2900 margin_mode: Option<OKXMarginMode>,
2901}
2902
2903impl OKXReportClient {
2904 async fn generate_order_status_report(
2905 &self,
2906 cmd: &GenerateOrderStatusReport,
2907 order_state: Option<CachedQueryOrderState>,
2908 ) -> anyhow::Result<Option<OrderStatusReport>> {
2909 let Some(instrument_id) = cmd.instrument_id else {
2910 anyhow::bail!("generate_order_status_report requires instrument_id");
2911 };
2912
2913 if cmd.client_order_id.is_none() && cmd.venue_order_id.is_none() {
2914 anyhow::bail!(
2915 "generate_order_status_report requires client_order_id or venue_order_id"
2916 );
2917 }
2918
2919 let cached_venue_order_id = order_state.and_then(|state| state.venue_order_id);
2920
2921 let regular_venue_order_id = order_state.and_then(|state| {
2922 if OKX_CONDITIONAL_ORDER_TYPES.contains(&state.order_type) {
2923 state.venue_order_id.or(cmd.venue_order_id)
2924 } else {
2925 state.venue_order_id
2926 }
2927 });
2928
2929 let selection_venue_order_id = cached_venue_order_id.or(cmd.venue_order_id);
2930 let route = query_order_route(
2931 instrument_id,
2932 order_state.map(|state| state.order_type),
2933 regular_venue_order_id.is_some(),
2934 self.http_client.is_spread(instrument_id),
2935 );
2936
2937 let mut reports = Vec::with_capacity(1);
2938 let mut query_algo = matches!(
2939 route,
2940 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo
2941 );
2942 let mut lookup_error = None;
2943
2944 match route {
2945 QueryOrderRoute::Spread => {
2946 let targeted_venue_order_id =
2947 cmd.venue_order_id.filter(|_| cmd.client_order_id.is_none());
2948
2949 match self
2950 .http_client
2951 .request_spread_order_status_report(
2952 self.account_id,
2953 instrument_id,
2954 cmd.client_order_id,
2955 targeted_venue_order_id,
2956 )
2957 .await
2958 {
2959 Ok(Some(report)) => reports.push(report),
2960 Ok(None) => {}
2961 Err(e) => lookup_error = Some(e),
2962 }
2963 }
2964 QueryOrderRoute::Regular | QueryOrderRoute::RegularThenAlgo => {
2965 let targeted_venue_order_id = regular_venue_order_id
2966 .or(cmd.venue_order_id.filter(|_| cmd.client_order_id.is_none()));
2967
2968 let result = if let Some(venue_order_id) = targeted_venue_order_id {
2969 self.http_client
2970 .request_order_status_report_by_venue_order_id(
2971 self.account_id,
2972 instrument_id,
2973 venue_order_id,
2974 )
2975 .await
2976 } else if let Some(client_order_id) = cmd.client_order_id {
2977 self.http_client
2978 .request_order_status_report(
2979 self.account_id,
2980 instrument_id,
2981 client_order_id,
2982 )
2983 .await
2984 } else {
2985 anyhow::bail!(
2986 "generate_order_status_report requires client_order_id or venue_order_id"
2987 );
2988 };
2989
2990 match result {
2991 Ok(Some(report)) => reports.push(report),
2992 Ok(None) => {
2993 query_algo |= route == QueryOrderRoute::RegularThenAlgo;
2994 }
2995 Err(e) => {
2996 lookup_error = Some(e);
2997 query_algo |= route == QueryOrderRoute::RegularThenAlgo;
2998 }
2999 }
3000 }
3001 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo => {}
3002 }
3003
3004 if query_algo {
3005 let (algo_id, algo_client_order_id) = match cmd.client_order_id {
3006 Some(client_order_id) => (None, Some(client_order_id)),
3007 None => (cmd.venue_order_id.map(|id| id.as_str().to_string()), None),
3008 };
3009
3010 match self
3011 .http_client
3012 .request_algo_order_status_reports(
3013 self.account_id,
3014 None,
3015 Some(instrument_id),
3016 algo_id,
3017 algo_client_order_id,
3018 None,
3019 Some(1),
3020 )
3021 .await
3022 {
3023 Ok(algo_reports) => merge_order_status_reports(&mut reports, algo_reports),
3024 Err(e) => {
3025 if lookup_error.is_none() {
3026 lookup_error = Some(e);
3027 }
3028 }
3029 }
3030 }
3031
3032 if reports.is_empty() {
3033 if let Some(e) = lookup_error {
3034 return Err(e);
3035 }
3036
3037 return Ok(None);
3038 }
3039
3040 if let Some(client_order_id) = cmd.client_order_id {
3041 Ok(select_query_order_report(
3042 reports,
3043 client_order_id,
3044 selection_venue_order_id,
3045 ))
3046 } else {
3047 Ok(Some(reports.remove(0)))
3048 }
3049 }
3050
3051 fn instrument_types(instrument_types: &[OKXInstrumentType]) -> Vec<OKXInstrumentType> {
3052 if instrument_types.is_empty() {
3053 vec![OKXInstrumentType::Spot]
3054 } else {
3055 instrument_types.to_vec()
3056 }
3057 }
3058
3059 fn report_scope<'a>(
3060 &'a self,
3061 instrument_types: &'a [OKXInstrumentType],
3062 ) -> ReportInstrumentScope<'a> {
3063 ReportInstrumentScope {
3064 instrument_types,
3065 load_spreads: self.load_spreads,
3066 margin_mode: self.margin_mode,
3067 }
3068 }
3069
3070 async fn collect_order_status_reports(
3071 &self,
3072 cmd: &GenerateOrderStatusReports,
3073 require_complete_active_coverage: bool,
3074 max_history_pages: Option<u32>,
3075 ) -> anyhow::Result<OrderReportSweep> {
3076 let instrument_types = Self::instrument_types(&self.instrument_types);
3077 let routing_types = order_routing_instrument_types(&instrument_types);
3078 let scope = self.report_scope(&routing_types);
3079 let start = nanos_to_datetime(cmd.start);
3080 let end = nanos_to_datetime(cmd.end);
3081 let mut reports = Vec::new();
3082 let mut regular_by_venue_order_id = AHashMap::new();
3083 let mut ambiguous_triggered_child_ids = AHashSet::new();
3084 let mut complete = true;
3085
3086 if let Some(instrument_id) = cmd.instrument_id {
3087 let sweep = self
3088 .http_client
3089 .request_order_status_reports_scoped(
3090 self.account_id,
3091 None,
3092 Some(instrument_id),
3093 start,
3094 end,
3095 cmd.open_only,
3096 None,
3097 Some(scope),
3098 max_history_pages,
3099 )
3100 .await?;
3101
3102 regular_by_venue_order_id.extend(
3103 sweep
3104 .reports
3105 .iter()
3106 .map(|report| (report.venue_order_id, report.clone())),
3107 );
3108
3109 reports.extend(sweep.reports);
3110 complete &= sweep.complete;
3111
3112 if !self.http_client.is_spread(instrument_id)
3113 && supports_algo_orders(okx_instrument_type_from_symbol(
3114 instrument_id.symbol.as_str(),
3115 ))
3116 {
3117 match self
3118 .http_client
3119 .request_algo_order_status_reports_sweep(
3120 self.account_id,
3121 None,
3122 Some(instrument_id),
3123 None,
3124 None,
3125 None,
3126 None,
3127 start,
3128 end,
3129 require_complete_active_coverage,
3130 max_history_pages,
3131 )
3132 .await
3133 {
3134 Ok(sweep) => {
3135 merge_algo_order_status_reports(
3136 &mut reports,
3137 sweep,
3138 &mut ambiguous_triggered_child_ids,
3139 &mut complete,
3140 );
3141 }
3142 Err(e)
3143 if require_complete_active_coverage
3144 && e.downcast_ref::<OKXPendingAlgoOrderReportsError>()
3145 .is_some() =>
3146 {
3147 return Err(e);
3148 }
3149 Err(e) if is_instrument_cache_miss(&e) => return Err(e),
3150 Err(e) => {
3151 log::warn!(
3152 "Failed to fetch algo order status reports for {instrument_id}: {e}"
3153 );
3154 complete = false;
3155 }
3156 }
3157 }
3158 } else {
3159 for inst_type in &routing_types {
3160 let sweep = self
3161 .http_client
3162 .request_order_status_reports_scoped(
3163 self.account_id,
3164 Some(*inst_type),
3165 None,
3166 start,
3167 end,
3168 cmd.open_only,
3169 None,
3170 Some(scope),
3171 max_history_pages,
3172 )
3173 .await?;
3174
3175 regular_by_venue_order_id.extend(
3176 sweep
3177 .reports
3178 .iter()
3179 .map(|report| (report.venue_order_id, report.clone())),
3180 );
3181
3182 reports.extend(sweep.reports);
3183 complete &= sweep.complete;
3184
3185 if supports_algo_orders(*inst_type) {
3186 match self
3187 .http_client
3188 .request_algo_order_status_reports_sweep(
3189 self.account_id,
3190 Some(*inst_type),
3191 None,
3192 None,
3193 None,
3194 None,
3195 None,
3196 start,
3197 end,
3198 require_complete_active_coverage,
3199 max_history_pages,
3200 )
3201 .await
3202 {
3203 Ok(sweep) => {
3204 merge_algo_order_status_reports(
3205 &mut reports,
3206 sweep,
3207 &mut ambiguous_triggered_child_ids,
3208 &mut complete,
3209 );
3210 }
3211 Err(e)
3212 if require_complete_active_coverage
3213 && e.downcast_ref::<OKXPendingAlgoOrderReportsError>()
3214 .is_some() =>
3215 {
3216 return Err(e);
3217 }
3218 Err(e) if is_instrument_cache_miss(&e) => return Err(e),
3219 Err(e) => {
3220 log::warn!(
3221 "Failed to fetch algo order status reports for {inst_type:?}: {e}"
3222 );
3223 complete = false;
3224 }
3225 }
3226 }
3227 }
3228 }
3229
3230 if cmd.instrument_id.is_none() && self.load_spreads {
3231 match self
3232 .http_client
3233 .request_order_status_reports_scoped(
3234 self.account_id,
3235 None,
3236 None,
3237 start,
3238 end,
3239 cmd.open_only,
3240 None,
3241 Some(scope),
3242 max_history_pages,
3243 )
3244 .await
3245 {
3246 Ok(sweep) => {
3247 reports.extend(sweep.reports);
3248 complete &= sweep.complete;
3249 }
3250 Err(e) if is_instrument_cache_miss(&e) => return Err(e),
3251 Err(e) => {
3252 log::warn!("Failed to fetch spread order status reports: {e}");
3253 complete = false;
3254 }
3255 }
3256 }
3257
3258 let mut fallback_parents = AHashMap::new();
3259
3260 if cmd.open_only {
3261 complete &= self
3262 .recover_triggered_child_order_reports(
3263 &mut reports,
3264 &ambiguous_triggered_child_ids,
3265 ®ular_by_venue_order_id,
3266 &mut fallback_parents,
3267 )
3268 .await;
3269 }
3270
3271 Ok(OrderReportSweep {
3272 fallback_parents,
3273 reports,
3274 complete,
3275 ambiguous_triggered_child_ids,
3276 regular_by_venue_order_id,
3277 })
3278 }
3279
3280 async fn collect_fill_reports(
3281 &self,
3282 cmd: GenerateFillReports,
3283 history: FillHistory,
3284 max_history_pages: Option<u32>,
3285 ) -> anyhow::Result<(Vec<FillReport>, bool)> {
3286 let instrument_types = Self::instrument_types(&self.instrument_types);
3287 let routing_types = order_routing_instrument_types(&instrument_types);
3288 let scope = self.report_scope(&routing_types);
3289 let start_dt = nanos_to_datetime(cmd.start);
3290 let end_dt = nanos_to_datetime(cmd.end);
3291 let mut reports = Vec::new();
3292 let mut complete = true;
3293
3294 if let Some(instrument_id) = cmd.instrument_id {
3295 let sweep = self
3296 .http_client
3297 .request_fill_reports_scoped(
3298 self.account_id,
3299 None,
3300 Some(instrument_id),
3301 start_dt,
3302 end_dt,
3303 None,
3304 history,
3305 Some(scope),
3306 max_history_pages,
3307 )
3308 .await?;
3309 reports.extend(sweep.reports);
3310 complete &= sweep.complete;
3311 } else {
3312 for inst_type in &routing_types {
3313 let sweep = self
3314 .http_client
3315 .request_fill_reports_scoped(
3316 self.account_id,
3317 Some(*inst_type),
3318 None,
3319 start_dt,
3320 end_dt,
3321 None,
3322 history,
3323 Some(scope),
3324 max_history_pages,
3325 )
3326 .await?;
3327 reports.extend(sweep.reports);
3328 complete &= sweep.complete;
3329 }
3330
3331 if self.load_spreads {
3332 let sweep = self
3333 .http_client
3334 .request_fill_reports_scoped(
3335 self.account_id,
3336 None,
3337 None,
3338 start_dt,
3339 end_dt,
3340 None,
3341 history,
3342 Some(scope),
3343 max_history_pages,
3344 )
3345 .await?;
3346 reports.extend(sweep.reports);
3347 complete &= sweep.complete;
3348 }
3349 }
3350
3351 if let Some(venue_order_id) = cmd.venue_order_id {
3352 reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
3353 }
3354
3355 Ok((reports, complete))
3356 }
3357
3358 async fn collect_position_status_reports(
3359 &self,
3360 cmd: &GeneratePositionStatusReports,
3361 ) -> anyhow::Result<(Vec<PositionStatusReport>, bool)> {
3362 let instrument_types = Self::instrument_types(&self.instrument_types);
3363 let scope = self.report_scope(&instrument_types);
3364 let mut reports = Vec::new();
3365 let mut complete = true;
3366
3367 if let Some(instrument_id) = cmd.instrument_id {
3368 if self.http_client.is_spread(instrument_id) {
3369 return Ok((reports, complete));
3370 }
3371
3372 let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
3373 if inst_type != OKXInstrumentType::Spot && inst_type != OKXInstrumentType::Margin {
3374 let sweep = self
3375 .http_client
3376 .request_position_status_reports_scoped(
3377 self.account_id,
3378 None,
3379 Some(instrument_id),
3380 Some(scope),
3381 )
3382 .await?;
3383 reports.extend(sweep.reports);
3384 complete &= sweep.complete;
3385 }
3386 } else {
3387 for inst_type in &instrument_types {
3388 if *inst_type == OKXInstrumentType::Spot || *inst_type == OKXInstrumentType::Margin
3389 {
3390 continue;
3391 }
3392
3393 let sweep = self
3394 .http_client
3395 .request_position_status_reports_scoped(
3396 self.account_id,
3397 Some(*inst_type),
3398 None,
3399 Some(scope),
3400 )
3401 .await?;
3402 reports.extend(sweep.reports);
3403 complete &= sweep.complete;
3404 }
3405 }
3406
3407 let mut margin_reports = self
3408 .http_client
3409 .request_spot_margin_position_reports(self.account_id)
3410 .await?;
3411
3412 if let Some(instrument_id) = cmd.instrument_id {
3413 margin_reports.retain(|report| report.instrument_id == instrument_id);
3414 }
3415
3416 reports.append(&mut margin_reports);
3417
3418 Ok((reports, complete))
3419 }
3420
3421 async fn recover_triggered_child_order_reports(
3422 &self,
3423 reports: &mut Vec<OrderStatusReport>,
3424 ambiguous_triggered_child_ids: &AHashSet<VenueOrderId>,
3425 regular_by_venue_order_id: &AHashMap<VenueOrderId, OrderStatusReport>,
3426 fallback_parents: &mut AHashMap<usize, OrderStatusReport>,
3427 ) -> bool {
3428 let mut recovered = Vec::with_capacity(reports.len());
3429 let recovery_candidate_count = reports
3430 .iter()
3431 .filter(|report| {
3432 report.order_status == OrderStatus::Triggered
3433 && !ambiguous_triggered_child_ids.contains(&report.venue_order_id)
3434 })
3435 .count();
3436 let mut complete = true;
3437
3438 if recovery_candidate_count > MAX_TRIGGERED_CHILD_RECOVERIES {
3439 log::warn!(
3440 "Triggered child recovery hit {MAX_TRIGGERED_CHILD_RECOVERIES} request cap; omitting unresolved reports"
3441 );
3442 }
3443
3444 let mut request_count = 0;
3445
3446 for mut parent_report in reports.drain(..) {
3447 if parent_report.order_status != OrderStatus::Triggered {
3448 recovered.push(parent_report);
3449 continue;
3450 }
3451
3452 if ambiguous_triggered_child_ids.contains(&parent_report.venue_order_id) {
3453 log::warn!(
3454 "Omitting triggered algo order with ambiguous child identifiers: {} {}",
3455 parent_report.instrument_id,
3456 parent_report.venue_order_id,
3457 );
3458 self.push_external_regular_order_fallback(
3459 &parent_report,
3460 regular_by_venue_order_id,
3461 &mut recovered,
3462 fallback_parents,
3463 );
3464 complete = false;
3465 continue;
3466 }
3467
3468 if request_count >= MAX_TRIGGERED_CHILD_RECOVERIES {
3469 self.push_external_regular_order_fallback(
3470 &parent_report,
3471 regular_by_venue_order_id,
3472 &mut recovered,
3473 fallback_parents,
3474 );
3475 complete = false;
3476 continue;
3477 }
3478 request_count += 1;
3479
3480 match self
3481 .http_client
3482 .request_order_status_report_by_venue_order_id(
3483 self.account_id,
3484 parent_report.instrument_id,
3485 parent_report.venue_order_id,
3486 )
3487 .await
3488 {
3489 Ok(Some(mut child_report)) => {
3490 if child_report.order_status == OrderStatus::Accepted {
3491 parent_report.quantity = child_report.quantity;
3492 parent_report.price = child_report.price.or(parent_report.price);
3493 parent_report.reduce_only |= child_report.reduce_only;
3494 parent_report.ts_last = child_report.ts_last;
3495 recovered.push(parent_report);
3496 } else {
3497 child_report.client_order_id = parent_report.client_order_id;
3498 recovered.push(child_report);
3499 }
3500 }
3501 Ok(None) => {
3502 log::warn!(
3503 "Triggered child order {} {} was not found",
3504 parent_report.instrument_id,
3505 parent_report.venue_order_id,
3506 );
3507 self.push_external_regular_order_fallback(
3508 &parent_report,
3509 regular_by_venue_order_id,
3510 &mut recovered,
3511 fallback_parents,
3512 );
3513 complete = false;
3514 }
3515 Err(e) => {
3516 log::warn!(
3517 "Failed to recover triggered child order {} {}: {e}",
3518 parent_report.instrument_id,
3519 parent_report.venue_order_id,
3520 );
3521 self.push_external_regular_order_fallback(
3522 &parent_report,
3523 regular_by_venue_order_id,
3524 &mut recovered,
3525 fallback_parents,
3526 );
3527 complete = false;
3528 }
3529 }
3530 }
3531
3532 *reports = recovered;
3533 complete
3534 }
3535
3536 fn push_external_regular_order_fallback(
3537 &self,
3538 parent_report: &OrderStatusReport,
3539 regular_by_venue_order_id: &AHashMap<VenueOrderId, OrderStatusReport>,
3540 recovered: &mut Vec<OrderStatusReport>,
3541 fallback_parents: &mut AHashMap<usize, OrderStatusReport>,
3542 ) {
3543 if let Some(regular_report) = regular_by_venue_order_id.get(&parent_report.venue_order_id) {
3544 fallback_parents.insert(recovered.len(), parent_report.clone());
3545 recovered.push(regular_report.clone());
3546 }
3547 }
3548}
3549
3550fn finish_order_report_sweep(
3551 core: &ExecutionClientCore,
3552 mut sweep: OrderReportSweep,
3553 command: &GenerateOrderStatusReports,
3554) -> OrderReportSweep {
3555 retain_external_regular_order_fallbacks(core, &mut sweep.reports, &sweep.fallback_parents);
3556 sweep.fallback_parents.clear();
3557 retain_order_status_reports(&mut sweep.reports, command);
3558 sweep
3559}
3560
3561fn retain_external_regular_order_fallbacks(
3562 core: &ExecutionClientCore,
3563 reports: &mut Vec<OrderStatusReport>,
3564 fallback_parents: &AHashMap<usize, OrderStatusReport>,
3565) {
3566 if fallback_parents.is_empty() {
3567 return;
3568 }
3569
3570 let cache = core.cache();
3571 let mut index = 0;
3572 reports.retain(|_| {
3573 let parent = fallback_parents.get(&index);
3574 index += 1;
3575 parent.is_none_or(|parent| {
3576 let cached_by_client = parent
3577 .client_order_id
3578 .is_some_and(|id| cache.order_exists(&id));
3579 let cached_by_venue = cache
3580 .client_order_id(&parent.venue_order_id)
3581 .is_some_and(|id| cache.order_exists(id));
3582 !cached_by_client && !cached_by_venue
3583 })
3584 });
3585}
3586
3587fn validate_submit_position_id(
3588 instrument_id: InstrumentId,
3589 position_side: Option<PositionSide>,
3590 position_id: Option<PositionId>,
3591) -> Result<(), OrderDeniedReason> {
3592 if position_id.is_none() {
3593 return Ok(());
3594 }
3595
3596 let expected = position_side.and_then(|side| venue_position_id(instrument_id, side.into()));
3597 if let (Some(position_id), Some(expected)) = (position_id, expected)
3598 && position_id != expected
3599 {
3600 return Err(OrderDeniedReason::InvalidPositionId {
3601 position_id,
3602 detail: format!("expected OKX venue position ID {expected}"),
3603 });
3604 }
3605
3606 Ok(())
3607}
3608
3609fn validate_order(
3610 order: &impl Order,
3611 trade_mode: OKXTradeMode,
3612 submission: OrderSubmission,
3613 is_spread: bool,
3614) -> Result<(), OrderDeniedReason> {
3615 if let Err(detail) = validate_okx_client_order_id(order.client_order_id().as_str()) {
3616 return Err(OrderDeniedReason::InvalidClientOrderId { detail });
3617 }
3618
3619 if is_spread && order.is_reduce_only() {
3620 return Err(OrderDeniedReason::UnsupportedReduceOnly);
3621 }
3622
3623 okx_algo_target_currency(
3624 okx_instrument_type_from_symbol(order.instrument_id().symbol.as_str()),
3625 trade_mode,
3626 order.order_side(),
3627 order.order_type(),
3628 order.is_quote_quantity(),
3629 )?;
3630
3631 validate_algo_trigger_type(
3632 okx_instrument_type_from_symbol(order.instrument_id().symbol.as_str()),
3633 trade_mode,
3634 order.order_type(),
3635 order.trigger_type(),
3636 )
3637 .map_err(|detail| OrderDeniedReason::ValidationFailed { detail })?;
3638
3639 if order.is_reduce_only()
3640 && okx_reduce_only_wire_value(
3641 okx_instrument_type_from_symbol(order.instrument_id().symbol.as_str()),
3642 trade_mode,
3643 order.order_side(),
3644 None,
3645 Some(true),
3646 )
3647 .is_err()
3648 {
3649 return Err(OrderDeniedReason::UnsupportedReduceOnly);
3650 }
3651
3652 if matches!(submission, OrderSubmission::List) {
3653 if OKX_CONDITIONAL_ORDER_TYPES.contains(&order.order_type()) {
3654 return Err(OrderDeniedReason::UnsupportedOrderList {
3655 detail: format!(
3656 "conditional order {} is not supported",
3657 order.client_order_id()
3658 ),
3659 });
3660 }
3661
3662 if order.time_in_force() != TimeInForce::Gtc {
3663 return Err(OrderDeniedReason::UnsupportedOrderList {
3664 detail: format!(
3665 "order {} has unsupported time in force {}",
3666 order.client_order_id(),
3667 order.time_in_force()
3668 ),
3669 });
3670 }
3671 }
3672
3673 Ok(())
3674}
3675
3676#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3677enum OrderCommandRoute {
3678 RegularWs,
3679 AlgoHttp,
3680 SpreadHttp,
3681}
3682
3683#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3684enum QueryOrderRoute {
3685 Regular,
3686 Algo,
3687 RegularThenAlgo,
3688 RegularAndAlgo,
3689 Spread,
3690}
3691
3692#[derive(Debug, Clone, Copy)]
3693struct CachedQueryOrderState {
3694 order_type: OrderType,
3695 venue_order_id: Option<VenueOrderId>,
3696}
3697
3698#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3699enum CancelAllOrdersRoute {
3700 Batch,
3701 MassCancelHttp,
3702 SpreadHttp,
3703}
3704
3705#[derive(Clone, Copy)]
3706enum OrderSubmission {
3707 Single,
3708 List,
3709}
3710
3711fn emit_submit_failure(
3712 failure: CommandFailure,
3713 emitter: &ExecutionEventEmitter,
3714 clock: &'static AtomicTime,
3715 strategy_id: StrategyId,
3716 instrument_id: InstrumentId,
3717 client_order_id: ClientOrderId,
3718) {
3719 match failure {
3720 CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
3721 emitter.emit_order_rejected_event(
3722 strategy_id,
3723 instrument_id,
3724 client_order_id,
3725 &reason,
3726 clock.get_time_ns(),
3727 false,
3728 );
3729 }
3730 CommandFailure::Ambiguous(reason) => {
3731 log::warn!(
3732 "Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {reason}"
3733 );
3734 }
3735 }
3736}
3737
3738fn emit_modify_failure(
3739 failure: CommandFailure,
3740 emitter: &ExecutionEventEmitter,
3741 clock: &'static AtomicTime,
3742 strategy_id: StrategyId,
3743 instrument_id: InstrumentId,
3744 client_order_id: ClientOrderId,
3745 venue_order_id: Option<VenueOrderId>,
3746) {
3747 match failure {
3748 CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
3749 emitter.emit_order_modify_rejected_event(
3750 strategy_id,
3751 instrument_id,
3752 client_order_id,
3753 venue_order_id,
3754 &reason,
3755 clock.get_time_ns(),
3756 );
3757 }
3758 CommandFailure::Ambiguous(reason) => {
3759 log::warn!(
3760 "Ambiguous modify failure for {client_order_id}, awaiting reconciliation: {reason}"
3761 );
3762 }
3763 }
3764}
3765
3766fn algo_amend_request(
3767 cmd: &ModifyOrder,
3768 algo_type: Option<OKXAlgoOrderType>,
3769 context: Option<OrderContext>,
3770) -> Result<OKXAmendAlgoOrderRequest, &'static str> {
3771 let algo_type = algo_type.ok_or("Cannot determine OKX algo order terms for modification")?;
3772 let context = context.ok_or("Cannot determine OKX algo order terms for modification")?;
3773 let algo_id = cmd
3774 .venue_order_id
3775 .ok_or("No algo ID available for pending conditional order")?;
3776 let trigger_price = cmd.trigger_price.map(|price| price.to_string());
3777 let order_price = cmd.price.map(|price| price.to_string());
3778
3779 let trigger_px_type = cmd.trigger_price.map(|_| {
3780 context
3781 .trigger_type
3782 .map_or(OKXTriggerType::Last, Into::into)
3783 .to_string()
3784 });
3785
3786 let mut request = OKXAmendAlgoOrderRequest {
3787 inst_id: cmd.instrument_id.symbol.to_string(),
3788 algo_id: algo_id.to_string(),
3789 algo_cl_ord_id: None,
3790 new_sz: cmd.quantity.map(|quantity| quantity.to_string()),
3791 new_trigger_px: None,
3792 new_trigger_px_type: None,
3793 new_tp_trigger_px: None,
3794 new_tp_ord_px: None,
3795 new_tp_trigger_px_type: None,
3796 new_sl_trigger_px: None,
3797 new_sl_ord_px: None,
3798 new_sl_trigger_px_type: None,
3799 new_order_px: None,
3800 new_callback_ratio: None,
3801 new_callback_spread: None,
3802 new_active_px: None,
3803 };
3804
3805 match (algo_type, context.identity.order_type) {
3806 (OKXAlgoOrderType::Trigger, _) => {
3807 request.new_trigger_px = trigger_price;
3808 request.new_order_px = order_price;
3809 request.new_trigger_px_type = trigger_px_type;
3810 }
3811 (OKXAlgoOrderType::Conditional, OrderType::StopMarket | OrderType::StopLimit) => {
3812 request.new_sl_trigger_px = trigger_price;
3813 request.new_sl_ord_px = order_price;
3814 request.new_sl_trigger_px_type = trigger_px_type;
3815 }
3816 (OKXAlgoOrderType::Conditional, OrderType::MarketIfTouched | OrderType::LimitIfTouched) => {
3817 request.new_tp_trigger_px = trigger_price;
3818 request.new_tp_ord_px = order_price;
3819 request.new_tp_trigger_px_type = trigger_px_type;
3820 }
3821 _ => return Err("OKX does not support modification of this algo order type"),
3822 }
3823
3824 Ok(request)
3825}
3826
3827fn emit_cancel_failure(
3828 failure: CommandFailure,
3829 emit_venue: Option<(&ExecutionEventEmitter, &'static AtomicTime)>,
3830 client_order_id: ClientOrderId,
3831 instrument_id: InstrumentId,
3832 strategy_id: StrategyId,
3833 venue_order_id: Option<VenueOrderId>,
3834) {
3835 match failure {
3836 CommandFailure::VenueRejected(reason) => {
3837 if let Some((emitter, clock)) = emit_venue {
3838 emitter.emit_order_cancel_rejected_event(
3839 strategy_id,
3840 instrument_id,
3841 client_order_id,
3842 venue_order_id,
3843 &reason,
3844 clock.get_time_ns(),
3845 );
3846 } else {
3847 log::warn!(
3848 "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
3849 );
3850 }
3851 }
3852 CommandFailure::NotSent(reason) => {
3853 log::warn!("Cancel command failed local validation for {client_order_id}: {reason}");
3854 }
3855 CommandFailure::Ambiguous(reason) => {
3856 log::warn!(
3857 "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
3858 );
3859 }
3860 }
3861}
3862
3863fn log_batch_cancel_failure(failure: CommandFailure, order_count: usize) {
3864 match failure {
3865 CommandFailure::NotSent(reason) => {
3866 log::warn!(
3867 "Batch cancel command failed local validation for {order_count} orders: {reason}"
3868 );
3869 }
3870 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
3871 log::warn!(
3872 "Ambiguous batch cancel failure for {order_count} orders, awaiting reconciliation: {reason}"
3873 );
3874 }
3875 }
3876}
3877
3878fn log_mass_cancel_failure(failure: CommandFailure, instrument_id: InstrumentId) {
3879 match failure {
3880 CommandFailure::NotSent(reason) => {
3881 log::warn!("Mass cancel command failed local validation for {instrument_id}: {reason}");
3882 }
3883 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
3884 log::warn!(
3885 "Ambiguous mass cancel failure for {instrument_id}, awaiting reconciliation: {reason}"
3886 );
3887 }
3888 }
3889}
3890
3891fn log_algo_batch_cancel_failure(failure: CommandFailure, contexts: &[AlgoCancelContext]) {
3892 match failure {
3893 CommandFailure::NotSent(reason) => {
3894 for ctx in contexts {
3895 log::warn!(
3896 "Algo batch cancel command failed local validation for {}: {reason}",
3897 ctx.client_order_id
3898 );
3899 }
3900 }
3901 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
3902 for ctx in contexts {
3903 log::warn!(
3904 "Ambiguous algo batch cancel failure for {}, awaiting reconciliation: {reason}",
3905 ctx.client_order_id
3906 );
3907 }
3908 }
3909 }
3910}
3911
3912fn get_param_as_string(params: Option<&Params>, key: &str) -> Option<String> {
3913 params.and_then(|p| {
3914 p.get(key).and_then(|v| {
3915 v.as_str()
3916 .map(ToString::to_string)
3917 .or_else(|| v.as_f64().map(|n| n.to_string()))
3918 })
3919 })
3920}
3921
3922fn get_param_as_bool(params: Option<&Params>, key: &str) -> Option<bool> {
3923 params.and_then(|params| params.get_bool(key))
3924}
3925
3926fn supports_algo_orders(instrument_type: OKXInstrumentType) -> bool {
3927 !matches!(
3928 instrument_type,
3929 OKXInstrumentType::Option | OKXInstrumentType::Events
3930 )
3931}
3932
3933fn order_routing_instrument_types(
3934 instrument_types: &[OKXInstrumentType],
3935) -> Vec<OKXInstrumentType> {
3936 let mut routing_types = instrument_types.to_vec();
3937
3938 if routing_types.contains(&OKXInstrumentType::Margin)
3940 && !routing_types.contains(&OKXInstrumentType::Spot)
3941 && !routing_types.contains(&OKXInstrumentType::Any)
3942 {
3943 routing_types.push(OKXInstrumentType::Spot);
3944 }
3945
3946 routing_types
3947}
3948
3949fn query_order_route(
3950 instrument_id: InstrumentId,
3951 order_type: Option<OrderType>,
3952 has_cached_venue_order_id: bool,
3953 is_spread: bool,
3954) -> QueryOrderRoute {
3955 if is_spread {
3956 return QueryOrderRoute::Spread;
3957 }
3958
3959 let supports_algo = supports_algo_orders(okx_instrument_type_from_symbol(
3960 instrument_id.symbol.as_str(),
3961 ));
3962
3963 match order_type {
3964 Some(order_type)
3965 if supports_algo
3966 && OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
3967 && has_cached_venue_order_id =>
3968 {
3969 QueryOrderRoute::RegularAndAlgo
3970 }
3971 Some(order_type) if supports_algo && OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type) => {
3972 QueryOrderRoute::Algo
3973 }
3974 None if supports_algo => QueryOrderRoute::RegularThenAlgo,
3975 _ => QueryOrderRoute::Regular,
3976 }
3977}
3978
3979fn is_instrument_cache_miss(error: &anyhow::Error) -> bool {
3980 error
3981 .chain()
3982 .any(|cause| cause.to_string().contains("missing from cache"))
3983}
3984
3985fn merge_algo_order_status_reports(
3986 reports: &mut Vec<OrderStatusReport>,
3987 sweep: AlgoOrderReportSweep,
3988 ambiguous_triggered_child_ids: &mut AHashSet<VenueOrderId>,
3989 complete: &mut bool,
3990) {
3991 let AlgoOrderReportSweep {
3992 reports: incoming,
3993 complete: sweep_complete,
3994 ambiguous_triggered_child_ids: ambiguous,
3995 } = sweep;
3996
3997 *complete &= sweep_complete && ambiguous.is_empty();
3998 ambiguous_triggered_child_ids.extend(ambiguous);
3999 merge_order_status_reports(reports, incoming);
4000}
4001
4002fn merge_order_status_reports(
4003 reports: &mut Vec<OrderStatusReport>,
4004 incoming: Vec<OrderStatusReport>,
4005) {
4006 let mut indexes: AHashMap<VenueOrderId, usize> = reports
4007 .iter()
4008 .enumerate()
4009 .map(|(index, report)| (report.venue_order_id, index))
4010 .collect();
4011
4012 for report in incoming {
4013 if let Some(index) = indexes.get(&report.venue_order_id).copied() {
4014 if is_order_status_report_more_advanced(&report, &reports[index]) {
4015 reports[index] = report;
4016 }
4017 } else {
4018 indexes.insert(report.venue_order_id, reports.len());
4019 reports.push(report);
4020 }
4021 }
4022}
4023
4024fn select_query_order_report(
4035 reports: Vec<OrderStatusReport>,
4036 client_order_id: ClientOrderId,
4037 venue_order_id: Option<VenueOrderId>,
4038) -> Option<OrderStatusReport> {
4039 let mut by_client_id: Option<OrderStatusReport> = None;
4040 let mut by_vid: Option<OrderStatusReport> = None;
4041
4042 for report in reports {
4043 if report.client_order_id == Some(client_order_id) {
4044 if by_client_id
4045 .as_ref()
4046 .is_none_or(|current| is_order_status_report_more_advanced(&report, current))
4047 {
4048 by_client_id = Some(report);
4049 }
4050
4051 continue;
4052 }
4053
4054 if venue_order_id
4055 .as_ref()
4056 .is_some_and(|vid| report.venue_order_id.as_str() == vid.as_str())
4057 && by_vid
4058 .as_ref()
4059 .is_none_or(|current| is_order_status_report_more_advanced(&report, current))
4060 {
4061 by_vid = Some(report);
4062 }
4063 }
4064
4065 by_client_id.or(by_vid)
4066}
4067
4068#[cfg(test)]
4069mod tests {
4070 use std::{cell::RefCell, rc::Rc, sync::Mutex, time::Duration};
4071
4072 use axum::{
4073 Json, Router,
4074 extract::ws::{Message, WebSocketUpgrade},
4075 routing::{get, post},
4076 };
4077 use nautilus_common::{
4078 cache::Cache,
4079 messages::{ExecutionEvent, ExecutionReport as CommonExecutionReport},
4080 testing::wait_until_async,
4081 };
4082 use nautilus_core::{AtomicMap, UUID4};
4083 use nautilus_model::{
4084 enums::{LiquiditySide, OrderSide, OrderStatus, TriggerType},
4085 events::{OrderEventAny, OrderInitialized},
4086 identifiers::{OrderListId, Symbol, TradeId, TraderId},
4087 instruments::{
4088 CryptoFuture, Instrument,
4089 stubs::{crypto_future_btcusdt, currency_pair_ethusdt},
4090 },
4091 orders::{OrderList, OrderTestBuilder, stubs::TestOrderEventStubs},
4092 position::Position,
4093 types::{Money, Price, Quantity},
4094 };
4095 use rstest::rstest;
4096 use serde_json::Value;
4097 use ustr::Ustr;
4098
4099 use super::*;
4100 use crate::common::{
4101 consts::OKX_CLIENT_ID, enums::OKXEnvironment, models::OKXInstrument,
4102 parse::parse_instrument_any,
4103 };
4104
4105 struct DropSignal(Option<tokio::sync::oneshot::Sender<()>>);
4106
4107 impl Drop for DropSignal {
4108 fn drop(&mut self) {
4109 if let Some(sender) = self.0.take() {
4110 let _ = sender.send(());
4111 }
4112 }
4113 }
4114
4115 #[derive(Clone, Copy)]
4116 enum ExecutionTaskBoundary {
4117 Reset,
4118 Dispose,
4119 RepeatedStop,
4120 }
4121
4122 #[rstest]
4123 #[case(OrderType::Market, QueryOrderRoute::Regular)]
4124 #[case(OrderType::Limit, QueryOrderRoute::Regular)]
4125 #[case(OrderType::StopMarket, QueryOrderRoute::Algo)]
4126 #[case(OrderType::TrailingStopMarket, QueryOrderRoute::Algo)]
4127 fn test_query_order_route_for_known_order_type(
4128 #[case] order_type: OrderType,
4129 #[case] expected: QueryOrderRoute,
4130 ) {
4131 assert_eq!(
4132 query_order_route(
4133 InstrumentId::from("BTC-USDT.OKX"),
4134 Some(order_type),
4135 false,
4136 false
4137 ),
4138 expected
4139 );
4140 }
4141
4142 #[rstest]
4143 fn test_query_order_route_for_conditional_order_with_cached_venue_id() {
4144 assert_eq!(
4145 query_order_route(
4146 InstrumentId::from("BTC-USDT.OKX"),
4147 Some(OrderType::StopMarket),
4148 true,
4149 false,
4150 ),
4151 QueryOrderRoute::RegularAndAlgo,
4152 );
4153 }
4154
4155 #[rstest]
4156 fn test_query_order_route_for_unknown_order_type() {
4157 assert_eq!(
4158 query_order_route(InstrumentId::from("BTC-USDT.OKX"), None, false, false),
4159 QueryOrderRoute::RegularThenAlgo,
4160 );
4161 }
4162
4163 #[rstest]
4164 fn test_query_order_route_for_spread() {
4165 assert_eq!(
4166 query_order_route(
4167 InstrumentId::from("ETH-USD-SWAP_ETH-USD-231229.OKX"),
4168 None,
4169 false,
4170 true,
4171 ),
4172 QueryOrderRoute::Spread,
4173 );
4174 }
4175
4176 #[rstest]
4177 #[case("BTC-USD_UM_XPERP-310404.OKX")]
4178 #[case("BTC-USD_UM-261030.OKX")]
4179 fn test_usd_margined_futures_use_regular_routes(
4180 mut crypto_future_btcusdt: CryptoFuture,
4181 #[case] instrument_id: &str,
4182 #[values(false, true)] cached: bool,
4183 #[values(OrderSide::Buy, OrderSide::Sell)] side: OrderSide,
4184 ) {
4185 let client = build_test_exec_client();
4186 let instrument_id = InstrumentId::from(instrument_id);
4187 crypto_future_btcusdt.id = instrument_id;
4188 crypto_future_btcusdt.raw_symbol = instrument_id.symbol;
4189
4190 if cached {
4191 client
4192 .http_client
4193 .cache_instrument(InstrumentAny::CryptoFuture(crypto_future_btcusdt));
4194 }
4195
4196 let order = OrderTestBuilder::new(OrderType::Limit)
4197 .instrument_id(instrument_id)
4198 .client_order_id(ClientOrderId::from("OREDUCEFUTURE"))
4199 .side(side)
4200 .price(Price::from("65000.1"))
4201 .quantity(Quantity::from("3"))
4202 .reduce_only(true)
4203 .build();
4204 let is_spread = client.http_client.is_spread(instrument_id);
4205
4206 assert_eq!(
4207 client
4208 .submit_order_route(instrument_id, OrderType::Limit)
4209 .unwrap(),
4210 OrderCommandRoute::RegularWs
4211 );
4212 assert_eq!(
4213 client
4214 .submit_order_route(instrument_id, OrderType::StopMarket)
4215 .unwrap(),
4216 OrderCommandRoute::AlgoHttp
4217 );
4218 assert_eq!(
4219 client.order_command_route(instrument_id, Some((OrderType::Limit, None)), false),
4220 OrderCommandRoute::RegularWs
4221 );
4222 assert_eq!(
4223 client.cancel_all_orders_route(instrument_id, None),
4224 CancelAllOrdersRoute::Batch
4225 );
4226 assert_eq!(
4227 query_order_route(instrument_id, Some(OrderType::Limit), false, is_spread),
4228 QueryOrderRoute::Regular
4229 );
4230 assert_eq!(
4231 validate_order(
4232 &order,
4233 OKXTradeMode::Cross,
4234 OrderSubmission::Single,
4235 is_spread
4236 ),
4237 Ok(())
4238 );
4239 }
4240
4241 #[rstest]
4242 fn test_validate_order_allows_conditional_single_submission() {
4243 let order = OrderTestBuilder::new(OrderType::StopMarket)
4244 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
4245 .client_order_id(ClientOrderId::from("OCONDITIONALSINGLE"))
4246 .side(OrderSide::Sell)
4247 .quantity(Quantity::from("1"))
4248 .trigger_price(Price::from("1000.00"))
4249 .build();
4250
4251 assert_eq!(
4252 validate_order(&order, OKXTradeMode::Cross, OrderSubmission::Single, false),
4253 Ok(())
4254 );
4255 }
4256
4257 #[rstest]
4258 fn test_validate_order_denies_conditional_order_in_list() {
4259 let order = OrderTestBuilder::new(OrderType::StopMarket)
4260 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
4261 .client_order_id(ClientOrderId::from("OCONDITIONALLIST"))
4262 .side(OrderSide::Sell)
4263 .quantity(Quantity::from("2"))
4264 .trigger_price(Price::from("900.00"))
4265 .build();
4266
4267 assert_eq!(
4268 validate_order(&order, OKXTradeMode::Cross, OrderSubmission::List, false),
4269 Err(OrderDeniedReason::UnsupportedOrderList {
4270 detail: "conditional order OCONDITIONALLIST is not supported".to_string(),
4271 })
4272 );
4273 }
4274
4275 #[rstest]
4276 fn test_validate_order_denies_non_gtc_order_in_list() {
4277 let order = OrderTestBuilder::new(OrderType::Limit)
4278 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
4279 .client_order_id(ClientOrderId::from("OIOCLIST"))
4280 .side(OrderSide::Buy)
4281 .price(Price::from("2000.00"))
4282 .quantity(Quantity::from("3"))
4283 .time_in_force(TimeInForce::Ioc)
4284 .build();
4285
4286 assert_eq!(
4287 validate_order(&order, OKXTradeMode::Cross, OrderSubmission::List, false),
4288 Err(OrderDeniedReason::UnsupportedOrderList {
4289 detail: "order OIOCLIST has unsupported time in force IOC".to_string(),
4290 })
4291 );
4292 }
4293
4294 #[rstest]
4295 #[case::cash("BTC-USDT.OKX", OKXTradeMode::Cash)]
4296 #[case::option("BTC-USD-241217-92000-C.OKX", OKXTradeMode::Cross)]
4297 #[case::event("BTC-ABOVE-DAILY-260224-1600-65000.OKX", OKXTradeMode::Cross)]
4298 fn test_validate_order_denies_unsupported_reduce_only(
4299 #[case] instrument_id: &str,
4300 #[case] trade_mode: OKXTradeMode,
4301 ) {
4302 let order = OrderTestBuilder::new(OrderType::Limit)
4303 .instrument_id(InstrumentId::from(instrument_id))
4304 .client_order_id(ClientOrderId::from("OREDUCEUNSUPPORTED"))
4305 .side(OrderSide::Sell)
4306 .price(Price::from("2000.00"))
4307 .quantity(Quantity::from("1"))
4308 .reduce_only(true)
4309 .build();
4310
4311 assert_eq!(
4312 validate_order(&order, trade_mode, OrderSubmission::Single, false),
4313 Err(OrderDeniedReason::UnsupportedReduceOnly)
4314 );
4315 }
4316
4317 fn build_config(
4318 margin_mode: Option<OKXMarginMode>,
4319 use_spot_margin: bool,
4320 ) -> OKXExecutionClientConfig {
4321 OKXExecutionClientConfig {
4322 margin_mode,
4323 use_spot_margin,
4324 ..OKXExecutionClientConfig::default()
4325 }
4326 }
4327
4328 #[rstest]
4329 #[case::spot(OKXInstrumentType::Spot, true)]
4330 #[case::margin(OKXInstrumentType::Margin, true)]
4331 #[case::swap(OKXInstrumentType::Swap, true)]
4332 #[case::futures(OKXInstrumentType::Futures, true)]
4333 #[case::option(OKXInstrumentType::Option, false)]
4334 #[case::events(OKXInstrumentType::Events, false)]
4335 fn test_supports_algo_orders(
4336 #[case] instrument_type: OKXInstrumentType,
4337 #[case] expected: bool,
4338 ) {
4339 assert_eq!(supports_algo_orders(instrument_type), expected);
4340 }
4341
4342 #[rstest]
4343 #[case::margin(
4344 vec![OKXInstrumentType::Margin],
4345 vec![OKXInstrumentType::Margin, OKXInstrumentType::Spot]
4346 )]
4347 #[case::spot_margin(
4348 vec![OKXInstrumentType::Spot, OKXInstrumentType::Margin],
4349 vec![OKXInstrumentType::Spot, OKXInstrumentType::Margin]
4350 )]
4351 #[case::any_margin(
4352 vec![OKXInstrumentType::Any, OKXInstrumentType::Margin],
4353 vec![OKXInstrumentType::Any, OKXInstrumentType::Margin]
4354 )]
4355 #[case::swap(
4356 vec![OKXInstrumentType::Swap],
4357 vec![OKXInstrumentType::Swap]
4358 )]
4359 fn test_order_routing_instrument_types(
4360 #[case] instrument_types: Vec<OKXInstrumentType>,
4361 #[case] expected: Vec<OKXInstrumentType>,
4362 ) {
4363 assert_eq!(order_routing_instrument_types(&instrument_types), expected);
4364 }
4365
4366 #[rstest]
4367 #[case::cash_no_spot_margin(AccountType::Cash, None, false, OKXTradeMode::Cash)]
4368 #[case::cash_spot_margin_cross(
4369 AccountType::Cash,
4370 Some(OKXMarginMode::Cross),
4371 true,
4372 OKXTradeMode::Cross
4373 )]
4374 #[case::cash_spot_margin_isolated(
4375 AccountType::Cash,
4376 Some(OKXMarginMode::Isolated),
4377 true,
4378 OKXTradeMode::Isolated
4379 )]
4380 #[case::cash_spot_margin_none(AccountType::Cash, None, true, OKXTradeMode::Isolated)]
4381 #[case::margin_cross(
4382 AccountType::Margin,
4383 Some(OKXMarginMode::Cross),
4384 false,
4385 OKXTradeMode::Cross
4386 )]
4387 #[case::margin_isolated(
4388 AccountType::Margin,
4389 Some(OKXMarginMode::Isolated),
4390 false,
4391 OKXTradeMode::Isolated
4392 )]
4393 #[case::margin_none(AccountType::Margin, None, false, OKXTradeMode::Isolated)]
4394 fn test_derive_default_trade_mode(
4395 #[case] account_type: AccountType,
4396 #[case] margin_mode: Option<OKXMarginMode>,
4397 #[case] use_spot_margin: bool,
4398 #[case] expected: OKXTradeMode,
4399 ) {
4400 let config = build_config(margin_mode, use_spot_margin);
4401
4402 let result = OKXExecutionClient::derive_default_trade_mode(account_type, &config);
4403
4404 assert_eq!(result, expected);
4405 }
4406
4407 #[rstest]
4408 #[case::spot_no_margin("BTC-USDT", None, false, OKXTradeMode::Cash)]
4409 #[case::spot_cross_margin("BTC-USDT", Some(OKXMarginMode::Cross), true, OKXTradeMode::Cross)]
4410 #[case::spot_isolated_margin(
4411 "ETH-USDT",
4412 Some(OKXMarginMode::Isolated),
4413 true,
4414 OKXTradeMode::Isolated
4415 )]
4416 #[case::spot_margin_no_mode("BTC-USDT", None, true, OKXTradeMode::Isolated)]
4417 #[case::swap_cross(
4418 "BTC-USDT-SWAP",
4419 Some(OKXMarginMode::Cross),
4420 false,
4421 OKXTradeMode::Cross
4422 )]
4423 #[case::swap_isolated(
4424 "BTC-USDT-SWAP",
4425 Some(OKXMarginMode::Isolated),
4426 false,
4427 OKXTradeMode::Isolated
4428 )]
4429 #[case::swap_no_mode("ETH-USDT-SWAP", None, false, OKXTradeMode::Isolated)]
4430 #[case::futures_cross(
4431 "BTC-USDT-250328",
4432 Some(OKXMarginMode::Cross),
4433 false,
4434 OKXTradeMode::Cross
4435 )]
4436 #[case::futures_isolated("BTC-USDT-250328", None, false, OKXTradeMode::Isolated)]
4437 #[case::option_cross(
4438 "BTC-USD-250328-50000-C",
4439 Some(OKXMarginMode::Cross),
4440 false,
4441 OKXTradeMode::Cross
4442 )]
4443 #[case::option_isolated("BTC-USD-250328-50000-C", None, false, OKXTradeMode::Isolated)]
4444 fn test_derive_trade_mode_for_instrument(
4445 #[case] symbol: &str,
4446 #[case] margin_mode: Option<OKXMarginMode>,
4447 #[case] use_spot_margin: bool,
4448 #[case] expected: OKXTradeMode,
4449 ) {
4450 let instrument_id = InstrumentId::from(format!("{symbol}.OKX").as_str());
4451
4452 let result = derive_trade_mode_for_instrument(instrument_id, margin_mode, use_spot_margin);
4453
4454 assert_eq!(result, expected);
4455 }
4456
4457 #[rstest]
4458 #[case::override_to_cross("cross", OKXTradeMode::Cross)]
4459 #[case::override_to_cash("cash", OKXTradeMode::Cash)]
4460 #[case::override_to_isolated("isolated", OKXTradeMode::Isolated)]
4461 #[case::override_to_spot_isolated("spot_isolated", OKXTradeMode::SpotIsolated)]
4462 #[case::case_insensitive("CROSS", OKXTradeMode::Cross)]
4463 fn test_td_mode_param_override(#[case] td_mode_value: &str, #[case] expected: OKXTradeMode) {
4464 let params = Params::new(
4465 [(
4466 "td_mode".to_string(),
4467 Value::String(td_mode_value.to_string()),
4468 )]
4469 .into(),
4470 );
4471
4472 let result = get_param_as_string(Some(¶ms), "td_mode")
4473 .and_then(|s| s.parse::<OKXTradeMode>().ok());
4474
4475 assert_eq!(result, Some(expected));
4476 }
4477
4478 #[rstest]
4479 fn test_td_mode_param_invalid_falls_through() {
4480 let params =
4481 Params::new([("td_mode".to_string(), Value::String("invalid".to_string()))].into());
4482
4483 let result = get_param_as_string(Some(¶ms), "td_mode")
4484 .and_then(|s| s.parse::<OKXTradeMode>().ok());
4485
4486 assert_eq!(result, None);
4487 }
4488
4489 #[rstest]
4490 fn test_td_mode_param_absent_falls_through() {
4491 let result = get_param_as_string(None, "td_mode");
4492
4493 assert_eq!(result, None);
4494 }
4495
4496 #[rstest]
4497 fn test_close_fraction_present_sets_reduce_only_true() {
4498 let params =
4499 Params::new([("close_fraction".to_string(), Value::String("1".to_string()))].into());
4500 let params = Some(params);
4501
4502 let close_fraction = get_param_as_string(params.as_ref(), "close_fraction");
4503 let is_reduce_only = false;
4504 let reduce_only = if close_fraction.is_some() {
4505 Some(true)
4506 } else {
4507 Some(is_reduce_only)
4508 };
4509
4510 assert_eq!(close_fraction, Some("1".to_string()));
4511 assert_eq!(reduce_only, Some(true));
4512 }
4513
4514 #[rstest]
4515 fn test_close_fraction_absent_preserves_reduce_only() {
4516 let params: Option<Params> = None;
4517
4518 let close_fraction = get_param_as_string(params.as_ref(), "close_fraction");
4519 let is_reduce_only = false;
4520 let reduce_only = if close_fraction.is_some() {
4521 Some(true)
4522 } else {
4523 Some(is_reduce_only)
4524 };
4525
4526 assert_eq!(close_fraction, None);
4527 assert_eq!(reduce_only, Some(false));
4528 }
4529
4530 #[rstest]
4531 fn test_close_fraction_absent_with_reduce_only_true() {
4532 let params: Option<Params> = None;
4533
4534 let close_fraction = get_param_as_string(params.as_ref(), "close_fraction");
4535 let is_reduce_only = true;
4536 let reduce_only = if close_fraction.is_some() {
4537 Some(true)
4538 } else {
4539 Some(is_reduce_only)
4540 };
4541
4542 assert_eq!(close_fraction, None);
4543 assert_eq!(reduce_only, Some(true));
4544 }
4545
4546 fn make_query_order_report(cid: Option<&str>, vid: &str) -> OrderStatusReport {
4547 OrderStatusReport::new(
4548 AccountId::from("OKX-001"),
4549 InstrumentId::from("BTC-USDT.OKX"),
4550 cid.map(ClientOrderId::from),
4551 VenueOrderId::from(vid),
4552 OrderSide::Buy.into(),
4553 OrderType::Limit,
4554 TimeInForce::Gtc,
4555 OrderStatus::Accepted,
4556 Quantity::new(1.0, 0),
4557 Quantity::zero(0),
4558 UnixNanos::default(),
4559 UnixNanos::default(),
4560 UnixNanos::default(),
4561 None,
4562 )
4563 }
4564
4565 fn with_linked(mut report: OrderStatusReport, linked: &[&str]) -> OrderStatusReport {
4566 report.linked_order_ids = Some(linked.iter().map(|s| ClientOrderId::from(*s)).collect());
4567 report
4568 }
4569
4570 #[rstest]
4571 fn test_merge_order_status_reports_keeps_filled_regular_child() {
4572 let mut filled = make_query_order_report(Some("O-PARENT"), "V-CHILD");
4573 filled.order_status = OrderStatus::Filled;
4574 filled.filled_qty = Quantity::new(1.0, 0);
4575 filled.ts_last = UnixNanos::from(100);
4576
4577 let mut triggered = make_query_order_report(Some("O-PARENT"), "V-CHILD");
4578 triggered.order_status = OrderStatus::Triggered;
4579 triggered.filled_qty = Quantity::new(1.0, 0);
4580 triggered.ts_last = UnixNanos::from(200);
4581
4582 let mut reports = vec![filled];
4583 merge_order_status_reports(&mut reports, vec![triggered]);
4584
4585 assert_eq!(reports.len(), 1);
4586 assert_eq!(reports[0].order_status, OrderStatus::Filled);
4587 assert_eq!(
4588 reports[0].client_order_id,
4589 Some(ClientOrderId::from("O-PARENT"))
4590 );
4591 assert_eq!(reports[0].venue_order_id, VenueOrderId::from("V-CHILD"));
4592 }
4593
4594 #[rstest]
4595 fn test_merge_order_status_reports_replaces_pending_parent_with_triggered_child() {
4596 let mut accepted = make_query_order_report(Some("O-PARENT"), "V-CHILD");
4597 accepted.ts_last = UnixNanos::from(100);
4598
4599 let mut triggered = make_query_order_report(Some("O-PARENT"), "V-CHILD");
4600 triggered.order_status = OrderStatus::Triggered;
4601 triggered.ts_last = UnixNanos::from(200);
4602
4603 let mut reports = vec![accepted];
4604 merge_order_status_reports(&mut reports, vec![triggered]);
4605
4606 assert_eq!(reports.len(), 1);
4607 assert_eq!(reports[0].order_status, OrderStatus::Triggered);
4608 assert_eq!(reports[0].ts_last, UnixNanos::from(200));
4609 }
4610
4611 #[rstest]
4612 fn test_select_query_order_report_matches_client_order_id() {
4613 let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
4614 let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
4615 assert_eq!(
4616 selected.and_then(|r| r.client_order_id),
4617 Some(ClientOrderId::from("O-001"))
4618 );
4619 }
4620
4621 #[rstest]
4622 fn test_select_query_order_report_client_wins_over_venue_mismatch() {
4623 let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
4624 let selected = select_query_order_report(
4625 reports,
4626 ClientOrderId::from("O-001"),
4627 Some(VenueOrderId::from("V-OTHER")),
4628 );
4629 assert_eq!(
4630 selected.and_then(|r| r.client_order_id),
4631 Some(ClientOrderId::from("O-001"))
4632 );
4633 }
4634
4635 #[rstest]
4636 #[case(
4637 OrderStatus::Accepted,
4638 Quantity::zero(0),
4639 OrderStatus::Triggered,
4640 Quantity::zero(0),
4641 OrderStatus::Triggered
4642 )]
4643 #[case(
4644 OrderStatus::Triggered,
4645 Quantity::zero(0),
4646 OrderStatus::PartiallyFilled,
4647 Quantity::new(0.5, 1),
4648 OrderStatus::PartiallyFilled
4649 )]
4650 #[case(
4651 OrderStatus::PartiallyFilled,
4652 Quantity::new(0.5, 1),
4653 OrderStatus::Filled,
4654 Quantity::new(1.0, 0),
4655 OrderStatus::Filled
4656 )]
4657 #[case(
4658 OrderStatus::Triggered,
4659 Quantity::zero(0),
4660 OrderStatus::Canceled,
4661 Quantity::zero(0),
4662 OrderStatus::Canceled
4663 )]
4664 #[case(
4665 OrderStatus::Triggered,
4666 Quantity::zero(0),
4667 OrderStatus::Rejected,
4668 Quantity::zero(0),
4669 OrderStatus::Rejected
4670 )]
4671 fn test_select_query_order_report_chooses_most_advanced_client_match_regardless_of_order(
4672 #[case] first_status: OrderStatus,
4673 #[case] first_filled_qty: Quantity,
4674 #[case] second_status: OrderStatus,
4675 #[case] second_filled_qty: Quantity,
4676 #[case] expected_status: OrderStatus,
4677 ) {
4678 let mut first = make_query_order_report(Some("O-001"), "V-PARENT");
4679 first.order_status = first_status;
4680 first.filled_qty = first_filled_qty;
4681 let mut second = make_query_order_report(Some("O-001"), "V-CHILD");
4682 second.order_status = second_status;
4683 second.filled_qty = second_filled_qty;
4684
4685 for reports in [vec![first.clone(), second.clone()], vec![second, first]] {
4686 let selected = select_query_order_report(
4687 reports,
4688 ClientOrderId::from("O-001"),
4689 Some(VenueOrderId::from("V-PARENT")),
4690 )
4691 .unwrap();
4692
4693 assert_eq!(selected.order_status, expected_status);
4694 }
4695 }
4696
4697 #[rstest]
4698 fn test_select_query_order_report_falls_back_to_venue_order_id() {
4699 let reports = vec![make_query_order_report(Some("O-CHILD"), "V-1")];
4702 let selected = select_query_order_report(
4703 reports,
4704 ClientOrderId::from("O-PARENT"),
4705 Some(VenueOrderId::from("V-1")),
4706 );
4707 assert_eq!(
4708 selected.map(|r| r.venue_order_id.as_str().to_string()),
4709 Some("V-1".to_string()),
4710 );
4711 }
4712
4713 #[rstest]
4714 fn test_select_query_order_report_rejects_when_nothing_matches() {
4715 let reports = vec![make_query_order_report(Some("O-OTHER"), "V-OTHER")];
4716 let selected = select_query_order_report(
4717 reports,
4718 ClientOrderId::from("O-001"),
4719 Some(VenueOrderId::from("V-1")),
4720 );
4721 assert!(selected.is_none());
4722 }
4723
4724 #[rstest]
4725 fn test_select_query_order_report_rejects_when_client_differs_and_no_vid_provided() {
4726 let reports = vec![make_query_order_report(Some("O-OTHER"), "V-1")];
4727 let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
4728 assert!(selected.is_none());
4729 }
4730
4731 #[rstest]
4732 fn test_select_query_order_report_ignores_linked_order_ids_for_parent_with_attached_tp() {
4733 let child_cid = "O-CHILD-TP";
4737 let reports = vec![with_linked(
4738 make_query_order_report(Some("O-PARENT"), "V-PARENT"),
4739 &[child_cid, "O-CHILD-SL"],
4740 )];
4741 let selected = select_query_order_report(reports, ClientOrderId::from(child_cid), None);
4742 assert!(selected.is_none());
4743 }
4744
4745 #[rstest]
4746 fn test_select_query_order_report_client_match_wins_over_vid_match_elsewhere() {
4747 let reports = vec![
4750 make_query_order_report(Some("O-OTHER"), "V-1"),
4751 make_query_order_report(Some("O-001"), "V-2"),
4752 ];
4753 let selected = select_query_order_report(
4754 reports,
4755 ClientOrderId::from("O-001"),
4756 Some(VenueOrderId::from("V-1")),
4757 );
4758 assert_eq!(
4759 selected.and_then(|r| r.client_order_id),
4760 Some(ClientOrderId::from("O-001")),
4761 );
4762 }
4763
4764 fn build_test_exec_client() -> OKXExecutionClient {
4765 build_test_exec_client_with_cache().0
4766 }
4767
4768 #[rstest]
4769 #[case::stop_limit(OrderType::StopLimit, OKXAlgoOrderType::Trigger)]
4770 #[case::stop_market(OrderType::StopMarket, OKXAlgoOrderType::Trigger)]
4771 #[case::touched_limit(OrderType::LimitIfTouched, OKXAlgoOrderType::Trigger)]
4772 #[case::touched_market(OrderType::MarketIfTouched, OKXAlgoOrderType::Trigger)]
4773 #[case::sl_limit(OrderType::StopLimit, OKXAlgoOrderType::Conditional)]
4774 #[case::sl_market(OrderType::StopMarket, OKXAlgoOrderType::Conditional)]
4775 #[case::tp_limit(OrderType::LimitIfTouched, OKXAlgoOrderType::Conditional)]
4776 #[case::tp_market(OrderType::MarketIfTouched, OKXAlgoOrderType::Conditional)]
4777 #[tokio::test]
4778 async fn test_modify_algo_order_http(
4779 #[case] order_type: OrderType,
4780 #[case] algo_type: OKXAlgoOrderType,
4781 #[values("all", "trigger", "quantity")] amendment: &str,
4782 #[values(("0", "0"), ("0", "51000"), ("0", "50004"), ("1", "51000"), ("1", "50004"))]
4783 response: (&'static str, &'static str),
4784 ) {
4785 let (code, s_code) = response;
4786 let (client, cache) = build_test_exec_client_with_cache();
4787 let (submit_tx, mut submit_rx) = tokio::sync::mpsc::unbounded_channel();
4788 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
4789
4790 let router = Router::new()
4791 .route(
4792 "/api/v5/trade/order-algo",
4793 post(move |Json(request): Json<Value>| {
4794 let tx = submit_tx.clone();
4795 async move {
4796 tx.send(request).unwrap();
4797 Json(serde_json::json!({"code": "0", "msg": "", "data": [{
4798 "algoId": "7003", "sCode": "0", "sMsg": ""
4799 }]}))
4800 }
4801 }),
4802 )
4803 .route(
4804 "/api/v5/trade/amend-algos",
4805 post(move |Json(request): Json<Value>| {
4806 let tx = request_tx.clone();
4807 async move {
4808 tx.send(request).unwrap();
4809 Json(serde_json::json!({"code": code, "msg": "All operations failed", "data": [{
4810 "algoId": "7003", "sCode": s_code, "sMsg": "test message"
4811 }]}))
4812 }
4813 }),
4814 );
4815
4816 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4817 let base_url = format!("http://{}", listener.local_addr().unwrap());
4818
4819 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
4820 let mut client = client;
4821 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
4822 client.emitter.set_sender(event_tx);
4823 client.http_client = OKXHttpClient::with_credentials(
4824 Some("test_key".into()),
4825 Some("test_secret".into()),
4826 Some("test_pass".into()),
4827 Some(base_url),
4828 5,
4829 0,
4830 1,
4831 1,
4832 crate::common::enums::OKXEnvironment::Live,
4833 None,
4834 )
4835 .unwrap();
4836 let client_order_id = ClientOrderId::from("OMODIFYALGO");
4837 let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
4838 let order = OrderTestBuilder::new(order_type)
4839 .instrument_id(instrument_id)
4840 .client_order_id(client_order_id)
4841 .side(OrderSide::Sell)
4842 .quantity(Quantity::from("0.03"))
4843 .price(Price::from("54900.0"))
4844 .trigger_price(Price::from("55000.0"))
4845 .trigger_type(TriggerType::MarkPrice)
4846 .build();
4847 let accepted = TestOrderEventStubs::accepted(
4848 &order,
4849 client.core.account_id,
4850 VenueOrderId::from("7003"),
4851 );
4852 cache
4853 .borrow_mut()
4854 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
4855 .unwrap();
4856 let mut submit = SubmitOrder::from_order(
4857 &order,
4858 TraderId::from("TESTER-001"),
4859 Some(*OKX_CLIENT_ID),
4860 None,
4861 UUID4::new(),
4862 UnixNanos::default(),
4863 );
4864
4865 if algo_type == OKXAlgoOrderType::Conditional {
4866 let params = Params::new([("close_fraction".into(), Value::String("1".into()))].into());
4867 submit.params = Some(params);
4868 }
4869
4870 client.submit_order(submit).unwrap();
4871 let submitted = tokio::time::timeout(Duration::from_secs(5), submit_rx.recv())
4872 .await
4873 .unwrap()
4874 .unwrap();
4875 assert_eq!(
4876 submitted["ordType"],
4877 serde_json::to_value(algo_type).unwrap()
4878 );
4879 wait_until_async(
4880 || async { client.pending_tasks.is_empty() },
4881 Duration::from_secs(5),
4882 )
4883 .await;
4884 assert!(matches!(
4885 event_rx.try_recv().unwrap(),
4886 ExecutionEvent::Order(OrderEventAny::Submitted(_))
4887 ));
4888 assert!(event_rx.try_recv().is_err());
4889 cache.borrow_mut().update_order(&accepted).unwrap();
4890 let limit = matches!(order_type, OrderType::StopLimit | OrderType::LimitIfTouched);
4891 client
4892 .modify_order(ModifyOrder::new(
4893 TraderId::from("TESTER-001"),
4894 Some(*OKX_CLIENT_ID),
4895 order.strategy_id(),
4896 instrument_id,
4897 client_order_id,
4898 Some(VenueOrderId::from("stale-parent")),
4899 (amendment != "trigger").then(|| Quantity::from("0.04")),
4900 (limit && amendment == "all").then(|| Price::from("53900.0")),
4901 (amendment != "quantity").then(|| Price::from("54000.0")),
4902 UUID4::new(),
4903 UnixNanos::default(),
4904 None,
4905 None,
4906 ))
4907 .unwrap();
4908 let request = tokio::time::timeout(Duration::from_secs(5), request_rx.recv())
4909 .await
4910 .unwrap()
4911 .unwrap();
4912 let mut expected = serde_json::json!({"instId": "BTC-USDT-SWAP", "algoId": "7003"});
4913 if amendment != "trigger" {
4914 expected["newSz"] = Value::String("0.04".into());
4915 }
4916
4917 let (trigger_field, type_field, price_field) = if algo_type == OKXAlgoOrderType::Trigger {
4918 ("newTriggerPx", "newTriggerPxType", "newOrdPx")
4919 } else if matches!(order_type, OrderType::StopLimit | OrderType::StopMarket) {
4920 ("newSlTriggerPx", "newSlTriggerPxType", "newSlOrdPx")
4921 } else {
4922 ("newTpTriggerPx", "newTpTriggerPxType", "newTpOrdPx")
4923 };
4924
4925 if amendment != "quantity" {
4926 expected[trigger_field] = Value::String("54000.0".into());
4927 expected[type_field] = Value::String("mark".into());
4928 }
4929
4930 if limit && amendment == "all" {
4931 expected[price_field] = Value::String("53900.0".into());
4932 }
4933
4934 assert_eq!(request, expected);
4935 wait_until_async(
4936 || async { client.pending_tasks.is_empty() },
4937 Duration::from_secs(5),
4938 )
4939 .await;
4940
4941 if s_code == "51000" {
4942 let ExecutionEvent::Order(OrderEventAny::ModifyRejected(rejected)) =
4943 event_rx.try_recv().unwrap()
4944 else {
4945 panic!("expected OrderModifyRejected");
4946 };
4947
4948 assert_eq!(rejected.client_order_id, client_order_id);
4949 assert_eq!(rejected.venue_order_id, Some(VenueOrderId::from("7003")));
4950 assert_eq!(rejected.instrument_id, instrument_id);
4951 assert_eq!(rejected.strategy_id, order.strategy_id());
4952 assert_eq!(rejected.reason.as_str(), "OKX error 51000: test message");
4953 }
4954
4955 assert!(event_rx.try_recv().is_err());
4956 server.abort();
4957 }
4958
4959 #[rstest]
4960 #[case::trigger(
4961 OrderType::StopLimit,
4962 OKXAlgoOrderType::Trigger,
4963 "newOrdPx",
4964 "newTriggerPx",
4965 "newTriggerPxType"
4966 )]
4967 #[case::sl(
4968 OrderType::StopLimit,
4969 OKXAlgoOrderType::Conditional,
4970 "newSlOrdPx",
4971 "newSlTriggerPx",
4972 "newSlTriggerPxType"
4973 )]
4974 #[case::tp(
4975 OrderType::LimitIfTouched,
4976 OKXAlgoOrderType::Conditional,
4977 "newTpOrdPx",
4978 "newTpTriggerPx",
4979 "newTpTriggerPxType"
4980 )]
4981 fn test_algo_amend_request_preserves_unconfirmed_terms(
4982 #[case] order_type: OrderType,
4983 #[case] algo_type: OKXAlgoOrderType,
4984 #[case] price_field: &str,
4985 #[case] trigger_field: &str,
4986 #[case] type_field: &str,
4987 ) {
4988 let order = OrderTestBuilder::new(order_type)
4989 .instrument_id(InstrumentId::from("BTC-USDT-SWAP.OKX"))
4990 .client_order_id(ClientOrderId::from("OUNCONFIRMED"))
4991 .side(OrderSide::Sell)
4992 .quantity(Quantity::from("0.03"))
4993 .price(Price::from("54900.0"))
4994 .trigger_price(Price::from("55000.0"))
4995 .trigger_type(TriggerType::MarkPrice)
4996 .build();
4997 let context = OrderContext::from(&order);
4998
4999 let mut command = ModifyOrder::new(
5000 TraderId::from("TESTER-001"),
5001 Some(*OKX_CLIENT_ID),
5002 order.strategy_id(),
5003 order.instrument_id(),
5004 order.client_order_id(),
5005 Some(VenueOrderId::from("7003")),
5006 None,
5007 Some(Price::from("53900.0")),
5008 None,
5009 UUID4::new(),
5010 UnixNanos::default(),
5011 None,
5012 None,
5013 );
5014 let first = algo_amend_request(&command, Some(algo_type), Some(context)).unwrap();
5015 command.price = None;
5016 command.trigger_price = Some(Price::from("54000.0"));
5017 let second = algo_amend_request(&command, Some(algo_type), Some(context)).unwrap();
5018 command.trigger_price = None;
5019 command.quantity = Some(Quantity::from("0.04"));
5020 let third = algo_amend_request(&command, Some(algo_type), Some(context)).unwrap();
5021 let mut first_expected = serde_json::json!({"instId": "BTC-USDT-SWAP", "algoId": "7003"});
5022 first_expected[price_field] = Value::String("53900.0".into());
5023 let mut second_expected = serde_json::json!({"instId": "BTC-USDT-SWAP", "algoId": "7003"});
5024 second_expected[trigger_field] = Value::String("54000.0".into());
5025 second_expected[type_field] = Value::String("mark".into());
5026 assert_eq!(serde_json::to_value(first).unwrap(), first_expected);
5027 assert_eq!(serde_json::to_value(second).unwrap(), second_expected);
5028 assert_eq!(
5029 serde_json::to_value(third).unwrap(),
5030 serde_json::json!({
5031 "instId": "BTC-USDT-SWAP", "algoId": "7003", "newSz": "0.04"
5032 })
5033 );
5034 }
5035
5036 #[rstest]
5037 #[case::unknown(
5038 None,
5039 true,
5040 true,
5041 "Cannot determine OKX algo order terms for modification"
5042 )]
5043 #[case::missing_context(
5044 Some(OKXAlgoOrderType::Trigger),
5045 false,
5046 true,
5047 "Cannot determine OKX algo order terms for modification"
5048 )]
5049 #[case::missing_id(
5050 Some(OKXAlgoOrderType::Trigger),
5051 true,
5052 false,
5053 "No algo ID available for pending conditional order"
5054 )]
5055 #[case::trailing(
5056 Some(OKXAlgoOrderType::MoveOrderStop),
5057 true,
5058 true,
5059 "OKX does not support modification of this algo order type"
5060 )]
5061 fn test_modify_algo_order_rejects_locally(
5062 #[case] algo_type: Option<OKXAlgoOrderType>,
5063 #[case] has_context: bool,
5064 #[case] has_id: bool,
5065 #[case] reason: &str,
5066 ) {
5067 let mut client = build_test_exec_client();
5068 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
5069 client.emitter.set_sender(event_tx);
5070 let client_order_id = ClientOrderId::from("OMODIFYINVALID");
5071 let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
5072 let order = OrderTestBuilder::new(OrderType::StopMarket)
5073 .instrument_id(instrument_id)
5074 .client_order_id(client_order_id)
5075 .side(OrderSide::Sell)
5076 .quantity(Quantity::from("0.03"))
5077 .trigger_price(Price::from("55000.0"))
5078 .build();
5079
5080 if has_context {
5081 client
5082 .ws_dispatch_state
5083 .track_order_context(OrderContext::from(&order));
5084 }
5085
5086 if let Some(order_type) = algo_type {
5087 client.ws_dispatch_state.algo_order_states.insert(
5088 client_order_id,
5089 AlgoOrderState {
5090 order_type,
5091 ts_last: UnixNanos::default(),
5092 },
5093 );
5094 }
5095
5096 let venue_order_id = has_id.then(|| VenueOrderId::from("7004"));
5097 client.modify_algo_order(&ModifyOrder::new(
5098 TraderId::from("TESTER-001"),
5099 Some(*OKX_CLIENT_ID),
5100 order.strategy_id(),
5101 instrument_id,
5102 client_order_id,
5103 venue_order_id,
5104 None,
5105 None,
5106 Some(Price::from("54000.0")),
5107 UUID4::new(),
5108 UnixNanos::default(),
5109 None,
5110 None,
5111 ));
5112
5113 let ExecutionEvent::Order(OrderEventAny::ModifyRejected(rejected)) =
5114 event_rx.try_recv().unwrap()
5115 else {
5116 panic!("expected OrderModifyRejected");
5117 };
5118
5119 assert_eq!(rejected.client_order_id, client_order_id);
5120 assert_eq!(rejected.venue_order_id, venue_order_id);
5121 assert_eq!(rejected.instrument_id, instrument_id);
5122 assert_eq!(rejected.strategy_id, order.strategy_id());
5123 assert_eq!(rejected.reason.as_str(), reason);
5124 assert!(event_rx.try_recv().is_err());
5125 assert!(client.pending_tasks.is_empty());
5126 }
5127
5128 #[rstest]
5129 #[case::client_id(true)]
5130 #[case::venue_id(false)]
5131 fn test_report_fallback_uses_current_cache(#[case] match_client_id: bool) {
5132 let (client, cache) = build_test_exec_client_with_cache();
5133 let parent = make_query_order_report(Some("O-PARENT"), "V-PARENT");
5134 let regular = make_query_order_report(Some("O-REGULAR"), "V-REGULAR");
5135 let fallback = make_query_order_report(None, "V-CHILD");
5136 let external = make_query_order_report(None, "V-EXTERNAL");
5137 let fallback_parents = AHashMap::from_iter([
5138 (1, parent.clone()),
5139 (
5140 2,
5141 make_query_order_report(Some("O-EXTERNAL"), "V-EXTERNAL-PARENT"),
5142 ),
5143 ]);
5144 let mut reports = vec![regular.clone(), fallback.clone(), external.clone()];
5145 retain_external_regular_order_fallbacks(&client.core, &mut reports, &fallback_parents);
5146 assert_eq!(reports, [regular.clone(), fallback, external.clone()]);
5147
5148 let order = OrderTestBuilder::new(OrderType::Limit)
5149 .instrument_id(parent.instrument_id)
5150 .side(OrderSide::Buy)
5151 .price(Price::from("100.00"))
5152 .quantity(Quantity::from("1"))
5153 .client_order_id(ClientOrderId::from(if match_client_id {
5154 "O-PARENT"
5155 } else {
5156 "O-OTHER"
5157 }))
5158 .build();
5159
5160 let accepted =
5161 TestOrderEventStubs::accepted(&order, client.core.account_id, parent.venue_order_id);
5162 cache
5163 .borrow_mut()
5164 .add_order(order, None, Some(*OKX_CLIENT_ID), false)
5165 .unwrap();
5166 cache.borrow_mut().update_order(&accepted).unwrap();
5167 retain_external_regular_order_fallbacks(&client.core, &mut reports, &fallback_parents);
5168
5169 assert_eq!(reports, [regular, external]);
5170 }
5171
5172 fn build_test_exec_client_with_cache() -> (OKXExecutionClient, Rc<RefCell<Cache>>) {
5173 let config = OKXExecutionClientConfig {
5174 api_key: Some("test_key".into()),
5175 api_secret: Some("test_secret".into()),
5176 api_passphrase: Some("test_pass".into()),
5177 ..OKXExecutionClientConfig::default()
5178 };
5179
5180 let cache = Rc::new(RefCell::new(Cache::default()));
5181 let core = ExecutionClientCore::new(
5182 TraderId::from("TESTER-001"),
5183 ClientId::from("OKX-TEST"),
5184 *OKX_VENUE,
5185 OmsType::Hedging,
5186 config.account_id,
5187 AccountType::Cash,
5188 None,
5189 Rc::clone(&cache),
5190 );
5191
5192 (
5193 OKXExecutionClient::new(core, config).expect("failed to build test client"),
5194 cache,
5195 )
5196 }
5197
5198 #[rstest]
5199 #[case::reset(ExecutionTaskBoundary::Reset)]
5200 #[case::dispose(ExecutionTaskBoundary::Dispose)]
5201 #[case::repeated_stop(ExecutionTaskBoundary::RepeatedStop)]
5202 #[tokio::test]
5203 async fn lifecycle_boundary_terminates_owned_execution_task(
5204 #[case] boundary: ExecutionTaskBoundary,
5205 ) {
5206 let mut client = build_test_exec_client();
5207
5208 if matches!(boundary, ExecutionTaskBoundary::RepeatedStop) {
5209 client.stop().expect("initial stop");
5210 }
5211
5212 let (drop_tx, drop_rx) = tokio::sync::oneshot::channel();
5213 let signal = DropSignal(Some(drop_tx));
5214 client.spawn_task("pending lifecycle task", async move {
5215 let _signal = signal;
5216 std::future::pending::<anyhow::Result<()>>().await
5217 });
5218
5219 match boundary {
5220 ExecutionTaskBoundary::Reset => client.reset().expect("reset"),
5221 ExecutionTaskBoundary::Dispose => client.dispose().expect("dispose"),
5222 ExecutionTaskBoundary::RepeatedStop => client.stop().expect("repeated stop"),
5223 }
5224
5225 tokio::time::timeout(Duration::from_secs(1), drop_rx)
5226 .await
5227 .expect("lifecycle boundary must drop the owned task")
5228 .expect("drop signal");
5229 terminate_tasks(&client.pending_tasks, "test execution client")
5230 .await
5231 .expect("execution task terminated");
5232 assert!(client.pending_tasks.is_empty());
5233 }
5234
5235 #[rstest]
5236 #[case("BTC-USDT.OKX", false, None, CancelAllOrdersRoute::Batch)]
5237 #[case("BTC-USDT.OKX", true, None, CancelAllOrdersRoute::MassCancelHttp)]
5238 #[case(
5239 "BTC-USDT.OKX",
5240 true,
5241 Some(OrderSide::Buy),
5242 CancelAllOrdersRoute::Batch
5243 )]
5244 #[case(
5245 "BTC-USDT.OKX",
5246 true,
5247 Some(OrderSide::Sell),
5248 CancelAllOrdersRoute::Batch
5249 )]
5250 #[case(
5251 "ETH-USD-SWAP_ETH-USD-231229.OKX",
5252 false,
5253 None,
5254 CancelAllOrdersRoute::SpreadHttp
5255 )]
5256 #[case(
5257 "ETH-USD-SWAP_ETH-USD-231229.OKX",
5258 false,
5259 Some(OrderSide::Buy),
5260 CancelAllOrdersRoute::Batch
5261 )]
5262 #[case(
5263 "ETH-USD-SWAP_ETH-USD-231229.OKX",
5264 true,
5265 Some(OrderSide::Sell),
5266 CancelAllOrdersRoute::Batch
5267 )]
5268 fn test_cancel_all_orders_route(
5269 #[case] instrument_id: &str,
5270 #[case] use_mm_mass_cancel: bool,
5271 #[case] order_side: Option<OrderSide>,
5272 #[case] expected: CancelAllOrdersRoute,
5273 ) {
5274 let mut client = build_test_exec_client();
5275 client.config.use_mm_mass_cancel = use_mm_mass_cancel;
5276
5277 assert_eq!(
5278 client.cancel_all_orders_route(InstrumentId::from(instrument_id), order_side),
5279 expected
5280 );
5281 }
5282
5283 #[rstest]
5284 #[case::buy(Some(OrderSide::Buy), false, &["OCANCEL0", "OCANCEL2"])]
5285 #[case::sell(Some(OrderSide::Sell), false, &["OCANCEL1", "OCANCEL3"])]
5286 #[case::both(None, false, &["OCANCEL0", "OCANCEL1", "OCANCEL2", "OCANCEL3"])]
5287 #[case::mass_buy(Some(OrderSide::Buy), true, &["OCANCEL0", "OCANCEL2"])]
5288 #[case::mass_sell(Some(OrderSide::Sell), true, &["OCANCEL1", "OCANCEL3"])]
5289 #[tokio::test]
5290 async fn test_cancel_all_orders_filters_cached_orders(
5291 #[case] order_side: Option<OrderSide>,
5292 #[case] use_mm_mass_cancel: bool,
5293 #[case] expected_ids: &[&str],
5294 #[values(1, 10, 11)] copies: usize,
5295 ) {
5296 let (mut client, cache) = build_test_exec_client_with_cache();
5297 client.config.use_mm_mass_cancel = use_mm_mass_cancel;
5298 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
5299
5300 let router = Router::new().route(
5301 "/ws",
5302 get(move |ws: WebSocketUpgrade| {
5303 let tx = request_tx.clone();
5304 async move {
5305 ws.on_upgrade(move |mut socket| async move {
5306 while let Some(Ok(Message::Text(text))) = socket.next().await {
5307 let request: Value = serde_json::from_str(&text).unwrap();
5308 if request["op"] == "login" {
5309 let response =
5310 serde_json::json!({"event": "login", "code": "0", "msg": "", "connId": "test-conn"});
5311 socket
5312 .send(Message::Text(response.to_string().into()))
5313 .await
5314 .unwrap();
5315 } else {
5316 tx.send(request).unwrap();
5317 }
5318 }
5319 })
5320 }
5321 }),
5322 );
5323
5324 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
5325 let url = format!("ws://{}/ws", listener.local_addr().unwrap());
5326
5327 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
5328 client.ws_private = OKXWebSocketClient::with_credentials(
5329 Some(url),
5330 Some("test_key".into()),
5331 Some("test_secret".into()),
5332 Some("test_pass".into()),
5333 Some(client.core.account_id),
5334 None,
5335 None,
5336 client.config.transport_backend,
5337 None,
5338 )
5339 .unwrap();
5340 client
5341 .ws_private
5342 .cache_inst_id_code(ustr::Ustr::from("BTC-USDT"), 10459);
5343 client.ws_private.connect().await.unwrap();
5344 client.ws_private.wait_until_active(5.0).await.unwrap();
5345 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
5346
5347 let expected_ids: Vec<_> = (0..copies)
5348 .flat_map(|copy| {
5349 expected_ids.iter().map(move |id| {
5350 let index: usize = id.strip_prefix("OCANCEL").unwrap().parse().unwrap();
5351 format!("OCANCEL{}", copy * 8 + index)
5352 })
5353 })
5354 .collect();
5355
5356 let mut expected: Vec<_> = expected_ids
5357 .iter()
5358 .map(|id| ClientOrderId::from(id.as_str()))
5359 .collect();
5360
5361 for copy in 0..copies {
5362 for (index, (symbol, side, strategy, open)) in [
5363 ("BTC-USDT.OKX", OrderSide::Buy, "S-001", true),
5364 ("BTC-USDT.OKX", OrderSide::Sell, "S-001", true),
5365 ("BTC-USDT.OKX", OrderSide::Buy, "S-002", true),
5366 ("BTC-USDT.OKX", OrderSide::Sell, "S-002", true),
5367 ("ETH-USDT.OKX", OrderSide::Buy, "S-001", true),
5368 ("ETH-USDT.OKX", OrderSide::Sell, "S-001", true),
5369 ("BTC-USDT.OKX", OrderSide::Buy, "S-001", false),
5370 ("BTC-USDT.OKX", OrderSide::Sell, "S-001", false),
5371 ]
5372 .into_iter()
5373 .enumerate()
5374 {
5375 let index = copy * 8 + index;
5376 let client_order_id = ClientOrderId::from(format!("OCANCEL{index}"));
5377 let mut order = OrderTestBuilder::new(OrderType::Limit)
5378 .instrument_id(InstrumentId::from(symbol))
5379 .client_order_id(client_order_id)
5380 .strategy_id(StrategyId::from(strategy))
5381 .side(side)
5382 .price(Price::from("100.00"))
5383 .quantity(Quantity::from("1"))
5384 .build();
5385 let accepted = TestOrderEventStubs::accepted(
5386 &order,
5387 client.core.account_id,
5388 VenueOrderId::from(format!("{index}")),
5389 );
5390 cache
5391 .borrow_mut()
5392 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
5393 .unwrap();
5394 order = cache.borrow_mut().update_order(&accepted).unwrap();
5395
5396 if !open {
5397 let canceled = TestOrderEventStubs::canceled(
5398 &order,
5399 client.core.account_id,
5400 order.venue_order_id(),
5401 );
5402 cache.borrow_mut().update_order(&canceled).unwrap();
5403 }
5404 }
5405 }
5406
5407 client
5408 .cancel_all_orders(CancelAllOrders::new(
5409 TraderId::from("TESTER-001"),
5410 Some(*OKX_CLIENT_ID),
5411 StrategyId::from("S-001"),
5412 instrument_id,
5413 order_side,
5414 UUID4::new(),
5415 UnixNanos::default(),
5416 None,
5417 None,
5418 ))
5419 .unwrap();
5420 let mut actual: Vec<_> = client
5421 .ws_dispatch_state
5422 .order_identities
5423 .iter()
5424 .map(|entry| *entry.key())
5425 .collect();
5426 actual.sort();
5427
5428 let expected_sizes: &[usize] = match (copies, order_side) {
5429 (1, Some(_)) => &[2],
5430 (1, None) => &[4],
5431 (10, Some(_)) => &[20],
5432 (10, None) => &[20, 20],
5433 (11, Some(_)) => &[20, 2],
5434 (11, None) => &[20, 20, 4],
5435 _ => unreachable!(),
5436 };
5437
5438 let mut payload = Vec::new();
5439 let mut batch_sizes = Vec::new();
5440
5441 for _ in expected_sizes {
5442 let request = tokio::time::timeout(Duration::from_secs(5), request_rx.recv())
5443 .await
5444 .unwrap()
5445 .unwrap();
5446 assert_eq!(request["op"], "batch-cancel-orders");
5447 let args = request["args"].as_array().unwrap();
5448 batch_sizes.push(args.len());
5449 payload.extend(args.iter().cloned());
5450 }
5451
5452 payload.sort_by_key(|item| item["clOrdId"].as_str().unwrap().to_string());
5453
5454 let mut expected_payload: Vec<_> = expected_ids
5455 .iter()
5456 .map(|id| {
5457 serde_json::json!({
5458 "instIdCode": 10459,
5459 "clOrdId": id,
5460 "ordId": id.strip_prefix("OCANCEL").unwrap(),
5461 })
5462 })
5463 .collect();
5464
5465 expected.sort();
5466 expected_payload.sort_by_key(|item| item["clOrdId"].as_str().unwrap().to_string());
5467 client.ws_private.close().await.unwrap();
5468 server.abort();
5469
5470 assert_eq!(batch_sizes, expected_sizes);
5471 assert_eq!(payload, expected_payload);
5472 assert_eq!(actual, expected);
5473 terminate_tasks(&client.pending_tasks, "test execution client")
5474 .await
5475 .unwrap();
5476 }
5477
5478 #[rstest]
5479 #[case::algo_buy("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Buy), &["0", "2"], false)]
5480 #[case::algo_sell("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Sell), &["1", "3"], false)]
5481 #[case::algo_both("BTC-USDT.OKX", OrderType::StopLimit, None, &["0", "1", "2", "3"], false)]
5482 #[case::spread_buy("ETH-USD-SWAP_ETH-USD-231229.OKX", OrderType::Limit, Some(OrderSide::Buy), &["0", "2"], false)]
5483 #[case::spread_sell("ETH-USD-SWAP_ETH-USD-231229.OKX", OrderType::Limit, Some(OrderSide::Sell), &["1", "3"], false)]
5484 #[case::advance_buy("BTC-USDT.OKX", OrderType::TrailingStopMarket, Some(OrderSide::Buy), &["0", "2"], false)]
5485 #[case::advance_sell("BTC-USDT.OKX", OrderType::TrailingStopMarket, Some(OrderSide::Sell), &["1", "3"], false)]
5486 #[case::advance_both("BTC-USDT.OKX", OrderType::TrailingStopMarket, None, &["0", "1", "2", "3"], false)]
5487 #[case::mixed_buy("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Buy), &["0", "2"], true)]
5488 #[case::mixed_sell("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Sell), &["1", "3"], true)]
5489 #[case::mixed_both("BTC-USDT.OKX", OrderType::StopLimit, None, &["0", "1", "2", "3"], true)]
5490 #[tokio::test]
5491 async fn test_cancel_all_orders_filters_http_requests(
5492 #[case] symbol: &str,
5493 #[case] order_type: OrderType,
5494 #[case] order_side: Option<OrderSide>,
5495 #[case] expected_ids: &[&str],
5496 #[case] mixed_types: bool,
5497 #[values(1, 5, 6)] copies: usize,
5498 ) {
5499 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
5500
5501 let router = Router::new().fallback(post(
5502 move |uri: axum::http::Uri, Json(request): Json<Value>| {
5503 let tx = request_tx.clone();
5504 async move {
5505 let mut response = serde_json::from_str::<Value>(include_str!(
5506 "../test_data/http_cancel_algo_order_response.json"
5507 ))
5508 .unwrap();
5509
5510 if let Some(ord_id) = request.get("ordId") {
5511 response["data"][0]
5512 .as_object_mut()
5513 .unwrap()
5514 .remove("algoId");
5515 response["data"][0]["ordId"] = ord_id.clone();
5516 } else {
5517 let template = response["data"][0].clone();
5518 response["data"] = request
5519 .as_array()
5520 .unwrap()
5521 .iter()
5522 .map(|item| {
5523 let mut result = template.clone();
5524 result["algoId"] = item["algoId"].clone();
5525 result
5526 })
5527 .collect();
5528 }
5529
5530 tx.send((uri.path().to_string(), request)).unwrap();
5531 Json(response)
5532 }
5533 },
5534 ));
5535
5536 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
5537 let base_url = format!("http://{}", listener.local_addr().unwrap());
5538
5539 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
5540 let (mut client, cache) = build_test_exec_client_with_cache();
5541 client.http_client = OKXHttpClient::with_credentials(
5542 Some("test_key".into()),
5543 Some("test_secret".into()),
5544 Some("test_pass".into()),
5545 Some(base_url),
5546 5,
5547 0,
5548 1,
5549 1,
5550 client.config.environment,
5551 None,
5552 )
5553 .unwrap();
5554 let instrument_id = InstrumentId::from(symbol);
5555
5556 let mut expected_ids: Vec<_> = (0..copies)
5557 .flat_map(|copy| {
5558 expected_ids.iter().map(move |id| {
5559 let index: usize = id.parse().unwrap();
5560 (copy * 4 + index).to_string()
5561 })
5562 })
5563 .collect();
5564
5565 for copy in 0..copies {
5566 for (index, side) in [
5567 OrderSide::Buy,
5568 OrderSide::Sell,
5569 OrderSide::Buy,
5570 OrderSide::Sell,
5571 ]
5572 .into_iter()
5573 .enumerate()
5574 {
5575 let index = copy * 4 + index;
5576
5577 let cached_order_type = if mixed_types && index % 4 >= 2 {
5578 OrderType::TrailingStopMarket
5579 } else {
5580 order_type
5581 };
5582
5583 let order = OrderTestBuilder::new(cached_order_type)
5584 .instrument_id(instrument_id)
5585 .client_order_id(ClientOrderId::from(format!("OHTTP{index}")))
5586 .strategy_id(StrategyId::from(if index % 4 < 2 {
5587 "S-001"
5588 } else {
5589 "S-002"
5590 }))
5591 .side(side)
5592 .price(Price::from("100.00"))
5593 .trigger_price(Price::from("101.00"))
5594 .trailing_offset(rust_decimal::Decimal::ONE)
5595 .trailing_offset_type(TrailingOffsetType::Price)
5596 .quantity(Quantity::from("1"))
5597 .build();
5598
5599 let accepted = TestOrderEventStubs::accepted(
5600 &order,
5601 client.core.account_id,
5602 VenueOrderId::from(format!("{index}")),
5603 );
5604 cache
5605 .borrow_mut()
5606 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
5607 .unwrap();
5608 cache.borrow_mut().update_order(&accepted).unwrap();
5609 }
5610 }
5611
5612 client
5613 .cancel_all_orders(CancelAllOrders::new(
5614 TraderId::from("TESTER-001"),
5615 Some(*OKX_CLIENT_ID),
5616 StrategyId::from("S-001"),
5617 instrument_id,
5618 order_side,
5619 UUID4::new(),
5620 UnixNanos::default(),
5621 None,
5622 None,
5623 ))
5624 .unwrap();
5625 wait_until_async(
5626 || async { client.pending_tasks.is_empty() },
5627 Duration::from_secs(5),
5628 )
5629 .await;
5630
5631 let mut expected_sizes = if order_type == OrderType::Limit {
5632 vec![1; expected_ids.len()]
5633 } else if mixed_types {
5634 match (copies, order_side) {
5635 (1, Some(_)) => vec![1, 1],
5636 (1, None) => vec![2, 2],
5637 (5, Some(_)) => vec![5, 5],
5638 (5, None) => vec![10, 10],
5639 (6, Some(_)) => vec![6, 6],
5640 (6, None) => vec![10, 2, 10, 2],
5641 _ => unreachable!(),
5642 }
5643 } else {
5644 match (copies, order_side) {
5645 (1, Some(_)) => vec![2],
5646 (1, None) => vec![4],
5647 (5, Some(_)) => vec![10],
5648 (5, None) => vec![10, 10],
5649 (6, Some(_)) => vec![10, 2],
5650 (6, None) => vec![10, 10, 4],
5651 _ => unreachable!(),
5652 }
5653 };
5654
5655 let mut actual = Vec::new();
5656 let mut batch_sizes = Vec::new();
5657
5658 while let Ok((path, request)) = request_rx.try_recv() {
5659 if order_type == OrderType::Limit {
5660 assert_eq!(path, "/api/v5/sprd/cancel-order");
5661 batch_sizes.push(1);
5662 actual.push(request["ordId"].as_str().unwrap().to_string());
5663 } else {
5664 batch_sizes.push(request.as_array().unwrap().len());
5665 for item in request.as_array().unwrap() {
5666 let index: usize = item["algoId"].as_str().unwrap().parse().unwrap();
5667
5668 let expected_path = if order_type == OrderType::TrailingStopMarket
5669 || (mixed_types && index % 4 >= 2)
5670 {
5671 "/api/v5/trade/cancel-advance-algos"
5672 } else {
5673 "/api/v5/trade/cancel-algos"
5674 };
5675
5676 assert_eq!(path, expected_path);
5677 assert_eq!(item["instId"], "BTC-USDT");
5678 actual.push(item["algoId"].as_str().unwrap().to_string());
5679 }
5680 }
5681 }
5682
5683 server.abort();
5684 actual.sort();
5685 expected_ids.sort();
5686 batch_sizes.sort_unstable();
5687 expected_sizes.sort_unstable();
5688
5689 assert_eq!(actual, expected_ids);
5690 assert_eq!(batch_sizes, expected_sizes);
5691 }
5692
5693 #[rstest]
5694 #[case("BTC-USDT.OKX", false)]
5695 #[case("BTC-USDT.OKX", true)]
5696 #[case("ETH-USD-SWAP_ETH-USD-231229.OKX", false)]
5697 #[tokio::test]
5698 async fn test_cancel_all_orders_empty_cache_sends_nothing(
5699 #[case] instrument_id: &str,
5700 #[case] use_mm_mass_cancel: bool,
5701 ) {
5702 let mut client = build_test_exec_client();
5703 client.config.use_mm_mass_cancel = use_mm_mass_cancel;
5704 client
5705 .cancel_all_orders(CancelAllOrders::new(
5706 TraderId::from("TESTER-001"),
5707 Some(*OKX_CLIENT_ID),
5708 StrategyId::from("S-001"),
5709 InstrumentId::from(instrument_id),
5710 Some(OrderSide::Buy),
5711 UUID4::new(),
5712 UnixNanos::default(),
5713 None,
5714 None,
5715 ))
5716 .unwrap();
5717
5718 assert!(client.pending_tasks.is_empty());
5719 assert!(client.ws_dispatch_state.order_identities.is_empty());
5720 }
5721
5722 #[rstest]
5723 fn test_order_command_route_uses_bound_child_before_engine_trigger_state_updates() {
5724 let client = build_test_exec_client();
5725 let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
5726 let pending_trigger = Some((OrderType::StopLimit, Some(false)));
5727
5728 assert_eq!(
5729 client.order_command_route(instrument_id, pending_trigger, false),
5730 OrderCommandRoute::AlgoHttp
5731 );
5732 assert_eq!(
5733 client.order_command_route(instrument_id, pending_trigger, true),
5734 OrderCommandRoute::RegularWs
5735 );
5736 assert_eq!(
5737 client.order_command_route(
5738 instrument_id,
5739 Some((OrderType::StopLimit, Some(true))),
5740 false
5741 ),
5742 OrderCommandRoute::RegularWs
5743 );
5744 }
5745
5746 #[rstest]
5747 #[case(
5748 OrderType::StopMarket,
5749 OrderSide::Sell,
5750 true,
5751 "UNSUPPORTED_QUOTE_QUANTITY"
5752 )]
5753 #[case(
5754 OrderType::MarketIfTouched,
5755 OrderSide::Sell,
5756 true,
5757 "UNSUPPORTED_QUOTE_QUANTITY"
5758 )]
5759 #[case(
5760 OrderType::TrailingStopMarket,
5761 OrderSide::Sell,
5762 true,
5763 "UNSUPPORTED_QUOTE_QUANTITY"
5764 )]
5765 #[case(
5766 OrderType::StopLimit,
5767 OrderSide::Sell,
5768 true,
5769 "UNSUPPORTED_QUOTE_QUANTITY"
5770 )]
5771 #[case(
5772 OrderType::LimitIfTouched,
5773 OrderSide::Sell,
5774 true,
5775 "UNSUPPORTED_QUOTE_QUANTITY"
5776 )]
5777 #[case(
5778 OrderType::StopLimit,
5779 OrderSide::Buy,
5780 true,
5781 "UNSUPPORTED_QUOTE_QUANTITY"
5782 )]
5783 #[case(
5784 OrderType::LimitIfTouched,
5785 OrderSide::Buy,
5786 true,
5787 "UNSUPPORTED_QUOTE_QUANTITY"
5788 )]
5789 #[case(
5790 OrderType::TrailingStopMarket,
5791 OrderSide::Buy,
5792 false,
5793 "OKX Spot TRAILING_STOP_MARKET algo buys require quote quantity"
5794 )]
5795 #[case(
5796 OrderType::StopMarket,
5797 OrderSide::Buy,
5798 false,
5799 "OKX Spot STOP_MARKET algo orders do not support BID_ASK triggers"
5800 )]
5801 fn test_submit_order_denies_unsupported_spot_algo_terms_before_submission(
5802 #[case] order_type: OrderType,
5803 #[case] side: OrderSide,
5804 #[case] quote_quantity: bool,
5805 #[case] expected_reason: &str,
5806 #[values("cash", "cross", "isolated")] mode: &str,
5807 ) {
5808 let (mut client, cache) = build_test_exec_client_with_cache();
5809 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
5810 client.emitter.set_sender(tx);
5811
5812 let order = OrderTestBuilder::new(order_type)
5813 .instrument_id(InstrumentId::from("ETH-USDT.OKX"))
5814 .client_order_id(ClientOrderId::from("OQUOTEDENIED"))
5815 .side(side)
5816 .quantity(Quantity::from("12.34"))
5817 .quote_quantity(quote_quantity)
5818 .price(Price::from(if side == OrderSide::Buy {
5819 "2350.12"
5820 } else {
5821 "2340.12"
5822 }))
5823 .trigger_price(Price::from("2345.67"))
5824 .trigger_type(if expected_reason.ends_with("triggers") {
5825 TriggerType::BidAsk
5826 } else {
5827 TriggerType::LastPrice
5828 })
5829 .trailing_offset(Decimal::new(25, 0))
5830 .trailing_offset_type(TrailingOffsetType::BasisPoints)
5831 .build();
5832
5833 cache
5834 .borrow_mut()
5835 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
5836 .unwrap();
5837 let mut command = SubmitOrder::from_order(
5838 &order,
5839 client.core.trader_id,
5840 Some(*OKX_CLIENT_ID),
5841 None,
5842 UUID4::new(),
5843 UnixNanos::default(),
5844 );
5845
5846 command.params = Some(Params::new(
5847 [("td_mode".to_string(), Value::String(mode.to_string()))].into(),
5848 ));
5849 let trade_mode = mode.parse::<OKXTradeMode>().unwrap();
5850
5851 client.submit_order(command).unwrap();
5852 let event = rx.try_recv().unwrap();
5853
5854 let ExecutionEvent::Order(OrderEventAny::Denied(denied)) = event else {
5855 panic!("Expected denial, received {event:?}");
5856 };
5857
5858 assert_eq!(denied.client_order_id, order.client_order_id());
5859
5860 let expected = if expected_reason == "UNSUPPORTED_QUOTE_QUANTITY" {
5861 expected_reason.to_string()
5862 } else if expected_reason.ends_with("triggers") {
5863 format!("VALIDATION_FAILED: {expected_reason}")
5864 } else {
5865 format!("VALIDATION_FAILED: {expected_reason} in {trade_mode} mode")
5866 };
5867
5868 assert_eq!(denied.reason.as_str(), expected);
5869 assert_eq!(
5870 rx.try_recv().unwrap_err(),
5871 tokio::sync::mpsc::error::TryRecvError::Empty
5872 );
5873 assert_eq!(
5874 client
5875 .ws_dispatch_state
5876 .order_context(order.client_order_id()),
5877 None
5878 );
5879 assert!(client.ws_dispatch_state.algo_order_states.is_empty());
5880 }
5881
5882 #[rstest]
5883 #[case(OrderType::StopMarket, "slTriggerPx", "newSlTriggerPx")]
5884 #[case(OrderType::MarketIfTouched, "tpTriggerPx", "newTpTriggerPx")]
5885 #[tokio::test]
5886 async fn test_submit_cash_spot_buy_records_conditional_before_stream_update(
5887 #[case] order_type: OrderType,
5888 #[case] trigger_field: &str,
5889 #[case] amend_field: &str,
5890 #[values(false, true)] quote_quantity: bool,
5891 ) {
5892 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
5893 let router = Router::new().route("/api/v5/trade/order-algo", post(
5894 move |Json(request): Json<Value>| {
5895 let tx = request_tx.clone();
5896 async move {
5897 tx.send(request).unwrap();
5898 Json(serde_json::json!({"code": "0", "msg": "", "data": [{
5899 "algoId": "7003", "algoClOrdId": "OSPOTCONDITIONAL", "sCode": "0", "sMsg": ""
5900 }]}))
5901 }
5902 },
5903 ));
5904
5905 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
5906 let base_url = format!("http://{}", listener.local_addr().unwrap());
5907
5908 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
5909 let (mut client, cache) = build_test_exec_client_with_cache();
5910 client.http_client = OKXHttpClient::with_credentials(
5911 Some("test_key".into()),
5912 Some("test_secret".into()),
5913 Some("test_pass".into()),
5914 Some(base_url),
5915 5,
5916 0,
5917 1,
5918 1,
5919 OKXEnvironment::Live,
5920 None,
5921 )
5922 .unwrap();
5923 let client_order_id = ClientOrderId::from("OSPOTCONDITIONAL");
5924 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
5925 client.emitter.set_sender(event_tx);
5926
5927 let quantity = Quantity::from(if quote_quantity {
5928 "12.34500"
5929 } else {
5930 "0.12345"
5931 });
5932
5933 let order = OrderTestBuilder::new(order_type)
5934 .instrument_id(InstrumentId::from("ETH-USDT.OKX"))
5935 .client_order_id(client_order_id)
5936 .side(OrderSide::Buy)
5937 .quantity(quantity)
5938 .quote_quantity(quote_quantity)
5939 .trigger_price(Price::from("2345.67"))
5940 .trigger_type(TriggerType::LastPrice)
5941 .build();
5942 cache
5943 .borrow_mut()
5944 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
5945 .unwrap();
5946 let submit = SubmitOrder::from_order(
5947 &order,
5948 client.core.trader_id,
5949 Some(*OKX_CLIENT_ID),
5950 None,
5951 UUID4::new(),
5952 UnixNanos::default(),
5953 );
5954 client.submit_order(submit).unwrap();
5955 let request = tokio::time::timeout(Duration::from_secs(5), request_rx.recv())
5956 .await
5957 .unwrap()
5958 .unwrap();
5959 let algo_type = client
5960 .ws_dispatch_state
5961 .algo_order_states
5962 .get(&client_order_id)
5963 .unwrap()
5964 .order_type;
5965 let context = client.ws_dispatch_state.order_context(client_order_id);
5966
5967 let amend = ModifyOrder::new(
5968 client.core.trader_id,
5969 Some(*OKX_CLIENT_ID),
5970 order.strategy_id(),
5971 order.instrument_id(),
5972 client_order_id,
5973 Some(VenueOrderId::from("7003")),
5974 None,
5975 None,
5976 Some(Price::from("2456.78")),
5977 UUID4::new(),
5978 UnixNanos::default(),
5979 None,
5980 None,
5981 );
5982 let amendment = algo_amend_request(&amend, Some(algo_type), context).unwrap();
5983 let amendment = serde_json::to_value(amendment).unwrap();
5984 wait_until_async(
5985 || async { client.pending_tasks.all_finished() },
5986 Duration::from_secs(5),
5987 )
5988 .await;
5989 server.abort();
5990
5991 assert_eq!(request["ordType"], "conditional");
5992 assert_eq!(
5993 request["tgtCcy"],
5994 if quote_quantity {
5995 "quote_ccy"
5996 } else {
5997 "base_ccy"
5998 }
5999 );
6000 assert_eq!(request["sz"], quantity.to_string());
6001 assert_eq!(request[trigger_field], "2345.67");
6002 assert_eq!(algo_type, OKXAlgoOrderType::Conditional);
6003 assert_eq!(context.unwrap().is_quote_quantity, quote_quantity);
6004 assert_eq!(amendment[amend_field], "2456.78");
6005 assert_eq!(amendment.get("newTriggerPx"), None);
6006
6007 let mut instrument = currency_pair_ethusdt();
6008 instrument.id = order.instrument_id();
6009 instrument.raw_symbol = Symbol::from("ETH-USDT");
6010 let instruments = AtomicMap::new();
6011 instruments.insert(
6012 Ustr::from("ETH-USDT"),
6013 InstrumentAny::CurrencyPair(instrument),
6014 );
6015 let mut fee_cache = FeeCache::new();
6016 let mut fills = FilledQtyCache::new();
6017 let mut snapshots = AHashMap::new();
6018 let mut parent: Value =
6019 serde_json::from_str(include_str!("../test_data/ws_orders_algo.json")).unwrap();
6020 let msg = &mut parent["data"][0];
6021 for (key, value) in [
6022 ("algoId", Value::from("7003")),
6023 ("algoClOrdId", Value::from(client_order_id.as_str())),
6024 ("instId", Value::from("ETH-USDT")),
6025 ("instType", Value::from("SPOT")),
6026 ("side", Value::from("buy")),
6027 ("posSide", Value::from("net")),
6028 ("tdMode", Value::from("cash")),
6029 ("reduceOnly", Value::from("false")),
6030 ("ordType", request["ordType"].clone()),
6031 ("sz", request["sz"].clone()),
6032 ("tgtCcy", request["tgtCcy"].clone()),
6033 (trigger_field, request[trigger_field].clone()),
6034 ("triggerPx", Value::from("")),
6035 ] {
6036 msg[key] = value;
6037 }
6038
6039 let (price_field, type_field) = if order_type == OrderType::StopMarket {
6040 ("slOrdPx", "slTriggerPxType")
6041 } else {
6042 ("tpOrdPx", "tpTriggerPxType")
6043 };
6044
6045 msg[price_field] = Value::from("-1");
6046 msg[type_field] = Value::from("last");
6047 let pending = OKXWsMessage::AlgoOrders(
6048 serde_json::from_value(serde_json::json!([msg.clone()])).unwrap(),
6049 );
6050 let pending_replay = OKXWsMessage::AlgoOrders(
6051 serde_json::from_value(serde_json::json!([msg.clone()])).unwrap(),
6052 );
6053 msg[trigger_field] = Value::from("2456.78");
6054 msg["uTime"] = Value::from("1737400000050");
6055 let amended = OKXWsMessage::AlgoOrders(
6056 serde_json::from_value(serde_json::json!([msg.clone()])).unwrap(),
6057 );
6058 let amended_replay = OKXWsMessage::AlgoOrders(
6059 serde_json::from_value(serde_json::json!([msg.clone()])).unwrap(),
6060 );
6061 msg["state"] = Value::from("order_placed");
6062 msg["ordId"] = Value::from("7004");
6063 msg["uTime"] = Value::from("1737400000100");
6064 let triggered = OKXWsMessage::AlgoOrders(
6065 serde_json::from_value(serde_json::json!([msg.clone()])).unwrap(),
6066 );
6067 let mut child: Value = serde_json::from_str(include_str!(
6068 "../test_data/ws_orders_algo_child_filled_empty_cl_ord_id.json"
6069 ))
6070 .unwrap();
6071 let msg = &mut child["data"][0];
6072
6073 let filled_qty = if quote_quantity { "0.00500" } else { "0.12345" };
6074
6075 for (key, value) in [
6076 ("algoId", "7003"),
6077 ("algoClOrdId", client_order_id.as_str()),
6078 ("ordId", "7004"),
6079 ("ordType", "market"),
6080 ("tdMode", "cash"),
6081 (
6082 "tgtCcy",
6083 if quote_quantity {
6084 "quote_ccy"
6085 } else {
6086 "base_ccy"
6087 },
6088 ),
6089 (
6090 "sz",
6091 if quote_quantity {
6092 "12.34500"
6093 } else {
6094 "0.12345"
6095 },
6096 ),
6097 ("accFillSz", filled_qty),
6098 ("fillSz", filled_qty),
6099 ("avgPx", "2469.00"),
6100 ("fillPx", "2469.00"),
6101 ("px", ""),
6102 ("uTime", "1737400000200"),
6103 ("cTime", "1737400000000"),
6104 ("fillTime", "1737400000200"),
6105 ] {
6106 msg[key] = Value::from(value);
6107 }
6108
6109 let filled = OKXWsMessage::Orders(serde_json::from_value(child["data"].clone()).unwrap());
6110 let filled_replay =
6111 OKXWsMessage::Orders(serde_json::from_value(child["data"].clone()).unwrap());
6112
6113 for message in [
6114 pending,
6115 pending_replay,
6116 amended,
6117 amended_replay,
6118 triggered,
6119 filled,
6120 filled_replay,
6121 ] {
6122 dispatch_ws_message(
6123 message,
6124 &client.emitter,
6125 &client.ws_dispatch_state,
6126 client.core.account_id,
6127 client.core.account_type,
6128 &instruments,
6129 &mut fee_cache,
6130 &mut fills,
6131 &mut snapshots,
6132 client.clock,
6133 );
6134 }
6135
6136 let events: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()).collect();
6137
6138 let [
6139 ExecutionEvent::Order(OrderEventAny::Submitted(submitted)),
6140 ExecutionEvent::Order(OrderEventAny::Accepted(accepted)),
6141 ExecutionEvent::Order(OrderEventAny::Updated(amended)),
6142 ExecutionEvent::Order(OrderEventAny::Updated(bound)),
6143 ExecutionEvent::Order(OrderEventAny::Filled(filled)),
6144 reports @ ..,
6145 ] = events.as_slice()
6146 else {
6147 panic!(
6148 "Expected submitted, accepted, amended, bound, filled, and optional status, was {events:?}"
6149 );
6150 };
6151
6152 let mut applied = order.clone();
6153
6154 for event in &events {
6155 if let ExecutionEvent::Order(event) = event {
6156 applied.apply(event.clone()).unwrap();
6157 assert_eq!(applied.is_quote_quantity(), quote_quantity);
6158 }
6159 }
6160
6161 assert_eq!(submitted.client_order_id, client_order_id);
6162 assert_eq!(accepted.client_order_id, client_order_id);
6163 assert_eq!(accepted.venue_order_id, VenueOrderId::from("7003"));
6164 assert_eq!(amended.client_order_id, client_order_id);
6165 assert_eq!(amended.venue_order_id, Some(VenueOrderId::from("7003")));
6166 assert_eq!(amended.quantity, quantity);
6167 assert_eq!(amended.price, None);
6168 assert_eq!(amended.trigger_price, Some(Price::from("2456.78")));
6169 assert_eq!(amended.is_quote_quantity, quote_quantity);
6170 assert_eq!(bound.client_order_id, client_order_id);
6171 assert_eq!(bound.venue_order_id, Some(VenueOrderId::from("7004")));
6172 assert_eq!(bound.quantity, quantity);
6173 assert_eq!(bound.price, None);
6174 assert_eq!(bound.trigger_price, Some(Price::from("2456.78")));
6175 assert_eq!(bound.is_quote_quantity, quote_quantity);
6176 assert_eq!(filled.client_order_id, client_order_id);
6177 assert_eq!(filled.venue_order_id, VenueOrderId::from("7004"));
6178 assert_eq!(filled.last_qty, Quantity::from(filled_qty));
6179 assert_eq!(filled.last_px, Price::from("2469.00"));
6180
6181 if quote_quantity {
6182 let [ExecutionEvent::Report(CommonExecutionReport::Order(report))] = reports else {
6183 panic!("Expected one terminal quote-sized status report, was {reports:?}");
6184 };
6185
6186 assert_eq!(report.client_order_id, Some(client_order_id));
6187 assert_eq!(report.venue_order_id, VenueOrderId::from("7004"));
6188 assert_eq!(report.order_status, OrderStatus::Filled);
6189 assert_eq!(report.quantity, Quantity::from(filled_qty));
6190 assert_eq!(report.filled_qty, Quantity::from(filled_qty));
6191 } else {
6192 assert!(reports.is_empty(), "{reports:?}");
6193 assert_eq!(applied.status(), OrderStatus::Filled);
6194 }
6195
6196 assert_eq!(
6197 client.ws_dispatch_state.order_identity(client_order_id),
6198 None
6199 );
6200 }
6201
6202 #[rstest]
6203 #[tokio::test]
6204 async fn test_submit_conditional_order_resolves_routing_context() {
6205 let router = Router::new().route(
6206 "/api/v5/trade/order-algo",
6207 post(|Json(request): Json<Value>| async move {
6208 let data = match request["algoClOrdId"].as_str() {
6209 Some("ORESTBIND001") => serde_json::json!([{
6210 "algoId": "3796251408639365120",
6211 "algoClOrdId": "ORESTBIND001",
6212 "sCode": "0",
6213 "sMsg": "",
6214 }]),
6215 Some("ORESTAMBIGUOUS002") => serde_json::json!([{
6216 "algoId": "",
6217 "algoClOrdId": "ORESTAMBIGUOUS002",
6218 "sCode": "51149",
6219 "sMsg": "Order timed out. Please try again.",
6220 }]),
6221 _ => serde_json::json!([]),
6222 };
6223 Json(serde_json::json!({
6224 "code": "0",
6225 "msg": "",
6226 "data": data,
6227 }))
6228 }),
6229 );
6230 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6231 let base_url_http = format!("http://{}", listener.local_addr().unwrap());
6232 tokio::spawn(async move {
6233 axum::serve(listener, router.into_make_service())
6234 .await
6235 .unwrap();
6236 });
6237
6238 let config = OKXExecutionClientConfig {
6239 api_key: Some("test_key".into()),
6240 api_secret: Some("test_secret".into()),
6241 api_passphrase: Some("test_pass".into()),
6242 base_url_http: Some(base_url_http),
6243 ..OKXExecutionClientConfig::default()
6244 };
6245 let cache = Rc::new(RefCell::new(Cache::default()));
6246 let core = ExecutionClientCore::new(
6247 TraderId::from("TESTER-001"),
6248 ClientId::from("OKX-TEST"),
6249 *OKX_VENUE,
6250 OmsType::Hedging,
6251 config.account_id,
6252 AccountType::Cash,
6253 None,
6254 Rc::clone(&cache),
6255 );
6256 let mut client = OKXExecutionClient::new(core, config).unwrap();
6257 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
6258 client.emitter.set_sender(event_tx);
6259 let client_order_id = ClientOrderId::from("ORESTBIND001");
6260 let order = OrderTestBuilder::new(OrderType::StopLimit)
6261 .client_order_id(client_order_id)
6262 .strategy_id(StrategyId::from("S-REST-BIND-001"))
6263 .instrument_id(InstrumentId::from("BTC-USDT-SWAP.OKX"))
6264 .side(OrderSide::Sell)
6265 .quantity(Quantity::from("0.01"))
6266 .price(Price::from("94900"))
6267 .trigger_price(Price::from("95000"))
6268 .build();
6269 cache
6270 .borrow_mut()
6271 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
6272 .unwrap();
6273 let command = SubmitOrder::from_order(
6274 &order,
6275 TraderId::from("TESTER-001"),
6276 Some(*OKX_CLIENT_ID),
6277 None,
6278 UUID4::new(),
6279 UnixNanos::default(),
6280 );
6281
6282 client.submit_order(command).unwrap();
6283 wait_until_async(
6284 || async {
6285 client
6286 .ws_dispatch_state
6287 .order_venue_binding(client_order_id)
6288 == Some((VenueOrderId::from("3796251408639365120"), false))
6289 },
6290 Duration::from_secs(5),
6291 )
6292 .await;
6293
6294 assert_eq!(
6295 client.ws_dispatch_state.order_identity(client_order_id),
6296 Some(OrderIdentity {
6297 client_order_id,
6298 strategy_id: StrategyId::from("S-REST-BIND-001"),
6299 instrument_id: InstrumentId::from("BTC-USDT-SWAP.OKX"),
6300 order_side: OrderSide::Sell,
6301 order_type: OrderType::StopLimit,
6302 })
6303 );
6304
6305 let ambiguous_client_order_id = ClientOrderId::from("ORESTAMBIGUOUS002");
6306 let ambiguous_order = OrderTestBuilder::new(OrderType::StopLimit)
6307 .client_order_id(ambiguous_client_order_id)
6308 .strategy_id(StrategyId::from("S-REST-BIND-001"))
6309 .instrument_id(InstrumentId::from("BTC-USDT-SWAP.OKX"))
6310 .side(OrderSide::Sell)
6311 .quantity(Quantity::from("0.02"))
6312 .price(Price::from("94800"))
6313 .trigger_price(Price::from("95100"))
6314 .build();
6315 cache
6316 .borrow_mut()
6317 .add_order(ambiguous_order.clone(), None, Some(*OKX_CLIENT_ID), false)
6318 .unwrap();
6319 let ambiguous_command = SubmitOrder::from_order(
6320 &ambiguous_order,
6321 TraderId::from("TESTER-001"),
6322 Some(*OKX_CLIENT_ID),
6323 None,
6324 UUID4::new(),
6325 UnixNanos::default(),
6326 );
6327
6328 client.submit_order(ambiguous_command).unwrap();
6329 wait_until_async(
6330 || async {
6331 client
6332 .ws_dispatch_state
6333 .order_identity(ambiguous_client_order_id)
6334 .is_none()
6335 },
6336 Duration::from_secs(5),
6337 )
6338 .await;
6339
6340 assert_eq!(
6341 client
6342 .ws_dispatch_state
6343 .order_venue_binding(ambiguous_client_order_id),
6344 None
6345 );
6346
6347 wait_until_async(
6348 || async { client.pending_tasks.all_finished() },
6349 Duration::from_secs(5),
6350 )
6351 .await;
6352 terminate_tasks(&client.pending_tasks, "test execution client")
6353 .await
6354 .expect("execution tasks terminated");
6355
6356 let mut events = Vec::new();
6357 while let Ok(event) = event_rx.try_recv() {
6358 events.push(event);
6359 }
6360
6361 for expected_client_order_id in [client_order_id, ambiguous_client_order_id] {
6362 assert_eq!(
6363 events
6364 .iter()
6365 .filter(|event| matches!(
6366 event,
6367 ExecutionEvent::Order(OrderEventAny::Submitted(submitted))
6368 if submitted.client_order_id == expected_client_order_id
6369 ))
6370 .count(),
6371 1,
6372 );
6373 }
6374 assert!(
6375 !events.iter().any(|event| matches!(
6376 event,
6377 ExecutionEvent::Order(OrderEventAny::Rejected(rejected))
6378 if rejected.client_order_id == ambiguous_client_order_id
6379 )),
6380 "ambiguous algo submit failure should not emit OrderRejected: {events:?}",
6381 );
6382 }
6383
6384 #[rstest]
6385 #[case::single(1, false)]
6386 #[case::batch(23, true)]
6387 #[tokio::test]
6388 async fn closed_handler_rejects_every_unsent_cancel(#[case] count: usize, #[case] batch: bool) {
6389 let (mut client, cache) = build_test_exec_client_with_cache();
6390 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
6391 client.emitter.set_sender(event_tx);
6392 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
6393 let strategy_id = StrategyId::from("S-CLOSED-HANDLER");
6394 client
6395 .ws_private
6396 .cache_inst_id_code(Ustr::from("BTC-USDT"), 10459);
6397 let mut cancels = Vec::new();
6398
6399 for index in 0..count {
6400 let client_order_id = ClientOrderId::from(format!("OCLOSED{index}"));
6401 let venue_order_id = VenueOrderId::from(format!("closed-venue-{index}"));
6402 let mut order = OrderTestBuilder::new(OrderType::Limit)
6403 .client_order_id(client_order_id)
6404 .strategy_id(strategy_id)
6405 .instrument_id(instrument_id)
6406 .side(OrderSide::Sell)
6407 .quantity(Quantity::from("0.005"))
6408 .price(Price::from("110000"))
6409 .build();
6410 let accepted =
6411 TestOrderEventStubs::accepted(&order, client.core.account_id, venue_order_id);
6412 order.apply(accepted).unwrap();
6413 cache
6414 .borrow_mut()
6415 .add_order(order, None, Some(*OKX_CLIENT_ID), false)
6416 .unwrap();
6417
6418 if index % 2 == 0 {
6419 client.ws_dispatch_state.bind_order(
6420 client_order_id,
6421 VenueOrderId::from(format!("authoritative-venue-{index}")),
6422 );
6423 }
6424
6425 cancels.push(CancelOrder::new(
6426 TraderId::from("TESTER-001"),
6427 Some(*OKX_CLIENT_ID),
6428 strategy_id,
6429 instrument_id,
6430 client_order_id,
6431 Some(venue_order_id),
6432 UUID4::new(),
6433 UnixNanos::default(),
6434 None,
6435 None,
6436 ));
6437 }
6438
6439 if batch {
6440 let command = BatchCancelOrders::new(
6441 TraderId::from("TESTER-001"),
6442 Some(*OKX_CLIENT_ID),
6443 strategy_id,
6444 instrument_id,
6445 cancels.clone(),
6446 UUID4::new(),
6447 UnixNanos::default(),
6448 None,
6449 None,
6450 );
6451 client.batch_cancel_orders(command).unwrap();
6452 } else {
6453 client.cancel_order(cancels[0].clone()).unwrap();
6454 }
6455
6456 wait_until_async(
6457 || async { client.pending_tasks.all_finished() },
6458 Duration::from_secs(5),
6459 )
6460 .await;
6461 terminate_tasks(&client.pending_tasks, "test execution client")
6462 .await
6463 .unwrap();
6464 let events: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()).collect();
6465
6466 assert_eq!(events.len(), count);
6467
6468 for (index, (event, cancel)) in events.iter().zip(cancels).enumerate() {
6469 let ExecutionEvent::Order(OrderEventAny::CancelRejected(rejected)) = event else {
6470 panic!("Expected cancel rejection, received {event:?}");
6471 };
6472
6473 assert_eq!(rejected.trader_id, client.core.trader_id);
6474 assert_eq!(rejected.strategy_id, strategy_id);
6475 assert_eq!(rejected.instrument_id, instrument_id);
6476 assert_eq!(rejected.client_order_id, cancel.client_order_id);
6477
6478 let expected_venue_order_id = if index % 2 == 0 {
6479 Some(VenueOrderId::from(format!("authoritative-venue-{index}")))
6480 } else {
6481 cancel.venue_order_id
6482 };
6483
6484 assert_eq!(rejected.venue_order_id, expected_venue_order_id);
6485 assert_eq!(rejected.account_id, Some(client.core.account_id));
6486 assert_eq!(
6487 rejected.reason.as_str(),
6488 "Handler not available: channel closed"
6489 );
6490 assert!(!rejected.reconciliation);
6491 assert_eq!(rejected.causation_id, None);
6492 assert_eq!(
6493 cache
6494 .borrow()
6495 .order(&cancel.client_order_id)
6496 .unwrap()
6497 .status(),
6498 OrderStatus::Accepted
6499 );
6500 }
6501
6502 assert!(client.ws_private.pending_cancels.is_empty());
6503 }
6504
6505 #[rstest]
6506 #[tokio::test]
6507 async fn test_local_cancel_validation_failure_does_not_emit_order_cancel_rejected() {
6508 let (mut client, cache) = build_test_exec_client_with_cache();
6509 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
6510 client.emitter.set_sender(event_tx);
6511 let client_order_id = ClientOrderId::from("OLOCALCANCELINVALID001");
6512 let strategy_id = StrategyId::from("S-LOCAL-CANCEL-INVALID-001");
6513 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
6514 let order = OrderTestBuilder::new(OrderType::Limit)
6515 .client_order_id(client_order_id)
6516 .strategy_id(strategy_id)
6517 .instrument_id(instrument_id)
6518 .side(OrderSide::Buy)
6519 .quantity(Quantity::from("1"))
6520 .price(Price::from("100000"))
6521 .build();
6522 cache
6523 .borrow_mut()
6524 .add_order(order, None, Some(*OKX_CLIENT_ID), false)
6525 .expect("cache order");
6526 let command = CancelOrder {
6527 trader_id: TraderId::from("TESTER-001"),
6528 client_id: Some(*OKX_CLIENT_ID),
6529 strategy_id,
6530 instrument_id,
6531 client_order_id,
6532 venue_order_id: Some(VenueOrderId::from("v-1")),
6533 command_id: UUID4::new(),
6534 ts_init: UnixNanos::default(),
6535 params: None,
6536 correlation_id: None,
6537 causation_id: None,
6538 };
6539
6540 client.cancel_order(command).expect("cancel order");
6541 wait_until_async(
6542 || async { client.pending_tasks.all_finished() },
6543 Duration::from_secs(5),
6544 )
6545 .await;
6546 terminate_tasks(&client.pending_tasks, "test execution client")
6547 .await
6548 .expect("execution task terminated");
6549
6550 let events: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()).collect();
6551 assert!(
6552 !events.iter().any(|event| matches!(
6553 event,
6554 ExecutionEvent::Order(OrderEventAny::CancelRejected(rejected))
6555 if rejected.client_order_id == client_order_id
6556 )),
6557 "local cancel validation failure should not emit OrderCancelRejected: {events:?}",
6558 );
6559 }
6560
6561 #[rstest]
6562 fn test_imported_conditional_order_preserves_report_routing() {
6563 let (mut client, cache) = build_test_exec_client_with_cache();
6564 let cid = ClientOrderId::from("OEEADEMOSTOPLIMIT001");
6565 let iid = InstrumentId::from("ETH-USDT.OKX");
6566 let mut instrument = currency_pair_ethusdt();
6567 instrument.id = iid;
6568 instrument.raw_symbol = Symbol::from("ETH-USDT");
6569 let instruments = AtomicMap::new();
6570 instruments.insert(
6571 Ustr::from("ETH-USDT"),
6572 InstrumentAny::CurrencyPair(instrument),
6573 );
6574 let order = OrderTestBuilder::new(OrderType::StopLimit)
6575 .instrument_id(iid)
6576 .client_order_id(cid)
6577 .side(OrderSide::Buy)
6578 .quantity(Quantity::from("0.003"))
6579 .price(Price::from("1886.00"))
6580 .trigger_price(Price::from("1900.00"))
6581 .build();
6582 cache
6583 .borrow_mut()
6584 .add_order(order.clone(), None, Some(client.core.client_id), false)
6585 .unwrap();
6586 client.register_external_order(
6587 cid,
6588 VenueOrderId::from("7003"),
6589 iid,
6590 order.strategy_id(),
6591 UnixNanos::default(),
6592 );
6593 client.ensure_order_identity(cid, order.strategy_id(), iid);
6594 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
6595 client.emitter.set_sender(tx);
6596 let mut fee_cache = FeeCache::new();
6597 let mut fills = FilledQtyCache::new();
6598 let mut snapshots = AHashMap::new();
6599 let mut parent: Value =
6600 serde_json::from_str(include_str!("../test_data/ws_orders_algo.json")).unwrap();
6601 let msg = &mut parent["data"][0];
6602 msg["algoClOrdId"] = serde_json::json!(cid.as_str());
6603 msg["algoId"] = serde_json::json!("7003");
6604 msg["instId"] = serde_json::json!("ETH-USDT");
6605 msg["ordType"] = serde_json::json!("conditional");
6606 msg["state"] = serde_json::json!("live");
6607 let message =
6608 OKXWsMessage::AlgoOrders(serde_json::from_value(parent["data"].clone()).unwrap());
6609 dispatch_ws_message(
6610 message,
6611 &client.emitter,
6612 &client.ws_dispatch_state,
6613 client.core.account_id,
6614 client.core.account_type,
6615 &instruments,
6616 &mut fee_cache,
6617 &mut fills,
6618 &mut snapshots,
6619 client.clock,
6620 );
6621 let event = rx.try_recv().unwrap();
6622
6623 let ExecutionEvent::Report(CommonExecutionReport::Order(report)) = event else {
6624 panic!("expected imported parent report, was {event:?}");
6625 };
6626
6627 assert_eq!(report.client_order_id, Some(cid));
6628 assert_eq!(report.venue_order_id, VenueOrderId::from("7003"));
6629 assert!(rx.try_recv().is_err());
6630 let child: Value = serde_json::from_str(include_str!(
6631 "../test_data/ws_orders_algo_child_filled_empty_cl_ord_id.json"
6632 ))
6633 .unwrap();
6634 let message = OKXWsMessage::Orders(serde_json::from_value(child["data"].clone()).unwrap());
6635 dispatch_ws_message(
6636 message,
6637 &client.emitter,
6638 &client.ws_dispatch_state,
6639 client.core.account_id,
6640 client.core.account_type,
6641 &instruments,
6642 &mut fee_cache,
6643 &mut fills,
6644 &mut snapshots,
6645 client.clock,
6646 );
6647 let event = rx.try_recv().unwrap();
6648
6649 let ExecutionEvent::Report(CommonExecutionReport::Fill(report)) = event else {
6650 panic!("expected imported child fill report, was {event:?}");
6651 };
6652
6653 assert_eq!(report.client_order_id, Some(cid));
6654 assert_eq!(
6655 report.venue_order_id,
6656 VenueOrderId::from("2497956918703120501")
6657 );
6658 assert_eq!(report.last_qty, Quantity::from("0.003"));
6659 assert_eq!(report.last_px, Price::from("1886.00"));
6660 assert!(rx.try_recv().is_err());
6661 assert_eq!(client.ws_dispatch_state.order_identity(cid), None);
6662 }
6663
6664 #[rstest]
6665 fn test_register_external_order_without_cached_order() {
6666 let client = build_test_exec_client();
6667 let cid = ClientOrderId::from("O-MISSING");
6668 client.register_external_order(
6669 cid,
6670 VenueOrderId::from("7002"),
6671 InstrumentId::from("BTC-USDT.OKX"),
6672 StrategyId::from("S-MISSING"),
6673 UnixNanos::default(),
6674 );
6675 assert_eq!(client.ws_dispatch_state.order_identity(cid), None);
6676 assert!(!client.ws_dispatch_state.contains_accepted(&cid));
6677 }
6678
6679 #[rstest]
6680 fn test_register_external_order_tracks_initialized_import() {
6681 let (client, cache) = build_test_exec_client_with_cache();
6682 let order = OrderTestBuilder::new(OrderType::Limit)
6683 .instrument_id(InstrumentId::from("BTC-USDT.OKX"))
6684 .client_order_id(ClientOrderId::from("O-IMPORTED"))
6685 .strategy_id(StrategyId::from("S-IMPORTED"))
6686 .side(OrderSide::Sell)
6687 .quantity(Quantity::from("0.02"))
6688 .price(Price::from("50000.0"))
6689 .build();
6690 let venue_order_id = VenueOrderId::from("7001");
6691 cache
6692 .borrow_mut()
6693 .add_order(order.clone(), None, Some(client.core.client_id), false)
6694 .unwrap();
6695
6696 client.register_external_order(
6697 order.client_order_id(),
6698 venue_order_id,
6699 order.instrument_id(),
6700 order.strategy_id(),
6701 UnixNanos::default(),
6702 );
6703
6704 assert_eq!(
6705 client
6706 .ws_dispatch_state
6707 .order_identity(order.client_order_id()),
6708 Some(OrderIdentity::from(&order))
6709 );
6710 assert!(
6711 client
6712 .ws_dispatch_state
6713 .contains_accepted(&order.client_order_id())
6714 );
6715 assert!(
6716 !client
6717 .ws_dispatch_state
6718 .contains_triggered(&order.client_order_id())
6719 );
6720 assert_eq!(order.status(), OrderStatus::Initialized);
6721 assert_eq!(order.account_id(), None);
6722 assert_eq!(order.venue_order_id(), None);
6723 client
6724 .ws_dispatch_state
6725 .remove_order_tracking(order.client_order_id());
6726 assert_eq!(
6727 client
6728 .ws_dispatch_state
6729 .order_identity(order.client_order_id()),
6730 None
6731 );
6732 }
6733
6734 #[rstest]
6735 fn test_ensure_order_identity_skips_order_without_cached_side() {
6736 let client = build_test_exec_client();
6737 let client_order_id = ClientOrderId::from("O-RESTORED-001");
6738 let strategy_id = StrategyId::from("S-RESTORED-002");
6739 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
6740
6741 client.ensure_order_identity(client_order_id, strategy_id, instrument_id);
6742
6743 assert!(
6744 client
6745 .ws_dispatch_state
6746 .order_identities
6747 .get(&client_order_id)
6748 .is_none()
6749 );
6750 }
6751
6752 #[rstest]
6753 fn test_on_instrument_writes_through_to_client_caches() {
6754 use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
6757
6758 let mut client = build_test_exec_client();
6759 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
6760 let symbol = instrument.symbol().inner();
6761 let raw_symbol = instrument.raw_symbol().inner();
6762
6763 client.on_instrument(instrument.clone());
6764
6765 let private_cache = client.ws_private.instruments_cache_arc();
6766 let business_cache = client.ws_business.instruments_cache_arc();
6767 assert_eq!(
6768 client
6769 .http_client
6770 .get_instrument(&raw_symbol)
6771 .map(|i| i.id()),
6772 Some(instrument.id()),
6773 );
6774 assert_eq!(
6775 private_cache.load().get(&symbol).map(Instrument::id),
6776 Some(instrument.id()),
6777 );
6778 assert_eq!(
6779 business_cache.load().get(&symbol).map(Instrument::id),
6780 Some(instrument.id()),
6781 );
6782 }
6783
6784 #[rstest]
6785 #[case(OrderSide::Buy, false, PositionSide::Long)]
6786 #[case(OrderSide::Sell, false, PositionSide::Short)]
6787 #[case(OrderSide::Buy, true, PositionSide::Short)]
6788 #[case(OrderSide::Sell, true, PositionSide::Long)]
6789 fn test_order_position_side(
6790 #[case] side: OrderSide,
6791 #[case] reduce_only: bool,
6792 #[case] expected: PositionSide,
6793 #[values("ETH-USDT-SWAP.OKX", "ETH-USDT-271231.OKX")] instrument_id: &str,
6794 ) {
6795 let mut client = build_test_exec_client();
6796 let order = OrderTestBuilder::new(OrderType::Limit)
6797 .instrument_id(InstrumentId::from(instrument_id))
6798 .side(side)
6799 .price(Price::from("123.45"))
6800 .quantity(Quantity::from("2"))
6801 .reduce_only(reduce_only)
6802 .build();
6803 assert_eq!(client.order_position_side(&order, None), None);
6804 client.position_mode = OKXPositionMode::LongShortMode;
6805 assert_eq!(client.order_position_side(&order, None), Some(expected));
6806 }
6807
6808 #[rstest]
6809 #[case(OrderSide::Buy, PositionSide::Short)]
6810 #[case(OrderSide::Sell, PositionSide::Long)]
6811 fn test_order_position_side_close_fraction(
6812 #[case] side: OrderSide,
6813 #[case] expected: PositionSide,
6814 ) {
6815 let mut client = build_test_exec_client();
6816 client.position_mode = OKXPositionMode::LongShortMode;
6817 let order = OrderTestBuilder::new(OrderType::StopMarket)
6818 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
6819 .side(side)
6820 .trigger_price(Price::from("123.45"))
6821 .quantity(Quantity::from("2"))
6822 .build();
6823 let params =
6824 Params::new([("close_fraction".to_string(), Value::String("1".to_string()))].into());
6825 assert_eq!(
6826 client.order_position_side(&order, Some(¶ms)),
6827 Some(expected)
6828 );
6829 }
6830
6831 #[rstest]
6832 #[case("ETH-USDT.OKX")]
6833 #[case("ETH-USD-271231-2000-C.OKX")]
6834 fn test_order_position_side_other_products(#[case] instrument_id: &str) {
6835 let mut client = build_test_exec_client();
6836 client.position_mode = OKXPositionMode::LongShortMode;
6837 let order = OrderTestBuilder::new(OrderType::Limit)
6838 .instrument_id(InstrumentId::from(instrument_id))
6839 .side(OrderSide::Sell)
6840 .price(Price::from("123.45"))
6841 .quantity(Quantity::from("2"))
6842 .build();
6843 assert_eq!(client.order_position_side(&order, None), None);
6844 }
6845
6846 #[rstest]
6847 #[case(None)]
6848 #[case(Some("ETH-USDT-SWAP.OKX-LONG"))]
6849 fn test_validate_submit_position_id_accepts_leg(#[case] position_id: Option<&str>) {
6850 assert_eq!(
6851 validate_submit_position_id(
6852 InstrumentId::from("ETH-USDT-SWAP.OKX"),
6853 Some(PositionSide::Long),
6854 position_id.map(PositionId::from),
6855 ),
6856 Ok(()),
6857 );
6858 }
6859
6860 #[rstest]
6861 fn test_validate_submit_position_id_rejects_other_leg() {
6862 let position_id = PositionId::from("ETH-USDT-SWAP.OKX-SHORT");
6863 let result = validate_submit_position_id(
6864 InstrumentId::from("ETH-USDT-SWAP.OKX"),
6865 Some(PositionSide::Long),
6866 Some(position_id),
6867 );
6868 assert_eq!(
6869 result,
6870 Err(OrderDeniedReason::InvalidPositionId {
6871 position_id,
6872 detail: "expected OKX venue position ID ETH-USDT-SWAP.OKX-LONG".to_string(),
6873 })
6874 );
6875 assert_eq!(
6876 validate_submit_position_id(
6877 InstrumentId::from("ETH-USDT-SWAP.OKX"),
6878 None,
6879 Some(position_id),
6880 ),
6881 Ok(())
6882 );
6883 }
6884
6885 #[rstest]
6886 #[case(None, OKXTradeMode::Cross, None)]
6887 #[case(Some(PositionSide::Long), OKXTradeMode::Isolated, None)]
6888 #[case(
6889 Some(PositionSide::Short),
6890 OKXTradeMode::Cross,
6891 Some(OKXMarginMode::Cross)
6892 )]
6893 fn test_validate_trade_mode_accepts_configured_or_unrestricted(
6894 #[case] position_side: Option<PositionSide>,
6895 #[case] trade_mode: OKXTradeMode,
6896 #[case] margin_mode: Option<OKXMarginMode>,
6897 ) {
6898 let mut client = build_test_exec_client();
6899 client.config.margin_mode = margin_mode;
6900 assert_eq!(
6901 client.validate_trade_mode(
6902 InstrumentId::from("BTC-USDT-SWAP.OKX"),
6903 position_side,
6904 trade_mode
6905 ),
6906 Ok(())
6907 );
6908 }
6909
6910 #[rstest]
6911 #[case(None)]
6912 #[case(Some(OKXMarginMode::None))]
6913 #[case(Some(OKXMarginMode::Isolated))]
6914 #[tokio::test]
6915 async fn test_update_position_mode_on_reconnect(#[case] margin_mode: Option<OKXMarginMode>) {
6916 let response = Arc::new(Mutex::new(
6917 serde_json::from_str::<Value>(include_str!(
6918 "../test_data/http_get_account_configuration.json"
6919 ))
6920 .unwrap(),
6921 ));
6922
6923 let positions = Arc::new(Mutex::new(
6924 serde_json::json!({"code":"0", "data":[], "msg":""}),
6925 ));
6926 let positions_response = Arc::clone(&positions);
6927 let handler_response = Arc::clone(&response);
6928 let router = Router::new()
6929 .route(
6930 "/api/v5/account/config",
6931 get(move || {
6932 let response = Arc::clone(&handler_response);
6933 async move { Json(response.lock().unwrap().clone()) }
6934 }),
6935 )
6936 .route(
6937 "/api/v5/account/positions",
6938 get(move || {
6939 let response = Arc::clone(&positions_response);
6940 async move { Json(response.lock().unwrap().clone()) }
6941 }),
6942 )
6943 .route(
6944 "/api/v5/account/balance",
6945 get(|| async { Json(serde_json::json!({"code":"0", "data":[], "msg":""})) }),
6946 );
6947
6948 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
6949 let address = listener.local_addr().unwrap();
6950 let server = tokio::spawn(async move {
6951 axum::serve(listener, router).await.unwrap();
6952 });
6953
6954 let mut client = build_test_exec_client();
6955 client.config.instrument_types = vec![OKXInstrumentType::Swap];
6956 client.config.margin_mode = margin_mode;
6957 client.http_client = OKXHttpClient::with_credentials(
6958 Some("key".into()),
6959 Some("secret".into()),
6960 Some("pass".into()),
6961 Some(format!("http://{address}")),
6962 5,
6963 0,
6964 1,
6965 1,
6966 OKXEnvironment::Demo,
6967 None,
6968 )
6969 .unwrap();
6970 response.lock().unwrap()["data"][0]["posMode"] = Value::String("long_short_mode".into());
6971 client.update_position_mode().await.unwrap();
6972 assert_eq!(client.position_mode, OKXPositionMode::LongShortMode);
6973 *positions.lock().unwrap() = serde_json::from_str(include_str!(
6974 "../test_data/http_get_positions_long_short.json"
6975 ))
6976 .unwrap();
6977 client.update_position_mode().await.unwrap();
6978 client.config.margin_mode = Some(OKXMarginMode::Cross);
6979 assert_eq!(
6980 client.update_position_mode().await.unwrap_err().to_string(),
6981 "OKX long/short position BTC-USDT-SWAP uses margin mode Isolated, expected Cross; mixed margin modes are not supported"
6982 );
6983 let instruments: Value =
6984 serde_json::from_str(include_str!("../test_data/http_get_instruments_swap.json"))
6985 .unwrap();
6986 let instrument: OKXInstrument = serde_json::from_value(
6987 instruments["data"]
6988 .as_array()
6989 .unwrap()
6990 .iter()
6991 .find(|row| row["instId"] == "BTC-USDT-SWAP")
6992 .unwrap()
6993 .clone(),
6994 )
6995 .unwrap();
6996 client.on_instrument(
6997 parse_instrument_any(&instrument, None, None, UnixNanos::default())
6998 .unwrap()
6999 .unwrap(),
7000 );
7001 let command = GeneratePositionStatusReportsBuilder::default()
7002 .ts_init(UnixNanos::from(98765))
7003 .build()
7004 .unwrap();
7005 assert_eq!(
7006 client
7007 .collect_position_status_reports(&command)
7008 .await
7009 .unwrap_err()
7010 .to_string(),
7011 "OKX long/short position BTC-USDT-SWAP uses margin mode Isolated, expected Cross; mixed margin modes are not supported"
7012 );
7013 client.config.margin_mode = margin_mode;
7014 let (reports, complete) = client
7015 .collect_position_status_reports(&command)
7016 .await
7017 .unwrap();
7018 assert!(complete);
7019 assert_eq!(reports.len(), 2);
7020 assert_eq!(reports[0].quantity, Quantity::from("5.00"));
7021 assert_eq!(
7022 reports[0].venue_position_id,
7023 Some(PositionId::from("BTC-USDT-SWAP.OKX-LONG"))
7024 );
7025 assert_eq!(reports[1].quantity, Quantity::from("2.00"));
7026 assert_eq!(
7027 reports[1].venue_position_id,
7028 Some(PositionId::from("BTC-USDT-SWAP.OKX-SHORT"))
7029 );
7030 client.config.margin_mode = Some(OKXMarginMode::Cross);
7031
7032 for position in positions.lock().unwrap()["data"].as_array_mut().unwrap() {
7033 position["pos"] = Value::String("0".to_string());
7034 }
7035
7036 client.update_position_mode().await.unwrap();
7037 client.config.margin_mode = Some(OKXMarginMode::Isolated);
7038 client.update_position_mode().await.unwrap();
7039 response.lock().unwrap()["data"][0]["posMode"] = Value::String("net_mode".into());
7040 client.update_position_mode().await.unwrap();
7041 assert_eq!(client.position_mode, OKXPositionMode::NetMode);
7042 response.lock().unwrap()["data"] = Value::Array(Vec::new());
7043 assert_eq!(
7044 client.update_position_mode().await.unwrap_err().to_string(),
7045 "OKX account configuration response is empty"
7046 );
7047 assert_eq!(
7048 client.connect().await.unwrap_err().to_string(),
7049 "OKX account configuration response is empty"
7050 );
7051 client.config.instrument_types = vec![OKXInstrumentType::Spot, OKXInstrumentType::Option];
7052 client.update_position_mode().await.unwrap();
7053 server.abort();
7054 }
7055
7056 #[rstest]
7057 #[case(false)]
7058 #[case(true)]
7059 #[tokio::test]
7060 async fn test_submit_spread_preserves_quote_quantity_handling(#[case] quote_quantity: bool) {
7061 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
7062 let router = Router::new().route(
7063 "/api/v5/sprd/order",
7064 post(move |Json(request): Json<Value>| {
7065 let tx = request_tx.clone();
7066 async move {
7067 tx.send(request).unwrap();
7068 let order: Value = serde_json::from_str(include_str!(
7069 "../test_data/http_place_order_response.json"
7070 ))
7071 .unwrap();
7072 Json(serde_json::json!({"code":"0", "msg":"", "data":[order]}))
7073 }
7074 }),
7075 );
7076
7077 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
7078 let address = listener.local_addr().unwrap();
7079 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
7080 let (mut client, cache) = build_test_exec_client_with_cache();
7081 client.http_client = OKXHttpClient::with_credentials(
7082 Some("key".into()),
7083 Some("secret".into()),
7084 Some("pass".into()),
7085 Some(format!("http://{address}")),
7086 5,
7087 0,
7088 1,
7089 1,
7090 OKXEnvironment::Demo,
7091 None,
7092 )
7093 .unwrap();
7094 let (event_tx, _event_rx) = tokio::sync::mpsc::unbounded_channel();
7095 client.emitter.set_sender(event_tx);
7096 let order = OrderTestBuilder::new(OrderType::Limit)
7097 .instrument_id(InstrumentId::from("BCH-USDT_BCH-USDT-SWAP.OKX"))
7098 .client_order_id(ClientOrderId::from("OSPREADQUOTE"))
7099 .side(OrderSide::Sell)
7100 .quantity(Quantity::from("3"))
7101 .price(Price::from("12.34"))
7102 .quote_quantity(quote_quantity)
7103 .build();
7104 cache
7105 .borrow_mut()
7106 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
7107 .unwrap();
7108 client
7109 .submit_order(SubmitOrder::from_order(
7110 &order,
7111 client.core.trader_id,
7112 Some(*OKX_CLIENT_ID),
7113 None,
7114 UUID4::new(),
7115 UnixNanos::default(),
7116 ))
7117 .unwrap();
7118 let request = tokio::time::timeout(Duration::from_secs(5), request_rx.recv())
7119 .await
7120 .unwrap()
7121 .unwrap();
7122 client.disconnect().await.unwrap();
7123 server.abort();
7124
7125 assert_eq!(
7126 request,
7127 serde_json::json!({
7128 "sprdId":"BCH-USDT_BCH-USDT-SWAP",
7129 "clOrdId":"OSPREADQUOTE",
7130 "tag":crate::common::consts::OKX_NAUTILUS_BROKER_ID,
7131 "side":"sell",
7132 "ordType":"limit",
7133 "sz":"3",
7134 "px":"12.34"
7135 })
7136 );
7137 }
7138
7139 fn swap_instrument() -> InstrumentAny {
7140 let response: Value =
7141 serde_json::from_str(include_str!("../test_data/http_get_instruments_swap.json"))
7142 .unwrap();
7143 let model: OKXInstrument = serde_json::from_value(response["data"][0].clone()).unwrap();
7144 parse_instrument_any(&model, None, None, UnixNanos::default())
7145 .unwrap()
7146 .unwrap()
7147 }
7148
7149 #[rstest]
7150 #[case(OrderSide::Buy, false, "long")]
7151 #[case(OrderSide::Sell, false, "short")]
7152 #[case(OrderSide::Buy, true, "short")]
7153 #[case(OrderSide::Sell, true, "long")]
7154 #[tokio::test]
7155 async fn test_submit_order_position_side_wire(
7156 #[case] side: OrderSide,
7157 #[case] reduce_only: bool,
7158 #[case] expected_side: &str,
7159 #[values("ws", "batch", "http", "algo", "close_fraction")] transport: &str,
7160 ) {
7161 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
7162 let ws_tx = request_tx.clone();
7163 let router = Router::new().route("/ws", get(move |ws: WebSocketUpgrade| {
7164 let tx = ws_tx.clone();
7165 async move { ws.on_upgrade(move |mut socket| async move {
7166 while let Some(Ok(Message::Text(text))) = socket.next().await {
7167 let request: Value = serde_json::from_str(&text).unwrap();
7168 if request["op"] == "login" {
7169 let response = serde_json::json!({"event":"login", "code":"0", "msg":"", "connId":"test"});
7170 socket.send(Message::Text(response.to_string().into())).await.unwrap();
7171 } else {
7172 tx.send(request).unwrap();
7173 }
7174 }
7175 }) }
7176 })).route("/api/v5/trade/order", post({
7177 let tx = request_tx.clone();
7178 move |Json(request): Json<Value>| {
7179 let tx = tx.clone();
7180 async move {
7181 tx.send(request).unwrap();
7182 Json(serde_json::from_str::<Value>(include_str!("../test_data/http_place_order_response.json")).unwrap())
7183 }
7184 }
7185 })).route("/api/v5/trade/order-algo", post(move |Json(request): Json<Value>| {
7186 let tx = request_tx.clone();
7187 async move {
7188 tx.send(request).unwrap();
7189 Json(serde_json::from_str::<Value>(include_str!("../test_data/http_place_algo_order_response.json")).unwrap())
7190 }
7191 }));
7192
7193 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
7194 let address = listener.local_addr().unwrap();
7195 let server = tokio::spawn(async move {
7196 axum::serve(listener, router).await.unwrap();
7197 });
7198
7199 let (mut client, cache) = build_test_exec_client_with_cache();
7200 client.position_mode = OKXPositionMode::LongShortMode;
7201 client.trade_mode = OKXTradeMode::Cross;
7202 client.http_client = OKXHttpClient::with_credentials(
7203 Some("key".into()),
7204 Some("secret".into()),
7205 Some("pass".into()),
7206 Some(format!("http://{address}")),
7207 5,
7208 0,
7209 1,
7210 1,
7211 OKXEnvironment::Demo,
7212 None,
7213 )
7214 .unwrap();
7215 client.ws_private = OKXWebSocketClient::with_credentials(
7216 Some(format!("ws://{address}/ws")),
7217 Some("key".into()),
7218 Some("secret".into()),
7219 Some("pass".into()),
7220 Some(client.core.account_id),
7221 None,
7222 None,
7223 client.config.transport_backend,
7224 None,
7225 )
7226 .unwrap();
7227 let instrument = swap_instrument();
7228 client.on_instrument(instrument.clone());
7229 client
7230 .ws_private
7231 .cache_inst_id_codes([(instrument.raw_symbol().inner(), 10458)]);
7232
7233 if transport == "ws" || transport == "batch" {
7234 client.ws_private.connect().await.unwrap();
7235 client.ws_private.wait_until_active(5.0).await.unwrap();
7236 }
7237
7238 let (event_tx, _event_rx) = tokio::sync::mpsc::unbounded_channel();
7239 client.emitter.set_sender(event_tx);
7240
7241 let order =
7242 OrderTestBuilder::new(if transport == "algo" || transport == "close_fraction" {
7243 OrderType::StopMarket
7244 } else {
7245 OrderType::Limit
7246 })
7247 .instrument_id(instrument.id())
7248 .client_order_id(ClientOrderId::from("OHEDGEWIRE"))
7249 .side(side)
7250 .quantity(Quantity::from("2"))
7251 .price(Price::from("123.45"))
7252 .trigger_price(Price::from("120.00"))
7253 .reduce_only(reduce_only)
7254 .build();
7255
7256 cache
7257 .borrow_mut()
7258 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
7259 .unwrap();
7260 let mut command = SubmitOrder::from_order(
7261 &order,
7262 client.core.trader_id,
7263 Some(*OKX_CLIENT_ID),
7264 None,
7265 UUID4::new(),
7266 UnixNanos::default(),
7267 );
7268
7269 if transport == "close_fraction" {
7270 let params = Params::new([("close_fraction".into(), Value::String("1".into()))].into());
7271 command.params = Some(params);
7272 }
7273
7274 match transport {
7275 "batch" => {
7276 let order_list = OrderList::new(
7277 OrderListId::from("OL-HEDGE"),
7278 instrument.id(),
7279 order.strategy_id(),
7280 vec![order.client_order_id()],
7281 UnixNanos::default(),
7282 );
7283 client
7284 .submit_order_list(SubmitOrderList::new(
7285 client.core.trader_id,
7286 Some(*OKX_CLIENT_ID),
7287 order.strategy_id(),
7288 order_list,
7289 vec![OrderInitialized::from(&order)],
7290 None,
7291 None,
7292 None,
7293 UUID4::new(),
7294 UnixNanos::default(),
7295 None,
7296 ))
7297 .unwrap();
7298 }
7299 "http" => client
7300 .submit_order_http(&command, client.order_position_side(&order, None))
7301 .unwrap(),
7302 _ => client.submit_order(command).unwrap(),
7303 }
7304
7305 let request = tokio::time::timeout(Duration::from_secs(5), request_rx.recv())
7306 .await
7307 .unwrap()
7308 .unwrap();
7309
7310 let body = if transport == "ws" || transport == "batch" {
7311 &request["args"][0]
7312 } else {
7313 &request
7314 };
7315
7316 let expected_side = if transport == "close_fraction" {
7317 if side == OrderSide::Buy {
7318 "short"
7319 } else {
7320 "long"
7321 }
7322 } else {
7323 expected_side
7324 };
7325
7326 assert_eq!(body["posSide"], expected_side);
7327 assert_eq!(
7328 body["side"],
7329 if side == OrderSide::Buy {
7330 "buy"
7331 } else {
7332 "sell"
7333 }
7334 );
7335 assert_eq!(body.get("reduceOnly"), None);
7336
7337 if transport == "close_fraction" {
7338 assert_eq!(body.get("sz"), None);
7339 assert_eq!(body["closeFraction"], "1");
7340 } else {
7341 assert_eq!(body["sz"], "2");
7342 }
7343
7344 client.disconnect().await.unwrap();
7345 server.abort();
7346 }
7347
7348 #[rstest]
7349 #[case(OrderSide::Buy, OKXPositionSide::Long)]
7350 #[case(OrderSide::Sell, OKXPositionSide::Short)]
7351 fn test_missing_hedge_leg_reports_flat(
7352 #[case] side: OrderSide,
7353 #[case] leg: OKXPositionSide,
7354 #[values(false, true)] other_leg_present: bool,
7355 #[values(OKXInstrumentType::Swap, OKXInstrumentType::Futures)]
7356 configured_type: OKXInstrumentType,
7357 ) {
7358 let (mut client, cache) = build_test_exec_client_with_cache();
7359 client.position_mode = OKXPositionMode::LongShortMode;
7360 client.config.instrument_types = vec![configured_type];
7361 let instrument = swap_instrument();
7362 let position_id = venue_position_id(instrument.id(), leg).unwrap();
7363 let order = OrderTestBuilder::new(OrderType::Market)
7364 .instrument_id(instrument.id())
7365 .side(side)
7366 .quantity(Quantity::from("2"))
7367 .build();
7368 let fill = TestOrderEventStubs::filled(
7369 &order,
7370 &instrument,
7371 Some(TradeId::from("T-HEDGE")),
7372 Some(position_id),
7373 Some(Price::from("123.4")),
7374 Some(Quantity::from("2")),
7375 Some(LiquiditySide::Taker),
7376 Some(Money::from("0.01 USDT")),
7377 None,
7378 Some(client.core.account_id),
7379 );
7380 let position = Position::new(&instrument, fill.into());
7381 cache
7382 .borrow_mut()
7383 .add_position(&position, OmsType::Hedging)
7384 .unwrap();
7385 let ts_init = UnixNanos::from(98765);
7386
7387 let other_side = if leg == OKXPositionSide::Long {
7388 OKXPositionSide::Short
7389 } else {
7390 OKXPositionSide::Long
7391 };
7392
7393 let mut reports = if other_leg_present {
7394 vec![PositionStatusReport::new(
7395 client.core.account_id,
7396 instrument.id(),
7397 other_side.into(),
7398 Quantity::from("3"),
7399 ts_init,
7400 ts_init,
7401 None,
7402 venue_position_id(instrument.id(), other_side),
7403 None,
7404 )]
7405 } else {
7406 Vec::new()
7407 };
7408
7409 let original_reports = reports.clone();
7410 client.add_flat_derivative_position_reports(&[], &[], &mut reports, ts_init);
7411
7412 if configured_type != OKXInstrumentType::Swap {
7413 assert_eq!(reports, original_reports);
7414 return;
7415 }
7416
7417 assert_eq!(reports.len(), 1 + usize::from(other_leg_present));
7418 let report = reports.last().unwrap();
7419 assert_eq!(report.account_id, client.core.account_id);
7420 assert_eq!(report.instrument_id, instrument.id());
7421 assert_eq!(report.venue_position_id, Some(position_id));
7422 assert_eq!(report.position_side, PositionSide::Flat);
7423 assert_eq!(report.quantity, Quantity::zero(position.quantity.precision));
7424 assert_eq!(report.ts_last, ts_init);
7425 assert_eq!(report.ts_init, ts_init);
7426 client.add_flat_derivative_position_reports(&[], &[], &mut reports, ts_init);
7427 assert_eq!(reports.len(), 1 + usize::from(other_leg_present));
7428 }
7429
7430 #[rstest]
7431 #[case(OKXPositionSide::Long, OrderSide::Sell)]
7432 #[case(OKXPositionSide::Short, OrderSide::Buy)]
7433 fn test_missing_hedge_history_leg_reports_flat(
7434 #[case] leg: OKXPositionSide,
7435 #[case] closing_side: OrderSide,
7436 #[values("order", "fill", "both")] source: &str,
7437 #[values(false, true)] other_leg_present: bool,
7438 ) {
7439 let mut client = build_test_exec_client();
7440 client.position_mode = OKXPositionMode::LongShortMode;
7441 client.config.instrument_types = vec![OKXInstrumentType::Swap];
7442 let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
7443 let position_id = venue_position_id(instrument_id, leg).unwrap();
7444 let ts_init = UnixNanos::from(98765);
7445 let mut order = make_query_order_report(Some("O-CLOSE"), "V-CLOSE");
7446 order.instrument_id = instrument_id;
7447 order.order_side = closing_side.into();
7448 order.order_status = OrderStatus::Filled;
7449 order.quantity = Quantity::from("2.00");
7450 order.filled_qty = Quantity::from("2.00");
7451 order.venue_position_id = Some(position_id);
7452
7453 let fill = FillReport::new(
7454 client.core.account_id,
7455 instrument_id,
7456 VenueOrderId::from("V-CLOSE"),
7457 TradeId::from("T-CLOSE"),
7458 closing_side,
7459 Quantity::from("2.00"),
7460 Price::from("123.4"),
7461 Money::from("0.01 USDT"),
7462 LiquiditySide::Taker,
7463 Some(ClientOrderId::from("O-CLOSE")),
7464 Some(position_id),
7465 ts_init,
7466 ts_init,
7467 None,
7468 );
7469
7470 let orders = if source == "fill" {
7471 vec![]
7472 } else {
7473 vec![order]
7474 };
7475
7476 let fills = if source == "order" {
7477 vec![]
7478 } else {
7479 vec![fill]
7480 };
7481
7482 let other_leg = if leg == OKXPositionSide::Long {
7483 OKXPositionSide::Short
7484 } else {
7485 OKXPositionSide::Long
7486 };
7487
7488 let mut reports = if other_leg_present {
7489 vec![PositionStatusReport::new(
7490 client.core.account_id,
7491 instrument_id,
7492 other_leg.into(),
7493 Quantity::from("3.50"),
7494 ts_init,
7495 ts_init,
7496 None,
7497 venue_position_id(instrument_id, other_leg),
7498 None,
7499 )]
7500 } else {
7501 vec![]
7502 };
7503
7504 client.add_flat_derivative_position_reports(&orders, &fills, &mut reports, ts_init);
7505
7506 assert_eq!(reports.len(), 1 + usize::from(other_leg_present));
7507 let report = reports.last().unwrap();
7508 assert_eq!(report.account_id, client.core.account_id);
7509 assert_eq!(report.instrument_id, instrument_id);
7510 assert_eq!(report.venue_position_id, Some(position_id));
7511 assert_eq!(report.position_side, PositionSide::Flat);
7512 assert_eq!(report.quantity, Quantity::zero(2));
7513 assert_eq!(report.quantity.precision, 2);
7514 assert_eq!(report.ts_last, ts_init);
7515 assert_eq!(report.ts_init, ts_init);
7516 assert_eq!(report.avg_px_open, None);
7517
7518 if other_leg_present {
7519 assert_eq!(
7520 reports[0].venue_position_id,
7521 venue_position_id(instrument_id, other_leg)
7522 );
7523 assert_eq!(reports[0].quantity, Quantity::from("3.50"));
7524 assert_eq!(reports[0].position_side, PositionSide::from(other_leg));
7525 }
7526
7527 client.add_flat_derivative_position_reports(&orders, &fills, &mut reports, ts_init);
7528 assert_eq!(reports.len(), 1 + usize::from(other_leg_present));
7529 }
7530
7531 #[rstest]
7532 #[case(false)]
7533 #[case(true)]
7534 fn test_submit_order_denies_hedge_conflict(#[case] margin_conflict: bool) {
7535 let (mut client, cache) = build_test_exec_client_with_cache();
7536 client.position_mode = OKXPositionMode::LongShortMode;
7537 client.trade_mode = OKXTradeMode::Cross;
7538 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7539 client.emitter.set_sender(tx);
7540 let order = OrderTestBuilder::new(OrderType::Limit)
7541 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
7542 .client_order_id(ClientOrderId::from("ODENYHEDGE"))
7543 .side(OrderSide::Sell)
7544 .price(Price::from("123.45"))
7545 .quantity(Quantity::from("2"))
7546 .build();
7547 cache
7548 .borrow_mut()
7549 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
7550 .unwrap();
7551 let mut command = SubmitOrder::from_order(
7552 &order,
7553 client.core.trader_id,
7554 Some(*OKX_CLIENT_ID),
7555 Some(PositionId::from("ETH-USDT-SWAP.OKX-LONG")),
7556 UUID4::new(),
7557 UnixNanos::default(),
7558 );
7559
7560 if margin_conflict {
7561 command.position_id = None;
7562 let params =
7563 Params::new([("td_mode".to_string(), Value::String("cross".to_string()))].into());
7564 command.params = Some(params);
7565 }
7566
7567 client.submit_order(command).unwrap();
7568 let event = rx.try_recv().unwrap();
7569
7570 let ExecutionEvent::Order(OrderEventAny::Denied(denied)) = event else {
7571 panic!("Expected denial, received {event:?}");
7572 };
7573
7574 assert_eq!(denied.client_order_id, order.client_order_id());
7575 assert_eq!(
7576 denied.reason.as_str(),
7577 if margin_conflict {
7578 "VALIDATION_FAILED: OKX long/short mode requires configured trade mode Isolated, received Cross; mixed margin modes are not supported"
7579 } else {
7580 "INVALID_POSITION_ID: ETH-USDT-SWAP.OKX-LONG; expected OKX venue position ID ETH-USDT-SWAP.OKX-SHORT"
7581 }
7582 );
7583 assert_eq!(
7584 rx.try_recv().unwrap_err(),
7585 tokio::sync::mpsc::error::TryRecvError::Empty
7586 );
7587 }
7588
7589 #[rstest]
7590 #[case(false)]
7591 #[case(true)]
7592 fn test_submit_order_list_denies_hedge_conflict_atomically(#[case] margin_conflict: bool) {
7593 let (mut client, cache) = build_test_exec_client_with_cache();
7594 client.position_mode = OKXPositionMode::LongShortMode;
7595 client.trade_mode = OKXTradeMode::Cross;
7596 let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
7597 client.emitter.set_sender(tx);
7598 let instrument_id = InstrumentId::from("ETH-USDT-SWAP.OKX");
7599
7600 let orders: Vec<_> = [OrderSide::Buy, OrderSide::Sell]
7601 .into_iter()
7602 .enumerate()
7603 .map(|(i, side)| {
7604 OrderTestBuilder::new(OrderType::Limit)
7605 .instrument_id(instrument_id)
7606 .client_order_id(ClientOrderId::from(format!("OHEDGE{i}")))
7607 .side(side)
7608 .price(Price::from("123.45"))
7609 .quantity(Quantity::from("2"))
7610 .build()
7611 })
7612 .collect();
7613
7614 for order in &orders {
7615 cache
7616 .borrow_mut()
7617 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
7618 .unwrap();
7619 }
7620
7621 let order_list = OrderList::new(
7622 OrderListId::from("OL-HEDGE"),
7623 instrument_id,
7624 orders[0].strategy_id(),
7625 orders.iter().map(Order::client_order_id).collect(),
7626 UnixNanos::default(),
7627 );
7628
7629 let mut command = SubmitOrderList::new(
7630 client.core.trader_id,
7631 Some(*OKX_CLIENT_ID),
7632 orders[0].strategy_id(),
7633 order_list,
7634 orders.iter().map(OrderInitialized::from).collect(),
7635 None,
7636 Some(PositionId::from("ETH-USDT-SWAP.OKX-LONG")),
7637 None,
7638 UUID4::new(),
7639 UnixNanos::default(),
7640 None,
7641 );
7642
7643 if margin_conflict {
7644 command.position_id = None;
7645 let params =
7646 Params::new([("td_mode".to_string(), Value::String("cross".to_string()))].into());
7647 command.params = Some(params);
7648 }
7649
7650 client.submit_order_list(command).unwrap();
7651
7652 for order in orders {
7653 let event = rx.try_recv().unwrap();
7654
7655 let ExecutionEvent::Order(OrderEventAny::Denied(denied)) = event else {
7656 panic!("Expected denial, received {event:?}");
7657 };
7658
7659 assert_eq!(denied.client_order_id, order.client_order_id());
7660
7661 let expected = if margin_conflict {
7662 "VALIDATION_FAILED: OKX long/short mode requires configured trade mode Isolated, received Cross; mixed margin modes are not supported"
7663 } else if order.order_side() == OrderSide::Sell {
7664 "INVALID_POSITION_ID: ETH-USDT-SWAP.OKX-LONG; expected OKX venue position ID ETH-USDT-SWAP.OKX-SHORT"
7665 } else {
7666 "ORDER_LIST_DENIED: OL-HEDGE"
7667 };
7668
7669 assert_eq!(denied.reason.as_str(), expected);
7670 }
7671
7672 assert_eq!(
7673 rx.try_recv().unwrap_err(),
7674 tokio::sync::mpsc::error::TryRecvError::Empty
7675 );
7676 }
7677}