degenbot.uniswap.concentrated.snapshot_readers

Generic snapshot readers and liquidity-snapshot facade shared by Uniswap V3/V4.

Module Contents

class degenbot.uniswap.concentrated.snapshot_readers.LiquidityMap

Bases: TypedDict

LiquidityMap class.

tick_bitmap: dict[int, degenbot.uniswap.concentrated.types.BitmapAtWord]

tick_data: dict[int, degenbot.uniswap.concentrated.types.LiquidityAtTick]

class degenbot.uniswap.concentrated.snapshot_readers.LiquidityEvent

Bases: Protocol

The fields a liquidity event exposes to the snapshot update flow.

block_number: int

liquidity: int

tick_lower: int

tick_upper: int

class degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotSource

Bases: 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.

storage_kind: str

chain_id: int

get_newest_block() → degenbot.types.aliases.BlockNumber | None

Return newest block.

Returns:

The newest block number, or None if unavailable.

get_pools() → set[Any]

Return pools.

Returns:

The set of pool keys.

class degenbot.uniswap.concentrated.snapshot_readers.MonolithicJsonFileSnapshotBase(path: pathlib.Path | str)

A pool liquidity source backed by a single JSON file.

{

“snapshot_block”: int, “chain_id”: int, “0xPoolKey1”: {

“tick_bitmap”: {

<word>: {

‘bitmap’: <value>, ‘block’: <value>,

}, “tick_data”: {

<tick>: {

‘liquidity_gross: <value>, ‘liquidity_net’: <value>, ‘block: <value>,

}

}

}, “0xPoolKey2”: { … }, “0xPoolKey3”: { … }, …

}.

storage_kind = 'file'

chain_id: int

get_newest_block() → degenbot.types.aliases.BlockNumber | None

Return newest block.

Returns:

The snapshot block number, or None if unavailable.

get_pools() → set[str]

Return pools.

Returns:

The set of pool keys from the snapshot.

class degenbot.uniswap.concentrated.snapshot_readers.IndividualJsonFileSnapshotBase(path: pathlib.Path | str)

Snapshot source backed by a directory of JSON files with this tree structure.

/path/to/snapshots/ |– _metadata.json -> { “block”: int, “chain_id”: int } |– 0xPoolKey1.json -> { “tick_bitmap”: {…}, “tick_data”: {…} } |– 0xPoolKey2.json -> { “tick_bitmap”: {…}, “tick_data”: {…} } |– 0xPoolKey3.json -> { “tick_bitmap”: {…}, “tick_data”: {…} }

Each pool file contains the same structure as the monolithic snapshot’s per-pool entries.

storage_kind = 'dir'

chain_id: int

get_newest_block() → degenbot.types.aliases.BlockNumber | None

Return newest block.

Returns:

The block number from metadata, or None if unavailable.

get_pools() → set[str]

Return pools.

Returns:

The set of pool keys from file stems.

get_liquidity_map(pool_address: str) → LiquidityMap | None

Return liquidity map.

Returns:

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

class degenbot.uniswap.concentrated.snapshot_readers.DatabaseSnapshotBase[K](chain_id: degenbot.types.aliases.ChainId, *, database_path: pathlib.Path)

Snapshot source backed by built-in SQLite database.

Routes every read through the Rust degenbot-db core crate via the _EngineSnapshot PyO3 seam (ADR-005 three-layer architecture). The explicit file path is retained as the authority for the lazy Rust handle.

storage_kind = 'db'

database_path

chain_id

close() → None

Release the Rust handle and its SQLite connection.

Closing is idempotent. A closed snapshot cannot lazily reopen its database handle.

get_all_liquidity_maps() → dict[K, dict[int, tuple[int, int]]]

Return all tick data as plain dicts.

Delegates the bulk read to the Rust core (GIL released during the SQLite scan). Returns {pool_key: {tick_index: (liquidity_gross, liquidity_net)}}.

Returns:

A dict mapping pool keys to tick data dicts.

get_newest_block() → degenbot.types.aliases.BlockNumber | None

Return newest block.

Returns:

The newest block number across all exchanges, or None if unavailable.

get_pools() → set[str]

Return pools.

Returns:

The set of pool keys from the database.

class degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase[K, U, E: LiquidityEvent, S: LiquiditySnapshotSource](source: S)

Retrieve and maintain liquidity positions for concentrated-liquidity pools.

newest_block: degenbot.types.aliases.BlockNumber

property chain_id: int

Chain id.

property pools: set[Any]

Pools.