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,
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, StrategyId, Venue, VenueOrderId,
61 },
62 instruments::InstrumentAny,
63 orders::Order,
64 reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
65 types::{AccountBalance, MarginBalance, Quantity},
66};
67use rust_decimal::Decimal;
68
69use crate::{
70 common::{
71 consts::{
72 OKX_CONDITIONAL_ORDER_TYPES, OKX_RECONCILIATION_LOOKBACK_DEFAULT_MINS,
73 OKX_RECONCILIATION_LOOKBACK_MAX_MINS, OKX_SUCCESS_CODE, OKX_VENUE,
74 OKX_WS_HEARTBEAT_SECS, okx_reduce_only_wire_value, resolve_instrument_families,
75 validate_okx_client_order_id,
76 },
77 enums::{OKXInstrumentType, OKXMarginMode, OKXTradeMode, is_advance_algo_order},
78 failure::{classify_okx_http_failure, classify_okx_venue_code, classify_okx_ws_failure},
79 parse::{
80 is_okx_spread_symbol, is_order_status_report_more_advanced, nanos_to_datetime,
81 okx_instrument_type_from_symbol,
82 },
83 task::{spawn_task, terminate_tasks},
84 },
85 config::OKXExecutionClientConfig,
86 http::{
87 client::{
88 AlgoOrderReportSweep, FillHistory, OKXHttpClient, OKXPendingAlgoOrderReportsError,
89 ReportInstrumentScope,
90 },
91 models::OKXCancelAlgoOrderRequest,
92 },
93 websocket::{
94 client::OKXWebSocketClient,
95 dispatch::{
96 AlgoCancelContext, WsDispatchState, dispatch_ws_message, emit_algo_cancel_rejections,
97 },
98 messages::OKXWsMessage,
99 parse::{FeeCache, FilledQtyCache, OrderStateSnapshot},
100 },
101};
102
103#[derive(Debug)]
104pub struct OKXExecutionClient {
105 core: ExecutionClientCore,
106 clock: &'static AtomicTime,
107 config: OKXExecutionClientConfig,
108 emitter: ExecutionEventEmitter,
109 http_client: OKXHttpClient,
110 ws_private: OKXWebSocketClient,
111 ws_business: OKXWebSocketClient,
112 trade_mode: OKXTradeMode,
113 ws_dispatch_state: Arc<WsDispatchState>,
114 session_tasks: TaskGroup,
115 pending_tasks: TaskGroup,
116}
117
118impl OKXExecutionClient {
119 pub fn new(
125 core: ExecutionClientCore,
126 config: OKXExecutionClientConfig,
127 ) -> anyhow::Result<Self> {
128 let api_key = config
129 .api_key
130 .as_ref()
131 .map(|value| value.expose_secret().to_owned());
132 let api_secret = config
133 .api_secret
134 .as_ref()
135 .map(|value| value.expose_secret().to_owned());
136 let api_passphrase = config
137 .api_passphrase
138 .as_ref()
139 .map(|value| value.expose_secret().to_owned());
140 let proxy_url = config
141 .proxy_url
142 .as_ref()
143 .map(|value| value.expose_secret().to_owned());
144 let http_client = OKXHttpClient::with_credentials(
145 api_key.clone(),
146 api_secret.clone(),
147 api_passphrase.clone(),
148 Some(config.http_base_url()),
149 config.http_timeout_secs,
150 config.max_retries,
151 config.retry_delay_initial_ms,
152 config.retry_delay_max_ms,
153 config.environment,
154 proxy_url.clone(),
155 )?;
156
157 let account_id = core.account_id;
158
159 let ws_private = OKXWebSocketClient::with_credentials(
160 Some(config.ws_private_url()),
161 api_key.clone(),
162 api_secret.clone(),
163 api_passphrase.clone(),
164 Some(account_id),
165 Some(OKX_WS_HEARTBEAT_SECS),
166 config.auth_timeout_secs,
167 config.transport_backend,
168 proxy_url.clone(),
169 )
170 .context("failed to construct OKX private websocket client")?
171 .with_socket_control(SocketControl::new(
172 core.client_id,
173 Some(*OKX_VENUE),
174 "okx-private-user-streams",
175 ));
176
177 let ws_business = OKXWebSocketClient::with_credentials(
178 Some(config.ws_business_url()),
179 api_key,
180 api_secret,
181 api_passphrase,
182 Some(account_id),
183 Some(OKX_WS_HEARTBEAT_SECS),
184 config.auth_timeout_secs,
185 config.transport_backend,
186 proxy_url,
187 )
188 .context("failed to construct OKX business websocket client")?
189 .with_socket_control(SocketControl::new(
190 core.client_id,
191 Some(*OKX_VENUE),
192 "okx-business-user-streams",
193 ));
194
195 http_client.set_spot_trade_quote_ccy(config.spot_trade_quote_ccy.clone());
196 ws_private.set_spot_trade_quote_ccy(config.spot_trade_quote_ccy.clone());
197 ws_business.set_spot_trade_quote_ccy(config.spot_trade_quote_ccy.clone());
198
199 let trade_mode = Self::derive_default_trade_mode(core.account_type, &config);
200 let clock = get_atomic_clock_realtime();
201 let emitter = ExecutionEventEmitter::new(
202 clock,
203 core.trader_id,
204 core.account_id,
205 core.account_type,
206 None,
207 );
208
209 let ws_dispatch_state = Arc::new(WsDispatchState::with_pending_maps(
210 ws_private.pending_orders.clone(),
211 ws_private.pending_cancels.clone(),
212 ws_private.pending_amends.clone(),
213 ));
214
215 Ok(Self {
216 core,
217 clock,
218 config,
219 emitter,
220 http_client,
221 ws_private,
222 ws_business,
223 trade_mode,
224 ws_dispatch_state,
225 session_tasks: TaskGroup::new(),
226 pending_tasks: TaskGroup::new(),
227 })
228 }
229
230 fn derive_default_trade_mode(
231 account_type: AccountType,
232 config: &OKXExecutionClientConfig,
233 ) -> OKXTradeMode {
234 let is_cross_margin = config.margin_mode == Some(OKXMarginMode::Cross);
235
236 if account_type == AccountType::Cash {
237 if !config.use_spot_margin {
238 return OKXTradeMode::Cash;
239 }
240 return if is_cross_margin {
241 OKXTradeMode::Cross
242 } else {
243 OKXTradeMode::Isolated
244 };
245 }
246
247 if is_cross_margin {
248 OKXTradeMode::Cross
249 } else {
250 OKXTradeMode::Isolated
251 }
252 }
253
254 fn trade_mode_for_order(
255 &self,
256 instrument_id: InstrumentId,
257 params: Option<&Params>,
258 ) -> OKXTradeMode {
259 if let Some(td_mode_str) = get_param_as_string(params, "td_mode") {
260 match td_mode_str.parse::<OKXTradeMode>() {
261 Ok(mode) => return mode,
262 Err(_) => {
263 log::warn!("Invalid td_mode '{td_mode_str}', using derived trade mode");
264 }
265 }
266 }
267
268 derive_trade_mode_for_instrument(
269 instrument_id,
270 self.config.margin_mode,
271 self.config.use_spot_margin,
272 )
273 }
274
275 fn instrument_types(&self) -> Vec<OKXInstrumentType> {
276 if self.config.instrument_types.is_empty() {
277 vec![OKXInstrumentType::Spot]
278 } else {
279 self.config.instrument_types.clone()
280 }
281 }
282
283 fn report_scope<'a>(
284 &'a self,
285 instrument_types: &'a [OKXInstrumentType],
286 ) -> ReportInstrumentScope<'a> {
287 ReportInstrumentScope {
288 instrument_types,
289 load_spreads: self.config.load_spreads,
290 }
291 }
292
293 async fn collect_order_status_reports(
294 &self,
295 cmd: &GenerateOrderStatusReports,
296 require_complete_active_coverage: bool,
297 ) -> anyhow::Result<OrderReportSweep> {
298 let instrument_types = self.instrument_types();
299 let routing_types = order_routing_instrument_types(&instrument_types);
300 let scope = self.report_scope(&routing_types);
301 let start = nanos_to_datetime(cmd.start);
302 let end = nanos_to_datetime(cmd.end);
303 let mut reports = Vec::new();
304 let mut regular_by_venue_order_id = AHashMap::new();
305 let mut ambiguous_triggered_child_ids = AHashSet::new();
306 let mut complete = true;
307
308 if let Some(instrument_id) = cmd.instrument_id {
309 let sweep = self
310 .http_client
311 .request_order_status_reports_scoped(
312 self.core.account_id,
313 None,
314 Some(instrument_id),
315 start,
316 end,
317 false,
318 None,
319 Some(scope),
320 )
321 .await?;
322
323 regular_by_venue_order_id.extend(
324 sweep
325 .reports
326 .iter()
327 .map(|report| (report.venue_order_id, report.clone())),
328 );
329
330 reports.extend(sweep.reports);
331 complete &= sweep.complete;
332
333 if !is_spread_instrument(instrument_id)
334 && supports_algo_orders(okx_instrument_type_from_symbol(
335 instrument_id.symbol.as_str(),
336 ))
337 {
338 match self
339 .http_client
340 .request_algo_order_status_reports_sweep(
341 self.core.account_id,
342 None,
343 Some(instrument_id),
344 None,
345 None,
346 None,
347 None,
348 start,
349 end,
350 require_complete_active_coverage,
351 )
352 .await
353 {
354 Ok(sweep) => {
355 merge_algo_order_status_reports(
356 &mut reports,
357 sweep,
358 &mut ambiguous_triggered_child_ids,
359 &mut complete,
360 );
361 }
362 Err(e)
363 if require_complete_active_coverage
364 && e.downcast_ref::<OKXPendingAlgoOrderReportsError>()
365 .is_some() =>
366 {
367 return Err(e);
368 }
369 Err(e) if is_instrument_cache_miss(&e) => return Err(e),
370 Err(e) => {
371 log::warn!(
372 "Failed to fetch algo order status reports for {instrument_id}: {e}"
373 );
374 complete = false;
375 }
376 }
377 }
378 } else {
379 for inst_type in &routing_types {
380 let sweep = self
381 .http_client
382 .request_order_status_reports_scoped(
383 self.core.account_id,
384 Some(*inst_type),
385 None,
386 start,
387 end,
388 false,
389 None,
390 Some(scope),
391 )
392 .await?;
393
394 regular_by_venue_order_id.extend(
395 sweep
396 .reports
397 .iter()
398 .map(|report| (report.venue_order_id, report.clone())),
399 );
400
401 reports.extend(sweep.reports);
402 complete &= sweep.complete;
403
404 if supports_algo_orders(*inst_type) {
405 match self
406 .http_client
407 .request_algo_order_status_reports_sweep(
408 self.core.account_id,
409 Some(*inst_type),
410 None,
411 None,
412 None,
413 None,
414 None,
415 start,
416 end,
417 require_complete_active_coverage,
418 )
419 .await
420 {
421 Ok(sweep) => {
422 merge_algo_order_status_reports(
423 &mut reports,
424 sweep,
425 &mut ambiguous_triggered_child_ids,
426 &mut complete,
427 );
428 }
429 Err(e)
430 if require_complete_active_coverage
431 && e.downcast_ref::<OKXPendingAlgoOrderReportsError>()
432 .is_some() =>
433 {
434 return Err(e);
435 }
436 Err(e) if is_instrument_cache_miss(&e) => return Err(e),
437 Err(e) => {
438 log::warn!(
439 "Failed to fetch algo order status reports for {inst_type:?}: {e}"
440 );
441 complete = false;
442 }
443 }
444 }
445 }
446
447 if self.config.load_spreads {
448 match self
449 .http_client
450 .request_order_status_reports_scoped(
451 self.core.account_id,
452 None,
453 None,
454 start,
455 end,
456 false,
457 None,
458 Some(scope),
459 )
460 .await
461 {
462 Ok(sweep) => {
463 reports.extend(sweep.reports);
464 complete &= sweep.complete;
465 }
466 Err(e) if is_instrument_cache_miss(&e) => return Err(e),
467 Err(e) => {
468 log::warn!("Failed to fetch spread order status reports: {e}");
469 complete = false;
470 }
471 }
472 }
473 }
474
475 retain_order_status_reports(&mut reports, cmd);
476
477 Ok(OrderReportSweep {
478 reports,
479 complete,
480 ambiguous_triggered_child_ids,
481 regular_by_venue_order_id,
482 })
483 }
484
485 async fn collect_fill_reports(
486 &self,
487 cmd: GenerateFillReports,
488 history: FillHistory,
489 ) -> anyhow::Result<(Vec<FillReport>, bool)> {
490 let instrument_types = self.instrument_types();
491 let routing_types = order_routing_instrument_types(&instrument_types);
492 let scope = self.report_scope(&routing_types);
493 let start_dt = nanos_to_datetime(cmd.start);
494 let end_dt = nanos_to_datetime(cmd.end);
495 let mut reports = Vec::new();
496 let mut complete = true;
497
498 if let Some(instrument_id) = cmd.instrument_id {
499 let sweep = self
500 .http_client
501 .request_fill_reports_scoped(
502 self.core.account_id,
503 None,
504 Some(instrument_id),
505 start_dt,
506 end_dt,
507 None,
508 history,
509 Some(scope),
510 )
511 .await?;
512 reports.extend(sweep.reports);
513 complete &= sweep.complete;
514 } else {
515 for inst_type in &routing_types {
516 let sweep = self
517 .http_client
518 .request_fill_reports_scoped(
519 self.core.account_id,
520 Some(*inst_type),
521 None,
522 start_dt,
523 end_dt,
524 None,
525 history,
526 Some(scope),
527 )
528 .await?;
529 reports.extend(sweep.reports);
530 complete &= sweep.complete;
531 }
532
533 if self.config.load_spreads {
534 let sweep = self
535 .http_client
536 .request_fill_reports_scoped(
537 self.core.account_id,
538 None,
539 None,
540 start_dt,
541 end_dt,
542 None,
543 history,
544 Some(scope),
545 )
546 .await?;
547 reports.extend(sweep.reports);
548 complete &= sweep.complete;
549 }
550 }
551
552 if let Some(venue_order_id) = cmd.venue_order_id {
553 reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
554 }
555
556 Ok((reports, complete))
557 }
558
559 async fn collect_position_status_reports(
560 &self,
561 cmd: &GeneratePositionStatusReports,
562 ) -> anyhow::Result<(Vec<PositionStatusReport>, bool)> {
563 let instrument_types = self.instrument_types();
564 let scope = self.report_scope(&instrument_types);
565 let mut reports = Vec::new();
566 let mut complete = true;
567
568 if let Some(instrument_id) = cmd.instrument_id {
569 if is_spread_instrument(instrument_id) {
570 return Ok((reports, complete));
571 }
572
573 let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
574 if inst_type != OKXInstrumentType::Spot && inst_type != OKXInstrumentType::Margin {
575 let sweep = self
576 .http_client
577 .request_position_status_reports_scoped(
578 self.core.account_id,
579 None,
580 Some(instrument_id),
581 Some(scope),
582 )
583 .await?;
584 reports.extend(sweep.reports);
585 complete &= sweep.complete;
586 }
587 } else {
588 for inst_type in &instrument_types {
589 if *inst_type == OKXInstrumentType::Spot || *inst_type == OKXInstrumentType::Margin
590 {
591 continue;
592 }
593 let sweep = self
594 .http_client
595 .request_position_status_reports_scoped(
596 self.core.account_id,
597 Some(*inst_type),
598 None,
599 Some(scope),
600 )
601 .await?;
602 reports.extend(sweep.reports);
603 complete &= sweep.complete;
604 }
605 }
606
607 let mut margin_reports = self
608 .http_client
609 .request_spot_margin_position_reports(self.core.account_id)
610 .await?;
611
612 if let Some(instrument_id) = cmd.instrument_id {
613 margin_reports.retain(|report| report.instrument_id == instrument_id);
614 }
615
616 reports.append(&mut margin_reports);
617
618 Ok((reports, complete))
619 }
620
621 fn update_account_state(&self) {
622 let http_client = self.http_client.clone();
623 let account_id = self.core.account_id;
624 let account_type = self.core.account_type;
625 let emitter = self.emitter.clone();
626
627 self.spawn_task("query_account", async move {
628 let account_state = http_client
629 .request_account_state(account_id, account_type)
630 .await
631 .context("failed to request OKX account state")?;
632 emitter.send_account_state(account_state);
633 Ok(())
634 });
635 }
636
637 fn is_conditional_order(&self, order_type: OrderType) -> bool {
638 OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
639 }
640
641 fn submit_order_route(
642 &self,
643 instrument_id: InstrumentId,
644 order_type: OrderType,
645 ) -> anyhow::Result<OrderCommandRoute> {
646 if self.is_conditional_order(order_type) {
647 if is_spread_instrument(instrument_id) {
648 anyhow::bail!(
649 "Trigger/conditional orders ({order_type:?}) are not supported for OKX spreads"
650 );
651 }
652
653 let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
654 if inst_type == OKXInstrumentType::Option {
655 anyhow::bail!(
656 "Trigger/conditional orders ({order_type:?}) are not supported for OKX options"
657 );
658 }
659
660 return Ok(OrderCommandRoute::AlgoHttp);
661 }
662
663 if is_spread_instrument(instrument_id) {
664 Ok(OrderCommandRoute::SpreadHttp)
665 } else {
666 Ok(OrderCommandRoute::RegularWs)
667 }
668 }
669
670 fn cancel_order_route(
671 &self,
672 instrument_id: InstrumentId,
673 order_state: Option<(OrderType, Option<bool>)>,
674 has_bound_child: bool,
675 ) -> OrderCommandRoute {
676 if is_spread_instrument(instrument_id) {
677 return OrderCommandRoute::SpreadHttp;
678 }
679
680 if has_bound_child {
681 return OrderCommandRoute::RegularWs;
682 }
683
684 if order_state.is_some_and(|(order_type, is_triggered)| {
685 self.is_conditional_order(order_type) && is_triggered != Some(true)
686 }) {
687 OrderCommandRoute::AlgoHttp
688 } else {
689 OrderCommandRoute::RegularWs
690 }
691 }
692
693 fn cancel_all_orders_route(
694 &self,
695 instrument_id: InstrumentId,
696 order_side: Option<OrderSide>,
697 ) -> CancelAllOrdersRoute {
698 if order_side.is_some() {
699 CancelAllOrdersRoute::Batch
700 } else if is_spread_instrument(instrument_id) {
701 CancelAllOrdersRoute::SpreadHttp
702 } else if self.config.use_mm_mass_cancel {
703 CancelAllOrdersRoute::MassCancelHttp
704 } else {
705 CancelAllOrdersRoute::Batch
706 }
707 }
708
709 fn submit_regular_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
710 let order = {
711 let cache = self.core.cache();
712 cache.try_order_owned(&cmd.client_order_id)?
713 };
714 let ws_private = self.ws_private.clone();
715 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
716
717 let emitter = self.emitter.clone();
718 let clock = self.clock;
719 let trader_id = self.core.trader_id;
720 let context = OrderContext::from(&order);
721
722 self.ws_dispatch_state
723 .order_identities
724 .insert(context.identity.client_order_id, context.identity);
725 let client_order_id = context.identity.client_order_id;
726 let strategy_id = context.identity.strategy_id;
727 let instrument_id = context.identity.instrument_id;
728 let order_side = context.identity.order_side;
729 let order_type = context.identity.order_type;
730 let quantity = context.quantity;
731 let time_in_force = context.time_in_force;
732 let price = context.price;
733 let trigger_price = context.trigger_price;
734 let is_post_only = context.is_post_only;
735 let is_reduce_only = context.is_reduce_only;
736 let is_quote_quantity = context.is_quote_quantity;
737
738 let px_usd = get_param_as_string(cmd.params.as_ref(), "px_usd");
739 let px_vol = get_param_as_string(cmd.params.as_ref(), "px_vol");
740 let outcome = get_param_as_string(cmd.params.as_ref(), "outcome");
741 let slippage_pct = get_param_as_string(cmd.params.as_ref(), "slippage_pct");
742 let rpi = get_param_as_bool(cmd.params.as_ref(), "rpi");
743 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
744 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
745
746 self.spawn_task("submit_order", async move {
747 let result = ws_private
748 .submit_order(
749 trader_id,
750 strategy_id,
751 instrument_id,
752 trade_mode,
753 client_order_id,
754 order_side,
755 order_type,
756 quantity,
757 Some(time_in_force),
758 price,
759 trigger_price,
760 Some(is_post_only),
761 Some(is_reduce_only),
762 Some(is_quote_quantity),
763 None,
764 None,
765 px_usd,
766 px_vol,
767 outcome,
768 slippage_pct,
769 rpi,
770 rpi_taker_access,
771 rpi_px_round,
772 )
773 .await;
774
775 if let Err(e) = result {
776 emit_submit_failure(
777 classify_okx_ws_failure(&e),
778 &emitter,
779 clock,
780 strategy_id,
781 instrument_id,
782 client_order_id,
783 );
784 return Err(anyhow::Error::new(e).context("submit order failed"));
785 }
786
787 Ok(())
788 });
789
790 Ok(())
791 }
792
793 fn submit_order_http(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
794 let order = {
795 let cache = self.core.cache();
796 cache.try_order_owned(&cmd.client_order_id)?
797 };
798 let http_client = self.http_client.clone();
799 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
800
801 let emitter = self.emitter.clone();
802 let clock = self.clock;
803 let context = OrderContext::from(&order);
804
805 self.ws_dispatch_state
806 .order_identities
807 .insert(context.identity.client_order_id, context.identity);
808 let client_order_id = context.identity.client_order_id;
809 let strategy_id = context.identity.strategy_id;
810 let instrument_id = context.identity.instrument_id;
811 let order_side = context.identity.order_side;
812 let order_type = context.identity.order_type;
813 let quantity = context.quantity;
814 let time_in_force = context.time_in_force;
815 let price = context.price;
816 let is_post_only = context.is_post_only;
817 let rpi = get_param_as_bool(cmd.params.as_ref(), "rpi");
818 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
819 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
820
821 self.spawn_task("submit_order_http", async move {
822 let result = http_client
823 .place_order_with_domain_types(
824 instrument_id,
825 trade_mode,
826 client_order_id,
827 order_side,
828 order_type,
829 quantity,
830 Some(time_in_force),
831 price,
832 Some(is_post_only),
833 None,
834 None,
835 None,
836 None,
837 None,
838 None,
839 None,
840 None,
841 rpi,
842 rpi_taker_access,
843 rpi_px_round,
844 )
845 .await;
846
847 if let Err(e) = result {
848 emit_submit_failure(
849 classify_okx_http_failure(&e),
850 &emitter,
851 clock,
852 strategy_id,
853 instrument_id,
854 client_order_id,
855 );
856 return Err(anyhow::Error::new(e).context("submit order failed"));
857 }
858
859 Ok(())
860 });
861
862 Ok(())
863 }
864
865 fn submit_conditional_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
866 let order = {
867 let cache = self.core.cache();
868 cache.try_order_owned(&cmd.client_order_id)?
869 };
870 let http_client = self.http_client.clone();
871 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
872
873 let emitter = self.emitter.clone();
874 let clock = self.clock;
875 let context = OrderContext::from(&order);
876
877 let client_order_id = context.identity.client_order_id;
878 let strategy_id = context.identity.strategy_id;
879 let instrument_id = context.identity.instrument_id;
880 let order_side = context.identity.order_side;
881 let order_type = context.identity.order_type;
882 let quantity = context.quantity;
883 let trigger_type = context.trigger_type;
884 let trigger_price = context.trigger_price;
885 let price = context.price;
886 let is_reduce_only = context.is_reduce_only;
887
888 let trailing_offset = order.trailing_offset();
889 let trailing_offset_type = order.trailing_offset_type();
890 let activation_price = order.activation_price();
891
892 let close_fraction = get_param_as_string(cmd.params.as_ref(), "close_fraction");
893
894 let reduce_only = if close_fraction.is_some() {
895 Some(true)
896 } else {
897 Some(is_reduce_only)
898 };
899
900 let (callback_ratio, callback_spread) = if order_type == OrderType::TrailingStopMarket {
901 let offset = trailing_offset
902 .ok_or_else(|| anyhow::anyhow!("TrailingStopMarket requires trailing_offset"))?;
903 let offset_type = trailing_offset_type.ok_or_else(|| {
904 anyhow::anyhow!("TrailingStopMarket requires trailing_offset_type")
905 })?;
906
907 match offset_type {
908 TrailingOffsetType::BasisPoints => {
909 let ratio = offset / Decimal::from(10000);
911 (Some(ratio.to_string()), None)
912 }
913 TrailingOffsetType::Price => (None, Some(offset.to_string())),
914 _ => {
915 anyhow::bail!("Unsupported trailing_offset_type for OKX: {offset_type:?}");
916 }
917 }
918 } else {
919 (None, None)
920 };
921
922 self.ws_dispatch_state.track_order_context(context);
923 let dispatch_state = Arc::clone(&self.ws_dispatch_state);
924
925 self.spawn_task("submit_algo_order", async move {
926 let result = http_client
927 .place_algo_order_with_domain_types(
928 instrument_id,
929 trade_mode,
930 client_order_id,
931 order_side,
932 order_type,
933 quantity,
934 trigger_price,
935 trigger_type,
936 price,
937 reduce_only,
938 close_fraction,
939 callback_ratio,
940 callback_spread,
941 activation_price,
942 )
943 .await;
944
945 match result {
946 Ok(response) => {
947 dispatch_state.bind_algo_parent(
948 client_order_id,
949 VenueOrderId::new(response.algo_id.as_str()),
950 );
951 }
952 Err(e) => {
953 let failure = classify_okx_http_failure(&e);
954 dispatch_state.resolve_algo_submit_failure(client_order_id, &failure);
955 emit_submit_failure(
956 failure,
957 &emitter,
958 clock,
959 strategy_id,
960 instrument_id,
961 client_order_id,
962 );
963 return Err(anyhow::Error::new(e).context("submit algo order failed"));
964 }
965 }
966
967 Ok(())
968 });
969
970 Ok(())
971 }
972
973 fn cancel_ws_order(&self, cmd: &CancelOrder) {
974 self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
975
976 let ws_private = self.ws_private.clone();
977 let mut command = cmd.clone();
978 command.venue_order_id = self
979 .ws_dispatch_state
980 .order_venue_binding(cmd.client_order_id)
981 .map(|(venue_order_id, _)| venue_order_id)
982 .or(cmd.venue_order_id);
983
984 self.spawn_task("cancel_order", async move {
985 let result = ws_private
986 .cancel_order(
987 command.trader_id,
988 command.strategy_id,
989 command.instrument_id,
990 Some(command.client_order_id),
991 command.venue_order_id,
992 )
993 .await;
994
995 if let Err(e) = result {
996 emit_cancel_failure(
997 classify_okx_ws_failure(&e),
998 None,
999 command.client_order_id,
1000 command.instrument_id,
1001 command.strategy_id,
1002 command.venue_order_id,
1003 );
1004 return Err(anyhow::Error::new(e).context("cancel order failed"));
1005 }
1006
1007 Ok(())
1008 });
1009 }
1010
1011 fn cancel_order_http(&self, cmd: &CancelOrder) {
1012 self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
1013
1014 let http_client = self.http_client.clone();
1015 let command = cmd.clone();
1016 let emitter = self.emitter.clone();
1017 let clock = self.clock;
1018
1019 self.spawn_task("cancel_order_http", async move {
1020 let result = http_client
1021 .cancel_order(
1022 command.instrument_id,
1023 Some(command.client_order_id),
1024 command.venue_order_id,
1025 )
1026 .await;
1027
1028 if let Err(e) = result {
1029 emit_cancel_failure(
1030 classify_okx_http_failure(&e),
1031 Some((&emitter, clock)),
1032 command.client_order_id,
1033 command.instrument_id,
1034 command.strategy_id,
1035 command.venue_order_id,
1036 );
1037 return Err(anyhow::Error::new(e).context("cancel order failed"));
1038 }
1039
1040 Ok(())
1041 });
1042 }
1043
1044 fn cancel_algo_order(&self, cmd: &CancelOrder) {
1045 let http_client = self.http_client.clone();
1046 let command = cmd.clone();
1047 let emitter = self.emitter.clone();
1048 let clock = self.clock;
1049
1050 let cache = self.core.cache();
1051 let is_advance = cache
1052 .order(&cmd.client_order_id)
1053 .is_some_and(|o| is_advance_algo_order(o.order_type()));
1054 drop(cache);
1055
1056 let request = OKXCancelAlgoOrderRequest {
1057 inst_id: cmd.instrument_id.symbol.to_string(),
1058 inst_id_code: None,
1059 algo_id: cmd.venue_order_id.map(|id| id.to_string()),
1060 algo_cl_ord_id: if cmd.venue_order_id.is_none() {
1061 Some(cmd.client_order_id.to_string())
1062 } else {
1063 None
1064 },
1065 };
1066
1067 self.spawn_task("cancel_algo_order", async move {
1068 let responses = if is_advance {
1069 http_client.cancel_advance_algo_orders(vec![request]).await
1070 } else {
1071 http_client.cancel_algo_orders(vec![request]).await
1072 };
1073
1074 match responses {
1075 Err(e) => {
1076 emit_cancel_failure(
1077 classify_okx_http_failure(&e),
1078 Some((&emitter, clock)),
1079 command.client_order_id,
1080 command.instrument_id,
1081 command.strategy_id,
1082 command.venue_order_id,
1083 );
1084 return Err(anyhow::Error::new(e).context("cancel algo order failed"));
1085 }
1086 Ok(resps) => {
1087 if let Some((code, msg)) = resps.first().and_then(|r| {
1088 r.s_code.as_deref().and_then(|code| {
1089 (code != OKX_SUCCESS_CODE)
1090 .then_some((code, r.s_msg.as_deref().unwrap_or("unknown")))
1091 })
1092 }) {
1093 let reason =
1094 format!("cancel-algo-order-rejected: s_code={code}, s_msg={msg}");
1095 let failure = classify_okx_venue_code(code, reason.clone());
1096 let is_rejected = matches!(failure, CommandFailure::VenueRejected(_));
1097 emit_cancel_failure(
1098 failure,
1099 Some((&emitter, clock)),
1100 command.client_order_id,
1101 command.instrument_id,
1102 command.strategy_id,
1103 command.venue_order_id,
1104 );
1105
1106 if is_rejected {
1107 anyhow::bail!("{reason}");
1108 }
1109 }
1110 }
1111 }
1112
1113 Ok(())
1114 });
1115 }
1116
1117 fn mass_cancel_instrument(&self, instrument_id: InstrumentId) {
1118 if is_spread_instrument(instrument_id) {
1119 let http_client = self.http_client.clone();
1120 self.spawn_task("mass_cancel_orders_http", async move {
1121 if let Err(e) = http_client.cancel_all_orders(instrument_id).await {
1122 log_mass_cancel_failure(classify_okx_http_failure(&e), instrument_id);
1123 return Err(anyhow::Error::new(e).context("mass cancel orders failed"));
1124 }
1125 Ok(())
1126 });
1127 return;
1128 }
1129
1130 let ws_private = self.ws_private.clone();
1131
1132 self.spawn_task("mass_cancel_orders", async move {
1133 if let Err(e) = ws_private.mass_cancel_orders(instrument_id).await {
1134 log_mass_cancel_failure(classify_okx_ws_failure(&e), instrument_id);
1135 return Err(anyhow::Error::new(e).context("mass cancel orders failed"));
1136 }
1137 Ok(())
1138 });
1139 }
1140
1141 fn ensure_order_identity(
1150 &self,
1151 client_order_id: ClientOrderId,
1152 strategy_id: StrategyId,
1153 instrument_id: InstrumentId,
1154 ) {
1155 if self
1156 .ws_dispatch_state
1157 .order_identity(client_order_id)
1158 .is_some()
1159 {
1160 return;
1161 }
1162
1163 let cache = self.core.cache();
1164 let order_identity = cache.order(&client_order_id).map(|order| OrderIdentity {
1165 client_order_id,
1166 instrument_id,
1167 strategy_id,
1168 order_side: order.order_side(),
1169 order_type: order.order_type(),
1170 });
1171 drop(cache);
1172
1173 if let Some(order_identity) = order_identity {
1174 self.ws_dispatch_state
1175 .order_identities
1176 .entry(client_order_id)
1177 .or_insert(order_identity);
1178 }
1179 }
1180
1181 fn spawn_task<F>(&self, description: &'static str, fut: F)
1182 where
1183 F: Future<Output = anyhow::Result<()>> + Send + 'static,
1184 {
1185 let fut = async move {
1186 if let Err(e) = fut.await {
1187 log::warn!("{description} failed: {e:?}");
1188 }
1189 };
1190
1191 match self.pending_tasks.spawner() {
1192 Ok(spawner) => spawn_task(&spawner, fut),
1193 Err(e) => log::debug!("Skipping {description} after OKX shutdown began: {e}"),
1194 }
1195 }
1196
1197 fn dispatch_algo_cancels(&self, items: Vec<(OKXCancelAlgoOrderRequest, AlgoCancelContext)>) {
1200 const BATCH_SIZE: usize = 10;
1201
1202 let mut regular_requests = Vec::new();
1203 let mut regular_contexts = Vec::new();
1204 let mut advance_requests = Vec::new();
1205 let mut advance_contexts = Vec::new();
1206
1207 let cache = self.core.cache();
1208
1209 for (request, ctx) in items {
1210 let is_advance = cache
1211 .order(&ctx.client_order_id)
1212 .is_some_and(|o| is_advance_algo_order(o.order_type()));
1213
1214 if is_advance {
1215 advance_requests.push(request);
1216 advance_contexts.push(ctx);
1217 } else {
1218 regular_requests.push(request);
1219 regular_contexts.push(ctx);
1220 }
1221 }
1222
1223 drop(cache);
1224
1225 for (requests, contexts) in regular_requests
1226 .chunks(BATCH_SIZE)
1227 .zip(regular_contexts.chunks(BATCH_SIZE))
1228 {
1229 let regular_requests = requests.to_vec();
1230 let regular_contexts = contexts.to_vec();
1231 let client = self.http_client.clone();
1232 let emitter = self.emitter.clone();
1233 let clock = self.clock;
1234
1235 self.spawn_task("cancel_algo_orders", async move {
1236 match client.cancel_algo_orders(regular_requests).await {
1237 Ok(responses) => {
1238 emit_algo_cancel_rejections(&responses, ®ular_contexts, &emitter, clock);
1239 }
1240 Err(e) => {
1241 log_algo_batch_cancel_failure(
1242 classify_okx_http_failure(&e),
1243 ®ular_contexts,
1244 );
1245 return Err(anyhow::Error::new(e).context("cancel algo orders failed"));
1246 }
1247 }
1248 Ok(())
1249 });
1250 }
1251
1252 for (requests, contexts) in advance_requests
1253 .chunks(BATCH_SIZE)
1254 .zip(advance_contexts.chunks(BATCH_SIZE))
1255 {
1256 let advance_requests = requests.to_vec();
1257 let advance_contexts = contexts.to_vec();
1258 let client = self.http_client.clone();
1259 let emitter = self.emitter.clone();
1260 let clock = self.clock;
1261
1262 self.spawn_task("cancel_advance_algo_orders", async move {
1263 match client.cancel_advance_algo_orders(advance_requests).await {
1264 Ok(responses) => {
1265 emit_algo_cancel_rejections(&responses, &advance_contexts, &emitter, clock);
1266 }
1267 Err(e) => {
1268 log_algo_batch_cancel_failure(
1269 classify_okx_http_failure(&e),
1270 &advance_contexts,
1271 );
1272 return Err(
1273 anyhow::Error::new(e).context("cancel advance algo orders failed")
1274 );
1275 }
1276 }
1277 Ok(())
1278 });
1279 }
1280 }
1281
1282 fn begin_generation_shutdown(&self) {
1283 self.pending_tasks.begin_shutdown();
1284 self.session_tasks.begin_shutdown();
1285 self.ws_private.begin_shutdown();
1286 self.ws_business.begin_shutdown();
1287 self.core.set_disconnected();
1288 }
1289
1290 async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
1292 let account_id = self.core.account_id;
1293
1294 if self.core.cache().account(&account_id).is_some() {
1295 log::info!("Account {account_id} registered");
1296 return Ok(());
1297 }
1298
1299 let start = Instant::now();
1300 let timeout = Duration::from_secs_f64(timeout_secs);
1301 let interval = Duration::from_millis(10);
1302
1303 loop {
1304 time::sleep(interval).await;
1305
1306 if self.core.cache().account(&account_id).is_some() {
1307 log::info!("Account {account_id} registered");
1308 return Ok(());
1309 }
1310
1311 if start.elapsed() >= timeout {
1312 anyhow::bail!(
1313 "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
1314 );
1315 }
1316 }
1317 }
1318
1319 async fn establish_session(&mut self) -> anyhow::Result<()> {
1325 if !self.pending_tasks.is_empty()
1327 || !self.session_tasks.is_empty()
1328 || !self.pending_tasks.is_open()
1329 || !self.session_tasks.is_open()
1330 || self.ws_private.is_active()
1331 || self.ws_business.is_active()
1332 || self.ws_private.has_task()
1333 || self.ws_business.has_task()
1334 {
1335 self.teardown_session().await?;
1336 }
1337
1338 if !self.pending_tasks.is_open() {
1339 self.pending_tasks
1340 .start_generation()
1341 .context("failed to start OKX execution request task generation")?;
1342 }
1343
1344 if !self.session_tasks.is_open() {
1345 self.session_tasks
1346 .start_generation()
1347 .context("failed to start OKX execution stream task generation")?;
1348 }
1349 let instrument_types = self.instrument_types();
1350
1351 if !self.core.instruments_initialized() {
1352 let mut all_instruments = Vec::new();
1353 let mut all_inst_id_codes = Vec::new();
1354
1355 for instrument_type in &instrument_types {
1356 let Some(families) =
1357 resolve_instrument_families(&self.config.instrument_families, *instrument_type)
1358 else {
1359 continue;
1360 };
1361
1362 if families.is_empty() {
1363 let (instruments, inst_id_codes) = self
1364 .http_client
1365 .request_instruments(*instrument_type, None)
1366 .await
1367 .with_context(|| {
1368 format!("failed to request OKX instruments for {instrument_type:?}")
1369 })?;
1370
1371 if instruments.is_empty() {
1372 anyhow::bail!(
1373 "No usable instruments for {instrument_type:?}, \
1374 cannot initialize execution client"
1375 );
1376 }
1377
1378 log::debug!(
1379 "Loaded {} {instrument_type:?} instruments",
1380 instruments.len()
1381 );
1382
1383 self.http_client.cache_instruments(&instruments);
1384 all_instruments.extend(instruments);
1385 all_inst_id_codes.extend(inst_id_codes);
1386 } else {
1387 for family in &families {
1388 let (instruments, inst_id_codes) = self
1389 .http_client
1390 .request_instruments(*instrument_type, Some(family.clone()))
1391 .await
1392 .with_context(|| {
1393 format!(
1394 "failed to request OKX instruments for {instrument_type:?} family {family}"
1395 )
1396 })?;
1397
1398 if instruments.is_empty() {
1399 anyhow::bail!(
1400 "No usable instruments for {instrument_type:?} family {family}, \
1401 cannot initialize execution client"
1402 );
1403 }
1404
1405 log::debug!(
1406 "Loaded {} {instrument_type:?} instruments for family {family}",
1407 instruments.len()
1408 );
1409
1410 self.http_client.cache_instruments(&instruments);
1411 all_instruments.extend(instruments);
1412 all_inst_id_codes.extend(inst_id_codes);
1413 }
1414 }
1415 }
1416
1417 if all_instruments.is_empty() {
1418 anyhow::bail!(
1419 "No instruments loaded for configured types {instrument_types:?}, \
1420 cannot initialize execution client"
1421 );
1422 }
1423
1424 if instrument_types.contains(&OKXInstrumentType::Spot)
1425 && let Err(e) = self
1426 .http_client
1427 .refresh_account_trade_quote_ccy_lists(OKXInstrumentType::Spot, None)
1428 .await
1429 {
1430 log::warn!("Failed to refresh account tradeQuoteCcyList: {e}");
1431 }
1432
1433 let trade_quote_ccy_lists = self.http_client.trade_quote_ccy_lists_snapshot();
1434 self.ws_private.cache_instruments(&all_instruments);
1435 self.ws_private
1436 .cache_inst_id_codes(all_inst_id_codes.clone());
1437 self.ws_private
1438 .cache_trade_quote_ccy_lists(trade_quote_ccy_lists.clone());
1439 self.ws_business.cache_instruments(&all_instruments);
1440 self.ws_business.cache_inst_id_codes(all_inst_id_codes);
1441 self.ws_business
1442 .cache_trade_quote_ccy_lists(trade_quote_ccy_lists);
1443 self.core.set_instruments_initialized();
1444 }
1445
1446 self.ws_private.connect().await?;
1447 self.ws_private.wait_until_active(10.0).await?;
1448 log::info!("Connected to private WebSocket");
1449
1450 {
1451 let stream = self.ws_private.stream();
1452 let emitter = self.emitter.clone();
1453 let state = Arc::clone(&self.ws_dispatch_state);
1454 let account_id = self.core.account_id;
1455 let account_type = self.core.account_type;
1456 let instruments = self.ws_private.instruments_cache_arc();
1457 let tasks = self
1458 .session_tasks
1459 .spawner()
1460 .context("OKX execution stream task admission is closed")?;
1461 let cancel = tasks.cancellation_token();
1462 let clock = self.clock;
1463
1464 spawn_task(&tasks, async move {
1465 let mut fee_cache = FeeCache::new();
1466 let mut filled_qty_cache = FilledQtyCache::new();
1467 let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
1468 AHashMap::new();
1469
1470 pin_mut!(stream);
1471
1472 loop {
1473 tokio::select! {
1474 biased;
1475 () = cancel.cancelled() => break,
1476 message = stream.next() => {
1477 let Some(message) = message else {
1478 break;
1479 };
1480 dispatch_ws_message(
1481 message,
1482 &emitter,
1483 &state,
1484 account_id,
1485 account_type,
1486 &instruments,
1487 &mut fee_cache,
1488 &mut filled_qty_cache,
1489 &mut order_state_cache,
1490 clock,
1491 );
1492 }
1493 () = state.wait_for_linked_child_route() => {
1494 dispatch_ws_message(
1495 OKXWsMessage::Orders(Vec::new()),
1496 &emitter,
1497 &state,
1498 account_id,
1499 account_type,
1500 &instruments,
1501 &mut fee_cache,
1502 &mut filled_qty_cache,
1503 &mut order_state_cache,
1504 clock,
1505 );
1506 }
1507 }
1508 }
1509 });
1510 }
1511
1512 self.ws_business.connect().await?;
1513 self.ws_business.wait_until_active(10.0).await?;
1514 log::info!("Connected to business WebSocket");
1515
1516 {
1517 let stream = self.ws_business.stream();
1518 let emitter = self.emitter.clone();
1519 let state = Arc::clone(&self.ws_dispatch_state);
1520 let account_id = self.core.account_id;
1521 let account_type = self.core.account_type;
1522 let instruments = self.ws_business.instruments_cache_arc();
1523 let tasks = self
1524 .session_tasks
1525 .spawner()
1526 .context("OKX execution stream task admission is closed")?;
1527 let cancel = tasks.cancellation_token();
1528 let clock = self.clock;
1529
1530 spawn_task(&tasks, async move {
1531 let mut fee_cache = FeeCache::new();
1532 let mut filled_qty_cache = FilledQtyCache::new();
1533 let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
1534 AHashMap::new();
1535
1536 pin_mut!(stream);
1537
1538 loop {
1539 tokio::select! {
1540 biased;
1541 () = cancel.cancelled() => break,
1542 message = stream.next() => {
1543 let Some(message) = message else {
1544 break;
1545 };
1546 dispatch_ws_message(
1547 message,
1548 &emitter,
1549 &state,
1550 account_id,
1551 account_type,
1552 &instruments,
1553 &mut fee_cache,
1554 &mut filled_qty_cache,
1555 &mut order_state_cache,
1556 clock,
1557 );
1558 }
1559 }
1560 }
1561 });
1562 }
1563
1564 let order_routing_types = order_routing_instrument_types(&instrument_types);
1565
1566 for inst_type in &order_routing_types {
1567 log::debug!("Subscribing to orders channel for {inst_type:?}");
1568 self.ws_private.subscribe_orders(*inst_type).await?;
1569 }
1570
1571 self.ws_private.subscribe_account().await?;
1572
1573 if order_routing_types.iter().any(|t| {
1575 matches!(
1576 t,
1577 OKXInstrumentType::Margin
1578 | OKXInstrumentType::Swap
1579 | OKXInstrumentType::Futures
1580 | OKXInstrumentType::Option
1581 )
1582 }) {
1583 log::debug!("Subscribing to liquidation warning channel");
1584 self.ws_private
1585 .subscribe_liquidation_warning(OKXInstrumentType::Any)
1586 .await?;
1587 }
1588
1589 if self.config.load_spreads {
1590 log::debug!("Subscribing to Nitro spread orders channel");
1591 self.ws_business.subscribe_spread_orders().await?;
1592 }
1593
1594 for inst_type in &order_routing_types {
1596 if supports_algo_orders(*inst_type) {
1597 self.ws_business.subscribe_orders_algo(*inst_type).await?;
1598 self.ws_business.subscribe_algo_advance(*inst_type).await?;
1599 }
1600 }
1601
1602 let account_state = self
1603 .http_client
1604 .request_account_state(self.core.account_id, self.core.account_type)
1605 .await
1606 .context("failed to request OKX account state")?;
1607
1608 if !account_state.balances.is_empty() {
1609 log::debug!(
1610 "Received account state with {} balance(s)",
1611 account_state.balances.len()
1612 );
1613 }
1614 self.emitter.send_account_state(account_state);
1615
1616 self.await_account_registered(30.0).await?;
1618
1619 Ok(())
1620 }
1621
1622 async fn teardown_session(&mut self) -> anyhow::Result<()> {
1625 self.begin_generation_shutdown();
1626 self.ws_private.request_close().await;
1627 self.ws_business.request_close().await;
1628 let pending_result = terminate_tasks(&self.pending_tasks, "OKX execution request").await;
1629 let session_result = terminate_tasks(&self.session_tasks, "OKX execution stream").await;
1630
1631 let private_result = self
1632 .ws_private
1633 .close()
1634 .await
1635 .context("failed to close private websocket");
1636 let business_result = self
1637 .ws_business
1638 .close()
1639 .await
1640 .context("failed to close business websocket");
1641
1642 self.core.set_disconnected();
1643
1644 let mut errors = Vec::new();
1645 if let Err(e) = pending_result {
1646 errors.push(e.to_string());
1647 }
1648
1649 if let Err(e) = session_result {
1650 errors.push(e.to_string());
1651 }
1652
1653 if let Err(e) = private_result {
1654 errors.push(e.to_string());
1655 }
1656
1657 if let Err(e) = business_result {
1658 errors.push(e.to_string());
1659 }
1660
1661 if errors.is_empty() {
1662 Ok(())
1663 } else {
1664 anyhow::bail!(errors.join("; "))
1665 }
1666 }
1667}
1668
1669fn derive_trade_mode_for_instrument(
1670 instrument_id: InstrumentId,
1671 margin_mode: Option<OKXMarginMode>,
1672 use_spot_margin: bool,
1673) -> OKXTradeMode {
1674 let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
1675 let is_cross_margin = margin_mode == Some(OKXMarginMode::Cross);
1676
1677 match inst_type {
1678 OKXInstrumentType::Spot => {
1679 if use_spot_margin {
1680 if is_cross_margin {
1681 OKXTradeMode::Cross
1682 } else {
1683 OKXTradeMode::Isolated
1684 }
1685 } else {
1686 OKXTradeMode::Cash
1687 }
1688 }
1689 _ => {
1690 if is_cross_margin {
1691 OKXTradeMode::Cross
1692 } else {
1693 OKXTradeMode::Isolated
1694 }
1695 }
1696 }
1697}
1698
1699#[async_trait(?Send)]
1700impl ExecutionClient for OKXExecutionClient {
1701 fn is_connected(&self) -> bool {
1702 self.core.is_connected()
1703 }
1704
1705 fn client_id(&self) -> ClientId {
1706 self.core.client_id
1707 }
1708
1709 fn account_id(&self) -> AccountId {
1710 self.core.account_id
1711 }
1712
1713 fn venue(&self) -> Venue {
1714 *OKX_VENUE
1715 }
1716
1717 fn oms_type(&self) -> OmsType {
1718 self.core.oms_type
1719 }
1720
1721 fn get_account(&self) -> Option<AccountAny> {
1722 self.core.cache().account_owned(&self.core.account_id)
1723 }
1724
1725 async fn connect(&mut self) -> anyhow::Result<()> {
1726 if self.core.is_connected() && self.pending_tasks.is_open() && self.session_tasks.is_open()
1727 {
1728 return Ok(());
1729 }
1730 let ws_private = self.ws_private.clone();
1731 let ws_business = self.ws_business.clone();
1732 let setup_guard =
1733 TaskGroupGuard::new(&[&self.session_tasks, &self.pending_tasks], move || {
1734 ws_private.begin_shutdown();
1735 ws_business.begin_shutdown();
1736 });
1737
1738 if let Err(e) = self.establish_session().await {
1739 if let Err(teardown_error) = self.teardown_session().await {
1740 return Err(e.context(format!(
1741 "OKX execution startup teardown failed: {teardown_error}"
1742 )));
1743 }
1744 return Err(e);
1745 }
1746
1747 self.core.set_connected();
1748 setup_guard.disarm();
1749 log::info!("Connected: client_id={}", self.core.client_id);
1750 Ok(())
1751 }
1752
1753 async fn disconnect(&mut self) -> anyhow::Result<()> {
1754 if self.core.is_disconnected()
1755 && self.pending_tasks.is_empty()
1756 && self.session_tasks.is_empty()
1757 && !self.ws_private.has_task()
1758 && !self.ws_business.has_task()
1759 {
1760 return Ok(());
1761 }
1762
1763 self.teardown_session().await?;
1764 log::info!("Disconnected: client_id={}", self.core.client_id);
1765 Ok(())
1766 }
1767
1768 fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
1769 self.update_account_state();
1770 Ok(())
1771 }
1772
1773 fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
1774 let http_client = self.http_client.clone();
1775 let account_id = self.core.account_id;
1776 let emitter = self.emitter.clone();
1777 let instrument_id = cmd.instrument_id;
1778 let client_order_id = cmd.client_order_id;
1779 let venue_order_id = cmd.venue_order_id;
1780 let order_state = {
1781 let cache = self.core.cache();
1782 cache
1783 .order(&client_order_id)
1784 .map(|order| CachedQueryOrderState {
1785 order_type: order.order_type(),
1786 venue_order_id: order.venue_order_id(),
1787 })
1788 };
1789
1790 let venue_binding = self.ws_dispatch_state.order_venue_binding(client_order_id);
1791 let authoritative_venue_order_id = venue_binding.map(|(venue_order_id, _)| venue_order_id);
1792 let has_bound_child = venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child);
1793 let cached_venue_order_id = order_state.and_then(|state| state.venue_order_id);
1794 let regular_venue_order_id = if has_bound_child {
1795 authoritative_venue_order_id
1796 } else {
1797 order_state.and_then(|state| {
1798 if OKX_CONDITIONAL_ORDER_TYPES.contains(&state.order_type) {
1799 state.venue_order_id.or(venue_order_id)
1800 } else {
1801 state.venue_order_id
1802 }
1803 })
1804 };
1805
1806 let selection_venue_order_id = authoritative_venue_order_id
1807 .or(cached_venue_order_id)
1808 .or(venue_order_id);
1809 let route = query_order_route(
1810 instrument_id,
1811 order_state.map(|state| state.order_type),
1812 regular_venue_order_id.is_some(),
1813 );
1814 self.spawn_task("query_order", async move {
1815 let mut reports = Vec::with_capacity(1);
1816 let mut query_algo = matches!(
1817 route,
1818 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo
1819 );
1820
1821 match route {
1822 QueryOrderRoute::Spread => {
1823 match http_client
1824 .request_order_status_reports(
1825 account_id,
1826 None,
1827 Some(instrument_id),
1828 None,
1829 None,
1830 false,
1831 None,
1832 )
1833 .await
1834 {
1835 Ok(spread_reports) => reports.extend(spread_reports),
1836 Err(e) => {
1837 log::error!("OKX query_order failed to fetch spread order: {e}");
1838 }
1839 }
1840 }
1841 QueryOrderRoute::Regular | QueryOrderRoute::RegularThenAlgo => {
1842 let result = if let Some(venue_order_id) = regular_venue_order_id {
1843 http_client
1844 .request_order_status_report_by_venue_order_id(
1845 account_id,
1846 instrument_id,
1847 venue_order_id,
1848 )
1849 .await
1850 } else {
1851 http_client
1852 .request_order_status_report(
1853 account_id,
1854 instrument_id,
1855 client_order_id,
1856 )
1857 .await
1858 };
1859
1860 match result {
1861 Ok(Some(report)) => reports.push(report),
1862 Ok(None) => {
1863 query_algo |= route == QueryOrderRoute::RegularThenAlgo;
1864 }
1865 Err(e) => {
1866 log::error!("OKX query_order failed to fetch regular order: {e}");
1867 }
1868 }
1869 }
1870 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo => {}
1871 }
1872
1873 if query_algo {
1878 let mut regular_child_venue_order_id = None;
1879
1880 match http_client
1881 .request_algo_order_status_reports(
1882 account_id,
1883 None,
1884 Some(instrument_id),
1885 None,
1886 Some(client_order_id),
1887 None,
1888 Some(1),
1889 )
1890 .await
1891 {
1892 Ok(algo_reports) => {
1893 if matches!(
1894 route,
1895 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo
1896 ) {
1897 regular_child_venue_order_id = algo_reports
1898 .iter()
1899 .find(|report| {
1900 matches!(
1901 report.order_status,
1902 OrderStatus::Triggered | OrderStatus::Filled
1903 )
1904 })
1905 .map(|report| report.venue_order_id)
1906 .or_else(|| {
1907 regular_venue_order_id.filter(|venue_order_id| {
1908 algo_reports.first().is_none_or(|report| {
1909 report.venue_order_id != *venue_order_id
1910 })
1911 })
1912 });
1913 }
1914
1915 merge_order_status_reports(&mut reports, algo_reports);
1916 }
1917 Err(e) => {
1918 if route == QueryOrderRoute::RegularAndAlgo {
1919 regular_child_venue_order_id = regular_venue_order_id;
1920 }
1921
1922 log::warn!("OKX query_order algo lookup failed for {instrument_id}: {e}");
1923 }
1924 }
1925
1926 if let Some(child_venue_order_id) = regular_child_venue_order_id {
1927 match http_client
1928 .request_order_status_report_by_venue_order_id(
1929 account_id,
1930 instrument_id,
1931 child_venue_order_id,
1932 )
1933 .await
1934 {
1935 Ok(Some(child_report)) => {
1936 merge_order_status_reports(&mut reports, vec![child_report]);
1937 }
1938 Ok(None) => {}
1939 Err(e) => {
1940 log::error!(
1941 "OKX query_order failed to fetch regular child order: {e}"
1942 );
1943 }
1944 }
1945 }
1946 }
1947
1948 let Some(report) = select_query_order_report(
1949 reports,
1950 client_order_id,
1951 selection_venue_order_id,
1952 ) else {
1953 log::warn!(
1954 "OKX query_order found no order for client_order_id={client_order_id}, venue_order_id={selection_venue_order_id:?}",
1955 );
1956 return Ok(());
1957 };
1958
1959 emitter.send_order_status_report(report);
1960 Ok(())
1961 });
1962 Ok(())
1963 }
1964
1965 fn generate_account_state(
1966 &self,
1967 balances: Vec<AccountBalance>,
1968 margins: Vec<MarginBalance>,
1969 reported: bool,
1970 ts_event: UnixNanos,
1971 info: Option<Params>,
1972 ) -> anyhow::Result<()> {
1973 self.emitter
1974 .emit_account_state(balances, margins, reported, ts_event, info);
1975 Ok(())
1976 }
1977
1978 fn on_instrument(&mut self, instrument: InstrumentAny) {
1979 self.http_client.cache_instrument(instrument.clone());
1980 self.ws_private.cache_instrument(instrument.clone());
1981 self.ws_business.cache_instrument(instrument);
1982 }
1983
1984 fn start(&mut self) -> anyhow::Result<()> {
1985 if self.core.is_started() {
1986 return Ok(());
1987 }
1988
1989 let sender = get_exec_event_sender();
1990 self.emitter.set_sender(sender);
1991 self.core.set_started();
1992
1993 log::info!(
1994 "Started: client_id={}, account_id={}, account_type={:?}, trade_mode={:?}, instrument_types={:?}, environment={}, proxy_url={:?}",
1995 self.core.client_id,
1996 self.core.account_id,
1997 self.core.account_type,
1998 self.trade_mode,
1999 self.config.instrument_types,
2000 self.config.environment,
2001 self.config.proxy_url,
2002 );
2003 Ok(())
2004 }
2005
2006 fn stop(&mut self) -> anyhow::Result<()> {
2007 let was_started = self.core.is_started();
2008 self.core.set_stopped();
2009 self.begin_generation_shutdown();
2010
2011 if was_started {
2012 log::info!("Stopped: client_id={}", self.core.client_id);
2013 }
2014 Ok(())
2015 }
2016
2017 fn reset(&mut self) -> anyhow::Result<()> {
2018 self.begin_generation_shutdown();
2019 Ok(())
2020 }
2021
2022 fn dispose(&mut self) -> anyhow::Result<()> {
2023 self.begin_generation_shutdown();
2024 Ok(())
2025 }
2026
2027 async fn generate_order_status_report(
2028 &self,
2029 cmd: &GenerateOrderStatusReport,
2030 ) -> anyhow::Result<Option<OrderStatusReport>> {
2031 let Some(instrument_id) = cmd.instrument_id else {
2032 anyhow::bail!("generate_order_status_report requires instrument_id");
2033 };
2034
2035 if cmd.client_order_id.is_none() && cmd.venue_order_id.is_none() {
2036 anyhow::bail!(
2037 "generate_order_status_report requires client_order_id or venue_order_id"
2038 );
2039 }
2040
2041 let order_state = {
2042 let cache = self.core.cache();
2043 cmd.client_order_id.and_then(|client_order_id| {
2044 cache
2045 .order(&client_order_id)
2046 .map(|order| CachedQueryOrderState {
2047 order_type: order.order_type(),
2048 venue_order_id: order.venue_order_id(),
2049 })
2050 })
2051 };
2052 let cached_venue_order_id = order_state.and_then(|state| state.venue_order_id);
2053 let regular_venue_order_id = order_state.and_then(|state| {
2054 if OKX_CONDITIONAL_ORDER_TYPES.contains(&state.order_type) {
2055 state.venue_order_id.or(cmd.venue_order_id)
2056 } else {
2057 state.venue_order_id
2058 }
2059 });
2060 let selection_venue_order_id = cached_venue_order_id.or(cmd.venue_order_id);
2061 let route = query_order_route(
2062 instrument_id,
2063 order_state.map(|state| state.order_type),
2064 regular_venue_order_id.is_some(),
2065 );
2066
2067 let mut reports = Vec::with_capacity(1);
2068 let mut query_algo = matches!(
2069 route,
2070 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo
2071 );
2072 let mut lookup_error = None;
2073
2074 match route {
2075 QueryOrderRoute::Spread => {
2076 let targeted_venue_order_id =
2077 cmd.venue_order_id.filter(|_| cmd.client_order_id.is_none());
2078
2079 match self
2080 .http_client
2081 .request_spread_order_status_report(
2082 self.core.account_id,
2083 instrument_id,
2084 cmd.client_order_id,
2085 targeted_venue_order_id,
2086 )
2087 .await
2088 {
2089 Ok(Some(report)) => reports.push(report),
2090 Ok(None) => {}
2091 Err(e) => lookup_error = Some(e),
2092 }
2093 }
2094 QueryOrderRoute::Regular | QueryOrderRoute::RegularThenAlgo => {
2095 let targeted_venue_order_id = regular_venue_order_id
2096 .or(cmd.venue_order_id.filter(|_| cmd.client_order_id.is_none()));
2097 let result = if let Some(venue_order_id) = targeted_venue_order_id {
2098 self.http_client
2099 .request_order_status_report_by_venue_order_id(
2100 self.core.account_id,
2101 instrument_id,
2102 venue_order_id,
2103 )
2104 .await
2105 } else if let Some(client_order_id) = cmd.client_order_id {
2106 self.http_client
2107 .request_order_status_report(
2108 self.core.account_id,
2109 instrument_id,
2110 client_order_id,
2111 )
2112 .await
2113 } else {
2114 anyhow::bail!(
2115 "generate_order_status_report requires client_order_id or venue_order_id"
2116 );
2117 };
2118
2119 match result {
2120 Ok(Some(report)) => reports.push(report),
2121 Ok(None) => {
2122 query_algo |= route == QueryOrderRoute::RegularThenAlgo;
2123 }
2124 Err(e) => {
2125 lookup_error = Some(e);
2126 query_algo |= route == QueryOrderRoute::RegularThenAlgo;
2127 }
2128 }
2129 }
2130 QueryOrderRoute::Algo | QueryOrderRoute::RegularAndAlgo => {}
2131 }
2132
2133 if query_algo {
2134 let (algo_id, algo_client_order_id) = match cmd.client_order_id {
2135 Some(client_order_id) => (None, Some(client_order_id)),
2136 None => (cmd.venue_order_id.map(|id| id.as_str().to_string()), None),
2137 };
2138
2139 match self
2140 .http_client
2141 .request_algo_order_status_reports(
2142 self.core.account_id,
2143 None,
2144 Some(instrument_id),
2145 algo_id,
2146 algo_client_order_id,
2147 None,
2148 Some(1),
2149 )
2150 .await
2151 {
2152 Ok(algo_reports) => merge_order_status_reports(&mut reports, algo_reports),
2153 Err(e) => {
2154 if lookup_error.is_none() {
2155 lookup_error = Some(e);
2156 }
2157 }
2158 }
2159 }
2160
2161 if reports.is_empty() {
2162 if let Some(e) = lookup_error {
2163 return Err(e);
2164 }
2165 return Ok(None);
2166 }
2167
2168 if let Some(client_order_id) = cmd.client_order_id {
2169 Ok(select_query_order_report(
2170 reports,
2171 client_order_id,
2172 selection_venue_order_id,
2173 ))
2174 } else {
2175 Ok(Some(reports.remove(0)))
2176 }
2177 }
2178
2179 async fn generate_order_status_reports(
2180 &self,
2181 cmd: &GenerateOrderStatusReports,
2182 ) -> anyhow::Result<Vec<OrderStatusReport>> {
2183 Ok(self.collect_order_status_reports(cmd, false).await?.reports)
2184 }
2185
2186 async fn generate_fill_reports(
2187 &self,
2188 cmd: GenerateFillReports,
2189 ) -> anyhow::Result<Vec<FillReport>> {
2190 Ok(self.collect_fill_reports(cmd, FillHistory::Recent).await?.0)
2191 }
2192
2193 async fn generate_position_status_reports(
2194 &self,
2195 cmd: &GeneratePositionStatusReports,
2196 ) -> anyhow::Result<Vec<PositionStatusReport>> {
2197 Ok(self.collect_position_status_reports(cmd).await?.0)
2198 }
2199
2200 async fn generate_mass_status(
2201 &self,
2202 lookback_mins: Option<u64>,
2203 ) -> anyhow::Result<Option<ExecutionMassStatus>> {
2204 log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
2205
2206 let ts_now = self.clock.get_time_ns();
2207
2208 let lookback_mins = lookback_mins
2209 .unwrap_or(OKX_RECONCILIATION_LOOKBACK_DEFAULT_MINS)
2210 .min(OKX_RECONCILIATION_LOOKBACK_MAX_MINS);
2211 let fill_history = if lookback_mins <= OKX_RECONCILIATION_LOOKBACK_DEFAULT_MINS {
2212 FillHistory::Recent
2213 } else {
2214 FillHistory::Extended
2215 };
2216 let lookback = DurationNanos::try_from_mins(lookback_mins)?;
2217 let start = Some(ts_now.saturating_sub(lookback));
2218
2219 let order_cmd = GenerateOrderStatusReportsBuilder::default()
2220 .ts_init(ts_now)
2221 .open_only(false) .start(start)
2223 .build()
2224 .map_err(|e| anyhow::anyhow!("{e}"))?;
2225
2226 let fill_cmd = GenerateFillReportsBuilder::default()
2227 .ts_init(ts_now)
2228 .start(start)
2229 .build()
2230 .map_err(|e| anyhow::anyhow!("{e}"))?;
2231
2232 let position_cmd = GeneratePositionStatusReportsBuilder::default()
2233 .ts_init(ts_now)
2234 .start(start)
2235 .build()
2236 .map_err(|e| anyhow::anyhow!("{e}"))?;
2237
2238 let (
2239 order_sweep,
2240 (fill_reports, fills_complete),
2241 (mut position_reports, positions_complete),
2242 ) = tokio::try_join!(
2243 self.collect_order_status_reports(&order_cmd, true),
2244 self.collect_fill_reports(fill_cmd, fill_history),
2245 self.collect_position_status_reports(&position_cmd),
2246 )?;
2247 let OrderReportSweep {
2248 reports: mut order_reports,
2249 complete: mut orders_complete,
2250 ambiguous_triggered_child_ids,
2251 regular_by_venue_order_id,
2252 } = order_sweep;
2253 orders_complete &= self
2254 .recover_triggered_child_order_reports(
2255 &mut order_reports,
2256 &ambiguous_triggered_child_ids,
2257 ®ular_by_venue_order_id,
2258 )
2259 .await;
2260
2261 if positions_complete {
2262 self.add_flat_derivative_position_reports(
2263 &order_reports,
2264 &fill_reports,
2265 &mut position_reports,
2266 ts_now,
2267 );
2268 }
2269 let reports_complete = orders_complete && fills_complete && positions_complete;
2270
2271 log::info!("Received {} OrderStatusReports", order_reports.len());
2272 log::info!("Received {} FillReports", fill_reports.len());
2273 log::info!("Received {} PositionReports", position_reports.len());
2274
2275 let mut mass_status = ExecutionMassStatus::new(
2276 self.core.client_id,
2277 self.core.account_id,
2278 *OKX_VENUE,
2279 ts_now,
2280 None,
2281 );
2282 mass_status.set_report_window(start, reports_complete);
2283 mass_status.add_order_reports(order_reports);
2284 mass_status.add_fill_reports(fill_reports);
2285 mass_status.add_position_reports(position_reports);
2286
2287 Ok(Some(mass_status))
2288 }
2289
2290 fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
2291 let route = {
2292 let cache = self.core.cache();
2293 let order = cache.try_order(&cmd.client_order_id)?;
2294
2295 if order.is_closed() {
2296 log::warn!("Cannot submit closed order {}", order.client_order_id());
2297 return Ok(());
2298 }
2299
2300 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
2301 if let Err(reason) = validate_order(&*order, trade_mode, OrderSubmission::Single) {
2302 self.emitter.emit_order_denied(&order, &reason.to_string());
2303 return Ok(());
2304 }
2305
2306 let order_type = order.order_type();
2307 let route = self.submit_order_route(cmd.instrument_id, order_type)?;
2308
2309 log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
2310 self.emitter.emit_order_submitted(&order);
2311
2312 route
2313 };
2314
2315 match route {
2316 OrderCommandRoute::RegularWs => self.submit_regular_order(&cmd),
2317 OrderCommandRoute::AlgoHttp => self.submit_conditional_order(&cmd),
2318 OrderCommandRoute::SpreadHttp => self.submit_order_http(&cmd),
2319 }
2320 }
2321
2322 fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
2323 if is_spread_instrument(cmd.instrument_id) {
2324 let cache = self.core.cache();
2325 let denied = OrderDeniedReason::UnsupportedOrderList {
2326 detail: "spread instruments are not supported in order lists".to_string(),
2327 }
2328 .to_string();
2329
2330 for client_order_id in &cmd.order_list.client_order_ids {
2331 let order = cache.try_order(client_order_id)?;
2332 self.emitter.emit_order_denied(&order, &denied);
2333 }
2334 return Ok(());
2335 }
2336
2337 let inst_type = okx_instrument_type_from_symbol(cmd.instrument_id.symbol.as_str());
2338 let trade_mode = self.trade_mode_for_order(cmd.instrument_id, cmd.params.as_ref());
2339
2340 let orders = self.core.get_orders_for_list(&cmd.order_list)?;
2342
2343 let invalid: Vec<(ClientOrderId, OrderDeniedReason)> = orders
2346 .iter()
2347 .filter_map(|order| {
2348 validate_order(order, trade_mode, OrderSubmission::List)
2349 .err()
2350 .map(|reason| (order.client_order_id(), reason))
2351 })
2352 .collect();
2353
2354 if !invalid.is_empty() {
2355 let order_list_id = cmd.order_list.id;
2356
2357 for order in &orders {
2358 let denied = invalid
2359 .iter()
2360 .find(|(client_order_id, _)| client_order_id == &order.client_order_id())
2361 .map_or_else(
2362 || OrderDeniedReason::OrderListDenied { order_list_id },
2363 |(_, reason)| reason.clone(),
2364 );
2365 self.emitter.emit_order_denied(order, &denied.to_string());
2366 }
2367 return Ok(());
2368 }
2369
2370 let mut batch_orders = Vec::new();
2372 let outcome = get_param_as_string(cmd.params.as_ref(), "outcome");
2373 let rpi = get_param_as_bool(cmd.params.as_ref(), "rpi");
2374 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
2375 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
2376
2377 for order in &orders {
2378 let context = OrderContext::from(order);
2379
2380 batch_orders.push((
2381 inst_type,
2382 cmd.instrument_id,
2383 trade_mode,
2384 context.identity.client_order_id,
2385 context.identity.order_side,
2386 None, context.identity.order_type,
2388 context.quantity,
2389 context.price,
2390 context.trigger_price,
2391 Some(context.is_post_only),
2392 Some(context.is_reduce_only),
2393 outcome.clone(),
2394 rpi,
2395 rpi_taker_access,
2396 rpi_px_round,
2397 ));
2398
2399 self.ws_dispatch_state
2400 .order_identities
2401 .insert(context.identity.client_order_id, context.identity);
2402
2403 log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
2404 self.emitter.emit_order_submitted(order);
2405 }
2406
2407 let ws_private = self.ws_private.clone();
2408 let emitter = self.emitter.clone();
2409 let clock = self.clock;
2410 let instrument_id = cmd.instrument_id;
2411 let strategy_id = cmd.strategy_id;
2412 let client_order_ids: Vec<_> = cmd.order_list.client_order_ids;
2413 let dispatch_state = Arc::clone(&self.ws_dispatch_state);
2414
2415 self.spawn_task("batch_submit_orders", async move {
2416 let result = ws_private
2417 .batch_submit_orders(batch_orders)
2418 .await;
2419
2420 if let Err(e) = result {
2421 match classify_okx_ws_failure(&e) {
2422 CommandFailure::NotSent(reason) => {
2423 let ts_event = clock.get_time_ns();
2424
2425 for cid in &client_order_ids {
2426 dispatch_state.order_identities.remove(cid);
2427 emitter.emit_order_rejected_event(
2428 strategy_id,
2429 instrument_id,
2430 *cid,
2431 &reason,
2432 ts_event,
2433 false,
2434 );
2435 }
2436 }
2437 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
2438 log::warn!(
2439 "Ambiguous batch submit failure for {} orders on {instrument_id}, awaiting reconciliation: {reason}",
2440 client_order_ids.len()
2441 );
2442 }
2443 }
2444 return Err(anyhow::Error::new(e).context("batch submit orders failed"));
2445 }
2446
2447 Ok(())
2448 });
2449
2450 Ok(())
2451 }
2452
2453 fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
2454 if is_spread_instrument(cmd.instrument_id) {
2455 self.emitter.emit_order_modify_rejected_event(
2456 cmd.strategy_id,
2457 cmd.instrument_id,
2458 cmd.client_order_id,
2459 cmd.venue_order_id,
2460 "OKX spread orders do not support modify requests",
2461 self.clock.get_time_ns(),
2462 );
2463 return Ok(());
2464 }
2465
2466 self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
2467
2468 let ws_private = self.ws_private.clone();
2469 let mut command = cmd.clone();
2470 command.venue_order_id = self
2471 .ws_dispatch_state
2472 .order_venue_binding(cmd.client_order_id)
2473 .map(|(venue_order_id, _)| venue_order_id)
2474 .or(cmd.venue_order_id);
2475
2476 let new_px_usd = get_param_as_string(cmd.params.as_ref(), "px_usd");
2477 let new_px_vol = get_param_as_string(cmd.params.as_ref(), "px_vol");
2478 let rpi_taker_access = get_param_as_bool(cmd.params.as_ref(), "rpi_taker_access");
2479 let rpi_px_round = get_param_as_bool(cmd.params.as_ref(), "rpi_px_round");
2480
2481 let emitter = self.emitter.clone();
2482 let clock = self.clock;
2483
2484 self.spawn_task("modify_order", async move {
2485 let result = ws_private
2486 .modify_order(
2487 command.trader_id,
2488 command.strategy_id,
2489 command.instrument_id,
2490 Some(command.client_order_id),
2491 command.price,
2492 command.quantity,
2493 command.venue_order_id,
2494 new_px_usd,
2495 new_px_vol,
2496 rpi_taker_access,
2497 rpi_px_round,
2498 )
2499 .await;
2500
2501 if let Err(e) = result {
2502 emit_modify_failure(
2503 classify_okx_ws_failure(&e),
2504 &emitter,
2505 clock,
2506 command.strategy_id,
2507 command.instrument_id,
2508 command.client_order_id,
2509 command.venue_order_id,
2510 );
2511 return Err(anyhow::Error::new(e).context("modify order failed"));
2512 }
2513
2514 Ok(())
2515 });
2516
2517 Ok(())
2518 }
2519
2520 fn cancel_order(&self, mut cmd: CancelOrder) -> anyhow::Result<()> {
2521 let venue_binding = self
2522 .ws_dispatch_state
2523 .order_venue_binding(cmd.client_order_id);
2524 let route = {
2525 let cache = self.core.cache();
2526 let order_state = cache
2527 .order(&cmd.client_order_id)
2528 .map(|order| (order.order_type(), order.is_triggered()));
2529 self.cancel_order_route(
2530 cmd.instrument_id,
2531 order_state,
2532 venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child),
2533 )
2534 };
2535
2536 cmd.venue_order_id = venue_binding
2537 .map(|(venue_order_id, _)| venue_order_id)
2538 .or(cmd.venue_order_id);
2539
2540 match route {
2541 OrderCommandRoute::RegularWs => self.cancel_ws_order(&cmd),
2542 OrderCommandRoute::AlgoHttp => self.cancel_algo_order(&cmd),
2543 OrderCommandRoute::SpreadHttp => self.cancel_order_http(&cmd),
2544 }
2545 Ok(())
2546 }
2547
2548 fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
2549 match self.cancel_all_orders_route(cmd.instrument_id, cmd.order_side) {
2550 CancelAllOrdersRoute::SpreadHttp | CancelAllOrdersRoute::MassCancelHttp => {
2551 self.mass_cancel_instrument(cmd.instrument_id);
2552 Ok(())
2553 }
2554 CancelAllOrdersRoute::Batch => {
2555 let cancels: Vec<_> = {
2556 let cache = self.core.cache();
2557 cache
2558 .orders_open(None, Some(&cmd.instrument_id), None, None, cmd.order_side)
2559 .iter()
2560 .map(|order| CancelOrder {
2561 trader_id: order.trader_id(),
2562 client_id: cmd.client_id,
2563 strategy_id: order.strategy_id(),
2564 instrument_id: order.instrument_id(),
2565 client_order_id: order.client_order_id(),
2566 venue_order_id: order.venue_order_id(),
2567 command_id: cmd.command_id,
2568 ts_init: cmd.ts_init,
2569 params: cmd.params.clone(),
2570 correlation_id: cmd.correlation_id,
2571 causation_id: cmd.causation_id,
2572 })
2573 .collect()
2574 };
2575
2576 if cancels.is_empty() {
2577 log::debug!("No open orders to cancel for {}", cmd.instrument_id);
2578 return Ok(());
2579 }
2580
2581 self.batch_cancel_orders(BatchCancelOrders {
2582 trader_id: cmd.trader_id,
2583 client_id: cmd.client_id,
2584 strategy_id: cmd.strategy_id,
2585 instrument_id: cmd.instrument_id,
2586 cancels,
2587 command_id: cmd.command_id,
2588 ts_init: cmd.ts_init,
2589 params: cmd.params,
2590 correlation_id: cmd.correlation_id,
2591 causation_id: cmd.causation_id,
2592 })
2593 }
2594 }
2595 }
2596
2597 fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
2598 let cache = self.core.cache();
2599
2600 let mut regular_payload = Vec::new();
2601 let mut algo_orders = Vec::new();
2602 let mut http_orders = Vec::new();
2603
2604 for cancel in &cmd.cancels {
2605 let order_state = cache
2606 .order(&cancel.client_order_id)
2607 .map(|order| (order.order_type(), order.is_triggered()));
2608
2609 let venue_binding = self
2610 .ws_dispatch_state
2611 .order_venue_binding(cancel.client_order_id);
2612 let authoritative_venue_order_id = venue_binding
2613 .map(|(venue_order_id, _)| venue_order_id)
2614 .or(cancel.venue_order_id);
2615 match self.cancel_order_route(
2616 cancel.instrument_id,
2617 order_state,
2618 venue_binding.is_some_and(|(_, has_bound_child)| has_bound_child),
2619 ) {
2620 OrderCommandRoute::RegularWs => {
2621 self.ensure_order_identity(
2622 cancel.client_order_id,
2623 cancel.strategy_id,
2624 cancel.instrument_id,
2625 );
2626 regular_payload.push((
2627 cancel.instrument_id,
2628 Some(cancel.client_order_id),
2629 authoritative_venue_order_id,
2630 ));
2631 }
2632 OrderCommandRoute::AlgoHttp => {
2633 let mut cancel = cancel.clone();
2634 cancel.venue_order_id = authoritative_venue_order_id;
2635 algo_orders.push(cancel);
2636 }
2637 OrderCommandRoute::SpreadHttp => {
2638 self.ensure_order_identity(
2639 cancel.client_order_id,
2640 cancel.strategy_id,
2641 cancel.instrument_id,
2642 );
2643 http_orders.push((
2644 cancel.client_order_id,
2645 cancel.instrument_id,
2646 cancel.strategy_id,
2647 authoritative_venue_order_id,
2648 ));
2649 }
2650 }
2651 }
2652 drop(cache);
2653
2654 if !regular_payload.is_empty() {
2655 let ws_private = self.ws_private.clone();
2656 self.spawn_task("batch_cancel_orders", async move {
2657 for batch in regular_payload.chunks(20) {
2658 if let Err(e) = ws_private.batch_cancel_orders(batch.to_vec()).await {
2659 log_batch_cancel_failure(classify_okx_ws_failure(&e), batch.len());
2660 return Err(anyhow::Error::new(e).context("batch cancel orders failed"));
2661 }
2662 }
2663 Ok(())
2664 });
2665 }
2666
2667 if !algo_orders.is_empty() {
2669 let items: Vec<_> = algo_orders
2670 .into_iter()
2671 .map(|cancel| {
2672 let request = OKXCancelAlgoOrderRequest {
2673 inst_id: cancel.instrument_id.symbol.to_string(),
2674 inst_id_code: None,
2675 algo_id: cancel.venue_order_id.map(|id| id.to_string()),
2676 algo_cl_ord_id: if cancel.venue_order_id.is_none() {
2677 Some(cancel.client_order_id.to_string())
2678 } else {
2679 None
2680 },
2681 };
2682 let ctx = AlgoCancelContext {
2683 client_order_id: cancel.client_order_id,
2684 instrument_id: cancel.instrument_id,
2685 strategy_id: cancel.strategy_id,
2686 venue_order_id: cancel.venue_order_id,
2687 };
2688 (request, ctx)
2689 })
2690 .collect();
2691 self.dispatch_algo_cancels(items);
2692 }
2693
2694 if !http_orders.is_empty() {
2695 let client = self.http_client.clone();
2696 let emitter = self.emitter.clone();
2697 let clock = self.clock;
2698
2699 self.spawn_task("cancel_http_orders", async move {
2700 for (client_order_id, instrument_id, strategy_id, venue_order_id) in http_orders {
2701 if let Err(e) = client
2702 .cancel_order(instrument_id, Some(client_order_id), venue_order_id)
2703 .await
2704 {
2705 emit_cancel_failure(
2706 classify_okx_http_failure(&e),
2707 Some((&emitter, clock)),
2708 client_order_id,
2709 instrument_id,
2710 strategy_id,
2711 venue_order_id,
2712 );
2713 }
2714 }
2715 Ok(())
2716 });
2717 }
2718
2719 Ok(())
2720 }
2721}
2722
2723const MAX_TRIGGERED_CHILD_RECOVERIES: usize = 100;
2724
2725struct OrderReportSweep {
2726 reports: Vec<OrderStatusReport>,
2727 complete: bool,
2728 ambiguous_triggered_child_ids: AHashSet<VenueOrderId>,
2729 regular_by_venue_order_id: AHashMap<VenueOrderId, OrderStatusReport>,
2730}
2731
2732impl OKXExecutionClient {
2733 fn add_flat_derivative_position_reports(
2734 &self,
2735 order_reports: &[OrderStatusReport],
2736 fill_reports: &[FillReport],
2737 position_reports: &mut Vec<PositionStatusReport>,
2738 ts_init: UnixNanos,
2739 ) {
2740 let mut instrument_ids: AHashSet<InstrumentId> = order_reports
2741 .iter()
2742 .map(|report| report.instrument_id)
2743 .chain(fill_reports.iter().map(|report| report.instrument_id))
2744 .collect();
2745 let mut hedging_instrument_ids = AHashSet::new();
2746
2747 {
2748 let cache = self.core.cache();
2749 for position in cache.positions_open(
2750 Some(&OKX_VENUE),
2751 None,
2752 None,
2753 Some(&self.core.account_id),
2754 None,
2755 ) {
2756 instrument_ids.insert(position.instrument_id);
2757 if cache.oms_type(&position.id) == Some(OmsType::Hedging) {
2758 hedging_instrument_ids.insert(position.instrument_id);
2759 }
2760 }
2761 }
2762
2763 instrument_ids.retain(|instrument_id| {
2764 !is_spread_instrument(*instrument_id)
2765 && matches!(
2766 okx_instrument_type_from_symbol(instrument_id.symbol.as_str()),
2767 OKXInstrumentType::Swap
2768 | OKXInstrumentType::Futures
2769 | OKXInstrumentType::Option
2770 )
2771 && !hedging_instrument_ids.contains(instrument_id)
2772 && !position_reports
2773 .iter()
2774 .any(|report| report.instrument_id == *instrument_id)
2775 });
2776
2777 for instrument_id in instrument_ids {
2778 position_reports.push(PositionStatusReport::new(
2780 self.core.account_id,
2781 instrument_id,
2782 PositionSide::Flat,
2783 Quantity::zero(0),
2784 ts_init,
2785 ts_init,
2786 None,
2787 None,
2788 None,
2789 ));
2790 }
2791 }
2792
2793 async fn recover_triggered_child_order_reports(
2794 &self,
2795 reports: &mut Vec<OrderStatusReport>,
2796 ambiguous_triggered_child_ids: &AHashSet<VenueOrderId>,
2797 regular_by_venue_order_id: &AHashMap<VenueOrderId, OrderStatusReport>,
2798 ) -> bool {
2799 let mut recovered = Vec::with_capacity(reports.len());
2800 let recovery_candidate_count = reports
2801 .iter()
2802 .filter(|report| {
2803 report.order_status == OrderStatus::Triggered
2804 && !ambiguous_triggered_child_ids.contains(&report.venue_order_id)
2805 })
2806 .count();
2807 let mut complete = true;
2808
2809 if recovery_candidate_count > MAX_TRIGGERED_CHILD_RECOVERIES {
2810 log::warn!(
2811 "Triggered child recovery hit {MAX_TRIGGERED_CHILD_RECOVERIES} request cap; omitting unresolved reports"
2812 );
2813 }
2814
2815 let mut request_count = 0;
2816
2817 for mut parent_report in reports.drain(..) {
2818 if parent_report.order_status != OrderStatus::Triggered {
2819 recovered.push(parent_report);
2820 continue;
2821 }
2822
2823 if ambiguous_triggered_child_ids.contains(&parent_report.venue_order_id) {
2824 log::warn!(
2825 "Omitting triggered algo order with ambiguous child identifiers: {} {}",
2826 parent_report.instrument_id,
2827 parent_report.venue_order_id,
2828 );
2829 self.push_external_regular_order_fallback(
2830 &parent_report,
2831 regular_by_venue_order_id,
2832 &mut recovered,
2833 );
2834 complete = false;
2835 continue;
2836 }
2837
2838 if request_count >= MAX_TRIGGERED_CHILD_RECOVERIES {
2839 self.push_external_regular_order_fallback(
2840 &parent_report,
2841 regular_by_venue_order_id,
2842 &mut recovered,
2843 );
2844 complete = false;
2845 continue;
2846 }
2847 request_count += 1;
2848
2849 match self
2850 .http_client
2851 .request_order_status_report_by_venue_order_id(
2852 self.core.account_id,
2853 parent_report.instrument_id,
2854 parent_report.venue_order_id,
2855 )
2856 .await
2857 {
2858 Ok(Some(mut child_report)) => {
2859 if child_report.order_status == OrderStatus::Accepted {
2860 parent_report.quantity = child_report.quantity;
2861 parent_report.price = child_report.price.or(parent_report.price);
2862 parent_report.reduce_only |= child_report.reduce_only;
2863 parent_report.ts_last = child_report.ts_last;
2864 recovered.push(parent_report);
2865 } else {
2866 child_report.client_order_id = parent_report.client_order_id;
2867 recovered.push(child_report);
2868 }
2869 }
2870 Ok(None) => {
2871 log::warn!(
2872 "Triggered child order {} {} was not found",
2873 parent_report.instrument_id,
2874 parent_report.venue_order_id,
2875 );
2876 self.push_external_regular_order_fallback(
2877 &parent_report,
2878 regular_by_venue_order_id,
2879 &mut recovered,
2880 );
2881 complete = false;
2882 }
2883 Err(e) => {
2884 log::warn!(
2885 "Failed to recover triggered child order {} {}: {e}",
2886 parent_report.instrument_id,
2887 parent_report.venue_order_id,
2888 );
2889 self.push_external_regular_order_fallback(
2890 &parent_report,
2891 regular_by_venue_order_id,
2892 &mut recovered,
2893 );
2894 complete = false;
2895 }
2896 }
2897 }
2898
2899 *reports = recovered;
2900 complete
2901 }
2902
2903 fn push_external_regular_order_fallback(
2904 &self,
2905 parent_report: &OrderStatusReport,
2906 regular_by_venue_order_id: &AHashMap<VenueOrderId, OrderStatusReport>,
2907 recovered: &mut Vec<OrderStatusReport>,
2908 ) {
2909 let cache = self.core.cache();
2910 let cached_by_client = parent_report
2911 .client_order_id
2912 .is_some_and(|client_order_id| cache.order_exists(&client_order_id));
2913 let cached_by_venue = cache
2914 .client_order_id(&parent_report.venue_order_id)
2915 .is_some_and(|client_order_id| cache.order_exists(client_order_id));
2916
2917 if !cached_by_client
2918 && !cached_by_venue
2919 && let Some(regular_report) =
2920 regular_by_venue_order_id.get(&parent_report.venue_order_id)
2921 {
2922 recovered.push(regular_report.clone());
2923 }
2924 }
2925}
2926
2927fn validate_order(
2928 order: &impl Order,
2929 trade_mode: OKXTradeMode,
2930 submission: OrderSubmission,
2931) -> Result<(), OrderDeniedReason> {
2932 if let Err(detail) = validate_okx_client_order_id(order.client_order_id().as_str()) {
2933 return Err(OrderDeniedReason::InvalidClientOrderId { detail });
2934 }
2935
2936 if is_spread_instrument(order.instrument_id()) && order.is_reduce_only() {
2937 return Err(OrderDeniedReason::UnsupportedReduceOnly);
2938 }
2939
2940 if order.is_reduce_only()
2941 && okx_reduce_only_wire_value(
2942 okx_instrument_type_from_symbol(order.instrument_id().symbol.as_str()),
2943 trade_mode,
2944 order.order_side(),
2945 None,
2946 Some(true),
2947 )
2948 .is_err()
2949 {
2950 return Err(OrderDeniedReason::UnsupportedReduceOnly);
2951 }
2952
2953 if matches!(submission, OrderSubmission::List) {
2954 if OKX_CONDITIONAL_ORDER_TYPES.contains(&order.order_type()) {
2955 return Err(OrderDeniedReason::UnsupportedOrderList {
2956 detail: format!(
2957 "conditional order {} is not supported",
2958 order.client_order_id()
2959 ),
2960 });
2961 }
2962
2963 if order.time_in_force() != TimeInForce::Gtc {
2964 return Err(OrderDeniedReason::UnsupportedOrderList {
2965 detail: format!(
2966 "order {} has unsupported time in force {}",
2967 order.client_order_id(),
2968 order.time_in_force()
2969 ),
2970 });
2971 }
2972 }
2973
2974 Ok(())
2975}
2976
2977#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2978enum OrderCommandRoute {
2979 RegularWs,
2980 AlgoHttp,
2981 SpreadHttp,
2982}
2983
2984#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2985enum QueryOrderRoute {
2986 Regular,
2987 Algo,
2988 RegularThenAlgo,
2989 RegularAndAlgo,
2990 Spread,
2991}
2992
2993#[derive(Debug, Clone, Copy)]
2994struct CachedQueryOrderState {
2995 order_type: OrderType,
2996 venue_order_id: Option<VenueOrderId>,
2997}
2998
2999#[derive(Debug, Clone, Copy, PartialEq, Eq)]
3000enum CancelAllOrdersRoute {
3001 Batch,
3002 MassCancelHttp,
3003 SpreadHttp,
3004}
3005
3006#[derive(Clone, Copy)]
3007enum OrderSubmission {
3008 Single,
3009 List,
3010}
3011
3012fn emit_submit_failure(
3013 failure: CommandFailure,
3014 emitter: &ExecutionEventEmitter,
3015 clock: &'static AtomicTime,
3016 strategy_id: StrategyId,
3017 instrument_id: InstrumentId,
3018 client_order_id: ClientOrderId,
3019) {
3020 match failure {
3021 CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
3022 emitter.emit_order_rejected_event(
3023 strategy_id,
3024 instrument_id,
3025 client_order_id,
3026 &reason,
3027 clock.get_time_ns(),
3028 false,
3029 );
3030 }
3031 CommandFailure::Ambiguous(reason) => {
3032 log::warn!(
3033 "Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {reason}"
3034 );
3035 }
3036 }
3037}
3038
3039fn emit_modify_failure(
3040 failure: CommandFailure,
3041 emitter: &ExecutionEventEmitter,
3042 clock: &'static AtomicTime,
3043 strategy_id: StrategyId,
3044 instrument_id: InstrumentId,
3045 client_order_id: ClientOrderId,
3046 venue_order_id: Option<VenueOrderId>,
3047) {
3048 match failure {
3049 CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason) => {
3050 emitter.emit_order_modify_rejected_event(
3051 strategy_id,
3052 instrument_id,
3053 client_order_id,
3054 venue_order_id,
3055 &reason,
3056 clock.get_time_ns(),
3057 );
3058 }
3059 CommandFailure::Ambiguous(reason) => {
3060 log::warn!(
3061 "Ambiguous modify failure for {client_order_id}, awaiting reconciliation: {reason}"
3062 );
3063 }
3064 }
3065}
3066
3067fn emit_cancel_failure(
3068 failure: CommandFailure,
3069 emit_venue: Option<(&ExecutionEventEmitter, &'static AtomicTime)>,
3070 client_order_id: ClientOrderId,
3071 instrument_id: InstrumentId,
3072 strategy_id: StrategyId,
3073 venue_order_id: Option<VenueOrderId>,
3074) {
3075 match failure {
3076 CommandFailure::VenueRejected(reason) => {
3077 if let Some((emitter, clock)) = emit_venue {
3078 emitter.emit_order_cancel_rejected_event(
3079 strategy_id,
3080 instrument_id,
3081 client_order_id,
3082 venue_order_id,
3083 &reason,
3084 clock.get_time_ns(),
3085 );
3086 } else {
3087 log::warn!(
3088 "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
3089 );
3090 }
3091 }
3092 CommandFailure::NotSent(reason) => {
3093 log::warn!("Cancel command failed local validation for {client_order_id}: {reason}");
3094 }
3095 CommandFailure::Ambiguous(reason) => {
3096 log::warn!(
3097 "Ambiguous cancel failure for {client_order_id}, awaiting reconciliation: {reason}"
3098 );
3099 }
3100 }
3101}
3102
3103fn log_batch_cancel_failure(failure: CommandFailure, order_count: usize) {
3104 match failure {
3105 CommandFailure::NotSent(reason) => {
3106 log::warn!(
3107 "Batch cancel command failed local validation for {order_count} orders: {reason}"
3108 );
3109 }
3110 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
3111 log::warn!(
3112 "Ambiguous batch cancel failure for {order_count} orders, awaiting reconciliation: {reason}"
3113 );
3114 }
3115 }
3116}
3117
3118fn log_mass_cancel_failure(failure: CommandFailure, instrument_id: InstrumentId) {
3119 match failure {
3120 CommandFailure::NotSent(reason) => {
3121 log::warn!("Mass cancel command failed local validation for {instrument_id}: {reason}");
3122 }
3123 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
3124 log::warn!(
3125 "Ambiguous mass cancel failure for {instrument_id}, awaiting reconciliation: {reason}"
3126 );
3127 }
3128 }
3129}
3130
3131fn log_algo_batch_cancel_failure(failure: CommandFailure, contexts: &[AlgoCancelContext]) {
3132 match failure {
3133 CommandFailure::NotSent(reason) => {
3134 for ctx in contexts {
3135 log::warn!(
3136 "Algo batch cancel command failed local validation for {}: {reason}",
3137 ctx.client_order_id
3138 );
3139 }
3140 }
3141 CommandFailure::Ambiguous(reason) | CommandFailure::VenueRejected(reason) => {
3142 for ctx in contexts {
3143 log::warn!(
3144 "Ambiguous algo batch cancel failure for {}, awaiting reconciliation: {reason}",
3145 ctx.client_order_id
3146 );
3147 }
3148 }
3149 }
3150}
3151
3152fn get_param_as_string(params: Option<&Params>, key: &str) -> Option<String> {
3153 params.and_then(|p| {
3154 p.get(key).and_then(|v| {
3155 v.as_str()
3156 .map(ToString::to_string)
3157 .or_else(|| v.as_f64().map(|n| n.to_string()))
3158 })
3159 })
3160}
3161
3162fn get_param_as_bool(params: Option<&Params>, key: &str) -> Option<bool> {
3163 params.and_then(|params| params.get_bool(key))
3164}
3165
3166fn supports_algo_orders(instrument_type: OKXInstrumentType) -> bool {
3167 !matches!(
3168 instrument_type,
3169 OKXInstrumentType::Option | OKXInstrumentType::Events
3170 )
3171}
3172
3173fn order_routing_instrument_types(
3174 instrument_types: &[OKXInstrumentType],
3175) -> Vec<OKXInstrumentType> {
3176 let mut routing_types = instrument_types.to_vec();
3177
3178 if routing_types.contains(&OKXInstrumentType::Margin)
3180 && !routing_types.contains(&OKXInstrumentType::Spot)
3181 && !routing_types.contains(&OKXInstrumentType::Any)
3182 {
3183 routing_types.push(OKXInstrumentType::Spot);
3184 }
3185
3186 routing_types
3187}
3188
3189fn query_order_route(
3190 instrument_id: InstrumentId,
3191 order_type: Option<OrderType>,
3192 has_cached_venue_order_id: bool,
3193) -> QueryOrderRoute {
3194 if is_spread_instrument(instrument_id) {
3195 return QueryOrderRoute::Spread;
3196 }
3197
3198 let supports_algo = supports_algo_orders(okx_instrument_type_from_symbol(
3199 instrument_id.symbol.as_str(),
3200 ));
3201
3202 match order_type {
3203 Some(order_type)
3204 if supports_algo
3205 && OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
3206 && has_cached_venue_order_id =>
3207 {
3208 QueryOrderRoute::RegularAndAlgo
3209 }
3210 Some(order_type) if supports_algo && OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type) => {
3211 QueryOrderRoute::Algo
3212 }
3213 None if supports_algo => QueryOrderRoute::RegularThenAlgo,
3214 _ => QueryOrderRoute::Regular,
3215 }
3216}
3217
3218fn is_spread_instrument(instrument_id: InstrumentId) -> bool {
3219 is_okx_spread_symbol(instrument_id.symbol.as_str())
3220}
3221
3222fn is_instrument_cache_miss(error: &anyhow::Error) -> bool {
3223 error
3224 .chain()
3225 .any(|cause| cause.to_string().contains("missing from cache"))
3226}
3227
3228fn merge_algo_order_status_reports(
3229 reports: &mut Vec<OrderStatusReport>,
3230 sweep: AlgoOrderReportSweep,
3231 ambiguous_triggered_child_ids: &mut AHashSet<VenueOrderId>,
3232 complete: &mut bool,
3233) {
3234 let AlgoOrderReportSweep {
3235 reports: incoming,
3236 complete: sweep_complete,
3237 ambiguous_triggered_child_ids: ambiguous,
3238 } = sweep;
3239
3240 *complete &= sweep_complete && ambiguous.is_empty();
3241 ambiguous_triggered_child_ids.extend(ambiguous);
3242 merge_order_status_reports(reports, incoming);
3243}
3244
3245fn merge_order_status_reports(
3246 reports: &mut Vec<OrderStatusReport>,
3247 incoming: Vec<OrderStatusReport>,
3248) {
3249 let mut indexes: AHashMap<VenueOrderId, usize> = reports
3250 .iter()
3251 .enumerate()
3252 .map(|(index, report)| (report.venue_order_id, index))
3253 .collect();
3254
3255 for report in incoming {
3256 if let Some(index) = indexes.get(&report.venue_order_id).copied() {
3257 if is_order_status_report_more_advanced(&report, &reports[index]) {
3258 reports[index] = report;
3259 }
3260 } else {
3261 indexes.insert(report.venue_order_id, reports.len());
3262 reports.push(report);
3263 }
3264 }
3265}
3266
3267fn select_query_order_report(
3278 reports: Vec<OrderStatusReport>,
3279 client_order_id: ClientOrderId,
3280 venue_order_id: Option<VenueOrderId>,
3281) -> Option<OrderStatusReport> {
3282 let mut by_client_id: Option<OrderStatusReport> = None;
3283 let mut by_vid: Option<OrderStatusReport> = None;
3284
3285 for report in reports {
3286 if report.client_order_id == Some(client_order_id) {
3287 if by_client_id
3288 .as_ref()
3289 .is_none_or(|current| is_order_status_report_more_advanced(&report, current))
3290 {
3291 by_client_id = Some(report);
3292 }
3293
3294 continue;
3295 }
3296
3297 if venue_order_id
3298 .as_ref()
3299 .is_some_and(|vid| report.venue_order_id.as_str() == vid.as_str())
3300 && by_vid
3301 .as_ref()
3302 .is_none_or(|current| is_order_status_report_more_advanced(&report, current))
3303 {
3304 by_vid = Some(report);
3305 }
3306 }
3307
3308 by_client_id.or(by_vid)
3309}
3310
3311#[cfg(test)]
3312mod tests {
3313 use std::{cell::RefCell, rc::Rc, time::Duration};
3314
3315 use axum::{
3316 Json, Router,
3317 extract::ws::{Message, WebSocketUpgrade},
3318 routing::{get, post},
3319 };
3320 use nautilus_common::{cache::Cache, messages::ExecutionEvent, testing::wait_until_async};
3321 use nautilus_core::UUID4;
3322 use nautilus_model::{
3323 enums::{OrderSide, OrderStatus},
3324 events::OrderEventAny,
3325 identifiers::TraderId,
3326 instruments::Instrument,
3327 orders::{OrderTestBuilder, stubs::TestOrderEventStubs},
3328 types::{Price, Quantity},
3329 };
3330 use rstest::rstest;
3331 use serde_json::Value;
3332
3333 use super::*;
3334 use crate::common::consts::OKX_CLIENT_ID;
3335
3336 struct DropSignal(Option<tokio::sync::oneshot::Sender<()>>);
3337
3338 impl Drop for DropSignal {
3339 fn drop(&mut self) {
3340 if let Some(sender) = self.0.take() {
3341 let _ = sender.send(());
3342 }
3343 }
3344 }
3345
3346 #[derive(Clone, Copy)]
3347 enum ExecutionTaskBoundary {
3348 Reset,
3349 Dispose,
3350 RepeatedStop,
3351 }
3352
3353 #[rstest]
3354 #[case(OrderType::Market, QueryOrderRoute::Regular)]
3355 #[case(OrderType::Limit, QueryOrderRoute::Regular)]
3356 #[case(OrderType::StopMarket, QueryOrderRoute::Algo)]
3357 #[case(OrderType::TrailingStopMarket, QueryOrderRoute::Algo)]
3358 fn test_query_order_route_for_known_order_type(
3359 #[case] order_type: OrderType,
3360 #[case] expected: QueryOrderRoute,
3361 ) {
3362 assert_eq!(
3363 query_order_route(InstrumentId::from("BTC-USDT.OKX"), Some(order_type), false,),
3364 expected
3365 );
3366 }
3367
3368 #[rstest]
3369 fn test_query_order_route_for_conditional_order_with_cached_venue_id() {
3370 assert_eq!(
3371 query_order_route(
3372 InstrumentId::from("BTC-USDT.OKX"),
3373 Some(OrderType::StopMarket),
3374 true,
3375 ),
3376 QueryOrderRoute::RegularAndAlgo,
3377 );
3378 }
3379
3380 #[rstest]
3381 fn test_query_order_route_for_unknown_order_type() {
3382 assert_eq!(
3383 query_order_route(InstrumentId::from("BTC-USDT.OKX"), None, false),
3384 QueryOrderRoute::RegularThenAlgo,
3385 );
3386 }
3387
3388 #[rstest]
3389 fn test_query_order_route_for_spread() {
3390 assert_eq!(
3391 query_order_route(
3392 InstrumentId::from("ETH-USD-SWAP_ETH-USD-231229.OKX"),
3393 None,
3394 false,
3395 ),
3396 QueryOrderRoute::Spread,
3397 );
3398 }
3399
3400 #[rstest]
3401 fn test_validate_order_allows_conditional_single_submission() {
3402 let order = OrderTestBuilder::new(OrderType::StopMarket)
3403 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
3404 .client_order_id(ClientOrderId::from("OCONDITIONALSINGLE"))
3405 .side(OrderSide::Sell)
3406 .quantity(Quantity::from("1"))
3407 .trigger_price(Price::from("1000.00"))
3408 .build();
3409
3410 assert_eq!(
3411 validate_order(&order, OKXTradeMode::Cross, OrderSubmission::Single),
3412 Ok(())
3413 );
3414 }
3415
3416 #[rstest]
3417 fn test_validate_order_denies_conditional_order_in_list() {
3418 let order = OrderTestBuilder::new(OrderType::StopMarket)
3419 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
3420 .client_order_id(ClientOrderId::from("OCONDITIONALLIST"))
3421 .side(OrderSide::Sell)
3422 .quantity(Quantity::from("2"))
3423 .trigger_price(Price::from("900.00"))
3424 .build();
3425
3426 assert_eq!(
3427 validate_order(&order, OKXTradeMode::Cross, OrderSubmission::List),
3428 Err(OrderDeniedReason::UnsupportedOrderList {
3429 detail: "conditional order OCONDITIONALLIST is not supported".to_string(),
3430 })
3431 );
3432 }
3433
3434 #[rstest]
3435 fn test_validate_order_denies_non_gtc_order_in_list() {
3436 let order = OrderTestBuilder::new(OrderType::Limit)
3437 .instrument_id(InstrumentId::from("ETH-USDT-SWAP.OKX"))
3438 .client_order_id(ClientOrderId::from("OIOCLIST"))
3439 .side(OrderSide::Buy)
3440 .price(Price::from("2000.00"))
3441 .quantity(Quantity::from("3"))
3442 .time_in_force(TimeInForce::Ioc)
3443 .build();
3444
3445 assert_eq!(
3446 validate_order(&order, OKXTradeMode::Cross, OrderSubmission::List),
3447 Err(OrderDeniedReason::UnsupportedOrderList {
3448 detail: "order OIOCLIST has unsupported time in force IOC".to_string(),
3449 })
3450 );
3451 }
3452
3453 #[rstest]
3454 #[case::cash("BTC-USDT.OKX", OKXTradeMode::Cash)]
3455 #[case::option("BTC-USD-241217-92000-C.OKX", OKXTradeMode::Cross)]
3456 #[case::event("BTC-ABOVE-DAILY-260224-1600-65000.OKX", OKXTradeMode::Cross)]
3457 fn test_validate_order_denies_unsupported_reduce_only(
3458 #[case] instrument_id: &str,
3459 #[case] trade_mode: OKXTradeMode,
3460 ) {
3461 let order = OrderTestBuilder::new(OrderType::Limit)
3462 .instrument_id(InstrumentId::from(instrument_id))
3463 .client_order_id(ClientOrderId::from("OREDUCEUNSUPPORTED"))
3464 .side(OrderSide::Sell)
3465 .price(Price::from("2000.00"))
3466 .quantity(Quantity::from("1"))
3467 .reduce_only(true)
3468 .build();
3469
3470 assert_eq!(
3471 validate_order(&order, trade_mode, OrderSubmission::Single),
3472 Err(OrderDeniedReason::UnsupportedReduceOnly)
3473 );
3474 }
3475
3476 fn build_config(
3477 margin_mode: Option<OKXMarginMode>,
3478 use_spot_margin: bool,
3479 ) -> OKXExecutionClientConfig {
3480 OKXExecutionClientConfig {
3481 margin_mode,
3482 use_spot_margin,
3483 ..OKXExecutionClientConfig::default()
3484 }
3485 }
3486
3487 #[rstest]
3488 #[case::spot(OKXInstrumentType::Spot, true)]
3489 #[case::margin(OKXInstrumentType::Margin, true)]
3490 #[case::swap(OKXInstrumentType::Swap, true)]
3491 #[case::futures(OKXInstrumentType::Futures, true)]
3492 #[case::option(OKXInstrumentType::Option, false)]
3493 #[case::events(OKXInstrumentType::Events, false)]
3494 fn test_supports_algo_orders(
3495 #[case] instrument_type: OKXInstrumentType,
3496 #[case] expected: bool,
3497 ) {
3498 assert_eq!(supports_algo_orders(instrument_type), expected);
3499 }
3500
3501 #[rstest]
3502 #[case::margin(
3503 vec![OKXInstrumentType::Margin],
3504 vec![OKXInstrumentType::Margin, OKXInstrumentType::Spot]
3505 )]
3506 #[case::spot_margin(
3507 vec![OKXInstrumentType::Spot, OKXInstrumentType::Margin],
3508 vec![OKXInstrumentType::Spot, OKXInstrumentType::Margin]
3509 )]
3510 #[case::any_margin(
3511 vec![OKXInstrumentType::Any, OKXInstrumentType::Margin],
3512 vec![OKXInstrumentType::Any, OKXInstrumentType::Margin]
3513 )]
3514 #[case::swap(
3515 vec![OKXInstrumentType::Swap],
3516 vec![OKXInstrumentType::Swap]
3517 )]
3518 fn test_order_routing_instrument_types(
3519 #[case] instrument_types: Vec<OKXInstrumentType>,
3520 #[case] expected: Vec<OKXInstrumentType>,
3521 ) {
3522 assert_eq!(order_routing_instrument_types(&instrument_types), expected);
3523 }
3524
3525 #[rstest]
3526 #[case::cash_no_spot_margin(AccountType::Cash, None, false, OKXTradeMode::Cash)]
3527 #[case::cash_spot_margin_cross(
3528 AccountType::Cash,
3529 Some(OKXMarginMode::Cross),
3530 true,
3531 OKXTradeMode::Cross
3532 )]
3533 #[case::cash_spot_margin_isolated(
3534 AccountType::Cash,
3535 Some(OKXMarginMode::Isolated),
3536 true,
3537 OKXTradeMode::Isolated
3538 )]
3539 #[case::cash_spot_margin_none(AccountType::Cash, None, true, OKXTradeMode::Isolated)]
3540 #[case::margin_cross(
3541 AccountType::Margin,
3542 Some(OKXMarginMode::Cross),
3543 false,
3544 OKXTradeMode::Cross
3545 )]
3546 #[case::margin_isolated(
3547 AccountType::Margin,
3548 Some(OKXMarginMode::Isolated),
3549 false,
3550 OKXTradeMode::Isolated
3551 )]
3552 #[case::margin_none(AccountType::Margin, None, false, OKXTradeMode::Isolated)]
3553 fn test_derive_default_trade_mode(
3554 #[case] account_type: AccountType,
3555 #[case] margin_mode: Option<OKXMarginMode>,
3556 #[case] use_spot_margin: bool,
3557 #[case] expected: OKXTradeMode,
3558 ) {
3559 let config = build_config(margin_mode, use_spot_margin);
3560
3561 let result = OKXExecutionClient::derive_default_trade_mode(account_type, &config);
3562
3563 assert_eq!(result, expected);
3564 }
3565
3566 #[rstest]
3567 #[case::spot_no_margin("BTC-USDT", None, false, OKXTradeMode::Cash)]
3568 #[case::spot_cross_margin("BTC-USDT", Some(OKXMarginMode::Cross), true, OKXTradeMode::Cross)]
3569 #[case::spot_isolated_margin(
3570 "ETH-USDT",
3571 Some(OKXMarginMode::Isolated),
3572 true,
3573 OKXTradeMode::Isolated
3574 )]
3575 #[case::spot_margin_no_mode("BTC-USDT", None, true, OKXTradeMode::Isolated)]
3576 #[case::swap_cross(
3577 "BTC-USDT-SWAP",
3578 Some(OKXMarginMode::Cross),
3579 false,
3580 OKXTradeMode::Cross
3581 )]
3582 #[case::swap_isolated(
3583 "BTC-USDT-SWAP",
3584 Some(OKXMarginMode::Isolated),
3585 false,
3586 OKXTradeMode::Isolated
3587 )]
3588 #[case::swap_no_mode("ETH-USDT-SWAP", None, false, OKXTradeMode::Isolated)]
3589 #[case::futures_cross(
3590 "BTC-USDT-250328",
3591 Some(OKXMarginMode::Cross),
3592 false,
3593 OKXTradeMode::Cross
3594 )]
3595 #[case::futures_isolated("BTC-USDT-250328", None, false, OKXTradeMode::Isolated)]
3596 #[case::option_cross(
3597 "BTC-USD-250328-50000-C",
3598 Some(OKXMarginMode::Cross),
3599 false,
3600 OKXTradeMode::Cross
3601 )]
3602 #[case::option_isolated("BTC-USD-250328-50000-C", None, false, OKXTradeMode::Isolated)]
3603 fn test_derive_trade_mode_for_instrument(
3604 #[case] symbol: &str,
3605 #[case] margin_mode: Option<OKXMarginMode>,
3606 #[case] use_spot_margin: bool,
3607 #[case] expected: OKXTradeMode,
3608 ) {
3609 let instrument_id = InstrumentId::from(format!("{symbol}.OKX").as_str());
3610
3611 let result = derive_trade_mode_for_instrument(instrument_id, margin_mode, use_spot_margin);
3612
3613 assert_eq!(result, expected);
3614 }
3615
3616 #[rstest]
3617 #[case::override_to_cross("cross", OKXTradeMode::Cross)]
3618 #[case::override_to_cash("cash", OKXTradeMode::Cash)]
3619 #[case::override_to_isolated("isolated", OKXTradeMode::Isolated)]
3620 #[case::override_to_spot_isolated("spot_isolated", OKXTradeMode::SpotIsolated)]
3621 #[case::case_insensitive("CROSS", OKXTradeMode::Cross)]
3622 fn test_td_mode_param_override(#[case] td_mode_value: &str, #[case] expected: OKXTradeMode) {
3623 let mut params = Params::new();
3624 params.insert(
3625 "td_mode".to_string(),
3626 Value::String(td_mode_value.to_string()),
3627 );
3628
3629 let result = get_param_as_string(Some(¶ms), "td_mode")
3630 .and_then(|s| s.parse::<OKXTradeMode>().ok());
3631
3632 assert_eq!(result, Some(expected));
3633 }
3634
3635 #[rstest]
3636 fn test_td_mode_param_invalid_falls_through() {
3637 let mut params = Params::new();
3638 params.insert("td_mode".to_string(), Value::String("invalid".to_string()));
3639
3640 let result = get_param_as_string(Some(¶ms), "td_mode")
3641 .and_then(|s| s.parse::<OKXTradeMode>().ok());
3642
3643 assert_eq!(result, None);
3644 }
3645
3646 #[rstest]
3647 fn test_td_mode_param_absent_falls_through() {
3648 let result = get_param_as_string(None, "td_mode");
3649
3650 assert_eq!(result, None);
3651 }
3652
3653 #[rstest]
3654 fn test_close_fraction_present_sets_reduce_only_true() {
3655 let mut params = Params::new();
3656 params.insert("close_fraction".to_string(), Value::String("1".to_string()));
3657 let params = Some(params);
3658
3659 let close_fraction = get_param_as_string(params.as_ref(), "close_fraction");
3660 let is_reduce_only = false;
3661 let reduce_only = if close_fraction.is_some() {
3662 Some(true)
3663 } else {
3664 Some(is_reduce_only)
3665 };
3666
3667 assert_eq!(close_fraction, Some("1".to_string()));
3668 assert_eq!(reduce_only, Some(true));
3669 }
3670
3671 #[rstest]
3672 fn test_close_fraction_absent_preserves_reduce_only() {
3673 let params: Option<Params> = None;
3674
3675 let close_fraction = get_param_as_string(params.as_ref(), "close_fraction");
3676 let is_reduce_only = false;
3677 let reduce_only = if close_fraction.is_some() {
3678 Some(true)
3679 } else {
3680 Some(is_reduce_only)
3681 };
3682
3683 assert_eq!(close_fraction, None);
3684 assert_eq!(reduce_only, Some(false));
3685 }
3686
3687 #[rstest]
3688 fn test_close_fraction_absent_with_reduce_only_true() {
3689 let params: Option<Params> = None;
3690
3691 let close_fraction = get_param_as_string(params.as_ref(), "close_fraction");
3692 let is_reduce_only = true;
3693 let reduce_only = if close_fraction.is_some() {
3694 Some(true)
3695 } else {
3696 Some(is_reduce_only)
3697 };
3698
3699 assert_eq!(close_fraction, None);
3700 assert_eq!(reduce_only, Some(true));
3701 }
3702
3703 fn make_query_order_report(cid: Option<&str>, vid: &str) -> OrderStatusReport {
3704 OrderStatusReport::new(
3705 AccountId::from("OKX-001"),
3706 InstrumentId::from("BTC-USDT.OKX"),
3707 cid.map(ClientOrderId::from),
3708 VenueOrderId::from(vid),
3709 OrderSide::Buy.into(),
3710 OrderType::Limit,
3711 TimeInForce::Gtc,
3712 OrderStatus::Accepted,
3713 Quantity::new(1.0, 0),
3714 Quantity::zero(0),
3715 UnixNanos::default(),
3716 UnixNanos::default(),
3717 UnixNanos::default(),
3718 None,
3719 )
3720 }
3721
3722 fn with_linked(mut report: OrderStatusReport, linked: &[&str]) -> OrderStatusReport {
3723 report.linked_order_ids = Some(linked.iter().map(|s| ClientOrderId::from(*s)).collect());
3724 report
3725 }
3726
3727 #[rstest]
3728 fn test_merge_order_status_reports_keeps_filled_regular_child() {
3729 let mut filled = make_query_order_report(Some("O-PARENT"), "V-CHILD");
3730 filled.order_status = OrderStatus::Filled;
3731 filled.filled_qty = Quantity::new(1.0, 0);
3732 filled.ts_last = UnixNanos::from(100);
3733
3734 let mut triggered = make_query_order_report(Some("O-PARENT"), "V-CHILD");
3735 triggered.order_status = OrderStatus::Triggered;
3736 triggered.filled_qty = Quantity::new(1.0, 0);
3737 triggered.ts_last = UnixNanos::from(200);
3738
3739 let mut reports = vec![filled];
3740 merge_order_status_reports(&mut reports, vec![triggered]);
3741
3742 assert_eq!(reports.len(), 1);
3743 assert_eq!(reports[0].order_status, OrderStatus::Filled);
3744 assert_eq!(
3745 reports[0].client_order_id,
3746 Some(ClientOrderId::from("O-PARENT"))
3747 );
3748 assert_eq!(reports[0].venue_order_id, VenueOrderId::from("V-CHILD"));
3749 }
3750
3751 #[rstest]
3752 fn test_merge_order_status_reports_replaces_pending_parent_with_triggered_child() {
3753 let mut accepted = make_query_order_report(Some("O-PARENT"), "V-CHILD");
3754 accepted.ts_last = UnixNanos::from(100);
3755
3756 let mut triggered = make_query_order_report(Some("O-PARENT"), "V-CHILD");
3757 triggered.order_status = OrderStatus::Triggered;
3758 triggered.ts_last = UnixNanos::from(200);
3759
3760 let mut reports = vec![accepted];
3761 merge_order_status_reports(&mut reports, vec![triggered]);
3762
3763 assert_eq!(reports.len(), 1);
3764 assert_eq!(reports[0].order_status, OrderStatus::Triggered);
3765 assert_eq!(reports[0].ts_last, UnixNanos::from(200));
3766 }
3767
3768 #[rstest]
3769 fn test_select_query_order_report_matches_client_order_id() {
3770 let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
3771 let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
3772 assert_eq!(
3773 selected.and_then(|r| r.client_order_id),
3774 Some(ClientOrderId::from("O-001"))
3775 );
3776 }
3777
3778 #[rstest]
3779 fn test_select_query_order_report_client_wins_over_venue_mismatch() {
3780 let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
3781 let selected = select_query_order_report(
3782 reports,
3783 ClientOrderId::from("O-001"),
3784 Some(VenueOrderId::from("V-OTHER")),
3785 );
3786 assert_eq!(
3787 selected.and_then(|r| r.client_order_id),
3788 Some(ClientOrderId::from("O-001"))
3789 );
3790 }
3791
3792 #[rstest]
3793 #[case(
3794 OrderStatus::Accepted,
3795 Quantity::zero(0),
3796 OrderStatus::Triggered,
3797 Quantity::zero(0),
3798 OrderStatus::Triggered
3799 )]
3800 #[case(
3801 OrderStatus::Triggered,
3802 Quantity::zero(0),
3803 OrderStatus::PartiallyFilled,
3804 Quantity::new(0.5, 1),
3805 OrderStatus::PartiallyFilled
3806 )]
3807 #[case(
3808 OrderStatus::PartiallyFilled,
3809 Quantity::new(0.5, 1),
3810 OrderStatus::Filled,
3811 Quantity::new(1.0, 0),
3812 OrderStatus::Filled
3813 )]
3814 #[case(
3815 OrderStatus::Triggered,
3816 Quantity::zero(0),
3817 OrderStatus::Canceled,
3818 Quantity::zero(0),
3819 OrderStatus::Canceled
3820 )]
3821 #[case(
3822 OrderStatus::Triggered,
3823 Quantity::zero(0),
3824 OrderStatus::Rejected,
3825 Quantity::zero(0),
3826 OrderStatus::Rejected
3827 )]
3828 fn test_select_query_order_report_chooses_most_advanced_client_match_regardless_of_order(
3829 #[case] first_status: OrderStatus,
3830 #[case] first_filled_qty: Quantity,
3831 #[case] second_status: OrderStatus,
3832 #[case] second_filled_qty: Quantity,
3833 #[case] expected_status: OrderStatus,
3834 ) {
3835 let mut first = make_query_order_report(Some("O-001"), "V-PARENT");
3836 first.order_status = first_status;
3837 first.filled_qty = first_filled_qty;
3838 let mut second = make_query_order_report(Some("O-001"), "V-CHILD");
3839 second.order_status = second_status;
3840 second.filled_qty = second_filled_qty;
3841
3842 for reports in [vec![first.clone(), second.clone()], vec![second, first]] {
3843 let selected = select_query_order_report(
3844 reports,
3845 ClientOrderId::from("O-001"),
3846 Some(VenueOrderId::from("V-PARENT")),
3847 )
3848 .unwrap();
3849
3850 assert_eq!(selected.order_status, expected_status);
3851 }
3852 }
3853
3854 #[rstest]
3855 fn test_select_query_order_report_falls_back_to_venue_order_id() {
3856 let reports = vec![make_query_order_report(Some("O-CHILD"), "V-1")];
3859 let selected = select_query_order_report(
3860 reports,
3861 ClientOrderId::from("O-PARENT"),
3862 Some(VenueOrderId::from("V-1")),
3863 );
3864 assert_eq!(
3865 selected.map(|r| r.venue_order_id.as_str().to_string()),
3866 Some("V-1".to_string()),
3867 );
3868 }
3869
3870 #[rstest]
3871 fn test_select_query_order_report_rejects_when_nothing_matches() {
3872 let reports = vec![make_query_order_report(Some("O-OTHER"), "V-OTHER")];
3873 let selected = select_query_order_report(
3874 reports,
3875 ClientOrderId::from("O-001"),
3876 Some(VenueOrderId::from("V-1")),
3877 );
3878 assert!(selected.is_none());
3879 }
3880
3881 #[rstest]
3882 fn test_select_query_order_report_rejects_when_client_differs_and_no_vid_provided() {
3883 let reports = vec![make_query_order_report(Some("O-OTHER"), "V-1")];
3884 let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
3885 assert!(selected.is_none());
3886 }
3887
3888 #[rstest]
3889 fn test_select_query_order_report_ignores_linked_order_ids_for_parent_with_attached_tp() {
3890 let child_cid = "O-CHILD-TP";
3894 let reports = vec![with_linked(
3895 make_query_order_report(Some("O-PARENT"), "V-PARENT"),
3896 &[child_cid, "O-CHILD-SL"],
3897 )];
3898 let selected = select_query_order_report(reports, ClientOrderId::from(child_cid), None);
3899 assert!(selected.is_none());
3900 }
3901
3902 #[rstest]
3903 fn test_select_query_order_report_client_match_wins_over_vid_match_elsewhere() {
3904 let reports = vec![
3907 make_query_order_report(Some("O-OTHER"), "V-1"),
3908 make_query_order_report(Some("O-001"), "V-2"),
3909 ];
3910 let selected = select_query_order_report(
3911 reports,
3912 ClientOrderId::from("O-001"),
3913 Some(VenueOrderId::from("V-1")),
3914 );
3915 assert_eq!(
3916 selected.and_then(|r| r.client_order_id),
3917 Some(ClientOrderId::from("O-001")),
3918 );
3919 }
3920
3921 fn build_test_exec_client() -> OKXExecutionClient {
3922 build_test_exec_client_with_cache().0
3923 }
3924
3925 fn build_test_exec_client_with_cache() -> (OKXExecutionClient, Rc<RefCell<Cache>>) {
3926 let config = OKXExecutionClientConfig {
3927 api_key: Some("test_key".into()),
3928 api_secret: Some("test_secret".into()),
3929 api_passphrase: Some("test_pass".into()),
3930 ..OKXExecutionClientConfig::default()
3931 };
3932
3933 let cache = Rc::new(RefCell::new(Cache::default()));
3934 let core = ExecutionClientCore::new(
3935 TraderId::from("TESTER-001"),
3936 ClientId::from("OKX-TEST"),
3937 *OKX_VENUE,
3938 OmsType::Hedging,
3939 config.account_id,
3940 AccountType::Cash,
3941 None,
3942 Rc::clone(&cache),
3943 );
3944
3945 (
3946 OKXExecutionClient::new(core, config).expect("failed to build test client"),
3947 cache,
3948 )
3949 }
3950
3951 #[rstest]
3952 #[case::reset(ExecutionTaskBoundary::Reset)]
3953 #[case::dispose(ExecutionTaskBoundary::Dispose)]
3954 #[case::repeated_stop(ExecutionTaskBoundary::RepeatedStop)]
3955 #[tokio::test]
3956 async fn lifecycle_boundary_terminates_owned_execution_task(
3957 #[case] boundary: ExecutionTaskBoundary,
3958 ) {
3959 let mut client = build_test_exec_client();
3960
3961 if matches!(boundary, ExecutionTaskBoundary::RepeatedStop) {
3962 client.stop().expect("initial stop");
3963 }
3964
3965 let (drop_tx, drop_rx) = tokio::sync::oneshot::channel();
3966 let signal = DropSignal(Some(drop_tx));
3967 client.spawn_task("pending lifecycle task", async move {
3968 let _signal = signal;
3969 std::future::pending::<anyhow::Result<()>>().await
3970 });
3971
3972 match boundary {
3973 ExecutionTaskBoundary::Reset => client.reset().expect("reset"),
3974 ExecutionTaskBoundary::Dispose => client.dispose().expect("dispose"),
3975 ExecutionTaskBoundary::RepeatedStop => client.stop().expect("repeated stop"),
3976 }
3977
3978 tokio::time::timeout(Duration::from_secs(1), drop_rx)
3979 .await
3980 .expect("lifecycle boundary must drop the owned task")
3981 .expect("drop signal");
3982 terminate_tasks(&client.pending_tasks, "test execution client")
3983 .await
3984 .expect("execution task terminated");
3985 assert!(client.pending_tasks.is_empty());
3986 }
3987
3988 #[rstest]
3989 #[case("BTC-USDT.OKX", false, None, CancelAllOrdersRoute::Batch)]
3990 #[case("BTC-USDT.OKX", true, None, CancelAllOrdersRoute::MassCancelHttp)]
3991 #[case(
3992 "BTC-USDT.OKX",
3993 true,
3994 Some(OrderSide::Buy),
3995 CancelAllOrdersRoute::Batch
3996 )]
3997 #[case(
3998 "BTC-USDT.OKX",
3999 true,
4000 Some(OrderSide::Sell),
4001 CancelAllOrdersRoute::Batch
4002 )]
4003 #[case(
4004 "ETH-USD-SWAP_ETH-USD-231229.OKX",
4005 false,
4006 None,
4007 CancelAllOrdersRoute::SpreadHttp
4008 )]
4009 #[case(
4010 "ETH-USD-SWAP_ETH-USD-231229.OKX",
4011 false,
4012 Some(OrderSide::Buy),
4013 CancelAllOrdersRoute::Batch
4014 )]
4015 #[case(
4016 "ETH-USD-SWAP_ETH-USD-231229.OKX",
4017 true,
4018 Some(OrderSide::Sell),
4019 CancelAllOrdersRoute::Batch
4020 )]
4021 fn test_cancel_all_orders_route(
4022 #[case] instrument_id: &str,
4023 #[case] use_mm_mass_cancel: bool,
4024 #[case] order_side: Option<OrderSide>,
4025 #[case] expected: CancelAllOrdersRoute,
4026 ) {
4027 let mut client = build_test_exec_client();
4028 client.config.use_mm_mass_cancel = use_mm_mass_cancel;
4029
4030 assert_eq!(
4031 client.cancel_all_orders_route(InstrumentId::from(instrument_id), order_side),
4032 expected
4033 );
4034 }
4035
4036 #[rstest]
4037 #[case::buy(Some(OrderSide::Buy), false, &["OCANCEL0", "OCANCEL2"])]
4038 #[case::sell(Some(OrderSide::Sell), false, &["OCANCEL1", "OCANCEL3"])]
4039 #[case::both(None, false, &["OCANCEL0", "OCANCEL1", "OCANCEL2", "OCANCEL3"])]
4040 #[case::mass_buy(Some(OrderSide::Buy), true, &["OCANCEL0", "OCANCEL2"])]
4041 #[case::mass_sell(Some(OrderSide::Sell), true, &["OCANCEL1", "OCANCEL3"])]
4042 #[tokio::test]
4043 async fn test_cancel_all_orders_filters_cached_orders(
4044 #[case] order_side: Option<OrderSide>,
4045 #[case] use_mm_mass_cancel: bool,
4046 #[case] expected_ids: &[&str],
4047 #[values(1, 10, 11)] copies: usize,
4048 ) {
4049 let (mut client, cache) = build_test_exec_client_with_cache();
4050 client.config.use_mm_mass_cancel = use_mm_mass_cancel;
4051 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
4052
4053 let router = Router::new().route(
4054 "/ws",
4055 get(move |ws: WebSocketUpgrade| {
4056 let tx = request_tx.clone();
4057 async move {
4058 ws.on_upgrade(move |mut socket| async move {
4059 while let Some(Ok(Message::Text(text))) = socket.next().await {
4060 let request: Value = serde_json::from_str(&text).unwrap();
4061 if request["op"] == "login" {
4062 let response =
4063 serde_json::json!({"event": "login", "code": "0", "msg": "", "connId": "test-conn"});
4064 socket
4065 .send(Message::Text(response.to_string().into()))
4066 .await
4067 .unwrap();
4068 } else {
4069 tx.send(request).unwrap();
4070 }
4071 }
4072 })
4073 }
4074 }),
4075 );
4076
4077 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4078 let url = format!("ws://{}/ws", listener.local_addr().unwrap());
4079
4080 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
4081 client.ws_private = OKXWebSocketClient::with_credentials(
4082 Some(url),
4083 Some("test_key".into()),
4084 Some("test_secret".into()),
4085 Some("test_pass".into()),
4086 Some(client.core.account_id),
4087 None,
4088 None,
4089 client.config.transport_backend,
4090 None,
4091 )
4092 .unwrap();
4093 client
4094 .ws_private
4095 .cache_inst_id_code(ustr::Ustr::from("BTC-USDT"), 10459);
4096 client.ws_private.connect().await.unwrap();
4097 client.ws_private.wait_until_active(5.0).await.unwrap();
4098 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
4099
4100 let expected_ids: Vec<_> = (0..copies)
4101 .flat_map(|copy| {
4102 expected_ids.iter().map(move |id| {
4103 let index: usize = id.strip_prefix("OCANCEL").unwrap().parse().unwrap();
4104 format!("OCANCEL{}", copy * 8 + index)
4105 })
4106 })
4107 .collect();
4108
4109 let mut expected: Vec<_> = expected_ids
4110 .iter()
4111 .map(|id| ClientOrderId::from(id.as_str()))
4112 .collect();
4113
4114 for copy in 0..copies {
4115 for (index, (symbol, side, strategy, open)) in [
4116 ("BTC-USDT.OKX", OrderSide::Buy, "S-001", true),
4117 ("BTC-USDT.OKX", OrderSide::Sell, "S-001", true),
4118 ("BTC-USDT.OKX", OrderSide::Buy, "S-002", true),
4119 ("BTC-USDT.OKX", OrderSide::Sell, "S-002", true),
4120 ("ETH-USDT.OKX", OrderSide::Buy, "S-001", true),
4121 ("ETH-USDT.OKX", OrderSide::Sell, "S-001", true),
4122 ("BTC-USDT.OKX", OrderSide::Buy, "S-001", false),
4123 ("BTC-USDT.OKX", OrderSide::Sell, "S-001", false),
4124 ]
4125 .into_iter()
4126 .enumerate()
4127 {
4128 let index = copy * 8 + index;
4129 let client_order_id = ClientOrderId::from(format!("OCANCEL{index}"));
4130 let mut order = OrderTestBuilder::new(OrderType::Limit)
4131 .instrument_id(InstrumentId::from(symbol))
4132 .client_order_id(client_order_id)
4133 .strategy_id(StrategyId::from(strategy))
4134 .side(side)
4135 .price(Price::from("100.00"))
4136 .quantity(Quantity::from("1"))
4137 .build();
4138 let accepted = TestOrderEventStubs::accepted(
4139 &order,
4140 client.core.account_id,
4141 VenueOrderId::from(format!("{index}")),
4142 );
4143 cache
4144 .borrow_mut()
4145 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
4146 .unwrap();
4147 order = cache.borrow_mut().update_order(&accepted).unwrap();
4148
4149 if !open {
4150 let canceled = TestOrderEventStubs::canceled(
4151 &order,
4152 client.core.account_id,
4153 order.venue_order_id(),
4154 );
4155 cache.borrow_mut().update_order(&canceled).unwrap();
4156 }
4157 }
4158 }
4159
4160 client
4161 .cancel_all_orders(CancelAllOrders::new(
4162 TraderId::from("TESTER-001"),
4163 Some(*OKX_CLIENT_ID),
4164 StrategyId::from("S-001"),
4165 instrument_id,
4166 order_side,
4167 UUID4::new(),
4168 UnixNanos::default(),
4169 None,
4170 None,
4171 ))
4172 .unwrap();
4173 let mut actual: Vec<_> = client
4174 .ws_dispatch_state
4175 .order_identities
4176 .iter()
4177 .map(|entry| *entry.key())
4178 .collect();
4179 actual.sort();
4180
4181 let expected_sizes: &[usize] = match (copies, order_side) {
4182 (1, Some(_)) => &[2],
4183 (1, None) => &[4],
4184 (10, Some(_)) => &[20],
4185 (10, None) => &[20, 20],
4186 (11, Some(_)) => &[20, 2],
4187 (11, None) => &[20, 20, 4],
4188 _ => unreachable!(),
4189 };
4190
4191 let mut payload = Vec::new();
4192 let mut batch_sizes = Vec::new();
4193
4194 for _ in expected_sizes {
4195 let request = tokio::time::timeout(Duration::from_secs(5), request_rx.recv())
4196 .await
4197 .unwrap()
4198 .unwrap();
4199 assert_eq!(request["op"], "batch-cancel-orders");
4200 let args = request["args"].as_array().unwrap();
4201 batch_sizes.push(args.len());
4202 payload.extend(args.iter().cloned());
4203 }
4204
4205 payload.sort_by_key(|item| item["clOrdId"].as_str().unwrap().to_string());
4206
4207 let mut expected_payload: Vec<_> = expected_ids
4208 .iter()
4209 .map(|id| {
4210 serde_json::json!({
4211 "instIdCode": 10459,
4212 "clOrdId": id,
4213 "ordId": id.strip_prefix("OCANCEL").unwrap(),
4214 })
4215 })
4216 .collect();
4217
4218 expected.sort();
4219 expected_payload.sort_by_key(|item| item["clOrdId"].as_str().unwrap().to_string());
4220 client.ws_private.close().await.unwrap();
4221 server.abort();
4222
4223 assert_eq!(batch_sizes, expected_sizes);
4224 assert_eq!(payload, expected_payload);
4225 assert_eq!(actual, expected);
4226 terminate_tasks(&client.pending_tasks, "test execution client")
4227 .await
4228 .unwrap();
4229 }
4230
4231 #[rstest]
4232 #[case::algo_buy("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Buy), &["0", "2"], false)]
4233 #[case::algo_sell("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Sell), &["1", "3"], false)]
4234 #[case::algo_both("BTC-USDT.OKX", OrderType::StopLimit, None, &["0", "1", "2", "3"], false)]
4235 #[case::spread_buy("ETH-USD-SWAP_ETH-USD-231229.OKX", OrderType::Limit, Some(OrderSide::Buy), &["0", "2"], false)]
4236 #[case::spread_sell("ETH-USD-SWAP_ETH-USD-231229.OKX", OrderType::Limit, Some(OrderSide::Sell), &["1", "3"], false)]
4237 #[case::advance_buy("BTC-USDT.OKX", OrderType::TrailingStopMarket, Some(OrderSide::Buy), &["0", "2"], false)]
4238 #[case::advance_sell("BTC-USDT.OKX", OrderType::TrailingStopMarket, Some(OrderSide::Sell), &["1", "3"], false)]
4239 #[case::advance_both("BTC-USDT.OKX", OrderType::TrailingStopMarket, None, &["0", "1", "2", "3"], false)]
4240 #[case::mixed_buy("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Buy), &["0", "2"], true)]
4241 #[case::mixed_sell("BTC-USDT.OKX", OrderType::StopLimit, Some(OrderSide::Sell), &["1", "3"], true)]
4242 #[case::mixed_both("BTC-USDT.OKX", OrderType::StopLimit, None, &["0", "1", "2", "3"], true)]
4243 #[tokio::test]
4244 async fn test_cancel_all_orders_filters_http_requests(
4245 #[case] symbol: &str,
4246 #[case] order_type: OrderType,
4247 #[case] order_side: Option<OrderSide>,
4248 #[case] expected_ids: &[&str],
4249 #[case] mixed_types: bool,
4250 #[values(1, 5, 6)] copies: usize,
4251 ) {
4252 let (request_tx, mut request_rx) = tokio::sync::mpsc::unbounded_channel();
4253
4254 let router = Router::new().fallback(post(
4255 move |uri: axum::http::Uri, Json(request): Json<Value>| {
4256 let tx = request_tx.clone();
4257 async move {
4258 let mut response = serde_json::from_str::<Value>(include_str!(
4259 "../test_data/http_cancel_algo_order_response.json"
4260 ))
4261 .unwrap();
4262
4263 if let Some(ord_id) = request.get("ordId") {
4264 response["data"][0]
4265 .as_object_mut()
4266 .unwrap()
4267 .remove("algoId");
4268 response["data"][0]["ordId"] = ord_id.clone();
4269 } else {
4270 let template = response["data"][0].clone();
4271 response["data"] = request
4272 .as_array()
4273 .unwrap()
4274 .iter()
4275 .map(|item| {
4276 let mut result = template.clone();
4277 result["algoId"] = item["algoId"].clone();
4278 result
4279 })
4280 .collect();
4281 }
4282
4283 tx.send((uri.path().to_string(), request)).unwrap();
4284 Json(response)
4285 }
4286 },
4287 ));
4288
4289 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4290 let base_url = format!("http://{}", listener.local_addr().unwrap());
4291
4292 let server = tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
4293 let (mut client, cache) = build_test_exec_client_with_cache();
4294 client.http_client = OKXHttpClient::with_credentials(
4295 Some("test_key".into()),
4296 Some("test_secret".into()),
4297 Some("test_pass".into()),
4298 Some(base_url),
4299 5,
4300 0,
4301 1,
4302 1,
4303 client.config.environment,
4304 None,
4305 )
4306 .unwrap();
4307 let instrument_id = InstrumentId::from(symbol);
4308
4309 let mut expected_ids: Vec<_> = (0..copies)
4310 .flat_map(|copy| {
4311 expected_ids.iter().map(move |id| {
4312 let index: usize = id.parse().unwrap();
4313 (copy * 4 + index).to_string()
4314 })
4315 })
4316 .collect();
4317
4318 for copy in 0..copies {
4319 for (index, side) in [
4320 OrderSide::Buy,
4321 OrderSide::Sell,
4322 OrderSide::Buy,
4323 OrderSide::Sell,
4324 ]
4325 .into_iter()
4326 .enumerate()
4327 {
4328 let index = copy * 4 + index;
4329
4330 let cached_order_type = if mixed_types && index % 4 >= 2 {
4331 OrderType::TrailingStopMarket
4332 } else {
4333 order_type
4334 };
4335
4336 let order = OrderTestBuilder::new(cached_order_type)
4337 .instrument_id(instrument_id)
4338 .client_order_id(ClientOrderId::from(format!("OHTTP{index}")))
4339 .strategy_id(StrategyId::from(if index % 4 < 2 {
4340 "S-001"
4341 } else {
4342 "S-002"
4343 }))
4344 .side(side)
4345 .price(Price::from("100.00"))
4346 .trigger_price(Price::from("101.00"))
4347 .trailing_offset(rust_decimal::Decimal::ONE)
4348 .trailing_offset_type(TrailingOffsetType::Price)
4349 .quantity(Quantity::from("1"))
4350 .build();
4351
4352 let accepted = TestOrderEventStubs::accepted(
4353 &order,
4354 client.core.account_id,
4355 VenueOrderId::from(format!("{index}")),
4356 );
4357 cache
4358 .borrow_mut()
4359 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
4360 .unwrap();
4361 cache.borrow_mut().update_order(&accepted).unwrap();
4362 }
4363 }
4364
4365 client
4366 .cancel_all_orders(CancelAllOrders::new(
4367 TraderId::from("TESTER-001"),
4368 Some(*OKX_CLIENT_ID),
4369 StrategyId::from("S-001"),
4370 instrument_id,
4371 order_side,
4372 UUID4::new(),
4373 UnixNanos::default(),
4374 None,
4375 None,
4376 ))
4377 .unwrap();
4378 wait_until_async(
4379 || async { client.pending_tasks.is_empty() },
4380 Duration::from_secs(5),
4381 )
4382 .await;
4383
4384 let mut expected_sizes = if order_type == OrderType::Limit {
4385 vec![1; expected_ids.len()]
4386 } else if mixed_types {
4387 match (copies, order_side) {
4388 (1, Some(_)) => vec![1, 1],
4389 (1, None) => vec![2, 2],
4390 (5, Some(_)) => vec![5, 5],
4391 (5, None) => vec![10, 10],
4392 (6, Some(_)) => vec![6, 6],
4393 (6, None) => vec![10, 2, 10, 2],
4394 _ => unreachable!(),
4395 }
4396 } else {
4397 match (copies, order_side) {
4398 (1, Some(_)) => vec![2],
4399 (1, None) => vec![4],
4400 (5, Some(_)) => vec![10],
4401 (5, None) => vec![10, 10],
4402 (6, Some(_)) => vec![10, 2],
4403 (6, None) => vec![10, 10, 4],
4404 _ => unreachable!(),
4405 }
4406 };
4407
4408 let mut actual = Vec::new();
4409 let mut batch_sizes = Vec::new();
4410
4411 while let Ok((path, request)) = request_rx.try_recv() {
4412 if order_type == OrderType::Limit {
4413 assert_eq!(path, "/api/v5/sprd/cancel-order");
4414 batch_sizes.push(1);
4415 actual.push(request["ordId"].as_str().unwrap().to_string());
4416 } else {
4417 batch_sizes.push(request.as_array().unwrap().len());
4418 for item in request.as_array().unwrap() {
4419 let index: usize = item["algoId"].as_str().unwrap().parse().unwrap();
4420
4421 let expected_path = if order_type == OrderType::TrailingStopMarket
4422 || (mixed_types && index % 4 >= 2)
4423 {
4424 "/api/v5/trade/cancel-advance-algos"
4425 } else {
4426 "/api/v5/trade/cancel-algos"
4427 };
4428
4429 assert_eq!(path, expected_path);
4430 assert_eq!(item["instId"], "BTC-USDT");
4431 actual.push(item["algoId"].as_str().unwrap().to_string());
4432 }
4433 }
4434 }
4435
4436 server.abort();
4437 actual.sort();
4438 expected_ids.sort();
4439 batch_sizes.sort_unstable();
4440 expected_sizes.sort_unstable();
4441
4442 assert_eq!(actual, expected_ids);
4443 assert_eq!(batch_sizes, expected_sizes);
4444 }
4445
4446 #[rstest]
4447 #[case("BTC-USDT.OKX", false)]
4448 #[case("BTC-USDT.OKX", true)]
4449 #[case("ETH-USD-SWAP_ETH-USD-231229.OKX", false)]
4450 #[tokio::test]
4451 async fn test_cancel_all_orders_empty_cache_sends_nothing(
4452 #[case] instrument_id: &str,
4453 #[case] use_mm_mass_cancel: bool,
4454 ) {
4455 let mut client = build_test_exec_client();
4456 client.config.use_mm_mass_cancel = use_mm_mass_cancel;
4457 client
4458 .cancel_all_orders(CancelAllOrders::new(
4459 TraderId::from("TESTER-001"),
4460 Some(*OKX_CLIENT_ID),
4461 StrategyId::from("S-001"),
4462 InstrumentId::from(instrument_id),
4463 Some(OrderSide::Buy),
4464 UUID4::new(),
4465 UnixNanos::default(),
4466 None,
4467 None,
4468 ))
4469 .unwrap();
4470
4471 assert!(client.pending_tasks.is_empty());
4472 assert!(client.ws_dispatch_state.order_identities.is_empty());
4473 }
4474
4475 #[rstest]
4476 fn test_cancel_order_route_uses_bound_child_before_engine_trigger_state_updates() {
4477 let client = build_test_exec_client();
4478 let instrument_id = InstrumentId::from("BTC-USDT-SWAP.OKX");
4479 let pending_trigger = Some((OrderType::StopLimit, Some(false)));
4480
4481 assert_eq!(
4482 client.cancel_order_route(instrument_id, pending_trigger, false),
4483 OrderCommandRoute::AlgoHttp
4484 );
4485 assert_eq!(
4486 client.cancel_order_route(instrument_id, pending_trigger, true),
4487 OrderCommandRoute::RegularWs
4488 );
4489 }
4490
4491 #[rstest]
4492 #[tokio::test]
4493 async fn test_submit_conditional_order_resolves_routing_context() {
4494 let router = Router::new().route(
4495 "/api/v5/trade/order-algo",
4496 post(|Json(request): Json<Value>| async move {
4497 let data = match request["algoClOrdId"].as_str() {
4498 Some("ORESTBIND001") => serde_json::json!([{
4499 "algoId": "3796251408639365120",
4500 "algoClOrdId": "ORESTBIND001",
4501 "sCode": "0",
4502 "sMsg": "",
4503 }]),
4504 Some("ORESTAMBIGUOUS002") => serde_json::json!([{
4505 "algoId": "",
4506 "algoClOrdId": "ORESTAMBIGUOUS002",
4507 "sCode": "51149",
4508 "sMsg": "Order timed out. Please try again.",
4509 }]),
4510 _ => serde_json::json!([]),
4511 };
4512 Json(serde_json::json!({
4513 "code": "0",
4514 "msg": "",
4515 "data": data,
4516 }))
4517 }),
4518 );
4519 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
4520 let base_url_http = format!("http://{}", listener.local_addr().unwrap());
4521 tokio::spawn(async move {
4522 axum::serve(listener, router.into_make_service())
4523 .await
4524 .unwrap();
4525 });
4526
4527 let config = OKXExecutionClientConfig {
4528 api_key: Some("test_key".into()),
4529 api_secret: Some("test_secret".into()),
4530 api_passphrase: Some("test_pass".into()),
4531 base_url_http: Some(base_url_http),
4532 ..OKXExecutionClientConfig::default()
4533 };
4534 let cache = Rc::new(RefCell::new(Cache::default()));
4535 let core = ExecutionClientCore::new(
4536 TraderId::from("TESTER-001"),
4537 ClientId::from("OKX-TEST"),
4538 *OKX_VENUE,
4539 OmsType::Hedging,
4540 config.account_id,
4541 AccountType::Cash,
4542 None,
4543 Rc::clone(&cache),
4544 );
4545 let mut client = OKXExecutionClient::new(core, config).unwrap();
4546 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
4547 client.emitter.set_sender(event_tx);
4548 let client_order_id = ClientOrderId::from("ORESTBIND001");
4549 let order = OrderTestBuilder::new(OrderType::StopLimit)
4550 .client_order_id(client_order_id)
4551 .strategy_id(StrategyId::from("S-REST-BIND-001"))
4552 .instrument_id(InstrumentId::from("BTC-USDT-SWAP.OKX"))
4553 .side(OrderSide::Sell)
4554 .quantity(Quantity::from("0.01"))
4555 .price(Price::from("94900"))
4556 .trigger_price(Price::from("95000"))
4557 .build();
4558 cache
4559 .borrow_mut()
4560 .add_order(order.clone(), None, Some(*OKX_CLIENT_ID), false)
4561 .unwrap();
4562 let command = SubmitOrder::from_order(
4563 &order,
4564 TraderId::from("TESTER-001"),
4565 Some(*OKX_CLIENT_ID),
4566 None,
4567 UUID4::new(),
4568 UnixNanos::default(),
4569 );
4570
4571 client.submit_order(command).unwrap();
4572 wait_until_async(
4573 || async {
4574 client
4575 .ws_dispatch_state
4576 .order_venue_binding(client_order_id)
4577 == Some((VenueOrderId::from("3796251408639365120"), false))
4578 },
4579 Duration::from_secs(5),
4580 )
4581 .await;
4582
4583 assert_eq!(
4584 client.ws_dispatch_state.order_identity(client_order_id),
4585 Some(OrderIdentity {
4586 client_order_id,
4587 strategy_id: StrategyId::from("S-REST-BIND-001"),
4588 instrument_id: InstrumentId::from("BTC-USDT-SWAP.OKX"),
4589 order_side: OrderSide::Sell,
4590 order_type: OrderType::StopLimit,
4591 })
4592 );
4593
4594 let ambiguous_client_order_id = ClientOrderId::from("ORESTAMBIGUOUS002");
4595 let ambiguous_order = OrderTestBuilder::new(OrderType::StopLimit)
4596 .client_order_id(ambiguous_client_order_id)
4597 .strategy_id(StrategyId::from("S-REST-BIND-001"))
4598 .instrument_id(InstrumentId::from("BTC-USDT-SWAP.OKX"))
4599 .side(OrderSide::Sell)
4600 .quantity(Quantity::from("0.02"))
4601 .price(Price::from("94800"))
4602 .trigger_price(Price::from("95100"))
4603 .build();
4604 cache
4605 .borrow_mut()
4606 .add_order(ambiguous_order.clone(), None, Some(*OKX_CLIENT_ID), false)
4607 .unwrap();
4608 let ambiguous_command = SubmitOrder::from_order(
4609 &ambiguous_order,
4610 TraderId::from("TESTER-001"),
4611 Some(*OKX_CLIENT_ID),
4612 None,
4613 UUID4::new(),
4614 UnixNanos::default(),
4615 );
4616
4617 client.submit_order(ambiguous_command).unwrap();
4618 wait_until_async(
4619 || async {
4620 client
4621 .ws_dispatch_state
4622 .order_identity(ambiguous_client_order_id)
4623 .is_none()
4624 },
4625 Duration::from_secs(5),
4626 )
4627 .await;
4628
4629 assert_eq!(
4630 client
4631 .ws_dispatch_state
4632 .order_venue_binding(ambiguous_client_order_id),
4633 None
4634 );
4635
4636 wait_until_async(
4637 || async { client.pending_tasks.all_finished() },
4638 Duration::from_secs(5),
4639 )
4640 .await;
4641 terminate_tasks(&client.pending_tasks, "test execution client")
4642 .await
4643 .expect("execution tasks terminated");
4644
4645 let mut events = Vec::new();
4646 while let Ok(event) = event_rx.try_recv() {
4647 events.push(event);
4648 }
4649
4650 for expected_client_order_id in [client_order_id, ambiguous_client_order_id] {
4651 assert_eq!(
4652 events
4653 .iter()
4654 .filter(|event| matches!(
4655 event,
4656 ExecutionEvent::Order(OrderEventAny::Submitted(submitted))
4657 if submitted.client_order_id == expected_client_order_id
4658 ))
4659 .count(),
4660 1,
4661 );
4662 }
4663 assert!(
4664 !events.iter().any(|event| matches!(
4665 event,
4666 ExecutionEvent::Order(OrderEventAny::Rejected(rejected))
4667 if rejected.client_order_id == ambiguous_client_order_id
4668 )),
4669 "ambiguous algo submit failure should not emit OrderRejected: {events:?}",
4670 );
4671 }
4672
4673 #[rstest]
4674 #[tokio::test]
4675 async fn test_local_cancel_validation_failure_does_not_emit_order_cancel_rejected() {
4676 let (mut client, cache) = build_test_exec_client_with_cache();
4677 let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
4678 client.emitter.set_sender(event_tx);
4679 let client_order_id = ClientOrderId::from("OLOCALCANCELINVALID001");
4680 let strategy_id = StrategyId::from("S-LOCAL-CANCEL-INVALID-001");
4681 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
4682 let order = OrderTestBuilder::new(OrderType::Limit)
4683 .client_order_id(client_order_id)
4684 .strategy_id(strategy_id)
4685 .instrument_id(instrument_id)
4686 .side(OrderSide::Buy)
4687 .quantity(Quantity::from("1"))
4688 .price(Price::from("100000"))
4689 .build();
4690 cache
4691 .borrow_mut()
4692 .add_order(order, None, Some(*OKX_CLIENT_ID), false)
4693 .expect("cache order");
4694 let command = CancelOrder {
4695 trader_id: TraderId::from("TESTER-001"),
4696 client_id: Some(*OKX_CLIENT_ID),
4697 strategy_id,
4698 instrument_id,
4699 client_order_id,
4700 venue_order_id: Some(VenueOrderId::from("v-1")),
4701 command_id: UUID4::new(),
4702 ts_init: UnixNanos::default(),
4703 params: None,
4704 correlation_id: None,
4705 causation_id: None,
4706 };
4707
4708 client.cancel_order(command).expect("cancel order");
4709 wait_until_async(
4710 || async { client.pending_tasks.all_finished() },
4711 Duration::from_secs(5),
4712 )
4713 .await;
4714 terminate_tasks(&client.pending_tasks, "test execution client")
4715 .await
4716 .expect("execution task terminated");
4717
4718 let events: Vec<_> = std::iter::from_fn(|| event_rx.try_recv().ok()).collect();
4719 assert!(
4720 !events.iter().any(|event| matches!(
4721 event,
4722 ExecutionEvent::Order(OrderEventAny::CancelRejected(rejected))
4723 if rejected.client_order_id == client_order_id
4724 )),
4725 "local cancel validation failure should not emit OrderCancelRejected: {events:?}",
4726 );
4727 }
4728
4729 #[rstest]
4730 fn test_ensure_order_identity_skips_order_without_cached_side() {
4731 let client = build_test_exec_client();
4732 let client_order_id = ClientOrderId::from("O-RESTORED-001");
4733 let strategy_id = StrategyId::from("S-RESTORED-002");
4734 let instrument_id = InstrumentId::from("BTC-USDT.OKX");
4735
4736 client.ensure_order_identity(client_order_id, strategy_id, instrument_id);
4737
4738 assert!(
4739 client
4740 .ws_dispatch_state
4741 .order_identities
4742 .get(&client_order_id)
4743 .is_none()
4744 );
4745 }
4746
4747 #[rstest]
4748 fn test_on_instrument_writes_through_to_client_caches() {
4749 use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
4752
4753 let mut client = build_test_exec_client();
4754 let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
4755 let symbol = instrument.symbol().inner();
4756 let raw_symbol = instrument.raw_symbol().inner();
4757
4758 client.on_instrument(instrument.clone());
4759
4760 let private_cache = client.ws_private.instruments_cache_arc();
4761 let business_cache = client.ws_business.instruments_cache_arc();
4762 assert_eq!(
4763 client
4764 .http_client
4765 .get_instrument(&raw_symbol)
4766 .map(|i| i.id()),
4767 Some(instrument.id()),
4768 );
4769 assert_eq!(
4770 private_cache.load().get(&symbol).map(Instrument::id),
4771 Some(instrument.id()),
4772 );
4773 assert_eq!(
4774 business_cache.load().get(&symbol).map(Instrument::id),
4775 Some(instrument.id()),
4776 );
4777 }
4778}