nautilus_tardis/http/
client.rs1use std::{env, time::Duration};
17
18use nautilus_core::{UnixNanos, consts::USER_AGENT};
19use nautilus_model::instruments::InstrumentAny;
20use reqwest::Response;
21
22use super::{
23 TARDIS_BASE_URL,
24 error::{Error, TardisErrorResponse},
25 models::InstrumentInfo,
26 parse::parse_instrument_any,
27 query::InstrumentFilter,
28};
29use crate::enums::Exchange;
30
31pub type Result<T> = std::result::Result<T, Error>;
32
33#[cfg_attr(
36 feature = "python",
37 pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.adapters")
38)]
39#[derive(Debug, Clone)]
40pub struct TardisHttpClient {
41 base_url: String,
42 api_key: String,
43 client: reqwest::Client,
44 normalize_symbols: bool,
45}
46
47impl TardisHttpClient {
48 pub fn new(
50 api_key: Option<&str>,
51 base_url: Option<&str>,
52 timeout_secs: Option<u64>,
53 normalize_symbols: bool,
54 ) -> anyhow::Result<Self> {
55 let api_key = match api_key {
56 Some(key) => key.to_string(),
57 None => env::var("TARDIS_API_KEY").map_err(|_| {
58 anyhow::anyhow!(
59 "API key must be provided or set in the 'TARDIS_API_KEY' environment variable"
60 )
61 })?,
62 };
63
64 let base_url = base_url.map_or_else(|| TARDIS_BASE_URL.to_string(), ToString::to_string);
65 let timeout = timeout_secs.map_or_else(|| Duration::from_secs(60), Duration::from_secs);
66
67 let client = reqwest::Client::builder()
68 .user_agent(USER_AGENT)
69 .timeout(timeout)
70 .build()?;
71
72 Ok(Self {
73 base_url,
74 api_key,
75 client,
76 normalize_symbols,
77 })
78 }
79
80 async fn handle_error_response<T>(resp: Response) -> Result<T> {
81 let status = resp.status().as_u16();
82 let error_text = resp.text().await.unwrap_or_default();
83
84 if let Ok(error) = serde_json::from_str::<TardisErrorResponse>(&error_text) {
85 Err(Error::ApiError {
86 status,
87 code: error.code,
88 message: error.message,
89 })
90 } else {
91 Err(Error::ApiError {
92 status,
93 code: 0,
94 message: error_text,
95 })
96 }
97 }
98
99 pub async fn instruments_info(
103 &self,
104 exchange: Exchange,
105 symbol: Option<&str>,
106 filter: Option<&InstrumentFilter>,
107 ) -> Result<Vec<InstrumentInfo>> {
108 let mut url = format!("{}/instruments/{exchange}", &self.base_url);
109 if let Some(symbol) = symbol {
110 url.push_str(&format!("/{symbol}"));
111 }
112 if let Some(filter) = filter {
113 if let Ok(filter_json) = serde_json::to_string(filter) {
114 url.push_str(&format!("?filter={}", urlencoding::encode(&filter_json)));
115 }
116 }
117 tracing::debug!("Requesting: {url}");
118
119 let resp = self
120 .client
121 .get(url)
122 .bearer_auth(&self.api_key)
123 .send()
124 .await?;
125 tracing::debug!("Response status: {}", resp.status());
126
127 if !resp.status().is_success() {
128 return Self::handle_error_response(resp).await;
129 }
130
131 let body = resp.text().await?;
132 tracing::trace!("{body}");
133
134 if let Ok(instrument) = serde_json::from_str::<InstrumentInfo>(&body) {
135 return Ok(vec![instrument]);
136 }
137
138 match serde_json::from_str(&body) {
139 Ok(parsed) => Ok(parsed),
140 Err(e) => {
141 tracing::error!("Failed to parse response: {e}");
142 tracing::debug!("Response body was: {body}");
143 Err(Error::ResponseParse(e.to_string()))
144 }
145 }
146 }
147
148 pub async fn instruments(
152 &self,
153 exchange: Exchange,
154 symbol: Option<&str>,
155 filter: Option<&InstrumentFilter>,
156 effective: Option<UnixNanos>,
157 ts_init: Option<UnixNanos>,
158 ) -> Result<Vec<InstrumentAny>> {
159 let response = self.instruments_info(exchange, symbol, filter).await?;
160
161 Ok(response
162 .into_iter()
163 .flat_map(|info| parse_instrument_any(info, effective, ts_init, self.normalize_symbols))
164 .collect())
165 }
166}