nautilus_derive/
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 for the Derive adapter.
17
18use std::{collections::HashMap, fmt::Debug};
19
20use async_trait::async_trait;
21use nautilus_common::providers::{InstrumentProvider, InstrumentStore};
22use nautilus_core::time::get_atomic_clock_realtime;
23use nautilus_model::{
24    identifiers::InstrumentId,
25    instruments::{Instrument, InstrumentAny},
26};
27
28use crate::{
29    common::{
30        consts::DERIVE_VENUE, enums::DeriveInstrumentType, parse::parse_derive_instrument_any,
31    },
32    http::{DeriveHttpClient, error::DeriveHttpError, models::DeriveInstrument},
33};
34
35const INSTRUMENT_NOT_FOUND_CODE: i64 = 12001;
36
37/// Provides Derive instruments via the REST API.
38///
39/// The Derive `public/get_instruments` endpoint is scoped by underlying
40/// currency, so callers configure the currency set up front or pass a
41/// `currency`/`currencies` filter to `load_all()`.
42pub struct DeriveInstrumentProvider {
43    store: InstrumentStore,
44    http_client: DeriveHttpClient,
45    currencies: Vec<String>,
46    include_expired: bool,
47}
48
49impl Debug for DeriveInstrumentProvider {
50    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
51        f.debug_struct(stringify!(DeriveInstrumentProvider))
52            .field("store", &self.store)
53            .field("http_client", &self.http_client)
54            .field("currencies", &self.currencies)
55            .field("include_expired", &self.include_expired)
56            .finish()
57    }
58}
59
60impl DeriveInstrumentProvider {
61    /// Creates a new provider with an empty store.
62    #[must_use]
63    pub fn new(http_client: DeriveHttpClient, currencies: Vec<String>) -> Self {
64        Self {
65            store: InstrumentStore::new(),
66            http_client,
67            currencies,
68            include_expired: false,
69        }
70    }
71
72    /// Creates a new provider and controls whether expired instruments load.
73    #[must_use]
74    pub fn with_expired(
75        http_client: DeriveHttpClient,
76        currencies: Vec<String>,
77        include_expired: bool,
78    ) -> Self {
79        Self {
80            store: InstrumentStore::new(),
81            http_client,
82            currencies,
83            include_expired,
84        }
85    }
86
87    /// Returns the configured currency filters.
88    #[must_use]
89    pub fn currencies(&self) -> &[String] {
90        &self.currencies
91    }
92
93    /// Returns whether `load_all()` includes expired instruments by default.
94    #[must_use]
95    pub const fn include_expired(&self) -> bool {
96        self.include_expired
97    }
98
99    /// Returns a reference to the underlying HTTP client.
100    #[must_use]
101    pub const fn http_client(&self) -> &DeriveHttpClient {
102        &self.http_client
103    }
104
105    /// Adds instruments to the store.
106    pub fn add_instruments(&mut self, instruments: Vec<InstrumentAny>) {
107        self.store.add_bulk(instruments);
108    }
109
110    async fn fetch_instruments(
111        &self,
112        currencies: &[String],
113        expired: bool,
114    ) -> anyhow::Result<Vec<InstrumentAny>> {
115        let mut instruments = Vec::new();
116
117        for currency in currencies {
118            let definitions =
119                fetch_instrument_definitions(&self.http_client, currency, expired).await?;
120            instruments.extend(parse_instrument_definitions(definitions));
121        }
122
123        Ok(instruments)
124    }
125}
126
127#[async_trait(?Send)]
128impl InstrumentProvider for DeriveInstrumentProvider {
129    fn store(&self) -> &InstrumentStore {
130        &self.store
131    }
132
133    fn store_mut(&mut self) -> &mut InstrumentStore {
134        &mut self.store
135    }
136
137    async fn load_all(&mut self, filters: Option<&HashMap<String, String>>) -> anyhow::Result<()> {
138        let (currencies, expired) =
139            resolve_load_filters(&self.currencies, self.include_expired, filters)?;
140        let instruments = self.fetch_instruments(&currencies, expired).await?;
141
142        self.store.clear();
143        self.add_instruments(instruments);
144        self.store.set_initialized();
145
146        Ok(())
147    }
148
149    async fn load_ids(
150        &mut self,
151        instrument_ids: &[InstrumentId],
152        filters: Option<&HashMap<String, String>>,
153    ) -> anyhow::Result<()> {
154        let missing: Vec<_> = instrument_ids
155            .iter()
156            .filter(|id| !self.store.contains(id))
157            .collect();
158
159        if missing.is_empty() {
160            return Ok(());
161        }
162
163        let expired = resolve_expired_filter(self.include_expired, filters)?;
164        let mut currencies: Vec<String> = missing
165            .iter()
166            .filter_map(|id| currency_from_instrument_id(id).map(ToOwned::to_owned))
167            .collect();
168        currencies.sort();
169        currencies.dedup();
170
171        if !currencies.is_empty() {
172            let instruments = self.fetch_instruments(&currencies, expired).await?;
173            self.add_instruments(instruments);
174        }
175
176        if missing.iter().all(|id| self.store.contains(id)) {
177            return Ok(());
178        }
179
180        if !self.store.is_initialized() {
181            let existing = self.store.get_all().values().cloned().collect::<Vec<_>>();
182            self.load_all(filters).await?;
183
184            for instrument in existing {
185                if !self.store.contains(&instrument.id()) {
186                    self.store.add(instrument);
187                }
188            }
189        }
190
191        let missing_ids: Vec<_> = instrument_ids
192            .iter()
193            .filter(|id| !self.store.contains(id))
194            .collect();
195
196        if missing_ids.is_empty() {
197            Ok(())
198        } else {
199            anyhow::bail!("Derive instruments not found: {missing_ids:?}")
200        }
201    }
202
203    async fn load(
204        &mut self,
205        instrument_id: &InstrumentId,
206        filters: Option<&HashMap<String, String>>,
207    ) -> anyhow::Result<()> {
208        if self.store.contains(instrument_id) {
209            return Ok(());
210        }
211
212        self.load_ids(&[*instrument_id], filters).await
213    }
214}
215
216pub(crate) fn parse_instrument_definitions(
217    definitions: Vec<DeriveInstrument>,
218) -> Vec<InstrumentAny> {
219    let ts_init = get_atomic_clock_realtime().get_time_ns();
220    let mut instruments = Vec::with_capacity(definitions.len());
221
222    for definition in definitions {
223        match parse_derive_instrument_any(&definition, ts_init) {
224            Ok(Some(instrument)) => instruments.push(instrument),
225            Ok(None) => {}
226            Err(e) => log::warn!(
227                "Failed to parse Derive instrument {}: {e}",
228                definition.instrument_name,
229            ),
230        }
231    }
232
233    instruments
234}
235
236pub(crate) async fn fetch_instrument_definitions(
237    http_client: &DeriveHttpClient,
238    currency: &str,
239    expired: bool,
240) -> anyhow::Result<Vec<DeriveInstrument>> {
241    let (mut definitions, options, erc20s) = tokio::try_join!(
242        fetch_instruments_if_listed(http_client, currency, DeriveInstrumentType::Perp, expired),
243        fetch_instruments_if_listed(http_client, currency, DeriveInstrumentType::Option, expired,),
244        fetch_instruments_if_listed(http_client, currency, DeriveInstrumentType::Erc20, expired,),
245    )?;
246    definitions.extend(options);
247    definitions.extend(erc20s);
248
249    Ok(definitions)
250}
251
252// Venue returns JSON-RPC error 12001 (`Instrument not found`) when a currency
253// has no listing for the requested product type. Treat that leg as empty so
254// the other product types still load.
255async fn fetch_instruments_if_listed(
256    http_client: &DeriveHttpClient,
257    currency: &str,
258    instrument_type: DeriveInstrumentType,
259    expired: bool,
260) -> Result<Vec<DeriveInstrument>, DeriveHttpError> {
261    match http_client
262        .get_instruments(currency, instrument_type, expired)
263        .await
264    {
265        Ok(rows) => Ok(rows),
266        Err(DeriveHttpError::JsonRpc { code, .. }) if code == INSTRUMENT_NOT_FOUND_CODE => {
267            Ok(Vec::new())
268        }
269        Err(e) => Err(e),
270    }
271}
272
273fn resolve_load_filters(
274    default_currencies: &[String],
275    default_expired: bool,
276    filters: Option<&HashMap<String, String>>,
277) -> anyhow::Result<(Vec<String>, bool)> {
278    let currencies = filters
279        .and_then(|map| {
280            map.get("currency")
281                .map(|currency| vec![currency.trim().to_string()])
282                .or_else(|| map.get("currencies").map(|value| split_currencies(value)))
283        })
284        .unwrap_or_else(|| default_currencies.to_vec());
285
286    let currencies = normalize_currencies(currencies);
287
288    anyhow::ensure!(
289        !currencies.is_empty(),
290        "DeriveInstrumentProvider requires at least one currency",
291    );
292
293    let expired = resolve_expired_filter(default_expired, filters)?;
294
295    Ok((currencies, expired))
296}
297
298fn resolve_expired_filter(
299    default_expired: bool,
300    filters: Option<&HashMap<String, String>>,
301) -> anyhow::Result<bool> {
302    filters
303        .and_then(|map| map.get("expired"))
304        .map(|value| value.parse::<bool>())
305        .transpose()
306        .map_err(|e| anyhow::anyhow!("invalid Derive `expired` filter: {e}"))
307        .map(|value| value.unwrap_or(default_expired))
308}
309
310fn split_currencies(value: &str) -> Vec<String> {
311    normalize_currencies(value.split(',').map(ToOwned::to_owned).collect())
312}
313
314fn normalize_currencies(currencies: Vec<String>) -> Vec<String> {
315    let mut currencies: Vec<_> = currencies
316        .into_iter()
317        .map(|currency| currency.trim().to_string())
318        .filter(|currency| !currency.is_empty())
319        .collect();
320    currencies.sort();
321    currencies.dedup();
322    currencies
323}
324
325fn currency_from_instrument_id(instrument_id: &InstrumentId) -> Option<&str> {
326    if instrument_id.venue != *DERIVE_VENUE {
327        return None;
328    }
329
330    instrument_id
331        .symbol
332        .as_str()
333        .split_once('-')
334        .and_then(|(currency, _)| (!currency.is_empty()).then_some(currency))
335}
nautilus_derive/providers.rs

nautilus_derive/
providers.rs
