nautilus_tardis/
config.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 nautilus_common::config::{ConfigResult, check_unique_values};
17#[cfg(test)]
18use nautilus_core::string::secret::REDACTED;
19use nautilus_core::string::secret::SecretString;
20#[cfg(feature = "arrow")]
21use parquet::basic::{Compression, ZstdLevel};
22use serde::{Deserialize, Serialize};
23
24use super::machine::types::{ReplayNormalizedRequestOptions, StreamNormalizedRequestOptions};
25
26/// Determines the output format for Tardis `book_snapshot_*` messages.
27#[derive(Debug, Clone, Default, Serialize, Deserialize)]
28#[serde(rename_all = "snake_case")]
29pub enum BookSnapshotOutput {
30    /// Convert book snapshots to `OrderBookDeltas` and write to `order_book_deltas/`.
31    #[default]
32    Deltas,
33    /// Convert book snapshots to `OrderBookDepth` and write to `order_book_depths/`.
34    #[serde(alias = "depth10")]
35    Depth,
36}
37
38/// Determines the compression codec for Parquet files written by Tardis replay.
39#[derive(Debug, Clone, Default, Serialize, Deserialize)]
40#[serde(rename_all = "snake_case")]
41pub enum ParquetCompression {
42    /// Use Zstandard compression with level 3.
43    #[default]
44    Zstd,
45    /// Use Snappy compression.
46    Snappy,
47    /// Write uncompressed Parquet files.
48    Uncompressed,
49}
50
51#[cfg(feature = "arrow")]
52impl ParquetCompression {
53    /// Converts the replay config compression value to a Parquet compression value.
54    ///
55    /// # Panics
56    ///
57    /// Panics if the hard-coded Zstandard level 3 is rejected by the Parquet crate.
58    #[must_use]
59    pub fn as_parquet_compression(&self) -> Compression {
60        match self {
61            Self::Zstd => {
62                let level = ZstdLevel::try_new(3).expect("zstd level 3 is valid");
63                Compression::ZSTD(level)
64            }
65            Self::Snappy => Compression::SNAPPY,
66            Self::Uncompressed => Compression::UNCOMPRESSED,
67        }
68    }
69}
70
71/// Provides a configuration for a Tardis Machine -> Nautilus data -> Parquet replay run.
72#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
73#[serde(deny_unknown_fields)]
74pub struct TardisReplayConfig {
75    /// The Tardis HTTP API base URL override.
76    pub tardis_http_url: Option<SecretString>,
77    /// The Tardis Machine websocket url.
78    pub tardis_ws_url: Option<SecretString>,
79    /// Optional proxy URL for the Tardis HTTP API client.
80    /// The Tardis Machine WebSocket transport does not yet support proxying.
81    pub proxy_url: Option<SecretString>,
82    /// If symbols should be normalized with Nautilus conventions.
83    pub normalize_symbols: Option<bool>,
84    /// The output directory for writing Nautilus format Parquet files.
85    pub output_path: Option<String>,
86    /// The Tardis Machine replay options.
87    #[builder(default)]
88    #[serde(default)]
89    pub options: Vec<ReplayNormalizedRequestOptions>,
90    /// The output format for `book_snapshot_*` messages.
91    ///
92    /// - `deltas`: Convert to `OrderBookDeltas` and write to `order_book_deltas/` (default).
93    /// - `depth`: Convert to `OrderBookDepth` and write to `order_book_depths/`.
94    pub book_snapshot_output: Option<BookSnapshotOutput>,
95    /// If best bid/offer fields from Tardis `option_summary` messages should emit `QuoteTick`.
96    pub extract_bbo_as_quotes: Option<bool>,
97    /// The compression codec for written data files.
98    ///
99    /// - `zstd`: Use Zstandard compression level 3 (default).
100    /// - `snappy`: Use Snappy compression.
101    /// - `uncompressed`: Write uncompressed Parquet files.
102    pub compression: Option<ParquetCompression>,
103}
104
105/// Configuration for the Tardis data client.
106#[derive(Debug, Clone, Serialize, Deserialize, bon::Builder)]
107#[serde(default, deny_unknown_fields)]
108#[cfg_attr(
109    feature = "python",
110    pyo3::pyclass(module = "nautilus_trader.adapters.tardis", from_py_object)
111)]
112#[cfg_attr(
113    feature = "python",
114    pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.adapters.tardis")
115)]
116pub struct TardisDataClientConfig {
117    /// Tardis API key for HTTP instrument fetching.
118    /// Falls back to `TARDIS_API_KEY` env var if not set.
119    pub api_key: Option<SecretString>,
120    /// The Tardis HTTP API base URL override.
121    pub tardis_http_url: Option<SecretString>,
122    /// Tardis Machine Server WebSocket URL.
123    /// Falls back to `TARDIS_MACHINE_WS_URL` env var if not set.
124    pub tardis_ws_url: Option<SecretString>,
125    /// Optional proxy URL for the Tardis HTTP API client.
126    /// The Tardis Machine WebSocket transport does not yet support proxying.
127    pub proxy_url: Option<SecretString>,
128    /// Whether to normalize symbols to Nautilus conventions.
129    #[builder(default = true)]
130    pub normalize_symbols: bool,
131    /// Output format for `book_snapshot_*` messages.
132    #[builder(default)]
133    pub book_snapshot_output: BookSnapshotOutput,
134    /// Whether to emit `QuoteTick` from Tardis `option_summary` best bid/offer fields.
135    #[builder(default)]
136    pub extract_bbo_as_quotes: bool,
137    /// Replay options defining exchanges, symbols, date ranges, and data types.
138    /// When non-empty the client connects to `ws-replay-normalized`.
139    #[builder(default)]
140    pub options: Vec<ReplayNormalizedRequestOptions>,
141    /// Live stream options defining exchanges, symbols, and data types.
142    /// When non-empty (and `options` is empty) the client connects to
143    /// `ws-stream-normalized` with automatic reconnection.
144    #[builder(default)]
145    pub stream_options: Vec<StreamNormalizedRequestOptions>,
146}
147
148#[cfg(feature = "python")]
149nautilus_core::impl_pyo3_config_getters!(TardisDataClientConfig {
150    normalize_symbols: bool,
151    extract_bbo_as_quotes: bool,
152    options: Vec<ReplayNormalizedRequestOptions>,
153    stream_options: Vec<StreamNormalizedRequestOptions>,
154});
155
156impl Default for TardisDataClientConfig {
157    fn default() -> Self {
158        Self::builder().build()
159    }
160}
161
162impl TardisReplayConfig {
163    /// Validates that configured selections contain no duplicates.
164    ///
165    /// # Errors
166    ///
167    /// Returns an error if a selection repeats a value.
168    pub fn validate(&self) -> ConfigResult<()> {
169        validate_replay_options(&self.options)
170    }
171}
172
173impl TardisDataClientConfig {
174    /// Validates that configured selections contain no duplicates.
175    ///
176    /// # Errors
177    ///
178    /// Returns an error if a selection repeats a value.
179    pub fn validate(&self) -> ConfigResult<()> {
180        validate_replay_options(&self.options)?;
181        check_unique_values("stream_options", &self.stream_options)?;
182
183        for (index, option) in self.stream_options.iter().enumerate() {
184            if let Some(symbols) = &option.symbols {
185                check_unique_values(format!("stream_options[{index}].symbols"), symbols)?;
186            }
187
188            check_unique_values(
189                format!("stream_options[{index}].data_types"),
190                &option.data_types,
191            )?;
192        }
193
194        Ok(())
195    }
196}
197
198fn validate_replay_options(options: &[ReplayNormalizedRequestOptions]) -> ConfigResult<()> {
199    check_unique_values("options", options)?;
200
201    for (index, option) in options.iter().enumerate() {
202        if let Some(symbols) = &option.symbols {
203            check_unique_values(format!("options[{index}].symbols"), symbols)?;
204        }
205
206        check_unique_values(format!("options[{index}].data_types"), &option.data_types)?;
207    }
208
209    Ok(())
210}
211
212#[cfg(test)]
213mod tests {
214    use nautilus_common::config::ConfigError;
215    use rstest::rstest;
216
217    use super::*;
218
219    #[rstest]
220    fn test_default_config_values() {
221        let config = TardisDataClientConfig::default();
222        assert!(config.api_key.is_none());
223        assert!(config.tardis_ws_url.is_none());
224        assert!(config.tardis_http_url.is_none());
225        assert!(config.proxy_url.is_none());
226        assert!(config.normalize_symbols);
227        assert!(matches!(
228            config.book_snapshot_output,
229            BookSnapshotOutput::Deltas
230        ));
231        assert!(!config.extract_bbo_as_quotes);
232        assert!(config.options.is_empty());
233        assert!(config.stream_options.is_empty());
234    }
235
236    #[rstest]
237    fn test_config_debug_redacts_credentials_and_urls() {
238        let config = TardisDataClientConfig {
239            api_key: Some("api-key-value".into()),
240            tardis_ws_url: Some("wss://user:ws-secret@localhost".into()),
241            tardis_http_url: Some("https://user:http-secret@localhost".into()),
242            proxy_url: Some("http://user:proxy-secret@localhost".into()),
243            ..Default::default()
244        };
245
246        let formatted = format!("{config:?}");
247
248        assert_eq!(formatted.matches(REDACTED).count(), 4);
249        assert!(!formatted.contains("api-key-value"));
250        assert!(!formatted.contains("ws-secret"));
251        assert!(!formatted.contains("http-secret"));
252        assert!(!formatted.contains("proxy-secret"));
253    }
254
255    #[rstest]
256    fn test_book_snapshot_output_default_is_deltas() {
257        assert!(matches!(
258            BookSnapshotOutput::default(),
259            BookSnapshotOutput::Deltas
260        ));
261    }
262
263    #[rstest]
264    fn test_book_snapshot_output_serde_roundtrip_deltas() {
265        let json = serde_json::to_string(&BookSnapshotOutput::Deltas).unwrap();
266        assert_eq!(json, "\"deltas\"");
267
268        let deserialized: BookSnapshotOutput = serde_json::from_str(&json).unwrap();
269        assert!(matches!(deserialized, BookSnapshotOutput::Deltas));
270    }
271
272    #[rstest]
273    fn test_book_snapshot_output_serde_roundtrip_depth() {
274        let json = serde_json::to_string(&BookSnapshotOutput::Depth).unwrap();
275        assert_eq!(json, "\"depth\"");
276
277        let deserialized: BookSnapshotOutput = serde_json::from_str(&json).unwrap();
278        assert!(matches!(deserialized, BookSnapshotOutput::Depth));
279    }
280
281    #[rstest]
282    fn test_book_snapshot_output_accepts_legacy_depth10_spelling() {
283        let deserialized: BookSnapshotOutput = serde_json::from_str("\"depth10\"").unwrap();
284        assert!(matches!(deserialized, BookSnapshotOutput::Depth));
285    }
286
287    #[cfg(feature = "arrow")]
288    #[rstest]
289    fn test_parquet_compression_default_is_zstd() {
290        assert!(matches!(
291            ParquetCompression::default(),
292            ParquetCompression::Zstd
293        ));
294        assert!(matches!(
295            ParquetCompression::default().as_parquet_compression(),
296            Compression::ZSTD(_)
297        ));
298    }
299
300    #[cfg(feature = "arrow")]
301    #[rstest]
302    fn test_parquet_compression_serde_roundtrip() {
303        let cases = [
304            (ParquetCompression::Zstd, "\"zstd\""),
305            (ParquetCompression::Snappy, "\"snappy\""),
306            (ParquetCompression::Uncompressed, "\"uncompressed\""),
307        ];
308
309        for (compression, expected_json) in cases {
310            let json = serde_json::to_string(&compression).unwrap();
311            assert_eq!(json, expected_json);
312
313            let deserialized: ParquetCompression = serde_json::from_str(&json).unwrap();
314            assert_eq!(
315                compression.as_parquet_compression(),
316                deserialized.as_parquet_compression()
317            );
318        }
319    }
320
321    #[rstest]
322    fn test_data_client_config_toml_minimal() {
323        let config: TardisDataClientConfig = toml::from_str(
324            r#"
325normalize_symbols = false
326book_snapshot_output = "depth"
327"#,
328        )
329        .unwrap();
330
331        assert!(!config.normalize_symbols);
332        assert!(matches!(
333            config.book_snapshot_output,
334            BookSnapshotOutput::Depth
335        ));
336        assert!(!config.extract_bbo_as_quotes);
337        assert!(config.options.is_empty());
338    }
339
340    #[rstest]
341    fn test_replay_config_omits_options_uses_empty_default() {
342        let config: TardisReplayConfig = toml::from_str(
343            r#"
344tardis_ws_url = "wss://example.com"
345normalize_symbols = false
346"#,
347        )
348        .unwrap();
349
350        assert!(config.options.is_empty());
351        assert_eq!(
352            config
353                .tardis_ws_url
354                .as_ref()
355                .map(SecretString::expose_secret),
356            Some("wss://example.com")
357        );
358        assert_eq!(config.normalize_symbols, Some(false));
359        assert_eq!(config.extract_bbo_as_quotes, None);
360    }
361
362    #[rstest]
363    fn test_replay_config_deserializes_compression() {
364        let json = r#"{
365            "tardis_ws_url": null,
366            "normalize_symbols": true,
367            "output_path": null,
368            "options": [],
369            "proxy_url": null,
370            "book_snapshot_output": "depth",
371            "extract_bbo_as_quotes": true,
372            "compression": "zstd"
373        }"#;
374
375        let config: TardisReplayConfig = serde_json::from_str(json).unwrap();
376
377        assert!(matches!(
378            config.book_snapshot_output,
379            Some(BookSnapshotOutput::Depth)
380        ));
381        assert_eq!(config.extract_bbo_as_quotes, Some(true));
382        assert!(matches!(config.compression, Some(ParquetCompression::Zstd)));
383    }
384
385    #[rstest]
386    #[case::replay_symbols(false, "symbols")]
387    #[case::replay_types(false, "data_types")]
388    #[case::replay_options(false, "options")]
389    #[case::stream_symbols(true, "symbols")]
390    #[case::stream_types(true, "data_types")]
391    #[case::stream_options(true, "stream_options")]
392    fn test_option_selection_duplicates(#[case] stream: bool, #[case] field: &str) {
393        let mut option = serde_json::json!({
394            "exchange": "binance-futures", "symbols": ["BTCUSDT"],
395            "dataTypes": ["trade"], "from": "2024-01-01", "to": "2024-01-02"
396        });
397
398        if stream {
399            option.as_object_mut().unwrap().remove("from");
400            option.as_object_mut().unwrap().remove("to");
401        }
402
403        match field {
404            "symbols" => option["symbols"] = serde_json::json!(["BTCUSDT", "ETHUSDT", "BTCUSDT"]),
405            "data_types" => option["dataTypes"] = serde_json::json!(["trade", "quote", "trade"]),
406            _ => {}
407        }
408
409        let mut options = vec![option.clone()];
410        if field == "options" || field == "stream_options" {
411            options.push(option);
412        }
413
414        let key = if stream { "stream_options" } else { "options" };
415        let mut input = serde_json::json!({});
416        input[key] = serde_json::json!(options);
417        let config: TardisDataClientConfig = serde_json::from_value(input.clone()).unwrap();
418
419        let expected_field = if field == key {
420            key.to_string()
421        } else {
422            format!("{key}[0].{field}")
423        };
424
425        let expected = Err(ConfigError::duplicate(expected_field, None));
426        let result = config.validate();
427
428        let replay_result = if stream {
429            None
430        } else {
431            Some(
432                serde_json::from_value::<TardisReplayConfig>(input)
433                    .unwrap()
434                    .validate(),
435            )
436        };
437
438        assert_eq!(result, expected);
439
440        if let Some(result) = replay_result {
441            assert_eq!(result, expected);
442        }
443    }
444}
nautilus_tardis/config.rs

nautilus_tardis/
config.rs
