nautilus_serialization/arrow/
funding.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;
17
18use arrow::{datatypes::Schema, error::ArrowError, record_batch::RecordBatch};
19use nautilus_model::data::{Data, FundingRateUpdate};
20
21use super::{
22    ArrowSchemaProvider, DecodeDataFromRecordBatch, DecodeFromRecordBatch, EncodeToRecordBatch,
23    EncodingError, KEY_INSTRUMENT_ID,
24    json::{
25        JsonFieldSpec, decode_batch_with_metadata_fields, encode_batch_with_identifier,
26        metadata_for_type, schema_for_type_with_identifier,
27    },
28};
29
30const FUNDING_RATE_UPDATE_FIELDS: &[JsonFieldSpec] = &[
31    JsonFieldSpec::utf8("rate", false),
32    JsonFieldSpec::u64("interval", true),
33    JsonFieldSpec::timestamp("next_funding_ns", true),
34    JsonFieldSpec::timestamp("ts_event", false),
35    JsonFieldSpec::timestamp("ts_init", false),
36];
37
38impl ArrowSchemaProvider for FundingRateUpdate {
39    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
40        schema_for_type_with_identifier("FundingRateUpdate", metadata, FUNDING_RATE_UPDATE_FIELDS)
41    }
42}
43
44impl EncodeToRecordBatch for FundingRateUpdate {
45    fn encode_batch<T>(
46        metadata: &HashMap<String, String>,
47        data: &[T],
48    ) -> Result<RecordBatch, ArrowError>
49    where
50        T: std::borrow::Borrow<Self>,
51    {
52        encode_batch_with_identifier(
53            "FundingRateUpdate",
54            metadata,
55            data.iter().map(std::borrow::Borrow::borrow),
56            FUNDING_RATE_UPDATE_FIELDS,
57            data.iter()
58                .map(std::borrow::Borrow::borrow)
59                .map(|update| update.instrument_id),
60        )
61    }
62
63    fn metadata(&self) -> HashMap<String, String> {
64        let mut metadata = metadata_for_type("FundingRateUpdate");
65        metadata.insert(
66            KEY_INSTRUMENT_ID.to_string(),
67            self.instrument_id.to_string(),
68        );
69        metadata
70    }
71}
72
73impl DecodeFromRecordBatch for FundingRateUpdate {
74    fn decode_batch(
75        metadata: &HashMap<String, String>,
76        record_batch: RecordBatch,
77    ) -> Result<Vec<Self>, EncodingError> {
78        decode_batch_with_metadata_fields(
79            metadata,
80            &record_batch,
81            FUNDING_RATE_UPDATE_FIELDS,
82            &[KEY_INSTRUMENT_ID],
83            Some("FundingRateUpdate"),
84        )
85    }
86}
87
88impl DecodeDataFromRecordBatch for FundingRateUpdate {
89    fn decode_data_batch(
90        metadata: &HashMap<String, String>,
91        record_batch: RecordBatch,
92    ) -> Result<Vec<Data>, EncodingError> {
93        let updates = Self::decode_batch(metadata, record_batch)?;
94        Ok(updates.into_iter().map(Data::from).collect())
95    }
96}
97
98#[cfg(test)]
99mod tests {
100    use std::str::FromStr;
101
102    use arrow::array::StringArray;
103    use nautilus_core::UnixNanos;
104    use nautilus_model::identifiers::InstrumentId;
105    use rstest::rstest;
106    use rust_decimal::Decimal;
107
108    use super::*;
109
110    #[rstest]
111    fn test_funding_rate_update_round_trip_preserves_decimal_precision() {
112        let update = FundingRateUpdate::new(
113            InstrumentId::from("BTCUSDT-PERP.BINANCE"),
114            Decimal::from_str("0.000123456789123456789").unwrap(),
115            Some(480),
116            Some(UnixNanos::from(9_000_000_000)),
117            UnixNanos::from(1_000_000_000),
118            UnixNanos::from(2_000_000_000),
119        );
120        let metadata = update.metadata();
121        let batch = FundingRateUpdate::encode_batch(&metadata, &[update]).unwrap();
122        let identifiers = batch
123            .column_by_name("identifier")
124            .unwrap()
125            .as_any()
126            .downcast_ref::<StringArray>()
127            .unwrap();
128        assert_eq!(FundingRateUpdate::get_fields()["rate"], "Utf8");
129        assert_eq!(
130            batch.schema().field_with_name("rate").unwrap().data_type(),
131            &arrow::datatypes::DataType::Utf8
132        );
133        assert_eq!(identifiers.value(0), "BTCUSDT-PERP.BINANCE");
134        assert!(batch.schema().field_with_name("instrument_id").is_err());
135        assert_eq!(
136            batch.schema().metadata(),
137            &HashMap::from([
138                ("type_name".to_string(), "FundingRateUpdate".to_string()),
139                (
140                    KEY_INSTRUMENT_ID.to_string(),
141                    "BTCUSDT-PERP.BINANCE".to_string()
142                ),
143            ])
144        );
145        let decoded = FundingRateUpdate::decode_batch(batch.schema().metadata(), batch).unwrap();
146
147        assert_eq!(decoded, vec![update]);
148    }
149
150    #[rstest]
151    fn test_funding_rate_update_round_trip_null_optionals() {
152        let update = FundingRateUpdate::new(
153            InstrumentId::from("BTCUSDT-PERP.BINANCE"),
154            Decimal::from_str("0.0001").unwrap(),
155            None,
156            None,
157            UnixNanos::from(1_000_000_000),
158            UnixNanos::from(2_000_000_000),
159        );
160        let metadata = update.metadata();
161        let batch = FundingRateUpdate::encode_batch(&metadata, &[update]).unwrap();
162        let decoded = FundingRateUpdate::decode_batch(batch.schema().metadata(), batch).unwrap();
163
164        assert_eq!(decoded, vec![update]);
165        assert!(decoded[0].interval.is_none());
166        assert!(decoded[0].next_funding_ns.is_none());
167    }
168}
nautilus_serialization/arrow/funding.rs

nautilus_serialization/arrow/
funding.rs
