nautilus_common/
providers.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 provider trait and shared instrument storage.
17//!
18//! Defines the [`InstrumentProvider`] trait for loading instrument definitions
19//! from venue APIs, and the [`InstrumentStore`] struct for caching them locally.
20
21use std::collections::HashMap;
22
23use ahash::RandomState;
24use async_trait::async_trait;
25use indexmap::IndexMap;
26use nautilus_model::{
27    identifiers::InstrumentId,
28    instruments::{Instrument, InstrumentAny},
29};
30
31/// Local instrument storage with initialization tracking.
32///
33/// Provides `add`/`find`/`get_all` operations for instrument caching.
34/// Not thread-safe by itself; wrap in `Arc<RwLock<InstrumentStore>>` when
35/// sharing across async tasks or WebSocket handlers.
36///
37/// Storage preserves insertion order because adapters publish instruments to the data
38/// engine straight from this store, and that emission sequence must hold across runs.
39/// The map keeps the `ahash` hasher so ordered iteration costs nothing on lookup.
40#[derive(Debug, Default)]
41pub struct InstrumentStore {
42    instruments: IndexMap<InstrumentId, InstrumentAny, RandomState>,
43    initialized: bool,
44}
45
46impl InstrumentStore {
47    /// Creates a new empty instrument store.
48    #[must_use]
49    pub fn new() -> Self {
50        Self::default()
51    }
52
53    /// Adds an instrument to the store, replacing any existing entry with the same ID.
54    pub fn add(&mut self, instrument: InstrumentAny) {
55        self.instruments.insert(instrument.id(), instrument);
56    }
57
58    /// Adds multiple instruments to the store.
59    pub fn add_bulk(&mut self, instruments: Vec<InstrumentAny>) {
60        for instrument in instruments {
61            self.add(instrument);
62        }
63    }
64
65    /// Returns the instrument for the given ID, if found.
66    #[must_use]
67    pub fn find(&self, instrument_id: &InstrumentId) -> Option<&InstrumentAny> {
68        self.instruments.get(instrument_id)
69    }
70
71    /// Returns whether the store contains the given instrument ID.
72    #[must_use]
73    pub fn contains(&self, instrument_id: &InstrumentId) -> bool {
74        self.instruments.contains_key(instrument_id)
75    }
76
77    /// Returns all instruments as a map keyed by instrument ID.
78    #[must_use]
79    pub fn get_all(&self) -> &IndexMap<InstrumentId, InstrumentAny, RandomState> {
80        &self.instruments
81    }
82
83    /// Returns all instruments as a vector.
84    #[must_use]
85    pub fn list_all(&self) -> Vec<&InstrumentAny> {
86        self.instruments.values().collect()
87    }
88
89    /// Returns the number of instruments in the store.
90    #[must_use]
91    pub fn count(&self) -> usize {
92        self.instruments.len()
93    }
94
95    /// Returns whether the store is empty.
96    #[must_use]
97    pub fn is_empty(&self) -> bool {
98        self.instruments.is_empty()
99    }
100
101    /// Returns whether the store has been marked as initialized.
102    #[must_use]
103    pub fn is_initialized(&self) -> bool {
104        self.initialized
105    }
106
107    /// Marks the store as initialized.
108    pub fn set_initialized(&mut self) {
109        self.initialized = true;
110    }
111
112    /// Clears all instruments and resets initialization state.
113    pub fn clear(&mut self) {
114        self.instruments.clear();
115        self.initialized = false;
116    }
117}
118
119/// Provides instrument definitions from a venue.
120///
121/// Implementations define how instruments are fetched from a venue API.
122/// The `store()` / `store_mut()` accessors expose the underlying
123/// [`InstrumentStore`] so that callers can query cached instruments.
124///
125/// # Thread safety
126///
127/// Provider instances are not intended to be sent across threads. The `?Send`
128/// bound allows implementations to hold non-Send state for Python interop.
129#[async_trait(?Send)]
130pub trait InstrumentProvider {
131    /// Returns a reference to the provider's instrument store.
132    fn store(&self) -> &InstrumentStore;
133
134    /// Returns a mutable reference to the provider's instrument store.
135    fn store_mut(&mut self) -> &mut InstrumentStore;
136
137    /// Loads all available instruments from the venue.
138    ///
139    /// Implementations should populate the store via `store_mut().add()`.
140    ///
141    /// # Errors
142    ///
143    /// Returns an error if the loading operation fails.
144    async fn load_all(&mut self, filters: Option<&HashMap<String, String>>) -> anyhow::Result<()>;
145
146    /// Loads specific instruments by their IDs.
147    ///
148    /// The default implementation calls [`load`](Self::load) for each ID
149    /// sequentially. Adapters with batch APIs should override this.
150    ///
151    /// # Errors
152    ///
153    /// Returns an error if any instrument fails to load.
154    async fn load_ids(
155        &mut self,
156        instrument_ids: &[InstrumentId],
157        filters: Option<&HashMap<String, String>>,
158    ) -> anyhow::Result<()> {
159        for instrument_id in instrument_ids {
160            self.load(instrument_id, filters).await?;
161        }
162        Ok(())
163    }
164
165    /// Loads a single instrument by its ID.
166    ///
167    /// # Errors
168    ///
169    /// Returns an error if the loading operation fails.
170    async fn load(
171        &mut self,
172        instrument_id: &InstrumentId,
173        filters: Option<&HashMap<String, String>>,
174    ) -> anyhow::Result<()>;
175}
176
177#[cfg(test)]
178mod tests {
179    use nautilus_model::instruments::{InstrumentAny, stubs::crypto_perpetual_ethusdt};
180    use rstest::rstest;
181
182    use super::*;
183
184    #[rstest]
185    fn test_instrument_store_default_is_empty() {
186        let store = InstrumentStore::new();
187        assert!(store.is_empty());
188        assert_eq!(store.count(), 0);
189        assert!(!store.is_initialized());
190    }
191
192    #[rstest]
193    fn test_instrument_store_add_and_find() {
194        let mut store = InstrumentStore::new();
195        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
196        let id = instrument.id();
197
198        store.add(instrument);
199
200        assert_eq!(store.count(), 1);
201        assert!(!store.is_empty());
202        assert!(store.contains(&id));
203        assert!(store.find(&id).is_some());
204    }
205
206    #[rstest]
207    fn test_instrument_store_add_bulk() {
208        let mut store = InstrumentStore::new();
209        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
210        let id = instrument.id();
211
212        store.add_bulk(vec![instrument]);
213
214        assert_eq!(store.count(), 1);
215        assert!(store.contains(&id));
216    }
217
218    #[rstest]
219    fn test_instrument_store_get_all() {
220        let mut store = InstrumentStore::new();
221        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
222
223        store.add(instrument);
224
225        let all = store.get_all();
226        assert_eq!(all.len(), 1);
227    }
228
229    #[rstest]
230    fn test_instrument_store_list_all() {
231        let mut store = InstrumentStore::new();
232        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
233
234        store.add(instrument);
235
236        let list = store.list_all();
237        assert_eq!(list.len(), 1);
238    }
239
240    #[rstest]
241    fn test_instrument_store_iterates_in_insertion_order() {
242        let mut store = InstrumentStore::new();
243        let base = crypto_perpetual_ethusdt();
244
245        // Insertion order is neither sorted nor reverse sorted, and six entries leave a
246        // 1-in-720 chance that hash-ordered iteration coincides with it.
247        let ids = [
248            "SOLUSDT-PERP.BINANCE",
249            "ADAUSDT-PERP.BINANCE",
250            "XRPUSDT-PERP.BINANCE",
251            "BTCUSDT-PERP.BINANCE",
252            "DOTUSDT-PERP.BINANCE",
253            "AVAXUSDT-PERP.BINANCE",
254        ];
255        let expected: Vec<InstrumentId> = ids.iter().map(|id| InstrumentId::from(*id)).collect();
256
257        for id in ids {
258            let mut variant = base.clone();
259            variant.id = InstrumentId::from(id);
260            store.add(InstrumentAny::CryptoPerpetual(variant));
261        }
262
263        let keys: Vec<InstrumentId> = store.get_all().keys().copied().collect();
264        let listed: Vec<InstrumentId> = store.list_all().into_iter().map(Instrument::id).collect();
265
266        assert_eq!(keys, expected);
267        assert_eq!(listed, expected);
268    }
269
270    #[rstest]
271    fn test_instrument_store_clear() {
272        let mut store = InstrumentStore::new();
273        let instrument = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
274
275        store.add(instrument);
276        store.set_initialized();
277        assert!(store.is_initialized());
278        assert_eq!(store.count(), 1);
279
280        store.clear();
281        assert!(!store.is_initialized());
282        assert!(store.is_empty());
283    }
284
285    #[rstest]
286    fn test_instrument_store_find_missing_returns_none() {
287        let store = InstrumentStore::new();
288        let id = InstrumentId::from("UNKNOWN-UNKNOWN.VENUE");
289        assert!(store.find(&id).is_none());
290        assert!(!store.contains(&id));
291    }
292
293    #[rstest]
294    fn test_instrument_store_add_replaces_existing() {
295        let mut store = InstrumentStore::new();
296        let instrument1 = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
297        let instrument2 = InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt());
298        let id = instrument1.id();
299
300        store.add(instrument1);
301        store.add(instrument2);
302
303        assert_eq!(store.count(), 1);
304        assert!(store.contains(&id));
305    }
306}
nautilus_common/providers.rs

nautilus_common/
providers.rs
