nautilus_infrastructure/sql/models/
positions.rs1use std::str::FromStr;
17
18use nautilus_core::UnixNanos;
19use nautilus_model::{
20 enums::{OrderSide, PositionSide},
21 events::PositionSnapshot,
22 identifiers::{AccountId, ClientOrderId, InstrumentId, PositionId, StrategyId, TraderId},
23 types::{Currency, Money, Quantity},
24};
25use sqlx::{FromRow, Row, postgres::PgRow};
26
27use crate::sql::models::decode_error;
28
29#[derive(Debug)]
30pub struct PositionSnapshotRow(pub PositionSnapshot);
31
32impl<'r> FromRow<'r, PgRow> for PositionSnapshotRow {
33 fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
34 let id = row.try_get::<&str, _>("id").map(PositionId::from)?;
35 let trader_id = row.try_get::<&str, _>("trader_id").map(TraderId::from)?;
36 let strategy_id = row
37 .try_get::<&str, _>("strategy_id")
38 .map(StrategyId::from)?;
39 let instrument_id = row
40 .try_get::<&str, _>("instrument_id")
41 .map(InstrumentId::from)?;
42 let account_id = row.try_get::<&str, _>("account_id").map(AccountId::from)?;
43 let opening_order_id = row
44 .try_get::<&str, _>("opening_order_id")
45 .map(ClientOrderId::from)?;
46 let closing_order_id = row
47 .try_get::<Option<&str>, _>("closing_order_id")
48 .ok()
49 .and_then(|x| x.map(ClientOrderId::from));
50 let entry = row
51 .try_get::<&str, _>("entry")
52 .map(OrderSide::from_str)?
53 .expect("Invalid `OrderSide`");
54 let side = row
55 .try_get::<&str, _>("side")
56 .map(PositionSide::from_str)?
57 .expect("Invalid `PositionSide`");
58 let signed_qty = row.try_get::<f64, _>("signed_qty")?;
59 let quantity = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
60 let peak_qty = row.try_get::<&str, _>("peak_qty").map(Quantity::from)?;
61 let quote_currency = row
62 .try_get::<&str, _>("quote_currency")
63 .map(Currency::from)?;
64 let base_currency = row
65 .try_get::<Option<&str>, _>("base_currency")
66 .ok()
67 .and_then(|x| x.map(Currency::from));
68 let settlement_currency = row
69 .try_get::<&str, _>("settlement_currency")
70 .map(Currency::from)?;
71 let avg_px_open = row.try_get::<f64, _>("avg_px_open")?;
72 let avg_px_close = row.try_get::<Option<f64>, _>("avg_px_close")?;
73 let realized_return = row.try_get::<Option<f64>, _>("realized_return")?;
74 let realized_pnl = row.try_get::<&str, _>("realized_pnl").map(Money::from)?;
75 let unrealized_pnl = row
76 .try_get::<Option<&str>, _>("unrealized_pnl")
77 .ok()
78 .and_then(|x| x.map(Money::from));
79 let commissions = row
80 .try_get::<Option<Vec<String>>, _>("commissions")?
81 .map_or_else(Vec::new, |c| {
82 c.into_iter().map(|s| Money::from(&s)).collect()
83 });
84 let duration_ns: Option<u64> = row
85 .try_get::<Option<&str>, _>("duration_ns")?
86 .map(|value| {
87 value
88 .parse()
89 .map_err(|e| decode_error("duration_ns", value, e))
90 })
91 .transpose()?;
92 let ts_opened = row.try_get::<String, _>("ts_opened").map(UnixNanos::from)?;
93 let ts_closed: Option<UnixNanos> = row
94 .try_get::<Option<String>, _>("ts_closed")?
95 .map(UnixNanos::from);
96 let ts_init = row.try_get::<String, _>("ts_init").map(UnixNanos::from)?;
97 let ts_last = row.try_get::<String, _>("ts_last").map(UnixNanos::from)?;
98 let replay_state = row.try_get::<Option<serde_json::Value>, _>("replay_state")?;
99
100 let snapshot = PositionSnapshot {
101 trader_id,
102 strategy_id,
103 instrument_id,
104 position_id: id,
105 account_id,
106 opening_order_id,
107 closing_order_id,
108 entry,
109 side,
110 signed_qty,
111 quantity,
112 peak_qty,
113 quote_currency,
114 base_currency,
115 settlement_currency,
116 avg_px_open,
117 avg_px_close,
118 realized_return,
119 realized_pnl: Some(realized_pnl), unrealized_pnl,
121 commissions,
122 duration_ns,
123 ts_opened,
124 ts_closed,
125 ts_last,
126 ts_init,
127 replay_state,
128 };
129
130 Ok(Self(snapshot))
131 }
132}