Uniswap V4¶

Field	Value
Module	`almanak.connectors.uniswap_v4`
Protocol kind	LP
Aliases	N/A

Supported Chains And Intents¶

Chain	Family	Supported Intents
Arbitrum	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
Avalanche	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
Base	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
BNB Chain	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
Ethereum	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
Optimism	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
Polygon	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`
Robinhood	EVM	`LP_CLOSE`, `LP_COLLECT_FEES`, `LP_OPEN`, `SWAP`

uniswap_v4 ¶

Uniswap V4 protocol connector.

Provides swap compilation, receipt parsing, and pool utilities for Uniswap V4's singleton PoolManager architecture.

Key differences from V3: - Singleton PoolManager contract (all pools in one contract) - Pool keys include hooks address (currency0, currency1, fee, tickSpacing, hooks) - Native ETH support (no mandatory WETH wrapping) - Flash accounting model - New Swap event signature from PoolManager

Exact pool selection

Static fees are integers from 0 to 1_000_000 in hundredths of a basis point. Tick spacing is independent of fee; custom pools require explicit spacing. A dynamic pool uses raw fee 0x800000 in its immutable key, while stored fees and per-swap overrides are mutable observations.

Pass all five canonical fields through swap_params['pool_key'], or resolve swap_params['pool_id'] through the gateway. Redundant pins must agree. Native currency is address zero; WETH is a distinct asset.

Example

from decimal import Decimal
from almanak.connectors.uniswap_v4 import PoolKey
from almanak.framework.intents import Intent

key = PoolKey(
    "0x0000000000000000000000000000000000000000",
    "0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913",
    fee=1000, tick_spacing=20,
)
intent = Intent.swap(
    from_token="USDC", to_token="ETH", amount=Decimal("3"),
    protocol="uniswap_v4", max_slippage=Decimal("0.005"),
    swap_params={"pool_key": key.to_wire()},
)

LP and hook qualification

LP entry accepts a pool ID and protocol_params['pool_key']. Withdrawal verifies the owned NFT's key and liquidity. Supply both explicit withdrawal minima or let measured principal establish the configured per-leg floors.

Hooked operations require explicit hook_data and a reviewed operation profile. The built-in profile admits callbacks that are unreachable for the requested operation, including compatible stored-dynamic pools. Per-swap override hooks need family-specific quote/execution evidence; deterministic ABI tests do not admit arbitrary deployed hooks. Unqualified custom accounting, subscribers and Safe contexts fail closed.

Execution rechecks bound identity, approvals, calldata, deployed route, expiry and block continuity before signing and submission. Mutable fees can invalidate a quote; on-chain input/output bounds remain authoritative.

UniswapV4Adapter ¶

UniswapV4Adapter(
    chain: str | None = None,
    config: UniswapV4Config | None = None,
    token_resolver: TokenResolver | None = None,
    gateway_client: GatewayClient | None = None,
    venue_verification_gateway_factory: Any = None,
)

Uniswap V4 swap adapter for intent compilation.

Compiles SwapIntents into ActionBundles containing approve + swap transactions targeting the V4 swap router.

Parameters:

Name	Type	Description	Default
`chain`	`str \| None`	Chain name.	`None`
`config`	`UniswapV4Config \| None`	Optional UniswapV4Config. If not provided, chain is used.	`None`
`token_resolver`	`TokenResolver \| None`	Optional TokenResolver for symbol -> address resolution.	`None`

get_position_liquidity ¶

get_position_liquidity(
    token_id: int, rpc_url: str | None = None
) -> int

Query on-chain liquidity for a V4 LP position.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the LP position.	required
`rpc_url`	`str \| None`	Optional RPC URL override.	`None`

Returns:

Type	Description
`int`	Liquidity amount (uint128). Raises ValueError if position is empty or query fails.

get_position_currencies ¶

get_position_currencies(
    token_id: int, rpc_url: str | None = None
) -> tuple[str, str]

Resolve a V4 position's (currency0, currency1) from its NFT id, on-chain.

Reads the position's PoolKey via PositionManager.getPoolAndPositionInfo and returns the two currency addresses in canonical sorted order. Lets a V4 position be closed from its id alone when the open-time currencies are not otherwise available (VIB-5361 operator recovery / ax lp-close).

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the LP position.	required
`rpc_url`	`str \| None`	Optional RPC URL override.	`None`

Returns:

Type	Description
`tuple[str, str]`	`(currency0, currency1)` lowercased EVM addresses, currency0 < currency1.

swap_exact_input ¶

swap_exact_input(
    token_in: str,
    token_out: str,
    amount_in: Decimal,
    slippage_bps: int | None = None,
    fee_tier: int | None = None,
    price_ratio: Decimal | None = None,
    *,
    max_price_impact: Decimal | None = None,
    config_max_price_impact: Decimal | None = None,
    offline_mode: bool = False,
    using_placeholders: bool = False,
    swap_params: dict[str, Any] | None = None,
) -> SwapResult

Build swap transactions for exact input amount.

Parameters:

Name	Type	Description	Default
`token_in`	`str`	Input token symbol or address.	required
`token_out`	`str`	Output token symbol or address.	required
`amount_in`	`Decimal`	Input amount in human-readable units.	required
`slippage_bps`	`int \| None`	Slippage tolerance in bps. Default from config.	`None`
`fee_tier`	`int \| None`	Fee tier. Default from config.	`None`
`price_ratio`	`Decimal \| None`	Price ratio (token_out per token_in) for cross-decimal quotes.	`None`
`max_price_impact`	`Decimal \| None`	Per-intent price-impact ceiling (`intent.max_price_impact`) or None to fall back to `config_max_price_impact` (VIB-2058).	`None`
`config_max_price_impact`	`Decimal \| None`	Compiler-config price-impact default (`ctx.max_price_impact_pct`). Defaults to 5% when unset.	`None`
`offline_mode`	`bool`	Permission-discovery / placeholder compile. When True an executable-quote failure degrades to the local estimate instead of failing closed (the swap is never broadcast in this mode).	`False`
`using_placeholders`	`bool`	Whether the price oracle holds placeholder prices — relaxes the price-impact guard (oracle estimate is not real).	`False`

Returns:

Type	Description
`SwapResult`	SwapResult with transactions list. `success=False` (no transactions)
`SwapResult`	when the executable quote is unavailable online (fail-closed, C1), the
`SwapResult`	executable quote returns zero output (C1), or the price impact exceeds
`SwapResult`	tolerance (C2) — see VIB-2058 (https://linear.app/almanak/issue/VIB-2058).

Raises:

Type Description

ValueError

for permanent compilation failures that should halt the strategy rather than retry — an unresolvable token, a missing wallet_address, or the VIB-3875 cross-decimal guard (get_quote_local with no price_ratio and mismatched token decimals). These deliberately propagate (not wrapped in a retryable SwapResult) and are caught + classified COMPILATION_PERMANENT at the compiler boundary (UniswapV4Compiler.compile_swap). The success=False returns above are the retryable failures.

compile_swap_intent ¶

compile_swap_intent(
    intent: SwapIntent,
    price_oracle: dict[str, Decimal] | None = None,
    *,
    config_max_price_impact: Decimal | None = None,
    permission_discovery: bool = False,
    using_placeholders: bool = False,
) -> ActionBundle

Compile a SwapIntent to an ActionBundle.

This method integrates with the intent system to convert high-level swap intents into executable transaction bundles.

Parameters:

Name	Type	Description	Default
`intent`	`SwapIntent`	The SwapIntent to compile.	required
`price_oracle`	`dict[str, Decimal] \| None`	Optional price map for USD conversions.	`None`
`config_max_price_impact`	`Decimal \| None`	Compiler-config price-impact default (`ctx.max_price_impact_pct`); the per-intent override is read from `intent.max_price_impact` (VIB-2058).	`None`
`permission_discovery`	`bool`	Permission-discovery compile — relaxes the executable-quote fail-closed rule (nothing is broadcast).	`False`
`using_placeholders`	`bool`	Price oracle holds placeholder prices — relaxes the price-impact guard.	`False`

Returns:

Type	Description
`ActionBundle`	ActionBundle containing transactions for execution.

compile_lp_open_intent ¶

compile_lp_open_intent(
    intent: LPOpenIntent,
    price_oracle: dict[str, Decimal] | None = None,
) -> ActionBundle

Compile an LPOpenIntent to an ActionBundle for V4 PositionManager.

Builds transactions for: 1-2. ERC-20 approve token0 + token1 to Permit2 3-4. Permit2.approve(PositionManager, token0/token1) 5. PositionManager.modifyLiquidities([MINT_POSITION, SETTLE_PAIR])

Parameters:

Name	Type	Description	Default
`intent`	`LPOpenIntent`	LPOpenIntent with pool, amounts, and price range.	required
`price_oracle`	`dict[str, Decimal] \| None`	Optional price map for liquidity estimation.	`None`

Returns:

Type	Description
`ActionBundle`	ActionBundle containing LP mint transactions.

compile_lp_close_intent ¶

compile_lp_close_intent(
    intent: LPCloseIntent,
    liquidity: int = 0,
    currency0: str = "",
    currency1: str = "",
) -> ActionBundle

Compile an LPCloseIntent to an ActionBundle for V4 PositionManager.

Builds a single transaction: PositionManager.modifyLiquidities([DECREASE_LIQUIDITY, TAKE_PAIR, BURN_POSITION])

Parameters:

Name	Type	Description	Default
`intent`	`LPCloseIntent`	LPCloseIntent with position_id.	required
`liquidity`	`int`	Total liquidity to withdraw (must be provided by caller, typically from on-chain position query).	`0`
`currency0`	`str`	Token0 address (sorted). Required for TAKE_PAIR.	`''`
`currency1`	`str`	Token1 address (sorted). Required for TAKE_PAIR.	`''`

Returns:

Type	Description
`ActionBundle`	ActionBundle containing LP close transactions.

compile_collect_fees_intent ¶

compile_collect_fees_intent(
    position_id: int,
    currency0: str,
    currency1: str,
    hook_data: bytes | None = None,
    *,
    pool: str | None = None,
    protocol_params: dict[str, Any] | None = None,
) -> ActionBundle

Compile a collect-fees operation for a V4 LP position.

Parameters:

Name	Type	Description	Default
`position_id`	`int`	NFT token ID.	required
`currency0`	`str`	Token0 address (sorted).	required
`currency1`	`str`	Token1 address (sorted).	required
`hook_data`	`bytes \| None`	Optional hook data for hooked pools.	`None`

Returns:

Type	Description
`ActionBundle`	ActionBundle containing fee collection transaction.

UniswapV4Config `dataclass` ¶

UniswapV4Config(
    chain: str,
    wallet_address: str = "",
    rpc_url: str | None = None,
    default_fee_tier: int = 3000,
    default_slippage_bps: int = 50,
    gateway_client: GatewayClient | None = None,
    managed_fork: bool | None = None,
    default_deadline_seconds: int = 300,
)

Configuration for UniswapV4Adapter.

Attributes:

Name	Type	Description
`chain`	`str`	Chain name (e.g. "arbitrum").
`wallet_address`	`str`	Wallet address for building transactions.
`rpc_url`	`str \| None`	Optional RPC URL for on-chain quotes (direct-HTTP fallback).
`default_fee_tier`	`int`	Default fee tier for swaps. Default 3000 (0.3%).
`default_slippage_bps`	`int`	Default slippage in basis points. Default 50 (0.5%).
`gateway_client`	`GatewayClient \| None`	Optional GatewayClient. When provided, on-chain eth_call queries route through `gateway_client.eth_call` and the `rpc_url` fallback is never exercised.
`managed_fork`	`bool \| None`	Tri-state managed-fork declaration threaded from the compiler context (ALM-3184). `None` means undeclared, in which case the price-impact guard probes the node rather than trusting the shape of `rpc_url`.
`default_deadline_seconds`	`int`	Transaction deadline in seconds.

UniswapV4UnsupportedPoolError ¶

Bases: UniswapV4FailLoudError

Pool shape is outside the V0 supported surface (hookless ERC20-ERC20).

Raised at compile time by the adapter before any transaction is built, so strategies fail loud on unsupported pool shapes instead of submitting transactions that the receipt parser / accounting layer cannot interpret.

V0 (VIB-4426) supports only: - hooks == 0x0000…0000 (no hook contract attached) - currency0 != 0x0000…0000 (no native-ETH currency leg)

Salt is intentionally NOT validated here: per VIB-4426 design §Q7, salt = bytes32(tokenId) is the canonical PositionManager._mint path, so a non-zero salt is the normal case and must not be rejected.

UniswapV4Compiler ¶

Bases: BaseProtocolCompiler[SwapCompilerContext]

Compiler for Uniswap V4 singleton PoolManager intents.

Declares :class:SwapCompilerContext (not the bare BaseCompilerContext) so the swap pipeline's price-impact / placeholder knobs (max_price_impact_pct, using_placeholders) reach the swap-safety guard (VIB-2058). V4 does not use the concentrated-liquidity adapter-factory machinery on CLCompilerContext — it owns its bespoke adapter — so it stops at SwapCompilerContext.

compile_swap ¶

compile_swap(
    ctx: SwapCompilerContext, intent: SwapIntent
) -> CompilationResult

Compile SWAP intent for Uniswap V4.

compile_lp_open ¶

compile_lp_open(
    ctx: BaseCompilerContext, intent: LPOpenIntent
) -> CompilationResult

Compile LP_OPEN intent for Uniswap V4 via PositionManager.

compile_lp_close ¶

compile_lp_close(
    ctx: BaseCompilerContext, intent: LPCloseIntent
) -> CompilationResult

Compile LP_CLOSE intent for Uniswap V4 via PositionManager.

compile_collect_fees ¶

compile_collect_fees(
    ctx: BaseCompilerContext, intent: CollectFeesIntent
) -> CompilationResult

Compile LP_COLLECT_FEES intent for Uniswap V4 via PositionManager.

HookDataEncoder ¶

Bases: ABC

Base class for encoding protocol-specific hookData.

Strategy authors subclass this to provide typed encoding for known hook contracts. The encoder validates inputs and produces ABI-encoded bytes that the hook contract expects.

Example

class DynamicFeeEncoder(HookDataEncoder):
    def encode(self, **kwargs) -> bytes:
        fee_override = kwargs.get("fee_override", 3000)
        return fee_override.to_bytes(32, "big")

    @property
    def hook_name(self) -> str:
        return "DynamicFeeHook"

encoder = DynamicFeeEncoder()
hook_data = encoder.encode(fee_override=500)

hook_name `abstractmethod` `property` ¶

hook_name: str

Human-readable name of the hook this encoder targets.

encode `abstractmethod` ¶

encode(**kwargs) -> bytes

Encode hookData for this specific hook contract.

Parameters:

Name	Type	Description	Default
`**kwargs`		Hook-specific parameters.	`{}`

Returns:

Type	Description
`bytes`	ABI-encoded bytes for the hookData field.

validate_flags ¶

validate_flags(flags: HookFlags) -> bool

Validate that hook flags are compatible with this encoder.

Override this method to enforce that the hook address has the expected capability bits set.

Parameters:

Name	Type	Description	Default
`flags`	`HookFlags`	Decoded HookFlags from the hook address.	required

Returns:

Type	Description
`bool`	True if flags are compatible, False otherwise.

HookFlags `dataclass` ¶

HookFlags(
    before_initialize: bool = False,
    after_initialize: bool = False,
    before_add_liquidity: bool = False,
    after_add_liquidity: bool = False,
    before_remove_liquidity: bool = False,
    after_remove_liquidity: bool = False,
    before_swap: bool = False,
    after_swap: bool = False,
    before_donate: bool = False,
    after_donate: bool = False,
    before_swap_returns_delta: bool = False,
    after_swap_returns_delta: bool = False,
    after_add_liquidity_returns_delta: bool = False,
    after_remove_liquidity_returns_delta: bool = False,
)

Decoded 14-bit hook capability flags from a V4 hook address.

In Uniswap V4, hook contract addresses encode their capabilities in the last 14 bits of the address. This is enforced by CREATE2 address mining -- the PoolManager validates that a hook's address matches its declared capabilities.

Usage

flags = HookFlags.from_address("0x...hook_address...") if flags.before_swap: print("Hook modifies swap behavior") if flags.has_any_swap_hooks: print("Hook participates in swaps")

has_any_swap_hooks `property` ¶

has_any_swap_hooks: bool

True if the hook participates in swap operations.

has_any_liquidity_hooks `property` ¶

has_any_liquidity_hooks: bool

True if the hook participates in liquidity operations.

has_any_delta_flags `property` ¶

has_any_delta_flags: bool

True if the hook returns balance deltas (modifies amounts).

is_empty `property` ¶

is_empty: bool

True if no hook capabilities are set (no-hook address).

active_flags `property` ¶

active_flags: list[str]

Return list of active hook flag names.

from_address `classmethod` ¶

from_address(hook_address: str) -> HookFlags

Decode hook capabilities from a hook contract address.

Parameters:

Name	Type	Description	Default
`hook_address`	`str`	Hook contract address (hex string with 0x prefix).	required

Returns:

Type	Description
`HookFlags`	HookFlags with decoded capability bits.

Raises:

Type	Description
`ValueError`	If the address is not a valid hex string.

from_bitmask `classmethod` ¶

from_bitmask(bitmask: int) -> HookFlags

Create HookFlags from a raw 14-bit bitmask.

Parameters:

Name	Type	Description	Default
`bitmask`	`int`	Integer with hook flags in the lower 14 bits.	required

Returns:

Type	Description
`HookFlags`	HookFlags with decoded capability bits.

to_bitmask ¶

to_bitmask() -> int

Convert flags back to a 14-bit integer bitmask.

requires_hook_data ¶

requires_hook_data() -> bool

True if this hook likely requires non-empty hookData.

Hooks with before_swap, after_swap, or delta-returning flags typically need hookData to function correctly. Empty hookData may cause reverts.

PoolDiscoveryResult `dataclass` ¶

PoolDiscoveryResult(
    pool_key: PoolKey,
    pool_id: str,
    hook_address: str,
    hook_flags: HookFlags,
    state: PoolState | None = None,
)

Result of pool discovery for a token pair.

Fields

pool_key: The resolved PoolKey. pool_id: Keccak256 hash of the ABI-encoded PoolKey. hook_address: Hook contract address (zero address if no hooks). hook_flags: Decoded hook capabilities. state: Pool state from StateView (None if not queried on-chain).

PoolState `dataclass` ¶

PoolState(
    sqrt_price_x96: int = 0,
    tick: int = 0,
    protocol_fee: int = 0,
    lp_fee: int = 0,
    exists: bool = False,
)

State of a V4 pool from StateView.getSlot0().

Fields

sqrt_price_x96: Current sqrt(price) as Q64.96 fixed-point. tick: Current tick index. protocol_fee: Protocol fee setting. lp_fee: LP fee in hundredths of a bip. exists: Whether the pool has been initialized.

PoolKey `dataclass` ¶

PoolKey(
    currency0: str,
    currency1: str,
    fee: int,
    tick_spacing: int,
    hooks: str = V4_ZERO_ADDRESS,
)

Canonical immutable currencies, raw fee field, independent spacing and hook.

UniswapV4ReceiptParser ¶

UniswapV4ReceiptParser(
    chain: str = "ethereum",
    pool_manager_address: str | None = None,
    position_manager_address: str | None = None,
    token_resolver: Any | None = None,
    pool_key_lookup: PoolKeyLookup | None = None,
)

Extract swaps, liquidity changes, transfers, and accounting identities.

parse_receipt ¶

parse_receipt(
    receipt: dict[str, Any],
    quoted_amount_out: int | None = None,
    *,
    swap_token_meta: dict[str, dict[str, Any]]
    | None = None,
    swap_pool_key: dict[str, Any] | None = None,
    swap_operation: dict[str, Any] | None = None,
) -> ParseResult

Decode supported events and build a swap summary when present.

Compiler token metadata supplies decimal hints when resolution fails.

extract_swap_amounts ¶

extract_swap_amounts(
    receipt: dict[str, Any],
    *,
    expected_out: Decimal | None = None,
    swap_token_meta: dict[str, dict[str, Any]]
    | None = None,
    swap_pool_key: dict[str, Any] | None = None,
    swap_operation: dict[str, Any] | None = None,
) -> SwapAmounts | None

Extract swap amounts for ResultEnricher integration.

expected_out is a human-unit pre-slippage quote. Token metadata has token_in/token_out entries containing address, symbol, and decimals and is used when the resolver misses.

extract_position_id ¶

extract_position_id(receipt: dict[str, Any]) -> int | None

Extract LP position NFT tokenId from ERC-721 Transfer event.

Prefer a zero-address mint from the configured PositionManager. A sole mint from another known V4 PositionManager is an address-mismatch fallback; unknown or multiple fallback emitters fail closed.

extract_liquidity ¶

extract_liquidity(receipt: dict[str, Any]) -> int | None

Return the first positive ModifyLiquidity delta.

extract_lp_open_data ¶

extract_lp_open_data(
    receipt: dict[str, Any],
) -> LPOpenData | None

Extract LP open data from a V4 mint receipt.

The first positive ModifyLiquidity must come from a known PositionManager; hook- or router-initiated mints are rejected. Its salt must equal bytes32(tokenId) from the PositionManager ERC-721 mint. The exact v4-core identity is keccak(packed(positionManager, tickLower, tickUpper, salt)).

ERC-20 deposits are raw base units summed by token and assigned in currency0 < currency1 order. A single observed currency requires a canonical PoolKey lookup; lookup failure or a token outside that key drops the result rather than guessing. An absent ERC-20 leg is measured zero, but native currency is None because msg.value emits no Transfer. With no transfers and no lookup, the legacy all-None shape remains fail-open for callers that use intent token order. The first same-pool Swap supplies current_tick.

extract_lp_close_data ¶

extract_lp_close_data(
    receipt: dict[str, Any],
) -> LPCloseData | None

Extract LP close data from a V4 burn receipt.

A unique negative ModifyLiquidity supplies the pool ID and removed liquidity. Raw base-unit withdrawals are summed only from Transfers leaving PoolManager and assigned by the looked-up PoolKey, never by log order. Lookup failure or any observed token outside the key fails closed; a missing ERC-20 key leg is measured zero.

Address-zero native currency returns raw ETH without an ERC-20 event, so its principal is unmeasured None and is filled from pre-burn position state. An empty observed set is valid only for a native pool; it remains an attribution failure for an all-ERC-20 pool. V4 does not separate fees from withdrawal Transfers here, so fees0/fees1 remain None and the runner measures fees separately before burning.

extract_registry_payload_open ¶

extract_registry_payload_open(
    receipt: dict[str, Any], *, fee_tier: int | None = None
) -> dict[str, Any] | None

Build an LP_OPEN registry payload from canonical V4 identity.

Physical identity requires tokenId and the chain's PositionManager; the 32-byte pool ID is the semantic grouping key. Missing fields fail closed rather than using zero or fabricated values. Fee tier is optional metadata, not identity.

registry_close_identity_matches ¶

registry_close_identity_matches(
    receipt: dict[str, Any],
    open_payload: dict[str, Any] | None,
) -> bool

Bind the burn to its canonical manager, pool and NFT salt before merging OPEN state.

extract_registry_payload_close ¶

extract_registry_payload_close(
    receipt: dict[str, Any],
    *,
    open_payload: dict[str, Any] | None = None,
    fee_tier: int | None = None,
) -> dict[str, Any] | None

Build an LP_CLOSE registry payload using its matched OPEN identity.

The matched OPEN tokenId must equal the canonical burn salt. Manager, pool and ticks must also agree before OPEN identity fields are merged.

build_extract_kwargs ¶

build_extract_kwargs(
    *, field: str, bundle_metadata: dict[str, Any]
) -> dict[str, Any]

Return canonical typed swap metadata for receipt extraction.

UniswapV4SDK ¶

UniswapV4SDK(
    chain: str,
    rpc_url: str | None = None,
    gateway_client: GatewayClient | None = None,
)

Uniswap V4 SDK for pool operations and swap encoding.

Routes swaps through the canonical UniversalRouter with Permit2 flow.

Parameters:

Name	Type	Description	Default
`chain`	`str`	Chain name (e.g. "arbitrum", "ethereum").	required
`rpc_url`	`str \| None`	Optional RPC URL for on-chain queries (direct HTTP fallback).	`None`
`gateway_client`	`GatewayClient \| None`	Optional GatewayClient. When provided, on-chain `eth_call` queries go through `gateway_client.eth_call` instead of direct urllib/HTTP. Strategy containers in production have no outbound HTTP so the gateway is required there; local dev and gateway-internal execution may still pass `rpc_url` instead.	`None`

get_position_liquidity ¶

get_position_liquidity(
    token_id: int, rpc_url: str | None = None
) -> int

Query on-chain liquidity for a V4 LP position via PositionManager.getPositionLiquidity(uint256).

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the LP position.	required
`rpc_url`	`str \| None`	RPC URL to use. Falls back to self.rpc_url.	`None`

Returns:

Type	Description
`int`	Liquidity amount (uint128) for the position.

Raises:

Type	Description
`ValueError`	If no RPC URL is available or the call fails.

get_position_pool_key ¶

get_position_pool_key(
    token_id: int, rpc_url: str | None = None
) -> PoolKey

Resolve a V4 position's PoolKey from its NFT token id, on-chain.

Calls PositionManager.getPoolAndPositionInfo(uint256) (selector 0x7ba03aad) and decodes the returned PoolKey struct. Unlike a V3 NFT (whose positions(tokenId) self-describes token0/token1/fee), a V4 position is keyed by a pool-id and the close path needs the underlying currencies — this read recovers them so a position can be closed from its id alone (VIB-5361 operator recovery).

The ABI return is (PoolKey poolKey, uint256 info). PoolKey has no dynamic fields, so it is encoded inline as 5 head words: (currency0, currency1, fee, tickSpacing, hooks). This mirrors the gateway-side decode in almanak/gateway/services/rpc_service.py (_decode_v4_pool_and_position_info) — kept byte-compatible.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the LP position.	required
`rpc_url`	`str \| None`	RPC URL to use. Falls back to `self.rpc_url`. When a gateway client is configured the call is routed through it.	`None`

Returns:

Type	Description
`PoolKey`	The position's :class:`PoolKey` (currencies returned in sorted order).

Raises:

Type	Description
`ValueError`	If no RPC URL/gateway is available or the call/decode fails.

get_pool_sqrt_price ¶

get_pool_sqrt_price(
    pool_key: PoolKey, rpc_url: str | None = None
) -> int | None

Query on-chain sqrtPriceX96 for a V4 pool via StateView.getSlot0().

Parameters:

Name	Type	Description	Default
`pool_key`	`PoolKey`	V4 PoolKey identifying the pool.	required
`rpc_url`	`str \| None`	RPC URL to use. Falls back to self.rpc_url.	`None`

Returns:

Type	Description
`int \| None`	sqrtPriceX96 (int) if successful, None if query fails or no RPC available.

compute_pool_key ¶

compute_pool_key(
    token0: str,
    token1: str,
    fee: int = 3000,
    tick_spacing: int | None = None,
    hooks: str = NATIVE_CURRENCY,
) -> PoolKey

Compute a V4 pool key for a token pair.

Parameters:

Name	Type	Description	Default
`token0`	`str`	First token address.	required
`token1`	`str`	Second token address.	required
`fee`	`int`	Fee tier in hundredths of a bip (e.g., 3000 = 0.3%).	`3000`
`tick_spacing`	`int \| None`	Custom tick spacing. Defaults to standard for fee tier.	`None`
`hooks`	`str`	Hooks contract address. Default: no hooks (zero address).	`NATIVE_CURRENCY`

Returns:

Type	Description
`PoolKey`	PoolKey with sorted currency addresses.

get_quote ¶

get_quote(
    token_in: str,
    token_out: str,
    amount_in: int,
    fee_tier: int = 3000,
    token_in_decimals: int = 18,
    token_out_decimals: int = 18,
    rpc_url: str | None = None,
    *,
    pool_key: PoolKey | None = None,
    hook_data: bytes = b"",
    block_number: int | None = None,
    from_address: str | None = None,
    quote_gateway: VenueVerificationGateway | None = None,
) -> SwapQuote

Get an executable exact-input quote from the V4 Quoter contract.

Parameters:

Name	Type	Description	Default
`token_in`	`str`	Input token address.	required
`token_out`	`str`	Output token address.	required
`amount_in`	`int`	Input amount in smallest units.	required
`fee_tier`	`int`	Fee tier (e.g. 3000 = 0.3%).	`3000`
`token_in_decimals`	`int`	Decimals for input token.	`18`
`token_out_decimals`	`int`	Decimals for output token.	`18`
`rpc_url`	`str \| None`	Optional direct RPC fallback for local-dev contexts.	`None`

Returns:

Type	Description
`SwapQuote`	SwapQuote with V4 Quoter amount_out.

get_quote_local ¶

get_quote_local(
    token_in: str,
    token_out: str,
    amount_in: int,
    fee_tier: int = 3000,
    token_in_decimals: int = 18,
    token_out_decimals: int = 18,
    price_ratio: Decimal | None = None,
) -> SwapQuote

Compute an offline swap quote estimate based on fee tier.

This is a best-effort estimate without on-chain data. For accurate quotes, use the V4 Quoter contract via RPC.

Parameters:

Name	Type	Description	Default
`token_in`	`str`	Input token address.	required
`token_out`	`str`	Output token address.	required
`amount_in`	`int`	Input amount in smallest units.	required
`fee_tier`	`int`	Fee tier (e.g. 3000 = 0.3%).	`3000`
`token_in_decimals`	`int`	Decimals for input token.	`18`
`token_out_decimals`	`int`	Decimals for output token.	`18`
`price_ratio`	`Decimal \| None`	Optional price ratio (token_in/token_out).	`None`

Returns:

Type	Description
`SwapQuote`	SwapQuote with estimated output.

build_approve_tx ¶

build_approve_tx(
    token_address: str, spender: str, amount: int
) -> SwapTransaction

Build an ERC-20 approve transaction.

Parameters:

Name	Type	Description	Default
`token_address`	`str`	Token contract address.	required
`spender`	`str`	Address to approve (Permit2 for V4 flow).	required
`amount`	`int`	Amount to approve.	required

Returns:

Type	Description
`SwapTransaction`	SwapTransaction with encoded approve calldata.

build_permit2_approve_tx ¶

build_permit2_approve_tx(
    token_address: str,
    spender: str,
    amount: int,
    expiration: int = 0,
) -> SwapTransaction

Build a Permit2.approve transaction to grant the UniversalRouter allowance.

Parameters:

Name	Type	Description	Default
`token_address`	`str`	Token address to approve.	required
`spender`	`str`	Address to grant allowance to (UniversalRouter).	required
`amount`	`int`	Amount to approve (uint160 max = 2^160-1).	required
`expiration`	`int`	Expiration timestamp (0 = default 30 days from now).	`0`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction targeting the Permit2 contract.

build_swap_tx ¶

build_swap_tx(
    quote: SwapQuote,
    recipient: str,
    slippage_bps: int = 50,
    deadline: int = 0,
) -> SwapTransaction

Build a V4 swap transaction via the UniversalRouter.

Uses the two-layer V4_SWAP encoding verified against real Ethereum mainnet txns

Outer: UniversalRouter.execute([V4_SWAP], [v4_input], deadline) Inner: v4_input = abi.encode(bytes actions, bytes[] params) actions = [SWAP_EXACT_IN_SINGLE, SETTLE, TAKE]

WETH retains its ERC-20 currency identity; the PoolKey binds both assets. Only a native address(0) leg adds SWEEP to forward native output or refund unspent native input.

Parameters:

Name	Type	Description	Default
`quote`	`SwapQuote`	Swap quote with amounts.	required
`recipient`	`str`	Address to receive output tokens.	required
`slippage_bps`	`int`	Slippage tolerance in basis points.	`50`
`deadline`	`int`	Transaction deadline (0 = 5 minutes from now).	`0`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction with encoded calldata.

build_mint_position_tx ¶

build_mint_position_tx(
    params: LPMintParams, deadline: int = 0
) -> SwapTransaction

Build a PositionManager.modifyLiquidities TX to mint a new LP position.

Encodes actions [MINT_POSITION, SETTLE_PAIR] to: 1. Create the position NFT with the specified liquidity 2. Settle (pay) both currencies via Permit2

Parameters:

Name	Type	Description	Default
`params`	`LPMintParams`	LPMintParams with pool key, tick range, liquidity, etc.	required
`deadline`	`int`	TX deadline (0 = 5 minutes from now).	`0`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction targeting PositionManager.

build_decrease_liquidity_tx ¶

build_decrease_liquidity_tx(
    params: LPDecreaseParams,
    currency0: str,
    currency1: str,
    recipient: str,
    deadline: int = 0,
    burn: bool = True,
) -> SwapTransaction

Build a PositionManager.modifyLiquidities TX to decrease/close an LP position.

Encodes actions [DECREASE_LIQUIDITY, TAKE_PAIR] and optionally [BURN_POSITION].

Parameters:

Name	Type	Description	Default
`params`	`LPDecreaseParams`	LPDecreaseParams with token ID, liquidity, minimums.	required
`currency0`	`str`	Token0 address (sorted).	required
`currency1`	`str`	Token1 address (sorted).	required
`recipient`	`str`	Address to receive withdrawn tokens.	required
`deadline`	`int`	TX deadline (0 = 5 minutes from now).	`0`
`burn`	`bool`	Whether to burn the NFT after withdrawal.	`True`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction targeting PositionManager.

build_collect_fees_tx ¶

build_collect_fees_tx(
    token_id: int,
    currency0: str,
    currency1: str,
    recipient: str,
    hook_data: bytes = b"",
    deadline: int = 0,
) -> SwapTransaction

Build a PositionManager.modifyLiquidities TX to collect fees only.

Decreases liquidity by 0 (triggers fee accrual update) then takes pair.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	Position NFT token ID.	required
`currency0`	`str`	Token0 address (sorted).	required
`currency1`	`str`	Token1 address (sorted).	required
`recipient`	`str`	Address to receive fees.	required
`hook_data`	`bytes`	Optional hook data for hooked pools.	`b''`
`deadline`	`int`	TX deadline (0 = 5 minutes from now).	`0`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction targeting PositionManager.

compute_liquidity_from_amounts `staticmethod` ¶

compute_liquidity_from_amounts(
    sqrt_price_x96: int,
    tick_lower: int,
    tick_upper: int,
    amount0: int,
    amount1: int,
) -> int

Compute liquidity from token amounts and price range.

Uses the same math as Uniswap V3/V4: - If current price is below range: liquidity from amount0 only - If current price is above range: liquidity from amount1 only - If current price is in range: min(liquidity from amount0, liquidity from amount1)

Parameters:

Name	Type	Description	Default
`sqrt_price_x96`	`int`	Current pool sqrtPriceX96 (or estimate).	required
`tick_lower`	`int`	Lower tick boundary.	required
`tick_upper`	`int`	Upper tick boundary.	required
`amount0`	`int`	Desired amount of token0 (in smallest units).	required
`amount1`	`int`	Desired amount of token1 (in smallest units).	required

Returns:

Type	Description
`int`	Estimated liquidity value.

estimate_sqrt_price_x96 `staticmethod` ¶

estimate_sqrt_price_x96(
    price: Decimal, decimals0: int = 18, decimals1: int = 18
) -> int

Estimate sqrtPriceX96 from a human-readable price (token1 per token0).

Parameters:

Name	Type	Description	Default
`price`	`Decimal`	Price of token0 in terms of token1.	required
`decimals0`	`int`	Decimals of token0.	`18`
`decimals1`	`int`	Decimals of token1.	`18`

Returns:

Type	Description
`int`	Estimated sqrtPriceX96 value.

tick_to_price `staticmethod` ¶

tick_to_price(
    tick: int, decimals0: int = 18, decimals1: int = 18
) -> Decimal

Convert tick to human-readable price.

Uses Decimal arithmetic to avoid float overflow at extreme ticks.

price_to_tick `staticmethod` ¶

price_to_tick(
    price: Decimal, decimals0: int = 18, decimals1: int = 18
) -> int

Convert human-readable price to tick.

Uses math.log for the inverse computation. Safe for typical price ranges.

discover_pool ¶

discover_pool(
    token0: str,
    token1: str,
    fee: int = 3000,
    tick_spacing: int | None = None,
    hooks: str = NO_HOOKS,
) -> PoolDiscoveryResult

Discover a V4 pool and decode its hook capabilities.

Two-step hook discovery: 1. Resolve PoolKey for the token pair/fee/tickSpacing to get the hook address 2. Decode the 14-bit capability bitmask from the hook address

Parameters:

Name	Type	Description	Default
`token0`	`str`	First token address.	required
`token1`	`str`	Second token address.	required
`fee`	`int`	Fee tier in hundredths of a bip.	`3000`
`tick_spacing`	`int \| None`	Custom tick spacing (defaults to standard for fee tier).	`None`
`hooks`	`str`	Hook contract address (zero address for no hooks).	`NO_HOOKS`

Returns:

Type	Description
`PoolDiscoveryResult`	PoolDiscoveryResult with pool key, ID, and hook capabilities.

Uniswap V4¶

Supported Chains And Intents¶

uniswap_v4 ¶

UniswapV4Adapter ¶

get_position_liquidity ¶

get_position_currencies ¶

swap_exact_input ¶

compile_swap_intent ¶

compile_lp_open_intent ¶

compile_lp_close_intent ¶

compile_collect_fees_intent ¶

UniswapV4Config dataclass ¶

UniswapV4UnsupportedPoolError ¶

UniswapV4Compiler ¶

compile_swap ¶

compile_lp_open ¶

compile_lp_close ¶

compile_collect_fees ¶

HookDataEncoder ¶

hook_name abstractmethod property ¶

encode abstractmethod ¶

validate_flags ¶

HookFlags dataclass ¶

has_any_swap_hooks property ¶

has_any_liquidity_hooks property ¶

has_any_delta_flags property ¶

is_empty property ¶

active_flags property ¶

from_address classmethod ¶

from_bitmask classmethod ¶

to_bitmask ¶

requires_hook_data ¶

PoolDiscoveryResult dataclass ¶

PoolState dataclass ¶

PoolKey dataclass ¶

UniswapV4ReceiptParser ¶

parse_receipt ¶

extract_swap_amounts ¶

extract_position_id ¶

extract_liquidity ¶

extract_lp_open_data ¶

extract_lp_close_data ¶

extract_registry_payload_open ¶

registry_close_identity_matches ¶

extract_registry_payload_close ¶

build_extract_kwargs ¶

UniswapV4SDK ¶

get_position_liquidity ¶

get_position_pool_key ¶

get_pool_sqrt_price ¶

compute_pool_key ¶

get_quote ¶

get_quote_local ¶

build_approve_tx ¶

build_permit2_approve_tx ¶

build_swap_tx ¶

build_mint_position_tx ¶

build_decrease_liquidity_tx ¶

build_collect_fees_tx ¶

compute_liquidity_from_amounts staticmethod ¶

estimate_sqrt_price_x96 staticmethod ¶

tick_to_price staticmethod ¶

price_to_tick staticmethod ¶

discover_pool ¶

UniswapV4Config `dataclass` ¶

hook_name `abstractmethod` `property` ¶

encode `abstractmethod` ¶

HookFlags `dataclass` ¶

has_any_swap_hooks `property` ¶

has_any_liquidity_hooks `property` ¶

has_any_delta_flags `property` ¶

is_empty `property` ¶

active_flags `property` ¶

from_address `classmethod` ¶

from_bitmask `classmethod` ¶

PoolDiscoveryResult `dataclass` ¶

PoolState `dataclass` ¶

PoolKey `dataclass` ¶

compute_liquidity_from_amounts `staticmethod` ¶

estimate_sqrt_price_x96 `staticmethod` ¶

tick_to_price `staticmethod` ¶

price_to_tick `staticmethod` ¶