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

nautilus_derive/
providers.rs
