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::{PoolFeeCollect, PoolFlash, PoolLiquidityUpdate, PoolLiquidityUpdateType, 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_profiler) = self
54            .cache
55            .borrow_mut()
56            .pool_profiler_mut(&self.instrument_id)
57            && let Err(e) = pool_profiler.process_swap(swap)
58        {
59            log::error!("Failed to process pool swap: {e}");
60        }
61    }
62
63    fn handle_pool_liquidity_update(&self, update: &PoolLiquidityUpdate) {
64        if let Some(pool_profiler) = self
65            .cache
66            .borrow_mut()
67            .pool_profiler_mut(&self.instrument_id)
68            && let Err(e) = match update.kind {
69                PoolLiquidityUpdateType::Mint => pool_profiler.process_mint(update),
70                PoolLiquidityUpdateType::Burn => pool_profiler.process_burn(update),
71                _ => panic!("Liquidity update operation {} not implemented", update.kind),
72            }
73        {
74            log::error!("Failed to process pool liquidity update: {e}");
75        }
76    }
77
78    fn handle_pool_fee_collect(&self, event: &PoolFeeCollect) {
79        if let Some(pool_profiler) = self
80            .cache
81            .borrow_mut()
82            .pool_profiler_mut(&self.instrument_id)
83            && let Err(e) = pool_profiler.process_collect(event)
84        {
85            log::error!("Failed to process pool fee collect: {e}");
86        }
87    }
88
89    fn handle_pool_flash(&self, event: &PoolFlash) {
90        if let Some(pool_profiler) = self
91            .cache
92            .borrow_mut()
93            .pool_profiler_mut(&self.instrument_id)
94            && let Err(e) = pool_profiler.process_flash(event)
95        {
96            log::error!("Failed to process pool flash: {e}");
97        }
98    }
99}
100
101impl MessageHandler for PoolUpdater {
102    fn id(&self) -> Ustr {
103        self.id
104    }
105
106    fn handle(&self, message: &dyn Any) {
107        if let Some(swap) = message.downcast_ref::<PoolSwap>() {
108            self.handle_pool_swap(swap);
109            return;
110        }
111
112        if let Some(update) = message.downcast_ref::<PoolLiquidityUpdate>() {
113            self.handle_pool_liquidity_update(update);
114            return;
115        }
116
117        if let Some(update) = message.downcast_ref::<PoolFeeCollect>() {
118            self.handle_pool_fee_collect(update);
119            return;
120        }
121
122        if let Some(flash) = message.downcast_ref::<PoolFlash>() {
123            self.handle_pool_flash(flash);
124        }
125    }
126
127    fn as_any(&self) -> &dyn Any {
128        self
129    }
130}
nautilus_data/engine/pool.rs

nautilus_data/engine/
pool.rs
