degenbot

degenbot: Ethereum DEX helper library.

Submodules

Package Contents

degenbot.logger
exception degenbot.AbiDecodeError(*, message: str | None = None)

Bases: degenbot.exceptions.base.DegenbotError

Raised when ABI decoding fails.

exception degenbot.AbiEncodeError(*, message: str | None = None)

Bases: degenbot.exceptions.base.DegenbotError

Raised when ABI encoding fails.

degenbot.abi_decode(types: collections.abc.Sequence[str], data: BytesLike) → tuple[Any, ...]

Decode ABI-encoded bytes into Python values.

Parameters:

  • types – ABI type strings (e.g., ["uint256", "address"]).

  • data – ABI-encoded bytes.

Returns:

Tuple of decoded values.

Raises:

AbiDecodeError – If decoding fails.

degenbot.abi_decode_single(abi_type: str, data: BytesLike) → Any

Decode a single ABI value.

Parameters:

  • abi_type – ABI type string (e.g., "uint256").

  • data – ABI-encoded bytes.

Returns:

The decoded value.

Raises:

AbiDecodeError – If decoding fails.

degenbot.abi_encode(types: collections.abc.Sequence[str], args: collections.abc.Sequence[Any]) → bytes

Encode values into ABI-encoded bytes.

Parameters:

  • types – ABI type strings (e.g., ["uint256", "address"]).

  • args – Values to encode.

Returns:

ABI-encoded bytes.

Raises:

AbiEncodeError – If encoding fails.

class degenbot.Bot(*, chain_id: int | None = None, node: str | None = None, database: str | None = None, provider: degenbot.provider.AlloyProvider | None = None, py_bot: degenbot._ffi.Bot | None = None, io: degenbot._ffi.BotIo | None = None, erc20_builder: degenbot.builders.erc20_builder.Erc20Builder | None = None, provider_factory: collections.abc.Callable[..., Any] | None = None)

Bases: degenbot.bot._account_queries.AccountQueryMixin

Explicit session object that owns the runtime state for a degenbot run.

Owns the per-session configuration, Rust database path, provider, registries, and engine handles instead of exposing module-level singletons.

Bot is: - Factory — creates pools/tokens via managers, doing all I/O to fetch data - Registry — tracks what it’s created - I/O boundary — all RPC calls and database access flow through Bot - Session — the lifetime scope for the entire run - Account queries — ERC-20/native balance, approval, and supply reads,

inherited from AccountQueryMixin so this class body stays under the public-method complexity bar; the facade surface is unchanged

database_path
managed_pools
pools
tokens
registration_fleet_hosted() → bool

Return the construction-time registration-intake stance (PRG-3/5).

Thin pass-through to the core: True once the engine has installed the fleet intake boot (the module-init stance holder is read at FFI init, the boot descriptor installs at engine construction). The crawl driver requires this hard-True from pipeline construction on (PRG-5: the legacy crawl shell is retired — no getattr-default fallback).

Returns:

True once the fleet intake boot is installed (stance = fleet).

submit_registration_unit(fn: object) → degenbot._ffi.IntakeReceipt

Submit ONE registration-intake unit for fleet-seat execution.

The receipt exposes done() / result() / wait() / wait_async(). The crawl’s per-path units (build + verify lifecycles + path registration) ride the fleet’s duty-counted PoolStateUpdater seats through this seam (PRG-5).

On a host whose fleet boot was refused (detected CPU budget below the pinned-role floor, or a boot invariant), this submit — and every submit after it — raises the typed, sticky BootRefused from the FFI seam: the library never aborts the host process on the boot-refusal arm, and nothing is enqueued into a pipe that will not be drained.

Returns:

The unit’s receipt (IntakeReceipt).

property chain_id: degenbot.types.aliases.ChainId

The single chain this Bot targets (ADR-006 D5).

property provider: degenbot.provider.AlloyProvider

The single RPC provider for this Bot’s chain.

add_tracker[M: degenbot.types.abstract.pool_tracker.AbstractPoolTracker[Any]](manager_cls: type[M], *, factory_address: str, **kwargs: Any) → M

Create a pool manager within this bot’s session.

Returns:

The computed value.

Raises:

TrackerAlreadyInitialized – See function documentation.

block_stream() → degenbot._ffi.BlockStream

Fresh async iterator over newHeads block notifications.

The settlement bot’s authoritative block clock: ticked once per accepted header by the pump — NOT derived from ResultBatch.solve_block, which lags by the send debounce.

ADR-027: the block-clock pipe is coordinator-owned, so this surfaces on the Bot (the pump-lifecycle handle) rather than on the arbitrage engine, which is out of the block path entirely. Once-only: a second call raises RuntimeError.

Returns:

Async iterator yielding one dict per accepted block header (number, timestamp, base_fee_per_gas, gas_used, gas_limit); ends when the pump stops.

Return type:

BlockStream

release_python_state() → None

Drop Python-side pool/token/tracker caches once Rust owns canonical state.

After the Rust engine has taken ownership of all pool state (snapshots streamed, pools registered, backfill complete), the Python-side tracker caches, snapshots, and pool/token registries are redundant — the hot loop only needs the engine and the async Alloy provider handle. This drops them so they stop pinning pool objects in memory.

