Add msgbus OptionGreeks binary codecs

- Add SBE and Cap'n Proto schemas for `OptionGreeks`
- Route `OptionGreeks` egress and ingress through binary codecs
- Reorder `OptionGreeks` wire and Python enum positions
- Cover binary egress, republish, and optional-field round trips
- Call out wire and Python enum breaks in release notes

Author: cjdsellers

RELEASES.md

@@ -20,6 +20,7 @@ This release includes many breaking changes across the user-facing Rust v2 APIs.
 - Added `RedisCacheConfig`, `PostgresCacheConfig`, and `RedisMessageBusConfig` for Rust factories
 - Added SEC1 EC private key support to socket TLS configuration (Rust)
 - Added SBE and Cap'n Proto encodings for Rust-native message bus publishers
+- Added SBE and Cap'n Proto external msgbus payload support for `OptionGreeks`
 - Added `order_position_index` Postgres table for the order-position index; run `make init-db` to migrate
 - Added negative price support for `Commodity` instruments in risk checks (#2330), thanks for reporting @fabz1
 - Added `add_native_exec_algorithm` and `ExecutionAlgorithmConfig` bindings to the Python v2 backtest engine
@@ -59,6 +60,7 @@ This release includes many breaking changes across the user-facing Rust v2 APIs.
 - Changed Parquet catalog write APIs to take borrowed slices instead of owned `Vec` (Rust) (#4296), thanks @sunlei
 - Changed `PoolProfiler.price_sqrt_ratio_x96` to return `int` instead of `str`
 - Changed PyO3 `DataActor`/`Strategy` historical request `start`/`end` to require UTC datetimes
+- Changed Python `NautilusDataType` enum order to put `OptionGreeks` before `InstrumentStatus`
 - Changed Redis cache account/order/position storage to event logs; clear old typed state (Rust)
 - Changed cache database and message bus backing construction to use factory-owned config structs (Rust)
 - Changed Rust actor `self.clock()` to return `ClockApi`; call methods directly instead of borrowing
@@ -68,6 +70,8 @@ This release includes many breaking changes across the user-facing Rust v2 APIs.
 - Changed `OrderAny::from_events` to return `OrderReplayError` instead of `anyhow::Error` (Rust)
 - Changed `OrderList::validate` to return `OrderListValidationError` instead of `anyhow::Error` (Rust)
 - Changed Rust message bus subscriber-count and presence queries to return invalid-topic errors instead of panicking
+- Changed Cap'n Proto `DataAny` ordinals to put `OptionGreeks` before instrument schemas
+- Changed SBE `DataAny` variants and template IDs to put `OptionGreeks` before instrument schemas
 - Changed `SyntheticInstrument` fallible methods to return `SyntheticInstrumentError` instead of `anyhow::Error` (Rust)
 - Changed tick scheme constructors and parsing to return `TickSchemeError` instead of `anyhow::Error` (Rust)
 - Changed WebSocket and socket `reconnect_timeout_ms` to bound only connection establishment (Rust)

crates/backtest/src/python/engine.rs

@@ -1191,14 +1191,14 @@ fn pyobject_to_data(_py: Python, obj: &Bound<'_, PyAny>) -> PyResult<Data> {
         return Ok(Data::FundingRateUpdate(funding_rate));
     }
 
-    if let Ok(status) = obj.extract::<InstrumentStatus>() {
-        return Ok(Data::InstrumentStatus(status));
-    }
-
     if let Ok(greeks) = obj.extract::<OptionGreeks>() {
         return Ok(Data::OptionGreeks(greeks));
     }
 
+    if let Ok(status) = obj.extract::<InstrumentStatus>() {
+        return Ok(Data::InstrumentStatus(status));
+    }
+
     if let Ok(close) = obj.extract::<InstrumentClose>() {
         return Ok(Data::InstrumentClose(close));
     }
@@ -1237,14 +1237,14 @@ fn pyobject_to_data(_py: Python, obj: &Bound<'_, PyAny>) -> PyResult<Data> {
         return Ok(Data::FundingRateUpdate(funding_rate));
     }
 
-    if let Ok(status) = InstrumentStatus::from_pyobject(obj) {
-        return Ok(Data::InstrumentStatus(status));
-    }
-
     if let Ok(greeks) = OptionGreeks::from_pyobject(obj) {
         return Ok(Data::OptionGreeks(greeks));
     }
 
+    if let Ok(status) = InstrumentStatus::from_pyobject(obj) {
+        return Ok(Data::InstrumentStatus(status));
+    }
+
     if let Ok(close) = InstrumentClose::from_pyobject(obj) {
         return Ok(Data::InstrumentClose(close));
     }

crates/common/src/msgbus/api.rs

@@ -1616,6 +1616,8 @@ mod tests {
     use nautilus_model::defi::{
         AmmType, Chain, Dex, DexType, PoolIdentifier, PoolLiquidityUpdateType, Token,
     };
+    #[cfg(any(feature = "sbe", feature = "capnp"))]
+    use nautilus_model::{data::OptionGreekValues, enums::GreeksConvention};
     use nautilus_model::{
         data::{
             Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks,
@@ -1949,6 +1951,28 @@ mod tests {
         )
     }
 
+    #[cfg(any(feature = "sbe", feature = "capnp"))]
+    fn option_greeks() -> OptionGreeks {
+        OptionGreeks {
+            instrument_id: InstrumentId::from("BTC-30JUN23-40000-C.DERIBIT"),
+            convention: GreeksConvention::PriceAdjusted,
+            greeks: OptionGreekValues {
+                delta: 0.525,
+                gamma: 0.00032,
+                vega: 12.25,
+                theta: -0.72,
+                rho: 0.18,
+            },
+            mark_iv: Some(0.0),
+            bid_iv: None,
+            ask_iv: Some(0.54),
+            underlying_price: Some(41_500.25),
+            open_interest: Some(0.0),
+            ts_event: UnixNanos::from(20),
+            ts_init: UnixNanos::from(21),
+        }
+    }
+
     fn portfolio_snapshot() -> PortfolioSnapshot {
         PortfolioSnapshot::new(
             AccountId::from("SIM-001"),
@@ -2529,6 +2553,15 @@ mod tests {
             subscribe_funding_rates,
             assert_eq_ref,
         );
+        assert_typed_external_round_trips(
+            encoding,
+            BusPayloadType::OptionGreeks,
+            "data.option_greeks.BTC-30JUN23-40000-C.DERIBIT",
+            option_greeks(),
+            publish_option_greeks,
+            subscribe_option_greeks,
+            assert_eq_ref,
+        );
     }
 
     #[cfg(feature = "sbe")]
@@ -2545,7 +2578,6 @@ mod tests {
 
     #[rstest]
     #[case(BusPayloadType::AccountState)]
-    #[case(BusPayloadType::OptionGreeks)]
     fn republish_external_message_skips_unsupported_binary_payload(
         #[case] payload_type: BusPayloadType,
     ) {
@@ -2560,16 +2592,6 @@ mod tests {
             }),
             None,
         );
-        let greeks_received = received.clone();
-        subscribe_option_greeks(
-            "events.unsupported.*".into(),
-            TypedHandler::from(move |greeks: &OptionGreeks| {
-                greeks_received
-                    .borrow_mut()
-                    .push(serde_json::to_value(greeks).unwrap());
-            }),
-            None,
-        );
 
         for encoding in [SerializationEncoding::Sbe, SerializationEncoding::Capnp] {
             let message = BusMessage::with_str_topic(
@@ -2631,6 +2653,26 @@ mod tests {
         reset_message_bus();
     }
 
+    #[cfg(feature = "sbe")]
+    #[rstest]
+    fn publish_option_greeks_sbe_forwards_decodable_payload_to_external_egress() {
+        let publications = install_capturing_external_egress(SerializationEncoding::Sbe);
+        let greeks = option_greeks();
+
+        publish_option_greeks("data.option_greeks.TEST".into(), &greeks);
+
+        let publications = publications.borrow();
+        assert_eq!(publications.len(), 1);
+        assert_eq!(publications[0].topic, "data.option_greeks.TEST");
+        assert_eq!(
+            OptionGreeks::from_sbe(&publications[0].payload)
+                .expect("SBE payload must decode as OptionGreeks"),
+            greeks
+        );
+        drop(publications);
+        reset_message_bus();
+    }
+
     #[cfg(not(feature = "sbe"))]
     #[rstest]
     fn publish_quote_sbe_without_feature_drops_payload() {
@@ -2669,6 +2711,32 @@ mod tests {
         reset_message_bus();
     }
 
+    #[cfg(feature = "capnp")]
+    #[rstest]
+    fn publish_option_greeks_capnp_forwards_decodable_payload_to_external_egress() {
+        let publications = install_capturing_external_egress(SerializationEncoding::Capnp);
+        let greeks = option_greeks();
+
+        publish_option_greeks("data.option_greeks.TEST".into(), &greeks);
+
+        let publications = publications.borrow();
+        assert_eq!(publications.len(), 1);
+        assert_eq!(publications[0].topic, "data.option_greeks.TEST");
+        let reader = capnp::serialize::read_message(
+            &mut &publications[0].payload[..],
+            capnp::message::ReaderOptions::new(),
+        )
+        .expect("Cap'n Proto payload must be readable");
+        let root = reader
+            .get_root::<market_capnp::option_greeks::Reader>()
+            .expect("Cap'n Proto payload must have an OptionGreeks root");
+        let decoded = OptionGreeks::from_capnp(root)
+            .expect("Cap'n Proto payload must decode as OptionGreeks");
+        assert_eq!(decoded, greeks);
+        drop(publications);
+        reset_message_bus();
+    }
+
     #[cfg(not(feature = "capnp"))]
     #[rstest]
     fn publish_quote_capnp_without_feature_drops_payload() {

crates/common/src/msgbus/external/codec/capnp.rs

@@ -18,8 +18,8 @@ use std::any::Any;
 use anyhow::Context;
 use bytes::Bytes;
 use nautilus_model::data::{
-    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas, OrderBookDepth10,
-    QuoteTick, TradeTick,
+    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks, OrderBookDeltas,
+    OrderBookDepth10, QuoteTick, TradeTick,
 };
 use nautilus_serialization::{
     capnp::{FromCapnp, ToCapnp},
@@ -95,6 +95,12 @@ define_deserializer!(
     "FundingRateUpdate",
     market_capnp::funding_rate_update::Reader
 );
+define_deserializer!(
+    deserialize_option_greeks,
+    OptionGreeks,
+    "OptionGreeks",
+    market_capnp::option_greeks::Reader
+);
 
 macro_rules! serialize_payload_as {
     ($message:expr, $type_name:expr, $ty:ty, $root:ty) => {{
@@ -171,6 +177,12 @@ pub(super) fn serialize_payload(
             FundingRateUpdate,
             market_capnp::funding_rate_update::Builder
         ),
+        BusPayloadType::OptionGreeks => serialize_payload_as!(
+            message,
+            type_name,
+            OptionGreeks,
+            market_capnp::option_greeks::Builder
+        ),
         _ => Err(PayloadCodecError::Dropped(format!(
             "Cap'n Proto serialization is not supported for {type_name}"
         ))),

crates/common/src/msgbus/external/codec/capnp_unavailable.rs

@@ -17,8 +17,8 @@ use std::any::Any;
 
 use bytes::Bytes;
 use nautilus_model::data::{
-    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas, OrderBookDepth10,
-    QuoteTick, TradeTick,
+    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks, OrderBookDeltas,
+    OrderBookDepth10, QuoteTick, TradeTick,
 };
 
 use super::PayloadCodecError;
@@ -40,6 +40,7 @@ define_deserializer!(deserialize_bar, Bar);
 define_deserializer!(deserialize_mark_price, MarkPriceUpdate);
 define_deserializer!(deserialize_index_price, IndexPriceUpdate);
 define_deserializer!(deserialize_funding_rate, FundingRateUpdate);
+define_deserializer!(deserialize_option_greeks, OptionGreeks);
 
 pub(super) fn serialize_payload(
     payload_type: BusPayloadType,

crates/common/src/msgbus/external/codec/mod.rs

@@ -19,8 +19,8 @@ use std::any::Any;
 
 use bytes::Bytes;
 use nautilus_model::data::{
-    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas, OrderBookDepth10,
-    QuoteTick, TradeTick,
+    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks, OrderBookDeltas,
+    OrderBookDepth10, QuoteTick, TradeTick,
 };
 use serde::de::DeserializeOwned;
 
@@ -127,6 +127,7 @@ define_market_data_deserializer!(
     FundingRateUpdate,
     FundingRateUpdate
 );
+define_market_data_deserializer!(deserialize_option_greeks, OptionGreeks, OptionGreeks);
 
 fn deserialize_market_data_payload<T>(
     payload_type: BusPayloadType,

crates/common/src/msgbus/external/codec/sbe.rs

@@ -17,8 +17,8 @@ use std::any::Any;
 
 use bytes::Bytes;
 use nautilus_model::data::{
-    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas, OrderBookDepth10,
-    QuoteTick, TradeTick,
+    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks, OrderBookDeltas,
+    OrderBookDepth10, QuoteTick, TradeTick,
 };
 use nautilus_serialization::sbe::{FromSbe, ToSbe};
 
@@ -64,6 +64,7 @@ define_deserializer!(
     FundingRateUpdate,
     "FundingRateUpdate"
 );
+define_deserializer!(deserialize_option_greeks, OptionGreeks, "OptionGreeks");
 
 pub(super) fn serialize_payload(
     payload_type: BusPayloadType,
@@ -89,6 +90,7 @@ pub(super) fn serialize_payload(
         BusPayloadType::FundingRateUpdate => {
             serialize_payload_as::<FundingRateUpdate>(type_name, message)
         }
+        BusPayloadType::OptionGreeks => serialize_payload_as::<OptionGreeks>(type_name, message),
         _ => Err(PayloadCodecError::Dropped(format!(
             "SBE serialization is not supported for {type_name}"
         ))),

crates/common/src/msgbus/external/codec/sbe_unavailable.rs

@@ -17,8 +17,8 @@ use std::any::Any;
 
 use bytes::Bytes;
 use nautilus_model::data::{
-    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OrderBookDeltas, OrderBookDepth10,
-    QuoteTick, TradeTick,
+    Bar, FundingRateUpdate, IndexPriceUpdate, MarkPriceUpdate, OptionGreeks, OrderBookDeltas,
+    OrderBookDepth10, QuoteTick, TradeTick,
 };
 
 use super::PayloadCodecError;
@@ -40,6 +40,7 @@ define_deserializer!(deserialize_bar, Bar);
 define_deserializer!(deserialize_mark_price, MarkPriceUpdate);
 define_deserializer!(deserialize_index_price, IndexPriceUpdate);
 define_deserializer!(deserialize_funding_rate, FundingRateUpdate);
+define_deserializer!(deserialize_option_greeks, OptionGreeks);
 
 pub(super) fn serialize_payload(
     payload_type: BusPayloadType,

crates/common/src/msgbus/external/mod.rs

@@ -190,11 +190,11 @@ pub fn republish_external_message(message: &BusMessage) -> anyhow::Result<()> {
             publish_funding_rate,
         )?,
         BusPayloadType::OptionGreeks => {
-            handle_json_msgpack(
+            handle_market_data(
                 topic,
-                message.payload_type,
                 message.encoding,
                 &message.payload,
+                codec::deserialize_option_greeks,
                 publish_option_greeks,
             )?;
         }

crates/common/src/msgbus/message.rs

@@ -196,6 +196,7 @@ impl BusPayloadType {
                 | Self::MarkPriceUpdate
                 | Self::IndexPriceUpdate
                 | Self::FundingRateUpdate
+                | Self::OptionGreeks
         )
     }
 }
@@ -357,7 +358,7 @@ mod tests {
     #[case(BusPayloadType::AccountState, BusPayloadCategory::BuiltIn)]
     #[case(BusPayloadType::OrderEvent, BusPayloadCategory::BuiltIn)]
     #[case(BusPayloadType::Instrument, BusPayloadCategory::Other)]
-    #[case(BusPayloadType::OptionGreeks, BusPayloadCategory::Other)]
+    #[case(BusPayloadType::OptionGreeks, BusPayloadCategory::MarketData)]
     #[case(
         BusPayloadType::Custom(Ustr::from("CustomPayload")),
         BusPayloadCategory::Other
@@ -391,7 +392,22 @@ mod tests {
     #[case(BusPayloadType::AccountState, SerializationEncoding::Json, true)]
     #[case(BusPayloadType::AccountState, SerializationEncoding::Capnp, false)]
     #[case(BusPayloadType::Instrument, SerializationEncoding::Sbe, false)]
-    #[case(BusPayloadType::OptionGreeks, SerializationEncoding::Capnp, false)]
+    #[cfg_attr(
+        feature = "sbe",
+        case(BusPayloadType::OptionGreeks, SerializationEncoding::Sbe, true)
+    )]
+    #[cfg_attr(
+        not(feature = "sbe"),
+        case(BusPayloadType::OptionGreeks, SerializationEncoding::Sbe, false)
+    )]
+    #[cfg_attr(
+        feature = "capnp",
+        case(BusPayloadType::OptionGreeks, SerializationEncoding::Capnp, true)
+    )]
+    #[cfg_attr(
+        not(feature = "capnp"),
+        case(BusPayloadType::OptionGreeks, SerializationEncoding::Capnp, false)
+    )]
     #[case(
         BusPayloadType::Custom(Ustr::from("CustomPayload")),
         SerializationEncoding::MsgPack,