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