Add support for DEXs, pools, and tokens in blockchain adapter (#2638)

Author: filipmacek

Cargo.lock

@@ -114,6 +114,11 @@ tokio = { version = "1.45.0", features = ["full"] }
 tokio-tungstenite = { version = "0.26.2", features = ["rustls-tls-native-roots"] }
 ustr = { version = "1.1.0", features = ["serde"] }
 uuid = { version = "1.16.0", features = ["v4", "serde"] }
+sqlx = { version = "0.8.5", features = [
+    "postgres",
+    "runtime-tokio",
+    "json"
+]}
 
 # dev-dependencies
 axum = "0.8.4"

Cargo.toml

@@ -31,6 +31,7 @@ nautilus-model = { workspace = true, features = ["defi"] }
 nautilus-core = { workspace = true }
 nautilus-network = { workspace = true, features = ["python"] }  # TODO: Untangle python feature
 nautilus-common = { workspace = true }
+nautilus-infrastructure = { workspace = true, features = ["postgres"] }
 
 anyhow = { workspace = true }
 async-trait = { workspace = true }
@@ -49,6 +50,10 @@ tracing = { workspace = true }
 ustr = { workspace = true }
 hypersync-client = { version = "0.18.2", optional = true }
 hypersync-schema = { version = "0.3.0", optional = true }
+alloy = { version = "1.0.3"}
+hex = { version = "0.4.3" }
+sqlx = {workspace = true}
+bytes = { workspace = true }
 
 [dev-dependencies]
 rstest = { workspace = true }
@@ -62,3 +67,8 @@ required-features = ["hypersync"]
 name = "live_blocks_hypersync"
 path = "bin/watch_hypersync_live_blocks.rs"
 required-features = ["hypersync"]
+
+[[bin]]
+name = "sync_tokens_pools"
+path = "bin/sync_tokens_pools.rs"
+required-features = ["hypersync"]

crates/adapters/blockchain/Cargo.toml

@@ -11,6 +11,7 @@ You can configure the required environment variables in two ways:
    ```
    CHAIN=Ethereum
    RPC_WSS_URL=wss://mainnet.infura.io/ws/v3/YOUR_INFURA_API_KEY
+   RPC_HTTP_URL=https://mainnet.infura.io/v3/YOUR_INFURA_API_KEY
    ```
 
 2. **Providing variables directly in the command line:**
@@ -24,11 +25,11 @@ You can configure the required environment variables in two ways:
 The scripts will connect to the specified blockchain and log information about each new block received for both the RPC version and only Hypersync.
 
 ```
-cargo run --bin live_blocks_rpc
+cargo run --bin live_blocks_rpc --features hypersync
 ```
 
 ```
-cargo run --bin live_blocks_hypersync
+cargo run --bin live_blocks_hypersync --features hypersync
 ```
 
 For RPC example, the output should be:
@@ -42,3 +43,10 @@ Running `target/debug/live_blocks_rpc`
 ^C2025-04-25T14:55:38.314022000Z [INFO] TRADER-001.live_blocks: Shutdown signal received, shutting down...
 
 ```
+
+### Sync dex, tokens and pool for Uniswap V3 on Ethereum
+This script demonstrates how to use the blockchain data client to discover and cache Uniswap V3 pools and their associated tokens. It queries the Ethereum blockchain for pool creation events emitted by the Uniswap V3 factory contract, retrieves token metadata (name, symbol, decimals) for each token in the pools via smart contract calls, and stores everything in a local Postgres database.
+
+```
+cargo run --bin sync_tokens_pools --features hypersync
+```

crates/adapters/blockchain/README.md

@@ -0,0 +1,109 @@
+// -------------------------------------------------------------------------------------------------
+//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
+//  https://nautechsystems.io
+//
+//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+//  You may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+// -------------------------------------------------------------------------------------------------
+
+use std::sync::Arc;
+
+use nautilus_blockchain::{config::BlockchainAdapterConfig, data::BlockchainDataClient, exchanges};
+use nautilus_common::logging::{
+    logger::{Logger, LoggerConfig},
+    writer::FileWriterConfig,
+};
+use nautilus_core::UUID4;
+use nautilus_model::{
+    defi::chain::{Blockchain, Chain, chains},
+    identifiers::TraderId,
+};
+use tokio::sync::Notify;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    dotenvy::dotenv().ok();
+    // Setup logger
+    let _logger_guard = Logger::init_with_config(
+        TraderId::default(),
+        UUID4::new(),
+        LoggerConfig::default(),
+        FileWriterConfig::new(None, None, None, None),
+    )?;
+
+    // Setup graceful shutdown with signal handling in different task
+    let notify = Arc::new(Notify::new());
+    let notifier = notify.clone();
+    tokio::spawn(async move {
+        let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
+            .expect("Failed to create SIGTERM listener");
+        let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())
+            .expect("Failed to create SIGINT listener");
+        tokio::select! {
+            _ = sigterm.recv() => {}
+            _ = sigint.recv() => {}
+        }
+        log::info!("Shutdown signal received, shutting down...");
+        notifier.notify_one();
+    });
+
+    // Initialize the blockchain data client, connect and subscribe to live blocks with RPC
+    let chain: Chain = match std::env::var("CHAIN") {
+        Ok(chain_str) => {
+            if let Ok(blockchain) = chain_str.parse::<Blockchain>() {
+                match blockchain {
+                    Blockchain::Ethereum => chains::ETHEREUM.clone(),
+                    Blockchain::Base => chains::BASE.clone(),
+                    Blockchain::Arbitrum => chains::ARBITRUM.clone(),
+                    Blockchain::Polygon => chains::POLYGON.clone(),
+                    _ => panic!("Invalid chain {chain_str}"),
+                }
+            } else {
+                panic!("Invalid chain {chain_str}");
+            }
+        }
+        Err(_) => chains::ETHEREUM.clone(), // default
+    };
+    let chain = Arc::new(chain);
+    let http_rpc_url = std::env::var("RPC_HTTP_URL").expect("RPC_HTTP_URL must be set");
+    let blockchain_adapter_config = BlockchainAdapterConfig::new(http_rpc_url, Some(3), None, true);
+    let mut data_client = BlockchainDataClient::new(chain, blockchain_adapter_config);
+    data_client.initialize_cache_database(None).await;
+
+    let univ3 = exchanges::ethereum::UNISWAP_V3.clone();
+    let dex_id = univ3.id();
+    data_client.connect().await?;
+    data_client.register_exchange(univ3.clone()).await?;
+    // Lets use block https://etherscan.io/block/22327045 from (Apr-22-2025 08:49:47 PM +UTC)
+    let from_block = Some(22327045);
+
+    // Main loop to keep the app running
+    loop {
+        tokio::select! {
+            () = notify.notified() => break,
+             result = data_client.sync_exchange_pools(dex_id.as_str(), from_block) => {
+                match result {
+                    Ok(_) => {
+                        // Exit after the tokens and pool are synced successfully
+                        log::info!("Successfully synced tokens and pools");
+                        break;
+                    },
+                    Err(e) => {
+                        // Handle error case
+                        log::error!("Error syncing tokens and pools: {}", e);
+                        break;
+                    }
+                }
+            }
+        }
+    }
+    data_client.disconnect()?;
+    Ok(())
+}

crates/adapters/blockchain/bin/sync_tokens_pools.rs

@@ -71,7 +71,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
         Err(_) => chains::ETHEREUM.clone(), // default
     };
-    let blockchain_adapter_config = BlockchainAdapterConfig::new(None, true);
+    let chain = Arc::new(chain);
+    let http_rpc_url = std::env::var("RPC_HTTP_URL").expect("RPC_HTTP_URL must be set");
+    let blockchain_adapter_config = BlockchainAdapterConfig::new(http_rpc_url, None, None, true);
     let mut data_client = BlockchainDataClient::new(chain.clone(), blockchain_adapter_config);
     data_client.connect().await?;
     data_client.subscribe_blocks().await;

crates/adapters/blockchain/bin/watch_hypersync_live_blocks.rs

@@ -71,8 +71,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
         Err(_) => chains::ETHEREUM.clone(), // default
     };
+    let chain = Arc::new(chain);
     let wss_rpc_url = std::env::var("RPC_WSS_URL").expect("RPC_WSS_URL must be set");
-    let blockchain_adapter_config = BlockchainAdapterConfig::new(Some(wss_rpc_url), false);
+    let http_rpc_url = std::env::var("RPC_HTTP_URL").expect("RPC_HTTP_URL must be set");
+    let blockchain_adapter_config =
+        BlockchainAdapterConfig::new(http_rpc_url, None, Some(wss_rpc_url), false);
     let mut data_client = BlockchainDataClient::new(chain.clone(), blockchain_adapter_config);
     data_client.connect().await?;
     data_client.subscribe_blocks().await;

crates/adapters/blockchain/bin/watch_rpc_live_blocks.rs

@@ -0,0 +1,168 @@
+// -------------------------------------------------------------------------------------------------
+//  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
+//  https://nautechsystems.io
+//
+//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+//  You may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+// -------------------------------------------------------------------------------------------------
+
+use nautilus_model::defi::{
+    amm::Pool,
+    chain::{Chain, SharedChain},
+    dex::Dex,
+    token::Token,
+};
+use sqlx::{PgPool, postgres::PgConnectOptions};
+
+use crate::cache::rows::TokenRow;
+
+/// Database interface for persisting and retrieving blockchain entities and domain objects.
+pub struct BlockchainCacheDatabase {
+    /// PostgreSQL connection pool used for database operations
+    pool: PgPool,
+}
+
+impl BlockchainCacheDatabase {
+    /// Initializes a new database instance by establishing a connection to PostgreSQL.
+    pub async fn init(pg_options: PgConnectOptions) -> Self {
+        let pool = PgPool::connect_with(pg_options)
+            .await
+            .expect("Error connecting to Postgres");
+        Self { pool }
+    }
+
+    /// Seeds the database with a blockchain chain record.
+    pub async fn seed_chain(&self, chain: &Chain) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"
+            INSERT INTO chain (
+                chain_id, name
+            ) VALUES ($1,$2)
+            ON CONFLICT (chain_id)
+            DO NOTHING
+        "#,
+        )
+        .bind(chain.chain_id as i32)
+        .bind(chain.name.to_string())
+        .execute(&self.pool)
+        .await
+        .map(|_| ())
+        .map_err(|e| anyhow::anyhow!("Failed to seed chain table: {e}"))
+    }
+
+    /// Adds or updates a DEX (Decentralized Exchange) record in the database.
+    pub async fn add_dex(&self, dex: &Dex) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"
+            INSERT INTO dex (
+                chain_id, name, factory_address
+            ) VALUES ($1, $2, $3)
+            ON CONFLICT (chain_id, name)
+            DO UPDATE
+            SET
+                factory_address = $3
+        "#,
+        )
+        .bind(dex.chain.chain_id as i32)
+        .bind(dex.name.as_ref())
+        .bind(dex.factory.as_ref())
+        .execute(&self.pool)
+        .await
+        .map(|_| ())
+        .map_err(|e| anyhow::anyhow!("Failed to insert into dex table: {e}"))
+    }
+
+    /// Adds or updates a liquidity pool/pair record in the database.
+    pub async fn add_pool(&self, pool: &Pool) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"
+            INSERT INTO pool (
+                chain_id, address, dex_name, creation_block,
+                token0_chain, token0_address,
+                token1_chain, token1_address,
+                fee, tick_spacing
+            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
+            ON CONFLICT (chain_id, address)
+            DO UPDATE
+            SET
+                dex_name = $3,
+                creation_block = $4,
+                token0_chain = $5,
+                token0_address = $6,
+                token1_chain = $7,
+                token1_address = $8,
+                fee = $9,
+                tick_spacing = $10
+        "#,
+        )
+        .bind(pool.chain.chain_id as i32)
+        .bind(pool.address.as_str())
+        .bind(pool.dex.name.as_ref())
+        .bind(pool.creation_block as i64)
+        .bind(pool.token0.chain.chain_id as i32)
+        .bind(pool.token0.address.as_str())
+        .bind(pool.token1.chain.chain_id as i32)
+        .bind(pool.token1.address.as_str())
+        .bind(pool.fee as i32)
+        .bind(pool.tick_spacing as i32)
+        .execute(&self.pool)
+        .await
+        .map(|_| ())
+        .map_err(|e| anyhow::anyhow!("Failed to insert into pool table: {e}"))
+    }
+
+    /// Adds or updates a token record in the database.
+    pub async fn add_token(&self, token: &Token) -> anyhow::Result<()> {
+        sqlx::query(
+            r#"
+            INSERT INTO token (
+                chain_id, address, name, symbol, decimals
+            ) VALUES ($1, $2, $3, $4, $5)
+            ON CONFLICT (chain_id, address)
+            DO UPDATE
+            SET
+                name = $3,
+                symbol = $4,
+                decimals = $5
+        "#,
+        )
+        .bind(token.chain.chain_id as i32)
+        .bind(token.address.as_str())
+        .bind(token.name.as_str())
+        .bind(token.symbol.as_str())
+        .bind(token.decimals as i32)
+        .execute(&self.pool)
+        .await
+        .map(|_| ())
+        .map_err(|e| anyhow::anyhow!("Failed to insert into token table: {e}"))
+    }
+
+    /// Retrieves all token records for the given chain and converts them into `Token` domain objects.
+    pub async fn load_tokens(&self, chain: SharedChain) -> anyhow::Result<Vec<Token>> {
+        sqlx::query_as::<_, TokenRow>("SELECT * FROM token WHERE chain_id = $1")
+            .bind(chain.chain_id as i32)
+            .fetch_all(&self.pool)
+            .await
+            .map(|rows| {
+                rows.into_iter()
+                    .map(|token_row| {
+                        Token::new(
+                            chain.clone(),
+                            token_row.address,
+                            token_row.name,
+                            token_row.symbol,
+                            token_row.decimals as u8,
+                        )
+                    })
+                    .collect::<Vec<_>>()
+            })
+            .map_err(|e| anyhow::anyhow!("Failed to load tokens: {e}"))
+    }
+}