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 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::{
33    client::DataClientAdapter,
34    subscription::{DefiSubscriptionKey, SubscriptionRelease},
35};
36
37impl DataClientAdapter {
38    #[inline]
39    pub fn execute_defi_subscribe(&mut self, cmd: DefiSubscribeCommand) {
40        self.execute_defi_subscribe_with_retained(cmd, false);
41    }
42
43    pub(crate) fn execute_defi_subscribe_with_retained(
44        &mut self,
45        cmd: DefiSubscribeCommand,
46        retain_on_failure: bool,
47    ) {
48        #[rustfmt::skip]
49        let key = match &cmd {
50            DefiSubscribeCommand::Blocks(command) => DefiSubscriptionKey::Blocks(command.chain),
51            DefiSubscribeCommand::Pool(command) => DefiSubscriptionKey::Pool(command.instrument_id),
52            DefiSubscribeCommand::PoolSwaps(command) => DefiSubscriptionKey::PoolSwaps(command.instrument_id),
53            DefiSubscribeCommand::PoolLiquidityUpdates(command) => DefiSubscriptionKey::PoolLiquidityUpdates(command.instrument_id),
54            DefiSubscribeCommand::PoolFeeCollects(command) => DefiSubscriptionKey::PoolFeeCollects(command.instrument_id),
55            DefiSubscribeCommand::PoolFlashEvents(command) => DefiSubscriptionKey::PoolFlashEvents(command.instrument_id),
56        };
57        #[rustfmt::skip]
58        let active = match &key {
59            DefiSubscriptionKey::Blocks(chain) => self.subscriptions_blocks.contains(chain),
60            DefiSubscriptionKey::Pool(id) => self.subscriptions_pools.contains(id),
61            DefiSubscriptionKey::PoolSwaps(id) => self.subscriptions_pool_swaps.contains(id),
62            DefiSubscriptionKey::PoolLiquidityUpdates(id) => self.subscriptions_pool_liquidity_updates.contains(id),
63            DefiSubscriptionKey::PoolFeeCollects(id) => self.subscriptions_pool_fee_collects.contains(id),
64            DefiSubscriptionKey::PoolFlashEvents(id) => self.subscriptions_pool_flash.contains(id),
65        };
66
67        if active {
68            self.subscriptions_active_defi
69                .retain(key, cmd.command_id(), cmd);
70            return;
71        }
72
73        let retained = cmd.clone();
74        let cmd_debug = format!("{cmd:?}");
75        #[rustfmt::skip]
76        let result = match cmd {
77            DefiSubscribeCommand::Blocks(cmd) => self.subscribe_blocks(cmd),
78            DefiSubscribeCommand::Pool(cmd) => self.subscribe_pool(cmd),
79            DefiSubscribeCommand::PoolSwaps(cmd) => self.subscribe_pool_swaps(cmd),
80            DefiSubscribeCommand::PoolLiquidityUpdates(cmd) => self.subscribe_pool_liquidity_updates(cmd),
81            DefiSubscribeCommand::PoolFeeCollects(cmd) => self.subscribe_pool_fee_collects(cmd),
82            DefiSubscribeCommand::PoolFlashEvents(cmd) => self.subscribe_pool_flash_events(cmd),
83        };
84
85        if let Err(e) = result {
86            if retain_on_failure {
87                self.subscriptions_active_defi
88                    .retain(key, retained.command_id(), retained);
89            }
90
91            log_command_error(&cmd_debug, &e);
92            return;
93        }
94
95        if let Some(subscription) = self.subscriptions_active_defi.get_mut(&key) {
96            subscription.command = retained.clone();
97        }
98        self.subscriptions_active_defi
99            .retain(key, retained.command_id(), retained);
100    }
101
102    #[inline]
103    pub fn execute_defi_unsubscribe(&mut self, cmd: &DefiUnsubscribeCommand) {
104        #[rustfmt::skip]
105        let key = match cmd {
106            DefiUnsubscribeCommand::Blocks(command) => DefiSubscriptionKey::Blocks(command.chain),
107            DefiUnsubscribeCommand::Pool(command) => DefiSubscriptionKey::Pool(command.instrument_id),
108            DefiUnsubscribeCommand::PoolSwaps(command) => DefiSubscriptionKey::PoolSwaps(command.instrument_id),
109            DefiUnsubscribeCommand::PoolLiquidityUpdates(command) => DefiSubscriptionKey::PoolLiquidityUpdates(command.instrument_id),
110            DefiUnsubscribeCommand::PoolFeeCollects(command) => DefiSubscriptionKey::PoolFeeCollects(command.instrument_id),
111            DefiUnsubscribeCommand::PoolFlashEvents(command) => DefiSubscriptionKey::PoolFlashEvents(command.instrument_id),
112        };
113        let command = match self.subscriptions_active_defi.release(&key) {
114            SubscriptionRelease::Retained => return,
115            SubscriptionRelease::Final(subscribe) => {
116                subscribe.into_unsubscribe(cmd.command_id(), cmd.ts_init())
117            }
118            SubscriptionRelease::Untracked => cmd.clone(),
119        };
120
121        #[rustfmt::skip]
122        let unsub_result = match &command {
123            DefiUnsubscribeCommand::Blocks(cmd) => self.unsubscribe_blocks(cmd),
124            DefiUnsubscribeCommand::Pool(cmd) => self.unsubscribe_pool(cmd),
125            DefiUnsubscribeCommand::PoolSwaps(cmd) => self.unsubscribe_pool_swaps(cmd),
126            DefiUnsubscribeCommand::PoolLiquidityUpdates(cmd) => self.unsubscribe_pool_liquidity_updates(cmd),
127            DefiUnsubscribeCommand::PoolFeeCollects(cmd) => self.unsubscribe_pool_fee_collects(cmd),
128            DefiUnsubscribeCommand::PoolFlashEvents(cmd) => self.unsubscribe_pool_flash_events(cmd),
129        };
130
131        if let Err(e) = unsub_result {
132            log_command_error(&command, &e);
133        } else {
134            self.subscriptions_active_defi.remove(&key);
135        }
136    }
137
138    /// Executes a DeFi data request command by dispatching to the appropriate handler.
139    ///
140    /// # Errors
141    ///
142    /// Returns an error if the underlying client request fails.
143    #[inline]
144    pub fn execute_defi_request(&self, cmd: DefiRequestCommand) -> anyhow::Result<()> {
145        match cmd {
146            DefiRequestCommand::PoolSnapshot(cmd) => self.request_pool_snapshot(cmd),
147        }
148    }
149
150    /// Subscribes to block events for the specified blockchain.
151    ///
152    /// # Errors
153    ///
154    /// Returns an error if the underlying client subscribe operation fails.
155    fn subscribe_blocks(&mut self, cmd: SubscribeBlocks) -> anyhow::Result<()> {
156        Self::execute_tracked_subscribe(
157            self.client.as_mut(),
158            &mut self.subscriptions_blocks,
159            cmd.chain,
160            "blocks",
161            |client| client.subscribe_blocks(cmd),
162        )
163    }
164
165    /// Unsubscribes from block events for the specified blockchain.
166    ///
167    /// # Errors
168    ///
169    /// Returns an error if the underlying client unsubscribe operation fails.
170    fn unsubscribe_blocks(&mut self, cmd: &UnsubscribeBlocks) -> anyhow::Result<()> {
171        Self::execute_tracked_unsubscribe(
172            self.client.as_mut(),
173            &mut self.subscriptions_blocks,
174            &cmd.chain,
175            "blocks",
176            |client| client.unsubscribe_blocks(cmd),
177        )
178    }
179
180    /// Subscribes to pool definition updates for the specified AMM pool.
181    ///
182    /// # Errors
183    ///
184    /// Returns an error if the underlying client subscribe operation fails.
185    fn subscribe_pool(&mut self, cmd: SubscribePool) -> anyhow::Result<()> {
186        Self::execute_tracked_subscribe(
187            self.client.as_mut(),
188            &mut self.subscriptions_pools,
189            cmd.instrument_id,
190            "pool",
191            |client| client.subscribe_pool(cmd),
192        )
193    }
194
195    /// Subscribes to pool swap events for the specified AMM pool.
196    ///
197    /// # Errors
198    ///
199    /// Returns an error if the underlying client subscribe operation fails.
200    fn subscribe_pool_swaps(&mut self, cmd: SubscribePoolSwaps) -> anyhow::Result<()> {
201        Self::execute_tracked_subscribe(
202            self.client.as_mut(),
203            &mut self.subscriptions_pool_swaps,
204            cmd.instrument_id,
205            "pool swaps",
206            |client| client.subscribe_pool_swaps(cmd),
207        )
208    }
209
210    /// Subscribes to pool liquidity update events for the specified AMM pool.
211    ///
212    /// # Errors
213    ///
214    /// Returns an error if the underlying client subscribe operation fails.
215    fn subscribe_pool_liquidity_updates(
216        &mut self,
217        cmd: SubscribePoolLiquidityUpdates,
218    ) -> anyhow::Result<()> {
219        Self::execute_tracked_subscribe(
220            self.client.as_mut(),
221            &mut self.subscriptions_pool_liquidity_updates,
222            cmd.instrument_id,
223            "pool liquidity updates",
224            |client| client.subscribe_pool_liquidity_updates(cmd),
225        )
226    }
227
228    /// Subscribes to pool fee collect events for the specified AMM pool.
229    ///
230    /// # Errors
231    ///
232    /// Returns an error if the underlying client subscribe operation fails.
233    fn subscribe_pool_fee_collects(&mut self, cmd: SubscribePoolFeeCollects) -> anyhow::Result<()> {
234        Self::execute_tracked_subscribe(
235            self.client.as_mut(),
236            &mut self.subscriptions_pool_fee_collects,
237            cmd.instrument_id,
238            "pool fee collects",
239            |client| client.subscribe_pool_fee_collects(cmd),
240        )
241    }
242
243    /// Subscribes to pool flash loan events for the specified AMM pool.
244    ///
245    /// # Errors
246    ///
247    /// Returns an error if the underlying client subscribe operation fails.
248    fn subscribe_pool_flash_events(&mut self, cmd: SubscribePoolFlashEvents) -> anyhow::Result<()> {
249        Self::execute_tracked_subscribe(
250            self.client.as_mut(),
251            &mut self.subscriptions_pool_flash,
252            cmd.instrument_id,
253            "pool flash events",
254            |client| client.subscribe_pool_flash_events(cmd),
255        )
256    }
257
258    /// Unsubscribes from pool definition updates for the specified AMM pool.
259    ///
260    /// # Errors
261    ///
262    /// Returns an error if the underlying client unsubscribe operation fails.
263    fn unsubscribe_pool(&mut self, cmd: &UnsubscribePool) -> anyhow::Result<()> {
264        Self::execute_tracked_unsubscribe(
265            self.client.as_mut(),
266            &mut self.subscriptions_pools,
267            &cmd.instrument_id,
268            "pool",
269            |client| client.unsubscribe_pool(cmd),
270        )
271    }
272
273    /// Unsubscribes from swap events for the specified AMM pool.
274    ///
275    /// # Errors
276    ///
277    /// Returns an error if the underlying client unsubscribe operation fails.
278    fn unsubscribe_pool_swaps(&mut self, cmd: &UnsubscribePoolSwaps) -> anyhow::Result<()> {
279        Self::execute_tracked_unsubscribe(
280            self.client.as_mut(),
281            &mut self.subscriptions_pool_swaps,
282            &cmd.instrument_id,
283            "pool swaps",
284            |client| client.unsubscribe_pool_swaps(cmd),
285        )
286    }
287
288    /// Unsubscribes from pool liquidity update events for the specified AMM pool.
289    ///
290    /// # Errors
291    ///
292    /// Returns an error if the underlying client unsubscribe operation fails.
293    fn unsubscribe_pool_liquidity_updates(
294        &mut self,
295        cmd: &UnsubscribePoolLiquidityUpdates,
296    ) -> anyhow::Result<()> {
297        Self::execute_tracked_unsubscribe(
298            self.client.as_mut(),
299            &mut self.subscriptions_pool_liquidity_updates,
300            &cmd.instrument_id,
301            "pool liquidity updates",
302            |client| client.unsubscribe_pool_liquidity_updates(cmd),
303        )
304    }
305
306    /// Unsubscribes from pool fee collect events for the specified AMM pool.
307    ///
308    /// # Errors
309    ///
310    /// Returns an error if the underlying client unsubscribe operation fails.
311    fn unsubscribe_pool_fee_collects(
312        &mut self,
313        cmd: &UnsubscribePoolFeeCollects,
314    ) -> anyhow::Result<()> {
315        Self::execute_tracked_unsubscribe(
316            self.client.as_mut(),
317            &mut self.subscriptions_pool_fee_collects,
318            &cmd.instrument_id,
319            "pool fee collects",
320            |client| client.unsubscribe_pool_fee_collects(cmd),
321        )
322    }
323
324    /// Unsubscribes from pool flash loan events for the specified AMM pool.
325    ///
326    /// # Errors
327    ///
328    /// Returns an error if the underlying client unsubscribe operation fails.
329    fn unsubscribe_pool_flash_events(
330        &mut self,
331        cmd: &UnsubscribePoolFlashEvents,
332    ) -> anyhow::Result<()> {
333        Self::execute_tracked_unsubscribe(
334            self.client.as_mut(),
335            &mut self.subscriptions_pool_flash,
336            &cmd.instrument_id,
337            "pool flash events",
338            |client| client.unsubscribe_pool_flash_events(cmd),
339        )
340    }
341
342    /// Sends a pool snapshot request for a given AMM pool.
343    ///
344    /// # Errors
345    ///
346    /// Returns an error if the client fails to process the pool snapshot request.
347    pub fn request_pool_snapshot(&self, req: RequestPoolSnapshot) -> anyhow::Result<()> {
348        self.client.request_pool_snapshot(req)
349    }
350}
nautilus_data/defi/client.rs

nautilus_data/defi/
client.rs