Idempotent; safe to call once, at the end of the startup handshake (after build_paths completes). Concrete trackers that carry a snapshot (e.g. UniswapV3StateTB) have unload_snapshot() called to release the snapshot reference.

close() → None

Release all Python handles owned by this Bot session.

End-of-life teardown that composes release_python_state() and closes the provider connection and drops the Rust Bot engine / provider references. Idempotent — safe to call directly and again from a with block’s __exit__.

The Rust Bot engine is reference-counted; closing this Python wrapper only drops this Bot’s ref. A running engine that took its own ref (via EngineRegistry(bot=bot) → ArbitrageEngine(py_bot=...)) is unaffected.

For the mid-lifecycle “drop redundant Python caches while the Bot keeps running” handshake, call release_python_state() directly; close() is for end-of-life.

register_builder(pool_class: type[degenbot.types.abstract.liquidity_pool.AbstractLiquidityPool], builder: degenbot.builders.protocol.PoolBuilder) → None

Register a builder for a concrete pool type.

After registration, update() will use type(pool) dict lookup instead of isinstance chains to find the right builder.

Parameters:

  • pool_class – The concrete pool class (e.g. UniswapV2Pool, AerodromeV2Pool).

  • builder – The builder instance that handles construction and updates for this pool type.

build_erc20token(address: str, *, silent: bool = False) → degenbot.erc20.erc20.Erc20Token

Fetch token metadata from DB/RPC and construct an I/O-free Erc20Token.

Returns:

The computed value.

get_token(address: str) → degenbot.erc20.erc20.Erc20Token

Get or create a token. Bot handles DB lookup, RPC calls, and registration.

Returns:

The computed value.

build_pool(address: str, *, state_block: int | None = None, silent: bool = False, tick_bitmap: dict[int, Any] | None = None, tick_data: dict[int, Any] | None = None, construction_route: degenbot.builders.request.ConstructionRoute | None = None) → degenbot.types.abstract.liquidity_pool.AbstractLiquidityPool

Build a pool from an address, automatically resolving its type.

V4 managed pools should use build_managed_pool() instead.

construction_route is the resolved construction-route policy the core route entry walks for the V3 arm (factory rungs + the generic builder); None = the generic-only route. The route is a driver VALUE — the core owns the walk and classifies every failure on the build-refusal taxonomy (an unsupported family raises the typed UnsupportedPoolFamilyError — the loud-abort rule).

Returns:

The computed value.

Raises:

DegenbotValueError – See function documentation.

build_managed_pool(address: str, request: degenbot.builders.request.BuildManagedPoolRequest) → degenbot.uniswap.v4_liquidity_pool.UniswapV4Pool

Build a V4 managed pool from a PoolManager address and pool ID.

address is the PoolManager contract; request carries the pool ID plus everything the pre-fetched identity may need (silent/ state_block common options, the V4 immutable data, and pre-fetched tick data) — see BuildManagedPoolRequest for the per-field contract: when the pool is not in the database, state_view_address, tokens, fee, tick_spacing must all be provided.

Returns:

The computed value.

update(pool: degenbot.types.abstract.liquidity_pool.AbstractLiquidityPool, *, block_number: degenbot.types.rpc_types.BlockIdentifier | None = None) → bool

Fetch the current state of a pool from the chain and apply it via.

pool.external_update().

Returns True if the state changed, False if unchanged.

Returns:

The computed boolean value.

degenbot.get_checksum_address(address: degenbot.types.chain.HexAddress | bytes) → degenbot._ffi.ChecksummedAddress

Return checksum address.

Returns:

The computed value.

class degenbot.AerodromeV2Pool(*args: Any, **kwargs: Any)

Bases: degenbot.aerodrome.v2_pool_state.AerodromeV2PoolState, degenbot.aerodrome.v2_pool_calc.AerodromeV2PoolCalc, degenbot.types.abstract.AbstractLiquidityPool

AerodromeV2Pool class.

variant: ClassVar[str | None] = 'aerodrome'
type PoolState = AerodromeV2PoolState
FEE_DENOMINATOR = 10000
address: degenbot.types.chain.ChecksummedAddress
factory: degenbot.types.chain.ChecksummedAddress
deployer_address: degenbot.types.chain.ChecksummedAddress
name: str
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). Every identity field (address, tokens, factory, fee, stable, variant) is read off the handle; reserves live in Rust (AerodromeV2PoolState) and are read via snapshot_aerodrome() — the Python StateCache is gone.

Returns:

A cls instance wrapping py_pool.

Raises:

DegenbotValueError – If the handle is not an Aerodrome V2 pool.

property reserves_token0: int

Reserves token0.

property reserves_token1: int

Reserves token1.

property state: PoolState

State.

Raises:

DegenbotValueError – If the Rust snapshot is absent.

property update_block: degenbot.types.aliases.BlockNumber

Update block.

static swap_is_viable(state: PoolState, vector: degenbot.uniswap.types.UniswapPoolSwapVector) → bool

Swap is viable.

Returns:

The computed boolean value.

external_update(update: degenbot.aerodrome.types.AerodromeV2PoolExternalUpdate) → None

External update.

Raises:

ExternalUpdateError – See function documentation.

discard_states_before_block(block: degenbot.types.aliases.BlockNumber) → None

Discard cached states earlier than the given block.

Raises:

NoPoolStateAvailable – See function documentation.

