nautilus_data/engine/
pool.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2025 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//! Message handler that maintains the `Pool` state stored in the global [`Cache`].
17//!
18//! The handler is functionally equivalent to `BookUpdater` but for DeFi liquidity
19//! pools. Whenever a [`PoolSwap`] or [`PoolLiquidityUpdate`] is published on the
20//! message bus the handler looks up the corresponding `Pool` instance in the
21//! cache and applies the change in-place (for now we only update the `ts_init`
22//! timestamp so that consumers can tell the pool has been touched).
23
24use std::{any::Any, cell::RefCell, rc::Rc};
25
26use nautilus_common::{cache::Cache, msgbus::handler::MessageHandler};
27use nautilus_model::{
28    defi::{PoolLiquidityUpdate, PoolSwap},
29    identifiers::InstrumentId,
30};
31use ustr::Ustr;
32
33/// Handles [`PoolSwap`]s and [`PoolLiquidityUpdate`]s for a single AMM pool.
34#[derive(Debug)]
35pub struct PoolUpdater {
36    id: Ustr,
37    instrument_id: InstrumentId,
38    cache: Rc<RefCell<Cache>>,
39}
40
41impl PoolUpdater {
42    /// Creates a new [`PoolUpdater`] bound to the given `instrument_id` and `cache`.
43    #[must_use]
44    pub fn new(instrument_id: &InstrumentId, cache: Rc<RefCell<Cache>>) -> Self {
45        Self {
46            id: Ustr::from(&format!("{}-{}", stringify!(PoolUpdater), instrument_id)),
47            instrument_id: *instrument_id,
48            cache,
49        }
50    }
51
52    fn handle_pool_swap(&self, _swap: &PoolSwap) {
53        if let Some(_pool) = self.cache.borrow_mut().pool_mut(&self.instrument_id) {
54            // TODO: Implement handling pool swap
55        }
56    }
57
58    fn handle_pool_liquidity_update(&self, _update: &PoolLiquidityUpdate) {
59        if let Some(_pool) = self.cache.borrow_mut().pool_mut(&self.instrument_id) {
60            // TODO: implement handling pool liquidity update
61        }
62    }
63}
64
65impl MessageHandler for PoolUpdater {
66    fn id(&self) -> Ustr {
67        self.id
68    }
69
70    fn handle(&self, message: &dyn Any) {
71        if let Some(swap) = message.downcast_ref::<PoolSwap>() {
72            self.handle_pool_swap(swap);
73            return;
74        }
75
76        if let Some(update) = message.downcast_ref::<PoolLiquidityUpdate>() {
77            self.handle_pool_liquidity_update(update);
78        }
79    }
80
81    fn as_any(&self) -> &dyn Any {
82        self
83    }
84}
nautilus_data/engine/pool.rs

nautilus_data/engine/
pool.rs
