Uniswap V4¶

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

almanak.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.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,
    gateway_client: GatewayClient | None = None,
)

Uniswap V4 swap adapter for intent compilation.

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

参数：

名称	类型	描述	默认
`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.

参数：

名称	类型	描述	默认
`token_id`	`int`	NFT token ID of the LP position.	必需
`rpc_url`	`str \| None`	Optional RPC URL override.	`None`

返回：

类型	描述
`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,
    *,
    max_price_impact: Decimal | None = None,
    config_max_price_impact: Decimal | None = None,
    offline_mode: bool = False,
    using_placeholders: bool = False,
) -> SwapResult

Build swap transactions for exact input amount.

参数：

名称	类型	描述	默认
`token_in`	`str`	Input token symbol or address.	必需
`token_out`	`str`	Output token symbol or address.	必需
`amount_in`	`Decimal`	Input amount in human-readable units.	必需
`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`

返回：

类型	描述
`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).

引发：

类型描述

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.

参数：

名称	类型	描述	默认
`intent`	`SwapIntent`	The SwapIntent to compile.	必需
`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`

返回：

类型	描述
`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])

参数：

名称	类型	描述	默认
`intent`	`LPOpenIntent`	LPOpenIntent with pool, amounts, and price range.	必需
`price_oracle`	`dict[str, Decimal] \| None`	Optional price map for liquidity estimation.	`None`

返回：

类型	描述
`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])

参数：

名称	类型	描述	默认
`intent`	`LPCloseIntent`	LPCloseIntent with position_id.	必需
`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.	`''`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`position_id`	`int`	NFT token ID.	必需
`currency0`	`str`	Token0 address (sorted).	必需
`currency1`	`str`	Token1 address (sorted).	必需
`hook_data`	`bytes`	Optional hook data for hooked pools.	`b''`

返回：

类型	描述
`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,
)

Configuration for UniswapV4Adapter.

属性：

名称	类型	描述
`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.

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.

参数：

名称	类型	描述	默认
`**kwargs`		Hook-specific parameters.	`{}`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`flags`	`HookFlags`	Decoded HookFlags from the hook address.	必需

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`hook_address`	`str`	Hook contract address (hex string with 0x prefix).	必需

返回：

类型	描述
`HookFlags`	HookFlags with decoded capability bits.

引发：

类型	描述
`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.

参数：

名称	类型	描述	默认
`bitmask`	`int`	Integer with hook flags in the lower 14 bits.	必需

返回：

类型	描述
`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,
    pool_key_lookup: PoolKeyLookup | None = None,
)

Parse Uniswap V4 transaction receipts.

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

参数：