restore_state_before_block(block: degenbot.types.aliases.BlockNumber) → None

Restore the last pool state recorded prior to a target block.

Use this method to maintain consistent state data following a chain re-organization.

Raises:

NoPoolStateAvailable – See function documentation.

class degenbot.AerodromeV2PoolState

Bases: degenbot.types.abstract.AbstractPoolState

AerodromeV2PoolState class.

reserves_token0: int
reserves_token1: int
class degenbot.AerodromeV2PoolTracker(*, factory_address: str, bot: degenbot.bot.Bot, chain_id: degenbot.types.aliases.ChainId | None = None, deployer_address: degenbot._ffi.ChecksummedAddress | str | None = None, pool_init_hash: str | None = None)

Bases: _AbstractAerodromeV2PoolTracker[degenbot.aerodrome.pools.AerodromeV2Pool]

Generate and track concrete instances of V2 liquidity pool helpers.

Tracks Uniswap V2 liquidity pool helpers or their child classes.

get_volatile_pool(token_addresses: tuple[str, str], *, silent: bool = False, pool_class_kwargs: dict[str, Any] | None = None) → degenbot.aerodrome.pools.AerodromeV2Pool

Get a volatile pool by its token addresses.

The token addresses may be passed in any order.

Returns:

The computed value.

class degenbot.AerodromeV3Pool(*args: Any, **kwargs: Any)

Bases: degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool

AerodromeV3Pool class.

variant: ClassVar[str | None] = 'aerodrome'
type PoolState = AerodromeV3PoolState
TICK_STRUCT_TYPES = ('uint128', 'int128', 'int128', 'uint256', 'uint256', 'uint256', 'int56', 'uint160', 'uint32', 'bool')
class degenbot.AerodromeV3PoolState

Bases: degenbot.uniswap.v3_types.UniswapV3PoolState

AerodromeV3PoolState class.

class degenbot.AerodromeV3PoolTracker(*args: Any, **kwargs: Any)

Bases: degenbot.uniswap.trackers.AbstractUniswapV3PoolTracker[degenbot.aerodrome.pools.AerodromeV3Pool]

AerodromeV3PoolTracker class.

abstractmethod get_pool_from_tokens_and_fee(*args: Any, **kwargs: Any) → Never

Return pool from tokens and fee.

get_pool_from_tokens_and_tick_spacing(token_addresses: tuple[str, str], tick_spacing: int, *, silent: bool = False, pool_class_kwargs: dict[str, Any] | None = None) → degenbot.aerodrome.pools.AerodromeV3Pool

Return pool from tokens and tick spacing.

Returns:

The computed value.

class degenbot.ChainlinkPriceContract(address: str, *, chain_id: degenbot.types.aliases.ChainId | None = None, decimals: int | None = None, bot: degenbot.bot.Bot | None = None)

Represents an on-chain Chainlink price oracle.

Constructed with the oracle address and chain_id. The price property fetches the current price and decimals on access.

The price property returns the decimal-corrected nominal token price in USD (e.g. 1 DAI = 1.0 USD).

This is a delegating shell over the Rust ChainlinkPriceFeed reader (the degenbot-price core crate, ADR-005). The inline provider.call_raw / abi_decode bodies are retired — eth_call + canonical ABI decode now run in Rust. The float price is computed here (display layer) from the raw int256 answer + decimals, preserving the prior float(answer / 10**decimals) behavior exactly.

address
property chain_id: degenbot.types.aliases.ChainId | None

Chain id.

property decimals: int

Decimals.

Raises:

ValueError – See function documentation.

property price: float

Price.

Raises:

ValueError – See function documentation.

class degenbot.CurveStableswapPool(*args: Any, **kwargs: Any)

Bases: degenbot.curve.stableswap_pool_state.StableswapPoolState, degenbot.types.abstract.AbstractLiquidityPool

CurveStableswapPool class.

type PoolState = CurveStableswapPoolState
PRECISION_DECIMALS: int = 18
PRECISION: int = 1000000000000000000
address: degenbot.types.chain.ChecksummedAddress
FEE_DENOMINATOR: int = 10000000000
A_PRECISION: int = 100
classmethod from_handle(py_pool: degenbot.types.Pool) → Self

Wrap a Rust-owned Pool handle as a Python companion.

Single-arg seam (ADR-005): reads every identity field + the stored data-provider trait object off the handle. The cross-pool references (base pool companion + underlying/LP tokens) are recovered from the handle too — the base pool via the Rust go-between curve_base_pool() (same shared BotState, no Python registry), wrapped in a _LazyBasePool that memoises construction.

Returns:

The companion wrapping the handle.

Raises:

DegenbotValueError – If the handle is not a Curve stableswap pool, or its tokens are not registered in the handle’s Bot.

property balances: tuple[int, ...]

Balances.

Read from the Rust core via the Pool handle (ADR-005 slice 11b). Rust BotState is the single source of truth for the mutable balances slot; this getter returns the live tuple.

property state: degenbot.curve.types.CurveStableswapPoolState

State.

Built from one atomic Rust snapshot (snapshot_curve() — (balances, block)) so callers see a coherent tuple (no torn read mid-external_update). Mirrors V3/V4’s snapshot_v3() contract.

Raises:

