nautilus_tardis/csv/
mod.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
16pub mod load;
17pub mod stream;
18
19#[cfg(feature = "arrow")]
20pub mod convert;
21
22mod record;
23
24use std::{
25    ffi::OsStr,
26    fs::File,
27    io::{BufReader, Read, Seek, SeekFrom},
28    path::Path,
29    time::Duration,
30};
31
32use csv::{Reader, ReaderBuilder};
33use flate2::read::GzDecoder;
34pub use load::{
35    load_deltas, load_depth_from_snapshot5, load_depth_from_snapshot25, load_funding_rates,
36    load_options_chain, load_quotes, load_trades,
37};
38use nautilus_model::{
39    data::{
40        BookOrder, FundingRateUpdate, NULL_ORDER, OptionGreekValues, OptionGreeks, OrderBookDelta,
41        QuoteTick, TradeTick,
42    },
43    enums::{BookAction, GreeksConvention, OrderSide},
44    identifiers::{InstrumentId, TradeId},
45    types::{Price, Quantity},
46};
47use rust_decimal::Decimal;
48pub use stream::{
49    stream_deltas, stream_depth_from_snapshot5, stream_depth_from_snapshot25, stream_funding_rates,
50    stream_options_chain, stream_quotes, stream_trades,
51};
52
53use super::csv::record::{
54    TardisBookUpdateRecord, TardisDerivativeTickerRecord, TardisOptionsChainRecord,
55    TardisQuoteRecord, TardisTradeRecord,
56};
57use crate::common::parse::{
58    derive_trade_id, parse_aggressor_side, parse_book_action, parse_instrument_id,
59    parse_order_side, parse_price, parse_timestamp, validate_non_zero_amount,
60};
61
62fn infer_precision(value: f64) -> u8 {
63    let mut buf = ryu::Buffer::new(); // Stack allocation
64    let s = buf.format(value);
65
66    match s.rsplit_once('.') {
67        Some((_, frac)) if frac != "0" => frac.len() as u8,
68        _ => 0,
69    }
70}
71
72fn create_csv_reader<P: AsRef<Path>>(
73    filepath: P,
74) -> anyhow::Result<Reader<Box<dyn std::io::Read>>> {
75    const MAX_RETRIES: u8 = 3;
76    const DELAY_MS: u64 = 100;
77    const BUFFER_SIZE: usize = 8 * 1024 * 1024; // 8MB buffer for large files
78
79    fn open_file_with_retry<P: AsRef<Path>>(
80        path: P,
81        max_retries: u8,
82        delay_ms: u64,
83    ) -> anyhow::Result<File> {
84        let path_ref = path.as_ref();
85        for attempt in 1..=max_retries {
86            match File::open(path_ref) {
87                Ok(file) => return Ok(file),
88                Err(e) => {
89                    if attempt == max_retries {
90                        anyhow::bail!(
91                            "Failed to open file '{}' after {max_retries} attempts: {e}",
92                            path_ref.display()
93                        );
94                    }
95
96                    log::debug!(
97                        "Attempt {attempt}/{max_retries} failed to open file '{}': {e}. Retrying after {delay_ms}ms...",
98                        path_ref.display()
99                    );
100                    std::thread::sleep(Duration::from_millis(delay_ms));
101                }
102            }
103        }
104        unreachable!("Loop should return either Ok or Err");
105    }
106
107    let filepath_ref = filepath.as_ref();
108    let mut file = open_file_with_retry(filepath_ref, MAX_RETRIES, DELAY_MS)?;
109
110    let is_gzipped = filepath_ref
111        .extension()
112        .and_then(OsStr::to_str)
113        .is_some_and(|ext| ext.eq_ignore_ascii_case("gz"));
114
115    if !is_gzipped {
116        let buf_reader = BufReader::with_capacity(BUFFER_SIZE, file);
117        return Ok(ReaderBuilder::new()
118            .has_headers(true)
119            .buffer_capacity(1024 * 1024) // 1MB CSV buffer
120            .from_reader(Box::new(buf_reader)));
121    }
122
123    let file_size = file.metadata()?.len();
124    if file_size < 2 {
125        anyhow::bail!("File too small to be a valid gzip file");
126    }
127
128    let mut header_buf = [0u8; 2];
129    for attempt in 1..=MAX_RETRIES {
130        match file.read_exact(&mut header_buf) {
131            Ok(()) => break,
132            Err(e) => {
133                if attempt == MAX_RETRIES {
134                    anyhow::bail!(
135                        "Failed to read gzip header from '{}' after {MAX_RETRIES} attempts: {e}",
136                        filepath_ref.display()
137                    );
138                }
139
140                log::debug!(
141                    "Attempt {attempt}/{MAX_RETRIES} failed to read header from '{}': {e}. Retrying after {DELAY_MS}ms...",
142                    filepath_ref.display()
143                );
144                std::thread::sleep(Duration::from_millis(DELAY_MS));
145            }
146        }
147    }
148
149    if header_buf[0] != 0x1f || header_buf[1] != 0x8b {
150        anyhow::bail!(
151            "File '{}' has .gz extension but invalid gzip header",
152            filepath_ref.display()
153        );
154    }
155
156    for attempt in 1..=MAX_RETRIES {
157        match file.seek(SeekFrom::Start(0)) {
158            Ok(_) => break,
159            Err(e) => {
160                if attempt == MAX_RETRIES {
161                    anyhow::bail!(
162                        "Failed to reset file position for '{}' after {MAX_RETRIES} attempts: {e}",
163                        filepath_ref.display()
164                    );
165                }
166
167                log::debug!(
168                    "Attempt {attempt}/{MAX_RETRIES} failed to seek in '{}': {e}. Retrying after {DELAY_MS}ms...",
169                    filepath_ref.display()
170                );
171                std::thread::sleep(Duration::from_millis(DELAY_MS));
172            }
173        }
174    }
175
176    let buf_reader = BufReader::with_capacity(BUFFER_SIZE, file);
177    let decoder = GzDecoder::new(buf_reader);
178
179    Ok(ReaderBuilder::new()
180        .has_headers(true)
181        .buffer_capacity(1024 * 1024) // 1MB CSV buffer
182        .from_reader(Box::new(decoder)))
183}
184
185fn create_book_order(
186    side: OrderSide,
187    price: Option<f64>,
188    amount: Option<f64>,
189    price_precision: u8,
190    size_precision: u8,
191) -> (BookOrder, u32) {
192    match price {
193        Some(price) => (
194            BookOrder::new(
195                side,
196                parse_price(price, price_precision),
197                Quantity::new(amount.unwrap_or(0.0), size_precision),
198                0,
199            ),
200            1, // Count set to 1 if order exists
201        ),
202        None => (NULL_ORDER, 0), // NULL_ORDER if price is None
203    }
204}
205
206fn parse_delta_record(
207    data: &TardisBookUpdateRecord,
208    price_precision: u8,
209    size_precision: u8,
210    instrument_id: Option<InstrumentId>,
211) -> anyhow::Result<OrderBookDelta> {
212    let instrument_id = match instrument_id {
213        Some(id) => id,
214        None => parse_instrument_id(&data.exchange, data.symbol),
215    };
216
217    let side = parse_order_side(&data.side);
218    let price = parse_price(data.price, price_precision);
219    let size = Quantity::new(data.amount, size_precision);
220    let order_id = 0; // Not applicable for L2 data
221    let order = BookOrder::new(side, price, size, order_id);
222
223    let action = parse_book_action(data.is_snapshot, size.as_f64());
224    let flags = 0; // Will be set later if needed
225    let sequence = 0; // Sequence not available
226    let ts_event = parse_timestamp(data.timestamp);
227    let ts_init = parse_timestamp(data.local_timestamp);
228
229    anyhow::ensure!(
230        !(action != BookAction::Delete && size.is_zero()),
231        "Invalid delta: action {action} when size zero, check size_precision ({size_precision}) vs data; {data:?}"
232    );
233
234    Ok(OrderBookDelta::new(
235        instrument_id,
236        action,
237        order,
238        flags,
239        sequence,
240        ts_event,
241        ts_init,
242    ))
243}
244
245fn parse_quote_record(
246    data: &TardisQuoteRecord,
247    price_precision: u8,
248    size_precision: u8,
249    instrument_id: Option<InstrumentId>,
250) -> QuoteTick {
251    let instrument_id = match instrument_id {
252        Some(id) => id,
253        None => parse_instrument_id(&data.exchange, data.symbol),
254    };
255
256    let bid_price = parse_price(data.bid_price.unwrap_or(0.0), price_precision);
257    let ask_price = parse_price(data.ask_price.unwrap_or(0.0), price_precision);
258    let bid_size = Quantity::new(data.bid_amount.unwrap_or(0.0), size_precision);
259    let ask_size = Quantity::new(data.ask_amount.unwrap_or(0.0), size_precision);
260    let ts_event = parse_timestamp(data.timestamp);
261    let ts_init = parse_timestamp(data.local_timestamp);
262
263    QuoteTick::new(
264        instrument_id,
265        bid_price,
266        ask_price,
267        bid_size,
268        ask_size,
269        ts_event,
270        ts_init,
271    )
272}
273
274fn parse_trade_record(
275    data: &TardisTradeRecord,
276    size: Quantity,
277    price_precision: u8,
278    instrument_id: Option<InstrumentId>,
279) -> TradeTick {
280    let instrument_id = match instrument_id {
281        Some(id) => id,
282        None => parse_instrument_id(&data.exchange, data.symbol),
283    };
284
285    let price = parse_price(data.price, price_precision);
286    let aggressor_side = parse_aggressor_side(&data.side);
287    let ts_event = parse_timestamp(data.timestamp);
288    let ts_init = parse_timestamp(data.local_timestamp);
289    let trade_id = if data.id.is_empty() {
290        derive_trade_id(
291            data.symbol,
292            ts_event.as_u64(),
293            &data.price.to_string(),
294            &data.amount.to_string(),
295            &data.side,
296        )
297    } else {
298        TradeId::new(&data.id)
299    };
300
301    TradeTick::new(
302        instrument_id,
303        price,
304        size,
305        aggressor_side,
306        trade_id,
307        ts_event,
308        ts_init,
309    )
310}
311
312fn parse_derivative_ticker_record(
313    data: &TardisDerivativeTickerRecord,
314    instrument_id: Option<InstrumentId>,
315) -> Option<FundingRateUpdate> {
316    // Only create funding rate update if we have funding rate data
317    let funding_rate = data.funding_rate?;
318
319    let instrument_id = match instrument_id {
320        Some(id) => id,
321        None => parse_instrument_id(&data.exchange, data.symbol),
322    };
323
324    let rate = Decimal::try_from(funding_rate).ok()?;
325    let next_funding_ns = data.funding_timestamp.map(parse_timestamp);
326    let ts_event = parse_timestamp(data.timestamp);
327    let ts_init = parse_timestamp(data.local_timestamp);
328
329    Some(FundingRateUpdate::new(
330        instrument_id,
331        rate,
332        None,
333        next_funding_ns,
334        ts_event,
335        ts_init,
336    ))
337}
338
339fn parse_options_chain_record(
340    data: &TardisOptionsChainRecord,
341    instrument_id: InstrumentId,
342) -> OptionGreeks {
343    OptionGreeks {
344        instrument_id,
345        convention: GreeksConvention::BlackScholes,
346        greeks: OptionGreekValues {
347            delta: data.delta.unwrap_or(0.0),
348            gamma: data.gamma.unwrap_or(0.0),
349            vega: data.vega.unwrap_or(0.0),
350            theta: data.theta.unwrap_or(0.0),
351            rho: data.rho.unwrap_or(0.0),
352        },
353        mark_iv: data.mark_iv,
354        bid_iv: data.bid_iv,
355        ask_iv: data.ask_iv,
356        underlying_price: data.underlying_price,
357        open_interest: data.open_interest,
358        ts_event: parse_timestamp(data.timestamp),
359        ts_init: parse_timestamp(data.local_timestamp),
360    }
361}
362
363fn parse_options_chain_record_as_quote(
364    data: &TardisOptionsChainRecord,
365    price_precision: u8,
366    size_precision: u8,
367    instrument_id: InstrumentId,
368) -> anyhow::Result<Option<QuoteTick>> {
369    let (Some(bid_price), Some(bid_amount), Some(ask_price), Some(ask_amount)) = (
370        data.bid_price,
371        data.bid_amount,
372        data.ask_price,
373        data.ask_amount,
374    ) else {
375        return Ok(None);
376    };
377
378    let bid_price = Price::new_checked(bid_price, price_precision)?;
379    let ask_price = Price::new_checked(ask_price, price_precision)?;
380    validate_non_zero_amount(bid_amount, size_precision)?;
381    let bid_size = Quantity::new_checked(bid_amount, size_precision)?;
382    validate_non_zero_amount(ask_amount, size_precision)?;
383    let ask_size = Quantity::new_checked(ask_amount, size_precision)?;
384
385    Ok(Some(QuoteTick::new(
386        instrument_id,
387        bid_price,
388        ask_price,
389        bid_size,
390        ask_size,
391        parse_timestamp(data.timestamp),
392        parse_timestamp(data.local_timestamp),
393    )))
394}
395
396fn matches_underlying_filter(symbol: &str, underlyings: Option<&[String]>) -> bool {
397    underlyings.is_none_or(|underlyings| underlyings.iter().any(|u| symbol.starts_with(u)))
398}
399
400fn normalize_underlying_filters(underlyings: Option<Vec<String>>) -> Option<Vec<String>> {
401    underlyings
402        .map(|values| {
403            values
404                .into_iter()
405                .map(|value| value.trim().to_uppercase())
406                .filter(|value| !value.is_empty())
407                .collect::<Vec<_>>()
408        })
409        .filter(|values| !values.is_empty())
410}
nautilus_tardis/csv/mod.rs

nautilus_tardis/csv/
mod.rs
