degenbot ======== .. py:module:: degenbot .. autoapi-nested-parse:: degenbot: Ethereum DEX helper library. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/degenbot/aave/index /autoapi/degenbot/abi/index /autoapi/degenbot/aerodrome/index /autoapi/degenbot/arbitrage/index /autoapi/degenbot/balancer/index /autoapi/degenbot/bot/index /autoapi/degenbot/bot_lifecycle/index /autoapi/degenbot/build_info/index /autoapi/degenbot/builders/index /autoapi/degenbot/calculations/index /autoapi/degenbot/camelot/index /autoapi/degenbot/chainlink/index /autoapi/degenbot/checksum_cache/index /autoapi/degenbot/config/index /autoapi/degenbot/constants/index /autoapi/degenbot/contract/index /autoapi/degenbot/crypto/index /autoapi/degenbot/curve/index /autoapi/degenbot/db/index /autoapi/degenbot/diagnostics/index /autoapi/degenbot/dispatch/index /autoapi/degenbot/erc20/index /autoapi/degenbot/exceptions/index /autoapi/degenbot/fleet/index /autoapi/degenbot/fork/index /autoapi/degenbot/logging/index /autoapi/degenbot/operator/index /autoapi/degenbot/pancakeswap/index /autoapi/degenbot/pathfinding/index /autoapi/degenbot/provider/index /autoapi/degenbot/registry/index /autoapi/degenbot/runner/index /autoapi/degenbot/runtime_status/index /autoapi/degenbot/strategy/index /autoapi/degenbot/sushiswap/index /autoapi/degenbot/telemetry/index /autoapi/degenbot/types/index /autoapi/degenbot/uniswap/index /autoapi/degenbot/updater/index /autoapi/degenbot/utils/index /autoapi/degenbot/validation/index /autoapi/degenbot/version/index Package Contents ---------------- .. py:data:: logger .. py:exception:: AbiDecodeError(*, message: str | None = None) Bases: :py:obj:`degenbot.exceptions.base.DegenbotError` Raised when ABI decoding fails. .. py:exception:: AbiEncodeError(*, message: str | None = None) Bases: :py:obj:`degenbot.exceptions.base.DegenbotError` Raised when ABI encoding fails. .. py:function:: abi_decode(types: collections.abc.Sequence[str], data: BytesLike) -> tuple[Any, ...] Decode ABI-encoded bytes into Python values. :param types: ABI type strings (e.g., ``["uint256", "address"]``). :param data: ABI-encoded bytes. :returns: Tuple of decoded values. :raises AbiDecodeError: If decoding fails. .. py:function:: abi_decode_single(abi_type: str, data: BytesLike) -> Any Decode a single ABI value. :param abi_type: ABI type string (e.g., ``"uint256"``). :param data: ABI-encoded bytes. :returns: The decoded value. :raises AbiDecodeError: If decoding fails. .. py:function:: abi_encode(types: collections.abc.Sequence[str], args: collections.abc.Sequence[Any]) -> bytes Encode values into ABI-encoded bytes. :param types: ABI type strings (e.g., ``["uint256", "address"]``). :param args: Values to encode. :returns: ABI-encoded bytes. :raises AbiEncodeError: If encoding fails. .. py:class:: 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: :py:obj:`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 .. py:attribute:: database_path .. py:attribute:: managed_pools .. py:attribute:: pools .. py:attribute:: tokens .. py:method:: 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). .. py:method:: 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``). .. py:property:: chain_id :type: degenbot.types.aliases.ChainId The single chain this Bot targets (ADR-006 D5). .. py:property:: provider :type: degenbot.provider.AlloyProvider The single RPC provider for this Bot's chain. .. py:method:: 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. .. py:method:: 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. :rtype: BlockStream .. py:method:: 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. .. py:method:: close() -> None Release all Python handles owned by this Bot session. End-of-life teardown that composes :meth:`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 :meth:`release_python_state` directly; ``close()`` is for end-of-life. .. py:method:: 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. :param pool_class: The concrete pool class (e.g. UniswapV2Pool, AerodromeV2Pool). :param builder: The builder instance that handles construction and updates for this pool type. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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 :class:`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. .. py:method:: 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. .. py:function:: get_checksum_address(address: degenbot.types.chain.HexAddress | bytes) -> degenbot._ffi.ChecksummedAddress Return checksum address. :returns: The computed value. .. py:class:: AerodromeV2Pool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.aerodrome.v2_pool_state.AerodromeV2PoolState`, :py:obj:`degenbot.aerodrome.v2_pool_calc.AerodromeV2PoolCalc`, :py:obj:`degenbot.types.abstract.AbstractLiquidityPool` AerodromeV2Pool class. .. py:attribute:: variant :type: ClassVar[str | None] :value: 'aerodrome' .. py:type:: PoolState :canonical: AerodromeV2PoolState .. py:attribute:: FEE_DENOMINATOR :value: 10000 .. py:attribute:: address :type: degenbot.types.chain.ChecksummedAddress .. py:attribute:: factory :type: degenbot.types.chain.ChecksummedAddress .. py:attribute:: deployer_address :type: degenbot.types.chain.ChecksummedAddress .. py:attribute:: name :type: str .. py:method:: from_handle(py_pool: degenbot.types.Pool) -> Self :classmethod: 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. .. py:property:: reserves_token0 :type: int Reserves token0. .. py:property:: reserves_token1 :type: int Reserves token1. .. py:property:: state :type: PoolState State. :raises DegenbotValueError: If the Rust snapshot is absent. .. py:property:: update_block :type: degenbot.types.aliases.BlockNumber Update block. .. py:method:: swap_is_viable(state: PoolState, vector: degenbot.uniswap.types.UniswapPoolSwapVector) -> bool :staticmethod: Swap is viable. :returns: The computed boolean value. .. py:method:: external_update(update: degenbot.aerodrome.types.AerodromeV2PoolExternalUpdate) -> None External update. :raises ExternalUpdateError: See function documentation. .. py:method:: discard_states_before_block(block: degenbot.types.aliases.BlockNumber) -> None Discard cached states earlier than the given block. :raises NoPoolStateAvailable: See function documentation. .. py:method:: 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. .. py:class:: AerodromeV2PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` AerodromeV2PoolState class. .. py:attribute:: reserves_token0 :type: int .. py:attribute:: reserves_token1 :type: int .. py:class:: 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: :py:obj:`_AbstractAerodromeV2PoolTracker`\ [\ :py:obj:`degenbot.aerodrome.pools.AerodromeV2Pool`\ ] Generate and track concrete instances of V2 liquidity pool helpers. Tracks Uniswap V2 liquidity pool helpers or their child classes. .. py:method:: 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. .. py:class:: AerodromeV3Pool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool` AerodromeV3Pool class. .. py:attribute:: variant :type: ClassVar[str | None] :value: 'aerodrome' .. py:type:: PoolState :canonical: AerodromeV3PoolState .. py:attribute:: TICK_STRUCT_TYPES :value: ('uint128', 'int128', 'int128', 'uint256', 'uint256', 'uint256', 'int56', 'uint160', 'uint32', 'bool') .. py:class:: AerodromeV3PoolState Bases: :py:obj:`degenbot.uniswap.v3_types.UniswapV3PoolState` AerodromeV3PoolState class. .. py:class:: AerodromeV3PoolTracker(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.uniswap.trackers.AbstractUniswapV3PoolTracker`\ [\ :py:obj:`degenbot.aerodrome.pools.AerodromeV3Pool`\ ] AerodromeV3PoolTracker class. .. py:method:: get_pool_from_tokens_and_fee(*args: Any, **kwargs: Any) -> Never :abstractmethod: Return pool from tokens and fee. .. py:method:: 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. .. py:class:: 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. .. py:attribute:: address .. py:property:: chain_id :type: degenbot.types.aliases.ChainId | None Chain id. .. py:property:: decimals :type: int Decimals. :raises ValueError: See function documentation. .. py:property:: price :type: float Price. :raises ValueError: See function documentation. .. py:class:: CurveStableswapPool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.curve.stableswap_pool_state.StableswapPoolState`, :py:obj:`degenbot.types.abstract.AbstractLiquidityPool` CurveStableswapPool class. .. py:type:: PoolState :canonical: CurveStableswapPoolState .. py:attribute:: PRECISION_DECIMALS :type: int :value: 18 .. py:attribute:: PRECISION :type: int :value: 1000000000000000000 .. py:attribute:: address :type: degenbot.types.chain.ChecksummedAddress .. py:attribute:: FEE_DENOMINATOR :type: int :value: 10000000000 .. py:attribute:: A_PRECISION :type: int :value: 100 .. py:method:: from_handle(py_pool: degenbot.types.Pool) -> Self :classmethod: 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 :class:`_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. .. py:property:: balances :type: 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. .. py:property:: state :type: 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). .. py:property:: update_block :type: degenbot.types.aliases.BlockNumber Update block (from Rust via the handle). .. py:property:: requires_io_at_calculation_time :type: 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. .. py:method:: 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). .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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 DegenbotValueError: See function documentation. :raises InvalidSwapInputAmount: See function documentation. :raises NoLiquidity: See function documentation. .. py:class:: CurveStableswapPoolSimulationResult CurveStableswapPoolSimulationResult class. .. py:attribute:: amount0_delta :type: int .. py:attribute:: amount1_delta :type: int .. py:attribute:: current_state :type: CurveStableswapPoolState .. py:class:: CurveStableswapPoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` CurveStableswapPoolState class. .. py:attribute:: balances :type: tuple[int, ...] .. py:attribute:: base :type: CurveStableswapPoolState | None :value: None .. py:class:: Erc20Token(*args: Any, **kwargs: Any) Bases: :py:obj:`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. .. py:method:: from_handle(py_token: degenbot._ffi.Erc20Token, *, oracle_address: str | None = None, state_cache_depth: int = 8) -> Self :classmethod: 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``. .. py:property:: address :type: 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. .. py:property:: name :type: str Token name (read from Rust-owned ``TokenEntry``). .. py:property:: symbol :type: str Token symbol (read from Rust-owned ``TokenEntry``). .. py:property:: decimals :type: int Token decimals (read from Rust-owned ``TokenEntry``). .. py:method:: get_cached_balance(address: degenbot.types.chain.ChecksummedAddress, block_number: int) -> int | None Return cached balance. :returns: The computed value. .. py:method:: set_cached_balance(address: degenbot.types.chain.ChecksummedAddress, block_number: int, balance: int) -> None Set cached balance. .. py:method:: 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. .. py:method:: set_cached_approval(block_number: int, owner: degenbot.types.chain.ChecksummedAddress, spender: degenbot.types.chain.ChecksummedAddress, amount: int) -> None Set cached approval. .. py:method:: get_cached_total_supply(block_number: int) -> int | None Return cached total supply. :returns: The computed value. .. py:method:: set_cached_total_supply(block_number: int, total_supply: int) -> None Set cached total supply. .. py:property:: price :type: float Price. :raises NoPriceOracle: See function documentation. .. py:property:: chain_id :type: int Chain ID (read from Rust-owned ``TokenEntry``). .. py:class:: EtherPlaceholder Bases: :py:obj:`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 :class:`Erc20Token`). Use :meth:`Erc20Token.from_handle` — which produces an ``EtherPlaceholder`` instance when called on the subclass — after registering the metadata in a ``Bot``. .. py:attribute:: addresses .. py:class:: 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 :class:`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 :attr:`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 :class:`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 :attr:`ipc_path`). The one breaking change for callers: ``self.w3`` is gone. Use :attr:`provider` (a `degenbot._ffi.AlloyProvider`) for general RPC. .. py:property:: provider :type: 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. .. py:property:: fork_url :type: str | None Fork URL the spawned anvil process forks from, or `None` for in-memory. .. py:property:: ipc_path :type: 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. .. py:property:: http_url :type: 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). .. py:property:: ws_url :type: str WebSocket endpoint of the spawned anvil node (e.g. ``ws://127.0.0.1:PORT``). .. py:property:: port :type: int TCP port the spawned anvil node listens on. .. py:method:: 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). .. py:method:: mine() -> None Mine a single block (`evm_mine`). :raises AnvilError: If the RPC call fails. .. py:method:: 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 AnvilError: If the in-place RPC call fails. :raises DegenbotValueError: If no reset options are provided. .. py:method:: set_balance(address: str, balance: degenbot.validation.evm_values.ValidatedUint256) -> None Set balance (`anvil_setBalance`). :raises AnvilError: If the RPC call fails. .. py:method:: set_code(address: str, bytecode: bytes) -> None Set code (`anvil_setCode`). :raises AnvilError: If the RPC call fails. .. py:method:: set_coinbase(address: str) -> None Set coinbase (`anvil_setCoinbase`). :raises AnvilError: If the RPC call fails. .. py:method:: set_block_timestamp_interval(interval: int) -> None Set block timestamp interval (`anvil_setBlockTimestampInterval`). :raises AnvilError: If the RPC call fails. .. py:method:: 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. .. py:method:: 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. .. py:method:: set_nonce(address: str, nonce: int) -> None Set nonce (`anvil_setNonce`). :raises AnvilError: If the RPC call fails. .. py:method:: 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. .. py:class:: PancakeswapV3Pool Bases: :py:obj:`degenbot.uniswap.v3_liquidity_pool.UniswapV3Pool` PancakeswapV3Pool class. .. py:attribute:: variant :type: ClassVar[str | None] :value: 'pancakeswap' .. py:class:: PancakeswapV3PoolTracker Bases: :py:obj:`degenbot.uniswap.trackers.UniswapV3PoolTracker` Track PancakeSwap V3 pool events. .. py:class:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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). .. py:method:: 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. .. py:method:: list_all() -> collections.abc.Iterator[degenbot.types.pool_protocols.ConcentratedLiquidityPool] Yield every registered V4 pool. :Yields: Each V4 pool companion filed in this registry. .. py:method:: reset() -> None Drop every V4 companion. The session's V4 identities are untouched. .. py:class:: 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. .. py:method:: 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. .. py:method:: 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. :raises DegenbotValueError: A companion is already registered for this identity. .. py:method:: 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 :meth:`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. .. py:method:: 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. .. py:method:: 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. .. py:method:: reset() -> None Drop every companion, V4 included. The session's identities are untouched. .. py:class:: 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. .. py:method:: 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. :param registration: The typed registration request; see :class:`PoolRegistration` for field semantics. :raises ValueError: If the factory is already registered for the given chain. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:property:: registrations :type: 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. .. py:class:: TokenRegistry(*, py_bot: degenbot._ffi.Bot) ERC-20 token companions, keyed by the session's token identities. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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). .. py:method:: 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`. .. py:method:: list_all() -> collections.abc.Iterator[degenbot.erc20.erc20.Erc20Token] Yield every registered token. :Yields: Each token companion filed in this registry. .. py:method:: reset() -> None Drop every token companion. The session's identities are untouched. .. py:data:: pool_type_registry .. py:class:: 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: :py:obj:`degenbot.uniswap.trackers.UniswapV3PoolTracker` SushiswapV3PoolTracker class. .. py:class:: UniswapV2Pool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.uniswap.v2_pool_state.V2PoolState`, :py:obj:`degenbot.uniswap.v2_pool_calc.UniswapV2PoolCalc`, :py:obj:`degenbot.types.abstract.AbstractLiquidityPool` A Uniswap V2-based liquidity pool implementing the x*y=k constant function invariant. .. py:attribute:: variant :type: ClassVar[str | None] :value: None .. py:attribute:: stable_swap :type: bool :value: False .. py:attribute:: fee_denominator :type: int | None :value: None .. py:attribute:: dex :type: degenbot.types.DexIdentity .. py:attribute:: address :type: degenbot._ffi.ChecksummedAddress .. py:attribute:: factory :type: degenbot._ffi.ChecksummedAddress .. py:attribute:: init_hash :type: str .. py:attribute:: deployer :type: degenbot._ffi.ChecksummedAddress .. py:attribute:: name :type: str .. py:type:: PoolState :canonical: UniswapV2PoolState .. py:method:: from_handle(py_pool: degenbot.types.Pool) -> Self :classmethod: 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. :raises 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). .. py:property:: update_block :type: degenbot.types.aliases.BlockNumber Update block. :returns: The block number of the most recent state update (from Rust). .. py:property:: reserves_token0 :type: int Reserves token0. :returns: The reserve amount for token0 (from Rust). .. py:property:: reserves_token1 :type: int Reserves token1. :returns: The reserve amount for token1 (from Rust). .. py:property:: state :type: 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). .. py:method:: external_update(update: degenbot.uniswap.v2_types.UniswapV2PoolExternalUpdate) -> None External update. :raises ExternalUpdateError: If the update is for a past block. .. py:method:: 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). .. py:method:: 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). .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:class:: UniswapV2PoolExternalUpdate UniswapV2PoolExternalUpdate class. .. py:attribute:: block_number :type: degenbot.types.aliases.BlockNumber .. py:attribute:: reserves_token0 :type: int .. py:attribute:: reserves_token1 :type: int .. py:class:: UniswapV2PoolSimulationResult Bases: :py:obj:`UniswapSimulationResult` UniswapV2PoolSimulationResult class. .. py:attribute:: initial_state :type: UniswapV2PoolState .. py:attribute:: final_state :type: UniswapV2PoolState .. py:class:: UniswapV2PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` UniswapV2PoolState class. .. py:attribute:: reserves_token0 :type: int .. py:attribute:: reserves_token1 :type: int .. py:class:: UniswapV2PoolTracker Bases: :py:obj:`AbstractUniswapV2PoolTracker`\ [\ :py:obj:`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. .. py:property:: pool_init_hash :type: str Pool init hash. :returns: The pool init hash string. .. py:method:: 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. .. py:class:: UniswapV3LiquiditySnapshot Bases: :py:obj:`degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase`\ [\ :py:obj:`degenbot.types.chain.ChecksummedAddress`\ , :py:obj:`degenbot.uniswap.v3_types.UniswapV3PoolLiquidityMappingUpdate`\ , :py:obj:`degenbot.uniswap.v3_types.UniswapV3LiquidityEvent`\ , :py:obj:`UniswapV3LiquiditySnapshotSource`\ ] Retrieve and maintain liquidity positions for Uniswap V3 pools. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:class:: UniswapV3PoolExternalUpdate UniswapV3PoolExternalUpdate class. .. py:attribute:: block_number :type: degenbot.types.aliases.BlockNumber .. py:attribute:: liquidity :type: Liquidity .. py:attribute:: sqrt_price_x96 :type: SqrtPriceX96 .. py:attribute:: tick :type: Tick .. py:class:: UniswapV3PoolSimulationResult Bases: :py:obj:`UniswapSimulationResult` UniswapV3PoolSimulationResult class. .. py:attribute:: initial_state :type: UniswapV3PoolState .. py:attribute:: final_state :type: UniswapV3PoolState .. py:class:: UniswapV3PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` UniswapV3PoolState class. .. py:attribute:: liquidity :type: Liquidity .. py:attribute:: sqrt_price_x96 :type: SqrtPriceX96 .. py:attribute:: tick :type: Tick .. py:attribute:: tick_bitmap :type: dict[BitmapWord, degenbot.uniswap.concentrated.types.BitmapAtWord] .. py:attribute:: tick_data :type: dict[Tick, degenbot.uniswap.concentrated.types.LiquidityAtTick] .. py:class:: UniswapV3PoolTracker Bases: :py:obj:`AbstractUniswapV3PoolTracker`\ [\ :py:obj:`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. .. py:method:: 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. .. py:class:: UniswapV4LiquiditySnapshot Bases: :py:obj:`degenbot.uniswap.concentrated.snapshot_readers.LiquiditySnapshotBase`\ [\ :py:obj:`ManagedPoolIdentifier`\ , :py:obj:`degenbot.uniswap.v4_types.UniswapV4PoolLiquidityMappingUpdate`\ , :py:obj:`degenbot.uniswap.v4_types.UniswapV4LiquidityEvent`\ , :py:obj:`UniswapV4LiquiditySnapshotSource`\ ] Retrieve and maintain liquidity positions for Uniswap V4 pools. .. py:property:: pools :type: set[ManagedPoolIdentifier] Pools. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:method:: 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. .. py:class:: UniswapV4Pool(*args: Any, **kwargs: Any) Bases: :py:obj:`degenbot.uniswap.v4_pool_state.V4PoolState`, :py:obj:`degenbot.uniswap.v4_pool_calc.UniswapV4PoolCalc`, :py:obj:`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. .. py:type:: PoolState :canonical: UniswapV4PoolState .. py:attribute:: hook_address :type: degenbot.types.chain.ChecksummedAddress .. py:attribute:: active_hooks :type: frozenset[Hooks] .. py:attribute:: name :type: str .. py:attribute:: protocol_fee :type: ProtocolFee .. py:attribute:: lp_fee :type: int .. py:method:: from_handle(py_pool: degenbot.types.Pool) -> Self :classmethod: 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. .. py:method:: 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 DegenbotValueError: If token_out is not held by this pool. :raises HookedPoolResult: If the pool has active hooks that affect the swap. :raises IncompleteSwap: If the swap cannot fulfill the full output amount. :raises LiquidityPoolError: If the simulated execution reverts. .. py:method:: 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 DegenbotValueError: If token_in is not held by this pool. :raises HookedPoolResult: If the pool has active hooks that affect the swap. :raises IncompleteSwap: If the swap cannot fulfill the full input amount. :raises LiquidityPoolError: If the simulated execution reverts. .. py:property:: address :type: degenbot.types.chain.ChecksummedAddress Address. :returns: The pool manager address. .. py:property:: pool_id :type: bytes Pool id. :returns: The pool ID bytes. .. py:property:: pool_key :type: degenbot.uniswap.v4_types.UniswapV4PoolKey Pool key. :returns: The V4 pool key struct. .. py:property:: sqrt_price_x96 :type: int Sqrt price x96. :returns: The current sqrt price as a Q64.96 value (from Rust). .. py:property:: state :type: 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. .. py:property:: tick_spacing :type: int Tick spacing. :returns: The tick spacing for the pool (Python-side identity). .. py:property:: fee :type: int Fee. :returns: The fee in pips (Python-side identity). .. py:class:: UniswapV4PoolExternalUpdate UniswapV4PoolExternalUpdate class. .. py:attribute:: block_number :type: degenbot.types.aliases.BlockNumber .. py:attribute:: liquidity :type: degenbot.uniswap.v3_types.Liquidity .. py:attribute:: sqrt_price_x96 :type: degenbot.uniswap.v3_types.SqrtPriceX96 .. py:attribute:: tick :type: degenbot.uniswap.v3_types.Tick .. py:class:: UniswapV4PoolState Bases: :py:obj:`degenbot.types.abstract.AbstractPoolState` UniswapV4PoolState class. .. py:attribute:: liquidity :type: degenbot.uniswap.v3_types.Liquidity .. py:attribute:: sqrt_price_x96 :type: degenbot.uniswap.v3_types.SqrtPriceX96 .. py:attribute:: tick :type: degenbot.uniswap.v3_types.Tick .. py:attribute:: tick_bitmap :type: dict[degenbot.uniswap.v3_types.BitmapWord, degenbot.uniswap.concentrated.types.BitmapAtWord] .. py:attribute:: tick_data :type: dict[degenbot.uniswap.v3_types.Tick, degenbot.uniswap.concentrated.types.LiquidityAtTick] .. py:attribute:: id :type: bytes .. py:attribute:: block :type: degenbot.types.aliases.BlockNumber | None