nautilus_persistence/python/
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
18use std::{
19    cell::RefCell,
20    collections::{HashMap, HashSet},
21    rc::Rc,
22};
23
24use nautilus_common::{
25    live::get_runtime,
26    python::{cache::PyCache, clock::PyClock},
27};
28use nautilus_core::{UnixNanos, datetime::get_timezone};
29use nautilus_model::{
30    data::{
31        Bar, Data, FundingRateUpdate, IndexPriceUpdate, InstrumentStatus, MarkPriceUpdate,
32        OptionGreeks, OrderBookDelta, OrderBookDepth10, QuoteTick, TradeTick,
33        close::InstrumentClose,
34    },
35    events::{
36        AccountState, OrderAccepted, OrderCancelRejected, OrderCanceled, OrderDenied,
37        OrderEmulated, OrderExpired, OrderFillVoided, OrderFilled, OrderInitialized,
38        OrderModifyRejected, OrderPendingCancel, OrderPendingUpdate, OrderRejected, OrderReleased,
39        OrderSnapshot, OrderSubmitted, OrderTriggered, OrderUpdated, PositionAdjusted,
40        PositionChanged, PositionClosed, PositionOpened, PositionSnapshot,
41    },
42    python::instruments::pyobject_to_instrument_any,
43    reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
44};
45use object_store::ObjectStoreExt;
46use pyo3::{exceptions::PyIOError, prelude::*};
47
48use crate::{
49    backend::feather::{
50        FeatherWriter, FeatherWriterSubscriptions, RotationConfig, default_per_instrument_types,
51    },
52    parquet::{ObjectStoreLocationKind, create_object_store_location_from_path},
53};
54
55/// Python binding for the Rust `FeatherWriter`.
56///
57/// This provides a streaming writer of Nautilus objects into feather files with rotation
58/// capabilities, matching the interface of Python's `StreamingFeatherWriter`.
59#[pyclass(
60    name = "StreamingFeatherWriter",
61    module = "nautilus_trader.persistence",
62    unsendable
63)]
64#[pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.persistence")]
65pub struct PyStreamingFeatherWriter {
66    writer: Rc<RefCell<FeatherWriter>>,
67    subscriptions: Option<FeatherWriterSubscriptions>,
68}
69
70#[pymethods]
71#[pyo3_stub_gen::derive::gen_stub_pymethods]
72impl PyStreamingFeatherWriter {
73    /// Creates a new `StreamingFeatherWriter` instance.
74    ///
75    /// # Parameters
76    ///
77    /// - `path`: The path to persist the stream to. Must be a directory.
78    /// - `cache`: The cache for query info (`PyCache`).
79    /// - `clock`: The clock to use for time-related operations (`PyClock`).
80    /// - `fs_protocol`: Optional filesystem protocol (default: "file").
81    /// - `fs_storage_options`: Optional storage options for cloud backends.
82    /// - `include_types`: Optional list of type names to include (e.g., `["quotes", "trades"]`).
83    /// - `rotation_mode`: Rotation mode (0=SIZE, 1=INTERVAL, `2=SCHEDULED_DATES`, `3=NO_ROTATION`).
84    /// - `max_file_size`: Maximum file size in bytes before rotation (for SIZE mode).
85    /// - `rotation_interval_ns`: Rotation interval in nanoseconds (for `INTERVAL/SCHEDULED_DATES` modes).
86    /// - `rotation_time_ns`: Scheduled rotation time in nanoseconds (for `SCHEDULED_DATES` mode).
87    /// - `flush_interval_ms`: Flush interval in milliseconds (default: 1000). Set to 0 to disable auto-flush.
88    /// - `replace`: If existing files at the given path should be replaced (default: False).
89    #[new]
90    #[pyo3(signature = (
91        path,
92        cache,
93        clock,
94        fs_protocol=None,
95        fs_storage_options=None,
96        include_types=None,
97        rotation_mode=3,
98        max_file_size=1_073_741_824,
99        rotation_interval_ns=None,
100        rotation_time_ns=None,
101        rotation_timezone="UTC",
102        flush_interval_ms=None,
103        replace=false
104    ))]
105    #[expect(clippy::too_many_arguments, clippy::needless_pass_by_value)]
106    pub fn py_new(
107        path: String,
108        cache: PyCache,
109        clock: PyClock,
110        fs_protocol: Option<&str>,
111        fs_storage_options: Option<HashMap<String, String>>,
112        include_types: Option<Vec<String>>,
113        rotation_mode: u8,
114        max_file_size: u64,
115        rotation_interval_ns: Option<u64>,
116        rotation_time_ns: Option<u64>,
117        rotation_timezone: &str,
118        flush_interval_ms: Option<u64>,
119        replace: bool,
120    ) -> PyResult<Self> {
121        // Create object store from path
122        // Use fs_protocol to construct the full path if it's a cloud protocol
123        let full_path = match fs_protocol {
124            Some(protocol) if protocol != "file" && !path.contains("://") => {
125                format!("{protocol}://{path}")
126            }
127            _ => path,
128        };
129
130        let storage_options = fs_storage_options
131            .map(|map| map.into_iter().collect::<ahash::AHashMap<String, String>>());
132
133        let location = create_object_store_location_from_path(&full_path, storage_options)
134            .map_err(|e| PyIOError::new_err(format!("Failed to create object store: {e}")))?;
135        let is_local_store = matches!(&location.kind, ObjectStoreLocationKind::Local);
136        let object_store = location.object_store;
137        let base_path = location.base_path;
138
139        // Handle replace parameter - delete existing files if requested
140        if replace {
141            let runtime = get_runtime();
142            let store_ref = object_store.clone();
143            let prefix = if base_path.is_empty() {
144                if !is_local_store {
145                    return Err(PyIOError::new_err(
146                        "replace=True for remote streaming paths requires a non-empty prefix",
147                    ));
148                }
149                None
150            } else {
151                Some(object_store::path::Path::from(base_path.clone()))
152            };
153            runtime
154                .block_on(async {
155                    let mut stream = store_ref.list(prefix.as_ref());
156                    let mut to_delete = Vec::new();
157
158                    while let Some(result) = futures::StreamExt::next(&mut stream).await {
159                        if let Ok(meta) = result {
160                            to_delete.push(meta.location);
161                        }
162                    }
163
164                    for path in to_delete {
165                        let _ = store_ref.delete(&path).await;
166                    }
167                    Ok::<(), anyhow::Error>(())
168                })
169                .map_err(|e| {
170                    PyIOError::new_err(format!("Failed to replace existing files: {e}"))
171                })?;
172        }
173
174        // Convert rotation mode to RotationConfig
175        // Python RotationMode: 0=SIZE, 1=INTERVAL, 2=SCHEDULED_DATES, 3=NO_ROTATION
176        let rotation_config = match rotation_mode {
177            0 => RotationConfig::Size {
178                max_size: max_file_size,
179            },
180            1 => {
181                let interval = rotation_interval_ns.unwrap_or(86_400_000_000_000); // Default 1 day
182                RotationConfig::Interval {
183                    interval_ns: interval,
184                }
185            }
186            2 => {
187                let interval = rotation_interval_ns.unwrap_or(86_400_000_000_000); // Default 1 day
188                let tz = get_timezone(rotation_timezone).map_err(|e| {
189                    PyIOError::new_err(format!("Failed to parse rotation_timezone: {e}"))
190                })?;
191                let time_ns = rotation_time_ns.unwrap_or(0);
192                RotationConfig::ScheduledDates {
193                    interval_ns: interval,
194                    rotation_time: UnixNanos::from(time_ns),
195                    rotation_timezone: tz,
196                }
197            }
198            _ => RotationConfig::NoRotation, // Default to no rotation for invalid values
199        };
200
201        // Convert include_types to HashSet
202        let type_filter = include_types.map(|types| types.into_iter().collect::<HashSet<String>>());
203
204        // Extract Clock from Python wrapper
205        // PyClock wraps Rc<RefCell<dyn Clock>>, we get the inner Rc
206        let clock_rc = clock.clock_rc();
207        // Note: Cache parameter is kept for API compatibility with Python StreamingFeatherWriter
208        // but is not directly used by FeatherWriter
209        let _cache = cache;
210
211        // Create FeatherWriter
212        let writer = FeatherWriter::new(
213            base_path,
214            object_store,
215            clock_rc,
216            rotation_config,
217            type_filter,
218            Some(default_per_instrument_types()),
219            flush_interval_ms, // Auto-flush interval in milliseconds
220        );
221
222        Ok(Self {
223            writer: Rc::new(RefCell::new(writer)),
224            subscriptions: None,
225        })
226    }
227
228    /// Subscribes to all messages on the message bus (pattern "*").
229    ///
230    /// This matches the behavior of Python's `StreamingFeatherWriter` when subscribed
231    /// via `trader.subscribe("*", writer.write)`.
232    pub fn subscribe(&mut self) -> PyResult<()> {
233        if self.subscriptions.is_some() {
234            // Already subscribed
235            return Ok(());
236        }
237
238        let handler = FeatherWriter::subscribe_to_message_bus(self.writer.clone())
239            .map_err(|e| PyIOError::new_err(format!("Failed to subscribe to message bus: {e}")))?;
240
241        self.subscriptions = Some(handler);
242        Ok(())
243    }
244
245    /// Unsubscribes from the message bus.
246    pub fn unsubscribe(&mut self) -> PyResult<()> {
247        if let Some(handler) = self.subscriptions.take() {
248            FeatherWriter::unsubscribe_from_message_bus(&handler);
249        }
250        Ok(())
251    }
252
253    /// Writes a data object to the stream.
254    ///
255    /// # Parameters
256    ///
257    /// - `data`: The data object to write (must be a Nautilus data type from pyo3).
258    #[expect(
259        clippy::needless_pass_by_value,
260        clippy::too_many_lines,
261        reason = "PyO3 writer binding must downcast supported data variants inline"
262    )]
263    pub fn write(&self, py: Python, data: Py<PyAny>) -> PyResult<()> {
264        macro_rules! try_write {
265            ($type:ty, $name:literal) => {
266                if let Ok(value) = data.extract::<$type>(py) {
267                    let mut writer = self.writer.borrow_mut();
268                    let runtime = get_runtime();
269                    return runtime
270                        .block_on(async { writer.write(value).await })
271                        .map_err(|e| {
272                            PyIOError::new_err(format!("Failed to write {}: {e}", $name))
273                        });
274                }
275            };
276        }
277
278        // Try to convert from common pyo3 data types
279        if let Ok(quote) = data.extract::<QuoteTick>(py) {
280            let mut writer = self.writer.borrow_mut();
281            let runtime = get_runtime();
282            return runtime
283                .block_on(async { writer.write_data(Data::Quote(quote)).await })
284                .map_err(|e| PyIOError::new_err(format!("Failed to write QuoteTick: {e}")));
285        }
286
287        if let Ok(trade) = data.extract::<TradeTick>(py) {
288            let mut writer = self.writer.borrow_mut();
289            let runtime = get_runtime();
290            return runtime
291                .block_on(async { writer.write_data(Data::Trade(trade)).await })
292                .map_err(|e| PyIOError::new_err(format!("Failed to write TradeTick: {e}")));
293        }
294
295        if let Ok(bar) = data.extract::<Bar>(py) {
296            let mut writer = self.writer.borrow_mut();
297            let runtime = get_runtime();
298            return runtime
299                .block_on(async { writer.write_data(Data::Bar(bar)).await })
300                .map_err(|e| PyIOError::new_err(format!("Failed to write Bar: {e}")));
301        }
302
303        if let Ok(delta) = data.extract::<OrderBookDelta>(py) {
304            let mut writer = self.writer.borrow_mut();
305            let runtime = get_runtime();
306            return runtime
307                .block_on(async { writer.write_data(Data::Delta(delta)).await })
308                .map_err(|e| PyIOError::new_err(format!("Failed to write OrderBookDelta: {e}")));
309        }
310
311        if let Ok(depth) = data.extract::<OrderBookDepth10>(py) {
312            let mut writer = self.writer.borrow_mut();
313            let runtime = get_runtime();
314            return runtime
315                .block_on(async { writer.write_data(Data::Depth10(Box::new(depth))).await })
316                .map_err(|e| PyIOError::new_err(format!("Failed to write OrderBookDepth10: {e}")));
317        }
318
319        if let Ok(price) = data.extract::<IndexPriceUpdate>(py) {
320            let mut writer = self.writer.borrow_mut();
321            let runtime = get_runtime();
322            return runtime
323                .block_on(async { writer.write_data(Data::IndexPrice(price)).await })
324                .map_err(|e| PyIOError::new_err(format!("Failed to write IndexPriceUpdate: {e}")));
325        }
326
327        if let Ok(price) = data.extract::<MarkPriceUpdate>(py) {
328            let mut writer = self.writer.borrow_mut();
329            let runtime = get_runtime();
330            return runtime
331                .block_on(async { writer.write_data(Data::MarkPrice(price)).await })
332                .map_err(|e| PyIOError::new_err(format!("Failed to write MarkPriceUpdate: {e}")));
333        }
334
335        if let Ok(greeks) = data.extract::<OptionGreeks>(py) {
336            let mut writer = self.writer.borrow_mut();
337            let runtime = get_runtime();
338            return runtime
339                .block_on(async { writer.write_data(Data::OptionGreeks(greeks)).await })
340                .map_err(|e| PyIOError::new_err(format!("Failed to write OptionGreeks: {e}")));
341        }
342
343        if let Ok(close) = data.extract::<InstrumentClose>(py) {
344            let mut writer = self.writer.borrow_mut();
345            let runtime = get_runtime();
346            return runtime
347                .block_on(async { writer.write_data(Data::InstrumentClose(close)).await })
348                .map_err(|e| PyIOError::new_err(format!("Failed to write InstrumentClose: {e}")));
349        }
350
351        try_write!(FundingRateUpdate, "FundingRateUpdate");
352        try_write!(InstrumentStatus, "InstrumentStatus");
353        try_write!(AccountState, "AccountState");
354        try_write!(OrderInitialized, "OrderInitialized");
355        try_write!(OrderDenied, "OrderDenied");
356        try_write!(OrderEmulated, "OrderEmulated");
357        try_write!(OrderSubmitted, "OrderSubmitted");
358        try_write!(OrderAccepted, "OrderAccepted");
359        try_write!(OrderRejected, "OrderRejected");
360        try_write!(OrderPendingCancel, "OrderPendingCancel");
361        try_write!(OrderCanceled, "OrderCanceled");
362        try_write!(OrderCancelRejected, "OrderCancelRejected");
363        try_write!(OrderExpired, "OrderExpired");
364        try_write!(OrderTriggered, "OrderTriggered");
365        try_write!(OrderPendingUpdate, "OrderPendingUpdate");
366        try_write!(OrderReleased, "OrderReleased");
367        try_write!(OrderModifyRejected, "OrderModifyRejected");
368        try_write!(OrderUpdated, "OrderUpdated");
369        try_write!(OrderFilled, "OrderFilled");
370        try_write!(OrderFillVoided, "OrderFillVoided");
371        try_write!(PositionOpened, "PositionOpened");
372        try_write!(PositionChanged, "PositionChanged");
373        try_write!(PositionClosed, "PositionClosed");
374        try_write!(PositionAdjusted, "PositionAdjusted");
375        try_write!(OrderSnapshot, "OrderSnapshot");
376        try_write!(PositionSnapshot, "PositionSnapshot");
377        try_write!(OrderStatusReport, "OrderStatusReport");
378        try_write!(FillReport, "FillReport");
379        try_write!(PositionStatusReport, "PositionStatusReport");
380        try_write!(ExecutionMassStatus, "ExecutionMassStatus");
381
382        // Try instrument types (uses type_str attribute for dispatch)
383        if let Ok(instrument) = pyobject_to_instrument_any(py, data.clone_ref(py)) {
384            let mut writer = self.writer.borrow_mut();
385            let runtime = get_runtime();
386            return runtime
387                .block_on(async { writer.write_instrument(instrument).await })
388                .map_err(|e| PyIOError::new_err(format!("Failed to write instrument: {e}")));
389        }
390
391        Err(PyIOError::new_err(
392            "Unsupported data type for feather writer",
393        ))
394    }
395
396    /// Flushes all active buffers by writing any remaining buffered bytes to the object store.
397    ///
398    /// This is called automatically based on `flush_interval_ms` if configured, but can also
399    /// be called manually by the client.
400    pub fn flush(&self) -> PyResult<()> {
401        let mut writer = self.writer.borrow_mut();
402        let runtime = get_runtime();
403
404        runtime
405            .block_on(async { writer.flush().await })
406            .map_err(|e| PyIOError::new_err(format!("Failed to flush: {e}")))
407    }
408
409    /// Closes all writers by flushing and removing them.
410    ///
411    /// After calling this, no further writes should be performed.
412    pub fn close(&self) -> PyResult<()> {
413        let mut writer = self.writer.borrow_mut();
414        let runtime = get_runtime();
415
416        runtime
417            .block_on(async { writer.close().await })
418            .map_err(|e| PyIOError::new_err(format!("Failed to close: {e}")))
419    }
420
421    /// Returns whether the writer has been closed (no active writers).
422    #[getter]
423    #[must_use]
424    pub fn is_closed(&self) -> bool {
425        self.writer.borrow().is_closed()
426    }
427
428    /// Returns information about the current files being written.
429    ///
430    /// Returns a dictionary mapping writer keys to (size, path) tuples.
431    #[must_use]
432    pub fn get_current_file_info(&self) -> HashMap<String, (u64, String)> {
433        self.writer.borrow().get_current_file_info()
434    }
435
436    /// Returns the next rotation time for a writer, or None if not set.
437    #[pyo3(signature = (type_str, instrument_id=None))]
438    #[must_use]
439    pub fn get_next_rotation_time(
440        &self,
441        type_str: &str,
442        instrument_id: Option<&str>,
443    ) -> Option<u64> {
444        self.writer
445            .borrow()
446            .get_next_rotation_time(type_str, instrument_id)
447            .map(|ns| ns.as_u64())
448    }
449}
nautilus_persistence/python/feather.rs

nautilus_persistence/python/
feather.rs
