degenbot.uniswap¶

Uniswap V2/V3/V4 pool type registration and exports.

Submodules¶

Package Contents¶

class degenbot.uniswap.UniswapV2PoolTracker¶

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

property pool_init_hash: str¶

Pool init hash.

Returns:

The pool init hash string.

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.

class degenbot.uniswap.UniswapV3PoolTracker¶

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

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.

class degenbot.uniswap.UniswapV2Pool(*args: Any, **kwargs: Any)¶

Bases: degenbot.uniswap.v2_pool_state.V2PoolState, degenbot.uniswap.v2_pool_calc.UniswapV2PoolCalc, degenbot.types.abstract.AbstractLiquidityPool

A Uniswap V2-based liquidity pool implementing the x*y=k constant function invariant.

variant: ClassVar[str | None] = None¶
stable_swap: bool = False¶
fee_denominator: int | None = None¶
dex: degenbot.types.DexIdentity¶
address: degenbot._ffi.ChecksummedAddress¶
factory: degenbot._ffi.ChecksummedAddress¶
init_hash: str¶
deployer: degenbot._ffi.ChecksummedAddress¶
name: str¶
type PoolState = UniswapV2PoolState¶
classmethod from_handle(py_pool: degenbot.types.Pool) → Self¶

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.

  • 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).

property update_block: degenbot.types.aliases.BlockNumber¶

Update block.

Returns:

The block number of the most recent state update (from Rust).

property reserves_token0: int¶

Reserves token0.

Returns:

The reserve amount for token0 (from Rust).

property reserves_token1: int¶

Reserves token1.

Returns:

The reserve amount for token1 (from Rust).

property state: 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).

external_update(update: degenbot.uniswap.v2_types.UniswapV2PoolExternalUpdate) → None¶

External update.

Raises:

ExternalUpdateError – If the update is for a past block.

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).

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).

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.

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.

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.

class degenbot.uniswap.UniswapV2PoolExternalUpdate¶

UniswapV2PoolExternalUpdate class.

block_number: degenbot.types.aliases.BlockNumber¶
reserves_token0: int¶
reserves_token1: int¶
class degenbot.uniswap.UniswapV2PoolSimulationResult¶

Bases: UniswapSimulationResult

UniswapV2PoolSimulationResult class.

initial_state: UniswapV2PoolState¶
final_state: UniswapV2PoolState¶
class degenbot.uniswap.UniswapV2PoolState¶

Bases: degenbot.types.abstract.AbstractPoolState

UniswapV2PoolState class.

reserves_token0: int¶
reserves_token1: int¶
class degenbot.uniswap.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.

class degenbot.uniswap.UniswapV3PoolExternalUpdate¶

UniswapV3PoolExternalUpdate class.

block_number: degenbot.types.aliases.BlockNumber¶
liquidity: Liquidity¶
sqrt_price_x96: SqrtPriceX96¶
tick: Tick¶
class degenbot.uniswap.UniswapV3PoolSimulationResult¶

Bases: UniswapSimulationResult

UniswapV3PoolSimulationResult class.

initial_state: UniswapV3PoolState¶
final_state: UniswapV3PoolState¶
class degenbot.uniswap.UniswapV3PoolState¶

Bases: degenbot.types.abstract.AbstractPoolState

UniswapV3PoolState class.

liquidity: Liquidity¶
sqrt_price_x96: SqrtPriceX96¶
tick: Tick¶
tick_bitmap: dict[BitmapWord, degenbot.uniswap.concentrated.types.BitmapAtWord]¶
tick_data: dict[Tick, degenbot.uniswap.concentrated.types.LiquidityAtTick]¶
class degenbot.uniswap.UniswapV4Pool(*args: Any, **kwargs: Any)¶

Bases: degenbot.uniswap.v4_pool_state.V4PoolState, degenbot.uniswap.v4_pool_calc.UniswapV4PoolCalc, 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.

type PoolState = UniswapV4PoolState¶
hook_address: degenbot.types.chain.ChecksummedAddress¶
active_hooks: frozenset[Hooks]¶
name: str¶
protocol_fee: ProtocolFee¶
lp_fee: int¶
classmethod from_handle(py_pool: degenbot.types.Pool) → Self¶

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.

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:
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:
property address: degenbot.types.chain.ChecksummedAddress¶

Address.

Returns:

The pool manager address.

property pool_id: bytes¶

Pool id.

Returns:

The pool ID bytes.

property pool_key: degenbot.uniswap.v4_types.UniswapV4PoolKey¶

Pool key.

Returns:

The V4 pool key struct.

property sqrt_price_x96: int¶

Sqrt price x96.

Returns:

The current sqrt price as a Q64.96 value (from Rust).

property state: 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.

property tick_spacing: int¶

Tick spacing.

Returns:

The tick spacing for the pool (Python-side identity).

property fee: int¶

Fee.

Returns:

The fee in pips (Python-side identity).

class degenbot.uniswap.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.

class degenbot.uniswap.UniswapV4PoolExternalUpdate¶

UniswapV4PoolExternalUpdate class.

block_number: degenbot.types.aliases.BlockNumber¶
liquidity: degenbot.uniswap.v3_types.Liquidity¶
sqrt_price_x96: degenbot.uniswap.v3_types.SqrtPriceX96¶
tick: degenbot.uniswap.v3_types.Tick¶
class degenbot.uniswap.UniswapV4PoolState¶

Bases: degenbot.types.abstract.AbstractPoolState

UniswapV4PoolState class.

liquidity: degenbot.uniswap.v3_types.Liquidity¶
sqrt_price_x96: degenbot.uniswap.v3_types.SqrtPriceX96¶
tick: degenbot.uniswap.v3_types.Tick¶
tick_bitmap: dict[degenbot.uniswap.v3_types.BitmapWord, degenbot.uniswap.concentrated.types.BitmapAtWord]¶
tick_data: dict[degenbot.uniswap.v3_types.Tick, degenbot.uniswap.concentrated.types.LiquidityAtTick]¶
id: bytes¶
block: degenbot.types.aliases.BlockNumber | None¶