Add psql cache database - instruments

nautilus_core/infrastructure/Cargo.toml

@@ -13,7 +13,7 @@ crate-type = ["rlib", "cdylib"]
 [dependencies]
 nautilus-common = { path = "../common", features = ["python"] }
 nautilus-core = { path = "../core" , features = ["python"] }
-nautilus-model = { path = "../model" , features = ["python"] }
+nautilus-model = { path = "../model" , features = ["python", "stubs"] }
 anyhow = { workspace = true }
 pyo3 = { workspace = true, optional = true }
 rmp-serde = { workspace = true }

nautilus_core/infrastructure/src/python/sql/cache_database.rs

@@ -17,7 +17,11 @@ use std::collections::HashMap;
 
 use nautilus_common::runtime::get_runtime;
 use nautilus_core::python::to_pyruntime_err;
-use nautilus_model::types::currency::Currency;
+use nautilus_model::{
+    identifiers::instrument_id::InstrumentId,
+    python::instruments::{convert_instrument_any_to_pyobject, convert_pyobject_to_instrument_any},
+    types::currency::Currency,
+};
 use pyo3::prelude::*;
 
 use crate::sql::{
@@ -74,6 +78,50 @@ impl PostgresCacheDatabase {
         result.map_err(to_pyruntime_err)
     }
 
+    #[pyo3(name = "load_instrument")]
+    fn py_load_instrument(
+        slf: PyRef<'_, Self>,
+        instrument_id: InstrumentId,
+        py: Python<'_>,
+    ) -> PyResult<Option<PyObject>> {
+        get_runtime().block_on(async {
+            let result = DatabaseQueries::load_instrument(&slf.pool, instrument_id)
+                .await
+                .unwrap();
+            match result {
+                Some(instrument) => {
+                    let py_object = convert_instrument_any_to_pyobject(py, instrument)?;
+                    Ok(Some(py_object))
+                }
+                None => Ok(None),
+            }
+        })
+    }
+
+    #[pyo3(name = "load_instruments")]
+    fn py_load_instruments(slf: PyRef<'_, Self>, py: Python<'_>) -> PyResult<Vec<PyObject>> {
+        get_runtime().block_on(async {
+            let result = DatabaseQueries::load_instruments(&slf.pool).await.unwrap();
+            let mut instruments = Vec::new();
+            for instrument in result {
+                let py_object = convert_instrument_any_to_pyobject(py, instrument)?;
+                instruments.push(py_object);
+            }
+            Ok(instruments)
+        })
+    }
+
+    #[pyo3(name = "add_instrument")]
+    fn py_add_instrument(
+        slf: PyRef<'_, Self>,
+        instrument: PyObject,
+        py: Python<'_>,
+    ) -> PyResult<()> {
+        let instrument_any = convert_pyobject_to_instrument_any(py, instrument)?;
+        let result = get_runtime().block_on(async { slf.add_instrument(instrument_any).await });
+        result.map_err(to_pyruntime_err)
+    }
+
     #[pyo3(name = "flush_db")]
     fn py_drop_schema(slf: PyRef<'_, Self>) -> PyResult<()> {
         let result =

nautilus_core/infrastructure/src/sql/cache_database.rs

@@ -18,7 +18,9 @@ use std::{
     time::{Duration, Instant},
 };
 
-use nautilus_model::types::currency::Currency;
+use nautilus_model::{
+    identifiers::instrument_id::InstrumentId, instruments::InstrumentAny, types::currency::Currency,
+};
 use sqlx::{postgres::PgConnectOptions, PgPool};
 use tokio::{
     sync::mpsc::{channel, error::TryRecvError, Receiver, Sender},
@@ -34,17 +36,19 @@ use crate::sql::{
 #[derive(Debug)]
 #[cfg_attr(
     feature = "python",
-    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.persistence")
+    pyo3::pyclass(module = "nautilus_trader.core.nautilus_pyo3.infrastructure")
 )]
 pub struct PostgresCacheDatabase {
     pub pool: PgPool,
     tx: Sender<DatabaseQuery>,
 }
 
+#[allow(clippy::large_enum_variant)]
 #[derive(Debug, Clone)]
 pub enum DatabaseQuery {
     Add(String, Vec<u8>),
     AddCurrency(Currency),
+    AddInstrument(InstrumentAny),
 }
 
 fn get_buffer_interval() -> Duration {
@@ -60,6 +64,64 @@ async fn drain_buffer(pool: &PgPool, buffer: &mut VecDeque<DatabaseQuery>) {
             DatabaseQuery::AddCurrency(currency) => {
                 DatabaseQueries::add_currency(pool, currency).await.unwrap();
             }
+            DatabaseQuery::AddInstrument(instrument) => match instrument {
+                InstrumentAny::CryptoFuture(crypto_future) => {
+                    DatabaseQueries::add_instrument(pool, "CRYPTO_FUTURE", Box::new(crypto_future))
+                        .await
+                        .unwrap()
+                }
+                InstrumentAny::CryptoPerpetual(crypto_perpetual) => {
+                    DatabaseQueries::add_instrument(
+                        pool,
+                        "CRYPTO_PERPETUAL",
+                        Box::new(crypto_perpetual),
+                    )
+                    .await
+                    .unwrap()
+                }
+                InstrumentAny::CurrencyPair(currency_pair) => {
+                    DatabaseQueries::add_instrument(pool, "CURRENCY_PAIR", Box::new(currency_pair))
+                        .await
+                        .unwrap()
+                }
+                InstrumentAny::Equity(equity) => {
+                    DatabaseQueries::add_instrument(pool, "EQUITY", Box::new(equity))
+                        .await
+                        .unwrap()
+                }
+                InstrumentAny::FuturesContract(futures_contract) => {
+                    DatabaseQueries::add_instrument(
+                        pool,
+                        "FUTURES_CONTRACT",
+                        Box::new(futures_contract),
+                    )
+                    .await
+                    .unwrap()
+                }
+                InstrumentAny::FuturesSpread(futures_spread) => DatabaseQueries::add_instrument(
+                    pool,
+                    "FUTURES_SPREAD",
+                    Box::new(futures_spread),
+                )
+                .await
+                .unwrap(),
+                InstrumentAny::OptionsContract(options_contract) => {
+                    DatabaseQueries::add_instrument(
+                        pool,
+                        "OPTIONS_CONTRACT",
+                        Box::new(options_contract),
+                    )
+                    .await
+                    .unwrap()
+                }
+                InstrumentAny::OptionsSpread(options_spread) => DatabaseQueries::add_instrument(
+                    pool,
+                    "OPTIONS_SPREAD",
+                    Box::new(options_spread),
+                )
+                .await
+                .unwrap(),
+            },
         }
     }
 }
@@ -141,4 +203,32 @@ impl PostgresCacheDatabase {
             anyhow::anyhow!("Failed to query add_currency to database message handler: {err}")
         })
     }
