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::utf8("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::utf8("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::utf8("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::utf8("avg_px_open", false),
64    JsonFieldSpec::utf8("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::utf8("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::utf8("avg_px_open", false),
91    JsonFieldSpec::utf8("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 rust_decimal_macros::dec;
186    use ustr::Ustr;
187
188    use super::*;
189
190    #[rstest]
191    fn test_position_adjusted_round_trip() {
192        let event = PositionAdjusted::new(
193            TraderId::from("TRADER-001"),
194            StrategyId::from("EMA-CROSS"),
195            InstrumentId::from("BTCUSDT.BINANCE"),
196            PositionId::from("P-001"),
197            AccountId::from("BINANCE-001"),
198            PositionAdjustmentType::Funding,
199            Some(Decimal::from_str("-0.123456789123456789").unwrap()),
200            Some(Money::new(-5.50, Currency::USD())),
201            Some(Ustr::from("funding_2024_01_15_08:00")),
202            UUID4::default(),
203            UnixNanos::from(1_000_000_000),
204            UnixNanos::from(2_000_000_000),
205        );
206        let metadata = event.metadata();
207        let batch = PositionAdjusted::encode_batch(&metadata, &[event]).unwrap();
208        let decoded =
209            PositionAdjusted::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
210
211        assert_eq!(decoded, vec![event]);
212    }
213
214    #[rstest]
215    fn test_position_opened_round_trip() {
216        let event = PositionOpened {
217            trader_id: TraderId::from("TRADER-001"),
218            strategy_id: StrategyId::from("EMA-CROSS"),
219            instrument_id: InstrumentId::from("EURUSD.SIM"),
220            position_id: PositionId::from("P-001"),
221            account_id: AccountId::from("SIM-001"),
222            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
223            entry: OrderSide::Buy,
224            side: PositionSide::Long,
225            signed_qty: dec!(150.0),
226            quantity: Quantity::from("150"),
227            last_qty: Quantity::from("150"),
228            last_px: Price::from("1.0525"),
229            currency: Currency::USD(),
230            avg_px_open: dec!(1.0525123456789012345678901234),
231            realized_pnl: Some(Money::new(-1.25, Currency::USD())),
232            event_id: UUID4::default(),
233            ts_event: UnixNanos::from(1_000_000_000),
234            ts_init: UnixNanos::from(1_000_000_001),
235        };
236        let metadata = event.metadata();
237        let batch = PositionOpened::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
238        let decoded = PositionOpened::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
239
240        assert_eq!(decoded, vec![event]);
241    }
242
243    #[rstest]
244    fn test_position_changed_round_trip() {
245        let event = PositionChanged {
246            trader_id: TraderId::from("TRADER-001"),
247            strategy_id: StrategyId::from("EMA-CROSS"),
248            instrument_id: InstrumentId::from("EURUSD.SIM"),
249            position_id: PositionId::from("P-001"),
250            account_id: AccountId::from("SIM-001"),
251            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
252            entry: OrderSide::Buy,
253            side: PositionSide::Long,
254            signed_qty: dec!(300.0),
255            quantity: Quantity::from("300"),
256            peak_quantity: Quantity::from("300"),
257            last_qty: Quantity::from("150"),
258            last_px: Price::from("1.0600"),
259            currency: Currency::USD(),
260            avg_px_open: dec!(1.0562),
261            avg_px_close: None,
262            realized_return: 0.0,
263            realized_pnl: None,
264            unrealized_pnl: Money::new(56.25, Currency::USD()),
265            event_id: UUID4::default(),
266            ts_opened: UnixNanos::from(1_000_000_000),
267            ts_event: UnixNanos::from(2_000_000_000),
268            ts_init: UnixNanos::from(2_000_000_001),
269        };
270        let metadata = event.metadata();
271        let batch = PositionChanged::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
272        let decoded =
273            PositionChanged::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
274
275        assert_eq!(decoded, vec![event]);
276    }
277
278    #[rstest]
279    fn test_position_closed_round_trip() {
280        let event = PositionClosed {
281            trader_id: TraderId::from("TRADER-001"),
282            strategy_id: StrategyId::from("EMA-CROSS"),
283            instrument_id: InstrumentId::from("EURUSD.SIM"),
284            position_id: PositionId::from("P-001"),
285            account_id: AccountId::from("SIM-001"),
286            opening_order_id: ClientOrderId::from("O-19700101-000000-001-001-1"),
287            closing_order_id: Some(ClientOrderId::from("O-19700101-000000-001-001-2")),
288            entry: OrderSide::Buy,
289            side: PositionSide::Flat,
290            signed_qty: dec!(0.0),
291            quantity: Quantity::from("0"),
292            peak_quantity: Quantity::from("150"),
293            last_qty: Quantity::from("150"),
294            last_px: Price::from("1.0600"),
295            currency: Currency::USD(),
296            avg_px_open: dec!(1.0525123456789012345678901234),
297            avg_px_close: Some(dec!(1.0600123456789012345678901234)),
298            realized_return: 0.0071,
299            realized_pnl: Some(Money::new(112.50, Currency::USD())),
300            unrealized_pnl: Money::new(0.0, Currency::USD()),
301            duration: DurationNanos::new(3_600_000_000_000),
302            event_id: UUID4::default(),
303            ts_opened: UnixNanos::from(1_000_000_000),
304            ts_closed: Some(UnixNanos::from(4_600_000_000)),
305            ts_event: UnixNanos::from(4_600_000_000),
306            ts_init: UnixNanos::from(5_000_000_000),
307        };
308        let metadata = event.metadata();
309        let batch = PositionClosed::encode_batch(&metadata, std::slice::from_ref(&event)).unwrap();
310        let decoded = PositionClosed::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
311
312        assert_eq!(decoded, vec![event]);
313    }
314}
nautilus_serialization/arrow/position_event.rs

nautilus_serialization/arrow/
position_event.rs