名称	类型	描述	默认
`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,
    *,
    swap_token_meta: dict[str, dict[str, Any]]
    | None = None,
) -> ParseResult

Parse a transaction receipt for V4 events.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field.	必需
`quoted_amount_out`	`int \| None`	Expected output for slippage calculation.	`None`
`swap_token_meta`	`dict[str, dict[str, Any]] \| None`	VIB-3164 — compiler-supplied token metadata. Forwarded to `_build_swap_result` so hints can resolve decimals when the TokenResolver misses.	`None`

返回：

类型	描述
`ParseResult`	ParseResult with decoded events and swap summary.

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

Extract swap amounts for ResultEnricher integration.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict.	必需
`expected_out`	`Decimal \| None`	VIB-3203 — pre-slippage-discount quote in human (Decimal) units from the compiler's ActionBundle metadata. Overrides the parser's internal `slippage_bps` when provided, since the enrichment path does not supply constructor-level quote data.	`None`
`swap_token_meta`	`dict[str, dict[str, Any]] \| None`	VIB-3164 — compiler-supplied token metadata threaded from `build_extract_kwargs` via the ResultEnricher hook. Forwarded to `parse_receipt` -> `_build_swap_result` so hints resolve decimals when the TokenResolver misses. Shape: `{"token_in": {"address": ..., "symbol": ..., "decimals": ...}, "token_out": {...}}`.	`None`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict.	必需

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict.	必需

返回：

类型	描述
`int \| None`	Liquidity amount or None if not found.

extract_lp_open_data ¶

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

Extract LP open data from a V4 mint receipt.

VIB-4474 / V4 LP accounting V0. Walks the receipt for the canonical PositionManager-mediated mint shape:

ModifyLiquidity with liquidity_delta > 0 (a mint, not a burn) and sender in POSITION_MANAGER_ADDRESS_SET (allowlist).
ERC-721 Transfer(from=0x0, ...) emitted by the PositionManager NFT contract to recover the position tokenId.
Salt/tokenId consistency check: salt == bytes32(tokenId) per v4-periphery PositionManager._mint(). Mismatched salt is the failure signal -- non-zero salt that matches the tokenId is the CANONICAL V4 path and must pass.
position_hash = keccak(packed(positionManager, tickLower, tickUpper, salt)) per v4-core Position.calculatePositionKey.

Amount attribution: sum ERC-20 Transfers landing in the PoolManager grouped by token, then assign by sorted-address order (currency0 < currency1 invariant). When only one currency is observed (e.g. a concentrated-liquidity position opened out of range, or a single-sided deposit), the gateway PoolKey lookup is invoked to resolve both currency addresses and stamp a measured zero on the unobserved leg (VIB-4535 — symmetric with T07's close-side extract_lp_close_data). On lookup failure the LPOpenData is dropped (telemetry counters: missing_pool_key_lookup / pool_key_not_found / pool_key_lookup_error) rather than emitted with ambiguous attribution.

Non-allowlisted sender or salt/tokenId mismatch → structured WARNING + returns None. The writer must not crash on a parser miss (Empty != Zero / blueprint 27).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field.	必需

返回：

类型	描述
`LPOpenData \| None`	`LPOpenData` with `pool_address` set to the 32-byte V4 pool_id
`LPOpenData \| None`	(66-char lowercase hex) and `position_hash` set to the v4-core
`LPOpenData \| None`	position key. `None` when no eligible mint was found or any
`LPOpenData \| None`	validation gate fired.

extract_lp_close_data ¶

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

Extract LP close data from a V4 burn receipt.

VIB-4476 / V4 LP accounting V0. Token attribution is driven by the canonical PoolKey resolved via the gateway LookupV4PoolKey RPC (T03), NOT by sorting observed Transfer logs. Sorted-Transfer attribution is broken for (a) native ETH (which emits no ERC-20 Transfer) and (b) any non-trivial pair ordering where the on-chain currency0 < currency1 invariant does not match the order the transfers happen to appear in.

Walks the receipt for:

ModifyLiquidity with liquidity_delta < 0 (a burn, not a mint). Pull pool_id from topics[1].
Canonical PoolKey for that pool_id via the injected pool_key_lookup callable.
Transfer-set integrity check: the set of token addresses in observed Transfer logs leaving the PoolManager MUST be a non-empty subset of {currency0, currency1} from the PoolKey. On a token outside that set: structured WARNING + return None (fail-loud over silent misattribution).
PoolKey-ordered amount assignment: amount0_collected = sum of transfers of currency0; amount1_collected = sum of transfers of currency1.

Native-ETH currency (VIB-4483 / P-V1-B) is supported: the native leg is returned as raw ETH (no ERC-20 Transfer) so its amount{0,1}_collected principal is None (unmeasured, Empty ≠ Zero — VIB-5117) rather than a fabricated 0; the runner fills it pre-burn from a QueryV4PositionState read (_stamp_v4_lp_close_native_principal). A fully-native single-sided close emits NO ERC-20 Transfer at all; the empty-observed-tokens gate below allows it through ONLY for a native-leg PoolKey (VIB-5119) so the LP_CLOSE event is still booked. Uncollected FEES are measured pre-burn via _stamp_v4_lp_close_fees (VIB-4482).

Emits:

pool_address = 32-byte canonical pool_id (66-char lowercase hex)
source = "modify_liquidity"
fees0 = None, fees1 = None — V4 bundles fees into the withdrawal Transfer in V0; explicit None is the honest signal (Empty ≠ Zero, blueprint 27). Separate fee measurement is V1 P-V1-A (VIB-4482).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field.	必需

返回：

类型	描述
`LPCloseData \| None`	`LPCloseData` with PoolKey-driven amount attribution, or
`LPCloseData \| None`	`None` when no eligible burn is found, the PoolKey lookup
`LPCloseData \| None`	fails, or the observed Transfer set does not match the PoolKey.

extract_registry_payload_open ¶

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

Build the V4 LP_OPEN position_registry.payload dict (VIB-4583).

Structural twin of the V3 parser's extract_registry_payload_open, with the V4 identity tuple: token_id (NFT tokenId), pool_id (the 32-byte PoolKey hash V4 reports as LPOpenData.pool_address), and the per-chain PositionManager address. Returns None — caller falls back to plain save_ledger_entry — when any load-bearing identity field is missing (Empty ≠ Zero: a zero-substituted token_id or fabricated pool_id would silently corrupt physical_identity_hash).

fee_tier is forwarded from the intent's compile-time metadata and is carried opaquely (V4 keys identity on PoolManager+tokenId, not fee tier). None ⇒ the key stays absent (never substitute 0).

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 the V4 LP_CLOSE position_registry.payload dict (VIB-4583).

A V4 close burn emits ModifyLiquidity (carrying the pool_id) and Transfer logs (the withdrawn amounts) but NO NFT tokenId — V4, like a V3 DecreaseLiquidity-less burn, does not re-emit the tokenId on the Transfer. The tokenId therefore comes from the matched OPEN-side registry row (open_payload), which the runner threads through after resolving the close intent's position_id against the prior open. Without an open_payload carrying a tokenId we CANNOT build the physical_identity_hash, so we return None (fall back to plain ledger write) rather than fabricate one.

The close receipt's pool_id is cross-checked against the open_payload pool_id — a disagreement means the wrong OPEN row was threaded, so we refuse the close (the close must not overwrite anchors with a mismatched identity).

build_extract_kwargs ¶

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

Return UniswapV4-owned kwargs for ResultEnricher extraction calls.

VIB-3164: the adapter records full token identity (from_token / to_token dicts with address, symbol, decimals — see uniswap_v4/adapter.py) in ActionBundle.metadata. Threading it here lets _build_swap_result resolve decimals when the TokenResolver misses or Transfer events cannot classify token addresses.

Native-token entries are skipped: the receipt's Transfer events carry the wrapped token's address, so a native entry can never match by address, and its decimals (18) equal the fallback anyway. Note: decimals may be None in the adapter when it missed during compilation; such entries are skipped by the decimals is None guard.

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.

参数：

名称	类型	描述	默认
`chain`	`str`	Chain name (e.g. "arbitrum", "ethereum").	必需
`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).

