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, StringBuilder, 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, extract_column,
33    extract_column_string, identifier_array_from_display,
34};
35
36const TYPE_NAME: &str = "OptionGreeks";
37
38impl ArrowSchemaProvider for OptionGreeks {
39    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
40        let fields = vec![
41            Field::new("instrument_id", DataType::Utf8, false),
42            Field::new("delta", DataType::Float64, false),
43            Field::new("gamma", DataType::Float64, false),
44            Field::new("vega", DataType::Float64, false),
45            Field::new("theta", DataType::Float64, false),
46            Field::new("rho", DataType::Float64, false),
47            Field::new("mark_iv", DataType::Float64, true),
48            Field::new("bid_iv", DataType::Float64, true),
49            Field::new("ask_iv", DataType::Float64, true),
50            Field::new("underlying_price", DataType::Float64, true),
51            Field::new("open_interest", DataType::Float64, true),
52            Field::new("ts_event", crate::arrow::timestamp_data_type(), false),
53            Field::new("ts_init", crate::arrow::timestamp_data_type(), false),
54            Field::new("convention", DataType::Utf8, false),
55            Field::new(KEY_IDENTIFIER, DataType::Utf8, true),
56        ];
57
58        let mut metadata = metadata.unwrap_or_default();
59        metadata.insert("type".to_string(), TYPE_NAME.to_string());
60        Schema::new_with_metadata(fields, metadata)
61    }
62}
63
64fn append_optional_f64(builder: &mut Float64Builder, value: Option<f64>) {
65    match value {
66        Some(value) => builder.append_value(value),
67        None => builder.append_null(),
68    }
69}
70
71fn optional_f64(values: &Float64Array, row: usize) -> Option<f64> {
72    if values.is_null(row) {
73        None
74    } else {
75        Some(values.value(row))
76    }
77}
78
79impl EncodeToRecordBatch for OptionGreeks {
80    fn encode_batch<T>(
81        metadata: &HashMap<String, String>,
82        data: &[T],
83    ) -> Result<RecordBatch, ArrowError>
84    where
85        T: std::borrow::Borrow<Self>,
86    {
87        let mut instrument_id_builder = StringBuilder::new();
88        let mut delta_builder = Float64Builder::new();
89        let mut gamma_builder = Float64Builder::new();
90        let mut vega_builder = Float64Builder::new();
91        let mut theta_builder = Float64Builder::new();
92        let mut rho_builder = Float64Builder::new();
93        let mut mark_iv_builder = Float64Builder::new();
94        let mut bid_iv_builder = Float64Builder::new();
95        let mut ask_iv_builder = Float64Builder::new();
96        let mut underlying_price_builder = Float64Builder::new();
97        let mut open_interest_builder = Float64Builder::new();
98        let mut ts_event_builder = UInt64Array::builder(data.len());
99        let mut ts_init_builder = UInt64Array::builder(data.len());
100        let mut convention_builder = StringBuilder::new();
101
102        for greeks in data.iter().map(std::borrow::Borrow::borrow) {
103            instrument_id_builder.append_value(greeks.instrument_id.to_string());
104            delta_builder.append_value(greeks.delta);
105            gamma_builder.append_value(greeks.gamma);
106            vega_builder.append_value(greeks.vega);
107            theta_builder.append_value(greeks.theta);
108            rho_builder.append_value(greeks.rho);
109            append_optional_f64(&mut mark_iv_builder, greeks.mark_iv);
110            append_optional_f64(&mut bid_iv_builder, greeks.bid_iv);
111            append_optional_f64(&mut ask_iv_builder, greeks.ask_iv);
112            append_optional_f64(&mut underlying_price_builder, greeks.underlying_price);
113            append_optional_f64(&mut open_interest_builder, greeks.open_interest);
114            ts_event_builder.append_value(greeks.ts_event.as_u64());
115            ts_init_builder.append_value(greeks.ts_init.as_u64());
116            convention_builder.append_value(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(instrument_id_builder.finish()),
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(convention_builder.finish()),
136                Arc::new(identifier_array_from_display(
137                    data.iter()
138                        .map(std::borrow::Borrow::borrow)
139                        .map(|greeks| greeks.instrument_id),
140                )),
141            ],
142        )
143    }
144
145    fn metadata(&self) -> HashMap<String, String> {
146        HashMap::from([
147            ("type".to_string(), TYPE_NAME.to_string()),
148            (
149                KEY_INSTRUMENT_ID.to_string(),
150                self.instrument_id.to_string(),
151            ),
152        ])
153    }
154}
155
156impl DecodeFromRecordBatch for OptionGreeks {
157    fn decode_batch(
158        _metadata: &HashMap<String, String>,
159        record_batch: RecordBatch,
160    ) -> Result<Vec<Self>, EncodingError> {
161        let record_batch = crate::arrow::record_batch_with_u64_timestamps(&record_batch)?;
162        let record_batch = &record_batch;
163        let cols = record_batch.columns();
164
165        let instrument_id_values = extract_column_string(cols, "instrument_id", 0)?;
166        let delta_values = extract_column::<Float64Array>(cols, "delta", 1, DataType::Float64)?;
167        let gamma_values = extract_column::<Float64Array>(cols, "gamma", 2, DataType::Float64)?;
168        let vega_values = extract_column::<Float64Array>(cols, "vega", 3, DataType::Float64)?;
169        let theta_values = extract_column::<Float64Array>(cols, "theta", 4, DataType::Float64)?;
170        let rho_values = extract_column::<Float64Array>(cols, "rho", 5, DataType::Float64)?;
171        let mark_iv_values = extract_column::<Float64Array>(cols, "mark_iv", 6, DataType::Float64)?;
172        let bid_iv_values = extract_column::<Float64Array>(cols, "bid_iv", 7, DataType::Float64)?;
173        let ask_iv_values = extract_column::<Float64Array>(cols, "ask_iv", 8, DataType::Float64)?;
174        let underlying_price_values =
175            extract_column::<Float64Array>(cols, "underlying_price", 9, DataType::Float64)?;
176        let open_interest_values =
177            extract_column::<Float64Array>(cols, "open_interest", 10, DataType::Float64)?;
178        let ts_event_values =
179            extract_column::<UInt64Array>(cols, "ts_event", 11, DataType::UInt64)?;
180        let ts_init_values = extract_column::<UInt64Array>(cols, "ts_init", 12, DataType::UInt64)?;
181        let convention_values = extract_column_string(cols, "convention", 13)?;
182
183        let result: Result<Vec<Self>, EncodingError> = (0..record_batch.num_rows())
184            .map(|row| {
185                let instrument_id = InstrumentId::from_str(instrument_id_values.value(row))
186                    .map_err(|e| EncodingError::ParseError("instrument_id", e.to_string()))?;
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
232    #[rstest]
233    fn test_encode_decode_round_trip() {
234        let instrument_id = InstrumentId::from("BTC-20260529-100000-C.OKX");
235        let original = vec![
236            OptionGreeks {
237                instrument_id,
238                convention: GreeksConvention::BlackScholes,
239                greeks: OptionGreekValues {
240                    delta: 0.55,
241                    gamma: 0.012,
242                    vega: 3.4,
243                    theta: -1.2,
244                    rho: 0.01,
245                },
246                mark_iv: Some(0.64),
247                bid_iv: Some(0.62),
248                ask_iv: Some(0.66),
249                underlying_price: Some(100_000.0),
250                open_interest: Some(42.0),
251                ts_event: 1.into(),
252                ts_init: 2.into(),
253            },
254            OptionGreeks {
255                instrument_id,
256                convention: GreeksConvention::PriceAdjusted,
257                greeks: OptionGreekValues {
258                    delta: 0.42,
259                    gamma: 0.009,
260                    vega: 2.9,
261                    theta: -0.9,
262                    rho: 0.02,
263                },
264                mark_iv: None,
265                bid_iv: None,
266                ask_iv: None,
267                underlying_price: None,
268                open_interest: None,
269                ts_event: 3.into(),
270                ts_init: 4.into(),
271            },
272        ];
273
274        let metadata = original[0].metadata();
275        let record_batch = OptionGreeks::encode_batch(&metadata, &original).unwrap();
276        let decoded = OptionGreeks::decode_batch(&metadata, record_batch).unwrap();
277
278        assert_eq!(decoded, original);
279    }
280}
nautilus_serialization/arrow/option_greeks.rs

nautilus_serialization/arrow/
option_greeks.rs
