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

nautilus_okx/websocket/
dispatch.rs
