nautilus_serialization/arrow/
position_event.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::events::{PositionAdjusted, PositionChanged, PositionClosed, PositionOpened};
20
21use super::{
22    ArrowSchemaProvider, DecodeTypedFromRecordBatch, EncodeToRecordBatch, EncodingError,
23    KEY_INSTRUMENT_ID,
24    json::{JsonFieldSpec, decode_batch, encode_batch, metadata_for_type, schema_for_type},
25};
26
27const POSITION_OPENED_FIELDS: &[JsonFieldSpec] = &[
28    JsonFieldSpec::utf8("trader_id", false),
29    JsonFieldSpec::utf8("strategy_id", false),
30    JsonFieldSpec::utf8("instrument_id", false),
31    JsonFieldSpec::utf8("position_id", false),
32    JsonFieldSpec::utf8("account_id", false),
33    JsonFieldSpec::utf8("opening_order_id", false),
34    JsonFieldSpec::utf8("entry", false),
35    JsonFieldSpec::utf8("side", false),
36    JsonFieldSpec::f64("signed_qty", false),
37    JsonFieldSpec::utf8("quantity", false),
38    JsonFieldSpec::utf8("last_qty", false),
39    JsonFieldSpec::utf8("last_px", false),
40    JsonFieldSpec::utf8("currency", false),
41    JsonFieldSpec::f64("avg_px_open", false),
42    JsonFieldSpec::utf8("realized_pnl", true),
43    JsonFieldSpec::utf8("event_id", false),
44    JsonFieldSpec::timestamp("ts_event", false),
45    JsonFieldSpec::timestamp("ts_init", false),
46];
47
48const POSITION_CHANGED_FIELDS: &[JsonFieldSpec] = &[
49    JsonFieldSpec::utf8("trader_id", false),
50    JsonFieldSpec::utf8("strategy_id", false),
51    JsonFieldSpec::utf8("instrument_id", false),
52    JsonFieldSpec::utf8("position_id", false),
53    JsonFieldSpec::utf8("account_id", false),
54    JsonFieldSpec::utf8("opening_order_id", false),
55    JsonFieldSpec::utf8("entry", false),
56    JsonFieldSpec::utf8("side", false),
57    JsonFieldSpec::f64("signed_qty", false),
58    JsonFieldSpec::utf8("quantity", false),
59    JsonFieldSpec::utf8("peak_quantity", false),
60    JsonFieldSpec::utf8("last_qty", false),
61    JsonFieldSpec::utf8("last_px", false),
62    JsonFieldSpec::utf8("currency", false),
63    JsonFieldSpec::f64("avg_px_open", false),
64    JsonFieldSpec::f64("avg_px_close", true),
65    JsonFieldSpec::f64("realized_return", false),
66    JsonFieldSpec::utf8("realized_pnl", true),
67    JsonFieldSpec::utf8("unrealized_pnl", false),
68    JsonFieldSpec::utf8("event_id", false),
69    JsonFieldSpec::timestamp("ts_opened", false),
70    JsonFieldSpec::timestamp("ts_event", false),
71    JsonFieldSpec::timestamp("ts_init", false),
72];
73
74const POSITION_CLOSED_FIELDS: &[JsonFieldSpec] = &[
75    JsonFieldSpec::utf8("trader_id", false),
76    JsonFieldSpec::utf8("strategy_id", false),
77    JsonFieldSpec::utf8("instrument_id", false),
78    JsonFieldSpec::utf8("position_id", false),
79    JsonFieldSpec::utf8("account_id", false),
80    JsonFieldSpec::utf8("opening_order_id", false),
81    JsonFieldSpec::utf8("closing_order_id", true),
82    JsonFieldSpec::utf8("entry", false),
83    JsonFieldSpec::utf8("side", false),
84    JsonFieldSpec::f64("signed_qty", false),
85    JsonFieldSpec::utf8("quantity", false),
86    JsonFieldSpec::utf8("peak_quantity", false),
87    JsonFieldSpec::utf8("last_qty", false),
88    JsonFieldSpec::utf8("last_px", false),
89    JsonFieldSpec::utf8("currency", false),
90    JsonFieldSpec::f64("avg_px_open", false),
91    JsonFieldSpec::f64("avg_px_close", true),
92    JsonFieldSpec::f64("realized_return", false),
93    JsonFieldSpec::utf8("realized_pnl", true),
94    JsonFieldSpec::utf8("unrealized_pnl", false),
95    JsonFieldSpec::u64("duration", false),
96    JsonFieldSpec::utf8("event_id", false),
97    JsonFieldSpec::timestamp("ts_opened", false),
98    JsonFieldSpec::timestamp("ts_closed", true),
99    JsonFieldSpec::timestamp("ts_event", false),
100    JsonFieldSpec::timestamp("ts_init", false),
101];
102
103const POSITION_ADJUSTED_FIELDS: &[JsonFieldSpec] = &[
104    JsonFieldSpec::utf8("trader_id", false),
105    JsonFieldSpec::utf8("strategy_id", false),
106    JsonFieldSpec::utf8("instrument_id", false),
107    JsonFieldSpec::utf8("position_id", false),
108    JsonFieldSpec::utf8("account_id", false),
109    JsonFieldSpec::utf8("adjustment_type", false),
110    JsonFieldSpec::utf8("quantity_change", true),
111    JsonFieldSpec::utf8("pnl_change", true),
112    JsonFieldSpec::utf8("reason", true),
113    JsonFieldSpec::utf8("event_id", false),
114    JsonFieldSpec::timestamp("ts_event", false),
115    JsonFieldSpec::timestamp("ts_init", false),
116];
117
118fn instrument_metadata(type_name: &'static str, instrument_id: &str) -> HashMap<String, String> {
119    let mut metadata = metadata_for_type(type_name);
120    metadata.insert(KEY_INSTRUMENT_ID.to_string(), instrument_id.to_string());
121    metadata
122}
123
124macro_rules! impl_position_event_arrow {
125    ($type:ty, $type_name:expr, $fields:expr) => {
126        impl ArrowSchemaProvider for $type {
127            fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
128                schema_for_type($type_name, metadata, $fields)
129            }
130        }
131
132        impl EncodeToRecordBatch for $type {
133            fn encode_batch<T>(
134                metadata: &HashMap<String, String>,
135                data: &[T],
136            ) -> Result<RecordBatch, ArrowError>
137            where
138                T: std::borrow::Borrow<Self>,
139            {
140                encode_batch(
141                    $type_name,
142                    metadata,
143                    data.iter().map(std::borrow::Borrow::borrow),
144                    $fields,
145                )
146            }
147
148            fn metadata(&self) -> HashMap<String, String> {
149                instrument_metadata($type_name, &self.instrument_id.to_string())
150            }
151        }
152
153        impl DecodeTypedFromRecordBatch for $type {
154            fn decode_typed_batch(
155                metadata: &HashMap<String, String>,
156                record_batch: RecordBatch,
157            ) -> Result<Vec<Self>, EncodingError> {
158                decode_batch(metadata, &record_batch, $fields, Some($type_name))
159            }
160        }
161    };
162}
163
164impl_position_event_arrow!(PositionOpened, "PositionOpened", POSITION_OPENED_FIELDS);
165impl_position_event_arrow!(PositionChanged, "PositionChanged", POSITION_CHANGED_FIELDS);
166impl_position_event_arrow!(PositionClosed, "PositionClosed", POSITION_CLOSED_FIELDS);
167impl_position_event_arrow!(
168    PositionAdjusted,
169    "PositionAdjusted",
170    POSITION_ADJUSTED_FIELDS
171);
172
173#[cfg(test)]
174mod tests {
175    use std::str::FromStr;
176
177    use nautilus_core::{DurationNanos, UUID4, UnixNanos};
178    use nautilus_model::{
179        enums::{OrderSide, PositionAdjustmentType, PositionSide},
180        identifiers::{AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TraderId},
181        types::{Currency, Money, Price, Quantity},
182    };
183    use rstest::rstest;
184    use rust_decimal::Decimal;
185    use ustr::Ustr;
186
187    use super::*;
188
189    #[rstest]
190    fn test_position_adjusted_round_trip() {
191        let event = PositionAdjusted::new(
192            TraderId::from("TRADER-001"),
193            StrategyId::from("EMA-CROSS"),
194            InstrumentId::from("BTCUSDT.BINANCE"),
195            PositionId::from("P-001"),
196            AccountId::from("BINANCE-001"),
197            PositionAdjustmentType::Funding,
198            Some(Decimal::from_str("-0.123456789123456789").unwrap()),
199            Some(Money::new(-5.50, Currency::USD())),
200            Some(Ustr::from("funding_2024_01_15_08:00")),
201            UUID4::default(),
202            UnixNanos::from(1_000_000_000),
203            UnixNanos::from(2_000_000_000),
204        );
205        let metadata = event.metadata();
206        let batch = PositionAdjusted::encode_batch(&metadata, &[event]).unwrap();
207        let decoded =
208            PositionAdjusted::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
209
210        assert_eq!(decoded, vec![event]);
211    }
212
213    #[rstest]
214    fn test_position_opened_round_trip() {
215        let event = PositionOpened {
216            trader_id: TraderId::from("TRADER-001"),
217            strategy_id: StrategyId::from("EMA-CROSS"),
218            instrument_id: InstrumentId::from("EURUSD.SIM"),
219            position_id: PositionId::from("P-001"),
220            account_id: AccountId::from("SIM-001"),
221            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
222            entry: OrderSide::Buy,
223            side: PositionSide::Long,
224            signed_qty: 150.0,
225            quantity: Quantity::from("150"),
226            last_qty: Quantity::from("150"),
227            last_px: Price::from("1.0525"),
228            currency: Currency::USD(),
229            avg_px_open: 1.0525,
230            realized_pnl: Some(Money::new(-1.25, Currency::USD())),
231            event_id: UUID4::default(),
232            ts_event: UnixNanos::from(1_000_000_000),
233            ts_init: UnixNanos::from(1_000_000_001),
234        };
235        let metadata = event.metadata();
236        let batch = PositionOpened::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
237        let decoded = PositionOpened::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
238
239        assert_eq!(decoded, vec![event]);
240    }
241
242    #[rstest]
243    fn test_position_changed_round_trip() {
244        let event = PositionChanged {
245            trader_id: TraderId::from("TRADER-001"),
246            strategy_id: StrategyId::from("EMA-CROSS"),
247            instrument_id: InstrumentId::from("EURUSD.SIM"),
248            position_id: PositionId::from("P-001"),
249            account_id: AccountId::from("SIM-001"),
250            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
251            entry: OrderSide::Buy,
252            side: PositionSide::Long,
253            signed_qty: 300.0,
254            quantity: Quantity::from("300"),
255            peak_quantity: Quantity::from("300"),
256            last_qty: Quantity::from("150"),
257            last_px: Price::from("1.0600"),
258            currency: Currency::USD(),
259            avg_px_open: 1.0562,
260            avg_px_close: None,
261            realized_return: 0.0,
262            realized_pnl: None,
263            unrealized_pnl: Money::new(56.25, Currency::USD()),
264            event_id: UUID4::default(),
265            ts_opened: UnixNanos::from(1_000_000_000),
266            ts_event: UnixNanos::from(2_000_000_000),
267            ts_init: UnixNanos::from(2_000_000_001),
268        };
269        let metadata = event.metadata();
270        let batch = PositionChanged::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
271        let decoded =
272            PositionChanged::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
273
274        assert_eq!(decoded, vec![event]);
275    }
276
277    #[rstest]
278    fn test_position_closed_round_trip() {
279        let event = PositionClosed {
280            trader_id: TraderId::from("TRADER-001"),
281            strategy_id: StrategyId::from("EMA-CROSS"),
282            instrument_id: InstrumentId::from("EURUSD.SIM"),
283            position_id: PositionId::from("P-001"),
284            account_id: AccountId::from("SIM-001"),
285            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
286            closing_order_id: Some(ClientOrderId::from("O-19700101-000000-001-001-2")),
287            entry: OrderSide::Buy,
288            side: PositionSide::Flat,
289            signed_qty: 0.0,
290            quantity: Quantity::from("0"),
291            peak_quantity: Quantity::from("150"),
292            last_qty: Quantity::from("150"),
293            last_px: Price::from("1.0600"),
294            currency: Currency::USD(),
295            avg_px_open: 1.0525,
296            avg_px_close: Some(1.0600),
297            realized_return: 0.0071,
298            realized_pnl: Some(Money::new(112.50, Currency::USD())),
299            unrealized_pnl: Money::new(0.0, Currency::USD()),
300            duration: DurationNanos::new(3_600_000_000_000),
301            event_id: UUID4::default(),
302            ts_opened: UnixNanos::from(1_000_000_000),
303            ts_closed: Some(UnixNanos::from(4_600_000_000)),
304            ts_event: UnixNanos::from(4_600_000_000),
305            ts_init: UnixNanos::from(5_000_000_000),
306        };
307        let metadata = event.metadata();
308        let batch = PositionClosed::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
309        let decoded = PositionClosed::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
310
311        assert_eq!(decoded, vec![event]);
312    }
313}
nautilus_serialization/arrow/position_event.rs

nautilus_serialization/arrow/
position_event.rs
