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) -> anyhow::Result<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        if let Some(instrument) = parse_derive_instrument_any(&definition, ts_init)? {
224            instruments.push(instrument);
225        }
226    }
227
228    Ok(instruments)
229}
230
231pub(crate) async fn fetch_instrument_definitions(
232    http_client: &DeriveHttpClient,
233    currency: &str,
234    expired: bool,
235) -> anyhow::Result<Vec<DeriveInstrument>> {
236    let (mut definitions, options, erc20s) = tokio::try_join!(
237        http_client.get_instruments(currency, DeriveInstrumentType::Perp, expired),
238        http_client.get_instruments(currency, DeriveInstrumentType::Option, expired),
239        fetch_erc20_instruments(http_client, currency, expired),
240    )?;
241    definitions.extend(options);
242    definitions.extend(erc20s);
243
244    Ok(definitions)
245}
246
247// Venue returns JSON-RPC error 12001 (`Instrument not found`) when querying
248// `erc20` for a currency with no spot listing (e.g. BTC has perp+option but
249// no spot). Treat as empty so a missing spot listing does not fail the
250// perp/option fetch.
251async fn fetch_erc20_instruments(
252    http_client: &DeriveHttpClient,
253    currency: &str,
254    expired: bool,
255) -> Result<Vec<DeriveInstrument>, DeriveHttpError> {
256    match http_client
257        .get_instruments(currency, DeriveInstrumentType::Erc20, expired)
258        .await
259    {
260        Ok(rows) => Ok(rows),
261        Err(DeriveHttpError::JsonRpc { code, .. }) if code == INSTRUMENT_NOT_FOUND_CODE => {
262            Ok(Vec::new())
263        }
264        Err(e) => Err(e),
265    }
266}
267
268fn resolve_load_filters(
269    default_currencies: &[String],
270    default_expired: bool,
271    filters: Option<&HashMap<String, String>>,
272) -> anyhow::Result<(Vec<String>, bool)> {
273    let currencies = filters
274        .and_then(|map| {
275            map.get("currency")
276                .map(|currency| vec![currency.trim().to_string()])
277                .or_else(|| map.get("currencies").map(|value| split_currencies(value)))
278        })
279        .unwrap_or_else(|| default_currencies.to_vec());
280
281    let currencies = normalize_currencies(currencies);
282
283    anyhow::ensure!(
284        !currencies.is_empty(),
285        "DeriveInstrumentProvider requires at least one currency",
286    );
287
288    let expired = resolve_expired_filter(default_expired, filters)?;
289
290    Ok((currencies, expired))
291}
292
293fn resolve_expired_filter(
294    default_expired: bool,
295    filters: Option<&HashMap<String, String>>,
296) -> anyhow::Result<bool> {
297    filters
298        .and_then(|map| map.get("expired"))
299        .map(|value| value.parse::<bool>())
300        .transpose()
301        .map_err(|e| anyhow::anyhow!("invalid Derive `expired` filter: {e}"))
302        .map(|value| value.unwrap_or(default_expired))
303}
304
305fn split_currencies(value: &str) -> Vec<String> {
306    normalize_currencies(value.split(',').map(ToOwned::to_owned).collect())
307}
308
309fn normalize_currencies(currencies: Vec<String>) -> Vec<String> {
310    let mut currencies: Vec<_> = currencies
311        .into_iter()
312        .map(|currency| currency.trim().to_string())
313        .filter(|currency| !currency.is_empty())
314        .collect();
315    currencies.sort();
316    currencies.dedup();
317    currencies
318}
319
320fn currency_from_instrument_id(instrument_id: &InstrumentId) -> Option<&str> {
321    if instrument_id.venue != *DERIVE_VENUE {
322        return None;
323    }
324
325    instrument_id
326        .symbol
327        .as_str()
328        .split_once('-')
329        .and_then(|(currency, _)| (!currency.is_empty()).then_some(currency))
330}
nautilus_derive/providers.rs

nautilus_derive/
providers.rs
