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::u64("ts_event", false),
45    JsonFieldSpec::u64("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::u64("ts_opened", false),
70    JsonFieldSpec::u64("ts_event", false),
71    JsonFieldSpec::u64("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::u64("ts_opened", false),
98    JsonFieldSpec::u64("ts_closed", true),
99    JsonFieldSpec::u64("ts_event", false),
100    JsonFieldSpec::u64("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::u64("ts_event", false),
115    JsonFieldSpec::u64("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(
134                metadata: &HashMap<String, String>,
135                data: &[Self],
136            ) -> Result<RecordBatch, ArrowError> {
137                encode_batch($type_name, metadata, data, $fields)
138            }
139
140            fn metadata(&self) -> HashMap<String, String> {
141                instrument_metadata($type_name, &self.instrument_id.to_string())
142            }
143        }
144
145        impl DecodeTypedFromRecordBatch for $type {
146            fn decode_typed_batch(
147                metadata: &HashMap<String, String>,
148                record_batch: RecordBatch,
149            ) -> Result<Vec<Self>, EncodingError> {
150                decode_batch(metadata, &record_batch, $fields, Some($type_name))
151            }
152        }
153    };
154}
155
156impl_position_event_arrow!(PositionOpened, "PositionOpened", POSITION_OPENED_FIELDS);
157impl_position_event_arrow!(PositionChanged, "PositionChanged", POSITION_CHANGED_FIELDS);
158impl_position_event_arrow!(PositionClosed, "PositionClosed", POSITION_CLOSED_FIELDS);
159impl_position_event_arrow!(
160    PositionAdjusted,
161    "PositionAdjusted",
162    POSITION_ADJUSTED_FIELDS
163);
164
165#[cfg(test)]
166mod tests {
167    use std::str::FromStr;
168
169    use nautilus_core::{UUID4, UnixNanos};
170    use nautilus_model::{
171        enums::{OrderSide, PositionAdjustmentType, PositionSide},
172        identifiers::{AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TraderId},
173        types::{Currency, Money, Price, Quantity},
174    };
175    use rstest::rstest;
176    use rust_decimal::Decimal;
177    use ustr::Ustr;
178
179    use super::*;
180
181    #[rstest]
182    fn test_position_adjusted_round_trip() {
183        let event = PositionAdjusted::new(
184            TraderId::from("TRADER-001"),
185            StrategyId::from("EMA-CROSS"),
186            InstrumentId::from("BTCUSDT.BINANCE"),
187            PositionId::from("P-001"),
188            AccountId::from("BINANCE-001"),
189            PositionAdjustmentType::Funding,
190            Some(Decimal::from_str("-0.123456789123456789").unwrap()),
191            Some(Money::new(-5.50, Currency::USD())),
192            Some(Ustr::from("funding_2024_01_15_08:00")),
193            UUID4::default(),
194            UnixNanos::from(1_000_000_000),
195            UnixNanos::from(2_000_000_000),
196        );
197        let metadata = event.metadata();
198        let batch = PositionAdjusted::encode_batch(&metadata, &[event]).unwrap();
199        let decoded =
200            PositionAdjusted::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
201
202        assert_eq!(decoded, vec![event]);
203    }
204
205    #[rstest]
206    fn test_position_opened_round_trip() {
207        let event = PositionOpened {
208            trader_id: TraderId::from("TRADER-001"),
209            strategy_id: StrategyId::from("EMA-CROSS"),
210            instrument_id: InstrumentId::from("EURUSD.SIM"),
211            position_id: PositionId::from("P-001"),
212            account_id: AccountId::from("SIM-001"),
213            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
214            entry: OrderSide::Buy,
215            side: PositionSide::Long,
216            signed_qty: 150.0,
217            quantity: Quantity::from("150"),
218            last_qty: Quantity::from("150"),
219            last_px: Price::from("1.0525"),
220            currency: Currency::USD(),
221            avg_px_open: 1.0525,
222            realized_pnl: Some(Money::new(-1.25, Currency::USD())),
223            event_id: UUID4::default(),
224            ts_event: UnixNanos::from(1_000_000_000),
225            ts_init: UnixNanos::from(1_000_000_001),
226        };
227        let metadata = event.metadata();
228        let batch = PositionOpened::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
229        let decoded = PositionOpened::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
230
231        assert_eq!(decoded, vec![event]);
232    }
233
234    #[rstest]
235    fn test_position_changed_round_trip() {
236        let event = PositionChanged {
237            trader_id: TraderId::from("TRADER-001"),
238            strategy_id: StrategyId::from("EMA-CROSS"),
239            instrument_id: InstrumentId::from("EURUSD.SIM"),
240            position_id: PositionId::from("P-001"),
241            account_id: AccountId::from("SIM-001"),
242            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
243            entry: OrderSide::Buy,
244            side: PositionSide::Long,
245            signed_qty: 300.0,
246            quantity: Quantity::from("300"),
247            peak_quantity: Quantity::from("300"),
248            last_qty: Quantity::from("150"),
249            last_px: Price::from("1.0600"),
250            currency: Currency::USD(),
251            avg_px_open: 1.0562,
252            avg_px_close: None,
253            realized_return: 0.0,
254            realized_pnl: None,
255            unrealized_pnl: Money::new(56.25, Currency::USD()),
256            event_id: UUID4::default(),
257            ts_opened: UnixNanos::from(1_000_000_000),
258            ts_event: UnixNanos::from(2_000_000_000),
259            ts_init: UnixNanos::from(2_000_000_001),
260        };
261        let metadata = event.metadata();
262        let batch = PositionChanged::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
263        let decoded =
264            PositionChanged::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
265
266        assert_eq!(decoded, vec![event]);
267    }
268
269    #[rstest]
270    fn test_position_closed_round_trip() {
271        let event = PositionClosed {
272            trader_id: TraderId::from("TRADER-001"),
273            strategy_id: StrategyId::from("EMA-CROSS"),
274            instrument_id: InstrumentId::from("EURUSD.SIM"),
275            position_id: PositionId::from("P-001"),
276            account_id: AccountId::from("SIM-001"),
277            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
278            closing_order_id: Some(ClientOrderId::from("O-19700101-000000-001-001-2")),
279            entry: OrderSide::Buy,
280            side: PositionSide::Flat,
281            signed_qty: 0.0,
282            quantity: Quantity::from("0"),
283            peak_quantity: Quantity::from("150"),
284            last_qty: Quantity::from("150"),
285            last_px: Price::from("1.0600"),
286            currency: Currency::USD(),
287            avg_px_open: 1.0525,
288            avg_px_close: Some(1.0600),
289            realized_return: 0.0071,
290            realized_pnl: Some(Money::new(112.50, Currency::USD())),
291            unrealized_pnl: Money::new(0.0, Currency::USD()),
292            duration: 3_600_000_000_000,
293            event_id: UUID4::default(),
294            ts_opened: UnixNanos::from(1_000_000_000),
295            ts_closed: Some(UnixNanos::from(4_600_000_000)),
296            ts_event: UnixNanos::from(4_600_000_000),
297            ts_init: UnixNanos::from(5_000_000_000),
298        };
299        let metadata = event.metadata();
300        let batch = PositionClosed::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
301        let decoded = PositionClosed::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
302
303        assert_eq!(decoded, vec![event]);
304    }
305}
nautilus_serialization/arrow/position_event.rs

nautilus_serialization/arrow/
position_event.rs
