nautilus_persistence/backend/parquet/
feather_session.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//! Feather-file session reading and stream-to-parquet conversion.
17//!
18//! Methods for reading per-run feather files written by the live/backtest writers and
19//! converting them to consolidated parquet for catalog ingest.
20
21#![expect(
22    clippy::unused_self,
23    reason = "session registration keeps backend-specific ordering logic together"
24)]
25
26use std::{borrow::Cow, sync::Arc};
27
28use datafusion::arrow::{datatypes::Schema, record_batch::RecordBatch};
29use futures::StreamExt;
30use indexmap::IndexMap;
31use nautilus_core::UnixNanos;
32use nautilus_model::data::{
33    Bar, Data, FundingRateUpdate, HasTsInit, IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate,
34    OptionGreeks, OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick, close::InstrumentClose,
35    to_variant,
36};
37use nautilus_serialization::arrow::{
38    DecodeDataFromRecordBatch, DecodeTypedFromRecordBatch, U64ColumnRef,
39};
40use object_store::path::Path as ObjectPath;
41
42use crate::{
43    backend::parquet::{
44        catalog::ParquetDataCatalog,
45        paths::{make_object_store_path, urisafe_instrument_id},
46    },
47    catalog::types::{CatalogDataType, parquet_data_path_prefix, record_path_prefix},
48    common::{
49        conversion::FeatherConversionSummary,
50        custom::decode_custom_batches_to_data,
51        datafusion::identifiers_from_record_batches,
52        paths::{identifier_from_session_feather_path, type_name_from_session_feather_path},
53    },
54    writer::{
55        materializer::{
56            StreamConversionOptions, apply_stream_conversion_transform,
57            coalesce_stream_conversion_batches, read_feather_record_batches,
58            restore_staged_record_batches,
59        },
60        run::FeatherSessionSource,
61    },
62};
63
64impl ParquetDataCatalog {
65    pub(crate) fn promote_feather_file(
66        &self,
67        source: &FeatherSessionSource,
68        feather_path: &str,
69        batches: Vec<RecordBatch>,
70        use_ts_event_for_ts_init: bool,
71        replay_identity: &str,
72    ) -> anyhow::Result<Option<FeatherConversionSummary>> {
73        if batches.is_empty() {
74            return Ok(None);
75        }
76        let batches = Self::restore_staged_batches(batches)?;
77        let type_name =
78            type_name_from_session_feather_path(feather_path, &source.kind, &source.instance_id)?;
79        let catalog_data_name = Self::canonical_stream_data_name(&type_name);
80        anyhow::ensure!(
81            Self::is_supported_stream_data_type(catalog_data_name),
82            "Unknown data class: {type_name}"
83        );
84        let identifier = Self::identifier_from_batch_or_path(
85            &batches[0],
86            feather_path,
87            &source.kind,
88            &source.instance_id,
89        )
90        .filter(|identifier| {
91            batches.iter().all(|batch| {
92                Self::identifier_from_batch_or_path(
93                    batch,
94                    feather_path,
95                    &source.kind,
96                    &source.instance_id,
97                )
98                .as_ref()
99                    == Some(identifier)
100            })
101        });
102
103        self.convert_feather_batches_to_parquet(
104            &source.kind,
105            &source.instance_id,
106            catalog_data_name,
107            feather_path,
108            &batches,
109            use_ts_event_for_ts_init,
110            Some(replay_identity),
111        )?;
112        Ok(Some(FeatherConversionSummary {
113            type_name,
114            identifier,
115            feather_path: feather_path.to_string(),
116            native_version: None,
117            unmatched_identifiers: None,
118        }))
119    }
120
121    /// Reads data from a live run instance.
122    ///
123    /// This method reads all data associated with a specific live run instance
124    /// from feather files stored in the catalog.
125    ///
126    /// # Parameters
127    ///
128    /// - `instance_id`: The ID of the live run instance to read.
129    ///
130    /// # Returns
131    ///
132    /// Returns a vector of `Data` objects from the live run, sorted by timestamp,
133    /// or an error if the operation fails.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if:
138    /// - The instance ID doesn't exist.
139    /// - Feather file reading fails.
140    /// - Data deserialization fails.
141    ///
142    /// # Note
143    ///
144    /// This method reads through the run reader: it lists the run's data-type directories, reads
145    /// every Feather file through the Arrow IPC stream reader with staged batch restoration, decodes
146    /// quotes, trades, order book deltas and depths, bars, index and mark prices, option Greeks,
147    /// funding rates, instrument status and closes, and custom data files into `Data` values, skips
148    /// unknown data types, and sorts the result by `ts_init`.
149    ///
150    /// # Examples
151    ///
152    /// ```rust,no_run
153    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
154    ///
155    /// let mut catalog = ParquetDataCatalog::new(
156    ///     std::path::Path::new("/tmp/nautilus_data"),
157    ///     None,
158    ///     None,
159    ///     None,
160    ///     None,
161    /// );
162    ///
163    /// // Read data from a live run
164    /// let data = catalog.read_live_run("instance-123")?;
165    /// for item in data {
166    ///     println!("Data: {:?}", item);
167    /// }
168    /// # Ok::<(), anyhow::Error>(())
169    /// ```
170    pub fn read_live_run(&self, instance_id: &str) -> anyhow::Result<Vec<Data>> {
171        self.read_run_data("live", instance_id)
172    }
173
174    /// Reads data from a backtest run instance.
175    ///
176    /// This method reads all data associated with a specific backtest run instance
177    /// from feather files stored in the catalog.
178    ///
179    /// # Parameters
180    ///
181    /// - `instance_id`: The ID of the backtest run instance to read.
182    ///
183    /// # Returns
184    ///
185    /// Returns a vector of `Data` objects from the backtest run, sorted by timestamp,
186    /// or an error if the operation fails.
187    ///
188    /// # Errors
189    ///
190    /// Returns an error if:
191    /// - The instance ID doesn't exist.
192    /// - Feather file reading fails.
193    /// - Data deserialization fails.
194    ///
195    /// # Examples
196    ///
197    /// ```rust,no_run
198    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
199    ///
200    /// let mut catalog = ParquetDataCatalog::new(
201    ///     std::path::Path::new("/tmp/nautilus_data"),
202    ///     None,
203    ///     None,
204    ///     None,
205    ///     None,
206    /// );
207    ///
208    /// // Read data from a backtest run
209    /// let data = catalog.read_backtest("instance-123")?;
210    /// for item in data {
211    ///     println!("Data: {:?}", item);
212    /// }
213    /// # Ok::<(), anyhow::Error>(())
214    /// ```
215    pub fn read_backtest(&self, instance_id: &str) -> anyhow::Result<Vec<Data>> {
216        self.read_run_data("backtest", instance_id)
217    }
218
219    /// Helper function to read data from a run instance (backtest or live).
220    ///
221    /// This function reads all data associated with a specific run instance
222    /// from feather files stored in the catalog.
223    ///
224    /// # Parameters
225    ///
226    /// - `subdirectory`: The subdirectory name ("backtest" or "live").
227    /// - `instance_id`: The ID of the run instance to read.
228    ///
229    /// # Returns
230    ///
231    /// Returns a vector of `Data` objects from the run, sorted by timestamp,
232    /// or an error if the operation fails.
233    fn read_run_data(&self, subdirectory: &str, instance_id: &str) -> anyhow::Result<Vec<Data>> {
234        // List all data type directories in the instance directory
235        let data_types = self.list_directory_stems(&format!("{subdirectory}/{instance_id}"))?;
236
237        if data_types.is_empty() {
238            // No data types found - return empty vector
239            return Ok(Vec::new());
240        }
241
242        let mut all_data: Vec<Data> = Vec::new();
243
244        // Process each persisted data type.
245        for data_cls in data_types {
246            // List all feather files for this data type
247            let feather_files = self.list_feather_files(
248                subdirectory,
249                instance_id,
250                &data_cls,
251                None, // No identifier filtering - read all
252            )?;
253
254            if feather_files.is_empty() {
255                continue; // Skip if no files found
256            }
257
258            // Process each feather file
259            for file_path in feather_files {
260                // Read the feather file (may contain multiple batches)
261                let batches = self.read_feather_file(&file_path)?;
262
263                if batches.is_empty() {
264                    continue; // Skip empty or invalid files
265                }
266
267                let decode_data_cls = Self::canonical_stream_data_name(&data_cls);
268
269                // Convert RecordBatches to Data objects based on data_cls
270                let file_data: Vec<Data> = match decode_data_cls {
271                    "quotes" => {
272                        let quotes: Vec<QuoteTick> =
273                            self.convert_record_batches_to_data(batches, false)?;
274                        quotes.into_iter().map(Data::from).collect()
275                    }
276                    "trades" => {
277                        let trades: Vec<TradeTick> =
278                            self.convert_record_batches_to_data(batches, false)?;
279                        trades.into_iter().map(Data::from).collect()
280                    }
281                    "order_book_deltas" => {
282                        let deltas: Vec<OrderBookDelta> =
283                            self.convert_record_batches_to_data(batches, false)?;
284                        deltas.into_iter().map(Data::from).collect()
285                    }
286                    "order_book_depths" => {
287                        let depths: Vec<OrderBookDepth> =
288                            self.convert_record_batches_to_data(batches, false)?;
289                        depths.into_iter().map(Data::from).collect()
290                    }
291                    "bars" => {
292                        let bars: Vec<Bar> = self.convert_record_batches_to_data(batches, false)?;
293                        bars.into_iter().map(Data::from).collect()
294                    }
295                    "index_prices" => {
296                        let prices: Vec<IndexPriceUpdate> =
297                            self.convert_record_batches_to_data(batches, false)?;
298                        prices.into_iter().map(Data::from).collect()
299                    }
300                    "mark_prices" => {
301                        let prices: Vec<MarkPriceUpdate> =
302                            self.convert_record_batches_to_data(batches, false)?;
303                        prices.into_iter().map(Data::from).collect()
304                    }
305                    "option_greeks" => {
306                        let greeks: Vec<OptionGreeks> =
307                            self.convert_record_batches_to_data(batches, false)?;
308                        greeks.into_iter().map(Data::from).collect()
309                    }
310                    "funding_rates" => {
311                        let funding_rates: Vec<FundingRateUpdate> =
312                            self.convert_record_batches_to_data(batches, false)?;
313                        funding_rates.into_iter().map(Data::from).collect()
314                    }
315                    "instrument_status" => {
316                        let statuses: Vec<InstrumentStatus> =
317                            self.convert_record_batches_to_data(batches, false)?;
318                        statuses.into_iter().map(Data::from).collect()
319                    }
320                    "instrument_closes" => {
321                        let closes: Vec<InstrumentClose> =
322                            self.convert_record_batches_to_data(batches, false)?;
323                        closes.into_iter().map(Data::from).collect()
324                    }
325                    _ => {
326                        if decode_data_cls.starts_with("custom/") {
327                            decode_custom_batches_to_data(batches, false)?
328                        } else {
329                            // Unknown data type - skip it
330                            continue;
331                        }
332                    }
333                };
334
335                all_data.extend(file_data);
336            }
337        }
338
339        // Sort all data by timestamp (ts_init)
340        all_data.sort_by_key(HasTsInit::ts_init);
341
342        Ok(all_data)
343    }
344
345    /// Lists feather files for a specific data class in a subdirectory.
346    ///
347    /// This function finds all `.feather` files in the specified subdirectory
348    /// (backtest or live) for the given instance ID and data class.
349    fn list_feather_files(
350        &self,
351        subdirectory: &str,
352        instance_id: &str,
353        data_name: &str,
354        identifiers: Option<&[String]>,
355    ) -> anyhow::Result<Vec<String>> {
356        let base_dir = make_object_store_path(&self.base_path, [subdirectory, instance_id]);
357
358        let mut files = Vec::new();
359
360        let list_result = self.execute_async(|| async {
361            let prefix = ObjectPath::from(format!("{base_dir}/"));
362            let mut stream = self.object_store.list(Some(&prefix));
363            let mut feather_files = Vec::new();
364
365            while let Some(object) = stream.next().await {
366                let object = object?;
367                let path_str = object.location.to_string();
368
369                if !path_str.ends_with(".feather") {
370                    continue;
371                }
372
373                let Ok(path_data_name) =
374                    type_name_from_session_feather_path(&path_str, subdirectory, instance_id)
375                else {
376                    continue;
377                };
378
379                if path_data_name != data_name {
380                    continue;
381                }
382
383                let path_identifier =
384                    identifier_from_session_feather_path(&path_str, subdirectory, instance_id);
385
386                if let (Some(identifiers), Some(path_identifier)) =
387                    (identifiers, path_identifier.as_deref())
388                    && !Self::stream_identifier_matches(path_identifier, identifiers)
389                {
390                    continue;
391                }
392
393                feather_files.push(path_str);
394            }
395
396            Ok::<Vec<String>, anyhow::Error>(feather_files)
397        })?;
398
399        files.extend(list_result);
400        files.sort();
401        Ok(files)
402    }
403
404    fn stream_identifier_matches(candidate: &str, identifiers: &[String]) -> bool {
405        identifiers.iter().any(|id| {
406            let safe_id = urisafe_instrument_id(id);
407            candidate.contains(id) || candidate.contains(&safe_id)
408        })
409    }
410
411    /// Reads a feather file and returns all `RecordBatches`.
412    fn read_feather_file(&self, file_path: &str) -> anyhow::Result<Vec<RecordBatch>> {
413        let path = ObjectPath::from(file_path);
414        let batches = self.execute_async(|| async {
415            read_feather_record_batches(self.object_store.clone(), &path).await
416        })?;
417        Self::restore_staged_batches(batches)
418    }
419
420    fn restore_staged_batches(batches: Vec<RecordBatch>) -> anyhow::Result<Vec<RecordBatch>> {
421        let mut restored = Vec::new();
422        for batch in batches {
423            restored.extend(restore_staged_record_batches(batch)?);
424        }
425        Ok(restored)
426    }
427
428    /// Converts `RecordBatches` to Data objects, optionally replacing `ts_init` with `ts_event`.
429    fn convert_record_batches_to_data<T>(
430        &self,
431        batches: Vec<RecordBatch>,
432        use_ts_event_for_ts_init: bool,
433    ) -> anyhow::Result<Vec<T>>
434    where
435        T: DecodeDataFromRecordBatch + TryFrom<Data>,
436    {
437        if batches.is_empty() {
438            return Ok(Vec::new());
439        }
440
441        let mut all_data = Vec::new();
442
443        for batch in batches {
444            let batch = apply_stream_conversion_transform(
445                &batch,
446                StreamConversionOptions {
447                    use_ts_event_for_ts_init,
448                    convert_bar_type_to_external: false,
449                },
450            )?;
451            let metadata = batch.schema().metadata().clone();
452
453            let data_vec = T::decode_data_batch(&metadata, batch)
454                .map_err(|e| anyhow::anyhow!("Failed to decode batch: {e}"))?;
455
456            all_data.extend(data_vec);
457        }
458
459        Ok(to_variant::<T>(all_data))
460    }
461
462    /// Converts `RecordBatches` directly to strongly typed values.
463    pub(crate) fn convert_record_batches_to_typed<T>(
464        &self,
465        batches: Vec<RecordBatch>,
466    ) -> anyhow::Result<Vec<T>>
467    where
468        T: DecodeTypedFromRecordBatch,
469    {
470        if batches.is_empty() {
471            return Ok(Vec::new());
472        }
473
474        let mut all_data = Vec::new();
475
476        for batch in batches {
477            let metadata = batch.schema().metadata().clone();
478            let decoded = T::decode_typed_batch(&metadata, batch)
479                .map_err(|e| anyhow::anyhow!("Failed to decode batch: {e}"))?;
480            all_data.extend(decoded);
481        }
482
483        Ok(all_data)
484    }
485
486    /// Converts stream data from feather files to parquet files.
487    ///
488    /// This method reads data from feather files generated during a backtest or live run
489    /// and writes it to the catalog in parquet format. It's useful for converting temporary
490    /// stream data into a more permanent and queryable format.
491    ///
492    /// # Parameters
493    ///
494    /// - `instance_id`: The ID of the backtest or live run instance.
495    /// - `data_cls`: The data class name (e.g., "quotes", "trades", "bars"), or
496    ///   `custom/{TypeName}` with the registered type name verbatim for custom data.
497    /// - `subdirectory`: The subdirectory containing the feather files. Either "backtest" or "live" (default: "backtest").
498    /// - `identifiers`: Optional list of identifiers to filter by (instrument IDs or bar types).
499    /// - `use_ts_event_for_ts_init`: If true, replaces the `ts_init` column with `ts_event` column values before deserializing.
500    ///
501    /// # Returns
502    ///
503    /// Returns `Ok(())` on success, or an error if the operation fails.
504    ///
505    /// # Errors
506    ///
507    /// Returns an error if:
508    /// - `data_type` is an instrument class selector, which has no staged stream name.
509    /// - `data_type` is a family streams do not support.
510    /// - Feather file listing fails.
511    /// - Feather file reading fails.
512    /// - Writing to parquet fails.
513    ///
514    /// # Note
515    ///
516    /// This method converts directly between Arrow IPC stream batches and Parquet batches without
517    /// materializing Nautilus data objects. An instance with no staged files for the family
518    /// converts nothing and returns success. It requires:
519    /// - Listing feather files in the specified subdirectory
520    /// - Reading feather files (Arrow IPC stream reading)
521    /// - Applying table-only stream conversion transforms
522    /// - Writing Arrow batches to the catalog
523    ///
524    /// # Examples
525    ///
526    /// ```rust,no_run
527    /// use nautilus_model::data::NautilusDataType;
528    /// use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
529    ///
530    /// let mut catalog = ParquetDataCatalog::new(
531    ///     std::path::Path::new("/tmp/nautilus_data"),
532    ///     None,
533    ///     None,
534    ///     None,
535    ///     None,
536    /// );
537    ///
538    /// // Convert backtest stream data to parquet
539    /// catalog.convert_stream_to_data(
540    ///     "instance-123",
541    ///     &NautilusDataType::QuoteTick.into(),
542    ///     Some("backtest"),
543    ///     None,
544    ///     false,
545    /// )?;
546    /// # Ok::<(), anyhow::Error>(())
547    /// ```
548    pub fn convert_stream_to_data(
549        &mut self,
550        instance_id: &str,
551        data_type: &CatalogDataType,
552        subdirectory: Option<&str>,
553        identifiers: Option<&[String]>,
554        use_ts_event_for_ts_init: bool,
555    ) -> anyhow::Result<()> {
556        let subdirectory = subdirectory.unwrap_or("backtest");
557
558        // Streams stage instruments under the single aggregate name,
559        // with the class carried per batch, so a class selector names no staged directory.
560        let stream_data_name: Cow<'static, str> = match data_type {
561            CatalogDataType::Data(data_type) => parquet_data_path_prefix(data_type),
562            CatalogDataType::Record(record_type) => record_path_prefix(record_type),
563            CatalogDataType::Instrument(class) => {
564                anyhow::bail!(
565                    "Stream conversion stages instruments under the aggregate family, not {class}; \
566                     pass the Instrument data type"
567                );
568            }
569        };
570
571        if !Self::is_supported_stream_data_type(&stream_data_name) {
572            anyhow::bail!("Stream conversion does not support {data_type}");
573        }
574
575        // List all feather files for this data class
576        let feather_files =
577            self.list_feather_files(subdirectory, instance_id, &stream_data_name, identifiers)?;
578
579        if feather_files.is_empty() {
580            return Ok(());
581        }
582
583        // Process each feather file independently so that each file's identifier
584        // (instrument_id or bar_type from schema metadata) is preserved when writing
585        // to parquet. Conversion then groups each file's restored batches by full schema
586        // before writing one catalog file per group.
587        for file_path in feather_files {
588            let batches = self.read_feather_file(&file_path)?;
589            self.convert_feather_batches_to_parquet(
590                subdirectory,
591                instance_id,
592                &stream_data_name,
593                &file_path,
594                &batches,
595                use_ts_event_for_ts_init,
596                None,
597            )?;
598        }
599
600        Ok(())
601    }
602
603    #[expect(
604        clippy::too_many_arguments,
605        reason = "the arguments describe one Feather source and its catalog destination"
606    )]
607    fn convert_feather_batches_to_parquet(
608        &self,
609        subdirectory: &str,
610        instance_id: &str,
611        catalog_data_name: &str,
612        feather_path: &str,
613        batches: &[RecordBatch],
614        use_ts_event_for_ts_init: bool,
615        replay_identity: Option<&str>,
616    ) -> anyhow::Result<()> {
617        let mut groups: IndexMap<Arc<Schema>, Vec<RecordBatch>> = IndexMap::new();
618
619        for batch in batches {
620            for restored in restore_staged_record_batches(batch.clone())? {
621                groups.entry(restored.schema()).or_default().push(restored);
622            }
623        }
624
625        for (index, group) in groups.into_values().enumerate() {
626            let Some(batch) =
627                Self::apply_stream_conversion_transforms(&group, use_ts_event_for_ts_init)
628                    .map_err(|e| {
629                        anyhow::anyhow!(
630                            "Failed to apply stream conversion transforms for {feather_path}: {e}"
631                        )
632                    })?
633            else {
634                continue;
635            };
636
637            let (start_ts, end_ts) = Self::ts_init_range(&batch).map_err(|e| {
638                anyhow::anyhow!("Failed to determine ts_init range for {feather_path}: {e}")
639            })?;
640
641            let identifier = Self::identifier_from_batch_or_path(
642                &batch,
643                feather_path,
644                subdirectory,
645                instance_id,
646            );
647
648            let instrument_prefix = if catalog_data_name == "instruments" {
649                let class = batch
650                    .schema()
651                    .metadata()
652                    .get("class")
653                    .cloned()
654                    .ok_or_else(|| anyhow::anyhow!("Staged instrument has no class metadata"))?;
655                Some(crate::catalog::types::instrument_path_prefix(
656                    &class.parse()?,
657                ))
658            } else {
659                None
660            };
661
662            let catalog_data_name = instrument_prefix.unwrap_or(catalog_data_name);
663
664            let directory = if let Some(type_name) = catalog_data_name.strip_prefix("custom/") {
665                self.make_path_custom_data(type_name, identifier.as_deref())?
666            } else {
667                self.make_path(catalog_data_name, identifier.as_deref())?
668            };
669
670            let batch = Self::with_catalog_identifier_metadata(
671                batch,
672                catalog_data_name,
673                identifier.as_deref(),
674            )?;
675            let batches = vec![batch];
676            let group_identity = format!("{}/{index}", replay_identity.unwrap_or(feather_path));
677            self.write_parquet_file_checked(
678                &directory,
679                UnixNanos::from(start_ts),
680                UnixNanos::from(end_ts),
681                &batches,
682                false,
683                "File",
684                None,
685                Some(&group_identity),
686            )?;
687        }
688
689        Ok(())
690    }
691
692    fn with_catalog_identifier_metadata(
693        batch: RecordBatch,
694        catalog_data_name: &str,
695        identifier: Option<&str>,
696    ) -> anyhow::Result<RecordBatch> {
697        let Some(identifier) = identifier else {
698            return Ok(batch);
699        };
700        let metadata_key = if catalog_data_name == "bars" {
701            "bar_type"
702        } else {
703            "instrument_id"
704        };
705
706        if batch.schema().metadata().contains_key(metadata_key) {
707            return Ok(batch);
708        }
709
710        let mut metadata = batch.schema().metadata().clone();
711        metadata.insert(metadata_key.to_string(), identifier.to_string());
712        let schema = Arc::new(Schema::new_with_metadata(
713            batch.schema().fields().clone(),
714            metadata,
715        ));
716        Ok(RecordBatch::try_new(schema, batch.columns().to_vec())?)
717    }
718
719    fn apply_stream_conversion_transforms(
720        batches: &[RecordBatch],
721        use_ts_event_for_ts_init: bool,
722    ) -> anyhow::Result<Option<RecordBatch>> {
723        coalesce_stream_conversion_batches(
724            batches,
725            StreamConversionOptions {
726                use_ts_event_for_ts_init,
727                convert_bar_type_to_external: true,
728            },
729        )
730    }
731
732    fn ts_init_range(batch: &RecordBatch) -> anyhow::Result<(u64, u64)> {
733        let ts_init = Self::ts_init_array(batch)?;
734        if ts_init.is_empty() {
735            anyhow::bail!("Cannot convert empty stream batch to parquet");
736        }
737
738        if (0..ts_init.len()).any(|row| ts_init.is_null(row)) {
739            anyhow::bail!("ts_init column contains null values");
740        }
741
742        let start = ts_init
743            .value(0)
744            .ok_or_else(|| anyhow::anyhow!("ts_init value cannot be negative"))?;
745        let end = ts_init
746            .value(ts_init.len() - 1)
747            .ok_or_else(|| anyhow::anyhow!("ts_init value cannot be negative"))?;
748        Ok((start, end))
749    }
750
751    fn ts_init_array(batch: &RecordBatch) -> anyhow::Result<U64ColumnRef<'_>> {
752        let ts_init_idx = batch
753            .schema()
754            .index_of("ts_init")
755            .map_err(|_| anyhow::anyhow!("ts_init column not found"))?;
756        U64ColumnRef::try_from_array(batch.column(ts_init_idx).as_ref())
757            .ok_or_else(|| anyhow::anyhow!("ts_init column has an unsupported type"))
758    }
759
760    fn identifier_from_batch_or_path(
761        batch: &RecordBatch,
762        feather_path: &str,
763        subdirectory: &str,
764        instance_id: &str,
765    ) -> Option<String> {
766        let metadata = batch.schema().metadata().clone();
767        if let Some(bar_type) = metadata.get("bar_type") {
768            return Some(bar_type.clone());
769        }
770
771        if let Some(instrument_id) = metadata.get("instrument_id") {
772            return Some(instrument_id.clone());
773        }
774
775        if let Ok(identifiers) = identifiers_from_record_batches(std::slice::from_ref(batch))
776            && identifiers.len() == 1
777        {
778            return identifiers.into_iter().next();
779        }
780
781        identifier_from_session_feather_path(feather_path, subdirectory, instance_id)
782    }
783
784    fn canonical_stream_data_name(data_name: &str) -> &str {
785        match data_name {
786            "quote_tick" => "quotes",
787            "trade_tick" => "trades",
788            "bar" => "bars",
789            "mark_price_update" => "mark_prices",
790            "index_price_update" => "index_prices",
791            "funding_rate_update" => "funding_rates",
792            "instrument_close" => "instrument_closes",
793            "order_book_delta" => "order_book_deltas",
794            other => other,
795        }
796    }
797
798    fn is_supported_stream_data_type(data_name: &str) -> bool {
799        data_name.starts_with("custom/")
800            || matches!(
801                data_name,
802                "instruments"
803                    | "quotes"
804                    | "trades"
805                    | "order_book_deltas"
806                    | "order_book_depths"
807                    | "bars"
808                    | "index_prices"
809                    | "mark_prices"
810                    | "option_greeks"
811                    | "instrument_status"
812                    | "instrument_closes"
813                    | "funding_rates"
814                    | "account_state"
815                    | "order_initialized"
816                    | "order_denied"
817                    | "order_emulated"
818                    | "order_submitted"
819                    | "order_accepted"
820                    | "order_rejected"
821                    | "order_pending_cancel"
822                    | "order_canceled"
823                    | "order_cancel_rejected"
824                    | "order_expired"
825                    | "order_triggered"
826                    | "order_pending_update"
827                    | "order_released"
828                    | "order_modify_rejected"
829                    | "order_updated"
830                    | "order_filled"
831                    | "order_fill_voided"
832                    | "position_opened"
833                    | "position_changed"
834                    | "position_closed"
835                    | "position_adjusted"
836                    | "order_snapshot"
837                    | "position_snapshot"
838                    | "order_status_report"
839                    | "fill_report"
840                    | "position_status_report"
841                    | "execution_mass_status"
842            )
843    }
844}
845
846#[cfg(test)]
847mod canonical_name_tests {
848    use rstest::rstest;
849
850    use super::ParquetDataCatalog;
851
852    #[rstest]
853    #[case("quotes", "quotes")]
854    #[case("trades", "trades")]
855    #[case("bars", "bars")]
856    #[case("order_book_delta", "order_book_deltas")]
857    #[case("mark_prices", "mark_prices")]
858    #[case("index_prices", "index_prices")]
859    #[case("funding_rates", "funding_rates")]
860    #[case("instrument_closes", "instrument_closes")]
861    #[case("quotes", "quotes")]
862    fn canonical_stream_data_aliases(#[case] input: &str, #[case] expected: &str) {
863        assert_eq!(
864            ParquetDataCatalog::canonical_stream_data_name(input),
865            expected,
866        );
867    }
868}
nautilus_persistence/backend/parquet/feather_session.rs

nautilus_persistence/backend/parquet/
feather_session.rs
