Skip to main content

nautilus_model/defi/pool_analysis/
profiler.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Pool profiling utilities for analyzing DeFi pool event data.
17
18use 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/// A DeFi pool state tracker and event processor for UniswapV3-style AMM pools.
54///
55/// The `PoolProfiler` provides complete pool state management including:
56/// - Liquidity position tracking and management.
57/// - Tick crossing and price movement simulation.
58/// - Fee accumulation and distribution tracking.
59/// - Protocol fee calculation.
60/// - Pool state validation and maintenance.
61///
62/// This profiler can both process historical events and execute new operations,
63/// making it suitable for both backtesting and simulation scenarios.
64///
65/// # Usage
66///
67/// Create a new profiler with a pool definition, initialize it with a starting price,
68/// then either process historical events or execute new pool operations to simulate
69/// trading activity and analyze pool behavior.
70#[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    /// Pool definition.
81    pub pool: SharedPool,
82    /// Position tracking by position key (`owner:tick_lower:tick_upper`).
83    positions: AHashMap<String, PoolPosition>,
84    /// Tick map managing liquidity distribution across price ranges.
85    pub tick_map: TickMap,
86    /// Global pool state including current price, tick, and cumulative flows with fees.
87    pub state: PoolState,
88    /// Analytics counters tracking pool operations and performance metrics.
89    pub analytics: PoolAnalytics,
90    /// The block position of the last processed event.
91    pub last_processed_event: Option<BlockPosition>,
92    /// The event timestamp of the last processed event.
93    pub last_processed_ts: Option<UnixNanos>,
94    /// Flag indicating whether the pool has been initialized with a starting price.
95    pub is_initialized: bool,
96    /// Optional progress reporter for tracking event processing.
97    reporter: Option<BlockchainSyncReporter>,
98    /// The last block number that was reported (used for progress tracking).
99    last_reported_block: u64,
100}
101
102impl PoolProfiler {
103    /// Creates a new [`PoolProfiler`] instance for tracking pool state and events.
104    ///
105    /// # Panics
106    ///
107    /// Panics if the pool's tick spacing is not set.
108    #[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    /// Initializes the pool with a starting price and activates the profiler.
134    ///
135    /// # Errors
136    ///
137    /// Returns an error if the profiler has already been initialized, or if the pool config carries
138    /// an `initial_tick` that disagrees with the tick derived from `price_sqrt_ratio_x96`.
139    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    /// Returns an error if the pool has not been initialized.
171    ///
172    /// # Errors
173    ///
174    /// Returns an error if [`Self::initialize`] or [`Self::restore_from_snapshot`] has not been
175    /// called yet.
176    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    /// Processes a historical pool event and updates internal state.
205    ///
206    /// Handles all types of pool events (swaps, mints, burns, fee collections),
207    /// and updates the profiler's internal state accordingly. This is the main
208    /// entry point for processing historical blockchain events.
209    ///
210    /// # Errors
211    ///
212    /// Returns an error if:
213    /// - Pool is not initialized.
214    /// - Event contains invalid data (tick ranges, amounts).
215    /// - Mathematical operations overflow.
216    /// - Swap replay cannot maintain the initialized tick partition.
217    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    // Checks if we need to skip events at or before the last processed event to prevent double-processing.
246    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    /// Auto-updates reporter if it's enabled.
266    fn update_reporter_if_enabled(&mut self, current_block: u64) {
267        // Auto-update reporter if enabled
268        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    /// Processes a historical swap event from blockchain data.
283    ///
284    /// Replays the swap by simulating it through [`Self::simulate_swap_through_ticks`],
285    /// then verifies the simulation results against the actual event data. Tick and
286    /// liquidity mismatches are corrected only when no simulated crossings occurred
287    /// and anchoring to the event tick preserves the initialized tick partition.
288    /// Otherwise, replay returns an error without changing the profiler. The event price
289    /// serves as the simulation limit, so wrong-side or out-of-bounds event prices also
290    /// return an error without changing the profiler. After a successful simulation, sqrt
291    /// price mismatches are corrected to match the event.
292    ///
293    /// # Use Case
294    ///
295    /// Historical event processing when rebuilding pool state from blockchain events.
296    ///
297    /// # Errors
298    ///
299    /// Returns an error if:
300    /// - Pool initialization checks fail.
301    /// - Swap simulation fails (see [`Self::simulate_swap_through_ticks`] errors).
302    /// - A structural mismatch involves simulated crossings or anchoring across an initialized
303    ///   tick.
304    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        // For price limit use the final sqrt price from swap, which is a
324        // good proxy to price limit
325        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        // Verify simulation against event data - correct with event values if mismatch detected
371        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    /// Executes a new simulated swap and returns the resulting event.
427    ///
428    /// This is the public API for forward simulation of swap operations. It delegates
429    /// the core swap mathematics to [`Self::simulate_swap_through_ticks`], then wraps
430    /// the results in a [`PoolSwap`] event structure with full metadata.
431    ///
432    /// # Errors
433    ///
434    /// Returns an error if:
435    /// - Pool metadata is missing or invalid.
436    /// - The price limit is violated.
437    /// - Fee or liquidity calculations overflow.
438    ///
439    /// Propagates failures from [`Self::simulate_swap_through_ticks`].
440    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, // ts_event (simulated; pool init time)
470            self.pool.ts_init, // ts_init
471            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    /// Core **read-only** swap simulation engine implementing `UniswapV3` mathematics.
483    ///
484    /// This method performs a complete swap simulation without modifying pool state,
485    /// working entirely on stack-allocated local copies of state variables. It returns
486    /// a [`SwapQuote`] containing all swap results and profiling data,
487    /// including a complete audit trail of crossed ticks.
488    ///
489    /// # Algorithm Overview
490    ///
491    /// 1. **Iterative price curve traversal**: Walks through liquidity ranges until
492    ///    the input/output amount is exhausted or the price limit is reached
493    /// 2. **Tick crossing tracking**: When crossing initialized tick boundaries, records
494    ///    the crossing in `crossed_ticks` vector with complete state snapshot (tick, direction, fee growth)
495    /// 3. **Local liquidity updates**: Tracks liquidity changes in local variables by reading
496    ///    `liquidity_net` from tick map (read-only, no mutations)
497    /// 4. **Fee calculation**: Splits fees between LPs and protocol, accumulates in local variables
498    /// 5. **Quote assembly**: Returns [`SwapQuote`] with amounts, prices, fees, and crossed tick data
499    ///
500    /// When `traverse_empty_ranges` is set, the walk continues across zero-liquidity ranges
501    /// to `sqrt_price_limit_x96` even after the amount is exhausted. This reproduces a
502    /// historical swap whose recorded amount (the on-chain consumed amount) runs out at the
503    /// last liquid tick before an empty range to the boundary; forward simulation leaves it
504    /// unset so the swap stops where the amount is spent, matching `UniswapV3`. With an exact
505    /// input, it also charges any input left at `sqrt_price_limit_x96` as fee, because the chain
506    /// consumed the recorded amount in full.
507    ///
508    /// # Errors
509    ///
510    /// Returns an error if:
511    /// - The price limit violates the swap direction or protocol bounds.
512    /// - Fee growth arithmetic overflows when scaling by liquidity.
513    /// - Swap step calculations fail.
514    /// - A swap iteration makes no progress: price, tick, and remaining amount are all unchanged.
515    ///
516    /// # Panics
517    ///
518    /// Panics if the pool fee has not been initialized.
519    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        // Swapping cache variables
541        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        // Track current fee growth during swap
545        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        // A replayed swap's recorded input was consumed in full, so input left at the event
552        // price is charged as fee even when that price equals the starting price.
553        let replay_exact_input = traverse_empty_ranges && exact_input;
554
555        // The replay clause keeps crossing empty ranges to the limit after the amount runs out
556        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            // Make sure we do not overshoot MIN/MAX tick
569            tick_next = tick_next.clamp(PoolTick::MIN_TICK, PoolTick::MAX_TICK);
570
571            // Get the price for the next tick
572            let sqrt_price_next = get_sqrt_ratio_at_tick(tick_next);
573
574            // Compute values to swap to the target tick, price limit, or point where input/output amount is exhausted
575            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            // Matches the on-chain step that ran out of input before its target; a step ending on
592            // a tick boundary keeps its fee because the chain crosses first.
593            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            // Update current price to the new price after this swap step (BEFORE amount updates, matching Solidity)
602            current_sqrt_price = swap_step_result.sqrt_ratio_next_x96;
603
604            // Update amounts based on swap direction and type
605            if exact_input {
606                // For exact input swaps: subtract input amount and fees from remaining, subtract output from calculated
607                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                // For exact output swaps: add output to remaining, add input and fees to calculated
613                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            // Calculate protocol fee if enabled
621            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            // Accumulate LP fee (protocol fee is already deducted if it exists).
634            lp_fee += step_fee_amount;
635
636            // Update global fee tracker
637            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            // Shift tick if we reached the next price
644            if swap_step_result.sqrt_ratio_next_x96 == sqrt_price_next {
645                // We have swapped all the way to the boundary of the next tick.
646                // Time to handle crossing into the next tick, which may change liquidity.
647                // If the tick is initialized, run the tick transition logic (liquidity changes, fee accumulators, etc.).
648                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                    // Update local liquidity tracking when crossing ticks
665                    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                // The price moved during this swap step, but didn't reach a tick boundary.
682                // So, update the tick to match the new price.
683                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        // Calculate final amounts
695        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    /// Applies a swap quote to the pool state (mutations only, no simulation).
725    ///
726    /// # Panics
727    ///
728    /// Panics if applying a tick-crossing liquidity delta overflows or underflows,
729    /// which indicates internal tick-map inconsistency rather than a recoverable
730    /// replay error.
731    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    /// Wraps a low-level [`LiquidityMathError`](super::error::LiquidityMathError) into a
772    /// [`PoolProfilerError`] carrying the supplied event location, leaving non-liquidity
773    /// errors untouched.
774    #[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    // panics-doc-ok (transitive via simulate_swap_through_ticks)
783    /// Returns a swap quote without modifying pool state.
784    ///
785    /// This method simulates a swap and provides detailed profiling metrics including:
786    /// - Amounts of tokens that would be exchanged
787    /// - Price before and after the swap
788    /// - Fee breakdown (LP fees and protocol fees)
789    /// - List of crossed ticks with state snapshots
790    ///
791    /// # Errors
792    ///
793    /// Returns an error if:
794    /// - The profiler is not initialized or the amount is zero.
795    /// - The price limit violates the swap direction or protocol bounds.
796    /// - Fee growth arithmetic overflows when scaling by liquidity.
797    /// - Swap step calculations fail.
798    /// - A swap iteration makes no progress.
799    ///
800    /// # Panics
801    ///
802    /// Panics if the pool fee has not been initialized.
803    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    /// Simulates an exact input swap (know input amount, calculate output amount).
831    ///
832    /// # Errors
833    /// Returns an error if pool is not initialized, input is zero, or price limit is invalid.
834    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        // Positive = exact input.
841        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    /// Simulates an exact output swap (know output amount, calculate required input amount).
848    ///
849    /// # Errors
850    /// Returns an error if pool is not initialized, output is zero, price limit is invalid, or
851    /// insufficient liquidity exists to fulfill the exact output amount.
852    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        // Negative = exact output.
859        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    /// Simulates a swap to move the pool price down to a target price.
867    ///
868    /// # Errors
869    /// Returns an error if pool is not initialized or price limit is invalid.
870    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    /// Simulates a swap to move the pool price up to a target price.
878    ///
879    /// # Errors
880    /// Returns an error if pool is not initialized or price limit is invalid.
881    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    /// Finds the maximum trade size that produces a target slippage (including fees).
889    ///
890    /// Uses binary search to find the largest trade size that results in slippage
891    /// at or below the target. The method iteratively simulates swaps at different
892    /// sizes until it converges to the optimal size within the specified tolerance.
893    ///
894    /// # Returns
895    /// The maximum trade size (U256) that produces the target slippage
896    ///
897    /// # Errors
898    /// Returns an error if:
899    /// - Impact is zero or exceeds 100% (10000 bps).
900    /// - Pool is not initialized.
901    /// - Swap simulations fail.
902    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    /// Finds the maximum trade size with search diagnostics.
908    /// This is the detailed version of [`Self::size_for_impact_bps`] that returns
909    /// extensive information about the search process.It is useful for debugging,
910    /// monitoring, and analyzing search behavior in production.
911    ///
912    /// # Returns
913    /// Detailed result with size and search diagnostics
914    ///
915    /// # Errors
916    /// Returns an error if:
917    /// - Impact is zero or exceeds 100% (10000 bps).
918    /// - Pool is not initialized.
919    /// - Swap simulations fail.
920    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    /// Validates that the price limit is in the swap direction and strictly inside the
930    /// `(MIN_SQRT_RATIO, MAX_SQRT_RATIO)` range. The current price is valid when `allow_equal` is
931    /// set for direct simulation or replay; otherwise the direction check matches Uniswap V3.
932    ///
933    /// # Errors
934    /// Returns an error if price limit violates swap direction or range constraints.
935    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            // Swapping token0 for token1: price must decrease
943            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            // Swapping token1 for token0: price must increase
956            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    /// Processes a mint (liquidity add) event from historical data.
975    ///
976    /// Updates pool state when liquidity is added to a position, validates ticks,
977    /// and delegates to internal liquidity management methods.
978    ///
979    /// # Errors
980    ///
981    /// Returns an error if:
982    /// - Pool is not initialized.
983    /// - Tick range is invalid or not properly spaced.
984    /// - Position updates fail.
985    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    /// Adds liquidity to a position.
1028    ///
1029    /// Updates position state, tracks deposited amounts, and manages tick maps.
1030    /// Called by both historical event processing and simulated operations.
1031    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        // Track deposited amounts
1052        self.analytics.total_amount0_deposited += amount0;
1053        self.analytics.total_amount1_deposited += amount1;
1054
1055        Ok(())
1056    }
1057
1058    /// Executes a simulated mint (liquidity addition) operation.
1059    ///
1060    /// Calculates required token amounts for the specified liquidity amount,
1061    /// updates pool state, and returns the resulting mint event.
1062    /// Rejects the mint before changing pool state if the resulting gross liquidity at either
1063    /// boundary tick exceeds the pool's maximum liquidity per tick.
1064    ///
1065    /// # Errors
1066    ///
1067    /// Returns an error if:
1068    /// - Pool is not initialized.
1069    /// - Tick range is invalid.
1070    /// - Amount calculations fail.
1071    /// - Resulting tick liquidity exceeds the per-tick limit, or a liquidity calculation overflows.
1072    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, // ts_event (simulated; pool init time)
1114            self.pool.ts_init, // ts_init
1115        );
1116
1117        Ok(event)
1118    }
1119
1120    /// Processes a burn (liquidity removal) event from historical data.
1121    ///
1122    /// Updates pool state when liquidity is removed from a position. Uses negative
1123    /// liquidity delta to reduce the position size and tracks withdrawn amounts.
1124    ///
1125    /// # Errors
1126    ///
1127    /// Returns an error if:
1128    /// - Pool is not initialized.
1129    /// - Tick range is invalid.
1130    /// - Position updates fail.
1131    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        // Update the position with a negative liquidity delta for the burn
1142        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    /// Executes a simulated burn (liquidity removal) operation.
1180    ///
1181    /// Calculates token amounts that would be withdrawn for the specified liquidity,
1182    /// updates pool state, and returns the resulting burn event.
1183    ///
1184    /// # Errors
1185    ///
1186    /// Returns an error if:
1187    /// - Pool is not initialized.
1188    /// - Tick range is invalid.
1189    /// - Amount calculations fail.
1190    /// - Insufficient liquidity in position.
1191    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        // Update the position with a negative liquidity delta for the burn
1211        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, // ts_event (simulated; pool init time)
1242            self.pool.ts_init, // ts_init
1243        );
1244
1245        Ok(event)
1246    }
1247
1248    /// Processes a fee collect event from historical data.
1249    ///
1250    /// Updates position state when accumulated fees are collected. Finds the
1251    /// position and delegates fee collection to the position object.
1252    ///
1253    /// Note: Tick validation is intentionally skipped to match Uniswap V3 behavior.
1254    /// Invalid positions have no fees to collect, so they're silently ignored.
1255    ///
1256    /// # Errors
1257    ///
1258    /// Returns an error if:
1259    /// - Pool is not initialized.
1260    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        // Cleanup position if it became empty after collecting all fees
1278        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    /// Applies a protocol-fee configuration change from a `SetFeeProtocol` event.
1301    ///
1302    /// Applies the DEX-specific protocol-fee representation so subsequent swap and flash fee
1303    /// splitting uses the correct setting. Not gated on pool initialization, since a protocol-fee
1304    /// change is independent of the pool's price/liquidity state.
1305    ///
1306    /// # Errors
1307    ///
1308    /// Returns an error if a Uniswap V3 fee protocol update cannot be packed into one byte.
1309    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    /// Applies a protocol-fee withdrawal from a `CollectProtocol` event.
1344    ///
1345    /// Decrements the accrued protocol-fee balances by the withdrawn amounts, leaving the on-chain
1346    /// remainder (Uniswap V3 keeps one wei in each slot to save gas). Saturating subtraction guards
1347    /// against replay accrual lagging behind the on-chain balance. Not gated on pool initialization,
1348    /// since the protocol-fee balances are independent of the pool's price/liquidity state.
1349    ///
1350    /// This function does not return an error; the `Result` keeps the signature uniform with the
1351    /// other `process_*` event handlers.
1352    #[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    /// Processes a flash loan event from historical data.
1393    ///
1394    /// # Errors
1395    ///
1396    /// Returns an error if:
1397    /// - Pool has no active liquidity.
1398    /// - Fee growth arithmetic overflows.
1399    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    /// Executes a simulated flash loan operation and returns the resulting event.
1426    ///
1427    /// # Errors
1428    ///
1429    /// Returns an error if:
1430    /// - Mathematical operations overflow when calculating fees.
1431    /// - Pool has no active liquidity.
1432    /// - Fee growth arithmetic overflows.
1433    ///
1434    /// # Panics
1435    ///
1436    /// Panics if the pool fee has not been set.
1437    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        // Calculate fees or paid0/paid1
1450        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, // ts_event (simulated; pool init time)
1475            self.pool.ts_init, // ts_init
1476            sender,
1477            recipient,
1478            amount0,
1479            amount1,
1480            paid0,
1481            paid1,
1482        );
1483
1484        Ok(flash_event)
1485    }
1486
1487    /// Core flash loan state update logic.
1488    ///
1489    /// # Errors
1490    ///
1491    /// Returns an error if:
1492    /// - No active liquidity in pool.
1493    /// - Fee growth arithmetic overflows.
1494    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        // Process token0 fees
1506        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        // Process token1 fees
1520        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    /// Updates position state and tick maps when liquidity changes.
1557    ///
1558    /// Core internal method that handles position updates for both mints and burns.
1559    /// Updates tick maps, position tracking, fee growth, and active liquidity.
1560    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        // Only validate when burning (negative liquidity_delta)
1573        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        // Pre-validate so an over/underflow error returns before mutating tick map
1587        // or position state.
1588        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        // Update tickmaps.
1617        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        // Clear the ticks if they are flipped and burned
1650        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    /// Removes position from tracking if it's completely empty.
1662    ///
1663    /// This prevents accumulation of positions in the memory that are not used anymore.
1664    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    /// Calculates the liquidity utilization rate for the pool.
1679    ///
1680    /// The utilization rate measures what fraction of total deployed liquidity
1681    /// is currently active (in-range and earning fees) at the current price tick.
1682    /// Returns zero when no position liquidity is tracked, otherwise truncates to six decimal places.
1683    /// Partial-history replay can produce values above one when active liquidity includes positions
1684    /// whose mint events were not replayed.
1685    #[must_use]
1686    pub fn liquidity_utilization_rate(&self) -> f64 {
1687        const PRECISION: u32 = 1_000_000; // 6 decimal places
1688
1689        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    /// Validates tick range for position operations.
1706    ///
1707    /// Ensures ticks are properly ordered, aligned to tick spacing, and within
1708    /// valid bounds. Used by all position-related operations.
1709    ///
1710    /// # Errors
1711    ///
1712    /// Returns an error if:
1713    /// - `tick_lower >= tick_upper` (invalid range).
1714    /// - Ticks are not multiples of pool's tick spacing.
1715    /// - Ticks are outside `MIN_TICK/MAX_TICK` bounds.
1716    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    /// Updates all liquidity analytics.
1736    fn update_liquidity_analytics(&mut self) {
1737        self.analytics.liquidity_utilization_rate = self.liquidity_utilization_rate();
1738    }
1739
1740    /// Returns the pool's active liquidity tracked by the tick map.
1741    ///
1742    /// This represents the effective liquidity available for trading at the current price.
1743    /// The tick map maintains this value efficiently by updating it during tick crossings
1744    /// as the price moves through different ranges.
1745    ///
1746    /// # Returns
1747    /// The active liquidity (u128) at the current tick from the tick map
1748    #[must_use]
1749    pub fn get_active_liquidity(&self) -> u128 {
1750        self.tick_map.liquidity
1751    }
1752
1753    /// Calculates total liquidity by summing all individual positions at the current tick.
1754    ///
1755    /// This computes liquidity by iterating through all positions and summing those that
1756    /// span the current tick. Unlike [`Self::get_active_liquidity`], which returns the maintained
1757    /// tick map value, this method performs a fresh calculation from position data.
1758    #[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    /// Calculates total liquidity across all positions, regardless of range status.
1772    #[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    /// Restores the profiler state from a saved snapshot.
1781    ///
1782    /// This method allows resuming profiling from a previously saved state,
1783    /// enabling incremental processing without reprocessing all historical events.
1784    ///
1785    /// # Errors
1786    ///
1787    /// Returns an error if:
1788    /// - Tick insertion into the tick map fails.
1789    ///
1790    /// # Panics
1791    ///
1792    /// Panics if the pool's tick spacing is not set.
1793    pub fn restore_from_snapshot(&mut self, snapshot: PoolSnapshot) -> anyhow::Result<()> {
1794        let liquidity = snapshot.state.liquidity;
1795
1796        // Restore state
1797        self.state = snapshot.state;
1798
1799        // Restore analytics (skip duration fields as they're debug-only)
1800        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        // Rebuild positions AHashMap
1811        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        // Rebuild tick_map
1822        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        // Restore active liquidity
1833        self.tick_map.liquidity = liquidity;
1834
1835        // Set last processed event
1836        self.last_processed_event = Some(snapshot.block_position);
1837        self.last_processed_ts = Some(snapshot.ts_event);
1838
1839        // Mark as initialized
1840        self.is_initialized = true;
1841
1842        // Recalculate analytics
1843        self.update_liquidity_analytics();
1844
1845        Ok(())
1846    }
1847
1848    /// Gets a list of all initialized tick values.
1849    ///
1850    /// Returns tick values that have been initialized (have liquidity positions).
1851    /// Useful for understanding the liquidity distribution across price ranges.
1852    #[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    /// Gets the number of active ticks.
1863    #[must_use]
1864    pub fn get_active_tick_count(&self) -> usize {
1865        self.tick_map.active_tick_count()
1866    }
1867
1868    /// Gets tick information for a specific tick value.
1869    ///
1870    /// Returns the tick data structure containing liquidity and fee information
1871    /// for the specified tick, if it exists.
1872    #[must_use]
1873    pub fn get_tick(&self, tick: i32) -> Option<&PoolTick> {
1874        self.tick_map.get_tick(tick)
1875    }
1876
1877    /// Gets the current tick position of the pool.
1878    ///
1879    /// Returns the tick that corresponds to the current pool price.
1880    /// The pool must be initialized before calling this method.
1881    #[must_use]
1882    pub fn get_current_tick(&self) -> i32 {
1883        self.state.current_tick
1884    }
1885
1886    /// Gets the total number of ticks tracked by the tick map.
1887    ///
1888    /// Returns count of all ticks that have ever been initialized,
1889    /// including those that may no longer have active liquidity.
1890    ///
1891    /// # Returns
1892    /// Total tick count in the tick map
1893    #[must_use]
1894    pub fn get_total_tick_count(&self) -> usize {
1895        self.tick_map.total_tick_count()
1896    }
1897
1898    /// Gets position information for a specific owner and tick range.
1899    ///
1900    /// Looks up a position by its unique key (owner + tick range) and returns
1901    /// the position data if it exists.
1902    #[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    /// Returns a list of all currently active positions.
1914    ///
1915    /// Active positions are those with liquidity > 0 whose tick range includes
1916    /// the current pool tick, meaning they have tokens actively deployed in the pool
1917    /// and are earning fees from trades at the current price.
1918    ///
1919    /// # Returns
1920    ///
1921    /// A vector of references to active [`PoolPosition`] objects.
1922    #[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    /// Returns a list of all positions tracked by the profiler.
1936    ///
1937    /// This includes both active and inactive positions, regardless of their
1938    /// liquidity or tick range relative to the current pool tick.
1939    ///
1940    /// # Returns
1941    ///
1942    /// A vector of references to all [`PoolPosition`] objects.
1943    #[must_use]
1944    pub fn get_all_positions(&self) -> Vec<&PoolPosition> {
1945        self.positions.values().collect()
1946    }
1947
1948    /// Returns position keys for all tracked positions.
1949    #[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    /// Extracts a complete snapshot of the current pool state.
1958    ///
1959    /// Extracts and bundles the complete pool state including global variables,
1960    /// all liquidity positions, and the full tick distribution into a portable
1961    /// [`PoolSnapshot`] structure. This snapshot can be serialized, persisted
1962    /// to database, or used to restore pool state later.
1963    ///
1964    /// # Errors
1965    ///
1966    /// Returns an error if no events have been processed yet, since there is no event watermark to
1967    /// anchor the snapshot to.
1968    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), // ts_event (last processed event)
1988            self.last_processed_ts.unwrap_or(self.pool.ts_init), // ts_init
1989        ))
1990    }
1991
1992    /// Gets the count of positions that are currently active.
1993    ///
1994    /// Active positions are those with liquidity > 0 and whose tick range
1995    /// includes the current pool tick (meaning they have tokens in the pool).
1996    #[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    /// Gets the count of positions that are currently inactive.
2010    ///
2011    /// Inactive positions are those that exist but don't span the current tick,
2012    /// meaning their liquidity is entirely in one token or the other.
2013    #[must_use]
2014    pub fn get_total_inactive_positions(&self) -> usize {
2015        self.positions.len() - self.get_total_active_positions()
2016    }
2017
2018    /// Estimates the total amount of token0 in the pool.
2019    ///
2020    /// Calculates token0 balance by summing:
2021    /// - Token0 amounts from all active liquidity positions
2022    /// - Accumulated trading fees (approximated from fee growth)
2023    /// - Protocol fees collected
2024    #[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        // 1. Calculate token0 from active liquidity positions
2031        for position in self.positions.values() {
2032            // Empty positions and positions below current price hold no token0
2033            if position.liquidity == 0 || position.tick_upper <= current_tick {
2034                continue;
2035            }
2036
2037            if position.tick_lower > current_tick {
2038                // Position is above current price - all token0
2039                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                // Position is active - token0 from current price to upper tick
2046                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        // 2. Add accumulated swap fees (fee_growth_global represents total fees accumulated)
2058        // Note: In a real pool, fees are distributed as liquidity, but for balance estimation
2059        // we can use a simplified approach by converting fee growth to token amounts
2060        let fee_growth_0 = self.state.fee_growth_global_0;
2061        if fee_growth_0 > U256::ZERO {
2062            // Convert fee growth to actual token amount using FullMath for precision
2063            // Fee growth is in Q128.128 format, so we need to scale it properly
2064            let active_liquidity = self.get_active_liquidity();
2065            if active_liquidity > 0 {
2066                // fee_growth_global is fees per unit of liquidity in Q128.128
2067                // To get total fees: mul_div(fee_growth, liquidity, 2^128)
2068                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        // 4. Add protocol fees
2077        total_amount0 += self.state.protocol_fees_token0;
2078
2079        total_amount0
2080    }
2081
2082    /// Estimates the total amount of token1 in the pool.
2083    ///
2084    /// Calculates token1 balance by summing:
2085    /// - Token1 amounts from all active liquidity positions
2086    /// - Accumulated trading fees (approximated from fee growth)
2087    /// - Protocol fees collected
2088    #[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        // 1. Calculate token1 from active liquidity positions
2095        for position in self.positions.values() {
2096            // Empty positions and positions above current price hold no token1
2097            if position.liquidity == 0 || position.tick_lower > current_tick {
2098                continue;
2099            }
2100
2101            if position.tick_upper <= current_tick {
2102                // Position is below current price - all token1
2103                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                // Position is active - token1 from lower tick to current price
2110                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        // 2. Add accumulated swap fees for token1
2122        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                // Convert fee growth to actual token amount using FullMath
2127                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        // 4. Add protocol fees
2136        total_amount1 += self.state.protocol_fees_token1;
2137
2138        total_amount1
2139    }
2140
2141    /// Sets the global fee growth for both tokens.
2142    ///
2143    /// This is primarily used for testing to simulate specific fee growth scenarios.
2144    /// In production, fee growth is updated through swap operations.
2145    ///
2146    /// # Arguments
2147    /// * `fee_growth_global_0` - New global fee growth for token0
2148    /// * `fee_growth_global_1` - New global fee growth for token1
2149    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    /// Returns the total number of events processed.
2155    #[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    /// Enables progress reporting for pool profiler event processing.
2165    ///
2166    /// When enabled, the profiler will automatically track and log progress
2167    /// as events are processed through the `process()` method.
2168    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    /// Finalizes reporting and logs final statistics.
2179    ///
2180    /// Should be called after all events have been processed to output
2181    /// the final summary of the profiler bootstrap operation.
2182    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}