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 have their identity registered in
19//! [`WsDispatchState`]. When user data stream messages arrive, the dispatch
20//! function checks for a registered identity:
21//! - Tracked orders produce proper order events (OrderAccepted, OrderFilled, etc.).
22//! - Untracked orders fall back to execution reports for reconciliation.
23
24use dashmap::DashMap;
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::{AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, VenueOrderId},
32    reports::OrderStatusReport,
33    types::{Price, Quantity},
34};
35use parking_lot::Mutex;
36
37/// The type of operation a pending WS API request represents.
38#[derive(Debug, Clone, Copy)]
39pub enum PendingOperation {
40    Place,
41    Cancel,
42    Modify,
43}
44
45/// A pending WS API request awaiting a response.
46///
47/// Stored in [`WsDispatchState::pending_requests`] after the WS client
48/// returns a request ID. When the venue responds (accepted or rejected),
49/// the pending request is removed and used to emit the correct order event.
50#[derive(Debug, Clone)]
51pub struct PendingRequest {
52    pub client_order_id: ClientOrderId,
53    pub venue_order_id: Option<VenueOrderId>,
54    pub operation: PendingOperation,
55}
56
57/// Outcome of a cancel-replace request, tracked by its `cancelNewClientOrderId`.
58///
59/// The cancel half's `CANCELED` report and the venue's response arrive in either
60/// order, so whichever comes second completes the picture.
61#[derive(Debug, Clone)]
62enum CancelReplaceOutcome {
63    Pending,
64    /// The cancel report arrived; withheld while the replacement is pending or succeeded.
65    ///
66    /// Parsed at that point so it can be replayed even for an order without a
67    /// dispatch identity, such as one recovered after restart.
68    Canceled(Box<OrderStatusReport>),
69    /// The venue rejected the request before any cancel report arrived.
70    Rejected,
71}
72
73/// Order identity context stored at submission time.
74///
75/// Provides the strategy and instrument metadata needed to construct proper
76/// order events without accessing the cache from the async dispatch task.
77/// The quantity and quote-sizing flag retain the submitted intent.
78#[derive(Debug, Clone)]
79pub struct OrderIdentity {
80    pub instrument_id: InstrumentId,
81    pub strategy_id: StrategyId,
82    pub order_side: OrderSide,
83    pub order_type: OrderType,
84    pub price: Option<Price>,
85    pub quantity: Quantity,
86    pub is_quote_quantity: bool,
87    pub venue_position_id: Option<PositionId>,
88}
89
90#[derive(Debug, Clone, Copy)]
91struct AlgoOrderIds {
92    algo: VenueOrderId,
93    current: VenueOrderId,
94}
95
96/// Tracks order lifecycle state for dispatch routing.
97///
98/// Orders with a registered identity (submitted through this client) produce
99/// proper order events. Orders without identity (external or pre-existing)
100/// fall back to execution reports for reconciliation.
101#[derive(Debug)]
102pub struct WsDispatchState {
103    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
104    pub pending_requests: DashMap<String, PendingRequest>,
105    algo_order_ids: DashMap<ClientOrderId, AlgoOrderIds>,
106    order_updates: DashMap<ClientOrderId, OrderUpdate>,
107    replacements: DashMap<ClientOrderId, PendingReplacement>,
108    emitted_accepted: Mutex<FifoCache<ClientOrderId, 10_000>>,
109    filled_orders: Mutex<FifoCache<ClientOrderId, 10_000>>,
110    /// Cancel-replace request IDs mapped to their `cancelNewClientOrderId`.
111    pub cancel_replace_request_ids: DashMap<String, String>,
112    cancel_replace_outcomes: Mutex<FifoCacheMap<String, CancelReplaceOutcome, 10_000>>,
113}
114
115impl Default for WsDispatchState {
116    fn default() -> Self {
117        Self {
118            order_identities: DashMap::new(),
119            pending_requests: DashMap::new(),
120            algo_order_ids: DashMap::new(),
121            order_updates: DashMap::new(),
122            replacements: DashMap::new(),
123            emitted_accepted: Mutex::new(FifoCache::new()),
124            filled_orders: Mutex::new(FifoCache::new()),
125            cancel_replace_request_ids: DashMap::new(),
126            cancel_replace_outcomes: Mutex::new(FifoCacheMap::new()),
127        }
128    }
129}
130
131impl WsDispatchState {
132    pub fn has_emitted_accepted(&self, cid: &ClientOrderId) -> bool {
133        self.emitted_accepted.lock().contains(cid)
134    }
135
136    /// Marks an order as having emitted an OrderAccepted event.
137    pub fn insert_accepted(&self, cid: ClientOrderId) {
138        self.emitted_accepted.lock().add(cid);
139    }
140
141    pub fn has_filled(&self, cid: &ClientOrderId) -> bool {
142        self.filled_orders.lock().contains(cid)
143    }
144
145    /// Marks an order as having received a fill.
146    pub fn insert_filled(&self, cid: ClientOrderId) {
147        self.filled_orders.lock().add(cid);
148    }
149
150    /// Records the `cancelNewClientOrderId` sent with a cancel-replace request.
151    pub fn insert_cancel_replace(&self, cancel_id: String) {
152        self.cancel_replace_outcomes
153            .lock()
154            .insert(cancel_id, CancelReplaceOutcome::Pending);
155    }
156
157    /// Returns `true` when `cancel_id` belongs to a cancel-replace this client issued.
158    pub fn has_cancel_replace(&self, cancel_id: &str) -> bool {
159        self.cancel_replace_outcomes
160            .lock()
161            .contains_key(&cancel_id.to_string())
162    }
163
164    /// Records the cancel half's `CANCELED` report for a cancel-replace request.
165    ///
166    /// Returns `true` when the report must be withheld because the replacement is
167    /// pending or succeeded, and `false` when it should dispatch as a standalone
168    /// cancel because the venue already rejected the replacement or the ID is not
169    /// one this client issued.
170    pub fn on_cancel_replace_canceled(&self, cancel_id: &str, report: OrderStatusReport) -> bool {
171        let mut outcomes = self.cancel_replace_outcomes.lock();
172        let Some(outcome) = outcomes.get_mut(&cancel_id.to_string()) else {
173            return false;
174        };
175
176        match outcome {
177            CancelReplaceOutcome::Pending => {
178                *outcome = CancelReplaceOutcome::Canceled(Box::new(report));
179                true
180            }
181            CancelReplaceOutcome::Canceled(_) => true,
182            CancelReplaceOutcome::Rejected => false,
183        }
184    }
185
186    /// Records a rejected cancel-replace request.
187    ///
188    /// Returns the withheld cancel report when it already arrived, so the caller
189    /// can emit the confirmed cancellation for the original order.
190    pub fn on_cancel_replace_rejected(&self, cancel_id: &str) -> Option<OrderStatusReport> {
191        let mut outcomes = self.cancel_replace_outcomes.lock();
192        let outcome = outcomes.get_mut(&cancel_id.to_string())?;
193
194        match outcome {
195            CancelReplaceOutcome::Pending => {
196                *outcome = CancelReplaceOutcome::Rejected;
197                None
198            }
199            CancelReplaceOutcome::Canceled(report) => Some((**report).clone()),
200            CancelReplaceOutcome::Rejected => None,
201        }
202    }
203
204    pub fn insert_algo_order_id(&self, cid: ClientOrderId, venue_order_id: VenueOrderId) {
205        self.algo_order_ids.entry(cid).or_insert(AlgoOrderIds {
206            algo: venue_order_id,
207            current: venue_order_id,
208        });
209    }
210
211    /// Promotes a known Algo order to its matching-engine venue order ID.
212    ///
213    /// Returns `None` for an unknown Algo order, `Some(true)` for a new ID, and
214    /// `Some(false)` when the ID was already current.
215    pub fn promote_algo_order_id(
216        &self,
217        cid: ClientOrderId,
218        venue_order_id: VenueOrderId,
219    ) -> Option<bool> {
220        let mut ids = self.algo_order_ids.get_mut(&cid)?;
221        let changed = ids.current != venue_order_id;
222        ids.current = venue_order_id;
223        Some(changed)
224    }
225
226    /// Returns the matching-engine venue order ID for a promoted Algo order.
227    pub fn promoted_algo_order_id(&self, cid: &ClientOrderId) -> Option<VenueOrderId> {
228        self.algo_order_ids
229            .get(cid)
230            .and_then(|ids| (ids.current != ids.algo).then_some(ids.current))
231    }
232
233    pub(crate) fn record_order_update(
234        &self,
235        cid: ClientOrderId,
236        venue_order_id: VenueOrderId,
237        quantity: Quantity,
238        price: Price,
239        trigger_price: Option<Price>,
240    ) -> bool {
241        let update = OrderUpdate {
242            venue_order_id,
243            quantity,
244            price,
245            trigger_price,
246        };
247        let changed = self
248            .order_updates
249            .insert(cid, update)
250            .is_none_or(|previous| previous != update);
251        self.replacements
252            .remove_if(&cid, |_, pending| pending.venue_order_id != venue_order_id);
253        changed
254    }
255
256    pub(crate) fn begin_replace(&self, cid: ClientOrderId, venue_order_id: VenueOrderId) {
257        self.replacements.insert(
258            cid,
259            PendingReplacement {
260                venue_order_id,
261                canceled: None,
262            },
263        );
264    }
265
266    pub(crate) fn defer_replace_cancel(&self, canceled: OrderCanceled) -> bool {
267        let cid = canceled.client_order_id;
268
269        if self
270            .order_updates
271            .get(&cid)
272            .is_some_and(|update| Some(update.venue_order_id) != canceled.venue_order_id)
273        {
274            return true;
275        }
276
277        if let Some(mut pending) = self.replacements.get_mut(&cid)
278            && Some(pending.venue_order_id) == canceled.venue_order_id
279        {
280            pending.canceled = Some(canceled);
281            return true;
282        }
283        false
284    }
285
286    pub(crate) fn reject_replace(&self, cid: ClientOrderId) -> Option<OrderCanceled> {
287        self.replacements
288            .remove(&cid)
289            .and_then(|(_, pending)| pending.canceled)
290    }
291
292    /// Removes all tracking state for a terminal order.
293    pub fn cleanup_terminal(&self, cid: ClientOrderId) {
294        self.order_identities.remove(&cid);
295        self.algo_order_ids.remove(&cid);
296        self.order_updates.remove(&cid);
297        self.replacements.remove(&cid);
298        self.emitted_accepted.lock().remove(&cid);
299        self.filled_orders.lock().remove(&cid);
300    }
301}
302
303#[derive(Debug)]
304struct PendingReplacement {
305    venue_order_id: VenueOrderId,
306    canceled: Option<OrderCanceled>,
307}
308
309#[derive(Debug, Clone, Copy, PartialEq, Eq)]
310struct OrderUpdate {
311    venue_order_id: VenueOrderId,
312    quantity: Quantity,
313    price: Price,
314    trigger_price: Option<Price>,
315}
316
317/// Synthesizes and emits OrderAccepted if one has not yet been emitted.
318///
319/// Handles fast-filling orders that skip the New state on Binance.
320pub fn ensure_accepted_emitted(
321    client_order_id: ClientOrderId,
322    account_id: AccountId,
323    venue_order_id: VenueOrderId,
324    identity: &OrderIdentity,
325    emitter: &ExecutionEventEmitter,
326    state: &WsDispatchState,
327    ts_init: UnixNanos,
328) {
329    if state.has_emitted_accepted(&client_order_id) {
330        return;
331    }
332    state.insert_accepted(client_order_id);
333    let accepted = OrderAccepted::new(
334        emitter.trader_id(),
335        identity.strategy_id,
336        identity.instrument_id,
337        client_order_id,
338        venue_order_id,
339        account_id,
340        UUID4::new(),
341        ts_init,
342        ts_init,
343        false,
344    );
345    emitter.send_order_event(OrderEventAny::Accepted(accepted));
346}
nautilus_binance/common/dispatch.rs

nautilus_binance/common/
dispatch.rs
