Continue Databento parsing in Rust

nautilus_core/adapters/Cargo.toml

@@ -16,13 +16,15 @@ nautilus-core = { path = "../core" }
 nautilus-model = { path = "../model", features = ["stubs"]}
 anyhow = { workspace = true }
 chrono = { workspace = true }
+indexmap = { workspace = true }
 itoa = { workspace = true }
 pyo3 = { workspace = true, optional = true }
 rand = { workspace = true }
 rust_decimal = { workspace = true }
 rust_decimal_macros = { workspace = true }
 tokio = { workspace = true }
 thiserror = { workspace = true }
+ustr = { workspace = true }
 dbn = { version = "0.14.2", optional = true }
 
 [features]

nautilus_core/adapters/src/databento/common.rs

@@ -0,0 +1,31 @@
+// -------------------------------------------------------------------------------------------------
+//  Copyright (C) 2015-2023 Nautech Systems Pty Ltd. All rights reserved.
+//  https://nautechsystems.io
+//
+//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+//  You may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+// -------------------------------------------------------------------------------------------------
+
+use nautilus_model::identifiers::{instrument_id::InstrumentId, symbol::Symbol, venue::Venue};
+use ustr::Ustr;
+
+use super::types::DatabentoPublisher;
+
+pub fn nautilus_instrument_id_from_databento(
+    raw_symbol: Ustr,
+    publisher: &DatabentoPublisher,
+) -> InstrumentId {
+    let symbol = Symbol { value: raw_symbol };
+    let venue = Venue {
+        value: Ustr::from(publisher.venue.as_str()),
+    }; // TODO: Optimize
+
+    InstrumentId::new(symbol, venue)
+}

nautilus_core/adapters/src/databento/mod.rs

@@ -13,4 +13,6 @@
 //  limitations under the License.
 // -------------------------------------------------------------------------------------------------
 
+pub mod common;
 pub mod parsing;
+pub mod types;

nautilus_core/adapters/src/databento/parsing.rs

@@ -17,6 +17,7 @@ use std::{
     cmp,
     ffi::{c_char, CStr},
     i64,
+    str::FromStr,
 };
 
 use anyhow::{anyhow, bail, Result};
@@ -36,10 +37,16 @@ use nautilus_model::{
         OptionKind, OrderSide, PriceType,
     },
     identifiers::{instrument_id::InstrumentId, trade_id::TradeId},
-    instruments::equity::Equity,
+    instruments::{
+        equity::Equity, futures_contract::FuturesContract, options_contract::OptionsContract,
+        Instrument,
+    },
     types::{currency::Currency, price::Price, quantity::Quantity},
 };
 use rust_decimal_macros::dec;
+use ustr::Ustr;
+
+use super::{common::nautilus_instrument_id_from_databento, types::DatabentoPublisher};
 
 pub fn parse_order_side(c: c_char) -> OrderSide {
     match c as u8 as char {
@@ -115,24 +122,34 @@ pub fn parse_min_price_increment(value: i64, currency: Currency) -> Result<Price
     }
 }
 