DegenbotValueError – If the Rust snapshot is absent (the pool is not registered in Rust as a Curve pool — unreachable for a companion built over a registered handle).

property update_block: degenbot.types.aliases.BlockNumber

Update block (from Rust via the handle).

property requires_io_at_calculation_time: bool

Whether this pool may call data_provider during swap calculations.

Returns True for pools that need per-block on-chain data (D, gamma, price_scale, lending rates, admin balances, virtual price for metapools, block timestamps for A ramping). Returns False only for plain pools with static rate multipliers and no A ramping.

external_update(update: degenbot.curve.types.CurveStableswapPoolExternalUpdate) → None

Apply an external state update with new balances.

Delegates to the Rust core (Pool.apply_curve_balance_update) which journals the prior balances (genesis-anchor V2-style discipline) and lands the new balances + update_block atomically (ADR-005 slice 11b). The StateCache temporal-navigation layer it used to write is gone — the Rust reorg journal handles rollback now.

Raises:

DegenbotValueError – If the Rust core rejects the update (the pool is not registered as a Curve pool — unreachable for a companion built over a registered handle).

calc_token_amount(*, amounts: collections.abc.Sequence[int], deposit: bool, block_identifier: degenbot.types.rpc_types.BlockIdentifier | None = None) → int

Simplified method to calculate addition or reduction in token supply at.

deposit or withdrawal without taking fees into account (but looking at slippage). Needed to prevent front-running, not for precise calculations!

Returns:

The computed integer value.

calc_withdraw_one_coin(_token_amount: int, i: int, block_identifier: degenbot.types.rpc_types.BlockIdentifier | None = None) → tuple[int, ...]

Calc withdraw one coin.

Returns:

The computed value.

get_dy(i: int, j: int, dx: int, block_identifier: degenbot.types.rpc_types.BlockIdentifier | None = None, override_state: degenbot.curve.types.CurveStableswapPoolState | None = None) → int

@notice Calculate the current output dy given input dx.

@dev Index values can be found via the coins public getter method @param i Index value for the coin to send @param j Index value of the coin to recieve @param dx Amount of i being exchanged @return Amount of j predicted.

Reference: https://github.com/curveresearch/notes/blob/main/stableswap.pdf

Delegates to the Rust-owned Pool.curve_get_dy: the I/O orchestration (amp/rates/xp + provider fetches) and the pure dy math both run in the Rust core, so this is a single handle call with no Python provider / cache / calculator on the swap path.

Returns:

The computed integer value.

Raises:

EVMRevertError – See function documentation.

calculate_tokens_out_from_tokens_in(token_in: degenbot.erc20.Erc20Token, token_out: degenbot.erc20.Erc20Token, token_in_quantity: int, override_state: degenbot.curve.types.CurveStableswapPoolState | None = None, block_identifier: degenbot.types.rpc_types.BlockIdentifier | None = None) → int

Calculate the expected token OUTPUT for a target INPUT at current pool reserves.

Returns:

The computed integer value.

Raises:
class degenbot.CurveStableswapPoolSimulationResult

CurveStableswapPoolSimulationResult class.

amount0_delta: int
amount1_delta: int
current_state: CurveStableswapPoolState
class degenbot.CurveStableswapPoolState

Bases: degenbot.types.abstract.AbstractPoolState

CurveStableswapPoolState class.

balances: tuple[int, ...]
base: CurveStableswapPoolState | None = None
class degenbot.Erc20Token(*args: Any, **kwargs: Any)

Bases: degenbot.types.abstract.AbstractErc20Token

An ERC-20 token contract.

Constructed from pre-fetched data only. Use Bot.build_erc20token() to fetch from chain. Balance, approval, and total supply queries go through Bot.get_token_balance() etc.

classmethod from_handle(py_token: degenbot._ffi.Erc20Token, *, oracle_address: str | None = None, state_cache_depth: int = 8) → Self

Wrap a Rust-owned _TokenHandle handle as a Python companion.

Internal seam (ADR-005, Polars-style _from_pydf pattern). Rust owns the token metadata (address, name, symbol, decimals, chain_id) as a TokenEntry; this companion reads it through self._py_token on every access and holds no metadata copy. Price oracle + balance/approval/total-supply caches stay Python (I/O constructs that cannot move to Rust).

Only Bot.get_token() / Bot.build_erc20token() (production) and make_erc20 (tests) should call this — they have already registered the token metadata in a Bot and obtained the handle. cls is used so subclasses that only set ClassVars inherit this seam and produce instances of the subclass.

Returns:

A cls instance wrapping py_token.

property address: degenbot.types.chain.ChecksummedAddress

Token contract address (EIP-55 checksum).

Rust holds the address bytes; get_checksum_address applies the codebase-wide EIP-55 display convention.

property name: str

Token name (read from Rust-owned TokenEntry).

property symbol: str

Token symbol (read from Rust-owned TokenEntry).

property decimals: int

Token decimals (read from Rust-owned TokenEntry).

get_cached_balance(address: degenbot.types.chain.ChecksummedAddress, block_number: int) → int | None

Return cached balance.

Returns:

The computed value.

set_cached_balance(address: degenbot.types.chain.ChecksummedAddress, block_number: int, balance: int) → None

Set cached balance.

get_cached_approval(block_number: int, owner: degenbot.types.chain.ChecksummedAddress, spender: degenbot.types.chain.ChecksummedAddress) → int | None

