nautilus_dydx/common/
instrument_cache.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//! Thread-safe instrument cache for dYdX adapter.
17//!
18//! This module provides a centralized cache for instrument data that is shared
19//! between HTTP client, WebSocket client, and execution client via `Arc`.
20//!
21//! # Design
22//!
23//! dYdX uses different identifiers in different contexts:
24//! - **InstrumentId** ("BTC-USD-PERP.DYDX"): Nautilus internal identifier (primary key)
25//! - **Market ticker** ("BTC-USD"): Used in public WebSocket channels
26//! - **clob_pair_id** (0, 1, 2...): Used in blockchain transactions and order messages
27//!
28//! This cache provides O(1) lookups by any of these identifiers through internal indices.
29//! Using `InstrumentId` as the primary key provides better type safety and eliminates
30//! redundant conversions.
31//!
32//! # Thread Safety
33//!
34//! All operations use `DashMap` for lock-free concurrent access. The cache can be
35//! safely shared across multiple async tasks via `Arc<InstrumentCache>`.
36
37use std::sync::atomic::{AtomicBool, Ordering};
38
39use dashmap::DashMap;
40use nautilus_model::{
41    identifiers::InstrumentId,
42    instruments::{Instrument, InstrumentAny},
43};
44use ustr::Ustr;
45
46use crate::{grpc::OrderMarketParams, http::models::PerpetualMarket};
47
48/// Thread-safe instrument cache with multiple lookup indices.
49///
50/// Shared between HTTP client, WebSocket client, and execution client via `Arc`.
51/// Provides O(1) lookups by `InstrumentId`, market ticker, or clob_pair_id.
52
53#[derive(Debug, Default)]
54pub struct InstrumentCache {
55    /// Primary storage: InstrumentId → InstrumentAny
56    instruments: DashMap<InstrumentId, InstrumentAny>,
57    /// Index: clob_pair_id (0, 1, 2...) → InstrumentId (direct lookup)
58    clob_pair_id_index: DashMap<u32, InstrumentId>,
59    /// Index: market ticker ("BTC-USD") → InstrumentId (direct lookup)
60    market_index: DashMap<Ustr, InstrumentId>,
61    /// Market parameters: InstrumentId → PerpetualMarket
62    market_params: DashMap<InstrumentId, PerpetualMarket>,
63    /// Whether cache has been initialized with instrument data
64    initialized: AtomicBool,
65}
66
67impl InstrumentCache {
68    /// Creates a new empty instrument cache.
69    #[must_use]
70    pub fn new() -> Self {
71        Self::default()
72    }
73
74    /// Inserts an instrument with its market data.
75    ///
76    /// This populates the primary storage and all lookup indices.
77    pub fn insert(&self, instrument: InstrumentAny, market: PerpetualMarket) {
78        let instrument_id = instrument.id();
79        let ticker = market.ticker;
80        let clob_pair_id = market.clob_pair_id;
81
82        // Primary storage
83        self.instruments.insert(instrument_id, instrument);
84
85        // Build indices for reverse lookups (now point directly to InstrumentId)
86        self.clob_pair_id_index.insert(clob_pair_id, instrument_id);
87        self.market_index.insert(ticker, instrument_id);
88
89        // Store full market params for order building
90        self.market_params.insert(instrument_id, market);
91    }
92
93    /// Bulk inserts instruments with their market data.
94    ///
95    /// Marks the cache as initialized after insertion.
96    pub fn insert_many(&self, items: Vec<(InstrumentAny, PerpetualMarket)>) {
97        for (instrument, market) in items {
98            self.insert(instrument, market);
99        }
100        self.initialized.store(true, Ordering::Release);
101    }
102
103    /// Clears all cached data.
104    ///
105    /// Useful for refreshing instruments from the API.
106    pub fn clear(&self) {
107        self.instruments.clear();
108        self.clob_pair_id_index.clear();
109        self.market_index.clear();
110        self.market_params.clear();
111        self.initialized.store(false, Ordering::Release);
112    }
113
114    /// Inserts an instrument without market data.
115    ///
116    /// Derives the market ticker from the instrument symbol by stripping the
117    /// "-PERP" suffix, so `get_by_market()` works. `get_by_clob_id()` requires
118    /// full market params and won't work for instruments inserted this way.
119    pub fn insert_instrument_only(&self, instrument: InstrumentAny) {
120        let instrument_id = instrument.id();
121        let symbol = instrument_id.symbol.as_str();
122        let ticker = symbol.strip_suffix("-PERP").unwrap_or(symbol);
123        self.market_index.insert(Ustr::from(ticker), instrument_id);
124        self.instruments.insert(instrument_id, instrument);
125    }
126
127    /// Bulk inserts instruments without market data (derives market tickers).
128    ///
129    /// Marks the cache as initialized after insertion.
130    pub fn insert_instruments_only(&self, instruments: Vec<InstrumentAny>) {
131        for instrument in instruments {
132            self.insert_instrument_only(instrument);
133        }
134        self.initialized.store(true, Ordering::Release);
135    }
136
137    /// Gets an instrument by InstrumentId.
138    #[must_use]
139    pub fn get(&self, instrument_id: &InstrumentId) -> Option<InstrumentAny> {
140        self.instruments.get(instrument_id).map(|r| r.clone())
141    }
142
143    /// Gets an instrument by market ticker (e.g., "BTC-USD").
144    ///
145    /// This is the identifier used in public WebSocket channels.
146    #[must_use]
147    pub fn get_by_market(&self, ticker: &str) -> Option<InstrumentAny> {
148        let ticker_ustr = Ustr::from(ticker);
149        self.market_index
150            .get(&ticker_ustr)
151            .and_then(|instrument_id| self.instruments.get(&*instrument_id).map(|r| r.clone()))
152    }
153
154    /// Gets an instrument by clob_pair_id (e.g., 0, 1, 2).
155    ///
156    /// This is the identifier used in blockchain transactions and order messages.
157    #[must_use]
158    pub fn get_by_clob_id(&self, clob_pair_id: u32) -> Option<InstrumentAny> {
159        self.clob_pair_id_index
160            .get(&clob_pair_id)
161            .and_then(|instrument_id| self.instruments.get(&*instrument_id).map(|r| r.clone()))
162    }
163
164    /// Gets an InstrumentId by clob_pair_id.
165    ///
166    /// Returns directly from index without cloning full instrument.
167    #[must_use]
168    pub fn get_id_by_clob_id(&self, clob_pair_id: u32) -> Option<InstrumentId> {
169        self.clob_pair_id_index.get(&clob_pair_id).map(|r| *r)
170    }
171
172    /// Gets an InstrumentId by market ticker.
173    ///
174    /// Returns directly from index without cloning full instrument.
175    #[must_use]
176    pub fn get_id_by_market(&self, ticker: &str) -> Option<InstrumentId> {
177        let ticker_ustr = Ustr::from(ticker);
178        self.market_index.get(&ticker_ustr).map(|r| *r)
179    }
180
181    /// Gets full market parameters by InstrumentId.
182    ///
183    /// Returns the complete `PerpetualMarket` data including margin requirements,
184    /// quantization parameters, and current oracle price.
185    #[must_use]
186    pub fn get_market_params(&self, instrument_id: &InstrumentId) -> Option<PerpetualMarket> {
187        self.market_params.get(instrument_id).map(|r| r.clone())
188    }
189
190    /// Gets order market parameters for order building.
191    ///
192    /// Returns the subset of market data needed for constructing orders
193    /// (quantization, clob_pair_id, etc.).
194    #[must_use]
195    pub fn get_order_market_params(
196        &self,
197        instrument_id: &InstrumentId,
198    ) -> Option<OrderMarketParams> {
199        self.get_market_params(instrument_id)
200            .map(|market| OrderMarketParams {
201                atomic_resolution: market.atomic_resolution,
202                clob_pair_id: market.clob_pair_id,
203                oracle_price: market.oracle_price,
204                quantum_conversion_exponent: market.quantum_conversion_exponent,
205                step_base_quantums: market.step_base_quantums,
206                subticks_per_tick: market.subticks_per_tick,
207            })
208    }
209
210    /// Updates oracle price for a market.
211    ///
212    /// Called when receiving price updates via WebSocket `v4_markets` channel.
213    pub fn update_oracle_price(&self, ticker: &str, oracle_price: rust_decimal::Decimal) {
214        let ticker_ustr = Ustr::from(ticker);
215        if let Some(instrument_id) = self.market_index.get(&ticker_ustr)
216            && let Some(mut market) = self.market_params.get_mut(&*instrument_id)
217        {
218            market.oracle_price = Some(oracle_price);
219        }
220    }
221
222    /// Returns whether the cache has been initialized with instrument data.
223    #[must_use]
224    pub fn is_initialized(&self) -> bool {
225        self.initialized.load(Ordering::Acquire)
226    }
227
228    /// Returns the number of cached instruments.
229    #[must_use]
230    pub fn len(&self) -> usize {
231        self.instruments.len()
232    }
233
234    /// Returns whether the cache is empty.
235    #[must_use]
236    pub fn is_empty(&self) -> bool {
237        self.instruments.is_empty()
238    }
239
240    /// Returns all cached instruments.
241    ///
242    /// Useful for WebSocket handler initialization and instrument replay.
243    #[must_use]
244    pub fn all_instruments(&self) -> Vec<InstrumentAny> {
245        self.instruments.iter().map(|r| r.clone()).collect()
246    }
247
248    /// Returns all InstrumentIds.
249    #[must_use]
250    pub fn all_instrument_ids(&self) -> Vec<InstrumentId> {
251        self.instruments.iter().map(|r| r.value().id()).collect()
252    }
253
254    /// Checks if an instrument exists by InstrumentId.
255    #[must_use]
256    pub fn contains(&self, instrument_id: &InstrumentId) -> bool {
257        self.instruments.contains_key(instrument_id)
258    }
259
260    /// Checks if an instrument exists by clob_pair_id.
261    #[must_use]
262    pub fn contains_clob_id(&self, clob_pair_id: u32) -> bool {
263        self.clob_pair_id_index.contains_key(&clob_pair_id)
264    }
265
266    /// Checks if an instrument exists by market ticker (e.g., "BTC-USD").
267    #[must_use]
268    pub fn contains_market(&self, ticker: &str) -> bool {
269        let ticker_ustr = Ustr::from(ticker);
270        self.market_index.contains_key(&ticker_ustr)
271    }
272
273    /// Returns a HashMap of all instruments keyed by InstrumentId.
274    ///
275    /// This is useful for parsing functions that expect `HashMap<InstrumentId, InstrumentAny>`.
276    /// Note: Creates a snapshot copy, so frequent calls should be avoided.
277    #[must_use]
278    pub fn to_instrument_id_map(&self) -> std::collections::HashMap<InstrumentId, InstrumentAny> {
279        self.instruments
280            .iter()
281            .map(|entry| (entry.value().id(), entry.value().clone()))
282            .collect()
283    }
284
285    /// Returns a HashMap of oracle prices keyed by InstrumentId.
286    ///
287    /// This is useful for parsing functions like `parse_account_state` that need oracle prices.
288    /// Note: Creates a snapshot copy, so frequent calls should be avoided.
289    #[must_use]
290    pub fn to_oracle_prices_map(
291        &self,
292    ) -> std::collections::HashMap<InstrumentId, rust_decimal::Decimal> {
293        self.market_params
294            .iter()
295            .filter_map(|entry| entry.value().oracle_price.map(|p| (*entry.key(), p)))
296            .collect()
297    }
298
299    /// Logs a warning about a missing instrument for a clob_pair_id, listing known mappings.
300    pub fn log_missing_clob_pair_id(&self, clob_pair_id: u32) {
301        let known: Vec<(u32, String)> = self
302            .clob_pair_id_index
303            .iter()
304            .map(|entry| (*entry.key(), entry.value().symbol.as_str().to_string()))
305            .collect();
306
307        log::warn!(
308            "Instrument for clob_pair_id {clob_pair_id} not found in cache. \
309             Known CLOB pair IDs and symbols: {known:?}"
310        );
311    }
312}
313
314#[cfg(test)]
315mod tests {
316    use nautilus_core::UnixNanos;
317    use nautilus_model::{
318        identifiers::{InstrumentId, Symbol},
319        instruments::{CryptoPerpetual, InstrumentAny},
320        types::{Currency, Price, Quantity},
321    };
322    use rstest::rstest;
323    use rust_decimal_macros::dec;
324    use ustr::Ustr;
325
326    use super::*;
327    use crate::common::{consts::DYDX_VENUE, enums::DydxMarketStatus};
328
329    fn create_test_instrument(symbol: &str) -> InstrumentAny {
330        let instrument_id = InstrumentId::new(Symbol::new(symbol), *DYDX_VENUE);
331        InstrumentAny::CryptoPerpetual(
332            CryptoPerpetual::builder()
333                .instrument_id(instrument_id)
334                .raw_symbol(instrument_id.symbol)
335                .base_currency(Currency::BTC())
336                .quote_currency(Currency::USD())
337                .settlement_currency(Currency::USD())
338                .is_inverse(false)
339                .price_precision(1)
340                .size_precision(3)
341                .price_increment(Price::new(0.1, 1))
342                .size_increment(Quantity::new(0.001, 3))
343                .ts_event(UnixNanos::default())
344                .ts_init(UnixNanos::default())
345                .build()
346                .unwrap(),
347        )
348    }
349
350    fn create_test_market(ticker: &str, clob_pair_id: u32) -> PerpetualMarket {
351        PerpetualMarket {
352            clob_pair_id,
353            ticker: Ustr::from(ticker),
354            status: DydxMarketStatus::Active,
355            base_asset: Some(Ustr::from("BTC")),
356            quote_asset: Some(Ustr::from("USD")),
357            step_size: dec!(0.001),
358            tick_size: dec!(0.1),
359            index_price: Some(dec!(50000)),
360            oracle_price: Some(dec!(50000)),
361            price_change_24h: dec!(0),
362            next_funding_rate: dec!(0),
363            next_funding_at: None,
364            min_order_size: Some(dec!(0.001)),
365            market_type: None,
366            initial_margin_fraction: dec!(0.05),
367            maintenance_margin_fraction: dec!(0.03),
368            base_position_notional: None,
369            incremental_position_size: None,
370            incremental_initial_margin_fraction: None,
371            max_position_size: None,
372            open_interest: dec!(1000),
373            atomic_resolution: -10,
374            quantum_conversion_exponent: -9,
375            subticks_per_tick: 1000000,
376            step_base_quantums: 1000000,
377            is_reduce_only: false,
378        }
379    }
380
381    #[rstest]
382    fn test_insert_and_get() {
383        let cache = InstrumentCache::new();
384        let instrument = create_test_instrument("BTC-USD-PERP");
385        let instrument_id = instrument.id();
386        let market = create_test_market("BTC-USD", 0);
387
388        cache.insert(instrument, market);
389
390        // Get by InstrumentId
391        let retrieved = cache.get(&instrument_id);
392        assert!(retrieved.is_some());
393        assert_eq!(retrieved.unwrap().id().symbol.as_str(), "BTC-USD-PERP");
394    }
395
396    #[rstest]
397    fn test_get_by_market() {
398        let cache = InstrumentCache::new();
399        let instrument = create_test_instrument("BTC-USD-PERP");
400        let market = create_test_market("BTC-USD", 0);
401
402        cache.insert(instrument, market);
403
404        // Get by market ticker
405        let retrieved = cache.get_by_market("BTC-USD");
406        assert!(retrieved.is_some());
407        assert_eq!(retrieved.unwrap().id().symbol.as_str(), "BTC-USD-PERP");
408    }
409
410    #[rstest]
411    fn test_get_by_clob_id() {
412        let cache = InstrumentCache::new();
413        let instrument = create_test_instrument("BTC-USD-PERP");
414        let market = create_test_market("BTC-USD", 0);
415
416        cache.insert(instrument, market);
417
418        // Get by clob_pair_id
419        let retrieved = cache.get_by_clob_id(0);
420        assert!(retrieved.is_some());
421        assert_eq!(retrieved.unwrap().id().symbol.as_str(), "BTC-USD-PERP");
422
423        // Non-existent clob_pair_id
424        assert!(cache.get_by_clob_id(999).is_none());
425    }
426
427    #[rstest]
428    fn test_insert_many() {
429        let cache = InstrumentCache::new();
430
431        let items = vec![
432            (
433                create_test_instrument("BTC-USD-PERP"),
434                create_test_market("BTC-USD", 0),
435            ),
436            (
437                create_test_instrument("ETH-USD-PERP"),
438                create_test_market("ETH-USD", 1),
439            ),
440        ];
441
442        assert!(!cache.is_initialized());
443        cache.insert_many(items);
444        assert!(cache.is_initialized());
445
446        assert_eq!(cache.len(), 2);
447        assert!(cache.get_by_market("BTC-USD").is_some());
448        assert!(cache.get_by_market("ETH-USD").is_some());
449        assert!(cache.get_by_clob_id(0).is_some());
450        assert!(cache.get_by_clob_id(1).is_some());
451    }
452
453    #[rstest]
454    fn test_clear() {
455        let cache = InstrumentCache::new();
456        let instrument = create_test_instrument("BTC-USD-PERP");
457        let market = create_test_market("BTC-USD", 0);
458
459        cache.insert(instrument, market);
460        assert_eq!(cache.len(), 1);
461
462        cache.clear();
463        assert_eq!(cache.len(), 0);
464        assert!(!cache.is_initialized());
465    }
466
467    #[rstest]
468    fn test_get_market_params() {
469        let cache = InstrumentCache::new();
470        let instrument = create_test_instrument("BTC-USD-PERP");
471        let market = create_test_market("BTC-USD", 0);
472
473        cache.insert(instrument.clone(), market);
474
475        let params = cache.get_market_params(&instrument.id());
476        assert!(params.is_some());
477        let params = params.unwrap();
478        assert_eq!(params.clob_pair_id, 0);
479        assert_eq!(params.ticker, "BTC-USD");
480    }
481
482    #[rstest]
483    fn test_update_oracle_price() {
484        let cache = InstrumentCache::new();
485        let instrument = create_test_instrument("BTC-USD-PERP");
486        let market = create_test_market("BTC-USD", 0);
487
488        cache.insert(instrument.clone(), market);
489
490        // Initial oracle price
491        let params = cache.get_market_params(&instrument.id()).unwrap();
492        assert_eq!(params.oracle_price, Some(dec!(50000)));
493
494        // Update oracle price
495        cache.update_oracle_price("BTC-USD", dec!(55000));
496
497        let params = cache.get_market_params(&instrument.id()).unwrap();
498        assert_eq!(params.oracle_price, Some(dec!(55000)));
499    }
500
501    #[rstest]
502    fn test_to_oracle_prices_map() {
503        let cache = InstrumentCache::new();
504
505        let items = vec![
506            (
507                create_test_instrument("BTC-USD-PERP"),
508                create_test_market("BTC-USD", 0),
509            ),
510            (
511                create_test_instrument("ETH-USD-PERP"),
512                create_test_market("ETH-USD", 1),
513            ),
514        ];
515
516        cache.insert_many(items);
517
518        // Update one oracle price
519        cache.update_oracle_price("ETH-USD", dec!(3000));
520
521        let oracle_map = cache.to_oracle_prices_map();
522        assert_eq!(oracle_map.len(), 2);
523
524        // BTC-USD should have default 50000
525        let btc_id = InstrumentId::new(Symbol::new("BTC-USD-PERP"), *DYDX_VENUE);
526        assert_eq!(oracle_map.get(&btc_id), Some(&dec!(50000)));
527
528        // ETH-USD should have updated price 3000
529        let eth_id = InstrumentId::new(Symbol::new("ETH-USD-PERP"), *DYDX_VENUE);
530        assert_eq!(oracle_map.get(&eth_id), Some(&dec!(3000)));
531    }
532
533    #[rstest]
534    fn test_get_order_market_params_with_none_oracle_price() {
535        let cache = InstrumentCache::new();
536        let instrument = create_test_instrument("WTI-USD-PERP");
537        let instrument_id = instrument.id();
538        let mut market = create_test_market("WTI-USD", 99);
539        market.oracle_price = None;
540
541        cache.insert(instrument, market);
542
543        let params = cache.get_order_market_params(&instrument_id).unwrap();
544        assert_eq!(params.oracle_price, None);
545        assert_eq!(params.clob_pair_id, 99);
546    }
547
548    #[rstest]
549    fn test_to_oracle_prices_map_excludes_none() {
550        let cache = InstrumentCache::new();
551
552        let mut market_no_oracle = create_test_market("WTI-USD", 99);
553        market_no_oracle.oracle_price = None;
554
555        let items = vec![
556            (
557                create_test_instrument("BTC-USD-PERP"),
558                create_test_market("BTC-USD", 0),
559            ),
560            (create_test_instrument("WTI-USD-PERP"), market_no_oracle),
561        ];
562
563        cache.insert_many(items);
564
565        let oracle_map = cache.to_oracle_prices_map();
566        assert_eq!(oracle_map.len(), 1);
567
568        let btc_id = InstrumentId::new(Symbol::new("BTC-USD-PERP"), *DYDX_VENUE);
569        assert_eq!(oracle_map.get(&btc_id), Some(&dec!(50000)));
570
571        let wti_id = InstrumentId::new(Symbol::new("WTI-USD-PERP"), *DYDX_VENUE);
572        assert_eq!(oracle_map.get(&wti_id), None);
573    }
574}
nautilus_dydx/common/instrument_cache.rs

nautilus_dydx/common/
instrument_cache.rs
