nautilus_tardis/http/
client.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
16use std::{borrow::Cow, collections::HashMap, fmt::Debug, io::Read, sync::Arc};
17
18use ahash::{AHashMap, AHashSet};
19use flate2::read::GzDecoder;
20use nautilus_core::{
21    DurationNanos, UnixNanos,
22    string::{parsing::precision_from_str, secret::REDACTED, urlencoding},
23};
24use nautilus_model::instruments::InstrumentAny;
25use nautilus_network::{
26    ConnectionUrl,
27    http::{HttpClient, HttpRedirectPolicy, HttpResponse, create_standard_nautilus_headers},
28    url_origin,
29};
30
31use super::{
32    error::{Error, TardisErrorResponse},
33    instruments::is_available,
34    models::TardisInstrumentInfo,
35    parse::parse_instrument_any,
36    query::InstrumentFilter,
37};
38use crate::{
39    common::{
40        consts::{TARDIS_REST_QUOTA, TARDIS_REST_RATE_KEY},
41        credential::Credential,
42        enums::TardisExchange,
43        parse::{normalize_instrument_id, parse_instrument_id},
44        urls::TARDIS_HTTP_BASE_URL,
45    },
46    machine::types::{TardisInstrumentKey, TardisInstrumentMiniInfo},
47};
48
49pub type Result<T> = std::result::Result<T, Error>;
50
51const CONTENT_ENCODING: &str = "content-encoding";
52const GZIP: &str = "gzip";
53
54// Deribit's all-symbol instrument list decompresses to about 284 MB
55const MAX_DECOMPRESSED_BYTES: u64 = 1024 * 1024 * 1024;
56
57/// A Tardis HTTP API client.
58/// See <https://docs.tardis.dev/api/http>.
59#[cfg_attr(
60    feature = "python",
61    pyo3::pyclass(module = "nautilus_trader.adapters.tardis", from_py_object)
62)]
63#[cfg_attr(
64    feature = "python",
65    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.tardis")
66)]
67#[derive(Clone)]
68pub struct TardisHttpClient {
69    base_url: ConnectionUrl,
70    credential: Option<Credential>,
71    client: HttpClient,
72    normalize_symbols: bool,
73}
74
75impl Debug for TardisHttpClient {
76    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
77        f.debug_struct(stringify!(TardisHttpClient))
78            .field("base_url", &self.base_url)
79            .field("credential", &self.credential.as_ref().map(|_| REDACTED))
80            .field("normalize_symbols", &self.normalize_symbols)
81            .finish()
82    }
83}
84
85impl TardisHttpClient {
86    /// Creates a new [`TardisHttpClient`] instance.
87    ///
88    /// # Errors
89    ///
90    /// Returns an error if no API key is provided (argument or `TARDIS_API_KEY` env var),
91    /// or if the HTTP client cannot be built.
92    pub fn new(
93        api_key: Option<&str>,
94        base_url: Option<&str>,
95        timeout_secs: Option<u64>,
96        normalize_symbols: bool,
97        proxy_url: Option<String>,
98    ) -> anyhow::Result<Self> {
99        let credential = Credential::resolve(api_key.map(ToString::to_string));
100
101        if credential.is_none() {
102            anyhow::bail!(
103                "API key must be provided or set in the 'TARDIS_API_KEY' environment variable"
104            );
105        }
106
107        let base_url =
108            base_url.map_or_else(|| TARDIS_HTTP_BASE_URL.to_string(), ToString::to_string);
109
110        let mut headers: HashMap<String, String> =
111            create_standard_nautilus_headers().into_iter().collect();
112
113        if let Some(ref cred) = credential {
114            headers.insert(
115                "Authorization".to_string(),
116                format!("Bearer {}", cred.api_key()),
117            );
118        }
119
120        // Uncompressed all-symbol instrument lists exceed the default HTTP response limit
121        headers.insert("Accept-Encoding".to_string(), GZIP.to_string());
122
123        let keyed_quotas = vec![(TARDIS_REST_RATE_KEY.to_string(), *TARDIS_REST_QUOTA)];
124        let client = HttpClient::builder()
125            .redirect_policy(HttpRedirectPolicy::Reject)
126            .headers(headers)
127            .keyed_quotas(keyed_quotas)
128            .default_quota(*TARDIS_REST_QUOTA)
129            .maybe_timeout_secs(timeout_secs.or(Some(60)))
130            .maybe_proxy_url(proxy_url)
131            .header_keys(vec![CONTENT_ENCODING.to_string()])
132            .build()?;
133
134        Ok(Self {
135            base_url: base_url.into(),
136            credential,
137            client,
138            normalize_symbols,
139        })
140    }
141
142    /// Returns the credential associated with this client.
143    #[must_use]
144    pub const fn credential(&self) -> Option<&Credential> {
145        self.credential.as_ref()
146    }
147
148    /// Returns all Tardis instrument definitions for the given `exchange`.
149    ///
150    /// # Errors
151    ///
152    /// Returns an error if the HTTP request fails or the response cannot be parsed.
153    ///
154    /// See <https://docs.tardis.dev/api/instruments-metadata-api>.
155    pub async fn instruments_info(
156        &self,
157        exchange: TardisExchange,
158        symbol: Option<&str>,
159        filter: Option<&InstrumentFilter>,
160    ) -> Result<Vec<TardisInstrumentInfo>> {
161        let mut url = format!("{}/instruments/{exchange}", self.base_url.expose());
162        if let Some(symbol) = symbol {
163            url.push_str(&format!("/{symbol}"));
164        }
165
166        if let Some(filter) = filter
167            && let Ok(filter_json) = serde_json::to_string(filter)
168        {
169            url.push_str(&format!("?filter={}", urlencoding::encode(&filter_json)));
170        }
171
172        log::debug!("Requesting: {}", url_origin(&url));
173
174        let rate_keys = Some(vec![TARDIS_REST_RATE_KEY.to_string()]);
175
176        let response = self
177            .client
178            .get(url, None, None, None, rate_keys)
179            .await
180            .map_err(|e| {
181                Error::Request(format!("Request to {} failed: {e}", self.base_url.origin()))
182            })?;
183
184        let status = response.status.as_u16();
185        log::debug!("Response status: {status}");
186
187        let body = decode_body(&response)?;
188
189        if !response.status.is_success() {
190            let body = String::from_utf8_lossy(&body).to_string();
191            return if let Ok(error) = serde_json::from_str::<TardisErrorResponse>(&body) {
192                Err(Error::ApiError {
193                    status,
194                    code: error.code,
195                    message: error.message,
196                })
197            } else {
198                Err(Error::ApiError {
199                    status,
200                    code: 0,
201                    message: body,
202                })
203            };
204        }
205
206        let body = String::from_utf8_lossy(&body);
207        log::trace!("{body}");
208
209        if let Ok(instrument) = serde_json::from_str::<TardisInstrumentInfo>(&body) {
210            return Ok(vec![instrument]);
211        }
212
213        match serde_json::from_str(&body) {
214            Ok(parsed) => Ok(parsed),
215            Err(e) => {
216                log::debug!("Failed to parse response: {e}");
217                log::debug!("Response body was: {body}");
218                Err(Error::ResponseParse(e.to_string()))
219            }
220        }
221    }
222
223    /// Returns all Nautilus instrument definitions for the given `exchange`, and filter params.
224    ///
225    /// # Errors
226    ///
227    /// Returns an error if fetching instrument info or parsing into domain types fails.
228    ///
229    /// See <https://docs.tardis.dev/api/instruments-metadata-api>.
230    #[expect(clippy::too_many_arguments)]
231    pub async fn instruments(
232        &self,
233        exchange: TardisExchange,
234        symbol: Option<&str>,
235        filter: Option<&InstrumentFilter>,
236        start: Option<UnixNanos>,
237        end: Option<UnixNanos>,
238        available_offset: Option<DurationNanos>,
239        effective: Option<UnixNanos>,
240        ts_init: Option<UnixNanos>,
241    ) -> Result<Vec<InstrumentAny>> {
242        let response = self.instruments_info(exchange, symbol, filter).await?;
243
244        Ok(response
245            .into_iter()
246            .filter(|info| is_available(info, start, end, available_offset, effective))
247            .flat_map(|info| {
248                parse_instrument_any(&info, effective, ts_init, self.normalize_symbols)
249            })
250            .collect())
251    }
252
253    /// Fetches instruments for the given exchanges, builds the mini-info map
254    /// for WS message parsing, and parses Nautilus instrument definitions.
255    ///
256    /// Returns a tuple of `(instrument_map, nautilus_instruments)`. The caller
257    /// decides how to use each half: `data.rs` emits instruments via the data
258    /// sender; `replay.rs` only needs the map.
259    ///
260    /// # Errors
261    ///
262    /// Returns an error if fetching instrument info for any exchange fails.
263    pub async fn bootstrap_instruments(
264        &self,
265        exchanges: &AHashSet<TardisExchange>,
266    ) -> Result<(
267        AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>>,
268        Vec<InstrumentAny>,
269    )> {
270        let mut instrument_map: AHashMap<TardisInstrumentKey, Arc<TardisInstrumentMiniInfo>> =
271            AHashMap::new();
272        let mut nautilus_instruments: Vec<InstrumentAny> = Vec::new();
273
274        for exchange in exchanges {
275            log::debug!("Fetching instruments for {exchange}");
276
277            let instruments_info = match self.instruments_info(*exchange, None, None).await {
278                Ok(info) => info,
279                Err(e) => {
280                    log::error!("Failed to fetch instruments for {exchange}: {e}");
281                    continue;
282                }
283            };
284
285            log::debug!(
286                "Received {} instruments for {exchange}",
287                instruments_info.len()
288            );
289
290            for inst in &instruments_info {
291                let instrument_type = inst.instrument_type;
292                let price_precision = precision_from_str(&inst.price_increment.to_string());
293                let size_precision = precision_from_str(&inst.amount_increment.to_string());
294
295                let instrument_id = if self.normalize_symbols {
296                    normalize_instrument_id(exchange, inst.id, &instrument_type, inst.inverse)
297                } else {
298                    parse_instrument_id(exchange, inst.id)
299                };
300
301                let info = TardisInstrumentMiniInfo::new(
302                    instrument_id,
303                    Some(inst.id),
304                    *exchange,
305                    price_precision,
306                    size_precision,
307                );
308                let key = info.as_tardis_instrument_key();
309                instrument_map.insert(key, Arc::new(info));
310            }
311
312            for inst in instruments_info {
313                nautilus_instruments.extend(parse_instrument_any(
314                    &inst,
315                    None,
316                    None,
317                    self.normalize_symbols,
318                ));
319            }
320        }
321
322        Ok((instrument_map, nautilus_instruments))
323    }
324}
325
326fn decode_body(response: &HttpResponse) -> Result<Cow<'_, [u8]>> {
327    match response.headers.get(CONTENT_ENCODING) {
328        None => Ok(Cow::Borrowed(response.body.as_ref())),
329        Some(encoding) if encoding.eq_ignore_ascii_case(GZIP) => {
330            decompress_gzip(&response.body, MAX_DECOMPRESSED_BYTES).map(Cow::Owned)
331        }
332        Some(encoding) => Err(Error::Request(format!(
333            "unsupported response content encoding '{encoding}'"
334        ))),
335    }
336}
337
338fn decompress_gzip(body: &[u8], max_bytes: u64) -> Result<Vec<u8>> {
339    let mut reader = GzDecoder::new(body).take(max_bytes + 1);
340    let mut decompressed = Vec::new();
341    reader
342        .read_to_end(&mut decompressed)
343        .map_err(|e| Error::Request(format!("failed to decompress response body: {e}")))?;
344
345    if reader.limit() == 0 {
346        return Err(Error::Request(format!(
347            "decompressed response body exceeds maximum of {max_bytes} bytes"
348        )));
349    }
350
351    Ok(decompressed)
352}
353
354#[cfg(test)]
355mod tests {
356    use std::io::Write;
357
358    use flate2::{Compression, write::GzEncoder};
359    use nautilus_network::http::{HttpStatus, StatusCode};
360    use nautilus_testkit::http::assert_http_redirect_rejected;
361    use rstest::rstest;
362
363    use super::*;
364
365    fn gzip(data: &[u8]) -> Vec<u8> {
366        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
367        encoder.write_all(data).unwrap();
368        encoder.finish().unwrap()
369    }
370
371    fn http_response(encoding: Option<&str>, body: Vec<u8>) -> HttpResponse {
372        HttpResponse {
373            status: HttpStatus::new(StatusCode::OK),
374            headers: encoding
375                .map(|encoding| (CONTENT_ENCODING.to_string(), encoding.to_string()))
376                .into_iter()
377                .collect(),
378            body: body.into(),
379        }
380    }
381
382    #[rstest]
383    #[case("https://api.example.com/v1", "https://api.example.com")]
384    #[case(
385        "https://SYN-user:SYN-password@api.example.com:8443/SYN-path?token=SYN-query#SYN-fragment",
386        "https://api.example.com:8443"
387    )]
388    fn test_debug_preserves_url_origin(#[case] url: &str, #[case] origin: &str) {
389        let client =
390            TardisHttpClient::new(Some("SYN-api-key"), Some(url), Some(1), true, None).unwrap();
391        let debug = format!("{client:?}");
392        assert!(debug.contains(&format!("base_url: {origin:?}")));
393        assert!(!debug.contains("SYN-"));
394        assert_eq!(client.base_url.expose(), url);
395    }
396
397    #[tokio::test]
398    async fn test_authenticated_client_rejects_redirects() {
399        let client = TardisHttpClient::new(Some("test-key"), None, Some(3), false, None)
400            .unwrap()
401            .client;
402        assert_http_redirect_rejected(|url| async move {
403            client
404                .get(url, None, None, Some(3), None)
405                .await
406                .unwrap()
407                .status
408                .as_u16()
409        })
410        .await;
411    }
412
413    #[rstest]
414    #[case::identity(None, b"[7]".to_vec())]
415    #[case::gzip(Some("gzip"), gzip(b"[7]"))]
416    #[case::gzip_uppercase(Some("GZIP"), gzip(b"[7]"))]
417    fn test_decode_body(#[case] encoding: Option<&str>, #[case] body: Vec<u8>) {
418        let response = http_response(encoding, body);
419
420        let decoded = decode_body(&response).unwrap();
421
422        assert_eq!(decoded.as_ref(), b"[7]");
423    }
424
425    #[rstest]
426    #[case::unsupported(Some("br"), b"[7]".to_vec(), "unsupported response content encoding 'br'")]
427    #[case::corrupt(
428        Some("gzip"),
429        b"[7]".to_vec(),
430        "failed to decompress response body: unexpected end of file"
431    )]
432    fn test_decode_body_rejects(
433        #[case] encoding: Option<&str>,
434        #[case] body: Vec<u8>,
435        #[case] expected: &str,
436    ) {
437        let response = http_response(encoding, body);
438
439        let error = decode_body(&response).unwrap_err();
440
441        let Error::Request(message) = error else {
442            panic!("expected request error, was {error:?}");
443        };
444
445        assert_eq!(message, expected);
446    }
447
448    #[rstest]
449    #[case::at_max(10)]
450    #[case::below_max(11)]
451    fn test_decompress_gzip_at_or_below_max_bytes(#[case] max_bytes: u64) {
452        let decompressed = decompress_gzip(&gzip(b"0123456789"), max_bytes).unwrap();
453
454        assert_eq!(decompressed, b"0123456789");
455    }
456
457    #[rstest]
458    fn test_decompress_gzip_above_max_bytes() {
459        let error = decompress_gzip(&gzip(b"0123456789"), 9).unwrap_err();
460
461        let Error::Request(message) = error else {
462            panic!("expected request error, was {error:?}");
463        };
464
465        assert_eq!(
466            message,
467            "decompressed response body exceeds maximum of 9 bytes"
468        );
469    }
470}
nautilus_tardis/http/client.rs

nautilus_tardis/http/
client.rs
