nautilus_data/defi/
client.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//! DeFi-specific data client functionality.
17//!
18//! This module provides DeFi subscription and request helper methods
19//! for the `DataClientAdapter`. All code in this module requires the `defi` feature flag.
20
21use nautilus_common::{
22    clients::log_command_error,
23    messages::defi::{
24        DefiRequestCommand, DefiSubscribeCommand, DefiUnsubscribeCommand, RequestPoolSnapshot,
25        SubscribeBlocks, SubscribePool, SubscribePoolFeeCollects, SubscribePoolFlashEvents,
26        SubscribePoolLiquidityUpdates, SubscribePoolSwaps, UnsubscribeBlocks, UnsubscribePool,
27        UnsubscribePoolFeeCollects, UnsubscribePoolFlashEvents, UnsubscribePoolLiquidityUpdates,
28        UnsubscribePoolSwaps,
29    },
30};
31
32use crate::client::DataClientAdapter;
33
34impl DataClientAdapter {
35    #[inline]
36    pub fn execute_defi_subscribe(&mut self, cmd: DefiSubscribeCommand) {
37        let cmd_debug = format!("{cmd:?}");
38        if let Err(e) = match cmd {
39            DefiSubscribeCommand::Blocks(cmd) => self.subscribe_blocks(cmd),
40            DefiSubscribeCommand::Pool(cmd) => self.subscribe_pool(cmd),
41            DefiSubscribeCommand::PoolSwaps(cmd) => self.subscribe_pool_swaps(cmd),
42            DefiSubscribeCommand::PoolLiquidityUpdates(cmd) => {
43                self.subscribe_pool_liquidity_updates(cmd)
44            }
45            DefiSubscribeCommand::PoolFeeCollects(cmd) => self.subscribe_pool_fee_collects(cmd),
46            DefiSubscribeCommand::PoolFlashEvents(cmd) => self.subscribe_pool_flash_events(cmd),
47        } {
48            log_command_error(&cmd_debug, &e);
49        }
50    }
51
52    #[inline]
53    pub fn execute_defi_unsubscribe(&mut self, cmd: &DefiUnsubscribeCommand) {
54        if let Err(e) = match cmd {
55            DefiUnsubscribeCommand::Blocks(cmd) => self.unsubscribe_blocks(cmd),
56            DefiUnsubscribeCommand::Pool(cmd) => self.unsubscribe_pool(cmd),
57            DefiUnsubscribeCommand::PoolSwaps(cmd) => self.unsubscribe_pool_swaps(cmd),
58            DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd) => {
59                self.unsubscribe_pool_liquidity_updates(cmd)
60            }
61            DefiUnsubscribeCommand::PoolFeeCollects(cmd) => self.unsubscribe_pool_fee_collects(cmd),
62            DefiUnsubscribeCommand::PoolFlashEvents(cmd) => self.unsubscribe_pool_flash_events(cmd),
63        } {
64            log_command_error(&cmd, &e);
65        }
66    }
67
68    /// Executes a DeFi data request command by dispatching to the appropriate handler.
69    ///
70    /// # Errors
71    ///
72    /// Returns an error if the underlying client request fails.
73    #[inline]
74    pub fn execute_defi_request(&self, cmd: DefiRequestCommand) -> anyhow::Result<()> {
75        match cmd {
76            DefiRequestCommand::PoolSnapshot(cmd) => self.request_pool_snapshot(cmd),
77        }
78    }
79
80    /// Subscribes to block events for the specified blockchain.
81    ///
82    /// # Errors
83    ///
84    /// Returns an error if the underlying client subscribe operation fails.
85    fn subscribe_blocks(&mut self, cmd: SubscribeBlocks) -> anyhow::Result<()> {
86        if Self::track_subscribe(&mut self.subscriptions_blocks, cmd.chain, "blocks") {
87            self.client.subscribe_blocks(cmd)?;
88        }
89        Ok(())
90    }
91
92    /// Unsubscribes from block events for the specified blockchain.
93    ///
94    /// # Errors
95    ///
96    /// Returns an error if the underlying client unsubscribe operation fails.
97    fn unsubscribe_blocks(&mut self, cmd: &UnsubscribeBlocks) -> anyhow::Result<()> {
98        if Self::track_unsubscribe(&mut self.subscriptions_blocks, cmd.chain, "blocks") {
99            self.client.unsubscribe_blocks(cmd)?;
100        }
101        Ok(())
102    }
103
104    /// Subscribes to pool definition updates for the specified AMM pool.
105    ///
106    /// # Errors
107    ///
108    /// Returns an error if the underlying client subscribe operation fails.
109    fn subscribe_pool(&mut self, cmd: SubscribePool) -> anyhow::Result<()> {
110        if Self::track_subscribe(&mut self.subscriptions_pools, cmd.instrument_id, "pool") {
111            self.client.subscribe_pool(cmd)?;
112        }
113        Ok(())
114    }
115
116    /// Subscribes to pool swap events for the specified AMM pool.
117    ///
118    /// # Errors
119    ///
120    /// Returns an error if the underlying client subscribe operation fails.
121    fn subscribe_pool_swaps(&mut self, cmd: SubscribePoolSwaps) -> anyhow::Result<()> {
122        if Self::track_subscribe(
123            &mut self.subscriptions_pool_swaps,
124            cmd.instrument_id,
125            "pool swaps",
126        ) {
127            self.client.subscribe_pool_swaps(cmd)?;
128        }
129        Ok(())
130    }
131
132    /// Subscribes to pool liquidity update events for the specified AMM pool.
133    ///
134    /// # Errors
135    ///
136    /// Returns an error if the underlying client subscribe operation fails.
137    fn subscribe_pool_liquidity_updates(
138        &mut self,
139        cmd: SubscribePoolLiquidityUpdates,
140    ) -> anyhow::Result<()> {
141        if Self::track_subscribe(
142            &mut self.subscriptions_pool_liquidity_updates,
143            cmd.instrument_id,
144            "pool liquidity updates",
145        ) {
146            self.client.subscribe_pool_liquidity_updates(cmd)?;
147        }
148        Ok(())
149    }
150
151    /// Subscribes to pool fee collect events for the specified AMM pool.
152    ///
153    /// # Errors
154    ///
155    /// Returns an error if the underlying client subscribe operation fails.
156    fn subscribe_pool_fee_collects(&mut self, cmd: SubscribePoolFeeCollects) -> anyhow::Result<()> {
157        if Self::track_subscribe(
158            &mut self.subscriptions_pool_fee_collects,
159            cmd.instrument_id,
160            "pool fee collects",
161        ) {
162            self.client.subscribe_pool_fee_collects(cmd)?;
163        }
164        Ok(())
165    }
166
167    /// Subscribes to pool flash loan events for the specified AMM pool.
168    ///
169    /// # Errors
170    ///
171    /// Returns an error if the underlying client subscribe operation fails.
172    fn subscribe_pool_flash_events(&mut self, cmd: SubscribePoolFlashEvents) -> anyhow::Result<()> {
173        if Self::track_subscribe(
174            &mut self.subscriptions_pool_flash,
175            cmd.instrument_id,
176            "pool flash events",
177        ) {
178            self.client.subscribe_pool_flash_events(cmd)?;
179        }
180        Ok(())
181    }
182
183    /// Unsubscribes from pool definition updates for the specified AMM pool.
184    ///
185    /// # Errors
186    ///
187    /// Returns an error if the underlying client unsubscribe operation fails.
188    fn unsubscribe_pool(&mut self, cmd: &UnsubscribePool) -> anyhow::Result<()> {
189        if Self::track_unsubscribe(&mut self.subscriptions_pools, cmd.instrument_id, "pool") {
190            self.client.unsubscribe_pool(cmd)?;
191        }
192        Ok(())
193    }
194
195    /// Unsubscribes from swap events for the specified AMM pool.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if the underlying client unsubscribe operation fails.
200    fn unsubscribe_pool_swaps(&mut self, cmd: &UnsubscribePoolSwaps) -> anyhow::Result<()> {
201        if Self::track_unsubscribe(
202            &mut self.subscriptions_pool_swaps,
203            cmd.instrument_id,
204            "pool swaps",
205        ) {
206            self.client.unsubscribe_pool_swaps(cmd)?;
207        }
208        Ok(())
209    }
210
211    /// Unsubscribes from pool liquidity update events for the specified AMM pool.
212    ///
213    /// # Errors
214    ///
215    /// Returns an error if the underlying client unsubscribe operation fails.
216    fn unsubscribe_pool_liquidity_updates(
217        &mut self,
218        cmd: &UnsubscribePoolLiquidityUpdates,
219    ) -> anyhow::Result<()> {
220        if Self::track_unsubscribe(
221            &mut self.subscriptions_pool_liquidity_updates,
222            cmd.instrument_id,
223            "pool liquidity updates",
224        ) {
225            self.client.unsubscribe_pool_liquidity_updates(cmd)?;
226        }
227        Ok(())
228    }
229
230    /// Unsubscribes from pool fee collect events for the specified AMM pool.
231    ///
232    /// # Errors
233    ///
234    /// Returns an error if the underlying client unsubscribe operation fails.
235    fn unsubscribe_pool_fee_collects(
236        &mut self,
237        cmd: &UnsubscribePoolFeeCollects,
238    ) -> anyhow::Result<()> {
239        if Self::track_unsubscribe(
240            &mut self.subscriptions_pool_fee_collects,
241            cmd.instrument_id,
242            "pool fee collects",
243        ) {
244            self.client.unsubscribe_pool_fee_collects(cmd)?;
245        }
246        Ok(())
247    }
248
249    /// Unsubscribes from pool flash loan events for the specified AMM pool.
250    ///
251    /// # Errors
252    ///
253    /// Returns an error if the underlying client unsubscribe operation fails.
254    fn unsubscribe_pool_flash_events(
255        &mut self,
256        cmd: &UnsubscribePoolFlashEvents,
257    ) -> anyhow::Result<()> {
258        if Self::track_unsubscribe(
259            &mut self.subscriptions_pool_flash,
260            cmd.instrument_id,
261            "pool flash events",
262        ) {
263            self.client.unsubscribe_pool_flash_events(cmd)?;
264        }
265        Ok(())
266    }
267
268    /// Sends a pool snapshot request for a given AMM pool.
269    ///
270    /// # Errors
271    ///
272    /// Returns an error if the client fails to process the pool snapshot request.
273    pub fn request_pool_snapshot(&self, req: RequestPoolSnapshot) -> anyhow::Result<()> {
274        self.client.request_pool_snapshot(req)
275    }
276}
nautilus_data/defi/client.rs

nautilus_data/defi/
client.rs
