nautilus_tardis/csv/
convert.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::{path::PathBuf, time::Duration};
17
18use ahash::AHashMap;
19use anyhow::Context;
20use csv::StringRecord;
21use nautilus_core::UnixNanos;
22use nautilus_model::{
23    data::{Data, HasTsInit},
24    enums::OptionKind,
25    identifiers::{InstrumentId, Symbol},
26    instruments::{CryptoOption, InstrumentAny},
27    types::{Currency, Price, Quantity, fixed::FIXED_PRECISION},
28};
29use nautilus_persistence::backend::catalog::ParquetDataCatalog;
30use rust_decimal::Decimal;
31
32use crate::{
33    common::{
34        enums::{TardisExchange, TardisOptionType},
35        parse::{parse_instrument_id, parse_option_kind, parse_timestamp},
36    },
37    csv::{
38        create_csv_reader, infer_precision, load::OptionsChainPrecision, matches_underlying_filter,
39        normalize_underlying_filters, parse_options_chain_record,
40        parse_options_chain_record_as_quote, record::TardisOptionsChainRecord,
41    },
42};
43
44const DATA_FLUSH_ROWS: usize = 100_000;
45
46/// Configuration for converting Tardis `options_chain` CSV files into a Nautilus catalog.
47#[derive(Debug, Clone, bon::Builder)]
48pub struct TardisOptionsChainCSVConverterConfig {
49    /// Tardis daily `options_chain` CSV file paths, processed in order.
50    pub filepaths: Vec<PathBuf>,
51    /// Nautilus catalog path.
52    pub catalog_path: PathBuf,
53    /// Optional underlying prefixes, such as `BTC-`.
54    pub underlyings: Option<Vec<String>>,
55    /// Optional thinning interval. Keeps the last row per instrument per bucket.
56    pub snapshot_interval: Option<Duration>,
57    /// Whether to emit quotes from best bid/offer fields.
58    #[builder(default = true)]
59    pub extract_bbo_as_quotes: bool,
60    /// Whether to derive and write instrument definitions from the CSV rows.
61    #[builder(default = true)]
62    pub write_instruments: bool,
63    /// Optional explicit price precision.
64    pub price_precision: Option<u8>,
65    /// Optional explicit size precision.
66    pub size_precision: Option<u8>,
67}
68
69/// Converts Tardis `options_chain` CSV files into `QuoteTick`, `OptionGreeks`, and instruments.
70///
71/// # Errors
72///
73/// Returns an error if a CSV file cannot be read, a row cannot be parsed, a complete best
74/// bid/offer row contains invalid values, instrument derivation fails, or catalog writes fail.
75pub fn convert_options_chain_csv(
76    config: &TardisOptionsChainCSVConverterConfig,
77) -> anyhow::Result<()> {
78    let underlyings = normalize_underlying_filters(config.underlyings.clone());
79    let catalog = ParquetDataCatalog::new(&config.catalog_path, None, None, None, None);
80    let mut precision_by_instrument: AHashMap<InstrumentId, OptionsChainPrecision> =
81        AHashMap::new();
82    let mut instrument_states: AHashMap<InstrumentId, InstrumentBuildState> = AHashMap::new();
83    let mut data_buffers: AHashMap<InstrumentId, DataBuffer> = AHashMap::new();
84    let mut pending_records: AHashMap<(InstrumentId, u64), TardisOptionsChainRecord> =
85        AHashMap::new();
86    let mut current_bucket = None;
87
88    for filepath in &config.filepaths {
89        let mut reader = create_csv_reader(filepath)
90            .with_context(|| format!("failed to open CSV file {}", filepath.display()))?;
91        let mut csv_record = StringRecord::new();
92
93        while reader
94            .read_record(&mut csv_record)
95            .with_context(|| format!("failed to read CSV file {}", filepath.display()))?
96        {
97            if let Some(underlyings) = underlyings.as_deref() {
98                let Some(symbol) = csv_record.get(1) else {
99                    continue;
100                };
101                let symbol = symbol.to_uppercase();
102                if !matches_underlying_filter(&symbol, Some(underlyings)) {
103                    continue;
104                }
105            }
106
107            let record: TardisOptionsChainRecord = csv_record
108                .deserialize(None)
109                .with_context(|| format!("failed to parse CSV file {}", filepath.display()))?;
110            let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
111            precision_by_instrument
112                .entry(instrument_id)
113                .or_insert_with(|| {
114                    OptionsChainPrecision::new(config.price_precision, config.size_precision)
115                })
116                .update(&record, config.price_precision, config.size_precision);
117            instrument_states
118                .entry(instrument_id)
119                .and_modify(|state| state.update_activation(record.local_timestamp))
120                .or_insert_with(|| InstrumentBuildState::new(record.clone()));
121
122            if let Some(interval) = config.snapshot_interval {
123                let interval_us = u64::try_from(interval.as_micros())
124                    .context("snapshot interval exceeds u64 microseconds")?;
125                anyhow::ensure!(interval_us > 0, "snapshot interval must be positive");
126                let bucket = record.local_timestamp / interval_us;
127
128                if let Some(current_bucket) = current_bucket {
129                    anyhow::ensure!(
130                        bucket >= current_bucket,
131                        "options_chain CSV rows must be ordered by local_timestamp when thinning"
132                    );
133                }
134
135                if current_bucket.is_none_or(|current| bucket > current) {
136                    flush_pending_records_before(
137                        &catalog,
138                        &mut pending_records,
139                        &mut data_buffers,
140                        &precision_by_instrument,
141                        bucket,
142                        config.extract_bbo_as_quotes,
143                    )?;
144                    current_bucket = Some(bucket);
145                }
146                pending_records
147                    .entry((instrument_id, bucket))
148                    .and_modify(|pending| {
149                        if record.local_timestamp >= pending.local_timestamp {
150                            *pending = record.clone();
151                        }
152                    })
153                    .or_insert(record);
154            } else {
155                flush_data_buffer_if_ready(
156                    &catalog,
157                    &mut data_buffers,
158                    instrument_id,
159                    parse_timestamp(record.local_timestamp),
160                )?;
161                let data = options_chain_record_to_data(
162                    &record,
163                    &precision_by_instrument,
164                    config.extract_bbo_as_quotes,
165                )?;
166                data_buffers
167                    .entry(instrument_id)
168                    .or_default()
169                    .extend(data, parse_timestamp(record.local_timestamp));
170            }
171        }
172
173        if config.snapshot_interval.is_some() {
174            flush_pending_records_before(
175                &catalog,
176                &mut pending_records,
177                &mut data_buffers,
178                &precision_by_instrument,
179                u64::MAX,
180                config.extract_bbo_as_quotes,
181            )?;
182            flush_data_buffers(&catalog, &mut data_buffers)?;
183            current_bucket = None;
184        }
185    }
186
187    flush_data_buffers(&catalog, &mut data_buffers)?;
188
189    if config.write_instruments {
190        let instruments = build_instruments(instrument_states, &precision_by_instrument)?;
191        catalog.write_instruments(instruments)?;
192    }
193
194    Ok(())
195}
196
197#[derive(Debug, Clone)]
198struct InstrumentBuildState {
199    record: TardisOptionsChainRecord,
200    activation: UnixNanos,
201}
202
203impl InstrumentBuildState {
204    fn new(record: TardisOptionsChainRecord) -> Self {
205        Self {
206            activation: parse_timestamp(record.local_timestamp),
207            record,
208        }
209    }
210
211    fn update_activation(&mut self, local_timestamp: u64) {
212        self.activation = self.activation.min(parse_timestamp(local_timestamp));
213    }
214}
215
216#[derive(Debug, Default)]
217struct DataBuffer {
218    data: Vec<Data>,
219    last_ts_init: Option<UnixNanos>,
220}
221
222impl DataBuffer {
223    fn extend(&mut self, mut data: Vec<Data>, ts_init: UnixNanos) {
224        self.data.append(&mut data);
225        self.last_ts_init = Some(ts_init);
226    }
227}
228
229fn flush_data_buffer_if_ready(
230    catalog: &ParquetDataCatalog,
231    data_buffers: &mut AHashMap<InstrumentId, DataBuffer>,
232    instrument_id: InstrumentId,
233    ts_init: UnixNanos,
234) -> anyhow::Result<()> {
235    let Some(buffer) = data_buffers.get_mut(&instrument_id) else {
236        return Ok(());
237    };
238
239    if buffer.data.len() >= DATA_FLUSH_ROWS
240        && buffer
241            .last_ts_init
242            .is_some_and(|last_ts_init| last_ts_init < ts_init)
243    {
244        write_data_buffer(catalog, &mut buffer.data)?;
245    }
246
247    Ok(())
248}
249
250fn flush_data_buffers(
251    catalog: &ParquetDataCatalog,
252    data_buffers: &mut AHashMap<InstrumentId, DataBuffer>,
253) -> anyhow::Result<()> {
254    for buffer in data_buffers.values_mut() {
255        write_data_buffer(catalog, &mut buffer.data)?;
256    }
257    data_buffers.clear();
258    Ok(())
259}
260
261fn flush_pending_records_before(
262    catalog: &ParquetDataCatalog,
263    pending_records: &mut AHashMap<(InstrumentId, u64), TardisOptionsChainRecord>,
264    data_buffers: &mut AHashMap<InstrumentId, DataBuffer>,
265    precision_by_instrument: &AHashMap<InstrumentId, OptionsChainPrecision>,
266    next_bucket: u64,
267    extract_bbo_as_quotes: bool,
268) -> anyhow::Result<()> {
269    let mut ready = Vec::new();
270    pending_records.retain(|(_, bucket), record| {
271        if *bucket < next_bucket {
272            ready.push(record.clone());
273            false
274        } else {
275            true
276        }
277    });
278    ready.sort_by_key(|record| (record.local_timestamp, record.symbol));
279
280    for record in ready {
281        let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
282        let ts_init = parse_timestamp(record.local_timestamp);
283        flush_data_buffer_if_ready(catalog, data_buffers, instrument_id, ts_init)?;
284        let data =
285            options_chain_record_to_data(&record, precision_by_instrument, extract_bbo_as_quotes)?;
286        data_buffers
287            .entry(instrument_id)
288            .or_default()
289            .extend(data, ts_init);
290    }
291    Ok(())
292}
293
294fn options_chain_record_to_data(
295    record: &TardisOptionsChainRecord,
296    precision_by_instrument: &AHashMap<InstrumentId, OptionsChainPrecision>,
297    extract_bbo_as_quotes: bool,
298) -> anyhow::Result<Vec<Data>> {
299    let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
300    let precision = precision_by_instrument
301        .get(&instrument_id)
302        .copied()
303        .unwrap_or_else(|| OptionsChainPrecision::new(None, None));
304    let mut data = Vec::with_capacity(2);
305
306    if extract_bbo_as_quotes
307        && let Some(quote) = parse_options_chain_record_as_quote(
308            record,
309            precision.price,
310            precision.size,
311            instrument_id,
312        )?
313    {
314        data.push(Data::Quote(quote));
315    }
316
317    data.push(Data::OptionGreeks(parse_options_chain_record(
318        record,
319        instrument_id,
320    )));
321    Ok(data)
322}
323
324fn write_data_buffer(catalog: &ParquetDataCatalog, data: &mut Vec<Data>) -> anyhow::Result<()> {
325    if data.is_empty() {
326        return Ok(());
327    }
328
329    let mut data_by_instrument: AHashMap<InstrumentId, Vec<Data>> = AHashMap::new();
330    for item in data.drain(..) {
331        data_by_instrument
332            .entry(item.instrument_id())
333            .or_default()
334            .push(item);
335    }
336
337    for instrument_data in data_by_instrument.values_mut() {
338        instrument_data.sort_by_key(HasTsInit::ts_init);
339        catalog.write_data_enum(instrument_data, None, None, None)?;
340    }
341
342    data.clear();
343    Ok(())
344}
345
346fn build_instruments(
347    instrument_states: AHashMap<InstrumentId, InstrumentBuildState>,
348    precision_by_instrument: &AHashMap<InstrumentId, OptionsChainPrecision>,
349) -> anyhow::Result<Vec<InstrumentAny>> {
350    let mut states = instrument_states.into_iter().collect::<Vec<_>>();
351    states.sort_by_key(|(instrument_id, _)| instrument_id.to_string());
352
353    states
354        .into_iter()
355        .map(|(instrument_id, state)| {
356            let precision = precision_by_instrument
357                .get(&instrument_id)
358                .copied()
359                .unwrap_or_else(|| OptionsChainPrecision::new(None, None));
360            create_crypto_option_from_options_chain_record(
361                &state.record,
362                instrument_id,
363                state.activation,
364                precision,
365            )
366        })
367        .collect()
368}
369
370fn create_crypto_option_from_options_chain_record(
371    record: &TardisOptionsChainRecord,
372    instrument_id: InstrumentId,
373    activation: UnixNanos,
374    precision: OptionsChainPrecision,
375) -> anyhow::Result<InstrumentAny> {
376    let underlying = record
377        .symbol
378        .as_str()
379        .split_once('-')
380        .map(|(underlying, _)| underlying)
381        .context("options_chain symbol missing underlying prefix")?;
382    let underlying_currency = Currency::get_or_create_crypto(underlying);
383    let (quote_currency, settlement_currency, is_inverse) =
384        option_currency_mapping(record.exchange, underlying_currency)?;
385    let instrument_price_precision = precision
386        .price
387        .max(infer_precision(record.strike_price).min(FIXED_PRECISION));
388    let price_increment = decimal_increment_price(instrument_price_precision)?;
389    let size_increment = decimal_increment_quantity(precision.size)?;
390    let strike_price = Price::from_decimal_dp(
391        Decimal::try_from(record.strike_price)?,
392        instrument_price_precision,
393    )?;
394    let expiration = parse_timestamp(record.expiration);
395    let option_kind = option_kind(record.option_type);
396
397    Ok(InstrumentAny::CryptoOption(CryptoOption::new_checked(
398        instrument_id,
399        Symbol::from_ustr_unchecked(record.symbol),
400        underlying_currency,
401        quote_currency,
402        settlement_currency,
403        is_inverse,
404        option_kind,
405        strike_price,
406        activation,
407        expiration,
408        instrument_price_precision,
409        precision.size,
410        price_increment,
411        size_increment,
412        None,
413        Some(size_increment),
414        None,
415        Some(size_increment),
416        None,
417        None,
418        None,
419        None,
420        None,
421        None,
422        None,
423        None,
424        None,
425        None,
426        parse_timestamp(record.timestamp),
427        parse_timestamp(record.local_timestamp),
428    )?))
429}
430
431fn option_currency_mapping(
432    exchange: TardisExchange,
433    underlying_currency: Currency,
434) -> anyhow::Result<(Currency, Currency, bool)> {
435    match exchange {
436        TardisExchange::Deribit => Ok((underlying_currency, underlying_currency, true)),
437        exchange => anyhow::bail!(
438            "options_chain instrument derivation supports Deribit only, received {exchange}"
439        ),
440    }
441}
442
443fn decimal_increment_price(precision: u8) -> anyhow::Result<Price> {
444    Ok(Price::from_decimal_dp(
445        Decimal::new(1, u32::from(precision)),
446        precision,
447    )?)
448}
449
450fn decimal_increment_quantity(precision: u8) -> anyhow::Result<Quantity> {
451    Ok(Quantity::from_decimal_dp(
452        Decimal::new(1, u32::from(precision)),
453        precision,
454    )?)
455}
456
457const fn option_kind(value: TardisOptionType) -> OptionKind {
458    parse_option_kind(value)
459}
460
461#[cfg(test)]
462mod tests {
463    use std::{
464        fs,
465        path::{Path, PathBuf},
466        time::Duration,
467    };
468
469    use nautilus_model::{
470        data::{OptionGreeks, QuoteTick},
471        enums::OptionKind,
472        instruments::Instrument,
473    };
474    use nautilus_persistence::backend::catalog::ParquetDataCatalog;
475    use rstest::rstest;
476    use tempfile::TempDir;
477    use ustr::Ustr;
478
479    use super::*;
480    use crate::common::testing::get_test_data_path;
481
482    #[rstest]
483    fn test_options_chain_converter_config_defaults_extract_bbo_as_quotes() {
484        let config = TardisOptionsChainCSVConverterConfig::builder()
485            .filepaths(Vec::<PathBuf>::new())
486            .catalog_path(PathBuf::from("/tmp/options-chain-catalog"))
487            .build();
488
489        assert!(config.extract_bbo_as_quotes);
490        assert!(config.write_instruments);
491    }
492
493    #[rstest]
494    fn test_convert_options_chain_csv_thins_and_round_trips_catalog() {
495        let temp_dir = TempDir::new().unwrap();
496        let filepath = get_test_data_path("options_chain.csv");
497        let config = TardisOptionsChainCSVConverterConfig {
498            filepaths: vec![filepath],
499            catalog_path: temp_dir.path().to_path_buf(),
500            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
501            snapshot_interval: Some(Duration::from_secs(60)),
502            extract_bbo_as_quotes: true,
503            write_instruments: true,
504            price_precision: None,
505            size_precision: None,
506        };
507
508        convert_options_chain_csv(&config).unwrap();
509
510        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
511        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
512        let quotes = catalog
513            .query_typed_data::<QuoteTick>(
514                Some(vec![instrument_id.clone()]),
515                None,
516                None,
517                None,
518                None,
519                true,
520            )
521            .unwrap();
522        let greeks = catalog
523            .query_typed_data::<OptionGreeks>(
524                Some(vec![instrument_id.clone()]),
525                None,
526                None,
527                None,
528                None,
529                true,
530            )
531            .unwrap();
532        let instruments = catalog
533            .query_instruments(Some(std::slice::from_ref(&instrument_id)))
534            .unwrap();
535
536        assert_eq!(quotes.len(), 2);
537        assert_eq!(greeks.len(), 2);
538        assert_eq!(instruments.len(), 1);
539
540        assert_eq!(quotes[0].bid_price, Price::from("0.0206"));
541        assert_eq!(quotes[0].ask_price, Price::from("0.0236"));
542        assert_eq!(quotes[0].ts_init.as_u64(), 1_591_574_400_473_112_000);
543        assert_eq!(quotes[1].bid_price, Price::from("0.0207"));
544        assert_eq!(quotes[1].ts_init.as_u64(), 1_591_574_460_473_112_000);
545
546        let InstrumentAny::CryptoOption(option) = &instruments[0] else {
547            panic!("Expected CryptoOption");
548        };
549        assert_eq!(option.id().to_string(), instrument_id);
550        assert_eq!(option.raw_symbol(), Symbol::from("BTC-9JUN20-9875-P"));
551        assert_eq!(option.option_kind, OptionKind::Put);
552        assert_eq!(option.strike_price, Price::from("9875.0000"));
553        assert_eq!(
554            option.expiration_ns,
555            UnixNanos::from(1_591_689_600_000_000_000)
556        );
557        assert_eq!(
558            option.activation_ns,
559            UnixNanos::from(1_591_574_400_196_008_000)
560        );
561        assert_eq!(option.price_increment, Price::from("0.0001"));
562        assert_eq!(option.size_increment, Quantity::from("0.1"));
563        assert!(option.is_inverse);
564        assert_eq!(option.quote_currency, Currency::from("BTC"));
565        assert_eq!(option.settlement_currency, Currency::from("BTC"));
566    }
567
568    #[rstest]
569    fn test_convert_options_chain_csv_thinned_multi_instrument_writes_separate_catalogs() {
570        let temp_dir = TempDir::new().unwrap();
571        let filepath = get_test_data_path("options_chain.csv");
572        let config = TardisOptionsChainCSVConverterConfig {
573            filepaths: vec![filepath],
574            catalog_path: temp_dir.path().to_path_buf(),
575            underlyings: Some(vec!["BTC-".to_string()]),
576            snapshot_interval: Some(Duration::from_secs(60)),
577            extract_bbo_as_quotes: true,
578            write_instruments: false,
579            price_precision: None,
580            size_precision: None,
581        };
582
583        convert_options_chain_csv(&config).unwrap();
584
585        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
586        let call_id = "BTC-9JUN20-10000-C.DERIBIT".to_string();
587        let next_expiry_id = "BTC-10JUN20-10000-C.DERIBIT".to_string();
588        let call_quotes = catalog
589            .query_typed_data::<QuoteTick>(
590                Some(vec![call_id.clone()]),
591                None,
592                None,
593                None,
594                None,
595                true,
596            )
597            .unwrap();
598        let next_expiry_greeks = catalog
599            .query_typed_data::<OptionGreeks>(
600                Some(vec![next_expiry_id.clone()]),
601                None,
602                None,
603                None,
604                None,
605                true,
606            )
607            .unwrap();
608
609        assert_eq!(call_quotes.len(), 1);
610        assert_eq!(call_quotes[0].instrument_id.to_string(), call_id);
611        assert_eq!(call_quotes[0].bid_price, Price::from("0.0305"));
612        assert_eq!(next_expiry_greeks.len(), 1);
613        assert_eq!(
614            next_expiry_greeks[0].instrument_id.to_string(),
615            next_expiry_id
616        );
617        assert_eq!(next_expiry_greeks[0].greeks.delta, 0.0);
618    }
619
620    #[rstest]
621    fn test_convert_options_chain_csv_unthinned_round_trips_catalog() {
622        let temp_dir = TempDir::new().unwrap();
623        let filepath = get_test_data_path("options_chain.csv");
624        let config = TardisOptionsChainCSVConverterConfig {
625            filepaths: vec![filepath],
626            catalog_path: temp_dir.path().to_path_buf(),
627            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
628            snapshot_interval: None,
629            extract_bbo_as_quotes: true,
630            write_instruments: false,
631            price_precision: None,
632            size_precision: None,
633        };
634
635        convert_options_chain_csv(&config).unwrap();
636
637        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
638        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
639        let quotes = catalog
640            .query_typed_data::<QuoteTick>(
641                Some(vec![instrument_id.clone()]),
642                None,
643                None,
644                None,
645                None,
646                true,
647            )
648            .unwrap();
649        let greeks = catalog
650            .query_typed_data::<OptionGreeks>(
651                Some(vec![instrument_id.clone()]),
652                None,
653                None,
654                None,
655                None,
656                true,
657            )
658            .unwrap();
659
660        assert_eq!(quotes.len(), 3);
661        assert_eq!(greeks.len(), 3);
662        assert_eq!(quotes[0].bid_price, Price::from("0.0205"));
663        assert_eq!(quotes[1].bid_price, Price::from("0.0206"));
664        assert_eq!(quotes[2].bid_price, Price::from("0.0207"));
665        assert_eq!(quotes[0].ts_init.as_u64(), 1_591_574_400_196_008_000);
666        assert_eq!(quotes[2].ts_init.as_u64(), 1_591_574_460_473_112_000);
667        assert!(
668            greeks
669                .iter()
670                .all(|greek| greek.instrument_id.to_string() == instrument_id)
671        );
672    }
673
674    #[rstest]
675    fn test_convert_options_chain_csv_can_suppress_bbo_quotes() {
676        let temp_dir = TempDir::new().unwrap();
677        let filepath = get_test_data_path("options_chain.csv");
678        let config = TardisOptionsChainCSVConverterConfig {
679            filepaths: vec![filepath],
680            catalog_path: temp_dir.path().to_path_buf(),
681            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
682            snapshot_interval: None,
683            extract_bbo_as_quotes: false,
684            write_instruments: false,
685            price_precision: None,
686            size_precision: None,
687        };
688
689        convert_options_chain_csv(&config).unwrap();
690
691        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
692        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
693        let quotes = catalog
694            .query_typed_data::<QuoteTick>(
695                Some(vec![instrument_id.clone()]),
696                None,
697                None,
698                None,
699                None,
700                true,
701            )
702            .unwrap();
703        let greeks = catalog
704            .query_typed_data::<OptionGreeks>(
705                Some(vec![instrument_id.clone()]),
706                None,
707                None,
708                None,
709                None,
710                true,
711            )
712            .unwrap();
713
714        assert!(quotes.is_empty());
715        assert_eq!(greeks.len(), 3);
716        assert!(
717            greeks
718                .iter()
719                .all(|greek| greek.instrument_id.to_string() == instrument_id)
720        );
721    }
722
723    #[rstest]
724    fn test_convert_options_chain_csv_thinned_multi_file_resets_bucket_state() {
725        let temp_dir = TempDir::new().unwrap();
726        let fixture = fs::read_to_string(get_test_data_path("options_chain.csv")).unwrap();
727        let lines = fixture.lines().collect::<Vec<_>>();
728        let first_filepath = temp_dir.path().join("late_btc_options_chain.csv");
729        let second_filepath = temp_dir.path().join("early_eth_options_chain.csv");
730        let catalog_path = temp_dir.path().join("catalog");
731        fs::create_dir(&catalog_path).unwrap();
732        write_options_chain_rows(&first_filepath, lines[0], &[lines[7]]);
733        write_options_chain_rows(&second_filepath, lines[0], &[lines[5]]);
734
735        let config = TardisOptionsChainCSVConverterConfig {
736            filepaths: vec![first_filepath, second_filepath],
737            catalog_path: catalog_path.clone(),
738            underlyings: None,
739            snapshot_interval: Some(Duration::from_secs(60)),
740            extract_bbo_as_quotes: true,
741            write_instruments: false,
742            price_precision: None,
743            size_precision: None,
744        };
745
746        convert_options_chain_csv(&config).unwrap();
747
748        let mut catalog = ParquetDataCatalog::new(&catalog_path, None, None, None, None);
749        let btc_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
750        let eth_id = "ETH-9JUN20-250-P.DERIBIT".to_string();
751        let btc_quotes = catalog
752            .query_typed_data::<QuoteTick>(Some(vec![btc_id.clone()]), None, None, None, None, true)
753            .unwrap();
754        let eth_quotes = catalog
755            .query_typed_data::<QuoteTick>(Some(vec![eth_id.clone()]), None, None, None, None, true)
756            .unwrap();
757        let eth_greeks = catalog
758            .query_typed_data::<OptionGreeks>(
759                Some(vec![eth_id.clone()]),
760                None,
761                None,
762                None,
763                None,
764                true,
765            )
766            .unwrap();
767
768        assert_eq!(btc_quotes.len(), 1);
769        assert_eq!(btc_quotes[0].instrument_id.to_string(), btc_id);
770        assert_eq!(btc_quotes[0].bid_price, Price::from("0.0207"));
771        assert_eq!(eth_quotes.len(), 1);
772        assert_eq!(eth_quotes[0].instrument_id.to_string(), eth_id);
773        assert_eq!(eth_quotes[0].bid_price, Price::from("0.12345"));
774        assert_eq!(eth_greeks.len(), 1);
775        assert_eq!(eth_greeks[0].instrument_id.to_string(), eth_id);
776    }
777
778    #[rstest]
779    fn test_create_crypto_option_preserves_fractional_strike_and_new_underlying() {
780        let record = TardisOptionsChainRecord {
781            exchange: TardisExchange::Deribit,
782            symbol: Ustr::from("NEW-9JUN20-2.05-C"),
783            timestamp: 1,
784            local_timestamp: 2,
785            option_type: TardisOptionType::Call,
786            strike_price: 2.05,
787            expiration: 1_591_689_600_000_000,
788            open_interest: None,
789            last_price: Some(0.1),
790            bid_price: Some(0.1),
791            bid_amount: Some(1.0),
792            bid_iv: None,
793            ask_price: Some(0.2),
794            ask_amount: Some(1.0),
795            ask_iv: None,
796            mark_price: None,
797            mark_iv: None,
798            underlying_index: "SYN.NEW-9JUN20".to_string(),
799            underlying_price: Some(2.0),
800            delta: None,
801            gamma: None,
802            vega: None,
803            theta: None,
804            rho: None,
805        };
806        let instrument_id = InstrumentId::from("NEW-9JUN20-2.05-C.DERIBIT");
807        let instrument = create_crypto_option_from_options_chain_record(
808            &record,
809            instrument_id,
810            UnixNanos::from(2_000),
811            OptionsChainPrecision { price: 1, size: 0 },
812        )
813        .unwrap();
814
815        let InstrumentAny::CryptoOption(option) = instrument else {
816            panic!("Expected CryptoOption");
817        };
818
819        assert_eq!(option.underlying, Currency::get_or_create_crypto("NEW"));
820        assert_eq!(option.strike_price, Price::from("2.05"));
821        assert_eq!(option.price_precision, 2);
822        assert_eq!(option.price_increment, Price::from("0.01"));
823    }
824
825    fn write_options_chain_rows(path: &Path, header: &str, rows: &[&str]) {
826        let mut contents = String::from(header);
827        contents.push('\n');
828        for row in rows {
829            contents.push_str(row);
830            contents.push('\n');
831        }
832        fs::write(path, contents).unwrap();
833    }
834}
nautilus_tardis/csv/convert.rs

nautilus_tardis/csv/
convert.rs