+
+    pub async fn load_currencies(&self) -> anyhow::Result<Vec<Currency>> {
+        DatabaseQueries::load_currencies(&self.pool).await
+    }
+
+    pub async fn load_currency(&self, code: &str) -> anyhow::Result<Option<Currency>> {
+        DatabaseQueries::load_currency(&self.pool, code).await
+    }
+
+    pub async fn add_instrument(&self, instrument: InstrumentAny) -> anyhow::Result<()> {
+        let query = DatabaseQuery::AddInstrument(instrument);
+        self.tx.send(query).await.map_err(|err| {
+            anyhow::anyhow!(
+                "Failed to send query add_instrument to database message handler: {err}"
+            )
+        })
+    }
+
+    pub async fn load_instrument(
+        &self,
+        instrument_id: InstrumentId,
+    ) -> anyhow::Result<Option<InstrumentAny>> {
+        DatabaseQueries::load_instrument(&self.pool, instrument_id).await
+    }
+
+    pub async fn load_instruments(&self) -> anyhow::Result<Vec<InstrumentAny>> {
+        DatabaseQueries::load_instruments(&self.pool).await
+    }
 }

nautilus_core/infrastructure/src/sql/mod.rs

@@ -14,7 +14,7 @@
 // -------------------------------------------------------------------------------------------------
 
 // Be careful about ordering and foreign key constraints when deleting data.
-pub const NAUTILUS_TABLES: [&str; 2] = ["general", "currency"];
+pub const NAUTILUS_TABLES: [&str; 3] = ["general", "instrument", "currency"];
 
 pub mod cache_database;
 pub mod models;

nautilus_core/infrastructure/src/sql/models/instruments.rs

@@ -35,55 +35,57 @@ use rust_decimal::Decimal;
 use sqlx::{postgres::PgRow, FromRow, Row};
 use ustr::Ustr;
 
-struct InstrumentAnyModel(InstrumentAny);
-struct CryptoFutureModel(CryptoFuture);
-struct CryptoPerpetualModel(CryptoPerpetual);
-struct CurrencyPairModel(CurrencyPair);
-struct EquityModel(Equity);
-struct FuturesContractModel(FuturesContract);
-struct FuturesSpreadModel(FuturesSpread);
-struct OptionsContractModel(OptionsContract);
-struct OptionsSpreadModel(OptionsSpread);
+pub struct InstrumentAnyModel(pub InstrumentAny);
+pub struct CryptoFutureModel(pub CryptoFuture);
+pub struct CryptoPerpetualModel(pub CryptoPerpetual);
+pub struct CurrencyPairModel(pub CurrencyPair);
+pub struct EquityModel(pub Equity);
+pub struct FuturesContractModel(pub FuturesContract);
+pub struct FuturesSpreadModel(pub FuturesSpread);
+pub struct OptionsContractModel(pub OptionsContract);
+pub struct OptionsSpreadModel(pub OptionsSpread);
 
 // TBD
