nautilus_persistence/python/backend/
feather.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//! Python bindings for the Rust `FeatherWriter` as `StreamingFeatherWriter`.
17
18#![expect(
19    clippy::too_many_lines,
20    reason = "PyO3 writer constructor mirrors Python keyword surface"
21)]
22
23use std::{
24    cell::RefCell,
25    collections::{HashMap, HashSet},
26    rc::Rc,
27    sync::{
28        Arc,
29        atomic::{AtomicU64, Ordering},
30    },
31};
32
33use nautilus_common::{
34    clock::Clock,
35    live::{block_on_nautilus_with, get_runtime},
36    python::{cache::PyCache, clock::PyClock},
37};
38use nautilus_core::{UnixNanos, datetime::get_timezone, python::to_pyruntime_err};
39use nautilus_model::{
40    data::{
41        Bar, CustomData, Data, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus,
42        MarkPriceUpdate, OptionGreeks, OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick,
43        close::InstrumentClose,
44    },
45    events::{
46        AccountState, OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied,
47        OrderEmulated, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
48        OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
49        OrderSnapshot, OrderSubmitted, OrderTriggered, OrderUpdated, PositionAdjusted,
50        PositionChanged, PositionClosed, PositionOpened, PositionSnapshot,
51    },
52    python::instruments::pyobject_to_instrument_any,
53    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
54};
55use object_store::ObjectStoreExt;
56use pyo3::{exceptions::PyIOError, prelude::*};
57
58use crate::{
59    common::{
60        paths::normalize_path_separators,
61        storage::{StorageBackend, create_storage_backend_from_path},
62    },
63    python::backend::writer_record_filter_from_py,
64    writer::{
65        feather::{FeatherWriter, RotationConfig, WriterClock},
66        subscription::StreamingSinkSubscription,
67    },
68};
69
70/// Source clock plus the shared atomic the writer reads time from.
71type ClockBridge = (Rc<RefCell<dyn Clock>>, Arc<AtomicU64>);
72
73/// Python binding for the Rust `FeatherWriter`.
74///
75/// This provides a streaming writer of Nautilus objects into feather files with rotation
76/// capabilities, matching the interface of Python's `StreamingFeatherWriter`.
77#[pyclass(
78    name = "StreamingFeatherWriter",
79    module = "nautilus_trader.persistence",
80    unsendable
81)]
82#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
83pub struct PyStreamingFeatherWriter {
84    writer: Rc<RefCell<FeatherWriter>>,
85    handler: Option<StreamingSinkSubscription>,
86    run_manifest: Option<(StorageBackend, String, String)>,
87    run_manifest_has_data: RefCell<bool>,
88    /// Present when constructed with a non-live clock: the source clock plus the
89    /// shared atomic the core writer reads, refreshed before each forwarded call.
90    /// Note: writes arriving via the message bus subscription do not refresh the
91    /// bridge; live usage pairs the subscription with a `LiveClock`.
92    clock_bridge: Option<ClockBridge>,
93}
94
95#[pymethods]
96#[pyo3_stub_gen::derive::gen_stub_pymethods]
97impl PyStreamingFeatherWriter {
98    /// Creates a new `StreamingFeatherWriter` instance.
99    ///
100    /// # Parameters
101    ///
102    /// - `path`: The path to persist the stream to. Must be a directory.
103    /// - `cache`: The cache for query info (`PyCache`).
104    /// - `clock`: The clock to use for time-related operations (`PyClock`).
105    /// - `fs_protocol`: Optional filesystem protocol (default: "file").
106    /// - `fs_storage_options`: Optional storage options for cloud backends.
107    /// - `include_types`: Optional list of type names to include (e.g., `["quotes", "trades"]`).
108    /// - `rotation_mode`: Rotation mode (0=SIZE, 1=INTERVAL, `2=SCHEDULED_DATES`, `3=NO_ROTATION`).
109    /// - `max_file_size`: Maximum file size in bytes before rotation (for SIZE mode).
110    /// - `rotation_interval_ns`: Rotation interval in nanoseconds (for `INTERVAL/SCHEDULED_DATES` modes).
111    /// - `rotation_time_ns`: Scheduled rotation time in nanoseconds (for `SCHEDULED_DATES` mode).
112    /// - `flush_interval_ms`: Flush interval in milliseconds (default: 1000). Set to 0 to disable auto-flush.
113    /// - `replace`: If existing files at the given path should be replaced (default: False).
114    #[new]
115    #[pyo3(signature = (
116        path,
117        cache,
118        clock,
119        fs_protocol=None,
120        fs_storage_options=None,
121        include_types=None,
122        record_types=None,
123        record_filters=None,
124        rotation_mode=3,
125        max_file_size=1_073_741_824,
126        rotation_interval_ns=None,
127        rotation_time_ns=None,
128        rotation_timezone="UTC",
129        flush_interval_ms=None,
130        replace=false
131    ))]
132    #[expect(
133        clippy::too_many_arguments,
134        clippy::needless_pass_by_value,
135        reason = "PyO3 constructor mirrors the writer configuration fields"
136    )]
137    pub fn py_new(
138        path: String,
139        cache: PyCache,
140        clock: PyClock,
141        fs_protocol: Option<&str>,
142        fs_storage_options: Option<HashMap<String, String>>,
143        include_types: Option<Vec<String>>,
144        record_types: Option<&Bound<'_, PyAny>>,
145        record_filters: Option<&Bound<'_, PyAny>>,
146        rotation_mode: u8,
147        max_file_size: u64,
148        rotation_interval_ns: Option<u64>,
149        rotation_time_ns: Option<u64>,
150        rotation_timezone: &str,
151        flush_interval_ms: Option<u64>,
152        replace: bool,
153    ) -> PyResult<Self> {
154        // Create object store from path
155        // Use fs_protocol to construct the full path if it's a cloud protocol
156        let full_path = if let Some(protocol) = fs_protocol {
157            if protocol != "file" && !path.contains("://") {
158                format!("{protocol}://{path}")
159            } else {
160                path
161            }
162        } else {
163            path
164        };
165
166        let storage_options = fs_storage_options
167            .map(|map| map.into_iter().collect::<ahash::AHashMap<String, String>>());
168
169        if replace
170            && url::Url::parse(&full_path).is_ok_and(|url| {
171                !matches!(url.scheme(), "file" | "memory")
172                    && url.path().trim_matches('/').is_empty()
173            })
174        {
175            return Err(PyIOError::new_err(
176                "replace=True for remote streaming paths requires a non-empty prefix",
177            ));
178        }
179
180        let storage = create_storage_backend_from_path(&full_path, storage_options)
181            .map_err(|e| PyIOError::new_err(format!("Failed to create storage backend: {e}")))?;
182        let object_store = storage.object_store.clone();
183
184        // Handle replace parameter - delete existing files if requested
185        if replace {
186            let store_ref = object_store.clone();
187            let base_path = storage.base_path.clone();
188            block_on_nautilus_with(move || async move {
189                let prefix = if base_path.is_empty() {
190                    None
191                } else {
192                    Some(object_store::path::Path::from(
193                        base_path.trim_start_matches('/'),
194                    ))
195                };
196                let mut stream = store_ref.list(prefix.as_ref());
197                let mut to_delete = Vec::new();
198
199                while let Some(result) = futures::StreamExt::next(&mut stream).await {
200                    if let Ok(meta) = result {
201                        to_delete.push(meta.location);
202                    }
203                }
204
205                for path in to_delete {
206                    let _ = store_ref.delete(&path).await;
207                }
208                Ok::<(), anyhow::Error>(())
209            })
210            .map_err(|e| PyIOError::new_err(format!("Failed to replace existing files: {e}")))?;
211        }
212
213        let run_manifest =
214            if let Some((kind, instance_id)) = run_kind_and_instance_id_from_path(&full_path) {
215                let manifest_storage = storage.clone();
216                let manifest_kind = kind.clone();
217                let manifest_instance_id = instance_id.clone();
218                block_on_nautilus_with(move || async move {
219                    manifest_storage
220                        .write_current_run_manifest(
221                            &manifest_kind,
222                            &manifest_instance_id,
223                            "in_progress",
224                            true,
225                        )
226                        .await
227                })
228                .map_err(|e| PyIOError::new_err(format!("Failed to write run manifest: {e}")))?;
229                Some((storage.clone(), kind, instance_id))
230            } else {
231                None
232            };
233
234        // Convert rotation mode to RotationConfig
235        // Python RotationMode: 0=SIZE, 1=INTERVAL, 2=SCHEDULED_DATES, 3=NO_ROTATION
236        let rotation_config = match rotation_mode {
237            0 => RotationConfig::Size {
238                max_size: max_file_size,
239            },
240            1 => {
241                let interval = rotation_interval_ns.unwrap_or(86_400_000_000_000); // Default 1 day
242                RotationConfig::Interval {
243                    interval_ns: interval,
244                }
245            }
246            2 => {
247                let interval = rotation_interval_ns.unwrap_or(86_400_000_000_000); // Default 1 day
248                let tz = get_timezone(rotation_timezone).map_err(|e| {
249                    PyIOError::new_err(format!("Failed to parse rotation_timezone: {e}"))
250                })?;
251                let time_ns = rotation_time_ns.unwrap_or(0);
252                RotationConfig::ScheduledDates {
253                    interval_ns: interval,
254                    rotation_time: UnixNanos::from(time_ns),
255                    rotation_timezone: tz,
256                }
257            }
258            _ => RotationConfig::NoRotation, // Default to no rotation for invalid values
259        };
260
261        // Convert include_types to HashSet
262        let type_filter = include_types.map(|types| types.into_iter().collect::<HashSet<String>>());
263        let record_filter = writer_record_filter_from_py(record_types, record_filters)?;
264
265        // Set up per-instrument types (matching Python's _per_instrument_writers)
266        let mut per_instrument_types = HashSet::new();
267        per_instrument_types.insert("bars".to_string());
268        per_instrument_types.insert("order_book_deltas".to_string());
269        per_instrument_types.insert("order_book_depths".to_string());
270        per_instrument_types.insert("option_greeks".to_string());
271        per_instrument_types.insert("quotes".to_string());
272        per_instrument_types.insert("trades".to_string());
273        per_instrument_types.insert("mark_prices".to_string());
274        per_instrument_types.insert("index_prices".to_string());
275        per_instrument_types.insert("funding_rates".to_string());
276
277        // Extract Clock from Python wrapper and translate it into the core
278        // writer's Send time source (live clocks read the wall clock directly;
279        // test clocks are bridged through a shared atomic)
280        let clock_rc = clock.clock_rc();
281        let (writer_clock, shared_time) = WriterClock::from_shared_clock(&clock_rc);
282        // Note: Cache parameter is kept for API compatibility with Python StreamingFeatherWriter
283        // but is not directly used by FeatherWriter
284        let _cache = cache;
285
286        // Create FeatherWriter
287        let writer = FeatherWriter::new(
288            storage.base_path,
289            object_store,
290            writer_clock,
291            rotation_config,
292            type_filter,
293            Some(per_instrument_types),
294            flush_interval_ms, // Auto-flush interval in milliseconds
295        )
296        .with_record_filter(record_filter);
297
298        Ok(Self {
299            writer: Rc::new(RefCell::new(writer)),
300            handler: None,
301            run_manifest,
302            run_manifest_has_data: RefCell::new(false),
303            clock_bridge: shared_time.map(|shared| (clock_rc, shared)),
304        })
305    }
306
307    /// Subscribes to all messages on the message bus (pattern "*").
308    ///
309    /// This matches the behavior of Python's `StreamingFeatherWriter` when subscribed
310    /// via `trader.subscribe("*", writer.write)`.
311    pub fn subscribe(&mut self) -> PyResult<()> {
312        if self.handler.is_some() {
313            // Already subscribed
314            return Ok(());
315        }
316
317        let handler = FeatherWriter::subscribe_to_message_bus(self.writer.clone())
318            .map_err(|e| PyIOError::new_err(format!("Failed to subscribe to message bus: {e}")))?;
319
320        self.handler = Some(handler);
321        Ok(())
322    }
323
324    /// Unsubscribes from the message bus.
325    pub fn unsubscribe(&mut self) -> PyResult<()> {
326        if let Some(handler) = self.handler.take() {
327            FeatherWriter::unsubscribe_from_message_bus(&handler);
328        }
329        Ok(())
330    }
331
332    /// Writes a data object to the stream.
333    ///
334    /// # Parameters
335    ///
336    /// - `data`: The data object to write (must be a Nautilus data type from pyo3).
337    #[expect(
338        clippy::needless_pass_by_value,
339        reason = "PyO3 writer binding must downcast supported data variants inline"
340    )]
341    pub fn write(&self, py: Python, data: Py<PyAny>) -> PyResult<()> {
342        self.refresh_writer_clock();
343
344        macro_rules! try_write {
345            ($type:ty, $name:literal) => {
346                if let Ok(value) = data.extract::<$type>(py) {
347                    let result =
348                        self.writer.borrow_mut().write(value).map_err(|e| {
349                            PyIOError::new_err(format!("Failed to write {}: {e}", $name))
350                        });
351                    return self.finish_write_result(result);
352                }
353            };
354        }
355
356        macro_rules! try_write_data {
357            ($data:expr, $name:literal) => {{
358                let result = self
359                    .writer
360                    .borrow_mut()
361                    .write_data($data)
362                    .map_err(|e| PyIOError::new_err(format!("Failed to write {}: {e}", $name)));
363                return self.finish_write_result(result);
364            }};
365        }
366
367        // Try to convert from common pyo3 data types
368        if let Ok(quote) = data.extract::<QuoteTick>(py) {
369            try_write_data!(Data::Quote(quote), "QuoteTick");
370        }
371
372        if let Ok(trade) = data.extract::<TradeTick>(py) {
373            try_write_data!(Data::Trade(trade), "TradeTick");
374        }
375
376        if let Ok(bar) = data.extract::<Bar>(py) {
377            try_write_data!(Data::Bar(bar), "Bar");
378        }
379
380        if let Ok(delta) = data.extract::<OrderBookDelta>(py) {
381            try_write_data!(Data::BookDelta(delta), "OrderBookDelta");
382        }
383
384        if let Ok(depth) = data.extract::<OrderBookDepth>(py) {
385            try_write_data!(Data::BookDepth(Box::new(depth)), "OrderBookDepth");
386        }
387
388        if let Ok(price) = data.extract::<IndexPriceUpdate>(py) {
389            try_write_data!(Data::IndexPrice(price), "IndexPriceUpdate");
390        }
391
392        if let Ok(price) = data.extract::<MarkPriceUpdate>(py) {
393            try_write_data!(Data::MarkPrice(price), "MarkPriceUpdate");
394        }
395
396        if let Ok(greeks) = data.extract::<OptionGreeks>(py) {
397            try_write_data!(Data::OptionGreeks(greeks), "OptionGreeks");
398        }
399
400        if let Ok(close) = data.extract::<InstrumentClose>(py) {
401            try_write_data!(Data::InstrumentClose(close), "InstrumentClose");
402        }
403
404        if let Ok(custom) = data.extract::<CustomData>(py) {
405            try_write_data!(Data::Custom(custom), "CustomData");
406        }
407
408        try_write!(FundingRateUpdate, "FundingRateUpdate");
409        try_write!(InstrumentStatus, "InstrumentStatus");
410        try_write!(AccountState, "AccountState");
411        try_write!(OrderInitialized, "OrderInitialized");
412        try_write!(OrderDenied, "OrderDenied");
413        try_write!(OrderEmulated, "OrderEmulated");
414        try_write!(OrderSubmitted, "OrderSubmitted");
415        try_write!(OrderAccepted, "OrderAccepted");
416        try_write!(OrderRejected, "OrderRejected");
417        try_write!(OrderPendingCancel, "OrderPendingCancel");
418        try_write!(OrderCanceled, "OrderCanceled");
419        try_write!(OrderCancelRejected, "OrderCancelRejected");
420        try_write!(OrderExpired, "OrderExpired");
421        try_write!(OrderTriggered, "OrderTriggered");
422        try_write!(OrderPendingUpdate, "OrderPendingUpdate");
423        try_write!(OrderReleased, "OrderReleased");
424        try_write!(OrderModifyRejected, "OrderModifyRejected");
425        try_write!(OrderUpdated, "OrderUpdated");
426        try_write!(OrderFilled, "OrderFilled");
427        try_write!(OrderFillVoided, "OrderFillVoided");
428        try_write!(PositionOpened, "PositionOpened");
429        try_write!(PositionChanged, "PositionChanged");
430        try_write!(PositionClosed, "PositionClosed");
431        try_write!(PositionAdjusted, "PositionAdjusted");
432        try_write!(OrderSnapshot, "OrderSnapshot");
433        try_write!(PositionSnapshot, "PositionSnapshot");
434        try_write!(OrderStatusReport, "OrderStatusReport");
435        try_write!(FillReport, "FillReport");
436        try_write!(PositionStatusReport, "PositionStatusReport");
437        try_write!(ExecutionMassStatus, "ExecutionMassStatus");
438
439        // Try instrument types (uses type_str attribute for dispatch)
440        if let Ok(instrument) = pyobject_to_instrument_any(py, data.clone_ref(py)) {
441            let result = self
442                .writer
443                .borrow_mut()
444                .write_instrument(instrument)
445                .map_err(|e| PyIOError::new_err(format!("Failed to write instrument: {e}")));
446            return self.finish_write_result(result);
447        }
448
449        Err(PyIOError::new_err(
450            "Unsupported data type for feather writer",
451        ))
452    }
453
454    /// Flushes all active buffers by writing any remaining buffered bytes to the object store.
455    ///
456    /// This is called automatically based on `flush_interval_ms` if configured, but can also
457    /// be called manually by the client.
458    pub fn flush(&self) -> PyResult<()> {
459        self.refresh_writer_clock();
460        let mut writer = self.writer.borrow_mut();
461
462        block_on_local("flush StreamingFeatherWriter", || async {
463            writer.flush().await
464        })?
465        .map_err(|e| PyIOError::new_err(format!("Failed to flush: {e}")))
466    }
467
468    /// Closes all writers by flushing and removing them.
469    ///
470    /// After calling this, no further writes should be performed.
471    pub fn close(&self) -> PyResult<()> {
472        self.refresh_writer_clock();
473        let mut writer = self.writer.borrow_mut();
474
475        block_on_local("close StreamingFeatherWriter", || async {
476            writer.close().await
477        })?
478        .map_err(|e| PyIOError::new_err(format!("Failed to close: {e}")))?;
479        drop(writer);
480        self.write_run_manifest(
481            "completed",
482            !*self.run_manifest_has_data.borrow(),
483            "complete",
484        )
485    }
486
487    /// Returns whether the writer has been closed (no active writers).
488    #[getter]
489    #[must_use]
490    pub fn is_closed(&self) -> bool {
491        self.writer.borrow().is_closed()
492    }
493
494    /// Returns information about the current files being written.
495    ///
496    /// Returns a dictionary mapping writer keys to (size, path) tuples.
497    #[must_use]
498    pub fn get_current_file_info(&self) -> HashMap<String, (u64, String)> {
499        self.writer.borrow().get_current_file_info()
500    }
501
502    /// Returns the next rotation time for a writer, or None if not set.
503    #[pyo3(signature = (type_str, instrument_id=None))]
504    #[must_use]
505    pub fn get_next_rotation_time(
506        &self,
507        type_str: &str,
508        instrument_id: Option<&str>,
509    ) -> Option<u64> {
510        self.writer
511            .borrow()
512            .get_next_rotation_time(type_str, instrument_id)
513            .map(|ns| ns.as_u64())
514    }
515}
516
517impl PyStreamingFeatherWriter {
518    // Pushes the source clock's current time into the core writer's shared
519    // atomic so test clocks drive flush/rotation cadence correctly.
520    fn refresh_writer_clock(&self) {
521        if let Some((clock, shared)) = &self.clock_bridge {
522            shared.store(clock.borrow().timestamp_ns().as_u64(), Ordering::Relaxed);
523        }
524    }
525
526    fn finish_write_result(&self, result: PyResult<()>) -> PyResult<()> {
527        result?;
528        self.mark_run_non_empty()
529    }
530
531    fn mark_run_non_empty(&self) -> PyResult<()> {
532        if *self.run_manifest_has_data.borrow() {
533            return Ok(());
534        }
535        self.write_run_manifest("in_progress", false, "update")?;
536        *self.run_manifest_has_data.borrow_mut() = true;
537        Ok(())
538    }
539
540    fn write_run_manifest(&self, status: &str, empty: bool, operation: &str) -> PyResult<()> {
541        let Some((storage, kind, instance_id)) = &self.run_manifest else {
542            return Ok(());
543        };
544
545        let storage = storage.clone();
546        let kind = kind.clone();
547        let instance_id = instance_id.clone();
548        let status = status.to_string();
549        block_on_nautilus_with(move || async move {
550            storage
551                .write_current_run_manifest(&kind, &instance_id, &status, empty)
552                .await
553        })
554        .map_err(|e| PyIOError::new_err(format!("Failed to {operation} run manifest: {e}")))
555    }
556}
557
558fn block_on_local<C, F>(operation: &str, create_future: C) -> PyResult<F::Output>
559where
560    C: FnOnce() -> F,
561    F: std::future::Future,
562{
563    let run = move || get_runtime().block_on(async move { create_future().await });
564    if tokio::runtime::Handle::try_current().is_err() {
565        return Ok(run());
566    }
567
568    Err(to_pyruntime_err(format!(
569        "Cannot {operation} from an active Tokio runtime"
570    )))
571}
572
573fn run_kind_and_instance_id_from_path(path: &str) -> Option<(String, String)> {
574    let normalized = normalize_path_separators(path);
575    let parsed_url = url::Url::parse(&normalized).ok();
576
577    let path = parsed_url.as_ref().map_or(normalized.as_str(), |url| {
578        url.path().trim_start_matches('/')
579    });
580
581    let components: Vec<&str> = path
582        .trim_matches('/')
583        .split('/')
584        .filter(|component| !component.is_empty())
585        .collect();
586    let instance_id = components.last()?;
587    let kind = components.get(components.len().checked_sub(2)?)?;
588
589    match *kind {
590        "backtest" | "live" | "sandbox" => Some(((*kind).to_string(), (*instance_id).to_string())),
591        _ => None,
592    }
593}
594
595#[cfg(test)]
596#[expect(
597    clippy::disallowed_types,
598    reason = "tests exercise direct Tokio LocalSet interoperability"
599)]
600mod tests {
601    use rstest::rstest;
602
603    use super::{block_on_local, run_kind_and_instance_id_from_path};
604
605    #[rstest]
606    fn block_on_local_rejects_current_thread_runtime() {
607        pyo3::Python::initialize();
608        let runtime = tokio::runtime::Builder::new_current_thread()
609            .enable_all()
610            .build()
611            .unwrap();
612
613        let error = runtime
614            .block_on(async { block_on_local("run test operation", || async { 42 }).unwrap_err() });
615
616        assert_eq!(
617            error.to_string(),
618            "RuntimeError: Cannot run test operation from an active Tokio runtime"
619        );
620    }
621
622    #[rstest]
623    fn block_on_local_rejects_multi_thread_local_set() {
624        pyo3::Python::initialize();
625        let runtime = tokio::runtime::Builder::new_multi_thread()
626            .worker_threads(2)
627            .enable_all()
628            .build()
629            .unwrap();
630        let local_set = tokio::task::LocalSet::new();
631
632        let error = runtime.block_on(local_set.run_until(async {
633            block_on_local("run test operation", || async { 42 }).unwrap_err()
634        }));
635
636        assert_eq!(
637            error.to_string(),
638            "RuntimeError: Cannot run test operation from an active Tokio runtime"
639        );
640    }
641
642    #[rstest]
643    #[case(
644        r"C:\Users\Administrator\AppData\Local\Temp\pytest-0\backtest\run-greeks",
645        "backtest",
646        "run-greeks"
647    )]
648    #[case("C:/catalog/backtest/run-1", "backtest", "run-1")]
649    #[case(r"\\server\share\live\run-2", "live", "run-2")]
650    #[case("/tmp/catalog/sandbox/run-3", "sandbox", "run-3")]
651    #[case("file:///C:/catalog/backtest/run-1", "backtest", "run-1")]
652    fn run_kind_and_instance_id_handles_platform_paths(
653        #[case] path: &str,
654        #[case] kind: &str,
655        #[case] instance_id: &str,
656    ) {
657        assert_eq!(
658            run_kind_and_instance_id_from_path(path),
659            Some((kind.to_string(), instance_id.to_string())),
660        );
661    }
662
663    #[rstest]
664    fn run_kind_and_instance_id_rejects_non_run_paths() {
665        assert_eq!(
666            run_kind_and_instance_id_from_path(r"C:\catalog\data\quotes"),
667            None
668        );
669        assert_eq!(run_kind_and_instance_id_from_path("/tmp/catalog"), None);
670    }
671}
nautilus_persistence/python/backend/feather.rs

nautilus_persistence/python/backend/
feather.rs
