degenbot.registry¶

Session-object registries and DEX deployment data for pool and token bookkeeping.

Submodules¶

Package Contents¶

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

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

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

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

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

Returns:

The registered V4 pool, or None if not found.

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

Register a V4 pool.

Raises:

DegenbotValueError – A companion is already registered for this identity.

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

Idempotently register a V4 pool, returning the stored instance.

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

Returns:

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

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

Drop the V4 companion for this identity.

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

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

Yield every registered V4 pool.

Yields:

Each V4 pool companion filed in this registry.

reset() → None¶

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

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

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

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

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

Retrieve a pool by chain and address.

Returns:

The registered pool, or None if not found.

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

Register a pool.

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

Raises:

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

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

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

Idempotently register a pool, returning the stored instance.

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

Returns:

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

Raises:

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

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

Remove a pool.

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

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

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

Yield every registered address-keyed pool.

Yields:

Each address-keyed pool companion filed in this registry.

reset() → None¶

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

class degenbot.registry.PoolTypeRegistry¶

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

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

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

Public API for external callers¶

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

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

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

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

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

Example:

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


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


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

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

register(registration: PoolRegistration) → None¶

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

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

Parameters:

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

Raises:

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

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

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

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

Raises:

KeyError – If no registration exists for the given key.

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

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

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

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

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

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

Returns:

True if a registration exists, False otherwise.

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

Get the pool class for (chain_id, factory).

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

Returns:

The pool class, or None if not found.

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

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

Returns:

The V2 pool class, or None if not found.

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

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

Returns:

The V3 pool class, or None if not found.

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

Get the PoolTypeDescriptor for (chain_id, factory).

Returns:

The descriptor, or None if not found.

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

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

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

Returns:

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

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

Get the deployment data for (chain_id, factory).

Returns:

The deployment data, or None if not found.

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

Get a PoolTypeDescriptor by its kind string.

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

Returns:

The descriptor, or None if not found.

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

A copy of all registrations.

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

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.

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.

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.

drop(handle: degenbot._ffi.SessionObject) → None¶

Forget the companion for handle’s identity; a no-op when absent.

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.

clear() → None¶

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

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

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.

Parameters:

  • chain_id – The session’s chain; a different chain is refused.

  • family – A registration family tag (“v3”, “balancer-weighted”, …).

  • address – The pool’s address, or its PoolManager when family is “v4”.

  • 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.

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.

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.

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.

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.

counts() → tuple[int, int]¶

Count the session’s canonical objects.

Returns:

(pool_count, token_count) — one entry per identity in this session.

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

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

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

Retrieve a token by chain and address.

Returns:

The registered token, or None if not found.

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

Register a token.

Raises:

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

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

Idempotently register a token, returning the stored instance.

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

Returns:

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

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

Drop the companion for this token.

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

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

Yield every registered token.

Yields:

Each token companion filed in this registry.

reset() → None¶

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