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, 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
40impl<'r> FromRow<'r, PgRow> for QuoteTickRow {
41    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
42        let instrument_id = row
43            .try_get::<&str, _>("instrument_id")
44            .map(InstrumentId::from)?;
45        let bid_price = row.try_get::<&str, _>("bid_price").map(Price::from)?;
46        let ask_price = row.try_get::<&str, _>("ask_price").map(Price::from)?;
47        let bid_size = row.try_get::<&str, _>("bid_size").map(Quantity::from)?;
48        let ask_size = row.try_get::<&str, _>("ask_size").map(Quantity::from)?;
49        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
50        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
51        let quote = QuoteTick::new(
52            instrument_id,
53            bid_price,
54            ask_price,
55            bid_size,
56            ask_size,
57            ts_event,
58            ts_init,
59        );
60        Ok(Self(quote))
61    }
62}
63
64impl<'r> FromRow<'r, PgRow> for TradeTickRow {
65    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
66        let instrument_id = row
67            .try_get::<&str, _>("instrument_id")
68            .map(InstrumentId::from)?;
69        let price = row.try_get::<&str, _>("price").map(Price::from)?;
70        let size = row.try_get::<&str, _>("quantity").map(Quantity::from)?;
71        let aggressor_side = row
72            .try_get::<AggressorSidePg, _>("aggressor_side")
73            .map(|x| x.0)?;
74        let trade_id = row
75            .try_get::<&str, _>("venue_trade_id")
76            .map(TradeId::from)?;
77        let ts_event = row.try_get::<&str, _>("ts_event").map(UnixNanos::from)?;
78        let ts_init = row.try_get::<&str, _>("ts_init").map(UnixNanos::from)?;
79        let trade = TradeTick::new(
80            instrument_id,
81            price,
82            size,
83            aggressor_side,
84            trade_id,
85            ts_event,
86            ts_init,
87        );
88        Ok(Self(trade))
89    }
90}
91
92impl<'r> FromRow<'r, PgRow> for BarRow {
93    fn from_row(row: &'r PgRow) -> Result<Self, Error> {
94        fn decode<T: FromStr>(row: &PgRow, column: &str) -> Result<T, Error>
95        where
96            T::Err: std::fmt::Display,
97        {
98            row.try_get::<&str, _>(column)?.parse::<T>().map_err(|e| {
99                Error::Decode(format!("Invalid `{column}` value in bar row: {e}").into())
100            })
101        }
102
103        let instrument_id: InstrumentId = decode(row, "instrument_id")?;
104        let step = read_usize(row, "step")?;
105        let price_type = row.try_get::<PriceTypePg, _>("price_type").map(|x| x.0)?;
106        let bar_aggregation = row
107            .try_get::<BarAggregationPg, _>("bar_aggregation")
108            .map(|x| x.0)?;
109        let aggregation_source = row
110            .try_get::<AggregationSourcePg, _>("aggregation_source")
111            .map(|x| x.0)?;
112        let spec = BarSpecification::new_checked(step, bar_aggregation, price_type)
113            .map_err(|e| Error::Decode(format!("Invalid bar specification in row: {e}").into()))?;
114        let bar_type = BarType::new(instrument_id, spec, aggregation_source);
115        let open: Price = decode(row, "open")?;
116        let high: Price = decode(row, "high")?;
117        let low: Price = decode(row, "low")?;
118        let close: Price = decode(row, "close")?;
119        let volume: Quantity = decode(row, "volume")?;
120        let ts_event: UnixNanos = decode(row, "ts_event")?;
121        let ts_init: UnixNanos = decode(row, "ts_init")?;
122        let bar = Bar::new_checked(bar_type, open, high, low, close, volume, ts_event, ts_init)
123            .map_err(|e| Error::Decode(format!("Invalid bar in row: {e}").into()))?;
124        Ok(Self(bar))
125    }
126}
nautilus_infrastructure/sql/models/data.rs

nautilus_infrastructure/sql/models/
data.rs
