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.
17
18use std::{
19    fmt::{Debug, Display},
20    fs, io,
21    sync::Arc,
22};
23
24use indexmap::IndexMap;
25use nautilus_core::Params;
26
27use super::{
28    feather::{RotationConfig, WriterClock},
29    filter::WriterRecordFilter,
30    traits::StreamingDataSink,
31};
32use crate::{
33    catalog::factory::CatalogConnectConfig,
34    common::{backend_name::backend_type, paths::local_writer_directory},
35};
36
37/// Built-in Feather streaming writer registry key.
38pub const FEATHER_WRITER_FACTORY_NAME: &str = "Feather";
39
40/// Built-in Parquet streaming writer registry key.
41pub const PARQUET_WRITER_FACTORY_NAME: &str = "Parquet";
42
43backend_type!(
44    /// Streaming writer backend used to persist run data, mirroring `CatalogBackendType`.
45    WriterBackendType {
46        Feather => FEATHER_WRITER_FACTORY_NAME,
47        Parquet => PARQUET_WRITER_FACTORY_NAME,
48    }
49);
50
51/// Connection settings handed to writer factories.
52#[derive(Clone, Debug)]
53pub struct WriterConnectConfig {
54    /// Local run-session directory the writer appends Feather files to.
55    pub uri: String,
56    /// Catalog that receives promoted data; required by every backend except `Feather`.
57    pub catalog: Option<CatalogConnectConfig>,
58    /// Rotation settings used by built-in streaming writer backends.
59    pub rotation_config: RotationConfig,
60    /// Optional automatic flush interval in milliseconds.
61    pub flush_interval_ms: Option<u64>,
62    /// Optional record-family and identifier filter.
63    pub record_filter: Option<WriterRecordFilter>,
64    /// Interval in milliseconds for promoting sealed files into `catalog`.
65    pub promotion_interval_ms: Option<u64>,
66    /// Whether closing the writer promotes remaining files into `catalog`.
67    pub promote_on_close: bool,
68    /// Whether Feather files are deleted after a successful promotion.
69    pub delete_feather_after_promotion: bool,
70    /// Whether promotion replaces `ts_init` with `ts_event`.
71    pub use_ts_event_for_ts_init: bool,
72    /// Backend-specific writer parameters.
73    pub params: Option<Params>,
74}
75
76impl WriterConnectConfig {
77    /// Creates a connect config for the given writer directory and promotion catalog.
78    #[must_use]
79    pub fn new(uri: impl Into<String>, catalog: Option<CatalogConnectConfig>) -> Self {
80        Self {
81            uri: uri.into(),
82            catalog,
83            rotation_config: RotationConfig::NoRotation,
84            flush_interval_ms: None,
85            record_filter: None,
86            promotion_interval_ms: None,
87            promote_on_close: true,
88            delete_feather_after_promotion: false,
89            use_ts_event_for_ts_init: false,
90            params: None,
91        }
92    }
93
94    /// Returns the catalog that receives promoted data.
95    ///
96    /// # Errors
97    ///
98    /// Returns an error if no catalog is configured. The error names `backend`.
99    pub fn required_catalog(&self, backend: &str) -> anyhow::Result<&CatalogConnectConfig> {
100        self.catalog
101            .as_ref()
102            .ok_or_else(|| anyhow::anyhow!("{backend} writer requires a promotion catalog"))
103    }
104}
105
106/// Factory constructing a streaming writer from connection settings and a clock.
107pub type WriterFactory =
108    Arc<dyn Fn(&WriterConnectConfig, WriterClock) -> anyhow::Result<StreamingDataSink>>;
109
110/// Ordered registry of writer factories keyed by name.
111pub type WriterFactoryRegistry = IndexMap<String, WriterFactory>;
112
113/// Resolves a writer backend through the factory registry and constructs the writer.
114///
115/// All variants resolve through the registry so built-ins and custom names share one code path.
116///
117/// # Errors
118///
119/// Returns an error if the backend's factory is not registered or construction fails.
120pub fn create_writer(
121    backend: &WriterBackendType,
122    config: &WriterConnectConfig,
123    clock: WriterClock,
124    factories: &WriterFactoryRegistry,
125) -> anyhow::Result<StreamingDataSink> {
126    let name = backend.to_string();
127    factories
128        .get(&name)
129        .ok_or_else(|| anyhow::anyhow!("No writer factory registered for '{name}'"))?(
130        config, clock
131    )
132}
133
134/// Deletes existing files below the writer directory.
135///
136/// # Errors
137///
138/// Returns an error if the writer directory is not local or deleting it fails.
139pub fn replace_existing_writer_data(config: &WriterConnectConfig) -> anyhow::Result<()> {
140    let directory = local_writer_directory(&config.uri)?;
141
142    // Keep the directory itself, which can be a mount point or a symlink to the real output
143    let entries = match fs::read_dir(&directory) {
144        Ok(entries) => entries,
145        Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),
146        Err(e) => return Err(e.into()),
147    };
148
149    for entry in entries {
150        let entry = entry?;
151        if entry.file_type()?.is_dir() {
152            fs::remove_dir_all(entry.path())?;
153        } else {
154            fs::remove_file(entry.path())?;
155        }
156    }
157
158    Ok(())
159}
160#[cfg(test)]
161mod tests {
162    use std::sync::atomic::AtomicU64;
163
164    use nautilus_common::enums::Environment;
165    use nautilus_core::UnixNanos;
166    use nautilus_model::{
167        data::{Data, InstrumentClose, InstrumentStatus, NautilusDataType, QuoteTick},
168        enums::{InstrumentCloseType, MarketStatusAction},
169        identifiers::InstrumentId,
170        instruments::{
171            InstrumentAny, NautilusInstrumentType,
172            stubs::{audusd_sim, futures_contract_es},
173        },
174        types::{Price, Quantity},
175    };
176    use rstest::rstest;
177    use tempfile::TempDir;
178
179    use super::*;
180    use crate::{
181        backend::{default_writer_factories, parquet::catalog::ParquetDataCatalog},
182        catalog::types::CatalogDataType,
183        writer::feather::FEATHER_PARTIAL_EXTENSION,
184    };
185
186    fn quote(ts_init: u64) -> QuoteTick {
187        QuoteTick::new(
188            InstrumentId::from("AUD/USD.SIM"),
189            Price::from("0.66"),
190            Price::from("0.67"),
191            Quantity::from("1000"),
192            Quantity::from("1000"),
193            UnixNanos::from(ts_init),
194            UnixNanos::from(ts_init),
195        )
196    }
197
198    #[rstest]
199    fn replace_existing_writer_data_removes_local_files() {
200        let directory = TempDir::new().unwrap();
201        let stale_file = directory.path().join("stale.feather");
202        let stale_type_directory = directory.path().join("quotes");
203        std::fs::write(&stale_file, b"stale").unwrap();
204        std::fs::create_dir(&stale_type_directory).unwrap();
205        std::fs::write(stale_type_directory.join("quotes_0.feather"), b"stale").unwrap();
206        let config =
207            WriterConnectConfig::new(format!("file://{}", directory.path().display()), None);
208
209        replace_existing_writer_data(&config).unwrap();
210
211        assert!(directory.path().is_dir());
212        assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 0);
213    }
214
215    #[rstest]
216    fn writer_record_filter_allows_empty_and_record_family_filters() {
217        let quotes = CatalogDataType::from(NautilusDataType::QuoteTick);
218        let trades = CatalogDataType::from(NautilusDataType::TradeTick);
219        let empty = WriterRecordFilter::new();
220        assert!(empty.contains(&quotes));
221        assert!(empty.allows(&quotes, None));
222        assert!(empty.allows(&trades, Some("AUD/USD.SIM")));
223
224        let mut filter = WriterRecordFilter::new();
225        filter.insert(NautilusDataType::QuoteTick, None);
226        assert!(filter.contains(&quotes));
227        assert!(!filter.contains(&trades));
228        assert!(filter.allows(&quotes, None));
229        assert!(filter.allows(&quotes, Some("AUD/USD.SIM")));
230        assert!(!filter.allows(&trades, Some("AUD/USD.SIM")));
231    }
232
233    #[rstest]
234    fn writer_record_filter_restricts_by_instrument_type() {
235        let mut filter = WriterRecordFilter::new();
236        filter.insert_instrument_type(NautilusInstrumentType::FuturesContract);
237
238        assert!(filter.contains(&NautilusDataType::Instrument.into()));
239        assert!(filter.allows(
240            &CatalogDataType::Instrument(NautilusInstrumentType::FuturesContract),
241            Some("ESM4.GLBX"),
242        ));
243        assert!(!filter.allows(
244            &CatalogDataType::Instrument(NautilusInstrumentType::Equity),
245            Some("AAPL.XNAS"),
246        ));
247        assert!(!filter.allows(&NautilusDataType::QuoteTick.into(), Some("ESM4.GLBX")));
248    }
249
250    #[rstest]
251    fn writer_record_filter_allows_data_and_instrument_type_union() {
252        let mut filter = WriterRecordFilter::new();
253        filter.insert(NautilusDataType::Bar, None);
254        filter.insert_instrument_type(NautilusInstrumentType::FuturesContract);
255
256        assert!(filter.allows(&NautilusDataType::Bar.into(), Some("ESM4.GLBX")));
257        assert!(filter.allows(
258            &CatalogDataType::Instrument(NautilusInstrumentType::FuturesContract),
259            Some("ESM4.GLBX"),
260        ));
261        assert!(!filter.allows(
262            &CatalogDataType::Instrument(NautilusInstrumentType::Equity),
263            Some("AAPL.XNAS"),
264        ));
265    }
266
267    #[rstest]
268    fn writer_record_filter_restricts_by_identifier() {
269        let quotes = CatalogDataType::from(NautilusDataType::QuoteTick);
270        let mut filter = WriterRecordFilter::new();
271        filter.insert(
272            NautilusDataType::QuoteTick,
273            Some(vec!["AUD/USD.SIM".to_string()]),
274        );
275
276        assert!(filter.contains(&quotes));
277        assert!(filter.allows(&quotes, Some("AUD/USD.SIM")));
278        assert!(!filter.allows(&quotes, Some("GBP/USD.SIM")));
279        assert!(!filter.allows(&quotes, None));
280        assert!(!filter.allows(&NautilusDataType::TradeTick.into(), Some("AUD/USD.SIM")));
281    }
282
283    #[rstest]
284    fn writer_backend_type_parses_built_ins_case_insensitively() {
285        assert_eq!(
286            "feather".parse::<WriterBackendType>().unwrap(),
287            WriterBackendType::Feather,
288        );
289        assert_eq!(
290            "parquet".parse::<WriterBackendType>().unwrap(),
291            WriterBackendType::Parquet,
292        );
293        assert_eq!(
294            "parquet".parse::<WriterBackendType>().unwrap(),
295            WriterBackendType::Parquet,
296        );
297        assert_eq!(WriterBackendType::Feather.to_string(), "Feather");
298        assert_eq!(WriterBackendType::Parquet.to_string(), "Parquet");
299        assert_eq!(WriterBackendType::Parquet.to_string(), "Parquet");
300    }
301
302    #[rstest]
303    fn writer_backend_type_preserves_custom_factory_case() {
304        assert_eq!(
305            "CustomSink".parse::<WriterBackendType>().unwrap(),
306            WriterBackendType::External("CustomSink".to_string()),
307        );
308        assert!("".parse::<WriterBackendType>().is_err());
309    }
310
311    #[rstest]
312    fn default_registry_contains_built_in_writer_factories() {
313        let registry = default_writer_factories();
314        assert!(registry.contains_key(FEATHER_WRITER_FACTORY_NAME));
315    }
316
317    #[rstest]
318    fn create_writer_builds_feather_writer_through_registry() {
319        let temp_dir = TempDir::new().unwrap();
320        let config = WriterConnectConfig::new(temp_dir.path().to_str().unwrap(), None);
321        let registry = default_writer_factories();
322
323        let mut writer = create_writer(
324            &WriterBackendType::Feather,
325            &config,
326            WriterClock::Live,
327            &registry,
328        )
329        .unwrap();
330        writer.write_data(Data::Quote(quote(100))).unwrap();
331        writer.flush().unwrap();
332        writer.close().unwrap();
333
334        assert_eq!(count_feather_files(temp_dir.path()), 1);
335    }
336
337    #[rstest]
338    #[case("backtest")]
339    #[case("live")]
340    fn create_writer_builds_feather_writer_for_run_session_kind(#[case] run_kind: &str) {
341        let temp_dir = TempDir::new().unwrap();
342        let session = temp_dir.path().join(run_kind).join("run-001");
343        let config = WriterConnectConfig::new(session.to_str().unwrap(), None);
344        let registry = default_writer_factories();
345
346        let mut writer = create_writer(
347            &WriterBackendType::Feather,
348            &config,
349            WriterClock::Live,
350            &registry,
351        )
352        .unwrap();
353        writer.write_data(Data::Quote(quote(100))).unwrap();
354        writer.close().unwrap();
355
356        assert_eq!(count_feather_files(&session), 1);
357    }
358
359    #[rstest]
360    fn create_writer_feather_recovers_partial_files_at_startup() {
361        let temp_dir = TempDir::new().unwrap();
362        let config = WriterConnectConfig::new(temp_dir.path().to_str().unwrap(), None);
363        let registry = default_writer_factories();
364        let start = || {
365            create_writer(
366                &WriterBackendType::Feather,
367                &config,
368                WriterClock::Test(Arc::new(AtomicU64::new(0))),
369                &registry,
370            )
371            .unwrap()
372        };
373
374        let mut crashed = start();
375        crashed.write_data(Data::Quote(quote(100))).unwrap();
376        crashed.close().unwrap();
377
378        // A writer that exited before sealing leaves its flushed stream as a partial file
379        let sealed = temp_dir.path().join("quotes").join("quotes_0.feather");
380        let partial = sealed.with_extension(FEATHER_PARTIAL_EXTENSION);
381        fs::rename(&sealed, &partial).unwrap();
382
383        let mut restarted = start();
384        let sealed_at_start = sealed.exists();
385        let partial_at_start = partial.exists();
386        restarted.close().unwrap();
387
388        assert!(sealed_at_start);
389        assert!(!partial_at_start);
390        assert_eq!(count_feather_files(temp_dir.path()), 1);
391    }
392
393    #[rstest]
394    fn create_writer_feather_round_trips_quotes_through_catalog() {
395        let temp_dir = TempDir::new().unwrap();
396        let session = temp_dir.path().join("backtest").join("run-001");
397        let expected = quote(100);
398        let config = WriterConnectConfig::new(session.to_str().unwrap(), None);
399        let registry = default_writer_factories();
400
401        let mut writer = create_writer(
402            &WriterBackendType::Feather,
403            &config,
404            WriterClock::Live,
405            &registry,
406        )
407        .unwrap();
408        writer.write_data(Data::Quote(expected)).unwrap();
409        writer.close().unwrap();
410
411        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
412        catalog
413            .convert_stream_to_data(
414                "run-001",
415                &CatalogDataType::from(NautilusDataType::QuoteTick),
416                Environment::Backtest,
417                None,
418                false,
419            )
420            .unwrap();
421
422        let loaded = catalog
423            .query::<QuoteTick>(None, None, None, None, None, true)
424            .unwrap();
425        assert_eq!(loaded, vec![expected]);
426    }
427
428    #[rstest]
429    fn create_writer_feather_round_trips_status_and_closes_for_each_instrument() {
430        let temp_dir = TempDir::new().unwrap();
431        let session = temp_dir.path().join("backtest").join("run-001");
432        let first = InstrumentId::from("AUD/USD.SIM");
433        let second = InstrumentId::from("GBP/USD.SIM");
434        let statuses = vec![
435            InstrumentStatus::new(
436                first,
437                MarketStatusAction::Trading,
438                UnixNanos::from(1),
439                UnixNanos::from(1),
440                None,
441                None,
442                Some(true),
443                None,
444                None,
445            ),
446            InstrumentStatus::new(
447                second,
448                MarketStatusAction::Halt,
449                UnixNanos::from(2),
450                UnixNanos::from(2),
451                None,
452                None,
453                Some(false),
454                None,
455                None,
456            ),
457        ];
458        let closes = vec![
459            InstrumentClose::new(
460                first,
461                Price::from("0.66"),
462                InstrumentCloseType::EndOfSession,
463                UnixNanos::from(3),
464                UnixNanos::from(3),
465            ),
466            InstrumentClose::new(
467                second,
468                Price::from("1.250"),
469                InstrumentCloseType::ContractExpired,
470                UnixNanos::from(4),
471                UnixNanos::from(4),
472            ),
473        ];
474        let config = WriterConnectConfig::new(session.to_str().unwrap(), None);
475        let registry = default_writer_factories();
476
477        let mut writer = create_writer(
478            &WriterBackendType::Feather,
479            &config,
480            WriterClock::Live,
481            &registry,
482        )
483        .unwrap();
484
485        for status in &statuses {
486            writer.write_data(Data::InstrumentStatus(*status)).unwrap();
487        }
488
489        for close in &closes {
490            writer.write_data(Data::InstrumentClose(*close)).unwrap();
491        }
492
493        writer.close().unwrap();
494
495        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
496
497        for data_type in [
498            NautilusDataType::InstrumentStatus,
499            NautilusDataType::InstrumentClose,
500        ] {
501            catalog
502                .convert_stream_to_data(
503                    "run-001",
504                    &CatalogDataType::from(data_type),
505                    Environment::Backtest,
506                    None,
507                    false,
508                )
509                .unwrap();
510        }
511
512        let loaded_statuses = catalog
513            .query::<InstrumentStatus>(None, None, None, None, None, true)
514            .unwrap();
515        let loaded_closes = catalog
516            .query::<InstrumentClose>(None, None, None, None, None, true)
517            .unwrap();
518        assert_eq!(loaded_statuses, statuses);
519        assert_eq!(loaded_closes, closes);
520    }
521
522    #[rstest]
523    fn create_writer_feather_filters_and_converts_instruments_by_type() {
524        let temp_dir = TempDir::new().unwrap();
525        let session = temp_dir.path().join("backtest").join("run-001");
526        let futures = InstrumentAny::FuturesContract(futures_contract_es(None, None));
527        let mut record_filter = WriterRecordFilter::new();
528        record_filter.insert_instrument_type(NautilusInstrumentType::FuturesContract);
529        let mut config = WriterConnectConfig::new(session.to_str().unwrap(), None);
530        config.record_filter = Some(record_filter);
531        let registry = default_writer_factories();
532
533        let mut writer = create_writer(
534            &WriterBackendType::Feather,
535            &config,
536            WriterClock::Live,
537            &registry,
538        )
539        .unwrap();
540        writer
541            .write_data(Data::Instrument(Box::new(InstrumentAny::CurrencyPair(
542                audusd_sim(),
543            ))))
544            .unwrap();
545        writer
546            .write_data(Data::Instrument(Box::new(futures.clone())))
547            .unwrap();
548        writer.close().unwrap();
549
550        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
551        catalog
552            .convert_stream_to_data(
553                "run-001",
554                &CatalogDataType::from(NautilusDataType::Instrument),
555                Environment::Backtest,
556                None,
557                false,
558            )
559            .unwrap();
560
561        let loaded = catalog.query_instruments(None).unwrap();
562        assert_eq!(
563            serde_json::to_value(&loaded).unwrap(),
564            serde_json::to_value(vec![futures]).unwrap()
565        );
566    }
567
568    #[rstest]
569    fn create_writer_errors_for_missing_external() {
570        let temp_dir = TempDir::new().unwrap();
571        let session = temp_dir.path().join("backtest").join("run-001");
572        let config = WriterConnectConfig::new(session.to_str().unwrap(), None);
573        let registry = default_writer_factories();
574
575        let result = create_writer(
576            &WriterBackendType::External("Missing".to_string()),
577            &config,
578            WriterClock::Live,
579            &registry,
580        );
581        assert!(
582            result
583                .unwrap_err()
584                .to_string()
585                .contains("No writer factory registered for 'Missing'"),
586        );
587    }
588
589    #[rstest]
590    #[case::feather(WriterBackendType::Feather)]
591    // Without the cloud feature, the Parquet writer rejects an S3 URI before the local-path check
592    #[cfg_attr(feature = "cloud", case::parquet(WriterBackendType::Parquet))]
593    fn create_writer_rejects_remote_staging_uri(#[case] backend: WriterBackendType) {
594        let catalog = TempDir::new().unwrap();
595
596        let config = WriterConnectConfig::new(
597            "s3://bucket/backtest/run-001",
598            Some(CatalogConnectConfig::new(
599                catalog.path().to_string_lossy(),
600                None,
601            )),
602        );
603
604        let error = create_writer(
605            &backend,
606            &config,
607            WriterClock::Live,
608            &default_writer_factories(),
609        )
610        .unwrap_err();
611
612        assert_eq!(
613            error.to_string(),
614            "Streaming writers append to local files, writer path must be local, was \
615             s3://bucket/backtest/run-001",
616        );
617    }
618
619    #[rstest]
620    fn replace_existing_writer_data_rejects_remote_uri() {
621        let config = WriterConnectConfig::new("s3://bucket/backtest/run-001", None);
622
623        let error = replace_existing_writer_data(&config).unwrap_err();
624
625        assert_eq!(
626            error.to_string(),
627            "Streaming writers append to local files, writer path must be local, was \
628             s3://bucket/backtest/run-001",
629        );
630    }
631
632    fn count_feather_files(dir: &std::path::Path) -> usize {
633        let mut count = 0;
634        let mut stack = vec![dir.to_path_buf()];
635        while let Some(path) = stack.pop() {
636            let Ok(entries) = std::fs::read_dir(&path) else {
637                continue;
638            };
639
640            for entry in entries.flatten() {
641                let path = entry.path();
642                if path.is_dir() {
643                    stack.push(path);
644                } else if path
645                    .extension()
646                    .is_some_and(|extension| extension == "feather")
647                {
648                    count += 1;
649                }
650            }
651        }
652
653        count
654    }
655}
nautilus_persistence/writer/factory.rs

nautilus_persistence/writer/
factory.rs
