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

nautilus_data/defi/
client.rs
