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;
26use nautilus_core::{UUID4, UnixNanos};
27use nautilus_live::ExecutionEventEmitter;
28use nautilus_model::{
29    enums::{OrderSide, OrderType},
30    events::{OrderAccepted, OrderEventAny},
31    identifiers::{AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, VenueOrderId},
32    types::{Price, Quantity},
33};
34use parking_lot::Mutex;
35
36/// The type of operation a pending WS API request represents.
37#[derive(Debug, Clone, Copy)]
38pub enum PendingOperation {
39    Place,
40    Cancel,
41    Modify,
42}
43
44/// A pending WS API request awaiting a response.
45///
46/// Stored in [`WsDispatchState::pending_requests`] after the WS client
47/// returns a request ID. When the venue responds (accepted or rejected),
48/// the pending request is removed and used to emit the correct order event.
49#[derive(Debug, Clone)]
50pub struct PendingRequest {
51    pub client_order_id: ClientOrderId,
52    pub venue_order_id: Option<VenueOrderId>,
53    pub operation: PendingOperation,
54}
55
56/// Order identity context stored at submission time.
57///
58/// Provides the strategy and instrument metadata needed to construct proper
59/// order events without accessing the cache from the async dispatch task.
60#[derive(Debug, Clone)]
61pub struct OrderIdentity {
62    pub instrument_id: InstrumentId,
63    pub strategy_id: StrategyId,
64    pub order_side: OrderSide,
65    pub order_type: OrderType,
66    pub price: Option<Price>,
67    pub quantity: Quantity,
68    pub venue_position_id: Option<PositionId>,
69}
70
71#[derive(Debug, Clone, Copy)]
72struct AlgoOrderIds {
73    algo: VenueOrderId,
74    current: VenueOrderId,
75}
76
77/// Tracks order lifecycle state for dispatch routing.
78///
79/// Orders with a registered identity (submitted through this client) produce
80/// proper order events. Orders without identity (external or pre-existing)
81/// fall back to execution reports for reconciliation.
82#[derive(Debug)]
83pub struct WsDispatchState {
84    pub order_identities: DashMap<ClientOrderId, OrderIdentity>,
85    pub pending_requests: DashMap<String, PendingRequest>,
86    algo_order_ids: DashMap<ClientOrderId, AlgoOrderIds>,
87    emitted_accepted: Mutex<FifoCache<ClientOrderId, 10_000>>,
88    filled_orders: Mutex<FifoCache<ClientOrderId, 10_000>>,
89}
90
91impl Default for WsDispatchState {
92    fn default() -> Self {
93        Self {
94            order_identities: DashMap::new(),
95            pending_requests: DashMap::new(),
96            algo_order_ids: DashMap::new(),
97            emitted_accepted: Mutex::new(FifoCache::new()),
98            filled_orders: Mutex::new(FifoCache::new()),
99        }
100    }
101}
102
103impl WsDispatchState {
104    pub fn has_emitted_accepted(&self, cid: &ClientOrderId) -> bool {
105        self.emitted_accepted.lock().contains(cid)
106    }
107
108    /// Marks an order as having emitted an OrderAccepted event.
109    pub fn insert_accepted(&self, cid: ClientOrderId) {
110        self.emitted_accepted.lock().add(cid);
111    }
112
113    pub fn has_filled(&self, cid: &ClientOrderId) -> bool {
114        self.filled_orders.lock().contains(cid)
115    }
116
117    /// Marks an order as having received a fill.
118    pub fn insert_filled(&self, cid: ClientOrderId) {
119        self.filled_orders.lock().add(cid);
120    }
121
122    pub fn insert_algo_order_id(&self, cid: ClientOrderId, venue_order_id: VenueOrderId) {
123        self.algo_order_ids.entry(cid).or_insert(AlgoOrderIds {
124            algo: venue_order_id,
125            current: venue_order_id,
126        });
127    }
128
129    /// Promotes a known Algo order to its matching-engine venue order ID.
130    ///
131    /// Returns `None` for an unknown Algo order, `Some(true)` for a new ID, and
132    /// `Some(false)` when the ID was already current.
133    pub fn promote_algo_order_id(
134        &self,
135        cid: ClientOrderId,
136        venue_order_id: VenueOrderId,
137    ) -> Option<bool> {
138        let mut ids = self.algo_order_ids.get_mut(&cid)?;
139        let changed = ids.current != venue_order_id;
140        ids.current = venue_order_id;
141        Some(changed)
142    }
143
144    /// Returns the matching-engine venue order ID for a promoted Algo order.
145    pub fn promoted_algo_order_id(&self, cid: &ClientOrderId) -> Option<VenueOrderId> {
146        self.algo_order_ids
147            .get(cid)
148            .and_then(|ids| (ids.current != ids.algo).then_some(ids.current))
149    }
150
151    /// Removes all tracking state for a terminal order.
152    pub fn cleanup_terminal(&self, cid: ClientOrderId) {
153        self.order_identities.remove(&cid);
154        self.algo_order_ids.remove(&cid);
155        self.emitted_accepted.lock().remove(&cid);
156        self.filled_orders.lock().remove(&cid);
157    }
158}
159
160/// Synthesizes and emits OrderAccepted if one has not yet been emitted.
161///
162/// Handles fast-filling orders that skip the New state on Binance.
163pub fn ensure_accepted_emitted(
164    client_order_id: ClientOrderId,
165    account_id: AccountId,
166    venue_order_id: VenueOrderId,
167    identity: &OrderIdentity,
168    emitter: &ExecutionEventEmitter,
169    state: &WsDispatchState,
170    ts_init: UnixNanos,
171) {
172    if state.has_emitted_accepted(&client_order_id) {
173        return;
174    }
175    state.insert_accepted(client_order_id);
176    let accepted = OrderAccepted::new(
177        emitter.trader_id(),
178        identity.strategy_id,
179        identity.instrument_id,
180        client_order_id,
181        venue_order_id,
182        account_id,
183        UUID4::new(),
184        ts_init,
185        ts_init,
186        false,
187    );
188    emitter.send_order_event(OrderEventAny::Accepted(accepted));
189}
nautilus_binance/common/dispatch.rs

nautilus_binance/common/
dispatch.rs