参数：

名称	类型	描述	默认
`token_id`	`int`	NFT token ID of the LP position.	必需
`rpc_url`	`str \| None`	RPC URL to use. Falls back to self.rpc_url.	`None`

返回：

类型	描述
`int`	Liquidity amount (uint128) for the position.

引发：

类型	描述
`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().

参数：

名称	类型	描述	默认
`pool_key`	`PoolKey`	V4 PoolKey identifying the pool.	必需
`rpc_url`	`str \| None`	RPC URL to use. Falls back to self.rpc_url.	`None`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`token0`	`str`	First token address.	必需
`token1`	`str`	Second token address.	必需
`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`

返回：

类型	描述
`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,
) -> SwapQuote

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

参数：

名称	类型	描述	默认
`token_in`	`str`	Input token address.	必需
`token_out`	`str`	Output token address.	必需
`amount_in`	`int`	Input amount in smallest units.	必需
`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`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`token_in`	`str`	Input token address.	必需
`token_out`	`str`	Output token address.	必需
`amount_in`	`int`	Input amount in smallest units.	必需
`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`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`token_address`	`str`	Token contract address.	必需
`spender`	`str`	Address to approve (Permit2 for V4 flow).	必需
`amount`	`int`	Amount to approve.	必需

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`token_address`	`str`	Token address to approve.	必需
`spender`	`str`	Address to grant allowance to (UniversalRouter).	必需
`amount`	`int`	Amount to approve (uint160 max = 2^160-1).	必需
`expiration`	`int`	Expiration timestamp (0 = default 30 days from now).	`0`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`quote`	`SwapQuote`	Swap quote with amounts.	必需
`recipient`	`str`	Address to receive output tokens.	必需
`slippage_bps`	`int`	Slippage tolerance in basis points.	`50`
`deadline`	`int`	Transaction deadline (0 = 30 minutes from now).	`0`

