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 std::sync::Mutex;
25
26use dashmap::DashMap;
27use nautilus_common::cache::fifo::FifoCache;
28use nautilus_core::{MUTEX_POISONED, UUID4, UnixNanos};
29use nautilus_live::ExecutionEventEmitter;
30use nautilus_model::{
31    enums::{OrderSide, OrderType},
32    events::{OrderAccepted, OrderEventAny},
33    identifiers::{AccountId, ClientOrderId, InstrumentId, StrategyId, VenueOrderId},
34    types::{Price, Quantity},
35};
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/// Order identity context stored at submission time.
58///
59/// Provides the strategy and instrument metadata needed to construct proper
60/// order events without accessing the cache from the async dispatch task.
61#[derive(Debug, Clone)]
62pub struct OrderIdentity {
63    pub instrument_id: InstrumentId,
64    pub strategy_id: StrategyId,
65    pub order_side: OrderSide,
66    pub order_type: OrderType,
67    pub price: Option<Price>,
68    pub quantity: Quantity,
69}
70
71/// Tracks order lifecycle state for dispatch routing.
72///
73/// Orders with a registered identity (submitted through this client) produce
74/// proper order events. Orders without identity (external or pre-existing)
75/// fall back to execution reports for reconciliation.
76#[derive(Debug)]
77pub struct WsDispatchState {
78    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
79    pub pending_requests: DashMap<String, PendingRequest>,
80    emitted_accepted: Mutex<FifoCache<ClientOrderId, 10_000>>,
81    filled_orders: Mutex<FifoCache<ClientOrderId, 10_000>>,
82}
83
84impl Default for WsDispatchState {
85    fn default() -> Self {
86        Self {
87            order_identities: DashMap::new(),
88            pending_requests: DashMap::new(),
89            emitted_accepted: Mutex::new(FifoCache::new()),
90            filled_orders: Mutex::new(FifoCache::new()),
91        }
92    }
93}
94
95#[expect(clippy::missing_panics_doc, reason = "mutex poisoning is not expected")]
96impl WsDispatchState {
97    pub fn has_emitted_accepted(&self, cid: &ClientOrderId) -> bool {
98        self.emitted_accepted
99            .lock()
100            .expect(MUTEX_POISONED)
101            .contains(cid)
102    }
103
104    /// Marks an order as having emitted an OrderAccepted event.
105    pub fn insert_accepted(&self, cid: ClientOrderId) {
106        self.emitted_accepted.lock().expect(MUTEX_POISONED).add(cid);
107    }
108
109    pub fn has_filled(&self, cid: &ClientOrderId) -> bool {
110        self.filled_orders
111            .lock()
112            .expect(MUTEX_POISONED)
113            .contains(cid)
114    }
115
116    /// Marks an order as having received a fill.
117    pub fn insert_filled(&self, cid: ClientOrderId) {
118        self.filled_orders.lock().expect(MUTEX_POISONED).add(cid);
119    }
120
121    /// Removes all tracking state for a terminal order.
122    pub fn cleanup_terminal(&self, cid: ClientOrderId) {
123        self.order_identities.remove(&cid);
124        self.emitted_accepted
125            .lock()
126            .expect(MUTEX_POISONED)
127            .remove(&cid);
128        self.filled_orders
129            .lock()
130            .expect(MUTEX_POISONED)
131            .remove(&cid);
132    }
133}
134
135/// Synthesizes and emits OrderAccepted if one has not yet been emitted.
136///
137/// Handles fast-filling orders that skip the New state on Binance.
138pub fn ensure_accepted_emitted(
139    client_order_id: ClientOrderId,
140    account_id: AccountId,
141    venue_order_id: VenueOrderId,
142    identity: &OrderIdentity,
143    emitter: &ExecutionEventEmitter,
144    state: &WsDispatchState,
145    ts_init: UnixNanos,
146) {
147    if state.has_emitted_accepted(&client_order_id) {
148        return;
149    }
150    state.insert_accepted(client_order_id);
151    let accepted = OrderAccepted::new(
152        emitter.trader_id(),
153        identity.strategy_id,
154        identity.instrument_id,
155        client_order_id,
156        venue_order_id,
157        account_id,
158        UUID4::new(),
159        ts_init,
160        ts_init,
161        false,
162    );
163    emitter.send_order_event(OrderEventAny::Accepted(accepted));
164}
nautilus_binance/common/dispatch.rs

nautilus_binance/common/
dispatch.rs
