nautilus_persistence/writer/
factory.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//! Streaming writer factory registry and connections.
17use std::{
18    fmt::{Debug, Display},
19    sync::Arc,
20};
21
22use ahash::AHashMap;
23use indexmap::IndexMap;
24use nautilus_common::live::block_on_nautilus_with;
25use nautilus_core::Params;
26use object_store::{ObjectStoreExt, path::Path as ObjectPath};
27
28use super::{
29    feather::{RotationConfig, WriterClock},
30    filter::WriterRecordFilter,
31    traits::StreamingDataSink,
32};
33use crate::common::{backend_name::backend_type, storage::create_storage_backend_from_path};
34
35/// Built-in Feather streaming writer registry key.
36pub const FEATHER_WRITER_FACTORY_NAME: &str = "Feather";
37
38/// Built-in Parquet streaming writer registry key.
39pub const PARQUET_WRITER_FACTORY_NAME: &str = "Parquet";
40
41backend_type!(
42    /// Streaming writer backend used to persist run data, mirroring `CatalogBackendType`.
43    WriterBackendType {
44        Feather => FEATHER_WRITER_FACTORY_NAME,
45        Parquet => PARQUET_WRITER_FACTORY_NAME,
46    }
47);
48
49/// Connection settings handed to writer factories.
50#[derive(Clone, Debug)]
51pub struct WriterConnectConfig {
52    /// Run-session storage URI the writer stages into.
53    pub uri: String,
54    /// Backend-specific storage options (credentials, endpoints).
55    pub storage_options: Option<AHashMap<String, String>>,
56    /// Rotation settings used by built-in streaming writer backends.
57    pub rotation_config: RotationConfig,
58    /// Optional automatic flush interval in milliseconds.
59    pub flush_interval_ms: Option<u64>,
60    /// Optional record-family and identifier filter.
61    pub record_filter: Option<WriterRecordFilter>,
62    /// Backend-specific writer parameters.
63    pub params: Option<Params>,
64}
65
66impl WriterConnectConfig {
67    /// Creates a connect config for the given URI.
68    #[must_use]
69    pub fn new(uri: impl Into<String>, storage_options: Option<AHashMap<String, String>>) -> Self {
70        Self {
71            uri: uri.into(),
72            storage_options,
73            rotation_config: RotationConfig::NoRotation,
74            flush_interval_ms: None,
75            record_filter: None,
76            params: None,
77        }
78    }
79}
80
81/// Factory constructing a streaming writer from connection settings and a clock.
82pub type WriterFactory =
83    Arc<dyn Fn(&WriterConnectConfig, WriterClock) -> anyhow::Result<StreamingDataSink>>;
84
85/// Ordered registry of writer factories keyed by name.
86pub type WriterFactoryRegistry = IndexMap<String, WriterFactory>;
87
88/// Resolves a writer backend through the factory registry and constructs the writer.
89///
90/// All variants resolve through the registry so built-ins and custom names share one code path.
91///
92/// # Errors
93///
94/// Returns an error if the backend's factory is not registered or construction fails.
95pub fn create_writer(
96    backend: &WriterBackendType,
97    config: &WriterConnectConfig,
98    clock: WriterClock,
99    factories: &WriterFactoryRegistry,
100) -> anyhow::Result<StreamingDataSink> {
101    let name = backend.to_string();
102    factories
103        .get(&name)
104        .ok_or_else(|| anyhow::anyhow!("No writer factory registered for '{name}'"))?(
105        config, clock
106    )
107}
108
109/// Deletes existing objects below writer connection URI.
110///
111/// # Errors
112///
113/// Returns an error if storage cannot be opened or listing/deleting objects fails.
114pub fn replace_existing_writer_data(config: &WriterConnectConfig) -> anyhow::Result<()> {
115    let storage = create_storage_backend_from_path(&config.uri, config.storage_options.clone())?;
116    block_on_nautilus_with(|| async {
117        for path in storage.list_files("", None).await? {
118            storage.object_store.delete(&ObjectPath::from(path)).await?;
119        }
120        Ok::<(), anyhow::Error>(())
121    })
122}
123#[cfg(test)]
124mod tests {
125    use nautilus_core::UnixNanos;
126    use nautilus_model::{
127        data::{Data, QuoteTick},
128        identifiers::InstrumentId,
129        instruments::{InstrumentAny, NautilusInstrumentType},
130        types::{Price, Quantity},
131    };
132    use rstest::rstest;
133    use tempfile::TempDir;
134
135    use super::*;
136    use crate::{backend::default_writer_factories, common::paths::CatalogPathPrefix};
137
138    fn quote(ts_init: u64) -> QuoteTick {
139        QuoteTick::new(
140            InstrumentId::from("AUD/USD.SIM"),
141            Price::from("0.66"),
142            Price::from("0.67"),
143            Quantity::from("1000"),
144            Quantity::from("1000"),
145            UnixNanos::from(ts_init),
146            UnixNanos::from(ts_init),
147        )
148    }
149
150    #[rstest]
151    fn replace_existing_writer_data_removes_local_files() {
152        let directory = TempDir::new().unwrap();
153        let stale_file = directory.path().join("stale.feather");
154        std::fs::write(&stale_file, b"stale").unwrap();
155        let config =
156            WriterConnectConfig::new(format!("file://{}", directory.path().display()), None);
157
158        replace_existing_writer_data(&config).unwrap();
159
160        assert!(!stale_file.exists());
161    }
162
163    #[rstest]
164    fn writer_record_filter_allows_empty_and_record_family_filters() {
165        let empty = WriterRecordFilter::new();
166        assert!(empty.contains_prefix("quotes"));
167        assert!(empty.allows("quotes", None, None));
168        assert!(empty.allows("trades", Some("AUD/USD.SIM"), None));
169
170        let mut filter = WriterRecordFilter::new();
171        filter.insert_prefix("quotes", None);
172        assert!(filter.contains_prefix("quotes"));
173        assert!(!filter.contains_prefix("trades"));
174        assert!(filter.allows("quotes", None, None));
175        assert!(filter.allows("quotes", Some("AUD/USD.SIM"), None));
176        assert!(!filter.allows("trades", Some("AUD/USD.SIM"), None));
177    }
178
179    #[rstest]
180    fn writer_record_filter_restricts_by_instrument_type() {
181        let mut filter = WriterRecordFilter::new();
182        filter.insert_instrument_type(&NautilusInstrumentType::FuturesContract);
183
184        assert!(filter.contains_prefix(InstrumentAny::path_prefix()));
185        assert!(filter.allows(
186            InstrumentAny::path_prefix(),
187            Some("ESM4.GLBX"),
188            Some("FuturesContract")
189        ));
190        assert!(!filter.allows(
191            InstrumentAny::path_prefix(),
192            Some("AAPL.XNAS"),
193            Some("Equity")
194        ));
195        assert!(!filter.allows("quotes", Some("ESM4.GLBX"), None));
196    }
197
198    #[rstest]
199    fn writer_record_filter_allows_data_and_instrument_type_union() {
200        let mut filter = WriterRecordFilter::new();
201        filter.insert_prefix("bars", None);
202        filter.insert_instrument_type(&NautilusInstrumentType::FuturesContract);
203
204        assert!(filter.allows("bars", Some("ESM4.GLBX"), None));
205        assert!(filter.allows(
206            InstrumentAny::path_prefix(),
207            Some("ESM4.GLBX"),
208            Some("FuturesContract")
209        ));
210        assert!(!filter.allows(
211            InstrumentAny::path_prefix(),
212            Some("AAPL.XNAS"),
213            Some("Equity")
214        ));
215    }
216
217    #[rstest]
218    fn writer_record_filter_restricts_by_identifier() {
219        let mut filter = WriterRecordFilter::new();
220        filter.insert_prefix("quotes", Some(vec!["AUD/USD.SIM".to_string()]));
221
222        assert!(filter.contains_prefix("quotes"));
223        assert!(filter.allows("quotes", Some("AUD/USD.SIM"), None));
224        assert!(!filter.allows("quotes", Some("GBP/USD.SIM"), None));
225        assert!(!filter.allows("quotes", None, None));
226        assert!(!filter.allows("trades", Some("AUD/USD.SIM"), None));
227    }
228
229    #[rstest]
230    fn writer_backend_type_parses_built_ins_case_insensitively() {
231        assert_eq!(
232            "feather".parse::<WriterBackendType>().unwrap(),
233            WriterBackendType::Feather,
234        );
235        assert_eq!(
236            "parquet".parse::<WriterBackendType>().unwrap(),
237            WriterBackendType::Parquet,
238        );
239        assert_eq!(
240            "parquet".parse::<WriterBackendType>().unwrap(),
241            WriterBackendType::Parquet,
242        );
243        assert_eq!(WriterBackendType::Feather.to_string(), "Feather");
244        assert_eq!(WriterBackendType::Parquet.to_string(), "Parquet");
245        assert_eq!(WriterBackendType::Parquet.to_string(), "Parquet");
246    }
247
248    #[rstest]
249    fn writer_backend_type_preserves_custom_factory_case() {
250        assert_eq!(
251            "CustomSink".parse::<WriterBackendType>().unwrap(),
252            WriterBackendType::External("CustomSink".to_string()),
253        );
254        assert!("".parse::<WriterBackendType>().is_err());
255    }
256
257    #[rstest]
258    fn default_registry_contains_built_in_writer_factories() {
259        let registry = default_writer_factories();
260        assert!(registry.contains_key(FEATHER_WRITER_FACTORY_NAME));
261    }
262
263    #[rstest]
264    fn create_writer_builds_feather_writer_through_registry() {
265        let temp_dir = TempDir::new().unwrap();
266        let config = WriterConnectConfig::new(temp_dir.path().to_str().unwrap(), None);
267        let registry = default_writer_factories();
268
269        let mut writer = create_writer(
270            &WriterBackendType::Feather,
271            &config,
272            WriterClock::Live,
273            &registry,
274        )
275        .unwrap();
276        writer.write_data(Data::Quote(quote(100))).unwrap();
277        writer.flush().unwrap();
278        writer.close().unwrap();
279
280        let feather_files = std::fs::read_dir(temp_dir.path())
281            .unwrap()
282            .filter_map(std::result::Result::ok)
283            .filter(|entry| {
284                entry
285                    .path()
286                    .extension()
287                    .is_some_and(|extension| extension == "feather")
288            })
289            .count();
290        assert_eq!(feather_files, 1);
291    }
292
293    #[rstest]
294    #[case("backtest")]
295    #[case("live")]
296    fn create_writer_builds_feather_writer_for_run_session_kind(#[case] run_kind: &str) {
297        let temp_dir = TempDir::new().unwrap();
298        let session = temp_dir.path().join(run_kind).join("run-001");
299        let config = WriterConnectConfig::new(session.to_str().unwrap(), None);
300        let registry = default_writer_factories();
301
302        let mut writer = create_writer(
303            &WriterBackendType::Feather,
304            &config,
305            WriterClock::Live,
306            &registry,
307        )
308        .unwrap();
309        writer.write_data(Data::Quote(quote(100))).unwrap();
310        writer.close().unwrap();
311
312        let feather_files = std::fs::read_dir(&session)
313            .unwrap()
314            .filter_map(Result::ok)
315            .filter(|entry| {
316                entry
317                    .path()
318                    .extension()
319                    .is_some_and(|extension| extension == "feather")
320            })
321            .count();
322        assert_eq!(feather_files, 1);
323    }
324
325    #[rstest]
326    fn create_writer_errors_for_missing_external() {
327        let temp_dir = TempDir::new().unwrap();
328        let session = temp_dir.path().join("backtest").join("run-001");
329        let config = WriterConnectConfig::new(session.to_str().unwrap(), None);
330        let registry = default_writer_factories();
331
332        let result = create_writer(
333            &WriterBackendType::External("Missing".to_string()),
334            &config,
335            WriterClock::Live,
336            &registry,
337        );
338        assert!(
339            result
340                .unwrap_err()
341                .to_string()
342                .contains("No writer factory registered for 'Missing'"),
343        );
344    }
345}
nautilus_persistence/writer/factory.rs

nautilus_persistence/writer/
factory.rs
