Skip to main content

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_init: Option<UnixNanos> = None;
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 message must also reset the previous snapshot state.
206        let starts_new_snapshot =
207            data.is_snapshot && (!last_is_snapshot || last_ts_init != Some(ts_init));
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_init != Some(ts_init)
214                && let Some(last_delta) = deltas.last_mut()
215            {
216                last_delta.flags = RecordFlag::F_LAST as u8;
217            }
218            last_ts_init = Some(ts_init);
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_init = delta.ts_init;
245        if last_ts_init != Some(ts_init)
246            && let Some(last_delta) = deltas.last_mut()
247        {
248            last_delta.flags = RecordFlag::F_LAST as u8;
249        }
250
251        last_ts_init = Some(ts_init);
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` and `funding_timestamp` fields from derivative ticker
763/// 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, OrderSide},
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    use rust_decimal_macros::dec;
894
895    use super::*;
896    use crate::common::{parse::parse_price, testing::get_test_data_path};
897
898    #[rstest]
899    #[case(0.0, 0)]
900    #[case(42.0, 0)]
901    #[case(0.1, 1)]
902    #[case(0.25, 2)]
903    #[case(123.0001, 4)]
904    #[case(-42.987654321,       9)]
905    #[case(1.234_567_890_123, 12)]
906    fn test_infer_precision(#[case] input: f64, #[case] expected: u8) {
907        assert_eq!(infer_precision(input), expected);
908    }
909
910    #[rstest]
911    pub fn test_dynamic_precision_inference() {
912        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
913binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50000.0,1.0
914binance-futures,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.5,2.0
915binance-futures,BTCUSDT,1640995202000000,1640995202100000,false,ask,50000.12,1.5
916binance-futures,BTCUSDT,1640995203000000,1640995203100000,false,bid,49999.123,3.0
917binance-futures,BTCUSDT,1640995204000000,1640995204100000,false,ask,50000.1234,0.5";
918
919        let temp_file = std::env::temp_dir().join("test_dynamic_precision.csv");
920        std::fs::write(&temp_file, csv_data).unwrap();
921
922        let deltas = load_deltas(&temp_file, None, None, None, None).unwrap();
923
924        // 5 data rows + 1 CLEAR delta at start (first row is snapshot)
925        assert_eq!(deltas.len(), 6);
926
927        // Skip the CLEAR delta at index 0
928        for (i, delta) in deltas.iter().skip(1).enumerate() {
929            assert_eq!(
930                delta.order.price.precision, 4,
931                "Price precision should be 4 for delta {i}",
932            );
933            assert_eq!(
934                delta.order.size.precision, 1,
935                "Size precision should be 1 for delta {i}",
936            );
937        }
938
939        // Test exact values to ensure retroactive precision updates work correctly
940        // Index 0 is CLEAR, data starts at index 1
941        assert_eq!(deltas[0].action, BookAction::Clear);
942
943        assert_eq!(deltas[1].order.price, parse_price(50000.0, 4));
944        assert_eq!(deltas[1].order.size, Quantity::new(1.0, 1));
945
946        assert_eq!(deltas[2].order.price, parse_price(49999.5, 4));
947        assert_eq!(deltas[2].order.size, Quantity::new(2.0, 1));
948
949        assert_eq!(deltas[3].order.price, parse_price(50000.12, 4));
950        assert_eq!(deltas[3].order.size, Quantity::new(1.5, 1));
951
952        assert_eq!(deltas[4].order.price, parse_price(49999.123, 4));
953        assert_eq!(deltas[4].order.size, Quantity::new(3.0, 1));
954
955        assert_eq!(deltas[5].order.price, parse_price(50000.1234, 4));
956        assert_eq!(deltas[5].order.size, Quantity::new(0.5, 1));
957
958        assert_eq!(
959            deltas[1].order.price.precision,
960            deltas[5].order.price.precision
961        );
962        assert_eq!(
963            deltas[1].order.size.precision,
964            deltas[3].order.size.precision
965        );
966
967        std::fs::remove_file(&temp_file).ok();
968    }
969
970    #[rstest]
971    #[case(Some(1), Some(0))] // Explicit precisions
972    #[case(None, None)] // Inferred precisions
973    pub fn test_read_deltas(
974        #[case] price_precision: Option<u8>,
975        #[case] size_precision: Option<u8>,
976    ) {
977        let filepath = get_tardis_deribit_book_l2_path();
978        let deltas =
979            load_deltas(filepath, price_precision, size_precision, None, Some(100)).unwrap();
980
981        // 15 data rows + 1 CLEAR delta at start (first row is snapshot)
982        assert_eq!(deltas.len(), 16);
983
984        // Index 0 is CLEAR delta
985        assert_eq!(deltas[0].action, BookAction::Clear);
986
987        // Index 1 is first data delta
988        assert_eq!(
989            deltas[1].instrument_id,
990            InstrumentId::from("BTC-PERPETUAL.DERIBIT")
991        );
992        assert_eq!(deltas[1].action, BookAction::Add);
993        assert_eq!(deltas[1].order.side, OrderSide::Sell.into());
994        assert_eq!(deltas[1].order.price, Price::from("6421.5"));
995        assert_eq!(deltas[1].order.size, Quantity::from("18640"));
996        assert_eq!(deltas[1].flags, 0);
997        assert_eq!(deltas[1].sequence, 0);
998        assert_eq!(deltas[1].ts_event, 1585699200245000000);
999        assert_eq!(deltas[1].ts_init, 1585699200355684000);
1000    }
1001
1002    #[rstest]
1003    #[case(Some(2), Some(3))] // Explicit precisions
1004    #[case(None, None)] // Inferred precisions
1005    pub fn test_read_depth10s_from_snapshot5(
1006        #[case] price_precision: Option<u8>,
1007        #[case] size_precision: Option<u8>,
1008    ) {
1009        let filepath = get_tardis_binance_snapshot5_path();
1010        let depths =
1011            load_depth10_from_snapshot5(filepath, price_precision, size_precision, None, Some(100))
1012                .unwrap();
1013
1014        assert_eq!(depths.len(), 10);
1015        assert_eq!(
1016            depths[0].instrument_id,
1017            InstrumentId::from("BTCUSDT.BINANCE")
1018        );
1019        assert_eq!(depths[0].bids.len(), 10);
1020        assert_eq!(depths[0].bids[0].price, Price::from("11657.07"));
1021        assert_eq!(depths[0].bids[0].size, Quantity::from("10.896"));
1022        assert_eq!(depths[0].bids[0].side, OrderSide::Buy.into());
1023        assert_eq!(depths[0].bids[0].order_id, 0);
1024        assert_eq!(depths[0].asks.len(), 10);
1025        assert_eq!(depths[0].asks[0].price, Price::from("11657.08"));
1026        assert_eq!(depths[0].asks[0].size, Quantity::from("1.714"));
1027        assert_eq!(depths[0].asks[0].side, OrderSide::Sell.into());
1028        assert_eq!(depths[0].asks[0].order_id, 0);
1029        assert_eq!(depths[0].bid_counts[0], 1);
1030        assert_eq!(depths[0].ask_counts[0], 1);
1031        // F_SNAPSHOT (32) | F_LAST (128) = 160
1032        assert_eq!(
1033            depths[0].flags,
1034            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1035        );
1036        assert_eq!(depths[0].ts_event, 1598918403696000000);
1037        assert_eq!(depths[0].ts_init, 1598918403810979000);
1038        assert_eq!(depths[0].sequence, 0);
1039    }
1040
1041    #[rstest]
1042    #[case(Some(2), Some(3))] // Explicit precisions
1043    #[case(None, None)] // Inferred precisions
1044    pub fn test_read_depth10s_from_snapshot25(
1045        #[case] price_precision: Option<u8>,
1046        #[case] size_precision: Option<u8>,
1047    ) {
1048        let filepath = get_tardis_binance_snapshot25_path();
1049        let depths = load_depth10_from_snapshot25(
1050            filepath,
1051            price_precision,
1052            size_precision,
1053            None,
1054            Some(100),
1055        )
1056        .unwrap();
1057
1058        assert_eq!(depths.len(), 10);
1059        assert_eq!(
1060            depths[0].instrument_id,
1061            InstrumentId::from("BTCUSDT.BINANCE")
1062        );
1063        assert_eq!(depths[0].bids.len(), 10);
1064        assert_eq!(depths[0].bids[0].price, Price::from("11657.07"));
1065        assert_eq!(depths[0].bids[0].size, Quantity::from("10.896"));
1066        assert_eq!(depths[0].bids[0].side, OrderSide::Buy.into());
1067        assert_eq!(depths[0].bids[0].order_id, 0);
1068        assert_eq!(depths[0].asks.len(), 10);
1069        assert_eq!(depths[0].asks[0].price, Price::from("11657.08"));
1070        assert_eq!(depths[0].asks[0].size, Quantity::from("1.714"));
1071        assert_eq!(depths[0].asks[0].side, OrderSide::Sell.into());
1072        assert_eq!(depths[0].asks[0].order_id, 0);
1073        assert_eq!(depths[0].bid_counts[0], 1);
1074        assert_eq!(depths[0].ask_counts[0], 1);
1075        // F_SNAPSHOT (32) | F_LAST (128) = 160
1076        assert_eq!(
1077            depths[0].flags,
1078            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1079        );
1080        assert_eq!(depths[0].ts_event, 1598918403696000000);
1081        assert_eq!(depths[0].ts_init, 1598918403810979000);
1082        assert_eq!(depths[0].sequence, 0);
1083    }
1084
1085    #[rstest]
1086    #[case(Some(1), Some(0))] // Explicit precisions
1087    #[case(None, None)] // Inferred precisions
1088    pub fn test_read_quotes(
1089        #[case] price_precision: Option<u8>,
1090        #[case] size_precision: Option<u8>,
1091    ) {
1092        let filepath = get_tardis_huobi_quotes_path();
1093        let quotes =
1094            load_quotes(filepath, price_precision, size_precision, None, Some(100)).unwrap();
1095
1096        assert_eq!(quotes.len(), 10);
1097        assert_eq!(
1098            quotes[0].instrument_id,
1099            InstrumentId::from("BTC-USD.HUOBI_DELIVERY")
1100        );
1101        assert_eq!(quotes[0].bid_price, Price::from("8629.2"));
1102        assert_eq!(quotes[0].bid_size, Quantity::from("806"));
1103        assert_eq!(quotes[0].ask_price, Price::from("8629.3"));
1104        assert_eq!(quotes[0].ask_size, Quantity::from("5494"));
1105        assert_eq!(quotes[0].ts_event, 1588291201099000000);
1106        assert_eq!(quotes[0].ts_init, 1588291201234268000);
1107    }
1108
1109    #[rstest]
1110    fn test_load_options_chain_filters_underlying_and_emits_quote_then_greeks() {
1111        let filepath = get_test_data_path("options_chain.csv");
1112        let data =
1113            load_options_chain(filepath, Some(vec!["btc-".to_string()]), None, None, None).unwrap();
1114
1115        assert_eq!(data.len(), 9);
1116
1117        let Data::Quote(quote) = &data[0] else {
1118            panic!("Expected first data item to be Quote");
1119        };
1120        let Data::OptionGreeks(greeks) = &data[1] else {
1121            panic!("Expected second data item to be OptionGreeks");
1122        };
1123
1124        assert_eq!(
1125            quote.instrument_id,
1126            InstrumentId::from("BTC-9JUN20-9875-P.DERIBIT")
1127        );
1128        assert_eq!(quote.bid_price, Price::from("0.0205"));
1129        assert_eq!(quote.ask_price, Price::from("0.0235"));
1130        assert_eq!(quote.bid_size, Quantity::from("15.1"));
1131        assert_eq!(quote.ask_size, Quantity::from("15.2"));
1132        assert_eq!(quote.bid_price.precision, 4);
1133        assert_eq!(quote.bid_size.precision, 1);
1134
1135        assert_eq!(greeks.instrument_id, quote.instrument_id);
1136        assert_eq!(greeks.greeks.delta, -0.61752);
1137        assert_eq!(greeks.mark_iv, Some(62.89));
1138        assert_eq!(greeks.underlying_price, Some(9756.36));
1139    }
1140
1141    #[rstest]
1142    fn test_load_options_chain_missing_bbo_emits_greeks_only_with_default_greeks() {
1143        let filepath = get_test_data_path("options_chain.csv");
1144        let data = load_options_chain(
1145            filepath,
1146            Some(vec!["BTC-10JUN20".to_string()]),
1147            None,
1148            None,
1149            None,
1150        )
1151        .unwrap();
1152
1153        assert_eq!(data.len(), 1);
1154
1155        let Data::OptionGreeks(greeks) = &data[0] else {
1156            panic!("Expected OptionGreeks, was {:?}", data[0]);
1157        };
1158
1159        assert_eq!(
1160            greeks.instrument_id,
1161            InstrumentId::from("BTC-10JUN20-10000-C.DERIBIT")
1162        );
1163        assert_eq!(greeks.open_interest, None);
1164        assert_eq!(greeks.bid_iv, None);
1165        assert_eq!(greeks.ask_iv, None);
1166        assert_eq!(greeks.greeks.delta, 0.0);
1167        assert_eq!(greeks.greeks.gamma, 0.0);
1168        assert_eq!(greeks.greeks.vega, 0.0);
1169        assert_eq!(greeks.greeks.theta, 0.0);
1170        assert_eq!(greeks.greeks.rho, 0.0);
1171    }
1172
1173    #[rstest]
1174    fn test_load_options_chain_rejects_zero_bbo_size() {
1175        let temp_file = tempfile::NamedTempFile::new().unwrap();
1176        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
1177deribit,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";
1178        fs::write(temp_file.path(), csv_data).unwrap();
1179
1180        let error = load_options_chain(temp_file.path(), None, None, None, None).unwrap_err();
1181
1182        assert_eq!(error.to_string(), "value was zero");
1183    }
1184
1185    #[rstest]
1186    fn test_load_options_chain_infers_precision_per_instrument() {
1187        let filepath = get_test_data_path("options_chain.csv");
1188        let data =
1189            load_options_chain(filepath, Some(vec!["ETH-".to_string()]), None, None, None).unwrap();
1190
1191        assert_eq!(data.len(), 3);
1192
1193        let Data::Quote(quote) = &data[0] else {
1194            panic!("Expected first data item to be Quote");
1195        };
1196
1197        assert_eq!(
1198            quote.instrument_id,
1199            InstrumentId::from("ETH-9JUN20-250-P.DERIBIT")
1200        );
1201        assert_eq!(quote.bid_price, Price::from("0.12345"));
1202        assert_eq!(quote.ask_price, Price::from("0.12456"));
1203        assert_eq!(quote.bid_size, Quantity::from("0.123456"));
1204        assert_eq!(quote.ask_size, Quantity::from("0.223456"));
1205        assert_eq!(quote.bid_price.precision, 5);
1206        assert_eq!(quote.bid_size.precision, 6);
1207
1208        assert!(matches!(data[1], Data::OptionGreeks(_)));
1209        assert!(matches!(data[2], Data::OptionGreeks(_)));
1210    }
1211
1212    #[rstest]
1213    #[case(Some(1), Some(0))] // Explicit precisions
1214    #[case(None, None)] // Inferred precisions
1215    pub fn test_read_trades(
1216        #[case] price_precision: Option<u8>,
1217        #[case] size_precision: Option<u8>,
1218    ) {
1219        let filepath = get_tardis_bitmex_trades_path();
1220        let trades =
1221            load_trades(filepath, price_precision, size_precision, None, Some(100)).unwrap();
1222
1223        assert_eq!(trades.len(), 10);
1224        assert_eq!(trades[0].instrument_id, InstrumentId::from("XBTUSD.BITMEX"));
1225        assert_eq!(trades[0].price, Price::from("8531.5"));
1226        assert_eq!(trades[0].size, Quantity::from("2152"));
1227        assert_eq!(trades[0].aggressor_side, AggressorSide::Sell);
1228        assert_eq!(
1229            trades[0].trade_id,
1230            TradeId::new("ccc3c1fa-212c-e8b0-1706-9b9c4f3d5ecf")
1231        );
1232        assert_eq!(trades[0].ts_event, 1583020803145000000);
1233        assert_eq!(trades[0].ts_init, 1583020803307160000);
1234    }
1235
1236    #[rstest]
1237    pub fn test_load_trades_derives_id_when_csv_id_empty() {
1238        // Two rows with empty `id` column must both hash deterministically
1239        // to the same TradeId, and a row with differing price must hash differently.
1240        let csv_data = "exchange,symbol,timestamp,local_timestamp,id,side,price,amount
1241binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50000.0,1.0
1242binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50000.0,1.0
1243binance,BTCUSDT,1640995200000000,1640995200100000,,buy,50001.0,1.0";
1244
1245        let temp_file = std::env::temp_dir().join("test_load_trades_empty_id.csv");
1246        std::fs::write(&temp_file, csv_data).unwrap();
1247
1248        let trades = load_trades(&temp_file, Some(2), Some(1), None, None).unwrap();
1249        assert_eq!(trades.len(), 3);
1250
1251        assert_eq!(trades[0].trade_id, trades[1].trade_id);
1252        assert_eq!(trades[0].trade_id.as_str().len(), 16);
1253        assert_ne!(trades[0].trade_id, trades[2].trade_id);
1254
1255        std::fs::remove_file(&temp_file).ok();
1256    }
1257
1258    #[rstest]
1259    pub fn test_load_trades_with_zero_sized_trade() {
1260        // Create test CSV data with one zero-sized trade that should be skipped
1261        let csv_data = "exchange,symbol,timestamp,local_timestamp,id,side,price,amount
1262binance,BTCUSDT,1640995200000000,1640995200100000,trade1,buy,50000.0,1.0
1263binance,BTCUSDT,1640995201000000,1640995201100000,trade2,sell,49999.5,0.0
1264binance,BTCUSDT,1640995202000000,1640995202100000,trade3,buy,50000.12,1.5
1265binance,BTCUSDT,1640995203000000,1640995203100000,trade4,sell,49999.123,3.0";
1266
1267        let temp_file = std::env::temp_dir().join("test_load_trades_zero_size.csv");
1268        std::fs::write(&temp_file, csv_data).unwrap();
1269
1270        let trades = load_trades(
1271            &temp_file,
1272            Some(4),
1273            Some(1),
1274            None,
1275            None, // No limit, load all
1276        )
1277        .unwrap();
1278
1279        // Should have 3 trades (zero-sized trade skipped)
1280        assert_eq!(trades.len(), 3);
1281
1282        // Verify the correct trades were loaded (not the zero-sized one)
1283        assert_eq!(trades[0].size, Quantity::from("1.0"));
1284        assert_eq!(trades[1].size, Quantity::from("1.5"));
1285        assert_eq!(trades[2].size, Quantity::from("3.0"));
1286
1287        // Verify trade IDs to confirm correct trades were loaded
1288        assert_eq!(trades[0].trade_id, TradeId::new("trade1"));
1289        assert_eq!(trades[1].trade_id, TradeId::new("trade3"));
1290        assert_eq!(trades[2].trade_id, TradeId::new("trade4"));
1291
1292        std::fs::remove_file(&temp_file).ok();
1293    }
1294
1295    #[rstest]
1296    pub fn test_load_trades_from_local_file() {
1297        let filepath = get_test_data_path("csv/trades_1.csv");
1298        let trades = load_trades(filepath, Some(1), Some(0), None, None).unwrap();
1299        assert_eq!(trades.len(), 2);
1300        assert_eq!(trades[0].price, Price::from("8531.5"));
1301        assert_eq!(trades[1].size, Quantity::from("1000"));
1302    }
1303
1304    #[rstest]
1305    pub fn test_load_deltas_from_local_file() {
1306        let filepath = get_test_data_path("csv/deltas_1.csv");
1307        let deltas = load_deltas(filepath, Some(1), Some(0), None, None).unwrap();
1308
1309        // 2 data rows + 1 CLEAR delta at start (first row is snapshot)
1310        assert_eq!(deltas.len(), 3);
1311        assert_eq!(deltas[0].action, BookAction::Clear);
1312        assert_eq!(deltas[1].order.price, Price::from("6421.5"));
1313        assert_eq!(deltas[2].order.size, Quantity::from("10000"));
1314    }
1315
1316    #[rstest]
1317    fn test_load_deltas_groups_messages_by_local_timestamp() {
1318        let filepath = get_test_data_path("csv/deltas_message_boundaries.csv");
1319        let deltas = load_deltas(filepath, Some(1), Some(1), None, None).unwrap();
1320
1321        assert_eq!(deltas.len(), 4);
1322        assert_eq!(
1323            deltas.iter().map(|delta| delta.flags).collect::<Vec<_>>(),
1324            vec![0, RecordFlag::F_LAST as u8, 0, RecordFlag::F_LAST as u8]
1325        );
1326        assert_eq!(
1327            deltas
1328                .iter()
1329                .map(|delta| delta.ts_event)
1330                .collect::<Vec<_>>(),
1331            vec![
1332                UnixNanos::from(1_000_000),
1333                UnixNanos::from(1_000_000),
1334                UnixNanos::from(1_000_000),
1335                UnixNanos::from(1_010_000),
1336            ]
1337        );
1338        assert_eq!(
1339            deltas.iter().map(|delta| delta.ts_init).collect::<Vec<_>>(),
1340            vec![
1341                UnixNanos::from(2_000_000),
1342                UnixNanos::from(2_000_000),
1343                UnixNanos::from(2_010_000),
1344                UnixNanos::from(2_010_000),
1345            ]
1346        );
1347        assert_eq!(deltas[0].order.side, Some(OrderSide::Buy));
1348        assert_eq!(deltas[0].order.price, Price::from("100.0"));
1349        assert_eq!(deltas[0].order.size, Quantity::from("1.0"));
1350        assert_eq!(deltas[1].order.side, Some(OrderSide::Sell));
1351        assert_eq!(deltas[1].order.price, Price::from("101.0"));
1352        assert_eq!(deltas[1].order.size, Quantity::from("2.0"));
1353        assert_eq!(deltas[2].order.side, Some(OrderSide::Buy));
1354        assert_eq!(deltas[2].order.price, Price::from("99.0"));
1355        assert_eq!(deltas[2].order.size, Quantity::from("3.0"));
1356        assert_eq!(deltas[3].order.side, Some(OrderSide::Sell));
1357        assert_eq!(deltas[3].order.price, Price::from("102.0"));
1358        assert_eq!(deltas[3].order.size, Quantity::from("4.0"));
1359    }
1360
1361    #[rstest]
1362    fn test_load_funding_rates_okex_xperp() {
1363        let filepath = get_test_data_path("csv/okex_futures_xperp_derivative_ticker.csv");
1364        let funding_rates = load_funding_rates(filepath, None, None).unwrap();
1365
1366        let instrument_id = InstrumentId::from("BTC-USD_UM_XPERP-310404.OKEX");
1367
1368        assert_eq!(funding_rates.len(), 8);
1369        assert!(
1370            funding_rates
1371                .iter()
1372                .all(|f| f.instrument_id == instrument_id)
1373        );
1374
1375        // OKX X-Perps publish no predicted rate, so the funding timestamp is the only forward
1376        // reference Tardis carries, and the interval is not representable in a derivative ticker
1377        assert!(
1378            funding_rates
1379                .iter()
1380                .all(|f| f.next_funding_ns.is_some() && f.interval.is_none())
1381        );
1382
1383        let first = &funding_rates[0];
1384        let rolled = &funding_rates[3];
1385
1386        assert_eq!(first.rate, dec!(-0.0003972900658902));
1387        assert_eq!(
1388            first.next_funding_ns,
1389            Some(UnixNanos::from(1_786_320_000_000_000_000))
1390        );
1391        assert_eq!(first.ts_event, UnixNanos::from(1_786_320_006_952_000_000));
1392        assert_eq!(first.ts_init, UnixNanos::from(1_786_320_006_971_532_000));
1393
1394        assert_eq!(rolled.rate, dec!(-0.0003962534258591));
1395        assert_eq!(
1396            rolled.next_funding_ns,
1397            Some(UnixNanos::from(1_786_348_800_000_000_000))
1398        );
1399        assert_eq!(rolled.ts_event, UnixNanos::from(1_786_320_007_369_000_000));
1400        assert_eq!(rolled.ts_init, UnixNanos::from(1_786_320_007_402_423_000));
1401    }
1402
1403    #[rstest]
1404    fn test_load_funding_rates_without_funding_timestamp() {
1405        let filepath = get_test_data_path("csv/deribit_derivative_ticker.csv");
1406        let funding_rates = load_funding_rates(filepath, None, None).unwrap();
1407
1408        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1409
1410        assert_eq!(funding_rates.len(), 3);
1411        assert!(
1412            funding_rates
1413                .iter()
1414                .all(|f| f.instrument_id == instrument_id)
1415        );
1416
1417        // Deribit publishes no funding timestamp, so there is no forward reference to carry
1418        assert!(
1419            funding_rates
1420                .iter()
1421                .all(|f| f.next_funding_ns.is_none() && f.interval.is_none())
1422        );
1423
1424        let first = &funding_rates[0];
1425        let changed = &funding_rates[2];
1426
1427        assert_eq!(first.rate, dec!(0.00000459));
1428        assert_eq!(first.ts_event, UnixNanos::from(1_786_320_665_523_000_000));
1429        assert_eq!(first.ts_init, UnixNanos::from(1_786_320_665_533_324_000));
1430
1431        assert_eq!(changed.rate, dec!(0.00000452));
1432        assert_eq!(changed.ts_event, UnixNanos::from(1_786_320_665_645_000_000));
1433        assert_eq!(changed.ts_init, UnixNanos::from(1_786_320_665_661_106_000));
1434    }
1435
1436    #[rstest]
1437    fn test_load_funding_rates_okex_usdc_across_index_migration() {
1438        let filepath = get_test_data_path("csv/okex_swap_usdc_index_migration.csv");
1439        let funding_rates = load_funding_rates(filepath, None, None).unwrap();
1440
1441        let instrument_id = InstrumentId::from("BTC-USDC-SWAP.OKEX");
1442
1443        assert_eq!(funding_rates.len(), 6);
1444
1445        // Tardis remaps USDC-margined contracts to the USDC index after 2023-04-10T08:40Z, which
1446        // changes the index price feed but never the contract symbol, so replaying across the
1447        // migration must resolve to one instrument
1448        assert!(
1449            funding_rates
1450                .iter()
1451                .all(|f| f.instrument_id == instrument_id)
1452        );
1453
1454        let pre = &funding_rates[0];
1455        let post = &funding_rates[3];
1456
1457        assert_eq!(pre.rate, dec!(0.0001035718476117));
1458        assert_eq!(
1459            pre.next_funding_ns,
1460            Some(UnixNanos::from(1_680_336_000_000_000_000))
1461        );
1462        assert_eq!(pre.ts_event, UnixNanos::from(1_680_309_048_427_000_000));
1463        assert_eq!(pre.ts_init, UnixNanos::from(1_680_309_048_450_728_000));
1464
1465        assert_eq!(post.rate, dec!(-0.000055804472025));
1466        assert_eq!(
1467            post.next_funding_ns,
1468            Some(UnixNanos::from(1_682_928_000_000_000_000))
1469        );
1470        assert_eq!(post.ts_event, UnixNanos::from(1_682_900_658_676_000_000));
1471        assert_eq!(post.ts_init, UnixNanos::from(1_682_900_658_698_855_000));
1472    }
1473
1474    #[rstest]
1475    fn test_load_depth10_from_snapshot5_comprehensive() {
1476        let filepath = get_tardis_binance_snapshot5_path();
1477        let depths = load_depth10_from_snapshot5(&filepath, None, None, None, Some(100)).unwrap();
1478
1479        assert_eq!(depths.len(), 10);
1480
1481        let first = &depths[0];
1482        assert_eq!(first.instrument_id.to_string(), "BTCUSDT.BINANCE");
1483        assert_eq!(first.bids.len(), 10);
1484        assert_eq!(first.asks.len(), 10);
1485
1486        // Check all bid levels (first 5 from data, rest empty)
1487        assert_eq!(first.bids[0].price, Price::from("11657.07"));
1488        assert_eq!(first.bids[0].size, Quantity::from("10.896"));
1489        assert_eq!(first.bids[0].side, OrderSide::Buy.into());
1490
1491        assert_eq!(first.bids[1].price, Price::from("11656.97"));
1492        assert_eq!(first.bids[1].size, Quantity::from("0.2"));
1493        assert_eq!(first.bids[1].side, OrderSide::Buy.into());
1494
1495        assert_eq!(first.bids[2].price, Price::from("11655.78"));
1496        assert_eq!(first.bids[2].size, Quantity::from("0.2"));
1497        assert_eq!(first.bids[2].side, OrderSide::Buy.into());
1498
1499        assert_eq!(first.bids[3].price, Price::from("11655.77"));
1500        assert_eq!(first.bids[3].size, Quantity::from("0.98"));
1501        assert_eq!(first.bids[3].side, OrderSide::Buy.into());
1502
1503        assert_eq!(first.bids[4].price, Price::from("11655.68"));
1504        assert_eq!(first.bids[4].size, Quantity::from("0.111"));
1505        assert_eq!(first.bids[4].side, OrderSide::Buy.into());
1506
1507        // Empty levels
1508        for i in 5..10 {
1509            assert_eq!(first.bids[i].price.raw, 0);
1510            assert_eq!(first.bids[i].size.raw, 0);
1511            assert_eq!(first.bids[i].side, None);
1512        }
1513
1514        // Check all ask levels (first 5 from data, rest empty)
1515        assert_eq!(first.asks[0].price, Price::from("11657.08"));
1516        assert_eq!(first.asks[0].size, Quantity::from("1.714"));
1517        assert_eq!(first.asks[0].side, OrderSide::Sell.into());
1518
1519        assert_eq!(first.asks[1].price, Price::from("11657.54"));
1520        assert_eq!(first.asks[1].size, Quantity::from("5.4"));
1521        assert_eq!(first.asks[1].side, OrderSide::Sell.into());
1522
1523        assert_eq!(first.asks[2].price, Price::from("11657.56"));
1524        assert_eq!(first.asks[2].size, Quantity::from("0.238"));
1525        assert_eq!(first.asks[2].side, OrderSide::Sell.into());
1526
1527        assert_eq!(first.asks[3].price, Price::from("11657.61"));
1528        assert_eq!(first.asks[3].size, Quantity::from("0.077"));
1529        assert_eq!(first.asks[3].side, OrderSide::Sell.into());
1530
1531        assert_eq!(first.asks[4].price, Price::from("11657.92"));
1532        assert_eq!(first.asks[4].size, Quantity::from("0.918"));
1533        assert_eq!(first.asks[4].side, OrderSide::Sell.into());
1534
1535        // Empty levels
1536        for i in 5..10 {
1537            assert_eq!(first.asks[i].price.raw, 0);
1538            assert_eq!(first.asks[i].size.raw, 0);
1539            assert_eq!(first.asks[i].side, None);
1540        }
1541
1542        // Logical checks: bid prices should decrease
1543        for i in 1..5 {
1544            assert!(
1545                first.bids[i].price < first.bids[i - 1].price,
1546                "Bid price at level {} should be less than level {}",
1547                i,
1548                i - 1
1549            );
1550        }
1551
1552        // Logical checks: ask prices should increase
1553        for i in 1..5 {
1554            assert!(
1555                first.asks[i].price > first.asks[i - 1].price,
1556                "Ask price at level {} should be greater than level {}",
1557                i,
1558                i - 1
1559            );
1560        }
1561
1562        // Logical check: spread should be positive
1563        assert!(
1564            first.asks[0].price > first.bids[0].price,
1565            "Best ask should be greater than best bid"
1566        );
1567
1568        // Check counts
1569        for i in 0..5 {
1570            assert_eq!(first.bid_counts[i], 1);
1571            assert_eq!(first.ask_counts[i], 1);
1572        }
1573
1574        for i in 5..10 {
1575            assert_eq!(first.bid_counts[i], 0);
1576            assert_eq!(first.ask_counts[i], 0);
1577        }
1578
1579        // Check metadata - F_SNAPSHOT (32) | F_LAST (128) = 160
1580        assert_eq!(
1581            first.flags,
1582            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1583        );
1584        assert_eq!(first.ts_event.as_u64(), 1598918403696000000);
1585        assert_eq!(first.ts_init.as_u64(), 1598918403810979000);
1586        assert_eq!(first.sequence, 0);
1587    }
1588
1589    #[rstest]
1590    fn test_load_depth10_from_snapshot25_comprehensive() {
1591        let filepath = get_tardis_binance_snapshot25_path();
1592        let depths = load_depth10_from_snapshot25(&filepath, None, None, None, Some(100)).unwrap();
1593
1594        assert_eq!(depths.len(), 10);
1595
1596        let first = &depths[0];
1597        assert_eq!(first.instrument_id.to_string(), "BTCUSDT.BINANCE");
1598        assert_eq!(first.bids.len(), 10);
1599        assert_eq!(first.asks.len(), 10);
1600
1601        // Check all 10 bid levels from snapshot25
1602        let expected_bids = vec![
1603            ("11657.07", "10.896"),
1604            ("11656.97", "0.2"),
1605            ("11655.78", "0.2"),
1606            ("11655.77", "0.98"),
1607            ("11655.68", "0.111"),
1608            ("11655.66", "0.077"),
1609            ("11655.57", "0.34"),
1610            ("11655.48", "0.4"),
1611            ("11655.26", "1.185"),
1612            ("11654.86", "0.195"),
1613        ];
1614
1615        for (i, (price, size)) in expected_bids.iter().enumerate() {
1616            assert_eq!(first.bids[i].price, Price::from(*price));
1617            assert_eq!(first.bids[i].size, Quantity::from(*size));
1618            assert_eq!(first.bids[i].side, OrderSide::Buy.into());
1619        }
1620
1621        // Check all 10 ask levels from snapshot25
1622        let expected_asks = vec![
1623            ("11657.08", "1.714"),
1624            ("11657.54", "5.4"),
1625            ("11657.56", "0.238"),
1626            ("11657.61", "0.077"),
1627            ("11657.92", "0.918"),
1628            ("11658.09", "1.015"),
1629            ("11658.12", "0.665"),
1630            ("11658.19", "0.583"),
1631            ("11658.28", "0.255"),
1632            ("11658.29", "0.656"),
1633        ];
1634
1635        for (i, (price, size)) in expected_asks.iter().enumerate() {
1636            assert_eq!(first.asks[i].price, Price::from(*price));
1637            assert_eq!(first.asks[i].size, Quantity::from(*size));
1638            assert_eq!(first.asks[i].side, OrderSide::Sell.into());
1639        }
1640
1641        // Logical checks: bid prices should strictly decrease
1642        for i in 1..10 {
1643            assert!(
1644                first.bids[i].price < first.bids[i - 1].price,
1645                "Bid price at level {} ({}) should be less than level {} ({})",
1646                i,
1647                first.bids[i].price,
1648                i - 1,
1649                first.bids[i - 1].price
1650            );
1651        }
1652
1653        // Logical checks: ask prices should strictly increase
1654        for i in 1..10 {
1655            assert!(
1656                first.asks[i].price > first.asks[i - 1].price,
1657                "Ask price at level {} ({}) should be greater than level {} ({})",
1658                i,
1659                first.asks[i].price,
1660                i - 1,
1661                first.asks[i - 1].price
1662            );
1663        }
1664
1665        // Logical check: spread should be positive
1666        assert!(
1667            first.asks[0].price > first.bids[0].price,
1668            "Best ask ({}) should be greater than best bid ({})",
1669            first.asks[0].price,
1670            first.bids[0].price
1671        );
1672
1673        // Check counts (all should be 1 for snapshot data)
1674        for i in 0..10 {
1675            assert_eq!(first.bid_counts[i], 1);
1676            assert_eq!(first.ask_counts[i], 1);
1677        }
1678
1679        // Check metadata - F_SNAPSHOT (32) | F_LAST (128) = 160
1680        assert_eq!(
1681            first.flags,
1682            RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8
1683        );
1684        assert_eq!(first.ts_event.as_u64(), 1598918403696000000);
1685        assert_eq!(first.ts_init.as_u64(), 1598918403810979000);
1686        assert_eq!(first.sequence, 0);
1687    }
1688
1689    #[rstest]
1690    fn test_snapshot_csv_field_order_interleaved() {
1691        // This test verifies that the CSV structs correctly handle the interleaved
1692        // asks/bids field ordering from Tardis CSV files
1693
1694        let csv_data = "exchange,symbol,timestamp,local_timestamp,\
1695asks[0].price,asks[0].amount,bids[0].price,bids[0].amount,\
1696asks[1].price,asks[1].amount,bids[1].price,bids[1].amount,\
1697asks[2].price,asks[2].amount,bids[2].price,bids[2].amount,\
1698asks[3].price,asks[3].amount,bids[3].price,bids[3].amount,\
1699asks[4].price,asks[4].amount,bids[4].price,bids[4].amount
1700binance-futures,BTCUSDT,1000000,2000000,\
1701100.5,1.0,100.4,2.0,\
1702100.6,1.1,100.3,2.1,\
1703100.7,1.2,100.2,2.2,\
1704100.8,1.3,100.1,2.3,\
1705100.9,1.4,100.0,2.4";
1706
1707        let temp_file = std::env::temp_dir().join("test_interleaved_snapshot5.csv");
1708        std::fs::write(&temp_file, csv_data).unwrap();
1709
1710        let depths = load_depth10_from_snapshot5(&temp_file, None, None, None, Some(1)).unwrap();
1711        assert_eq!(depths.len(), 1);
1712
1713        let depth = &depths[0];
1714
1715        // Verify bids are correctly parsed (should be decreasing)
1716        assert_eq!(depth.bids[0].price, Price::from("100.4"));
1717        assert_eq!(depth.bids[1].price, Price::from("100.3"));
1718        assert_eq!(depth.bids[2].price, Price::from("100.2"));
1719        assert_eq!(depth.bids[3].price, Price::from("100.1"));
1720        assert_eq!(depth.bids[4].price, Price::from("100.0"));
1721
1722        // Verify asks are correctly parsed (should be increasing)
1723        assert_eq!(depth.asks[0].price, Price::from("100.5"));
1724        assert_eq!(depth.asks[1].price, Price::from("100.6"));
1725        assert_eq!(depth.asks[2].price, Price::from("100.7"));
1726        assert_eq!(depth.asks[3].price, Price::from("100.8"));
1727        assert_eq!(depth.asks[4].price, Price::from("100.9"));
1728
1729        // Verify sizes
1730        assert_eq!(depth.bids[0].size, Quantity::from("2.0"));
1731        assert_eq!(depth.asks[0].size, Quantity::from("1.0"));
1732
1733        std::fs::remove_file(temp_file).unwrap();
1734    }
1735
1736    #[rstest]
1737    fn test_load_deltas_limit_includes_clear_deltas() {
1738        // Test that limit counts total emitted deltas (including CLEARs)
1739        // When limit=5, we should get exactly 5 deltas: 1 CLEAR + 4 data deltas
1740        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
1741binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
1742binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0
1743binance-futures,BTCUSDT,1640995201000000,1640995201100000,false,bid,49999.0,0.5
1744binance-futures,BTCUSDT,1640995202000000,1640995202100000,false,ask,50002.0,1.5
1745binance-futures,BTCUSDT,1640995203000000,1640995203100000,false,bid,49998.0,0.5
1746binance-futures,BTCUSDT,1640995204000000,1640995204100000,false,ask,50003.0,2.0
1747binance-futures,BTCUSDT,1640995205000000,1640995205100000,false,bid,49997.0,0.5";
1748
1749        let temp_file = std::env::temp_dir().join("test_load_deltas_limit.csv");
1750        std::fs::write(&temp_file, csv_data).unwrap();
1751
1752        // Load with limit=5 (should emit exactly 5 deltas including CLEAR)
1753        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, Some(5)).unwrap();
1754
1755        // Should have exactly 5 deltas: 1 CLEAR + 4 data deltas
1756        assert_eq!(deltas.len(), 5);
1757        assert_eq!(deltas[0].action, BookAction::Clear);
1758        assert_eq!(deltas[1].action, BookAction::Add);
1759        assert_eq!(deltas[2].action, BookAction::Add);
1760        assert_eq!(deltas[3].action, BookAction::Update);
1761        assert_eq!(deltas[4].action, BookAction::Update);
1762
1763        // Verify the last delta is from the 4th CSV record (49999.0 bid)
1764        assert_eq!(deltas[3].order.price, parse_price(49999.0, 1));
1765
1766        std::fs::remove_file(&temp_file).ok();
1767    }
1768
1769    #[rstest]
1770    fn test_load_deltas_limit_stops_at_clear() {
1771        // Test that limit=1 with snapshot data returns only the CLEAR delta
1772        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
1773binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,bid,50000.0,1.0
1774binance-futures,BTCUSDT,1640995200000000,1640995200100000,true,ask,50001.0,2.0";
1775
1776        let temp_file = std::env::temp_dir().join("test_load_deltas_limit_stops_at_clear.csv");
1777        std::fs::write(&temp_file, csv_data).unwrap();
1778
1779        // Load with limit=1 should only get the CLEAR delta
1780        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, Some(1)).unwrap();
1781
1782        assert_eq!(deltas.len(), 1);
1783        assert_eq!(deltas[0].action, BookAction::Clear);
1784
1785        std::fs::remove_file(&temp_file).ok();
1786    }
1787
1788    #[rstest]
1789    fn test_load_deltas_with_consecutive_snapshots_inserts_clear() {
1790        let csv_data = "exchange,symbol,timestamp,local_timestamp,is_snapshot,side,price,amount
1791hyperliquid,BTC,1640995200000000,1640995200100000,true,bid,50000.0,1.0
1792hyperliquid,BTC,1640995200000001,1640995200100000,true,ask,50001.0,2.0
1793hyperliquid,BTC,1640995201000000,1640995201100000,true,bid,49990.0,3.0
1794hyperliquid,BTC,1640995201000001,1640995201100000,true,ask,49991.0,4.0";
1795
1796        let temp_file = std::env::temp_dir().join("test_load_deltas_consecutive_snapshots.csv");
1797        std::fs::write(&temp_file, csv_data).unwrap();
1798
1799        let deltas = load_deltas(&temp_file, Some(1), Some(1), None, None).unwrap();
1800        let clear_count = deltas
1801            .iter()
1802            .filter(|d| d.action == BookAction::Clear)
1803            .count();
1804
1805        assert_eq!(clear_count, 2);
1806        assert_eq!(deltas[0].action, BookAction::Clear);
1807        assert_eq!(deltas[3].action, BookAction::Clear);
1808        assert_eq!(
1809            deltas[2].flags & RecordFlag::F_LAST as u8,
1810            RecordFlag::F_LAST as u8
1811        );
1812        assert_eq!(deltas[3].flags & RecordFlag::F_LAST as u8, 0);
1813        assert_eq!(
1814            deltas
1815                .iter()
1816                .map(|delta| (delta.action, delta.flags, delta.ts_event, delta.ts_init))
1817                .collect::<Vec<_>>(),
1818            vec![
1819                (
1820                    BookAction::Clear,
1821                    RecordFlag::F_SNAPSHOT as u8,
1822                    UnixNanos::from(1_640_995_200_000_000_000),
1823                    UnixNanos::from(1_640_995_200_100_000_000),
1824                ),
1825                (
1826                    BookAction::Add,
1827                    0,
1828                    UnixNanos::from(1_640_995_200_000_000_000),
1829                    UnixNanos::from(1_640_995_200_100_000_000),
1830                ),
1831                (
1832                    BookAction::Add,
1833                    RecordFlag::F_LAST as u8,
1834                    UnixNanos::from(1_640_995_200_000_001_000),
1835                    UnixNanos::from(1_640_995_200_100_000_000),
1836                ),
1837                (
1838                    BookAction::Clear,
1839                    RecordFlag::F_SNAPSHOT as u8,
1840                    UnixNanos::from(1_640_995_201_000_000_000),
1841                    UnixNanos::from(1_640_995_201_100_000_000),
1842                ),
1843                (
1844                    BookAction::Add,
1845                    0,
1846                    UnixNanos::from(1_640_995_201_000_000_000),
1847                    UnixNanos::from(1_640_995_201_100_000_000),
1848                ),
1849                (
1850                    BookAction::Add,
1851                    RecordFlag::F_LAST as u8,
1852                    UnixNanos::from(1_640_995_201_000_001_000),
1853                    UnixNanos::from(1_640_995_201_100_000_000),
1854                ),
1855            ]
1856        );
1857
1858        std::fs::remove_file(&temp_file).ok();
1859    }
1860
1861    #[rstest]
1862    fn test_load_deltas_limit_with_mid_day_snapshot() {
1863        // Test limit behavior when there's a mid-day snapshot
1864        // The limit counts total emitted deltas including CLEARs
1865        let filepath = get_test_data_path("csv/deltas_with_snapshot.csv");
1866        let deltas = load_deltas(filepath, Some(1), Some(1), None, Some(5)).unwrap();
1867
1868        // With limit=5, we get exactly 5 deltas
1869        // First snapshot inserts CLEAR, then we get 4 more data deltas
1870        assert_eq!(deltas.len(), 5);
1871        assert_eq!(deltas[0].action, BookAction::Clear);
1872    }
1873
1874    // Curates the large Tardis Deribit CSV.gz into NautilusTrader Parquet format.
1875    // Run manually: `cargo test -p nautilus-tardis test_curate_deribit_deltas -- --ignored --nocapture`
1876    #[rstest]
1877    #[ignore = "one-time dataset curation, not for routine CI"]
1878    fn test_curate_deribit_deltas() {
1879        let csv_path = get_test_data_root()
1880            .join("large")
1881            .join("tardis_deribit_incremental_book_L2_2020-04-01_BTC-PERPETUAL.csv.gz");
1882
1883        let instrument_id = InstrumentId::from("BTC-PERPETUAL.DERIBIT");
1884        let parquet_path = "/tmp/tardis_BTC-PERPETUAL.DERIBIT_2020-04-01_deltas.parquet";
1885
1886        println!("Loading deltas from {}", csv_path.display());
1887        let deltas = load_deltas(&csv_path, None, None, Some(instrument_id), None).unwrap();
1888        let count = deltas.len();
1889        println!("Loaded {count} deltas");
1890
1891        let sample = deltas
1892            .iter()
1893            .find(|d| d.order.price.precision > 0)
1894            .expect("Should have at least one non-CLEAR delta");
1895        let price_precision = sample.order.price.precision;
1896        let size_precision = sample.order.size.precision;
1897        println!("Precision: price={price_precision}, size={size_precision}");
1898
1899        // Write in chunks to avoid stack overflow on large batches
1900        let metadata =
1901            OrderBookDelta::get_metadata(&instrument_id, price_precision, size_precision);
1902        let schema = OrderBookDelta::get_schema(Some(metadata.clone()));
1903
1904        println!("Writing Parquet to {parquet_path}");
1905        let file = File::create(parquet_path).unwrap();
1906        let zstd_level = parquet::basic::ZstdLevel::try_new(3).unwrap();
1907        let props = WriterProperties::builder()
1908            .set_compression(parquet::basic::Compression::ZSTD(zstd_level))
1909            .set_max_row_group_row_count(Some(1_000_000))
1910            .build();
1911        let mut writer = ArrowWriter::try_new(file, Arc::new(schema), Some(props)).unwrap();
1912
1913        let chunk_size = 1_000_000;
1914        for (i, chunk) in deltas.chunks(chunk_size).enumerate() {
1915            println!("  Encoding chunk {} ({} records)...", i + 1, chunk.len());
1916            let batch = OrderBookDelta::encode_batch(&metadata, chunk).unwrap();
1917            writer.write(&batch).unwrap();
1918        }
1919        writer.close().unwrap();
1920
1921        let file_size = fs::metadata(parquet_path).unwrap().len();
1922        println!("\n=== CURATION COMPLETE ===");
1923        println!("Records: {count}");
1924        println!("Price precision: {price_precision}");
1925        println!("Size precision: {size_precision}");
1926        println!(
1927            "File size: {} bytes ({:.1} MB)",
1928            file_size,
1929            file_size as f64 / 1_048_576.0
1930        );
1931        println!("Output: {parquet_path}");
1932        println!("\nNext steps:");
1933        println!("  sha256sum {parquet_path}");
1934    }
1935}