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

nautilus_data/defi/
client.rs
