nautilus_serialization/arrow/
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
16//! Native Arrow records for calculated Greeks and yield curves.
17
18use std::{borrow::Borrow, collections::HashMap, sync::Arc};
19
20use arrow::{
21    array::{
22        Array, ArrayRef, BooleanArray, Float64Array, Float64Builder, Int32Array, ListArray,
23        ListBuilder, StringArray, UInt64Array,
24    },
25    datatypes::{DataType, Field, Schema},
26    error::ArrowError,
27    record_batch::RecordBatch,
28};
29use nautilus_model::{
30    data::{
31        CustomData, CustomDataTrait, Data,
32        greeks::{GreeksData, OptionGreekValues, PortfolioGreeks, YieldCurve},
33    },
34    identifiers::InstrumentId,
35};
36
37use super::{
38    ArrowSchemaProvider, DecodeDataFromRecordBatch, DecodeTypedFromRecordBatch,
39    EncodeToRecordBatch, EncodingError, KEY_INSTRUMENT_ID, StringColumnRef, extract_column_by_name,
40    extract_column_string, identifier_array_from_display, metadata_with_type_name,
41    record_batch_with_timestamps, record_batch_with_u64_timestamps, timestamp_data_type,
42};
43
44impl ArrowSchemaProvider for GreeksData {
45    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
46        let mut fields = vec![
47            Field::new("ts_init", timestamp_data_type(), false),
48            Field::new("ts_event", timestamp_data_type(), false),
49            Field::new("instrument_id", DataType::Utf8, false),
50            Field::new("is_call", DataType::Boolean, false),
51            Field::new("strike", DataType::Float64, false),
52            Field::new("expiry", DataType::Int32, false),
53            Field::new("expiry_in_days", DataType::Int32, false),
54            Field::new("expiry_in_years", DataType::Float64, false),
55            Field::new("multiplier", DataType::Float64, false),
56            Field::new("quantity", DataType::Float64, false),
57            Field::new("underlying_price", DataType::Float64, false),
58            Field::new("interest_rate", DataType::Float64, false),
59            Field::new("cost_of_carry", DataType::Float64, false),
60            Field::new("vol", DataType::Float64, false),
61            Field::new("pnl", DataType::Float64, false),
62            Field::new("price", DataType::Float64, false),
63        ];
64
65        fields.extend(greek_fields());
66        fields.push(Field::new("itm_prob", DataType::Float64, false));
67        Schema::new_with_metadata(fields, metadata_with_type_name("GreeksData", metadata))
68    }
69}
70
71impl EncodeToRecordBatch for GreeksData {
72    fn encode_batch<T: Borrow<Self>>(
73        metadata: &HashMap<String, String>,
74        data: &[T],
75    ) -> Result<RecordBatch, ArrowError> {
76        let mut columns: Vec<ArrayRef> = vec![
77            Arc::new(UInt64Array::from_iter_values(
78                data.iter().map(|value| value.borrow().ts_init.as_u64()),
79            )),
80            Arc::new(UInt64Array::from_iter_values(
81                data.iter().map(|value| value.borrow().ts_event.as_u64()),
82            )),
83            Arc::new(identifier_array_from_display(
84                data.iter().map(|value| value.borrow().instrument_id),
85            )),
86            Arc::new(BooleanArray::from(
87                data.iter()
88                    .map(|value| value.borrow().is_call)
89                    .collect::<Vec<_>>(),
90            )),
91            Arc::new(Float64Array::from_iter_values(
92                data.iter().map(|value| value.borrow().strike),
93            )),
94            Arc::new(Int32Array::from_iter_values(
95                data.iter().map(|value| value.borrow().expiry),
96            )),
97            Arc::new(Int32Array::from_iter_values(
98                data.iter().map(|value| value.borrow().expiry_in_days),
99            )),
100            Arc::new(Float64Array::from_iter_values(
101                data.iter().map(|value| value.borrow().expiry_in_years),
102            )),
103            Arc::new(Float64Array::from_iter_values(
104                data.iter().map(|value| value.borrow().multiplier),
105            )),
106            Arc::new(Float64Array::from_iter_values(
107                data.iter().map(|value| value.borrow().quantity),
108            )),
109            Arc::new(Float64Array::from_iter_values(
110                data.iter().map(|value| value.borrow().underlying_price),
111            )),
112            Arc::new(Float64Array::from_iter_values(
113                data.iter().map(|value| value.borrow().interest_rate),
114            )),
115            Arc::new(Float64Array::from_iter_values(
116                data.iter().map(|value| value.borrow().cost_of_carry),
117            )),
118            Arc::new(Float64Array::from_iter_values(
119                data.iter().map(|value| value.borrow().vol),
120            )),
121            Arc::new(Float64Array::from_iter_values(
122                data.iter().map(|value| value.borrow().pnl),
123            )),
124            Arc::new(Float64Array::from_iter_values(
125                data.iter().map(|value| value.borrow().price),
126            )),
127        ];
128
129        columns.extend(greek_columns(
130            data.iter().map(|value| value.borrow().greeks),
131        ));
132        columns.push(Arc::new(Float64Array::from_iter_values(
133            data.iter().map(|value| value.borrow().itm_prob),
134        )));
135        record_batch_with_timestamps(Arc::new(Self::get_schema(Some(metadata.clone()))), columns)
136    }
137
138    fn metadata(&self) -> HashMap<String, String> {
139        let mut metadata = metadata_with_type_name("GreeksData", None);
140        metadata.insert(
141            KEY_INSTRUMENT_ID.to_string(),
142            self.instrument_id.to_string(),
143        );
144        metadata
145    }
146}
147
148impl DecodeTypedFromRecordBatch for GreeksData {
149    fn decode_typed_batch(
150        _metadata: &HashMap<String, String>,
151        batch: RecordBatch,
152    ) -> Result<Vec<Self>, EncodingError> {
153        let batch = record_batch_with_u64_timestamps(&batch)?;
154        let ts_init = required_column::<UInt64Array>(&batch, "ts_init", DataType::UInt64)?;
155        let ts_event = required_column::<UInt64Array>(&batch, "ts_event", DataType::UInt64)?;
156        let instrument_id = required_string_column(&batch, "instrument_id")?;
157        let is_call = required_column::<BooleanArray>(&batch, "is_call", DataType::Boolean)?;
158        let strike = required_column::<Float64Array>(&batch, "strike", DataType::Float64)?;
159        let expiry = required_column::<Int32Array>(&batch, "expiry", DataType::Int32)?;
160        let expiry_in_days =
161            required_column::<Int32Array>(&batch, "expiry_in_days", DataType::Int32)?;
162        let expiry_in_years =
163            required_column::<Float64Array>(&batch, "expiry_in_years", DataType::Float64)?;
164        let multiplier = required_column::<Float64Array>(&batch, "multiplier", DataType::Float64)?;
165        let quantity = required_column::<Float64Array>(&batch, "quantity", DataType::Float64)?;
166        let underlying_price =
167            required_column::<Float64Array>(&batch, "underlying_price", DataType::Float64)?;
168        let interest_rate =
169            required_column::<Float64Array>(&batch, "interest_rate", DataType::Float64)?;
170        let cost_of_carry =
171            required_column::<Float64Array>(&batch, "cost_of_carry", DataType::Float64)?;
172        let vol = required_column::<Float64Array>(&batch, "vol", DataType::Float64)?;
173        let pnl = required_column::<Float64Array>(&batch, "pnl", DataType::Float64)?;
174        let price = required_column::<Float64Array>(&batch, "price", DataType::Float64)?;
175        let itm_prob = required_column::<Float64Array>(&batch, "itm_prob", DataType::Float64)?;
176        let greeks = decode_greeks(&batch)?;
177
178        (0..batch.num_rows())
179            .map(|row| {
180                Ok(Self {
181                    instrument_id: instrument_id
182                        .value(row)
183                        .parse::<InstrumentId>()
184                        .map_err(|e| EncodingError::ParseError("instrument_id", e.to_string()))?,
185                    is_call: is_call.value(row),
186                    strike: strike.value(row),
187                    expiry: expiry.value(row),
188                    expiry_in_days: expiry_in_days.value(row),
189                    expiry_in_years: expiry_in_years.value(row),
190                    multiplier: multiplier.value(row),
191                    quantity: quantity.value(row),
192                    underlying_price: underlying_price.value(row),
193                    interest_rate: interest_rate.value(row),
194                    cost_of_carry: cost_of_carry.value(row),
195                    vol: vol.value(row),
196                    pnl: pnl.value(row),
197                    price: price.value(row),
198                    greeks: greeks[row],
199                    itm_prob: itm_prob.value(row),
200                    ts_event: ts_event.value(row).into(),
201                    ts_init: ts_init.value(row).into(),
202                })
203            })
204            .collect()
205    }
206}
207
208impl ArrowSchemaProvider for PortfolioGreeks {
209    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
210        let mut fields = vec![
211            Field::new("ts_init", timestamp_data_type(), false),
212            Field::new("ts_event", timestamp_data_type(), false),
213            Field::new("pnl", DataType::Float64, false),
214            Field::new("price", DataType::Float64, false),
215        ];
216
217        fields.extend(greek_fields());
218        Schema::new_with_metadata(fields, metadata_with_type_name("PortfolioGreeks", metadata))
219    }
220}
221
222impl EncodeToRecordBatch for PortfolioGreeks {
223    fn encode_batch<T: Borrow<Self>>(
224        metadata: &HashMap<String, String>,
225        data: &[T],
226    ) -> Result<RecordBatch, ArrowError> {
227        let mut columns: Vec<ArrayRef> = vec![
228            Arc::new(UInt64Array::from_iter_values(
229                data.iter().map(|value| value.borrow().ts_init.as_u64()),
230            )),
231            Arc::new(UInt64Array::from_iter_values(
232                data.iter().map(|value| value.borrow().ts_event.as_u64()),
233            )),
234            Arc::new(Float64Array::from_iter_values(
235                data.iter().map(|value| value.borrow().pnl),
236            )),
237            Arc::new(Float64Array::from_iter_values(
238                data.iter().map(|value| value.borrow().price),
239            )),
240        ];
241
242        columns.extend(greek_columns(
243            data.iter().map(|value| value.borrow().greeks),
244        ));
245        record_batch_with_timestamps(Arc::new(Self::get_schema(Some(metadata.clone()))), columns)
246    }
247
248    fn metadata(&self) -> HashMap<String, String> {
249        metadata_with_type_name("PortfolioGreeks", None)
250    }
251}
252
253impl DecodeTypedFromRecordBatch for PortfolioGreeks {
254    fn decode_typed_batch(
255        _metadata: &HashMap<String, String>,
256        batch: RecordBatch,
257    ) -> Result<Vec<Self>, EncodingError> {
258        let batch = record_batch_with_u64_timestamps(&batch)?;
259        let ts_init = required_column::<UInt64Array>(&batch, "ts_init", DataType::UInt64)?;
260        let ts_event = required_column::<UInt64Array>(&batch, "ts_event", DataType::UInt64)?;
261        let pnl = required_column::<Float64Array>(&batch, "pnl", DataType::Float64)?;
262        let price = required_column::<Float64Array>(&batch, "price", DataType::Float64)?;
263        let greeks = decode_greeks(&batch)?;
264
265        (0..batch.num_rows())
266            .map(|row| {
267                Ok(Self {
268                    pnl: pnl.value(row),
269                    price: price.value(row),
270                    greeks: greeks[row],
271                    ts_event: ts_event.value(row).into(),
272                    ts_init: ts_init.value(row).into(),
273                })
274            })
275            .collect()
276    }
277}
278
279impl ArrowSchemaProvider for YieldCurve {
280    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
281        let fields = vec![
282            Field::new("ts_init", timestamp_data_type(), false),
283            Field::new("ts_event", timestamp_data_type(), false),
284            Field::new("curve_name", DataType::Utf8, false),
285            Field::new("tenors", float_list_data_type(), false),
286            Field::new("interest_rates", float_list_data_type(), false),
287        ];
288        Schema::new_with_metadata(
289            fields,
290            metadata_with_type_name(Self::type_name_static(), metadata),
291        )
292    }
293}
294
295impl EncodeToRecordBatch for YieldCurve {
296    fn encode_batch<T: Borrow<Self>>(
297        metadata: &HashMap<String, String>,
298        data: &[T],
299    ) -> Result<RecordBatch, ArrowError> {
300        let columns: Vec<ArrayRef> = vec![
301            Arc::new(UInt64Array::from_iter_values(
302                data.iter().map(|value| value.borrow().ts_init.as_u64()),
303            )),
304            Arc::new(UInt64Array::from_iter_values(
305                data.iter().map(|value| value.borrow().ts_event.as_u64()),
306            )),
307            Arc::new(StringArray::from_iter_values(
308                data.iter().map(|value| value.borrow().curve_name.as_str()),
309            )),
310            Arc::new(float_lists(
311                data.iter().map(|value| value.borrow().tenors.as_slice()),
312            )),
313            Arc::new(float_lists(
314                data.iter()
315                    .map(|value| value.borrow().interest_rates.as_slice()),
316            )),
317        ];
318        record_batch_with_timestamps(Arc::new(Self::get_schema(Some(metadata.clone()))), columns)
319    }
320
321    fn metadata(&self) -> HashMap<String, String> {
322        metadata_with_type_name(Self::type_name_static(), None)
323    }
324}
325
326impl DecodeTypedFromRecordBatch for YieldCurve {
327    fn decode_typed_batch(
328        _metadata: &HashMap<String, String>,
329        batch: RecordBatch,
330    ) -> Result<Vec<Self>, EncodingError> {
331        let batch = record_batch_with_u64_timestamps(&batch)?;
332        let ts_init = required_column::<UInt64Array>(&batch, "ts_init", DataType::UInt64)?;
333        let ts_event = required_column::<UInt64Array>(&batch, "ts_event", DataType::UInt64)?;
334        let curve_name = required_string_column(&batch, "curve_name")?;
335        let tenors = required_column::<ListArray>(&batch, "tenors", float_list_data_type())?;
336        let interest_rates =
337            required_column::<ListArray>(&batch, "interest_rates", float_list_data_type())?;
338        (0..batch.num_rows())
339            .map(|row| {
340                Ok(Self {
341                    curve_name: curve_name.value(row).to_string(),
342                    tenors: decode_float_list(tenors, "tenors", row)?,
343                    interest_rates: decode_float_list(interest_rates, "interest_rates", row)?,
344                    ts_event: ts_event.value(row).into(),
345                    ts_init: ts_init.value(row).into(),
346                })
347            })
348            .collect()
349    }
350}
351
352impl DecodeDataFromRecordBatch for YieldCurve {
353    fn decode_data_batch(
354        metadata: &HashMap<String, String>,
355        record_batch: RecordBatch,
356    ) -> Result<Vec<Data>, EncodingError> {
357        Ok(Self::decode_typed_batch(metadata, record_batch)?
358            .into_iter()
359            .map(|curve| Data::Custom(CustomData::from_arc(Arc::new(curve))))
360            .collect())
361    }
362}
363
364fn greek_fields() -> Vec<Field> {
365    ["delta", "gamma", "vega", "theta", "rho"]
366        .into_iter()
367        .map(|name| Field::new(name, DataType::Float64, false))
368        .collect()
369}
370
371fn greek_columns(values: impl Iterator<Item = OptionGreekValues> + Clone) -> Vec<ArrayRef> {
372    vec![
373        Arc::new(Float64Array::from_iter_values(
374            values.clone().map(|value| value.delta),
375        )),
376        Arc::new(Float64Array::from_iter_values(
377            values.clone().map(|value| value.gamma),
378        )),
379        Arc::new(Float64Array::from_iter_values(
380            values.clone().map(|value| value.vega),
381        )),
382        Arc::new(Float64Array::from_iter_values(
383            values.clone().map(|value| value.theta),
384        )),
385        Arc::new(Float64Array::from_iter_values(
386            values.map(|value| value.rho),
387        )),
388    ]
389}
390
391fn decode_greeks(batch: &RecordBatch) -> Result<Vec<OptionGreekValues>, EncodingError> {
392    let delta = required_column::<Float64Array>(batch, "delta", DataType::Float64)?;
393    let gamma = required_column::<Float64Array>(batch, "gamma", DataType::Float64)?;
394    let vega = required_column::<Float64Array>(batch, "vega", DataType::Float64)?;
395    let theta = required_column::<Float64Array>(batch, "theta", DataType::Float64)?;
396    let rho = required_column::<Float64Array>(batch, "rho", DataType::Float64)?;
397    Ok((0..batch.num_rows())
398        .map(|row| OptionGreekValues {
399            delta: delta.value(row),
400            gamma: gamma.value(row),
401            vega: vega.value(row),
402            theta: theta.value(row),
403            rho: rho.value(row),
404        })
405        .collect())
406}
407
408fn required_column<'a, A: Array + 'static>(
409    batch: &'a RecordBatch,
410    name: &'static str,
411    data_type: DataType,
412) -> Result<&'a A, EncodingError> {
413    let column = extract_column_by_name::<A>(batch, name, data_type)?;
414
415    if column.null_count() != 0 {
416        return Err(EncodingError::ParseError(
417            name,
418            "required column contains nulls".to_string(),
419        ));
420    }
421
422    Ok(column)
423}
424
425fn required_string_column<'a>(
426    batch: &'a RecordBatch,
427    name: &'static str,
428) -> Result<StringColumnRef<'a>, EncodingError> {
429    let index = batch.schema().index_of(name)?;
430    let column = extract_column_string(batch.columns(), name, index)?;
431
432    if batch.column(index).logical_null_count() != 0 {
433        return Err(EncodingError::ParseError(
434            name,
435            "required column contains nulls".to_string(),
436        ));
437    }
438
439    Ok(column)
440}
441
442fn float_list_data_type() -> DataType {
443    DataType::List(Arc::new(Field::new("item", DataType::Float64, true)))
444}
445
446fn float_lists<'a>(values: impl Iterator<Item = &'a [f64]>) -> ListArray {
447    let mut builder = ListBuilder::new(Float64Builder::new());
448    for values in values {
449        builder.values().append_slice(values);
450        builder.append(true);
451    }
452
453    builder.finish()
454}
455
456fn decode_float_list(
457    values: &ListArray,
458    name: &'static str,
459    row: usize,
460) -> Result<Vec<f64>, EncodingError> {
461    let values = values.value(row);
462
463    let values = values
464        .as_any()
465        .downcast_ref::<Float64Array>()
466        .ok_or_else(|| {
467            EncodingError::ParseError(name, "expected Float64 list values".to_string())
468        })?;
469
470    if values.null_count() != 0 {
471        return Err(EncodingError::ParseError(
472            name,
473            format!("row {row}: list contains nulls"),
474        ));
475    }
476
477    Ok(values.values().to_vec())
478}
479
480#[cfg(test)]
481mod tests {
482    use nautilus_model::data::{decode_custom_from_arrow, encode_custom_to_arrow};
483    use rstest::rstest;
484
485    use super::*;
486
487    fn sensitivities() -> OptionGreekValues {
488        OptionGreekValues {
489            delta: 0.41,
490            gamma: 0.0012,
491            vega: 11.2,
492            theta: -2.3,
493            rho: 1.7,
494        }
495    }
496
497    fn assert_sensitivities(actual: OptionGreekValues, expected: OptionGreekValues) {
498        assert_eq!(actual.delta.to_bits(), expected.delta.to_bits());
499        assert_eq!(actual.gamma.to_bits(), expected.gamma.to_bits());
500        assert_eq!(actual.vega.to_bits(), expected.vega.to_bits());
501        assert_eq!(actual.theta.to_bits(), expected.theta.to_bits());
502        assert_eq!(actual.rho.to_bits(), expected.rho.to_bits());
503    }
504
505    #[rstest]
506    fn test_greeks_data_round_trip() {
507        let first = GreeksData::new(
508            "ESZ26 C7000.XCME".parse().unwrap(),
509            false,
510            7000.5,
511            20_261_218,
512            -68,
513            0.186,
514            50.0,
515            -3.0,
516            6950.25,
517            0.0425,
518            0.0125,
519            0.175,
520            12.5,
521            98.75,
522            sensitivities(),
523            0.62,
524            11.into(),
525            13.into(),
526        );
527        let mut second = first.clone();
528        second.instrument_id = "ESZ26 P6900.XCME".parse().unwrap();
529        second.is_call = true;
530        second.ts_init = 31.into();
531        second.ts_event = 29.into();
532        second.greeks.rho = -0.0;
533        second.greeks.delta = f64::from_bits(0x3fe0_0000_0000_0001);
534        let values = [first, second];
535        let batch = GreeksData::encode_batch(&values[0].metadata(), &values).unwrap();
536        let decoded =
537            GreeksData::decode_typed_batch(batch.schema().metadata(), batch.clone()).unwrap();
538
539        assert_eq!(
540            batch
541                .schema()
542                .field_with_name("expiry")
543                .unwrap()
544                .data_type(),
545            &DataType::Int32
546        );
547        assert_eq!(
548            batch.schema().field_with_name("rho").unwrap().data_type(),
549            &DataType::Float64
550        );
551        assert_eq!(decoded.len(), values.len());
552
553        for (actual, expected) in decoded.iter().zip(&values) {
554            assert_eq!(actual.instrument_id, expected.instrument_id);
555            assert_eq!(actual.is_call, expected.is_call);
556            assert_eq!(actual.expiry, expected.expiry);
557            assert_eq!(actual.expiry_in_days, expected.expiry_in_days);
558            assert_eq!(actual.ts_init, expected.ts_init);
559            assert_eq!(actual.ts_event, expected.ts_event);
560            macro_rules! assert_float { ($($field:ident),+) => { $(assert_eq!(actual.$field.to_bits(), expected.$field.to_bits(), stringify!($field));)+ }; }
561            assert_float!(
562                strike,
563                expiry_in_years,
564                multiplier,
565                quantity,
566                underlying_price,
567                interest_rate,
568                cost_of_carry,
569                vol,
570                pnl,
571                price,
572                itm_prob
573            );
574            assert_sensitivities(actual.greeks, expected.greeks);
575        }
576    }
577
578    #[rstest]
579    fn test_portfolio_greeks_round_trip() {
580        let values = [PortfolioGreeks {
581            pnl: -125.5,
582            price: 4310.25,
583            greeks: sensitivities(),
584            ts_event: 19.into(),
585            ts_init: 17.into(),
586        }];
587
588        let batch = PortfolioGreeks::encode_batch(&values[0].metadata(), &values).unwrap();
589        let decoded =
590            PortfolioGreeks::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
591
592        assert_eq!(decoded.len(), values.len());
593        assert_eq!(decoded[0].ts_init, values[0].ts_init);
594        assert_eq!(decoded[0].ts_event, values[0].ts_event);
595        assert_eq!(decoded[0].pnl.to_bits(), values[0].pnl.to_bits());
596        assert_eq!(decoded[0].price.to_bits(), values[0].price.to_bits());
597        assert_sensitivities(decoded[0].greeks, values[0].greeks);
598    }
599
600    #[rstest]
601    fn test_yield_curve_registered_custom_data_round_trip() {
602        let curve = YieldCurve::new(
603            "USD-SOFR".to_string(),
604            vec![0.25, 1.0, 5.0],
605            vec![0.043, 0.041, 0.038],
606            23.into(),
607            29.into(),
608        );
609        let item: Arc<dyn CustomDataTrait> = Arc::new(curve.clone());
610
611        crate::ensure_custom_data_registered::<YieldCurve>();
612        let batch = encode_custom_to_arrow("YieldCurve", &[item])
613            .unwrap()
614            .unwrap();
615        let metadata = batch.schema().metadata().clone();
616        let decoded = decode_custom_from_arrow("YieldCurve", &metadata, batch)
617            .unwrap()
618            .unwrap();
619
620        let [Data::Custom(custom)] = decoded.as_slice() else {
621            panic!("expected one custom data row, was {decoded:?}");
622        };
623
624        assert_eq!(custom.data_type.type_name(), "YieldCurve");
625        assert_eq!(
626            custom.data.as_any().downcast_ref::<YieldCurve>(),
627            Some(&curve)
628        );
629    }
630
631    #[rstest]
632    #[case(vec![0.25, 1.0, 5.0], vec![0.043, 0.041, 0.038])]
633    #[case(vec![], vec![-0.0, f64::from_bits(0x3fe0_0000_0000_0001)])]
634    fn test_yield_curve_round_trip(#[case] tenors: Vec<f64>, #[case] interest_rates: Vec<f64>) {
635        let curve = YieldCurve::new(
636            "USD-SOFR".to_string(),
637            tenors,
638            interest_rates,
639            23.into(),
640            29.into(),
641        );
642        let batch =
643            YieldCurve::encode_batch(&curve.metadata(), std::slice::from_ref(&curve)).unwrap();
644        let decoded = YieldCurve::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
645
646        assert_eq!(decoded.len(), 1);
647        assert_eq!(decoded[0].ts_init, curve.ts_init);
648        assert_eq!(decoded[0].ts_event, curve.ts_event);
649        assert_eq!(decoded[0].curve_name, curve.curve_name);
650        assert_eq!(
651            decoded[0]
652                .tenors
653                .iter()
654                .map(|v| v.to_bits())
655                .collect::<Vec<_>>(),
656            curve.tenors.iter().map(|v| v.to_bits()).collect::<Vec<_>>()
657        );
658        assert_eq!(
659            decoded[0]
660                .interest_rates
661                .iter()
662                .map(|v| v.to_bits())
663                .collect::<Vec<_>>(),
664            curve
665                .interest_rates
666                .iter()
667                .map(|v| v.to_bits())
668                .collect::<Vec<_>>()
669        );
670    }
671
672    #[rstest]
673    fn test_native_greeks_empty_batches() {
674        macro_rules! assert_empty {
675            ($type:ty) => {{
676                let batch = <$type>::encode_batch(&HashMap::new(), &Vec::<$type>::new()).unwrap();
677                let decoded =
678                    <$type>::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
679                assert!(decoded.is_empty());
680            }};
681        }
682        assert_empty!(GreeksData);
683        assert_empty!(PortfolioGreeks);
684        assert_empty!(YieldCurve);
685    }
686
687    #[rstest]
688    #[case("pnl")]
689    #[case("rho")]
690    fn test_portfolio_greeks_rejects_null_values(#[case] name: &'static str) {
691        let value = PortfolioGreeks {
692            pnl: -125.5,
693            price: 4310.25,
694            greeks: sensitivities(),
695            ts_event: 19.into(),
696            ts_init: 17.into(),
697        };
698
699        let batch = PortfolioGreeks::encode_batch(&value.metadata(), &[value]).unwrap();
700        let index = batch.schema().index_of(name).unwrap();
701        let mut fields = batch.schema().fields().to_vec();
702        fields[index] = Arc::new(fields[index].as_ref().clone().with_nullable(true));
703        let mut columns = batch.columns().to_vec();
704        columns[index] = Arc::new(Float64Array::from(vec![None::<f64>]));
705        let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
706        let error = PortfolioGreeks::decode_typed_batch(&HashMap::new(), batch).unwrap_err();
707
708        assert!(
709            matches!(error, EncodingError::ParseError(field, ref message)
710            if field == name && message == "required column contains nulls")
711        );
712    }
713
714    #[rstest]
715    #[case(Some("USD-SOFR"))]
716    #[case(None)]
717    fn test_yield_curve_dictionary_name(#[case] name: Option<&str>) {
718        use arrow::{
719            array::{DictionaryArray, Int8Array},
720            datatypes::Int8Type,
721        };
722
723        let value = YieldCurve::new(
724            "USD-SOFR".to_string(),
725            vec![0.25],
726            vec![0.043],
727            23.into(),
728            29.into(),
729        );
730        let batch = YieldCurve::encode_batch(&value.metadata(), &[value]).unwrap();
731        let index = batch.schema().index_of("curve_name").unwrap();
732        let names = DictionaryArray::<Int8Type>::try_new(
733            Int8Array::from(vec![0]),
734            Arc::new(StringArray::from(vec![name])),
735        )
736        .unwrap();
737        let mut fields = batch.schema().fields().to_vec();
738        fields[index] = Arc::new(Field::new("curve_name", names.data_type().clone(), true));
739        let mut columns = batch.columns().to_vec();
740        columns[index] = Arc::new(names);
741        let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
742        let decoded = YieldCurve::decode_typed_batch(&HashMap::new(), batch);
743
744        match name {
745            Some(name) => assert_eq!(decoded.unwrap()[0].curve_name, name),
746            None => assert!(
747                matches!(decoded.unwrap_err(), EncodingError::ParseError("curve_name", ref message) if message == "required column contains nulls")
748            ),
749        }
750    }
751
752    #[rstest]
753    #[case("curve_name")]
754    #[case("tenors")]
755    #[case("interest_rates")]
756    fn test_yield_curve_rejects_null_values(#[case] name: &'static str) {
757        let value = YieldCurve::new(
758            "USD-SOFR".to_string(),
759            vec![0.25],
760            vec![0.043],
761            23.into(),
762            29.into(),
763        );
764        let batch = YieldCurve::encode_batch(&value.metadata(), &[value]).unwrap();
765        let index = batch.schema().index_of(name).unwrap();
766        let mut fields = batch.schema().fields().to_vec();
767        let mut columns = batch.columns().to_vec();
768
769        let expected = if name == "curve_name" {
770            fields[index] = Arc::new(fields[index].as_ref().clone().with_nullable(true));
771            columns[index] = Arc::new(StringArray::from(vec![None::<&str>]));
772            "required column contains nulls"
773        } else {
774            let mut builder = ListBuilder::new(Float64Builder::new());
775            builder.values().append_null();
776            builder.append(true);
777            columns[index] = Arc::new(builder.finish());
778            "row 0: list contains nulls"
779        };
780
781        let batch = RecordBatch::try_new(Arc::new(Schema::new(fields)), columns).unwrap();
782        let error = YieldCurve::decode_typed_batch(&HashMap::new(), batch).unwrap_err();
783
784        assert!(
785            matches!(error, EncodingError::ParseError(field, ref message)
786            if field == name && message == expected)
787        );
788    }
789}
nautilus_serialization/arrow/greeks.rs

nautilus_serialization/arrow/
greeks.rs
