degenbot.arbitrage.policy¶

Path-composition rejection predicates.

A EngineRegistry can refuse an register_path() candidate by policy — token denylist/allowlist, hop-count min/max, min-liquidity gate, duplicate-pool guard. This is the distinct, policy-layer concern; the Rust core’s pool admission floor (hooked/dynamic-fee V4 refusal, surfaced as HookedPoolRejectedError / DynamicFeePoolRejectedError) stays in the Rust core (ADR-005 standalone-core constraint).

Rejections surface as the typed PathRejectedError family (subclassing ArbitrageError, NOT ValueError) so callers classify by type — mirroring the Plan 102 typed-exception pattern now used for pool admission + verification.

The predicate is a pluggable protocol: callers inject a PathPolicy (or any structural implementation of PathCompositionPredicate) into EngineRegistry.

Module Contents¶

class degenbot.arbitrage.policy.PathCompositionPredicate¶

Bases: Protocol

Pluggable path-composition policy.

Evaluate a path candidate (pool, zero_for_one) sequence. Accept by returning None; reject by raising a PathRejectedError subtype. Rejections propagate out of register_path() so callers classify by type.

evaluate(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) → None¶

Evaluate the path; raise PathRejectedError to reject.

class degenbot.arbitrage.policy.NoOpPathPredicate¶

Default predicate — accepts every path (no policy enforced).

evaluate(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) → None¶

Accept the path (no-op).

degenbot.arbitrage.policy.touched_tokens(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) → list[str]¶

Return the ordered, de-duplicated token addresses a path touches.

Captures the input token (first hop’s token_in) + every hop’s token_out (which includes all intermediate connecting tokens and the final profit token). Order: input first, then each hop’s output in path order; duplicates removed preserving first occurrence.

Returns:

The ordered, de-duplicated list of touched checksummed token addresses (input + intermediates + profit token).

class degenbot.arbitrage.policy.PathPolicy¶

Composable path-composition policy implementing the predicate protocol.

Every rule is opt-in via its fields; the defaults accept any path. Rules evaluate in a fixed order so the first violation raises (deterministic for callers’ classification):

  1. hop-count bounds (min then max)

  2. token denylist / allowlist

  3. duplicate-pool guard

  4. min-liquidity gate (only when liquidity_of is supplied)

Token-address inputs are normalized by checksum so denylist/allowlist entries supplied in any case match the pool’s checksummed token addresses.

The library deliberately does NOT encode pool-type-specific liquidity math (V2 reserves vs. V3/V4 liquidity/sqrt_price_x96). Supply a liquidity_of callable that returns the liquidity proxy for a pool, or leave it None to skip the gate.

disallowed_tokens: frozenset[str]¶
allowed_tokens: frozenset[str] | None = None¶
min_hops: int = 1¶
max_hops: int = 9223372036854775807¶
min_liquidity: int = 0¶
liquidity_of: collections.abc.Callable[[ArbPathPool], int] | None = None¶
reject_duplicate_pools: bool = True¶
evaluate(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) → None¶

Evaluate all enabled rules; raise on the first violation.

Raises: