Skip to main content

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, Mutex},
21    time::{Duration, Instant},
22};
23
24use ahash::AHashMap;
25use anyhow::Context;
26use async_trait::async_trait;
27use futures_util::{StreamExt, pin_mut};
28use nautilus_common::{
29    clients::ExecutionClient,
30    live::{get_runtime, 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    MUTEX_POISONED, UnixNanos,
41    params::Params,
42    time::{AtomicTime, get_atomic_clock_realtime},
43};
44use nautilus_live::{ExecutionClientCore, ExecutionEventEmitter};
45use nautilus_model::{
46    accounts::AccountAny,
47    enums::{AccountType, OmsType, OrderSide, OrderType, TimeInForce, TrailingOffsetType},
48    events::OrderDeniedReason,
49    identifiers::{
50        AccountId, ClientId, ClientOrderId, InstrumentId, StrategyId, TraderId, Venue, VenueOrderId,
51    },
52    instruments::InstrumentAny,
53    orders::Order,
54    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
55    types::{AccountBalance, MarginBalance, Money, Quantity},
56};
57use rust_decimal::Decimal;
58use tokio::task::JoinHandle;
59use ustr::Ustr;
60
61use crate::{
62    common::{
63        consts::{
64            OKX_CONDITIONAL_ORDER_TYPES, OKX_SUCCESS_CODE, OKX_VENUE, OKX_WS_HEARTBEAT_SECS,
65            resolve_instrument_families, validate_okx_client_order_id,
66        },
67        enums::{OKXInstrumentType, OKXMarginMode, OKXTradeMode, is_advance_algo_order},
68        parse::{is_okx_spread_symbol, nanos_to_datetime, okx_instrument_type_from_symbol},
69    },
70    config::OKXExecClientConfig,
71    http::{client::OKXHttpClient, error::OKXHttpError, models::OKXCancelAlgoOrderRequest},
72    websocket::{
73        client::OKXWebSocketClient,
74        dispatch::{
75            AlgoCancelContext, OrderIdentity, WsDispatchState, dispatch_ws_message,
76            emit_algo_cancel_rejections, emit_batch_cancel_failure,
77        },
78        error::OKXWsError,
79        parse::OrderStateSnapshot,
80    },
81};
82
83#[derive(Debug)]
84pub struct OKXExecutionClient {
85    core: ExecutionClientCore,
86    clock: &'static AtomicTime,
87    config: OKXExecClientConfig,
88    emitter: ExecutionEventEmitter,
89    http_client: OKXHttpClient,
90    ws_private: OKXWebSocketClient,
91    ws_business: OKXWebSocketClient,
92    trade_mode: OKXTradeMode,
93    ws_stream_handle: Option<JoinHandle<()>>,
94    ws_business_stream_handle: Option<JoinHandle<()>>,
95    ws_dispatch_state: Arc<WsDispatchState>,
96    pending_tasks: Mutex<Vec<JoinHandle<()>>>,
97}
98
99impl OKXExecutionClient {
100    /// Creates a new [`OKXExecutionClient`].
101    ///
102    /// # Errors
103    ///
104    /// Returns an error if the client fails to initialize.
105    pub fn new(core: ExecutionClientCore, config: OKXExecClientConfig) -> anyhow::Result<Self> {
106        let http_client = OKXHttpClient::with_credentials(
107            config.api_key.clone(),
108            config.api_secret.clone(),
109            config.api_passphrase.clone(),
110            Some(config.http_base_url()),
111            config.http_timeout_secs,
112            config.max_retries,
113            config.retry_delay_initial_ms,
114            config.retry_delay_max_ms,
115            config.environment,
116            config.proxy_url.clone(),
117        )?;
118
119        let account_id = core.account_id;
120
121        let ws_private = OKXWebSocketClient::with_credentials(
122            Some(config.ws_private_url()),
123            config.api_key.clone(),
124            config.api_secret.clone(),
125            config.api_passphrase.clone(),
126            Some(account_id),
127            Some(OKX_WS_HEARTBEAT_SECS),
128            None,
129            config.transport_backend,
130            config.proxy_url.clone(),
131        )
132        .context("failed to construct OKX private websocket client")?;
133
134        let ws_business = OKXWebSocketClient::with_credentials(
135            Some(config.ws_business_url()),
136            config.api_key.clone(),
137            config.api_secret.clone(),
138            config.api_passphrase.clone(),
139            Some(account_id),
140            Some(OKX_WS_HEARTBEAT_SECS),
141            None,
142            config.transport_backend,
143            config.proxy_url.clone(),
144        )
145        .context("failed to construct OKX business websocket client")?;
146
147        let trade_mode = Self::derive_default_trade_mode(core.account_type, &config);
148        let clock = get_atomic_clock_realtime();
149        let emitter = ExecutionEventEmitter::new(
150            clock,
151            core.trader_id,
152            core.account_id,
153            core.account_type,
154            None,
155        );
156
157        let ws_dispatch_state = Arc::new(WsDispatchState::with_pending_maps(
158            ws_private.pending_orders.clone(),
159            ws_private.pending_cancels.clone(),
160            ws_private.pending_amends.clone(),
161        ));
162
163        Ok(Self {
164            core,
165            clock,
166            config,
167            emitter,
168            http_client,
169            ws_private,
170            ws_business,
171            trade_mode,
172            ws_stream_handle: None,
173            ws_business_stream_handle: None,
174            ws_dispatch_state,
175            pending_tasks: Mutex::new(Vec::new()),
176        })
177    }
178
179    fn derive_default_trade_mode(
180        account_type: AccountType,
181        config: &OKXExecClientConfig,
182    ) -> OKXTradeMode {
183        let is_cross_margin = config.margin_mode == Some(OKXMarginMode::Cross);
184
185        if account_type == AccountType::Cash {
186            if !config.use_spot_margin {
187                return OKXTradeMode::Cash;
188            }
189            return if is_cross_margin {
190                OKXTradeMode::Cross
191            } else {
192                OKXTradeMode::Isolated
193            };
194        }
195
196        if is_cross_margin {
197            OKXTradeMode::Cross
198        } else {
199            OKXTradeMode::Isolated
200        }
201    }
202
203    fn trade_mode_for_order(
204        &self,
205        instrument_id: InstrumentId,
206        params: &Option<Params>,
207    ) -> OKXTradeMode {
208        if let Some(td_mode_str) = get_param_as_string(params, "td_mode") {
209            match td_mode_str.parse::<OKXTradeMode>() {
210                Ok(mode) => return mode,
211                Err(_) => {
212                    log::warn!("Invalid td_mode '{td_mode_str}', using derived trade mode");
213                }
214            }
215        }
216
217        derive_trade_mode_for_instrument(
218            instrument_id,
219            self.config.margin_mode,
220            self.config.use_spot_margin,
221        )
222    }
223
224    fn instrument_types(&self) -> Vec<OKXInstrumentType> {
225        if self.config.instrument_types.is_empty() {
226            vec![OKXInstrumentType::Spot]
227        } else {
228            self.config.instrument_types.clone()
229        }
230    }
231
232    fn update_account_state(&self) {
233        let http_client = self.http_client.clone();
234        let account_id = self.core.account_id;
235        let emitter = self.emitter.clone();
236
237        self.spawn_task("query_account", async move {
238            let account_state = http_client
239                .request_account_state(account_id)
240                .await
241                .context("failed to request OKX account state")?;
242            emitter.send_account_state(account_state);
243            Ok(())
244        });
245    }
246
247    fn is_conditional_order(&self, order_type: OrderType) -> bool {
248        OKX_CONDITIONAL_ORDER_TYPES.contains(&order_type)
249    }
250
251    fn submit_order_route(
252        &self,
253        instrument_id: InstrumentId,
254        order_type: OrderType,
255    ) -> anyhow::Result<OrderCommandRoute> {
256        if self.is_conditional_order(order_type) {
257            if is_spread_instrument(instrument_id) {
258                anyhow::bail!(
259                    "Trigger/conditional orders ({order_type:?}) are not supported for OKX spreads"
260                );
261            }
262
263            let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
264            if inst_type == OKXInstrumentType::Option {
265                anyhow::bail!(
266                    "Trigger/conditional orders ({order_type:?}) are not supported for OKX options"
267                );
268            }
269
270            return Ok(OrderCommandRoute::AlgoHttp);
271        }
272
273        if is_spread_instrument(instrument_id) {
274            Ok(OrderCommandRoute::SpreadHttp)
275        } else {
276            Ok(OrderCommandRoute::RegularWs)
277        }
278    }
279
280    fn cancel_order_route(
281        &self,
282        instrument_id: InstrumentId,
283        order_state: Option<(OrderType, Option<bool>)>,
284    ) -> OrderCommandRoute {
285        if is_spread_instrument(instrument_id) {
286            return OrderCommandRoute::SpreadHttp;
287        }
288
289        if order_state.is_some_and(|(order_type, is_triggered)| {
290            self.is_conditional_order(order_type) && is_triggered != Some(true)
291        }) {
292            OrderCommandRoute::AlgoHttp
293        } else {
294            OrderCommandRoute::RegularWs
295        }
296    }
297
298    fn cancel_all_orders_route(&self, instrument_id: InstrumentId) -> CancelAllOrdersRoute {
299        if is_spread_instrument(instrument_id) {
300            CancelAllOrdersRoute::SpreadHttp
301        } else if self.config.use_mm_mass_cancel {
302            CancelAllOrdersRoute::MassCancelHttp
303        } else {
304            CancelAllOrdersRoute::BatchWs
305        }
306    }
307
308    fn submit_regular_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
309        let order = {
310            let cache = self.core.cache();
311            cache.try_order_owned(&cmd.client_order_id)?
312        };
313        let ws_private = self.ws_private.clone();
314        let trade_mode = self.trade_mode_for_order(cmd.instrument_id, &cmd.params);
315
316        let emitter = self.emitter.clone();
317        let clock = self.clock;
318        let trader_id = self.core.trader_id;
319        let client_order_id = order.client_order_id();
320        let strategy_id = order.strategy_id();
321        let instrument_id = order.instrument_id();
322
323        self.ws_dispatch_state.order_identities.insert(
324            client_order_id,
325            OrderIdentity {
326                instrument_id,
327                strategy_id,
328                order_side: order.order_side(),
329                order_type: order.order_type(),
330            },
331        );
332        let order_side = order.order_side();
333        let order_type = order.order_type();
334        let quantity = order.quantity();
335        let time_in_force = order.time_in_force();
336        let price = order.price();
337        let trigger_price = order.trigger_price();
338        let is_post_only = order.is_post_only();
339        let is_reduce_only = order.is_reduce_only();
340        let is_quote_quantity = order.is_quote_quantity();
341
342        let px_usd = get_param_as_string(&cmd.params, "px_usd");
343        let px_vol = get_param_as_string(&cmd.params, "px_vol");
344        let speed_bump = get_param_as_string(&cmd.params, "speed_bump");
345        let outcome = get_param_as_string(&cmd.params, "outcome");
346        let slippage_pct = get_param_as_string(&cmd.params, "slippage_pct");
347
348        self.spawn_task("submit_order", async move {
349            let result = ws_private
350                .submit_order(
351                    trader_id,
352                    strategy_id,
353                    instrument_id,
354                    trade_mode,
355                    client_order_id,
356                    order_side,
357                    order_type,
358                    quantity,
359                    Some(time_in_force),
360                    price,
361                    trigger_price,
362                    Some(is_post_only),
363                    Some(is_reduce_only),
364                    Some(is_quote_quantity),
365                    None,
366                    None,
367                    px_usd,
368                    px_vol,
369                    speed_bump,
370                    outcome,
371                    slippage_pct,
372                )
373                .await;
374
375            if let Err(e) = result {
376                if is_okx_ws_local_command_failure(&e) {
377                    let ts_event = clock.get_time_ns();
378                    emitter.emit_order_rejected_event(
379                        strategy_id,
380                        instrument_id,
381                        client_order_id,
382                        &format!("submit-order-error: {e}"),
383                        ts_event,
384                        false,
385                    );
386                } else {
387                    log::warn!(
388                        "Ambiguous submit failure for {client_order_id}, awaiting reconciliation: {e}"
389                    );
390                }
391                return Err(anyhow::Error::new(e).context("submit order failed"));
392            }
393
394            Ok(())
395        });
396
397        Ok(())
398    }
399
400    fn submit_order_http(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
401        let order = {
402            let cache = self.core.cache();
403            cache.try_order_owned(&cmd.client_order_id)?
404        };
405        let http_client = self.http_client.clone();
406        let trade_mode = self.trade_mode_for_order(cmd.instrument_id, &cmd.params);
407
408        let emitter = self.emitter.clone();
409        let clock = self.clock;
410        let client_order_id = order.client_order_id();
411        let strategy_id = order.strategy_id();
412        let instrument_id = order.instrument_id();
413
414        self.ws_dispatch_state.order_identities.insert(
415            client_order_id,
416            OrderIdentity {
417                instrument_id,
418                strategy_id,
419                order_side: order.order_side(),
420                order_type: order.order_type(),
421            },
422        );
423        let order_side = order.order_side();
424        let order_type = order.order_type();
425        let quantity = order.quantity();
426        let time_in_force = order.time_in_force();
427        let price = order.price();
428        let is_post_only = order.is_post_only();
429
430        self.spawn_task("submit_order_http", async move {
431            let result = http_client
432                .place_order_with_domain_types(
433                    instrument_id,
434                    trade_mode,
435                    client_order_id,
436                    order_side,
437                    order_type,
438                    quantity,
439                    Some(time_in_force),
440                    price,
441                    Some(is_post_only),
442                    None,
443                    None,
444                    None,
445                    None,
446                    None,
447                    None,
448                    None,
449                    None,
450                    None,
451                )
452                .await;
453
454            if let Err(e) = result {
455                if is_okx_http_structured_venue_rejection(&e)
456                    || is_okx_http_local_command_failure(&e)
457                {
458                    let ts_event = clock.get_time_ns();
459                    emitter.emit_order_rejected_event(
460                        strategy_id,
461                        instrument_id,
462                        client_order_id,
463                        &format!("submit-order-error: {e}"),
464                        ts_event,
465                        false,
466                    );
467                } else {
468                    log::warn!(
469                        "Ambiguous HTTP submit failure for {client_order_id}, awaiting reconciliation: {e}"
470                    );
471                }
472                return Err(anyhow::Error::new(e).context("submit order failed"));
473            }
474
475            Ok(())
476        });
477
478        Ok(())
479    }
480
481    fn submit_conditional_order(&self, cmd: &SubmitOrder) -> anyhow::Result<()> {
482        let order = {
483            let cache = self.core.cache();
484            cache.try_order_owned(&cmd.client_order_id)?
485        };
486        let http_client = self.http_client.clone();
487        let trade_mode = self.trade_mode_for_order(cmd.instrument_id, &cmd.params);
488
489        let emitter = self.emitter.clone();
490        let clock = self.clock;
491        let client_order_id = order.client_order_id();
492        let strategy_id = order.strategy_id();
493        let instrument_id = order.instrument_id();
494        let order_side = order.order_side();
495        let order_type = order.order_type();
496
497        self.ws_dispatch_state.order_identities.insert(
498            client_order_id,
499            OrderIdentity {
500                instrument_id,
501                strategy_id,
502                order_side,
503                order_type,
504            },
505        );
506        let quantity = order.quantity();
507        let trigger_type = order.trigger_type();
508        let trigger_price = order.trigger_price();
509        let price = order.price();
510        let is_reduce_only = order.is_reduce_only();
511
512        let trailing_offset = order.trailing_offset();
513        let trailing_offset_type = order.trailing_offset_type();
514        let activation_price = order.activation_price();
515
516        let close_fraction = get_param_as_string(&cmd.params, "close_fraction");
517        let reduce_only = if close_fraction.is_some() {
518            Some(true)
519        } else {
520            Some(is_reduce_only)
521        };
522
523        let (callback_ratio, callback_spread) = if order_type == OrderType::TrailingStopMarket {
524            let offset = trailing_offset
525                .ok_or_else(|| anyhow::anyhow!("TrailingStopMarket requires trailing_offset"))?;
526            let offset_type = trailing_offset_type.ok_or_else(|| {
527                anyhow::anyhow!("TrailingStopMarket requires trailing_offset_type")
528            })?;
529
530            match offset_type {
531                TrailingOffsetType::BasisPoints => {
532                    // Convert basis points to ratio (e.g., 100 bps = 0.01)
533                    let ratio = offset / Decimal::from(10000);
534                    (Some(ratio.to_string()), None)
535                }
536                TrailingOffsetType::Price => (None, Some(offset.to_string())),
537                _ => {
538                    anyhow::bail!("Unsupported trailing_offset_type for OKX: {offset_type:?}");
539                }
540            }
541        } else {
542            (None, None)
543        };
544
545        self.spawn_task("submit_algo_order", async move {
546            let result = http_client
547                .place_algo_order_with_domain_types(
548                    instrument_id,
549                    trade_mode,
550                    client_order_id,
551                    order_side,
552                    order_type,
553                    quantity,
554                    trigger_price,
555                    trigger_type,
556                    price,
557                    reduce_only,
558                    close_fraction,
559                    callback_ratio,
560                    callback_spread,
561                    activation_price,
562                )
563                .await;
564
565            if let Err(e) = result {
566                if is_okx_http_structured_venue_rejection(&e)
567                    || is_okx_http_local_command_failure(&e)
568                {
569                    let ts_event = clock.get_time_ns();
570                    emitter.emit_order_rejected_event(
571                        strategy_id,
572                        instrument_id,
573                        client_order_id,
574                        &format!("submit-order-error: {e}"),
575                        ts_event,
576                        false,
577                    );
578                } else {
579                    log::warn!(
580                        "Ambiguous algo submit failure for {client_order_id}, awaiting reconciliation: {e}"
581                    );
582                }
583                return Err(anyhow::Error::new(e).context("submit algo order failed"));
584            }
585
586            Ok(())
587        });
588
589        Ok(())
590    }
591
592    fn cancel_ws_order(&self, cmd: &CancelOrder) {
593        self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
594
595        let ws_private = self.ws_private.clone();
596        let command = cmd.clone();
597
598        self.spawn_task("cancel_order", async move {
599            let result = ws_private
600                .cancel_order(
601                    command.trader_id,
602                    command.strategy_id,
603                    command.instrument_id,
604                    Some(command.client_order_id),
605                    command.venue_order_id,
606                )
607                .await;
608
609            if let Err(e) = result {
610                if is_okx_ws_local_command_failure(&e) {
611                    log::warn!(
612                        "Cancel command failed local validation for {}: {e}",
613                        command.client_order_id
614                    );
615                } else {
616                    log::warn!(
617                        "Ambiguous cancel failure for {}, awaiting reconciliation: {e}",
618                        command.client_order_id
619                    );
620                }
621                return Err(anyhow::Error::new(e).context("cancel order failed"));
622            }
623
624            Ok(())
625        });
626    }
627
628    fn cancel_order_http(&self, cmd: &CancelOrder) {
629        self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
630
631        let http_client = self.http_client.clone();
632        let command = cmd.clone();
633        let emitter = self.emitter.clone();
634        let clock = self.clock;
635
636        self.spawn_task("cancel_order_http", async move {
637            let result = http_client
638                .cancel_order(
639                    command.instrument_id,
640                    Some(command.client_order_id),
641                    command.venue_order_id,
642                )
643                .await;
644
645            if let Err(e) = result {
646                if is_okx_http_structured_venue_rejection(&e) {
647                    let ts_event = clock.get_time_ns();
648                    emitter.emit_order_cancel_rejected_event(
649                        command.strategy_id,
650                        command.instrument_id,
651                        command.client_order_id,
652                        command.venue_order_id,
653                        &format!("cancel-order-error: {e}"),
654                        ts_event,
655                    );
656                } else if is_okx_http_local_command_failure(&e) {
657                    log::warn!(
658                        "HTTP cancel command failed local validation for {}: {e}",
659                        command.client_order_id
660                    );
661                } else {
662                    log::warn!(
663                        "Ambiguous HTTP cancel failure for {}, awaiting reconciliation: {e}",
664                        command.client_order_id
665                    );
666                }
667                return Err(anyhow::Error::new(e).context("cancel order failed"));
668            }
669
670            Ok(())
671        });
672    }
673
674    fn cancel_algo_order(&self, cmd: &CancelOrder) {
675        let http_client = self.http_client.clone();
676        let command = cmd.clone();
677        let emitter = self.emitter.clone();
678        let clock = self.clock;
679
680        let cache = self.core.cache();
681        let is_advance = cache
682            .order(&cmd.client_order_id)
683            .is_some_and(|o| is_advance_algo_order(o.order_type()));
684        drop(cache);
685
686        let request = OKXCancelAlgoOrderRequest {
687            inst_id: cmd.instrument_id.symbol.to_string(),
688            inst_id_code: None,
689            algo_id: cmd.venue_order_id.map(|id| id.to_string()),
690            algo_cl_ord_id: if cmd.venue_order_id.is_none() {
691                Some(cmd.client_order_id.to_string())
692            } else {
693                None
694            },
695        };
696
697        self.spawn_task("cancel_algo_order", async move {
698            let responses = if is_advance {
699                http_client.cancel_advance_algo_orders(vec![request]).await
700            } else {
701                http_client.cancel_algo_orders(vec![request]).await
702            };
703
704            let reject_reason = match &responses {
705                Err(e) if is_okx_http_structured_venue_rejection(e) => {
706                    Some(format!("cancel-algo-order-error: {e}"))
707                }
708                Err(e) if is_okx_http_local_command_failure(e) => {
709                    log::warn!(
710                        "Algo cancel command failed local validation for {}: {e}",
711                        command.client_order_id
712                    );
713                    None
714                }
715                Err(e) => {
716                    log::warn!(
717                        "Ambiguous algo cancel failure for {}, awaiting reconciliation: {e}",
718                        command.client_order_id
719                    );
720                    None
721                }
722                Ok(resps) => {
723                    // Check per-order business status code
724                    resps.first().and_then(|r| {
725                        r.s_code.as_deref().and_then(|code| {
726                            if code == OKX_SUCCESS_CODE {
727                                None
728                            } else {
729                                let msg = r.s_msg.as_deref().unwrap_or("unknown");
730                                Some(format!(
731                                    "cancel-algo-order-rejected: s_code={code}, s_msg={msg}"
732                                ))
733                            }
734                        })
735                    })
736                }
737            };
738
739            if let Some(reason) = reject_reason {
740                let ts_event = clock.get_time_ns();
741                emitter.emit_order_cancel_rejected_event(
742                    command.strategy_id,
743                    command.instrument_id,
744                    command.client_order_id,
745                    command.venue_order_id,
746                    &reason,
747                    ts_event,
748                );
749                anyhow::bail!("{reason}");
750            }
751
752            if let Err(e) = responses {
753                return Err(anyhow::Error::new(e).context("cancel algo order failed"));
754            }
755
756            Ok(())
757        });
758    }
759
760    fn mass_cancel_instrument(&self, instrument_id: InstrumentId) {
761        if is_spread_instrument(instrument_id) {
762            let http_client = self.http_client.clone();
763            self.spawn_task("mass_cancel_orders_http", async move {
764                http_client
765                    .cancel_all_orders(instrument_id)
766                    .await
767                    .map_err(|e| anyhow::anyhow!("Mass cancel orders failed: {e}"))?;
768                Ok(())
769            });
770            return;
771        }
772
773        let ws_private = self.ws_private.clone();
774
775        self.spawn_task("mass_cancel_orders", async move {
776            ws_private.mass_cancel_orders(instrument_id).await?;
777            Ok(())
778        });
779    }
780
781    /// Populates `order_identities` for an order if not already present.
782    ///
783    /// Needed for cancel/modify commands on orders loaded via reconciliation
784    /// (which bypass `submit_order` and therefore have no identity entry).
785    /// Uses `DashMap::entry().or_insert_with` to keep the check-and-insert
786    /// atomic; without it, two concurrent reconciliation tasks could race
787    /// past a `contains_key` check and overwrite each other with stale
788    /// cache state.
789    fn ensure_order_identity(
790        &self,
791        client_order_id: ClientOrderId,
792        strategy_id: StrategyId,
793        instrument_id: InstrumentId,
794    ) {
795        self.ws_dispatch_state
796            .order_identities
797            .entry(client_order_id)
798            .or_insert_with(|| {
799                let cache = self.core.cache();
800                let (order_side, order_type) = cache
801                    .order(&client_order_id)
802                    .map_or((OrderSide::NoOrderSide, OrderType::Market), |o| {
803                        (o.order_side(), o.order_type())
804                    });
805                drop(cache);
806
807                OrderIdentity {
808                    instrument_id,
809                    strategy_id,
810                    order_side,
811                    order_type,
812                }
813            });
814    }
815
816    fn spawn_task<F>(&self, description: &'static str, fut: F)
817    where
818        F: Future<Output = anyhow::Result<()>> + Send + 'static,
819    {
820        let runtime = get_runtime();
821        let handle = runtime.spawn(async move {
822            if let Err(e) = fut.await {
823                log::warn!("{description} failed: {e:?}");
824            }
825        });
826
827        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
828        tasks.retain(|handle| !handle.is_finished());
829        tasks.push(handle);
830    }
831
832    // Partitions algo cancel orders into regular and advance, then spawns
833    // HTTP tasks for each group with per-item and batch-level rejection handling.
834    fn dispatch_algo_cancels(&self, items: Vec<(OKXCancelAlgoOrderRequest, AlgoCancelContext)>) {
835        let mut regular_requests = Vec::new();
836        let mut regular_contexts = Vec::new();
837        let mut advance_requests = Vec::new();
838        let mut advance_contexts = Vec::new();
839
840        let cache = self.core.cache();
841
842        for (request, ctx) in items {
843            let is_advance = cache
844                .order(&ctx.client_order_id)
845                .is_some_and(|o| is_advance_algo_order(o.order_type()));
846
847            if is_advance {
848                advance_requests.push(request);
849                advance_contexts.push(ctx);
850            } else {
851                regular_requests.push(request);
852                regular_contexts.push(ctx);
853            }
854        }
855
856        drop(cache);
857
858        if !regular_requests.is_empty() {
859            let client = self.http_client.clone();
860            let emitter = self.emitter.clone();
861            let clock = self.clock;
862
863            self.spawn_task("cancel_algo_orders", async move {
864                match client.cancel_algo_orders(regular_requests).await {
865                    Ok(responses) => {
866                        emit_algo_cancel_rejections(&responses, &regular_contexts, &emitter, clock);
867                    }
868                    Err(e) => {
869                        if is_okx_http_local_command_failure(&e) {
870                            for ctx in &regular_contexts {
871                                log::warn!(
872                                    "Algo batch cancel command failed local validation for {}: {e}",
873                                    ctx.client_order_id
874                                );
875                            }
876                        } else {
877                            let msg = format!("{e}");
878                            emit_batch_cancel_failure(&regular_contexts, &msg, &emitter, clock);
879                        }
880                        return Err(anyhow::Error::new(e).context("cancel algo orders failed"));
881                    }
882                }
883                Ok(())
884            });
885        }
886
887        if !advance_requests.is_empty() {
888            let client = self.http_client.clone();
889            let emitter = self.emitter.clone();
890            let clock = self.clock;
891
892            self.spawn_task("cancel_advance_algo_orders", async move {
893                match client.cancel_advance_algo_orders(advance_requests).await {
894                    Ok(responses) => {
895                        emit_algo_cancel_rejections(&responses, &advance_contexts, &emitter, clock);
896                    }
897                    Err(e) => {
898                        if is_okx_http_local_command_failure(&e) {
899                            for ctx in &advance_contexts {
900                                log::warn!(
901                                    "Advance algo batch cancel command failed local validation for {}: {e}",
902                                    ctx.client_order_id
903                                );
904                            }
905                        } else {
906                            let msg = format!("{e}");
907                            emit_batch_cancel_failure(&advance_contexts, &msg, &emitter, clock);
908                        }
909                        return Err(
910                            anyhow::Error::new(e).context("cancel advance algo orders failed")
911                        );
912                    }
913                }
914                Ok(())
915            });
916        }
917    }
918
919    fn abort_pending_tasks(&self) {
920        let mut tasks = self.pending_tasks.lock().expect(MUTEX_POISONED);
921
922        for handle in tasks.drain(..) {
923            handle.abort();
924        }
925    }
926
927    /// Polls the cache until the account is registered or timeout is reached.
928    async fn await_account_registered(&self, timeout_secs: f64) -> anyhow::Result<()> {
929        let account_id = self.core.account_id;
930
931        if self.core.cache().account(&account_id).is_some() {
932            log::info!("Account {account_id} registered");
933            return Ok(());
934        }
935
936        let start = Instant::now();
937        let timeout = Duration::from_secs_f64(timeout_secs);
938        let interval = Duration::from_millis(10);
939
940        loop {
941            tokio::time::sleep(interval).await;
942
943            if self.core.cache().account(&account_id).is_some() {
944                log::info!("Account {account_id} registered");
945                return Ok(());
946            }
947
948            if start.elapsed() >= timeout {
949                anyhow::bail!(
950                    "Timeout waiting for account {account_id} to be registered after {timeout_secs}s"
951                );
952            }
953        }
954    }
955}
956
957fn derive_trade_mode_for_instrument(
958    instrument_id: InstrumentId,
959    margin_mode: Option<OKXMarginMode>,
960    use_spot_margin: bool,
961) -> OKXTradeMode {
962    let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
963    let is_cross_margin = margin_mode == Some(OKXMarginMode::Cross);
964
965    match inst_type {
966        OKXInstrumentType::Spot => {
967            if use_spot_margin {
968                if is_cross_margin {
969                    OKXTradeMode::Cross
970                } else {
971                    OKXTradeMode::Isolated
972                }
973            } else {
974                OKXTradeMode::Cash
975            }
976        }
977        _ => {
978            if is_cross_margin {
979                OKXTradeMode::Cross
980            } else {
981                OKXTradeMode::Isolated
982            }
983        }
984    }
985}
986
987#[async_trait(?Send)]
988impl ExecutionClient for OKXExecutionClient {
989    fn is_connected(&self) -> bool {
990        self.core.is_connected()
991    }
992
993    fn client_id(&self) -> ClientId {
994        self.core.client_id
995    }
996
997    fn account_id(&self) -> AccountId {
998        self.core.account_id
999    }
1000
1001    fn venue(&self) -> Venue {
1002        *OKX_VENUE
1003    }
1004
1005    fn oms_type(&self) -> OmsType {
1006        self.core.oms_type
1007    }
1008
1009    fn get_account(&self) -> Option<AccountAny> {
1010        self.core.cache().account_owned(&self.core.account_id)
1011    }
1012
1013    async fn connect(&mut self) -> anyhow::Result<()> {
1014        if self.core.is_connected() {
1015            return Ok(());
1016        }
1017
1018        let instrument_types = self.instrument_types();
1019
1020        if !self.core.instruments_initialized() {
1021            let mut all_instruments = Vec::new();
1022            let mut all_inst_id_codes = Vec::new();
1023
1024            for instrument_type in &instrument_types {
1025                let Some(families) =
1026                    resolve_instrument_families(&self.config.instrument_families, *instrument_type)
1027                else {
1028                    continue;
1029                };
1030
1031                if families.is_empty() {
1032                    let (instruments, inst_id_codes) = self
1033                        .http_client
1034                        .request_instruments(*instrument_type, None)
1035                        .await
1036                        .with_context(|| {
1037                            format!("failed to request OKX instruments for {instrument_type:?}")
1038                        })?;
1039
1040                    if instruments.is_empty() {
1041                        log::warn!("No instruments returned for {instrument_type:?}");
1042                        continue;
1043                    }
1044
1045                    log::debug!(
1046                        "Loaded {} {instrument_type:?} instruments",
1047                        instruments.len()
1048                    );
1049
1050                    self.http_client.cache_instruments(&instruments);
1051                    all_instruments.extend(instruments);
1052                    all_inst_id_codes.extend(inst_id_codes);
1053                } else {
1054                    for family in &families {
1055                        let (instruments, inst_id_codes) = self
1056                            .http_client
1057                            .request_instruments(*instrument_type, Some(family.clone()))
1058                            .await
1059                            .with_context(|| {
1060                                format!(
1061                                    "failed to request OKX instruments for {instrument_type:?} family {family}"
1062                                )
1063                            })?;
1064
1065                        if instruments.is_empty() {
1066                            log::warn!(
1067                                "No instruments returned for {instrument_type:?} family {family}"
1068                            );
1069                            continue;
1070                        }
1071
1072                        log::debug!(
1073                            "Loaded {} {instrument_type:?} instruments for family {family}",
1074                            instruments.len()
1075                        );
1076
1077                        self.http_client.cache_instruments(&instruments);
1078                        all_instruments.extend(instruments);
1079                        all_inst_id_codes.extend(inst_id_codes);
1080                    }
1081                }
1082            }
1083
1084            if all_instruments.is_empty() {
1085                anyhow::bail!(
1086                    "No instruments loaded for configured types {instrument_types:?}, \
1087                     cannot initialize execution client"
1088                );
1089            }
1090
1091            self.ws_private.cache_instruments(&all_instruments);
1092            self.ws_private
1093                .cache_inst_id_codes(all_inst_id_codes.clone());
1094            self.ws_business.cache_instruments(&all_instruments);
1095            self.ws_business.cache_inst_id_codes(all_inst_id_codes);
1096            self.core.set_instruments_initialized();
1097        }
1098
1099        self.ws_private.connect().await?;
1100        self.ws_private.wait_until_active(10.0).await?;
1101        log::info!("Connected to private WebSocket");
1102
1103        if self.ws_stream_handle.is_none() {
1104            let stream = self.ws_private.stream();
1105            let emitter = self.emitter.clone();
1106            let state = Arc::clone(&self.ws_dispatch_state);
1107            let account_id = self.core.account_id;
1108            let instruments = self.ws_private.instruments_cache_arc();
1109            let clock = self.clock;
1110
1111            let handle = get_runtime().spawn(async move {
1112                let mut fee_cache: AHashMap<Ustr, Money> = AHashMap::new();
1113                let mut filled_qty_cache: AHashMap<Ustr, Quantity> = AHashMap::new();
1114                let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
1115                    AHashMap::new();
1116
1117                pin_mut!(stream);
1118
1119                while let Some(message) = stream.next().await {
1120                    dispatch_ws_message(
1121                        message,
1122                        &emitter,
1123                        &state,
1124                        account_id,
1125                        &instruments,
1126                        &mut fee_cache,
1127                        &mut filled_qty_cache,
1128                        &mut order_state_cache,
1129                        clock,
1130                    );
1131                }
1132            });
1133            self.ws_stream_handle = Some(handle);
1134        }
1135
1136        self.ws_business.connect().await?;
1137        self.ws_business.wait_until_active(10.0).await?;
1138        log::info!("Connected to business WebSocket");
1139
1140        if self.ws_business_stream_handle.is_none() {
1141            let stream = self.ws_business.stream();
1142            let emitter = self.emitter.clone();
1143            let state = Arc::clone(&self.ws_dispatch_state);
1144            let account_id = self.core.account_id;
1145            let instruments = self.ws_business.instruments_cache_arc();
1146            let clock = self.clock;
1147
1148            let handle = get_runtime().spawn(async move {
1149                let mut fee_cache: AHashMap<Ustr, Money> = AHashMap::new();
1150                let mut filled_qty_cache: AHashMap<Ustr, Quantity> = AHashMap::new();
1151                let mut order_state_cache: AHashMap<ClientOrderId, OrderStateSnapshot> =
1152                    AHashMap::new();
1153
1154                pin_mut!(stream);
1155
1156                while let Some(message) = stream.next().await {
1157                    dispatch_ws_message(
1158                        message,
1159                        &emitter,
1160                        &state,
1161                        account_id,
1162                        &instruments,
1163                        &mut fee_cache,
1164                        &mut filled_qty_cache,
1165                        &mut order_state_cache,
1166                        clock,
1167                    );
1168                }
1169            });
1170
1171            self.ws_business_stream_handle = Some(handle);
1172        }
1173
1174        for inst_type in &instrument_types {
1175            log::debug!("Subscribing to orders channel for {inst_type:?}");
1176            self.ws_private.subscribe_orders(*inst_type).await?;
1177
1178            if self.config.use_fills_channel {
1179                log::debug!("Subscribing to fills channel for {inst_type:?}");
1180                if let Err(e) = self.ws_private.subscribe_fills(*inst_type).await {
1181                    log::warn!("Failed to subscribe to fills channel ({inst_type:?}): {e}");
1182                }
1183            }
1184        }
1185
1186        self.ws_private.subscribe_account().await?;
1187
1188        if self.config.load_spreads {
1189            log::debug!("Subscribing to Nitro spread orders channel");
1190            self.ws_business.subscribe_spread_orders().await?;
1191        }
1192
1193        // Subscribe to algo orders on business WebSocket (OKX requires this endpoint)
1194        for inst_type in &instrument_types {
1195            if supports_algo_orders(*inst_type) {
1196                self.ws_business.subscribe_orders_algo(*inst_type).await?;
1197                self.ws_business.subscribe_algo_advance(*inst_type).await?;
1198            }
1199        }
1200
1201        let account_state = self
1202            .http_client
1203            .request_account_state(self.core.account_id)
1204            .await
1205            .context("failed to request OKX account state")?;
1206
1207        if !account_state.balances.is_empty() {
1208            log::debug!(
1209                "Received account state with {} balance(s)",
1210                account_state.balances.len()
1211            );
1212        }
1213        self.emitter.send_account_state(account_state);
1214
1215        // Wait for account to be registered in cache before completing connect
1216        self.await_account_registered(30.0).await?;
1217
1218        self.core.set_connected();
1219        log::info!("Connected: client_id={}", self.core.client_id);
1220        Ok(())
1221    }
1222
1223    async fn disconnect(&mut self) -> anyhow::Result<()> {
1224        if self.core.is_disconnected() {
1225            return Ok(());
1226        }
1227
1228        self.abort_pending_tasks();
1229        self.http_client.cancel_all_requests();
1230
1231        if let Err(e) = self.ws_private.close().await {
1232            log::warn!("Error closing private websocket: {e:?}");
1233        }
1234
1235        if let Err(e) = self.ws_business.close().await {
1236            log::warn!("Error closing business websocket: {e:?}");
1237        }
1238
1239        if let Some(handle) = self.ws_stream_handle.take() {
1240            handle.abort();
1241        }
1242
1243        if let Some(handle) = self.ws_business_stream_handle.take() {
1244            handle.abort();
1245        }
1246
1247        self.core.set_disconnected();
1248        log::info!("Disconnected: client_id={}", self.core.client_id);
1249        Ok(())
1250    }
1251
1252    fn query_account(&self, _cmd: QueryAccount) -> anyhow::Result<()> {
1253        self.update_account_state();
1254        Ok(())
1255    }
1256
1257    fn query_order(&self, cmd: QueryOrder) -> anyhow::Result<()> {
1258        let http_client = self.http_client.clone();
1259        let account_id = self.core.account_id;
1260        let emitter = self.emitter.clone();
1261        let instrument_id = cmd.instrument_id;
1262        let client_order_id = cmd.client_order_id;
1263        let venue_order_id = cmd.venue_order_id;
1264        let should_query_algo = !is_spread_instrument(instrument_id)
1265            && supports_algo_orders(okx_instrument_type_from_symbol(
1266                instrument_id.symbol.as_str(),
1267            ));
1268
1269        self.spawn_task("query_order", async move {
1270            let mut reports = match http_client
1271                .request_order_status_reports(
1272                    account_id,
1273                    None,
1274                    Some(instrument_id),
1275                    None,
1276                    None,
1277                    false,
1278                    None,
1279                )
1280                .await
1281            {
1282                Ok(r) => r,
1283                Err(e) => {
1284                    log::error!("OKX query_order failed to fetch orders: {e}");
1285                    Vec::new()
1286                }
1287            };
1288
1289            // Merge algo orders (stop, OCO, TP/SL, trailing) so query_order can
1290            // resolve conditional orders as well.
1291            if should_query_algo {
1292                match http_client
1293                    .request_algo_order_status_reports(
1294                        account_id,
1295                        None,
1296                        Some(instrument_id),
1297                        None,
1298                        Some(client_order_id),
1299                        None,
1300                        None,
1301                    )
1302                    .await
1303                {
1304                    Ok(mut algo) => reports.append(&mut algo),
1305                    Err(e) => {
1306                        log::warn!("OKX query_order algo lookup failed for {instrument_id}: {e}");
1307                    }
1308                }
1309            }
1310
1311            let Some(report) = select_query_order_report(reports, client_order_id, venue_order_id)
1312            else {
1313                log::warn!(
1314                    "OKX query_order found no order for client_order_id={client_order_id}, venue_order_id={venue_order_id:?}",
1315                );
1316                return Ok(());
1317            };
1318
1319            emitter.send_order_status_report(report);
1320            Ok(())
1321        });
1322        Ok(())
1323    }
1324
1325    fn generate_account_state(
1326        &self,
1327        balances: Vec<AccountBalance>,
1328        margins: Vec<MarginBalance>,
1329        reported: bool,
1330        ts_event: UnixNanos,
1331    ) -> anyhow::Result<()> {
1332        self.emitter
1333            .emit_account_state(balances, margins, reported, ts_event);
1334        Ok(())
1335    }
1336
1337    fn on_instrument(&mut self, instrument: InstrumentAny) {
1338        self.http_client.cache_instrument(instrument.clone());
1339        self.ws_private.cache_instrument(instrument.clone());
1340        self.ws_business.cache_instrument(instrument);
1341    }
1342
1343    fn start(&mut self) -> anyhow::Result<()> {
1344        if self.core.is_started() {
1345            return Ok(());
1346        }
1347
1348        let sender = get_exec_event_sender();
1349        self.emitter.set_sender(sender);
1350        self.core.set_started();
1351
1352        let http_client = self.http_client.clone();
1353        let ws_private = self.ws_private.clone();
1354        let ws_business = self.ws_business.clone();
1355        let instrument_types = self.config.instrument_types.clone();
1356        let instrument_families = self.config.instrument_families.clone();
1357
1358        get_runtime().spawn(async move {
1359            let mut all_instruments = Vec::new();
1360            let mut all_inst_id_codes = Vec::new();
1361
1362            for instrument_type in instrument_types {
1363                let Some(families) =
1364                    resolve_instrument_families(&instrument_families, instrument_type)
1365                else {
1366                    continue;
1367                };
1368
1369                if families.is_empty() {
1370                    match http_client.request_instruments(instrument_type, None).await {
1371                        Ok((instruments, inst_id_codes)) => {
1372                            if instruments.is_empty() {
1373                                log::warn!("No instruments returned for {instrument_type:?}");
1374                                continue;
1375                            }
1376                            http_client.cache_instruments(&instruments);
1377                            all_instruments.extend(instruments);
1378                            all_inst_id_codes.extend(inst_id_codes);
1379                        }
1380                        Err(e) => {
1381                            log::error!(
1382                                "Failed to request instruments for {instrument_type:?}: {e}"
1383                            );
1384                        }
1385                    }
1386                } else {
1387                    for family in &families {
1388                        match http_client
1389                            .request_instruments(instrument_type, Some(family.clone()))
1390                            .await
1391                        {
1392                            Ok((instruments, inst_id_codes)) => {
1393                                if instruments.is_empty() {
1394                                    log::warn!(
1395                                        "No instruments returned for {instrument_type:?} family {family}"
1396                                    );
1397                                    continue;
1398                                }
1399                                http_client.cache_instruments(&instruments);
1400                                all_instruments.extend(instruments);
1401                                all_inst_id_codes.extend(inst_id_codes);
1402                            }
1403                            Err(e) => {
1404                                log::error!(
1405                                    "Failed to request instruments for {instrument_type:?} family {family}: {e}"
1406                                );
1407                            }
1408                        }
1409                    }
1410                }
1411            }
1412
1413            if all_instruments.is_empty() {
1414                log::error!(
1415                    "Instrument bootstrap yielded no instruments, order submissions will fail"
1416                );
1417            } else {
1418                ws_private.cache_instruments(&all_instruments);
1419                ws_private.cache_inst_id_codes(all_inst_id_codes.clone());
1420                ws_business.cache_instruments(&all_instruments);
1421                ws_business.cache_inst_id_codes(all_inst_id_codes);
1422                log::debug!("Instruments initialized");
1423            }
1424        });
1425
1426        log::info!(
1427            "Started: client_id={}, account_id={}, account_type={:?}, trade_mode={:?}, instrument_types={:?}, use_fills_channel={}, environment={}, proxy_url={:?}",
1428            self.core.client_id,
1429            self.core.account_id,
1430            self.core.account_type,
1431            self.trade_mode,
1432            self.config.instrument_types,
1433            self.config.use_fills_channel,
1434            self.config.environment,
1435            self.config.proxy_url,
1436        );
1437        Ok(())
1438    }
1439
1440    fn stop(&mut self) -> anyhow::Result<()> {
1441        if self.core.is_stopped() {
1442            return Ok(());
1443        }
1444
1445        self.core.set_stopped();
1446        self.core.set_disconnected();
1447
1448        if let Some(handle) = self.ws_stream_handle.take() {
1449            handle.abort();
1450        }
1451
1452        if let Some(handle) = self.ws_business_stream_handle.take() {
1453            handle.abort();
1454        }
1455        self.abort_pending_tasks();
1456        log::info!("Stopped: client_id={}", self.core.client_id);
1457        Ok(())
1458    }
1459
1460    async fn generate_order_status_report(
1461        &self,
1462        cmd: &GenerateOrderStatusReport,
1463    ) -> anyhow::Result<Option<OrderStatusReport>> {
1464        let Some(instrument_id) = cmd.instrument_id else {
1465            log::warn!("generate_order_status_report requires instrument_id: {cmd:?}");
1466            return Ok(None);
1467        };
1468
1469        let mut reports = self
1470            .http_client
1471            .request_order_status_reports(
1472                self.core.account_id,
1473                None,
1474                Some(instrument_id),
1475                None,
1476                None,
1477                false,
1478                None,
1479            )
1480            .await?;
1481
1482        if !is_spread_instrument(instrument_id)
1483            && supports_algo_orders(okx_instrument_type_from_symbol(
1484                instrument_id.symbol.as_str(),
1485            ))
1486        {
1487            // Merge algo orders (stop, OCO, TP/SL, trailing). They live on a
1488            // separate OKX endpoint and would otherwise be dropped from
1489            // reconciliation, leaving stop/conditional orders unrecovered after
1490            // a restart.
1491            match self
1492                .http_client
1493                .request_algo_order_status_reports(
1494                    self.core.account_id,
1495                    None,
1496                    Some(instrument_id),
1497                    None,
1498                    cmd.client_order_id,
1499                    None,
1500                    None,
1501                )
1502                .await
1503            {
1504                Ok(mut algo_reports) => reports.append(&mut algo_reports),
1505                Err(e) => {
1506                    log::warn!(
1507                        "Failed to fetch algo order status reports for {instrument_id}: {e}"
1508                    );
1509                }
1510            }
1511        }
1512
1513        if let Some(client_order_id) = cmd.client_order_id {
1514            reports.retain(|report| report.client_order_id == Some(client_order_id));
1515        }
1516
1517        if let Some(venue_order_id) = cmd.venue_order_id {
1518            reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
1519        }
1520
1521        Ok(reports.into_iter().next())
1522    }
1523
1524    async fn generate_order_status_reports(
1525        &self,
1526        cmd: &GenerateOrderStatusReports,
1527    ) -> anyhow::Result<Vec<OrderStatusReport>> {
1528        let mut reports = Vec::new();
1529
1530        if let Some(instrument_id) = cmd.instrument_id {
1531            let mut fetched = self
1532                .http_client
1533                .request_order_status_reports(
1534                    self.core.account_id,
1535                    None,
1536                    Some(instrument_id),
1537                    None,
1538                    None,
1539                    false,
1540                    None,
1541                )
1542                .await?;
1543            reports.append(&mut fetched);
1544
1545            if !is_spread_instrument(instrument_id)
1546                && supports_algo_orders(okx_instrument_type_from_symbol(
1547                    instrument_id.symbol.as_str(),
1548                ))
1549            {
1550                // Merge algo orders for the requested instrument so reconciliation
1551                // recovers stop, OCO, TP/SL, and trailing orders alongside regular
1552                // ones. Failure here is logged but does not abort the regular
1553                // reconciliation; an algo-endpoint outage should not blank the
1554                // entire status report.
1555                match self
1556                    .http_client
1557                    .request_algo_order_status_reports(
1558                        self.core.account_id,
1559                        None,
1560                        Some(instrument_id),
1561                        None,
1562                        None,
1563                        None,
1564                        None,
1565                    )
1566                    .await
1567                {
1568                    Ok(mut algo) => reports.append(&mut algo),
1569                    Err(e) => {
1570                        log::warn!(
1571                            "Failed to fetch algo order status reports for {instrument_id}: {e}"
1572                        );
1573                    }
1574                }
1575            }
1576        } else {
1577            for inst_type in self.instrument_types() {
1578                let mut fetched = self
1579                    .http_client
1580                    .request_order_status_reports(
1581                        self.core.account_id,
1582                        Some(inst_type),
1583                        None,
1584                        None,
1585                        None,
1586                        false,
1587                        None,
1588                    )
1589                    .await?;
1590                reports.append(&mut fetched);
1591
1592                if supports_algo_orders(inst_type) {
1593                    match self
1594                        .http_client
1595                        .request_algo_order_status_reports(
1596                            self.core.account_id,
1597                            Some(inst_type),
1598                            None,
1599                            None,
1600                            None,
1601                            None,
1602                            None,
1603                        )
1604                        .await
1605                    {
1606                        Ok(mut algo) => reports.append(&mut algo),
1607                        Err(e) => log::warn!(
1608                            "Failed to fetch algo order status reports for {inst_type:?}: {e}"
1609                        ),
1610                    }
1611                }
1612            }
1613
1614            if self.config.load_spreads {
1615                match self
1616                    .http_client
1617                    .request_order_status_reports(
1618                        self.core.account_id,
1619                        None,
1620                        None,
1621                        None,
1622                        None,
1623                        false,
1624                        None,
1625                    )
1626                    .await
1627                {
1628                    Ok(mut spreads) => reports.append(&mut spreads),
1629                    Err(e) => log::warn!("Failed to fetch spread order status reports: {e}"),
1630                }
1631            }
1632        }
1633
1634        if cmd.open_only {
1635            reports.retain(|r| r.order_status.is_open());
1636        }
1637
1638        if let Some(start) = cmd.start {
1639            reports.retain(|r| r.ts_last >= start);
1640        }
1641
1642        if let Some(end) = cmd.end {
1643            reports.retain(|r| r.ts_last <= end);
1644        }
1645
1646        Ok(reports)
1647    }
1648
1649    async fn generate_fill_reports(
1650        &self,
1651        cmd: GenerateFillReports,
1652    ) -> anyhow::Result<Vec<FillReport>> {
1653        let start_dt = nanos_to_datetime(cmd.start);
1654        let end_dt = nanos_to_datetime(cmd.end);
1655        let mut reports = Vec::new();
1656
1657        if let Some(instrument_id) = cmd.instrument_id {
1658            let mut fetched = self
1659                .http_client
1660                .request_fill_reports(
1661                    self.core.account_id,
1662                    None,
1663                    Some(instrument_id),
1664                    start_dt,
1665                    end_dt,
1666                    None,
1667                )
1668                .await?;
1669            reports.append(&mut fetched);
1670        } else {
1671            for inst_type in self.instrument_types() {
1672                let mut fetched = self
1673                    .http_client
1674                    .request_fill_reports(
1675                        self.core.account_id,
1676                        Some(inst_type),
1677                        None,
1678                        start_dt,
1679                        end_dt,
1680                        None,
1681                    )
1682                    .await?;
1683                reports.append(&mut fetched);
1684            }
1685
1686            if self.config.load_spreads {
1687                match self
1688                    .http_client
1689                    .request_fill_reports(self.core.account_id, None, None, start_dt, end_dt, None)
1690                    .await
1691                {
1692                    Ok(mut spreads) => reports.append(&mut spreads),
1693                    Err(e) => log::warn!("Failed to fetch spread fill reports: {e}"),
1694                }
1695            }
1696        }
1697
1698        if let Some(venue_order_id) = cmd.venue_order_id {
1699            reports.retain(|report| report.venue_order_id.as_str() == venue_order_id.as_str());
1700        }
1701
1702        Ok(reports)
1703    }
1704
1705    async fn generate_position_status_reports(
1706        &self,
1707        cmd: &GeneratePositionStatusReports,
1708    ) -> anyhow::Result<Vec<PositionStatusReport>> {
1709        let mut reports = Vec::new();
1710
1711        // Query derivative positions (SWAP/FUTURES/OPTION) from /api/v5/account/positions
1712        // Note: The positions endpoint does not support Spot or Margin - those are handled separately
1713        if let Some(instrument_id) = cmd.instrument_id {
1714            if is_spread_instrument(instrument_id) {
1715                return Ok(reports);
1716            }
1717
1718            let inst_type = okx_instrument_type_from_symbol(instrument_id.symbol.as_str());
1719            if inst_type != OKXInstrumentType::Spot && inst_type != OKXInstrumentType::Margin {
1720                let mut fetched = self
1721                    .http_client
1722                    .request_position_status_reports(
1723                        self.core.account_id,
1724                        None,
1725                        Some(instrument_id),
1726                    )
1727                    .await?;
1728                reports.append(&mut fetched);
1729            }
1730        } else {
1731            for inst_type in self.instrument_types() {
1732                // Skip Spot and Margin - positions API only supports derivatives
1733                if inst_type == OKXInstrumentType::Spot || inst_type == OKXInstrumentType::Margin {
1734                    continue;
1735                }
1736                let mut fetched = self
1737                    .http_client
1738                    .request_position_status_reports(self.core.account_id, Some(inst_type), None)
1739                    .await?;
1740                reports.append(&mut fetched);
1741            }
1742        }
1743
1744        // Query spot margin positions from /api/v5/account/balance
1745        // Spot margin positions appear as balance sheet items (liab/spotInUseAmt fields)
1746        let mut margin_reports = self
1747            .http_client
1748            .request_spot_margin_position_reports(self.core.account_id)
1749            .await?;
1750
1751        if let Some(instrument_id) = cmd.instrument_id {
1752            margin_reports.retain(|report| report.instrument_id == instrument_id);
1753        }
1754
1755        reports.append(&mut margin_reports);
1756
1757        Ok(reports)
1758    }
1759
1760    async fn generate_mass_status(
1761        &self,
1762        lookback_mins: Option<u64>,
1763    ) -> anyhow::Result<Option<ExecutionMassStatus>> {
1764        log::info!("Generating ExecutionMassStatus (lookback_mins={lookback_mins:?})");
1765
1766        let ts_now = self.clock.get_time_ns();
1767
1768        let start = lookback_mins.map(|mins| {
1769            let lookback_ns = mins * 60 * 1_000_000_000;
1770            UnixNanos::from(ts_now.as_u64().saturating_sub(lookback_ns))
1771        });
1772
1773        let order_cmd = GenerateOrderStatusReportsBuilder::default()
1774            .ts_init(ts_now)
1775            .open_only(false) // get all orders for mass status
1776            .start(start)
1777            .build()
1778            .map_err(|e| anyhow::anyhow!("{e}"))?;
1779
1780        let fill_cmd = GenerateFillReportsBuilder::default()
1781            .ts_init(ts_now)
1782            .start(start)
1783            .build()
1784            .map_err(|e| anyhow::anyhow!("{e}"))?;
1785
1786        let position_cmd = GeneratePositionStatusReportsBuilder::default()
1787            .ts_init(ts_now)
1788            .start(start)
1789            .build()
1790            .map_err(|e| anyhow::anyhow!("{e}"))?;
1791
1792        let (order_reports, fill_reports, position_reports) = tokio::try_join!(
1793            self.generate_order_status_reports(&order_cmd),
1794            self.generate_fill_reports(fill_cmd),
1795            self.generate_position_status_reports(&position_cmd),
1796        )?;
1797
1798        log::info!("Received {} OrderStatusReports", order_reports.len());
1799        log::info!("Received {} FillReports", fill_reports.len());
1800        log::info!("Received {} PositionReports", position_reports.len());
1801
1802        let mut mass_status = ExecutionMassStatus::new(
1803            self.core.client_id,
1804            self.core.account_id,
1805            *OKX_VENUE,
1806            ts_now,
1807            None,
1808        );
1809
1810        mass_status.add_order_reports(order_reports);
1811        mass_status.add_fill_reports(fill_reports);
1812        mass_status.add_position_reports(position_reports);
1813
1814        Ok(Some(mass_status))
1815    }
1816
1817    fn submit_order(&self, cmd: SubmitOrder) -> anyhow::Result<()> {
1818        let route = {
1819            let cache = self.core.cache();
1820            let order = cache.try_order(&cmd.client_order_id)?;
1821
1822            if order.is_closed() {
1823                log::warn!("Cannot submit closed order {}", order.client_order_id());
1824                return Ok(());
1825            }
1826
1827            if let Err(reason) = validate_okx_client_order_id(cmd.client_order_id.as_str()) {
1828                let denied = OrderDeniedReason::InvalidClientOrderId { detail: reason };
1829                self.emitter.emit_order_denied(&order, &denied.to_string());
1830                return Ok(());
1831            }
1832
1833            let order_type = order.order_type();
1834            let route = self.submit_order_route(cmd.instrument_id, order_type)?;
1835
1836            log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
1837            self.emitter.emit_order_submitted(&order);
1838
1839            route
1840        };
1841
1842        match route {
1843            OrderCommandRoute::RegularWs => self.submit_regular_order(&cmd),
1844            OrderCommandRoute::AlgoHttp => self.submit_conditional_order(&cmd),
1845            OrderCommandRoute::SpreadHttp => self.submit_order_http(&cmd),
1846        }
1847    }
1848
1849    fn submit_order_list(&self, cmd: SubmitOrderList) -> anyhow::Result<()> {
1850        if is_spread_instrument(cmd.instrument_id) {
1851            let cache = self.core.cache();
1852            let denied = OrderDeniedReason::UnsupportedOrderList {
1853                detail: "spread instruments are not supported in order lists".to_string(),
1854            }
1855            .to_string();
1856
1857            for client_order_id in &cmd.order_list.client_order_ids {
1858                let order = cache.try_order(client_order_id)?;
1859                self.emitter.emit_order_denied(&order, &denied);
1860            }
1861            return Ok(());
1862        }
1863
1864        let inst_type = okx_instrument_type_from_symbol(cmd.instrument_id.symbol.as_str());
1865
1866        // Validate all orders before emitting any submitted events
1867        let cache = self.core.cache();
1868
1869        // Pre-validate every clOrdId so an invalid leg denies the whole list atomically;
1870        // otherwise sibling legs would be left in the cache without a terminal event.
1871        let invalid: Vec<(ClientOrderId, String)> = cmd
1872            .order_list
1873            .client_order_ids
1874            .iter()
1875            .filter_map(|cid| {
1876                validate_okx_client_order_id(cid.as_str())
1877                    .err()
1878                    .map(|r| (*cid, r))
1879            })
1880            .collect();
1881
1882        if !invalid.is_empty() {
1883            let order_list_id = cmd.order_list.id;
1884            for client_order_id in &cmd.order_list.client_order_ids {
1885                let order = cache.try_order(client_order_id)?;
1886                let denied = invalid
1887                    .iter()
1888                    .find(|(cid, _)| cid == client_order_id)
1889                    .map_or_else(
1890                        || OrderDeniedReason::OrderListDenied { order_list_id },
1891                        |(_, r)| OrderDeniedReason::InvalidClientOrderId { detail: r.clone() },
1892                    );
1893                self.emitter.emit_order_denied(&order, &denied.to_string());
1894            }
1895            return Ok(());
1896        }
1897
1898        for client_order_id in &cmd.order_list.client_order_ids {
1899            let order = cache.try_order(client_order_id)?;
1900
1901            if self.is_conditional_order(order.order_type()) {
1902                anyhow::bail!("Conditional orders not supported in order lists: {client_order_id}");
1903            }
1904
1905            if order.time_in_force() != TimeInForce::Gtc {
1906                anyhow::bail!(
1907                    "Only GTC orders supported in order lists: {client_order_id} has {:?}",
1908                    order.time_in_force()
1909                );
1910            }
1911        }
1912
1913        // Build batch payload and emit submitted events
1914        let mut batch_orders = Vec::new();
1915        let speed_bump = get_param_as_string(&cmd.params, "speed_bump");
1916        let outcome = get_param_as_string(&cmd.params, "outcome");
1917
1918        for client_order_id in &cmd.order_list.client_order_ids {
1919            let order = cache.order(client_order_id).expect("validated above");
1920
1921            batch_orders.push((
1922                inst_type,
1923                cmd.instrument_id,
1924                self.trade_mode_for_order(cmd.instrument_id, &cmd.params),
1925                order.client_order_id(),
1926                order.order_side(),
1927                None, // position_side: WS client defaults to Net for derivatives
1928                order.order_type(),
1929                order.quantity(),
1930                order.price(),
1931                order.trigger_price(),
1932                Some(order.is_post_only()),
1933                Some(order.is_reduce_only()),
1934                speed_bump.clone(),
1935                outcome.clone(),
1936            ));
1937
1938            self.ws_dispatch_state.order_identities.insert(
1939                order.client_order_id(),
1940                OrderIdentity {
1941                    instrument_id: cmd.instrument_id,
1942                    strategy_id: order.strategy_id(),
1943                    order_side: order.order_side(),
1944                    order_type: order.order_type(),
1945                },
1946            );
1947
1948            log::debug!("OrderSubmitted client_order_id={}", order.client_order_id());
1949            self.emitter.emit_order_submitted(&order);
1950        }
1951
1952        drop(cache);
1953
1954        let ws_private = self.ws_private.clone();
1955        let emitter = self.emitter.clone();
1956        let clock = self.clock;
1957        let instrument_id = cmd.instrument_id;
1958        let strategy_id = cmd.strategy_id;
1959        let client_order_ids: Vec<_> = cmd.order_list.client_order_ids;
1960        let dispatch_state = Arc::clone(&self.ws_dispatch_state);
1961
1962        self.spawn_task("batch_submit_orders", async move {
1963            let result = ws_private
1964                .batch_submit_orders(batch_orders)
1965                .await;
1966
1967            if let Err(e) = result {
1968                if is_okx_ws_local_command_failure(&e) {
1969                    let ts_event = clock.get_time_ns();
1970
1971                    for cid in &client_order_ids {
1972                        dispatch_state.order_identities.remove(cid);
1973                        emitter.emit_order_rejected_event(
1974                            strategy_id,
1975                            instrument_id,
1976                            *cid,
1977                            &format!("batch-submit-error: {e}"),
1978                            ts_event,
1979                            false,
1980                        );
1981                    }
1982                } else {
1983                    log::warn!(
1984                        "Ambiguous batch submit failure for {} orders on {instrument_id}, awaiting reconciliation: {e}",
1985                        client_order_ids.len()
1986                    );
1987                }
1988                return Err(anyhow::Error::new(e).context("batch submit orders failed"));
1989            }
1990
1991            Ok(())
1992        });
1993
1994        Ok(())
1995    }
1996
1997    fn modify_order(&self, cmd: ModifyOrder) -> anyhow::Result<()> {
1998        if is_spread_instrument(cmd.instrument_id) {
1999            self.emitter.emit_order_modify_rejected_event(
2000                cmd.strategy_id,
2001                cmd.instrument_id,
2002                cmd.client_order_id,
2003                cmd.venue_order_id,
2004                "OKX spread orders do not support modify requests",
2005                self.clock.get_time_ns(),
2006            );
2007            return Ok(());
2008        }
2009
2010        self.ensure_order_identity(cmd.client_order_id, cmd.strategy_id, cmd.instrument_id);
2011
2012        let ws_private = self.ws_private.clone();
2013        let command = cmd.clone();
2014
2015        let new_px_usd = get_param_as_string(&cmd.params, "px_usd");
2016        let new_px_vol = get_param_as_string(&cmd.params, "px_vol");
2017        let speed_bump = get_param_as_string(&cmd.params, "speed_bump");
2018
2019        let emitter = self.emitter.clone();
2020        let clock = self.clock;
2021
2022        self.spawn_task("modify_order", async move {
2023            let result = ws_private
2024                .modify_order(
2025                    command.trader_id,
2026                    command.strategy_id,
2027                    command.instrument_id,
2028                    Some(command.client_order_id),
2029                    command.price,
2030                    command.quantity,
2031                    command.venue_order_id,
2032                    new_px_usd,
2033                    new_px_vol,
2034                    speed_bump,
2035                )
2036                .await;
2037
2038            if let Err(e) = result {
2039                if is_okx_ws_local_command_failure(&e) {
2040                    let ts_event = clock.get_time_ns();
2041                    emitter.emit_order_modify_rejected_event(
2042                        command.strategy_id,
2043                        command.instrument_id,
2044                        command.client_order_id,
2045                        command.venue_order_id,
2046                        &format!("modify-order-error: {e}"),
2047                        ts_event,
2048                    );
2049                } else {
2050                    log::warn!(
2051                        "Ambiguous modify failure for {}, awaiting reconciliation: {e}",
2052                        command.client_order_id
2053                    );
2054                }
2055                return Err(anyhow::Error::new(e).context("modify order failed"));
2056            }
2057
2058            Ok(())
2059        });
2060
2061        Ok(())
2062    }
2063
2064    fn cancel_order(&self, cmd: CancelOrder) -> anyhow::Result<()> {
2065        let route = {
2066            let cache = self.core.cache();
2067            let order_state = cache
2068                .order(&cmd.client_order_id)
2069                .map(|order| (order.order_type(), order.is_triggered()));
2070            self.cancel_order_route(cmd.instrument_id, order_state)
2071        };
2072
2073        match route {
2074            OrderCommandRoute::RegularWs => self.cancel_ws_order(&cmd),
2075            OrderCommandRoute::AlgoHttp => self.cancel_algo_order(&cmd),
2076            OrderCommandRoute::SpreadHttp => self.cancel_order_http(&cmd),
2077        }
2078        Ok(())
2079    }
2080
2081    fn cancel_all_orders(&self, cmd: CancelAllOrders) -> anyhow::Result<()> {
2082        match self.cancel_all_orders_route(cmd.instrument_id) {
2083            CancelAllOrdersRoute::SpreadHttp | CancelAllOrdersRoute::MassCancelHttp => {
2084                self.mass_cancel_instrument(cmd.instrument_id);
2085                Ok(())
2086            }
2087            CancelAllOrdersRoute::BatchWs => {
2088                let cache = self.core.cache();
2089                let open_orders =
2090                    cache.orders_open(None, Some(&cmd.instrument_id), None, None, None);
2091
2092                if open_orders.is_empty() {
2093                    log::debug!("No open orders to cancel for {}", cmd.instrument_id);
2094                    return Ok(());
2095                }
2096
2097                let mut regular_payload = Vec::new();
2098                let mut regular_cancel_contexts = Vec::new();
2099                let mut algo_orders: Vec<(
2100                    InstrumentId,
2101                    ClientOrderId,
2102                    Option<VenueOrderId>,
2103                    TraderId,
2104                    StrategyId,
2105                )> = Vec::new();
2106
2107                for order in &open_orders {
2108                    let order_state = Some((order.order_type(), order.is_triggered()));
2109                    match self.cancel_order_route(order.instrument_id(), order_state) {
2110                        OrderCommandRoute::RegularWs => {
2111                            self.ensure_order_identity(
2112                                order.client_order_id(),
2113                                order.strategy_id(),
2114                                order.instrument_id(),
2115                            );
2116                            regular_payload.push((
2117                                order.instrument_id(),
2118                                Some(order.client_order_id()),
2119                                order.venue_order_id(),
2120                            ));
2121                            regular_cancel_contexts.push((
2122                                order.client_order_id(),
2123                                order.instrument_id(),
2124                                order.strategy_id(),
2125                            ));
2126                        }
2127                        OrderCommandRoute::AlgoHttp => {
2128                            algo_orders.push((
2129                                order.instrument_id(),
2130                                order.client_order_id(),
2131                                order.venue_order_id(),
2132                                order.trader_id(),
2133                                order.strategy_id(),
2134                            ));
2135                        }
2136                        OrderCommandRoute::SpreadHttp => {}
2137                    }
2138                }
2139                drop(open_orders);
2140                drop(cache);
2141
2142                log::debug!(
2143                    "Canceling {} regular orders and {} algo orders for {}",
2144                    regular_payload.len(),
2145                    algo_orders.len(),
2146                    cmd.instrument_id
2147                );
2148
2149                if !regular_payload.is_empty() {
2150                    let ws_private = self.ws_private.clone();
2151
2152                    self.spawn_task("batch_cancel_orders", async move {
2153                        if let Err(e) = ws_private.batch_cancel_orders(regular_payload).await {
2154                            if is_okx_ws_local_command_failure(&e) {
2155                                log::warn!(
2156                                    "Batch cancel command failed local validation for {} orders: {e}",
2157                                    regular_cancel_contexts.len()
2158                                );
2159                            } else {
2160                                log::warn!(
2161                                    "Ambiguous batch cancel failure for {} orders, awaiting reconciliation: {e}",
2162                                    regular_cancel_contexts.len()
2163                                );
2164                            }
2165                            return Err(anyhow::Error::new(e).context("batch cancel orders failed"));
2166                        }
2167                        Ok(())
2168                    });
2169                }
2170
2171                // OKX doesn't support algo cancel via private WebSocket, must use HTTP
2172                if !algo_orders.is_empty() {
2173                    let items: Vec<_> = algo_orders
2174                        .into_iter()
2175                        .map(
2176                            |(
2177                                instrument_id,
2178                                client_order_id,
2179                                venue_order_id,
2180                                _trader_id,
2181                                strategy_id,
2182                            )| {
2183                                let request = OKXCancelAlgoOrderRequest {
2184                                    inst_id: instrument_id.symbol.to_string(),
2185                                    inst_id_code: None,
2186                                    algo_id: venue_order_id.map(|id| id.to_string()),
2187                                    algo_cl_ord_id: if venue_order_id.is_none() {
2188                                        Some(client_order_id.to_string())
2189                                    } else {
2190                                        None
2191                                    },
2192                                };
2193                                let ctx = AlgoCancelContext {
2194                                    client_order_id,
2195                                    instrument_id,
2196                                    strategy_id,
2197                                    venue_order_id,
2198                                };
2199                                (request, ctx)
2200                            },
2201                        )
2202                        .collect();
2203                    self.dispatch_algo_cancels(items);
2204                }
2205
2206                Ok(())
2207            }
2208        }
2209    }
2210
2211    fn batch_cancel_orders(&self, cmd: BatchCancelOrders) -> anyhow::Result<()> {
2212        let cache = self.core.cache();
2213
2214        let mut regular_payload = Vec::new();
2215        let mut algo_orders = Vec::new();
2216        let mut http_orders = Vec::new();
2217
2218        for cancel in &cmd.cancels {
2219            let order_state = cache
2220                .order(&cancel.client_order_id)
2221                .map(|order| (order.order_type(), order.is_triggered()));
2222
2223            match self.cancel_order_route(cancel.instrument_id, order_state) {
2224                OrderCommandRoute::RegularWs => {
2225                    self.ensure_order_identity(
2226                        cancel.client_order_id,
2227                        cancel.strategy_id,
2228                        cancel.instrument_id,
2229                    );
2230                    regular_payload.push((
2231                        cancel.instrument_id,
2232                        Some(cancel.client_order_id),
2233                        cancel.venue_order_id,
2234                    ));
2235                }
2236                OrderCommandRoute::AlgoHttp => algo_orders.push(cancel.clone()),
2237                OrderCommandRoute::SpreadHttp => {
2238                    self.ensure_order_identity(
2239                        cancel.client_order_id,
2240                        cancel.strategy_id,
2241                        cancel.instrument_id,
2242                    );
2243                    http_orders.push((
2244                        cancel.client_order_id,
2245                        cancel.instrument_id,
2246                        cancel.strategy_id,
2247                        cancel.venue_order_id,
2248                    ));
2249                }
2250            }
2251        }
2252        drop(cache);
2253
2254        if !regular_payload.is_empty() {
2255            let ws_private = self.ws_private.clone();
2256            let cancel_contexts: Vec<_> = cmd
2257                .cancels
2258                .iter()
2259                .filter(|c| {
2260                    regular_payload
2261                        .iter()
2262                        .any(|(_, cid, _)| *cid == Some(c.client_order_id))
2263                })
2264                .map(|c| (c.client_order_id, c.instrument_id, c.strategy_id))
2265                .collect();
2266
2267            self.spawn_task("batch_cancel_orders", async move {
2268                if let Err(e) = ws_private.batch_cancel_orders(regular_payload).await {
2269                    if is_okx_ws_local_command_failure(&e) {
2270                        log::warn!(
2271                            "Batch cancel command failed local validation for {} orders: {e}",
2272                            cancel_contexts.len()
2273                        );
2274                    } else {
2275                        log::warn!(
2276                            "Ambiguous batch cancel failure for {} orders, awaiting reconciliation: {e}",
2277                            cancel_contexts.len()
2278                        );
2279                    }
2280                    return Err(anyhow::Error::new(e).context("batch cancel orders failed"));
2281                }
2282                Ok(())
2283            });
2284        }
2285
2286        // OKX doesn't support algo cancel via private WebSocket, must use HTTP
2287        if !algo_orders.is_empty() {
2288            let items: Vec<_> = algo_orders
2289                .into_iter()
2290                .map(|cancel| {
2291                    let request = OKXCancelAlgoOrderRequest {
2292                        inst_id: cancel.instrument_id.symbol.to_string(),
2293                        inst_id_code: None,
2294                        algo_id: cancel.venue_order_id.map(|id| id.to_string()),
2295                        algo_cl_ord_id: if cancel.venue_order_id.is_none() {
2296                            Some(cancel.client_order_id.to_string())
2297                        } else {
2298                            None
2299                        },
2300                    };
2301                    let ctx = AlgoCancelContext {
2302                        client_order_id: cancel.client_order_id,
2303                        instrument_id: cancel.instrument_id,
2304                        strategy_id: cancel.strategy_id,
2305                        venue_order_id: cancel.venue_order_id,
2306                    };
2307                    (request, ctx)
2308                })
2309                .collect();
2310            self.dispatch_algo_cancels(items);
2311        }
2312
2313        if !http_orders.is_empty() {
2314            let client = self.http_client.clone();
2315            let emitter = self.emitter.clone();
2316            let clock = self.clock;
2317
2318            self.spawn_task("cancel_http_orders", async move {
2319                for (client_order_id, instrument_id, strategy_id, venue_order_id) in http_orders {
2320                    if let Err(e) = client
2321                        .cancel_order(instrument_id, Some(client_order_id), venue_order_id)
2322                        .await
2323                    {
2324                        if is_okx_http_structured_venue_rejection(&e) {
2325                            let ts_event = clock.get_time_ns();
2326                            emitter.emit_order_cancel_rejected_event(
2327                                strategy_id,
2328                                instrument_id,
2329                                client_order_id,
2330                                venue_order_id,
2331                                &format!("cancel-http-order-error: {e}"),
2332                                ts_event,
2333                            );
2334                        } else if is_okx_http_local_command_failure(&e) {
2335                            log::warn!(
2336                                "HTTP cancel command failed local validation for {client_order_id}: {e}"
2337                            );
2338                        } else {
2339                            log::warn!(
2340                                "Ambiguous HTTP cancel failure for {client_order_id}, awaiting reconciliation: {e}"
2341                            );
2342                        }
2343                    }
2344                }
2345                Ok(())
2346            });
2347        }
2348
2349        Ok(())
2350    }
2351}
2352
2353#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2354enum OrderCommandRoute {
2355    RegularWs,
2356    AlgoHttp,
2357    SpreadHttp,
2358}
2359
2360#[derive(Debug, Clone, Copy, PartialEq, Eq)]
2361enum CancelAllOrdersRoute {
2362    BatchWs,
2363    MassCancelHttp,
2364    SpreadHttp,
2365}
2366
2367fn is_okx_http_structured_venue_rejection(error: &OKXHttpError) -> bool {
2368    matches!(error, OKXHttpError::OkxError { .. })
2369}
2370
2371fn is_okx_http_local_command_failure(error: &OKXHttpError) -> bool {
2372    match error {
2373        OKXHttpError::MissingCredentials => true,
2374        OKXHttpError::ValidationError(message) => !is_ambiguous_okx_http_failure(message),
2375        _ => false,
2376    }
2377}
2378
2379fn is_okx_ws_local_command_failure(error: &OKXWsError) -> bool {
2380    match error {
2381        OKXWsError::ClientError(message) => !is_ambiguous_okx_ws_client_failure(message),
2382        OKXWsError::JsonError(_) => true,
2383        _ => false,
2384    }
2385}
2386
2387fn is_ambiguous_okx_http_failure(message: &str) -> bool {
2388    contains_any_ignore_ascii_case(
2389        message,
2390        &[
2391            "empty response",
2392            "timeout",
2393            "timed out",
2394            "retry",
2395            "request canceled",
2396            "network",
2397            "unexpected http status",
2398            "status code",
2399            "failed to deserialize",
2400            "failed to parse",
2401        ],
2402    )
2403}
2404
2405fn is_ambiguous_okx_ws_client_failure(message: &str) -> bool {
2406    contains_any_ignore_ascii_case(
2407        message,
2408        &[
2409            "handler not available",
2410            "no active websocket client",
2411            "send failed",
2412            "timeout",
2413            "timed out",
2414            "retry",
2415            "connection",
2416            "network",
2417        ],
2418    )
2419}
2420
2421fn contains_any_ignore_ascii_case(value: &str, needles: &[&str]) -> bool {
2422    needles.iter().any(|needle| {
2423        value
2424            .as_bytes()
2425            .windows(needle.len())
2426            .any(|window| window.eq_ignore_ascii_case(needle.as_bytes()))
2427    })
2428}
2429
2430fn get_param_as_string(params: &Option<Params>, key: &str) -> Option<String> {
2431    params.as_ref().and_then(|p| {
2432        p.get(key).and_then(|v| {
2433            v.as_str()
2434                .map(ToString::to_string)
2435                .or_else(|| v.as_f64().map(|n| n.to_string()))
2436        })
2437    })
2438}
2439
2440fn supports_algo_orders(instrument_type: OKXInstrumentType) -> bool {
2441    !matches!(
2442        instrument_type,
2443        OKXInstrumentType::Option | OKXInstrumentType::Events
2444    )
2445}
2446
2447fn is_spread_instrument(instrument_id: InstrumentId) -> bool {
2448    is_okx_spread_symbol(instrument_id.symbol.as_str())
2449}
2450
2451// Picks the report that best answers the query. Tiered so a strong signal
2452// wins over a weak one regardless of ordering in the merged result set:
2453//   1. Exact `client_order_id` match.
2454//   2. Exact `venue_order_id` match (rare: only when the cached vid is
2455//      still valid; OKX rotates venue_order_id once an algo order triggers).
2456//
2457// Triggered-algo recovery is handled by the algo endpoint in the caller,
2458// which queries by algo_cl_ord_id and returns the parent's algo record
2459// directly. `linked_order_ids` is deliberately not consulted here because
2460// it is also populated with attached TP/SL child ids on the parent order,
2461// which would otherwise let a query for a child match the parent's report.
2462fn select_query_order_report(
2463    reports: Vec<OrderStatusReport>,
2464    client_order_id: ClientOrderId,
2465    venue_order_id: Option<VenueOrderId>,
2466) -> Option<OrderStatusReport> {
2467    let mut by_vid: Option<OrderStatusReport> = None;
2468
2469    for report in reports {
2470        if report.client_order_id == Some(client_order_id) {
2471            return Some(report);
2472        }
2473
2474        if by_vid.is_none()
2475            && venue_order_id
2476                .as_ref()
2477                .is_some_and(|vid| report.venue_order_id.as_str() == vid.as_str())
2478        {
2479            by_vid = Some(report);
2480        }
2481    }
2482
2483    by_vid
2484}
2485
2486#[cfg(test)]
2487mod tests {
2488    use std::{cell::RefCell, rc::Rc};
2489
2490    use nautilus_common::cache::Cache;
2491    use nautilus_model::{enums::OrderStatus, instruments::Instrument};
2492    use rstest::rstest;
2493    use serde_json::Value;
2494
2495    use super::*;
2496
2497    fn build_config(
2498        margin_mode: Option<OKXMarginMode>,
2499        use_spot_margin: bool,
2500    ) -> OKXExecClientConfig {
2501        OKXExecClientConfig {
2502            margin_mode,
2503            use_spot_margin,
2504            ..OKXExecClientConfig::default()
2505        }
2506    }
2507
2508    #[rstest]
2509    #[case::spot(OKXInstrumentType::Spot, true)]
2510    #[case::margin(OKXInstrumentType::Margin, true)]
2511    #[case::swap(OKXInstrumentType::Swap, true)]
2512    #[case::futures(OKXInstrumentType::Futures, true)]
2513    #[case::option(OKXInstrumentType::Option, false)]
2514    #[case::events(OKXInstrumentType::Events, false)]
2515    fn test_supports_algo_orders(
2516        #[case] instrument_type: OKXInstrumentType,
2517        #[case] expected: bool,
2518    ) {
2519        assert_eq!(supports_algo_orders(instrument_type), expected);
2520    }
2521
2522    #[rstest]
2523    #[case::cash_no_spot_margin(AccountType::Cash, None, false, OKXTradeMode::Cash)]
2524    #[case::cash_spot_margin_cross(
2525        AccountType::Cash,
2526        Some(OKXMarginMode::Cross),
2527        true,
2528        OKXTradeMode::Cross
2529    )]
2530    #[case::cash_spot_margin_isolated(
2531        AccountType::Cash,
2532        Some(OKXMarginMode::Isolated),
2533        true,
2534        OKXTradeMode::Isolated
2535    )]
2536    #[case::cash_spot_margin_none(AccountType::Cash, None, true, OKXTradeMode::Isolated)]
2537    #[case::margin_cross(
2538        AccountType::Margin,
2539        Some(OKXMarginMode::Cross),
2540        false,
2541        OKXTradeMode::Cross
2542    )]
2543    #[case::margin_isolated(
2544        AccountType::Margin,
2545        Some(OKXMarginMode::Isolated),
2546        false,
2547        OKXTradeMode::Isolated
2548    )]
2549    #[case::margin_none(AccountType::Margin, None, false, OKXTradeMode::Isolated)]
2550    fn test_derive_default_trade_mode(
2551        #[case] account_type: AccountType,
2552        #[case] margin_mode: Option<OKXMarginMode>,
2553        #[case] use_spot_margin: bool,
2554        #[case] expected: OKXTradeMode,
2555    ) {
2556        let config = build_config(margin_mode, use_spot_margin);
2557
2558        let result = OKXExecutionClient::derive_default_trade_mode(account_type, &config);
2559
2560        assert_eq!(result, expected);
2561    }
2562
2563    #[rstest]
2564    #[case::spot_no_margin("BTC-USDT", None, false, OKXTradeMode::Cash)]
2565    #[case::spot_cross_margin("BTC-USDT", Some(OKXMarginMode::Cross), true, OKXTradeMode::Cross)]
2566    #[case::spot_isolated_margin(
2567        "ETH-USDT",
2568        Some(OKXMarginMode::Isolated),
2569        true,
2570        OKXTradeMode::Isolated
2571    )]
2572    #[case::spot_margin_no_mode("BTC-USDT", None, true, OKXTradeMode::Isolated)]
2573    #[case::swap_cross(
2574        "BTC-USDT-SWAP",
2575        Some(OKXMarginMode::Cross),
2576        false,
2577        OKXTradeMode::Cross
2578    )]
2579    #[case::swap_isolated(
2580        "BTC-USDT-SWAP",
2581        Some(OKXMarginMode::Isolated),
2582        false,
2583        OKXTradeMode::Isolated
2584    )]
2585    #[case::swap_no_mode("ETH-USDT-SWAP", None, false, OKXTradeMode::Isolated)]
2586    #[case::futures_cross(
2587        "BTC-USDT-250328",
2588        Some(OKXMarginMode::Cross),
2589        false,
2590        OKXTradeMode::Cross
2591    )]
2592    #[case::futures_isolated("BTC-USDT-250328", None, false, OKXTradeMode::Isolated)]
2593    #[case::option_cross(
2594        "BTC-USD-250328-50000-C",
2595        Some(OKXMarginMode::Cross),
2596        false,
2597        OKXTradeMode::Cross
2598    )]
2599    #[case::option_isolated("BTC-USD-250328-50000-C", None, false, OKXTradeMode::Isolated)]
2600    fn test_derive_trade_mode_for_instrument(
2601        #[case] symbol: &str,
2602        #[case] margin_mode: Option<OKXMarginMode>,
2603        #[case] use_spot_margin: bool,
2604        #[case] expected: OKXTradeMode,
2605    ) {
2606        let instrument_id = InstrumentId::from(format!("{symbol}.OKX").as_str());
2607
2608        let result = derive_trade_mode_for_instrument(instrument_id, margin_mode, use_spot_margin);
2609
2610        assert_eq!(result, expected);
2611    }
2612
2613    #[rstest]
2614    #[case::override_to_cross("cross", OKXTradeMode::Cross)]
2615    #[case::override_to_cash("cash", OKXTradeMode::Cash)]
2616    #[case::override_to_isolated("isolated", OKXTradeMode::Isolated)]
2617    #[case::override_to_spot_isolated("spot_isolated", OKXTradeMode::SpotIsolated)]
2618    #[case::case_insensitive("CROSS", OKXTradeMode::Cross)]
2619    fn test_td_mode_param_override(#[case] td_mode_value: &str, #[case] expected: OKXTradeMode) {
2620        let mut params = Params::new();
2621        params.insert(
2622            "td_mode".to_string(),
2623            Value::String(td_mode_value.to_string()),
2624        );
2625
2626        let result = get_param_as_string(&Some(params), "td_mode")
2627            .and_then(|s| s.parse::<OKXTradeMode>().ok());
2628
2629        assert_eq!(result, Some(expected));
2630    }
2631
2632    #[rstest]
2633    fn test_td_mode_param_invalid_falls_through() {
2634        let mut params = Params::new();
2635        params.insert("td_mode".to_string(), Value::String("invalid".to_string()));
2636
2637        let result = get_param_as_string(&Some(params), "td_mode")
2638            .and_then(|s| s.parse::<OKXTradeMode>().ok());
2639
2640        assert_eq!(result, None);
2641    }
2642
2643    #[rstest]
2644    fn test_td_mode_param_absent_falls_through() {
2645        let result = get_param_as_string(&None, "td_mode");
2646
2647        assert_eq!(result, None);
2648    }
2649
2650    #[rstest]
2651    fn test_close_fraction_present_sets_reduce_only_true() {
2652        let mut params = Params::new();
2653        params.insert("close_fraction".to_string(), Value::String("1".to_string()));
2654        let params = Some(params);
2655
2656        let close_fraction = get_param_as_string(&params, "close_fraction");
2657        let is_reduce_only = false;
2658        let reduce_only = if close_fraction.is_some() {
2659            Some(true)
2660        } else {
2661            Some(is_reduce_only)
2662        };
2663
2664        assert_eq!(close_fraction, Some("1".to_string()));
2665        assert_eq!(reduce_only, Some(true));
2666    }
2667
2668    #[rstest]
2669    fn test_close_fraction_absent_preserves_reduce_only() {
2670        let params: Option<Params> = None;
2671
2672        let close_fraction = get_param_as_string(&params, "close_fraction");
2673        let is_reduce_only = false;
2674        let reduce_only = if close_fraction.is_some() {
2675            Some(true)
2676        } else {
2677            Some(is_reduce_only)
2678        };
2679
2680        assert_eq!(close_fraction, None);
2681        assert_eq!(reduce_only, Some(false));
2682    }
2683
2684    #[rstest]
2685    fn test_close_fraction_absent_with_reduce_only_true() {
2686        let params: Option<Params> = None;
2687
2688        let close_fraction = get_param_as_string(&params, "close_fraction");
2689        let is_reduce_only = true;
2690        let reduce_only = if close_fraction.is_some() {
2691            Some(true)
2692        } else {
2693            Some(is_reduce_only)
2694        };
2695
2696        assert_eq!(close_fraction, None);
2697        assert_eq!(reduce_only, Some(true));
2698    }
2699
2700    fn make_query_order_report(cid: Option<&str>, vid: &str) -> OrderStatusReport {
2701        OrderStatusReport::new(
2702            AccountId::from("OKX-001"),
2703            InstrumentId::from("BTC-USDT.OKX"),
2704            cid.map(ClientOrderId::from),
2705            VenueOrderId::from(vid),
2706            OrderSide::Buy,
2707            OrderType::Limit,
2708            TimeInForce::Gtc,
2709            OrderStatus::Accepted,
2710            Quantity::new(1.0, 0),
2711            Quantity::zero(0),
2712            UnixNanos::default(),
2713            UnixNanos::default(),
2714            UnixNanos::default(),
2715            None,
2716        )
2717    }
2718
2719    fn with_linked(mut report: OrderStatusReport, linked: &[&str]) -> OrderStatusReport {
2720        report.linked_order_ids = Some(linked.iter().map(|s| ClientOrderId::from(*s)).collect());
2721        report
2722    }
2723
2724    #[rstest]
2725    fn test_select_query_order_report_matches_client_order_id() {
2726        let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
2727        let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
2728        assert_eq!(
2729            selected.and_then(|r| r.client_order_id),
2730            Some(ClientOrderId::from("O-001"))
2731        );
2732    }
2733
2734    #[rstest]
2735    fn test_select_query_order_report_client_wins_over_venue_mismatch() {
2736        let reports = vec![make_query_order_report(Some("O-001"), "V-1")];
2737        let selected = select_query_order_report(
2738            reports,
2739            ClientOrderId::from("O-001"),
2740            Some(VenueOrderId::from("V-OTHER")),
2741        );
2742        assert_eq!(
2743            selected.and_then(|r| r.client_order_id),
2744            Some(ClientOrderId::from("O-001"))
2745        );
2746    }
2747
2748    #[rstest]
2749    fn test_select_query_order_report_falls_back_to_venue_order_id() {
2750        // Algo child trigger: report's client_order_id is the child, the
2751        // command still carries the pre-trigger venue_order_id.
2752        let reports = vec![make_query_order_report(Some("O-CHILD"), "V-1")];
2753        let selected = select_query_order_report(
2754            reports,
2755            ClientOrderId::from("O-PARENT"),
2756            Some(VenueOrderId::from("V-1")),
2757        );
2758        assert_eq!(
2759            selected.map(|r| r.venue_order_id.as_str().to_string()),
2760            Some("V-1".to_string()),
2761        );
2762    }
2763
2764    #[rstest]
2765    fn test_select_query_order_report_rejects_when_nothing_matches() {
2766        let reports = vec![make_query_order_report(Some("O-OTHER"), "V-OTHER")];
2767        let selected = select_query_order_report(
2768            reports,
2769            ClientOrderId::from("O-001"),
2770            Some(VenueOrderId::from("V-1")),
2771        );
2772        assert!(selected.is_none());
2773    }
2774
2775    #[rstest]
2776    fn test_select_query_order_report_rejects_when_client_differs_and_no_vid_provided() {
2777        let reports = vec![make_query_order_report(Some("O-OTHER"), "V-1")];
2778        let selected = select_query_order_report(reports, ClientOrderId::from("O-001"), None);
2779        assert!(selected.is_none());
2780    }
2781
2782    #[rstest]
2783    fn test_select_query_order_report_ignores_linked_order_ids_for_parent_with_attached_tp() {
2784        // Parent order has attached TP/SL children listed in its
2785        // linked_order_ids. A query for one of those children must NOT
2786        // resolve to the parent's report via the linked_order_ids.
2787        let child_cid = "O-CHILD-TP";
2788        let reports = vec![with_linked(
2789            make_query_order_report(Some("O-PARENT"), "V-PARENT"),
2790            &[child_cid, "O-CHILD-SL"],
2791        )];
2792        let selected = select_query_order_report(reports, ClientOrderId::from(child_cid), None);
2793        assert!(selected.is_none());
2794    }
2795
2796    #[rstest]
2797    fn test_select_query_order_report_client_match_wins_over_vid_match_elsewhere() {
2798        // Ordering invariant: the client_order_id match beats a vid match on
2799        // a different report regardless of which appears first in the list.
2800        let reports = vec![
2801            make_query_order_report(Some("O-OTHER"), "V-1"),
2802            make_query_order_report(Some("O-001"), "V-2"),
2803        ];
2804        let selected = select_query_order_report(
2805            reports,
2806            ClientOrderId::from("O-001"),
2807            Some(VenueOrderId::from("V-1")),
2808        );
2809        assert_eq!(
2810            selected.and_then(|r| r.client_order_id),
2811            Some(ClientOrderId::from("O-001")),
2812        );
2813    }
2814
2815    fn build_test_exec_client() -> OKXExecutionClient {
2816        let config = OKXExecClientConfig {
2817            api_key: Some("test_key".to_string()),
2818            api_secret: Some("test_secret".to_string()),
2819            api_passphrase: Some("test_pass".to_string()),
2820            ..OKXExecClientConfig::default()
2821        };
2822
2823        let cache = Rc::new(RefCell::new(Cache::default()));
2824        let core = ExecutionClientCore::new(
2825            config.trader_id,
2826            ClientId::from("OKX-TEST"),
2827            *OKX_VENUE,
2828            OmsType::Hedging,
2829            config.account_id,
2830            AccountType::Cash,
2831            None,
2832            cache,
2833        );
2834
2835        OKXExecutionClient::new(core, config).expect("failed to build test client")
2836    }
2837
2838    #[rstest]
2839    fn test_on_instrument_writes_through_to_client_caches() {
2840        // Bus-delivered instrument updates must land in both the private and
2841        // business WebSocket caches, and in the HTTP cache used by reconciliation.
2842        use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
2843
2844        let mut client = build_test_exec_client();
2845        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
2846        let symbol = instrument.symbol().inner();
2847        let raw_symbol = instrument.raw_symbol().inner();
2848
2849        client.on_instrument(instrument.clone());
2850
2851        let private_cache = client.ws_private.instruments_cache_arc();
2852        let business_cache = client.ws_business.instruments_cache_arc();
2853        assert_eq!(
2854            client
2855                .http_client
2856                .get_instrument(&raw_symbol)
2857                .map(|i| i.id()),
2858            Some(instrument.id()),
2859        );
2860        assert_eq!(
2861            private_cache.load().get(&symbol).map(|i| i.id()),
2862            Some(instrument.id()),
2863        );
2864        assert_eq!(
2865            business_cache.load().get(&symbol).map(|i| i.id()),
2866            Some(instrument.id()),
2867        );
2868    }
2869}