degenbot.uniswap.v3_snapshot

Uniswap V3 pool snapshot and subscription handler.

Module Contents

class degenbot.uniswap.v3_snapshot.UniswapV3LiquiditySnapshotSource

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

A minimal protocol for retrieving pool data from a generic source.

Any class implementing the protocol must implement these methods, transforming data as necessary to return the specified types.

get_liquidity_map(pool_address: degenbot.types.chain.ChecksummedAddress) → degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap | None

Return liquidity map.

Returns:

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

get_pools() → set[degenbot.types.chain.ChecksummedAddress]

Return pools.

Returns:

The set of pool addresses.

class degenbot.uniswap.v3_snapshot.MonolithicJsonFileSnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.MonolithicJsonFileSnapshotBase

A pool liquidity source backed by a single JSON file.

get_liquidity_map(pool_address: degenbot.types.chain.ChecksummedAddress) → degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap | None

Return liquidity map.

Returns:

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

class degenbot.uniswap.v3_snapshot.IndividualJsonFileSnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.IndividualJsonFileSnapshotBase

Snapshot source backed by a directory of JSON files.

class degenbot.uniswap.v3_snapshot.DatabaseSnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.DatabaseSnapshotBase[degenbot.types.chain.ChecksummedAddress]

Snapshot source backed by the built-in SQLite database.

get_liquidity_map(pool_address: degenbot.types.chain.ChecksummedAddress) → degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap | None

Return liquidity map.

Returns:

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

class degenbot.uniswap.v3_snapshot.UniswapV3LiquiditySnapshot

Bases: degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase[degenbot.types.chain.ChecksummedAddress, degenbot.uniswap.v3_types.UniswapV3PoolLiquidityMappingUpdate, degenbot.uniswap.v3_types.UniswapV3LiquidityEvent, UniswapV3LiquiditySnapshotSource]

Retrieve and maintain liquidity positions for Uniswap V3 pools.

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.

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.

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.

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.