nautilus_bybit/common/
status.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//! Instrument status mapping and polling for the Bybit adapter.
17
18use ahash::{AHashMap, AHashSet};
19use nautilus_common::{live::sender::EventSender, messages::DataEvent};
20use nautilus_core::UnixNanos;
21use nautilus_model::{
22    data::InstrumentStatus, enums::MarketStatusAction, identifiers::InstrumentId,
23};
24
25use super::enums::BybitInstrumentStatus;
26
27impl From<BybitInstrumentStatus> for MarketStatusAction {
28    fn from(status: BybitInstrumentStatus) -> Self {
29        match status {
30            BybitInstrumentStatus::PreLaunch | BybitInstrumentStatus::PendingOpen => Self::PreOpen,
31            BybitInstrumentStatus::Trading => Self::Trading,
32            BybitInstrumentStatus::Delivering => Self::PreClose,
33            BybitInstrumentStatus::Closed => Self::Close,
34            BybitInstrumentStatus::Other => Self::NotAvailableForTrading,
35        }
36    }
37}
38
39/// Compares new status snapshot against cached state, emitting [`InstrumentStatus`]
40/// events for changes and removals.
41///
42/// The cache is always updated to reflect the full API state. Emissions are gated
43/// by `subscriptions`: only instruments present in the subscription set produce
44/// events. Pass `None` to emit for all changes unconditionally.
45///
46/// Symbols present in the cache but absent from the new snapshot are treated as
47/// removed and emit `NotAvailableForTrading` (if subscribed).
48pub fn diff_and_emit_statuses(
49    new_statuses: &AHashMap<InstrumentId, MarketStatusAction>,
50    cached_statuses: &mut AHashMap<InstrumentId, MarketStatusAction>,
51    subscriptions: Option<&AHashSet<InstrumentId>>,
52    sender: &EventSender<DataEvent>,
53    ts_event: UnixNanos,
54    ts_init: UnixNanos,
55) {
56    let is_subscribed = |id: &InstrumentId| subscriptions.is_none_or(|subs| subs.contains(id));
57
58    for (instrument_id, &new_action) in new_statuses {
59        let changed = cached_statuses
60            .get(instrument_id)
61            .is_none_or(|&prev| prev != new_action);
62
63        if changed {
64            cached_statuses.insert(*instrument_id, new_action);
65            if is_subscribed(instrument_id) {
66                emit_status(sender, *instrument_id, new_action, ts_event, ts_init);
67            }
68        }
69    }
70
71    // Detect symbols removed from the exchange info snapshot
72    let removed: Vec<InstrumentId> = cached_statuses
73        .keys()
74        .filter(|id| !new_statuses.contains_key(id))
75        .copied()
76        .collect();
77
78    for instrument_id in removed {
79        cached_statuses.remove(&instrument_id);
80        if is_subscribed(&instrument_id) {
81            emit_status(
82                sender,
83                instrument_id,
84                MarketStatusAction::NotAvailableForTrading,
85                ts_event,
86                ts_init,
87            );
88        }
89    }
90}
91
92pub(crate) fn emit_status(
93    sender: &EventSender<DataEvent>,
94    instrument_id: InstrumentId,
95    action: MarketStatusAction,
96    ts_event: UnixNanos,
97    ts_init: UnixNanos,
98) {
99    let is_trading = Some(matches!(action, MarketStatusAction::Trading));
100    let status = InstrumentStatus::new(
101        instrument_id,
102        action,
103        ts_event,
104        ts_init,
105        None,
106        None,
107        is_trading,
108        None,
109        None,
110    );
111
112    if let Err(e) = sender.send(DataEvent::InstrumentStatus(status)) {
113        log::error!("Failed to emit instrument status event: {e}");
114    }
115}
116
117#[cfg(test)]
118mod tests {
119    use nautilus_model::identifiers::InstrumentId;
120    use rstest::rstest;
121
122    use super::*;
123
124    #[rstest]
125    #[case(BybitInstrumentStatus::Trading, MarketStatusAction::Trading)]
126    #[case(BybitInstrumentStatus::PreLaunch, MarketStatusAction::PreOpen)]
127    #[case(BybitInstrumentStatus::PendingOpen, MarketStatusAction::PreOpen)]
128    #[case(BybitInstrumentStatus::Delivering, MarketStatusAction::PreClose)]
129    #[case(BybitInstrumentStatus::Closed, MarketStatusAction::Close)]
130    #[case(
131        BybitInstrumentStatus::Other,
132        MarketStatusAction::NotAvailableForTrading
133    )]
134    fn test_bybit_instrument_status_to_market_action(
135        #[case] input: BybitInstrumentStatus,
136        #[case] expected: MarketStatusAction,
137    ) {
138        assert_eq!(MarketStatusAction::from(input), expected);
139    }
140
141    #[rstest]
142    fn test_diff_emits_on_change() {
143        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
144        let id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
145
146        let mut cached = AHashMap::new();
147        cached.insert(id, MarketStatusAction::Trading);
148
149        let mut new_statuses = AHashMap::new();
150        new_statuses.insert(id, MarketStatusAction::Halt);
151
152        diff_and_emit_statuses(
153            &new_statuses,
154            &mut cached,
155            None,
156            &tx.into(),
157            UnixNanos::default(),
158            UnixNanos::default(),
159        );
160
161        let event = rx.try_recv().expect("expected status event");
162        match event {
163            DataEvent::InstrumentStatus(status) => {
164                assert_eq!(status.instrument_id, id);
165                assert_eq!(status.action, MarketStatusAction::Halt);
166                assert_eq!(status.is_trading, Some(false));
167            }
168            _ => panic!("expected InstrumentStatus event"),
169        }
170
171        assert_eq!(cached.get(&id), Some(&MarketStatusAction::Halt));
172    }
173
174    #[rstest]
175    fn test_diff_no_emit_when_unchanged() {
176        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
177        let id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
178
179        let mut cached = AHashMap::new();
180        cached.insert(id, MarketStatusAction::Trading);
181
182        let mut new_statuses = AHashMap::new();
183        new_statuses.insert(id, MarketStatusAction::Trading);
184
185        diff_and_emit_statuses(
186            &new_statuses,
187            &mut cached,
188            None,
189            &tx.into(),
190            UnixNanos::default(),
191            UnixNanos::default(),
192        );
193
194        rx.try_recv().unwrap_err();
195    }
196
197    #[rstest]
198    fn test_diff_emits_for_new_symbol() {
199        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
200        let id = InstrumentId::from("ETHUSDT-LINEAR.BYBIT");
201
202        let mut cached = AHashMap::new();
203        let mut new_statuses = AHashMap::new();
204        new_statuses.insert(id, MarketStatusAction::Trading);
205
206        diff_and_emit_statuses(
207            &new_statuses,
208            &mut cached,
209            None,
210            &tx.into(),
211            UnixNanos::default(),
212            UnixNanos::default(),
213        );
214
215        let event = rx.try_recv().expect("expected status event for new symbol");
216        match event {
217            DataEvent::InstrumentStatus(status) => {
218                assert_eq!(status.instrument_id, id);
219                assert_eq!(status.action, MarketStatusAction::Trading);
220                assert_eq!(status.is_trading, Some(true));
221            }
222            _ => panic!("expected InstrumentStatus event"),
223        }
224    }
225
226    #[rstest]
227    fn test_diff_emits_not_available_for_removed_symbol() {
228        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
229        let id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
230
231        let mut cached = AHashMap::new();
232        cached.insert(id, MarketStatusAction::Trading);
233
234        let new_statuses = AHashMap::new(); // Symbol disappeared
235
236        diff_and_emit_statuses(
237            &new_statuses,
238            &mut cached,
239            None,
240            &tx.into(),
241            UnixNanos::default(),
242            UnixNanos::default(),
243        );
244
245        let event = rx
246            .try_recv()
247            .expect("expected status event for removed symbol");
248        match event {
249            DataEvent::InstrumentStatus(status) => {
250                assert_eq!(status.instrument_id, id);
251                assert_eq!(status.action, MarketStatusAction::NotAvailableForTrading);
252                assert_eq!(status.is_trading, Some(false));
253            }
254            _ => panic!("expected InstrumentStatus event"),
255        }
256
257        assert!(!cached.contains_key(&id));
258    }
259
260    #[rstest]
261    fn test_diff_subscription_gating_only_emits_for_subscribed() {
262        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
263        let subscribed_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
264        let unsubscribed_id = InstrumentId::from("ETHUSDT-LINEAR.BYBIT");
265
266        let mut subs = AHashSet::new();
267        subs.insert(subscribed_id);
268
269        let mut cached = AHashMap::new();
270        cached.insert(subscribed_id, MarketStatusAction::Trading);
271        cached.insert(unsubscribed_id, MarketStatusAction::Trading);
272
273        // Both change status
274        let mut new_statuses = AHashMap::new();
275        new_statuses.insert(subscribed_id, MarketStatusAction::Halt);
276        new_statuses.insert(unsubscribed_id, MarketStatusAction::Halt);
277
278        diff_and_emit_statuses(
279            &new_statuses,
280            &mut cached,
281            Some(&subs),
282            &tx.into(),
283            UnixNanos::default(),
284            UnixNanos::default(),
285        );
286
287        // Only subscribed instrument emits
288        let event = rx.try_recv().expect("expected status event");
289        match event {
290            DataEvent::InstrumentStatus(status) => {
291                assert_eq!(status.instrument_id, subscribed_id);
292                assert_eq!(status.action, MarketStatusAction::Halt);
293            }
294            _ => panic!("expected InstrumentStatus event"),
295        }
296        assert!(rx.try_recv().is_err(), "should not emit for unsubscribed");
297
298        // But cache is updated for both
299        assert_eq!(cached.get(&subscribed_id), Some(&MarketStatusAction::Halt));
300        assert_eq!(
301            cached.get(&unsubscribed_id),
302            Some(&MarketStatusAction::Halt)
303        );
304    }
305
306    #[rstest]
307    fn test_diff_removal_only_emits_for_subscribed() {
308        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
309        let subscribed_id = InstrumentId::from("BTCUSDT-LINEAR.BYBIT");
310        let unsubscribed_id = InstrumentId::from("ETHUSDT-LINEAR.BYBIT");
311
312        let mut subs = AHashSet::new();
313        subs.insert(subscribed_id);
314
315        let mut cached = AHashMap::new();
316        cached.insert(subscribed_id, MarketStatusAction::Trading);
317        cached.insert(unsubscribed_id, MarketStatusAction::Trading);
318
319        let new_statuses = AHashMap::new(); // Both removed from API
320
321        diff_and_emit_statuses(
322            &new_statuses,
323            &mut cached,
324            Some(&subs),
325            &tx.into(),
326            UnixNanos::default(),
327            UnixNanos::default(),
328        );
329
330        // Only subscribed instrument emits NotAvailableForTrading
331        let event = rx.try_recv().expect("expected removal event");
332        match event {
333            DataEvent::InstrumentStatus(status) => {
334                assert_eq!(status.instrument_id, subscribed_id);
335                assert_eq!(status.action, MarketStatusAction::NotAvailableForTrading);
336            }
337            _ => panic!("expected InstrumentStatus event"),
338        }
339        assert!(rx.try_recv().is_err(), "should not emit for unsubscribed");
340
341        // Both removed from cache
342        assert!(cached.is_empty());
343    }
344}
nautilus_bybit/common/status.rs

nautilus_bybit/common/
status.rs
