nautilus_blockchain/python/
cache.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
16//! Python bindings for loading blockchain cache state.
17
18use nautilus_common::live::get_runtime;
19use nautilus_core::python::{to_pyruntime_err, to_pyvalue_err};
20use nautilus_infrastructure::sql::pg::PostgresConnectOptions;
21use nautilus_model::defi::{PoolIdentifier, pool_analysis::PoolSnapshot};
22use pyo3::prelude::*;
23
24use crate::cache::database::BlockchainCacheDatabase;
25
26/// Loads the latest pool snapshot from the Postgres cache for backtest replay.
27///
28/// Connects to the cache database and reconstructs the most recent snapshot for the pool,
29/// including its full position and tick state, so a pool profiler can initialize before
30/// historical swaps and liquidity updates are replayed. When `before_block` is set, only
31/// snapshots at or before that block are considered, so the snapshot aligns with the replay
32/// start. When `require_valid` is `true`, only snapshots validated against on-chain state are
33/// returned. Returns `None` when no matching snapshot exists in the cache.
34///
35/// # Errors
36///
37/// Returns a `PyErr` if `pool_address` is not a valid pool identifier, or if the database
38/// connection or query fails.
39#[pyfunction]
40#[pyo3_stub_gen::derive::gen_stub_pyfunction(module = "nautilus_trader.adapters.blockchain")]
41#[pyo3(name = "load_pool_snapshot")]
42#[pyo3(signature = (pg_config, chain_id, pool_address, before_block=None, require_valid=true))]
43#[gen_stub(
44    override_return_type(
45        type_repr = "typing.Optional[nautilus_trader.model.PoolSnapshot]",
46        imports = ("typing", "nautilus_trader.model"),
47    ),
48)]
49pub fn py_load_pool_snapshot(
50    #[gen_stub(
51        override_type(
52            type_repr = "nautilus_trader.infrastructure.PostgresConnectOptions",
53            imports = ("nautilus_trader.infrastructure",),
54        ),
55    )]
56    pg_config: PostgresConnectOptions,
57    chain_id: u32,
58    pool_address: &str,
59    before_block: Option<u64>,
60    require_valid: bool,
61) -> PyResult<Option<PoolSnapshot>> {
62    let pool_identifier = PoolIdentifier::new_checked(pool_address).map_err(to_pyvalue_err)?;
63
64    get_runtime().block_on(async move {
65        let database = BlockchainCacheDatabase::connect(pg_config.into())
66            .await
67            .map_err(to_pyruntime_err)?;
68        database
69            .load_latest_pool_snapshot(chain_id, &pool_identifier, before_block, require_valid)
70            .await
71            .map_err(to_pyruntime_err)
72    })
73}
nautilus_blockchain/python/cache.rs

nautilus_blockchain/python/
cache.rs
