feat: pull market metadata on startup

candle-creator/Cargo.lock

@@ -3136,6 +3136,7 @@ dependencies = [
  "solana-rpc",
  "solana-sdk",
  "solana-transaction-status",
+ "spl-token",
  "sqlx",
  "tokio",
  "tokio-stream",

candle-creator/Cargo.toml

@@ -19,6 +19,7 @@ solana-account-decoder = "=1.14.13"
 solana-transaction-status = "=1.14.13"
 solana-sdk = "=1.14.13"
 solana-rpc = "=1.14.13"
+spl-token = "3.5.0"
 anchor-client = "=0.26.0"
 borsh = "0.9"
 

candle-creator/src/database/database.rs

@@ -12,7 +12,7 @@ use tokio::sync::mpsc::{error::TryRecvError, Receiver};
 
 use crate::{
     trade_fetching::parsing::OpenBookFillEventLog,
-    utils::{AnyhowWrap, Config},
+    utils::{AnyhowWrap, Config, MarketInfo},
 };
 
 pub async fn connect_to_database(config: &Config) -> anyhow::Result<Pool<Postgres>> {
@@ -30,10 +30,11 @@ pub async fn connect_to_database(config: &Config) -> anyhow::Result<Pool<Postgre
     }
 }
 
-pub async fn setup_database(pool: &Pool<Postgres>) -> anyhow::Result<()> {
+pub async fn setup_database(pool: &Pool<Postgres>, markets: Vec<MarketInfo>) -> anyhow::Result<()> {
     let candles_table_fut = create_candles_table(pool);
     let fills_table_fut = create_fills_table(pool);
-    let result = tokio::try_join!(candles_table_fut, fills_table_fut);
+    let markets_table_fut = create_markets_table(pool, markets);
+    let result = tokio::try_join!(candles_table_fut, fills_table_fut, markets_table_fut);
     match result {
         Ok(_) => {
             println!("Successfully configured database");
@@ -109,6 +110,58 @@ pub async fn create_fills_table(pool: &Pool<Postgres>) -> anyhow::Result<()> {
     tx.commit().await.map_err_anyhow()
 }
 
+pub async fn create_markets_table(
+    pool: &Pool<Postgres>,
+    markets: Vec<MarketInfo>,
+) -> anyhow::Result<()> {
+    let mut tx = pool.begin().await.map_err_anyhow()?;
+
+    sqlx::query(
+        "CREATE TABLE IF NOT EXISTS markets (
+            market_name text PRIMARY KEY,
+            address text,
+            base_decimals numeric,
+            quote_decimals numeric,
+            base_lot_size numeric,
+            quote_lot_size numeric
+        )",
+    )
+    .execute(&mut tx)
+    .await?;
+
+    let insert_statement = build_markets_insert_statement(markets);
+    sqlx::query(&insert_statement).execute(&mut tx).await?;
+
+    tx.commit().await.map_err_anyhow()
+}
+
+fn build_markets_insert_statement(markets: Vec<MarketInfo>) -> String {
+    let mut stmt = String::from("INSERT INTO markets (market_name, address, base_decimals, quote_decimals, base_lot_size, quote_lot_size) VALUES");
+    for (idx, market) in markets.iter().enumerate() {
+        let val_str = format!(
+            "(\'{}\', \'{}\', {}, {}, {}, {})",
+            market.name,
+            market.address,
+            market.base_decimals,
+            market.quote_decimals,
+            market.base_lot_size,
+            market.quote_lot_size,
+        );
+
+        if idx == 0 {
+            stmt = format!("{} {}", &stmt, val_str);
+        } else {
+            stmt = format!("{}, {}", &stmt, val_str);
+        }
+    }
+
+    let handle_conflict = "ON CONFLICT (market_name) DO UPDATE SET address=excluded.address";
+
+    stmt = format!("{} {}", stmt, handle_conflict);
+    print!("{}", stmt);
+    stmt
+}
+
 pub async fn save_candles() {
     unimplemented!("TODO");
 }
@@ -184,8 +237,6 @@ fn build_fills_upsert_statement(events: Vec<OpenBookFillEventLog>) -> String {
     stmt
 }
 
-// pub async fn create_markets_table() {}
-
 #[cfg(test)]
 mod tests {
     use super::*;

candle-creator/src/main.rs

@@ -1,4 +1,4 @@
-use crate::{trade_fetching::parsing::OpenBookFillEventLog, utils::Config};
+use crate::{trade_fetching::{parsing::OpenBookFillEventLog, scrape::fetch_market_infos}, utils::Config};
 use database::database::{connect_to_database, setup_database};
 use dotenv;
 use tokio::sync::mpsc;
@@ -14,17 +14,18 @@ async fn main() -> anyhow::Result<()> {
     let rpc_url: String = dotenv::var("RPC_URL").unwrap();
     let database_url: String = dotenv::var("DATABASE_URL").unwrap();
 
+
     let config = Config {
         rpc_url,
         database_url,
         max_pg_pool_connections: 5,
-        markets: utils::load_markets("/Users/dboures/dev/openbook-candles/markets.json"),
     };
 
-    println!("{:?}", config);
+    let markets = utils::load_markets("/Users/dboures/dev/openbook-candles/markets.json");
+    let market_infos = fetch_market_infos(&config, markets).await?;
 
     let pool = connect_to_database(&config).await?;
-    setup_database(&pool).await?;
+    setup_database(&pool, market_infos).await?;
 
     let (fill_sender, fill_receiver) = mpsc::channel::<OpenBookFillEventLog>(1000);
 

candle-creator/src/trade_fetching/parsing.rs

@@ -27,6 +27,61 @@ pub struct OpenBookFillEventLog {
     pub referrer_rebate: Option<u64>,
 }
 
+#[derive(Copy, Clone, AnchorDeserialize)]
+#[cfg_attr(target_endian = "little", derive(Debug))]
+#[repr(packed)]
+pub struct MarketState {
+    // 0
+    pub account_flags: u64, // Initialized, Market
+
+    // 1
+    pub own_address: [u64; 4],
+
+    // 5
+    pub vault_signer_nonce: u64,
+    // 6
+    pub coin_mint: [u64; 4],
+    // 10
+    pub pc_mint: [u64; 4],
+
+    // 14
+    pub coin_vault: [u64; 4],
+    // 18
+    pub coin_deposits_total: u64,
+    // 19
+    pub coin_fees_accrued: u64,
+
+    // 20
+    pub pc_vault: [u64; 4],
+    // 24
+    pub pc_deposits_total: u64,
+    // 25
+    pub pc_fees_accrued: u64,
+
+    // 26
+    pub pc_dust_threshold: u64,
+
+    // 27
+    pub req_q: [u64; 4],
+    // 31
+    pub event_q: [u64; 4],
+
+    // 35
+    pub bids: [u64; 4],
+    // 39
+    pub asks: [u64; 4],
+
+    // 43
+    pub coin_lot_size: u64,
+    // 44
+    pub pc_lot_size: u64,
+
+    // 45
+    pub fee_rate_bps: u64,
+    // 46
+    pub referrer_rebates_accrued: u64,
+}
+
 pub fn parse_trades_from_openbook_txns(
     txns: &mut Vec<ClientResult<EncodedConfirmedTransactionWithStatusMeta>>,
 ) -> Vec<OpenBookFillEventLog> {

candle-creator/src/trade_fetching/scrape.rs

@@ -1,25 +1,24 @@
+use anchor_lang::AnchorDeserialize;
+use solana_account_decoder::UiAccountEncoding;
 use solana_client::{
     client_error::Result as ClientResult,
     rpc_client::{GetConfirmedSignaturesForAddress2Config, RpcClient},
-    rpc_config::RpcTransactionConfig,
+    rpc_config::{RpcTransactionConfig, RpcAccountInfoConfig},
 };
-use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Signature};
+use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Signature, program_pack::Pack};
 use solana_transaction_status::{EncodedConfirmedTransactionWithStatusMeta, UiTransactionEncoding};
+use spl_token::state::Mint;
 use std::{str::FromStr, time::Duration};
 use tokio::sync::mpsc::Sender;
 
-use crate::utils::Config;
+use crate::{utils::{Config, MarketInfo, MarketConfig}, trade_fetching::parsing::{MarketState}};
 
 use super::parsing::{parse_trades_from_openbook_txns, OpenBookFillEventLog};
 
-// use serde::{Deserialize, Serialize};
-
 pub async fn scrape(config: &Config, fill_sender: Sender<OpenBookFillEventLog>) {
     let url = &config.rpc_url;
     let rpc_client = RpcClient::new_with_commitment(url, CommitmentConfig::processed());
 
-    fetch_market(&rpc_client).await;
-
     let before_slot = None;
     loop {
         scrape_transactions(&rpc_client, before_slot, &fill_sender).await;
@@ -106,14 +105,66 @@ pub async fn scrape_transactions(
     Signature::from_str(&last_sig.signature).unwrap()
 }
 
-async fn fetch_market(rpc_client: &RpcClient) {
-    let data = rpc_client
-        .get_account_data(
-            &Pubkey::from_str("8BnEgHoWFysVcuFFX7QztDmzuH8r5ZFvyP3sYwn1XTh6").unwrap(),
-        )
-        .unwrap();
+pub async fn fetch_market_infos(config: &Config, markets: Vec<MarketConfig>) -> anyhow::Result<Vec<MarketInfo>> {
+    let url = &config.rpc_url;
+    let rpc_client = RpcClient::new_with_commitment(url, CommitmentConfig::processed());
 
-    println!("{}", data.len());
+    let rpc_config = RpcAccountInfoConfig {
+        encoding: Some(UiAccountEncoding::Base64),
+        data_slice: None,
+        commitment: Some(CommitmentConfig::confirmed()),
+        min_context_slot: None,
+    };
 
-    // simply the market object in TS
+    let market_keys = markets.iter().map(|x| Pubkey::from_str(&x.address).unwrap()).collect::<Vec<Pubkey>>();
+    let mut market_results = rpc_client.get_multiple_accounts_with_config(&market_keys, rpc_config.clone()).unwrap().value;
+
+    let mut mint_keys = Vec::new();
+
+    let mut market_infos = market_results.iter_mut().map(|mut r| {
+        let get_account_result = r.as_mut().unwrap();
+
+        let mut market_bytes: &[u8] = &mut get_account_result.data[5..];
+        let raw_market: MarketState = AnchorDeserialize::deserialize(&mut market_bytes).unwrap();
+
+        let base_mint = serum_bytes_to_pubkey(raw_market.coin_mint);
+        let quote_mint = serum_bytes_to_pubkey(raw_market.pc_mint);
+        mint_keys.push(base_mint);
+        mint_keys.push(quote_mint);
+
+        MarketInfo {
+            name: "".to_string(),
+            address: serum_bytes_to_pubkey(raw_market.own_address).to_string(),
+            base_decimals: 0,
+            quote_decimals: 0,
+            base_lot_size: raw_market.coin_lot_size,
+            quote_lot_size: raw_market.pc_lot_size,
+        }
+    }).collect::<Vec<MarketInfo>>();
+
+    let mint_results = rpc_client.get_multiple_accounts_with_config(&mint_keys, rpc_config).unwrap().value;
+    for i in (0..mint_results.len()).step_by(2) {
+        let mut base_mint_account = mint_results[i].as_ref().unwrap().clone();
+        let mut quote_mint_account = mint_results[i+1].as_ref().unwrap().clone();
+
+        let mut base_mint_bytes: &[u8] = &mut base_mint_account.data[..];
+        let mut quote_mint_bytes: &[u8] = &mut quote_mint_account.data[..];
+
+        let base_mint = Mint::unpack_from_slice(&mut base_mint_bytes).unwrap();
+        let quote_mint = Mint::unpack_from_slice(&mut quote_mint_bytes).unwrap();
+
+        market_infos[i / 2].name = markets[i / 2].name.clone();
+        market_infos[i / 2].base_decimals = base_mint.decimals;
+        market_infos[i / 2].quote_decimals = quote_mint.decimals;
+    }
+
+    Ok(market_infos)
+}
+
+fn serum_bytes_to_pubkey(data: [u64; 4]) -> Pubkey {
+    let mut res = [0; 32];
+    for i in 0..4 {
+        res[8*i..][..8].copy_from_slice(&data[i].to_le_bytes());
+    }
+    Pubkey::new_from_array(res)
 }

candle-creator/src/utils/mod.rs

@@ -18,13 +18,22 @@ pub struct Config {
     pub rpc_url: String,
     pub database_url: String,
     pub max_pg_pool_connections: u32,
-    pub markets: Vec<MarketConfig>,
 }
 
 #[derive(Clone, Debug, Deserialize)]
 pub struct MarketConfig {
     pub name: String,
-    pub market: String,
+    pub address: String,
+}
+
+#[derive(Debug)]
+pub struct MarketInfo {
+    pub name: String,
+    pub address: String,
+    pub base_decimals: u8,
+    pub quote_decimals: u8,
+    pub base_lot_size: u64,
+    pub quote_lot_size: u64,
 }
 
 pub fn load_markets(path: &str) -> Vec<MarketConfig> {

markets.json

@@ -1,6 +1,18 @@
 [
     {
-    "name" : "SOL/USDC",
-    "market" : "8BnEgHoWFysVcuFFX7QztDmzuH8r5ZFvyP3sYwn1XTh6"
+        "name" : "SOL/USDC",
+        "address" : "8BnEgHoWFysVcuFFX7QztDmzuH8r5ZFvyP3sYwn1XTh6"
+    },
+    {
+        "name" : "RLB/USDC",
+        "address" : "72h8rWaWwfPUL36PAFqyQZU8RT1V3FKG7Nc45aK89xTs"
+    },
+    {
+        "name" : "MNGO/USDC",
+        "address" : "3NnxQvDcZXputNMxaxsGvqiKpqgPfSYXpNigZNFcknmD"
+    },
+    {
+        "name" : "BONK/SOL",
+        "address" : "Hs97TCZeuYiJxooo3U73qEHXg3dKpRL4uYKYRryEK9CF"
     }
 ]