跳转至

Uniswap V4

Connector for Uniswap V4 with UniversalRouter swaps, PositionManager LP, and hook discovery.

almanak.framework.connectors.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

Example

from almanak.framework.connectors.uniswap_v4 import UniswapV4Adapter

adapter = UniswapV4Adapter(chain="arbitrum") bundle = adapter.compile_swap_intent(intent, price_oracle)

UniswapV4Adapter

UniswapV4Adapter(
    chain: str | None = None,
    config: UniswapV4Config | None = None,
    token_resolver: TokenResolver | None = 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.

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,
) -> 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

Returns:

Type Description
SwapResult

SwapResult with transactions list.

compile_swap_intent

compile_swap_intent(
    intent: SwapIntent,
    price_oracle: dict[str, Decimal] | None = None,
) -> 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

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 = b"",
) -> 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

Optional hook data for hooked pools.

b''

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,
)

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.

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%).

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.

UniswapV4ReceiptParser

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

Parse Uniswap V4 transaction receipts.

Extracts swap amounts, effective prices, and balance deltas from V4 PoolManager events.

Parameters:

Name Type Description Default
chain str

Chain name for context.

'ethereum'
pool_manager_address str | None

PoolManager address to filter events.

None

parse_receipt

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

Parse a transaction receipt for V4 events.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field.

required
quoted_amount_out int | None

Expected output for slippage calculation.

None

Returns:

Type Description
ParseResult

ParseResult with decoded events and swap summary.

extract_swap_amounts

extract_swap_amounts(
    receipt: dict[str, Any],
) -> SwapAmounts | None

Extract swap amounts for ResultEnricher integration.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict.

required

Returns:

Type Description
SwapAmounts | None

SwapAmounts or None if no swap event found.

extract_position_id

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

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

Looks for a Transfer event emitted by the PositionManager contract where from_address is the zero address (indicating a mint).

Falls back to ERC-721 mint Transfers from other known V4 PositionManager addresses if no exact chain match is found (handles address mismatches or proxy patterns). Rejects mints from unknown contracts to fail closed.

Called by ResultEnricher for LP_OPEN intents.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict.

required

Returns:

Type Description
int | None

Position ID (tokenId) or None if not found.

extract_liquidity

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

Extract liquidity delta from ModifyLiquidity event.

Called by ResultEnricher for LP_OPEN intents.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict.

required

Returns:

Type Description
int | None

Liquidity amount or None if not found.

extract_lp_close_data

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

Extract LP close data from ModifyLiquidity and Transfer events.

Called by ResultEnricher for LP_CLOSE intents.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict.

required

Returns:

Type Description
LPCloseData | None

LPCloseData with collected amounts, or None if not found.

UniswapV4SDK

UniswapV4SDK(chain: str, rpc_url: str | 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.

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_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_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 routing: V4 pools primarily use native ETH (address(0)), not WETH. When token_in or token_out is WETH, the swap routes through the native ETH pool and adds UNWRAP_WETH or WRAP_ETH commands at the UniversalRouter level.

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 = 30 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 = 30 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 = 30 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 = 30 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.