返回：

类型	描述
`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

参数：

名称	类型	描述	默认
`params`	`LPMintParams`	LPMintParams with pool key, tick range, liquidity, etc.	必需
`deadline`	`int`	TX deadline (0 = 30 minutes from now).	`0`

返回：

类型	描述
`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].

参数：

名称	类型	描述	默认
`params`	`LPDecreaseParams`	LPDecreaseParams with token ID, liquidity, minimums.	必需
`currency0`	`str`	Token0 address (sorted).	必需
`currency1`	`str`	Token1 address (sorted).	必需
`recipient`	`str`	Address to receive withdrawn tokens.	必需
`deadline`	`int`	TX deadline (0 = 30 minutes from now).	`0`
`burn`	`bool`	Whether to burn the NFT after withdrawal.	`True`

返回：

类型	描述
`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.

参数：

名称	类型	描述	默认
`token_id`	`int`	Position NFT token ID.	必需
`currency0`	`str`	Token0 address (sorted).	必需
`currency1`	`str`	Token1 address (sorted).	必需
`recipient`	`str`	Address to receive fees.	必需
`hook_data`	`bytes`	Optional hook data for hooked pools.	`b''`
`deadline`	`int`	TX deadline (0 = 30 minutes from now).	`0`

返回：

类型	描述
`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)

参数：

名称	类型	描述	默认
`sqrt_price_x96`	`int`	Current pool sqrtPriceX96 (or estimate).	必需
`tick_lower`	`int`	Lower tick boundary.	必需
`tick_upper`	`int`	Upper tick boundary.	必需
`amount0`	`int`	Desired amount of token0 (in smallest units).	必需
`amount1`	`int`	Desired amount of token1 (in smallest units).	必需

返回：

类型	描述
`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).

参数：

名称	类型	描述	默认
`price`	`Decimal`	Price of token0 in terms of token1.	必需
`decimals0`	`int`	Decimals of token0.	`18`
`decimals1`	`int`	Decimals of token1.	`18`

返回：

类型	描述
`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

参数：

名称	类型	描述	默认
`token0`	`str`	First token address.	必需
`token1`	`str`	Second token address.	必需
`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`

返回：

类型	描述
`PoolDiscoveryResult`	PoolDiscoveryResult with pool key, ID, and hook capabilities.

getattr ¶

__getattr__(name: str) -> Any

PEP 562 lazy attribute access.

Uniswap V4¶

almanak.connectors.uniswap_v4 ¶

UniswapV4Adapter ¶

get_position_liquidity ¶

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 ¶

UniswapV4ReceiptParser ¶

parse_receipt ¶

extract_swap_amounts ¶

extract_position_id ¶

extract_liquidity ¶

extract_lp_open_data ¶

extract_lp_close_data ¶

extract_registry_payload_open ¶

extract_registry_payload_close ¶

build_extract_kwargs ¶

UniswapV4SDK ¶

get_position_liquidity ¶

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 ¶

__getattr__ ¶

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` ¶

compute_liquidity_from_amounts `staticmethod` ¶

estimate_sqrt_price_x96 `staticmethod` ¶

tick_to_price `staticmethod` ¶

price_to_tick `staticmethod` ¶

getattr ¶