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

nautilus_bybit/common/
instruments.rs
