nautilus_infrastructure/sql/models/
data.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::str::FromStr;
17
18use nautilus_core::UnixNanos;
19use nautilus_model::{
20    data::{Bar, BarSpecification, BarType, InstrumentClose, QuoteTick, TradeTick},
21    identifiers::{InstrumentId, TradeId},
22    types::{Price, Quantity},
23};
24use sqlx::{Error, FromRow, Row, postgres::PgRow};
25
26use crate::sql::models::{
27    enums::{AggregationSourcePg, AggressorSidePg, BarAggregationPg, PriceTypePg},
28    read_usize,
29};
30
31#[derive(Debug)]
32pub struct QuoteTickRow(pub QuoteTick);
33
34#[derive(Debug)]
35pub struct TradeTickRow(pub TradeTick);
36
37#[derive(Debug)]
38pub struct BarRow(pub Bar);
39
40#[derive(Debug)]
41pub struct InstrumentCloseRow(pub InstrumentClose);
42
43impl<'r> FromRow<'r, PgRow> for QuoteTickRow {
44    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
45        let instrument_id = row
46            .try_get::<&str, _>("instrument_id")
47            .map(InstrumentId::from)?;
48        let bid_price = row.try_get::<&str, _>("bid_price").map(Price::from)?;
49        let ask_price = row.try_get::<&str, _>("ask_price").map(Price::from)?;
50        let bid_size = row.try_get::<&str, _>("bid_size").map(Quantity::from)?;
51        let ask_size = row.try_get::<&str, _>("ask_size").map(Quantity::from)?;
52        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
53        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
54        let quote = QuoteTick::new(
55            instrument_id,
56            bid_price,
57            ask_price,
58            bid_size,
59            ask_size,
60            ts_event,
61            ts_init,
62        );
63        Ok(Self(quote))
64    }
65}
66
67impl<'r> FromRow<'r, PgRow> for TradeTickRow {
68    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
69        let instrument_id = row
70            .try_get::<&str, _>("instrument_id")
71            .map(InstrumentId::from)?;
72        let price = row.try_get::<&str, _>("price").map(Price::from)?;
73        let size = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
74        let aggressor_side = row
75            .try_get::<AggressorSidePg, _>("aggressor_side")
76            .map(|x| x.0)?;
77        let trade_id = row
78            .try_get::<&str, _>("venue_trade_id")
79            .map(TradeId::from)?;
80        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
81        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
82        let trade = TradeTick::new(
83            instrument_id,
84            price,
85            size,
86            aggressor_side,
87            trade_id,
88            ts_event,
89            ts_init,
90        );
91        Ok(Self(trade))
92    }
93}
94
95impl<'r> FromRow<'r, PgRow> for BarRow {
96    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
97        let instrument_id: InstrumentId = decode_text_column(row, "instrument_id", "bar")?;
98        let step = read_usize(row, "step")?;
99        let price_type = row.try_get::<PriceTypePg, _>("price_type").map(|x| x.0)?;
100        let bar_aggregation = row
101            .try_get::<BarAggregationPg, _>("bar_aggregation")
102            .map(|x| x.0)?;
103        let aggregation_source = row
104            .try_get::<AggregationSourcePg, _>("aggregation_source")
105            .map(|x| x.0)?;
106        let spec = BarSpecification::new_checked(step, bar_aggregation, price_type)
107            .map_err(|e| Error::Decode(format!("Invalid bar specification in row: {e}").into()))?;
108        let bar_type = BarType::new(instrument_id, spec, aggregation_source);
109        let open: Price = decode_text_column(row, "open", "bar")?;
110        let high: Price = decode_text_column(row, "high", "bar")?;
111        let low: Price = decode_text_column(row, "low", "bar")?;
112        let close: Price = decode_text_column(row, "close", "bar")?;
113        let volume: Quantity = decode_text_column(row, "volume", "bar")?;
114        let ts_event: UnixNanos = decode_text_column(row, "ts_event", "bar")?;
115        let ts_init: UnixNanos = decode_text_column(row, "ts_init", "bar")?;
116        let bar = Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
117            .map_err(|e| Error::Decode(format!("Invalid bar in row: {e}").into()))?;
118        Ok(Self(bar))
119    }
120}
121
122impl<'r> FromRow<'r, PgRow> for InstrumentCloseRow {
123    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
124        let instrument_id = decode_text_column(row, "instrument_id", "instrument close")?;
125        let close_price = decode_text_column(row, "close_price", "instrument close")?;
126        let close_type = decode_text_column(row, "close_type", "instrument close")?;
127        let ts_event = decode_text_column(row, "ts_event", "instrument close")?;
128        let ts_init = decode_text_column(row, "ts_init", "instrument close")?;
129        Ok(Self(InstrumentClose::new(
130            instrument_id,
131            close_price,
132            close_type,
133            ts_event,
134            ts_init,
135        )))
136    }
137}
138
139fn decode_text_column<T: FromStr>(row: &PgRow, column: &str, record: &str) -> Result<T, Error>
140where
141    T::Err: std::fmt::Display,
142{
143    row.try_get::<&str, _>(column)?
144        .parse::<T>()
145        .map_err(|e| Error::Decode(format!("Invalid `{column}` value in {record} row: {e}").into()))
146}
nautilus_infrastructure/sql/models/data.rs

nautilus_infrastructure/sql/models/
data.rs
