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

nautilus_data/defi/
client.rs
