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

nautilus_binance/common/
dispatch.rs