-pub fn parse_raw_symbol_to_string(raw_symbol: [c_char; dbn::SYMBOL_CSTR_LEN]) -> Result<String> {
-    let c_str: &CStr = unsafe { CStr::from_ptr(raw_symbol.as_ptr()) };
+/// # Safety
+///
+/// - Assumes `ptr` is a valid C string pointer.
+pub unsafe fn parse_raw_ptr_to_string(ptr: *const c_char) -> Result<String> {
+    let c_str: &CStr = unsafe { CStr::from_ptr(ptr) };
     let str_slice: &str = c_str.to_str().map_err(|e| anyhow!(e))?;
     Ok(str_slice.to_owned())
 }
 
+/// # Safety
+///
+/// - Assumes `ptr` is a valid C string pointer.
+pub unsafe fn parse_raw_ptr_to_ustr(ptr: *const c_char) -> Result<Ustr> {
+    let c_str: &CStr = unsafe { CStr::from_ptr(ptr) };
+    let str_slice: &str = c_str.to_str().map_err(|e| anyhow!(e))?;
+    Ok(Ustr::from(str_slice))
+}
+
 pub fn parse_equity(
     record: dbn::InstrumentDefMsg,
     instrument_id: InstrumentId,
     ts_init: UnixNanos,
 ) -> Result<Equity> {
-    // Use USD for all US equities venues for now
-    let currency = Currency::USD();
+    let currency = Currency::USD(); // TODO: Temporary hard coding of US equities for now
 
     Equity::new(
         instrument_id,
         instrument_id.symbol,
-        // Symbol::from_str(&parse_raw_symbol_to_string(record.raw_symbol)?)?,
         None, // No ISIN available yet
         currency,
         currency.precision,
@@ -151,6 +168,78 @@ pub fn parse_equity(
     )
 }
 
+pub fn parse_futures_contract(
+    record: dbn::InstrumentDefMsg,
+    instrument_id: InstrumentId,
+    ts_init: UnixNanos,
+) -> Result<FuturesContract> {
+    let currency = Currency::USD(); // TODO: Temporary hard coding of US futures for now
+    let cfi_str = unsafe { parse_raw_ptr_to_string(record.cfi.as_ptr())? };
+    let asset = unsafe { parse_raw_ptr_to_ustr(record.asset.as_ptr())? };
+    let (asset_class, _) = parse_cfi_iso10926(&cfi_str)?;
+
+    FuturesContract::new(
+        instrument_id,
+        instrument_id.symbol,
+        asset_class.unwrap_or(AssetClass::Commodity),
+        asset,
+        record.activation,
+        record.expiration,
+        currency,
+        currency.precision,
+        parse_min_price_increment(record.min_price_increment, currency)?,
+        dec!(0), // TBD
+        dec!(0), // TBD
+        dec!(0), // TBD
+        dec!(0), // TBD
+        Quantity::new(record.contract_multiplier.into(), 0)?,
+        None,           // TBD
+        None,           // TBD
+        None,           // TBD
+        None,           // TBD
+        None,           // TBD
+        record.ts_recv, // More accurate and reliable timestamp
+        ts_init,
+    )
+}
+
+pub fn parse_options_contract(
+    record: dbn::InstrumentDefMsg,
+    instrument_id: InstrumentId,
+    ts_init: UnixNanos,
+) -> Result<OptionsContract> {
+    let currency_str = unsafe { parse_raw_ptr_to_string(record.currency.as_ptr())? };
+    let cfi_str = unsafe { parse_raw_ptr_to_string(record.cfi.as_ptr())? };
+    let currency = Currency::from_str(&currency_str)?;
+    let (asset_class, _) = parse_cfi_iso10926(&cfi_str)?;
+    let lot_size = Quantity::new(1.0, 0)?;
+
+    OptionsContract::new(
+        instrument_id,
+        instrument_id.symbol,
+        asset_class.unwrap_or(AssetClass::Commodity),
+        unsafe { parse_raw_ptr_to_ustr(record.asset.as_ptr())? },
+        parse_option_kind(record.instrument_class)?,
+        record.activation,
+        record.expiration,
+        Price::from_raw(record.strike_price, currency.precision)?,
+        currency,
+        currency.precision,
+        parse_min_price_increment(record.min_price_increment, currency)?,
+        dec!(0), // TBD
+        dec!(0), // TBD
+        dec!(0), // TBD
+        dec!(0), // TBD
+        Some(lot_size),
+        None,           // TBD
+        None,           // TBD
+        None,           // TBD
+        None,           // TBD
+        record.ts_recv, // More accurate and reliable timestamp
+        ts_init,
+    )
+}
+
 pub fn is_trade_msg(order_side: OrderSide, action: c_char) -> bool {
     order_side == OrderSide::NoOrderSide || action as u8 as char == 'T'
 }
@@ -409,3 +498,49 @@ pub fn parse_ohlcv_msg(
 
     Ok(bar)
 }
+
+// pub fn parse_record_with_metadata<T>(
+//     record: T,
+//     publishers: IndexMap<PublisherId, DatabentoPublisher>,
+//     ts_init: UnixNanos,
+// ) -> Result<Data>
+// where
+//     T: dbn::Record,
+// {
+//     let publisher_id: PublisherId = record.header().publisher_id;
+//     let publisher = publishers
+//         .get(&record.header().publisher_id)
+//         .ok_or_else(|| anyhow!("Publisher ID {publisher_id} not found in map"))?;
+//     match record.rtype() {
+//         dbn::RType::InstrumentDef => parse_instrument_def_msg(record, publisher, ts_init)?,
+//         _ => bail!("OOPS!"),
+//     }
+// }
+
+pub fn parse_instrument_def_msg(
+    record: dbn::InstrumentDefMsg,
+    publisher: &DatabentoPublisher,
+    ts_init: UnixNanos,
+) -> Result<Box<dyn Instrument>> {
+    let raw_symbol = unsafe { parse_raw_ptr_to_ustr(record.raw_symbol.as_ptr())? };
+    let instrument_id = nautilus_instrument_id_from_databento(raw_symbol, publisher);
+
+    match record.instrument_class as u8 as char {
+        'C' | 'P' => Ok(Box::new(parse_options_contract(
+            record,
+            instrument_id,
+            ts_init,
+        )?)),
+        'K' => Ok(Box::new(parse_equity(record, instrument_id, ts_init)?)),
+        'F' => Ok(Box::new(parse_futures_contract(
+            record,
+            instrument_id,
+            ts_init,
+        )?)),
+        'X' => bail!("Unsupported `instrument_class` 'X' (FX_SPOT)"),
+        _ => bail!(
+            "Invalid `instrument_class`, was {}",
+            record.instrument_class
+        ),
+    }
+}