degenbot.bot¶

Bot: central session manager for pool/token construction and registries.

Re-exports the driver Bot session class from _bot plus the driver_boot seam (the explicit install of the shared runtime + telemetry stack, called by Bot.__init__) from degenbot._ffi. The Rust engine handles — the same-named degenbot._ffi.Bot / BotIo pyclasses — are imported directly from degenbot._ffi by first-party code: the module path disambiguates the driver class from the engine handle (ADR-032; the ADR-013 init-only rule exempts the engine-handle types).

Package Contents¶

class degenbot.bot.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¶

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.