degenbot.bot ============ .. py:module:: degenbot.bot .. autoapi-nested-parse:: Bot: central session manager for pool/token construction and registries. Re-exports the driver ``Bot`` session class from :mod:`._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 ---------------- .. 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.