degenbot.arbitrage.policy
=========================

.. py:module:: degenbot.arbitrage.policy

.. autoapi-nested-parse::

   Path-composition rejection predicates.

   A :class:`~degenbot.arbitrage.EngineRegistry` can refuse an
   :func:`~degenbot.arbitrage.EngineRegistry.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 :class:`~degenbot.exceptions.PathRejectedError`
   family (subclassing :class:`~degenbot.exceptions.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
   :class:`PathPolicy` (or any structural implementation of
   :class:`PathCompositionPredicate`) into ``EngineRegistry``.







Module Contents
---------------

.. py:class:: PathCompositionPredicate

   Bases: :py:obj:`Protocol`


   Pluggable path-composition policy.

   Evaluate a path candidate ``(pool, zero_for_one)`` sequence. Accept by
   returning ``None``; reject by raising a
   :class:`~degenbot.exceptions.PathRejectedError` subtype. Rejections
   propagate out of
   :meth:`~degenbot.arbitrage.EngineRegistry.register_path` so callers
   classify by type.


   .. py:method:: evaluate(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) -> None

      Evaluate the path; raise ``PathRejectedError`` to reject.



.. py:class:: NoOpPathPredicate

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


   .. py:method:: evaluate(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) -> None

      Accept the path (no-op).



.. py:function:: 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).


.. py:class:: 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.


   .. py:attribute:: disallowed_tokens
      :type:  frozenset[str]


   .. py:attribute:: allowed_tokens
      :type:  frozenset[str] | None
      :value: None



   .. py:attribute:: min_hops
      :type:  int
      :value: 1



   .. py:attribute:: max_hops
      :type:  int
      :value: 9223372036854775807



   .. py:attribute:: min_liquidity
      :type:  int
      :value: 0



   .. py:attribute:: liquidity_of
      :type:  collections.abc.Callable[[ArbPathPool], int] | None
      :value: None



   .. py:attribute:: reject_duplicate_pools
      :type:  bool
      :value: True



   .. py:method:: evaluate(pools_and_zfos: collections.abc.Sequence[tuple[ArbPathPool, bool]]) -> None

      Evaluate all enabled rules; raise on the first violation.

      :raises HopCountInsufficientError: If the hop count is below ``min_hops``.
      :raises HopCountExceededError: If the hop count exceeds ``max_hops``.
      :raises TokenDenylistedError: If a touched token is denylisted, or (when an
          allowlist is set) absent from the allowlist.
      :raises DuplicatePoolError: If the same pool appears more than once and
          ``reject_duplicate_pools`` is set.
      :raises InsufficientLiquidityError: If a pool's ``liquidity_of`` proxy is
          below ``min_liquidity`` (only when ``liquidity_of`` is set).