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