nautilus_binance/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 Binance adapter.
17
18use ahash::AHashMap;
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 crate::spot::sbe::generated::symbol_status::SymbolStatus;
28
29impl From<SymbolStatus> for MarketStatusAction {
30    fn from(status: SymbolStatus) -> Self {
31        match status {
32            SymbolStatus::Trading => Self::Trading,
33            SymbolStatus::EndOfDay => Self::Close,
34            SymbolStatus::Halt => Self::Halt,
35            SymbolStatus::Break => Self::Pause,
36            SymbolStatus::CancelOnly => Self::Halt,
37            SymbolStatus::NonRepresentable | SymbolStatus::NullVal => Self::NotAvailableForTrading,
38        }
39    }
40}
41
42/// Compares new status snapshot against cached state, emitting [`InstrumentStatus`]
43/// events for changes and removals.
44///
45/// Symbols present in the cache but absent from the new snapshot are treated as
46/// removed and emit `NotAvailableForTrading`.
47pub fn diff_and_emit_statuses(
48    new_statuses: &AHashMap<InstrumentId, MarketStatusAction>,
49    cached_statuses: &mut AHashMap<InstrumentId, MarketStatusAction>,
50    sender: &EventSender<DataEvent>,
51    ts_event: UnixNanos,
52    ts_init: UnixNanos,
53) {
54    for (instrument_id, &new_action) in new_statuses {
55        let changed = cached_statuses
56            .get(instrument_id)
57            .is_none_or(|&prev| prev != new_action);
58
59        if changed {
60            cached_statuses.insert(*instrument_id, new_action);
61            emit_status(sender, *instrument_id, new_action, ts_event, ts_init);
62        }
63    }
64
65    // Detect symbols removed from the exchange info snapshot
66    let removed: Vec<InstrumentId> = cached_statuses
67        .keys()
68        .filter(|id| !new_statuses.contains_key(id))
69        .copied()
70        .collect();
71
72    for instrument_id in removed {
73        cached_statuses.remove(&instrument_id);
74        emit_status(
75            sender,
76            instrument_id,
77            MarketStatusAction::NotAvailableForTrading,
78            ts_event,
79            ts_init,
80        );
81    }
82}
83
84fn emit_status(
85    sender: &EventSender<DataEvent>,
86    instrument_id: InstrumentId,
87    action: MarketStatusAction,
88    ts_event: UnixNanos,
89    ts_init: UnixNanos,
90) {
91    let is_trading = Some(matches!(action, MarketStatusAction::Trading));
92    let status = InstrumentStatus::new(
93        instrument_id,
94        action,
95        ts_event,
96        ts_init,
97        None,
98        None,
99        is_trading,
100        None,
101        None,
102    );
103
104    if let Err(e) = sender.send(DataEvent::Data(Data::InstrumentStatus(status))) {
105        log::error!("Failed to emit instrument status event: {e}");
106    }
107}
108
109#[cfg(test)]
110mod tests {
111    use nautilus_model::identifiers::InstrumentId;
112    use rstest::rstest;
113
114    use super::{
115        super::enums::{BinanceContractStatus, BinanceTradingStatus},
116        *,
117    };
118
119    #[rstest]
120    #[case(SymbolStatus::Trading, MarketStatusAction::Trading)]
121    #[case(SymbolStatus::EndOfDay, MarketStatusAction::Close)]
122    #[case(SymbolStatus::Halt, MarketStatusAction::Halt)]
123    #[case(SymbolStatus::Break, MarketStatusAction::Pause)]
124    #[case(SymbolStatus::CancelOnly, MarketStatusAction::Halt)]
125    #[case(
126        SymbolStatus::NonRepresentable,
127        MarketStatusAction::NotAvailableForTrading
128    )]
129    #[case(SymbolStatus::NullVal, MarketStatusAction::NotAvailableForTrading)]
130    fn test_symbol_status_to_market_action(
131        #[case] input: SymbolStatus,
132        #[case] expected: MarketStatusAction,
133    ) {
134        assert_eq!(MarketStatusAction::from(input), expected);
135    }
136
137    #[rstest]
138    #[case(BinanceTradingStatus::Trading, MarketStatusAction::Trading)]
139    #[case(BinanceTradingStatus::PendingTrading, MarketStatusAction::PreOpen)]
140    #[case(BinanceTradingStatus::PreTrading, MarketStatusAction::PreOpen)]
141    #[case(BinanceTradingStatus::PostTrading, MarketStatusAction::PostClose)]
142    #[case(BinanceTradingStatus::EndOfDay, MarketStatusAction::Close)]
143    #[case(BinanceTradingStatus::Halt, MarketStatusAction::Halt)]
144    #[case(BinanceTradingStatus::AuctionMatch, MarketStatusAction::Cross)]
145    #[case(BinanceTradingStatus::Break, MarketStatusAction::Pause)]
146    #[case(BinanceTradingStatus::PreDelivering, MarketStatusAction::PreClose)]
147    #[case(BinanceTradingStatus::Delivering, MarketStatusAction::Close)]
148    #[case(BinanceTradingStatus::Delivered, MarketStatusAction::Close)]
149    #[case(BinanceTradingStatus::PreSettle, MarketStatusAction::PreClose)]
150    #[case(BinanceTradingStatus::Settling, MarketStatusAction::Close)]
151    #[case(BinanceTradingStatus::Close, MarketStatusAction::Close)]
152    #[case(BinanceTradingStatus::TradingHalt, MarketStatusAction::Halt)]
153    #[case(BinanceTradingStatus::TradingCancelOnly, MarketStatusAction::Halt)]
154    #[case(
155        BinanceTradingStatus::Unknown,
156        MarketStatusAction::NotAvailableForTrading
157    )]
158    fn test_trading_status_to_market_action(
159        #[case] input: BinanceTradingStatus,
160        #[case] expected: MarketStatusAction,
161    ) {
162        assert_eq!(MarketStatusAction::from(input), expected);
163    }
164
165    #[rstest]
166    #[case(BinanceContractStatus::Trading, MarketStatusAction::Trading)]
167    #[case(BinanceContractStatus::PendingTrading, MarketStatusAction::PreOpen)]
168    #[case(BinanceContractStatus::PreDelivering, MarketStatusAction::PreClose)]
169    #[case(BinanceContractStatus::Delivering, MarketStatusAction::Close)]
170    #[case(BinanceContractStatus::Delivered, MarketStatusAction::Close)]
171    #[case(BinanceContractStatus::PreSettle, MarketStatusAction::PreClose)]
172    #[case(BinanceContractStatus::Settling, MarketStatusAction::Close)]
173    #[case(BinanceContractStatus::Close, MarketStatusAction::Close)]
174    #[case(BinanceContractStatus::TradingHalt, MarketStatusAction::Halt)]
175    #[case(BinanceContractStatus::TradingCancelOnly, MarketStatusAction::Halt)]
176    #[case(BinanceContractStatus::PreDelisting, MarketStatusAction::PreClose)]
177    #[case(BinanceContractStatus::Delisting, MarketStatusAction::Suspend)]
178    #[case(
179        BinanceContractStatus::Down,
180        MarketStatusAction::NotAvailableForTrading
181    )]
182    #[case(
183        BinanceContractStatus::Unknown,
184        MarketStatusAction::NotAvailableForTrading
185    )]
186    fn test_contract_status_to_market_action(
187        #[case] input: BinanceContractStatus,
188        #[case] expected: MarketStatusAction,
189    ) {
190        assert_eq!(MarketStatusAction::from(input), expected);
191    }
192
193    #[rstest]
194    fn test_diff_emits_on_change() {
195        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
196        let id = InstrumentId::from("BTCUSDT.BINANCE");
197
198        let mut cached = AHashMap::new();
199        cached.insert(id, MarketStatusAction::Trading);
200
201        let mut new_statuses = AHashMap::new();
202        new_statuses.insert(id, MarketStatusAction::Halt);
203
204        diff_and_emit_statuses(
205            &new_statuses,
206            &mut cached,
207            &tx.into(),
208            UnixNanos::default(),
209            UnixNanos::default(),
210        );
211
212        let event = rx.try_recv().expect("expected status event");
213        match event {
214            DataEvent::Data(Data::InstrumentStatus(status)) => {
215                assert_eq!(status.instrument_id, id);
216                assert_eq!(status.action, MarketStatusAction::Halt);
217                assert_eq!(status.is_trading, Some(false));
218            }
219            _ => panic!("expected InstrumentStatus event"),
220        }
221
222        assert_eq!(cached.get(&id), Some(&MarketStatusAction::Halt));
223    }
224
225    #[rstest]
226    fn test_diff_no_emit_when_unchanged() {
227        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
228        let id = InstrumentId::from("BTCUSDT.BINANCE");
229
230        let mut cached = AHashMap::new();
231        cached.insert(id, MarketStatusAction::Trading);
232
233        let mut new_statuses = AHashMap::new();
234        new_statuses.insert(id, MarketStatusAction::Trading);
235
236        diff_and_emit_statuses(
237            &new_statuses,
238            &mut cached,
239            &tx.into(),
240            UnixNanos::default(),
241            UnixNanos::default(),
242        );
243
244        rx.try_recv().unwrap_err();
245    }
246
247    #[rstest]
248    fn test_diff_emits_for_new_symbol() {
249        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
250        let id = InstrumentId::from("ETHUSDT.BINANCE");
251
252        let mut cached = AHashMap::new();
253        let mut new_statuses = AHashMap::new();
254        new_statuses.insert(id, MarketStatusAction::Trading);
255
256        diff_and_emit_statuses(
257            &new_statuses,
258            &mut cached,
259            &tx.into(),
260            UnixNanos::default(),
261            UnixNanos::default(),
262        );
263
264        let event = rx.try_recv().expect("expected status event for new symbol");
265        match event {
266            DataEvent::Data(Data::InstrumentStatus(status)) => {
267                assert_eq!(status.instrument_id, id);
268                assert_eq!(status.action, MarketStatusAction::Trading);
269                assert_eq!(status.is_trading, Some(true));
270            }
271            _ => panic!("expected InstrumentStatus event"),
272        }
273    }
274
275    #[rstest]
276    fn test_diff_emits_not_available_for_removed_symbol() {
277        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
278        let id = InstrumentId::from("BTCUSDT.BINANCE");
279
280        let mut cached = AHashMap::new();
281        cached.insert(id, MarketStatusAction::Trading);
282
283        let new_statuses = AHashMap::new(); // Symbol disappeared
284
285        diff_and_emit_statuses(
286            &new_statuses,
287            &mut cached,
288            &tx.into(),
289            UnixNanos::default(),
290            UnixNanos::default(),
291        );
292
293        let event = rx
294            .try_recv()
295            .expect("expected status event for removed symbol");
296
297        match event {
298            DataEvent::Data(Data::InstrumentStatus(status)) => {
299                assert_eq!(status.instrument_id, id);
300                assert_eq!(status.action, MarketStatusAction::NotAvailableForTrading);
301                assert_eq!(status.is_trading, Some(false));
302            }
303            _ => panic!("expected InstrumentStatus event"),
304        }
305
306        assert!(!cached.contains_key(&id));
307    }
308}
nautilus_binance/common/status.rs

nautilus_binance/common/
status.rs
