nautilus_infrastructure/sql/
queries.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 ahash::AHashMap;
17use nautilus_common::signal::Signal;
18use nautilus_model::{
19    accounts::AccountAny,
20    data::{Bar, CustomData, DataType, HasTsInit, QuoteTick, TradeTick},
21    events::{
22        AccountState, OrderEvent, OrderEventAny, OrderFilled, OrderInitialized, OrderSnapshot,
23        position::snapshot::PositionSnapshot,
24    },
25    identifiers::{AccountId, ClientId, ClientOrderId, InstrumentId, PositionId},
26    instruments::{Instrument, InstrumentAny},
27    orders::OrderAny,
28    position::Position,
29    types::{AccountBalance, Currency, MarginBalance},
30};
31use sqlx::{PgPool, Postgres, Row, Transaction};
32
33use super::models::{
34    orders::OrderSnapshotModel, positions::PositionSnapshotModel, types::SignalModel,
35};
36use crate::sql::models::{
37    accounts::AccountEventModel,
38    data::{BarModel, QuoteTickModel, TradeTickModel},
39    enums::{
40        AggregationSourceModel, AggressorSideModel, AssetClassModel, BarAggregationModel,
41        CurrencyTypeModel, PriceTypeModel, TrailingOffsetTypeModel,
42    },
43    general::{GeneralRow, OrderEventOrderClientIdCombination, OrderPositionIndexRow},
44    instruments::InstrumentAnyModel,
45    orders::{OrderEventAnyModel, OrderFilledModel},
46    types::CurrencyModel,
47};
48
49#[derive(Debug)]
50pub struct DatabaseQueries;
51
52impl DatabaseQueries {
53    /// Truncates all tables in the cache database via the provided Postgres `pool`.
54    ///
55    /// # Errors
56    ///
57    /// Returns an error if the TRUNCATE operation fails.
58    pub async fn truncate(pool: &PgPool) -> anyhow::Result<()> {
59        sqlx::query("SELECT truncate_all_tables()")
60            .execute(pool)
61            .await
62            .map(|_| ())
63            .map_err(|e| anyhow::anyhow!("Failed to truncate tables: {e}"))
64    }
65
66    /// Inserts a raw key-value entry into the `general` table via the provided `pool`.
67    ///
68    /// # Errors
69    ///
70    /// Returns an error if the INSERT operation fails.
71    pub async fn add(pool: &PgPool, key: String, value: Vec<u8>) -> anyhow::Result<()> {
72        sqlx::query("INSERT INTO general (id, value) VALUES ($1, $2)")
73            .bind(key)
74            .bind(value)
75            .execute(pool)
76            .await
77            .map(|_| ())
78            .map_err(|e| anyhow::anyhow!("Failed to insert into general table: {e}"))
79    }
80
81    /// Loads all entries from the `general` table via the provided `pool`.
82    ///
83    /// # Errors
84    ///
85    /// Returns an error if the SELECT operation fails.
86    pub async fn load(pool: &PgPool) -> anyhow::Result<AHashMap<String, Vec<u8>>> {
87        sqlx::query_as::<_, GeneralRow>("SELECT * FROM general")
88            .fetch_all(pool)
89            .await
90            .map(|rows| {
91                let mut cache: AHashMap<String, Vec<u8>> = AHashMap::new();
92                for row in rows {
93                    cache.insert(row.id, row.value);
94                }
95                cache
96            })
97            .map_err(|e| anyhow::anyhow!("Failed to load general table: {e}"))
98    }
99
100    /// Inserts or ignores a `Currency` row via the provided `pool`.
101    ///
102    /// # Errors
103    ///
104    /// Returns an error if the INSERT operation fails.
105    pub async fn add_currency(pool: &PgPool, currency: Currency) -> anyhow::Result<()> {
106        sqlx::query(
107            "INSERT INTO currency (id, precision, iso4217, name, currency_type) VALUES ($1, $2, $3, $4, $5::currency_type) ON CONFLICT (id) DO NOTHING"
108        )
109            .bind(currency.code.as_str())
110            .bind(i32::from(currency.precision))
111            .bind(i32::from(currency.iso4217))
112            .bind(currency.name.as_str())
113            .bind(CurrencyTypeModel(currency.currency_type))
114            .execute(pool)
115            .await
116            .map(|_| ())
117            .map_err(|e| anyhow::anyhow!("Failed to insert into currency table: {e}"))
118    }
119
120    /// Loads all `Currency` entries via the provided `pool`.
121    ///
122    /// # Errors
123    ///
124    /// Returns an error if the SELECT operation fails.
125    pub async fn load_currencies(pool: &PgPool) -> anyhow::Result<Vec<Currency>> {
126        sqlx::query_as::<_, CurrencyModel>("SELECT * FROM currency ORDER BY id ASC")
127            .fetch_all(pool)
128            .await
129            .map(|rows| rows.into_iter().map(|row| row.0).collect())
130            .map_err(|e| anyhow::anyhow!("Failed to load currencies: {e}"))
131    }
132
133    /// Loads a single `Currency` entry by `code` via the provided `pool`.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if the SELECT operation fails.
138    pub async fn load_currency(pool: &PgPool, code: &str) -> anyhow::Result<Option<Currency>> {
139        sqlx::query_as::<_, CurrencyModel>("SELECT * FROM currency WHERE id = $1")
140            .bind(code)
141            .fetch_optional(pool)
142            .await
143            .map(|currency| currency.map(|row| row.0))
144            .map_err(|e| anyhow::anyhow!("Failed to load currency: {e}"))
145    }
146
147    /// Inserts or updates an `InstrumentAny` entry via the provided `pool`.
148    ///
149    /// # Errors
150    ///
151    /// Returns an error if the INSERT or UPDATE operation fails.
152    pub async fn add_instrument(
153        pool: &PgPool,
154        kind: &str,
155        instrument: Box<dyn Instrument>,
156    ) -> anyhow::Result<()> {
157        sqlx::query(r#"
158            INSERT INTO "instrument" (
159                id, kind, raw_symbol, base_currency, underlying, quote_currency, settlement_currency, isin, asset_class, exchange,
160                strategy_type, multiplier, option_kind, is_inverse, strike_price, activation_ns, expiration_ns, price_precision, size_precision,
161                price_increment, size_increment, maker_fee, taker_fee, margin_init, margin_maint, lot_size, max_quantity, min_quantity, max_notional,
162                min_notional, max_price, min_price, ts_init, ts_event, created_at, updated_at
163            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9::asset_class, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28, $29, $30, $31, $32, $33, $34, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
164            ON CONFLICT (id)
165            DO UPDATE
166            SET
167                kind = $2, raw_symbol = $3, base_currency= $4, underlying = $5, quote_currency = $6, settlement_currency = $7, isin = $8, asset_class = $9, exchange = $10,
168                 strategy_type = $11, multiplier = $12, option_kind = $13, is_inverse = $14, strike_price = $15, activation_ns = $16, expiration_ns = $17 , price_precision = $18, size_precision = $19,
169                 price_increment = $20, size_increment = $21, maker_fee = $22, taker_fee = $23, margin_init = $24, margin_maint = $25, lot_size = $26, max_quantity = $27,
170                 min_quantity = $28, max_notional = $29, min_notional = $30, max_price = $31, min_price = $32, ts_init = $33,  ts_event = $34, updated_at = CURRENT_TIMESTAMP
171            "#)
172            .bind(instrument.id().to_string())
173            .bind(kind)
174            .bind(instrument.raw_symbol().to_string())
175            .bind(instrument.base_currency().map(|x| x.code.as_str()))
176            .bind(instrument.underlying().map(|x| x.to_string()))
177            .bind(instrument.quote_currency().code.as_str())
178            .bind(instrument.settlement_currency().code.as_str())
179            .bind(instrument.isin().map(|x| x.to_string()))
180            .bind(AssetClassModel(instrument.asset_class()))
181            .bind(instrument.exchange().map(|x| x.to_string()))
182            .bind(instrument.strategy_type().map(|x| x.to_string()))
183            .bind(instrument.multiplier().to_string())
184            .bind(instrument.option_kind().map(|x| x.to_string()))
185            .bind(instrument.is_inverse())
186            .bind(instrument.strike_price().map(|x| x.to_string()))
187            .bind(instrument.activation_ns().map(|x| x.to_string()))
188            .bind(instrument.expiration_ns().map(|x| x.to_string()))
189            .bind(i32::from(instrument.price_precision()))
190            .bind(i32::from(instrument.size_precision()))
191            .bind(instrument.price_increment().to_string())
192            .bind(instrument.size_increment().to_string())
193            .bind(instrument.maker_fee().to_string())
194            .bind(instrument.taker_fee().to_string())
195            .bind(instrument.margin_init().to_string())
196            .bind(instrument.margin_maint().to_string())
197            .bind(instrument.lot_size().map(|x| x.to_string()))
198            .bind(instrument.max_quantity().map(|x| x.to_string()))
199            .bind(instrument.min_quantity().map(|x| x.to_string()))
200            .bind(instrument.max_notional().map(|x| x.to_string()))
201            .bind(instrument.min_notional().map(|x| x.to_string()))
202            .bind(instrument.max_price().map(|x| x.to_string()))
203            .bind(instrument.min_price().map(|x| x.to_string()))
204            .bind(instrument.ts_init().to_string())
205            .bind(instrument.ts_event().to_string())
206            .execute(pool)
207            .await
208            .map(|_| ())
209            .map_err(|e| anyhow::anyhow!("Failed to insert item {} into instrument table: {:?}", instrument.id(), e))
210    }
211
212    /// Loads a single `InstrumentAny` entry by `instrument_id` via the provided `pool`.
213    ///
214    /// # Errors
215    ///
216    /// Returns an error if the SELECT operation fails.
217    pub async fn load_instrument(
218        pool: &PgPool,
219        instrument_id: &InstrumentId,
220    ) -> anyhow::Result<Option<InstrumentAny>> {
221        sqlx::query_as::<_, InstrumentAnyModel>("SELECT * FROM instrument WHERE id = $1")
222            .bind(instrument_id.to_string())
223            .fetch_optional(pool)
224            .await
225            .map(|instrument| instrument.map(|row| row.0))
226            .map_err(|e| {
227                anyhow::anyhow!("Failed to load instrument with id {instrument_id},error is: {e}")
228            })
229    }
230
231    /// Loads all `InstrumentAny` entries via the provided `pool`.
232    ///
233    /// # Errors
234    ///
235    /// Returns an error if the SELECT operation fails.
236    pub async fn load_instruments(pool: &PgPool) -> anyhow::Result<Vec<InstrumentAny>> {
237        sqlx::query_as::<_, InstrumentAnyModel>("SELECT * FROM instrument")
238            .fetch_all(pool)
239            .await
240            .map(|rows| rows.into_iter().map(|row| row.0).collect())
241            .map_err(|e| anyhow::anyhow!("Failed to load instruments: {e}"))
242    }
243
244    /// Inserts an `OrderInitialized` event via the provided `pool`.
245    ///
246    /// # Errors
247    ///
248    /// Returns an error if the SQL INSERT or UPDATE operation fails.
249    pub async fn add_order(
250        pool: &PgPool,
251        event: OrderInitialized,
252        client_id: Option<ClientId>,
253    ) -> anyhow::Result<()> {
254        Self::add_order_event(pool, Box::new(event), client_id).await
255    }
256
257    /// Inserts an `OrderSnapshot` entry via the provided `pool`.
258    ///
259    /// # Errors
260    ///
261    /// Returns an error if the SQL INSERT operation fails.
262    ///
263    /// # Panics
264    ///
265    /// Panics if serialization of `snapshot.exec_algorithm_params` fails.
266    #[expect(
267        clippy::too_many_lines,
268        reason = "order snapshot persistence maps the full database schema in one transaction"
269    )]
270    pub async fn add_order_snapshot(pool: &PgPool, snapshot: OrderSnapshot) -> anyhow::Result<()> {
271        let mut transaction = pool.begin().await?;
272
273        // Insert trader if it does not exist
274        // TODO remove this when node and trader initialization is implemented
275        sqlx::query(
276            r#"
277            INSERT INTO "trader" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
278            "#,
279        )
280        .bind(snapshot.trader_id.to_string())
281        .execute(&mut *transaction)
282        .await
283        .map(|_| ())
284        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
285
286        sqlx::query(
287            r#"
288            INSERT INTO "order" (
289                id, trader_id, strategy_id, instrument_id, client_order_id, venue_order_id, position_id,
290                account_id, last_trade_id, order_type, order_side, quantity, price, trigger_price,
291                trigger_type, limit_offset, trailing_offset, trailing_offset_type, time_in_force,
292                expire_time, filled_qty, liquidity_side, avg_px, slippage, commissions, status,
293                is_post_only, is_reduce_only, is_quote_quantity, display_qty, emulation_trigger,
294                trigger_instrument_id, contingency_type, order_list_id, linked_order_ids,
295                parent_order_id, exec_algorithm_id, exec_algorithm_params, exec_spawn_id, tags, init_id, ts_init, ts_last,
296                created_at, updated_at
297            ) VALUES (
298                $1, $2, $3, $4, $1, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16,
299                $17::TRAILING_OFFSET_TYPE, $18, $19, $20, $21, $22, $23, $24, $25, $26, $27, $28,
300                $29, $30, $31, $32, $33, $34, $35, $36, $37, $38, $39, $40, $41, $42,
301                CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
302            )
303            ON CONFLICT (id)
304            DO UPDATE SET
305                trader_id = $2,
306                strategy_id = $3,
307                instrument_id = $4,
308                venue_order_id = $5,
309                position_id = $6,
310                account_id = $7,
311                last_trade_id = $8,
312                order_type = $9,
313                order_side = $10,
314                quantity = $11,
315                price = $12,
316                trigger_price = $13,
317                trigger_type = $14,
318                limit_offset = $15,
319                trailing_offset = $16,
320                trailing_offset_type = $17::TRAILING_OFFSET_TYPE,
321                time_in_force = $18,
322                expire_time = $19,
323                filled_qty = $20,
324                liquidity_side = $21,
325                avg_px = $22,
326                slippage = $23,
327                commissions = $24,
328                status = $25,
329                is_post_only = $26,
330                is_reduce_only = $27,
331                is_quote_quantity = $28,
332                display_qty = $29,
333                emulation_trigger = $30,
334                trigger_instrument_id = $31,
335                contingency_type = $32,
336                order_list_id = $33,
337                linked_order_ids = $34,
338                parent_order_id = $35,
339                exec_algorithm_id = $36,
340                exec_algorithm_params = $37,
341                exec_spawn_id = $38,
342                tags = $39,
343                init_id = $40,
344                ts_init = $41,
345                ts_last = $42,
346                updated_at = CURRENT_TIMESTAMP
347        "#)
348            .bind(snapshot.client_order_id.to_string())  // Used for both id and client_order_id
349            .bind(snapshot.trader_id.to_string())
350            .bind(snapshot.strategy_id.to_string())
351            .bind(snapshot.instrument_id.to_string())
352            .bind(snapshot.venue_order_id.map(|x| x.to_string()))
353            .bind(snapshot.position_id.map(|x| x.to_string()))
354            .bind(snapshot.account_id.map(|x| x.to_string()))
355            .bind(snapshot.last_trade_id.map(|x| x.to_string()))
356            .bind(snapshot.order_type.to_string())
357            .bind(snapshot.order_side.to_string())
358            .bind(snapshot.quantity.to_string())
359            .bind(snapshot.price.map(|x| x.to_string()))
360            .bind(snapshot.trigger_price.map(|x| x.to_string()))
361            .bind(snapshot.trigger_type.map(|x| x.to_string()))
362            .bind(snapshot.limit_offset.map(|x| x.to_string()))
363            .bind(snapshot.trailing_offset.map(|x| x.to_string()))
364            .bind(snapshot.trailing_offset_type.map(|x| x.to_string()))
365            .bind(snapshot.time_in_force.to_string())
366            .bind(snapshot.expire_time.map(|x| x.to_string()))
367            .bind(snapshot.filled_qty.to_string())
368            .bind(snapshot.liquidity_side.map(|x| x.to_string()))
369            .bind(snapshot.avg_px)
370            .bind(snapshot.slippage)
371            .bind(snapshot.commissions.iter().map(ToString::to_string).collect::<Vec<String>>())
372            .bind(snapshot.status.to_string())
373            .bind(snapshot.is_post_only)
374            .bind(snapshot.is_reduce_only)
375            .bind(snapshot.is_quote_quantity)
376            .bind(snapshot.display_qty.map(|x| x.to_string()))
377            .bind(snapshot.emulation_trigger.map(|x| x.to_string()))
378            .bind(snapshot.trigger_instrument_id.map(|x| x.to_string()))
379            .bind(snapshot.contingency_type.map(|x| x.to_string()))
380            .bind(snapshot.order_list_id.map(|x| x.to_string()))
381            .bind(snapshot.linked_order_ids.map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
382            .bind(snapshot.parent_order_id.map(|x| x.to_string()))
383            .bind(snapshot.exec_algorithm_id.map(|x| x.to_string()))
384            .bind(snapshot.exec_algorithm_params.map(|x| serde_json::to_value(x).unwrap()))
385            .bind(snapshot.exec_spawn_id.map(|x| x.to_string()))
386            .bind(snapshot.tags.map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
387            .bind(snapshot.init_id.to_string())
388            .bind(snapshot.ts_init.to_string())
389            .bind(snapshot.ts_last.to_string())
390            .execute(&mut *transaction)
391            .await
392            .map(|_| ())
393            .map_err(|e| anyhow::anyhow!("Failed to insert into order table: {e}"))?;
394
395        transaction
396            .commit()
397            .await
398            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
399    }
400
401    /// Loads an `OrderSnapshot` entry by client order ID via the provided `pool`.
402    ///
403    /// # Errors
404    ///
405    /// Returns an error if the SQL SELECT or deserialization fails.
406    pub async fn load_order_snapshot(
407        pool: &PgPool,
408        client_order_id: &ClientOrderId,
409    ) -> anyhow::Result<Option<OrderSnapshot>> {
410        sqlx::query_as::<_, OrderSnapshotModel>(
411            r#"SELECT * FROM "order" WHERE client_order_id = $1"#,
412        )
413        .bind(client_order_id.to_string())
414        .fetch_optional(pool)
415        .await
416        .map(|model| model.map(|m| m.0))
417        .map_err(|e| anyhow::anyhow!("Failed to load order snapshot: {e}"))
418    }
419
420    /// Inserts or updates a `PositionSnapshot` entry via the provided `pool`.
421    ///
422    /// # Errors
423    ///
424    /// Returns an error if the SQL INSERT or UPDATE operation fails, or if beginning the transaction fails.
425    pub async fn add_position_snapshot(
426        pool: &PgPool,
427        snapshot: PositionSnapshot,
428    ) -> anyhow::Result<()> {
429        let mut transaction = pool.begin().await?;
430
431        // Insert trader if it does not exist
432        // TODO remove this when node and trader initialization is implemented
433        sqlx::query(
434            r#"
435            INSERT INTO "trader" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
436        "#,
437        )
438        .bind(snapshot.trader_id.to_string())
439        .execute(&mut *transaction)
440        .await
441        .map(|_| ())
442        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
443
444        sqlx::query(r#"
445            INSERT INTO "position" (
446                id, trader_id, strategy_id, instrument_id, account_id, opening_order_id, closing_order_id, entry, side, signed_qty, quantity, peak_qty,
447                quote_currency, base_currency, settlement_currency, avg_px_open, avg_px_close, realized_return, realized_pnl, unrealized_pnl, commissions,
448                duration_ns, ts_opened, ts_closed, ts_init, ts_last, created_at, updated_at
449            ) VALUES (
450                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
451                $21, $22, $23, $24, $25, $26, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
452            )
453            ON CONFLICT (id)
454            DO UPDATE
455            SET
456                trader_id = $2, strategy_id = $3, instrument_id = $4, account_id = $5, opening_order_id = $6, closing_order_id = $7, entry = $8, side = $9, signed_qty = $10, quantity = $11,
457                peak_qty = $12, quote_currency = $13, base_currency = $14, settlement_currency = $15, avg_px_open = $16, avg_px_close = $17, realized_return = $18, realized_pnl = $19, unrealized_pnl = $20,
458                commissions = $21, duration_ns = $22, ts_opened = $23, ts_closed = $24, ts_init = $25, ts_last = $26, updated_at = CURRENT_TIMESTAMP
459        "#)
460            .bind(snapshot.position_id.to_string())
461            .bind(snapshot.trader_id.to_string())
462            .bind(snapshot.strategy_id.to_string())
463            .bind(snapshot.instrument_id.to_string())
464            .bind(snapshot.account_id.to_string())
465            .bind(snapshot.opening_order_id.to_string())
466            .bind(snapshot.closing_order_id.map(|x| x.to_string()))
467            .bind(snapshot.entry.to_string())
468            .bind(snapshot.side.to_string())
469            .bind(snapshot.signed_qty)
470            .bind(snapshot.quantity.to_string())
471            .bind(snapshot.peak_qty.to_string())
472            .bind(snapshot.quote_currency.to_string())
473            .bind(snapshot.base_currency.map(|x| x.to_string()))
474            .bind(snapshot.settlement_currency.to_string())
475            .bind(snapshot.avg_px_open)
476            .bind(snapshot.avg_px_close)
477            .bind(snapshot.realized_return)
478            .bind(snapshot.realized_pnl.map(|x| x.to_string()))
479            .bind(snapshot.unrealized_pnl.map(|x| x.to_string()))
480            .bind(snapshot.commissions.iter().map(ToString::to_string).collect::<Vec<String>>())
481            .bind(snapshot.duration_ns.map(|x| x.to_string()))
482            .bind(snapshot.ts_opened.to_string())
483            .bind(snapshot.ts_closed.map(|x| x.to_string()))
484            .bind(snapshot.ts_init.to_string())
485            .bind(snapshot.ts_last.to_string())
486            .execute(&mut *transaction)
487            .await
488            .map(|_| ())
489            .map_err(|e| anyhow::anyhow!("Failed to insert into position table: {e}"))?;
490        transaction
491            .commit()
492            .await
493            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
494    }
495
496    /// Loads a `PositionSnapshot` entry by `position_id` via the provided `pool`.
497    ///
498    /// # Errors
499    ///
500    /// Returns an error if the SQL SELECT or deserialization fails.
501    pub async fn load_position_snapshot(
502        pool: &PgPool,
503        position_id: &PositionId,
504    ) -> anyhow::Result<Option<PositionSnapshot>> {
505        sqlx::query_as::<_, PositionSnapshotModel>(r#"SELECT * FROM "position" WHERE id = $1"#)
506            .bind(position_id.to_string())
507            .fetch_optional(pool)
508            .await
509            .map(|model| model.map(|m| m.0))
510            .map_err(|e| anyhow::anyhow!("Failed to load position snapshot: {e}"))
511    }
512
513    /// Checks if an `OrderInitialized` event exists for the given `client_order_id` via the provided `pool`.
514    ///
515    /// # Errors
516    ///
517    /// Returns an error if the SQL SELECT operation fails.
518    pub async fn check_if_order_initialized_exists(
519        pool: &PgPool,
520        client_order_id: ClientOrderId,
521    ) -> anyhow::Result<bool> {
522        sqlx::query(r#"
523            SELECT EXISTS(SELECT 1 FROM "order_event" WHERE client_order_id = $1 AND kind = 'OrderInitialized')
524        "#)
525            .bind(client_order_id.to_string())
526            .fetch_one(pool)
527            .await
528            .map(|row| row.get(0))
529            .map_err(|e| anyhow::anyhow!("Failed to check if order initialized exists: {e}"))
530    }
531
532    /// Checks if any account event exists for the given `account_id` via the provided `pool`.
533    ///
534    /// # Errors
535    ///
536    /// Returns an error if the SQL SELECT operation fails.
537    pub async fn check_if_account_event_exists(
538        pool: &PgPool,
539        account_id: AccountId,
540    ) -> anyhow::Result<bool> {
541        sqlx::query(
542            r#"
543            SELECT EXISTS(SELECT 1 FROM "account_event" WHERE account_id = $1)
544        "#,
545        )
546        .bind(account_id.to_string())
547        .fetch_one(pool)
548        .await
549        .map(|row| row.get(0))
550        .map_err(|e| anyhow::anyhow!("Failed to check if account event exists: {e}"))
551    }
552
553    /// Inserts or updates an order event entry via the provided `pool`.
554    ///
555    /// # Errors
556    ///
557    /// Returns an error if the SQL INSERT or UPDATE operation fails.
558    pub async fn add_order_event(
559        pool: &PgPool,
560        order_event: Box<dyn OrderEvent>,
561        client_id: Option<ClientId>,
562    ) -> anyhow::Result<()> {
563        let mut transaction = pool.begin().await?;
564
565        // Insert trader if it does not exist
566        // TODO remove this when node and trader initialization is implemented
567        sqlx::query(
568            r#"
569            INSERT INTO "trader" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
570        "#,
571        )
572        .bind(order_event.trader_id().to_string())
573        .execute(&mut *transaction)
574        .await
575        .map(|_| ())
576        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
577
578        // Insert client if it does not exist
579        // TODO remove this when client initialization is implemented
580        if let Some(client_id) = client_id {
581            sqlx::query(
582                r#"
583                INSERT INTO "client" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
584            "#,
585            )
586            .bind(client_id.to_string())
587            .execute(&mut *transaction)
588            .await
589            .map(|_| ())
590            .map_err(|e| anyhow::anyhow!("Failed to insert into client table: {e}"))?;
591        }
592
593        sqlx::query(r#"
594            INSERT INTO "order_event" (
595                id, kind, client_order_id, order_type, order_side, trader_id, client_id, reason, strategy_id, instrument_id, trade_id, currency, quantity, time_in_force, liquidity_side,
596                post_only, reduce_only, quote_quantity, reconciliation, price, last_px, last_qty, trigger_price, trigger_type, limit_offset, trailing_offset,
597                trailing_offset_type, expire_time, display_qty, emulation_trigger, trigger_instrument_id, contingency_type,
598                order_list_id, linked_order_ids, parent_order_id,
599                exec_algorithm_id, exec_spawn_id, venue_order_id, account_id, position_id, commission, ts_event, ts_init, created_at, updated_at
600            ) VALUES (
601                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20,
602                $21, $22, $23, $24, $25, $26::trailing_offset_type, $27, $28, $29, $30, $31, $32, $33, $34,
603                $35, $36, $37, $38, $39, $40, $41, $42, $43, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
604            )
605            ON CONFLICT (id)
606            DO UPDATE
607            SET
608                kind = $2, client_order_id = $3, order_type = $4, order_side=$5, trader_id = $6, client_id = $7, reason = $8, strategy_id = $9, instrument_id = $10, trade_id = $11, currency = $12,
609                quantity = $13, time_in_force = $14, liquidity_side = $15, post_only = $16, reduce_only = $17, quote_quantity = $18, reconciliation = $19, price = $20, last_px = $21,
610                last_qty = $22, trigger_price = $23, trigger_type = $24, limit_offset = $25, trailing_offset = $26, trailing_offset_type = $27, expire_time = $28, display_qty = $29,
611                emulation_trigger = $30, trigger_instrument_id = $31, contingency_type = $32, order_list_id = $33, linked_order_ids = $34, parent_order_id = $35, exec_algorithm_id = $36,
612                exec_spawn_id = $37, venue_order_id = $38, account_id = $39, position_id = $40, commission = $41, ts_event = $42, ts_init = $43, updated_at = CURRENT_TIMESTAMP
613
614        "#)
615            .bind(order_event.id().to_string())
616            .bind(order_event.type_name())
617            .bind(order_event.client_order_id().to_string())
618            .bind(order_event.order_type().map(|x| x.to_string()))
619            .bind(order_event.order_side().map(|x| x.to_string()))
620            .bind(order_event.trader_id().to_string())
621            .bind(client_id.map(|x| x.to_string()))
622            .bind(order_event.reason().map(|x| x.to_string()))
623            .bind(order_event.strategy_id().to_string())
624            .bind(order_event.instrument_id().to_string())
625            .bind(order_event.trade_id().map(|x| x.to_string()))
626            .bind(order_event.currency().map(|x| x.code.as_str()))
627            .bind(order_event.quantity().map(|x| x.to_string()))
628            .bind(order_event.time_in_force().map(|x| x.to_string()))
629            .bind(order_event.liquidity_side().map(|x| x.to_string()))
630            .bind(order_event.post_only())
631            .bind(order_event.reduce_only())
632            .bind(order_event.quote_quantity())
633            .bind(order_event.reconciliation())
634            .bind(order_event.price().map(|x| x.to_string()))
635            .bind(order_event.last_px().map(|x| x.to_string()))
636            .bind(order_event.last_qty().map(|x| x.to_string()))
637            .bind(order_event.trigger_price().map(|x| x.to_string()))
638            .bind(order_event.trigger_type().map(|x| x.to_string()))
639            .bind(order_event.limit_offset().map(|x| x.to_string()))
640            .bind(order_event.trailing_offset().map(|x| x.to_string()))
641            .bind(order_event.trailing_offset_type().map(TrailingOffsetTypeModel))
642            .bind(order_event.expire_time().map(|x| x.to_string()))
643            .bind(order_event.display_qty().map(|x| x.to_string()))
644            .bind(order_event.emulation_trigger().map(|x| x.to_string()))
645            .bind(order_event.trigger_instrument_id().map(|x| x.to_string()))
646            .bind(order_event.contingency_type().map(|x| x.to_string()))
647            .bind(order_event.order_list_id().map(|x| x.to_string()))
648            .bind(order_event.linked_order_ids().map(|x| x.iter().map(ToString::to_string).collect::<Vec<String>>()))
649            .bind(order_event.parent_order_id().map(|x| x.to_string()))
650            .bind(order_event.exec_algorithm_id().map(|x| x.to_string()))
651            .bind(order_event.exec_spawn_id().map(|x| x.to_string()))
652            .bind(order_event.venue_order_id().map(|x| x.to_string()))
653            .bind(order_event.account_id().map(|x| x.to_string()))
654            .bind(order_event.position_id().map(|x| x.to_string()))
655            .bind(order_event.commission().map(|x| x.to_string()))
656            .bind(order_event.ts_event().to_string())
657            .bind(order_event.ts_init().to_string())
658            .execute(&mut *transaction)
659            .await
660            .map(|_| ())
661            .map_err(|e| anyhow::anyhow!("Failed to insert into order_event table: {e}"))?;
662        transaction
663            .commit()
664            .await
665            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
666    }
667
668    /// Loads all order events for a `client_order_id` via the provided `pool`.
669    ///
670    /// # Errors
671    ///
672    /// Returns an error if the SQL SELECT or deserialization fails.
673    pub async fn load_order_events(
674        pool: &PgPool,
675        client_order_id: &ClientOrderId,
676    ) -> anyhow::Result<Vec<OrderEventAny>> {
677        sqlx::query_as::<_, OrderEventAnyModel>(r#"SELECT * FROM "order_event" event WHERE event.client_order_id = $1 ORDER BY created_at ASC"#)
678        .bind(client_order_id.to_string())
679        .fetch_all(pool)
680        .await
681        .map(|rows| rows.into_iter().map(|row| row.0).collect())
682        .map_err(|e| anyhow::anyhow!("Failed to load order events: {e}"))
683    }
684
685    /// Loads and assembles a complete `OrderAny` for a `client_order_id` via the provided `pool`.
686    ///
687    /// # Errors
688    ///
689    /// Returns an error if assembling events or SQL operations fail.
690    ///
691    pub async fn load_order(
692        pool: &PgPool,
693        client_order_id: &ClientOrderId,
694    ) -> anyhow::Result<Option<OrderAny>> {
695        let order_events = Self::load_order_events(pool, client_order_id).await;
696
697        match order_events {
698            Ok(order_events) => {
699                if order_events.is_empty() {
700                    return Ok(None);
701                }
702                let order = OrderAny::from_events(order_events).map_err(|e| {
703                    anyhow::anyhow!("Failed to assemble order {client_order_id} from events: {e}")
704                })?;
705                Ok(Some(order))
706            }
707            Err(e) => anyhow::bail!("Failed to load order events: {e}"),
708        }
709    }
710
711    /// Loads and assembles all `OrderAny` entries via the provided `pool`.
712    ///
713    /// # Errors
714    ///
715    /// Returns an error if loading events or SQL operations fail.
716    ///
717    pub async fn load_orders(pool: &PgPool) -> anyhow::Result<Vec<OrderAny>> {
718        let mut orders: Vec<OrderAny> = Vec::new();
719        let client_order_ids: Vec<ClientOrderId> = sqlx::query(
720            r#"
721            SELECT DISTINCT client_order_id FROM "order_event"
722        "#,
723        )
724        .fetch_all(pool)
725        .await
726        .map(|rows| {
727            rows.into_iter()
728                .map(|row| ClientOrderId::from(row.get::<&str, _>(0)))
729                .collect()
730        })
731        .map_err(|e| anyhow::anyhow!("Failed to load order ids: {e}"))?;
732        for id in client_order_ids {
733            let order = Self::load_order(pool, &id).await?;
734            if let Some(order) = order {
735                orders.push(order);
736            }
737        }
738        Ok(orders)
739    }
740
741    /// Replaces the fill event log for a `position_id` via the provided `pool`.
742    ///
743    /// # Errors
744    ///
745    /// Returns an error if the fill is invalid or if the SQL operations fail.
746    pub async fn add_position(
747        pool: &PgPool,
748        position_id: PositionId,
749        event: &OrderFilled,
750    ) -> anyhow::Result<()> {
751        let event_position_id = Self::event_position_id(event)?;
752        if event_position_id != position_id {
753            anyhow::bail!(
754                "Cannot persist position event {} for mismatched position_id: expected {}, was {}",
755                event.event_id,
756                position_id,
757                event_position_id
758            );
759        }
760
761        let mut transaction = pool.begin().await?;
762
763        sqlx::query(r#"DELETE FROM "position_event" WHERE position_id = $1"#)
764            .bind(position_id.to_string())
765            .execute(&mut *transaction)
766            .await
767            .map(|_| ())
768            .map_err(|e| anyhow::anyhow!("Failed to delete position_event rows: {e}"))?;
769
770        Self::insert_position_event(&mut transaction, event).await?;
771        transaction
772            .commit()
773            .await
774            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
775    }
776
777    /// Appends a fill event for a `Position` via the provided `pool`.
778    ///
779    /// # Errors
780    ///
781    /// Returns an error if the fill is invalid or if the SQL operations fail.
782    pub async fn update_position(pool: &PgPool, event: &OrderFilled) -> anyhow::Result<()> {
783        let mut transaction = pool.begin().await?;
784
785        Self::insert_position_event(&mut transaction, event).await?;
786        transaction
787            .commit()
788            .await
789            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
790    }
791
792    /// Appends an `OrderFilled` event to the position event log via the provided `pool`.
793    ///
794    /// # Errors
795    ///
796    /// Returns an error if the SQL INSERT operation fails.
797    pub async fn add_position_event(pool: &PgPool, event: &OrderFilled) -> anyhow::Result<()> {
798        let mut transaction = pool.begin().await?;
799
800        Self::insert_position_event(&mut transaction, event).await?;
801        transaction
802            .commit()
803            .await
804            .map_err(|e| anyhow::anyhow!("Failed to commit transaction: {e}"))
805    }
806
807    /// Loads all fill events for a `position_id` via the provided `pool`.
808    ///
809    /// # Errors
810    ///
811    /// Returns an error if the SQL SELECT or deserialization fails.
812    pub async fn load_position_events(
813        pool: &PgPool,
814        position_id: &PositionId,
815    ) -> anyhow::Result<Vec<OrderFilled>> {
816        sqlx::query_as::<_, OrderFilledModel>(
817            r#"
818            SELECT *
819            FROM "position_event"
820            WHERE position_id = $1
821            ORDER BY event_sequence ASC
822        "#,
823        )
824        .bind(position_id.to_string())
825        .fetch_all(pool)
826        .await
827        .map(|rows| rows.into_iter().map(|row| row.0).collect())
828        .map_err(|e| anyhow::anyhow!("Failed to load position events: {e}"))
829    }
830
831    /// Loads and replays a complete `Position` for a `position_id` via the provided `pool`.
832    ///
833    /// # Errors
834    ///
835    /// Returns an error if loading events, loading instruments, or replaying fills fails.
836    pub async fn load_position(
837        pool: &PgPool,
838        position_id: &PositionId,
839    ) -> anyhow::Result<Option<Position>> {
840        let fills = Self::load_position_events(pool, position_id).await?;
841        let Some((first_fill, remaining_fills)) = fills.split_first() else {
842            return Ok(None);
843        };
844        let Some(instrument) = Self::load_instrument(pool, &first_fill.instrument_id).await? else {
845            log::error!(
846                "Instrument not found for position {position_id}: {}",
847                first_fill.instrument_id
848            );
849            return Ok(None);
850        };
851
852        let mut position = Position::new(&instrument, *first_fill);
853        for fill in remaining_fills {
854            if position.trade_ids().contains(&fill.trade_id) {
855                anyhow::bail!(
856                    "Duplicate fill event for position {position_id}: {}",
857                    fill.trade_id
858                );
859            }
860            position.apply(fill);
861        }
862
863        Ok(Some(position))
864    }
865
866    /// Loads and replays all `Position` entries via the provided `pool`.
867    ///
868    /// # Errors
869    ///
870    /// Returns an error if loading position IDs or replaying any position fails.
871    pub async fn load_positions(pool: &PgPool) -> anyhow::Result<Vec<Position>> {
872        let position_ids: Vec<PositionId> = sqlx::query(
873            r#"
874            SELECT DISTINCT position_id
875            FROM "position_event"
876            ORDER BY position_id ASC
877        "#,
878        )
879        .fetch_all(pool)
880        .await
881        .map(|rows| {
882            rows.into_iter()
883                .map(|row| PositionId::from(row.get::<&str, _>(0)))
884                .collect()
885        })
886        .map_err(|e| anyhow::anyhow!("Failed to load position ids: {e}"))?;
887
888        let mut positions = Vec::new();
889
890        for id in position_ids {
891            match Self::load_position(pool, &id).await {
892                Ok(Some(position)) => positions.push(position),
893                Ok(None) => log::error!("Position not found: {id}"),
894                Err(e) => log::error!("Failed to load position {id}: {e}"),
895            }
896        }
897
898        Ok(positions)
899    }
900
901    async fn insert_position_event(
902        transaction: &mut Transaction<'_, Postgres>,
903        event: &OrderFilled,
904    ) -> anyhow::Result<()> {
905        let position_id = Self::event_position_id(event)?;
906
907        sqlx::query(
908            r#"
909            INSERT INTO "trader" (id)
910            VALUES ($1)
911            ON CONFLICT (id) DO NOTHING
912        "#,
913        )
914        .bind(event.trader_id.to_string())
915        .execute(&mut **transaction)
916        .await
917        .map(|_| ())
918        .map_err(|e| anyhow::anyhow!("Failed to insert into trader table: {e}"))?;
919
920        sqlx::query(
921            r#"
922            INSERT INTO "position_event" (
923                id, kind, trader_id, strategy_id, instrument_id, client_order_id, venue_order_id,
924                account_id, trade_id, currency, order_type, order_side, last_px, last_qty,
925                liquidity_side, position_id, commission, ts_event, ts_init, created_at, updated_at
926            ) VALUES (
927                $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17,
928                $18, $19, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
929            )
930        "#,
931        )
932        .bind(event.event_id.to_string())
933        .bind("OrderFilled")
934        .bind(event.trader_id.to_string())
935        .bind(event.strategy_id.to_string())
936        .bind(event.instrument_id.to_string())
937        .bind(event.client_order_id.to_string())
938        .bind(event.venue_order_id.to_string())
939        .bind(event.account_id.to_string())
940        .bind(event.trade_id.to_string())
941        .bind(event.currency.code.as_str())
942        .bind(event.order_type.to_string())
943        .bind(event.order_side.to_string())
944        .bind(event.last_px.to_string())
945        .bind(event.last_qty.to_string())
946        .bind(event.liquidity_side.to_string())
947        .bind(position_id.to_string())
948        .bind(event.commission.map(|commission| commission.to_string()))
949        .bind(event.ts_event.to_string())
950        .bind(event.ts_init.to_string())
951        .execute(&mut **transaction)
952        .await
953        .map(|_| ())
954        .map_err(|e| anyhow::anyhow!("Failed to insert into position_event table: {e}"))
955    }
956
957    fn event_position_id(event: &OrderFilled) -> anyhow::Result<PositionId> {
958        event.position_id.ok_or_else(|| {
959            anyhow::anyhow!(
960                "Cannot persist position event with no position_id: {}",
961                event.event_id
962            )
963        })
964    }
965
966    /// Inserts or updates an `AccountState` event via the provided `pool`.
967    ///
968    /// # Errors
969    ///
970    /// Returns an error if the SQL INSERT or UPDATE operation fails.
971    pub async fn add_account(
972        pool: &PgPool,
973        updated: bool,
974        account_event: AccountState,
975    ) -> anyhow::Result<()> {
976        if updated {
977            let exists =
978                Self::check_if_account_event_exists(pool, account_event.account_id).await?;
979
980            if !exists {
981                anyhow::bail!(
982                    "Account event does not exist for account: {}",
983                    account_event.account_id
984                );
985            }
986        }
987
988        let mut transaction = pool.begin().await?;
989        let balances = serde_json::to_value::<Vec<AccountBalance>>(account_event.balances)
990            .map_err(|e| anyhow::anyhow!("Failed to serialize account balances: {e}"))?;
991        let margins = serde_json::to_value::<Vec<MarginBalance>>(account_event.margins)
992            .map_err(|e| anyhow::anyhow!("Failed to serialize margin balances: {e}"))?;
993
994        sqlx::query(
995            r#"
996            INSERT INTO "account" (id) VALUES ($1) ON CONFLICT (id) DO NOTHING
997        "#,
998        )
999        .bind(account_event.account_id.to_string())
1000        .execute(&mut *transaction)
1001        .await
1002        .map(|_| ())
1003        .map_err(|e| anyhow::anyhow!("Failed to insert into account table: {e}"))?;
1004
1005        sqlx::query(r#"
1006            INSERT INTO "account_event" (
1007                id, kind, account_id, base_currency, balances, margins, is_reported, ts_event, ts_init, created_at, updated_at
1008            ) VALUES (
1009                $1, $2, $3, $4, $5, $6, $7, $8, $9, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1010            )
1011            ON CONFLICT (id)
1012            DO UPDATE
1013            SET
1014                kind = $2, account_id = $3, base_currency = $4, balances = $5, margins = $6, is_reported = $7,
1015                ts_event = $8, ts_init = $9, updated_at = CURRENT_TIMESTAMP
1016        "#)
1017            .bind(account_event.event_id.to_string())
1018            .bind(account_event.account_type.to_string())
1019            .bind(account_event.account_id.to_string())
1020            .bind(account_event.base_currency.map(|x| x.code.as_str()))
1021            .bind(balances)
1022            .bind(margins)
1023            .bind(account_event.is_reported)
1024            .bind(account_event.ts_event.to_string())
1025            .bind(account_event.ts_init.to_string())
1026            .execute(&mut *transaction)
1027            .await
1028            .map(|_| ())
1029            .map_err(|e| anyhow::anyhow!("Failed to insert into account_event table: {e}"))?;
1030        transaction
1031            .commit()
1032            .await
1033            .map_err(|e| anyhow::anyhow!("Failed to commit add_account transaction: {e}"))
1034    }
1035
1036    /// Loads all account events for `account_id` via the provided `pool`.
1037    ///
1038    /// # Errors
1039    ///
1040    /// Returns an error if the SQL SELECT or deserialization fails.
1041    pub async fn load_account_events(
1042        pool: &PgPool,
1043        account_id: &AccountId,
1044    ) -> anyhow::Result<Vec<AccountState>> {
1045        sqlx::query_as::<_, AccountEventModel>(
1046            r#"SELECT * FROM "account_event" WHERE account_id = $1 ORDER BY created_at ASC"#,
1047        )
1048        .bind(account_id.to_string())
1049        .fetch_all(pool)
1050        .await
1051        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1052        .map_err(|e| anyhow::anyhow!("Failed to load account events: {e}"))
1053    }
1054
1055    /// Loads and assembles a complete `AccountAny` for `account_id` via the provided `pool`.
1056    ///
1057    /// # Errors
1058    ///
1059    /// Returns an error if assembling events or SQL operations fail.
1060    ///
1061    pub async fn load_account(
1062        pool: &PgPool,
1063        account_id: &AccountId,
1064    ) -> anyhow::Result<Option<AccountAny>> {
1065        let account_events = Self::load_account_events(pool, account_id).await;
1066        match account_events {
1067            Ok(account_events) => {
1068                if account_events.is_empty() {
1069                    return Ok(None);
1070                }
1071                let account = AccountAny::from_events(&account_events).map_err(|e| {
1072                    anyhow::anyhow!("Failed to assemble account {account_id} from events: {e}")
1073                })?;
1074                Ok(Some(account))
1075            }
1076            Err(e) => anyhow::bail!("Failed to load account events: {e}"),
1077        }
1078    }
1079
1080    /// Loads and assembles all `AccountAny` entries via the provided `pool`.
1081    ///
1082    /// # Errors
1083    ///
1084    /// Returns an error if loading events or SQL operations fail.
1085    ///
1086    pub async fn load_accounts(pool: &PgPool) -> anyhow::Result<Vec<AccountAny>> {
1087        let mut accounts: Vec<AccountAny> = Vec::new();
1088        let account_ids: Vec<AccountId> = sqlx::query(
1089            r#"
1090            SELECT DISTINCT account_id FROM "account_event"
1091        "#,
1092        )
1093        .fetch_all(pool)
1094        .await
1095        .map(|rows| {
1096            rows.into_iter()
1097                .map(|row| AccountId::from(row.get::<&str, _>(0)))
1098                .collect()
1099        })
1100        .map_err(|e| anyhow::anyhow!("Failed to load account ids: {e}"))?;
1101        for id in account_ids {
1102            let account = Self::load_account(pool, &id).await?;
1103            if let Some(account) = account {
1104                accounts.push(account);
1105            }
1106        }
1107        Ok(accounts)
1108    }
1109
1110    /// Inserts a `TradeTick` entry via the provided `pool`.
1111    ///
1112    /// # Errors
1113    ///
1114    /// Returns an error if the SQL INSERT operation fails.
1115    pub async fn add_trade(pool: &PgPool, trade: &TradeTick) -> anyhow::Result<()> {
1116        sqlx::query(r#"
1117            INSERT INTO "trade" (
1118                instrument_id, price, quantity, aggressor_side, venue_trade_id,
1119                ts_event, ts_init, created_at, updated_at
1120            ) VALUES (
1121                $1, $2, $3, $4::aggressor_side, $5, $6, $7, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1122            )
1123            ON CONFLICT (id)
1124            DO UPDATE
1125            SET
1126                instrument_id = $1, price = $2, quantity = $3, aggressor_side = $4, venue_trade_id = $5,
1127                ts_event = $6, ts_init = $7, updated_at = CURRENT_TIMESTAMP
1128        "#)
1129            .bind(trade.instrument_id.to_string())
1130            .bind(trade.price.to_string())
1131            .bind(trade.size.to_string())
1132            .bind(AggressorSideModel(trade.aggressor_side))
1133            .bind(trade.trade_id.to_string())
1134            .bind(trade.ts_event.to_string())
1135            .bind(trade.ts_init.to_string())
1136            .execute(pool)
1137            .await
1138            .map(|_| ())
1139            .map_err(|e| anyhow::anyhow!("Failed to insert into trade table: {e}"))
1140    }
1141
1142    /// Loads all `TradeTick` entries for `instrument_id` via the provided `pool`.
1143    ///
1144    /// # Errors
1145    ///
1146    /// Returns an error if the SQL SELECT or deserialization fails.
1147    pub async fn load_trades(
1148        pool: &PgPool,
1149        instrument_id: &InstrumentId,
1150    ) -> anyhow::Result<Vec<TradeTick>> {
1151        sqlx::query_as::<_, TradeTickModel>(
1152            r#"SELECT * FROM "trade" WHERE instrument_id = $1 ORDER BY ts_event ASC"#,
1153        )
1154        .bind(instrument_id.to_string())
1155        .fetch_all(pool)
1156        .await
1157        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1158        .map_err(|e| anyhow::anyhow!("Failed to load trades: {e}"))
1159    }
1160
1161    /// Inserts a `QuoteTick` entry via the provided `pool`.
1162    ///
1163    /// # Errors
1164    ///
1165    /// Returns an error if the SQL INSERT operation fails.
1166    pub async fn add_quote(pool: &PgPool, quote: &QuoteTick) -> anyhow::Result<()> {
1167        sqlx::query(r#"
1168            INSERT INTO "quote" (
1169                instrument_id, bid_price, ask_price, bid_size, ask_size, ts_event, ts_init, created_at, updated_at
1170            ) VALUES (
1171                $1, $2, $3, $4, $5, $6, $7, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1172            )
1173            ON CONFLICT (id)
1174            DO UPDATE
1175            SET
1176                instrument_id = $1, bid_price = $2, ask_price = $3, bid_size = $4, ask_size = $5,
1177                ts_event = $6, ts_init = $7, updated_at = CURRENT_TIMESTAMP
1178        "#)
1179            .bind(quote.instrument_id.to_string())
1180            .bind(quote.bid_price.to_string())
1181            .bind(quote.ask_price.to_string())
1182            .bind(quote.bid_size.to_string())
1183            .bind(quote.ask_size.to_string())
1184            .bind(quote.ts_event.to_string())
1185            .bind(quote.ts_init.to_string())
1186            .execute(pool)
1187            .await
1188            .map(|_| ())
1189            .map_err(|e| anyhow::anyhow!("Failed to insert into quote table: {e}"))
1190    }
1191
1192    /// Loads all `QuoteTick` entries for `instrument_id` via the provided `pool`.
1193    ///
1194    /// # Errors
1195    ///
1196    /// Returns an error if the SQL SELECT or deserialization fails.
1197    pub async fn load_quotes(
1198        pool: &PgPool,
1199        instrument_id: &InstrumentId,
1200    ) -> anyhow::Result<Vec<QuoteTick>> {
1201        sqlx::query_as::<_, QuoteTickModel>(
1202            r#"SELECT * FROM "quote" WHERE instrument_id = $1 ORDER BY ts_event ASC"#,
1203        )
1204        .bind(instrument_id.to_string())
1205        .fetch_all(pool)
1206        .await
1207        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1208        .map_err(|e| anyhow::anyhow!("Failed to load quotes: {e}"))
1209    }
1210
1211    /// Inserts a `Bar` entry via the provided `pool`.
1212    ///
1213    /// # Errors
1214    ///
1215    /// Returns an error if the SQL INSERT operation fails.
1216    pub async fn add_bar(pool: &PgPool, bar: &Bar) -> anyhow::Result<()> {
1217        let bar_step = i32::try_from(bar.bar_type.spec().step.get())
1218            .map_err(|e| anyhow::anyhow!("invalid bar step: {e}"))?;
1219
1220        sqlx::query(r#"
1221            INSERT INTO "bar" (
1222                instrument_id, step, bar_aggregation, price_type, aggregation_source, open, high, low, close, volume, ts_event, ts_init, created_at, updated_at
1223            ) VALUES (
1224                $1, $2, $3::bar_aggregation, $4::price_type, $5::aggregation_source, $6, $7, $8, $9, $10, $11, $12, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1225            )
1226            ON CONFLICT (id)
1227            DO UPDATE
1228            SET
1229                instrument_id = $1, step = $2, bar_aggregation = $3::bar_aggregation, price_type = $4::price_type, aggregation_source = $5::aggregation_source,
1230                open = $6, high = $7, low = $8, close = $9, volume = $10, ts_event = $11, ts_init = $12, updated_at = CURRENT_TIMESTAMP
1231        "#)
1232            .bind(bar.bar_type.instrument_id().to_string())
1233            .bind(bar_step)
1234            .bind(BarAggregationModel(bar.bar_type.spec().aggregation))
1235            .bind(PriceTypeModel(bar.bar_type.spec().price_type))
1236            .bind(AggregationSourceModel(bar.bar_type.aggregation_source()))
1237            .bind(bar.open.to_string())
1238            .bind(bar.high.to_string())
1239            .bind(bar.low.to_string())
1240            .bind(bar.close.to_string())
1241            .bind(bar.volume.to_string())
1242            .bind(bar.ts_event.to_string())
1243            .bind(bar.ts_init.to_string())
1244            .execute(pool)
1245            .await
1246            .map(|_| ())
1247            .map_err(|e| anyhow::anyhow!("Failed to insert into bar table: {e}"))
1248    }
1249
1250    /// Loads all `Bar` entries for `instrument_id` via the provided `pool`.
1251    ///
1252    /// # Errors
1253    ///
1254    /// Returns an error if the SQL SELECT or deserialization fails.
1255    pub async fn load_bars(
1256        pool: &PgPool,
1257        instrument_id: &InstrumentId,
1258    ) -> anyhow::Result<Vec<Bar>> {
1259        sqlx::query_as::<_, BarModel>(
1260            r#"SELECT * FROM "bar" WHERE instrument_id = $1 ORDER BY ts_event ASC"#,
1261        )
1262        .bind(instrument_id.to_string())
1263        .fetch_all(pool)
1264        .await
1265        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1266        .map_err(|e| anyhow::anyhow!("Failed to load bars: {e}"))
1267    }
1268
1269    /// Loads all distinct client order IDs from order events via the provided `pool`.
1270    ///
1271    /// # Errors
1272    ///
1273    /// Returns an error if the SQL SELECT or iteration fails.
1274    pub async fn load_distinct_order_event_client_ids(
1275        pool: &PgPool,
1276    ) -> anyhow::Result<AHashMap<ClientOrderId, ClientId>> {
1277        let mut map: AHashMap<ClientOrderId, ClientId> = AHashMap::new();
1278        let result = sqlx::query_as::<_, OrderEventOrderClientIdCombination>(
1279            r#"
1280            SELECT DISTINCT ON (client_order_id)
1281                client_order_id AS "client_order_id",
1282                client_id AS "client_id"
1283            FROM "order_event"
1284            WHERE client_id IS NOT NULL
1285            ORDER BY client_order_id, created_at DESC
1286        "#,
1287        )
1288        .fetch_all(pool)
1289        .await
1290        .map_err(|e| anyhow::anyhow!("Failed to load account ids: {e}"))?;
1291
1292        for id in result {
1293            map.insert(id.client_order_id, id.client_id);
1294        }
1295        Ok(map)
1296    }
1297
1298    /// Inserts or updates an order ID to position ID index entry via the provided `pool`.
1299    ///
1300    /// # Errors
1301    ///
1302    /// Returns an error if the SQL INSERT or UPDATE operation fails.
1303    pub async fn index_order_position(
1304        pool: &PgPool,
1305        client_order_id: ClientOrderId,
1306        position_id: PositionId,
1307    ) -> anyhow::Result<()> {
1308        sqlx::query(
1309            r#"
1310            INSERT INTO "order_position_index" (
1311                client_order_id, position_id, created_at, updated_at
1312            ) VALUES (
1313                $1, $2, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1314            )
1315            ON CONFLICT (client_order_id)
1316            DO UPDATE
1317            SET
1318                position_id = $2, updated_at = CURRENT_TIMESTAMP
1319        "#,
1320        )
1321        .bind(client_order_id.to_string())
1322        .bind(position_id.to_string())
1323        .execute(pool)
1324        .await
1325        .map(|_| ())
1326        .map_err(|e| anyhow::anyhow!("Failed to insert into order_position_index table: {e}"))
1327    }
1328
1329    /// Loads the order ID to position ID index via the provided `pool`.
1330    ///
1331    /// # Errors
1332    ///
1333    /// Returns an error if the SQL SELECT or iteration fails.
1334    pub async fn load_index_order_position(
1335        pool: &PgPool,
1336    ) -> anyhow::Result<AHashMap<ClientOrderId, PositionId>> {
1337        let mut map: AHashMap<ClientOrderId, PositionId> = AHashMap::new();
1338        let result = sqlx::query_as::<_, OrderPositionIndexRow>(
1339            r#"
1340            SELECT
1341                client_order_id AS "client_order_id",
1342                position_id AS "position_id"
1343            FROM "order_position_index"
1344        "#,
1345        )
1346        .fetch_all(pool)
1347        .await
1348        .map_err(|e| anyhow::anyhow!("Failed to load order position index: {e}"))?;
1349
1350        for row in result {
1351            map.insert(row.client_order_id, row.position_id);
1352        }
1353        Ok(map)
1354    }
1355
1356    /// Inserts a `Signal` entry via the provided `pool`.
1357    ///
1358    /// # Errors
1359    ///
1360    /// Returns an error if the SQL INSERT operation fails.
1361    pub async fn add_signal(pool: &PgPool, signal: &Signal) -> anyhow::Result<()> {
1362        sqlx::query(
1363            r#"
1364            INSERT INTO "signal" (
1365                name, value, ts_event, ts_init, created_at, updated_at
1366            ) VALUES (
1367                $1, $2, $3, $4, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP
1368            )
1369            ON CONFLICT (id)
1370            DO UPDATE
1371            SET
1372                name = $1, value = $2, ts_event = $3, ts_init = $4,
1373                updated_at = CURRENT_TIMESTAMP
1374        "#,
1375        )
1376        .bind(signal.name.to_string())
1377        .bind(signal.value.clone())
1378        .bind(signal.ts_event.to_string())
1379        .bind(signal.ts_init.to_string())
1380        .execute(pool)
1381        .await
1382        .map(|_| ())
1383        .map_err(|e| anyhow::anyhow!("Failed to insert into signal table: {e}"))
1384    }
1385
1386    /// Loads all `Signal` entries by `name` via the provided `pool`.
1387    ///
1388    /// # Errors
1389    ///
1390    /// Returns an error if the SQL SELECT or deserialization fails.
1391    pub async fn load_signals(pool: &PgPool, name: &str) -> anyhow::Result<Vec<Signal>> {
1392        sqlx::query_as::<_, SignalModel>(
1393            r#"SELECT * FROM "signal" WHERE name = $1 ORDER BY ts_init ASC"#,
1394        )
1395        .bind(name)
1396        .fetch_all(pool)
1397        .await
1398        .map(|rows| rows.into_iter().map(|row| row.0).collect())
1399        .map_err(|e| anyhow::anyhow!("Failed to load signals: {e}"))
1400    }
1401
1402    /// Inserts a `CustomData` entry via the provided `pool`.
1403    ///
1404    /// Serializes the model `CustomData` to full JSON and stores it in the JSONB `value` column.
1405    ///
1406    /// # Errors
1407    ///
1408    /// Returns an error if the SQL INSERT operation fails.
1409    pub async fn add_custom_data(pool: &PgPool, data: &CustomData) -> anyhow::Result<()> {
1410        let json_bytes = serde_json::to_vec(data)
1411            .map_err(|e| anyhow::anyhow!("CustomData must be valid JSON: {e}"))?;
1412        let value_json: serde_json::Value = serde_json::from_slice(&json_bytes)
1413            .map_err(|e| anyhow::anyhow!("CustomData value must be valid JSON: {e}"))?;
1414        let data_type_obj = value_json
1415            .get("data_type")
1416            .and_then(|v| v.as_object())
1417            .ok_or_else(|| anyhow::anyhow!("CustomData JSON missing data_type"))?;
1418        let data_type_name = data_type_obj
1419            .get("type_name")
1420            .and_then(|v| v.as_str())
1421            .unwrap_or("");
1422        let metadata_json = data_type_obj
1423            .get("metadata")
1424            .cloned()
1425            .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new()));
1426        let identifier = data_type_obj
1427            .get("identifier")
1428            .and_then(|v| v.as_str())
1429            .unwrap_or("");
1430        sqlx::query(
1431            r#"
1432            INSERT INTO "custom" (data_type, metadata, identifier, value, ts_event, ts_init, created_at, updated_at)
1433            VALUES ($1, $2, $3, $4, $5, $6, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
1434            ON CONFLICT (id)
1435            DO UPDATE SET
1436                data_type = EXCLUDED.data_type,
1437                metadata = EXCLUDED.metadata,
1438                identifier = EXCLUDED.identifier,
1439                value = EXCLUDED.value,
1440                ts_event = EXCLUDED.ts_event,
1441                ts_init = EXCLUDED.ts_init,
1442                updated_at = CURRENT_TIMESTAMP
1443        "#,
1444        )
1445        .bind(data_type_name)
1446        .bind(&metadata_json)
1447        .bind(identifier)
1448        .bind(&value_json)
1449        .bind(
1450            value_json
1451                .get("ts_event")
1452                .and_then(serde_json::Value::as_u64)
1453                .unwrap_or_else(|| data.ts_init().as_u64())
1454                .to_string(),
1455        )
1456        .bind(data.ts_init().to_string())
1457        .execute(pool)
1458        .await
1459        .map(|_| ())
1460        .map_err(|e| anyhow::anyhow!("Failed to insert into custom table: {e}"))
1461    }
1462
1463    /// Loads all `CustomData` entries of `data_type` via the provided `pool`.
1464    ///
1465    /// Filters by `data_type`, `metadata`, and `identifier` columns to match the requested data type.
1466    ///
1467    /// # Errors
1468    ///
1469    /// Returns an error if the SQL SELECT or deserialization fails.
1470    pub async fn load_custom_data(
1471        pool: &PgPool,
1472        data_type: &DataType,
1473    ) -> anyhow::Result<Vec<CustomData>> {
1474        let metadata_json = data_type.metadata().as_ref().map_or(
1475            Ok(serde_json::Value::Object(serde_json::Map::new())),
1476            serde_json::to_value,
1477        )?;
1478
1479        let type_name = data_type.type_name();
1480        let short_type = type_name.rsplit([':', '.']).next().unwrap_or(type_name);
1481
1482        let rows = match data_type.identifier() {
1483            Some(identifier) => {
1484                sqlx::query(
1485                    r#"SELECT value, ts_event, ts_init FROM "custom"
1486                   WHERE (data_type = $1 OR data_type = $2)
1487                     AND metadata = $3
1488                     AND identifier = $4
1489                   ORDER BY ts_init ASC"#,
1490                )
1491                .bind(type_name)
1492                .bind(short_type)
1493                .bind(&metadata_json)
1494                .bind(identifier)
1495                .fetch_all(pool)
1496                .await
1497            }
1498            None => {
1499                sqlx::query(
1500                    r#"SELECT value, ts_event, ts_init FROM "custom"
1501                   WHERE (data_type = $1 OR data_type = $2)
1502                     AND metadata = $3
1503                     AND identifier = ''
1504                   ORDER BY ts_init ASC"#,
1505                )
1506                .bind(type_name)
1507                .bind(short_type)
1508                .bind(&metadata_json)
1509                .fetch_all(pool)
1510                .await
1511            }
1512        }
1513        .map_err(|e| anyhow::anyhow!("Failed to load custom data: {e}"))?;
1514
1515        let mut results = Vec::with_capacity(rows.len());
1516        for row in rows {
1517            let value_json: serde_json::Value = row.try_get("value")?;
1518            let json_bytes = serde_json::to_vec(&value_json)
1519                .map_err(|e| anyhow::anyhow!("Failed to serialize JSON: {e}"))?;
1520            let custom =
1521                CustomData::from_json_bytes(&json_bytes).map_err(|e| anyhow::anyhow!("{e}"))?;
1522            results.push(custom);
1523        }
1524        Ok(results)
1525    }
1526}
nautilus_infrastructure/sql/queries.rs

nautilus_infrastructure/sql/
queries.rs