Return cached approval.

Returns:

The computed value.

set_cached_approval(block_number: int, owner: degenbot.types.chain.ChecksummedAddress, spender: degenbot.types.chain.ChecksummedAddress, amount: int) → None

Set cached approval.

get_cached_total_supply(block_number: int) → int | None

Return cached total supply.

Returns:

The computed value.

set_cached_total_supply(block_number: int, total_supply: int) → None

Set cached total supply.

property price: float

Price.

Raises:

NoPriceOracle – See function documentation.

property chain_id: int

Chain ID (read from Rust-owned TokenEntry).

class degenbot.EtherPlaceholder

Bases: degenbot.erc20.Erc20Token

An Erc20Token-like adapter for the ‘all Es’ or zero address placeholder.

Used by pools to represent native Ether. Under ADR-005, metadata (name=”Ether Placeholder”, symbol=”ETH”, decimals=18) is registered in the Rust Bot when an EtherPlaceholder is built; the inherited delegating properties read it back through the Rust Erc20Token handle.

Direct construction is forbidden (inherited from Erc20Token). Use Erc20Token.from_handle() — which produces an EtherPlaceholder instance when called on the subclass — after registering the metadata in a Bot.

addresses
class degenbot.AnvilFork(*, fork_url: str | None = None, fork_block: degenbot.types.aliases.BlockNumber | None = None, launch: ForkLaunchConfig | None = None)

A rust-owned Anvil fork subprocess + IPC DynProvider + dev-RPC surface.

Companion shell (ADR-005 Python layer) over degenbot._ffi.AnvilFork (the PyO3 seam on degenbot_fork::AnvilFork). The rust core owns the subprocess lifecycle (drop = kill) + an alloy DynProvider (IPC transport) + the 12 anvil dev-RPC methods. The shell carries the legacy constructor knobs in ForkLaunchConfig, preserves the legacy dev-method names, raises the legacy AnvilError exception subclass on rust-side RPC failures, and exposes general RPC through the AlloyProvider pyclass at provider (replacing the legacy self.w3 Web3 handle — that handle is gone; all callers now use fork.provider.get_block_number() etc.).

The rust core (degenbot_fork::AnvilFork) owns:

  • the anvil subprocess lifecycle (drop = kill);

  • a connected alloy DynProvider over IPC (transport);

  • the 12 dev-RPC methods (evm_mine/anvil_reset/evm_snapshot/ evm_revert/anvil_set_balance/…), exposed on this shell as the legacy Python names (mine/reset/set_snapshot/ return_to_snapshot/set_balance/…).

The Python shell keeps:

  • the legacy constructor knobs (grouped in ForkLaunchConfig) + the multi-override queueing (balance_overrides / bytecode_overrides / nonce_overrides / storage_overrides / coinbase) — applied post-spawn via the dev-method delegations;

  • the legacy Python dev-method names so existing API surface doesn’t reshuffle (the rust core’s name for set_snapshot is snapshot, return_to_snapshot → revert, set_storage → set_storage_at, set_next_base_fee → set_next_base_fee — these translations happen in-shell);

  • the legacy AnvilError exception on rust RPC failures (translated from rust’s PyRuntimeError at the boundary, so the legacy AnvilError(method=..., error=...) shape + the exception-subclass identity stay intact);

  • the @validate_call + ValidatedUint256 annotations on set_balance / set_next_base_fee / set_next_block_timestamp so input-range validation raises pydantic ValidationError before the rust round-trip (matches the legacy pre-0.7 surface).

Dropped (no longer needed now that subprocess lifecycle is rust-owned): the –auto-impersonate CLI arg list construction, the _setup_process / _setup_w3 / _close_ipc_socket block, the socket-based free-port allocation, the tenacity spawn retry, the subprocess.Popen capture-file management (capture_path / preserve_capture / *_capture_filename properties), the web3.IPCProvider client + middleware_onion.inject(middleware) plumbing (alloy’s DynProvider is the only transport layer under the rust core), the ipc_provider_kwargs shim, and http_url / ws_url / port / ipc_filename (anvil-CLI control knobs that the rust core controls directly now — the IPC path is the only socket surfaced back, via ipc_path).

The one breaking change for callers: self.w3 is gone. Use provider (a degenbot._ffi.AlloyProvider) for general RPC.

property provider: degenbot.provider.AlloyProvider

The general-RPC AlloyProvider bound to the fork over HTTP.

Use for retry-aware eth_call / get_logs against the in-memory fork (mirrors the legacy self.w3 handle). HTTP (not the rust core’s IPC DynProvider) so that consumers which borrow it never hold an alloy IPC/WS pubsub that would reconnect into a dead socket when the fork closes. Raising once the fork is closed guarantees a provider over a dead subprocess is never silently handed out.

Raises:

AnvilError – If the fork was closed.

property fork_url: str | None

Fork URL the spawned anvil process forks from, or None for in-memory.

property ipc_path: str

Resolved IPC socket path the spawned anvil process listens on.

Use to construct a second AlloyProvider (or other transport) against the same forked node. Replaces the legacy AnvilFork.ipc_filename (a pathlib.Path derived from a user-supplied ipc_path parent + the allocated HTTP port — that bookkeeping is gone; the rust core owns the IPC socket directly + reports the resolved path here).

