Add arrow schema for instrument close

Author: cjdsellers

RELEASES.md

@@ -13,6 +13,9 @@ This release adds support for Python 3.13 (*not* yet compatible with free-thread
 - Added log file rotation with additional config options `max_file_size` and `max_backup_count` (#2468), thanks @xingyanan and @twitu
 - Added `bars_timestamp_on_close` config option for `BybitDataClientConfig` (default `True` to match Nautilus conventions)
 - Added `BetfairSequenceCompleted` custom data type for Betfair to mark the completion of a sequence of messages
+- Added Arrow schema for `MarkPriceUpdate` in Rust
+- Added Arrow schema for `IndexPriceUpdate` in Rust
+- Added Arrow schema for `InstrumentClose` in Rust
 - Improved robustness of in-flight order check for `LiveExecutionEngine`, once exceeded query retries will resolve submitted orders as rejected and pending orders as canceled
 - Improved logging for `BacktestNode` crashes with full stack trace and prettier config logging
 

crates/adapters/tardis/src/replay.rs

@@ -188,8 +188,7 @@ pub async fn run_tardis_machine_replay_from_config(config_filepath: &Path) -> an
             Data::Trade(msg) => handle_trade_msg(msg, &mut trades_map, &mut trades_cursors, &path),
             Data::Bar(msg) => handle_bar_msg(msg, &mut bars_map, &mut bars_cursors, &path),
             Data::Delta(_) => panic!("Individual delta message not implemented (or required)"),
-            Data::MarkPrice(_) => panic!("Not implemented"),
-            Data::IndexPrice(_) => panic!("Not implemented"),
+            _ => panic!("Not implemented"),
         }
 
         msg_count += 1;

crates/data/src/engine/mod.rs

@@ -76,6 +76,7 @@ use nautilus_model::{
     data::{
         Bar, BarType, Data, DataType, OrderBookDelta, OrderBookDeltas, OrderBookDepth10, QuoteTick,
         TradeTick,
+        close::InstrumentClose,
         prices::{IndexPriceUpdate, MarkPriceUpdate},
     },
     enums::{AggregationSource, BarAggregation, BookType, PriceType, RecordFlag},
@@ -442,8 +443,9 @@ impl DataEngine {
             Data::Quote(quote) => self.handle_quote(quote),
             Data::Trade(trade) => self.handle_trade(trade),
             Data::Bar(bar) => self.handle_bar(bar),
-            Data::MarkPrice(mark_price) => self.handle_mark_price(mark_price),
-            Data::IndexPrice(index_price) => self.handle_index_price(index_price),
+            Data::MarkPriceUpdate(mark_price) => self.handle_mark_price(mark_price),
+            Data::IndexPriceUpdate(index_price) => self.handle_index_price(index_price),
+            Data::InstrumentClose(close) => self.handle_instrument_close(close),
         }
     }
 
@@ -550,12 +552,12 @@ impl DataEngine {
         };
 
         let topic = switchboard::get_book_deltas_topic(deltas.instrument_id);
-        msgbus::publish(&topic, &deltas as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &deltas as &dyn Any);
     }
 
     fn handle_depth10(&mut self, depth: OrderBookDepth10) {
         let topic = switchboard::get_book_depth10_topic(depth.instrument_id);
-        msgbus::publish(&topic, &depth as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &depth as &dyn Any);
     }
 
     fn handle_quote(&mut self, quote: QuoteTick) {
@@ -566,7 +568,7 @@ impl DataEngine {
         // TODO: Handle synthetics
 
         let topic = switchboard::get_quotes_topic(quote.instrument_id);
-        msgbus::publish(&topic, &quote as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &quote as &dyn Any);
     }
 
     fn handle_trade(&mut self, trade: TradeTick) {
@@ -577,7 +579,7 @@ impl DataEngine {
         // TODO: Handle synthetics
 
         let topic = switchboard::get_trades_topic(trade.instrument_id);
-        msgbus::publish(&topic, &trade as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &trade as &dyn Any);
     }
 
     fn handle_bar(&mut self, bar: Bar) {
@@ -607,7 +609,7 @@ impl DataEngine {
         }
 
         let topic = switchboard::get_bars_topic(bar.bar_type);
-        msgbus::publish(&topic, &bar as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &bar as &dyn Any);
     }
 
     fn handle_mark_price(&mut self, mark_price: MarkPriceUpdate) {
@@ -616,7 +618,7 @@ impl DataEngine {
         }
 
         let topic = switchboard::get_mark_price_topic(mark_price.instrument_id);
-        msgbus::publish(&topic, &mark_price as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &mark_price as &dyn Any);
     }
 
     fn handle_index_price(&mut self, index_price: IndexPriceUpdate) {
@@ -630,7 +632,12 @@ impl DataEngine {
         }
 
         let topic = switchboard::get_index_price_topic(index_price.instrument_id);
-        msgbus::publish(&topic, &index_price as &dyn Any); // TODO: Optimize
+        msgbus::publish(&topic, &index_price as &dyn Any);
+    }
+
+    fn handle_instrument_close(&mut self, close: InstrumentClose) {
+        let topic = switchboard::get_instrument_close_topic(close.instrument_id);
+        msgbus::publish(&topic, &close as &dyn Any);
     }
 
     // -- SUBSCRIPTION HANDLERS -------------------------------------------------------------------

crates/data/src/engine/tests.rs

@@ -946,7 +946,7 @@ fn test_process_mark_price(
     msgbus::subscribe(topic, handler.clone(), None);
 
     let mut data_engine = data_engine.borrow_mut();
-    data_engine.process_data(Data::MarkPrice(mark_price));
+    data_engine.process_data(Data::MarkPriceUpdate(mark_price));
     let cache = &data_engine.get_cache();
     let messages = get_saved_messages::<MarkPriceUpdate>(handler);
 
@@ -1005,7 +1005,7 @@ fn test_process_index_price(
     msgbus::subscribe(topic, handler.clone(), None);
 
     let mut data_engine = data_engine.borrow_mut();
-    data_engine.process_data(Data::IndexPrice(index_price));
+    data_engine.process_data(Data::IndexPriceUpdate(index_price));
     let cache = &data_engine.get_cache();
     let messages = get_saved_messages::<IndexPriceUpdate>(handler);
 

crates/model/cbindgen.toml

@@ -70,3 +70,6 @@ exclude = [
 "UUID4" = "UUID4_t"
 "Venue" = "Venue_t"
 "VenueOrderId" = "VenueOrderId_t"
+"MarkPriceUpdate" = "MarkPriceUpdate_t"
+"IndexPriceUpdate" = "IndexPriceUpdate_t"
+"InstrumentClose" = "InstrumentClose_t"

crates/model/cbindgen_cython.toml

@@ -87,3 +87,6 @@ exclude = [
 "UUID4" = "UUID4_t"
 "Venue" = "Venue_t"
 "VenueOrderId" = "VenueOrderId_t"
+"MarkPriceUpdate" = "MarkPriceUpdate_t"
+"IndexPriceUpdate" = "IndexPriceUpdate_t"
+"InstrumentClose" = "InstrumentClose_t"

crates/model/src/data/close.rs

@@ -17,11 +17,16 @@
 
 use std::{collections::HashMap, fmt::Display, hash::Hash};
 
+use indexmap::IndexMap;
 use nautilus_core::{UnixNanos, serialization::Serializable};
 use serde::{Deserialize, Serialize};
 
 use super::GetTsInit;
-use crate::{enums::InstrumentCloseType, identifiers::InstrumentId, types::Price};
+use crate::{
+    enums::InstrumentCloseType,
+    identifiers::InstrumentId,
+    types::{Price, fixed::FIXED_SIZE_BINARY},
+};
 
 /// Represents an instrument close at a venue.
 #[repr(C)]
@@ -64,9 +69,24 @@ impl InstrumentClose {
 
     /// Returns the metadata for the type, for use with serialization formats.
     #[must_use]
-    pub fn get_metadata(instrument_id: &InstrumentId) -> HashMap<String, String> {
+    pub fn get_metadata(
+        instrument_id: &InstrumentId,
+        price_precision: u8,
+    ) -> HashMap<String, String> {
         let mut metadata = HashMap::new();
         metadata.insert("instrument_id".to_string(), instrument_id.to_string());
+        metadata.insert("price_precision".to_string(), price_precision.to_string());
+        metadata
+    }
+
+    /// Returns the field map for the type, for use with Arrow schemas.
+    #[must_use]
+    pub fn get_fields() -> IndexMap<String, String> {
+        let mut metadata = IndexMap::new();
+        metadata.insert("close_price".to_string(), FIXED_SIZE_BINARY.to_string());
+        metadata.insert("close_type".to_string(), "UInt8".to_string());
+        metadata.insert("ts_event".to_string(), "UInt64".to_string());
+        metadata.insert("ts_init".to_string(), "UInt64".to_string());
         metadata
     }
 }