degenbot.pathfinding

Pathfinding utilities for discovering arbitrage routes.

Barrier module (ADR-013: the Pydantic barrier): bridges the Rust build_path_graph / find_paths_rust seams from _ffi and re-exports the deep pathfinding logic from _pathfinding. Importers should use:

from degenbot.pathfinding import find_paths, find_paths_async

rather than reaching into degenbot._ffi directly.

Package Contents

class degenbot.pathfinding.PathfindingRequest

Search parameters shared by find_paths and find_paths_async.

Fields:

chain_id: Chain ID restricting pool and token queries. start_tokens: Token addresses that begin a path. end_tokens: Token addresses that end a path. database_path: File-backed SQLite database opened by Rust read seams. min_depth: Minimum hops in yielded paths. max_depth: Optional maximum hops in yielded paths. pool_types: Typed pool families to include (default V2/V3/V4). pool_type_per_depth: Optional per-depth allowed pool-family sets; a

None entry allows all kinds at that depth.

allowed_intermediate_tokens: Optional intermediate-token whitelist.

chain_id: int

start_tokens: collections.abc.Iterable[degenbot.types.chain.ChecksummedAddress | str]

end_tokens: collections.abc.Iterable[degenbot.types.chain.ChecksummedAddress | str]

database_path: pathlib.Path

min_depth: int = 2

max_depth: int | None = None

pool_types: collections.abc.Sequence[degenbot.pathfinding.PoolKind]

pool_type_per_depth: collections.abc.Sequence[set[degenbot.pathfinding.PoolKind] | None] | None = None

allowed_intermediate_tokens: collections.abc.Iterable[degenbot.types.chain.ChecksummedAddress | str] | None = None

class degenbot.pathfinding.PathStep

PathStep class.

address: degenbot.types.chain.ChecksummedAddress

type: degenbot.pathfinding.PoolKind

hash: str | None = None

degenbot.pathfinding.find_paths(*, request: PathfindingRequest) → collections.abc.Iterator[collections.abc.Sequence[PathStep]]

Find paths from each start token to each end token via the Rust DFS.

Parameters:

request – The graph scope + traversal constraints for this search.

Yields:

A valid arbitrage path from a start token to an end token.

async degenbot.pathfinding.find_paths_async(*, request: PathfindingRequest, batch_size: int = 1000, prepare_traversals: collections.abc.Callable[..., list[_Traversal]] = _prepare_traversals, find_paths_async_rust: collections.abc.Callable[..., Any] = find_paths_async_rust) → collections.abc.AsyncGenerator[collections.abc.Sequence[PathStep], None]

Async find_paths, driving the Rust batched async iterator.

The one-time prep runs on the shared tokio blocking pool, so neither the database resolution nor the Rust bulk read stalls the event loop.

Parameters:

• request – The graph scope + traversal constraints for this search.

• batch_size – Paths per Rust delivery batch (default 1000), clamped >= 1.

• prepare_traversals – The one-time prep seam called as prepare_traversals(request=request) on the blocking pool.

• find_paths_async_rust – The Rust batched async iterator factory, called with the traversal’s positional seam arguments.

Yields:

Sequences of PathStep objects representing arbitrage paths.

Raises:

The producer's exception, re-raised at the consumer. –