1use ahash::AHashMap;
19use alloy_primitives::{Address, I256, U160, U256};
20use nautilus_core::UnixNanos;
21
22use crate::defi::{
23 DexType, PoolLiquidityUpdate, PoolSwap, SharedPool,
24 data::{
25 DexPoolData, PoolFeeCollect, PoolFeeProtocolCollect, PoolFeeProtocolUpdate,
26 PoolLiquidityUpdateType, block::BlockPosition, flash::PoolFlash,
27 },
28 pool_analysis::{
29 error::{
30 PoolEventKind, PoolEventLocation, PoolProfilerError, liquidity_error_with_location,
31 },
32 position::PoolPosition,
33 quote::SwapQuote,
34 size_estimator,
35 snapshot::{PROTOCOL_FEE_BASIS_POINTS_DENOMINATOR, PoolAnalytics, PoolSnapshot, PoolState},
36 swap_math::compute_swap_step,
37 },
38 reporting::{BlockchainSyncReportItems, BlockchainSyncReporter},
39 tick_map::{
40 TickMap,
41 full_math::{FullMath, Q128},
42 liquidity_math::{liquidity_math_add, try_liquidity_math_add},
43 sqrt_price_math::{get_amount0_delta, get_amount1_delta, get_amounts_for_liquidity},
44 tick::{CrossedTick, PoolTick},
45 tick_math::{
46 MAX_SQRT_RATIO, MIN_SQRT_RATIO, get_sqrt_ratio_at_tick, get_tick_at_sqrt_ratio,
47 },
48 },
49};
50
51#[derive(Debug, Clone)]
69#[cfg_attr(
70 feature = "python",
71 pyo3::pyclass(module = "nautilus_trader.model", from_py_object)
72)]
73#[cfg_attr(
74 feature = "python",
75 pyo3_stub_gen::derive::gen_stub_pyclass(module = "nautilus_trader.model")
76)]
77pub struct PoolProfiler {
78 pub pool: SharedPool,
80 positions: AHashMap<String, PoolPosition>,
82 pub tick_map: TickMap,
84 pub state: PoolState,
86 pub analytics: PoolAnalytics,
88 pub last_processed_event: Option<BlockPosition>,
90 pub last_processed_ts: Option<UnixNanos>,
92 pub is_initialized: bool,
94 reporter: Option<BlockchainSyncReporter>,
96 last_reported_block: u64,
98}
99
100impl PoolProfiler {
101 #[must_use]
107 pub fn new(pool: SharedPool) -> Self {
108 let tick_spacing = pool.tick_spacing.expect("Pool tick spacing must be set");
109 let mut state = PoolState::default();
110
111 if let Some((fee_protocol0, fee_protocol1)) =
112 initial_protocol_fee_basis_points(pool.dex.name, pool.fee)
113 {
114 state.set_protocol_fee_basis_points(fee_protocol0, fee_protocol1);
115 }
116
117 Self {
118 pool,
119 positions: AHashMap::new(),
120 tick_map: TickMap::new(tick_spacing),
121 state,
122 analytics: PoolAnalytics::default(),
123 last_processed_event: None,
124 last_processed_ts: None,
125 is_initialized: false,
126 reporter: None,
127 last_reported_block: 0,
128 }
129 }
130
131 pub fn initialize(&mut self, price_sqrt_ratio_x96: U160) -> Result<(), PoolProfilerError> {
139 if self.is_initialized {
140 return Err(PoolProfilerError::AlreadyInitialized {
141 instrument_id: self.pool.instrument_id,
142 pool_identifier: self.pool.pool_identifier,
143 });
144 }
145
146 let calculated_tick = get_tick_at_sqrt_ratio(price_sqrt_ratio_x96);
147
148 if let Some(initial_tick) = self.pool.initial_tick
149 && initial_tick != calculated_tick
150 {
151 return Err(PoolProfilerError::InitialTickMismatch {
152 instrument_id: self.pool.instrument_id,
153 pool_identifier: self.pool.pool_identifier,
154 initial_tick,
155 calculated_tick,
156 });
157 }
158
159 log::info!(
160 "Initializing pool profiler with tick {calculated_tick} and price sqrt ratio {price_sqrt_ratio_x96}"
161 );
162
163 self.state.current_tick = calculated_tick;
164 self.state.price_sqrt_ratio_x96 = price_sqrt_ratio_x96;
165 self.is_initialized = true;
166 Ok(())
167 }
168
169 pub fn check_if_initialized(&self, event_kind: PoolEventKind) -> Result<(), PoolProfilerError> {
176 if !self.is_initialized {
177 return Err(PoolProfilerError::NotInitialized {
178 instrument_id: self.pool.instrument_id,
179 pool_identifier: self.pool.pool_identifier,
180 event_kind,
181 });
182 }
183 Ok(())
184 }
185
186 fn event_location(
187 &self,
188 event_kind: PoolEventKind,
189 block: u64,
190 transaction_index: u32,
191 log_index: u32,
192 ) -> PoolEventLocation {
193 PoolEventLocation {
194 instrument_id: self.pool.instrument_id,
195 pool_identifier: self.pool.pool_identifier,
196 block,
197 transaction_index,
198 log_index,
199 event_kind,
200 }
201 }
202
203 pub fn process(&mut self, event: &DexPoolData) -> anyhow::Result<()> {
216 if self.check_if_already_processed(
217 event.block_number(),
218 event.transaction_index(),
219 event.log_index(),
220 ) {
221 return Ok(());
222 }
223
224 match event {
225 DexPoolData::Swap(swap) => self.process_swap(swap)?,
226 DexPoolData::LiquidityUpdate(update) => match update.kind {
227 PoolLiquidityUpdateType::Mint => self.process_mint(update)?,
228 PoolLiquidityUpdateType::Burn => self.process_burn(update)?,
229 },
230 DexPoolData::FeeCollect(collect) => self.process_collect(collect)?,
231 DexPoolData::FeeProtocolUpdate(update) => self.process_fee_protocol_update(update)?,
232 DexPoolData::FeeProtocolCollect(collect) => {
233 self.process_fee_protocol_collect(collect)?;
234 }
235 DexPoolData::Flash(flash) => self.process_flash(flash)?,
236 }
237
238 self.update_reporter_if_enabled(event.block_number());
239
240 Ok(())
241 }
242
243 fn check_if_already_processed(&self, block: u64, tx_idx: u32, log_idx: u32) -> bool {
245 if let Some(last_event) = &self.last_processed_event {
246 let should_skip = block < last_event.number
247 || (block == last_event.number && tx_idx < last_event.transaction_index)
248 || (block == last_event.number
249 && tx_idx == last_event.transaction_index
250 && log_idx <= last_event.log_index);
251
252 if should_skip {
253 log::debug!(
254 "Skipping already processed event at block {block} tx {tx_idx} log {log_idx}"
255 );
256 }
257 return should_skip;
258 }
259
260 false
261 }
262
263 fn update_reporter_if_enabled(&mut self, current_block: u64) {
265 if let Some(reporter) = &mut self.reporter {
267 let blocks_processed = current_block.saturating_sub(self.last_reported_block);
268
269 if blocks_processed > 0 {
270 reporter.update(blocks_processed as usize);
271 self.last_reported_block = current_block;
272
273 if reporter.should_log_progress(current_block, current_block) {
274 reporter.log_progress(current_block);
275 }
276 }
277 }
278 }
279
280 pub fn process_swap(&mut self, swap: &PoolSwap) -> anyhow::Result<()> {
300 self.check_if_initialized(PoolEventKind::Swap)?;
301
302 if self.check_if_already_processed(swap.block, swap.transaction_index, swap.log_index) {
303 return Ok(());
304 }
305
306 let zero_for_one = swap.amount0.is_positive();
307 let amount_specified = if zero_for_one {
308 swap.amount0
309 } else {
310 swap.amount1
311 };
312
313 let sqrt_price_limit_x96 = swap.sqrt_price_x96;
316 let location = self.event_location(
317 PoolEventKind::Swap,
318 swap.block,
319 swap.transaction_index,
320 swap.log_index,
321 );
322 let swap_quote = self
323 .simulate_swap_through_ticks(amount_specified, zero_for_one, sqrt_price_limit_x96, true)
324 .map_err(|e| Self::wrap_liquidity_error(e, location))?;
325
326 let tick_mismatch = swap.tick != swap_quote.tick_after;
327 let liquidity_mismatch = swap.liquidity != swap_quote.liquidity_after;
328 let sqrt_mismatch = swap.sqrt_price_x96 != swap_quote.sqrt_price_after_x96;
329 let structural_mismatch = tick_mismatch || liquidity_mismatch;
330 if structural_mismatch && !swap_quote.crossed_ticks.is_empty() {
331 log::warn!(
332 "Replay swap simulation diverged after crossing {} ticks on block {}; anchoring event state without simulated tick-cross mutations",
333 swap_quote.crossed_ticks.len(),
334 swap.block
335 );
336 self.apply_swap_quote_without_crossed_ticks(&swap_quote);
337 } else {
338 self.apply_swap_quote(&swap_quote);
339 }
340
341 if tick_mismatch {
343 log::warn!(
344 "Inconsistency in swap processing: Current tick mismatch: simulated {}, event {} on block {}",
345 swap_quote.tick_after,
346 swap.tick,
347 swap.block
348 );
349 }
350
351 if swap.tick != self.state.current_tick {
352 self.state.current_tick = swap.tick;
353 }
354
355 if liquidity_mismatch {
356 log::warn!(
357 "Inconsistency in swap processing: Active liquidity mismatch: simulated {}, event {} on block {}",
358 swap_quote.liquidity_after,
359 swap.liquidity,
360 swap.block
361 );
362 }
363
364 if swap.liquidity != self.tick_map.liquidity {
365 self.tick_map.liquidity = swap.liquidity;
366 }
367
368 if sqrt_mismatch {
369 log::warn!(
370 "Inconsistency in swap processing: Sqrt price mismatch: simulated {}, event {} on block {}",
371 swap_quote.sqrt_price_after_x96,
372 swap.sqrt_price_x96,
373 swap.block
374 );
375 }
376
377 if swap.sqrt_price_x96 != self.state.price_sqrt_ratio_x96 {
378 self.state.price_sqrt_ratio_x96 = swap.sqrt_price_x96;
379 }
380
381 self.last_processed_event = Some(
382 BlockPosition::new(
383 swap.block,
384 swap.transaction_hash.clone(),
385 swap.transaction_index,
386 swap.log_index,
387 )
388 .with_block_hash(swap.block_hash.clone()),
389 );
390 self.last_processed_ts = Some(swap.ts_event);
391 self.update_reporter_if_enabled(swap.block);
392 self.update_liquidity_analytics();
393
394 Ok(())
395 }
396
397 pub fn execute_swap(
410 &mut self,
411 sender: Address,
412 recipient: Address,
413 block: BlockPosition,
414 zero_for_one: bool,
415 amount_specified: I256,
416 sqrt_price_limit_x96: U160,
417 ) -> anyhow::Result<PoolSwap> {
418 self.check_if_initialized(PoolEventKind::Swap)?;
419
420 let swap_quote = self.simulate_swap_through_ticks(
421 amount_specified,
422 zero_for_one,
423 sqrt_price_limit_x96,
424 false,
425 )?;
426
427 self.apply_swap_quote(&swap_quote);
428
429 let swap_event = PoolSwap::new(
430 self.pool.chain.clone(),
431 self.pool.dex.clone(),
432 self.pool.instrument_id,
433 self.pool.pool_identifier,
434 block.number,
435 block.transaction_hash,
436 block.transaction_index,
437 block.log_index,
438 self.pool.ts_init, self.pool.ts_init, sender,
441 recipient,
442 swap_quote.amount0,
443 swap_quote.amount1,
444 self.state.price_sqrt_ratio_x96,
445 self.tick_map.liquidity,
446 self.state.current_tick,
447 );
448 Ok(swap_event)
449 }
450
451 pub fn simulate_swap_through_ticks(
487 &self,
488 amount_specified: I256,
489 zero_for_one: bool,
490 sqrt_price_limit_x96: U160,
491 traverse_empty_ranges: bool,
492 ) -> anyhow::Result<SwapQuote> {
493 let exact_input = amount_specified.is_positive();
494 let fee_tier = self.pool.fee.expect("Pool fee should be initialized");
495
496 let mut current_sqrt_price = self.state.price_sqrt_ratio_x96;
497 let mut current_tick = self.state.current_tick;
498 let mut current_active_liquidity = self.tick_map.liquidity;
499 let mut amount_specified_remaining = amount_specified;
500 let mut amount_calculated = I256::ZERO;
501 let mut protocol_fee = U256::ZERO;
502 let mut lp_fee = U256::ZERO;
503 let mut crossed_ticks = Vec::new();
504
505 let fee_protocol = self.state.uniswap_v3_fee_protocol(zero_for_one);
507 let fee_protocol_basis_points = self.state.fee_protocol_basis_points(zero_for_one);
508
509 let mut current_fee_growth_global = if zero_for_one {
511 self.state.fee_growth_global_0
512 } else {
513 self.state.fee_growth_global_1
514 };
515
516 while (amount_specified_remaining != I256::ZERO
518 || (traverse_empty_ranges && current_active_liquidity == 0))
519 && sqrt_price_limit_x96 != current_sqrt_price
520 {
521 let sqrt_price_start_x96 = current_sqrt_price;
522
523 let (mut tick_next, initialized) = self
524 .tick_map
525 .next_initialized_tick(current_tick, zero_for_one);
526
527 tick_next = tick_next.clamp(PoolTick::MIN_TICK, PoolTick::MAX_TICK);
529
530 let sqrt_price_next = get_sqrt_ratio_at_tick(tick_next);
532
533 let sqrt_price_target = if (zero_for_one && sqrt_price_next < sqrt_price_limit_x96)
535 || (!zero_for_one && sqrt_price_next > sqrt_price_limit_x96)
536 {
537 sqrt_price_limit_x96
538 } else {
539 sqrt_price_next
540 };
541 let swap_step_result = compute_swap_step(
542 current_sqrt_price,
543 sqrt_price_target,
544 current_active_liquidity,
545 amount_specified_remaining,
546 fee_tier,
547 )?;
548
549 current_sqrt_price = swap_step_result.sqrt_ratio_next_x96;
551
552 if exact_input {
554 amount_specified_remaining -= FullMath::truncate_to_i256(
556 swap_step_result.amount_in + swap_step_result.fee_amount,
557 );
558 amount_calculated -= FullMath::truncate_to_i256(swap_step_result.amount_out);
559 } else {
560 amount_specified_remaining +=
562 FullMath::truncate_to_i256(swap_step_result.amount_out);
563 amount_calculated += FullMath::truncate_to_i256(
564 swap_step_result.amount_in + swap_step_result.fee_amount,
565 );
566 }
567
568 let mut step_fee_amount = swap_step_result.fee_amount;
570
571 if fee_protocol > 0 || fee_protocol_basis_points.is_some() {
572 let protocol_fee_delta = Self::protocol_fee_delta(
573 swap_step_result.fee_amount,
574 fee_protocol,
575 fee_protocol_basis_points,
576 )?;
577 step_fee_amount -= protocol_fee_delta;
578 protocol_fee += protocol_fee_delta;
579 }
580
581 lp_fee += step_fee_amount;
583
584 if current_active_liquidity > 0 {
586 let fee_growth_delta =
587 FullMath::mul_div(step_fee_amount, Q128, U256::from(current_active_liquidity))?;
588 current_fee_growth_global += fee_growth_delta;
589 }
590
591 if swap_step_result.sqrt_ratio_next_x96 == sqrt_price_next {
593 if initialized {
597 crossed_ticks.push(CrossedTick::new(
598 tick_next,
599 zero_for_one,
600 if zero_for_one {
601 current_fee_growth_global
602 } else {
603 self.state.fee_growth_global_0
604 },
605 if zero_for_one {
606 self.state.fee_growth_global_1
607 } else {
608 current_fee_growth_global
609 },
610 ));
611
612 if let Some(tick_data) = self.tick_map.get_tick(tick_next) {
614 let liquidity_net = tick_data.liquidity_net;
615 current_active_liquidity = if zero_for_one {
616 try_liquidity_math_add(current_active_liquidity, -liquidity_net)?
617 } else {
618 try_liquidity_math_add(current_active_liquidity, liquidity_net)?
619 };
620 }
621 }
622
623 current_tick = if zero_for_one {
624 tick_next - 1
625 } else {
626 tick_next
627 };
628 } else if swap_step_result.sqrt_ratio_next_x96 != sqrt_price_start_x96 {
629 current_tick = get_tick_at_sqrt_ratio(current_sqrt_price);
632 }
633 }
634
635 let (amount0, amount1) = if zero_for_one == exact_input {
637 (
638 amount_specified - amount_specified_remaining,
639 amount_calculated,
640 )
641 } else {
642 (
643 amount_calculated,
644 amount_specified - amount_specified_remaining,
645 )
646 };
647
648 let quote = SwapQuote::new(
649 self.pool.instrument_id,
650 amount0,
651 amount1,
652 self.state.price_sqrt_ratio_x96,
653 current_sqrt_price,
654 self.state.current_tick,
655 current_tick,
656 current_active_liquidity,
657 current_fee_growth_global,
658 lp_fee,
659 protocol_fee,
660 crossed_ticks,
661 );
662 Ok(quote)
663 }
664
665 pub fn apply_swap_quote(&mut self, swap_quote: &SwapQuote) {
673 self.state.current_tick = swap_quote.tick_after;
674 self.state.price_sqrt_ratio_x96 = swap_quote.sqrt_price_after_x96;
675
676 self.apply_swap_quote_fee_state(swap_quote);
677
678 for crossed in &swap_quote.crossed_ticks {
679 let liquidity_net =
680 self.tick_map
681 .cross_tick(crossed.tick, crossed.fee_growth_0, crossed.fee_growth_1);
682
683 self.tick_map.liquidity = if crossed.zero_for_one {
684 liquidity_math_add(self.tick_map.liquidity, -liquidity_net)
685 } else {
686 liquidity_math_add(self.tick_map.liquidity, liquidity_net)
687 };
688 }
689 self.analytics.total_swaps += 1;
690
691 debug_assert_eq!(
692 self.tick_map.liquidity, swap_quote.liquidity_after,
693 "Liquidity mismatch in apply_swap_quote: computed={}, quote={}",
694 self.tick_map.liquidity, swap_quote.liquidity_after
695 );
696 }
697
698 fn apply_swap_quote_without_crossed_ticks(&mut self, swap_quote: &SwapQuote) {
699 self.state.current_tick = swap_quote.tick_after;
700 self.state.price_sqrt_ratio_x96 = swap_quote.sqrt_price_after_x96;
701 self.apply_swap_quote_fee_state(swap_quote);
702 self.analytics.total_swaps += 1;
703 }
704
705 fn apply_swap_quote_fee_state(&mut self, swap_quote: &SwapQuote) {
706 if swap_quote.zero_for_one() {
707 self.state.fee_growth_global_0 = swap_quote.fee_growth_global_after;
708 self.state.protocol_fees_token0 += swap_quote.protocol_fee;
709 } else {
710 self.state.fee_growth_global_1 = swap_quote.fee_growth_global_after;
711 self.state.protocol_fees_token1 += swap_quote.protocol_fee;
712 }
713 }
714
715 #[must_use]
719 pub fn wrap_liquidity_error(err: anyhow::Error, location: PoolEventLocation) -> anyhow::Error {
720 match err.downcast::<super::error::LiquidityMathError>() {
721 Ok(math_err) => anyhow::Error::from(liquidity_error_with_location(math_err, location)),
722 Err(other) => other,
723 }
724 }
725
726 pub fn quote_swap(
741 &self,
742 amount_specified: I256,
743 zero_for_one: bool,
744 sqrt_price_limit_x96: Option<U160>,
745 ) -> anyhow::Result<SwapQuote> {
746 self.check_if_initialized(PoolEventKind::Swap)?;
747
748 if amount_specified.is_zero() {
749 anyhow::bail!("Cannot quote swap with zero amount");
750 }
751
752 if let Some(price_limit) = sqrt_price_limit_x96 {
753 self.validate_price_limit(price_limit, zero_for_one)?;
754 }
755
756 let limit = sqrt_price_limit_x96.unwrap_or_else(|| {
757 if zero_for_one {
758 MIN_SQRT_RATIO + U160::from(1)
759 } else {
760 MAX_SQRT_RATIO - U160::from(1)
761 }
762 });
763
764 self.simulate_swap_through_ticks(amount_specified, zero_for_one, limit, false)
765 }
766
767 pub fn swap_exact_in(
772 &self,
773 amount_in: U256,
774 zero_for_one: bool,
775 sqrt_price_limit_x96: Option<U160>,
776 ) -> anyhow::Result<SwapQuote> {
777 let amount_specified = I256::from(amount_in);
779 let quote = self.quote_swap(amount_specified, zero_for_one, sqrt_price_limit_x96)?;
780
781 Ok(quote)
782 }
783
784 pub fn swap_exact_out(
790 &self,
791 amount_out: U256,
792 zero_for_one: bool,
793 sqrt_price_limit_x96: Option<U160>,
794 ) -> anyhow::Result<SwapQuote> {
795 let amount_specified = -I256::from(amount_out);
797 let quote = self.quote_swap(amount_specified, zero_for_one, sqrt_price_limit_x96)?;
798 quote.validate_exact_output(amount_out)?;
799
800 Ok(quote)
801 }
802
803 pub fn swap_to_lower_sqrt_price(
808 &self,
809 sqrt_price_limit_x96: U160,
810 ) -> anyhow::Result<SwapQuote> {
811 self.quote_swap(I256::MAX, true, Some(sqrt_price_limit_x96))
812 }
813
814 pub fn swap_to_higher_sqrt_price(
819 &self,
820 sqrt_price_limit_x96: U160,
821 ) -> anyhow::Result<SwapQuote> {
822 self.quote_swap(I256::MAX, false, Some(sqrt_price_limit_x96))
823 }
824
825 pub fn size_for_impact_bps(&self, impact_bps: u32, zero_for_one: bool) -> anyhow::Result<U256> {
840 let config = size_estimator::EstimationConfig::default();
841 size_estimator::size_for_impact_bps(self, impact_bps, zero_for_one, &config)
842 }
843
844 pub fn size_for_impact_bps_detailed(
858 &self,
859 impact_bps: u32,
860 zero_for_one: bool,
861 ) -> anyhow::Result<size_estimator::SizeForImpactResult> {
862 let config = size_estimator::EstimationConfig::default();
863 size_estimator::size_for_impact_bps_detailed(self, impact_bps, zero_for_one, &config)
864 }
865
866 fn validate_price_limit(
871 &self,
872 limit_price_sqrt: U160,
873 zero_for_one: bool,
874 ) -> anyhow::Result<()> {
875 if zero_for_one {
876 if limit_price_sqrt >= self.state.price_sqrt_ratio_x96 {
878 anyhow::bail!("Price limit must be less than current price for zero_for_one swaps");
879 }
880 } else {
881 if limit_price_sqrt <= self.state.price_sqrt_ratio_x96 {
883 anyhow::bail!(
884 "Price limit must be greater than current price for one_for_zero swaps"
885 );
886 }
887 }
888
889 Ok(())
890 }
891
892 pub fn process_mint(&mut self, update: &PoolLiquidityUpdate) -> anyhow::Result<()> {
904 self.check_if_initialized(PoolEventKind::Mint)?;
905
906 if self.check_if_already_processed(update.block, update.transaction_index, update.log_index)
907 {
908 return Ok(());
909 }
910
911 self.validate_ticks(update.tick_lower, update.tick_upper)?;
912 let location = self.event_location(
913 PoolEventKind::Mint,
914 update.block,
915 update.transaction_index,
916 update.log_index,
917 );
918 self.add_liquidity(
919 &update.owner,
920 update.tick_lower,
921 update.tick_upper,
922 update.position_liquidity,
923 update.amount0,
924 update.amount1,
925 )
926 .map_err(|e| Self::wrap_liquidity_error(e, location))?;
927
928 self.analytics.total_mints += 1;
929 self.last_processed_event = Some(
930 BlockPosition::new(
931 update.block,
932 update.transaction_hash.clone(),
933 update.transaction_index,
934 update.log_index,
935 )
936 .with_block_hash(update.block_hash.clone()),
937 );
938 self.last_processed_ts = Some(update.ts_event);
939 self.update_reporter_if_enabled(update.block);
940 self.update_liquidity_analytics();
941
942 Ok(())
943 }
944
945 fn add_liquidity(
950 &mut self,
951 owner: &Address,
952 tick_lower: i32,
953 tick_upper: i32,
954 liquidity: u128,
955 amount0: U256,
956 amount1: U256,
957 ) -> anyhow::Result<()> {
958 let liquidity_delta = i128::try_from(liquidity)
959 .map_err(|_| anyhow::anyhow!("Liquidity {liquidity} exceeds i128::MAX"))?;
960 self.update_position(
961 owner,
962 tick_lower,
963 tick_upper,
964 liquidity_delta,
965 amount0,
966 amount1,
967 )?;
968
969 self.analytics.total_amount0_deposited += amount0;
971 self.analytics.total_amount1_deposited += amount1;
972
973 Ok(())
974 }
975
976 pub fn execute_mint(
988 &mut self,
989 recipient: Address,
990 block: BlockPosition,
991 tick_lower: i32,
992 tick_upper: i32,
993 liquidity: u128,
994 ) -> anyhow::Result<PoolLiquidityUpdate> {
995 self.check_if_initialized(PoolEventKind::Mint)?;
996
997 self.validate_ticks(tick_lower, tick_upper)?;
998 let (amount0, amount1) = get_amounts_for_liquidity(
999 self.state.price_sqrt_ratio_x96,
1000 tick_lower,
1001 tick_upper,
1002 liquidity,
1003 true,
1004 );
1005 self.add_liquidity(
1006 &recipient, tick_lower, tick_upper, liquidity, amount0, amount1,
1007 )?;
1008
1009 self.analytics.total_mints += 1;
1010
1011 let event = PoolLiquidityUpdate::new(
1012 self.pool.chain.clone(),
1013 self.pool.dex.clone(),
1014 self.pool.instrument_id,
1015 self.pool.pool_identifier,
1016 PoolLiquidityUpdateType::Mint,
1017 block.number,
1018 block.transaction_hash,
1019 block.transaction_index,
1020 block.log_index,
1021 None,
1022 recipient,
1023 liquidity,
1024 amount0,
1025 amount1,
1026 tick_lower,
1027 tick_upper,
1028 self.pool.ts_init, self.pool.ts_init, );
1031
1032 Ok(event)
1033 }
1034
1035 pub fn process_burn(&mut self, update: &PoolLiquidityUpdate) -> anyhow::Result<()> {
1047 self.check_if_initialized(PoolEventKind::Burn)?;
1048
1049 if self.check_if_already_processed(update.block, update.transaction_index, update.log_index)
1050 {
1051 return Ok(());
1052 }
1053
1054 self.validate_ticks(update.tick_lower, update.tick_upper)?;
1055
1056 let liquidity_delta = i128::try_from(update.position_liquidity).map_err(|_| {
1058 anyhow::anyhow!("Liquidity {} exceeds i128::MAX", update.position_liquidity)
1059 })?;
1060 let location = self.event_location(
1061 PoolEventKind::Burn,
1062 update.block,
1063 update.transaction_index,
1064 update.log_index,
1065 );
1066
1067 self.update_position(
1068 &update.owner,
1069 update.tick_lower,
1070 update.tick_upper,
1071 -liquidity_delta,
1072 update.amount0,
1073 update.amount1,
1074 )
1075 .map_err(|e| Self::wrap_liquidity_error(e, location))?;
1076
1077 self.analytics.total_burns += 1;
1078 self.last_processed_event = Some(
1079 BlockPosition::new(
1080 update.block,
1081 update.transaction_hash.clone(),
1082 update.transaction_index,
1083 update.log_index,
1084 )
1085 .with_block_hash(update.block_hash.clone()),
1086 );
1087 self.last_processed_ts = Some(update.ts_event);
1088 self.update_reporter_if_enabled(update.block);
1089 self.update_liquidity_analytics();
1090
1091 Ok(())
1092 }
1093
1094 pub fn execute_burn(
1107 &mut self,
1108 recipient: Address,
1109 block: BlockPosition,
1110 tick_lower: i32,
1111 tick_upper: i32,
1112 liquidity: u128,
1113 ) -> anyhow::Result<PoolLiquidityUpdate> {
1114 self.check_if_initialized(PoolEventKind::Burn)?;
1115
1116 self.validate_ticks(tick_lower, tick_upper)?;
1117 let (amount0, amount1) = get_amounts_for_liquidity(
1118 self.state.price_sqrt_ratio_x96,
1119 tick_lower,
1120 tick_upper,
1121 liquidity,
1122 false,
1123 );
1124
1125 let liquidity_delta = i128::try_from(liquidity)
1127 .map_err(|_| anyhow::anyhow!("Liquidity {liquidity} exceeds i128::MAX"))?;
1128 self.update_position(
1129 &recipient,
1130 tick_lower,
1131 tick_upper,
1132 -liquidity_delta,
1133 amount0,
1134 amount1,
1135 )?;
1136
1137 self.analytics.total_burns += 1;
1138
1139 let event = PoolLiquidityUpdate::new(
1140 self.pool.chain.clone(),
1141 self.pool.dex.clone(),
1142 self.pool.instrument_id,
1143 self.pool.pool_identifier,
1144 PoolLiquidityUpdateType::Burn,
1145 block.number,
1146 block.transaction_hash,
1147 block.transaction_index,
1148 block.log_index,
1149 None,
1150 recipient,
1151 liquidity,
1152 amount0,
1153 amount1,
1154 tick_lower,
1155 tick_upper,
1156 self.pool.ts_init, self.pool.ts_init, );
1159
1160 Ok(event)
1161 }
1162
1163 pub fn process_collect(&mut self, collect: &PoolFeeCollect) -> anyhow::Result<()> {
1176 self.check_if_initialized(PoolEventKind::Collect)?;
1177
1178 if self.check_if_already_processed(
1179 collect.block,
1180 collect.transaction_index,
1181 collect.log_index,
1182 ) {
1183 return Ok(());
1184 }
1185 let position_key =
1186 PoolPosition::get_position_key(&collect.owner, collect.tick_lower, collect.tick_upper);
1187
1188 if let Some(position) = self.positions.get_mut(&position_key) {
1189 position.collect_fees(collect.amount0, collect.amount1);
1190 }
1191
1192 self.cleanup_position_if_empty(&position_key);
1194
1195 self.analytics.total_amount0_collected += U256::from(collect.amount0);
1196 self.analytics.total_amount1_collected += U256::from(collect.amount1);
1197
1198 self.analytics.total_fee_collects += 1;
1199 self.last_processed_event = Some(
1200 BlockPosition::new(
1201 collect.block,
1202 collect.transaction_hash.clone(),
1203 collect.transaction_index,
1204 collect.log_index,
1205 )
1206 .with_block_hash(collect.block_hash.clone()),
1207 );
1208 self.last_processed_ts = Some(collect.ts_event);
1209 self.update_reporter_if_enabled(collect.block);
1210 self.update_liquidity_analytics();
1211
1212 Ok(())
1213 }
1214
1215 pub fn process_fee_protocol_update(
1226 &mut self,
1227 update: &PoolFeeProtocolUpdate,
1228 ) -> anyhow::Result<()> {
1229 if self.check_if_already_processed(update.block, update.transaction_index, update.log_index)
1230 {
1231 return Ok(());
1232 }
1233
1234 if update.dex.name == DexType::PancakeSwapV3 {
1235 self.state
1236 .set_protocol_fee_basis_points(update.fee_protocol0_new, update.fee_protocol1_new);
1237 } else {
1238 let fee_protocol = update
1239 .uniswap_v3_packed()
1240 .ok_or_else(|| anyhow::anyhow!("invalid Uniswap V3 fee protocol update"))?;
1241 self.state.set_uniswap_v3_fee_protocol(fee_protocol);
1242 }
1243
1244 self.last_processed_event = Some(
1245 BlockPosition::new(
1246 update.block,
1247 update.transaction_hash.clone(),
1248 update.transaction_index,
1249 update.log_index,
1250 )
1251 .with_block_hash(update.block_hash.clone()),
1252 );
1253 self.last_processed_ts = Some(update.ts_event);
1254 self.update_reporter_if_enabled(update.block);
1255
1256 Ok(())
1257 }
1258
1259 pub fn process_fee_protocol_collect(
1271 &mut self,
1272 collect: &PoolFeeProtocolCollect,
1273 ) -> anyhow::Result<()> {
1274 if self.check_if_already_processed(
1275 collect.block,
1276 collect.transaction_index,
1277 collect.log_index,
1278 ) {
1279 return Ok(());
1280 }
1281
1282 self.state.protocol_fees_token0 = self
1283 .state
1284 .protocol_fees_token0
1285 .saturating_sub(U256::from(collect.amount0));
1286 self.state.protocol_fees_token1 = self
1287 .state
1288 .protocol_fees_token1
1289 .saturating_sub(U256::from(collect.amount1));
1290
1291 self.last_processed_event = Some(
1292 BlockPosition::new(
1293 collect.block,
1294 collect.transaction_hash.clone(),
1295 collect.transaction_index,
1296 collect.log_index,
1297 )
1298 .with_block_hash(collect.block_hash.clone()),
1299 );
1300 self.last_processed_ts = Some(collect.ts_event);
1301 self.update_reporter_if_enabled(collect.block);
1302
1303 Ok(())
1304 }
1305
1306 pub fn process_flash(&mut self, flash: &PoolFlash) -> anyhow::Result<()> {
1314 self.check_if_initialized(PoolEventKind::Flash)?;
1315
1316 if self.check_if_already_processed(flash.block, flash.transaction_index, flash.log_index) {
1317 return Ok(());
1318 }
1319
1320 self.update_flash_state(flash.paid0, flash.paid1)?;
1321
1322 self.analytics.total_flashes += 1;
1323 self.last_processed_event = Some(
1324 BlockPosition::new(
1325 flash.block,
1326 flash.transaction_hash.clone(),
1327 flash.transaction_index,
1328 flash.log_index,
1329 )
1330 .with_block_hash(flash.block_hash.clone()),
1331 );
1332 self.last_processed_ts = Some(flash.ts_event);
1333 self.update_reporter_if_enabled(flash.block);
1334 self.update_liquidity_analytics();
1335
1336 Ok(())
1337 }
1338
1339 pub fn execute_flash(
1352 &mut self,
1353 sender: Address,
1354 recipient: Address,
1355 block: BlockPosition,
1356 amount0: U256,
1357 amount1: U256,
1358 ) -> anyhow::Result<PoolFlash> {
1359 self.check_if_initialized(PoolEventKind::Flash)?;
1360
1361 let fee_tier = self.pool.fee.expect("Pool fee should be initialized");
1362
1363 let paid0 = if amount0 > U256::ZERO {
1365 FullMath::mul_div_rounding_up(amount0, U256::from(fee_tier), U256::from(1_000_000))?
1366 } else {
1367 U256::ZERO
1368 };
1369
1370 let paid1 = if amount1 > U256::ZERO {
1371 FullMath::mul_div_rounding_up(amount1, U256::from(fee_tier), U256::from(1_000_000))?
1372 } else {
1373 U256::ZERO
1374 };
1375
1376 self.update_flash_state(paid0, paid1)?;
1377 self.analytics.total_flashes += 1;
1378
1379 let flash_event = PoolFlash::new(
1380 self.pool.chain.clone(),
1381 self.pool.dex.clone(),
1382 self.pool.instrument_id,
1383 self.pool.pool_identifier,
1384 block.number,
1385 block.transaction_hash,
1386 block.transaction_index,
1387 block.log_index,
1388 self.pool.ts_init, self.pool.ts_init, sender,
1391 recipient,
1392 amount0,
1393 amount1,
1394 paid0,
1395 paid1,
1396 );
1397
1398 Ok(flash_event)
1399 }
1400
1401 fn update_flash_state(&mut self, paid0: U256, paid1: U256) -> anyhow::Result<()> {
1409 let liquidity = self.tick_map.liquidity;
1410 if liquidity == 0 {
1411 anyhow::bail!("No liquidity")
1412 }
1413
1414 let fee_protocol_0 = self.state.uniswap_v3_fee_protocol(true);
1415 let fee_protocol_1 = self.state.uniswap_v3_fee_protocol(false);
1416 let fee_protocol0_basis_points = self.state.fee_protocol_basis_points(true);
1417 let fee_protocol1_basis_points = self.state.fee_protocol_basis_points(false);
1418
1419 if paid0 > U256::ZERO {
1421 let protocol_fee_0 =
1422 Self::protocol_fee_delta(paid0, fee_protocol_0, fee_protocol0_basis_points)?;
1423
1424 if protocol_fee_0 > U256::ZERO {
1425 self.state.protocol_fees_token0 += protocol_fee_0;
1426 }
1427
1428 let lp_fee_0 = paid0 - protocol_fee_0;
1429 let delta = FullMath::mul_div(lp_fee_0, Q128, U256::from(liquidity))?;
1430 self.state.fee_growth_global_0 += delta;
1431 }
1432
1433 if paid1 > U256::ZERO {
1435 let protocol_fee_1 =
1436 Self::protocol_fee_delta(paid1, fee_protocol_1, fee_protocol1_basis_points)?;
1437
1438 if protocol_fee_1 > U256::ZERO {
1439 self.state.protocol_fees_token1 += protocol_fee_1;
1440 }
1441
1442 let lp_fee_1 = paid1 - protocol_fee_1;
1443 let delta = FullMath::mul_div(lp_fee_1, Q128, U256::from(liquidity))?;
1444 self.state.fee_growth_global_1 += delta;
1445 }
1446
1447 Ok(())
1448 }
1449
1450 fn protocol_fee_delta(
1451 fee_amount: U256,
1452 uniswap_v3_fee_protocol: u8,
1453 fee_protocol_basis_points: Option<u32>,
1454 ) -> anyhow::Result<U256> {
1455 if let Some(basis_points) = fee_protocol_basis_points {
1456 return FullMath::mul_div(
1457 fee_amount,
1458 U256::from(basis_points),
1459 U256::from(PROTOCOL_FEE_BASIS_POINTS_DENOMINATOR),
1460 );
1461 }
1462
1463 if uniswap_v3_fee_protocol > 0 {
1464 Ok(fee_amount / U256::from(uniswap_v3_fee_protocol))
1465 } else {
1466 Ok(U256::ZERO)
1467 }
1468 }
1469
1470 fn update_position(
1475 &mut self,
1476 owner: &Address,
1477 tick_lower: i32,
1478 tick_upper: i32,
1479 liquidity_delta: i128,
1480 amount0: U256,
1481 amount1: U256,
1482 ) -> anyhow::Result<()> {
1483 let current_tick = self.state.current_tick;
1484 let position_key = PoolPosition::get_position_key(owner, tick_lower, tick_upper);
1485 let position = self
1486 .positions
1487 .entry(position_key)
1488 .or_insert(PoolPosition::new(*owner, tick_lower, tick_upper, 0));
1489
1490 if liquidity_delta < 0 {
1492 let burn_amount = liquidity_delta.unsigned_abs();
1493 if position.liquidity < burn_amount {
1494 anyhow::bail!(
1495 "Position liquidity {} is less than the requested burn amount of {}",
1496 position.liquidity,
1497 burn_amount
1498 );
1499 }
1500 }
1501
1502 let new_active_liquidity = if tick_lower <= current_tick && current_tick < tick_upper {
1505 Some(try_liquidity_math_add(
1506 self.tick_map.liquidity,
1507 liquidity_delta,
1508 )?)
1509 } else {
1510 None
1511 };
1512
1513 for tick_value in [tick_lower, tick_upper] {
1514 let liquidity_gross = self
1515 .tick_map
1516 .get_tick(tick_value)
1517 .map_or(0, |tick| tick.liquidity_gross);
1518 try_liquidity_math_add(liquidity_gross, liquidity_delta)?;
1519 }
1520
1521 let flipped_lower = self.tick_map.update(
1523 tick_lower,
1524 current_tick,
1525 liquidity_delta,
1526 false,
1527 self.state.fee_growth_global_0,
1528 self.state.fee_growth_global_1,
1529 );
1530 let flipped_upper = self.tick_map.update(
1531 tick_upper,
1532 current_tick,
1533 liquidity_delta,
1534 true,
1535 self.state.fee_growth_global_0,
1536 self.state.fee_growth_global_1,
1537 );
1538
1539 let (fee_growth_inside_0, fee_growth_inside_1) = self.tick_map.get_fee_growth_inside(
1540 tick_lower,
1541 tick_upper,
1542 current_tick,
1543 self.state.fee_growth_global_0,
1544 self.state.fee_growth_global_1,
1545 );
1546 position.update_liquidity(liquidity_delta);
1547 position.update_fees(fee_growth_inside_0, fee_growth_inside_1);
1548 position.update_amounts(liquidity_delta, amount0, amount1);
1549
1550 if let Some(active_liquidity) = new_active_liquidity {
1551 self.tick_map.liquidity = active_liquidity;
1552 }
1553
1554 if liquidity_delta < 0 && flipped_lower {
1556 self.tick_map.clear(tick_lower);
1557 }
1558
1559 if liquidity_delta < 0 && flipped_upper {
1560 self.tick_map.clear(tick_upper);
1561 }
1562
1563 Ok(())
1564 }
1565
1566 fn cleanup_position_if_empty(&mut self, position_key: &str) {
1570 if let Some(position) = self.positions.get(position_key)
1571 && position.is_empty()
1572 {
1573 log::debug!(
1574 "CLEANING UP EMPTY POSITION: owner={}, ticks=[{}, {}]",
1575 position.owner,
1576 position.tick_lower,
1577 position.tick_upper,
1578 );
1579 self.positions.remove(position_key);
1580 }
1581 }
1582
1583 #[must_use]
1588 pub fn liquidity_utilization_rate(&self) -> f64 {
1589 const PRECISION: u32 = 1_000_000; let total_liquidity = self.get_total_liquidity();
1592 let active_liquidity = self.get_active_liquidity();
1593
1594 if total_liquidity == U256::ZERO {
1595 return 0.0;
1596 }
1597 let ratio = FullMath::mul_div(
1598 U256::from(active_liquidity),
1599 U256::from(PRECISION),
1600 total_liquidity,
1601 )
1602 .unwrap_or(U256::ZERO);
1603
1604 ratio.to::<u64>() as f64 / f64::from(PRECISION)
1607 }
1608
1609 fn validate_ticks(&self, tick_lower: i32, tick_upper: i32) -> anyhow::Result<()> {
1621 if tick_lower >= tick_upper {
1622 anyhow::bail!("Invalid tick range: {tick_lower} >= {tick_upper}")
1623 }
1624
1625 if tick_lower % self.pool.tick_spacing.unwrap() as i32 != 0
1626 || tick_upper % self.pool.tick_spacing.unwrap() as i32 != 0
1627 {
1628 anyhow::bail!(
1629 "Ticks {tick_lower} and {tick_upper} must be multiples of the tick spacing"
1630 )
1631 }
1632
1633 if tick_lower < PoolTick::MIN_TICK || tick_upper > PoolTick::MAX_TICK {
1634 anyhow::bail!("Invalid tick bounds for {tick_lower} and {tick_upper}");
1635 }
1636 Ok(())
1637 }
1638
1639 fn update_liquidity_analytics(&mut self) {
1641 self.analytics.liquidity_utilization_rate = self.liquidity_utilization_rate();
1642 }
1643
1644 #[must_use]
1653 pub fn get_active_liquidity(&self) -> u128 {
1654 self.tick_map.liquidity
1655 }
1656
1657 #[must_use]
1663 pub fn get_total_liquidity_from_active_positions(&self) -> u128 {
1664 self.positions
1665 .values()
1666 .filter(|position| {
1667 position.liquidity > 0
1668 && position.tick_lower <= self.state.current_tick
1669 && self.state.current_tick < position.tick_upper
1670 })
1671 .map(|position| position.liquidity)
1672 .sum()
1673 }
1674
1675 #[must_use]
1677 pub fn get_total_liquidity(&self) -> U256 {
1678 self.positions
1679 .values()
1680 .map(|position| U256::from(position.liquidity))
1681 .fold(U256::ZERO, |acc, liq| acc + liq)
1682 }
1683
1684 pub fn restore_from_snapshot(&mut self, snapshot: PoolSnapshot) -> anyhow::Result<()> {
1698 let liquidity = snapshot.state.liquidity;
1699
1700 self.state = snapshot.state;
1702
1703 self.analytics.total_amount0_deposited = snapshot.analytics.total_amount0_deposited;
1705 self.analytics.total_amount1_deposited = snapshot.analytics.total_amount1_deposited;
1706 self.analytics.total_amount0_collected = snapshot.analytics.total_amount0_collected;
1707 self.analytics.total_amount1_collected = snapshot.analytics.total_amount1_collected;
1708 self.analytics.total_swaps = snapshot.analytics.total_swaps;
1709 self.analytics.total_mints = snapshot.analytics.total_mints;
1710 self.analytics.total_burns = snapshot.analytics.total_burns;
1711 self.analytics.total_fee_collects = snapshot.analytics.total_fee_collects;
1712 self.analytics.total_flashes = snapshot.analytics.total_flashes;
1713
1714 self.positions.clear();
1716 for position in snapshot.positions {
1717 let key = PoolPosition::get_position_key(
1718 &position.owner,
1719 position.tick_lower,
1720 position.tick_upper,
1721 );
1722 self.positions.insert(key, position);
1723 }
1724
1725 self.tick_map = TickMap::new(
1727 self.pool
1728 .tick_spacing
1729 .expect("Pool tick spacing must be set"),
1730 );
1731
1732 for tick in snapshot.ticks {
1733 self.tick_map.restore_tick(tick);
1734 }
1735
1736 self.tick_map.liquidity = liquidity;
1738
1739 self.last_processed_event = Some(snapshot.block_position);
1741 self.last_processed_ts = Some(snapshot.ts_event);
1742
1743 self.is_initialized = true;
1745
1746 self.update_liquidity_analytics();
1748
1749 Ok(())
1750 }
1751
1752 #[must_use]
1757 pub fn get_active_tick_values(&self) -> Vec<i32> {
1758 self.tick_map
1759 .get_all_ticks()
1760 .iter()
1761 .filter(|(_, tick)| self.tick_map.is_tick_initialized(tick.value))
1762 .map(|(tick_value, _)| *tick_value)
1763 .collect()
1764 }
1765
1766 #[must_use]
1768 pub fn get_active_tick_count(&self) -> usize {
1769 self.tick_map.active_tick_count()
1770 }
1771
1772 #[must_use]
1777 pub fn get_tick(&self, tick: i32) -> Option<&PoolTick> {
1778 self.tick_map.get_tick(tick)
1779 }
1780
1781 #[must_use]
1786 pub fn get_current_tick(&self) -> i32 {
1787 self.state.current_tick
1788 }
1789
1790 #[must_use]
1798 pub fn get_total_tick_count(&self) -> usize {
1799 self.tick_map.total_tick_count()
1800 }
1801
1802 #[must_use]
1807 pub fn get_position(
1808 &self,
1809 owner: &Address,
1810 tick_lower: i32,
1811 tick_upper: i32,
1812 ) -> Option<&PoolPosition> {
1813 let position_key = PoolPosition::get_position_key(owner, tick_lower, tick_upper);
1814 self.positions.get(&position_key)
1815 }
1816
1817 #[must_use]
1827 pub fn get_active_positions(&self) -> Vec<&PoolPosition> {
1828 self.positions
1829 .values()
1830 .filter(|position| {
1831 let current_tick = self.get_current_tick();
1832 position.liquidity > 0
1833 && position.tick_lower <= current_tick
1834 && current_tick < position.tick_upper
1835 })
1836 .collect()
1837 }
1838
1839 #[must_use]
1848 pub fn get_all_positions(&self) -> Vec<&PoolPosition> {
1849 self.positions.values().collect()
1850 }
1851
1852 #[must_use]
1854 pub fn get_all_position_keys(&self) -> Vec<(Address, i32, i32)> {
1855 self.get_all_positions()
1856 .iter()
1857 .map(|position| (position.owner, position.tick_lower, position.tick_upper))
1858 .collect()
1859 }
1860
1861 pub fn extract_snapshot(&self) -> anyhow::Result<PoolSnapshot> {
1873 let positions: Vec<_> = self.positions.values().cloned().collect();
1874 let ticks: Vec<_> = self.tick_map.get_all_ticks().values().copied().collect();
1875
1876 let mut state = self.state.clone();
1877 state.liquidity = self.tick_map.liquidity;
1878
1879 let last_processed_event = self
1880 .last_processed_event
1881 .clone()
1882 .ok_or_else(|| anyhow::anyhow!("Cannot extract snapshot: no events processed yet"))?;
1883
1884 Ok(PoolSnapshot::new(
1885 self.pool.instrument_id,
1886 state,
1887 positions,
1888 ticks,
1889 self.analytics.clone(),
1890 last_processed_event,
1891 self.last_processed_ts.unwrap_or(self.pool.ts_init), self.last_processed_ts.unwrap_or(self.pool.ts_init), ))
1894 }
1895
1896 #[must_use]
1901 pub fn get_total_active_positions(&self) -> usize {
1902 self.positions
1903 .iter()
1904 .filter(|(_, position)| {
1905 let current_tick = self.get_current_tick();
1906 position.liquidity > 0
1907 && position.tick_lower <= current_tick
1908 && current_tick < position.tick_upper
1909 })
1910 .count()
1911 }
1912
1913 #[must_use]
1918 pub fn get_total_inactive_positions(&self) -> usize {
1919 self.positions.len() - self.get_total_active_positions()
1920 }
1921
1922 #[must_use]
1929 pub fn estimate_balance_of_token0(&self) -> U256 {
1930 let mut total_amount0 = U256::ZERO;
1931 let current_sqrt_price = self.state.price_sqrt_ratio_x96;
1932 let current_tick = self.state.current_tick;
1933 let mut total_fees_0_collected: u128 = 0;
1934
1935 for position in self.positions.values() {
1937 if position.liquidity > 0 {
1938 if position.tick_upper <= current_tick {
1939 continue;
1941 } else if position.tick_lower > current_tick {
1942 let sqrt_ratio_a = get_sqrt_ratio_at_tick(position.tick_lower);
1944 let sqrt_ratio_b = get_sqrt_ratio_at_tick(position.tick_upper);
1945 let amount0 =
1946 get_amount0_delta(sqrt_ratio_a, sqrt_ratio_b, position.liquidity, true);
1947 total_amount0 += amount0;
1948 } else {
1949 let sqrt_ratio_upper = get_sqrt_ratio_at_tick(position.tick_upper);
1951 let amount0 = get_amount0_delta(
1952 current_sqrt_price,
1953 sqrt_ratio_upper,
1954 position.liquidity,
1955 true,
1956 );
1957 total_amount0 += amount0;
1958 }
1959 }
1960
1961 total_fees_0_collected += position.total_amount0_collected;
1962 }
1963
1964 let fee_growth_0 = self.state.fee_growth_global_0;
1968 if fee_growth_0 > U256::ZERO {
1969 let active_liquidity = self.get_active_liquidity();
1972 if active_liquidity > 0 {
1973 if let Ok(total_fees_0) =
1976 FullMath::mul_div(fee_growth_0, U256::from(active_liquidity), Q128)
1977 {
1978 total_amount0 += total_fees_0;
1979 }
1980 }
1981 }
1982
1983 let total_fees_0_left = fee_growth_0 - U256::from(total_fees_0_collected);
1984
1985 total_amount0 += self.state.protocol_fees_token0;
1987
1988 total_amount0 + total_fees_0_left
1989 }
1990
1991 #[must_use]
1998 pub fn estimate_balance_of_token1(&self) -> U256 {
1999 let mut total_amount1 = U256::ZERO;
2000 let current_sqrt_price = self.state.price_sqrt_ratio_x96;
2001 let current_tick = self.state.current_tick;
2002 let mut total_fees_1_collected: u128 = 0;
2003
2004 for position in self.positions.values() {
2006 if position.liquidity > 0 {
2007 if position.tick_lower > current_tick {
2008 continue;
2010 } else if position.tick_upper <= current_tick {
2011 let sqrt_ratio_a = get_sqrt_ratio_at_tick(position.tick_lower);
2013 let sqrt_ratio_b = get_sqrt_ratio_at_tick(position.tick_upper);
2014 let amount1 =
2015 get_amount1_delta(sqrt_ratio_a, sqrt_ratio_b, position.liquidity, true);
2016 total_amount1 += amount1;
2017 } else {
2018 let sqrt_ratio_lower = get_sqrt_ratio_at_tick(position.tick_lower);
2020 let amount1 = get_amount1_delta(
2021 sqrt_ratio_lower,
2022 current_sqrt_price,
2023 position.liquidity,
2024 true,
2025 );
2026 total_amount1 += amount1;
2027 }
2028 }
2029
2030 total_fees_1_collected += position.total_amount1_collected;
2032 }
2033
2034 let fee_growth_1 = self.state.fee_growth_global_1;
2036 if fee_growth_1 > U256::ZERO {
2037 let active_liquidity = self.get_active_liquidity();
2038 if active_liquidity > 0 {
2039 if let Ok(total_fees_1) =
2041 FullMath::mul_div(fee_growth_1, U256::from(active_liquidity), Q128)
2042 {
2043 total_amount1 += total_fees_1;
2044 }
2045 }
2046 }
2047
2048 let total_fees_1_left = fee_growth_1 - U256::from(total_fees_1_collected);
2049
2050 total_amount1 += self.state.protocol_fees_token1;
2052
2053 total_amount1 + total_fees_1_left
2054 }
2055
2056 pub fn set_fee_growth_global(&mut self, fee_growth_global_0: U256, fee_growth_global_1: U256) {
2065 self.state.fee_growth_global_0 = fee_growth_global_0;
2066 self.state.fee_growth_global_1 = fee_growth_global_1;
2067 }
2068
2069 #[must_use]
2071 pub fn get_total_events(&self) -> u64 {
2072 self.analytics.total_swaps
2073 + self.analytics.total_mints
2074 + self.analytics.total_burns
2075 + self.analytics.total_fee_collects
2076 + self.analytics.total_flashes
2077 }
2078
2079 pub fn enable_reporting(&mut self, from_block: u64, total_blocks: u64, update_interval: u64) {
2084 self.reporter = Some(BlockchainSyncReporter::new(
2085 BlockchainSyncReportItems::PoolProfiling,
2086 from_block,
2087 total_blocks,
2088 update_interval,
2089 ));
2090 self.last_reported_block = from_block;
2091 }
2092
2093 pub fn finalize_reporting(&mut self) {
2098 if let Some(reporter) = &self.reporter {
2099 reporter.log_final_stats();
2100 }
2101 self.reporter = None;
2102 }
2103}
2104
2105fn initial_protocol_fee_basis_points(
2106 dex_type: DexType,
2107 pool_fee: Option<u32>,
2108) -> Option<(u32, u32)> {
2109 if dex_type != DexType::PancakeSwapV3 {
2110 return None;
2111 }
2112
2113 let fee_protocol = match pool_fee {
2114 Some(100) => 3_300,
2115 Some(500) => 3_400,
2116 _ => 3_200,
2117 };
2118
2119 Some((fee_protocol, fee_protocol))
2120}