nautilus_serialization/arrow/
option_greeks.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::{collections::HashMap, str::FromStr, sync::Arc};
17
18use arrow::{
19    array::{Array, Float64Array, Float64Builder, UInt64Array},
20    datatypes::{DataType, Field, Schema},
21    error::ArrowError,
22    record_batch::RecordBatch,
23};
24use nautilus_model::{
25    data::{Data, greeks::OptionGreekValues, option_chain::OptionGreeks},
26    enums::GreeksConvention,
27    identifiers::InstrumentId,
28};
29
30use super::{
31    ArrowSchemaProvider, DecodeDataFromRecordBatch, DecodeFromRecordBatch, EncodeToRecordBatch,
32    EncodingError, KEY_IDENTIFIER, KEY_INSTRUMENT_ID, decode_required_timestamp,
33    enum_dictionary_array, enum_dictionary_data_type, extract_column, extract_column_string,
34    identifier_array_from_display, json::metadata_for_type, metadata_with_type_name,
35};
36
37const TYPE_NAME: &str = "OptionGreeks";
38
39impl ArrowSchemaProvider for OptionGreeks {
40    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
41        let fields = vec![
42            Field::new("convention", enum_dictionary_data_type(), false),
43            Field::new("delta", DataType::Float64, false),
44            Field::new("gamma", DataType::Float64, false),
45            Field::new("vega", DataType::Float64, false),
46            Field::new("theta", DataType::Float64, false),
47            Field::new("rho", DataType::Float64, false),
48            Field::new("mark_iv", DataType::Float64, true),
49            Field::new("bid_iv", DataType::Float64, true),
50            Field::new("ask_iv", DataType::Float64, true),
51            Field::new("underlying_price", DataType::Float64, true),
52            Field::new("open_interest", DataType::Float64, true),
53            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
54            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
55            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
56        ];
57
58        Schema::new_with_metadata(fields, metadata_with_type_name(TYPE_NAME, metadata))
59    }
60}
61
62fn append_optional_f64(builder: &mut Float64Builder, value: Option<f64>) {
63    match value {
64        Some(value) => builder.append_value(value),
65        None => builder.append_null(),
66    }
67}
68
69fn optional_f64(values: &Float64Array, row: usize) -> Option<f64> {
70    if values.is_null(row) {
71        None
72    } else {
73        Some(values.value(row))
74    }
75}
76
77impl EncodeToRecordBatch for OptionGreeks {
78    fn encode_batch<T>(
79        metadata: &HashMap<String, String>,
80        data: &[T],
81    ) -> Result<RecordBatch, ArrowError>
82    where
83        T: std::borrow::Borrow<Self>,
84    {
85        let mut delta_builder = Float64Builder::new();
86        let mut gamma_builder = Float64Builder::new();
87        let mut vega_builder = Float64Builder::new();
88        let mut theta_builder = Float64Builder::new();
89        let mut rho_builder = Float64Builder::new();
90        let mut mark_iv_builder = Float64Builder::new();
91        let mut bid_iv_builder = Float64Builder::new();
92        let mut ask_iv_builder = Float64Builder::new();
93        let mut underlying_price_builder = Float64Builder::new();
94        let mut open_interest_builder = Float64Builder::new();
95        let mut ts_event_builder = UInt64Array::builder(data.len());
96        let mut ts_init_builder = UInt64Array::builder(data.len());
97
98        for greeks in data.iter().map(std::borrow::Borrow::borrow) {
99            delta_builder.append_value(greeks.delta);
100            gamma_builder.append_value(greeks.gamma);
101            vega_builder.append_value(greeks.vega);
102            theta_builder.append_value(greeks.theta);
103            rho_builder.append_value(greeks.rho);
104            append_optional_f64(&mut mark_iv_builder, greeks.mark_iv);
105            append_optional_f64(&mut bid_iv_builder, greeks.bid_iv);
106            append_optional_f64(&mut ask_iv_builder, greeks.ask_iv);
107            append_optional_f64(&mut underlying_price_builder, greeks.underlying_price);
108            append_optional_f64(&mut open_interest_builder, greeks.open_interest);
109            ts_event_builder.append_value(greeks.ts_event.as_u64());
110            ts_init_builder.append_value(greeks.ts_init.as_u64());
111        }
112
113        let conventions = enum_dictionary_array(
114            data.iter()
115                .map(std::borrow::Borrow::borrow)
116                .map(|greeks| greeks.convention),
117        )?;
118
119        crate::arrow::record_batch_with_timestamps(
120            Arc::new(Self::get_schema(Some(metadata.clone()))),
121            vec![
122                Arc::new(conventions),
123                Arc::new(delta_builder.finish()),
124                Arc::new(gamma_builder.finish()),
125                Arc::new(vega_builder.finish()),
126                Arc::new(theta_builder.finish()),
127                Arc::new(rho_builder.finish()),
128                Arc::new(mark_iv_builder.finish()),
129                Arc::new(bid_iv_builder.finish()),
130                Arc::new(ask_iv_builder.finish()),
131                Arc::new(underlying_price_builder.finish()),
132                Arc::new(open_interest_builder.finish()),
133                Arc::new(ts_event_builder.finish()),
134                Arc::new(ts_init_builder.finish()),
135                Arc::new(identifier_array_from_display(
136                    data.iter()
137                        .map(std::borrow::Borrow::borrow)
138                        .map(|greeks| greeks.instrument_id),
139                )),
140            ],
141        )
142    }
143
144    fn metadata(&self) -> HashMap<String, String> {
145        let mut metadata = metadata_for_type(TYPE_NAME);
146        metadata.insert(
147            KEY_INSTRUMENT_ID.to_string(),
148            self.instrument_id.to_string(),
149        );
150        metadata
151    }
152}
153
154impl DecodeFromRecordBatch for OptionGreeks {
155    fn decode_batch(
156        metadata: &HashMap<String, String>,
157        record_batch: RecordBatch,
158    ) -> Result<Vec<Self>, EncodingError> {
159        let instrument_id = metadata
160            .get(KEY_INSTRUMENT_ID)
161            .ok_or(EncodingError::MissingMetadata(KEY_INSTRUMENT_ID))?
162            .parse::<InstrumentId>()
163            .map_err(|e| EncodingError::ParseError(KEY_INSTRUMENT_ID, e.to_string()))?;
164        let record_batch = crate::arrow::record_batch_with_u64_timestamps(&record_batch)?;
165        let record_batch = &record_batch;
166        let cols = record_batch.columns();
167
168        let convention_values = extract_column_string(cols, "convention", 0)?;
169        let delta_values = extract_column::<Float64Array>(cols, "delta", 1, DataType::Float64)?;
170        let gamma_values = extract_column::<Float64Array>(cols, "gamma", 2, DataType::Float64)?;
171        let vega_values = extract_column::<Float64Array>(cols, "vega", 3, DataType::Float64)?;
172        let theta_values = extract_column::<Float64Array>(cols, "theta", 4, DataType::Float64)?;
173        let rho_values = extract_column::<Float64Array>(cols, "rho", 5, DataType::Float64)?;
174        let mark_iv_values = extract_column::<Float64Array>(cols, "mark_iv", 6, DataType::Float64)?;
175        let bid_iv_values = extract_column::<Float64Array>(cols, "bid_iv", 7, DataType::Float64)?;
176        let ask_iv_values = extract_column::<Float64Array>(cols, "ask_iv", 8, DataType::Float64)?;
177        let underlying_price_values =
178            extract_column::<Float64Array>(cols, "underlying_price", 9, DataType::Float64)?;
179        let open_interest_values =
180            extract_column::<Float64Array>(cols, "open_interest", 10, DataType::Float64)?;
181        let ts_event_values =
182            extract_column::<UInt64Array>(cols, "ts_event", 11, DataType::UInt64)?;
183        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 12, DataType::UInt64)?;
184
185        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
186            .map(|row| {
187                let convention = GreeksConvention::from_str(convention_values.value(row))
188                    .map_err(|e| EncodingError::ParseError("convention", e.to_string()))?;
189
190                Ok(Self {
191                    instrument_id,
192                    convention,
193                    greeks: OptionGreekValues {
194                        delta: delta_values.value(row),
195                        gamma: gamma_values.value(row),
196                        vega: vega_values.value(row),
197                        theta: theta_values.value(row),
198                        rho: rho_values.value(row),
199                    },
200                    mark_iv: optional_f64(mark_iv_values, row),
201                    bid_iv: optional_f64(bid_iv_values, row),
202                    ask_iv: optional_f64(ask_iv_values, row),
203                    underlying_price: optional_f64(underlying_price_values, row),
204                    open_interest: optional_f64(open_interest_values, row),
205                    ts_event: decode_required_timestamp(ts_event_values, "ts_event", row)?,
206                    ts_init: decode_required_timestamp(ts_init_values, "ts_init", row)?,
207                })
208            })
209            .collect();
210
211        result
212    }
213}
214
215impl DecodeDataFromRecordBatch for OptionGreeks {
216    fn decode_data_batch(
217        metadata: &HashMap<String, String>,
218        record_batch: RecordBatch,
219    ) -> Result<Vec<Data>, EncodingError> {
220        let greeks = Self::decode_batch(metadata, record_batch)?;
221        Ok(greeks.into_iter().map(Data::from).collect())
222    }
223}
224
225#[cfg(test)]
226mod tests {
227    use nautilus_model::{enums::GreeksConvention, identifiers::InstrumentId};
228    use rstest::rstest;
229
230    use super::*;
231    use crate::arrow::KEY_TYPE_NAME;
232
233    #[rstest]
234    fn test_encode_decode_round_trip() {
235        let instrument_id = InstrumentId::from("BTC-20260529-100000-C.OKX");
236        let original = vec![
237            OptionGreeks {
238                instrument_id,
239                convention: GreeksConvention::BlackScholes,
240                greeks: OptionGreekValues {
241                    delta: 0.55,
242                    gamma: 0.012,
243                    vega: 3.4,
244                    theta: -1.2,
245                    rho: 0.01,
246                },
247                mark_iv: Some(0.64),
248                bid_iv: Some(0.62),
249                ask_iv: Some(0.66),
250                underlying_price: Some(100_000.0),
251                open_interest: Some(42.0),
252                ts_event: 1.into(),
253                ts_init: 2.into(),
254            },
255            OptionGreeks {
256                instrument_id,
257                convention: GreeksConvention::PriceAdjusted,
258                greeks: OptionGreekValues {
259                    delta: 0.42,
260                    gamma: 0.009,
261                    vega: 2.9,
262                    theta: -0.9,
263                    rho: 0.02,
264                },
265                mark_iv: None,
266                bid_iv: None,
267                ask_iv: None,
268                underlying_price: None,
269                open_interest: None,
270                ts_event: 3.into(),
271                ts_init: 4.into(),
272            },
273        ];
274
275        let metadata = original[0].metadata();
276        let record_batch = OptionGreeks::encode_batch(&metadata, &original).unwrap();
277        let schema = record_batch.schema();
278        let decoded = OptionGreeks::decode_batch(schema.metadata(), record_batch).unwrap();
279
280        assert_eq!(
281            schema
282                .fields()
283                .iter()
284                .map(|field| field.name().as_str())
285                .collect::<Vec<_>>(),
286            vec![
287                "convention",
288                "delta",
289                "gamma",
290                "vega",
291                "theta",
292                "rho",
293                "mark_iv",
294                "bid_iv",
295                "ask_iv",
296                "underlying_price",
297                "open_interest",
298                "ts_event",
299                "ts_init",
300                KEY_IDENTIFIER,
301            ]
302        );
303        assert_eq!(schema.field(0).data_type(), &enum_dictionary_data_type());
304        assert_eq!(
305            schema.metadata(),
306            &HashMap::from([
307                (KEY_TYPE_NAME.to_string(), "OptionGreeks".to_string()),
308                (KEY_INSTRUMENT_ID.to_string(), instrument_id.to_string()),
309            ])
310        );
311        assert_eq!(decoded, original);
312    }
313}
nautilus_serialization/arrow/option_greeks.rs

nautilus_serialization/arrow/
option_greeks.rs
