nautilus_bybit/common/
instruments.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 definition diffing and emission for the Bybit adapter.
17
18use ahash::{AHashMap, AHashSet};
19use nautilus_common::{live::sender::EventSender, messages::DataEvent};
20use nautilus_model::{
21    identifiers::InstrumentId,
22    instruments::{Instrument, InstrumentAny},
23};
24
25/// Returns `true` if the economically meaningful fields of two instruments differ.
26///
27/// [`InstrumentAny`]'s `PartialEq` compares only the instrument ID, and definitions carry
28/// per-fetch timestamps that always change - so neither is usable for detecting "real" updates.
29/// This compares the fields a strategy prices and sizes against, ignoring timestamps.
30fn economics_differ(a: &InstrumentAny, b: &InstrumentAny) -> bool {
31    a.margin_init() != b.margin_init()
32        || a.margin_maint() != b.margin_maint()
33        || a.price_precision() != b.price_precision()
34        || a.size_precision() != b.size_precision()
35        || a.price_increment() != b.price_increment()
36        || a.size_increment() != b.size_increment()
37        || a.multiplier() != b.multiplier()
38        || a.lot_size() != b.lot_size()
39        || a.min_quantity() != b.min_quantity()
40        || a.max_quantity() != b.max_quantity()
41        || a.min_notional() != b.min_notional()
42        || a.max_notional() != b.max_notional()
43        || a.min_price() != b.min_price()
44        || a.max_price() != b.max_price()
45}
46
47/// Compares a fresh instrument snapshot against cached definitions, emitting [`DataEvent::Data`]
48/// events for new and economically-changed instruments.
49///
50/// The cache is updated to reflect each meaningful change regardless of subscription, while emissions
51/// are gated by `subscriptions`: only subscribed instruments produce events. Pass `None` to emit for
52/// all changes (e.g. a venue-wide subscription).
53pub fn diff_and_emit_instruments(
54    new_instruments: &[InstrumentAny],
55    cached: &mut AHashMap<InstrumentId, InstrumentAny>,
56    subscriptions: Option<&AHashSet<InstrumentId>>,
57    sender: &EventSender<DataEvent>,
58) {
59    let is_subscribed = |id: &InstrumentId| subscriptions.is_none_or(|subs| subs.contains(id));
60
61    for instrument in new_instruments {
62        let id = instrument.id();
63        let changed = cached
64            .get(&id)
65            .is_none_or(|prev| economics_differ(prev, instrument));
66
67        if changed {
68            cached.insert(id, instrument.clone());
69            if is_subscribed(&id)
70                && let Err(e) = sender.send(DataEvent::Data(instrument.clone().into()))
71            {
72                log::error!("Failed to emit instrument event: {e}");
73            }
74        }
75    }
76}
77
78#[cfg(test)]
79mod tests {
80    use nautilus_core::UnixNanos;
81    use nautilus_model::{
82        data::Data,
83        identifiers::{InstrumentId, Symbol},
84        instruments::{CryptoPerpetual, InstrumentAny},
85        types::{Currency, Money, Price, Quantity},
86    };
87    use rstest::rstest;
88    use rust_decimal::Decimal;
89
90    use super::*;
91
92    /// Builds a BTCUSDT linear perp with the given economic fields; everything else is fixed so a
93    /// single varied field is what the diff sees.
94    fn perp(
95        margin_init: Decimal,
96        size_increment: Quantity,
97        min_notional: Option<Money>,
98    ) -> InstrumentAny {
99        InstrumentAny::CryptoPerpetual(
100            CryptoPerpetual::builder()
101                .instrument_id(InstrumentId::from("BTCUSDT-LINEAR.BYBIT"))
102                .raw_symbol(Symbol::from("BTCUSDT-LINEAR"))
103                .base_currency(Currency::BTC())
104                .quote_currency(Currency::USDT())
105                .settlement_currency(Currency::USDT())
106                .is_inverse(false)
107                .price_precision(1)
108                .size_precision(3)
109                .price_increment(Price::from("0.1"))
110                .size_increment(size_increment)
111                .min_quantity(Quantity::from("0.001"))
112                .margin_init(margin_init)
113                .maybe_min_notional(min_notional)
114                .ts_event(UnixNanos::default())
115                .ts_init(UnixNanos::default())
116                .build()
117                .unwrap(),
118        )
119    }
120
121    fn default_perp() -> InstrumentAny {
122        perp(Decimal::ZERO, Quantity::from("0.001"), None)
123    }
124
125    #[rstest]
126    fn test_emits_for_new_instrument() {
127        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
128        let instrument = default_perp();
129        let mut cached = AHashMap::new();
130
131        diff_and_emit_instruments(
132            std::slice::from_ref(&instrument),
133            &mut cached,
134            None,
135            &tx.into(),
136        );
137
138        match rx.try_recv().expect("expected instrument event") {
139            DataEvent::Data(Data::Instrument(emitted)) => assert_eq!(emitted.id(), instrument.id()),
140            _ => panic!("expected Instrument event"),
141        }
142        assert!(cached.contains_key(&instrument.id()));
143    }
144
145    #[rstest]
146    fn test_no_emit_when_unchanged() {
147        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
148        let instrument = default_perp();
149        let mut cached = AHashMap::new();
150        cached.insert(instrument.id(), default_perp());
151
152        diff_and_emit_instruments(&[instrument], &mut cached, None, &tx.into());
153
154        assert!(rx.try_recv().is_err());
155    }
156
157    #[rstest]
158    fn test_emits_on_margin_change() {
159        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
160        let id = default_perp().id();
161        let mut cached = AHashMap::new();
162        cached.insert(id, default_perp());
163
164        // Same instrument, higher initial margin.
165        let updated = perp(Decimal::ONE, Quantity::from("0.001"), None);
166        diff_and_emit_instruments(&[updated], &mut cached, None, &tx.into());
167
168        match rx
169            .try_recv()
170            .expect("expected instrument event on margin change")
171        {
172            DataEvent::Data(Data::Instrument(emitted)) => {
173                assert_eq!(emitted.margin_init(), Decimal::ONE);
174            }
175            _ => panic!("expected Instrument event"),
176        }
177    }
178
179    #[rstest]
180    fn test_emits_on_size_increment_and_min_notional_change() {
181        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
182        let id = default_perp().id();
183
184        // size_increment change
185        let mut cached = AHashMap::new();
186        cached.insert(id, default_perp());
187        let bigger_step = perp(Decimal::ZERO, Quantity::from("0.002"), None);
188        diff_and_emit_instruments(&[bigger_step], &mut cached, None, &tx.clone().into());
189        assert!(rx.try_recv().is_ok(), "size_increment change should emit");
190
191        // min_notional change
192        let mut cached = AHashMap::new();
193        cached.insert(id, default_perp());
194        let with_min = perp(
195            Decimal::ZERO,
196            Quantity::from("0.001"),
197            Some(Money::new(5.0, Currency::USDT())),
198        );
199        diff_and_emit_instruments(&[with_min], &mut cached, None, &tx.into());
200        assert!(rx.try_recv().is_ok(), "min_notional change should emit");
201    }
202
203    #[rstest]
204    fn test_subscription_gating_updates_cache_but_only_emits_for_subscribed() {
205        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
206        let id = default_perp().id();
207
208        // Not subscribed: cache still updates, but no event.
209        let empty_subs = AHashSet::new();
210        let mut cached = AHashMap::new();
211        cached.insert(id, default_perp());
212        let updated = perp(Decimal::ONE, Quantity::from("0.001"), None);
213
214        diff_and_emit_instruments(
215            &[updated],
216            &mut cached,
217            Some(&empty_subs),
218            &tx.clone().into(),
219        );
220
221        assert!(rx.try_recv().is_err(), "unsubscribed should not emit");
222        assert_eq!(
223            cached.get(&id).unwrap().margin_init(),
224            Decimal::ONE,
225            "cache should update regardless of subscription"
226        );
227
228        // Subscribed: emits.
229        let mut subs = AHashSet::new();
230        subs.insert(id);
231        let mut cached = AHashMap::new();
232        cached.insert(id, default_perp());
233        let updated = perp(Decimal::ONE, Quantity::from("0.001"), None);
234
235        diff_and_emit_instruments(&[updated], &mut cached, Some(&subs), &tx.into());
236
237        assert!(rx.try_recv().is_ok(), "subscribed should emit");
238    }
239}
nautilus_bybit/common/instruments.rs

nautilus_bybit/common/
instruments.rs
