nautilus_betfair/stream/
ocm.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//! Shared OCM stream handler state.
17
18use ahash::{AHashMap, AHashSet};
19use nautilus_model::identifiers::{ClientOrderId, StrategyId};
20use rust_decimal::Decimal;
21
22use crate::{
23    common::{
24        parse::{make_customer_order_ref, make_customer_order_ref_legacy},
25        types::OrderSyncEntry,
26    },
27    stream::parse::FillTracker,
28};
29
30/// Shared mutable state for the OCM stream handler.
31///
32/// Accessed by both the TCP reader closure and the execution client methods
33/// (submit, modify, connect/disconnect). All access goes through `Arc<Mutex<>>`.
34#[derive(Debug, Default)]
35pub struct OcmState {
36    pub fill_tracker: FillTracker,
37    /// Maps customer_order_ref (rfo) to ClientOrderId for stream resolution.
38    pub customer_order_refs: AHashMap<String, ClientOrderId>,
39    /// Maps client_order_id to submitting strategy. Captured at submit so the stream task
40    /// builds direct events for tracked orders without cache access.
41    pub order_strategies: AHashMap<ClientOrderId, StrategyId>,
42    /// Client order IDs that already had an `OrderAccepted` emitted (via the HTTP
43    /// place response or stream synthesis), so acceptance is applied exactly once.
44    pub accepted_orders: AHashSet<ClientOrderId>,
45    /// Client order IDs that already received an OCM order status update.
46    pub stream_reported_client_orders: AHashSet<ClientOrderId>,
47    /// Bet IDs that have received a terminal event (cancel, lapse, fill-complete).
48    pub terminal_orders: AHashSet<String>,
49    /// Old bet IDs from replace operations, to suppress late stream updates.
50    pub replaced_venue_order_ids: AHashSet<String>,
51    /// (client_order_id, old_bet_id) pairs for in-flight replace operations.
52    pub pending_update_keys: AHashSet<(ClientOrderId, String)>,
53}
54
55impl OcmState {
56    /// Registers a customer_order_ref mapping for a new order.
57    pub fn register_customer_order_ref(&mut self, client_order_id: ClientOrderId) {
58        let rfo = make_customer_order_ref(client_order_id.as_str());
59        self.customer_order_refs.insert(rfo, client_order_id);
60    }
61
62    /// Registers both current and legacy customer_order_ref truncations.
63    ///
64    /// Used during reconnect sync for pre-existing orders that may
65    /// have been placed with either truncation format.
66    pub fn register_customer_order_ref_with_legacy(&mut self, client_order_id: ClientOrderId) {
67        let rfo = make_customer_order_ref(client_order_id.as_str());
68        let rfo_legacy = make_customer_order_ref_legacy(client_order_id.as_str());
69        self.customer_order_refs.insert(rfo, client_order_id);
70
71        if rfo_legacy != client_order_id.as_str() {
72            self.customer_order_refs.insert(rfo_legacy, client_order_id);
73        }
74    }
75
76    /// Records the submitting strategy for a tracked order.
77    pub fn register_order_identity(
78        &mut self,
79        client_order_id: ClientOrderId,
80        strategy_id: StrategyId,
81    ) {
82        self.order_strategies.insert(client_order_id, strategy_id);
83    }
84
85    /// Returns the submitting strategy for a tracked order, if known.
86    pub fn order_strategy_id(&self, client_order_id: &ClientOrderId) -> Option<StrategyId> {
87        self.order_strategies.get(client_order_id).copied()
88    }
89
90    /// Records that acceptance has been emitted for a tracked order.
91    ///
92    /// Returns `true` when this call newly marks the order accepted (the caller
93    /// should emit `OrderAccepted`), or `false` when acceptance was already emitted.
94    pub fn mark_accepted(&mut self, client_order_id: ClientOrderId) -> bool {
95        self.accepted_orders.insert(client_order_id)
96    }
97
98    /// Removes customer_order_ref mappings for a client_order_id.
99    pub fn remove_customer_order_refs(&mut self, client_order_id: &ClientOrderId) {
100        let rfo = make_customer_order_ref(client_order_id.as_str());
101        let rfo_legacy = make_customer_order_ref_legacy(client_order_id.as_str());
102        self.customer_order_refs.remove(&rfo);
103        self.customer_order_refs.remove(&rfo_legacy);
104        self.order_strategies.remove(client_order_id);
105        self.accepted_orders.remove(client_order_id);
106    }
107
108    /// Resolves a client_order_id from the unmatched order's rfo field.
109    pub fn resolve_client_order_id(&self, rfo: Option<&str>) -> Option<ClientOrderId> {
110        rfo.and_then(|r| self.customer_order_refs.get(r).copied())
111    }
112
113    /// Returns `true` if a cancel/lapse for this bet should be suppressed
114    /// because a replace operation is pending or the bet was already replaced.
115    pub fn should_suppress_cancel(&self, client_order_id: &ClientOrderId, bet_id: &str) -> bool {
116        if self.replaced_venue_order_ids.contains(bet_id) {
117            return true;
118        }
119
120        self.pending_update_keys
121            .contains(&(*client_order_id, bet_id.to_string()))
122    }
123
124    /// Cleans up customer_order_ref mappings for a terminal order,
125    /// unless a pending replace exists for this client_order_id.
126    pub fn cleanup_terminal_order(&mut self, client_order_id: &ClientOrderId) {
127        let has_pending = self
128            .pending_update_keys
129            .iter()
130            .any(|(cid, _)| cid == client_order_id);
131
132        if !has_pending {
133            self.remove_customer_order_refs(client_order_id);
134        }
135    }
136
137    /// Anchors the fill tracker against cached orders so the post-reconnect
138    /// image neither treats cumulative size as a new fill nor re-emits a
139    /// fill that was published via another channel.
140    pub fn sync_from_orders(&mut self, orders: &[OrderSyncEntry]) {
141        for entry in orders {
142            if entry.is_closed {
143                self.terminal_orders.insert(entry.bet_id.clone());
144            } else {
145                self.register_customer_order_ref_with_legacy(entry.client_order_id);
146            }
147
148            if entry.filled_qty > Decimal::ZERO {
149                self.fill_tracker
150                    .sync_order(&entry.bet_id, entry.filled_qty, entry.avg_px);
151            }
152
153            if !entry.trade_ids.is_empty() {
154                self.fill_tracker
155                    .seed_published_trade_ids(entry.trade_ids.iter().cloned());
156            }
157        }
158    }
159}
nautilus_betfair/stream/ocm.rs

nautilus_betfair/stream/
ocm.rs
