nautilus_tardis/csv/
load.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
16use std::{error::Error, path::Path};
17
18use ahash::AHashMap;
19use csv::StringRecord;
20use nautilus_core::UnixNanos;
21use nautilus_model::{
22    data::{
23        DEPTH10_LEN, Data, FundingRateUpdate, NULL_ORDER, OrderBookDelta, OrderBookDepth10,
24        QuoteTick, TradeTick,
25    },
26    enums::{OrderSide, RecordFlag},
27    identifiers::InstrumentId,
28    types::{Quantity, fixed::FIXED_PRECISION},
29};
30
31use crate::{
32    common::parse::{parse_instrument_id, parse_timestamp},
33    csv::{
34        create_book_order, create_csv_reader, infer_precision, matches_underlying_filter,
35        normalize_underlying_filters, parse_delta_record, parse_derivative_ticker_record,
36        parse_options_chain_record, parse_options_chain_record_as_quote, parse_quote_record,
37        parse_trade_record,
38        record::{
39            TardisBookUpdateRecord, TardisDerivativeTickerRecord, TardisOptionsChainRecord,
40            TardisOrderBookSnapshot5Record, TardisOrderBookSnapshot25Record, TardisQuoteRecord,
41            TardisTradeRecord,
42        },
43    },
44};
45
46#[derive(Debug, Clone, Copy)]
47pub(in crate::csv) struct OptionsChainPrecision {
48    pub(in crate::csv) price: u8,
49    pub(in crate::csv) size: u8,
50}
51
52impl OptionsChainPrecision {
53    pub(in crate::csv) const fn new(
54        price_precision: Option<u8>,
55        size_precision: Option<u8>,
56    ) -> Self {
57        Self {
58            price: match price_precision {
59                Some(precision) => precision,
60                None => 0,
61            },
62            size: match size_precision {
63                Some(precision) => precision,
64                None => 0,
65            },
66        }
67    }
68
69    pub(in crate::csv) fn update(
70        &mut self,
71        record: &TardisOptionsChainRecord,
72        price_precision: Option<u8>,
73        size_precision: Option<u8>,
74    ) {
75        if price_precision.is_none() {
76            for value in [record.last_price, record.bid_price, record.ask_price]
77                .into_iter()
78                .flatten()
79            {
80                update_precision_if_needed(&mut self.price, value, price_precision);
81            }
82        }
83
84        if size_precision.is_none() {
85            for value in [record.bid_amount, record.ask_amount].into_iter().flatten() {
86                update_precision_if_needed(&mut self.size, value, size_precision);
87            }
88        }
89    }
90}
91
92fn update_precision_if_needed(current: &mut u8, value: f64, explicit: Option<u8>) -> bool {
93    if explicit.is_some() {
94        return false;
95    }
96
97    let inferred = infer_precision(value).min(FIXED_PRECISION);
98    if inferred > *current {
99        *current = inferred;
100        true
101    } else {
102        false
103    }
104}
105
106fn update_deltas_precision(
107    deltas: &mut [OrderBookDelta],
108    price_precision: Option<u8>,
109    size_precision: Option<u8>,
110    current_price_precision: u8,
111    current_size_precision: u8,
112) {
113    for delta in deltas {
114        if price_precision.is_none() {
115            delta.order.price.precision = current_price_precision;
116        }
117
118        if size_precision.is_none() {
119            delta.order.size.precision = current_size_precision;
120        }
121    }
122}
123
124fn update_quotes_precision(
125    quotes: &mut [QuoteTick],
126    price_precision: Option<u8>,
127    size_precision: Option<u8>,
128    current_price_precision: u8,
129    current_size_precision: u8,
130) {
131    for quote in quotes {
132        if price_precision.is_none() {
133            quote.bid_price.precision = current_price_precision;
134            quote.ask_price.precision = current_price_precision;
135        }
136
137        if size_precision.is_none() {
138            quote.bid_size.precision = current_size_precision;
139            quote.ask_size.precision = current_size_precision;
140        }
141    }
142}
143
144fn update_trades_precision(
145    trades: &mut [TradeTick],
146    price_precision: Option<u8>,
147    size_precision: Option<u8>,
148    current_price_precision: u8,
149    current_size_precision: u8,
150) {
151    for trade in trades {
152        if price_precision.is_none() {
153            trade.price.precision = current_price_precision;
154        }
155
156        if size_precision.is_none() {
157            trade.size.precision = current_size_precision;
158        }
159    }
160}
161
162/// Loads [`OrderBookDelta`]s from a Tardis format CSV at the given `filepath`,
163/// automatically applying `GZip` decompression for files ending in ".gz".
164/// Load order book delta records from a CSV or gzipped CSV file.
165///
166/// # Errors
167///
168/// Returns an error if the file cannot be opened, read, or parsed as CSV.
169pub fn load_deltas<P: AsRef<Path>>(
170    filepath: P,
171    price_precision: Option<u8>,
172    size_precision: Option<u8>,
173    instrument_id: Option<InstrumentId>,
174    limit: Option<usize>,
175) -> Result<Vec<OrderBookDelta>, Box<dyn Error>> {
176    // Estimate capacity for Vec pre-allocation
177    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
178    let mut deltas: Vec<OrderBookDelta> = Vec::with_capacity(estimated_capacity);
179
180    let mut current_price_precision = price_precision.unwrap_or(0);
181    let mut current_size_precision = size_precision.unwrap_or(0);
182    let mut last_ts_event = UnixNanos::default();
183    let mut last_is_snapshot = false;
184
185    let mut reader = create_csv_reader(filepath)?;
186    let mut record = StringRecord::new();
187
188    while reader.read_record(&mut record)? {
189        if let Some(limit) = limit
190            && deltas.len() >= limit
191        {
192            break;
193        }
194
195        let data: TardisBookUpdateRecord = record.deserialize(None)?;
196
197        update_precision_if_needed(&mut current_price_precision, data.price, price_precision);
198        update_precision_if_needed(&mut current_size_precision, data.amount, size_precision);
199
200        let ts_event = parse_timestamp(data.timestamp);
201        let ts_init = parse_timestamp(data.local_timestamp);
202
203        // Insert CLEAR on snapshot boundary to reset order book state.
204        // Some venues emit every book event as a full snapshot, so a new
205        // snapshot timestamp must also reset the previous snapshot state.
206        let starts_new_snapshot =
207            data.is_snapshot && (!last_is_snapshot || last_ts_event != ts_event);
208
209        if starts_new_snapshot {
210            let clear_instrument_id =
211                instrument_id.unwrap_or_else(|| parse_instrument_id(&data.exchange, data.symbol));
212
213            if last_ts_event != ts_event
214                && let Some(last_delta) = deltas.last_mut()
215            {
216                last_delta.flags = RecordFlag::F_LAST as u8;
217            }
218            last_ts_event = ts_event;
219
220            let clear_delta = OrderBookDelta::clear(clear_instrument_id, 0, ts_event, ts_init);
221            deltas.push(clear_delta);
222
223            if let Some(limit) = limit
224                && deltas.len() >= limit
225            {
226                break;
227            }
228        }
229        last_is_snapshot = data.is_snapshot;
230
231        let delta = match parse_delta_record(
232            &data,
233            current_price_precision,
234            current_size_precision,
235            instrument_id,
236        ) {
237            Ok(d) => d,
238            Err(e) => {
239                log::warn!("Skipping invalid delta record: {e}");
240                continue;
241            }
242        };
243
244        let ts_event = delta.ts_event;
245        if last_ts_event != ts_event
246            && let Some(last_delta) = deltas.last_mut()
247        {
248            last_delta.flags = RecordFlag::F_LAST as u8;
249        }
250
251        last_ts_event = ts_event;
252
253        deltas.push(delta);
254    }
255
256    // Set F_LAST flag for final delta
257    if let Some(last_delta) = deltas.last_mut() {
258        last_delta.flags = RecordFlag::F_LAST as u8;
259    }
260
261    // Update all deltas to use the final (maximum) precision discovered
262    // This is done once at the end instead of on every precision change (O(n) vs O(n²))
263    update_deltas_precision(
264        &mut deltas,
265        price_precision,
266        size_precision,
267        current_price_precision,
268        current_size_precision,
269    );
270
271    Ok(deltas)
272}
273
274/// Loads [`OrderBookDepth10`]s from a Tardis format CSV at the given `filepath`,
275/// automatically applying `GZip` decompression for files ending in ".gz".
276/// Load order book depth-10 snapshots (5-level) from a CSV or gzipped CSV file.
277///
278/// # Errors
279///
280/// Returns an error if the file cannot be opened, read, or parsed as CSV.
281///
282/// # Panics
283///
284/// Panics if a record level cannot be parsed to depth-10.
285pub fn load_depth10_from_snapshot5<P: AsRef<Path>>(
286    filepath: P,
287    price_precision: Option<u8>,
288    size_precision: Option<u8>,
289    instrument_id: Option<InstrumentId>,
290    limit: Option<usize>,
291) -> Result<Vec<OrderBookDepth10>, Box<dyn Error>> {
292    // Estimate capacity for Vec pre-allocation
293    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
294    let mut depths: Vec<OrderBookDepth10> = Vec::with_capacity(estimated_capacity);
295
296    let mut current_price_precision = price_precision.unwrap_or(0);
297    let mut current_size_precision = size_precision.unwrap_or(0);
298
299    let mut reader = create_csv_reader(filepath)?;
300    let mut record = StringRecord::new();
301
302    while reader.read_record(&mut record)? {
303        let data: TardisOrderBookSnapshot5Record = record.deserialize(None)?;
304
305        // Update precisions dynamically if not explicitly set
306        let mut precision_updated = false;
307
308        if price_precision.is_none()
309            && let Some(bid_price) = data.bids_0_price
310        {
311            let inferred_price_precision = infer_precision(bid_price).min(FIXED_PRECISION);
312            if inferred_price_precision > current_price_precision {
313                current_price_precision = inferred_price_precision;
314                precision_updated = true;
315            }
316        }
317
318        if size_precision.is_none()
319            && let Some(bid_amount) = data.bids_0_amount
320        {
321            let inferred_size_precision = infer_precision(bid_amount).min(FIXED_PRECISION);
322            if inferred_size_precision > current_size_precision {
323                current_size_precision = inferred_size_precision;
324                precision_updated = true;
325            }
326        }
327
328        // If precision increased, update all previous depths
329        if precision_updated {
330            for depth in &mut depths {
331                for i in 0..DEPTH10_LEN {
332                    if price_precision.is_none() {
333                        depth.bids[i].price.precision = current_price_precision;
334                        depth.asks[i].price.precision = current_price_precision;
335                    }
336
337                    if size_precision.is_none() {
338                        depth.bids[i].size.precision = current_size_precision;
339                        depth.asks[i].size.precision = current_size_precision;
340                    }
341                }
342            }
343        }
344
345        let instrument_id = match &instrument_id {
346            Some(id) => *id,
347            None => parse_instrument_id(&data.exchange, data.symbol),
348        };
349        // Mark as both snapshot and last (consistent with streaming implementation)
350        let flags = RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8;
351        let sequence = 0; // Sequence not available
352        let ts_event = parse_timestamp(data.timestamp);
353        let ts_init = parse_timestamp(data.local_timestamp);
354
355        // Initialize empty arrays
356        let mut bids = [NULL_ORDER; DEPTH10_LEN];
357        let mut asks = [NULL_ORDER; DEPTH10_LEN];
358        let mut bid_counts = [0u32; DEPTH10_LEN];
359        let mut ask_counts = [0u32; DEPTH10_LEN];
360
361        for i in 0..=4 {
362            // Create bids
363            let (bid_order, bid_count) = create_book_order(
364                OrderSide::Buy,
365                match i {
366                    0 => data.bids_0_price,
367                    1 => data.bids_1_price,
368                    2 => data.bids_2_price,
369                    3 => data.bids_3_price,
370                    4 => data.bids_4_price,
371                    _ => unreachable!("i is constrained to 0..=4 by loop"),
372                },
373                match i {
374                    0 => data.bids_0_amount,
375                    1 => data.bids_1_amount,
376                    2 => data.bids_2_amount,
377                    3 => data.bids_3_amount,
378                    4 => data.bids_4_amount,
379                    _ => unreachable!("i is constrained to 0..=4 by loop"),
380                },
381                current_price_precision,
382                current_size_precision,
383            );
384            bids[i] = bid_order;
385            bid_counts[i] = bid_count;
386
387            // Create asks
388            let (ask_order, ask_count) = create_book_order(
389                OrderSide::Sell,
390                match i {
391                    0 => data.asks_0_price,
392                    1 => data.asks_1_price,
393                    2 => data.asks_2_price,
394                    3 => data.asks_3_price,
395                    4 => data.asks_4_price,
396                    _ => None, // Unreachable, but for safety
397                },
398                match i {
399                    0 => data.asks_0_amount,
400                    1 => data.asks_1_amount,
401                    2 => data.asks_2_amount,
402                    3 => data.asks_3_amount,
403                    4 => data.asks_4_amount,
404                    _ => None, // Unreachable, but for safety
405                },
406                current_price_precision,
407                current_size_precision,
408            );
409            asks[i] = ask_order;
410            ask_counts[i] = ask_count;
411        }
412
413        let depth = OrderBookDepth10::new(
414            instrument_id,
415            bids,
416            asks,
417            bid_counts,
418            ask_counts,
419            flags,
420            sequence,
421            ts_event,
422            ts_init,
423        );
424
425        depths.push(depth);
426
427        if let Some(limit) = limit
428            && depths.len() >= limit
429        {
430            break;
431        }
432    }
433
434    Ok(depths)
435}
436
437/// Loads [`OrderBookDepth10`]s from a Tardis format CSV at the given `filepath`,
438/// automatically applying `GZip` decompression for files ending in ".gz".
439/// Load order book depth-10 snapshots (25-level) from a CSV or gzipped CSV file.
440///
441/// # Errors
442///
443/// Returns an error if the file cannot be opened, read, or parsed as CSV.
444pub fn load_depth10_from_snapshot25<P: AsRef<Path>>(
445    filepath: P,
446    price_precision: Option<u8>,
447    size_precision: Option<u8>,
448    instrument_id: Option<InstrumentId>,
449    limit: Option<usize>,
450) -> Result<Vec<OrderBookDepth10>, Box<dyn Error>> {
451    // Estimate capacity for Vec pre-allocation
452    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
453    let mut depths: Vec<OrderBookDepth10> = Vec::with_capacity(estimated_capacity);
454
455    let mut current_price_precision = price_precision.unwrap_or(0);
456    let mut current_size_precision = size_precision.unwrap_or(0);
457    let mut reader = create_csv_reader(filepath)?;
458    let mut record = StringRecord::new();
459
460    while reader.read_record(&mut record)? {
461        let data: TardisOrderBookSnapshot25Record = record.deserialize(None)?;
462
463        // Update precisions dynamically if not explicitly set
464        let mut precision_updated = false;
465
466        if price_precision.is_none()
467            && let Some(bid_price) = data.bids_0_price
468        {
469            let inferred_price_precision = infer_precision(bid_price).min(FIXED_PRECISION);
470            if inferred_price_precision > current_price_precision {
471                current_price_precision = inferred_price_precision;
472                precision_updated = true;
473            }
474        }
475
476        if size_precision.is_none()
477            && let Some(bid_amount) = data.bids_0_amount
478        {
479            let inferred_size_precision = infer_precision(bid_amount).min(FIXED_PRECISION);
480            if inferred_size_precision > current_size_precision {
481                current_size_precision = inferred_size_precision;
482                precision_updated = true;
483            }
484        }
485
486        // If precision increased, update all previous depths
487        if precision_updated {
488            for depth in &mut depths {
489                for i in 0..DEPTH10_LEN {
490                    if price_precision.is_none() {
491                        depth.bids[i].price.precision = current_price_precision;
492                        depth.asks[i].price.precision = current_price_precision;
493                    }
494
495                    if size_precision.is_none() {
496                        depth.bids[i].size.precision = current_size_precision;
497                        depth.asks[i].size.precision = current_size_precision;
498                    }
499                }
500            }
501        }
502
503        let instrument_id = match &instrument_id {
504            Some(id) => *id,
505            None => parse_instrument_id(&data.exchange, data.symbol),
506        };
507        // Mark as both snapshot and last (consistent with streaming implementation)
508        let flags = RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8;
509        let sequence = 0; // Sequence not available
510        let ts_event = parse_timestamp(data.timestamp);
511        let ts_init = parse_timestamp(data.local_timestamp);
512
513        // Initialize empty arrays for the first 10 levels only
514        let mut bids = [NULL_ORDER; DEPTH10_LEN];
515        let mut asks = [NULL_ORDER; DEPTH10_LEN];
516        let mut bid_counts = [0u32; DEPTH10_LEN];
517        let mut ask_counts = [0u32; DEPTH10_LEN];
518
519        // Fill only the first 10 levels from the 25-level record
520        for i in 0..DEPTH10_LEN {
521            // Create bids
522            let (bid_order, bid_count) = create_book_order(
523                OrderSide::Buy,
524                match i {
525                    0 => data.bids_0_price,
526                    1 => data.bids_1_price,
527                    2 => data.bids_2_price,
528                    3 => data.bids_3_price,
529                    4 => data.bids_4_price,
530                    5 => data.bids_5_price,
531                    6 => data.bids_6_price,
532                    7 => data.bids_7_price,
533                    8 => data.bids_8_price,
534                    9 => data.bids_9_price,
535                    _ => unreachable!("i is constrained to 0..10 by loop"),
536                },
537                match i {
538                    0 => data.bids_0_amount,
539                    1 => data.bids_1_amount,
540                    2 => data.bids_2_amount,
541                    3 => data.bids_3_amount,
542                    4 => data.bids_4_amount,
543                    5 => data.bids_5_amount,
544                    6 => data.bids_6_amount,
545                    7 => data.bids_7_amount,
546                    8 => data.bids_8_amount,
547                    9 => data.bids_9_amount,
548                    _ => unreachable!("i is constrained to 0..10 by loop"),
549                },
550                current_price_precision,
551                current_size_precision,
552            );
553            bids[i] = bid_order;
554            bid_counts[i] = bid_count;
555
556            // Create asks
557            let (ask_order, ask_count) = create_book_order(
558                OrderSide::Sell,
559                match i {
560                    0 => data.asks_0_price,
561                    1 => data.asks_1_price,
562                    2 => data.asks_2_price,
563                    3 => data.asks_3_price,
564                    4 => data.asks_4_price,
565                    5 => data.asks_5_price,
566                    6 => data.asks_6_price,
567                    7 => data.asks_7_price,
568                    8 => data.asks_8_price,
569                    9 => data.asks_9_price,
570                    _ => unreachable!("i is constrained to 0..10 by loop"),
571                },
572                match i {
573                    0 => data.asks_0_amount,
574                    1 => data.asks_1_amount,
575                    2 => data.asks_2_amount,
576                    3 => data.asks_3_amount,
577                    4 => data.asks_4_amount,
578                    5 => data.asks_5_amount,
579                    6 => data.asks_6_amount,
580                    7 => data.asks_7_amount,
581                    8 => data.asks_8_amount,
582                    9 => data.asks_9_amount,
583                    _ => unreachable!("i is constrained to 0..10 by loop"),
584                },
585                current_price_precision,
586                current_size_precision,
587            );
588            asks[i] = ask_order;
589            ask_counts[i] = ask_count;
590        }
591
592        let depth = OrderBookDepth10::new(
593            instrument_id,
594            bids,
595            asks,
596            bid_counts,
597            ask_counts,
598            flags,
599            sequence,
600            ts_event,
601            ts_init,
602        );
603
604        depths.push(depth);
605
606        if let Some(limit) = limit
607            && depths.len() >= limit
608        {
609            break;
610        }
611    }
612
613    Ok(depths)
614}
615
616/// Loads [`QuoteTick`]s from a Tardis format CSV at the given `filepath`,
617/// automatically applying `GZip` decompression for files ending in ".gz".
618/// Load quote ticks from a CSV or gzipped CSV file.
619///
620/// # Errors
621///
622/// Returns an error if the file cannot be opened, read, or parsed as CSV.
623pub fn load_quotes<P: AsRef<Path>>(
624    filepath: P,
625    price_precision: Option<u8>,
626    size_precision: Option<u8>,
627    instrument_id: Option<InstrumentId>,
628    limit: Option<usize>,
629) -> Result<Vec<QuoteTick>, Box<dyn Error>> {
630    // Estimate capacity for Vec pre-allocation
631    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
632    let mut quotes: Vec<QuoteTick> = Vec::with_capacity(estimated_capacity);
633
634    let mut current_price_precision = price_precision.unwrap_or(0);
635    let mut current_size_precision = size_precision.unwrap_or(0);
636    let mut reader = create_csv_reader(filepath)?;
637    let mut record = StringRecord::new();
638
639    while reader.read_record(&mut record)? {
640        let data: TardisQuoteRecord = record.deserialize(None)?;
641
642        if price_precision.is_none()
643            && let Some(bid_price) = data.bid_price
644        {
645            let inferred_price_precision = infer_precision(bid_price).min(FIXED_PRECISION);
646            if inferred_price_precision > current_price_precision {
647                current_price_precision = inferred_price_precision;
648            }
649        }
650
651        if size_precision.is_none()
652            && let Some(bid_amount) = data.bid_amount
653        {
654            let inferred_size_precision = infer_precision(bid_amount).min(FIXED_PRECISION);
655            if inferred_size_precision > current_size_precision {
656                current_size_precision = inferred_size_precision;
657            }
658        }
659
660        let quote = parse_quote_record(
661            &data,
662            current_price_precision,
663            current_size_precision,
664            instrument_id,
665        );
666
667        quotes.push(quote);
668
669        if let Some(limit) = limit
670            && quotes.len() >= limit
671        {
672            break;
673        }
674    }
675
676    // Update all quotes to use the final (maximum) precision discovered
677    // This is done once at the end instead of on every precision change (O(n) vs O(n²))
678    update_quotes_precision(
679        &mut quotes,
680        price_precision,
681        size_precision,
682        current_price_precision,
683        current_size_precision,
684    );
685
686    Ok(quotes)
687}
688
689/// Loads [`TradeTick`]s from a Tardis format CSV at the given `filepath`,
690/// automatically applying `GZip` decompression for files ending in ".gz".
691/// Load trade ticks from a CSV or gzipped CSV file.
692///
693/// # Errors
694///
695/// Returns an error if the file cannot be opened, read, or parsed as CSV.
696pub fn load_trades<P: AsRef<Path>>(
697    filepath: P,
698    price_precision: Option<u8>,
699    size_precision: Option<u8>,
700    instrument_id: Option<InstrumentId>,
701    limit: Option<usize>,
702) -> Result<Vec<TradeTick>, Box<dyn Error>> {
703    // Estimate capacity for Vec pre-allocation
704    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
705    let mut trades: Vec<TradeTick> = Vec::with_capacity(estimated_capacity);
706
707    let mut current_price_precision = price_precision.unwrap_or(0);
708    let mut current_size_precision = size_precision.unwrap_or(0);
709    let mut reader = create_csv_reader(filepath)?;
710    let mut record = StringRecord::new();
711
712    while reader.read_record(&mut record)? {
713        let data: TardisTradeRecord = record.deserialize(None)?;
714
715        if price_precision.is_none() {
716            let inferred_price_precision = infer_precision(data.price).min(FIXED_PRECISION);
717            if inferred_price_precision > current_price_precision {
718                current_price_precision = inferred_price_precision;
719            }
720        }
721
722        if size_precision.is_none() {
723            let inferred_size_precision = infer_precision(data.amount).min(FIXED_PRECISION);
724            if inferred_size_precision > current_size_precision {
725                current_size_precision = inferred_size_precision;
726            }
727        }
728
729        let size = Quantity::new_checked(data.amount, current_size_precision)?;
730
731        if size.is_positive() {
732            let trade = parse_trade_record(&data, size, current_price_precision, instrument_id);
733
734            trades.push(trade);
735
736            if let Some(limit) = limit
737                && trades.len() >= limit
738            {
739                break;
740            }
741        } else {
742            log::warn!("Skipping zero-sized trade: {data:?}");
743        }
744    }
745
746    // Update all trades to use the final (maximum) precision discovered
747    // This is done once at the end instead of on every precision change (O(n) vs O(n²))
748    update_trades_precision(
749        &mut trades,
750        price_precision,
751        size_precision,
752        current_price_precision,
753        current_size_precision,
754    );
755
756    Ok(trades)
757}
758
759/// Loads [`FundingRateUpdate`]s from a Tardis format derivative ticker CSV at the given `filepath`,
760/// automatically applying `GZip` decompression for files ending in ".gz".
761///
762/// This function parses the `funding_rate`, `predicted_funding_rate`, and `funding_timestamp`
763/// fields from derivative ticker data to create funding rate updates.
764///
765/// # Errors
766///
767/// Returns an error if the file cannot be opened, read, or parsed as CSV.
768pub fn load_funding_rates<P: AsRef<Path>>(
769    filepath: P,
770    instrument_id: Option<InstrumentId>,
771    limit: Option<usize>,
772) -> Result<Vec<FundingRateUpdate>, Box<dyn Error>> {
773    // Estimate capacity for Vec pre-allocation
774    let estimated_capacity = limit.unwrap_or(100_000).min(1_000_000);
775    let mut funding_rates: Vec<FundingRateUpdate> = Vec::with_capacity(estimated_capacity);
776
777    let mut reader = create_csv_reader(filepath)?;
778    let mut record = StringRecord::new();
779
780    while reader.read_record(&mut record)? {
781        let data: TardisDerivativeTickerRecord = record.deserialize(None)?;
782
783        // Parse to funding rate update (returns None if no funding data)
784        if let Some(funding_rate) = parse_derivative_ticker_record(&data, instrument_id) {
785            funding_rates.push(funding_rate);
786
787            if let Some(limit) = limit
788                && funding_rates.len() >= limit
789            {
790                break;
791            }
792        }
793    }
794
795    Ok(funding_rates)
796}
797
798/// Loads option chain rows from a Tardis `options_chain` CSV file.
799///
800/// Returns quote ticks before option greeks for rows with a complete best bid/offer. Rows missing
801/// any best bid/offer field still return option greeks.
802///
803/// # Errors
804///
805/// Returns an error if the file cannot be opened, read, or parsed as CSV, or if a complete best
806/// bid/offer row contains invalid price or size values.
807pub fn load_options_chain<P: AsRef<Path>>(
808    filepath: P,
809    underlyings: Option<Vec<String>>,
810    price_precision: Option<u8>,
811    size_precision: Option<u8>,
812    limit: Option<usize>,
813) -> Result<Vec<Data>, Box<dyn Error>> {
814    let underlyings = normalize_underlying_filters(underlyings);
815    let estimated_capacity = limit.unwrap_or(1_000_000).min(10_000_000);
816    let mut records: Vec<TardisOptionsChainRecord> = Vec::with_capacity(estimated_capacity);
817    let mut precision_by_instrument: AHashMap<InstrumentId, OptionsChainPrecision> =
818        AHashMap::new();
819
820    let mut reader = create_csv_reader(filepath)?;
821    let mut record = StringRecord::new();
822
823    while reader.read_record(&mut record)? {
824        if let Some(underlyings) = underlyings.as_deref() {
825            let Some(symbol) = record.get(1) else {
826                continue;
827            };
828            let symbol = symbol.to_uppercase();
829            if !matches_underlying_filter(&symbol, Some(underlyings)) {
830                continue;
831            }
832        }
833
834        let data: TardisOptionsChainRecord = record.deserialize(None)?;
835        let instrument_id = parse_instrument_id(&data.exchange, data.symbol);
836        precision_by_instrument
837            .entry(instrument_id)
838            .or_insert_with(|| OptionsChainPrecision::new(price_precision, size_precision))
839            .update(&data, price_precision, size_precision);
840        records.push(data);
841
842        if let Some(limit) = limit
843            && records.len() >= limit
844        {
845            break;
846        }
847    }
848
849    let mut output = Vec::with_capacity(records.len() * 2);
850    for record in records {
851        let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
852        let precision = precision_by_instrument
853            .get(&instrument_id)
854            .copied()
855            .unwrap_or_else(|| OptionsChainPrecision::new(price_precision, size_precision));
856
857        if let Some(quote) = parse_options_chain_record_as_quote(
858            &record,
859            precision.price,
860            precision.size,
861            instrument_id,
862        )? {
863            output.push(Data::Quote(quote));
864        }
865
866        output.push(Data::OptionGreeks(parse_options_chain_record(
867            &record,
868            instrument_id,
869        )));
870    }
871
872    Ok(output)
873}
874
875#[cfg(test)]
876mod tests {
877    use std::{fs, fs::File, sync::Arc};
878
879    use nautilus_core::paths::get_test_data_path as get_test_data_root;
880    use nautilus_model::{
881        enums::{AggressorSide, BookAction},
882        identifiers::{InstrumentId, TradeId},
883        types::Price,
884    };
885    use nautilus_serialization::arrow::{ArrowSchemaProvider, EncodeToRecordBatch};
886    use nautilus_testkit::common::{
887        get_tardis_binance_snapshot5_path, get_tardis_binance_snapshot25_path,
888        get_tardis_bitmex_trades_path, get_tardis_deribit_book_l2_path,
889        get_tardis_huobi_quotes_path,
890    };
891    use parquet::{arrow::ArrowWriter, file::properties::WriterProperties};
892    use rstest::*;
893
894    use super::*;
895    use crate::common::{parse::parse_price, testing::get_test_data_path};
896
897    #[rstest]
898    #[case(0.0, 0)]
899    #[case(42.0, 0)]
900    #[case(0.1, 1)]
901    #[case(0.25, 2)]
902    #[case(123.0001, 4)]
903    #[case(-42.987654321,       9)]
904    #[case(1.234_567_890_123, 12)]
905    fn test_infer_precision(#[case] input: f64, #[case] expected: u8) {
906        assert_eq!(infer_precision(input), expected);
907    }
908
909    #[rstest]
910    pub fn test_dynamic_precision_inference() {
911        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
912binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50000.0,1.0
913binance-futures,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.5,2.0
914binance-futures,BTCUSDT,1640995202000000,1640995202100000,false,ask,50000.12,1.5
915binance-futures,BTCUSDT,1640995203000000,1640995203100000,false,bid,49999.123,3.0
916binance-futures,BTCUSDT,1640995204000000,1640995204100000,false,ask,50000.1234,0.5";
917
918        let temp_file = std::env::temp_dir().join("test_dynamic_precision.csv");
919        std::fs::write(&temp_file, csv_data).unwrap();
920
921        let deltas = load_deltas(&temp_file, None, None, None, None).unwrap();
922
923        // 5 data rows + 1 CLEAR delta at start (first row is snapshot)
924        assert_eq!(deltas.len(), 6);
925
926        // Skip the CLEAR delta at index 0
927        for (i, delta) in deltas.iter().skip(1).enumerate() {
928            assert_eq!(
929                delta.order.price.precision, 4,
930                "Price precision should be 4 for delta {i}",
931            );
932            assert_eq!(
933                delta.order.size.precision, 1,
934                "Size precision should be 1 for delta {i}",
935            );
936        }
937
938        // Test exact values to ensure retroactive precision updates work correctly
939        // Index 0 is CLEAR, data starts at index 1
940        assert_eq!(deltas[0].action, BookAction::Clear);
941
942        assert_eq!(deltas[1].order.price, parse_price(50000.0, 4));
943        assert_eq!(deltas[1].order.size, Quantity::new(1.0, 1));
944
945        assert_eq!(deltas[2].order.price, parse_price(49999.5, 4));
946        assert_eq!(deltas[2].order.size, Quantity::new(2.0, 1));
947
948        assert_eq!(deltas[3].order.price, parse_price(50000.12, 4));
949        assert_eq!(deltas[3].order.size, Quantity::new(1.5, 1));
950
951        assert_eq!(deltas[4].order.price, parse_price(49999.123, 4));
952        assert_eq!(deltas[4].order.size, Quantity::new(3.0, 1));
953
954        assert_eq!(deltas[5].order.price, parse_price(50000.1234, 4));
955        assert_eq!(deltas[5].order.size, Quantity::new(0.5, 1));
956
957        assert_eq!(
958            deltas[1].order.price.precision,
959            deltas[5].order.price.precision
960        );
961        assert_eq!(
962            deltas[1].order.size.precision,
963            deltas[3].order.size.precision
964        );
965
966        std::fs::remove_file(&temp_file).ok();
967    }
968
969    #[rstest]
970    #[case(Some(1), Some(0))] // Explicit precisions
971    #[case(None, None)] // Inferred precisions
972    pub fn test_read_deltas(
973        #[case] price_precision: Option<u8>,
974        #[case] size_precision: Option<u8>,
975    ) {
976        let filepath = get_tardis_deribit_book_l2_path();
977        let deltas =
978            load_deltas(filepath, price_precision, size_precision, None, Some(100)).unwrap();
979
980        // 15 data rows + 1 CLEAR delta at start (first row is snapshot)
981        assert_eq!(deltas.len(), 16);
982
983        // Index 0 is CLEAR delta
984        assert_eq!(deltas[0].action, BookAction::Clear);
985
986        // Index 1 is first data delta
987        assert_eq!(
988            deltas[1].instrument_id,
989            InstrumentId::from("BTC-PERPETUAL.DERIBIT")
990        );
991        assert_eq!(deltas[1].action, BookAction::Add);
992        assert_eq!(deltas[1].order.side, OrderSide::Sell);
993        assert_eq!(deltas[1].order.price, Price::from("6421.5"));
994        assert_eq!(deltas[1].order.size, Quantity::from("18640"));
995        assert_eq!(deltas[1].flags, 0);
996        assert_eq!(deltas[1].sequence, 0);
997        assert_eq!(deltas[1].ts_event, 1585699200245000000);
998        assert_eq!(deltas[1].ts_init, 1585699200355684000);
999    }
1000
1001    #[rstest]
1002    #[case(Some(2), Some(3))] // Explicit precisions
1003    #[case(None, None)] // Inferred precisions
1004    pub fn test_read_depth10s_from_snapshot5(
1005        #[case] price_precision: Option<u8>,
1006        #[case] size_precision: Option<u8>,
1007    ) {
1008        let filepath = get_tardis_binance_snapshot5_path();
1009        let depths =
1010            load_depth10_from_snapshot5(filepath, price_precision, size_precision, None, Some(100))
1011                .unwrap();
1012
1013        assert_eq!(depths.len(), 10);
1014        assert_eq!(
1015            depths[0].instrument_id,
1016            InstrumentId::from("BTCUSDT.BINANCE")
1017        );
1018        assert_eq!(depths[0].bids.len(), 10);
1019        assert_eq!(depths[0].bids[0].price, Price::from("11657.07"));
1020        assert_eq!(depths[0].bids[0].size, Quantity::from("10.896"));
1021        assert_eq!(depths[0].bids[0].side, OrderSide::Buy);
1022        assert_eq!(depths[0].bids[0].order_id, 0);
1023        assert_eq!(depths[0].asks.len(), 10);
1024        assert_eq!(depths[0].asks[0].price, Price::from("11657.08"));
1025        assert_eq!(depths[0].asks[0].size, Quantity::from("1.714"));
1026        assert_eq!(depths[0].asks[0].side, OrderSide::Sell);
1027        assert_eq!(depths[0].asks[0].order_id, 0);
1028        assert_eq!(depths[0].bid_counts[0], 1);
1029        assert_eq!(depths[0].ask_counts[0], 1);
1030        // F_SNAPSHOT (32) | F_LAST (128) = 160
1031        assert_eq!(
1032            depths[0].flags,
1033            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1034        );
1035        assert_eq!(depths[0].ts_event, 1598918403696000000);
1036        assert_eq!(depths[0].ts_init, 1598918403810979000);
1037        assert_eq!(depths[0].sequence, 0);
1038    }
1039
1040    #[rstest]
1041    #[case(Some(2), Some(3))] // Explicit precisions
1042    #[case(None, None)] // Inferred precisions
1043    pub fn test_read_depth10s_from_snapshot25(
1044        #[case] price_precision: Option<u8>,
1045        #[case] size_precision: Option<u8>,
1046    ) {
1047        let filepath = get_tardis_binance_snapshot25_path();
1048        let depths = load_depth10_from_snapshot25(
1049            filepath,
1050            price_precision,
1051            size_precision,
1052            None,
1053            Some(100),
1054        )
1055        .unwrap();
1056
1057        assert_eq!(depths.len(), 10);
1058        assert_eq!(
1059            depths[0].instrument_id,
1060            InstrumentId::from("BTCUSDT.BINANCE")
1061        );
1062        assert_eq!(depths[0].bids.len(), 10);
1063        assert_eq!(depths[0].bids[0].price, Price::from("11657.07"));
1064        assert_eq!(depths[0].bids[0].size, Quantity::from("10.896"));
1065        assert_eq!(depths[0].bids[0].side, OrderSide::Buy);
1066        assert_eq!(depths[0].bids[0].order_id, 0);
1067        assert_eq!(depths[0].asks.len(), 10);
1068        assert_eq!(depths[0].asks[0].price, Price::from("11657.08"));
1069        assert_eq!(depths[0].asks[0].size, Quantity::from("1.714"));
1070        assert_eq!(depths[0].asks[0].side, OrderSide::Sell);
1071        assert_eq!(depths[0].asks[0].order_id, 0);
1072        assert_eq!(depths[0].bid_counts[0], 1);
1073        assert_eq!(depths[0].ask_counts[0], 1);
1074        // F_SNAPSHOT (32) | F_LAST (128) = 160
1075        assert_eq!(
1076            depths[0].flags,
1077            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1078        );
1079        assert_eq!(depths[0].ts_event, 1598918403696000000);
1080        assert_eq!(depths[0].ts_init, 1598918403810979000);
1081        assert_eq!(depths[0].sequence, 0);
1082    }
1083
1084    #[rstest]
1085    #[case(Some(1), Some(0))] // Explicit precisions
1086    #[case(None, None)] // Inferred precisions
1087    pub fn test_read_quotes(
1088        #[case] price_precision: Option<u8>,
1089        #[case] size_precision: Option<u8>,
1090    ) {
1091        let filepath = get_tardis_huobi_quotes_path();
1092        let quotes =
1093            load_quotes(filepath, price_precision, size_precision, None, Some(100)).unwrap();
1094
1095        assert_eq!(quotes.len(), 10);
1096        assert_eq!(
1097            quotes[0].instrument_id,
1098            InstrumentId::from("BTC-USD.HUOBI_DELIVERY")
1099        );
1100        assert_eq!(quotes[0].bid_price, Price::from("8629.2"));
1101        assert_eq!(quotes[0].bid_size, Quantity::from("806"));
1102        assert_eq!(quotes[0].ask_price, Price::from("8629.3"));
1103        assert_eq!(quotes[0].ask_size, Quantity::from("5494"));
1104        assert_eq!(quotes[0].ts_event, 1588291201099000000);
1105        assert_eq!(quotes[0].ts_init, 1588291201234268000);
1106    }
1107
1108    #[rstest]
1109    fn test_load_options_chain_filters_underlying_and_emits_quote_then_greeks() {
1110        let filepath = get_test_data_path("options_chain.csv");
1111        let data =
1112            load_options_chain(filepath, Some(vec!["btc-".to_string()]), None, None, None).unwrap();
1113
1114        assert_eq!(data.len(), 9);
1115
1116        let Data::Quote(quote) = &data[0] else {
1117            panic!("Expected first data item to be Quote");
1118        };
1119        let Data::OptionGreeks(greeks) = &data[1] else {
1120            panic!("Expected second data item to be OptionGreeks");
1121        };
1122
1123        assert_eq!(
1124            quote.instrument_id,
1125            InstrumentId::from("BTC-9JUN20-9875-P.DERIBIT")
1126        );
1127        assert_eq!(quote.bid_price, Price::from("0.0205"));
1128        assert_eq!(quote.ask_price, Price::from("0.0235"));
1129        assert_eq!(quote.bid_size, Quantity::from("15.1"));
1130        assert_eq!(quote.ask_size, Quantity::from("15.2"));
1131        assert_eq!(quote.bid_price.precision, 4);
1132        assert_eq!(quote.bid_size.precision, 1);
1133
1134        assert_eq!(greeks.instrument_id, quote.instrument_id);
1135        assert_eq!(greeks.greeks.delta, -0.61752);
1136        assert_eq!(greeks.mark_iv, Some(62.89));
1137        assert_eq!(greeks.underlying_price, Some(9756.36));
1138    }
1139
1140    #[rstest]
1141    fn test_load_options_chain_missing_bbo_emits_greeks_only_with_default_greeks() {
1142        let filepath = get_test_data_path("options_chain.csv");
1143        let data = load_options_chain(
1144            filepath,
1145            Some(vec!["BTC-10JUN20".to_string()]),
1146            None,
1147            None,
1148            None,
1149        )
1150        .unwrap();
1151
1152        assert_eq!(data.len(), 1);
1153
1154        let Data::OptionGreeks(greeks) = &data[0] else {
1155            panic!("Expected OptionGreeks, was {:?}", data[0]);
1156        };
1157
1158        assert_eq!(
1159            greeks.instrument_id,
1160            InstrumentId::from("BTC-10JUN20-10000-C.DERIBIT")
1161        );
1162        assert_eq!(greeks.open_interest, None);
1163        assert_eq!(greeks.bid_iv, None);
1164        assert_eq!(greeks.ask_iv, None);
1165        assert_eq!(greeks.greeks.delta, 0.0);
1166        assert_eq!(greeks.greeks.gamma, 0.0);
1167        assert_eq!(greeks.greeks.vega, 0.0);
1168        assert_eq!(greeks.greeks.theta, 0.0);
1169        assert_eq!(greeks.greeks.rho, 0.0);
1170    }
1171
1172    #[rstest]
1173    fn test_load_options_chain_rejects_zero_bbo_size() {
1174        let temp_file = tempfile::NamedTempFile::new().unwrap();
1175        let csv_data = "exchange,symbol,timestamp,local_timestamp,type,strike_price,expiration,open_interest,last_price,bid_price,bid_amount,bid_iv,ask_price,ask_amount,ask_iv,mark_price,mark_iv,underlying_index,underlying_price,delta,gamma,vega,theta,rho
1176deribit,BTC-9JUN20-9875-P,1591574399413000,1591574400196008,put,9875,1591689600000000,0.1,0.0295,0.0205,0,55.91,0.0235,15.2,68.94,0.02210436,62.89,SYN.BTC-9JUN20,9756.36,-0.61752,0.00103,2.24964,-53.05655,-0.22796";
1177        fs::write(temp_file.path(), csv_data).unwrap();
1178
1179        let error = load_options_chain(temp_file.path(), None, None, None, None).unwrap_err();
1180
1181        assert_eq!(error.to_string(), "value was zero");
1182    }
1183
1184    #[rstest]
1185    fn test_load_options_chain_infers_precision_per_instrument() {
1186        let filepath = get_test_data_path("options_chain.csv");
1187        let data =
1188            load_options_chain(filepath, Some(vec!["ETH-".to_string()]), None, None, None).unwrap();
1189
1190        assert_eq!(data.len(), 3);
1191
1192        let Data::Quote(quote) = &data[0] else {
1193            panic!("Expected first data item to be Quote");
1194        };
1195
1196        assert_eq!(
1197            quote.instrument_id,
1198            InstrumentId::from("ETH-9JUN20-250-P.DERIBIT")
1199        );
1200        assert_eq!(quote.bid_price, Price::from("0.12345"));
1201        assert_eq!(quote.ask_price, Price::from("0.12456"));
1202        assert_eq!(quote.bid_size, Quantity::from("0.123456"));
1203        assert_eq!(quote.ask_size, Quantity::from("0.223456"));
1204        assert_eq!(quote.bid_price.precision, 5);
1205        assert_eq!(quote.bid_size.precision, 6);
1206
1207        assert!(matches!(data[1], Data::OptionGreeks(_)));
1208        assert!(matches!(data[2], Data::OptionGreeks(_)));
1209    }
1210
1211    #[rstest]
1212    #[case(Some(1), Some(0))] // Explicit precisions
1213    #[case(None, None)] // Inferred precisions
1214    pub fn test_read_trades(
1215        #[case] price_precision: Option<u8>,
1216        #[case] size_precision: Option<u8>,
1217    ) {
1218        let filepath = get_tardis_bitmex_trades_path();
1219        let trades =
1220            load_trades(filepath, price_precision, size_precision, None, Some(100)).unwrap();
1221
1222        assert_eq!(trades.len(), 10);
1223        assert_eq!(trades[0].instrument_id, InstrumentId::from("XBTUSD.BITMEX"));
1224        assert_eq!(trades[0].price, Price::from("8531.5"));
1225        assert_eq!(trades[0].size, Quantity::from("2152"));
1226        assert_eq!(trades[0].aggressor_side, AggressorSide::Seller);
1227        assert_eq!(
1228            trades[0].trade_id,
1229            TradeId::new("ccc3c1fa-212c-e8b0-1706-9b9c4f3d5ecf")
1230        );
1231        assert_eq!(trades[0].ts_event, 1583020803145000000);
1232        assert_eq!(trades[0].ts_init, 1583020803307160000);
1233    }
1234
1235    #[rstest]
1236    pub fn test_load_trades_derives_id_when_csv_id_empty() {
1237        // Two rows with empty `id` column must both hash deterministically
1238        // to the same TradeId, and a row with differing price must hash differently.
1239        let csv_data = "exchange,symbol,timestamp,local_timestamp,id,side,price,amount
1240binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50000.0,1.0
1241binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50000.0,1.0
1242binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50001.0,1.0";
1243
1244        let temp_file = std::env::temp_dir().join("test_load_trades_empty_id.csv");
1245        std::fs::write(&temp_file, csv_data).unwrap();
1246
1247        let trades = load_trades(&temp_file, Some(2), Some(1), None, None).unwrap();
1248        assert_eq!(trades.len(), 3);
1249
1250        assert_eq!(trades[0].trade_id, trades[1].trade_id);
1251        assert_eq!(trades[0].trade_id.as_str().len(), 16);
1252        assert_ne!(trades[0].trade_id, trades[2].trade_id);
1253
1254        std::fs::remove_file(&temp_file).ok();
1255    }
1256
1257    #[rstest]
1258    pub fn test_load_trades_with_zero_sized_trade() {
1259        // Create test CSV data with one zero-sized trade that should be skipped
1260        let csv_data = "exchange,symbol,timestamp,local_timestamp,id,side,price,amount
1261binance,BTCUSDT,1640995200000000,1640995200100000,trade1,buy,50000.0,1.0
1262binance,BTCUSDT,1640995201000000,1640995201100000,trade2,sell,49999.5,0.0
1263binance,BTCUSDT,1640995202000000,1640995202100000,trade3,buy,50000.12,1.5
1264binance,BTCUSDT,1640995203000000,1640995203100000,trade4,sell,49999.123,3.0";
1265
1266        let temp_file = std::env::temp_dir().join("test_load_trades_zero_size.csv");
1267        std::fs::write(&temp_file, csv_data).unwrap();
1268
1269        let trades = load_trades(
1270            &temp_file,
1271            Some(4),
1272            Some(1),
1273            None,
1274            None, // No limit, load all
1275        )
1276        .unwrap();
1277
1278        // Should have 3 trades (zero-sized trade skipped)
1279        assert_eq!(trades.len(), 3);
1280
1281        // Verify the correct trades were loaded (not the zero-sized one)
1282        assert_eq!(trades[0].size, Quantity::from("1.0"));
1283        assert_eq!(trades[1].size, Quantity::from("1.5"));
1284        assert_eq!(trades[2].size, Quantity::from("3.0"));
1285
1286        // Verify trade IDs to confirm correct trades were loaded
1287        assert_eq!(trades[0].trade_id, TradeId::new("trade1"));
1288        assert_eq!(trades[1].trade_id, TradeId::new("trade3"));
1289        assert_eq!(trades[2].trade_id, TradeId::new("trade4"));
1290
1291        std::fs::remove_file(&temp_file).ok();
1292    }
1293
1294    #[rstest]
1295    pub fn test_load_trades_from_local_file() {
1296        let filepath = get_test_data_path("csv/trades_1.csv");
1297        let trades = load_trades(filepath, Some(1), Some(0), None, None).unwrap();
1298        assert_eq!(trades.len(), 2);
1299        assert_eq!(trades[0].price, Price::from("8531.5"));
1300        assert_eq!(trades[1].size, Quantity::from("1000"));
1301    }
1302
1303    #[rstest]
1304    pub fn test_load_deltas_from_local_file() {
1305        let filepath = get_test_data_path("csv/deltas_1.csv");
1306        let deltas = load_deltas(filepath, Some(1), Some(0), None, None).unwrap();
1307
1308        // 2 data rows + 1 CLEAR delta at start (first row is snapshot)
1309        assert_eq!(deltas.len(), 3);
1310        assert_eq!(deltas[0].action, BookAction::Clear);
1311        assert_eq!(deltas[1].order.price, Price::from("6421.5"));
1312        assert_eq!(deltas[2].order.size, Quantity::from("10000"));
1313    }
1314
1315    #[rstest]
1316    fn test_load_depth10_from_snapshot5_comprehensive() {
1317        let filepath = get_tardis_binance_snapshot5_path();
1318        let depths = load_depth10_from_snapshot5(&filepath, None, None, None, Some(100)).unwrap();
1319
1320        assert_eq!(depths.len(), 10);
1321
1322        let first = &depths[0];
1323        assert_eq!(first.instrument_id.to_string(), "BTCUSDT.BINANCE");
1324        assert_eq!(first.bids.len(), 10);
1325        assert_eq!(first.asks.len(), 10);
1326
1327        // Check all bid levels (first 5 from data, rest empty)
1328        assert_eq!(first.bids[0].price, Price::from("11657.07"));
1329        assert_eq!(first.bids[0].size, Quantity::from("10.896"));
1330        assert_eq!(first.bids[0].side, OrderSide::Buy);
1331
1332        assert_eq!(first.bids[1].price, Price::from("11656.97"));
1333        assert_eq!(first.bids[1].size, Quantity::from("0.2"));
1334        assert_eq!(first.bids[1].side, OrderSide::Buy);
1335
1336        assert_eq!(first.bids[2].price, Price::from("11655.78"));
1337        assert_eq!(first.bids[2].size, Quantity::from("0.2"));
1338        assert_eq!(first.bids[2].side, OrderSide::Buy);
1339
1340        assert_eq!(first.bids[3].price, Price::from("11655.77"));
1341        assert_eq!(first.bids[3].size, Quantity::from("0.98"));
1342        assert_eq!(first.bids[3].side, OrderSide::Buy);
1343
1344        assert_eq!(first.bids[4].price, Price::from("11655.68"));
1345        assert_eq!(first.bids[4].size, Quantity::from("0.111"));
1346        assert_eq!(first.bids[4].side, OrderSide::Buy);
1347
1348        // Empty levels
1349        for i in 5..10 {
1350            assert_eq!(first.bids[i].price.raw, 0);
1351            assert_eq!(first.bids[i].size.raw, 0);
1352            assert_eq!(first.bids[i].side, OrderSide::NoOrderSide);
1353        }
1354
1355        // Check all ask levels (first 5 from data, rest empty)
1356        assert_eq!(first.asks[0].price, Price::from("11657.08"));
1357        assert_eq!(first.asks[0].size, Quantity::from("1.714"));
1358        assert_eq!(first.asks[0].side, OrderSide::Sell);
1359
1360        assert_eq!(first.asks[1].price, Price::from("11657.54"));
1361        assert_eq!(first.asks[1].size, Quantity::from("5.4"));
1362        assert_eq!(first.asks[1].side, OrderSide::Sell);
1363
1364        assert_eq!(first.asks[2].price, Price::from("11657.56"));
1365        assert_eq!(first.asks[2].size, Quantity::from("0.238"));
1366        assert_eq!(first.asks[2].side, OrderSide::Sell);
1367
1368        assert_eq!(first.asks[3].price, Price::from("11657.61"));
1369        assert_eq!(first.asks[3].size, Quantity::from("0.077"));
1370        assert_eq!(first.asks[3].side, OrderSide::Sell);
1371
1372        assert_eq!(first.asks[4].price, Price::from("11657.92"));
1373        assert_eq!(first.asks[4].size, Quantity::from("0.918"));
1374        assert_eq!(first.asks[4].side, OrderSide::Sell);
1375
1376        // Empty levels
1377        for i in 5..10 {
1378            assert_eq!(first.asks[i].price.raw, 0);
1379            assert_eq!(first.asks[i].size.raw, 0);
1380            assert_eq!(first.asks[i].side, OrderSide::NoOrderSide);
1381        }
1382
1383        // Logical checks: bid prices should decrease
1384        for i in 1..5 {
1385            assert!(
1386                first.bids[i].price < first.bids[i - 1].price,
1387                "Bid price at level {} should be less than level {}",
1388                i,
1389                i - 1
1390            );
1391        }
1392
1393        // Logical checks: ask prices should increase
1394        for i in 1..5 {
1395            assert!(
1396                first.asks[i].price > first.asks[i - 1].price,
1397                "Ask price at level {} should be greater than level {}",
1398                i,
1399                i - 1
1400            );
1401        }
1402
1403        // Logical check: spread should be positive
1404        assert!(
1405            first.asks[0].price > first.bids[0].price,
1406            "Best ask should be greater than best bid"
1407        );
1408
1409        // Check counts
1410        for i in 0..5 {
1411            assert_eq!(first.bid_counts[i], 1);
1412            assert_eq!(first.ask_counts[i], 1);
1413        }
1414
1415        for i in 5..10 {
1416            assert_eq!(first.bid_counts[i], 0);
1417            assert_eq!(first.ask_counts[i], 0);
1418        }
1419
1420        // Check metadata - F_SNAPSHOT (32) | F_LAST (128) = 160
1421        assert_eq!(
1422            first.flags,
1423            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1424        );
1425        assert_eq!(first.ts_event.as_u64(), 1598918403696000000);
1426        assert_eq!(first.ts_init.as_u64(), 1598918403810979000);
1427        assert_eq!(first.sequence, 0);
1428    }
1429
1430    #[rstest]
1431    fn test_load_depth10_from_snapshot25_comprehensive() {
1432        let filepath = get_tardis_binance_snapshot25_path();
1433        let depths = load_depth10_from_snapshot25(&filepath, None, None, None, Some(100)).unwrap();
1434
1435        assert_eq!(depths.len(), 10);
1436
1437        let first = &depths[0];
1438        assert_eq!(first.instrument_id.to_string(), "BTCUSDT.BINANCE");
1439        assert_eq!(first.bids.len(), 10);
1440        assert_eq!(first.asks.len(), 10);
1441
1442        // Check all 10 bid levels from snapshot25
1443        let expected_bids = vec![
1444            ("11657.07", "10.896"),
1445            ("11656.97", "0.2"),
1446            ("11655.78", "0.2"),
1447            ("11655.77", "0.98"),
1448            ("11655.68", "0.111"),
1449            ("11655.66", "0.077"),
1450            ("11655.57", "0.34"),
1451            ("11655.48", "0.4"),
1452            ("11655.26", "1.185"),
1453            ("11654.86", "0.195"),
1454        ];
1455
1456        for (i, (price, size)) in expected_bids.iter().enumerate() {
1457            assert_eq!(first.bids[i].price, Price::from(*price));
1458            assert_eq!(first.bids[i].size, Quantity::from(*size));
1459            assert_eq!(first.bids[i].side, OrderSide::Buy);
1460        }
1461
1462        // Check all 10 ask levels from snapshot25
1463        let expected_asks = vec![
1464            ("11657.08", "1.714"),
1465            ("11657.54", "5.4"),
1466            ("11657.56", "0.238"),
1467            ("11657.61", "0.077"),
1468            ("11657.92", "0.918"),
1469            ("11658.09", "1.015"),
1470            ("11658.12", "0.665"),
1471            ("11658.19", "0.583"),
1472            ("11658.28", "0.255"),
1473            ("11658.29", "0.656"),
1474        ];
1475
1476        for (i, (price, size)) in expected_asks.iter().enumerate() {
1477            assert_eq!(first.asks[i].price, Price::from(*price));
1478            assert_eq!(first.asks[i].size, Quantity::from(*size));
1479            assert_eq!(first.asks[i].side, OrderSide::Sell);
1480        }
1481
1482        // Logical checks: bid prices should strictly decrease
1483        for i in 1..10 {
1484            assert!(
1485                first.bids[i].price < first.bids[i - 1].price,
1486                "Bid price at level {} ({}) should be less than level {} ({})",
1487                i,
1488                first.bids[i].price,
1489                i - 1,
1490                first.bids[i - 1].price
1491            );
1492        }
1493
1494        // Logical checks: ask prices should strictly increase
1495        for i in 1..10 {
1496            assert!(
1497                first.asks[i].price > first.asks[i - 1].price,
1498                "Ask price at level {} ({}) should be greater than level {} ({})",
1499                i,
1500                first.asks[i].price,
1501                i - 1,
1502                first.asks[i - 1].price
1503            );
1504        }
1505
1506        // Logical check: spread should be positive
1507        assert!(
1508            first.asks[0].price > first.bids[0].price,
1509            "Best ask ({}) should be greater than best bid ({})",
1510            first.asks[0].price,
1511            first.bids[0].price
1512        );
1513
1514        // Check counts (all should be 1 for snapshot data)
1515        for i in 0..10 {
1516            assert_eq!(first.bid_counts[i], 1);
1517            assert_eq!(first.ask_counts[i], 1);
1518        }
1519
1520        // Check metadata - F_SNAPSHOT (32) | F_LAST (128) = 160
1521        assert_eq!(
1522            first.flags,
1523            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1524        );
1525        assert_eq!(first.ts_event.as_u64(), 1598918403696000000);
1526        assert_eq!(first.ts_init.as_u64(), 1598918403810979000);
1527        assert_eq!(first.sequence, 0);
1528    }
1529
1530    #[rstest]
1531    fn test_snapshot_csv_field_order_interleaved() {
1532        // This test verifies that the CSV structs correctly handle the interleaved
1533        // asks/bids field ordering from Tardis CSV files
1534
1535        let csv_data = "exchange,symbol,timestamp,local_timestamp,\
1536asks[0].price,asks[0].amount,bids[0].price,bids[0].amount,\
1537asks[1].price,asks[1].amount,bids[1].price,bids[1].amount,\
1538asks[2].price,asks[2].amount,bids[2].price,bids[2].amount,\
1539asks[3].price,asks[3].amount,bids[3].price,bids[3].amount,\
1540asks[4].price,asks[4].amount,bids[4].price,bids[4].amount
1541binance-futures,BTCUSDT,1000000,2000000,\
1542100.5,1.0,100.4,2.0,\
1543100.6,1.1,100.3,2.1,\
1544100.7,1.2,100.2,2.2,\
1545100.8,1.3,100.1,2.3,\
1546100.9,1.4,100.0,2.4";
1547
1548        let temp_file = std::env::temp_dir().join("test_interleaved_snapshot5.csv");
1549        std::fs::write(&temp_file, csv_data).unwrap();
1550
1551        let depths = load_depth10_from_snapshot5(&temp_file, None, None, None, Some(1)).unwrap();
1552        assert_eq!(depths.len(), 1);
1553
1554        let depth = &depths[0];
1555
1556        // Verify bids are correctly parsed (should be decreasing)
1557        assert_eq!(depth.bids[0].price, Price::from("100.4"));
1558        assert_eq!(depth.bids[1].price, Price::from("100.3"));
1559        assert_eq!(depth.bids[2].price, Price::from("100.2"));
1560        assert_eq!(depth.bids[3].price, Price::from("100.1"));
1561        assert_eq!(depth.bids[4].price, Price::from("100.0"));
1562
1563        // Verify asks are correctly parsed (should be increasing)
1564        assert_eq!(depth.asks[0].price, Price::from("100.5"));
1565        assert_eq!(depth.asks[1].price, Price::from("100.6"));
1566        assert_eq!(depth.asks[2].price, Price::from("100.7"));
1567        assert_eq!(depth.asks[3].price, Price::from("100.8"));
1568        assert_eq!(depth.asks[4].price, Price::from("100.9"));
1569
1570        // Verify sizes
1571        assert_eq!(depth.bids[0].size, Quantity::from("2.0"));
1572        assert_eq!(depth.asks[0].size, Quantity::from("1.0"));
1573
1574        std::fs::remove_file(temp_file).unwrap();
1575    }
1576
1577    #[rstest]
1578    fn test_load_deltas_limit_includes_clear_deltas() {
1579        // Test that limit counts total emitted deltas (including CLEARs)
1580        // When limit=5, we should get exactly 5 deltas: 1 CLEAR + 4 data deltas
1581        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
1582binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
1583binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0
1584binance-futures,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.0,0.5
1585binance-futures,BTCUSDT,1640995202000000,1640995202100000,false,ask,50002.0,1.5
1586binance-futures,BTCUSDT,1640995203000000,1640995203100000,false,bid,49998.0,0.5
1587binance-futures,BTCUSDT,1640995204000000,1640995204100000,false,ask,50003.0,2.0
1588binance-futures,BTCUSDT,1640995205000000,1640995205100000,false,bid,49997.0,0.5";
1589
1590        let temp_file = std::env::temp_dir().join("test_load_deltas_limit.csv");
1591        std::fs::write(&temp_file, csv_data).unwrap();
1592
1593        // Load with limit=5 (should emit exactly 5 deltas including CLEAR)
1594        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, Some(5)).unwrap();
1595
1596        // Should have exactly 5 deltas: 1 CLEAR + 4 data deltas
1597        assert_eq!(deltas.len(), 5);
1598        assert_eq!(deltas[0].action, BookAction::Clear);
1599        assert_eq!(deltas[1].action, BookAction::Add);
1600        assert_eq!(deltas[2].action, BookAction::Add);
1601        assert_eq!(deltas[3].action, BookAction::Update);
1602        assert_eq!(deltas[4].action, BookAction::Update);
1603
1604        // Verify the last delta is from the 4th CSV record (49999.0 bid)
1605        assert_eq!(deltas[3].order.price, parse_price(49999.0, 1));
1606
1607        std::fs::remove_file(&temp_file).ok();
1608    }
1609
1610    #[rstest]
1611    fn test_load_deltas_limit_stops_at_clear() {
1612        // Test that limit=1 with snapshot data returns only the CLEAR delta
1613        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
1614binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
1615binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0";
1616
1617        let temp_file = std::env::temp_dir().join("test_load_deltas_limit_stops_at_clear.csv");
1618        std::fs::write(&temp_file, csv_data).unwrap();
1619
1620        // Load with limit=1 should only get the CLEAR delta
1621        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, Some(1)).unwrap();
1622
1623        assert_eq!(deltas.len(), 1);
1624        assert_eq!(deltas[0].action, BookAction::Clear);
1625
1626        std::fs::remove_file(&temp_file).ok();
1627    }
1628
1629    #[rstest]
1630    fn test_load_deltas_with_consecutive_snapshots_inserts_clear() {
1631        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
1632hyperliquid,BTC,1640995200000000,1640995200100000,true,bid,50000.0,1.0
1633hyperliquid,BTC,1640995200000000,1640995200100000,true,ask,50001.0,2.0
1634hyperliquid,BTC,1640995201000000,1640995201100000,true,bid,49990.0,3.0
1635hyperliquid,BTC,1640995201000000,1640995201100000,true,ask,49991.0,4.0";
1636
1637        let temp_file = std::env::temp_dir().join("test_load_deltas_consecutive_snapshots.csv");
1638        std::fs::write(&temp_file, csv_data).unwrap();
1639
1640        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, None).unwrap();
1641        let clear_count = deltas
1642            .iter()
1643            .filter(|d| d.action == BookAction::Clear)
1644            .count();
1645
1646        assert_eq!(clear_count, 2);
1647        assert_eq!(deltas[0].action, BookAction::Clear);
1648        assert_eq!(deltas[3].action, BookAction::Clear);
1649        assert_eq!(
1650            deltas[2].flags & RecordFlag::F_LAST as u8,
1651            RecordFlag::F_LAST as u8
1652        );
1653        assert_eq!(deltas[3].flags & RecordFlag::F_LAST as u8, 0);
1654
1655        std::fs::remove_file(&temp_file).ok();
1656    }
1657
1658    #[rstest]
1659    fn test_load_deltas_limit_with_mid_day_snapshot() {
1660        // Test limit behavior when there's a mid-day snapshot
1661        // The limit counts total emitted deltas including CLEARs
1662        let filepath = get_test_data_path("csv/deltas_with_snapshot.csv");
1663        let deltas = load_deltas(filepath, Some(1), Some(1), None, Some(5)).unwrap();
1664
1665        // With limit=5, we get exactly 5 deltas
1666        // First snapshot inserts CLEAR, then we get 4 more data deltas
1667        assert_eq!(deltas.len(), 5);
1668        assert_eq!(deltas[0].action, BookAction::Clear);
1669    }
1670
1671    // Curates the large Tardis Deribit CSV.gz into NautilusTrader Parquet format.
1672    // Run manually: `cargo test -p nautilus-tardis test_curate_deribit_deltas -- --ignored --nocapture`
1673    #[rstest]
1674    #[ignore = "one-time dataset curation, not for routine CI"]
1675    fn test_curate_deribit_deltas() {
1676        let csv_path = get_test_data_root()
1677            .join("large")
1678            .join("tardis_deribit_incremental_book_L2_2020-04-01_BTC-PERPETUAL.csv.gz");
1679
1680        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1681        let parquet_path = "/tmp/tardis_BTC-PERPETUAL.DERIBIT_2020-04-01_deltas.parquet";
1682
1683        println!("Loading deltas from {}", csv_path.display());
1684        let deltas = load_deltas(&csv_path, None, None, Some(instrument_id), None).unwrap();
1685        let count = deltas.len();
1686        println!("Loaded {count} deltas");
1687
1688        let sample = deltas
1689            .iter()
1690            .find(|d| d.order.price.precision > 0)
1691            .expect("Should have at least one non-CLEAR delta");
1692        let price_precision = sample.order.price.precision;
1693        let size_precision = sample.order.size.precision;
1694        println!("Precision: price={price_precision}, size={size_precision}");
1695
1696        // Write in chunks to avoid stack overflow on large batches
1697        let metadata =
1698            OrderBookDelta::get_metadata(&instrument_id, price_precision, size_precision);
1699        let schema = OrderBookDelta::get_schema(Some(metadata.clone()));
1700
1701        println!("Writing Parquet to {parquet_path}");
1702        let file = File::create(parquet_path).unwrap();
1703        let zstd_level = parquet::basic::ZstdLevel::try_new(3).unwrap();
1704        let props = WriterProperties::builder()
1705            .set_compression(parquet::basic::Compression::ZSTD(zstd_level))
1706            .set_max_row_group_row_count(Some(1_000_000))
1707            .build();
1708        let mut writer = ArrowWriter::try_new(file, Arc::new(schema), Some(props)).unwrap();
1709
1710        let chunk_size = 1_000_000;
1711        for (i, chunk) in deltas.chunks(chunk_size).enumerate() {
1712            println!("  Encoding chunk {} ({} records)...", i + 1, chunk.len());
1713            let batch = OrderBookDelta::encode_batch(&metadata, chunk).unwrap();
1714            writer.write(&batch).unwrap();
1715        }
1716        writer.close().unwrap();
1717
1718        let file_size = fs::metadata(parquet_path).unwrap().len();
1719        println!("\n=== CURATION COMPLETE ===");
1720        println!("Records: {count}");
1721        println!("Price precision: {price_precision}");
1722        println!("Size precision: {size_precision}");
1723        println!(
1724            "File size: {} bytes ({:.1} MB)",
1725            file_size,
1726            file_size as f64 / 1_048_576.0
1727        );
1728        println!("Output: {parquet_path}");
1729        println!("\nNext steps:");
1730        println!("  sha256sum {parquet_path}");
1731    }
1732}
nautilus_tardis/csv/load.rs

nautilus_tardis/csv/
load.rs
