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(
398        CryptoOption::builder()
399            .instrument_id(instrument_id)
400            .raw_symbol(Symbol::from_ustr_unchecked(record.symbol))
401            .underlying(underlying_currency)
402            .quote_currency(quote_currency)
403            .settlement_currency(settlement_currency)
404            .is_inverse(is_inverse)
405            .option_kind(option_kind)
406            .strike_price(strike_price)
407            .activation_ns(activation)
408            .expiration_ns(expiration)
409            .price_precision(instrument_price_precision)
410            .size_precision(precision.size)
411            .price_increment(price_increment)
412            .size_increment(size_increment)
413            .lot_size(size_increment)
414            .min_quantity(size_increment)
415            .ts_event(parse_timestamp(record.timestamp))
416            .ts_init(parse_timestamp(record.local_timestamp))
417            .build()?,
418    ))
419}
420
421fn option_currency_mapping(
422    exchange: TardisExchange,
423    underlying_currency: Currency,
424) -> anyhow::Result<(Currency, Currency, bool)> {
425    match exchange {
426        TardisExchange::Deribit => Ok((underlying_currency, underlying_currency, true)),
427        exchange => anyhow::bail!(
428            "options_chain instrument derivation supports Deribit only, received {exchange}"
429        ),
430    }
431}
432
433fn decimal_increment_price(precision: u8) -> anyhow::Result<Price> {
434    Ok(Price::from_decimal_dp(
435        Decimal::new(1, u32::from(precision)),
436        precision,
437    )?)
438}
439
440fn decimal_increment_quantity(precision: u8) -> anyhow::Result<Quantity> {
441    Ok(Quantity::from_decimal_dp(
442        Decimal::new(1, u32::from(precision)),
443        precision,
444    )?)
445}
446
447const fn option_kind(value: TardisOptionType) -> OptionKind {
448    parse_option_kind(value)
449}
450
451#[cfg(test)]
452mod tests {
453    use std::{
454        fs,
455        path::{Path, PathBuf},
456        time::Duration,
457    };
458
459    use nautilus_model::{
460        data::{OptionGreeks, QuoteTick},
461        enums::OptionKind,
462        instruments::Instrument,
463    };
464    use nautilus_persistence::backend::catalog::ParquetDataCatalog;
465    use rstest::rstest;
466    use tempfile::TempDir;
467    use ustr::Ustr;
468
469    use super::*;
470    use crate::common::testing::get_test_data_path;
471
472    #[rstest]
473    fn test_options_chain_converter_config_defaults_extract_bbo_as_quotes() {
474        let config = TardisOptionsChainCSVConverterConfig::builder()
475            .filepaths(Vec::<PathBuf>::new())
476            .catalog_path(PathBuf::from("/tmp/options-chain-catalog"))
477            .build();
478
479        assert!(config.extract_bbo_as_quotes);
480        assert!(config.write_instruments);
481    }
482
483    #[rstest]
484    fn test_convert_options_chain_csv_thins_and_round_trips_catalog() {
485        let temp_dir = TempDir::new().unwrap();
486        let filepath = get_test_data_path("options_chain.csv");
487        let config = TardisOptionsChainCSVConverterConfig {
488            filepaths: vec![filepath],
489            catalog_path: temp_dir.path().to_path_buf(),
490            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
491            snapshot_interval: Some(Duration::from_secs(60)),
492            extract_bbo_as_quotes: true,
493            write_instruments: true,
494            price_precision: None,
495            size_precision: None,
496        };
497
498        convert_options_chain_csv(&config).unwrap();
499
500        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
501        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
502        let quotes = catalog
503            .query_typed_data::<QuoteTick>(
504                Some(vec![instrument_id.clone()]),
505                None,
506                None,
507                None,
508                None,
509                true,
510            )
511            .unwrap();
512        let greeks = catalog
513            .query_typed_data::<OptionGreeks>(
514                Some(vec![instrument_id.clone()]),
515                None,
516                None,
517                None,
518                None,
519                true,
520            )
521            .unwrap();
522        let instruments = catalog
523            .query_instruments(Some(std::slice::from_ref(&instrument_id)))
524            .unwrap();
525
526        assert_eq!(quotes.len(), 2);
527        assert_eq!(greeks.len(), 2);
528        assert_eq!(instruments.len(), 1);
529
530        assert_eq!(quotes[0].bid_price, Price::from("0.0206"));
531        assert_eq!(quotes[0].ask_price, Price::from("0.0236"));
532        assert_eq!(quotes[0].ts_init.as_u64(), 1_591_574_400_473_112_000);
533        assert_eq!(quotes[1].bid_price, Price::from("0.0207"));
534        assert_eq!(quotes[1].ts_init.as_u64(), 1_591_574_460_473_112_000);
535
536        let InstrumentAny::CryptoOption(option) = &instruments[0] else {
537            panic!("Expected CryptoOption");
538        };
539        assert_eq!(option.id().to_string(), instrument_id);
540        assert_eq!(option.raw_symbol(), Symbol::from("BTC-9JUN20-9875-P"));
541        assert_eq!(option.option_kind, OptionKind::Put);
542        assert_eq!(option.strike_price, Price::from("9875.0000"));
543        assert_eq!(
544            option.expiration_ns,
545            UnixNanos::from(1_591_689_600_000_000_000)
546        );
547        assert_eq!(
548            option.activation_ns,
549            UnixNanos::from(1_591_574_400_196_008_000)
550        );
551        assert_eq!(option.price_increment, Price::from("0.0001"));
552        assert_eq!(option.size_increment, Quantity::from("0.1"));
553        assert!(option.is_inverse);
554        assert_eq!(option.quote_currency, Currency::from("BTC"));
555        assert_eq!(option.settlement_currency, Currency::from("BTC"));
556    }
557
558    #[rstest]
559    fn test_convert_options_chain_csv_thinned_multi_instrument_writes_separate_catalogs() {
560        let temp_dir = TempDir::new().unwrap();
561        let filepath = get_test_data_path("options_chain.csv");
562        let config = TardisOptionsChainCSVConverterConfig {
563            filepaths: vec![filepath],
564            catalog_path: temp_dir.path().to_path_buf(),
565            underlyings: Some(vec!["BTC-".to_string()]),
566            snapshot_interval: Some(Duration::from_secs(60)),
567            extract_bbo_as_quotes: true,
568            write_instruments: false,
569            price_precision: None,
570            size_precision: None,
571        };
572
573        convert_options_chain_csv(&config).unwrap();
574
575        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
576        let call_id = "BTC-9JUN20-10000-C.DERIBIT".to_string();
577        let next_expiry_id = "BTC-10JUN20-10000-C.DERIBIT".to_string();
578        let call_quotes = catalog
579            .query_typed_data::<QuoteTick>(
580                Some(vec![call_id.clone()]),
581                None,
582                None,
583                None,
584                None,
585                true,
586            )
587            .unwrap();
588        let next_expiry_greeks = catalog
589            .query_typed_data::<OptionGreeks>(
590                Some(vec![next_expiry_id.clone()]),
591                None,
592                None,
593                None,
594                None,
595                true,
596            )
597            .unwrap();
598
599        assert_eq!(call_quotes.len(), 1);
600        assert_eq!(call_quotes[0].instrument_id.to_string(), call_id);
601        assert_eq!(call_quotes[0].bid_price, Price::from("0.0305"));
602        assert_eq!(next_expiry_greeks.len(), 1);
603        assert_eq!(
604            next_expiry_greeks[0].instrument_id.to_string(),
605            next_expiry_id
606        );
607        assert_eq!(next_expiry_greeks[0].greeks.delta, 0.0);
608    }
609
610    #[rstest]
611    fn test_convert_options_chain_csv_unthinned_round_trips_catalog() {
612        let temp_dir = TempDir::new().unwrap();
613        let filepath = get_test_data_path("options_chain.csv");
614        let config = TardisOptionsChainCSVConverterConfig {
615            filepaths: vec![filepath],
616            catalog_path: temp_dir.path().to_path_buf(),
617            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
618            snapshot_interval: None,
619            extract_bbo_as_quotes: true,
620            write_instruments: false,
621            price_precision: None,
622            size_precision: None,
623        };
624
625        convert_options_chain_csv(&config).unwrap();
626
627        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
628        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
629        let quotes = catalog
630            .query_typed_data::<QuoteTick>(
631                Some(vec![instrument_id.clone()]),
632                None,
633                None,
634                None,
635                None,
636                true,
637            )
638            .unwrap();
639        let greeks = catalog
640            .query_typed_data::<OptionGreeks>(
641                Some(vec![instrument_id.clone()]),
642                None,
643                None,
644                None,
645                None,
646                true,
647            )
648            .unwrap();
649
650        assert_eq!(quotes.len(), 3);
651        assert_eq!(greeks.len(), 3);
652        assert_eq!(quotes[0].bid_price, Price::from("0.0205"));
653        assert_eq!(quotes[1].bid_price, Price::from("0.0206"));
654        assert_eq!(quotes[2].bid_price, Price::from("0.0207"));
655        assert_eq!(quotes[0].ts_init.as_u64(), 1_591_574_400_196_008_000);
656        assert_eq!(quotes[2].ts_init.as_u64(), 1_591_574_460_473_112_000);
657        assert!(
658            greeks
659                .iter()
660                .all(|greek| greek.instrument_id.to_string() == instrument_id)
661        );
662    }
663
664    #[rstest]
665    fn test_convert_options_chain_csv_can_suppress_bbo_quotes() {
666        let temp_dir = TempDir::new().unwrap();
667        let filepath = get_test_data_path("options_chain.csv");
668        let config = TardisOptionsChainCSVConverterConfig {
669            filepaths: vec![filepath],
670            catalog_path: temp_dir.path().to_path_buf(),
671            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
672            snapshot_interval: None,
673            extract_bbo_as_quotes: false,
674            write_instruments: false,
675            price_precision: None,
676            size_precision: None,
677        };
678
679        convert_options_chain_csv(&config).unwrap();
680
681        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
682        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
683        let quotes = catalog
684            .query_typed_data::<QuoteTick>(
685                Some(vec![instrument_id.clone()]),
686                None,
687                None,
688                None,
689                None,
690                true,
691            )
692            .unwrap();
693        let greeks = catalog
694            .query_typed_data::<OptionGreeks>(
695                Some(vec![instrument_id.clone()]),
696                None,
697                None,
698                None,
699                None,
700                true,
701            )
702            .unwrap();
703
704        assert!(quotes.is_empty());
705        assert_eq!(greeks.len(), 3);
706        assert!(
707            greeks
708                .iter()
709                .all(|greek| greek.instrument_id.to_string() == instrument_id)
710        );
711    }
712
713    #[rstest]
714    fn test_convert_options_chain_csv_thinned_multi_file_resets_bucket_state() {
715        let temp_dir = TempDir::new().unwrap();
716        let fixture = fs::read_to_string(get_test_data_path("options_chain.csv")).unwrap();
717        let lines = fixture.lines().collect::<Vec<_>>();
718        let first_filepath = temp_dir.path().join("late_btc_options_chain.csv");
719        let second_filepath = temp_dir.path().join("early_eth_options_chain.csv");
720        let catalog_path = temp_dir.path().join("catalog");
721        fs::create_dir(&catalog_path).unwrap();
722        write_options_chain_rows(&first_filepath, lines[0], &[lines[7]]);
723        write_options_chain_rows(&second_filepath, lines[0], &[lines[5]]);
724
725        let config = TardisOptionsChainCSVConverterConfig {
726            filepaths: vec![first_filepath, second_filepath],
727            catalog_path: catalog_path.clone(),
728            underlyings: None,
729            snapshot_interval: Some(Duration::from_secs(60)),
730            extract_bbo_as_quotes: true,
731            write_instruments: false,
732            price_precision: None,
733            size_precision: None,
734        };
735
736        convert_options_chain_csv(&config).unwrap();
737
738        let mut catalog = ParquetDataCatalog::new(&catalog_path, None, None, None, None);
739        let btc_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
740        let eth_id = "ETH-9JUN20-250-P.DERIBIT".to_string();
741        let btc_quotes = catalog
742            .query_typed_data::<QuoteTick>(Some(vec![btc_id.clone()]), None, None, None, None, true)
743            .unwrap();
744        let eth_quotes = catalog
745            .query_typed_data::<QuoteTick>(Some(vec![eth_id.clone()]), None, None, None, None, true)
746            .unwrap();
747        let eth_greeks = catalog
748            .query_typed_data::<OptionGreeks>(
749                Some(vec![eth_id.clone()]),
750                None,
751                None,
752                None,
753                None,
754                true,
755            )
756            .unwrap();
757
758        assert_eq!(btc_quotes.len(), 1);
759        assert_eq!(btc_quotes[0].instrument_id.to_string(), btc_id);
760        assert_eq!(btc_quotes[0].bid_price, Price::from("0.0207"));
761        assert_eq!(eth_quotes.len(), 1);
762        assert_eq!(eth_quotes[0].instrument_id.to_string(), eth_id);
763        assert_eq!(eth_quotes[0].bid_price, Price::from("0.12345"));
764        assert_eq!(eth_greeks.len(), 1);
765        assert_eq!(eth_greeks[0].instrument_id.to_string(), eth_id);
766    }
767
768    #[rstest]
769    fn test_create_crypto_option_preserves_fractional_strike_and_new_underlying() {
770        let record = TardisOptionsChainRecord {
771            exchange: TardisExchange::Deribit,
772            symbol: Ustr::from("NEW-9JUN20-2.05-C"),
773            timestamp: 1,
774            local_timestamp: 2,
775            option_type: TardisOptionType::Call,
776            strike_price: 2.05,
777            expiration: 1_591_689_600_000_000,
778            open_interest: None,
779            last_price: Some(0.1),
780            bid_price: Some(0.1),
781            bid_amount: Some(1.0),
782            bid_iv: None,
783            ask_price: Some(0.2),
784            ask_amount: Some(1.0),
785            ask_iv: None,
786            mark_price: None,
787            mark_iv: None,
788            underlying_index: "SYN.NEW-9JUN20".to_string(),
789            underlying_price: Some(2.0),
790            delta: None,
791            gamma: None,
792            vega: None,
793            theta: None,
794            rho: None,
795        };
796        let instrument_id = InstrumentId::from("NEW-9JUN20-2.05-C.DERIBIT");
797        let instrument = create_crypto_option_from_options_chain_record(
798            &record,
799            instrument_id,
800            UnixNanos::from(2_000),
801            OptionsChainPrecision { price: 1, size: 0 },
802        )
803        .unwrap();
804
805        let InstrumentAny::CryptoOption(option) = instrument else {
806            panic!("Expected CryptoOption");
807        };
808
809        assert_eq!(option.underlying, Currency::get_or_create_crypto("NEW"));
810        assert_eq!(option.strike_price, Price::from("2.05"));
811        assert_eq!(option.price_precision, 2);
812        assert_eq!(option.price_increment, Price::from("0.01"));
813    }
814
815    fn write_options_chain_rows(path: &Path, header: &str, rows: &[&str]) {
816        let mut contents = String::from(header);
817        contents.push('\n');
818        for row in rows {
819            contents.push_str(row);
820            contents.push('\n');
821        }
822        fs::write(path, contents).unwrap();
823    }
824}
nautilus_tardis/csv/convert.rs

nautilus_tardis/csv/
convert.rs