-// impl<'r> FromRow<'r, PgRow> for InstrumentAnyModel {
-//     fn from_row(row: &'r PgRow) -> Result<Self, Error> {
-//         let kind = row.get::<String, _>("kind");
-//         if kind == "CRYPTO_FUTURE" {
-//             Ok(InstrumentAny::CryptoFuture(
-//                 CryptoFutureModel::from_row(row).unwrap().0,
-//             ))
-//         } else if kind == "CRYPTO_PERPETUAL" {
-//             Ok(InstrumentAny::CryptoPerpetual(
-//                 CryptoPerpetual::from_row(row).unwrap(),
-//             ))
-//         } else if kind == "CURRENCY_PAIR" {
-//             Ok(InstrumentAny::CurrencyPair(
-//                 CurrencyPair::from_row(row).unwrap(),
-//             ))
-//         } else if kind == "EQUITY" {
-//             Ok(InstrumentAny::Equity(Equity::from_row(row).unwrap()))
-//         } else if kind == "FUTURES_CONTRACT" {
-//             Ok(InstrumentAny::FuturesContract(
-//                 FuturesContract::from_row(row).unwrap(),
-//             ))
-//         } else if kind == "FUTURES_SPREAD" {
-//             Ok(InstrumentAny::FuturesSpread(
-//                 FuturesSpread::from_row(row).unwrap(),
-//             ))
-//         } else if kind == "OPTIONS_CONTRACT" {
-//             Ok(InstrumentAny::OptionsContract(
-//                 OptionsContract::from_row(row).unwrap(),
-//             ))
-//         } else if kind == "OPTIONS_SPREAD" {
-//             Ok(InstrumentAny::OptionsSpread(
-//                 OptionsSpread::from_row(row).unwrap(),
-//             ))
-//         } else {
-//             panic!("Unknown instrument type")
-//         }
-//     }
-// }
+impl<'r> FromRow<'r, PgRow> for InstrumentAnyModel {
+    fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {
+        let kind = row.get::<String, _>("kind");
+        if kind == "CRYPTO_FUTURE" {
+            Ok(InstrumentAnyModel(InstrumentAny::CryptoFuture(
+                CryptoFutureModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "CRYPTO_PERPETUAL" {
+            Ok(InstrumentAnyModel(InstrumentAny::CryptoPerpetual(
+                CryptoPerpetualModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "CURRENCY_PAIR" {
+            Ok(InstrumentAnyModel(InstrumentAny::CurrencyPair(
+                CurrencyPairModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "EQUITY" {
+            Ok(InstrumentAnyModel(InstrumentAny::Equity(
+                EquityModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "FUTURES_CONTRACT" {
+            Ok(InstrumentAnyModel(InstrumentAny::FuturesContract(
+                FuturesContractModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "FUTURES_SPREAD" {
+            Ok(InstrumentAnyModel(InstrumentAny::FuturesSpread(
+                FuturesSpreadModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "OPTIONS_CONTRACT" {
+            Ok(InstrumentAnyModel(InstrumentAny::OptionsContract(
+                OptionsContractModel::from_row(row).unwrap().0,
+            )))
+        } else if kind == "OPTIONS_SPREAD" {
+            Ok(InstrumentAnyModel(InstrumentAny::OptionsSpread(
+                OptionsSpreadModel::from_row(row).unwrap().0,
+            )))
+        } else {
+            panic!("Unknown instrument type")
+        }
+    }
+}
 
 impl<'r> FromRow<'r, PgRow> for CryptoFutureModel {
     fn from_row(row: &'r PgRow) -> Result<Self, sqlx::Error> {

nautilus_core/infrastructure/src/sql/pg.rs

@@ -105,7 +105,7 @@ pub async fn delete_nautilus_postgres_tables(db: &PgPool) -> anyhow::Result<()>
         query(format!("DELETE FROM \"{}\" WHERE true", table).as_str())
             .execute(db)
             .await
-            .unwrap_or_else(|_| panic!("Failed to delete table {}", table));
+            .unwrap_or_else(|err| panic!("Failed to delete table {} because: {}", table, err));
     }
     Ok(())
 }