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

nautilus_bybit/common/
status.rs
