degenbot.uniswap.v4_snapshot

Uniswap V4 pool snapshot and subscription handler.

Module Contents

type degenbot.uniswap.v4_snapshot.PoolId = str

type degenbot.uniswap.v4_snapshot.PoolManagerAddress = ChecksummedAddress

type degenbot.uniswap.v4_snapshot.ManagedPoolIdentifier = tuple[PoolManagerAddress, PoolId]

class degenbot.uniswap.v4_snapshot.UniswapV4LiquiditySnapshotSource

Bases: degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotSource, Protocol

A minimal protocol allowing the UniswapV4LiquiditySnapshot class to retrieve pool data.

from a generic source.

get_liquidity_map(pool_manager: degenbot.types.chain.ChecksummedAddress, pool_id: bytes | str) → degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap | None

Return liquidity map.

get_pools() → set[PoolId]

Return pools.

class degenbot.uniswap.v4_snapshot.MonolithicJsonFileSnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.MonolithicJsonFileSnapshotBase

A pool liquidity source backed by a single JSON file.

get_liquidity_map(pool_manager: degenbot.types.chain.ChecksummedAddress, pool_id: bytes | str) → degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap | None

Return liquidity map.

Returns:

The liquidity map for the pool, or None if the pool is not found.

class degenbot.uniswap.v4_snapshot.DatabaseSnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.DatabaseSnapshotBase[ManagedPoolIdentifier]

Snapshot source backed by the built-in SQLite database.

get_liquidity_map(pool_manager: degenbot.types.chain.ChecksummedAddress, pool_id: bytes | str) → degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap | None

Return liquidity map.

Returns:

The liquidity map for the pool, or None if not found in the database.

class degenbot.uniswap.v4_snapshot.UniswapV4LiquiditySnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase[ManagedPoolIdentifier, degenbot.uniswap.v4_types.UniswapV4PoolLiquidityMappingUpdate, degenbot.uniswap.v4_types.UniswapV4LiquidityEvent, UniswapV4LiquiditySnapshotSource]

Retrieve and maintain liquidity positions for Uniswap V4 pools.

property pools: set[ManagedPoolIdentifier]

Pools.

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.

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.

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.

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.