Raises:

AnvilError – If the fork was closed.

property http_url: str

HTTP endpoint of the spawned anvil node (e.g. http://127.0.0.1:PORT).

Use to construct a second AlloyProvider over the rust-owned anvil subprocess when the IPC-bound provider is not enough (e.g. a plain HTTP transport from another Python process).

property ws_url: str

WebSocket endpoint of the spawned anvil node (e.g. ws://127.0.0.1:PORT).

property port: int

TCP port the spawned anvil node listens on.

close() → None

Drop the rust-owned subprocess + IPC handle.

The rust core (degenbot_fork::AnvilFork) owns the subprocess via alloy’s AnvilInstance; dropping the _FFIAnvilFork pyclass handle terminates the anvil process + closes the IPC transport. This method just clears the Python-side reference so the GC finalizes the rust handle — there is no separate IPC-socket cleanup needed (alloy’s AnvilInstance Drop impl handles it).

mine() → None

Mine a single block (evm_mine).

Raises:

AnvilError – If the RPC call fails.

reset(fork_url: str | None = None, block_number: degenbot.types.aliases.BlockNumber | None = None, transaction_hash: str | None = None) → None

Fork from a new endpoint, block number, or transaction hash.

Resetting to a new block number only is done in-place (no subprocess relaunch). Resetting to a new endpoint or from a transaction hash tears down the existing rust-owned subprocess + spawns a new one with the new fork parameters. Slower than in-place.

Raises:
set_balance(address: str, balance: degenbot.validation.evm_values.ValidatedUint256) → None

Set balance (anvil_setBalance).

Raises:

AnvilError – If the RPC call fails.

set_code(address: str, bytecode: bytes) → None

Set code (anvil_setCode).

Raises:

AnvilError – If the RPC call fails.

set_coinbase(address: str) → None

Set coinbase (anvil_setCoinbase).

Raises:

AnvilError – If the RPC call fails.

set_block_timestamp_interval(interval: int) → None

Set block timestamp interval (anvil_setBlockTimestampInterval).

Raises:

AnvilError – If the RPC call fails.

set_next_base_fee(fee: degenbot.validation.evm_values.ValidatedUint256) → None

Set the next block base fee (anvil_setNextBlockBaseFeePerGas).

Raises:

AnvilError – If the RPC call fails.

set_next_block_timestamp(timestamp: degenbot.validation.evm_values.ValidatedUint256) → None

Set the next block timestamp (evm_setNextBlockTimestamp).

Raises:

AnvilError – If the RPC call fails.

set_nonce(address: str, nonce: int) → None

Set nonce (anvil_setNonce).

Raises:

AnvilError – If the RPC call fails.

set_storage(address: degenbot.types.chain.HexAddress | bytes, position: int, value: degenbot.types.chain.HexStr | bytes | int) → None

Set storage (anvil_setStorageAt).

Raises:

AnvilError – If the RPC call fails.

class degenbot.PancakeswapV3Pool

Bases: degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool

PancakeswapV3Pool class.

variant: ClassVar[str | None] = 'pancakeswap'
class degenbot.PancakeswapV3PoolTracker

Bases: degenbot.uniswap.trackers.UniswapV3PoolTracker

Track PancakeSwap V3 pool events.

class degenbot.ManagedPoolRegistry(*, py_bot: degenbot._ffi.Bot)

V4 pool companions, keyed by the session’s (PoolManager, pool_id) identities.

A V4 pool is named by its pair, never by the PoolManager address alone, so every read here carries both. This is the registry Bot.managed_pools and the one PoolRegistry delegates its V4 branch to, so the two are one companion store for the session’s V4 pools.

get(chain_id: degenbot.types.aliases.ChainId, pool_manager_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId) → degenbot.types.pool_protocols.ConcentratedLiquidityPool | None

Retrieve a V4 pool by chain, manager address, and pool ID.

Returns:

The registered V4 pool, or None if not found.

add(pool: degenbot.types.pool_protocols.ConcentratedLiquidityPool, chain_id: degenbot.types.aliases.ChainId, pool_manager_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId) → None

Register a V4 pool.

Raises:

DegenbotValueError – A companion is already registered for this identity.

get_or_add(pool: degenbot.types.pool_protocols.ConcentratedLiquidityPool, chain_id: degenbot.types.aliases.ChainId, pool_manager_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId) → degenbot.types.pool_protocols.ConcentratedLiquidityPool

Idempotently register a V4 pool, returning the stored instance.

If a concurrent registration worker already built this pool, return the canonical stored instance instead of raising — a distinct path sharing this pool is not lossily skipped.

Returns:

The stored pool instance (the existing canonical one on a duplicate).

remove(chain_id: degenbot.types.aliases.ChainId, pool_manager_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId) → None

Drop the V4 companion for this identity.

The session’s identity for the pool is session-lifetime and stays; only the Python companion goes.

list_all() → collections.abc.Iterator[degenbot.types.pool_protocols.ConcentratedLiquidityPool]

Yield every registered V4 pool.

Yields:

Each V4 pool companion filed in this registry.

reset() → None

Drop every V4 companion. The session’s V4 identities are untouched.

class degenbot.PoolRegistry(*, py_bot: degenbot._ffi.Bot, managed_pool_registry: ManagedPoolRegistry | None = None)

Address-keyed pool companions, delegating identity to the session registry.

The non-V4 families are address-keyed, which is why reads here are family-agnostic: the session resolves the address to whichever family registered it first. Registering a pool, by contrast, names the family — read off the companion’s live handle — so a V3 pool and a Balancer pool at one address stay two identities, as they are in the core.

get(chain_id: degenbot.types.aliases.ChainId, pool_address: degenbot.types.chain.ChecksummedAddress, pool_id: None = None) → degenbot.types.abstract.AbstractLiquidityPool | None
get(chain_id: degenbot.types.aliases.ChainId, pool_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId) → degenbot.types.pool_protocols.ConcentratedLiquidityPool | None

Retrieve a pool by chain and address.

Returns:

The registered pool, or None if not found.

add(pool: degenbot.types.abstract.AbstractLiquidityPool, chain_id: degenbot.types.aliases.ChainId, pool_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId | None = None) → None

Register a pool.

When pool_id is provided, the pool must satisfy the ConcentratedLiquidityPool protocol and is registered in the managed pool sub-registry. Otherwise, it is registered as a standard pool.

Raises:

  • TypeError – If pool_id is provided but pool does not satisfy ConcentratedLiquidityPool.

  • DegenbotValueError – A companion is already registered for this identity.

get_or_add(pool: degenbot.types.abstract.AbstractLiquidityPool, chain_id: degenbot.types.aliases.ChainId, pool_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId | None = None) → degenbot.types.abstract.AbstractLiquidityPool | degenbot.types.pool_protocols.ConcentratedLiquidityPool

Idempotently register a pool, returning the stored instance.

Used by the concurrent registration build path: if another worker already built this pool, return the canonical stored instance instead of raising, so a distinct path sharing the pool is not lossily skipped. Mirrors add()’s managed/V4 dispatch.

Returns:

The stored pool instance (the existing canonical one on a duplicate).

Raises:

TypeError – If pool_id is provided but pool does not satisfy ConcentratedLiquidityPool.

remove(chain_id: degenbot.types.aliases.ChainId, pool_address: degenbot.types.chain.ChecksummedAddress, pool_id: PoolId) → None
remove(chain_id: degenbot.types.aliases.ChainId, pool_address: degenbot.types.chain.ChecksummedAddress, pool_id: None = None) → None

Remove a pool.

For V2/V3 pools (pool_id is None), propagates to the Rust BotState via py_bot.unregister_pool so the Rust-owned state stays symmetric with the Python registry (ADR-007). V4 pools (pool_id is bytes) are Python-only here — V4 unregister is engine-side (see ADR-007 Deferred).

The removal is of the companion and the live pool state; the session’s identity for the pool is session-lifetime and is not withdrawn, so a later build of the same address re-files a companion under the same canonical identity.

list_all() → collections.abc.Iterator[degenbot.types.abstract.AbstractLiquidityPool]

Yield every registered address-keyed pool.

Yields:

Each address-keyed pool companion filed in this registry.

reset() → None

Drop every companion, V4 included. The session’s identities are untouched.

class degenbot.PoolTypeRegistry

Unified registry mapping (chain_id, factory_address) → pool type identity.

Each DEX module registers its pool subclass at import time via register(). Builders consult this registry to select the concrete class and its deployment data.

Family, variant, and kind are auto-derived from the class hierarchy and the class’s variant attribute.

Public API for external callers

Library users who want to register a custom DEX pool class should:

  1. Subclass a pool shape protocol (ConstantProductPool, ConcentratedLiquidityPool, or StableswapPool). This is typically done by inheriting from an existing pool class that already satisfies the protocol (e.g., UniswapV2Pool).

  2. Add a variant: ClassVar[str | None] = "your_dex_name" class attribute. Use the bare DEX name without a _v2/_v3 suffix — the suffix is derived automatically from the family.

  3. If the pool has a non-standard constructor (e.g. requires chain fetches for extra parameters), add a builder method (e.g. _build_aerodrome_v2, _build_camelot) on the appropriate pool builder class. The builder will be dispatched via issubclass checks in the build() method.

  4. Call pool_type_registry.register() with the class, chain ID, factory address, and optional deployment data.

Example:

from degenbot.registry import pool_type_registry
from degenbot.uniswap.v2_liquidity_pool import UniswapV2Pool


class MyCustomPool(UniswapV2Pool):
    variant: ClassVar[str | None] = "my_dex"


pool_type_registry.register(
    PoolRegistration(
        pool_class=MyCustomPool,
        chain_id=1,
        factory_address="0x...",
        pool_init_hash="0x...",
    )
)

After registration, Bot.build_pool() will automatically select MyCustomPool for any pool whose factory() returns the registered address on the given chain.

register(registration: PoolRegistration) → None

Register a pool class for a specific (chain_id, factory) deployment.

Identity (family, variant, kind) is auto-derived from the class unless overridden on the request. Deployment data (chain_id, factory, deployer, init_hash) is stored alongside for lookup.

Parameters:

registration – The typed registration request; see PoolRegistration for field semantics.

Raises:

ValueError – If the factory is already registered for the given chain.

unregister(*, chain_id: degenbot.types.aliases.ChainId, factory_address: str) → None

Remove a previously-registered (chain_id, factory) entry.

Used by tests to clean up after calling register() so the module-level singleton is not permanently polluted.

Raises:

KeyError – If no registration exists for the given key.

set_default_v2_class(pool_class: type[degenbot.types.pool_protocols.ConstantProductPool]) → None

Set the default V2 pool class when no factory-specific mapping exists.

set_default_v3_class(pool_class: type[degenbot.types.pool_protocols.ConcentratedLiquidityPool]) → None

Set the default V3 pool class when no factory-specific mapping exists.

has_registration(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → bool

Whether a pool class is registered for (chain_id, factory).

Returns:

True if a registration exists, False otherwise.

get_class(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → type[degenbot.types.abstract.liquidity_pool.AbstractLiquidityPool] | None

Get the pool class for (chain_id, factory).

Returns None if no specific registration exists and no default is set.

Returns:

The pool class, or None if not found.

get_v2_class(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → type[degenbot.types.pool_protocols.ConstantProductPool] | None

Get the V2 pool class for (chain_id, factory), with default fallback.

Returns:

The V2 pool class, or None if not found.

get_v3_class(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → type[degenbot.types.pool_protocols.ConcentratedLiquidityPool] | None

Get the V3 pool class for (chain_id, factory), with default fallback.

Returns:

The V3 pool class, or None if not found.

get_descriptor(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → degenbot.types.pool_type.PoolTypeDescriptor | None

Get the PoolTypeDescriptor for (chain_id, factory).

Returns:

The descriptor, or None if not found.

get_v2_identity(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → degenbot.types.DexIdentity | None

Get the DexIdentity preset for (chain_id, factory), or None.

Returns None if no registration exists OR if the registration was made without a dex_identity (e.g. Aerodrome V2 — deferred per TODO-e30504ed).

Returns:

The DexIdentity preset, or None if not found / not set.

get_deployment(chain_id: degenbot.types.aliases.ChainId, factory_address: str) → PoolDeploymentData | None

Get the deployment data for (chain_id, factory).

Returns:

The deployment data, or None if not found.

get_descriptor_by_kind(kind: str) → degenbot.types.pool_type.PoolTypeDescriptor | None

Get a PoolTypeDescriptor by its kind string.

Used for DB lookups where the kind is known but the factory address is not. When multiple deployments share a kind, returns the descriptor from the last registration.

Returns:

The descriptor, or None if not found.

property registrations: dict[tuple[degenbot.types.aliases.ChainId, str], tuple[type[degenbot.types.abstract.liquidity_pool.AbstractLiquidityPool], degenbot.types.pool_type.PoolTypeDescriptor, PoolDeploymentData]]

A copy of all registrations.

class degenbot.TokenRegistry(*, py_bot: degenbot._ffi.Bot)

ERC-20 token companions, keyed by the session’s token identities.

get(token_address: str, chain_id: degenbot.types.aliases.ChainId) → degenbot.erc20.erc20.Erc20Token | None

Retrieve a token by chain and address.

Returns:

The registered token, or None if not found.

add(token_address: str, chain_id: degenbot.types.aliases.ChainId, token: degenbot.erc20.erc20.Erc20Token) → None

Register a token.

Raises:

DegenbotValueError – A companion is already registered for this token identity.

get_or_add(token_address: str, chain_id: degenbot.types.aliases.ChainId, token: degenbot.erc20.erc20.Erc20Token) → degenbot.erc20.erc20.Erc20Token

Idempotently register a token, returning the stored instance.

If a concurrent registration worker already built this token, return the canonical stored instance instead of raising — a distinct path sharing this token is not lossily skipped.

Returns:

The stored token instance (the existing canonical one on a duplicate).

remove(token_address: str, chain_id: degenbot.types.aliases.ChainId) → None

Drop the companion for this token.

The session’s identity for the token is session-lifetime and stays, so a later build of the same address re-files a companion under the same canonical identity. No live token state is touched: the Rust BotState token entry belongs to register_token / build_erc20_token.

list_all() → collections.abc.Iterator[degenbot.erc20.erc20.Erc20Token]

Yield every registered token.

Yields:

Each token companion filed in this registry.

reset() → None

Drop every token companion. The session’s identities are untouched.

degenbot.pool_type_registry
class degenbot.SushiswapV3PoolTracker(factory_address: degenbot.types.chain.ChecksummedAddress | str, bot: degenbot.bot.Bot, *, chain_id: degenbot.types.aliases.ChainId | None = None, overrides: DeploymentOverrides | None = None, snapshot: degenbot.uniswap.v3_snapshot.UniswapV3LiquiditySnapshot | None = None)

Bases: degenbot.uniswap.trackers.UniswapV3PoolTracker

SushiswapV3PoolTracker class.

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

UniswapV2PoolExternalUpdate class.

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

Bases: UniswapSimulationResult

UniswapV2PoolSimulationResult class.

initial_state: UniswapV2PoolState
final_state: UniswapV2PoolState
class degenbot.UniswapV2PoolState

Bases: degenbot.types.abstract.AbstractPoolState

UniswapV2PoolState class.

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

UniswapV3PoolExternalUpdate class.

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

Bases: UniswapSimulationResult

UniswapV3PoolSimulationResult class.

initial_state: UniswapV3PoolState
final_state: UniswapV3PoolState
class degenbot.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.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.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.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.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.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