1use std::sync::Arc;
19
20use ahash::AHashMap;
21use alloy_primitives::{Address, I256, U160, U256};
22use nautilus_core::UnixNanos;
23
24use crate::defi::{
25 DexType, PoolLiquidityUpdate, PoolSwap, SharedPool,
26 data::{
27 DexPoolData, PoolFeeCollect, PoolFeeProtocolCollect, PoolFeeProtocolUpdate,
28 PoolLiquidityUpdateType, block::BlockPosition, flash::PoolFlash,
29 },
30 pool_analysis::{
31 error::{
32 PoolEventKind, PoolEventLocation, PoolProfilerError, liquidity_error_with_location,
33 },
34 position::PoolPosition,
35 quote::SwapQuote,
36 size_estimator,
37 snapshot::{PROTOCOL_FEE_BASIS_POINTS_DENOMINATOR, PoolAnalytics, PoolSnapshot, PoolState},
38 swap_math::compute_swap_step,
39 },
40 reporting::{BlockchainSyncReportItems, BlockchainSyncReporter},
41 tick_map::{
42 TickMap,
43 full_math::{FullMath, Q128},
44 liquidity_math::{liquidity_math_add, try_liquidity_math_add},
45 sqrt_price_math::{get_amount0_delta, get_amount1_delta, get_amounts_for_liquidity},
46 tick::{CrossedTick, PoolTick},
47 tick_math::{
48 MAX_SQRT_RATIO, MIN_SQRT_RATIO, get_sqrt_ratio_at_tick, get_tick_at_sqrt_ratio,
49 },
50 },
51};
52
53#[derive(Debug, Clone)]
71#[cfg_attr(
72 feature = "python",
73 pyo3::pyclass(module = "nautilus_trader.model", from_py_object)
74)]
75#[cfg_attr(
76 feature = "python",
77 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.model")
78)]
79pub struct PoolProfiler {
80 pub pool: SharedPool,
82 positions: AHashMap<String, PoolPosition>,
84 pub tick_map: TickMap,
86 pub state: PoolState,
88 pub analytics: PoolAnalytics,
90 pub last_processed_event: Option<BlockPosition>,
92 pub last_processed_ts: Option<UnixNanos>,
94 pub is_initialized: bool,
96 reporter: Option<BlockchainSyncReporter>,
98 last_reported_block: u64,
100}
101
102impl PoolProfiler {
103 #[must_use]
109 pub fn new(pool: SharedPool) -> Self {
110 let tick_spacing = pool.tick_spacing.expect("Pool tick spacing must be set");
111 let mut state = PoolState::default();
112
113 if let Some((fee_protocol0, fee_protocol1)) =
114 initial_protocol_fee_basis_points(pool.dex.name, pool.fee)
115 {
116 state.set_protocol_fee_basis_points(fee_protocol0, fee_protocol1);
117 }
118
119 Self {
120 pool,
121 positions: AHashMap::new(),
122 tick_map: TickMap::new(tick_spacing),
123 state,
124 analytics: PoolAnalytics::default(),
125 last_processed_event: None,
126 last_processed_ts: None,
127 is_initialized: false,
128 reporter: None,
129 last_reported_block: 0,
130 }
131 }
132
133 pub fn initialize(&mut self, price_sqrt_ratio_x96: U160) -> Result<(), PoolProfilerError> {
140 if self.is_initialized {
141 return Err(PoolProfilerError::AlreadyInitialized {
142 instrument_id: self.pool.instrument_id,
143 pool_identifier: self.pool.pool_identifier,
144 });
145 }
146
147 let calculated_tick = get_tick_at_sqrt_ratio(price_sqrt_ratio_x96);
148
149 if let Some(initial_tick) = self.pool.initial_tick
150 && initial_tick != calculated_tick
151 {
152 return Err(PoolProfilerError::InitialTickMismatch {
153 instrument_id: self.pool.instrument_id,
154 pool_identifier: self.pool.pool_identifier,
155 initial_tick,
156 calculated_tick,
157 });
158 }
159
160 log::info!(
161 "Initializing pool profiler with tick {calculated_tick} and price sqrt ratio {price_sqrt_ratio_x96}"
162 );
163
164 self.state.current_tick = calculated_tick;
165 self.state.price_sqrt_ratio_x96 = price_sqrt_ratio_x96;
166 self.is_initialized = true;
167 Ok(())
168 }
169
170 pub fn check_if_initialized(&self, event_kind: PoolEventKind) -> Result<(), PoolProfilerError> {
177 if !self.is_initialized {
178 return Err(PoolProfilerError::NotInitialized {
179 instrument_id: self.pool.instrument_id,
180 pool_identifier: self.pool.pool_identifier,
181 event_kind,
182 });
183 }
184 Ok(())
185 }
186
187 fn event_location(
188 &self,
189 event_kind: PoolEventKind,
190 block: u64,
191 transaction_index: u32,
192 log_index: u32,
193 ) -> PoolEventLocation {
194 PoolEventLocation {
195 instrument_id: self.pool.instrument_id,
196 pool_identifier: self.pool.pool_identifier,
197 block,
198 transaction_index,
199 log_index,
200 event_kind,
201 }
202 }
203
204 pub fn process(&mut self, event: &DexPoolData) -> anyhow::Result<()> {
218 if self.check_if_already_processed(
219 event.block_number(),
220 event.transaction_index(),
221 event.log_index(),
222 ) {
223 return Ok(());
224 }
225
226 match event {
227 DexPoolData::Swap(swap) => self.process_swap(swap)?,
228 DexPoolData::LiquidityUpdate(update) => match update.kind {
229 PoolLiquidityUpdateType::Mint => self.process_mint(update)?,
230 PoolLiquidityUpdateType::Burn => self.process_burn(update)?,
231 },
232 DexPoolData::FeeCollect(collect) => self.process_collect(collect)?,
233 DexPoolData::FeeProtocolUpdate(update) => self.process_fee_protocol_update(update)?,
234 DexPoolData::FeeProtocolCollect(collect) => {
235 self.process_fee_protocol_collect(collect)?;
236 }
237 DexPoolData::Flash(flash) => self.process_flash(flash)?,
238 }
239
240 self.update_reporter_if_enabled(event.block_number());
241
242 Ok(())
243 }
244
245 fn check_if_already_processed(&self, block: u64, tx_idx: u32, log_idx: u32) -> bool {
247 if let Some(last_event) = &self.last_processed_event {
248 let should_skip = block < last_event.number
249 || (block == last_event.number && tx_idx < last_event.transaction_index)
250 || (block == last_event.number
251 && tx_idx == last_event.transaction_index
252 && log_idx <= last_event.log_index);
253
254 if should_skip {
255 log::debug!(
256 "Skipping already processed event at block {block} tx {tx_idx} log {log_idx}"
257 );
258 }
259 return should_skip;
260 }
261
262 false
263 }
264
265 fn update_reporter_if_enabled(&mut self, current_block: u64) {
267 if let Some(reporter) = &mut self.reporter {
269 let blocks_processed = current_block.saturating_sub(self.last_reported_block);
270
271 if blocks_processed > 0 {
272 reporter.update(blocks_processed as usize);
273 self.last_reported_block = current_block;
274
275 if reporter.should_log_progress(current_block, current_block) {
276 reporter.log_progress(current_block);
277 }
278 }
279 }
280 }
281
282 pub fn process_swap(&mut self, swap: &PoolSwap) -> anyhow::Result<()> {
305 self.check_if_initialized(PoolEventKind::Swap)?;
306
307 if self.check_if_already_processed(swap.block, swap.transaction_index, swap.log_index) {
308 return Ok(());
309 }
310
311 let zero_for_one = if swap.amount0.is_zero() && swap.amount1.is_zero() {
312 swap.sqrt_price_x96 < self.state.price_sqrt_ratio_x96
313 } else {
314 swap.amount0.is_positive()
315 };
316
317 let amount_specified = if zero_for_one {
318 swap.amount0
319 } else {
320 swap.amount1
321 };
322
323 let sqrt_price_limit_x96 = swap.sqrt_price_x96;
326 let location = self.event_location(
327 PoolEventKind::Swap,
328 swap.block,
329 swap.transaction_index,
330 swap.log_index,
331 );
332 let swap_quote = self
333 .simulate_swap_through_ticks(amount_specified, zero_for_one, sqrt_price_limit_x96, true)
334 .map_err(|e| Self::wrap_liquidity_error(e, location.clone()))?;
335
336 let tick_mismatch = swap.tick != swap_quote.tick_after;
337 let liquidity_mismatch = swap.liquidity != swap_quote.liquidity_after;
338 let sqrt_mismatch = swap.sqrt_price_x96 != swap_quote.sqrt_price_after_x96;
339 let structural_mismatch = tick_mismatch || liquidity_mismatch;
340 let anchoring_crossed_tick = if tick_mismatch {
341 self.tick_map
342 .get_all_ticks()
343 .keys()
344 .copied()
345 .filter(|&tick| {
346 self.tick_map.is_tick_initialized(tick)
347 && (swap_quote.tick_after >= tick) != (swap.tick >= tick)
348 })
349 .min()
350 } else {
351 None
352 };
353
354 if structural_mismatch
355 && (!swap_quote.crossed_ticks.is_empty() || anchoring_crossed_tick.is_some())
356 {
357 return Err(PoolProfilerError::SwapReplayMismatch {
358 location,
359 simulated_tick: swap_quote.tick_after,
360 event_tick: swap.tick,
361 simulated_liquidity: swap_quote.liquidity_after,
362 event_liquidity: swap.liquidity,
363 simulated_crossed_tick_count: swap_quote.crossed_ticks.len(),
364 anchoring_crossed_tick,
365 }
366 .into());
367 }
368 self.apply_swap_quote(&swap_quote);
369
370 if tick_mismatch {
372 log::warn!(
373 "Inconsistency in swap processing: Current tick mismatch: simulated {}, event {} on block {}",
374 swap_quote.tick_after,
375 swap.tick,
376 swap.block
377 );
378 }
379
380 if swap.tick != self.state.current_tick {
381 self.state.current_tick = swap.tick;
382 }
383
384 if liquidity_mismatch {
385 log::warn!(
386 "Inconsistency in swap processing: Active liquidity mismatch: simulated {}, event {} on block {}",
387 swap_quote.liquidity_after,
388 swap.liquidity,
389 swap.block
390 );
391 }
392
393 if swap.liquidity != self.tick_map.liquidity {
394 self.tick_map.liquidity = swap.liquidity;
395 }
396
397 if sqrt_mismatch {
398 log::warn!(
399 "Inconsistency in swap processing: Sqrt price mismatch: simulated {}, event {} on block {}",
400 swap_quote.sqrt_price_after_x96,
401 swap.sqrt_price_x96,
402 swap.block
403 );
404 }
405
406 if swap.sqrt_price_x96 != self.state.price_sqrt_ratio_x96 {
407 self.state.price_sqrt_ratio_x96 = swap.sqrt_price_x96;
408 }
409
410 self.last_processed_event = Some(
411 BlockPosition::new(
412 swap.block,
413 swap.transaction_hash.clone(),
414 swap.transaction_index,
415 swap.log_index,
416 )
417 .with_block_hash(swap.block_hash.clone()),
418 );
419 self.last_processed_ts = Some(swap.ts_event);
420 self.update_reporter_if_enabled(swap.block);
421 self.update_liquidity_analytics();
422
423 Ok(())
424 }
425
426 pub fn execute_swap(
441 &mut self,
442 sender: Address,
443 recipient: Address,
444 block: BlockPosition,
445 zero_for_one: bool,
446 amount_specified: I256,
447 sqrt_price_limit_x96: U160,
448 ) -> anyhow::Result<PoolSwap> {
449 self.check_if_initialized(PoolEventKind::Swap)?;
450
451 let swap_quote = self.simulate_swap_through_ticks(
452 amount_specified,
453 zero_for_one,
454 sqrt_price_limit_x96,
455 false,
456 )?;
457
458 self.apply_swap_quote(&swap_quote);
459
460 let swap_event = PoolSwap::new(
461 Arc::clone(&self.pool.chain),
462 Arc::clone(&self.pool.dex),
463 self.pool.instrument_id,
464 self.pool.pool_identifier,
465 block.number,
466 block.transaction_hash,
467 block.transaction_index,
468 block.log_index,
469 self.pool.ts_init, self.pool.ts_init, sender,
472 recipient,
473 swap_quote.amount0,
474 swap_quote.amount1,
475 self.state.price_sqrt_ratio_x96,
476 self.tick_map.liquidity,
477 self.state.current_tick,
478 );
479 Ok(swap_event)
480 }
481
482 pub fn simulate_swap_through_ticks(
520 &self,
521 amount_specified: I256,
522 zero_for_one: bool,
523 sqrt_price_limit_x96: U160,
524 traverse_empty_ranges: bool,
525 ) -> anyhow::Result<SwapQuote> {
526 self.validate_price_limit(sqrt_price_limit_x96, zero_for_one, true)?;
527
528 let exact_input = amount_specified.is_positive();
529 let fee_tier = self.pool.fee.expect("Pool fee should be initialized");
530
531 let mut current_sqrt_price = self.state.price_sqrt_ratio_x96;
532 let mut current_tick = self.state.current_tick;
533 let mut current_active_liquidity = self.tick_map.liquidity;
534 let mut amount_specified_remaining = amount_specified;
535 let mut amount_calculated = I256::ZERO;
536 let mut protocol_fee = U256::ZERO;
537 let mut lp_fee = U256::ZERO;
538 let mut crossed_ticks = Vec::new();
539
540 let fee_protocol = self.state.uniswap_v3_fee_protocol(zero_for_one);
542 let fee_protocol_basis_points = self.state.fee_protocol_basis_points(zero_for_one);
543
544 let mut current_fee_growth_global = if zero_for_one {
546 self.state.fee_growth_global_0
547 } else {
548 self.state.fee_growth_global_1
549 };
550
551 let replay_exact_input = traverse_empty_ranges && exact_input;
554
555 while (amount_specified_remaining != I256::ZERO
557 || (traverse_empty_ranges && current_active_liquidity == 0))
558 && (sqrt_price_limit_x96 != current_sqrt_price
559 || (replay_exact_input && amount_specified_remaining.is_positive()))
560 {
561 let iteration_start = (current_sqrt_price, current_tick, amount_specified_remaining);
562 let sqrt_price_start_x96 = current_sqrt_price;
563
564 let (mut tick_next, initialized) = self
565 .tick_map
566 .next_initialized_tick(current_tick, zero_for_one);
567
568 tick_next = tick_next.clamp(PoolTick::MIN_TICK, PoolTick::MAX_TICK);
570
571 let sqrt_price_next = get_sqrt_ratio_at_tick(tick_next);
573
574 let sqrt_price_target = if (zero_for_one && sqrt_price_next < sqrt_price_limit_x96)
576 || (!zero_for_one && sqrt_price_next > sqrt_price_limit_x96)
577 {
578 sqrt_price_limit_x96
579 } else {
580 sqrt_price_next
581 };
582
583 let mut swap_step_result = compute_swap_step(
584 current_sqrt_price,
585 sqrt_price_target,
586 current_active_liquidity,
587 amount_specified_remaining,
588 fee_tier,
589 )?;
590
591 if replay_exact_input
594 && swap_step_result.sqrt_ratio_next_x96 == sqrt_price_limit_x96
595 && swap_step_result.sqrt_ratio_next_x96 != sqrt_price_next
596 {
597 swap_step_result.fee_amount =
598 amount_specified_remaining.into_raw() - swap_step_result.amount_in;
599 }
600
601 current_sqrt_price = swap_step_result.sqrt_ratio_next_x96;
603
604 if exact_input {
606 amount_specified_remaining -= FullMath::truncate_to_i256(
608 swap_step_result.amount_in + swap_step_result.fee_amount,
609 );
610 amount_calculated -= FullMath::truncate_to_i256(swap_step_result.amount_out);
611 } else {
612 amount_specified_remaining +=
614 FullMath::truncate_to_i256(swap_step_result.amount_out);
615 amount_calculated += FullMath::truncate_to_i256(
616 swap_step_result.amount_in + swap_step_result.fee_amount,
617 );
618 }
619
620 let mut step_fee_amount = swap_step_result.fee_amount;
622
623 if fee_protocol > 0 || fee_protocol_basis_points.is_some() {
624 let protocol_fee_delta = Self::protocol_fee_delta(
625 swap_step_result.fee_amount,
626 fee_protocol,
627 fee_protocol_basis_points,
628 )?;
629 step_fee_amount -= protocol_fee_delta;
630 protocol_fee += protocol_fee_delta;
631 }
632
633 lp_fee += step_fee_amount;
635
636 if current_active_liquidity > 0 {
638 let fee_growth_delta =
639 FullMath::mul_div(step_fee_amount, Q128, U256::from(current_active_liquidity))?;
640 current_fee_growth_global += fee_growth_delta;
641 }
642
643 if swap_step_result.sqrt_ratio_next_x96 == sqrt_price_next {
645 if initialized {
649 crossed_ticks.push(CrossedTick::new(
650 tick_next,
651 zero_for_one,
652 if zero_for_one {
653 current_fee_growth_global
654 } else {
655 self.state.fee_growth_global_0
656 },
657 if zero_for_one {
658 self.state.fee_growth_global_1
659 } else {
660 current_fee_growth_global
661 },
662 ));
663
664 if let Some(tick_data) = self.tick_map.get_tick(tick_next) {
666 let liquidity_net = tick_data.liquidity_net;
667 current_active_liquidity = if zero_for_one {
668 try_liquidity_math_add(current_active_liquidity, -liquidity_net)?
669 } else {
670 try_liquidity_math_add(current_active_liquidity, liquidity_net)?
671 };
672 }
673 }
674
675 current_tick = if zero_for_one {
676 tick_next - 1
677 } else {
678 tick_next
679 };
680 } else if swap_step_result.sqrt_ratio_next_x96 != sqrt_price_start_x96 {
681 current_tick = get_tick_at_sqrt_ratio(current_sqrt_price);
684 }
685
686 if (current_sqrt_price, current_tick, amount_specified_remaining) == iteration_start {
687 anyhow::bail!(
688 "Swap simulation made no progress at tick {current_tick} with sqrt price \
689 {current_sqrt_price} and remaining amount {amount_specified_remaining}"
690 );
691 }
692 }
693
694 let (amount0, amount1) = if zero_for_one == exact_input {
696 (
697 amount_specified - amount_specified_remaining,
698 amount_calculated,
699 )
700 } else {
701 (
702 amount_calculated,
703 amount_specified - amount_specified_remaining,
704 )
705 };
706
707 let quote = SwapQuote::new(
708 self.pool.instrument_id,
709 amount0,
710 amount1,
711 self.state.price_sqrt_ratio_x96,
712 current_sqrt_price,
713 self.state.current_tick,
714 current_tick,
715 current_active_liquidity,
716 current_fee_growth_global,
717 lp_fee,
718 protocol_fee,
719 crossed_ticks,
720 );
721 Ok(quote)
722 }
723
724 pub fn apply_swap_quote(&mut self, swap_quote: &SwapQuote) {
732 self.state.current_tick = swap_quote.tick_after;
733 self.state.price_sqrt_ratio_x96 = swap_quote.sqrt_price_after_x96;
734
735 self.apply_swap_quote_fee_state(swap_quote);
736
737 for crossed in &swap_quote.crossed_ticks {
738 let liquidity_net =
739 self.tick_map
740 .cross_tick(crossed.tick, crossed.fee_growth_0, crossed.fee_growth_1);
741
742 self.tick_map.liquidity = if crossed.zero_for_one {
743 liquidity_math_add(self.tick_map.liquidity, -liquidity_net)
744 } else {
745 liquidity_math_add(self.tick_map.liquidity, liquidity_net)
746 };
747 }
748 self.analytics.total_swaps += 1;
749
750 debug_assert_eq!(
751 self.tick_map.liquidity, swap_quote.liquidity_after,
752 "Liquidity mismatch in apply_swap_quote: computed={}, quote={}",
753 self.tick_map.liquidity, swap_quote.liquidity_after
754 );
755 }
756
757 fn apply_swap_quote_fee_state(&mut self, swap_quote: &SwapQuote) {
758 if swap_quote.amount0.is_zero() && swap_quote.amount1.is_zero() {
759 return;
760 }
761
762 if swap_quote.zero_for_one() {
763 self.state.fee_growth_global_0 = swap_quote.fee_growth_global_after;
764 self.state.protocol_fees_token0 += swap_quote.protocol_fee;
765 } else {
766 self.state.fee_growth_global_1 = swap_quote.fee_growth_global_after;
767 self.state.protocol_fees_token1 += swap_quote.protocol_fee;
768 }
769 }
770
771 #[must_use]
775 pub fn wrap_liquidity_error(err: anyhow::Error, location: PoolEventLocation) -> anyhow::Error {
776 match err.downcast::<super::error::LiquidityMathError>() {
777 Ok(math_err) => anyhow::Error::from(liquidity_error_with_location(math_err, location)),
778 Err(other) => other,
779 }
780 }
781
782 pub fn quote_swap(
804 &self,
805 amount_specified: I256,
806 zero_for_one: bool,
807 sqrt_price_limit_x96: Option<U160>,
808 ) -> anyhow::Result<SwapQuote> {
809 self.check_if_initialized(PoolEventKind::Swap)?;
810
811 if amount_specified.is_zero() {
812 anyhow::bail!("Cannot quote swap with zero amount");
813 }
814
815 if let Some(price_limit) = sqrt_price_limit_x96 {
816 self.validate_price_limit(price_limit, zero_for_one, false)?;
817 }
818
819 let limit = sqrt_price_limit_x96.unwrap_or_else(|| {
820 if zero_for_one {
821 MIN_SQRT_RATIO + U160::from(1)
822 } else {
823 MAX_SQRT_RATIO - U160::from(1)
824 }
825 });
826
827 self.simulate_swap_through_ticks(amount_specified, zero_for_one, limit, false)
828 }
829
830 pub fn swap_exact_in(
835 &self,
836 amount_in: U256,
837 zero_for_one: bool,
838 sqrt_price_limit_x96: Option<U160>,
839 ) -> anyhow::Result<SwapQuote> {
840 let amount_specified = I256::from(amount_in);
842 let quote = self.quote_swap(amount_specified, zero_for_one, sqrt_price_limit_x96)?;
843
844 Ok(quote)
845 }
846
847 pub fn swap_exact_out(
853 &self,
854 amount_out: U256,
855 zero_for_one: bool,
856 sqrt_price_limit_x96: Option<U160>,
857 ) -> anyhow::Result<SwapQuote> {
858 let amount_specified = -I256::from(amount_out);
860 let quote = self.quote_swap(amount_specified, zero_for_one, sqrt_price_limit_x96)?;
861 quote.validate_exact_output(amount_out)?;
862
863 Ok(quote)
864 }
865
866 pub fn swap_to_lower_sqrt_price(
871 &self,
872 sqrt_price_limit_x96: U160,
873 ) -> anyhow::Result<SwapQuote> {
874 self.quote_swap(I256::MAX, true, Some(sqrt_price_limit_x96))
875 }
876
877 pub fn swap_to_higher_sqrt_price(
882 &self,
883 sqrt_price_limit_x96: U160,
884 ) -> anyhow::Result<SwapQuote> {
885 self.quote_swap(I256::MAX, false, Some(sqrt_price_limit_x96))
886 }
887
888 pub fn size_for_impact_bps(&self, impact_bps: u32, zero_for_one: bool) -> anyhow::Result<U256> {
903 let config = size_estimator::EstimationConfig::default();
904 size_estimator::size_for_impact_bps(self, impact_bps, zero_for_one, &config)
905 }
906
907 pub fn size_for_impact_bps_detailed(
921 &self,
922 impact_bps: u32,
923 zero_for_one: bool,
924 ) -> anyhow::Result<size_estimator::SizeForImpactResult> {
925 let config = size_estimator::EstimationConfig::default();
926 size_estimator::size_for_impact_bps_detailed(self, impact_bps, zero_for_one, &config)
927 }
928
929 fn validate_price_limit(
936 &self,
937 limit_price_sqrt: U160,
938 zero_for_one: bool,
939 allow_equal: bool,
940 ) -> anyhow::Result<()> {
941 if zero_for_one {
942 if limit_price_sqrt > self.state.price_sqrt_ratio_x96
944 || (!allow_equal && limit_price_sqrt == self.state.price_sqrt_ratio_x96)
945 {
946 anyhow::bail!("Price limit must be less than current price for zero_for_one swaps");
947 }
948
949 if limit_price_sqrt <= MIN_SQRT_RATIO {
950 anyhow::bail!(
951 "Price limit {limit_price_sqrt} must be greater than MIN_SQRT_RATIO {MIN_SQRT_RATIO}"
952 );
953 }
954 } else {
955 if limit_price_sqrt < self.state.price_sqrt_ratio_x96
957 || (!allow_equal && limit_price_sqrt == self.state.price_sqrt_ratio_x96)
958 {
959 anyhow::bail!(
960 "Price limit must be greater than current price for one_for_zero swaps"
961 );
962 }
963
964 if limit_price_sqrt >= MAX_SQRT_RATIO {
965 anyhow::bail!(
966 "Price limit {limit_price_sqrt} must be less than MAX_SQRT_RATIO {MAX_SQRT_RATIO}"
967 );
968 }
969 }
970
971 Ok(())
972 }
973
974 pub fn process_mint(&mut self, update: &PoolLiquidityUpdate) -> anyhow::Result<()> {
986 self.check_if_initialized(PoolEventKind::Mint)?;
987
988 if self.check_if_already_processed(update.block, update.transaction_index, update.log_index)
989 {
990 return Ok(());
991 }
992
993 self.validate_ticks(update.tick_lower, update.tick_upper)?;
994 let location = self.event_location(
995 PoolEventKind::Mint,
996 update.block,
997 update.transaction_index,
998 update.log_index,
999 );
1000 self.add_liquidity(
1001 &update.owner,
1002 update.tick_lower,
1003 update.tick_upper,
1004 update.position_liquidity,
1005 update.amount0,
1006 update.amount1,
1007 )
1008 .map_err(|e| Self::wrap_liquidity_error(e, location))?;
1009
1010 self.analytics.total_mints += 1;
1011 self.last_processed_event = Some(
1012 BlockPosition::new(
1013 update.block,
1014 update.transaction_hash.clone(),
1015 update.transaction_index,
1016 update.log_index,
1017 )
1018 .with_block_hash(update.block_hash.clone()),
1019 );
1020 self.last_processed_ts = Some(update.ts_event);
1021 self.update_reporter_if_enabled(update.block);
1022 self.update_liquidity_analytics();
1023
1024 Ok(())
1025 }
1026
1027 fn add_liquidity(
1032 &mut self,
1033 owner: &Address,
1034 tick_lower: i32,
1035 tick_upper: i32,
1036 liquidity: u128,
1037 amount0: U256,
1038 amount1: U256,
1039 ) -> anyhow::Result<()> {
1040 let liquidity_delta = i128::try_from(liquidity)
1041 .map_err(|_| anyhow::anyhow!("Liquidity {liquidity} exceeds i128::MAX"))?;
1042 self.update_position(
1043 owner,
1044 tick_lower,
1045 tick_upper,
1046 liquidity_delta,
1047 amount0,
1048 amount1,
1049 )?;
1050
1051 self.analytics.total_amount0_deposited += amount0;
1053 self.analytics.total_amount1_deposited += amount1;
1054
1055 Ok(())
1056 }
1057
1058 pub fn execute_mint(
1073 &mut self,
1074 recipient: Address,
1075 block: BlockPosition,
1076 tick_lower: i32,
1077 tick_upper: i32,
1078 liquidity: u128,
1079 ) -> anyhow::Result<PoolLiquidityUpdate> {
1080 self.check_if_initialized(PoolEventKind::Mint)?;
1081
1082 self.validate_ticks(tick_lower, tick_upper)?;
1083 let (amount0, amount1) = get_amounts_for_liquidity(
1084 self.state.price_sqrt_ratio_x96,
1085 tick_lower,
1086 tick_upper,
1087 liquidity,
1088 true,
1089 );
1090 self.add_liquidity(
1091 &recipient, tick_lower, tick_upper, liquidity, amount0, amount1,
1092 )?;
1093
1094 self.analytics.total_mints += 1;
1095
1096 let event = PoolLiquidityUpdate::new(
1097 Arc::clone(&self.pool.chain),
1098 Arc::clone(&self.pool.dex),
1099 self.pool.instrument_id,
1100 self.pool.pool_identifier,
1101 PoolLiquidityUpdateType::Mint,
1102 block.number,
1103 block.transaction_hash,
1104 block.transaction_index,
1105 block.log_index,
1106 None,
1107 recipient,
1108 liquidity,
1109 amount0,
1110 amount1,
1111 tick_lower,
1112 tick_upper,
1113 self.pool.ts_init, self.pool.ts_init, );
1116
1117 Ok(event)
1118 }
1119
1120 pub fn process_burn(&mut self, update: &PoolLiquidityUpdate) -> anyhow::Result<()> {
1132 self.check_if_initialized(PoolEventKind::Burn)?;
1133
1134 if self.check_if_already_processed(update.block, update.transaction_index, update.log_index)
1135 {
1136 return Ok(());
1137 }
1138
1139 self.validate_ticks(update.tick_lower, update.tick_upper)?;
1140
1141 let liquidity_delta = i128::try_from(update.position_liquidity).map_err(|_| {
1143 anyhow::anyhow!("Liquidity {} exceeds i128::MAX", update.position_liquidity)
1144 })?;
1145 let location = self.event_location(
1146 PoolEventKind::Burn,
1147 update.block,
1148 update.transaction_index,
1149 update.log_index,
1150 );
1151
1152 self.update_position(
1153 &update.owner,
1154 update.tick_lower,
1155 update.tick_upper,
1156 -liquidity_delta,
1157 update.amount0,
1158 update.amount1,
1159 )
1160 .map_err(|e| Self::wrap_liquidity_error(e, location))?;
1161
1162 self.analytics.total_burns += 1;
1163 self.last_processed_event = Some(
1164 BlockPosition::new(
1165 update.block,
1166 update.transaction_hash.clone(),
1167 update.transaction_index,
1168 update.log_index,
1169 )
1170 .with_block_hash(update.block_hash.clone()),
1171 );
1172 self.last_processed_ts = Some(update.ts_event);
1173 self.update_reporter_if_enabled(update.block);
1174 self.update_liquidity_analytics();
1175
1176 Ok(())
1177 }
1178
1179 pub fn execute_burn(
1192 &mut self,
1193 recipient: Address,
1194 block: BlockPosition,
1195 tick_lower: i32,
1196 tick_upper: i32,
1197 liquidity: u128,
1198 ) -> anyhow::Result<PoolLiquidityUpdate> {
1199 self.check_if_initialized(PoolEventKind::Burn)?;
1200
1201 self.validate_ticks(tick_lower, tick_upper)?;
1202 let (amount0, amount1) = get_amounts_for_liquidity(
1203 self.state.price_sqrt_ratio_x96,
1204 tick_lower,
1205 tick_upper,
1206 liquidity,
1207 false,
1208 );
1209
1210 let liquidity_delta = i128::try_from(liquidity)
1212 .map_err(|_| anyhow::anyhow!("Liquidity {liquidity} exceeds i128::MAX"))?;
1213 self.update_position(
1214 &recipient,
1215 tick_lower,
1216 tick_upper,
1217 -liquidity_delta,
1218 amount0,
1219 amount1,
1220 )?;
1221
1222 self.analytics.total_burns += 1;
1223
1224 let event = PoolLiquidityUpdate::new(
1225 Arc::clone(&self.pool.chain),
1226 Arc::clone(&self.pool.dex),
1227 self.pool.instrument_id,
1228 self.pool.pool_identifier,
1229 PoolLiquidityUpdateType::Burn,
1230 block.number,
1231 block.transaction_hash,
1232 block.transaction_index,
1233 block.log_index,
1234 None,
1235 recipient,
1236 liquidity,
1237 amount0,
1238 amount1,
1239 tick_lower,
1240 tick_upper,
1241 self.pool.ts_init, self.pool.ts_init, );
1244
1245 Ok(event)
1246 }
1247
1248 pub fn process_collect(&mut self, collect: &PoolFeeCollect) -> anyhow::Result<()> {
1261 self.check_if_initialized(PoolEventKind::Collect)?;
1262
1263 if self.check_if_already_processed(
1264 collect.block,
1265 collect.transaction_index,
1266 collect.log_index,
1267 ) {
1268 return Ok(());
1269 }
1270 let position_key =
1271 PoolPosition::get_position_key(&collect.owner, collect.tick_lower, collect.tick_upper);
1272
1273 if let Some(position) = self.positions.get_mut(&position_key) {
1274 position.collect_fees(collect.amount0, collect.amount1);
1275 }
1276
1277 self.cleanup_position_if_empty(&position_key);
1279
1280 self.analytics.total_amount0_collected += U256::from(collect.amount0);
1281 self.analytics.total_amount1_collected += U256::from(collect.amount1);
1282
1283 self.analytics.total_fee_collects += 1;
1284 self.last_processed_event = Some(
1285 BlockPosition::new(
1286 collect.block,
1287 collect.transaction_hash.clone(),
1288 collect.transaction_index,
1289 collect.log_index,
1290 )
1291 .with_block_hash(collect.block_hash.clone()),
1292 );
1293 self.last_processed_ts = Some(collect.ts_event);
1294 self.update_reporter_if_enabled(collect.block);
1295 self.update_liquidity_analytics();
1296
1297 Ok(())
1298 }
1299
1300 pub fn process_fee_protocol_update(
1310 &mut self,
1311 update: &PoolFeeProtocolUpdate,
1312 ) -> anyhow::Result<()> {
1313 if self.check_if_already_processed(update.block, update.transaction_index, update.log_index)
1314 {
1315 return Ok(());
1316 }
1317
1318 if update.dex.name == DexType::PancakeSwapV3 {
1319 self.state
1320 .set_protocol_fee_basis_points(update.fee_protocol0_new, update.fee_protocol1_new);
1321 } else {
1322 let fee_protocol = update
1323 .uniswap_v3_packed()
1324 .ok_or_else(|| anyhow::anyhow!("invalid Uniswap V3 fee protocol update"))?;
1325 self.state.set_uniswap_v3_fee_protocol(fee_protocol);
1326 }
1327
1328 self.last_processed_event = Some(
1329 BlockPosition::new(
1330 update.block,
1331 update.transaction_hash.clone(),
1332 update.transaction_index,
1333 update.log_index,
1334 )
1335 .with_block_hash(update.block_hash.clone()),
1336 );
1337 self.last_processed_ts = Some(update.ts_event);
1338 self.update_reporter_if_enabled(update.block);
1339
1340 Ok(())
1341 }
1342
1343 #[allow(
1353 clippy::missing_errors_doc,
1354 reason = "protocol fee collection only updates in-memory state"
1355 )]
1356 pub fn process_fee_protocol_collect(
1357 &mut self,
1358 collect: &PoolFeeProtocolCollect,
1359 ) -> anyhow::Result<()> {
1360 if self.check_if_already_processed(
1361 collect.block,
1362 collect.transaction_index,
1363 collect.log_index,
1364 ) {
1365 return Ok(());
1366 }
1367
1368 self.state.protocol_fees_token0 = self
1369 .state
1370 .protocol_fees_token0
1371 .saturating_sub(U256::from(collect.amount0));
1372 self.state.protocol_fees_token1 = self
1373 .state
1374 .protocol_fees_token1
1375 .saturating_sub(U256::from(collect.amount1));
1376
1377 self.last_processed_event = Some(
1378 BlockPosition::new(
1379 collect.block,
1380 collect.transaction_hash.clone(),
1381 collect.transaction_index,
1382 collect.log_index,
1383 )
1384 .with_block_hash(collect.block_hash.clone()),
1385 );
1386 self.last_processed_ts = Some(collect.ts_event);
1387 self.update_reporter_if_enabled(collect.block);
1388
1389 Ok(())
1390 }
1391
1392 pub fn process_flash(&mut self, flash: &PoolFlash) -> anyhow::Result<()> {
1400 self.check_if_initialized(PoolEventKind::Flash)?;
1401
1402 if self.check_if_already_processed(flash.block, flash.transaction_index, flash.log_index) {
1403 return Ok(());
1404 }
1405
1406 self.update_flash_state(flash.paid0, flash.paid1)?;
1407
1408 self.analytics.total_flashes += 1;
1409 self.last_processed_event = Some(
1410 BlockPosition::new(
1411 flash.block,
1412 flash.transaction_hash.clone(),
1413 flash.transaction_index,
1414 flash.log_index,
1415 )
1416 .with_block_hash(flash.block_hash.clone()),
1417 );
1418 self.last_processed_ts = Some(flash.ts_event);
1419 self.update_reporter_if_enabled(flash.block);
1420 self.update_liquidity_analytics();
1421
1422 Ok(())
1423 }
1424
1425 pub fn execute_flash(
1438 &mut self,
1439 sender: Address,
1440 recipient: Address,
1441 block: BlockPosition,
1442 amount0: U256,
1443 amount1: U256,
1444 ) -> anyhow::Result<PoolFlash> {
1445 self.check_if_initialized(PoolEventKind::Flash)?;
1446
1447 let fee_tier = self.pool.fee.expect("Pool fee should be initialized");
1448
1449 let paid0 = if amount0 > U256::ZERO {
1451 FullMath::mul_div_rounding_up(amount0, U256::from(fee_tier), U256::from(1_000_000))?
1452 } else {
1453 U256::ZERO
1454 };
1455
1456 let paid1 = if amount1 > U256::ZERO {
1457 FullMath::mul_div_rounding_up(amount1, U256::from(fee_tier), U256::from(1_000_000))?
1458 } else {
1459 U256::ZERO
1460 };
1461
1462 self.update_flash_state(paid0, paid1)?;
1463 self.analytics.total_flashes += 1;
1464
1465 let flash_event = PoolFlash::new(
1466 Arc::clone(&self.pool.chain),
1467 Arc::clone(&self.pool.dex),
1468 self.pool.instrument_id,
1469 self.pool.pool_identifier,
1470 block.number,
1471 block.transaction_hash,
1472 block.transaction_index,
1473 block.log_index,
1474 self.pool.ts_init, self.pool.ts_init, sender,
1477 recipient,
1478 amount0,
1479 amount1,
1480 paid0,
1481 paid1,
1482 );
1483
1484 Ok(flash_event)
1485 }
1486
1487 fn update_flash_state(&mut self, paid0: U256, paid1: U256) -> anyhow::Result<()> {
1495 let liquidity = self.tick_map.liquidity;
1496 if liquidity == 0 {
1497 anyhow::bail!("No liquidity")
1498 }
1499
1500 let fee_protocol_0 = self.state.uniswap_v3_fee_protocol(true);
1501 let fee_protocol_1 = self.state.uniswap_v3_fee_protocol(false);
1502 let fee_protocol0_basis_points = self.state.fee_protocol_basis_points(true);
1503 let fee_protocol1_basis_points = self.state.fee_protocol_basis_points(false);
1504
1505 if paid0 > U256::ZERO {
1507 let protocol_fee_0 =
1508 Self::protocol_fee_delta(paid0, fee_protocol_0, fee_protocol0_basis_points)?;
1509
1510 if protocol_fee_0 > U256::ZERO {
1511 self.state.protocol_fees_token0 += protocol_fee_0;
1512 }
1513
1514 let lp_fee_0 = paid0 - protocol_fee_0;
1515 let delta = FullMath::mul_div(lp_fee_0, Q128, U256::from(liquidity))?;
1516 self.state.fee_growth_global_0 += delta;
1517 }
1518
1519 if paid1 > U256::ZERO {
1521 let protocol_fee_1 =
1522 Self::protocol_fee_delta(paid1, fee_protocol_1, fee_protocol1_basis_points)?;
1523
1524 if protocol_fee_1 > U256::ZERO {
1525 self.state.protocol_fees_token1 += protocol_fee_1;
1526 }
1527
1528 let lp_fee_1 = paid1 - protocol_fee_1;
1529 let delta = FullMath::mul_div(lp_fee_1, Q128, U256::from(liquidity))?;
1530 self.state.fee_growth_global_1 += delta;
1531 }
1532
1533 Ok(())
1534 }
1535
1536 fn protocol_fee_delta(
1537 fee_amount: U256,
1538 uniswap_v3_fee_protocol: u8,
1539 fee_protocol_basis_points: Option<u32>,
1540 ) -> anyhow::Result<U256> {
1541 if let Some(basis_points) = fee_protocol_basis_points {
1542 return FullMath::mul_div(
1543 fee_amount,
1544 U256::from(basis_points),
1545 U256::from(PROTOCOL_FEE_BASIS_POINTS_DENOMINATOR),
1546 );
1547 }
1548
1549 if uniswap_v3_fee_protocol > 0 {
1550 Ok(fee_amount / U256::from(uniswap_v3_fee_protocol))
1551 } else {
1552 Ok(U256::ZERO)
1553 }
1554 }
1555
1556 fn update_position(
1561 &mut self,
1562 owner: &Address,
1563 tick_lower: i32,
1564 tick_upper: i32,
1565 liquidity_delta: i128,
1566 amount0: U256,
1567 amount1: U256,
1568 ) -> anyhow::Result<()> {
1569 let current_tick = self.state.current_tick;
1570 let position_key = PoolPosition::get_position_key(owner, tick_lower, tick_upper);
1571
1572 if liquidity_delta < 0 {
1574 let position_liquidity = self
1575 .positions
1576 .get(&position_key)
1577 .map_or(0, |position| position.liquidity);
1578 let burn_amount = liquidity_delta.unsigned_abs();
1579 if position_liquidity < burn_amount {
1580 anyhow::bail!(
1581 "Position liquidity {position_liquidity} is less than the requested burn amount of {burn_amount}"
1582 );
1583 }
1584 }
1585
1586 let new_active_liquidity = if tick_lower <= current_tick && current_tick < tick_upper {
1589 Some(try_liquidity_math_add(
1590 self.tick_map.liquidity,
1591 liquidity_delta,
1592 )?)
1593 } else {
1594 None
1595 };
1596
1597 for tick_value in [tick_lower, tick_upper] {
1598 let liquidity_gross = self
1599 .tick_map
1600 .get_tick(tick_value)
1601 .map_or(0, |tick| tick.liquidity_gross);
1602 let liquidity_after = try_liquidity_math_add(liquidity_gross, liquidity_delta)?;
1603 if liquidity_after > self.tick_map.max_liquidity_per_tick {
1604 anyhow::bail!(
1605 "Liquidity {liquidity_after} exceeds maximum per tick {} at tick {tick_value}",
1606 self.tick_map.max_liquidity_per_tick,
1607 );
1608 }
1609 }
1610
1611 let position = self
1612 .positions
1613 .entry(position_key)
1614 .or_insert(PoolPosition::new(*owner, tick_lower, tick_upper, 0));
1615
1616 let flipped_lower = self.tick_map.update(
1618 tick_lower,
1619 current_tick,
1620 liquidity_delta,
1621 false,
1622 self.state.fee_growth_global_0,
1623 self.state.fee_growth_global_1,
1624 );
1625 let flipped_upper = self.tick_map.update(
1626 tick_upper,
1627 current_tick,
1628 liquidity_delta,
1629 true,
1630 self.state.fee_growth_global_0,
1631 self.state.fee_growth_global_1,
1632 );
1633
1634 let (fee_growth_inside_0, fee_growth_inside_1) = self.tick_map.get_fee_growth_inside(
1635 tick_lower,
1636 tick_upper,
1637 current_tick,
1638 self.state.fee_growth_global_0,
1639 self.state.fee_growth_global_1,
1640 );
1641 position.update_fees(fee_growth_inside_0, fee_growth_inside_1);
1642 position.update_liquidity(liquidity_delta);
1643 position.update_amounts(liquidity_delta, amount0, amount1);
1644
1645 if let Some(active_liquidity) = new_active_liquidity {
1646 self.tick_map.liquidity = active_liquidity;
1647 }
1648
1649 if liquidity_delta < 0 && flipped_lower {
1651 self.tick_map.clear(tick_lower);
1652 }
1653
1654 if liquidity_delta < 0 && flipped_upper {
1655 self.tick_map.clear(tick_upper);
1656 }
1657
1658 Ok(())
1659 }
1660
1661 fn cleanup_position_if_empty(&mut self, position_key: &str) {
1665 if let Some(position) = self.positions.get(position_key)
1666 && position.is_empty()
1667 {
1668 log::debug!(
1669 "CLEANING UP EMPTY POSITION: owner={}, ticks=[{}, {}]",
1670 position.owner,
1671 position.tick_lower,
1672 position.tick_upper,
1673 );
1674 self.positions.remove(position_key);
1675 }
1676 }
1677
1678 #[must_use]
1686 pub fn liquidity_utilization_rate(&self) -> f64 {
1687 const PRECISION: u32 = 1_000_000; let total_liquidity = self.get_total_liquidity();
1690 let active_liquidity = self.get_active_liquidity();
1691
1692 if total_liquidity == U256::ZERO {
1693 return 0.0;
1694 }
1695 let ratio = FullMath::mul_div(
1696 U256::from(active_liquidity),
1697 U256::from(PRECISION),
1698 total_liquidity,
1699 )
1700 .unwrap_or(U256::ZERO);
1701
1702 f64::from(ratio) / f64::from(PRECISION)
1703 }
1704
1705 fn validate_ticks(&self, tick_lower: i32, tick_upper: i32) -> anyhow::Result<()> {
1717 if tick_lower >= tick_upper {
1718 anyhow::bail!("Invalid tick range: {tick_lower} >= {tick_upper}")
1719 }
1720
1721 if tick_lower % self.pool.tick_spacing.unwrap() as i32 != 0
1722 || tick_upper % self.pool.tick_spacing.unwrap() as i32 != 0
1723 {
1724 anyhow::bail!(
1725 "Ticks {tick_lower} and {tick_upper} must be multiples of the tick spacing"
1726 )
1727 }
1728
1729 if tick_lower < PoolTick::MIN_TICK || tick_upper > PoolTick::MAX_TICK {
1730 anyhow::bail!("Invalid tick bounds for {tick_lower} and {tick_upper}");
1731 }
1732 Ok(())
1733 }
1734
1735 fn update_liquidity_analytics(&mut self) {
1737 self.analytics.liquidity_utilization_rate = self.liquidity_utilization_rate();
1738 }
1739
1740 #[must_use]
1749 pub fn get_active_liquidity(&self) -> u128 {
1750 self.tick_map.liquidity
1751 }
1752
1753 #[must_use]
1759 pub fn get_total_liquidity_from_active_positions(&self) -> u128 {
1760 self.positions
1761 .values()
1762 .filter(|position| {
1763 position.liquidity > 0
1764 && position.tick_lower <= self.state.current_tick
1765 && self.state.current_tick < position.tick_upper
1766 })
1767 .map(|position| position.liquidity)
1768 .sum()
1769 }
1770
1771 #[must_use]
1773 pub fn get_total_liquidity(&self) -> U256 {
1774 self.positions
1775 .values()
1776 .map(|position| U256::from(position.liquidity))
1777 .fold(U256::ZERO, |acc, liq| acc + liq)
1778 }
1779
1780 pub fn restore_from_snapshot(&mut self, snapshot: PoolSnapshot) -> anyhow::Result<()> {
1794 let liquidity = snapshot.state.liquidity;
1795
1796 self.state = snapshot.state;
1798
1799 self.analytics.total_amount0_deposited = snapshot.analytics.total_amount0_deposited;
1801 self.analytics.total_amount1_deposited = snapshot.analytics.total_amount1_deposited;
1802 self.analytics.total_amount0_collected = snapshot.analytics.total_amount0_collected;
1803 self.analytics.total_amount1_collected = snapshot.analytics.total_amount1_collected;
1804 self.analytics.total_swaps = snapshot.analytics.total_swaps;
1805 self.analytics.total_mints = snapshot.analytics.total_mints;
1806 self.analytics.total_burns = snapshot.analytics.total_burns;
1807 self.analytics.total_fee_collects = snapshot.analytics.total_fee_collects;
1808 self.analytics.total_flashes = snapshot.analytics.total_flashes;
1809
1810 self.positions.clear();
1812 for position in snapshot.positions {
1813 let key = PoolPosition::get_position_key(
1814 &position.owner,
1815 position.tick_lower,
1816 position.tick_upper,
1817 );
1818 self.positions.insert(key, position);
1819 }
1820
1821 self.tick_map = TickMap::new(
1823 self.pool
1824 .tick_spacing
1825 .expect("Pool tick spacing must be set"),
1826 );
1827
1828 for tick in snapshot.ticks {
1829 self.tick_map.restore_tick(tick);
1830 }
1831
1832 self.tick_map.liquidity = liquidity;
1834
1835 self.last_processed_event = Some(snapshot.block_position);
1837 self.last_processed_ts = Some(snapshot.ts_event);
1838
1839 self.is_initialized = true;
1841
1842 self.update_liquidity_analytics();
1844
1845 Ok(())
1846 }
1847
1848 #[must_use]
1853 pub fn get_active_tick_values(&self) -> Vec<i32> {
1854 self.tick_map
1855 .get_all_ticks()
1856 .iter()
1857 .filter(|(_, tick)| self.tick_map.is_tick_initialized(tick.value))
1858 .map(|(tick_value, _)| *tick_value)
1859 .collect()
1860 }
1861
1862 #[must_use]
1864 pub fn get_active_tick_count(&self) -> usize {
1865 self.tick_map.active_tick_count()
1866 }
1867
1868 #[must_use]
1873 pub fn get_tick(&self, tick: i32) -> Option<&PoolTick> {
1874 self.tick_map.get_tick(tick)
1875 }
1876
1877 #[must_use]
1882 pub fn get_current_tick(&self) -> i32 {
1883 self.state.current_tick
1884 }
1885
1886 #[must_use]
1894 pub fn get_total_tick_count(&self) -> usize {
1895 self.tick_map.total_tick_count()
1896 }
1897
1898 #[must_use]
1903 pub fn get_position(
1904 &self,
1905 owner: &Address,
1906 tick_lower: i32,
1907 tick_upper: i32,
1908 ) -> Option<&PoolPosition> {
1909 let position_key = PoolPosition::get_position_key(owner, tick_lower, tick_upper);
1910 self.positions.get(&position_key)
1911 }
1912
1913 #[must_use]
1923 pub fn get_active_positions(&self) -> Vec<&PoolPosition> {
1924 self.positions
1925 .values()
1926 .filter(|position| {
1927 let current_tick = self.get_current_tick();
1928 position.liquidity > 0
1929 && position.tick_lower <= current_tick
1930 && current_tick < position.tick_upper
1931 })
1932 .collect()
1933 }
1934
1935 #[must_use]
1944 pub fn get_all_positions(&self) -> Vec<&PoolPosition> {
1945 self.positions.values().collect()
1946 }
1947
1948 #[must_use]
1950 pub fn get_all_position_keys(&self) -> Vec<(Address, i32, i32)> {
1951 self.get_all_positions()
1952 .iter()
1953 .map(|position| (position.owner, position.tick_lower, position.tick_upper))
1954 .collect()
1955 }
1956
1957 pub fn extract_snapshot(&self) -> anyhow::Result<PoolSnapshot> {
1969 let positions: Vec<_> = self.positions.values().cloned().collect();
1970 let ticks: Vec<_> = self.tick_map.get_all_ticks().values().copied().collect();
1971
1972 let mut state = self.state.clone();
1973 state.liquidity = self.tick_map.liquidity;
1974
1975 let last_processed_event = self
1976 .last_processed_event
1977 .clone()
1978 .ok_or_else(|| anyhow::anyhow!("Cannot extract snapshot: no events processed yet"))?;
1979
1980 Ok(PoolSnapshot::new(
1981 self.pool.instrument_id,
1982 state,
1983 positions,
1984 ticks,
1985 self.analytics.clone(),
1986 last_processed_event,
1987 self.last_processed_ts.unwrap_or(self.pool.ts_init), self.last_processed_ts.unwrap_or(self.pool.ts_init), ))
1990 }
1991
1992 #[must_use]
1997 pub fn get_total_active_positions(&self) -> usize {
1998 self.positions
1999 .iter()
2000 .filter(|(_, position)| {
2001 let current_tick = self.get_current_tick();
2002 position.liquidity > 0
2003 && position.tick_lower <= current_tick
2004 && current_tick < position.tick_upper
2005 })
2006 .count()
2007 }
2008
2009 #[must_use]
2014 pub fn get_total_inactive_positions(&self) -> usize {
2015 self.positions.len() - self.get_total_active_positions()
2016 }
2017
2018 #[must_use]
2025 pub fn estimate_balance_of_token0(&self) -> U256 {
2026 let mut total_amount0 = U256::ZERO;
2027 let current_sqrt_price = self.state.price_sqrt_ratio_x96;
2028 let current_tick = self.state.current_tick;
2029
2030 for position in self.positions.values() {
2032 if position.liquidity == 0 || position.tick_upper <= current_tick {
2034 continue;
2035 }
2036
2037 if position.tick_lower > current_tick {
2038 let sqrt_ratio_a = get_sqrt_ratio_at_tick(position.tick_lower);
2040 let sqrt_ratio_b = get_sqrt_ratio_at_tick(position.tick_upper);
2041 let amount0 =
2042 get_amount0_delta(sqrt_ratio_a, sqrt_ratio_b, position.liquidity, true);
2043 total_amount0 += amount0;
2044 } else {
2045 let sqrt_ratio_upper = get_sqrt_ratio_at_tick(position.tick_upper);
2047 let amount0 = get_amount0_delta(
2048 current_sqrt_price,
2049 sqrt_ratio_upper,
2050 position.liquidity,
2051 true,
2052 );
2053 total_amount0 += amount0;
2054 }
2055 }
2056
2057 let fee_growth_0 = self.state.fee_growth_global_0;
2061 if fee_growth_0 > U256::ZERO {
2062 let active_liquidity = self.get_active_liquidity();
2065 if active_liquidity > 0 {
2066 if let Ok(total_fees_0) =
2069 FullMath::mul_div(fee_growth_0, U256::from(active_liquidity), Q128)
2070 {
2071 total_amount0 += total_fees_0;
2072 }
2073 }
2074 }
2075
2076 total_amount0 += self.state.protocol_fees_token0;
2078
2079 total_amount0
2080 }
2081
2082 #[must_use]
2089 pub fn estimate_balance_of_token1(&self) -> U256 {
2090 let mut total_amount1 = U256::ZERO;
2091 let current_sqrt_price = self.state.price_sqrt_ratio_x96;
2092 let current_tick = self.state.current_tick;
2093
2094 for position in self.positions.values() {
2096 if position.liquidity == 0 || position.tick_lower > current_tick {
2098 continue;
2099 }
2100
2101 if position.tick_upper <= current_tick {
2102 let sqrt_ratio_a = get_sqrt_ratio_at_tick(position.tick_lower);
2104 let sqrt_ratio_b = get_sqrt_ratio_at_tick(position.tick_upper);
2105 let amount1 =
2106 get_amount1_delta(sqrt_ratio_a, sqrt_ratio_b, position.liquidity, true);
2107 total_amount1 += amount1;
2108 } else {
2109 let sqrt_ratio_lower = get_sqrt_ratio_at_tick(position.tick_lower);
2111 let amount1 = get_amount1_delta(
2112 sqrt_ratio_lower,
2113 current_sqrt_price,
2114 position.liquidity,
2115 true,
2116 );
2117 total_amount1 += amount1;
2118 }
2119 }
2120
2121 let fee_growth_1 = self.state.fee_growth_global_1;
2123 if fee_growth_1 > U256::ZERO {
2124 let active_liquidity = self.get_active_liquidity();
2125 if active_liquidity > 0 {
2126 if let Ok(total_fees_1) =
2128 FullMath::mul_div(fee_growth_1, U256::from(active_liquidity), Q128)
2129 {
2130 total_amount1 += total_fees_1;
2131 }
2132 }
2133 }
2134
2135 total_amount1 += self.state.protocol_fees_token1;
2137
2138 total_amount1
2139 }
2140
2141 pub fn set_fee_growth_global(&mut self, fee_growth_global_0: U256, fee_growth_global_1: U256) {
2150 self.state.fee_growth_global_0 = fee_growth_global_0;
2151 self.state.fee_growth_global_1 = fee_growth_global_1;
2152 }
2153
2154 #[must_use]
2156 pub fn get_total_events(&self) -> u64 {
2157 self.analytics.total_swaps
2158 + self.analytics.total_mints
2159 + self.analytics.total_burns
2160 + self.analytics.total_fee_collects
2161 + self.analytics.total_flashes
2162 }
2163
2164 pub fn enable_reporting(&mut self, from_block: u64, total_blocks: u64, update_interval: u64) {
2169 self.reporter = Some(BlockchainSyncReporter::new(
2170 BlockchainSyncReportItems::PoolProfiling,
2171 from_block,
2172 total_blocks,
2173 update_interval,
2174 ));
2175 self.last_reported_block = from_block;
2176 }
2177
2178 pub fn finalize_reporting(&mut self) {
2183 if let Some(reporter) = &self.reporter {
2184 reporter.log_final_stats();
2185 }
2186 self.reporter = None;
2187 }
2188}
2189
2190fn initial_protocol_fee_basis_points(
2191 dex_type: DexType,
2192 pool_fee: Option<u32>,
2193) -> Option<(u32, u32)> {
2194 if dex_type != DexType::PancakeSwapV3 {
2195 return None;
2196 }
2197
2198 let fee_protocol = match pool_fee {
2199 Some(100) => 3_300,
2200 Some(500) => 3_400,
2201 _ => 3_200,
2202 };
2203
2204 Some((fee_protocol, fee_protocol))
2205}