nautilus_common/defi/
data_actor.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 actor functionality.
17//!
18//! This module provides DeFi subscription and unsubscription methods
19//! for the `DataActorCore`. All code in this module requires the `defi` feature flag.
20
21use nautilus_core::{Params, UUID4};
22use nautilus_model::{
23    defi::{Block, Blockchain, Pool, PoolFeeCollect, PoolFlash, PoolLiquidityUpdate, PoolSwap},
24    identifiers::{ClientId, InstrumentId},
25};
26
27use crate::{
28    actor::DataActorCore,
29    defi::{
30        DefiSubscribeCommand, DefiUnsubscribeCommand, SubscribeBlocks, SubscribePool,
31        SubscribePoolFeeCollects, SubscribePoolFlashEvents, SubscribePoolLiquidityUpdates,
32        SubscribePoolSwaps, UnsubscribeBlocks, UnsubscribePool, UnsubscribePoolFeeCollects,
33        UnsubscribePoolFlashEvents, UnsubscribePoolLiquidityUpdates, UnsubscribePoolSwaps,
34        switchboard::{
35            get_defi_blocks_topic, get_defi_collect_topic, get_defi_flash_topic,
36            get_defi_liquidity_topic, get_defi_pool_swaps_topic, get_defi_pool_topic,
37        },
38    },
39    messages::data::DataCommand,
40    msgbus::{MStr, Topic, TypedHandler},
41};
42
43impl DataActorCore {
44    /// Subscribes the actor to block.
45    pub fn subscribe_blocks(
46        &mut self,
47        topic: MStr<Topic>,
48        handler: TypedHandler<Block>,
49        chain: Blockchain,
50        client_id: Option<ClientId>,
51        params: Option<Params>,
52    ) {
53        self.check_registered();
54
55        let command = DataCommand::DefiSubscribe(DefiSubscribeCommand::Blocks(SubscribeBlocks {
56            chain,
57            client_id,
58            command_id: UUID4::new(),
59            ts_init: self.timestamp_ns(),
60            params,
61        }));
62
63        if self.add_block_subscription(topic, handler, command.clone()) {
64            self.send_data_cmd(command);
65        }
66    }
67
68    /// Subscribes the actor to pool.
69    pub fn subscribe_pool(
70        &mut self,
71        topic: MStr<Topic>,
72        handler: TypedHandler<Pool>,
73        instrument_id: InstrumentId,
74        client_id: Option<ClientId>,
75        params: Option<Params>,
76    ) {
77        self.check_registered();
78
79        let command = DataCommand::DefiSubscribe(DefiSubscribeCommand::Pool(SubscribePool {
80            instrument_id,
81            client_id,
82            command_id: UUID4::new(),
83            ts_init: self.timestamp_ns(),
84            params,
85        }));
86
87        if self.add_pool_subscription(topic, handler, command.clone()) {
88            self.send_data_cmd(command);
89        }
90    }
91
92    /// Subscribes the actor to pool swap.
93    pub fn subscribe_pool_swaps(
94        &mut self,
95        topic: MStr<Topic>,
96        handler: TypedHandler<PoolSwap>,
97        instrument_id: InstrumentId,
98        client_id: Option<ClientId>,
99        params: Option<Params>,
100    ) {
101        self.check_registered();
102
103        let command =
104            DataCommand::DefiSubscribe(DefiSubscribeCommand::PoolSwaps(SubscribePoolSwaps {
105                instrument_id,
106                client_id,
107                command_id: UUID4::new(),
108                ts_init: self.timestamp_ns(),
109                params,
110            }));
111
112        if self.add_pool_swap_subscription(topic, handler, command.clone()) {
113            self.send_data_cmd(command);
114        }
115    }
116
117    /// Subscribes the actor to pool liquidity update.
118    pub fn subscribe_pool_liquidity_updates(
119        &mut self,
120        topic: MStr<Topic>,
121        handler: TypedHandler<PoolLiquidityUpdate>,
122        instrument_id: InstrumentId,
123        client_id: Option<ClientId>,
124        params: Option<Params>,
125    ) {
126        self.check_registered();
127
128        let command = DataCommand::DefiSubscribe(DefiSubscribeCommand::PoolLiquidityUpdates(
129            SubscribePoolLiquidityUpdates {
130                instrument_id,
131                client_id,
132                command_id: UUID4::new(),
133                ts_init: self.timestamp_ns(),
134                params,
135            },
136        ));
137
138        if self.add_pool_liquidity_subscription(topic, handler, command.clone()) {
139            self.send_data_cmd(command);
140        }
141    }
142
143    /// Subscribes the actor to pool fee collect.
144    pub fn subscribe_pool_fee_collects(
145        &mut self,
146        topic: MStr<Topic>,
147        handler: TypedHandler<PoolFeeCollect>,
148        instrument_id: InstrumentId,
149        client_id: Option<ClientId>,
150        params: Option<Params>,
151    ) {
152        self.check_registered();
153
154        let command = DataCommand::DefiSubscribe(DefiSubscribeCommand::PoolFeeCollects(
155            SubscribePoolFeeCollects {
156                instrument_id,
157                client_id,
158                command_id: UUID4::new(),
159                ts_init: self.timestamp_ns(),
160                params,
161            },
162        ));
163
164        if self.add_pool_collect_subscription(topic, handler, command.clone()) {
165            self.send_data_cmd(command);
166        }
167    }
168
169    /// Subscribes the actor to pool flash event.
170    pub fn subscribe_pool_flash_events(
171        &mut self,
172        topic: MStr<Topic>,
173        handler: TypedHandler<PoolFlash>,
174        instrument_id: InstrumentId,
175        client_id: Option<ClientId>,
176        params: Option<Params>,
177    ) {
178        self.check_registered();
179
180        let command = DataCommand::DefiSubscribe(DefiSubscribeCommand::PoolFlashEvents(
181            SubscribePoolFlashEvents {
182                instrument_id,
183                client_id,
184                command_id: UUID4::new(),
185                ts_init: self.timestamp_ns(),
186                params,
187            },
188        ));
189
190        if self.add_pool_flash_subscription(topic, handler, command.clone()) {
191            self.send_data_cmd(command);
192        }
193    }
194
195    /// Unsubscribes the actor from blocks.
196    pub fn unsubscribe_blocks(
197        &mut self,
198        chain: Blockchain,
199        client_id: Option<ClientId>,
200        params: Option<Params>,
201    ) {
202        self.check_registered();
203
204        let topic = get_defi_blocks_topic(chain);
205        let retained = self.remove_block_subscription(topic);
206
207        let command = DefiUnsubscribeCommand::Blocks(UnsubscribeBlocks {
208            chain,
209            client_id,
210            command_id: UUID4::new(),
211            ts_init: self.timestamp_ns(),
212            params,
213        });
214
215        self.send_unsubscribe_cmd(retained, DataCommand::DefiUnsubscribe(command));
216    }
217
218    /// Unsubscribes the actor from pool definition updates.
219    pub fn unsubscribe_pool(
220        &mut self,
221        instrument_id: InstrumentId,
222        client_id: Option<ClientId>,
223        params: Option<Params>,
224    ) {
225        self.check_registered();
226
227        let topic = get_defi_pool_topic(instrument_id);
228        let retained = self.remove_pool_subscription(topic);
229
230        let command = DefiUnsubscribeCommand::Pool(UnsubscribePool {
231            instrument_id,
232            client_id,
233            command_id: UUID4::new(),
234            ts_init: self.timestamp_ns(),
235            params,
236        });
237
238        self.send_unsubscribe_cmd(retained, DataCommand::DefiUnsubscribe(command));
239    }
240
241    /// Unsubscribes the actor from pool swaps.
242    pub fn unsubscribe_pool_swaps(
243        &mut self,
244        instrument_id: InstrumentId,
245        client_id: Option<ClientId>,
246        params: Option<Params>,
247    ) {
248        self.check_registered();
249
250        let topic = get_defi_pool_swaps_topic(instrument_id);
251        let retained = self.remove_pool_swap_subscription(topic);
252
253        let command = DefiUnsubscribeCommand::PoolSwaps(UnsubscribePoolSwaps {
254            instrument_id,
255            client_id,
256            command_id: UUID4::new(),
257            ts_init: self.timestamp_ns(),
258            params,
259        });
260
261        self.send_unsubscribe_cmd(retained, DataCommand::DefiUnsubscribe(command));
262    }
263
264    /// Unsubscribes the actor from pool liquidity updates.
265    pub fn unsubscribe_pool_liquidity_updates(
266        &mut self,
267        instrument_id: InstrumentId,
268        client_id: Option<ClientId>,
269        params: Option<Params>,
270    ) {
271        self.check_registered();
272
273        let topic = get_defi_liquidity_topic(instrument_id);
274        let retained = self.remove_pool_liquidity_subscription(topic);
275
276        let command =
277            DefiUnsubscribeCommand::PoolLiquidityUpdates(UnsubscribePoolLiquidityUpdates {
278                instrument_id,
279                client_id,
280                command_id: UUID4::new(),
281                ts_init: self.timestamp_ns(),
282                params,
283            });
284
285        self.send_unsubscribe_cmd(retained, DataCommand::DefiUnsubscribe(command));
286    }
287
288    /// Unsubscribes the actor from pool fee collects.
289    pub fn unsubscribe_pool_fee_collects(
290        &mut self,
291        instrument_id: InstrumentId,
292        client_id: Option<ClientId>,
293        params: Option<Params>,
294    ) {
295        self.check_registered();
296
297        let topic = get_defi_collect_topic(instrument_id);
298        let retained = self.remove_pool_collect_subscription(topic);
299
300        let command = DefiUnsubscribeCommand::PoolFeeCollects(UnsubscribePoolFeeCollects {
301            instrument_id,
302            client_id,
303            command_id: UUID4::new(),
304            ts_init: self.timestamp_ns(),
305            params,
306        });
307
308        self.send_unsubscribe_cmd(retained, DataCommand::DefiUnsubscribe(command));
309    }
310
311    /// Unsubscribes the actor from pool flash events.
312    pub fn unsubscribe_pool_flash_events(
313        &mut self,
314        instrument_id: InstrumentId,
315        client_id: Option<ClientId>,
316        params: Option<Params>,
317    ) {
318        self.check_registered();
319
320        let topic = get_defi_flash_topic(instrument_id);
321        let retained = self.remove_pool_flash_subscription(topic);
322
323        let command = DefiUnsubscribeCommand::PoolFlashEvents(UnsubscribePoolFlashEvents {
324            instrument_id,
325            client_id,
326            command_id: UUID4::new(),
327            ts_init: self.timestamp_ns(),
328            params,
329        });
330
331        self.send_unsubscribe_cmd(retained, DataCommand::DefiUnsubscribe(command));
332    }
333}
nautilus_common/defi/data_actor.rs

nautilus_common/defi/
data_actor.rs
