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

nautilus_tardis/csv/
mod.rs
