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::{
24    collections::VecDeque,
25    hash::Hash,
26    sync::{Arc, Mutex},
27};
28
29use ahash::AHashMap;
30use dashmap::DashMap;
31use nautilus_core::{AtomicMap, MUTEX_POISONED, UUID4, UnixNanos, time::AtomicTime};
32use nautilus_live::ExecutionEventEmitter;
33use nautilus_model::{
34    enums::{OrderSide, OrderStatus, OrderType},
35    events::{OrderAccepted, OrderEventAny, OrderFilled, OrderRejected},
36    identifiers::{
37        AccountId, ClientOrderId, InstrumentId, StrategyId, TradeId, TraderId, VenueOrderId,
38    },
39    instruments::{Instrument, InstrumentAny},
40    orders::TRIGGERABLE_ORDER_TYPES,
41    reports::FillReport,
42    types::{Currency, Money, Quantity},
43};
44use ustr::Ustr;
45
46use crate::{
47    common::{
48        consts::{
49            OKX_FIELD_CLORDID, OKX_FIELD_SCODE, OKX_FIELD_SMSG, OKX_FIELD_SUBCODE,
50            OKX_POST_ONLY_CANCEL_REASON, OKX_POST_ONLY_CANCEL_SOURCE, OKX_SUCCESS_CODE,
51        },
52        enums::OKXOrderStatus,
53        parse::{
54            is_market_price, parse_client_order_id, parse_millisecond_timestamp, parse_price,
55            parse_quantity,
56        },
57    },
58    http::models::{OKXAccount, OKXCancelAlgoOrderResponse, OKXPosition, OKXSpreadOrder},
59    websocket::{
60        client::PendingOrderInfo,
61        enums::OKXWsOperation,
62        handler::is_post_only_auto_cancel,
63        messages::{ExecutionReport, OKXOrderMsg, OKXWsMessage},
64        parse::{
65            OrderStateSnapshot, ParsedOrderEvent, parse_algo_order_msg, parse_order_event,
66            parse_order_msg, parse_spread_order_event, parse_spread_order_msg,
67            update_fee_fill_caches,
68        },
69    },
70};
71
72/// Maximum entries held by the dedup sets before the oldest is evicted.
73const DEDUP_CAPACITY: usize = 10_000;
74
75/// Bounded deduplication set with FIFO eviction.
76///
77/// Insertions are tagged with a sequence number so a stale marker left
78/// behind by `remove` or by re-insertion of the same key cannot evict the
79/// live entry when it reaches the front of the queue.
80#[derive(Debug)]
81pub struct BoundedDedup<K> {
82    inner: Mutex<BoundedDedupInner<K>>,
83    capacity: usize,
84}
85
86#[derive(Debug)]
87struct BoundedDedupInner<K> {
88    set: AHashMap<K, u64>,
89    queue: VecDeque<(K, u64)>,
90    next_seq: u64,
91}
92
93impl<K> BoundedDedup<K>
94where
95    K: Eq + Hash + Clone,
96{
97    /// Creates a new dedup set with the given maximum capacity.
98    pub fn new(capacity: usize) -> Self {
99        Self {
100            inner: Mutex::new(BoundedDedupInner {
101                set: AHashMap::with_capacity(capacity),
102                queue: VecDeque::with_capacity(capacity),
103                next_seq: 0,
104            }),
105            capacity,
106        }
107    }
108
109    /// Returns `true` if the key is currently present.
110    #[allow(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
111    pub fn contains(&self, key: &K) -> bool {
112        self.inner
113            .lock()
114            .expect(MUTEX_POISONED)
115            .set
116            .contains_key(key)
117    }
118
119    /// Inserts the key. Returns `true` when the key was already present
120    /// (duplicate) and `false` when this call was the first insertion.
121    #[allow(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
122    pub fn check_and_insert(&self, key: K) -> bool {
123        let mut inner = self.inner.lock().expect(MUTEX_POISONED);
124        if inner.set.contains_key(&key) {
125            return true;
126        }
127        let seq = inner.next_seq;
128        inner.next_seq = inner.next_seq.wrapping_add(1);
129        inner.set.insert(key.clone(), seq);
130        inner.queue.push_back((key, seq));
131        while inner.queue.len() > self.capacity
132            && let Some((old_key, old_seq)) = inner.queue.pop_front()
133        {
134            // Skip if a fresher insertion has superseded this marker.
135            if inner.set.get(&old_key) == Some(&old_seq) {
136                inner.set.remove(&old_key);
137            }
138        }
139        false
140    }
141
142    /// Inserts the key without reporting whether it was already present.
143    pub fn insert(&self, key: K) {
144        let _ = self.check_and_insert(key);
145    }
146
147    /// Drops the key from the set if present. Returns `true` if it was found.
148    #[allow(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
149    pub fn remove(&self, key: &K) -> bool {
150        self.inner
151            .lock()
152            .expect(MUTEX_POISONED)
153            .set
154            .remove(key)
155            .is_some()
156    }
157}
158
159/// Order identity context stored at submission time, used by the WS dispatch
160/// task to produce proper order events without Cache access.
161///
162/// These fields are immutable for the lifetime of an order and are used to
163/// construct proper order events (OrderAccepted, OrderFilled, etc.) instead
164/// of execution reports.
165#[derive(Debug, Clone)]
166pub struct OrderIdentity {
167    pub instrument_id: InstrumentId,
168    pub strategy_id: StrategyId,
169    pub order_side: OrderSide,
170    pub order_type: OrderType,
171}
172
173/// Shared state for cross-stream event deduplication between the private
174/// and business WebSocket dispatch loops.
175#[derive(Debug)]
176pub struct WsDispatchState {
177    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
178    pub emitted_accepted: BoundedDedup<ClientOrderId>,
179    pub triggered_orders: BoundedDedup<ClientOrderId>,
180    pub filled_orders: BoundedDedup<ClientOrderId>,
181    pub terminal_orders: BoundedDedup<ClientOrderId>,
182    pub emitted_trades: BoundedDedup<TradeId>,
183    pub(crate) pending_orders: Arc<DashMap<String, PendingOrderInfo>>,
184    pub(crate) pending_cancels: Arc<DashMap<String, PendingOrderInfo>>,
185    pub(crate) pending_amends: Arc<DashMap<String, PendingOrderInfo>>,
186}
187
188impl Default for WsDispatchState {
189    fn default() -> Self {
190        Self {
191            order_identities: DashMap::new(),
192            emitted_accepted: BoundedDedup::new(DEDUP_CAPACITY),
193            triggered_orders: BoundedDedup::new(DEDUP_CAPACITY),
194            filled_orders: BoundedDedup::new(DEDUP_CAPACITY),
195            terminal_orders: BoundedDedup::new(DEDUP_CAPACITY),
196            emitted_trades: BoundedDedup::new(DEDUP_CAPACITY),
197            pending_orders: Arc::new(DashMap::new()),
198            pending_cancels: Arc::new(DashMap::new()),
199            pending_amends: Arc::new(DashMap::new()),
200        }
201    }
202}
203
204impl WsDispatchState {
205    // Creates a dispatch state sharing the pending operation maps
206    // with the WebSocket client that populates them
207    pub(crate) fn with_pending_maps(
208        pending_orders: Arc<DashMap<String, PendingOrderInfo>>,
209        pending_cancels: Arc<DashMap<String, PendingOrderInfo>>,
210        pending_amends: Arc<DashMap<String, PendingOrderInfo>>,
211    ) -> Self {
212        Self {
213            pending_orders,
214            pending_cancels,
215            pending_amends,
216            ..Default::default()
217        }
218    }
219}
220
221impl WsDispatchState {
222    pub(crate) fn insert_accepted(&self, cid: ClientOrderId) {
223        self.emitted_accepted.insert(cid);
224    }
225
226    pub(crate) fn insert_triggered(&self, cid: ClientOrderId) {
227        self.triggered_orders.insert(cid);
228    }
229
230    pub(crate) fn insert_filled(&self, cid: ClientOrderId) {
231        self.filled_orders.insert(cid);
232    }
233
234    pub(crate) fn insert_terminal(&self, cid: ClientOrderId) {
235        self.terminal_orders.insert(cid);
236    }
237
238    /// Returns `true` if this trade was already emitted (duplicate).
239    /// Uses atomic insert to avoid TOCTOU races between concurrent streams.
240    pub fn check_and_insert_trade(&self, trade_id: TradeId) -> bool {
241        self.emitted_trades.check_and_insert(trade_id)
242    }
243}
244
245/// Dispatches a WebSocket message with cross-stream deduplication.
246///
247/// For orders with a tracked identity (submitted through this client), produces
248/// proper order events (OrderAccepted, OrderCanceled, OrderFilled, etc.).
249/// For untracked orders (external or pre-existing), falls back to execution
250/// reports for downstream reconciliation.
251#[expect(clippy::too_many_arguments)]
252pub fn dispatch_ws_message(
253    message: OKXWsMessage,
254    emitter: &ExecutionEventEmitter,
255    state: &WsDispatchState,
256    account_id: AccountId,
257    instruments: &AtomicMap<Ustr, InstrumentAny>,
258    fee_cache: &mut AHashMap<Ustr, Money>,
259    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
260    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
261    clock: &AtomicTime,
262) {
263    let guard = instruments.load();
264    let instruments: &AHashMap<Ustr, InstrumentAny> = &guard;
265
266    match message {
267        OKXWsMessage::Orders(order_msgs) => {
268            let ts_init = clock.get_time_ns();
269            dispatch_order_messages(
270                &order_msgs,
271                emitter,
272                state,
273                account_id,
274                instruments,
275                fee_cache,
276                filled_qty_cache,
277                order_state_cache,
278                ts_init,
279            );
280        }
281        OKXWsMessage::SpreadOrders(order_msgs) => {
282            let ts_init = clock.get_time_ns();
283            dispatch_spread_order_messages(
284                &order_msgs,
285                emitter,
286                state,
287                account_id,
288                instruments,
289                filled_qty_cache,
290                order_state_cache,
291                ts_init,
292            );
293        }
294        OKXWsMessage::AlgoOrders(algo_msgs) => {
295            let ts_init = clock.get_time_ns();
296            let mut reports = Vec::new();
297
298            for msg in algo_msgs {
299                match parse_algo_order_msg(&msg, account_id, instruments, ts_init) {
300                    Ok(Some(report)) => reports.push(report),
301                    Ok(None) => {}
302                    Err(e) => log::error!("Failed to parse algo order message: {e}"),
303                }
304            }
305            dispatch_execution_reports(reports, emitter, state);
306        }
307        OKXWsMessage::Account(data) => {
308            let ts_init = clock.get_time_ns();
309
310            match serde_json::from_value::<Vec<OKXAccount>>(data) {
311                Ok(accounts) => {
312                    for account in &accounts {
313                        match crate::common::parse::parse_account_state(
314                            account, account_id, ts_init,
315                        ) {
316                            Ok(account_state) => emitter.send_account_state(account_state),
317                            Err(e) => log::error!("Failed to parse account state: {e}"),
318                        }
319                    }
320                }
321                Err(e) => log::error!("Failed to deserialize account data: {e}"),
322            }
323        }
324        OKXWsMessage::Positions(data) => {
325            let ts_init = clock.get_time_ns();
326
327            match serde_json::from_value::<Vec<OKXPosition>>(data) {
328                Ok(positions) => {
329                    for position in positions {
330                        let Some(instrument) = instruments.get(&position.inst_id) else {
331                            log::warn!("No cached instrument for position: {}", position.inst_id);
332                            continue;
333                        };
334                        let instrument_id = instrument.id();
335                        let size_precision = instrument.size_precision();
336
337                        match crate::common::parse::parse_position_status_report(
338                            &position,
339                            account_id,
340                            instrument_id,
341                            size_precision,
342                            ts_init,
343                        ) {
344                            Ok(report) => emitter.send_position_report(report),
345                            Err(e) => log::error!("Failed to parse position report: {e}"),
346                        }
347                    }
348                }
349                Err(e) => log::error!("Failed to deserialize positions data: {e}"),
350            }
351        }
352        OKXWsMessage::OrderResponse {
353            id,
354            op,
355            code,
356            msg,
357            data,
358        } => {
359            let ts_init = clock.get_time_ns();
360
361            for item in &data {
362                let s_code = item
363                    .get(OKX_FIELD_SCODE)
364                    .and_then(|v| v.as_str())
365                    .unwrap_or("");
366                let s_msg = item
367                    .get(OKX_FIELD_SMSG)
368                    .and_then(|v| v.as_str())
369                    .unwrap_or("");
370                let sub_code = item
371                    .get(OKX_FIELD_SUBCODE)
372                    .and_then(|v| v.as_str())
373                    .unwrap_or("");
374                let reason = format_order_response_reason(s_code, s_msg, sub_code);
375                let cl_ord_id = item
376                    .get(OKX_FIELD_CLORDID)
377                    .and_then(|v| v.as_str())
378                    .unwrap_or("");
379
380                if s_code == OKX_SUCCESS_CODE {
381                    log::debug!("Order response ok: op={op:?} cl_ord_id={cl_ord_id}");
382                    match op {
383                        OKXWsOperation::Order
384                        | OKXWsOperation::BatchOrders
385                        | OKXWsOperation::OrderAlgo => {
386                            state.pending_orders.remove(cl_ord_id);
387                        }
388                        OKXWsOperation::CancelOrder
389                        | OKXWsOperation::BatchCancelOrders
390                        | OKXWsOperation::MassCancel
391                        | OKXWsOperation::CancelAlgos => {
392                            state.pending_cancels.remove(cl_ord_id);
393                        }
394                        OKXWsOperation::AmendOrder | OKXWsOperation::BatchAmendOrders => {
395                            state.pending_amends.remove(cl_ord_id);
396                        }
397                        _ => {}
398                    }
399                    continue;
400                }
401
402                let Some(client_order_id) = parse_client_order_id(cl_ord_id) else {
403                    log::warn!(
404                        "Order response error without client_order_id: \
405                         op={op:?} s_code={s_code} s_msg={s_msg}"
406                    );
407                    continue;
408                };
409
410                let Some(ident) = state
411                    .order_identities
412                    .get(&client_order_id)
413                    .map(|entry| entry.clone())
414                else {
415                    log::warn!(
416                        "Order response error for untracked order: \
417                         op={op:?} cl_ord_id={cl_ord_id} s_code={s_code} s_msg={s_msg}"
418                    );
419                    continue;
420                };
421
422                let venue_order_id = item
423                    .get("ordId")
424                    .and_then(|v| v.as_str())
425                    .filter(|s| !s.is_empty())
426                    .map(VenueOrderId::new);
427
428                match op {
429                    OKXWsOperation::Order | OKXWsOperation::BatchOrders => {
430                        state.order_identities.remove(&client_order_id);
431                        state.pending_orders.remove(cl_ord_id);
432                        emitter.emit_order_rejected_event(
433                            ident.strategy_id,
434                            ident.instrument_id,
435                            client_order_id,
436                            &reason,
437                            ts_init,
438                            false,
439                        );
440                    }
441                    OKXWsOperation::CancelOrder
442                    | OKXWsOperation::BatchCancelOrders
443                    | OKXWsOperation::MassCancel => {
444                        state.pending_cancels.remove(cl_ord_id);
445                        emitter.emit_order_cancel_rejected_event(
446                            ident.strategy_id,
447                            ident.instrument_id,
448                            client_order_id,
449                            venue_order_id,
450                            &reason,
451                            ts_init,
452                        );
453                    }
454                    OKXWsOperation::AmendOrder | OKXWsOperation::BatchAmendOrders => {
455                        state.pending_amends.remove(cl_ord_id);
456                        emitter.emit_order_modify_rejected_event(
457                            ident.strategy_id,
458                            ident.instrument_id,
459                            client_order_id,
460                            venue_order_id,
461                            &reason,
462                            ts_init,
463                        );
464                    }
465                    _ => {
466                        log::warn!(
467                            "Order response error for unhandled op: \
468                             op={op:?} cl_ord_id={cl_ord_id} s_code={s_code} s_msg={s_msg}"
469                        );
470                    }
471                }
472            }
473
474            if code != "0" && data.is_empty() {
475                log::warn!(
476                    "Order response error (no data): id={id:?} op={op:?} code={code} msg={msg}"
477                );
478            }
479        }
480        OKXWsMessage::SendFailed {
481            request_id,
482            client_order_id,
483            op,
484            error,
485        } => {
486            log::warn!(
487                "WebSocket send failed without structured venue response: \
488                 request_id={request_id}, client_order_id={client_order_id:?}, \
489                 op={op:?}, awaiting reconciliation: {error}"
490            );
491
492            if let Some(client_order_id) = client_order_id {
493                let key = client_order_id.as_str();
494
495                match op {
496                    Some(
497                        OKXWsOperation::Order
498                        | OKXWsOperation::BatchOrders
499                        | OKXWsOperation::OrderAlgo,
500                    ) => {
501                        state.pending_orders.remove(key);
502                    }
503                    Some(
504                        OKXWsOperation::CancelOrder
505                        | OKXWsOperation::BatchCancelOrders
506                        | OKXWsOperation::MassCancel
507                        | OKXWsOperation::CancelAlgos,
508                    ) => {
509                        state.pending_cancels.remove(key);
510                    }
511                    Some(OKXWsOperation::AmendOrder | OKXWsOperation::BatchAmendOrders) => {
512                        state.pending_amends.remove(key);
513                    }
514                    _ => {}
515                }
516            }
517        }
518        OKXWsMessage::ChannelData { channel, .. } => {
519            log::debug!("Ignoring data channel message on execution client: {channel:?}");
520        }
521        OKXWsMessage::BookData { .. } | OKXWsMessage::Instruments(_) => {
522            log::debug!("Ignoring data message on execution client");
523        }
524        OKXWsMessage::Error(e) => {
525            log::warn!(
526                "Websocket error: code={} message={} conn_id={:?}",
527                e.code,
528                e.message,
529                e.conn_id
530            );
531        }
532        OKXWsMessage::Reconnected => {
533            log::info!("Websocket reconnected");
534        }
535        OKXWsMessage::Authenticated => {
536            log::debug!("Websocket authenticated");
537        }
538    }
539}
540
541/// Dispatches order messages, producing proper order events for tracked orders
542/// and falling back to execution reports for untracked/external orders.
543#[expect(clippy::too_many_arguments)]
544fn dispatch_order_messages(
545    order_msgs: &[OKXOrderMsg],
546    emitter: &ExecutionEventEmitter,
547    state: &WsDispatchState,
548    account_id: AccountId,
549    instruments: &AHashMap<Ustr, InstrumentAny>,
550    fee_cache: &mut AHashMap<Ustr, Money>,
551    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
552    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
553    ts_init: UnixNanos,
554) {
555    for msg in order_msgs {
556        let Some(instrument) = instruments.get(&msg.inst_id) else {
557            log::warn!("No instrument for {}, skipping order message", msg.inst_id);
558            continue;
559        };
560
561        let Some(client_order_id) = parse_client_order_id(&msg.cl_ord_id) else {
562            log::debug!(
563                "Order without client_order_id (ord_id={}), sending as report",
564                msg.ord_id
565            );
566            dispatch_order_msg_as_report(
567                msg,
568                account_id,
569                instruments,
570                fee_cache,
571                filled_qty_cache,
572                emitter,
573                state,
574                ts_init,
575            );
576            continue;
577        };
578
579        // Resolve identity: check direct match first, then fall back to the
580        // parent algo order ID for triggered child orders. OKX assigns a new
581        // cl_ord_id to child orders when an algo/stop triggers, preserving the
582        // parent's client order ID in algo_cl_ord_id.
583        let (client_order_id, identity) = match state
584            .order_identities
585            .get(&client_order_id)
586            .map(|r| r.clone())
587        {
588            Some(ident) => (client_order_id, Some(ident)),
589            None => {
590                if let Some(parent_id) = msg
591                    .algo_cl_ord_id
592                    .as_deref()
593                    .and_then(parse_client_order_id)
594                {
595                    let parent_ident = state.order_identities.get(&parent_id).map(|r| r.clone());
596
597                    if parent_ident.is_some() {
598                        (parent_id, parent_ident)
599                    } else {
600                        (client_order_id, None)
601                    }
602                } else {
603                    (client_order_id, None)
604                }
605            }
606        };
607
608        if let Some(ident) = identity {
609            if is_post_only_auto_cancel(msg) {
610                let ts_event = parse_millisecond_timestamp(msg.u_time);
611                let rejected = OrderRejected::new(
612                    emitter.trader_id(),
613                    ident.strategy_id,
614                    instrument.id(),
615                    client_order_id,
616                    account_id,
617                    Ustr::from("Post-only order would have taken liquidity"),
618                    UUID4::new(),
619                    ts_event,
620                    ts_init,
621                    false,
622                    true, // due_post_only
623                );
624                state.order_identities.remove(&client_order_id);
625                order_state_cache.remove(&client_order_id);
626                fee_cache.remove(&msg.ord_id);
627                filled_qty_cache.remove(&msg.ord_id);
628                emitter.send_order_event(OrderEventAny::Rejected(rejected));
629                continue;
630            }
631
632            let previous_fee = fee_cache.get(&msg.ord_id).copied();
633            let previous_filled_qty = filled_qty_cache.get(&msg.ord_id).copied();
634            let previous_state = order_state_cache.get(&client_order_id);
635
636            match parse_order_event(
637                msg,
638                client_order_id,
639                account_id,
640                emitter.trader_id(),
641                ident.strategy_id,
642                instrument,
643                previous_fee,
644                previous_filled_qty,
645                previous_state,
646                ts_init,
647            ) {
648                Ok(event) => {
649                    update_order_caches(
650                        msg,
651                        instrument,
652                        client_order_id,
653                        fee_cache,
654                        filled_qty_cache,
655                        order_state_cache,
656                    );
657                    dispatch_parsed_order_event(
658                        event,
659                        client_order_id,
660                        account_id,
661                        VenueOrderId::new(msg.ord_id),
662                        &ident,
663                        instrument,
664                        msg.state,
665                        emitter,
666                        state,
667                        order_state_cache,
668                        ts_init,
669                    );
670                }
671                Err(e) => log::error!("Failed to parse order event for {client_order_id}: {e}"),
672            }
673        } else {
674            log::debug!(
675                "Untracked order {client_order_id} (ord_id={}), sending as report for reconciliation",
676                msg.ord_id
677            );
678            dispatch_order_msg_as_report(
679                msg,
680                account_id,
681                instruments,
682                fee_cache,
683                filled_qty_cache,
684                emitter,
685                state,
686                ts_init,
687            );
688        }
689    }
690}
691
692#[expect(clippy::too_many_arguments)]
693fn dispatch_spread_order_messages(
694    order_msgs: &[OKXSpreadOrder],
695    emitter: &ExecutionEventEmitter,
696    state: &WsDispatchState,
697    account_id: AccountId,
698    instruments: &AHashMap<Ustr, InstrumentAny>,
699    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
700    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
701    ts_init: UnixNanos,
702) {
703    for msg in order_msgs {
704        let Some(instrument) = instruments.get(&msg.sprd_id) else {
705            log::warn!(
706                "No instrument for {}, skipping spread order message",
707                msg.sprd_id
708            );
709            continue;
710        };
711
712        let Some(client_order_id) = parse_client_order_id(msg.cl_ord_id.as_str()) else {
713            log::debug!(
714                "Spread order without client_order_id (ord_id={}), sending as report",
715                msg.ord_id
716            );
717            dispatch_spread_order_msg_as_report(
718                msg,
719                account_id,
720                instruments,
721                filled_qty_cache,
722                emitter,
723                state,
724                ts_init,
725            );
726            continue;
727        };
728
729        let identity = state
730            .order_identities
731            .get(&client_order_id)
732            .map(|r| r.clone());
733
734        if let Some(ident) = identity {
735            if is_spread_post_only_auto_cancel(msg) {
736                let ts_event = msg
737                    .u_time
738                    .or(msg.c_time)
739                    .map_or(ts_init, parse_millisecond_timestamp);
740                let rejected = OrderRejected::new(
741                    emitter.trader_id(),
742                    ident.strategy_id,
743                    instrument.id(),
744                    client_order_id,
745                    account_id,
746                    Ustr::from(OKX_POST_ONLY_CANCEL_REASON),
747                    UUID4::new(),
748                    ts_event,
749                    ts_init,
750                    false,
751                    true,
752                );
753                state.order_identities.remove(&client_order_id);
754                order_state_cache.remove(&client_order_id);
755                filled_qty_cache.remove(&msg.ord_id);
756                emitter.send_order_event(OrderEventAny::Rejected(rejected));
757                continue;
758            }
759
760            let previous_filled_qty = filled_qty_cache.get(&msg.ord_id).copied();
761            let previous_state = order_state_cache.get(&client_order_id);
762
763            match parse_spread_order_event(
764                msg,
765                client_order_id,
766                account_id,
767                emitter.trader_id(),
768                ident.strategy_id,
769                instrument,
770                previous_filled_qty,
771                previous_state,
772                ts_init,
773            ) {
774                Ok(event) => {
775                    update_spread_order_caches(
776                        msg,
777                        instrument,
778                        client_order_id,
779                        filled_qty_cache,
780                        order_state_cache,
781                    );
782                    dispatch_parsed_order_event(
783                        event,
784                        client_order_id,
785                        account_id,
786                        VenueOrderId::new(msg.ord_id.as_str()),
787                        &ident,
788                        instrument,
789                        msg.state,
790                        emitter,
791                        state,
792                        order_state_cache,
793                        ts_init,
794                    );
795                }
796                Err(e) => {
797                    log::error!("Failed to parse spread order event for {client_order_id}: {e}");
798                }
799            }
800        } else {
801            log::debug!(
802                "Untracked spread order {client_order_id} (ord_id={}), sending as report for reconciliation",
803                msg.ord_id
804            );
805            dispatch_spread_order_msg_as_report(
806                msg,
807                account_id,
808                instruments,
809                filled_qty_cache,
810                emitter,
811                state,
812                ts_init,
813            );
814        }
815    }
816}
817
818/// Dispatches a parsed order event as a proper `OrderEventAny`.
819///
820/// Guarantees the `Submitted -> Accepted -> ...` lifecycle by synthesizing
821/// `OrderAccepted` before any other event when one has not yet been emitted.
822/// Duplicate `Accepted` events (e.g. from reconnect replays) are suppressed.
823#[expect(clippy::too_many_arguments)]
824fn dispatch_parsed_order_event(
825    event: ParsedOrderEvent,
826    client_order_id: ClientOrderId,
827    account_id: AccountId,
828    venue_order_id: VenueOrderId,
829    identity: &OrderIdentity,
830    instrument: &InstrumentAny,
831    venue_status: OKXOrderStatus,
832    emitter: &ExecutionEventEmitter,
833    state: &WsDispatchState,
834    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
835    ts_init: UnixNanos,
836) {
837    let is_terminal;
838
839    match event {
840        ParsedOrderEvent::Accepted(e) => {
841            if state.emitted_accepted.contains(&client_order_id)
842                || state.filled_orders.contains(&client_order_id)
843                || state.triggered_orders.contains(&client_order_id)
844                || state.terminal_orders.contains(&client_order_id)
845            {
846                log::debug!("Skipping duplicate Accepted for {client_order_id}");
847                return;
848            }
849            state.insert_accepted(client_order_id);
850            is_terminal = false;
851            emitter.send_order_event(OrderEventAny::Accepted(e));
852        }
853        ParsedOrderEvent::Triggered(e) => {
854            if state.filled_orders.contains(&client_order_id) {
855                log::debug!("Skipping stale Triggered for {client_order_id} (already filled)");
856                return;
857            }
858
859            if !TRIGGERABLE_ORDER_TYPES.contains(&identity.order_type) {
860                log::debug!(
861                    "Skipping OrderTriggered for {} order {client_order_id}: market-style stops have no TRIGGERED state",
862                    identity.order_type,
863                );
864                state.insert_triggered(client_order_id);
865                return;
866            }
867
868            ensure_accepted_emitted(
869                client_order_id,
870                account_id,
871                venue_order_id,
872                identity,
873                emitter,
874                state,
875                ts_init,
876            );
877            state.insert_triggered(client_order_id);
878            is_terminal = false;
879            emitter.send_order_event(OrderEventAny::Triggered(e));
880        }
881        ParsedOrderEvent::Canceled(e) => {
882            ensure_accepted_emitted(
883                client_order_id,
884                account_id,
885                venue_order_id,
886                identity,
887                emitter,
888                state,
889                ts_init,
890            );
891            state.triggered_orders.remove(&client_order_id);
892            state.filled_orders.remove(&client_order_id);
893            is_terminal = true;
894            emitter.send_order_event(OrderEventAny::Canceled(e));
895        }
896        ParsedOrderEvent::Expired(e) => {
897            ensure_accepted_emitted(
898                client_order_id,
899                account_id,
900                venue_order_id,
901                identity,
902                emitter,
903                state,
904                ts_init,
905            );
906            state.triggered_orders.remove(&client_order_id);
907            state.filled_orders.remove(&client_order_id);
908            is_terminal = true;
909            emitter.send_order_event(OrderEventAny::Expired(e));
910        }
911        ParsedOrderEvent::Updated(e) => {
912            ensure_accepted_emitted(
913                client_order_id,
914                account_id,
915                venue_order_id,
916                identity,
917                emitter,
918                state,
919                ts_init,
920            );
921            is_terminal = false;
922            emitter.send_order_event(OrderEventAny::Updated(e));
923        }
924        ParsedOrderEvent::Fill(fill_report) => {
925            let is_duplicate = state.check_and_insert_trade(fill_report.trade_id);
926            is_terminal = venue_status == OKXOrderStatus::Filled;
927
928            if is_duplicate {
929                log::debug!(
930                    "Skipping duplicate fill for {client_order_id}: trade_id={}",
931                    fill_report.trade_id
932                );
933            } else {
934                ensure_accepted_emitted(
935                    client_order_id,
936                    account_id,
937                    venue_order_id,
938                    identity,
939                    emitter,
940                    state,
941                    ts_init,
942                );
943                state.insert_filled(client_order_id);
944                state.triggered_orders.remove(&client_order_id);
945                let filled = fill_report_to_order_filled(
946                    &fill_report,
947                    emitter.trader_id(),
948                    identity,
949                    instrument.quote_currency(),
950                );
951                emitter.send_order_event(OrderEventAny::Filled(filled));
952            }
953        }
954        ParsedOrderEvent::StatusOnly(report) => {
955            is_terminal = matches!(
956                report.order_status,
957                OrderStatus::Filled | OrderStatus::Canceled | OrderStatus::Expired
958            );
959            emitter.send_order_status_report(*report);
960        }
961        ParsedOrderEvent::Skipped => return,
962    }
963
964    if is_terminal {
965        state.insert_terminal(client_order_id);
966        state.order_identities.remove(&client_order_id);
967        state.emitted_accepted.remove(&client_order_id);
968        order_state_cache.remove(&client_order_id);
969        // Keep fee_cache and filled_qty_cache entries: replayed terminal
970        // messages go through the untracked report path and need prior
971        // cumulative state to avoid re-emitting the full fill quantity
972    }
973}
974
975/// Synthesizes and emits `OrderAccepted` if one has not yet been emitted for
976/// this order. Handles fast-filling orders that skip the `Live` state on OKX.
977fn ensure_accepted_emitted(
978    client_order_id: ClientOrderId,
979    account_id: AccountId,
980    venue_order_id: VenueOrderId,
981    identity: &OrderIdentity,
982    emitter: &ExecutionEventEmitter,
983    state: &WsDispatchState,
984    ts_init: UnixNanos,
985) {
986    if state.emitted_accepted.contains(&client_order_id) {
987        return;
988    }
989    state.insert_accepted(client_order_id);
990    let accepted = OrderAccepted::new(
991        emitter.trader_id(),
992        identity.strategy_id,
993        identity.instrument_id,
994        client_order_id,
995        venue_order_id,
996        account_id,
997        UUID4::new(),
998        ts_init,
999        ts_init,
1000        false,
1001    );
1002    emitter.send_order_event(OrderEventAny::Accepted(accepted));
1003}
1004
1005/// Converts a [`FillReport`] into an [`OrderFilled`] event using tracked identity.
1006fn fill_report_to_order_filled(
1007    report: &FillReport,
1008    trader_id: TraderId,
1009    identity: &OrderIdentity,
1010    quote_currency: Currency,
1011) -> OrderFilled {
1012    OrderFilled::new(
1013        trader_id,
1014        identity.strategy_id,
1015        report.instrument_id,
1016        report
1017            .client_order_id
1018            .expect("tracked order has client_order_id"),
1019        report.venue_order_id,
1020        report.account_id,
1021        report.trade_id,
1022        identity.order_side,
1023        identity.order_type,
1024        report.last_qty,
1025        report.last_px,
1026        quote_currency,
1027        report.liquidity_side,
1028        UUID4::new(),
1029        report.ts_event,
1030        report.ts_init,
1031        false,
1032        report.venue_position_id,
1033        Some(report.commission),
1034    )
1035}
1036
1037/// Falls back to the report path for a single order message.
1038#[expect(clippy::too_many_arguments)]
1039fn dispatch_order_msg_as_report(
1040    msg: &OKXOrderMsg,
1041    account_id: AccountId,
1042    instruments: &AHashMap<Ustr, InstrumentAny>,
1043    fee_cache: &mut AHashMap<Ustr, Money>,
1044    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
1045    emitter: &ExecutionEventEmitter,
1046    state: &WsDispatchState,
1047    ts_init: UnixNanos,
1048) {
1049    match parse_order_msg(
1050        msg,
1051        account_id,
1052        instruments,
1053        fee_cache,
1054        filled_qty_cache,
1055        ts_init,
1056    ) {
1057        Ok(report) => {
1058            if let Some(instrument) = instruments.get(&msg.inst_id) {
1059                update_fee_fill_caches(msg, instrument, fee_cache, filled_qty_cache);
1060            }
1061            dispatch_execution_reports(vec![report], emitter, state);
1062        }
1063        Err(e) => log::error!("Failed to parse order message as report: {e}"),
1064    }
1065}
1066
1067fn dispatch_spread_order_msg_as_report(
1068    msg: &OKXSpreadOrder,
1069    account_id: AccountId,
1070    instruments: &AHashMap<Ustr, InstrumentAny>,
1071    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
1072    emitter: &ExecutionEventEmitter,
1073    state: &WsDispatchState,
1074    ts_init: UnixNanos,
1075) {
1076    match parse_spread_order_msg(msg, account_id, instruments, filled_qty_cache, ts_init) {
1077        Ok(report) => {
1078            if let Some(instrument) = instruments.get(&msg.sprd_id) {
1079                update_spread_fill_cache(msg, instrument, filled_qty_cache);
1080            }
1081            dispatch_execution_reports(vec![report], emitter, state);
1082        }
1083        Err(e) => log::error!("Failed to parse spread order message as report: {e}"),
1084    }
1085}
1086
1087/// Updates fee, fill, and order state caches from a raw OKX order message.
1088fn update_order_caches(
1089    msg: &OKXOrderMsg,
1090    instrument: &InstrumentAny,
1091    client_order_id: ClientOrderId,
1092    fee_cache: &mut AHashMap<Ustr, Money>,
1093    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
1094    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
1095) {
1096    update_fee_fill_caches(msg, instrument, fee_cache, filled_qty_cache);
1097
1098    let venue_order_id = VenueOrderId::new(msg.ord_id);
1099    let quantity = parse_quantity(&msg.sz, instrument.size_precision()).unwrap_or_default();
1100    let price = if is_market_price(&msg.px) {
1101        None
1102    } else {
1103        parse_price(&msg.px, instrument.price_precision()).ok()
1104    };
1105
1106    order_state_cache.insert(
1107        client_order_id,
1108        OrderStateSnapshot {
1109            venue_order_id,
1110            quantity,
1111            price,
1112        },
1113    );
1114}
1115
1116fn update_spread_order_caches(
1117    msg: &OKXSpreadOrder,
1118    instrument: &InstrumentAny,
1119    client_order_id: ClientOrderId,
1120    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
1121    order_state_cache: &mut AHashMap<ClientOrderId, OrderStateSnapshot>,
1122) {
1123    update_spread_fill_cache(msg, instrument, filled_qty_cache);
1124
1125    let venue_order_id = VenueOrderId::new(msg.ord_id.as_str());
1126    let quantity = parse_quantity(&msg.sz, instrument.size_precision()).unwrap_or_default();
1127    let price = if is_market_price(&msg.px) {
1128        None
1129    } else {
1130        parse_price(&msg.px, instrument.price_precision()).ok()
1131    };
1132
1133    order_state_cache.insert(
1134        client_order_id,
1135        OrderStateSnapshot {
1136            venue_order_id,
1137            quantity,
1138            price,
1139        },
1140    );
1141}
1142
1143fn update_spread_fill_cache(
1144    msg: &OKXSpreadOrder,
1145    instrument: &InstrumentAny,
1146    filled_qty_cache: &mut AHashMap<Ustr, Quantity>,
1147) {
1148    if !msg.acc_fill_sz.is_empty()
1149        && msg.acc_fill_sz != "0"
1150        && let Ok(qty) = parse_quantity(&msg.acc_fill_sz, instrument.size_precision())
1151    {
1152        filled_qty_cache.insert(msg.ord_id, qty);
1153    }
1154}
1155
1156fn is_spread_post_only_auto_cancel(msg: &OKXSpreadOrder) -> bool {
1157    msg.state == OKXOrderStatus::Canceled && msg.cancel_source == OKX_POST_ONLY_CANCEL_SOURCE
1158}
1159
1160/// Dispatches execution reports with cross-stream deduplication.
1161pub fn dispatch_execution_reports(
1162    reports: Vec<ExecutionReport>,
1163    emitter: &ExecutionEventEmitter,
1164    state: &WsDispatchState,
1165) {
1166    log::debug!("Processing {} execution report(s)", reports.len());
1167
1168    for report in reports {
1169        match report {
1170            ExecutionReport::Order(order_report) => {
1171                if let Some(cid) = order_report.client_order_id {
1172                    match order_report.order_status {
1173                        // Guard form reformats awkwardly across multiple lines
1174                        #[allow(clippy::collapsible_match)]
1175                        OrderStatus::Accepted => {
1176                            if state.terminal_orders.contains(&cid)
1177                                || state.filled_orders.contains(&cid)
1178                                || state.triggered_orders.contains(&cid)
1179                            {
1180                                log::debug!(
1181                                    "Skipping stale OrderStatusReport(Accepted) \
1182                                     for {cid} (order already terminal)"
1183                                );
1184                                continue;
1185                            }
1186                        }
1187                        OrderStatus::Triggered => {
1188                            if state.filled_orders.contains(&cid) {
1189                                log::debug!(
1190                                    "Skipping stale OrderStatusReport(Triggered) \
1191                                     for {cid} (already filled)"
1192                                );
1193                                continue;
1194                            }
1195                            state.insert_triggered(cid);
1196                        }
1197                        OrderStatus::Filled => {
1198                            state.insert_filled(cid);
1199                            state.insert_terminal(cid);
1200                            state.triggered_orders.remove(&cid);
1201                        }
1202                        OrderStatus::Canceled | OrderStatus::Expired | OrderStatus::Rejected => {
1203                            state.insert_terminal(cid);
1204                            state.triggered_orders.remove(&cid);
1205                            state.filled_orders.remove(&cid);
1206                        }
1207                        _ => {}
1208                    }
1209                }
1210                emitter.send_order_status_report(order_report);
1211            }
1212            ExecutionReport::Fill(fill_report) => {
1213                if state.check_and_insert_trade(fill_report.trade_id) {
1214                    log::debug!(
1215                        "Skipping duplicate fill report: trade_id={}",
1216                        fill_report.trade_id
1217                    );
1218                    continue;
1219                }
1220
1221                if let Some(cid) = fill_report.client_order_id {
1222                    state.insert_filled(cid);
1223                    state.triggered_orders.remove(&cid);
1224                }
1225                emitter.send_fill_report(fill_report);
1226            }
1227        }
1228    }
1229}
1230
1231fn format_order_response_reason(s_code: &str, s_msg: &str, sub_code: &str) -> String {
1232    match (s_msg.is_empty(), sub_code.is_empty(), s_code.is_empty()) {
1233        (false, true, _) => s_msg.to_string(),
1234        (false, false, _) => format!("{s_msg} (subCode={sub_code})"),
1235        (true, false, false) => format!("sCode={s_code} subCode={sub_code}"),
1236        (true, false, true) => format!("subCode={sub_code}"),
1237        (true, true, false) => format!("sCode={s_code}"),
1238        (true, true, true) => String::new(),
1239    }
1240}
1241
1242#[derive(Debug, Clone)]
1243pub struct AlgoCancelContext {
1244    pub client_order_id: ClientOrderId,
1245    pub instrument_id: InstrumentId,
1246    pub strategy_id: StrategyId,
1247    pub venue_order_id: Option<VenueOrderId>,
1248}
1249
1250// Contexts must correspond 1:1 with the requests that produced
1251// the responses (OKX preserves request order in batch responses).
1252pub fn emit_algo_cancel_rejections(
1253    responses: &[OKXCancelAlgoOrderResponse],
1254    contexts: &[AlgoCancelContext],
1255    emitter: &ExecutionEventEmitter,
1256    clock: &'static AtomicTime,
1257) {
1258    for (i, item) in responses.iter().enumerate() {
1259        let code = item.s_code.as_deref().unwrap_or(OKX_SUCCESS_CODE);
1260        if code == OKX_SUCCESS_CODE {
1261            continue;
1262        }
1263
1264        let msg = item.s_msg.as_deref().unwrap_or("");
1265
1266        if let Some(ctx) = contexts.get(i) {
1267            let ts = clock.get_time_ns();
1268            emitter.emit_order_cancel_rejected_event(
1269                ctx.strategy_id,
1270                ctx.instrument_id,
1271                ctx.client_order_id,
1272                ctx.venue_order_id,
1273                msg,
1274                ts,
1275            );
1276        } else {
1277            log::warn!(
1278                "Algo cancel rejected but no context at index {i}: \
1279                 algo_id={} sCode={code} sMsg={msg}",
1280                item.algo_id
1281            );
1282        }
1283    }
1284}
1285
1286pub fn emit_batch_cancel_failure(
1287    contexts: &[AlgoCancelContext],
1288    error: &str,
1289    _emitter: &ExecutionEventEmitter,
1290    _clock: &'static AtomicTime,
1291) {
1292    for ctx in contexts {
1293        log::warn!(
1294            "Ambiguous algo batch cancel failure for {}, awaiting reconciliation: {error}",
1295            ctx.client_order_id
1296        );
1297    }
1298}
1299
1300#[cfg(test)]
1301mod tests {
1302    use rstest::rstest;
1303
1304    use super::{BoundedDedup, format_order_response_reason};
1305
1306    #[rstest]
1307    #[case("51000", "Rejected", "", "Rejected")]
1308    #[case("51000", "Rejected", "51004", "Rejected (subCode=51004)")]
1309    #[case("51000", "", "51004", "sCode=51000 subCode=51004")]
1310    #[case("51000", "", "", "sCode=51000")]
1311    #[case("", "", "51004", "subCode=51004")]
1312    #[case("", "", "", "")]
1313    fn test_format_order_response_reason(
1314        #[case] s_code: &str,
1315        #[case] s_msg: &str,
1316        #[case] sub_code: &str,
1317        #[case] expected: &str,
1318    ) {
1319        assert_eq!(
1320            format_order_response_reason(s_code, s_msg, sub_code),
1321            expected
1322        );
1323    }
1324
1325    #[rstest]
1326    fn test_bounded_dedup_check_and_insert_returns_false_on_first_insert() {
1327        let dedup = BoundedDedup::<u32>::new(4);
1328        assert!(!dedup.check_and_insert(1));
1329        assert!(dedup.contains(&1));
1330    }
1331
1332    #[rstest]
1333    fn test_bounded_dedup_check_and_insert_returns_true_on_duplicate() {
1334        let dedup = BoundedDedup::<u32>::new(4);
1335        dedup.insert(1);
1336        assert!(dedup.check_and_insert(1));
1337    }
1338
1339    #[rstest]
1340    fn test_bounded_dedup_evicts_oldest_on_overflow() {
1341        let dedup = BoundedDedup::<u32>::new(3);
1342        dedup.insert(1);
1343        dedup.insert(2);
1344        dedup.insert(3);
1345        dedup.insert(4);
1346
1347        assert!(!dedup.contains(&1));
1348        assert!(dedup.contains(&2));
1349        assert!(dedup.contains(&3));
1350        assert!(dedup.contains(&4));
1351    }
1352
1353    #[rstest]
1354    fn test_bounded_dedup_evicted_key_is_not_treated_as_duplicate() {
1355        let dedup = BoundedDedup::<u32>::new(2);
1356        dedup.insert(1);
1357        dedup.insert(2);
1358        dedup.insert(3);
1359
1360        assert!(!dedup.check_and_insert(1));
1361        assert!(dedup.contains(&1));
1362    }
1363
1364    #[rstest]
1365    fn test_bounded_dedup_remove_drops_entry() {
1366        let dedup = BoundedDedup::<u32>::new(4);
1367        dedup.insert(1);
1368        assert!(dedup.remove(&1));
1369        assert!(!dedup.contains(&1));
1370        assert!(!dedup.remove(&1));
1371    }
1372
1373    #[rstest]
1374    fn test_bounded_dedup_remove_then_reinsert_does_not_double_count() {
1375        let dedup = BoundedDedup::<u32>::new(2);
1376        for k in 0u32..1000 {
1377            dedup.insert(k);
1378            dedup.remove(&k);
1379        }
1380        assert!(!dedup.contains(&0));
1381        assert!(!dedup.contains(&500));
1382    }
1383
1384    #[rstest]
1385    fn test_bounded_dedup_reinsert_survives_stale_marker_eviction() {
1386        let dedup = BoundedDedup::<u32>::new(3);
1387        dedup.insert(1);
1388        dedup.remove(&1);
1389
1390        dedup.insert(2);
1391        dedup.insert(3);
1392        dedup.insert(1);
1393
1394        assert!(dedup.contains(&1));
1395        assert!(dedup.contains(&2));
1396        assert!(dedup.contains(&3));
1397    }
1398}
nautilus_okx/websocket/dispatch.rs

nautilus_okx/websocket/
dispatch.rs
