degenbot.uniswap ================ .. py:module:: degenbot.uniswap .. autoapi-nested-parse:: Uniswap V2/V3/V4 pool type registration and exports. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/degenbot/uniswap/cl_companion/index /autoapi/degenbot/uniswap/cl_pool_state/index /autoapi/degenbot/uniswap/concentrated/index /autoapi/degenbot/uniswap/deployments/index /autoapi/degenbot/uniswap/math/index /autoapi/degenbot/uniswap/snapshot_binary/index /autoapi/degenbot/uniswap/trackers/index /autoapi/degenbot/uniswap/types/index /autoapi/degenbot/uniswap/v2_functions/index /autoapi/degenbot/uniswap/v2_liquidity_pool/index /autoapi/degenbot/uniswap/v2_pool_calc/index /autoapi/degenbot/uniswap/v2_pool_state/index /autoapi/degenbot/uniswap/v2_types/index /autoapi/degenbot/uniswap/v3_functions/index /autoapi/degenbot/uniswap/v3_liquidity_pool/index /autoapi/degenbot/uniswap/v3_pool_calc/index /autoapi/degenbot/uniswap/v3_pool_state/index /autoapi/degenbot/uniswap/v3_snapshot/index /autoapi/degenbot/uniswap/v3_types/index /autoapi/degenbot/uniswap/v4_liquidity_pool/index /autoapi/degenbot/uniswap/v4_pool_calc/index /autoapi/degenbot/uniswap/v4_pool_state/index /autoapi/degenbot/uniswap/v4_snapshot/index /autoapi/degenbot/uniswap/v4_types/index Package Contents ---------------- .. py:class:: UniswapV2PoolTracker Bases: :py:obj:`AbstractUniswapV2PoolTracker`\ [\ :py:obj:`degenbot.uniswap.v2_liquidity_pool.UniswapV2Pool`\ ] A class that generates and tracks concrete instances of a Uniswap V2. liquidity pool helper or one of its child classes. .. py:property:: pool_init_hash :type: str Pool init hash. :returns: The pool init hash string. .. py:method:: get_pool_from_tokens(token_addresses: tuple[str, str], *, silent: bool = False) -> degenbot.uniswap.v2_liquidity_pool.UniswapV2Pool Get a pool by its token addresses. :returns: The pool instance. .. py:class:: UniswapV3PoolTracker Bases: :py:obj:`AbstractUniswapV3PoolTracker`\ [\ :py:obj:`degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool`\ ] A class that generates and tracks concrete instances of a Uniswap V3. liquidity pool helper or one of its child classes. .. py:method:: get_pool_from_tokens_and_fee(token_addresses: tuple[degenbot.types.chain.ChecksummedAddress | str, degenbot.types.chain.ChecksummedAddress | str], pool_fee: int, *, silent: bool = False) -> degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool Get a pool by its token addresses and fee. :returns: The pool instance. .. py:class:: UniswapV2Pool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.uniswap.v2_pool_state.V2PoolState`, :py:obj:`degenbot.uniswap.v2_pool_calc.UniswapV2PoolCalc`, :py:obj:`degenbot.types.abstract.AbstractLiquidityPool` A Uniswap V2-based liquidity pool implementing the x*y=k constant function invariant. .. py:attribute:: variant :type: ClassVar[str | None] :value: None .. py:attribute:: stable_swap :type: bool :value: False .. py:attribute:: fee_denominator :type: int | None :value: None .. py:attribute:: dex :type: degenbot.types.DexIdentity .. py:attribute:: address :type: degenbot._ffi.ChecksummedAddress .. py:attribute:: factory :type: degenbot._ffi.ChecksummedAddress .. py:attribute:: init_hash :type: str .. py:attribute:: deployer :type: degenbot._ffi.ChecksummedAddress .. py:attribute:: name :type: str .. py:type:: PoolState :canonical: UniswapV2PoolState .. py:method:: from_handle(py_pool: degenbot.types.Pool) -> Self :classmethod: Wrap a Rust-owned ``Pool`` handle as a Python companion. Internal seam (ADR-005, Polars-style ``_from_pydf`` pattern). The handle is self-describing: every identity field (address, factory, fees, tokens, dex preset, stable strategy) is read off it — no identity is passed as constructor args. Rust owns the mutable state (reserves + reorg journal) as ``V2PoolState`` and the immutable registration metadata as ``V2PoolDescriptor``; this companion reads both through ``self._py_pool``. Only ``Bot.build_pool()`` (production) and ``make_v2_pool`` (tests) should call this — they have already registered the pool (and, per ADR-006, its tokens in the same ``Bot``) and obtained the handle. ``cls`` is used so subclasses that only set ClassVars (the documented extension contract) inherit this seam and produce instances of the subclass. :returns: A ``cls`` instance wrapping ``py_pool``. :raises DegenbotValueError: If the handle is not a V2-family pool (``py_pool.variant`` is empty — the ``PoolEntry`` is not ``V2``), so the union-handle V2 getters would return empty/default identity. :raises DegenbotValueError: If the handle has no ``DexIdentity`` preset (the pool was not registered with a variant) or the pool's tokens are not registered in the same ``Bot`` (ADR-006). .. py:property:: update_block :type: degenbot.types.aliases.BlockNumber Update block. :returns: The block number of the most recent state update (from Rust). .. py:property:: reserves_token0 :type: int Reserves token0. :returns: The reserve amount for token0 (from Rust). .. py:property:: reserves_token1 :type: int Reserves token1. :returns: The reserve amount for token1 (from Rust). .. py:property:: state :type: PoolState State. :returns: The current pool state, built from one atomic Rust snapshot (``_py_pool.snapshot()``) so a Rust-side ``sync_reserves`` (pump update) can't interleave between the reserve reads. :raises DegenbotValueError: If the pool is not registered in Rust (no V2 state to snapshot). .. py:method:: external_update(update: degenbot.uniswap.v2_types.UniswapV2PoolExternalUpdate) -> None External update. :raises ExternalUpdateError: If the update is for a past block. .. py:method:: discard_states_before_block(block: degenbot.types.aliases.BlockNumber) -> None Discard cached V2 reorg journal deltas earlier than the given block. Delegates to ``Pool.discard_before_block`` (Rust pops journal deltas strictly earlier than the target, keeping the genesis delta + everything at/after the target). The current state is unchanged when the target is at/after the newest delta. :raises NoPoolStateAvailable: If the target is past the newest delta (would remove every known state). .. py:method:: restore_state_before_block(block: degenbot.types.aliases.BlockNumber) -> None Restore the V2 pool to the landed-at state just before the target block. Delegates to ``Pool.restore_before_block`` (Rust pops journal deltas at/after the target + reverse-applies them, writing back the pre-target reserves in one write guard). The journal's ``update_block`` lands at the oldest popped delta's block (the target convention); the restored reserves are the pre-target state. :raises NoPoolStateAvailable: If no state exists prior to the target block (the target is at or before the registration block). .. py:method:: simulate_exact_input_swap(token_in: degenbot.erc20.Erc20Token, token_in_quantity: int, override_state: PoolState | None = None) -> degenbot.uniswap.v2_types.UniswapV2PoolSimulationResult Simulate an exact input swap. :returns: The simulation result with delta amounts and state transitions. :raises DegenbotValueError: If token_in is unknown. .. py:method:: simulate_exact_output_swap(token_out: degenbot.erc20.Erc20Token, token_out_quantity: int, override_state: PoolState | None = None) -> degenbot.uniswap.v2_types.UniswapV2PoolSimulationResult Simulate exact output swap. :returns: The simulation result with delta amounts and state transitions. :raises DegenbotValueError: If token_out is unknown. .. py:method:: calculate_tokens_out_from_tokens_in(token_in: degenbot.erc20.Erc20Token, token_in_quantity: int, override_state: degenbot.uniswap.v2_types.UniswapV2PoolState | None = None) -> int Calculate the expected token OUTPUT for a target INPUT at current reserves. Strategy dispatch (ADR-005 slice 7 step 4a fold): Camelot stable pools (``stable_swap=True``) use the solidly-stable invariant with Camelot's k/get_y; all other V2 pools fall through to ``super()`` — the ``UniswapV2PoolCalc`` Rust-delegation path (slice 5) is unperturbed for the volatile majority. :returns: The expected output token amount. .. py:class:: UniswapV2PoolExternalUpdate UniswapV2PoolExternalUpdate class. .. py:attribute:: block_number :type: degenbot.types.aliases.BlockNumber .. py:attribute:: reserves_token0 :type: int .. py:attribute:: reserves_token1 :type: int .. py:class:: UniswapV2PoolSimulationResult Bases: :py:obj:`UniswapSimulationResult` UniswapV2PoolSimulationResult class. .. py:attribute:: initial_state :type: UniswapV2PoolState .. py:attribute:: final_state :type: UniswapV2PoolState .. py:class:: UniswapV2PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` UniswapV2PoolState class. .. py:attribute:: reserves_token0 :type: int .. py:attribute:: reserves_token1 :type: int .. py:class:: UniswapV3LiquiditySnapshot Bases: :py:obj:`degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase`\ [\ :py:obj:`degenbot.types.chain.ChecksummedAddress`\ , :py:obj:`degenbot.uniswap.v3_types.UniswapV3PoolLiquidityMappingUpdate`\ , :py:obj:`degenbot.uniswap.v3_types.UniswapV3LiquidityEvent`\ , :py:obj:`UniswapV3LiquiditySnapshotSource`\ ] Retrieve and maintain liquidity positions for Uniswap V3 pools. .. py:method:: pending_updates(pool_address: str) -> tuple[degenbot.uniswap.v3_types.UniswapV3PoolLiquidityMappingUpdate, ...] Consume pending liquidity updates for the pool. :returns: A tuple of liquidity mapping updates for the pool. .. py:method:: tick_bitmap(pool_address: str | bytes) -> dict[int, degenbot.uniswap.concentrated.types.BitmapAtWord] | None Consume the tick initialization bitmaps for the pool. :returns: The tick bitmap dict, or None if no snapshot exists. .. py:method:: tick_data(pool_address: str | bytes) -> dict[int, degenbot.uniswap.concentrated.types.LiquidityAtTick] | None Consume the liquidity mapping for the pool. :returns: The tick data dict, or None if no snapshot exists. .. py:method:: update(pool: degenbot.types.chain.HexAddress, tick_data: dict[int, degenbot.uniswap.concentrated.types.LiquidityAtTick], tick_bitmap: dict[int, degenbot.uniswap.concentrated.types.BitmapAtWord]) -> None Update the liquidity mapping for the pool. :raises UnknownPool: If the pool has no snapshot. .. py:class:: UniswapV3PoolExternalUpdate UniswapV3PoolExternalUpdate class. .. py:attribute:: block_number :type: degenbot.types.aliases.BlockNumber .. py:attribute:: liquidity :type: Liquidity .. py:attribute:: sqrt_price_x96 :type: SqrtPriceX96 .. py:attribute:: tick :type: Tick .. py:class:: UniswapV3PoolSimulationResult Bases: :py:obj:`UniswapSimulationResult` UniswapV3PoolSimulationResult class. .. py:attribute:: initial_state :type: UniswapV3PoolState .. py:attribute:: final_state :type: UniswapV3PoolState .. py:class:: UniswapV3PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` UniswapV3PoolState class. .. py:attribute:: liquidity :type: Liquidity .. py:attribute:: sqrt_price_x96 :type: SqrtPriceX96 .. py:attribute:: tick :type: Tick .. py:attribute:: tick_bitmap :type: dict[BitmapWord, degenbot.uniswap.concentrated.types.BitmapAtWord] .. py:attribute:: tick_data :type: dict[Tick, degenbot.uniswap.concentrated.types.LiquidityAtTick] .. py:class:: UniswapV4Pool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.uniswap.v4_pool_state.V4PoolState`, :py:obj:`degenbot.uniswap.v4_pool_calc.UniswapV4PoolCalc`, :py:obj:`degenbot.uniswap.cl_companion.ConcentratedLiquidityCompanion` A Uniswap V4 concentrated-liquidity pool companion over a ``Pool`` handle. Rust owns the mutable state (scalars + tick data + reorg journal) as ``V4PoolState``; this companion reads it through ``self._py_pool`` (one atomic ``snapshot_v3()`` for scalars — already V3/V4-generic via ``get_v3_or_v4_pool`` — + ``tick_data_snapshot()`` / ``tick_bitmap_snapshot()`` for the tick maps) and delegates ``external_update`` (Swap) / ``update_liquidity_map`` (ModifyLiquidity) / ``update_tick_data`` (sparse-map backfill) / discard / restore to the handle. V4-specific identity (pool_id, pool_manager, pool_key, hooks, protocol_fee, lp_fee, state_view_address) stays Python-side — matches V3. Construct via the V4 builder (which registers in Rust and hands the handle here); tests use ``make_v4_pool``. Hook admission floor: pools with amount-modifying hooks (`hook_flags & 0xCC != 0`) or dynamic fees (`fee == 0x100000`) are rejected in Rust (`BotState::register_v4_pool`), surfaced at `Bot.register_v4_pool` as typed exceptions — so this companion never holds a hooked/dynamic-fee pool. .. py:type:: PoolState :canonical: UniswapV4PoolState .. py:attribute:: hook_address :type: degenbot.types.chain.ChecksummedAddress .. py:attribute:: active_hooks :type: frozenset[Hooks] .. py:attribute:: name :type: str .. py:attribute:: protocol_fee :type: ProtocolFee .. py:attribute:: lp_fee :type: int .. py:method:: from_handle(py_pool: degenbot.types.Pool) -> Self :classmethod: Wrap a Rust-owned ``Pool`` handle as a Python companion. Internal seam (ADR-005, Polars-style ``_from_pydf`` pattern). The handle is self-describing: every identity field (pool_manager, pool_id, pool_key, hooks, tokens, fee, tick_spacing) is read off it — no identity is passed as constructor args. Rust owns the mutable state (slot0 + tick_data + reorg journal) as ``V4PoolState`` and the immutable registration metadata as ``V4PoolIdentity``; this companion reads both through ``self._py_pool``. Protocol fee / LP fee / state_view_address are builder-supplied values the seam defaults; the builder overrides them after ``from_handle`` (matches V3's deployer/init_hash override). :returns: A ``cls`` instance wrapping ``py_pool``. :raises DegenbotValueError: If the handle is not a V4-family pool. .. py:method:: calculate_tokens_in_from_tokens_out(token_out: degenbot.erc20.Erc20Token, token_out_quantity: int, override_state: degenbot.uniswap.v4_types.UniswapV4PoolState | None = None) -> int Calculate tokens in from tokens out. :returns: The required input token amount. :raises DegenbotValueError: If token_out is not held by this pool. :raises HookedPoolResult: If the pool has active hooks that affect the swap. :raises IncompleteSwap: If the swap cannot fulfill the full output amount. :raises LiquidityPoolError: If the simulated execution reverts. .. py:method:: calculate_tokens_out_from_tokens_in(token_in: degenbot.erc20.Erc20Token, token_in_quantity: int, override_state: degenbot.uniswap.v4_types.UniswapV4PoolState | None = None) -> int Calculate tokens out from tokens in. :returns: The expected output token amount. :raises DegenbotValueError: If token_in is not held by this pool. :raises HookedPoolResult: If the pool has active hooks that affect the swap. :raises IncompleteSwap: If the swap cannot fulfill the full input amount. :raises LiquidityPoolError: If the simulated execution reverts. .. py:property:: address :type: degenbot.types.chain.ChecksummedAddress Address. :returns: The pool manager address. .. py:property:: pool_id :type: bytes Pool id. :returns: The pool ID bytes. .. py:property:: pool_key :type: degenbot.uniswap.v4_types.UniswapV4PoolKey Pool key. :returns: The V4 pool key struct. .. py:property:: sqrt_price_x96 :type: int Sqrt price x96. :returns: The current sqrt price as a Q64.96 value (from Rust). .. py:property:: state :type: degenbot.uniswap.v4_types.UniswapV4PoolState State. :returns: The current pool state, built from one atomic Rust scalar snapshot (``_py_pool.snapshot_v3()`` — V3/V4-generic — ) + the tick-map snapshots. :raises DegenbotValueError: If the pool is not registered in Rust. .. py:property:: tick_spacing :type: int Tick spacing. :returns: The tick spacing for the pool (Python-side identity). .. py:property:: fee :type: int Fee. :returns: The fee in pips (Python-side identity). .. py:class:: UniswapV4LiquiditySnapshot Bases: :py:obj:`degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase`\ [\ :py:obj:`ManagedPoolIdentifier`\ , :py:obj:`degenbot.uniswap.v4_types.UniswapV4PoolLiquidityMappingUpdate`\ , :py:obj:`degenbot.uniswap.v4_types.UniswapV4LiquidityEvent`\ , :py:obj:`UniswapV4LiquiditySnapshotSource`\ ] Retrieve and maintain liquidity positions for Uniswap V4 pools. .. py:property:: pools :type: set[ManagedPoolIdentifier] Pools. .. py:method:: pending_updates(pool_manager: degenbot.types.chain.HexAddress | bytes, pool_id: degenbot.types.chain.HexStr | bytes) -> tuple[degenbot.uniswap.v4_types.UniswapV4PoolLiquidityMappingUpdate, ...] Consume and return all pending liquidity events for this pool. :returns: Tuple of pending liquidity mapping updates for the pool. .. py:method:: tick_bitmap(pool_manager: degenbot.types.chain.HexAddress | bytes, pool_id: degenbot.types.chain.HexStr | bytes) -> dict[int, degenbot.uniswap.concentrated.types.BitmapAtWord] | None Consume the tick initialization bitmaps for the pool. :returns: The tick bitmap dict, or None if the pool snapshot is unavailable. .. py:method:: tick_data(pool_manager: degenbot.types.chain.HexAddress | bytes, pool_id: degenbot.types.chain.HexStr | bytes) -> dict[int, degenbot.uniswap.concentrated.types.LiquidityAtTick] | None Consume the liquidity mapping for the pool. :returns: The tick data dict, or None if the pool snapshot is unavailable. .. py:method:: update(pool_manager: degenbot.types.chain.HexAddress | bytes, pool_id: degenbot.types.chain.HexStr | bytes, tick_data: dict[int, degenbot.uniswap.concentrated.types.LiquidityAtTick], tick_bitmap: dict[int, degenbot.uniswap.concentrated.types.BitmapAtWord]) -> None Update the liquidity mapping for the pool. :raises UnknownPoolId: If the pool is not found in the snapshot. .. py:class:: UniswapV4PoolExternalUpdate UniswapV4PoolExternalUpdate class. .. py:attribute:: block_number :type: degenbot.types.aliases.BlockNumber .. py:attribute:: liquidity :type: degenbot.uniswap.v3_types.Liquidity .. py:attribute:: sqrt_price_x96 :type: degenbot.uniswap.v3_types.SqrtPriceX96 .. py:attribute:: tick :type: degenbot.uniswap.v3_types.Tick .. py:class:: UniswapV4PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` UniswapV4PoolState class. .. py:attribute:: liquidity :type: degenbot.uniswap.v3_types.Liquidity .. py:attribute:: sqrt_price_x96 :type: degenbot.uniswap.v3_types.SqrtPriceX96 .. py:attribute:: tick :type: degenbot.uniswap.v3_types.Tick .. py:attribute:: tick_bitmap :type: dict[degenbot.uniswap.v3_types.BitmapWord, degenbot.uniswap.concentrated.types.BitmapAtWord] .. py:attribute:: tick_data :type: dict[degenbot.uniswap.v3_types.Tick, degenbot.uniswap.concentrated.types.LiquidityAtTick] .. py:attribute:: id :type: bytes .. py:attribute:: block :type: degenbot.types.aliases.BlockNumber | None