degenbot.registry ================= .. py:module:: degenbot.registry .. autoapi-nested-parse:: Session-object registries and DEX deployment data for pool and token bookkeeping. Submodules ---------- .. toctree:: :maxdepth: 1 /autoapi/degenbot/registry/deployment_loader/index /autoapi/degenbot/registry/deployment_records/index /autoapi/degenbot/registry/pool/index /autoapi/degenbot/registry/pool_type/index /autoapi/degenbot/registry/session/index /autoapi/degenbot/registry/token/index Package Contents ---------------- .. 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:data:: pool_type_registry .. py:class:: CompanionCache[T] The presentation objects this registry hands out, keyed by session key. Read, store, and drop all take a `SessionObject` handle, so the only keys that can enter this cache are the ones the Rust registry minted. A miss is `None` and never a `KeyError`, so a registry read reads as a question rather than an assertion. .. py:method:: resolve(handle: degenbot._ffi.SessionObject) -> T | None Return the companion filed under `handle`'s identity, if any. :returns: The companion, or None when the identity has none filed. .. py:method:: store(handle: degenbot._ffi.SessionObject, item: T) -> None File `item` under `handle`'s identity, replacing any entry there. Callers that must not replace use [`get_or_store`], or check [`resolve`] first — this is the unconditional write. .. py:method:: get_or_store(handle: degenbot._ffi.SessionObject, item: T) -> T File `item` unless the identity already has a companion; return the canonical one. The build path's concurrent-build primitive: a second worker that built the same pool or token gets the companion the first worker registered rather than a twin. `setdefault` is a single atomic call under the GIL, so racing workers converge on one entry. :returns: The filed companion — the existing one on a duplicate, `item` otherwise. .. py:method:: drop(handle: degenbot._ffi.SessionObject) -> None Forget the companion for `handle`'s identity; a no-op when absent. .. py:method:: entries() -> collections.abc.Iterator[SessionCompanion[T]] Iterate the filed companions with the handle that names each. :returns: An iterator over one ``(handle, companion)`` pair per filed companion, snapshotted so a caller may mutate the cache while iterating. .. py:method:: clear() -> None Drop every companion. The session's identities are untouched. .. py:class:: SessionObjects(py_bot: degenbot._ffi.Bot) Typed adapter over the session-registry seam of one `Bot` handle. Every method is one canonical-identity question or get-or-create, with arguments already typed: hex strings and family tags in, a session object handle (or `None` for a resolve miss) out. No state lives here — the registry behind the handle is the session's one registry. .. py:method:: get_or_create_pool(*, chain_id: int, family: str, address: str, pool_id: bytes | None = None) -> degenbot._ffi.SessionObject Return the session's canonical pool object, registering it if new. :param chain_id: The session's chain; a different chain is refused. :param family: A registration family tag (`"v3"`, `"balancer-weighted"`, …). :param address: The pool's address, or its `PoolManager` when `family` is `"v4"`. :param pool_id: The on-chain V4 pool id; required for `"v4"` and refused for every other family. :returns: The one canonical object for this identity. .. py:method:: resolve_pool(*, chain_id: int, family: str, address: str, pool_id: bytes | None = None) -> degenbot._ffi.SessionObject | None Return the session's canonical pool object for this identity, or `None`. Registers nothing. :returns: The object, or None when the session does not hold the identity. .. py:method:: resolve_pool_by_address(*, chain_id: int, address: str) -> degenbot._ffi.SessionObject | None Return the session's canonical pool object at `address`, or `None`. The family-agnostic read: it names whichever family registered that address first in this session. A V4 pool is not address-keyed and is never named here. :returns: The object, or None when the session holds no address-keyed pool at that address. .. py:method:: get_or_create_token(*, chain_id: int, address: str) -> degenbot._ffi.SessionObject Return the session's canonical token object, registering it if new. :returns: The one canonical object for this token identity. .. py:method:: resolve_token(*, chain_id: int, address: str) -> degenbot._ffi.SessionObject | None Return the session's canonical token object for `address`, or `None`. :returns: The object, or None when the session does not hold it. .. py:method:: counts() -> tuple[int, int] Count the session's canonical objects. :returns: ``(pool_count, token_count)`` — one entry per identity in this session. .. 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.