nautilus_binance/common/
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 dispatch state for tracked/external order routing.
17//!
18//! Orders submitted through this client and active local orders restored from the cache have their
19//! identity registered in [`WsDispatchState`]. User data stream messages route by registered identity:
20//! - Tracked orders produce typed order events (OrderAccepted, OrderFilled, etc.).
21//! - Untracked orders fall back to execution reports for reconciliation.
22
23use ahash::AHashSet;
24use dashmap::{DashMap, mapref::entry::Entry};
25use nautilus_common::cache::fifo::{FifoCache, FifoCacheMap};
26use nautilus_core::{UUID4, UnixNanos};
27use nautilus_live::ExecutionEventEmitter;
28use nautilus_model::{
29    enums::{OrderSide, OrderType},
30    events::{OrderAccepted, OrderCanceled, OrderEventAny},
31    identifiers::{
32        AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TradeId, VenueOrderId,
33    },
34    orders::{Order, OrderAny},
35    reports::OrderStatusReport,
36    types::{Price, Quantity},
37};
38use parking_lot::Mutex;
39use ustr::Ustr;
40
41/// The type of operation a pending WS API request represents.
42#[derive(Debug, Clone, Copy)]
43pub enum PendingOperation {
44    Place,
45    Cancel,
46    Modify,
47}
48
49/// A pending WS API request awaiting a response.
50///
51/// Stored in [`WsDispatchState::pending_requests`] after the WS client
52/// returns a request ID. When the venue responds (accepted or rejected),
53/// the pending request is removed and used to emit the correct order event.
54#[derive(Debug, Clone)]
55pub struct PendingRequest {
56    pub client_order_id: ClientOrderId,
57    pub venue_order_id: Option<VenueOrderId>,
58    pub operation: PendingOperation,
59}
60
61/// Outcome of a cancel-replace request, tracked by its `cancelNewClientOrderId`.
62///
63/// The cancel half's `CANCELED` report and the venue's response arrive in either
64/// order, so whichever comes second completes the picture.
65#[derive(Debug, Clone)]
66enum CancelReplaceOutcome {
67    Pending,
68    /// The cancel report arrived; withheld while the replacement is pending or succeeded.
69    ///
70    /// Parsed at that point so it can be replayed even for an order without a
71    /// dispatch identity, such as an external order.
72    Canceled(Box<OrderStatusReport>),
73    /// The venue rejected the request before any cancel report arrived.
74    Rejected,
75}
76
77/// Order identity context registered on submission or restored for an active local order.
78///
79/// Provides the strategy and instrument metadata needed to construct
80/// typed order events without accessing the cache from the async dispatch task.
81/// The quantity and quote-sizing flag retain the order's sizing intent.
82#[derive(Debug, Clone)]
83pub struct OrderIdentity {
84    pub instrument_id: InstrumentId,
85    pub strategy_id: StrategyId,
86    pub order_side: OrderSide,
87    pub order_type: OrderType,
88    pub price: Option<Price>,
89    pub quantity: Quantity,
90    pub is_quote_quantity: bool,
91    pub venue_position_id: Option<PositionId>,
92}
93
94#[derive(Debug, Clone, Copy)]
95struct AlgoOrderIds {
96    algo: VenueOrderId,
97    current: VenueOrderId,
98}
99
100/// Tracks order lifecycle state for dispatch routing.
101///
102/// Orders with a registered identity produce typed order events. Untracked orders
103/// fall back to execution reports for reconciliation. For restored orders, accepted and filled
104/// state, plus the trade IDs known at restoration, remain available until terminal cleanup,
105/// independent of the bounded deduplication caches.
106#[derive(Debug)]
107pub struct WsDispatchState {
108    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
109    pub pending_requests: DashMap<String, PendingRequest>,
110    pub(crate) order_lists: DashMap<ClientOrderId, OrderListIdentity>,
111    algo_order_ids: DashMap<ClientOrderId, AlgoOrderIds>,
112    order_updates: DashMap<ClientOrderId, OrderBinding>,
113    orders_restored: DashMap<ClientOrderId, RestoredOrderState>,
114    replacements: DashMap<ClientOrderId, PendingReplacement>,
115    emitted_accepted: Mutex<FifoCache<ClientOrderId, 10_000>>,
116    filled_orders: Mutex<FifoCache<ClientOrderId, 10_000>>,
117    /// Cancel-replace request IDs mapped to their `cancelNewClientOrderId`.
118    pub cancel_replace_request_ids: DashMap<String, String>,
119    cancel_replace_outcomes: Mutex<FifoCacheMap<String, CancelReplaceOutcome, 10_000>>,
120}
121
122impl Default for WsDispatchState {
123    fn default() -> Self {
124        Self {
125            order_identities: DashMap::new(),
126            pending_requests: DashMap::new(),
127            order_lists: DashMap::new(),
128            algo_order_ids: DashMap::new(),
129            order_updates: DashMap::new(),
130            orders_restored: DashMap::new(),
131            replacements: DashMap::new(),
132            emitted_accepted: Mutex::new(FifoCache::new()),
133            filled_orders: Mutex::new(FifoCache::new()),
134            cancel_replace_request_ids: DashMap::new(),
135            cancel_replace_outcomes: Mutex::new(FifoCacheMap::new()),
136        }
137    }
138}
139
140impl WsDispatchState {
141    pub(crate) fn restore_order(
142        &self,
143        order: &OrderAny,
144        venue_position_id: Option<PositionId>,
145        price_precision: u8,
146    ) {
147        let cid = order.client_order_id();
148
149        let Entry::Vacant(entry) = self.order_identities.entry(cid) else {
150            return;
151        };
152
153        let accepted = order.ts_accepted().is_some();
154        self.orders_restored.insert(
155            cid,
156            RestoredOrderState {
157                trade_ids: order.trade_ids().into_iter().copied().collect(),
158                accepted,
159                filled: order.filled_qty().is_positive(),
160            },
161        );
162
163        if let Some(venue_order_id) = order.venue_order_id()
164            && !order.is_quote_quantity()
165        {
166            let current = OrderUpdate {
167                venue_order_id,
168                quantity: order.quantity(),
169                price: order
170                    .price()
171                    .unwrap_or_else(|| Price::zero(price_precision)),
172                trigger_price: order.trigger_price(),
173            };
174
175            let superseded = order
176                .venue_order_ids()
177                .into_iter()
178                .copied()
179                .filter(|id| *id != venue_order_id)
180                .collect();
181            self.order_updates.insert(
182                cid,
183                OrderBinding {
184                    current,
185                    superseded,
186                },
187            );
188        }
189
190        entry.insert(OrderIdentity {
191            instrument_id: order.instrument_id(),
192            strategy_id: order.strategy_id(),
193            order_side: order.order_side(),
194            order_type: order.order_type(),
195            price: order.price(),
196            quantity: order.quantity(),
197            is_quote_quantity: order.is_quote_quantity(),
198            venue_position_id,
199        });
200    }
201
202    pub fn has_emitted_accepted(&self, cid: &ClientOrderId) -> bool {
203        self.orders_restored
204            .get(cid)
205            .is_some_and(|order| order.accepted)
206            || self.order_lists.contains_key(cid)
207            || self.emitted_accepted.lock().contains(cid)
208    }
209
210    /// Marks an order as having emitted an OrderAccepted event.
211    pub fn insert_accepted(&self, cid: ClientOrderId) {
212        if let Some(mut order) = self.orders_restored.get_mut(&cid) {
213            order.accepted = true;
214        }
215
216        self.emitted_accepted.lock().add(cid);
217    }
218
219    pub fn has_filled(&self, cid: &ClientOrderId) -> bool {
220        self.orders_restored
221            .get(cid)
222            .is_some_and(|order| order.filled)
223            || self.filled_orders.lock().contains(cid)
224    }
225
226    pub(crate) fn is_order_new_stale(
227        &self,
228        cid: &ClientOrderId,
229        venue_order_id: VenueOrderId,
230    ) -> bool {
231        self.has_filled(cid)
232            && self
233                .order_updates
234                .get(cid)
235                .is_none_or(|binding| binding.current.venue_order_id == venue_order_id)
236    }
237
238    pub(crate) fn has_trade(
239        &self,
240        cid: &ClientOrderId,
241        key: &(Ustr, i64),
242        seen_trade_ids: &Mutex<FifoCache<(Ustr, i64), 10_000>>,
243    ) -> bool {
244        self.orders_restored.get(cid).is_some_and(|order| {
245            let trade_id = TradeId::new(key.1.to_string());
246            order.trade_ids.contains(&trade_id)
247        }) || seen_trade_ids.lock().contains(key)
248    }
249
250    /// Marks an order as having received a fill.
251    pub fn insert_filled(&self, cid: ClientOrderId) {
252        if let Some(mut order) = self.orders_restored.get_mut(&cid) {
253            order.filled = true;
254        }
255
256        self.filled_orders.lock().add(cid);
257    }
258
259    /// Records the `cancelNewClientOrderId` sent with a cancel-replace request.
260    pub fn insert_cancel_replace(&self, cancel_id: String) {
261        self.cancel_replace_outcomes
262            .lock()
263            .insert(cancel_id, CancelReplaceOutcome::Pending);
264    }
265
266    /// Returns `true` when `cancel_id` belongs to a cancel-replace this client issued.
267    pub fn has_cancel_replace(&self, cancel_id: &str) -> bool {
268        self.cancel_replace_outcomes
269            .lock()
270            .contains_key(&cancel_id.to_string())
271    }
272
273    /// Records the cancel half's `CANCELED` report for a cancel-replace request.
274    ///
275    /// Returns `true` when the report must be withheld because the replacement is
276    /// pending or succeeded, and `false` when it should dispatch as a standalone
277    /// cancel because the venue already rejected the replacement or the ID is not
278    /// one this client issued.
279    pub fn on_cancel_replace_canceled(&self, cancel_id: &str, report: OrderStatusReport) -> bool {
280        let mut outcomes = self.cancel_replace_outcomes.lock();
281        let Some(outcome) = outcomes.get_mut(&cancel_id.to_string()) else {
282            return false;
283        };
284
285        match outcome {
286            CancelReplaceOutcome::Pending => {
287                *outcome = CancelReplaceOutcome::Canceled(Box::new(report));
288                true
289            }
290            CancelReplaceOutcome::Canceled(_) => true,
291            CancelReplaceOutcome::Rejected => false,
292        }
293    }
294
295    /// Records a rejected cancel-replace request.
296    ///
297    /// Returns the withheld cancel report when it already arrived, so the caller
298    /// can emit the confirmed cancellation for the original order.
299    pub fn on_cancel_replace_rejected(&self, cancel_id: &str) -> Option<OrderStatusReport> {
300        let mut outcomes = self.cancel_replace_outcomes.lock();
301        let outcome = outcomes.get_mut(&cancel_id.to_string())?;
302
303        match outcome {
304            CancelReplaceOutcome::Pending => {
305                *outcome = CancelReplaceOutcome::Rejected;
306                None
307            }
308            CancelReplaceOutcome::Canceled(report) => Some((**report).clone()),
309            CancelReplaceOutcome::Rejected => None,
310        }
311    }
312
313    pub fn insert_algo_order_id(&self, cid: ClientOrderId, venue_order_id: VenueOrderId) {
314        self.algo_order_ids.entry(cid).or_insert(AlgoOrderIds {
315            algo: venue_order_id,
316            current: venue_order_id,
317        });
318    }
319
320    /// Promotes a known Algo order to its matching-engine venue order ID.
321    ///
322    /// Returns `None` for an unknown Algo order, `Some(true)` for a new ID, and
323    /// `Some(false)` when the ID was already current.
324    pub fn promote_algo_order_id(
325        &self,
326        cid: ClientOrderId,
327        venue_order_id: VenueOrderId,
328    ) -> Option<bool> {
329        let mut ids = self.algo_order_ids.get_mut(&cid)?;
330        let changed = ids.current != venue_order_id;
331        ids.current = venue_order_id;
332        Some(changed)
333    }
334
335    /// Returns the matching-engine venue order ID for a promoted Algo order.
336    pub fn promoted_algo_order_id(&self, cid: &ClientOrderId) -> Option<VenueOrderId> {
337        self.algo_order_ids
338            .get(cid)
339            .and_then(|ids| (ids.current != ids.algo).then_some(ids.current))
340    }
341
342    /// Returns `None` for a superseded venue order ID.
343    ///
344    /// Otherwise, returns `Some(true)` for the first update or a changed venue order ID,
345    /// quantity, price, or trigger price, and `Some(false)` for unchanged values.
346    pub(crate) fn record_order_update(
347        &self,
348        cid: ClientOrderId,
349        venue_order_id: VenueOrderId,
350        quantity: Quantity,
351        price: Price,
352        trigger_price: Option<Price>,
353    ) -> Option<bool> {
354        let update = OrderUpdate {
355            venue_order_id,
356            quantity,
357            price,
358            trigger_price,
359        };
360
361        let changed = match self.order_updates.entry(cid) {
362            Entry::Occupied(mut entry) => {
363                let binding = entry.get_mut();
364                if binding.superseded.contains(&venue_order_id) {
365                    return None;
366                }
367
368                let changed = binding.current != update;
369                if binding.current.venue_order_id != venue_order_id {
370                    binding.superseded.push(binding.current.venue_order_id);
371                }
372
373                binding.current = update;
374                changed
375            }
376            Entry::Vacant(entry) => {
377                entry.insert(OrderBinding {
378                    current: update,
379                    superseded: Vec::new(),
380                });
381
382                true
383            }
384        };
385
386        self.replacements
387            .remove_if(&cid, |_, pending| pending.venue_order_id != venue_order_id);
388        Some(changed)
389    }
390
391    pub(crate) fn record_order_update_from_fill(
392        &self,
393        cid: ClientOrderId,
394        venue_order_id: VenueOrderId,
395        quantity: Quantity,
396        price: Price,
397        trigger_price: Option<Price>,
398        identity: &OrderIdentity,
399    ) -> Option<bool> {
400        if !identity.is_quote_quantity {
401            self.order_updates
402                .entry(cid)
403                .or_insert_with(|| OrderBinding {
404                    current: OrderUpdate {
405                        venue_order_id,
406                        quantity: identity.quantity,
407                        price: identity.price.unwrap_or(price),
408                        trigger_price,
409                    },
410                    superseded: Vec::new(),
411                });
412        }
413
414        self.record_order_update(cid, venue_order_id, quantity, price, trigger_price)
415    }
416
417    pub(crate) fn begin_replace(&self, cid: ClientOrderId, venue_order_id: VenueOrderId) {
418        self.replacements.insert(
419            cid,
420            PendingReplacement {
421                venue_order_id,
422                canceled: None,
423            },
424        );
425    }
426
427    pub(crate) fn defer_replace_cancel(&self, canceled: OrderCanceled) -> bool {
428        let cid = canceled.client_order_id;
429
430        if self.order_updates.get(&cid).is_some_and(|binding| {
431            canceled
432                .venue_order_id
433                .is_some_and(|venue_order_id| binding.superseded.contains(&venue_order_id))
434        }) {
435            return true;
436        }
437
438        if let Some(mut pending) = self.replacements.get_mut(&cid)
439            && Some(pending.venue_order_id) == canceled.venue_order_id
440        {
441            pending.canceled = Some(canceled);
442            return true;
443        }
444        false
445    }
446
447    pub(crate) fn reject_replace(&self, cid: ClientOrderId) -> Option<OrderCanceled> {
448        self.replacements
449            .remove(&cid)
450            .and_then(|(_, pending)| pending.canceled)
451    }
452
453    /// Removes all tracking state for a terminal order.
454    pub fn cleanup_terminal(&self, cid: ClientOrderId) {
455        self.order_identities.remove(&cid);
456        self.order_lists.remove(&cid);
457        self.algo_order_ids.remove(&cid);
458        self.order_updates.remove(&cid);
459        self.orders_restored.remove(&cid);
460        self.replacements.remove(&cid);
461        self.emitted_accepted.lock().remove(&cid);
462        self.filled_orders.lock().remove(&cid);
463    }
464}
465
466#[derive(Debug)]
467struct RestoredOrderState {
468    trade_ids: AHashSet<TradeId>,
469    accepted: bool,
470    filled: bool,
471}
472
473#[derive(Debug, Clone, Copy)]
474pub(crate) struct OrderListIdentity {
475    pub order_list_id: i64,
476    pub venue_order_id: VenueOrderId,
477    pub execution_report_seen: bool,
478}
479
480#[derive(Debug)]
481struct PendingReplacement {
482    venue_order_id: VenueOrderId,
483    canceled: Option<OrderCanceled>,
484}
485
486#[derive(Debug, Clone, Copy, PartialEq, Eq)]
487struct OrderUpdate {
488    venue_order_id: VenueOrderId,
489    quantity: Quantity,
490    price: Price,
491    trigger_price: Option<Price>,
492}
493
494#[derive(Debug)]
495struct OrderBinding {
496    current: OrderUpdate,
497    superseded: Vec<VenueOrderId>,
498}
499
500/// Synthesizes and emits OrderAccepted if one has not yet been emitted.
501///
502/// Handles fast-filling orders that skip the New state on Binance.
503pub fn ensure_accepted_emitted(
504    client_order_id: ClientOrderId,
505    account_id: AccountId,
506    venue_order_id: VenueOrderId,
507    identity: &OrderIdentity,
508    emitter: &ExecutionEventEmitter,
509    state: &WsDispatchState,
510    ts_init: UnixNanos,
511) {
512    if state.has_emitted_accepted(&client_order_id) {
513        return;
514    }
515    state.insert_accepted(client_order_id);
516    let accepted = OrderAccepted::new(
517        emitter.trader_id(),
518        identity.strategy_id,
519        identity.instrument_id,
520        client_order_id,
521        venue_order_id,
522        account_id,
523        UUID4::new(),
524        ts_init,
525        ts_init,
526        false,
527    );
528    emitter.send_order_event(OrderEventAny::Accepted(accepted));
529}
530
531#[cfg(test)]
532mod tests {
533    use nautilus_model::{
534        enums::OrderType,
535        events::order::spec::{OrderFilledSpec, OrderUpdatedSpec},
536        identifiers::AccountId,
537        orders::{OrderTestBuilder, stubs::TestOrderEventStubs},
538    };
539    use rstest::rstest;
540
541    use super::*;
542
543    #[rstest]
544    #[case::submitted(false)]
545    #[case::accepted(true)]
546    fn test_restore_order_retains_identity_and_lifecycle_beyond_replay_capacity(
547        #[case] accepted: bool,
548        #[values(false, true)] quote_quantity: bool,
549    ) {
550        let order_type = if quote_quantity {
551            OrderType::Market
552        } else {
553            OrderType::Limit
554        };
555
556        let mut order = OrderTestBuilder::new(order_type)
557            .client_order_id(ClientOrderId::from("restored-active"))
558            .strategy_id(StrategyId::from("RETAINED-007"))
559            .instrument_id(InstrumentId::from("ETHUSDT.BINANCE"))
560            .side(OrderSide::Sell)
561            .quantity(Quantity::from("2.003"))
562            .price(Price::from("3456.78"))
563            .quote_quantity(quote_quantity)
564            .build();
565        let account_id = AccountId::from("BINANCE-009");
566        let submitted = TestOrderEventStubs::submitted(&order, account_id);
567        order.apply(submitted).unwrap();
568
569        if accepted {
570            let event =
571                TestOrderEventStubs::accepted(&order, account_id, VenueOrderId::from("9182"));
572            order.apply(event).unwrap();
573        }
574
575        let cid = order.client_order_id();
576        let position_id = Some(PositionId::from("ETHUSDT.BINANCE-SHORT"));
577        let state = WsDispatchState::default();
578        state.restore_order(&order, position_id, 2);
579
580        assert_eq!(state.has_emitted_accepted(&cid), accepted);
581        assert!(!state.has_filled(&cid));
582        state.insert_accepted(cid);
583        state.insert_filled(cid);
584
585        for index in 0..10_001 {
586            let other = ClientOrderId::from(format!("other-{index}"));
587            state.insert_accepted(other);
588            state.insert_filled(other);
589        }
590
591        state.restore_order(&order, None, 2);
592
593        let identity = state.order_identities.get(&cid).unwrap().clone();
594        assert_eq!(identity.instrument_id, order.instrument_id());
595        assert_eq!(identity.strategy_id, order.strategy_id());
596        assert_eq!(identity.order_side, OrderSide::Sell);
597        assert_eq!(identity.order_type, order_type);
598        assert_eq!(
599            identity.price,
600            (!quote_quantity).then(|| Price::from("3456.78"))
601        );
602        assert_eq!(identity.quantity, Quantity::from("2.003"));
603        assert_eq!(identity.is_quote_quantity, quote_quantity);
604        assert_eq!(identity.venue_position_id, position_id);
605        assert!(state.has_emitted_accepted(&cid));
606        assert!(state.has_filled(&cid));
607
608        state.cleanup_terminal(cid);
609        assert!(!state.order_identities.contains_key(&cid));
610        assert!(!state.orders_restored.contains_key(&cid));
611        assert!(!state.has_emitted_accepted(&cid));
612        assert!(!state.has_filled(&cid));
613    }
614
615    #[rstest]
616    fn test_restore_market_order_seeds_zero_price_update(
617        #[values(false, true)] quote_quantity: bool,
618    ) {
619        let mut order = OrderTestBuilder::new(OrderType::Market)
620            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
621            .quantity(Quantity::from("0.002"))
622            .quote_quantity(quote_quantity)
623            .build();
624        let event = TestOrderEventStubs::accepted(
625            &order,
626            AccountId::from("BINANCE-001"),
627            VenueOrderId::from("312"),
628        );
629        order.apply(event).unwrap();
630        let state = WsDispatchState::default();
631        state.restore_order(&order, None, 2);
632
633        assert_eq!(
634            state.record_order_update(
635                order.client_order_id(),
636                VenueOrderId::from("312"),
637                order.quantity(),
638                Price::zero(2),
639                None
640            ),
641            Some(quote_quantity)
642        );
643    }
644
645    #[rstest]
646    fn test_restore_order_retains_partial_fills() {
647        let mut order = OrderTestBuilder::new(OrderType::Limit)
648            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
649            .quantity(Quantity::from("3"))
650            .price(Price::from("50000.00"))
651            .build();
652        let accepted = TestOrderEventStubs::accepted(
653            &order,
654            AccountId::from("BINANCE-001"),
655            VenueOrderId::from("312"),
656        );
657        order.apply(accepted).unwrap();
658        let fill = OrderFilledSpec::builder()
659            .trader_id(order.trader_id())
660            .strategy_id(order.strategy_id())
661            .instrument_id(order.instrument_id())
662            .client_order_id(order.client_order_id())
663            .venue_order_id(VenueOrderId::from("312"))
664            .account_id(AccountId::from("BINANCE-001"))
665            .trade_id(TradeId::from("9182"))
666            .last_qty(Quantity::from("1"))
667            .last_px(Price::from("50000.00"))
668            .build();
669        order.apply(OrderEventAny::Filled(fill)).unwrap();
670        let state = WsDispatchState::default();
671        state.restore_order(&order, None, 2);
672
673        assert!(state.has_emitted_accepted(&order.client_order_id()));
674        assert!(state.has_filled(&order.client_order_id()));
675        let seen_trade_ids = Mutex::new(FifoCache::new());
676        assert!(state.has_trade(
677            &order.client_order_id(),
678            &(Ustr::from("BTCUSDT"), 9182),
679            &seen_trade_ids
680        ));
681        assert!(!state.has_trade(
682            &order.client_order_id(),
683            &(Ustr::from("BTCUSDT"), 9183),
684            &seen_trade_ids
685        ));
686    }
687
688    #[rstest]
689    fn test_restore_order_same_shape_replacement_updates_venue_id() {
690        let mut order = OrderTestBuilder::new(OrderType::Limit)
691            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
692            .quantity(Quantity::from("3"))
693            .price(Price::from("50000.00"))
694            .build();
695        let account_id = AccountId::from("BINANCE-001");
696        let old_id = VenueOrderId::from("312");
697        let current_id = VenueOrderId::from("313");
698        let accepted = TestOrderEventStubs::accepted(&order, account_id, old_id);
699        order.apply(accepted).unwrap();
700        let state = WsDispatchState::default();
701        state.restore_order(&order, None, 2);
702
703        let canceled = OrderCanceled::new(
704            order.trader_id(),
705            order.strategy_id(),
706            order.instrument_id(),
707            order.client_order_id(),
708            UUID4::new(),
709            UnixNanos::default(),
710            UnixNanos::default(),
711            false,
712            Some(current_id),
713            Some(account_id),
714            None,
715        );
716        assert!(!state.defer_replace_cancel(canceled));
717        assert_eq!(
718            state.record_order_update(
719                order.client_order_id(),
720                current_id,
721                order.quantity(),
722                order.price().unwrap(),
723                None
724            ),
725            Some(true)
726        );
727        assert_eq!(
728            state.record_order_update(
729                order.client_order_id(),
730                current_id,
731                order.quantity(),
732                order.price().unwrap(),
733                None
734            ),
735            Some(false)
736        );
737
738        let stale_canceled = OrderCanceled::new(
739            order.trader_id(),
740            order.strategy_id(),
741            order.instrument_id(),
742            order.client_order_id(),
743            UUID4::new(),
744            UnixNanos::default(),
745            UnixNanos::default(),
746            false,
747            Some(old_id),
748            Some(account_id),
749            None,
750        );
751        assert!(state.defer_replace_cancel(stale_canceled));
752        assert!(!state.defer_replace_cancel(canceled));
753    }
754
755    #[rstest]
756    fn test_restore_order_retains_superseded_venue_ids() {
757        let mut order = OrderTestBuilder::new(OrderType::Limit)
758            .instrument_id(InstrumentId::from("BTCUSDT.BINANCE"))
759            .quantity(Quantity::from("3"))
760            .price(Price::from("50000.00"))
761            .build();
762        let account_id = AccountId::from("BINANCE-001");
763        let old_id = VenueOrderId::from("312");
764        let current_id = VenueOrderId::from("313");
765        let unknown_id = VenueOrderId::from("314");
766        let accepted = TestOrderEventStubs::accepted(&order, account_id, old_id);
767        order.apply(accepted).unwrap();
768        let updated = OrderUpdatedSpec::builder()
769            .trader_id(order.trader_id())
770            .strategy_id(order.strategy_id())
771            .instrument_id(order.instrument_id())
772            .client_order_id(order.client_order_id())
773            .quantity(order.quantity())
774            .price(order.price().unwrap())
775            .venue_order_id(current_id)
776            .account_id(account_id)
777            .build();
778        order.apply(OrderEventAny::Updated(updated)).unwrap();
779        let state = WsDispatchState::default();
780        state.restore_order(&order, None, 2);
781
782        for (venue_order_id, suppressed) in
783            [(old_id, true), (current_id, false), (unknown_id, false)]
784        {
785            let canceled = OrderCanceled::new(
786                order.trader_id(),
787                order.strategy_id(),
788                order.instrument_id(),
789                order.client_order_id(),
790                UUID4::new(),
791                UnixNanos::default(),
792                UnixNanos::default(),
793                false,
794                Some(venue_order_id),
795                Some(account_id),
796                None,
797            );
798            assert_eq!(state.defer_replace_cancel(canceled), suppressed);
799        }
800
801        state.begin_replace(order.client_order_id(), current_id);
802
803        let pending = OrderCanceled::new(
804            order.trader_id(),
805            order.strategy_id(),
806            order.instrument_id(),
807            order.client_order_id(),
808            UUID4::new(),
809            UnixNanos::default(),
810            UnixNanos::default(),
811            false,
812            Some(current_id),
813            Some(account_id),
814            None,
815        );
816        assert!(state.defer_replace_cancel(pending));
817        assert_eq!(
818            state.record_order_update(
819                order.client_order_id(),
820                old_id,
821                Quantity::from("2"),
822                Price::from("49000.00"),
823                None
824            ),
825            None
826        );
827        assert_eq!(state.reject_replace(order.client_order_id()), Some(pending));
828    }
829}
nautilus_binance/common/dispatch.rs

nautilus_binance/common/
dispatch.rs
