nautilus_okx/
execution.rs

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

nautilus_okx/
execution.rs
