nautilus_okx/websocket/
dispatch.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//! WebSocket message dispatch for the OKX execution client.
17//!
18//! Routes incoming [`OKXWsMessage`] variants to the appropriate parsing and
19//! event emission paths. Orders with a tracked identity, including submitted
20//! orders and registered external limit orders, produce order events. Untracked
21//! orders fall back to execution reports for downstream reconciliation.
22
23use std::{collections::VecDeque, fmt::Debug, hash::Hash, sync::Arc};
24
25use ahash::AHashMap;
26use dashmap::DashMap;
27use nautilus_common::cache::fifo::{FifoCache, FifoCacheMap};
28use nautilus_core::{AtomicMap, UUID4, UnixNanos, time::AtomicTime};
29use nautilus_live::{
30    ExecutionEventEmitter,
31    execution::{
32        context::{OrderContext, OrderIdentity},
33        failure::CommandFailure,
34    },
35};
36use nautilus_model::{
37    enums::{AccountType, OrderStatus, OrderType},
38    events::{
39        OrderAccepted, OrderCanceled, OrderEventAny, OrderFilled, OrderRejected, OrderTriggered,
40        OrderUpdated,
41    },
42    identifiers::{
43        AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
44    },
45    instruments::{Instrument, InstrumentAny},
46    orders::TRIGGERABLE_ORDER_TYPES,
47    reports::FillReport,
48    types::Currency,
49};
50use parking_lot::Mutex;
51use ustr::Ustr;
52
53use crate::{
54    common::{
55        consts::{
56            OKX_CONDITIONAL_ORDER_TYPES, OKX_FIELD_CLORDID, OKX_FIELD_SCODE, OKX_FIELD_SMSG,
57            OKX_FIELD_SUBCODE, OKX_POST_ONLY_CANCEL_REASON, OKX_POST_ONLY_CANCEL_SOURCE,
58            OKX_SUCCESS_CODE,
59        },
60        enums::{OKXAlgoOrderStatus, OKXAlgoOrderType, OKXOrderStatus, OKXOrderType},
61        failure::{classify_okx_venue_code, classify_okx_ws_failure},
62        parse::{
63            is_market_price, parse_client_order_id, parse_millisecond_timestamp, parse_price,
64            parse_quantity,
65        },
66    },
67    http::models::{OKXAccount, OKXCancelAlgoOrderResponse, OKXPosition, OKXSpreadOrder},
68    websocket::{
69        client::PendingOrderInfo,
70        enums::OKXWsOperation,
71        handler::{is_post_only_auto_cancel, is_unfilled_rpi_cancel},
72        messages::{ExecutionReport, OKXAlgoOrderMsg, OKXOrderMsg, OKXWsMessage},
73        parse::{
74            FeeCache, FilledQtyCache, OrderStateSnapshot, ParsedOrderEvent,
75            is_terminal_order_state, parse_algo_order_msg, parse_algo_order_status_report,
76            parse_order_event, parse_order_msg, parse_order_status_report,
77            parse_spread_order_event, parse_spread_order_msg, update_fee_fill_caches,
78        },
79    },
80};
81
82/// Maximum entries held by the dedup sets before the oldest is evicted.
83const DEDUP_CAPACITY: usize = 10_000;
84
85#[derive(Clone, Copy, Debug, PartialEq, Eq)]
86struct OrderVenueBinding {
87    parent: VenueOrderId,
88    child: Option<VenueOrderId>,
89}
90
91#[derive(Debug)]
92struct OrderLifecycleBindings {
93    client_by_parent: AHashMap<VenueOrderId, ClientOrderId>,
94    venue_by_client: AHashMap<ClientOrderId, OrderVenueBinding>,
95    terminal_client_by_parent: FifoCacheMap<VenueOrderId, ClientOrderId, DEDUP_CAPACITY>,
96}
97
98impl Default for OrderLifecycleBindings {
99    fn default() -> Self {
100        Self {
101            client_by_parent: AHashMap::new(),
102            venue_by_client: AHashMap::new(),
103            terminal_client_by_parent: FifoCacheMap::new(),
104        }
105    }
106}
107
108impl OrderLifecycleBindings {
109    fn client_order_id(&self, parent: &VenueOrderId) -> Option<ClientOrderId> {
110        self.client_by_parent
111            .get(parent)
112            .or_else(|| self.terminal_client_by_parent.get(parent))
113            .copied()
114    }
115
116    fn finish(&mut self, client_order_id: ClientOrderId, binding: OrderVenueBinding) {
117        self.client_by_parent.remove(&binding.parent);
118        self.venue_by_client.remove(&client_order_id);
119        self.terminal_client_by_parent
120            .insert(binding.parent, client_order_id);
121    }
122}
123
124#[derive(Clone, Copy, Debug, PartialEq, Eq)]
125enum ExecutionUpdateRoute {
126    Tracked(ClientOrderId, OrderContext),
127    External,
128    Suppressed,
129}
130
131#[derive(Clone, Copy, Debug, PartialEq, Eq)]
132enum LinkedChildResolution {
133    Bound(ClientOrderId),
134    Held,
135    External,
136}
137
138#[derive(Debug)]
139struct PendingLinkedChild {
140    parent_venue_order_id: VenueOrderId,
141    candidate_client_order_ids: Vec<ClientOrderId>,
142    message: OKXOrderMsg,
143}
144
145#[derive(Debug)]
146struct DedupCache<K>
147where
148    K: Clone + Debug + Eq + Hash,
149{
150    inner: Mutex<FifoCache<K, DEDUP_CAPACITY>>,
151}
152
153impl<K> DedupCache<K>
154where
155    K: Clone + Debug + Eq + Hash,
156{
157    fn new() -> Self {
158        Self {
159            inner: Mutex::new(FifoCache::new()),
160        }
161    }
162
163    fn contains(&self, key: &K) -> bool {
164        self.inner.lock().contains(key)
165    }
166
167    fn insert(&self, key: K) -> bool {
168        self.inner.lock().insert(key)
169    }
170
171    fn remove(&self, key: &K) {
172        self.inner.lock().remove(key);
173    }
174}
175
176/// Shared state for cross-stream event deduplication between the private
177/// and business WebSocket dispatch loops.
178#[derive(Debug)]
179pub struct WsDispatchState {
180    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
181    order_contexts: DashMap<ClientOrderId, OrderContext>,
182    pub(crate) algo_order_states: DashMap<ClientOrderId, AlgoOrderState>,
183    pub(crate) pending_orders: Arc<DashMap<String, PendingOrderInfo>>,
184    pub(crate) pending_cancels: Arc<DashMap<String, PendingOrderInfo>>,
185    pub(crate) pending_amends: Arc<DashMap<String, PendingOrderInfo>>,
186    accepted_venue_order_ids: Mutex<FifoCacheMap<ClientOrderId, VenueOrderId, DEDUP_CAPACITY>>,
187    triggered_orders: DedupCache<ClientOrderId>,
188    filled_orders: DedupCache<ClientOrderId>,
189    terminal_orders: DedupCache<ClientOrderId>,
190    emitted_trades: DedupCache<TradeId>,
191    post_only_rejections: DedupCache<Ustr>,
192    lifecycle_bindings: Mutex<OrderLifecycleBindings>,
193    pending_linked_children: Mutex<VecDeque<PendingLinkedChild>>,
194    linked_child_notify: tokio::sync::Notify,
195}
196
197#[derive(Clone, Copy, Debug)]
198pub(crate) struct AlgoOrderState {
199    pub(crate) order_type: OKXAlgoOrderType,
200    pub(crate) ts_last: UnixNanos,
201}
202
203impl Default for WsDispatchState {
204    fn default() -> Self {
205        Self {
206            order_identities: DashMap::new(),
207            order_contexts: DashMap::new(),
208            algo_order_states: DashMap::new(),
209            pending_orders: Arc::new(DashMap::new()),
210            pending_cancels: Arc::new(DashMap::new()),
211            pending_amends: Arc::new(DashMap::new()),
212            accepted_venue_order_ids: Mutex::new(FifoCacheMap::new()),
213            triggered_orders: DedupCache::new(),
214            filled_orders: DedupCache::new(),
215            terminal_orders: DedupCache::new(),
216            emitted_trades: DedupCache::new(),
217            post_only_rejections: DedupCache::new(),
218            lifecycle_bindings: Mutex::new(OrderLifecycleBindings::default()),
219            pending_linked_children: Mutex::new(VecDeque::new()),
220            linked_child_notify: tokio::sync::Notify::new(),
221        }
222    }
223}
224
225impl WsDispatchState {
226    // Creates a dispatch state sharing the pending operation maps
227    // with the WebSocket client that populates them
228    pub(crate) fn with_pending_maps(
229        pending_orders: Arc<DashMap<String, PendingOrderInfo>>,
230        pending_cancels: Arc<DashMap<String, PendingOrderInfo>>,
231        pending_amends: Arc<DashMap<String, PendingOrderInfo>>,
232    ) -> Self {
233        Self {
234            pending_orders,
235            pending_cancels,
236            pending_amends,
237            ..Default::default()
238        }
239    }
240
241    pub(crate) fn track_order_context(&self, context: OrderContext) {
242        self.order_contexts
243            .insert(context.identity.client_order_id, context);
244    }
245
246    pub(crate) fn register_external_order(
247        &self,
248        context: OrderContext,
249        venue_order_id: VenueOrderId,
250    ) {
251        let client_order_id = context.identity.client_order_id;
252        self.track_order_context(context);
253        self.bind_order(client_order_id, venue_order_id);
254        self.insert_accepted(client_order_id, venue_order_id);
255    }
256
257    pub(crate) fn bind_order(&self, client_order_id: ClientOrderId, venue_order_id: VenueOrderId) {
258        let mut bindings = self.lifecycle_bindings.lock();
259        let binding = bindings.venue_by_client.get(&client_order_id).copied();
260        if binding.is_some_and(|binding| binding.parent != venue_order_id) {
261            log::error!(
262                "Ignoring conflicting order binding for {client_order_id}: expected={:?} received={venue_order_id}",
263                binding.map(|binding| binding.parent),
264            );
265            return;
266        }
267
268        bindings
269            .client_by_parent
270            .insert(venue_order_id, client_order_id);
271        bindings.venue_by_client.insert(
272            client_order_id,
273            binding.unwrap_or(OrderVenueBinding {
274                parent: venue_order_id,
275                child: None,
276            }),
277        );
278        self.linked_child_notify.notify_one();
279    }
280
281    pub(crate) fn order_venue_binding(
282        &self,
283        client_order_id: ClientOrderId,
284    ) -> Option<(VenueOrderId, bool)> {
285        let bindings = self.lifecycle_bindings.lock();
286        bindings
287            .venue_by_client
288            .get(&client_order_id)
289            .map(|binding| {
290                (
291                    binding.child.unwrap_or(binding.parent),
292                    binding.child.is_some(),
293                )
294            })
295    }
296
297    pub(crate) fn order_identity(&self, client_order_id: ClientOrderId) -> Option<OrderIdentity> {
298        self.order_context(client_order_id)
299            .map(|context| context.identity)
300            .or_else(|| {
301                self.order_identities
302                    .get(&client_order_id)
303                    .map(|entry| *entry)
304            })
305    }
306
307    pub(crate) fn order_context(&self, client_order_id: ClientOrderId) -> Option<OrderContext> {
308        self.order_contexts
309            .get(&client_order_id)
310            .map(|entry| *entry)
311    }
312
313    pub(crate) fn remove_order_tracking(&self, client_order_id: ClientOrderId) {
314        let pending = self.pending_linked_children.lock();
315        let removed_context = self.order_contexts.remove(&client_order_id);
316        self.order_identities.remove(&client_order_id);
317        self.algo_order_states.remove(&client_order_id);
318        drop(pending);
319
320        let Some((_, context)) = removed_context else {
321            return;
322        };
323
324        if context.identity.order_type != OrderType::Limit {
325            self.linked_child_notify.notify_one();
326            return;
327        }
328
329        let mut bindings = self.lifecycle_bindings.lock();
330        if let Some(binding) = bindings.venue_by_client.remove(&client_order_id) {
331            bindings.client_by_parent.remove(&binding.parent);
332        }
333    }
334
335    pub(crate) fn resolve_algo_submit_failure(
336        &self,
337        client_order_id: ClientOrderId,
338        failure: &CommandFailure,
339    ) {
340        let bindings = self.lifecycle_bindings.lock();
341        if matches!(failure, CommandFailure::Ambiguous(_))
342            && bindings.venue_by_client.contains_key(&client_order_id)
343        {
344            return;
345        }
346
347        self.remove_order_tracking(client_order_id);
348    }
349
350    pub(crate) async fn wait_for_linked_child_route(&self) {
351        self.linked_child_notify.notified().await;
352    }
353
354    fn resolve_or_hold_linked_child(
355        &self,
356        parent_venue_order_id: VenueOrderId,
357        instrument_id: InstrumentId,
358        message: &OKXOrderMsg,
359    ) -> LinkedChildResolution {
360        let candidate_client_order_ids = self
361            .order_contexts
362            .iter()
363            .filter_map(|entry| {
364                (entry.identity.instrument_id == instrument_id
365                    && OKX_CONDITIONAL_ORDER_TYPES.contains(&entry.identity.order_type))
366                .then_some(entry.identity.client_order_id)
367            })
368            .collect::<Vec<_>>();
369        let bindings = self.lifecycle_bindings.lock();
370        if let Some(client_order_id) = bindings.client_order_id(&parent_venue_order_id) {
371            return LinkedChildResolution::Bound(client_order_id);
372        }
373
374        let mut pending = self.pending_linked_children.lock();
375        let candidate_client_order_ids = candidate_client_order_ids
376            .iter()
377            .filter(|client_order_id| {
378                self.order_contexts.contains_key(client_order_id)
379                    && !bindings.venue_by_client.contains_key(client_order_id)
380            })
381            .copied()
382            .collect::<Vec<_>>();
383
384        if candidate_client_order_ids.is_empty() {
385            return LinkedChildResolution::External;
386        }
387
388        pending.push_back(PendingLinkedChild {
389            parent_venue_order_id,
390            candidate_client_order_ids,
391            message: message.clone(),
392        });
393        LinkedChildResolution::Held
394    }
395
396    fn take_routable_linked_children(&self) -> Vec<OKXOrderMsg> {
397        if self.pending_linked_children.lock().is_empty() {
398            return Vec::new();
399        }
400
401        let bindings = self.lifecycle_bindings.lock();
402        let mut pending = self.pending_linked_children.lock();
403        let mut held = VecDeque::with_capacity(pending.len());
404        let mut routable = Vec::new();
405
406        while let Some(child) = pending.pop_front() {
407            let is_bound = bindings
408                .client_order_id(&child.parent_venue_order_id)
409                .is_some();
410            let is_pending = child
411                .candidate_client_order_ids
412                .iter()
413                .any(|client_order_id| {
414                    self.order_contexts.contains_key(client_order_id)
415                        && !bindings.venue_by_client.contains_key(client_order_id)
416                });
417
418            if is_bound || !is_pending {
419                routable.push(child.message);
420            } else {
421                held.push_back(child);
422            }
423        }
424
425        *pending = held;
426        routable
427    }
428}
429
430impl WsDispatchState {
431    /// Returns whether acceptance was already emitted for the order.
432    #[must_use]
433    pub fn contains_accepted(&self, cid: &ClientOrderId) -> bool {
434        self.accepted_venue_order_ids.lock().contains_key(cid)
435    }
436
437    /// Records that acceptance was emitted for the order.
438    pub fn insert_accepted(&self, cid: ClientOrderId, venue_order_id: VenueOrderId) {
439        self.accepted_venue_order_ids
440            .lock()
441            .insert(cid, venue_order_id);
442    }
443
444    fn accepted_venue_order_id(&self, cid: &ClientOrderId) -> Option<VenueOrderId> {
445        self.accepted_venue_order_ids.lock().get(cid).copied()
446    }
447
448    /// Returns whether the order was already triggered.
449    #[must_use]
450    pub fn contains_triggered(&self, cid: &ClientOrderId) -> bool {
451        self.triggered_orders.contains(cid)
452    }
453
454    /// Records that the order was triggered.
455    pub fn insert_triggered(&self, cid: ClientOrderId) {
456        let _ = self.triggered_orders.insert(cid);
457    }
458
459    /// Returns whether the order was already filled.
460    #[must_use]
461    pub fn contains_filled(&self, cid: &ClientOrderId) -> bool {
462        self.filled_orders.contains(cid)
463    }
464
465    /// Records that the order was filled.
466    pub fn insert_filled(&self, cid: ClientOrderId) {
467        let _ = self.filled_orders.insert(cid);
468    }
469
470    /// Returns whether the order already reached a terminal state.
471    #[must_use]
472    pub fn contains_terminal(&self, cid: &ClientOrderId) -> bool {
473        self.terminal_orders.contains(cid)
474    }
475
476    /// Records that the order reached a terminal state.
477    pub fn insert_terminal(&self, cid: ClientOrderId) {
478        let _ = self.terminal_orders.insert(cid);
479    }
480
481    /// Returns `true` if this trade was already emitted (duplicate).
482    /// Uses atomic insert to avoid TOCTOU races between concurrent streams.
483    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
484        !self.emitted_trades.insert(trade_id)
485    }
486
487    #[must_use]
488    pub fn contains_trade(&self, trade_id: &TradeId) -> bool {
489        self.emitted_trades.contains(trade_id)
490    }
491
492    fn remove_accepted(&self, cid: &ClientOrderId) {
493        self.accepted_venue_order_ids.lock().remove(cid);
494    }
495
496    fn remove_triggered(&self, cid: &ClientOrderId) {
497        self.triggered_orders.remove(cid);
498    }
499
500    fn remove_filled(&self, cid: &ClientOrderId) {
501        self.filled_orders.remove(cid);
502    }
503
504    fn insert_post_only_rejection(&self, order_id: Ustr) {
505        let _ = self.post_only_rejections.insert(order_id);
506    }
507
508    fn contains_post_only_rejection(&self, order_id: &Ustr) -> bool {
509        self.post_only_rejections.contains(order_id)
510    }
511}
512
513/// Dispatches a WebSocket message with cross-stream deduplication.
514///
515/// Orders with a tracked identity, including submitted orders and registered
516/// external limit orders, produce order events (`OrderAccepted`, `OrderCanceled`,
517/// `OrderFilled`, etc.). Untracked orders produce execution reports for downstream
518/// reconciliation.
519#[expect(clippy::too_many_arguments)]
520pub fn dispatch_ws_message(
521    message: OKXWsMessage,
522    emitter: &ExecutionEventEmitter,
523    state: &WsDispatchState,
524    account_id: AccountId,
525    account_type: AccountType,
526    instruments: &AtomicMap<Ustr, InstrumentAny>,
527    fee_cache: &mut FeeCache,
528    filled_qty_cache: &mut FilledQtyCache,
529    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
530    clock: &AtomicTime,
531) {
532    let guard = instruments.load();
533    let instruments: &AHashMap<Ustr, InstrumentAny> = &guard;
534
535    match message {
536        OKXWsMessage::Orders(order_msgs) => {
537            let ts_init = clock.get_time_ns();
538            let pending_order_msgs = state.take_routable_linked_children();
539            if !pending_order_msgs.is_empty() {
540                dispatch_order_messages(
541                    &pending_order_msgs,
542                    emitter,
543                    state,
544                    account_id,
545                    instruments,
546                    fee_cache,
547                    filled_qty_cache,
548                    order_state_cache,
549                    ts_init,
550                );
551            }
552            dispatch_order_messages(
553                &order_msgs,
554                emitter,
555                state,
556                account_id,
557                instruments,
558                fee_cache,
559                filled_qty_cache,
560                order_state_cache,
561                ts_init,
562            );
563        }
564        OKXWsMessage::SpreadOrders(order_msgs) => {
565            let ts_init = clock.get_time_ns();
566            dispatch_spread_order_messages(
567                &order_msgs,
568                emitter,
569                state,
570                account_id,
571                instruments,
572                filled_qty_cache,
573                order_state_cache,
574                ts_init,
575            );
576        }
577        OKXWsMessage::AlgoOrders(algo_msgs) => {
578            let ts_init = clock.get_time_ns();
579            for msg in &algo_msgs {
580                dispatch_algo_order_message(msg, emitter, state, account_id, instruments, ts_init);
581            }
582        }
583        OKXWsMessage::Account(data) => {
584            let ts_init = clock.get_time_ns();
585
586            match serde_json::from_value::<Vec<OKXAccount>>(data) {
587                Ok(accounts) => {
588                    for account in &accounts {
589                        match crate::common::parse::parse_account_state(
590                            account,
591                            account_id,
592                            account_type,
593                            ts_init,
594                        ) {
595                            Ok(account_state) => emitter.send_account_state(account_state),
596                            Err(e) => log::error!("Failed to parse account state: {e}"),
597                        }
598                    }
599                }
600                Err(e) => log::error!("Failed to deserialize account data: {e}"),
601            }
602        }
603        OKXWsMessage::Positions(data) => {
604            let ts_init = clock.get_time_ns();
605
606            match serde_json::from_value::<Vec<OKXPosition>>(data) {
607                Ok(positions) => {
608                    for position in positions {
609                        let Some(instrument) = instruments.get(&position.inst_id) else {
610                            log::warn!("No cached instrument for position: {}", position.inst_id);
611                            continue;
612                        };
613                        let instrument_id = instrument.id();
614                        let size_precision = instrument.size_precision();
615
616                        match crate::common::parse::parse_position_status_report(
617                            &position,
618                            account_id,
619                            instrument_id,
620                            size_precision,
621                            ts_init,
622                        ) {
623                            Ok(report) => emitter.send_position_report(report),
624                            Err(e) => log::error!("Failed to parse position report: {e}"),
625                        }
626                    }
627                }
628                Err(e) => log::error!("Failed to deserialize positions data: {e}"),
629            }
630        }
631        OKXWsMessage::OrderResponse {
632            id,
633            op,
634            code,
635            msg,
636            data,
637        } => {
638            let ts_init = clock.get_time_ns();
639
640            for item in &data {
641                let s_code = item
642                    .get(OKX_FIELD_SCODE)
643                    .and_then(|v| v.as_str())
644                    .unwrap_or("");
645                let s_msg = item
646                    .get(OKX_FIELD_SMSG)
647                    .and_then(|v| v.as_str())
648                    .unwrap_or("");
649                let sub_code = item
650                    .get(OKX_FIELD_SUBCODE)
651                    .and_then(|v| v.as_str())
652                    .unwrap_or("");
653                let reason = format_order_response_reason(s_code, s_msg, sub_code);
654                let cl_ord_id = item
655                    .get(OKX_FIELD_CLORDID)
656                    .and_then(|v| v.as_str())
657                    .unwrap_or("");
658
659                if s_code == OKX_SUCCESS_CODE {
660                    log::debug!("Order response ok: op={op:?} cl_ord_id={cl_ord_id}");
661                    match op {
662                        OKXWsOperation::Order
663                        | OKXWsOperation::BatchOrders
664                        | OKXWsOperation::OrderAlgo => {
665                            state.pending_orders.remove(cl_ord_id);
666                        }
667                        OKXWsOperation::CancelOrder
668                        | OKXWsOperation::BatchCancelOrders
669                        | OKXWsOperation::MassCancel
670                        | OKXWsOperation::CancelAlgos => {
671                            state.pending_cancels.remove(cl_ord_id);
672                        }
673                        OKXWsOperation::AmendOrder | OKXWsOperation::BatchAmendOrders => {
674                            state.pending_amends.remove(cl_ord_id);
675                        }
676                        _ => {}
677                    }
678                    continue;
679                }
680
681                let Some(client_order_id) = parse_client_order_id(cl_ord_id) else {
682                    log::warn!(
683                        "Order response error without client_order_id: \
684                         op={op:?} s_code={s_code} s_msg={s_msg}"
685                    );
686                    continue;
687                };
688
689                let Some(ident) = state.order_identity(client_order_id) else {
690                    log::warn!(
691                        "Order response error for untracked order: \
692                         op={op:?} cl_ord_id={cl_ord_id} s_code={s_code} s_msg={s_msg}"
693                    );
694                    continue;
695                };
696
697                let venue_order_id = item
698                    .get("ordId")
699                    .and_then(|v| v.as_str())
700                    .filter(|s| !s.is_empty())
701                    .map(VenueOrderId::new);
702
703                match classify_okx_venue_code(s_code, reason.clone()) {
704                    CommandFailure::Ambiguous(reason) => {
705                        log::warn!(
706                            "Ambiguous order response for {client_order_id}, awaiting reconciliation: \
707                             op={op:?} s_code={s_code} {reason}"
708                        );
709                        continue;
710                    }
711                    CommandFailure::NotSent(_) => {
712                        log::warn!(
713                            "Unexpected NotSent classification for venue order response: \
714                             op={op:?} cl_ord_id={cl_ord_id} s_code={s_code}"
715                        );
716                        continue;
717                    }
718                    CommandFailure::VenueRejected(_) => {}
719                }
720
721                match op {
722                    OKXWsOperation::Order | OKXWsOperation::BatchOrders => {
723                        state.remove_order_tracking(client_order_id);
724                        state.pending_orders.remove(cl_ord_id);
725                        emitter.emit_order_rejected_event(
726                            ident.strategy_id,
727                            ident.instrument_id,
728                            client_order_id,
729                            &reason,
730                            ts_init,
731                            false,
732                        );
733                    }
734                    OKXWsOperation::CancelOrder
735                    | OKXWsOperation::BatchCancelOrders
736                    | OKXWsOperation::MassCancel => {
737                        state.pending_cancels.remove(cl_ord_id);
738                        emitter.emit_order_cancel_rejected_event(
739                            ident.strategy_id,
740                            ident.instrument_id,
741                            client_order_id,
742                            venue_order_id,
743                            &reason,
744                            ts_init,
745                        );
746                    }
747                    OKXWsOperation::AmendOrder | OKXWsOperation::BatchAmendOrders => {
748                        state.pending_amends.remove(cl_ord_id);
749                        emitter.emit_order_modify_rejected_event(
750                            ident.strategy_id,
751                            ident.instrument_id,
752                            client_order_id,
753                            venue_order_id,
754                            &reason,
755                            ts_init,
756                        );
757                    }
758                    _ => {
759                        log::warn!(
760                            "Order response error for unhandled op: \
761                             op={op:?} cl_ord_id={cl_ord_id} s_code={s_code} s_msg={s_msg}"
762                        );
763                    }
764                }
765            }
766
767            if code != "0" && data.is_empty() {
768                log::warn!(
769                    "Order response error (no data): id={id:?} op={op:?} code={code} msg={msg}"
770                );
771            }
772        }
773        OKXWsMessage::SendFailed {
774            request_id,
775            client_order_ids,
776            op,
777            error,
778        } => {
779            let failure = classify_okx_ws_failure(&error);
780            let is_ambiguous = matches!(failure, CommandFailure::Ambiguous(_));
781            log::warn!(
782                "WebSocket send failed without structured venue response: \
783                 request_id={request_id}, client_order_ids={client_order_ids:?}, \
784                 op={op:?}, {failure:?}"
785            );
786
787            for client_order_id in client_order_ids {
788                let key = client_order_id.as_str();
789
790                match op {
791                    Some(
792                        OKXWsOperation::Order
793                        | OKXWsOperation::BatchOrders
794                        | OKXWsOperation::OrderAlgo,
795                    ) => {
796                        if !is_ambiguous {
797                            state.pending_orders.remove(key);
798                        }
799                        emit_send_failed_submit(&failure, state, emitter, clock, client_order_id);
800                    }
801                    Some(
802                        OKXWsOperation::CancelOrder
803                        | OKXWsOperation::BatchCancelOrders
804                        | OKXWsOperation::MassCancel
805                        | OKXWsOperation::CancelAlgos,
806                    ) => {
807                        if !is_ambiguous {
808                            state.pending_cancels.remove(key);
809                        }
810
811                        emit_send_failed_cancel(
812                            &failure,
813                            state,
814                            emitter,
815                            clock,
816                            client_order_id,
817                            None,
818                        );
819                    }
820                    Some(OKXWsOperation::AmendOrder | OKXWsOperation::BatchAmendOrders) => {
821                        if !is_ambiguous {
822                            state.pending_amends.remove(key);
823                        }
824                        emit_send_failed_modify(&failure, state, emitter, clock, client_order_id);
825                    }
826                    _ => {}
827                }
828            }
829        }
830        OKXWsMessage::ChannelData { channel, .. } => {
831            log::debug!("Ignoring data channel message on execution client: {channel:?}");
832        }
833        OKXWsMessage::SubscriptionFailed {
834            channel,
835            inst_id,
836            code,
837            msg,
838        } => {
839            log::error!(
840                "OKX rejected {channel:?} subscription for {inst_id:?} \
841                 (code={code}, msg={msg}); execution updates for it will not flow"
842            );
843        }
844        OKXWsMessage::LiquidationWarnings(warnings) => {
845            for warning in warnings {
846                log::warn!(
847                    "Liquidation warning: inst_id={}, pos_side={:?}, pos={}, mgn_ratio={}, mark_px={}, mgn_mode={:?}",
848                    warning.inst_id,
849                    warning.pos_side,
850                    warning.pos,
851                    warning.mgn_ratio,
852                    warning.mark_px,
853                    warning.mgn_mode,
854                );
855            }
856        }
857        OKXWsMessage::BookData { .. }
858        | OKXWsMessage::RpiBookData { .. }
859        | OKXWsMessage::Instruments(_) => {
860            log::debug!("Ignoring data message on execution client");
861        }
862        OKXWsMessage::Error(_) => {}
863        OKXWsMessage::Reconnected => {
864            log::info!("Websocket reconnected");
865        }
866        OKXWsMessage::Authenticated => {
867            log::debug!("Websocket authenticated");
868        }
869    }
870}
871
872fn route_algo_order_message(
873    msg: &OKXAlgoOrderMsg,
874    state: &WsDispatchState,
875) -> ExecutionUpdateRoute {
876    if matches!(
877        msg.ord_type,
878        OKXAlgoOrderType::Iceberg
879            | OKXAlgoOrderType::SmartIceberg
880            | OKXAlgoOrderType::Twap
881            | OKXAlgoOrderType::Chase
882            | OKXAlgoOrderType::Other
883    ) || msg.state == OKXAlgoOrderStatus::Unknown
884    {
885        return ExecutionUpdateRoute::Suppressed;
886    }
887
888    let direct_client_order_id = parse_client_order_id(&msg.algo_cl_ord_id)
889        .or_else(|| parse_client_order_id(&msg.cl_ord_id));
890
891    if let Some(client_order_id) = direct_client_order_id {
892        if state.contains_terminal(&client_order_id) {
893            return ExecutionUpdateRoute::Suppressed;
894        }
895
896        if let Some(context) = state
897            .order_contexts
898            .get(&client_order_id)
899            .filter(|entry| OKX_CONDITIONAL_ORDER_TYPES.contains(&entry.identity.order_type))
900            .map(|entry| *entry)
901        {
902            return ExecutionUpdateRoute::Tracked(client_order_id, context);
903        }
904
905        if state.order_identity(client_order_id).is_some() {
906            return ExecutionUpdateRoute::Suppressed;
907        }
908    }
909
910    let parent_venue_order_id = VenueOrderId::new(msg.algo_id.as_str());
911    let client_order_id = {
912        let bindings = state.lifecycle_bindings.lock();
913        bindings.client_order_id(&parent_venue_order_id)
914    };
915
916    let Some(client_order_id) = client_order_id else {
917        return ExecutionUpdateRoute::External;
918    };
919
920    if state.contains_terminal(&client_order_id) {
921        return ExecutionUpdateRoute::Suppressed;
922    }
923
924    state
925        .order_contexts
926        .get(&client_order_id)
927        .filter(|entry| OKX_CONDITIONAL_ORDER_TYPES.contains(&entry.identity.order_type))
928        .map_or(ExecutionUpdateRoute::Suppressed, |entry| {
929            ExecutionUpdateRoute::Tracked(client_order_id, *entry)
930        })
931}
932
933fn dispatch_algo_order_message(
934    msg: &OKXAlgoOrderMsg,
935    emitter: &ExecutionEventEmitter,
936    state: &WsDispatchState,
937    account_id: AccountId,
938    instruments: &AHashMap<Ustr, InstrumentAny>,
939    ts_init: UnixNanos,
940) {
941    let route = route_algo_order_message(msg, state);
942
943    match route {
944        ExecutionUpdateRoute::External => {
945            match parse_algo_order_msg(msg, account_id, instruments, ts_init) {
946                Ok(Some(report)) => dispatch_execution_reports(vec![report], emitter, state),
947                Ok(None) => {}
948                Err(e) => log::error!("Failed to parse external algo order message: {e}"),
949            }
950        }
951        ExecutionUpdateRoute::Suppressed => {
952            log::debug!(
953                "Suppressing algo order update: algo_id={} state={:?}",
954                msg.algo_id,
955                msg.state,
956            );
957        }
958        ExecutionUpdateRoute::Tracked(client_order_id, context) => {
959            let Some(instrument) = instruments.get(&msg.inst_id) else {
960                log::warn!(
961                    "No instrument for {}, skipping algo order message",
962                    msg.inst_id
963                );
964                return;
965            };
966            dispatch_tracked_algo_order_message(
967                msg,
968                client_order_id,
969                context,
970                instrument,
971                emitter,
972                state,
973                account_id,
974                ts_init,
975            );
976        }
977    }
978}
979
980#[expect(
981    clippy::too_many_arguments,
982    reason = "tracked routing requires the resolved context, venue state, and event timestamps"
983)]
984fn dispatch_tracked_algo_order_message(
985    msg: &OKXAlgoOrderMsg,
986    client_order_id: ClientOrderId,
987    mut context: OrderContext,
988    instrument: &InstrumentAny,
989    emitter: &ExecutionEventEmitter,
990    state: &WsDispatchState,
991    account_id: AccountId,
992    ts_init: UnixNanos,
993) {
994    let parent_venue_order_id = VenueOrderId::new(msg.algo_id.as_str());
995    let ts_event = parse_millisecond_timestamp(msg.u_time);
996    let mut bindings = state.lifecycle_bindings.lock();
997    let mut is_terminal = false;
998    let mut binding = bindings
999        .venue_by_client
1000        .get(&client_order_id)
1001        .copied()
1002        .unwrap_or(OrderVenueBinding {
1003            parent: parent_venue_order_id,
1004            child: None,
1005        });
1006
1007    if binding.parent != parent_venue_order_id {
1008        log::error!(
1009            "Suppressing conflicting algo parent binding for {client_order_id}: expected={} received={parent_venue_order_id}",
1010            binding.parent,
1011        );
1012        return;
1013    }
1014
1015    if binding.child.is_none() {
1016        let Some(refreshed) = dispatch_pending_algo_order(
1017            msg, context, instrument, account_id, ts_event, ts_init, emitter, state,
1018        ) else {
1019            return;
1020        };
1021
1022        context = refreshed;
1023    }
1024
1025    bindings
1026        .client_by_parent
1027        .insert(parent_venue_order_id, client_order_id);
1028
1029    match msg.state {
1030        OKXAlgoOrderStatus::Live | OKXAlgoOrderStatus::Pause => {}
1031        OKXAlgoOrderStatus::Effective
1032        | OKXAlgoOrderStatus::OrderPlaced
1033        | OKXAlgoOrderStatus::PartiallyEffective
1034        | OKXAlgoOrderStatus::Filled
1035        | OKXAlgoOrderStatus::PartiallyFailed => {
1036            let child_venue_order_id = algo_child_venue_order_id(msg);
1037            if let Some(child_venue_order_id) = child_venue_order_id {
1038                bind_algo_child_and_emit_transition(
1039                    client_order_id,
1040                    parent_venue_order_id,
1041                    child_venue_order_id,
1042                    &mut binding,
1043                    context,
1044                    account_id,
1045                    ts_event,
1046                    ts_init,
1047                    emitter,
1048                    state,
1049                );
1050            } else {
1051                ensure_accepted_emitted(
1052                    client_order_id,
1053                    account_id,
1054                    parent_venue_order_id,
1055                    &context.identity,
1056                    emitter,
1057                    state,
1058                    ts_event,
1059                    ts_init,
1060                );
1061            }
1062
1063            if matches!(
1064                msg.state,
1065                OKXAlgoOrderStatus::Filled | OKXAlgoOrderStatus::PartiallyFailed
1066            ) {
1067                log::debug!(
1068                    "Deferring tracked algo {:?} update for {client_order_id} to regular child execution updates",
1069                    msg.state,
1070                );
1071            }
1072        }
1073        OKXAlgoOrderStatus::Canceled => {
1074            if binding.child.is_some() {
1075                log::debug!(
1076                    "Suppressing stale canceled algo parent for triggered order {client_order_id}"
1077                );
1078            } else {
1079                ensure_accepted_emitted(
1080                    client_order_id,
1081                    account_id,
1082                    parent_venue_order_id,
1083                    &context.identity,
1084                    emitter,
1085                    state,
1086                    ts_event,
1087                    ts_init,
1088                );
1089                let canceled = OrderCanceled::new(
1090                    emitter.trader_id(),
1091                    context.identity.strategy_id,
1092                    context.identity.instrument_id,
1093                    client_order_id,
1094                    UUID4::new(),
1095                    ts_event,
1096                    ts_init,
1097                    false,
1098                    Some(parent_venue_order_id),
1099                    Some(account_id),
1100                    None,
1101                );
1102                state.insert_terminal(client_order_id);
1103                state.remove_accepted(&client_order_id);
1104                state.remove_order_tracking(client_order_id);
1105                is_terminal = true;
1106                emitter.send_order_event(OrderEventAny::Canceled(canceled));
1107            }
1108        }
1109        OKXAlgoOrderStatus::OrderFailed => {
1110            if binding.child.is_some() {
1111                log::debug!(
1112                    "Suppressing stale failed algo parent for triggered order {client_order_id}"
1113                );
1114            } else {
1115                let reason = if msg.fail_code.is_empty() {
1116                    "OKX algo order failed".to_string()
1117                } else {
1118                    format_order_response_reason(&msg.fail_code, "", "")
1119                };
1120                let rejected = OrderRejected::new(
1121                    emitter.trader_id(),
1122                    context.identity.strategy_id,
1123                    context.identity.instrument_id,
1124                    client_order_id,
1125                    account_id,
1126                    Ustr::from(reason.as_str()),
1127                    UUID4::new(),
1128                    ts_event,
1129                    ts_init,
1130                    false,
1131                    false,
1132                );
1133                state.insert_terminal(client_order_id);
1134                state.remove_accepted(&client_order_id);
1135                state.remove_order_tracking(client_order_id);
1136                is_terminal = true;
1137                emitter.send_order_event(OrderEventAny::Rejected(rejected));
1138            }
1139        }
1140        OKXAlgoOrderStatus::Unknown => {}
1141    }
1142
1143    if is_terminal {
1144        bindings.finish(client_order_id, binding);
1145    } else {
1146        bindings.venue_by_client.insert(client_order_id, binding);
1147    }
1148
1149    drop(bindings);
1150    state.linked_child_notify.notify_one();
1151}
1152
1153fn algo_child_venue_order_id(msg: &OKXAlgoOrderMsg) -> Option<VenueOrderId> {
1154    if !msg.ord_id.is_empty() {
1155        return Some(VenueOrderId::new(msg.ord_id.as_str()));
1156    }
1157
1158    match msg.ord_id_list.as_slice() {
1159        [order_id] if !order_id.is_empty() => Some(VenueOrderId::new(order_id.as_str())),
1160        [] => None,
1161        order_ids => {
1162            log::warn!(
1163                "Cannot bind algo order {} to {} triggered child IDs",
1164                msg.algo_id,
1165                order_ids.len(),
1166            );
1167            None
1168        }
1169    }
1170}
1171
1172#[expect(clippy::too_many_arguments)]
1173fn dispatch_pending_algo_order(
1174    msg: &OKXAlgoOrderMsg,
1175    mut context: OrderContext,
1176    instrument: &InstrumentAny,
1177    account_id: AccountId,
1178    ts_event: UnixNanos,
1179    ts_init: UnixNanos,
1180    emitter: &ExecutionEventEmitter,
1181    state: &WsDispatchState,
1182) -> Option<OrderContext> {
1183    if state
1184        .algo_order_states
1185        .get(&context.identity.client_order_id)
1186        .is_some_and(|previous| ts_event < previous.ts_last)
1187    {
1188        return None;
1189    }
1190
1191    let client_order_id = context.identity.client_order_id;
1192    let pending = matches!(
1193        msg.state,
1194        OKXAlgoOrderStatus::Live | OKXAlgoOrderStatus::Pause
1195    );
1196    let parent_venue_order_id = VenueOrderId::new(msg.algo_id.as_str());
1197
1198    if pending {
1199        ensure_accepted_emitted(
1200            client_order_id,
1201            account_id,
1202            parent_venue_order_id,
1203            &context.identity,
1204            emitter,
1205            state,
1206            ts_event,
1207            ts_init,
1208        );
1209    }
1210
1211    let previous = context;
1212
1213    match parse_algo_order_status_report(msg, instrument, account_id, ts_init) {
1214        Ok(report) => {
1215            if !msg.sz.is_empty() {
1216                context.quantity = report.quantity;
1217            }
1218
1219            context.price = report.price;
1220            context.trigger_price = report.trigger_price;
1221            context.trigger_type = report.trigger_type;
1222        }
1223        Err(e) => {
1224            log::error!(
1225                "Failed to refresh tracked algo order context for {}: {e}",
1226                context.identity.client_order_id,
1227            );
1228            return Some(context);
1229        }
1230    }
1231
1232    if !msg.actual_sz.is_empty() && msg.actual_sz != "0" {
1233        match parse_quantity(msg.actual_sz.as_str(), instrument.size_precision()) {
1234            Ok(quantity) => context.quantity = quantity,
1235            Err(e) => log::error!(
1236                "Failed to refresh tracked algo actual quantity for {}: {e}",
1237                context.identity.client_order_id,
1238            ),
1239        }
1240    }
1241
1242    state.track_order_context(context);
1243    state.algo_order_states.insert(
1244        client_order_id,
1245        AlgoOrderState {
1246            order_type: msg.ord_type,
1247            ts_last: ts_event,
1248        },
1249    );
1250
1251    if pending
1252        && msg.ord_type != OKXAlgoOrderType::MoveOrderStop
1253        && (previous.quantity != context.quantity
1254            || previous.price != context.price
1255            || previous.trigger_price != context.trigger_price)
1256    {
1257        emit_order_update(
1258            parent_venue_order_id,
1259            context,
1260            account_id,
1261            ts_event,
1262            ts_init,
1263            false,
1264            emitter,
1265        );
1266    }
1267
1268    Some(context)
1269}
1270
1271fn refresh_regular_child_context(
1272    msg: &OKXOrderMsg,
1273    mut context: OrderContext,
1274    instrument: &InstrumentAny,
1275) -> OrderContext {
1276    match parse_quantity(&msg.sz, instrument.size_precision()) {
1277        Ok(quantity) => context.quantity = quantity,
1278        Err(e) => log::error!(
1279            "Failed to refresh tracked child quantity for {}: {e}",
1280            context.identity.client_order_id,
1281        ),
1282    }
1283
1284    context.price = if is_market_price(&msg.px) {
1285        None
1286    } else {
1287        match parse_price(&msg.px, instrument.price_precision()) {
1288            Ok(price) => Some(price),
1289            Err(e) => {
1290                log::error!(
1291                    "Failed to refresh tracked child price for {}: {e}",
1292                    context.identity.client_order_id,
1293                );
1294                context.price
1295            }
1296        }
1297    };
1298    context
1299}
1300
1301#[expect(clippy::too_many_arguments)]
1302fn bind_algo_child_and_emit_transition(
1303    client_order_id: ClientOrderId,
1304    parent_venue_order_id: VenueOrderId,
1305    child_venue_order_id: VenueOrderId,
1306    binding: &mut OrderVenueBinding,
1307    context: OrderContext,
1308    account_id: AccountId,
1309    ts_event: UnixNanos,
1310    ts_init: UnixNanos,
1311    emitter: &ExecutionEventEmitter,
1312    state: &WsDispatchState,
1313) {
1314    if let Some(bound_child) = binding.child {
1315        if bound_child != child_venue_order_id {
1316            log::error!(
1317                "Suppressing conflicting algo child binding for {client_order_id}: expected={bound_child} received={child_venue_order_id}"
1318            );
1319        }
1320        return;
1321    }
1322
1323    binding.child = Some(child_venue_order_id);
1324    state.track_order_context(context);
1325    ensure_accepted_emitted(
1326        client_order_id,
1327        account_id,
1328        parent_venue_order_id,
1329        &context.identity,
1330        emitter,
1331        state,
1332        ts_event,
1333        ts_init,
1334    );
1335
1336    if state.accepted_venue_order_id(&client_order_id) != Some(child_venue_order_id) {
1337        state.insert_accepted(client_order_id, child_venue_order_id);
1338        emit_order_update(
1339            child_venue_order_id,
1340            context,
1341            account_id,
1342            ts_event,
1343            ts_init,
1344            false,
1345            emitter,
1346        );
1347    }
1348
1349    if !state.contains_triggered(&client_order_id) {
1350        state.insert_triggered(client_order_id);
1351
1352        if TRIGGERABLE_ORDER_TYPES.contains(&context.identity.order_type) {
1353            let triggered = OrderTriggered::new(
1354                emitter.trader_id(),
1355                context.identity.strategy_id,
1356                context.identity.instrument_id,
1357                client_order_id,
1358                UUID4::new(),
1359                ts_event,
1360                ts_init,
1361                false,
1362                Some(child_venue_order_id),
1363                Some(account_id),
1364            );
1365            emitter.send_order_event(OrderEventAny::Triggered(triggered));
1366        }
1367    }
1368}
1369
1370#[expect(clippy::too_many_arguments)]
1371fn refresh_bound_child_and_emit_update(
1372    client_order_id: ClientOrderId,
1373    child_venue_order_id: VenueOrderId,
1374    context: OrderContext,
1375    account_id: AccountId,
1376    ts_event: UnixNanos,
1377    ts_init: UnixNanos,
1378    emitter: &ExecutionEventEmitter,
1379    state: &WsDispatchState,
1380    emit_update: bool,
1381) {
1382    let terms_changed = state
1383        .order_contexts
1384        .get(&client_order_id)
1385        .is_some_and(|previous| {
1386            previous.quantity != context.quantity
1387                || previous.price != context.price
1388                || previous.trigger_price != context.trigger_price
1389        });
1390    state.track_order_context(context);
1391
1392    if !terms_changed || !emit_update {
1393        return;
1394    }
1395
1396    state.insert_accepted(client_order_id, child_venue_order_id);
1397    emit_order_update(
1398        child_venue_order_id,
1399        context,
1400        account_id,
1401        ts_event,
1402        ts_init,
1403        false,
1404        emitter,
1405    );
1406}
1407
1408fn emit_order_update(
1409    child_venue_order_id: VenueOrderId,
1410    context: OrderContext,
1411    account_id: AccountId,
1412    ts_event: UnixNanos,
1413    ts_init: UnixNanos,
1414    reconciliation: bool,
1415    emitter: &ExecutionEventEmitter,
1416) {
1417    let updated = OrderUpdated::new(
1418        emitter.trader_id(),
1419        context.identity.strategy_id,
1420        context.identity.instrument_id,
1421        context.identity.client_order_id,
1422        context.quantity,
1423        UUID4::new(),
1424        ts_event,
1425        ts_init,
1426        reconciliation,
1427        Some(child_venue_order_id),
1428        Some(account_id),
1429        context.price,
1430        context.trigger_price,
1431        None,
1432        context.is_quote_quantity,
1433    );
1434    emitter.send_order_event(OrderEventAny::Updated(updated));
1435}
1436
1437/// Dispatches order messages, producing proper order events for tracked orders
1438/// and falling back to execution reports for untracked orders.
1439#[expect(clippy::too_many_arguments)]
1440fn dispatch_order_messages(
1441    order_msgs: &[OKXOrderMsg],
1442    emitter: &ExecutionEventEmitter,
1443    state: &WsDispatchState,
1444    account_id: AccountId,
1445    instruments: &AHashMap<Ustr, InstrumentAny>,
1446    fee_cache: &mut FeeCache,
1447    filled_qty_cache: &mut FilledQtyCache,
1448    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
1449    ts_init: UnixNanos,
1450) {
1451    for msg in order_msgs {
1452        let Some(instrument) = instruments.get(&msg.inst_id) else {
1453            log::warn!("No instrument for {}, skipping order message", msg.inst_id);
1454            continue;
1455        };
1456
1457        let direct_client_order_id = state
1458            .lifecycle_bindings
1459            .lock()
1460            .client_by_parent
1461            .get(&VenueOrderId::new(msg.ord_id))
1462            .copied()
1463            .or_else(|| parse_client_order_id(&msg.cl_ord_id));
1464        let parent_client_order_id = msg
1465            .algo_cl_ord_id
1466            .as_deref()
1467            .and_then(parse_client_order_id);
1468        let linked_parent_venue_order_id = msg
1469            .algo_id
1470            .as_deref()
1471            .filter(|value| !value.is_empty())
1472            .or_else(|| {
1473                msg.linked_algo_ord
1474                    .as_ref()
1475                    .map(|linked| linked.algo_id.as_str())
1476                    .filter(|value| !value.is_empty())
1477            })
1478            .map(VenueOrderId::new);
1479
1480        // Triggered child orders may have a generated or empty cl_ord_id.
1481        // Resolve the tracked parent before falling back to a report.
1482        let direct_resolution = [direct_client_order_id, parent_client_order_id]
1483            .into_iter()
1484            .flatten()
1485            .find_map(|client_order_id| {
1486                state
1487                    .order_identity(client_order_id)
1488                    .map(|identity| (client_order_id, Some(identity)))
1489            });
1490
1491        let bound_client_order_id = if direct_resolution.is_none()
1492            && let Some(parent_venue_order_id) = linked_parent_venue_order_id
1493        {
1494            match state.resolve_or_hold_linked_child(parent_venue_order_id, instrument.id(), msg) {
1495                LinkedChildResolution::Bound(client_order_id) => Some(client_order_id),
1496                LinkedChildResolution::Held => {
1497                    log::debug!(
1498                        "Holding linked child update during algo parent binding: ord_id={} parent_id={parent_venue_order_id}",
1499                        msg.ord_id,
1500                    );
1501                    continue;
1502                }
1503                LinkedChildResolution::External => None,
1504            }
1505        } else {
1506            None
1507        };
1508
1509        let resolved = direct_resolution
1510            .or_else(|| {
1511                bound_client_order_id.and_then(|client_order_id| {
1512                    state
1513                        .order_identity(client_order_id)
1514                        .map(|identity| (client_order_id, Some(identity)))
1515                })
1516            })
1517            .or_else(|| {
1518                bound_client_order_id
1519                    .or(parent_client_order_id)
1520                    .or(direct_client_order_id)
1521                    .map(|client_order_id| (client_order_id, None))
1522            });
1523
1524        let Some((client_order_id, identity)) = resolved else {
1525            log::debug!(
1526                "Order without client or algo client order ID (ord_id={}), sending as report",
1527                msg.ord_id
1528            );
1529            dispatch_order_msg_as_report(
1530                msg,
1531                account_id,
1532                instruments,
1533                fee_cache,
1534                filled_qty_cache,
1535                emitter,
1536                state,
1537                ts_init,
1538            );
1539            continue;
1540        };
1541
1542        if let Some(ident) = identity {
1543            let context = state
1544                .order_contexts
1545                .get(&client_order_id)
1546                .filter(|entry| OKX_CONDITIONAL_ORDER_TYPES.contains(&entry.identity.order_type))
1547                .map(|entry| refresh_regular_child_context(msg, *entry, instrument));
1548            let mut lifecycle_bindings = context.map(|_| state.lifecycle_bindings.lock());
1549
1550            if let (Some(context), Some(bindings)) = (context, lifecycle_bindings.as_mut()) {
1551                let parent_venue_order_id = linked_parent_venue_order_id.or_else(|| {
1552                    bindings
1553                        .venue_by_client
1554                        .get(&client_order_id)
1555                        .map(|binding| binding.parent)
1556                });
1557
1558                let child_venue_order_id = VenueOrderId::new(msg.ord_id);
1559                let parent_venue_order_id = parent_venue_order_id.unwrap_or(child_venue_order_id);
1560                let mut binding = bindings
1561                    .venue_by_client
1562                    .get(&client_order_id)
1563                    .copied()
1564                    .unwrap_or(OrderVenueBinding {
1565                        parent: parent_venue_order_id,
1566                        child: None,
1567                    });
1568
1569                if binding.parent != parent_venue_order_id {
1570                    log::error!(
1571                        "Suppressing conflicting child parent binding for {client_order_id}: expected={} received={parent_venue_order_id}",
1572                        binding.parent,
1573                    );
1574                    continue;
1575                }
1576
1577                bindings
1578                    .client_by_parent
1579                    .insert(parent_venue_order_id, client_order_id);
1580                let ts_event = parse_millisecond_timestamp(msg.u_time);
1581
1582                if binding.child == Some(child_venue_order_id) {
1583                    refresh_bound_child_and_emit_update(
1584                        client_order_id,
1585                        child_venue_order_id,
1586                        context,
1587                        account_id,
1588                        ts_event,
1589                        ts_init,
1590                        emitter,
1591                        state,
1592                        !order_state_cache.contains_key(&client_order_id),
1593                    );
1594                } else {
1595                    bind_algo_child_and_emit_transition(
1596                        client_order_id,
1597                        parent_venue_order_id,
1598                        child_venue_order_id,
1599                        &mut binding,
1600                        context,
1601                        account_id,
1602                        ts_event,
1603                        ts_init,
1604                        emitter,
1605                        state,
1606                    );
1607                }
1608
1609                bindings.venue_by_client.insert(client_order_id, binding);
1610            }
1611
1612            let is_post_only_cancel = is_post_only_auto_cancel(msg);
1613
1614            if is_post_only_cancel
1615                || (!state.contains_accepted(&client_order_id) && is_unfilled_rpi_cancel(msg))
1616            {
1617                if is_post_only_cancel {
1618                    state.insert_post_only_rejection(msg.ord_id);
1619                }
1620
1621                let ts_event = parse_millisecond_timestamp(msg.u_time);
1622                let reason = if msg.ord_type == OKXOrderType::Rpi {
1623                    msg.cancel_source_reason
1624                        .as_deref()
1625                        .filter(|reason| !reason.is_empty())
1626                        .unwrap_or("RPI order canceled before acceptance")
1627                } else {
1628                    "Post-only order would have taken liquidity"
1629                };
1630                let rejected = OrderRejected::new(
1631                    emitter.trader_id(),
1632                    ident.strategy_id,
1633                    instrument.id(),
1634                    client_order_id,
1635                    account_id,
1636                    Ustr::from(reason),
1637                    UUID4::new(),
1638                    ts_event,
1639                    ts_init,
1640                    false,
1641                    true, // due_post_only
1642                );
1643                state.remove_order_tracking(client_order_id);
1644                if let Some(bindings) = lifecycle_bindings.as_mut() {
1645                    state.insert_terminal(client_order_id);
1646                    if let Some(binding) = bindings.venue_by_client.get(&client_order_id).copied() {
1647                        bindings.finish(client_order_id, binding);
1648                    }
1649                }
1650
1651                order_state_cache.remove(&client_order_id);
1652                fee_cache.remove(&msg.ord_id);
1653                filled_qty_cache.remove(&msg.ord_id);
1654                emitter.send_order_event(OrderEventAny::Rejected(rejected));
1655                continue;
1656            }
1657
1658            let previous_fee = fee_cache.get(&msg.ord_id).copied();
1659            let previous_filled_qty = filled_qty_cache.get(&msg.ord_id).copied();
1660            let previous_state = order_state_cache.get(&client_order_id);
1661
1662            match parse_order_event(
1663                msg,
1664                client_order_id,
1665                account_id,
1666                emitter.trader_id(),
1667                ident.strategy_id,
1668                instrument,
1669                previous_fee,
1670                previous_filled_qty,
1671                previous_state,
1672                ts_init,
1673            ) {
1674                Ok(event) => {
1675                    let needs_status =
1676                        matches!(event, ParsedOrderEvent::Fill(_) | ParsedOrderEvent::Skipped);
1677                    update_order_state_cache(msg, instrument, client_order_id, order_state_cache);
1678                    dispatch_imported_order_update(
1679                        client_order_id,
1680                        msg.state,
1681                        parse_millisecond_timestamp(msg.u_time),
1682                        ts_init,
1683                        emitter,
1684                        state,
1685                        order_state_cache,
1686                        &event,
1687                    );
1688                    dispatch_parsed_order_event(
1689                        event,
1690                        client_order_id,
1691                        account_id,
1692                        VenueOrderId::new(msg.ord_id),
1693                        &ident,
1694                        instrument,
1695                        msg.state,
1696                        emitter,
1697                        state,
1698                        order_state_cache,
1699                        ts_init,
1700                    );
1701
1702                    dispatch_filled_order_status(
1703                        msg,
1704                        Some(client_order_id),
1705                        instrument,
1706                        emitter,
1707                        state,
1708                        ts_init,
1709                        needs_status,
1710                    );
1711
1712                    if state.contains_terminal(&client_order_id)
1713                        && let Some(bindings) = lifecycle_bindings.as_mut()
1714                        && let Some(binding) =
1715                            bindings.venue_by_client.get(&client_order_id).copied()
1716                    {
1717                        bindings.finish(client_order_id, binding);
1718                    }
1719
1720                    update_fee_fill_caches(msg, instrument, fee_cache, filled_qty_cache);
1721                }
1722                Err(e) => log::error!("Failed to parse order event for {client_order_id}: {e}"),
1723            }
1724        } else if state.contains_terminal(&client_order_id) {
1725            dispatch_terminal_order_fill_as_report(
1726                msg,
1727                client_order_id,
1728                account_id,
1729                instruments,
1730                fee_cache,
1731                filled_qty_cache,
1732                emitter,
1733                state,
1734                ts_init,
1735            );
1736        } else if is_post_only_auto_cancel(msg) && state.contains_post_only_rejection(&msg.ord_id) {
1737            log::debug!(
1738                "Skipping replayed post-only rejection for {client_order_id}: ord_id={}",
1739                msg.ord_id
1740            );
1741        } else {
1742            log::debug!(
1743                "Untracked order {client_order_id} (ord_id={}), sending as report for reconciliation",
1744                msg.ord_id
1745            );
1746            dispatch_order_msg_as_report(
1747                msg,
1748                account_id,
1749                instruments,
1750                fee_cache,
1751                filled_qty_cache,
1752                emitter,
1753                state,
1754                ts_init,
1755            );
1756        }
1757    }
1758}
1759
1760#[expect(clippy::too_many_arguments)]
1761fn dispatch_imported_order_update(
1762    client_order_id: ClientOrderId,
1763    venue_status: OKXOrderStatus,
1764    ts_event: UnixNanos,
1765    ts_init: UnixNanos,
1766    emitter: &ExecutionEventEmitter,
1767    state: &WsDispatchState,
1768    order_state_cache: &AHashMap<ClientOrderId, OrderStateSnapshot>,
1769    event: &ParsedOrderEvent,
1770) {
1771    let Some(mut context) = state
1772        .order_contexts
1773        .get_mut(&client_order_id)
1774        .filter(|entry| entry.identity.order_type == OrderType::Limit)
1775    else {
1776        return;
1777    };
1778
1779    let Some(snapshot) = order_state_cache.get(&client_order_id) else {
1780        return;
1781    };
1782
1783    let terms_changed = snapshot.quantity != context.quantity || snapshot.price != context.price;
1784    context.quantity = snapshot.quantity;
1785    context.price = snapshot.price;
1786
1787    if terms_changed
1788        && !matches!(event, ParsedOrderEvent::Updated(_))
1789        && matches!(
1790            venue_status,
1791            OKXOrderStatus::Live | OKXOrderStatus::PartiallyFilled | OKXOrderStatus::Filled
1792        )
1793    {
1794        emit_order_update(
1795            snapshot.venue_order_id,
1796            *context,
1797            emitter.account_id(),
1798            ts_event,
1799            ts_init,
1800            venue_status == OKXOrderStatus::Filled,
1801            emitter,
1802        );
1803    }
1804}
1805
1806#[expect(clippy::too_many_arguments)]
1807fn dispatch_spread_order_messages(
1808    order_msgs: &[OKXSpreadOrder],
1809    emitter: &ExecutionEventEmitter,
1810    state: &WsDispatchState,
1811    account_id: AccountId,
1812    instruments: &AHashMap<Ustr, InstrumentAny>,
1813    filled_qty_cache: &mut FilledQtyCache,
1814    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
1815    ts_init: UnixNanos,
1816) {
1817    for msg in order_msgs {
1818        let Some(instrument) = instruments.get(&msg.sprd_id) else {
1819            log::warn!(
1820                "No instrument for {}, skipping spread order message",
1821                msg.sprd_id
1822            );
1823            continue;
1824        };
1825
1826        let client_order_id = state
1827            .lifecycle_bindings
1828            .lock()
1829            .client_by_parent
1830            .get(&VenueOrderId::new(msg.ord_id.as_str()))
1831            .copied()
1832            .or_else(|| parse_client_order_id(msg.cl_ord_id.as_str()));
1833
1834        let Some(client_order_id) = client_order_id else {
1835            log::debug!(
1836                "Spread order without client_order_id (ord_id={}), sending as report",
1837                msg.ord_id
1838            );
1839            dispatch_spread_order_msg_as_report(
1840                msg,
1841                account_id,
1842                instruments,
1843                filled_qty_cache,
1844                emitter,
1845                state,
1846                ts_init,
1847            );
1848            continue;
1849        };
1850
1851        let identity = state.order_identity(client_order_id);
1852
1853        if let Some(ident) = identity {
1854            if is_spread_post_only_auto_cancel(msg) {
1855                let ts_event = msg
1856                    .u_time
1857                    .or(msg.c_time)
1858                    .map_or(ts_init, parse_millisecond_timestamp);
1859                let rejected = OrderRejected::new(
1860                    emitter.trader_id(),
1861                    ident.strategy_id,
1862                    instrument.id(),
1863                    client_order_id,
1864                    account_id,
1865                    Ustr::from(OKX_POST_ONLY_CANCEL_REASON),
1866                    UUID4::new(),
1867                    ts_event,
1868                    ts_init,
1869                    false,
1870                    true,
1871                );
1872                state.remove_order_tracking(client_order_id);
1873                order_state_cache.remove(&client_order_id);
1874                filled_qty_cache.remove(&msg.ord_id);
1875                emitter.send_order_event(OrderEventAny::Rejected(rejected));
1876                continue;
1877            }
1878
1879            let previous_filled_qty = filled_qty_cache.get(&msg.ord_id).copied();
1880            let previous_state = order_state_cache.get(&client_order_id);
1881
1882            match parse_spread_order_event(
1883                msg,
1884                client_order_id,
1885                account_id,
1886                emitter.trader_id(),
1887                ident.strategy_id,
1888                instrument,
1889                previous_filled_qty,
1890                previous_state,
1891                ts_init,
1892            ) {
1893                Ok(event) => {
1894                    update_spread_order_state_cache(
1895                        msg,
1896                        instrument,
1897                        client_order_id,
1898                        order_state_cache,
1899                    );
1900                    let ts_event = msg
1901                        .u_time
1902                        .or(msg.c_time)
1903                        .map_or(ts_init, parse_millisecond_timestamp);
1904                    dispatch_imported_order_update(
1905                        client_order_id,
1906                        msg.state,
1907                        ts_event,
1908                        ts_init,
1909                        emitter,
1910                        state,
1911                        order_state_cache,
1912                        &event,
1913                    );
1914                    dispatch_parsed_order_event(
1915                        event,
1916                        client_order_id,
1917                        account_id,
1918                        VenueOrderId::new(msg.ord_id.as_str()),
1919                        &ident,
1920                        instrument,
1921                        msg.state,
1922                        emitter,
1923                        state,
1924                        order_state_cache,
1925                        ts_init,
1926                    );
1927                    update_spread_fill_cache(msg, instrument, filled_qty_cache);
1928                }
1929                Err(e) => {
1930                    log::error!("Failed to parse spread order event for {client_order_id}: {e}");
1931                }
1932            }
1933        } else {
1934            log::debug!(
1935                "Untracked spread order {client_order_id} (ord_id={}), sending as report for reconciliation",
1936                msg.ord_id
1937            );
1938            dispatch_spread_order_msg_as_report(
1939                msg,
1940                account_id,
1941                instruments,
1942                filled_qty_cache,
1943                emitter,
1944                state,
1945                ts_init,
1946            );
1947        }
1948    }
1949}
1950
1951/// Dispatches a parsed order event as a proper `OrderEventAny`.
1952///
1953/// Guarantees the `Submitted -> Accepted -> ...` lifecycle by synthesizing
1954/// `OrderAccepted` before any other event when one has not yet been emitted.
1955/// Duplicate `Accepted` events (e.g. from reconnect replays) are suppressed.
1956#[expect(clippy::too_many_arguments)]
1957fn dispatch_parsed_order_event(
1958    event: ParsedOrderEvent,
1959    client_order_id: ClientOrderId,
1960    account_id: AccountId,
1961    venue_order_id: VenueOrderId,
1962    identity: &OrderIdentity,
1963    instrument: &InstrumentAny,
1964    venue_status: OKXOrderStatus,
1965    emitter: &ExecutionEventEmitter,
1966    state: &WsDispatchState,
1967    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
1968    ts_init: UnixNanos,
1969) {
1970    let is_terminal;
1971
1972    match event {
1973        ParsedOrderEvent::Accepted(e) => {
1974            if state.contains_filled(&client_order_id) || state.contains_terminal(&client_order_id)
1975            {
1976                log::debug!("Skipping duplicate Accepted for {client_order_id}");
1977                return;
1978            }
1979
1980            if state.contains_accepted(&client_order_id) {
1981                emit_venue_order_id_update_if_changed(
1982                    client_order_id,
1983                    account_id,
1984                    venue_order_id,
1985                    identity,
1986                    e.ts_event,
1987                    emitter,
1988                    state,
1989                    order_state_cache,
1990                    ts_init,
1991                );
1992                return;
1993            }
1994
1995            if state.contains_triggered(&client_order_id) {
1996                log::debug!("Skipping duplicate Accepted for {client_order_id}");
1997                return;
1998            }
1999
2000            state.insert_accepted(client_order_id, venue_order_id);
2001            is_terminal = false;
2002            emitter.send_order_event(OrderEventAny::Accepted(e));
2003        }
2004        ParsedOrderEvent::Triggered(e) => {
2005            if state.contains_filled(&client_order_id) {
2006                log::debug!("Skipping stale Triggered for {client_order_id} (already filled)");
2007                return;
2008            }
2009
2010            if !TRIGGERABLE_ORDER_TYPES.contains(&identity.order_type) {
2011                log::debug!(
2012                    "Skipping OrderTriggered for {} order {client_order_id}: market-style stops have no TRIGGERED state",
2013                    identity.order_type,
2014                );
2015                state.insert_triggered(client_order_id);
2016                return;
2017            }
2018
2019            ensure_accepted_emitted(
2020                client_order_id,
2021                account_id,
2022                venue_order_id,
2023                identity,
2024                emitter,
2025                state,
2026                ts_init,
2027                ts_init,
2028            );
2029            state.insert_triggered(client_order_id);
2030            is_terminal = false;
2031            emitter.send_order_event(OrderEventAny::Triggered(e));
2032        }
2033        ParsedOrderEvent::Canceled(e) => {
2034            ensure_accepted_emitted(
2035                client_order_id,
2036                account_id,
2037                venue_order_id,
2038                identity,
2039                emitter,
2040                state,
2041                ts_init,
2042                ts_init,
2043            );
2044            state.remove_triggered(&client_order_id);
2045            state.remove_filled(&client_order_id);
2046            is_terminal = true;
2047            emitter.send_order_event(OrderEventAny::Canceled(e));
2048        }
2049        ParsedOrderEvent::Expired(e) => {
2050            ensure_accepted_emitted(
2051                client_order_id,
2052                account_id,
2053                venue_order_id,
2054                identity,
2055                emitter,
2056                state,
2057                ts_init,
2058                ts_init,
2059            );
2060            state.remove_triggered(&client_order_id);
2061            state.remove_filled(&client_order_id);
2062            is_terminal = true;
2063            emitter.send_order_event(OrderEventAny::Expired(e));
2064        }
2065        ParsedOrderEvent::Updated(e) => {
2066            ensure_accepted_emitted(
2067                client_order_id,
2068                account_id,
2069                venue_order_id,
2070                identity,
2071                emitter,
2072                state,
2073                ts_init,
2074                ts_init,
2075            );
2076            is_terminal = false;
2077            emitter.send_order_event(OrderEventAny::Updated(e));
2078        }
2079        ParsedOrderEvent::Fill(fill_report) => {
2080            is_terminal = venue_status == OKXOrderStatus::Filled;
2081
2082            if state.check_and_insert_trade(fill_report.trade_id) {
2083                log::debug!(
2084                    "Skipping duplicate fill for {client_order_id}: trade_id={}",
2085                    fill_report.trade_id
2086                );
2087            } else {
2088                emit_venue_order_id_update_if_changed(
2089                    client_order_id,
2090                    account_id,
2091                    venue_order_id,
2092                    identity,
2093                    fill_report.ts_event,
2094                    emitter,
2095                    state,
2096                    order_state_cache,
2097                    ts_init,
2098                );
2099                ensure_accepted_emitted(
2100                    client_order_id,
2101                    account_id,
2102                    venue_order_id,
2103                    identity,
2104                    emitter,
2105                    state,
2106                    ts_init,
2107                    ts_init,
2108                );
2109                state.insert_filled(client_order_id);
2110                state.remove_triggered(&client_order_id);
2111                let filled = fill_report_to_order_filled(
2112                    &fill_report,
2113                    emitter.trader_id(),
2114                    identity,
2115                    instrument.quote_currency(),
2116                );
2117                emitter.send_order_event(OrderEventAny::Filled(filled));
2118            }
2119        }
2120        ParsedOrderEvent::StatusOnly(report) => {
2121            is_terminal = matches!(
2122                report.order_status,
2123                OrderStatus::Filled | OrderStatus::Canceled | OrderStatus::Expired
2124            );
2125            emitter.send_order_status_report(*report);
2126        }
2127        ParsedOrderEvent::Skipped => {
2128            is_terminal = venue_status == OKXOrderStatus::Filled;
2129        }
2130    }
2131
2132    if is_terminal {
2133        state.insert_terminal(client_order_id);
2134        state.remove_order_tracking(client_order_id);
2135        state.remove_accepted(&client_order_id);
2136        order_state_cache.remove(&client_order_id);
2137        // Keep fee_cache and filled_qty_cache entries: replayed terminal
2138        // messages go through the untracked report path and need prior
2139        // cumulative state to avoid re-emitting the full fill quantity
2140    }
2141}
2142
2143/// Synthesizes and emits `OrderAccepted` if one has not yet been emitted for
2144/// this order. Handles fast-filling orders that skip the `Live` state on OKX.
2145#[expect(
2146    clippy::too_many_arguments,
2147    reason = "acceptance reconstruction requires identity, venue state, and event timestamps"
2148)]
2149fn ensure_accepted_emitted(
2150    client_order_id: ClientOrderId,
2151    account_id: AccountId,
2152    venue_order_id: VenueOrderId,
2153    identity: &OrderIdentity,
2154    emitter: &ExecutionEventEmitter,
2155    state: &WsDispatchState,
2156    ts_event: UnixNanos,
2157    ts_init: UnixNanos,
2158) {
2159    if state.contains_accepted(&client_order_id)
2160        || state.contains_terminal(&client_order_id)
2161        || state.contains_filled(&client_order_id)
2162    {
2163        return;
2164    }
2165
2166    state.insert_accepted(client_order_id, venue_order_id);
2167    let accepted = OrderAccepted::new(
2168        emitter.trader_id(),
2169        identity.strategy_id,
2170        identity.instrument_id,
2171        client_order_id,
2172        venue_order_id,
2173        account_id,
2174        UUID4::new(),
2175        ts_event,
2176        ts_init,
2177        false,
2178    );
2179    emitter.send_order_event(OrderEventAny::Accepted(accepted));
2180}
2181
2182#[expect(clippy::too_many_arguments)]
2183fn emit_venue_order_id_update_if_changed(
2184    client_order_id: ClientOrderId,
2185    account_id: AccountId,
2186    venue_order_id: VenueOrderId,
2187    identity: &OrderIdentity,
2188    ts_event: UnixNanos,
2189    emitter: &ExecutionEventEmitter,
2190    state: &WsDispatchState,
2191    order_state_cache: &AHashMap<ClientOrderId, OrderStateSnapshot>,
2192    ts_init: UnixNanos,
2193) {
2194    let Some(accepted_venue_order_id) = state.accepted_venue_order_id(&client_order_id) else {
2195        return;
2196    };
2197
2198    if accepted_venue_order_id == venue_order_id {
2199        return;
2200    }
2201
2202    let Some(snapshot) = order_state_cache.get(&client_order_id) else {
2203        return;
2204    };
2205
2206    state.insert_accepted(client_order_id, venue_order_id);
2207    let updated = OrderUpdated::new(
2208        emitter.trader_id(),
2209        identity.strategy_id,
2210        identity.instrument_id,
2211        client_order_id,
2212        snapshot.quantity,
2213        UUID4::new(),
2214        ts_event,
2215        ts_init,
2216        false,
2217        Some(venue_order_id),
2218        Some(account_id),
2219        snapshot.price,
2220        None,
2221        None,
2222        false,
2223    );
2224    emitter.send_order_event(OrderEventAny::Updated(updated));
2225}
2226
2227/// Converts a [`FillReport`] into an [`OrderFilled`] event using tracked identity.
2228fn fill_report_to_order_filled(
2229    report: &FillReport,
2230    trader_id: TraderId,
2231    identity: &OrderIdentity,
2232    quote_currency: Currency,
2233) -> OrderFilled {
2234    OrderFilled::new(
2235        trader_id,
2236        identity.strategy_id,
2237        report.instrument_id,
2238        report
2239            .client_order_id
2240            .expect("tracked order has client_order_id"),
2241        report.venue_order_id,
2242        report.account_id,
2243        report.trade_id,
2244        identity.order_side,
2245        identity.order_type,
2246        report.last_qty,
2247        report.last_px,
2248        quote_currency,
2249        report.liquidity_side,
2250        UUID4::new(),
2251        report.ts_event,
2252        report.ts_init,
2253        false,
2254        report.venue_position_id,
2255        Some(report.commission),
2256        None,
2257    )
2258}
2259
2260/// Falls back to the report path for a single order message.
2261#[expect(clippy::too_many_arguments)]
2262fn dispatch_order_msg_as_report(
2263    msg: &OKXOrderMsg,
2264    account_id: AccountId,
2265    instruments: &AHashMap<Ustr, InstrumentAny>,
2266    fee_cache: &mut FeeCache,
2267    filled_qty_cache: &mut FilledQtyCache,
2268    emitter: &ExecutionEventEmitter,
2269    state: &WsDispatchState,
2270    ts_init: UnixNanos,
2271) {
2272    match parse_order_msg(
2273        msg,
2274        account_id,
2275        instruments,
2276        fee_cache,
2277        filled_qty_cache,
2278        ts_init,
2279    ) {
2280        Ok(report) => {
2281            let needs_status = matches!(report, ExecutionReport::Fill(_));
2282            dispatch_execution_reports(vec![report], emitter, state);
2283
2284            if let Some(instrument) = instruments.get(&msg.inst_id) {
2285                dispatch_filled_order_status(
2286                    msg,
2287                    None,
2288                    instrument,
2289                    emitter,
2290                    state,
2291                    ts_init,
2292                    needs_status,
2293                );
2294                update_fee_fill_caches(msg, instrument, fee_cache, filled_qty_cache);
2295            }
2296        }
2297        Err(e) => log::error!("Failed to parse order message as report: {e}"),
2298    }
2299}
2300
2301fn dispatch_filled_order_status(
2302    msg: &OKXOrderMsg,
2303    client_order_id: Option<ClientOrderId>,
2304    instrument: &InstrumentAny,
2305    emitter: &ExecutionEventEmitter,
2306    state: &WsDispatchState,
2307    ts_init: UnixNanos,
2308    needs_status: bool,
2309) {
2310    if !needs_status || msg.state != OKXOrderStatus::Filled {
2311        return;
2312    }
2313
2314    let report = parse_order_status_report(msg, instrument, emitter.account_id(), ts_init)
2315        .and_then(|report| {
2316            let requested = parse_quantity(&msg.sz, instrument.size_precision())?;
2317            Ok((report, requested))
2318        });
2319
2320    match report {
2321        Ok((mut report, requested)) => {
2322            if report.filled_qty.is_positive() && report.quantity != requested {
2323                report.client_order_id = client_order_id.or(report.client_order_id);
2324
2325                // The caller dispatches any real fill first; reconcile base shortfalls and quote sizing
2326                dispatch_execution_reports(vec![ExecutionReport::Order(report)], emitter, state);
2327            }
2328        }
2329        Err(e) => log::error!(
2330            "Failed to parse filled order status for {}: {e}",
2331            msg.ord_id
2332        ),
2333    }
2334}
2335
2336#[expect(clippy::too_many_arguments)]
2337fn dispatch_terminal_order_fill_as_report(
2338    msg: &OKXOrderMsg,
2339    client_order_id: ClientOrderId,
2340    account_id: AccountId,
2341    instruments: &AHashMap<Ustr, InstrumentAny>,
2342    fee_cache: &mut FeeCache,
2343    filled_qty_cache: &mut FilledQtyCache,
2344    emitter: &ExecutionEventEmitter,
2345    state: &WsDispatchState,
2346    ts_init: UnixNanos,
2347) {
2348    match parse_order_msg(
2349        msg,
2350        account_id,
2351        instruments,
2352        fee_cache,
2353        filled_qty_cache,
2354        ts_init,
2355    ) {
2356        Ok(ExecutionReport::Fill(mut report)) => {
2357            report.client_order_id = Some(client_order_id);
2358            dispatch_execution_reports(vec![ExecutionReport::Fill(report)], emitter, state);
2359
2360            if let Some(instrument) = instruments.get(&msg.inst_id) {
2361                update_fee_fill_caches(msg, instrument, fee_cache, filled_qty_cache);
2362            }
2363        }
2364        Ok(ExecutionReport::Order(_)) => {
2365            log::debug!(
2366                "Suppressing stale regular order status for terminal tracked order {client_order_id}: ord_id={}",
2367                msg.ord_id,
2368            );
2369        }
2370        Err(e) => log::error!("Failed to parse terminal order update: {e}"),
2371    }
2372}
2373
2374fn dispatch_spread_order_msg_as_report(
2375    msg: &OKXSpreadOrder,
2376    account_id: AccountId,
2377    instruments: &AHashMap<Ustr, InstrumentAny>,
2378    filled_qty_cache: &mut FilledQtyCache,
2379    emitter: &ExecutionEventEmitter,
2380    state: &WsDispatchState,
2381    ts_init: UnixNanos,
2382) {
2383    match parse_spread_order_msg(msg, account_id, instruments, filled_qty_cache, ts_init) {
2384        Ok(report) => {
2385            dispatch_execution_reports(vec![report], emitter, state);
2386
2387            if let Some(instrument) = instruments.get(&msg.sprd_id) {
2388                update_spread_fill_cache(msg, instrument, filled_qty_cache);
2389            }
2390        }
2391        Err(e) => log::error!("Failed to parse spread order message as report: {e}"),
2392    }
2393}
2394
2395/// Updates fee, fill, and order state caches from a raw OKX order message.
2396fn update_order_state_cache(
2397    msg: &OKXOrderMsg,
2398    instrument: &InstrumentAny,
2399    client_order_id: ClientOrderId,
2400    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
2401) {
2402    let venue_order_id = VenueOrderId::new(msg.ord_id);
2403    let quantity = parse_quantity(&msg.sz, instrument.size_precision()).unwrap_or_default();
2404    let price = if is_market_price(&msg.px) {
2405        None
2406    } else {
2407        parse_price(&msg.px, instrument.price_precision()).ok()
2408    };
2409
2410    order_state_cache.insert(
2411        client_order_id,
2412        OrderStateSnapshot {
2413            venue_order_id,
2414            quantity,
2415            price,
2416        },
2417    );
2418}
2419
2420fn update_spread_order_state_cache(
2421    msg: &OKXSpreadOrder,
2422    instrument: &InstrumentAny,
2423    client_order_id: ClientOrderId,
2424    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
2425) {
2426    let venue_order_id = VenueOrderId::new(msg.ord_id.as_str());
2427    let quantity = parse_quantity(&msg.sz, instrument.size_precision()).unwrap_or_default();
2428    let price = if is_market_price(&msg.px) {
2429        None
2430    } else {
2431        parse_price(&msg.px, instrument.price_precision()).ok()
2432    };
2433
2434    order_state_cache.insert(
2435        client_order_id,
2436        OrderStateSnapshot {
2437            venue_order_id,
2438            quantity,
2439            price,
2440        },
2441    );
2442}
2443
2444fn update_spread_fill_cache(
2445    msg: &OKXSpreadOrder,
2446    instrument: &InstrumentAny,
2447    filled_qty_cache: &mut FilledQtyCache,
2448) {
2449    if !msg.acc_fill_sz.is_empty()
2450        && msg.acc_fill_sz != "0"
2451        && let Ok(qty) = parse_quantity(&msg.acc_fill_sz, instrument.size_precision())
2452    {
2453        filled_qty_cache.record(msg.ord_id, qty, is_terminal_order_state(msg.state));
2454    }
2455}
2456
2457fn is_spread_post_only_auto_cancel(msg: &OKXSpreadOrder) -> bool {
2458    msg.state == OKXOrderStatus::Canceled && msg.cancel_source == OKX_POST_ONLY_CANCEL_SOURCE
2459}
2460
2461/// Dispatches execution reports with cross-stream deduplication.
2462pub fn dispatch_execution_reports(
2463    reports: Vec<ExecutionReport>,
2464    emitter: &ExecutionEventEmitter,
2465    state: &WsDispatchState,
2466) {
2467    log::debug!("Processing {} execution report(s)", reports.len());
2468
2469    for report in reports {
2470        match report {
2471            ExecutionReport::Order(order_report) => {
2472                if let Some(cid) = order_report.client_order_id {
2473                    match order_report.order_status {
2474                        // Guard form reformats awkwardly across multiple lines
2475                        #[allow(clippy::collapsible_match)]
2476                        OrderStatus::Accepted => {
2477                            if state.contains_terminal(&cid)
2478                                || state.contains_filled(&cid)
2479                                || state.contains_triggered(&cid)
2480                            {
2481                                log::debug!(
2482                                    "Skipping stale OrderStatusReport(Accepted) \
2483                                     for {cid} (order already terminal)"
2484                                );
2485                                continue;
2486                            }
2487
2488                            if !state.contains_accepted(&cid) {
2489                                state.insert_accepted(cid, order_report.venue_order_id);
2490                            }
2491                        }
2492                        OrderStatus::Triggered => {
2493                            if state.contains_filled(&cid) {
2494                                log::debug!(
2495                                    "Skipping stale OrderStatusReport(Triggered) \
2496                                     for {cid} (already filled)"
2497                                );
2498                                continue;
2499                            }
2500                            state.insert_triggered(cid);
2501                        }
2502                        OrderStatus::Filled => {
2503                            state.insert_filled(cid);
2504                            state.insert_terminal(cid);
2505                            state.remove_triggered(&cid);
2506                        }
2507                        OrderStatus::Canceled | OrderStatus::Expired | OrderStatus::Rejected => {
2508                            state.insert_terminal(cid);
2509                            state.remove_triggered(&cid);
2510                            state.remove_filled(&cid);
2511                        }
2512                        _ => {}
2513                    }
2514                }
2515                emitter.send_order_status_report(order_report);
2516            }
2517            ExecutionReport::Fill(fill_report) => {
2518                if state.check_and_insert_trade(fill_report.trade_id) {
2519                    log::debug!(
2520                        "Skipping duplicate fill report: trade_id={}",
2521                        fill_report.trade_id
2522                    );
2523                    continue;
2524                }
2525
2526                if let Some(cid) = fill_report.client_order_id {
2527                    state.insert_filled(cid);
2528                    state.remove_triggered(&cid);
2529                }
2530                emitter.send_fill_report(fill_report);
2531            }
2532        }
2533    }
2534}
2535
2536fn emit_send_failed_submit(
2537    failure: &CommandFailure,
2538    state: &WsDispatchState,
2539    emitter: &ExecutionEventEmitter,
2540    clock: &AtomicTime,
2541    client_order_id: ClientOrderId,
2542) {
2543    let (CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason)) = failure else {
2544        return;
2545    };
2546    let Some(ident) = state.order_identity(client_order_id) else {
2547        return;
2548    };
2549
2550    state.remove_order_tracking(client_order_id);
2551    emitter.emit_order_rejected_event(
2552        ident.strategy_id,
2553        ident.instrument_id,
2554        client_order_id,
2555        reason,
2556        clock.get_time_ns(),
2557        false,
2558    );
2559}
2560
2561fn emit_send_failed_modify(
2562    failure: &CommandFailure,
2563    state: &WsDispatchState,
2564    emitter: &ExecutionEventEmitter,
2565    clock: &AtomicTime,
2566    client_order_id: ClientOrderId,
2567) {
2568    let (CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason)) = failure else {
2569        return;
2570    };
2571    let Some(ident) = state.order_identity(client_order_id) else {
2572        return;
2573    };
2574
2575    emitter.emit_order_modify_rejected_event(
2576        ident.strategy_id,
2577        ident.instrument_id,
2578        client_order_id,
2579        None,
2580        reason,
2581        clock.get_time_ns(),
2582    );
2583}
2584
2585pub(crate) fn emit_send_failed_cancel(
2586    failure: &CommandFailure,
2587    state: &WsDispatchState,
2588    emitter: &ExecutionEventEmitter,
2589    clock: &AtomicTime,
2590    client_order_id: ClientOrderId,
2591    venue_order_id: Option<VenueOrderId>,
2592) {
2593    let (CommandFailure::NotSent(reason) | CommandFailure::VenueRejected(reason)) = failure else {
2594        return;
2595    };
2596
2597    let Some(ident) = state.order_identity(client_order_id) else {
2598        return;
2599    };
2600
2601    let venue_order_id = state
2602        .order_venue_binding(client_order_id)
2603        .map(|(venue_order_id, _)| venue_order_id)
2604        .or(venue_order_id);
2605
2606    emitter.emit_order_cancel_rejected_event(
2607        ident.strategy_id,
2608        ident.instrument_id,
2609        client_order_id,
2610        venue_order_id,
2611        reason,
2612        clock.get_time_ns(),
2613    );
2614}
2615
2616#[derive(Debug, Clone)]
2617pub struct AlgoCancelContext {
2618    pub client_order_id: ClientOrderId,
2619    pub instrument_id: InstrumentId,
2620    pub strategy_id: StrategyId,
2621    pub venue_order_id: Option<VenueOrderId>,
2622}
2623
2624// Contexts must correspond 1:1 with the requests that produced
2625// the responses (OKX preserves request order in batch responses).
2626pub fn emit_algo_cancel_rejections(
2627    responses: &[OKXCancelAlgoOrderResponse],
2628    contexts: &[AlgoCancelContext],
2629    emitter: &ExecutionEventEmitter,
2630    clock: &'static AtomicTime,
2631) {
2632    for (i, item) in responses.iter().enumerate() {
2633        let code = item.s_code.as_deref().unwrap_or(OKX_SUCCESS_CODE);
2634        if code == OKX_SUCCESS_CODE {
2635            continue;
2636        }
2637
2638        let msg = item.s_msg.as_deref().unwrap_or("");
2639
2640        if matches!(
2641            classify_okx_venue_code(code, msg),
2642            CommandFailure::Ambiguous(_) | CommandFailure::NotSent(_)
2643        ) {
2644            if let Some(ctx) = contexts.get(i) {
2645                log::warn!(
2646                    "Ambiguous algo cancel response for {}, awaiting reconciliation: \
2647                     algo_id={} sCode={code} sMsg={msg}",
2648                    ctx.client_order_id,
2649                    item.algo_id
2650                );
2651            } else {
2652                log::warn!(
2653                    "Ambiguous algo cancel response without context at index {i}: \
2654                     algo_id={} sCode={code} sMsg={msg}",
2655                    item.algo_id
2656                );
2657            }
2658            continue;
2659        }
2660
2661        if let Some(ctx) = contexts.get(i) {
2662            let reason = format_order_response_reason(code, msg, "");
2663            let ts = clock.get_time_ns();
2664            emitter.emit_order_cancel_rejected_event(
2665                ctx.strategy_id,
2666                ctx.instrument_id,
2667                ctx.client_order_id,
2668                ctx.venue_order_id,
2669                &reason,
2670                ts,
2671            );
2672        } else {
2673            log::warn!(
2674                "Algo cancel rejected but no context at index {i}: \
2675                 algo_id={} sCode={code} sMsg={msg}",
2676                item.algo_id
2677            );
2678        }
2679    }
2680}
2681
2682pub fn emit_batch_cancel_failure(
2683    contexts: &[AlgoCancelContext],
2684    error: &str,
2685    _emitter: &ExecutionEventEmitter,
2686    _clock: &'static AtomicTime,
2687) {
2688    for ctx in contexts {
2689        log::warn!(
2690            "Ambiguous algo batch cancel failure for {}, awaiting reconciliation: {error}",
2691            ctx.client_order_id
2692        );
2693    }
2694}
2695
2696pub(crate) fn format_order_response_reason(s_code: &str, s_msg: &str, sub_code: &str) -> String {
2697    match (s_code.is_empty(), s_msg.is_empty(), sub_code.is_empty()) {
2698        (false, false, true) => format!("OKX error {s_code}: {s_msg}"),
2699        (false, false, false) => format!("OKX error {s_code}: {s_msg} (subCode={sub_code})"),
2700        (false, true, true) => format!("OKX error {s_code}"),
2701        (false, true, false) => format!("OKX error {s_code} (subCode={sub_code})"),
2702        (true, false, true) => s_msg.to_string(),
2703        (true, false, false) => format!("{s_msg} (subCode={sub_code})"),
2704        (true, true, false) => format!("subCode={sub_code}"),
2705        (true, true, true) => String::new(),
2706    }
2707}
2708
2709#[cfg(test)]
2710mod tests {
2711    use std::time::Duration;
2712
2713    use nautilus_common::messages::{ExecutionEvent, ExecutionReport as CommonExecutionReport};
2714    use nautilus_core::time::get_atomic_clock_realtime;
2715    use nautilus_model::{
2716        enums::{AccountType, OrderSide, OrderType, TimeInForce, TriggerType},
2717        identifiers::Symbol,
2718        instruments::{CryptoPerpetual, CurrencyPair},
2719        orders::{Order, OrderTestBuilder, stubs::TestOrderEventStubs},
2720        types::{Price, Quantity},
2721    };
2722    use nautilus_network::error::SendError;
2723    use rstest::rstest;
2724
2725    use super::*;
2726    use crate::{
2727        common::{
2728            enums::{OKXInstrumentType, OKXSide, OKXTargetCurrency},
2729            parse::parse_spread_instrument,
2730        },
2731        websocket::{error::OKXWsError, messages::OKXWsFrame},
2732    };
2733
2734    fn load_algo_order_messages(fixture: &str) -> Vec<OKXAlgoOrderMsg> {
2735        let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
2736            .join("test_data")
2737            .join(fixture);
2738        let content = std::fs::read_to_string(path).unwrap();
2739        let frame: OKXWsFrame = serde_json::from_str(&content).unwrap();
2740        let OKXWsFrame::Data { data, .. } = frame else {
2741            panic!("Expected algo order data frame");
2742        };
2743        serde_json::from_value(data).unwrap()
2744    }
2745
2746    fn load_regular_order_messages(fixture: &str) -> Vec<OKXOrderMsg> {
2747        let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
2748            .join("test_data")
2749            .join(fixture);
2750        let content = std::fs::read_to_string(path).unwrap();
2751        let frame: OKXWsFrame = serde_json::from_str(&content).unwrap();
2752        let OKXWsFrame::Data { data, .. } = frame else {
2753            panic!("Expected regular order data frame");
2754        };
2755        serde_json::from_value(data).unwrap()
2756    }
2757
2758    fn test_algo_context(client_order_id: ClientOrderId) -> OrderContext {
2759        OrderContext {
2760            identity: OrderIdentity {
2761                client_order_id,
2762                strategy_id: StrategyId::from("STRATEGY-001"),
2763                instrument_id: InstrumentId::from("BTC-USDT-SWAP.OKX"),
2764                order_side: OrderSide::Sell,
2765                order_type: OrderType::StopLimit,
2766            },
2767            quantity: Quantity::from("0.01"),
2768            price: Some(Price::from("102900")),
2769            trigger_price: Some(Price::from("95000")),
2770            trigger_type: Some(TriggerType::LastPrice),
2771            time_in_force: TimeInForce::Gtc,
2772            is_post_only: false,
2773            is_reduce_only: true,
2774            is_quote_quantity: false,
2775        }
2776    }
2777
2778    fn test_algo_instruments() -> AtomicMap<Ustr, InstrumentAny> {
2779        let instrument = CryptoPerpetual::builder()
2780            .instrument_id(InstrumentId::from("BTC-USDT-SWAP.OKX"))
2781            .raw_symbol(Symbol::from("BTC-USDT-SWAP"))
2782            .base_currency(Currency::BTC())
2783            .quote_currency(Currency::USDT())
2784            .settlement_currency(Currency::USDT())
2785            .is_inverse(false)
2786            .price_precision(2)
2787            .size_precision(8)
2788            .price_increment(Price::from("0.01"))
2789            .size_increment(Quantity::from("0.00000001"))
2790            .ts_event(UnixNanos::default())
2791            .ts_init(UnixNanos::default())
2792            .build()
2793            .unwrap();
2794        let instruments = AtomicMap::new();
2795        instruments.insert(
2796            Ustr::from("BTC-USDT-SWAP"),
2797            InstrumentAny::CryptoPerpetual(instrument),
2798        );
2799        instruments
2800    }
2801
2802    fn test_execution_emitter() -> (
2803        ExecutionEventEmitter,
2804        tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2805    ) {
2806        let clock = get_atomic_clock_realtime();
2807        let mut emitter = ExecutionEventEmitter::new(
2808            clock,
2809            TraderId::from("TRADER-001"),
2810            AccountId::from("OKX-001"),
2811            AccountType::Margin,
2812            None,
2813        );
2814        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
2815        emitter.set_sender(sender);
2816        (emitter, receiver)
2817    }
2818
2819    fn dispatch_test_message(
2820        message: OKXWsMessage,
2821        emitter: &ExecutionEventEmitter,
2822        state: &WsDispatchState,
2823        instruments: &AtomicMap<Ustr, InstrumentAny>,
2824    ) {
2825        dispatch_ws_message(
2826            message,
2827            emitter,
2828            state,
2829            AccountId::from("OKX-001"),
2830            AccountType::Margin,
2831            instruments,
2832            &mut FeeCache::new(),
2833            &mut FilledQtyCache::new(),
2834            &mut AHashMap::new(),
2835            get_atomic_clock_realtime(),
2836        );
2837    }
2838
2839    fn drain_execution_events(
2840        receiver: &mut tokio::sync::mpsc::UnboundedReceiver<ExecutionEvent>,
2841    ) -> Vec<ExecutionEvent> {
2842        let mut events = Vec::new();
2843        while let Ok(event) = receiver.try_recv() {
2844            events.push(event);
2845        }
2846        events
2847    }
2848
2849    #[rstest]
2850    #[case::regular(false)]
2851    #[case::spread(true)]
2852    fn test_imported_order_unchanged_then_amended(#[case] spread: bool) {
2853        let cid = ClientOrderId::from("O-IMPORTED");
2854        let vid = VenueOrderId::from("7001");
2855
2856        let (instrument, mut payload) = imported_order_fixture(spread);
2857
2858        let instrument_id = instrument.id();
2859        let order = OrderTestBuilder::new(OrderType::Limit)
2860            .instrument_id(instrument_id)
2861            .client_order_id(cid)
2862            .side(OrderSide::Buy)
2863            .quantity(Quantity::from("0.01"))
2864            .price(Price::from("1.00"))
2865            .build();
2866        let state = WsDispatchState::default();
2867        state.register_external_order(OrderContext::from(&order), vid);
2868        let instruments = AtomicMap::new();
2869        instruments.insert(instrument_id.symbol.inner(), instrument);
2870        let (emitter, mut rx) = test_execution_emitter();
2871        let mut fee_cache = FeeCache::new();
2872        let mut fills = FilledQtyCache::new();
2873        let mut snapshots = AHashMap::new();
2874        let msg = &mut payload["data"][0];
2875        msg[if spread { "sprdId" } else { "instId" }] =
2876            serde_json::json!(instrument_id.symbol.as_str());
2877        msg["ordId"] = serde_json::json!("7001");
2878        msg["clOrdId"] = serde_json::json!("");
2879        msg["ordType"] = serde_json::json!("limit");
2880        msg["state"] = serde_json::json!("live");
2881        msg["sz"] = serde_json::json!("0.01");
2882        msg["px"] = serde_json::json!("1.00");
2883        msg["accFillSz"] = serde_json::json!("0");
2884        msg["fillSz"] = serde_json::json!("0");
2885        msg["tradeId"] = serde_json::json!("");
2886
2887        for amended in [false, true, true] {
2888            if amended {
2889                payload["data"][0]["sz"] = serde_json::json!("0.02");
2890                payload["data"][0]["px"] = serde_json::json!("1.10");
2891            }
2892
2893            let message = if spread {
2894                OKXWsMessage::SpreadOrders(serde_json::from_value(payload["data"].clone()).unwrap())
2895            } else {
2896                OKXWsMessage::Orders(serde_json::from_value(payload["data"].clone()).unwrap())
2897            };
2898
2899            dispatch_ws_message(
2900                message,
2901                &emitter,
2902                &state,
2903                emitter.account_id(),
2904                AccountType::Margin,
2905                &instruments,
2906                &mut fee_cache,
2907                &mut fills,
2908                &mut snapshots,
2909                get_atomic_clock_realtime(),
2910            );
2911
2912            if !amended {
2913                assert!(drain_execution_events(&mut rx).is_empty());
2914            }
2915        }
2916
2917        let events = drain_execution_events(&mut rx);
2918
2919        let [ExecutionEvent::Order(OrderEventAny::Updated(updated))] = events.as_slice() else {
2920            panic!("expected exactly one imported update, was {events:?}");
2921        };
2922
2923        assert_eq!(updated.client_order_id, cid);
2924        assert_eq!(updated.venue_order_id, Some(vid));
2925        assert_eq!(updated.quantity, Quantity::from("0.02"));
2926        assert_eq!(updated.price, Some(Price::from("1.10")));
2927        assert!(!state.contains_triggered(&cid));
2928        state.remove_order_tracking(cid);
2929        assert_eq!(state.order_context(cid), None);
2930        assert_eq!(state.order_venue_binding(cid), None);
2931        assert!(
2932            !state
2933                .lifecycle_bindings
2934                .lock()
2935                .client_by_parent
2936                .contains_key(&vid)
2937        );
2938    }
2939
2940    #[rstest]
2941    #[case::regular(false)]
2942    #[case::spread(true)]
2943    fn test_imported_order_reduced_to_filled_quantity_closes(#[case] spread: bool) {
2944        let (instrument, mut payload) = imported_order_fixture(spread);
2945        let instrument_id = instrument.id();
2946        let cid = ClientOrderId::from("O-IMPORTED-CLOSE");
2947        let vid = VenueOrderId::from("7001");
2948        let (emitter, mut rx) = test_execution_emitter();
2949        let mut order = OrderTestBuilder::new(OrderType::Limit)
2950            .instrument_id(instrument_id)
2951            .client_order_id(cid)
2952            .side(OrderSide::Buy)
2953            .quantity(Quantity::from("0.03"))
2954            .price(Price::from("1.00"))
2955            .build();
2956        let state = WsDispatchState::default();
2957        state.register_external_order(OrderContext::from(&order), vid);
2958        order
2959            .apply(TestOrderEventStubs::submitted(&order, emitter.account_id()))
2960            .unwrap();
2961        order
2962            .apply(TestOrderEventStubs::accepted(
2963                &order,
2964                emitter.account_id(),
2965                vid,
2966            ))
2967            .unwrap();
2968        let fill = TestOrderEventStubs::filled(
2969            &order,
2970            &instrument,
2971            Some(TradeId::from("7002")),
2972            None,
2973            Some(Price::from("1.00")),
2974            Some(Quantity::from("0.01")),
2975            None,
2976            None,
2977            None,
2978            Some(emitter.account_id()),
2979        );
2980        order.apply(fill).unwrap();
2981        let instruments = AtomicMap::new();
2982        instruments.insert(instrument_id.symbol.inner(), instrument);
2983        let mut fees = FeeCache::new();
2984        let mut fills = FilledQtyCache::new();
2985        fills.record(Ustr::from("7001"), Quantity::from("0.01"), false);
2986        let mut snapshots = AHashMap::new();
2987        let msg = &mut payload["data"][0];
2988        msg[if spread { "sprdId" } else { "instId" }] =
2989            serde_json::json!(instrument_id.symbol.as_str());
2990        msg["ordId"] = serde_json::json!("7001");
2991        msg["clOrdId"] = serde_json::json!("");
2992        msg["ordType"] = serde_json::json!("limit");
2993        msg["state"] = serde_json::json!("partially_filled");
2994        msg["sz"] = serde_json::json!("0.03");
2995        msg["px"] = serde_json::json!("1.00");
2996        msg["accFillSz"] = serde_json::json!("0.01");
2997        msg["fillSz"] = serde_json::json!("0");
2998        msg["fillPx"] = serde_json::json!("1.00");
2999        msg["tradeId"] = serde_json::json!("");
3000        msg["fee"] = serde_json::json!("0");
3001        msg["uTime"] = serde_json::json!("1781827200123");
3002
3003        for terminal in [false, true] {
3004            if terminal {
3005                let msg = &mut payload["data"][0];
3006                msg["state"] = serde_json::json!("filled");
3007                msg["sz"] = serde_json::json!("0.01");
3008                msg["fillSz"] = serde_json::json!("0");
3009                msg["tradeId"] = serde_json::json!("");
3010            }
3011
3012            let message = if spread {
3013                OKXWsMessage::SpreadOrders(serde_json::from_value(payload["data"].clone()).unwrap())
3014            } else {
3015                OKXWsMessage::Orders(serde_json::from_value(payload["data"].clone()).unwrap())
3016            };
3017
3018            dispatch_ws_message(
3019                message,
3020                &emitter,
3021                &state,
3022                emitter.account_id(),
3023                AccountType::Margin,
3024                &instruments,
3025                &mut fees,
3026                &mut fills,
3027                &mut snapshots,
3028                get_atomic_clock_realtime(),
3029            );
3030            let events = drain_execution_events(&mut rx);
3031
3032            if terminal {
3033                let [
3034                    ExecutionEvent::Order(OrderEventAny::Updated(updated)),
3035                    ExecutionEvent::Report(CommonExecutionReport::Order(report)),
3036                ] = events.as_slice()
3037                else {
3038                    panic!("expected terminal amendment and status report, was {events:?}");
3039                };
3040
3041                let expected = OrderUpdated::new(
3042                    emitter.trader_id(),
3043                    order.strategy_id(),
3044                    instrument_id,
3045                    cid,
3046                    Quantity::from("0.01"),
3047                    updated.event_id,
3048                    UnixNanos::from(1_781_827_200_123_000_000),
3049                    updated.ts_init,
3050                    true,
3051                    Some(vid),
3052                    Some(emitter.account_id()),
3053                    Some(Price::from("1.00")),
3054                    None,
3055                    None,
3056                    false,
3057                );
3058                assert_eq!(*updated, expected);
3059                assert_eq!(report.order_status, OrderStatus::Filled);
3060                assert_eq!(report.quantity, Quantity::from("0.01"));
3061                assert_eq!(report.filled_qty, Quantity::from("0.01"));
3062                order.apply(OrderEventAny::Updated(*updated)).unwrap();
3063            } else {
3064                let [ExecutionEvent::Report(CommonExecutionReport::Order(report))] =
3065                    events.as_slice()
3066                else {
3067                    panic!("expected unchanged partial status, was {events:?}");
3068                };
3069
3070                assert_eq!(report.order_status, OrderStatus::PartiallyFilled);
3071                assert_eq!(report.quantity, Quantity::from("0.03"));
3072                assert_eq!(report.filled_qty, Quantity::from("0.01"));
3073                assert_eq!(order.status(), OrderStatus::PartiallyFilled);
3074            }
3075        }
3076
3077        assert_eq!(order.status(), OrderStatus::Filled);
3078        assert_eq!(order.quantity(), Quantity::from("0.01"));
3079        assert_eq!(order.filled_qty(), Quantity::from("0.01"));
3080        assert_eq!(order.leaves_qty(), Quantity::from("0.00"));
3081        assert_eq!(order.price(), Some(Price::from("1.00")));
3082        assert_eq!(state.order_identity(cid), None);
3083        assert_eq!(state.order_venue_binding(cid), None);
3084        assert!(
3085            !state
3086                .lifecycle_bindings
3087                .lock()
3088                .client_by_parent
3089                .contains_key(&vid)
3090        );
3091    }
3092
3093    fn imported_order_fixture(spread: bool) -> (InstrumentAny, serde_json::Value) {
3094        if spread {
3095            let definitions: serde_json::Value =
3096                serde_json::from_str(include_str!("../../test_data/http_get_spreads.json"))
3097                    .unwrap();
3098            let definition = serde_json::from_value(definitions["data"][1].clone()).unwrap();
3099            let instrument =
3100                parse_spread_instrument(&definition, None, None, UnixNanos::default()).unwrap();
3101            let payload = serde_json::from_str::<serde_json::Value>(include_str!(
3102                "../../test_data/ws_sprd_orders.json"
3103            ))
3104            .unwrap();
3105            (instrument, payload)
3106        } else {
3107            let instruments = test_algo_instruments();
3108            let instrument = instruments
3109                .load()
3110                .get(&Ustr::from("BTC-USDT-SWAP"))
3111                .unwrap()
3112                .clone();
3113            let payload = serde_json::from_str::<serde_json::Value>(include_str!(
3114                "../../test_data/ws_orders.json"
3115            ))
3116            .unwrap();
3117            (instrument, payload)
3118        }
3119    }
3120
3121    #[rstest]
3122    #[case::single_fill(false, false)]
3123    #[case::partial_fill(true, false)]
3124    #[case::status_only(true, true)]
3125    fn short_final_fill_dispatches_terminal_status(
3126        #[values(false, true)] tracked: bool,
3127        #[values(OKXTargetCurrency::BaseCcy, OKXTargetCurrency::QuoteCcy)]
3128        target: OKXTargetCurrency,
3129        #[case] partial_first: bool,
3130        #[case] status_only: bool,
3131    ) {
3132        let mut msg = load_regular_order_messages("ws_orders.json").remove(0);
3133        msg.inst_id = Ustr::from("BTC-USDC");
3134        msg.inst_type = OKXInstrumentType::Spot;
3135        msg.ord_type = OKXOrderType::Market;
3136        msg.side = OKXSide::Buy;
3137        msg.tgt_ccy = Some(target);
3138        msg.sz = match target {
3139            OKXTargetCurrency::BaseCcy => "0.00296487",
3140            OKXTargetCurrency::QuoteCcy => "296.487",
3141        }
3142        .to_string();
3143
3144        msg.acc_fill_sz = Some("0.00296475".to_string());
3145
3146        let last_qty = if partial_first {
3147            "0.00196475"
3148        } else {
3149            "0.00296475"
3150        };
3151
3152        msg.fill_sz = last_qty.to_string();
3153        msg.fill_px = "100000".to_string();
3154        msg.avg_px = "100000".to_string();
3155        msg.px.clear();
3156        msg.state = OKXOrderStatus::Filled;
3157        let client_order_id = ClientOrderId::new(msg.cl_ord_id.as_str());
3158        let instrument_id = InstrumentId::from("BTC-USDC.OKX");
3159        let instrument = CurrencyPair::builder()
3160            .instrument_id(instrument_id)
3161            .raw_symbol(Symbol::from("BTC-USDC"))
3162            .base_currency(Currency::BTC())
3163            .quote_currency(Currency::USDC())
3164            .price_precision(2)
3165            .size_precision(8)
3166            .price_increment(Price::from("0.01"))
3167            .size_increment(Quantity::from("0.00000001"))
3168            .ts_event(UnixNanos::default())
3169            .ts_init(UnixNanos::default())
3170            .build()
3171            .unwrap();
3172        let instruments = AtomicMap::new();
3173        instruments.insert(msg.inst_id, InstrumentAny::CurrencyPair(instrument));
3174        let state = WsDispatchState::default();
3175
3176        if tracked {
3177            state.order_identities.insert(
3178                client_order_id,
3179                OrderIdentity {
3180                    client_order_id,
3181                    strategy_id: StrategyId::from("STRATEGY-001"),
3182                    instrument_id,
3183                    order_side: OrderSide::Buy,
3184                    order_type: OrderType::Market,
3185                },
3186            );
3187        }
3188
3189        let (emitter, mut receiver) = test_execution_emitter();
3190        let mut fee_cache = FeeCache::new();
3191        let mut filled_qty_cache = FilledQtyCache::new();
3192        let mut order_state_cache = AHashMap::new();
3193        let clock = get_atomic_clock_realtime();
3194
3195        if partial_first {
3196            let mut partial = msg.clone();
3197            partial.state = OKXOrderStatus::PartiallyFilled;
3198
3199            if status_only {
3200                partial.fill_sz = "0.00296475".to_string();
3201                msg.fill_sz = "0".to_string();
3202            } else {
3203                partial.acc_fill_sz = Some("0.001".to_string());
3204                partial.fill_sz = "0.001".to_string();
3205                partial.trade_id = "prior-trade".to_string();
3206            }
3207
3208            dispatch_ws_message(
3209                OKXWsMessage::Orders(vec![partial]),
3210                &emitter,
3211                &state,
3212                emitter.account_id(),
3213                AccountType::Cash,
3214                &instruments,
3215                &mut fee_cache,
3216                &mut filled_qty_cache,
3217                &mut order_state_cache,
3218                clock,
3219            );
3220            let events = drain_execution_events(&mut receiver);
3221            assert_eq!(events.len(), if tracked { 2 } else { 1 });
3222            assert!(!state.contains_terminal(&client_order_id));
3223        }
3224
3225        dispatch_ws_message(
3226            OKXWsMessage::Orders(vec![msg.clone()]),
3227            &emitter,
3228            &state,
3229            emitter.account_id(),
3230            AccountType::Cash,
3231            &instruments,
3232            &mut fee_cache,
3233            &mut filled_qty_cache,
3234            &mut order_state_cache,
3235            clock,
3236        );
3237
3238        let events = drain_execution_events(&mut receiver);
3239        let fill_index = usize::from(tracked && !partial_first);
3240        assert_eq!(events.len(), if status_only { 1 } else { fill_index + 2 });
3241
3242        if !status_only {
3243            match &events[fill_index] {
3244                ExecutionEvent::Order(OrderEventAny::Filled(fill)) => {
3245                    assert!(tracked);
3246                    assert_eq!(fill.last_qty, Quantity::from(last_qty));
3247                }
3248                ExecutionEvent::Report(CommonExecutionReport::Fill(fill)) => {
3249                    assert!(!tracked);
3250                    assert_eq!(fill.last_qty, Quantity::from(last_qty));
3251                }
3252                other => panic!("Expected fill before terminal status, was {other:?}"),
3253            }
3254        }
3255
3256        let report_index = if status_only { 0 } else { fill_index + 1 };
3257
3258        let ExecutionEvent::Report(CommonExecutionReport::Order(report)) = &events[report_index]
3259        else {
3260            panic!("Expected terminal order status report");
3261        };
3262
3263        assert_eq!(report.client_order_id, Some(client_order_id));
3264        assert_eq!(report.order_status, OrderStatus::Filled);
3265        assert_eq!(report.quantity, Quantity::from("0.00296475"));
3266        assert_eq!(report.filled_qty, Quantity::from("0.00296475"));
3267        assert!(state.contains_terminal(&client_order_id));
3268        assert!(!state.order_identities.contains_key(&client_order_id));
3269        assert!(!state.contains_accepted(&client_order_id));
3270        assert!(!order_state_cache.contains_key(&client_order_id));
3271
3272        dispatch_ws_message(
3273            OKXWsMessage::Orders(vec![msg]),
3274            &emitter,
3275            &state,
3276            emitter.account_id(),
3277            AccountType::Cash,
3278            &instruments,
3279            &mut fee_cache,
3280            &mut filled_qty_cache,
3281            &mut order_state_cache,
3282            clock,
3283        );
3284        assert!(drain_execution_events(&mut receiver).is_empty());
3285    }
3286
3287    #[rstest]
3288    #[case::margin(AccountType::Margin)]
3289    #[case::cash(AccountType::Cash)]
3290    fn account_updates_report_the_configured_account_type(#[case] account_type: AccountType) {
3291        let path = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR"))
3292            .join("test_data")
3293            .join("http_get_account_balance.json");
3294        let content = std::fs::read_to_string(path).unwrap();
3295        let payload: serde_json::Value = serde_json::from_str(&content).unwrap();
3296        let instruments = AtomicMap::new();
3297        let (emitter, mut receiver) = test_execution_emitter();
3298
3299        dispatch_ws_message(
3300            OKXWsMessage::Account(payload["data"].clone()),
3301            &emitter,
3302            &WsDispatchState::default(),
3303            AccountId::from("OKX-001"),
3304            account_type,
3305            &instruments,
3306            &mut FeeCache::new(),
3307            &mut FilledQtyCache::new(),
3308            &mut AHashMap::new(),
3309            get_atomic_clock_realtime(),
3310        );
3311
3312        let events = drain_execution_events(&mut receiver);
3313        assert_eq!(events.len(), 1);
3314
3315        let ExecutionEvent::Account(state) = &events[0] else {
3316            panic!("expected an account state event");
3317        };
3318
3319        assert_eq!(state.account_type, account_type);
3320        assert_eq!(state.account_id, AccountId::from("OKX-001"));
3321    }
3322
3323    #[rstest]
3324    fn tracked_algo_live_and_pause_emit_one_parent_acceptance() {
3325        let mut messages = load_algo_order_messages("ws_orders_algo.json");
3326        let live = messages.remove(0);
3327        let mut pause = live.clone();
3328        pause.state = OKXAlgoOrderStatus::Pause;
3329        pause.u_time += 1;
3330        let client_order_id = ClientOrderId::new(live.algo_cl_ord_id.as_str());
3331        let state = WsDispatchState::default();
3332        let mut context = test_algo_context(client_order_id);
3333        context.identity.order_type = OrderType::StopMarket;
3334        context.price = None;
3335        state.track_order_context(context);
3336        let instruments = test_algo_instruments();
3337        let (emitter, mut receiver) = test_execution_emitter();
3338
3339        dispatch_test_message(
3340            OKXWsMessage::AlgoOrders(vec![live, pause]),
3341            &emitter,
3342            &state,
3343            &instruments,
3344        );
3345
3346        let events = drain_execution_events(&mut receiver);
3347        assert_eq!(events.len(), 1);
3348        match &events[0] {
3349            ExecutionEvent::Order(OrderEventAny::Accepted(accepted)) => {
3350                assert_eq!(accepted.client_order_id, client_order_id);
3351                assert_eq!(
3352                    accepted.venue_order_id,
3353                    VenueOrderId::new("706620792746729472")
3354                );
3355            }
3356            other => panic!("Expected tracked algo acceptance, was {other:?}"),
3357        }
3358    }
3359
3360    #[rstest]
3361    #[case::after_acceptance(false)]
3362    #[case::first_acceptance(true)]
3363    fn tracked_algo_amend_emits_updated_terms_once(#[case] first_acceptance: bool) {
3364        let live = load_algo_order_messages("ws_orders_algo.json").remove(1);
3365        let client_order_id = ClientOrderId::new(live.algo_cl_ord_id.as_str());
3366        let state = WsDispatchState::default();
3367        let mut context = test_algo_context(client_order_id);
3368        context.identity.order_side = OrderSide::Buy;
3369        context.quantity = Quantity::from("0.02");
3370        context.price = Some(Price::from("106000.00"));
3371        context.trigger_price = Some(Price::from("105000.00"));
3372        context.trigger_type = Some(TriggerType::MarkPrice);
3373        state.track_order_context(context);
3374        let instruments = test_algo_instruments();
3375        let (emitter, mut receiver) = test_execution_emitter();
3376
3377        if !first_acceptance {
3378            dispatch_test_message(
3379                OKXWsMessage::AlgoOrders(vec![live.clone()]),
3380                &emitter,
3381                &state,
3382                &instruments,
3383            );
3384            let events = drain_execution_events(&mut receiver);
3385            assert_eq!(events.len(), 1);
3386            assert!(matches!(
3387                &events[0],
3388                ExecutionEvent::Order(OrderEventAny::Accepted(_))
3389            ));
3390        }
3391
3392        let mut amended = live.clone();
3393        amended.sz = "0.03".into();
3394        amended.trigger_px = "104000".into();
3395        amended.ord_px = "103000".into();
3396        amended.u_time += 2;
3397        dispatch_test_message(
3398            OKXWsMessage::AlgoOrders(vec![amended.clone(), amended, live]),
3399            &emitter,
3400            &state,
3401            &instruments,
3402        );
3403        let events = drain_execution_events(&mut receiver);
3404        assert_eq!(events.len(), if first_acceptance { 2 } else { 1 });
3405
3406        if first_acceptance {
3407            assert!(matches!(
3408                &events[0],
3409                ExecutionEvent::Order(OrderEventAny::Accepted(_))
3410            ));
3411        }
3412
3413        let ExecutionEvent::Order(OrderEventAny::Updated(updated)) = events.last().unwrap() else {
3414            panic!("expected OrderUpdated");
3415        };
3416
3417        assert_eq!(updated.client_order_id, client_order_id);
3418        assert_eq!(
3419            updated.venue_order_id,
3420            Some(VenueOrderId::from("706620792746729473"))
3421        );
3422        assert_eq!(updated.instrument_id, context.identity.instrument_id);
3423        assert_eq!(updated.strategy_id, context.identity.strategy_id);
3424        assert_eq!(updated.account_id, Some(AccountId::from("OKX-001")));
3425        assert_eq!(updated.quantity, Quantity::from("0.03"));
3426        assert_eq!(updated.price, Some(Price::from("103000.00")));
3427        assert_eq!(updated.trigger_price, Some(Price::from("104000.00")));
3428        let final_context = state.order_context(client_order_id).unwrap();
3429        assert_eq!(final_context.quantity, updated.quantity);
3430        assert_eq!(final_context.price, updated.price);
3431        assert_eq!(final_context.trigger_price, updated.trigger_price);
3432        assert_eq!(final_context.trigger_type, Some(TriggerType::MarkPrice));
3433    }
3434
3435    #[rstest]
3436    fn tracked_algo_trailing_acceptance_does_not_emit_update() {
3437        let message = load_algo_order_messages("ws_orders_algo_states.json").remove(2);
3438        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
3439        let state = WsDispatchState::default();
3440        let mut context = test_algo_context(client_order_id);
3441        context.identity.order_type = OrderType::TrailingStopMarket;
3442        context.price = None;
3443        state.track_order_context(context);
3444        let instruments = test_algo_instruments();
3445        let (emitter, mut receiver) = test_execution_emitter();
3446
3447        dispatch_test_message(
3448            OKXWsMessage::AlgoOrders(vec![message]),
3449            &emitter,
3450            &state,
3451            &instruments,
3452        );
3453
3454        let events = drain_execution_events(&mut receiver);
3455        assert_eq!(events.len(), 1);
3456
3457        let ExecutionEvent::Order(OrderEventAny::Accepted(accepted)) = &events[0] else {
3458            panic!("expected OrderAccepted");
3459        };
3460
3461        assert_eq!(accepted.client_order_id, client_order_id);
3462        assert_eq!(
3463            accepted.venue_order_id,
3464            VenueOrderId::from("706620792746729482")
3465        );
3466        assert_eq!(
3467            state.order_context(client_order_id).unwrap().trigger_price,
3468            None
3469        );
3470    }
3471
3472    #[rstest]
3473    #[case::live(OKXAlgoOrderStatus::Live)]
3474    #[case::canceled(OKXAlgoOrderStatus::Canceled)]
3475    #[case::failed(OKXAlgoOrderStatus::OrderFailed)]
3476    fn tracked_algo_malformed_terms_preserve_lifecycle(#[case] status: OKXAlgoOrderStatus) {
3477        let mut message = load_algo_order_messages("ws_orders_algo.json").remove(1);
3478        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
3479        let state = WsDispatchState::default();
3480        let context = test_algo_context(client_order_id);
3481        state.track_order_context(context);
3482        let instruments = test_algo_instruments();
3483        let (emitter, mut receiver) = test_execution_emitter();
3484        message.state = status;
3485        message.trigger_px = "invalid".into();
3486        message.fail_code = "51008".into();
3487
3488        dispatch_test_message(
3489            OKXWsMessage::AlgoOrders(vec![message]),
3490            &emitter,
3491            &state,
3492            &instruments,
3493        );
3494
3495        let events = drain_execution_events(&mut receiver);
3496        assert_eq!(
3497            events.len(),
3498            if status == OKXAlgoOrderStatus::Canceled {
3499                2
3500            } else {
3501                1
3502            }
3503        );
3504
3505        match status {
3506            OKXAlgoOrderStatus::Live => {
3507                assert!(
3508                    matches!(&events[0], ExecutionEvent::Order(OrderEventAny::Accepted(event))
3509                    if event.client_order_id == client_order_id
3510                        && event.venue_order_id == VenueOrderId::from("706620792746729473"))
3511                );
3512                let retained = state.order_context(client_order_id).unwrap();
3513                assert_eq!(retained.quantity, context.quantity);
3514                assert_eq!(retained.price, context.price);
3515                assert_eq!(retained.trigger_price, context.trigger_price);
3516                assert_eq!(retained.trigger_type, context.trigger_type);
3517                assert!(!state.contains_terminal(&client_order_id));
3518            }
3519            OKXAlgoOrderStatus::Canceled => {
3520                assert!(matches!(
3521                    &events[0],
3522                    ExecutionEvent::Order(OrderEventAny::Accepted(_))
3523                ));
3524                assert!(
3525                    matches!(&events[1], ExecutionEvent::Order(OrderEventAny::Canceled(event))
3526                    if event.client_order_id == client_order_id
3527                        && event.venue_order_id == Some(VenueOrderId::from("706620792746729473")))
3528                );
3529                assert!(state.contains_terminal(&client_order_id));
3530                assert!(state.order_context(client_order_id).is_none());
3531            }
3532            OKXAlgoOrderStatus::OrderFailed => {
3533                assert!(
3534                    matches!(&events[0], ExecutionEvent::Order(OrderEventAny::Rejected(event))
3535                    if event.client_order_id == client_order_id
3536                        && event.reason == Ustr::from("OKX error 51008"))
3537                );
3538                assert!(state.contains_terminal(&client_order_id));
3539                assert!(state.order_context(client_order_id).is_none());
3540            }
3541            _ => unreachable!(),
3542        }
3543    }
3544
3545    #[rstest]
3546    fn algo_update_routes_are_explicit_for_external_tracked_and_suppressed() {
3547        let mut message = load_algo_order_messages("ws_orders_algo.json").remove(0);
3548        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
3549        let state = WsDispatchState::default();
3550
3551        assert_eq!(
3552            route_algo_order_message(&message, &state),
3553            ExecutionUpdateRoute::External
3554        );
3555
3556        let context = test_algo_context(client_order_id);
3557        state.track_order_context(context);
3558        assert_eq!(
3559            route_algo_order_message(&message, &state),
3560            ExecutionUpdateRoute::Tracked(client_order_id, context)
3561        );
3562
3563        message.state = OKXAlgoOrderStatus::Unknown;
3564        assert_eq!(
3565            route_algo_order_message(&message, &state),
3566            ExecutionUpdateRoute::Suppressed
3567        );
3568    }
3569
3570    #[rstest]
3571    #[case::client_id(true)]
3572    #[case::venue_id(false)]
3573    fn imported_limit_context_does_not_route_as_algo(#[case] with_client_id: bool) {
3574        let mut message = load_algo_order_messages("ws_orders_algo.json").remove(0);
3575        let cid = ClientOrderId::new(message.algo_cl_ord_id.as_str());
3576        let vid = VenueOrderId::new(message.algo_id.as_str());
3577        let mut context = test_algo_context(cid);
3578        context.identity.order_type = OrderType::Limit;
3579        context.trigger_price = None;
3580        context.trigger_type = None;
3581        let state = WsDispatchState::default();
3582        state.register_external_order(context, vid);
3583
3584        if !with_client_id {
3585            message.algo_cl_ord_id.clear();
3586            message.cl_ord_id.clear();
3587        }
3588
3589        let route = route_algo_order_message(&message, &state);
3590
3591        assert_eq!(route, ExecutionUpdateRoute::Suppressed);
3592        assert_eq!(state.order_context(cid), Some(context));
3593        assert_eq!(state.order_venue_binding(cid), Some((vid, false)));
3594        assert!(!state.contains_triggered(&cid));
3595    }
3596
3597    #[rstest]
3598    #[case::limit(OrderType::Limit, false)]
3599    #[case::conditional(OrderType::StopLimit, true)]
3600    fn linked_child_waits_only_for_conditional_context(
3601        #[case] order_type: OrderType,
3602        #[case] held: bool,
3603    ) {
3604        let child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
3605        let parent = VenueOrderId::from("7002");
3606        let cid = ClientOrderId::from("O-PENDING-BINDING");
3607        let mut context = test_algo_context(cid);
3608        context.identity.order_type = order_type;
3609        if order_type == OrderType::Limit {
3610            context.trigger_price = None;
3611            context.trigger_type = None;
3612        }
3613
3614        let state = WsDispatchState::default();
3615        state.track_order_context(context);
3616
3617        let resolution =
3618            state.resolve_or_hold_linked_child(parent, context.identity.instrument_id, &child);
3619
3620        assert_eq!(
3621            resolution,
3622            if held {
3623                LinkedChildResolution::Held
3624            } else {
3625                LinkedChildResolution::External
3626            },
3627        );
3628        assert_eq!(
3629            state.pending_linked_children.lock().len(),
3630            usize::from(held)
3631        );
3632        assert_eq!(state.order_context(cid), Some(context));
3633        assert_eq!(state.order_venue_binding(cid), None);
3634    }
3635
3636    #[rstest]
3637    fn rest_parent_binding_routes_linked_child_before_parent_stream_update() {
3638        let mut child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
3639        let client_order_id = ClientOrderId::new("STOP003BTCUSDT20250120");
3640        let parent_venue_order_id = VenueOrderId::new("706620792746729474");
3641        let child_venue_order_id = VenueOrderId::new(child.ord_id);
3642        child.algo_cl_ord_id = None;
3643        let state = WsDispatchState::default();
3644        state.track_order_context(test_algo_context(client_order_id));
3645        state.bind_order(client_order_id, parent_venue_order_id);
3646        let instruments = test_algo_instruments();
3647        let (emitter, mut receiver) = test_execution_emitter();
3648
3649        dispatch_test_message(
3650            OKXWsMessage::Orders(vec![child.clone()]),
3651            &emitter,
3652            &state,
3653            &instruments,
3654        );
3655
3656        let events = drain_execution_events(&mut receiver);
3657        assert_eq!(events.len(), 4);
3658        assert!(matches!(
3659            &events[0],
3660            ExecutionEvent::Order(OrderEventAny::Accepted(accepted))
3661                if accepted.client_order_id == client_order_id
3662                    && accepted.venue_order_id == parent_venue_order_id
3663        ));
3664        assert!(matches!(
3665            &events[1],
3666            ExecutionEvent::Order(OrderEventAny::Updated(updated))
3667                if updated.client_order_id == client_order_id
3668                    && updated.venue_order_id == Some(child_venue_order_id)
3669        ));
3670        assert!(matches!(
3671            &events[2],
3672            ExecutionEvent::Order(OrderEventAny::Triggered(triggered))
3673                if triggered.client_order_id == client_order_id
3674                    && triggered.venue_order_id == Some(child_venue_order_id)
3675        ));
3676        assert!(matches!(
3677            &events[3],
3678            ExecutionEvent::Order(OrderEventAny::Filled(filled))
3679                if filled.client_order_id == client_order_id
3680                    && filled.venue_order_id == child_venue_order_id
3681        ));
3682
3683        dispatch_test_message(
3684            OKXWsMessage::Orders(vec![child]),
3685            &emitter,
3686            &state,
3687            &instruments,
3688        );
3689        assert!(drain_execution_events(&mut receiver).is_empty());
3690    }
3691
3692    #[rstest]
3693    #[tokio::test]
3694    async fn pre_binding_child_is_held_until_parent_binding() {
3695        let child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
3696        let parent_venue_order_id = VenueOrderId::new(
3697            child
3698                .linked_algo_ord
3699                .as_ref()
3700                .expect("Expected linked algo order")
3701                .algo_id
3702                .as_str(),
3703        );
3704        let client_order_id = ClientOrderId::new("STOP003BTCUSDT20250120");
3705        let child_venue_order_id = VenueOrderId::new(child.ord_id);
3706        let state = WsDispatchState::default();
3707        state.track_order_context(test_algo_context(client_order_id));
3708        let instruments = test_algo_instruments();
3709        let (emitter, mut receiver) = test_execution_emitter();
3710
3711        dispatch_test_message(
3712            OKXWsMessage::Orders(vec![child.clone()]),
3713            &emitter,
3714            &state,
3715            &instruments,
3716        );
3717        assert!(drain_execution_events(&mut receiver).is_empty());
3718
3719        state.bind_order(client_order_id, parent_venue_order_id);
3720        tokio::time::timeout(
3721            Duration::from_millis(100),
3722            state.wait_for_linked_child_route(),
3723        )
3724        .await
3725        .expect("Expected parent binding to wake the private dispatch loop");
3726        dispatch_test_message(
3727            OKXWsMessage::Orders(Vec::new()),
3728            &emitter,
3729            &state,
3730            &instruments,
3731        );
3732
3733        let events = drain_execution_events(&mut receiver);
3734        assert_eq!(events.len(), 4);
3735        assert!(matches!(
3736            &events[0],
3737            ExecutionEvent::Order(OrderEventAny::Accepted(accepted))
3738                if accepted.client_order_id == client_order_id
3739                    && accepted.venue_order_id == parent_venue_order_id
3740        ));
3741        assert!(matches!(
3742            &events[1],
3743            ExecutionEvent::Order(OrderEventAny::Updated(updated))
3744                if updated.client_order_id == client_order_id
3745                    && updated.venue_order_id == Some(child_venue_order_id)
3746        ));
3747        assert!(matches!(
3748            &events[2],
3749            ExecutionEvent::Order(OrderEventAny::Triggered(triggered))
3750                if triggered.client_order_id == client_order_id
3751                    && triggered.venue_order_id == Some(child_venue_order_id)
3752        ));
3753        assert!(matches!(
3754            &events[3],
3755            ExecutionEvent::Order(OrderEventAny::Filled(filled))
3756                if filled.client_order_id == client_order_id
3757                    && filled.venue_order_id == child_venue_order_id
3758        ));
3759
3760        dispatch_test_message(OKXWsMessage::Reconnected, &emitter, &state, &instruments);
3761        dispatch_test_message(
3762            OKXWsMessage::Orders(vec![child]),
3763            &emitter,
3764            &state,
3765            &instruments,
3766        );
3767        assert!(drain_execution_events(&mut receiver).is_empty());
3768    }
3769
3770    #[rstest]
3771    fn held_child_becomes_external_when_candidate_binds_another_parent() {
3772        let child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
3773        let client_order_id = ClientOrderId::new("STOP003BTCUSDT20250120");
3774        let child_venue_order_id = VenueOrderId::new(child.ord_id);
3775        let state = WsDispatchState::default();
3776        state.track_order_context(test_algo_context(client_order_id));
3777        let instruments = test_algo_instruments();
3778        let (emitter, mut receiver) = test_execution_emitter();
3779
3780        dispatch_test_message(
3781            OKXWsMessage::Orders(vec![child]),
3782            &emitter,
3783            &state,
3784            &instruments,
3785        );
3786        assert!(drain_execution_events(&mut receiver).is_empty());
3787
3788        let authoritative_parent = VenueOrderId::new("706620792746729475");
3789        state.bind_order(client_order_id, authoritative_parent);
3790        dispatch_test_message(
3791            OKXWsMessage::Orders(Vec::new()),
3792            &emitter,
3793            &state,
3794            &instruments,
3795        );
3796
3797        let events = drain_execution_events(&mut receiver);
3798        assert_eq!(events.len(), 1);
3799        assert!(matches!(
3800            &events[0],
3801            ExecutionEvent::Report(CommonExecutionReport::Fill(report))
3802                if report.client_order_id
3803                    == Some(ClientOrderId::new("706620792746729474_0"))
3804                    && report.venue_order_id == child_venue_order_id
3805        ));
3806        assert_eq!(
3807            state.order_venue_binding(client_order_id),
3808            Some((authoritative_parent, false))
3809        );
3810    }
3811
3812    #[rstest]
3813    #[case::algo(false)]
3814    #[case::imported(true)]
3815    fn active_order_bindings_are_not_evicted_with_replay_state(#[case] imported: bool) {
3816        let state = WsDispatchState::default();
3817        let first_client_order_id = ClientOrderId::new("STOP-ACTIVE-00000");
3818        let first_venue_order_id = VenueOrderId::new("700000000000000000");
3819
3820        for index in 0..=DEDUP_CAPACITY {
3821            let client_order_id = ClientOrderId::new(format!("STOP-ACTIVE-{index:05}").as_str());
3822            let venue_order_id = VenueOrderId::new(
3823                format!("{}", 700_000_000_000_000_000_u64 + index as u64).as_str(),
3824            );
3825
3826            if imported {
3827                let mut context = test_algo_context(client_order_id);
3828                context.identity.order_type = OrderType::Limit;
3829                context.trigger_price = None;
3830                context.trigger_type = None;
3831                state.register_external_order(context, venue_order_id);
3832            } else {
3833                state.bind_order(client_order_id, venue_order_id);
3834            }
3835        }
3836
3837        assert_eq!(
3838            state.order_venue_binding(first_client_order_id),
3839            Some((first_venue_order_id, false))
3840        );
3841        assert_eq!(
3842            state
3843                .lifecycle_bindings
3844                .lock()
3845                .client_by_parent
3846                .get(&first_venue_order_id)
3847                .copied(),
3848            Some(first_client_order_id),
3849        );
3850    }
3851
3852    #[rstest]
3853    fn ambiguous_submit_failure_preserves_only_bound_context() {
3854        let client_order_id = ClientOrderId::new("STOP-AMBIGUOUS-001");
3855        let failure = CommandFailure::Ambiguous("request timed out".to_string());
3856        let unbound_state = WsDispatchState::default();
3857        unbound_state.track_order_context(test_algo_context(client_order_id));
3858
3859        unbound_state.resolve_algo_submit_failure(client_order_id, &failure);
3860        assert_eq!(unbound_state.order_identity(client_order_id), None);
3861
3862        let bound_state = WsDispatchState::default();
3863        let context = test_algo_context(client_order_id);
3864        let parent_venue_order_id = VenueOrderId::new("706620792746729476");
3865        bound_state.track_order_context(context);
3866        bound_state.bind_order(client_order_id, parent_venue_order_id);
3867
3868        bound_state.resolve_algo_submit_failure(client_order_id, &failure);
3869        assert_eq!(
3870            bound_state.order_identity(client_order_id),
3871            Some(context.identity)
3872        );
3873        assert_eq!(
3874            bound_state.order_venue_binding(client_order_id),
3875            Some((parent_venue_order_id, false))
3876        );
3877    }
3878
3879    #[rstest]
3880    #[case::effective(0, false)]
3881    #[case::effective_order_id_list(0, true)]
3882    #[case::partially_effective(1, false)]
3883    #[case::order_placed(2, false)]
3884    fn tracked_algo_trigger_states_bind_child_before_trigger(
3885        #[case] index: usize,
3886        #[case] use_order_id_list: bool,
3887    ) {
3888        let mut messages = if index == 2 {
3889            load_algo_order_messages("ws_orders_algo.json")
3890        } else {
3891            load_algo_order_messages("ws_orders_algo_states.json")
3892        };
3893        let mut message = if index == 2 {
3894            messages.remove(2)
3895        } else {
3896            messages.remove(index)
3897        };
3898        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
3899        let child_venue_order_id = VenueOrderId::new(message.ord_id.as_str());
3900        if use_order_id_list {
3901            message.ord_id_list = vec![message.ord_id.clone()];
3902            message.ord_id.clear();
3903        }
3904
3905        let state = WsDispatchState::default();
3906        state.track_order_context(test_algo_context(client_order_id));
3907        let instruments = test_algo_instruments();
3908        let (emitter, mut receiver) = test_execution_emitter();
3909
3910        dispatch_test_message(
3911            OKXWsMessage::AlgoOrders(vec![message]),
3912            &emitter,
3913            &state,
3914            &instruments,
3915        );
3916
3917        let events = drain_execution_events(&mut receiver);
3918        assert_eq!(events.len(), 3);
3919        assert!(matches!(
3920            &events[0],
3921            ExecutionEvent::Order(OrderEventAny::Accepted(_))
3922        ));
3923        assert!(matches!(
3924            &events[1],
3925            ExecutionEvent::Order(OrderEventAny::Updated(updated))
3926                if updated.venue_order_id == Some(child_venue_order_id)
3927        ));
3928        assert!(matches!(
3929            &events[2],
3930            ExecutionEvent::Order(OrderEventAny::Triggered(triggered))
3931                if triggered.venue_order_id == Some(child_venue_order_id)
3932        ));
3933        assert_eq!(
3934            state.order_venue_binding(client_order_id),
3935            Some((child_venue_order_id, true))
3936        );
3937    }
3938
3939    #[rstest]
3940    fn tracked_algo_transition_uses_venue_actual_quantity() {
3941        let mut message = load_algo_order_messages("ws_orders_algo_states.json").remove(0);
3942        message.sz.clear();
3943        message.close_fraction = "1".to_string();
3944        message.actual_sz = "0.025".to_string();
3945        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
3946        let state = WsDispatchState::default();
3947        state.track_order_context(test_algo_context(client_order_id));
3948        let instruments = test_algo_instruments();
3949        let (emitter, mut receiver) = test_execution_emitter();
3950
3951        dispatch_test_message(
3952            OKXWsMessage::AlgoOrders(vec![message]),
3953            &emitter,
3954            &state,
3955            &instruments,
3956        );
3957
3958        let events = drain_execution_events(&mut receiver);
3959        assert_eq!(events.len(), 3);
3960        assert!(matches!(
3961            &events[1],
3962            ExecutionEvent::Order(OrderEventAny::Updated(updated))
3963                if updated.quantity == Quantity::from("0.025")
3964                    && updated.price.is_none()
3965        ));
3966        assert_eq!(
3967            state
3968                .order_contexts
3969                .get(&client_order_id)
3970                .map(|context| context.quantity),
3971            Some(Quantity::from("0.025"))
3972        );
3973    }
3974
3975    #[rstest]
3976    fn regular_child_refreshes_parent_transition_terms() {
3977        let mut parent = load_algo_order_messages("ws_orders_algo_states.json").remove(0);
3978        parent.ord_px = "94950".to_string();
3979        let parent_replay = parent.clone();
3980        let client_order_id = ClientOrderId::new(parent.algo_cl_ord_id.as_str());
3981        let parent_venue_order_id = parent.algo_id.clone();
3982        let child_venue_order_id = parent.ord_id.clone();
3983        let state = WsDispatchState::default();
3984        state.track_order_context(test_algo_context(client_order_id));
3985        let instruments = test_algo_instruments();
3986        let (emitter, mut receiver) = test_execution_emitter();
3987
3988        dispatch_test_message(
3989            OKXWsMessage::AlgoOrders(vec![parent]),
3990            &emitter,
3991            &state,
3992            &instruments,
3993        );
3994        assert_eq!(drain_execution_events(&mut receiver).len(), 3);
3995
3996        let mut child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
3997        child.algo_id = Some(parent_venue_order_id.clone());
3998        child.algo_cl_ord_id = None;
3999        child.linked_algo_ord = Some(crate::websocket::messages::OKXLinkedAlgoOrd {
4000            algo_id: parent_venue_order_id,
4001        });
4002        child.ord_id = Ustr::from(child_venue_order_id.as_str());
4003        child.ord_type = OKXOrderType::Limit;
4004        child.state = OKXOrderStatus::Live;
4005        child.sz = "0.025".to_string();
4006        child.px = "94950".to_string();
4007        child.acc_fill_sz = Some("0".to_string());
4008        child.fill_sz.clear();
4009        child.fill_px.clear();
4010        child.trade_id.clear();
4011
4012        dispatch_test_message(
4013            OKXWsMessage::Orders(vec![child]),
4014            &emitter,
4015            &state,
4016            &instruments,
4017        );
4018
4019        let events = drain_execution_events(&mut receiver);
4020        assert_eq!(events.len(), 1);
4021        assert!(matches!(
4022            &events[0],
4023            ExecutionEvent::Order(OrderEventAny::Updated(updated))
4024                if updated.quantity == Quantity::from("0.025")
4025                    && updated.price == Some(Price::from("94950"))
4026        ));
4027        assert_eq!(
4028            state
4029                .order_contexts
4030                .get(&client_order_id)
4031                .map(|context| (context.quantity, context.price)),
4032            Some((Quantity::from("0.025"), Some(Price::from("94950"))))
4033        );
4034
4035        dispatch_test_message(
4036            OKXWsMessage::AlgoOrders(vec![parent_replay]),
4037            &emitter,
4038            &state,
4039            &instruments,
4040        );
4041        assert!(drain_execution_events(&mut receiver).is_empty());
4042        assert_eq!(
4043            state
4044                .order_contexts
4045                .get(&client_order_id)
4046                .map(|context| context.quantity),
4047            Some(Quantity::from("0.025"))
4048        );
4049    }
4050
4051    #[rstest]
4052    fn tracked_algo_holds_trigger_until_linked_child_arrives() {
4053        let mut parent = load_algo_order_messages("ws_orders_algo_states.json").remove(0);
4054        parent.algo_id = "706620792746729474".to_string();
4055        parent.algo_cl_ord_id = "STOP003BTCUSDT20250120".to_string();
4056        parent.ord_id.clear();
4057        parent.ord_id_list.clear();
4058        let client_order_id = ClientOrderId::new(parent.algo_cl_ord_id.as_str());
4059        let state = WsDispatchState::default();
4060        state.track_order_context(test_algo_context(client_order_id));
4061        let instruments = test_algo_instruments();
4062        let (emitter, mut receiver) = test_execution_emitter();
4063
4064        dispatch_test_message(
4065            OKXWsMessage::AlgoOrders(vec![parent]),
4066            &emitter,
4067            &state,
4068            &instruments,
4069        );
4070
4071        let parent_events = drain_execution_events(&mut receiver);
4072        assert_eq!(parent_events.len(), 1);
4073        assert!(matches!(
4074            &parent_events[0],
4075            ExecutionEvent::Order(OrderEventAny::Accepted(_))
4076        ));
4077        assert!(!state.contains_triggered(&client_order_id));
4078
4079        let child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
4080        assert!(child.algo_cl_ord_id.is_none());
4081        assert_eq!(child.cl_ord_id, "706620792746729474_0");
4082        assert_eq!(
4083            child
4084                .linked_algo_ord
4085                .as_ref()
4086                .map(|linked| linked.algo_id.as_str()),
4087            Some("706620792746729474")
4088        );
4089        dispatch_test_message(
4090            OKXWsMessage::Orders(vec![child]),
4091            &emitter,
4092            &state,
4093            &instruments,
4094        );
4095
4096        let child_events = drain_execution_events(&mut receiver);
4097        assert_eq!(child_events.len(), 3);
4098        assert!(matches!(
4099            &child_events[0],
4100            ExecutionEvent::Order(OrderEventAny::Updated(_))
4101        ));
4102        assert!(matches!(
4103            &child_events[1],
4104            ExecutionEvent::Order(OrderEventAny::Triggered(_))
4105        ));
4106
4107        match &child_events[2] {
4108            ExecutionEvent::Order(OrderEventAny::Filled(filled)) => {
4109                assert_eq!(filled.client_order_id, client_order_id);
4110                assert_eq!(
4111                    filled.venue_order_id,
4112                    VenueOrderId::new("706620792746729999")
4113                );
4114                assert_eq!(filled.trade_id, TradeId::new("1518905530"));
4115            }
4116            other => panic!("Expected tracked child fill, was {other:?}"),
4117        }
4118
4119        assert!(state.contains_terminal(&client_order_id));
4120    }
4121
4122    #[rstest]
4123    fn tracked_algo_child_cancellation_routes_late_fill_report() {
4124        let parent = load_algo_order_messages("ws_orders_algo_states.json").remove(0);
4125        let client_order_id = ClientOrderId::new(parent.algo_cl_ord_id.as_str());
4126        let parent_venue_order_id = parent.algo_id.clone();
4127        let child_venue_order_id = parent.ord_id.clone();
4128        let state = WsDispatchState::default();
4129        state.track_order_context(test_algo_context(client_order_id));
4130        let instruments = test_algo_instruments();
4131        let (emitter, mut receiver) = test_execution_emitter();
4132
4133        dispatch_test_message(
4134            OKXWsMessage::AlgoOrders(vec![parent]),
4135            &emitter,
4136            &state,
4137            &instruments,
4138        );
4139        assert_eq!(drain_execution_events(&mut receiver).len(), 3);
4140
4141        let mut child = load_regular_order_messages("ws_orders_trigger.json").remove(0);
4142        child.algo_id = Some(parent_venue_order_id.clone());
4143        child.linked_algo_ord = Some(crate::websocket::messages::OKXLinkedAlgoOrd {
4144            algo_id: parent_venue_order_id,
4145        });
4146        child.ord_id = Ustr::from(child_venue_order_id.as_str());
4147        let mut canceled = child.clone();
4148        canceled.state = OKXOrderStatus::Canceled;
4149        canceled.acc_fill_sz = Some("0".to_string());
4150        canceled.fill_sz.clear();
4151        canceled.fill_px.clear();
4152        canceled.trade_id.clear();
4153        dispatch_test_message(
4154            OKXWsMessage::Orders(vec![canceled]),
4155            &emitter,
4156            &state,
4157            &instruments,
4158        );
4159
4160        let events = drain_execution_events(&mut receiver);
4161        assert_eq!(events.len(), 1);
4162        assert!(matches!(
4163            &events[0],
4164            ExecutionEvent::Order(OrderEventAny::Canceled(canceled))
4165                if canceled.client_order_id == client_order_id
4166                    && canceled.venue_order_id
4167                        == Some(VenueOrderId::new(child_venue_order_id.as_str()))
4168        ));
4169        assert!(state.contains_terminal(&client_order_id));
4170
4171        dispatch_test_message(
4172            OKXWsMessage::Orders(vec![child.clone()]),
4173            &emitter,
4174            &state,
4175            &instruments,
4176        );
4177        let events = drain_execution_events(&mut receiver);
4178        assert_eq!(events.len(), 1);
4179        assert!(matches!(
4180            &events[0],
4181            ExecutionEvent::Report(CommonExecutionReport::Fill(report))
4182                if report.client_order_id == Some(client_order_id)
4183                    && report.venue_order_id
4184                        == VenueOrderId::new(child_venue_order_id.as_str())
4185                    && report.trade_id == TradeId::new("1518905530")
4186        ));
4187
4188        dispatch_test_message(
4189            OKXWsMessage::Orders(vec![child]),
4190            &emitter,
4191            &state,
4192            &instruments,
4193        );
4194        assert!(drain_execution_events(&mut receiver).is_empty());
4195    }
4196
4197    #[rstest]
4198    fn reconnect_replay_and_stale_parent_acceptance_are_suppressed() {
4199        let message = load_algo_order_messages("ws_orders_algo_states.json").remove(0);
4200        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
4201        let state = WsDispatchState::default();
4202        state.track_order_context(test_algo_context(client_order_id));
4203        let instruments = test_algo_instruments();
4204        let (emitter, mut receiver) = test_execution_emitter();
4205
4206        dispatch_test_message(
4207            OKXWsMessage::AlgoOrders(vec![message.clone()]),
4208            &emitter,
4209            &state,
4210            &instruments,
4211        );
4212        assert_eq!(drain_execution_events(&mut receiver).len(), 3);
4213
4214        let mut stale_live = message.clone();
4215        stale_live.state = OKXAlgoOrderStatus::Live;
4216        stale_live.ord_id.clear();
4217        dispatch_test_message(OKXWsMessage::Reconnected, &emitter, &state, &instruments);
4218        dispatch_test_message(
4219            OKXWsMessage::AlgoOrders(vec![message, stale_live]),
4220            &emitter,
4221            &state,
4222            &instruments,
4223        );
4224
4225        assert!(drain_execution_events(&mut receiver).is_empty());
4226        assert_eq!(
4227            state
4228                .order_venue_binding(client_order_id)
4229                .map(|(venue_order_id, _)| venue_order_id),
4230            Some(VenueOrderId::new("706620792746730010"))
4231        );
4232    }
4233
4234    #[rstest]
4235    #[case::filled("ws_orders_algo.json", 4, 3)]
4236    #[case::partially_failed("ws_orders_algo_states.json", 4, 1)]
4237    fn tracked_algo_aggregate_state_never_enters_reconciliation(
4238        #[case] fixture: &str,
4239        #[case] index: usize,
4240        #[case] expected_order_events: usize,
4241    ) {
4242        let message = load_algo_order_messages(fixture).remove(index);
4243        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
4244        let state = WsDispatchState::default();
4245        state.track_order_context(test_algo_context(client_order_id));
4246        let instruments = test_algo_instruments();
4247        let (emitter, mut receiver) = test_execution_emitter();
4248
4249        dispatch_test_message(
4250            OKXWsMessage::AlgoOrders(vec![message]),
4251            &emitter,
4252            &state,
4253            &instruments,
4254        );
4255
4256        let events = drain_execution_events(&mut receiver);
4257        assert_eq!(events.len(), expected_order_events);
4258        assert!(
4259            events
4260                .iter()
4261                .all(|event| matches!(event, ExecutionEvent::Order(_)))
4262        );
4263        assert!(state.order_contexts.contains_key(&client_order_id));
4264        assert!(!state.contains_terminal(&client_order_id));
4265    }
4266
4267    #[rstest]
4268    fn tracked_algo_cancel_is_terminal_and_stale_live_is_suppressed() {
4269        let mut message = load_algo_order_messages("ws_orders_algo.json").remove(3);
4270        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
4271        let state = WsDispatchState::default();
4272        state.track_order_context(test_algo_context(client_order_id));
4273        let instruments = test_algo_instruments();
4274        let (emitter, mut receiver) = test_execution_emitter();
4275
4276        dispatch_test_message(
4277            OKXWsMessage::AlgoOrders(vec![message.clone()]),
4278            &emitter,
4279            &state,
4280            &instruments,
4281        );
4282        message.state = OKXAlgoOrderStatus::Live;
4283        message.algo_cl_ord_id.clear();
4284        message.cl_ord_id.clear();
4285        dispatch_test_message(
4286            OKXWsMessage::AlgoOrders(vec![message]),
4287            &emitter,
4288            &state,
4289            &instruments,
4290        );
4291
4292        let events = drain_execution_events(&mut receiver);
4293        assert_eq!(events.len(), 2);
4294        assert!(matches!(
4295            &events[0],
4296            ExecutionEvent::Order(OrderEventAny::Accepted(_))
4297        ));
4298        assert!(matches!(
4299            &events[1],
4300            ExecutionEvent::Order(OrderEventAny::Canceled(canceled))
4301                if canceled.client_order_id == client_order_id
4302        ));
4303        assert!(state.contains_terminal(&client_order_id));
4304        assert!(!state.order_contexts.contains_key(&client_order_id));
4305        assert_eq!(state.order_venue_binding(client_order_id), None);
4306    }
4307
4308    #[rstest]
4309    fn tracked_algo_failure_uses_typed_rejection() {
4310        let mut message = load_algo_order_messages("ws_orders_algo_states.json").remove(3);
4311        message.fail_code = "51008".to_string();
4312        let client_order_id = ClientOrderId::new(message.algo_cl_ord_id.as_str());
4313        let state = WsDispatchState::default();
4314        state.track_order_context(test_algo_context(client_order_id));
4315        let instruments = test_algo_instruments();
4316        let (emitter, mut receiver) = test_execution_emitter();
4317
4318        dispatch_test_message(
4319            OKXWsMessage::AlgoOrders(vec![message]),
4320            &emitter,
4321            &state,
4322            &instruments,
4323        );
4324
4325        let events = drain_execution_events(&mut receiver);
4326        assert_eq!(events.len(), 1);
4327        assert!(matches!(
4328            &events[0],
4329            ExecutionEvent::Order(OrderEventAny::Rejected(rejected))
4330                if rejected.client_order_id == client_order_id
4331                    && rejected.reason == Ustr::from("OKX error 51008")
4332        ));
4333        assert!(state.contains_terminal(&client_order_id));
4334    }
4335
4336    #[rstest]
4337    #[case("51000", "Rejected", "", "OKX error 51000: Rejected")]
4338    #[case(
4339        "51000",
4340        "Rejected",
4341        "51004",
4342        "OKX error 51000: Rejected (subCode=51004)"
4343    )]
4344    #[case("51000", "", "", "OKX error 51000")]
4345    #[case("51000", "", "51004", "OKX error 51000 (subCode=51004)")]
4346    #[case("", "Rejected", "", "Rejected")]
4347    #[case("", "Rejected", "51004", "Rejected (subCode=51004)")]
4348    #[case("", "", "51004", "subCode=51004")]
4349    #[case("", "", "", "")]
4350    fn test_format_order_response_reason(
4351        #[case] s_code: &str,
4352        #[case] s_msg: &str,
4353        #[case] sub_code: &str,
4354        #[case] expected: &str,
4355    ) {
4356        assert_eq!(
4357            format_order_response_reason(s_code, s_msg, sub_code),
4358            expected
4359        );
4360    }
4361
4362    #[rstest]
4363    #[case("order", false, false)]
4364    #[case("batch-orders", false, false)]
4365    #[case("batch-orders", false, true)]
4366    #[case("amend-order", true, false)]
4367    #[case("batch-amend-orders", true, false)]
4368    #[case("batch-amend-orders", true, true)]
4369    fn rpi_minimum_notional_rejection_preserves_order_lifecycle(
4370        #[case] case: &str,
4371        #[case] amend: bool,
4372        #[case] reverse: bool,
4373    ) {
4374        let fixtures: serde_json::Value =
4375            serde_json::from_str(include_str!("../../test_data/rpi_minimum_notional.json"))
4376                .unwrap();
4377        let mut response = fixtures[case]["response"].clone();
4378        if reverse {
4379            response["data"].as_array_mut().unwrap().reverse();
4380        }
4381        let frame: OKXWsFrame = serde_json::from_value(response.clone()).unwrap();
4382        let OKXWsFrame::OrderResponse {
4383            id,
4384            op,
4385            code,
4386            msg,
4387            data,
4388        } = frame
4389        else {
4390            panic!("Expected order response");
4391        };
4392        let state = WsDispatchState::default();
4393        let instrument_id = InstrumentId::from("BTC-USDT.OKX");
4394        let strategy_id = StrategyId::from("RPI-003");
4395        let pending = if amend {
4396            &state.pending_amends
4397        } else {
4398            &state.pending_orders
4399        };
4400        let unrelated = "ORPI003";
4401        pending.insert(
4402            unrelated.to_string(),
4403            PendingOrderInfo {
4404                trader_id: TraderId::from("TRADER-001"),
4405                strategy_id,
4406                instrument_id,
4407            },
4408        );
4409        let mut originals = Vec::new();
4410        for item in &data {
4411            let client_order_id = ClientOrderId::new(item["clOrdId"].as_str().unwrap());
4412            let context = OrderContext {
4413                identity: OrderIdentity {
4414                    client_order_id,
4415                    strategy_id,
4416                    instrument_id,
4417                    order_side: OrderSide::Sell,
4418                    order_type: OrderType::Limit,
4419                },
4420                quantity: Quantity::from("0.2"),
4421                price: Some(Price::from("65123")),
4422                trigger_price: None,
4423                trigger_type: None,
4424                time_in_force: TimeInForce::Gtc,
4425                is_post_only: true,
4426                is_reduce_only: false,
4427                is_quote_quantity: false,
4428            };
4429            state.track_order_context(context);
4430            state
4431                .order_identities
4432                .insert(client_order_id, context.identity);
4433            pending.insert(
4434                client_order_id.to_string(),
4435                PendingOrderInfo {
4436                    trader_id: TraderId::from("TRADER-001"),
4437                    strategy_id,
4438                    instrument_id,
4439                },
4440            );
4441            originals.push(context);
4442        }
4443        let (emitter, mut receiver) = test_execution_emitter();
4444
4445        dispatch_test_message(
4446            OKXWsMessage::OrderResponse {
4447                id,
4448                op,
4449                code,
4450                msg,
4451                data,
4452            },
4453            &emitter,
4454            &state,
4455            &AtomicMap::new(),
4456        );
4457
4458        let events = drain_execution_events(&mut receiver);
4459        assert_eq!(events.len(), 1);
4460        let rejected_id = ClientOrderId::from("ORPI002");
4461        let s_msg = response["data"]
4462            .as_array()
4463            .unwrap()
4464            .iter()
4465            .find(|item| item["sCode"] == "54051")
4466            .unwrap()["sMsg"]
4467            .as_str()
4468            .unwrap();
4469        let reason = format!("OKX error 54051: {s_msg}");
4470
4471        match &events[0] {
4472            ExecutionEvent::Order(OrderEventAny::Rejected(event)) if !amend => {
4473                assert_eq!(event.trader_id, TraderId::from("TRADER-001"));
4474                assert_eq!(event.strategy_id, strategy_id);
4475                assert_eq!(event.instrument_id, instrument_id);
4476                assert_eq!(event.client_order_id, rejected_id);
4477                assert_eq!(event.account_id, AccountId::from("OKX-001"));
4478                assert_eq!(event.reason.as_str(), reason);
4479                assert!(!event.reconciliation);
4480                assert!(!event.due_post_only);
4481            }
4482            ExecutionEvent::Order(OrderEventAny::ModifyRejected(event)) if amend => {
4483                assert_eq!(event.trader_id, TraderId::from("TRADER-001"));
4484                assert_eq!(event.strategy_id, strategy_id);
4485                assert_eq!(event.instrument_id, instrument_id);
4486                assert_eq!(event.client_order_id, rejected_id);
4487                assert_eq!(event.account_id, Some(AccountId::from("OKX-001")));
4488                assert_eq!(
4489                    event.venue_order_id,
4490                    Some(VenueOrderId::from("2500000000000000002"))
4491                );
4492                assert_eq!(event.reason.as_str(), reason);
4493                assert!(!event.reconciliation);
4494            }
4495            event => panic!("Unexpected event: {event:?}"),
4496        }
4497        assert_eq!(pending.len(), 1);
4498        assert!(pending.contains_key(unrelated));
4499
4500        for original in originals {
4501            let client_order_id = original.identity.client_order_id;
4502            if amend || client_order_id != rejected_id {
4503                assert_eq!(
4504                    *state.order_identities.get(&client_order_id).unwrap(),
4505                    original.identity
4506                );
4507                assert_eq!(
4508                    *state.order_contexts.get(&client_order_id).unwrap(),
4509                    original
4510                );
4511                assert!(!state.terminal_orders.contains(&client_order_id));
4512            } else {
4513                assert!(!state.order_identities.contains_key(&client_order_id));
4514                assert!(!state.order_contexts.contains_key(&client_order_id));
4515            }
4516        }
4517    }
4518
4519    #[rstest]
4520    fn cancel_send_failure_prefers_current_venue_binding() {
4521        let state = WsDispatchState::default();
4522        let client_order_id = ClientOrderId::from("O-rebound-cancel");
4523        let identity = test_algo_context(client_order_id).identity;
4524        state.order_identities.insert(client_order_id, identity);
4525        state.bind_order(client_order_id, VenueOrderId::from("current-venue"));
4526        let (emitter, mut receiver) = test_execution_emitter();
4527
4528        emit_send_failed_cancel(
4529            &CommandFailure::NotSent("handler closed".to_string()),
4530            &state,
4531            &emitter,
4532            get_atomic_clock_realtime(),
4533            client_order_id,
4534            Some(VenueOrderId::from("stale-venue")),
4535        );
4536        let events = drain_execution_events(&mut receiver);
4537
4538        assert_eq!(events.len(), 1);
4539
4540        let ExecutionEvent::Order(OrderEventAny::CancelRejected(event)) = &events[0] else {
4541            panic!("Expected cancel rejection, received {:?}", events[0]);
4542        };
4543
4544        assert_eq!(event.client_order_id, client_order_id);
4545        assert_eq!(
4546            event.venue_order_id,
4547            Some(VenueOrderId::from("current-venue"))
4548        );
4549        assert_eq!(event.instrument_id, identity.instrument_id);
4550        assert_eq!(event.strategy_id, identity.strategy_id);
4551        assert_eq!(event.reason.as_str(), "handler closed");
4552    }
4553
4554    #[rstest]
4555    #[case::no_client(OKXWsError::NoActiveClient, false)]
4556    #[case::connection_changed(OKXWsError::TransportSend(SendError::ConnectionChanged), false)]
4557    #[case::timeout(OKXWsError::TransportSend(SendError::Timeout), false)]
4558    #[case::write_timeout(OKXWsError::TransportSend(SendError::WriteTimeout), true)]
4559    #[case::broken_pipe(OKXWsError::TransportSend(SendError::BrokenPipe("reset".to_string())), true)]
4560    fn cancel_send_failure_resolves_only_definitely_unsent_orders(
4561        #[values(
4562            OKXWsOperation::CancelOrder,
4563            OKXWsOperation::BatchCancelOrders,
4564            OKXWsOperation::MassCancel,
4565            OKXWsOperation::CancelAlgos
4566        )]
4567        op: OKXWsOperation,
4568        #[case] error: OKXWsError,
4569        #[case] ambiguous: bool,
4570    ) {
4571        let state = WsDispatchState::default();
4572        let client_order_ids = [
4573            ClientOrderId::from("O-cancel-first"),
4574            ClientOrderId::from("O-cancel-second"),
4575        ];
4576        let instrument_id = InstrumentId::from("ETH-USDT-SWAP.OKX");
4577        let strategy_id = StrategyId::from("S-CANCEL-FAILURE");
4578        let reason = error.to_string();
4579
4580        for (index, client_order_id) in client_order_ids.into_iter().enumerate() {
4581            state.order_identities.insert(
4582                client_order_id,
4583                OrderIdentity {
4584                    client_order_id,
4585                    instrument_id,
4586                    strategy_id,
4587                    order_side: OrderSide::Sell,
4588                    order_type: OrderType::Limit,
4589                },
4590            );
4591
4592            state.bind_order(
4593                client_order_id,
4594                VenueOrderId::from(format!("cancel-venue-{index}")),
4595            );
4596            state.pending_cancels.insert(
4597                client_order_id.to_string(),
4598                PendingOrderInfo {
4599                    trader_id: TraderId::from("TRADER-001"),
4600                    strategy_id,
4601                    instrument_id,
4602                },
4603            );
4604        }
4605
4606        let clock = Box::leak(Box::new(AtomicTime::new(false, UnixNanos::from(123_456))));
4607
4608        let mut emitter = ExecutionEventEmitter::new(
4609            clock,
4610            TraderId::from("TRADER-001"),
4611            AccountId::from("OKX-001"),
4612            AccountType::Margin,
4613            None,
4614        );
4615        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
4616        emitter.set_sender(sender);
4617        dispatch_ws_message(
4618            OKXWsMessage::SendFailed {
4619                request_id: "cancel-failure".to_string(),
4620                client_order_ids: client_order_ids.to_vec(),
4621                op: Some(op),
4622                error,
4623            },
4624            &emitter,
4625            &state,
4626            AccountId::from("OKX-001"),
4627            AccountType::Margin,
4628            &AtomicMap::new(),
4629            &mut FeeCache::new(),
4630            &mut FilledQtyCache::new(),
4631            &mut AHashMap::new(),
4632            clock,
4633        );
4634
4635        let events = drain_execution_events(&mut receiver);
4636
4637        assert_eq!(events.len(), if ambiguous { 0 } else { 2 });
4638
4639        for (index, client_order_id) in client_order_ids.into_iter().enumerate() {
4640            assert_eq!(
4641                state.pending_cancels.contains_key(client_order_id.as_str()),
4642                ambiguous
4643            );
4644            assert_eq!(
4645                state.order_identity(client_order_id),
4646                Some(OrderIdentity {
4647                    client_order_id,
4648                    instrument_id,
4649                    strategy_id,
4650                    order_side: OrderSide::Sell,
4651                    order_type: OrderType::Limit,
4652                })
4653            );
4654            assert_eq!(
4655                state.order_venue_binding(client_order_id),
4656                Some((VenueOrderId::from(format!("cancel-venue-{index}")), false))
4657            );
4658
4659            if ambiguous {
4660                continue;
4661            }
4662
4663            let ExecutionEvent::Order(OrderEventAny::CancelRejected(event)) = &events[index] else {
4664                panic!("Expected cancel rejection, received {:?}", events[index]);
4665            };
4666
4667            assert_eq!(event.trader_id, TraderId::from("TRADER-001"));
4668            assert_eq!(event.strategy_id, strategy_id);
4669            assert_eq!(event.instrument_id, instrument_id);
4670            assert_eq!(event.client_order_id, client_order_id);
4671            assert_eq!(
4672                event.venue_order_id,
4673                Some(VenueOrderId::from(format!("cancel-venue-{index}")))
4674            );
4675            assert_eq!(event.account_id, Some(AccountId::from("OKX-001")));
4676            assert_eq!(event.reason.as_str(), reason);
4677            assert_eq!(event.ts_event, UnixNanos::from(123_456));
4678            assert_eq!(event.ts_init, UnixNanos::from(123_456));
4679            assert_eq!(event.causation_id, None);
4680            assert!(!event.reconciliation);
4681        }
4682    }
4683
4684    #[rstest]
4685    #[case::ambiguous(OKXWsError::SendFailed("connection reset".to_string()), true)]
4686    #[case::not_sent(OKXWsError::NoActiveClient, false)]
4687    fn send_failure_preserves_only_ambiguous_pending_orders(
4688        #[case] error: OKXWsError,
4689        #[case] expected_pending: bool,
4690    ) {
4691        let client_order_ids = [
4692            ClientOrderId::from("O-batch-pending-1"),
4693            ClientOrderId::from("O-batch-pending-2"),
4694        ];
4695        let state = WsDispatchState::default();
4696        let instrument_id = InstrumentId::from("ETH-USDT-SWAP.OKX");
4697        let strategy_id = StrategyId::from("STRATEGY-001");
4698
4699        for client_order_id in client_order_ids {
4700            state.pending_orders.insert(
4701                client_order_id.to_string(),
4702                PendingOrderInfo {
4703                    trader_id: TraderId::from("TRADER-001"),
4704                    strategy_id,
4705                    instrument_id,
4706                },
4707            );
4708            state.order_identities.insert(
4709                client_order_id,
4710                OrderIdentity {
4711                    client_order_id,
4712                    instrument_id,
4713                    strategy_id,
4714                    order_side: OrderSide::Buy,
4715                    order_type: OrderType::Limit,
4716                },
4717            );
4718        }
4719
4720        let clock = get_atomic_clock_realtime();
4721        let mut emitter = ExecutionEventEmitter::new(
4722            clock,
4723            TraderId::from("TRADER-001"),
4724            AccountId::from("OKX-001"),
4725            AccountType::Margin,
4726            None,
4727        );
4728        let (sender, _receiver) = tokio::sync::mpsc::unbounded_channel();
4729        emitter.set_sender(sender);
4730        let instruments = AtomicMap::new();
4731        let mut fee_cache = FeeCache::new();
4732        let mut filled_qty_cache = FilledQtyCache::new();
4733        let mut order_state_cache = AHashMap::new();
4734
4735        dispatch_ws_message(
4736            OKXWsMessage::SendFailed {
4737                request_id: "req-batch-send-failure".to_string(),
4738                client_order_ids: client_order_ids.to_vec(),
4739                op: Some(OKXWsOperation::BatchOrders),
4740                error,
4741            },
4742            &emitter,
4743            &state,
4744            AccountId::from("OKX-001"),
4745            AccountType::Margin,
4746            &instruments,
4747            &mut fee_cache,
4748            &mut filled_qty_cache,
4749            &mut order_state_cache,
4750            clock,
4751        );
4752
4753        for client_order_id in client_order_ids {
4754            assert_eq!(
4755                state.pending_orders.contains_key(client_order_id.as_str()),
4756                expected_pending,
4757                "pending state mismatch for {client_order_id}"
4758            );
4759        }
4760    }
4761}
nautilus_okx/websocket/dispatch.rs

nautilus_okx/websocket/
dispatch.rs
