Uniswap V3¶

Connector for Uniswap V3 concentrated liquidity DEX.

almanak.connectors.uniswap_v3 ¶

Uniswap V3 Connector (concentrated liquidity AMM).

This module provides the Uniswap V3 adapter for executing token swaps on Uniswap V3 across multiple chains.

Supported chains: - Ethereum - Arbitrum - Optimism - Polygon - Base - Avalanche - BNB Chain - Monad

Example

from almanak.connectors.uniswap_v3 import UniswapV3Adapter, UniswapV3Config

config = UniswapV3Config( chain="arbitrum", wallet_address="0x...", ) adapter = UniswapV3Adapter(config)

Execute a swap¶

result = adapter.swap_exact_input( token_in="USDC", token_out="WETH", amount_in=Decimal("1000"), )

Use SDK for lower-level operations¶

from almanak.connectors.uniswap_v3 import UniswapV3SDK

sdk = UniswapV3SDK(chain="arbitrum", rpc_url="https://arb1.arbitrum.io/rpc") pool = sdk.get_pool_address(weth_addr, usdc_addr, fee_tier=3000)

SwapQuote `dataclass` ¶

SwapQuote(
    token_in: str,
    token_out: str,
    amount_in: int,
    amount_out: int,
    fee_tier: int,
    sqrt_price_x96_after: int = 0,
    gas_estimate: int = UNISWAP_V3_GAS_ESTIMATES[
        "swap_exact_input"
    ],
    price_impact_bps: int = 0,
    effective_price: Decimal = Decimal("0"),
    quoted_at: datetime = (lambda: datetime.now(UTC))(),
)

Quote for a swap operation.

Attributes:

Name	Type	Description
`token_in`	`str`	Input token address
`token_out`	`str`	Output token address
`amount_in`	`int`	Input amount in wei
`amount_out`	`int`	Output amount in wei
`fee_tier`	`int`	Fee tier of the pool
`sqrt_price_x96_after`	`int`	Price after swap (sqrt format)
`gas_estimate`	`int`	Estimated gas for the swap
`price_impact_bps`	`int`	Price impact in basis points
`effective_price`	`Decimal`	Effective price of the swap
`quoted_at`	`datetime`	Timestamp when quote was fetched

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapResult `dataclass` ¶

SwapResult(
    success: bool,
    transactions: list[TransactionData] = list(),
    quote: SwapQuote | None = None,
    amount_in: int = 0,
    amount_out_minimum: int = 0,
    error: str | None = None,
    gas_estimate: int = 0,
)

Result of a swap operation.

Attributes:

Name	Type	Description
`success`	`bool`	Whether the swap was built successfully
`transactions`	`list[TransactionData]`	List of transactions to execute
`quote`	`SwapQuote \| None`	Quote used for the swap
`amount_in`	`int`	Actual input amount
`amount_out_minimum`	`int`	Minimum output amount (with slippage)
`error`	`str \| None`	Error message if failed
`gas_estimate`	`int`	Total gas estimate for all transactions

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapType ¶

Bases: Enum

Type of swap operation.

TransactionData `dataclass` ¶

TransactionData(
    to: str,
    value: int,
    data: str,
    gas_estimate: int,
    description: str,
    tx_type: str = "swap",
)

Transaction data for execution.

Attributes:

Name	Type	Description
`to`	`str`	Target contract address
`value`	`int`	Native token value to send
`data`	`str`	Encoded calldata
`gas_estimate`	`int`	Estimated gas
`description`	`str`	Human-readable description
`tx_type`	`str`	Type of transaction (approve, swap)

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

UniswapV3Adapter ¶

UniswapV3Adapter(
    config: UniswapV3Config,
    token_resolver: TokenResolver | None = None,
)

Adapter for Uniswap V3 DEX protocol.

This adapter provides methods for: - Executing token swaps (exact input and exact output) - Building swap transactions - Handling ERC-20 approvals - Managing slippage protection

Example

config = UniswapV3Config( chain="arbitrum", wallet_address="0x...", ) adapter = UniswapV3Adapter(config)

Execute a swap¶

result = adapter.swap_exact_input( token_in="USDC", token_out="WETH", amount_in=Decimal("1000"), # 1000 USDC slippage_bps=50, )

Compile a SwapIntent to ActionBundle¶

intent = SwapIntent( from_token="USDC", to_token="WETH", amount_usd=Decimal("1000"), ) bundle = adapter.compile_swap_intent(intent)

Initialize the adapter.

Parameters:

Name	Type	Description	Default
`config`	`UniswapV3Config`	Uniswap V3 adapter configuration	required
`token_resolver`	`TokenResolver \| None`	Optional TokenResolver instance. If None, uses singleton.	`None`

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,
    recipient: str | None = None,
) -> SwapResult

Build a swap transaction with exact input amount.

This is the most common swap type where you specify exactly how much you want to spend and accept variable output.

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`	Amount of input token (in token units, not wei)	required
`slippage_bps`	`int \| None`	Slippage tolerance in basis points (default from config)	`None`
`fee_tier`	`int \| None`	Pool fee tier (default from config)	`None`
`recipient`	`str \| None`	Address to receive output tokens (default: wallet_address)	`None`

Returns:

Type	Description
`SwapResult`	SwapResult with transaction data

swap_exact_output ¶

swap_exact_output(
    token_in: str,
    token_out: str,
    amount_out: Decimal,
    slippage_bps: int | None = None,
    fee_tier: int | None = None,
    recipient: str | None = None,
) -> SwapResult

Build a swap transaction with exact output amount.

This swap type specifies exactly how much you want to receive and accepts variable input.

Parameters:

Name	Type	Description	Default
`token_in`	`str`	Input token symbol or address	required
`token_out`	`str`	Output token symbol or address	required
`amount_out`	`Decimal`	Amount of output token (in token units, not wei)	required
`slippage_bps`	`int \| None`	Slippage tolerance in basis points (default from config)	`None`
`fee_tier`	`int \| None`	Pool fee tier (default from config)	`None`
`recipient`	`str \| None`	Address to receive output tokens (default: wallet_address)	`None`

Returns:

Type	Description
`SwapResult`	SwapResult with transaction data

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 oracle for USD conversions	`None`

Returns:

Type	Description
`ActionBundle`	ActionBundle containing transactions for execution

set_allowance ¶

set_allowance(
    token: str, spender: str, amount: int
) -> None

Set cached allowance (for testing).

Parameters:

Name	Type	Description	Default
`token`	`str`	Token address	required
`spender`	`str`	Spender address	required
`amount`	`int`	Allowance amount	required

clear_allowance_cache ¶

clear_allowance_cache() -> None

Clear the allowance cache.

UniswapV3Config `dataclass` ¶

UniswapV3Config(
    chain: str,
    wallet_address: str,
    default_slippage_bps: int = 50,
    default_fee_tier: int = DEFAULT_FEE_TIER,
    deadline_seconds: int = 300,
    price_provider: dict[str, Decimal] | None = None,
    allow_placeholder_prices: bool = False,
)

Configuration for UniswapV3Adapter.

Attributes:

Name	Type	Description
`chain`	`str`	Target blockchain (ethereum, arbitrum, optimism, polygon, base)
`wallet_address`	`str`	Address executing transactions
`default_slippage_bps`	`int`	Default slippage tolerance in basis points (default 50 = 0.5%)
`default_fee_tier`	`int`	Default fee tier for pools (default 3000 = 0.3%)
`deadline_seconds`	`int`	Transaction deadline in seconds (default 300 = 5 minutes)
`price_provider`	`dict[str, Decimal] \| None`	Price oracle dict (token symbol -> USD price). Required for production use to calculate accurate slippage amounts.
`allow_placeholder_prices`	`bool`	If False (default), raises ValueError when no price_provider is given. Set to True ONLY for unit tests.

__post_init__ ¶

__post_init__() -> None

Validate configuration.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

UniswapV3LPAdapter ¶

UniswapV3LPAdapter(
    chain: str, protocol: str = "uniswap_v3"
)

LP calldata adapter for Uniswap V3 and compatible forks.

get_position_manager_address ¶

get_position_manager_address() -> str

Get the NFT position manager address.

get_mint_calldata ¶

get_mint_calldata(
    token0: str,
    token1: str,
    fee: int,
    tick_lower: int,
    tick_upper: int,
    amount0_desired: int,
    amount1_desired: int,
    amount0_min: int,
    amount1_min: int,
    recipient: str,
    deadline: int,
) -> bytes

Generate calldata for NonfungiblePositionManager.mint.

get_decrease_liquidity_calldata ¶

get_decrease_liquidity_calldata(
    token_id: int,
    liquidity: int,
    amount0_min: int,
    amount1_min: int,
    deadline: int,
) -> bytes

Generate calldata for NonfungiblePositionManager.decreaseLiquidity.

get_collect_calldata ¶

get_collect_calldata(
    token_id: int,
    recipient: str,
    amount0_max: int,
    amount1_max: int,
) -> bytes

Generate calldata for NonfungiblePositionManager.collect.

get_burn_calldata ¶

get_burn_calldata(token_id: int) -> bytes

Generate calldata for NonfungiblePositionManager.burn.

estimate_mint_gas ¶

estimate_mint_gas() -> int

Estimate gas for minting a new position.

estimate_close_gas ¶

estimate_close_gas(collect_fees: bool) -> int

Estimate gas for closing a position.

ParsedSwapResult `dataclass` ¶

ParsedSwapResult(
    token_in: str,
    token_out: str,
    token_in_symbol: str,
    token_out_symbol: str,
    amount_in: int,
    amount_out: int,
    amount_in_decimal: Decimal,
    amount_out_decimal: Decimal,
    effective_price: Decimal,
    slippage_bps: int,
    pool_address: str,
    sqrt_price_x96_after: int = 0,
    tick_after: int = 0,
    token_in_decimals_resolved: bool = True,
    token_out_decimals_resolved: bool = True,
)

High-level swap result extracted from receipt.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> ParsedSwapResult

Create from dictionary.

to_swap_result_payload ¶

to_swap_result_payload() -> SwapResultPayload

Convert to SwapResultPayload for event emission.

ParseResult `dataclass` ¶

ParseResult(
    success: bool,
    events: list[UniswapV3Event] = list(),
    swap_events: list[SwapEventData] = list(),
    transfer_events: list[TransferEventData] = list(),
    swap_result: ParsedSwapResult | None = None,
    error: str | None = None,
    transaction_hash: str = "",
    block_number: int = 0,
    transaction_success: bool = True,
)

Result of parsing a receipt.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapEventData `dataclass` ¶

SwapEventData(
    sender: str,
    recipient: str,
    amount0: int,
    amount1: int,
    sqrt_price_x96: int,
    liquidity: int,
    tick: int,
    pool_address: str,
)

Parsed data from Swap event.

token0_is_input `property` ¶

token0_is_input: bool

Check if token0 is the input token.

token1_is_input `property` ¶

token1_is_input: bool

Check if token1 is the input token.

amount_in `property` ¶

amount_in: int

Get the absolute input amount.

amount_out `property` ¶

amount_out: int

Get the absolute output amount.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> SwapEventData

Create from dictionary.

TransferEventData `dataclass` ¶

TransferEventData(
    from_addr: str,
    to_addr: str,
    value: int,
    token_address: str,
)

Parsed data from Transfer event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

UniswapV3Event `dataclass` ¶

UniswapV3Event(
    event_type: UniswapV3EventType,
    event_name: str,
    log_index: int,
    transaction_hash: str,
    block_number: int,
    contract_address: str,
    data: dict[str, Any],
    raw_topics: list[str] = list(),
    raw_data: str = "",
    timestamp: datetime = (lambda: datetime.now(UTC))(),
)

Parsed Uniswap V3 event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> UniswapV3Event

Create from dictionary.

UniswapV3EventType ¶

Bases: Enum

Uniswap V3 event types.

UniswapV3ReceiptParser ¶

UniswapV3ReceiptParser(
    chain: str = "arbitrum",
    token0_address: str | None = None,
    token1_address: str | None = None,
    token0_symbol: str | None = None,
    token1_symbol: str | None = None,
    token0_decimals: int | None = None,
    token1_decimals: int | None = None,
    quoted_price: Decimal | None = None,
    **kwargs: Any,
)

Parser for Uniswap V3 transaction receipts.

Refactored to use base infrastructure utilities for hex decoding and event registry management. Maintains full backward compatibility.

Initialize the parser.

Parameters:

Name	Type	Description	Default
`chain`	`str`	Blockchain network (for token symbol resolution)	`'arbitrum'`
`token0_address`	`str \| None`	Address of token0 in the pool	`None`
`token1_address`	`str \| None`	Address of token1 in the pool	`None`
`token0_symbol`	`str \| None`	Symbol of token0	`None`
`token1_symbol`	`str \| None`	Symbol of token1	`None`
`token0_decimals`	`int \| None`	Decimals for token0	`None`
`token1_decimals`	`int \| None`	Decimals for token1	`None`
`quoted_price`	`Decimal \| None`	Expected price for slippage calculation	`None`

build_extract_kwargs ¶

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

Return Uniswap-V3-owned kwargs for ResultEnricher extraction calls.

VIB-3164: the compiler records full token identity (from_token / to_token dicts with address, symbol, decimals — see UniswapV3Compiler.compile_swap) in ActionBundle.metadata. Threading it here lets _build_swap_result resolve decimals when the TokenResolver misses or Transfer events cannot be classified, instead of falling through to the 18-decimal default.

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.

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.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict	required
`quoted_amount_out`	`int \| None`	Expected output amount for slippage calculation	`None`
`swap_token_meta`	`dict[str, dict[str, Any]] \| None`	Optional compiler-supplied token metadata (VIB-3164). Forwarded to `_build_swap_result` so decimals resolve when the TokenResolver misses or Transfer events cannot be classified. Shape: `{"token_in": {"address": ..., "symbol": ..., "decimals": ...}, "token_out": {...}}`.	`None`

Returns:

Type	Description
`ParseResult`	ParseResult with extracted events and swap data

parse_logs ¶

parse_logs(
    logs: list[dict[str, Any]],
) -> list[UniswapV3Event]

Parse a list of logs.

Parameters:

Name	Type	Description	Default
`logs`	`list[dict[str, Any]]`	List of log dicts	required

Returns:

Type	Description
`list[UniswapV3Event]`	List of parsed events

extract_position_id_result ¶

extract_position_id_result(
    receipt: dict[str, Any],
) -> ExtractResult[int]

Fail-closed variant of :meth:extract_position_id — see VIB-3159.

extract_swap_amounts_result ¶

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

Fail-closed variant of :meth:extract_swap_amounts — see VIB-3159.

VIB-3203: forwards expected_out so realized slippage is populated when the framework supplies the compiler's pre-slippage quote. VIB-3164: forwards swap_token_meta so compiler-supplied token decimals resolve when the TokenResolver misses.

extract_lp_close_data_result ¶

extract_lp_close_data_result(
    receipt: dict[str, Any],
) -> ExtractResult[LPCloseData]

Fail-closed variant of :meth:extract_lp_close_data — see VIB-3159.

extract_lp_open_data_result ¶

extract_lp_open_data_result(
    receipt: dict[str, Any],
) -> ExtractResult[LPOpenData]

Fail-closed variant of :meth:extract_lp_open_data — see VIB-3159.

Distinguishes "no IncreaseLiquidity event" (benign — e.g. LP_OPEN that failed mid-bundle) from "parser crashed". Both are returned as None by the legacy method, which forces the enricher to treat genuine parse failures as missing data — exactly the ghost-position class of bug VIB-3159 addresses.

extract_liquidity_result ¶

extract_liquidity_result(
    receipt: dict[str, Any],
) -> ExtractResult[int]

Fail-closed variant of :meth:extract_liquidity — see VIB-3159.

extract_position_id ¶

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

Extract LP position ID (NFT tokenId) from a transaction receipt.

Looks for ERC-721 Transfer events from the NonfungiblePositionManager where from=address(0), indicating a mint (new position created).

For ERC-721 Transfer events, the signature is: Transfer(address indexed from, address indexed to, uint256 indexed tokenId) All parameters are indexed, so tokenId is in topics[3], not in data.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Position ID (tokenId) if found, None otherwise

Example

parser = UniswapV3ReceiptParser(chain="arbitrum") position_id = parser.extract_position_id(receipt) if position_id: ... print(f"Opened position: {position_id}")

extract_position_id_from_logs `staticmethod` ¶

extract_position_id_from_logs(
    logs: list[dict[str, Any]], chain: str = "arbitrum"
) -> int | None

Static method to extract position ID from logs without instantiating parser.

Convenience method for cases where you just need to extract the position ID without parsing other events.

Parameters:

Name	Type	Description	Default
`logs`	`list[dict[str, Any]]`	List of log dicts from transaction receipt	required
`chain`	`str`	Chain name for position manager address lookup	`'arbitrum'`

Returns:

Type	Description
`int \| None`	Position ID (tokenId) if found, None otherwise

Example

position_id = UniswapV3ReceiptParser.extract_position_id_from_logs( ... receipt["logs"], chain="arbitrum" ... )

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 from a transaction receipt.

This method is called by the ResultEnricher to automatically populate ExecutionResult.swap_amounts for SWAP intents.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required
`expected_out`	`Decimal \| None`	VIB-3203 — pre-slippage-discount quote in human (Decimal) units, sourced from `ActionBundle.metadata["expected_output_human"]`. When provided, realized `slippage_bps` is computed as `(expected_out - amount_out_decimal) / expected_out * 10_000`. When absent, `slippage_bps` stays `None` (legacy behavior).	`None`
`swap_token_meta`	`dict[str, dict[str, Any]] \| None`	VIB-3164 — compiler-supplied token metadata. Shape: `{"token_in": {"address": ..., "symbol": ..., "decimals": ...}, "token_out": {...}}`. Forwarded to `parse_receipt` so the parser can resolve decimals when the TokenResolver misses or Transfer events cannot be classified against the pool address.	`None`

Returns:

Type	Description
`SwapAmounts \| None`	SwapAmounts dataclass if swap event found, None otherwise

Example

parser = UniswapV3ReceiptParser(chain="arbitrum") swap_amounts = parser.extract_swap_amounts(receipt) if swap_amounts: ... print(f"Swapped: {swap_amounts.amount_in_decimal}")

extract_tick_lower ¶

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

Extract tick lower from LP mint transaction receipt.

Looks for Mint events from Uniswap V3 pools. tickLower is an indexed parameter in topics[2].

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Tick lower value if found, None otherwise

extract_tick_upper ¶

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

Extract tick upper from LP mint transaction receipt.

Looks for Mint events from Uniswap V3 pools. tickUpper is an indexed parameter in topics[3].

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Tick upper value if found, None otherwise

extract_liquidity ¶

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

Extract liquidity from a Uniswap V3 Pool Mint event.

Pool Mint signature (only the non-indexed fields land in data)::

event Mint(
    address sender,             // non-indexed → data[ 0..32 ]
    address indexed owner,
    int24   indexed tickLower,
    int24   indexed tickUpper,
    uint128 amount,             // non-indexed → data[32..64 ]  (the liquidity)
    uint256 amount0,            // non-indexed → data[64..96 ]
    uint256 amount1             // non-indexed → data[96..128]
);

Sister parsers (sushiswap_v3, pancakeswap_v3) already read at offset 32 — this method used to read at offset 0 and surface the sender-address slot as "liquidity" (a ~50-digit garbage uint), which leaked into extracted_data['liquidity'] for every LP_OPEN (VIB-4395). The fix here aligns with the ABI; the canonical per-position liquidity is also available via :meth:extract_lp_open_data (NFT manager IncreaseLiquidity).

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`int \| None`	Liquidity amount if a Pool Mint event is found, `None` otherwise.

extract_lp_open_data ¶

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

Extract LP open data from a transaction receipt.

Looks for IncreaseLiquidity events emitted by the Uniswap V3 NonfungiblePositionManager (and its forks listed in POSITION_MANAGER_ADDRESSES) when an LP position is opened or topped up. The event signature is::

IncreaseLiquidity(
    uint256 indexed tokenId,
    uint128 liquidity,
    uint256 amount0,
    uint256 amount1,
)

Layout (after VIB-3658 / April 30 audit item #4):

topics[0]: keccak topic0 (constant)
topics[1]: NFT tokenId (indexed uint256)
data: liquidity (uint128, padded to 32 bytes)
- amount0 (uint256) + amount1 (uint256)

Behavior contract:

Returns LPOpenData populated with the raw on-chain ints (position_id, liquidity, amount0, amount1). The accounting handler (lp_handler.py) is responsible for decimal-scaling and USD valuation — this parser stays raw so tests don't have to mock the token resolver and the gateway price oracle.
Returns None when no IncreaseLiquidity log is present (e.g. an LP_OPEN that failed mid-bundle, or a non-NPM contract path) — never raises.
The IncreaseLiquidity event is emitted by the NPM for mint() AND increaseLiquidity() calls. We accept either: a fresh-mint NPM emits the same event right after the ERC-721 Transfer(0x0, owner, tokenId).

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field.	required

Returns:

Type	Description
`LPOpenData \| None`	`LPOpenData` if an `IncreaseLiquidity` event is present,
`LPOpenData \| None`	`None` otherwise.

extract_lp_close_data ¶

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

Extract LP close data from transaction receipt.

Decodes the Uniswap V3 LP close pattern: decreaseLiquidity emits Burn (which carries the principal amounts), and collect emits Collect (which carries principal PLUS earned fees). The accrued fees are the difference between the two.

Burn(address indexed owner, int24 indexed tickLower,

int24 indexed tickUpper, uint128 amount, uint256 amount0, uint256 amount1)

data layout: amount (uint128, left-padded to 32B) ‖ amount0 (uint256) ‖ amount1 (uint256)

Collect(address indexed owner, address recipient,

int24 indexed tickLower, int24 indexed tickUpper, uint128 amount0, uint128 amount1)

owner, tickLower, tickUpper are indexed (3 topics + topic0). recipient is non-indexed — it occupies the first 32-byte data slot, so amount0/amount1 start at offsets 32 and 64, not 0 and 32.
data layout: recipient (address, left-padded to 32B) ‖ amount0 (uint128, left-padded to 32B) ‖ amount1 (uint128, left-padded to 32B)

For a fee-only collect() (no decreaseLiquidity in the same TX — e.g. an in-range fee harvest), there is no Burn event. We treat the full Collect amounts as fees, with principal = 0.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	required

Returns:

Type	Description
`LPCloseData \| None`	LPCloseData if Burn or Collect event found, None otherwise.

extract_registry_payload_open ¶

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

Build the LP_OPEN position_registry.payload dict.

Reads the existing :meth:extract_lp_open_data output for token_id / tick_lower / tick_upper / liquidity / amount0 / amount1 / pool_address and composes the canonical 8-key shape (plus optional fee_tier and the per-chain nft_manager_addr). Returns None when any of the load-bearing identity fields are missing — the caller treats that as "fall back to accounting_only", per CLAUDE.md "Empty ≠ zero" (a zero-substituted token_id would silently corrupt the physical_identity_hash).

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with `logs` field.	required
`fee_tier`	`int \| None`	Optional pool fee tier (e.g. `500` for 0.05%); forwarded from the intent's compile-time metadata. `None` when unknown — the payload key stays absent rather than substituting `0` (Empty ≠ zero).	`None`

Returns:

Type	Description
`dict[str, Any] \| None`	`dict` JSON-serializable with the 8 (or 9 with fee_tier) keys
`dict[str, Any] \| None`	ratified by PRD §Registry Data Shape and the T08 golden, OR
`dict[str, Any] \| None`	`None` when the LP_OPEN data isn't extractable from the
`dict[str, Any] \| None`	receipt.

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 LP_CLOSE position_registry.payload dict.

Reads the existing :meth:extract_lp_close_data output (Burn / Collect amounts) and the close-side DecreaseLiquidity event for the NFT token_id, then composes the 13-key shape that the T08 lp_close/expected_registry_row.json golden specifies.

Audit M1 (CodeRabbit): a real UniV3 LP_CLOSE proves itself with DecreaseLiquidity on the receipt AND a Burn log carrying the pool address. A Collect-only receipt is NOT a close — it's a fee harvest. If we silently synthesized token_id / pool_address from open_payload here, a Collect-only receipt or a malformed close would produce a "successful" close payload with stale OPEN-side anchors, and the registry would mark a still-open NFT as closed (the cutover spec D3.F6 silent-error class).

The flow is:

Decode close-side events (extract_lp_close_data) and the DecreaseLiquidity log (_decreaseliquidity_token_id).
Verify the receipt-derived identity anchors are present and non-zero.
Cross-check against open_payload if supplied — refuse on any disagreement (_open_payload_disagrees).
Compose the receipt-only payload (_build_close_receipt_payload).
Merge OPEN-time fields the close receipt cannot re-derive (_merge_open_payload_fields) — ticks, OPEN-time amounts, original mint liquidity, fee tier, token labels.
Apply the fee_tier argument if open_payload didn't carry one.

Returns None when the close-side identity anchors (token_id + pool_address) cannot be derived OR cross-checks fail. The caller treats that as "fall back to accounting_only" with an ERROR log (no zero substitution).

extract_protocol_fees ¶

extract_protocol_fees(
    receipt: dict[str, Any],
    *,
    fee_tier_bps: int | None = None,
) -> ProtocolFees | None

Extract DEX protocol fees from a swap receipt.

Uniswap V3 swap fees are a pool-wide basis-points slice of amount_in. The numeric fee tier is resolved at compile time and forwarded by the ResultEnricher via bundle_metadata["selected_fee_tier"] so the parser does not need to re-read pool state.

Fee amount (token-denominated):

fee_amount_in = amount_in * fee_tier_bps / 1_000_000

(fee_tier_bps is Uniswap's pip-based tier, e.g. 500 = 0.05%, so the divisor is 1_000_000 — not 10_000.)

VIB-3204 audit contract: until a price oracle is plumbed through to this layer, the parser returns None (unknown) — never a ProtocolFees(total_usd=Decimal(0)), which would falsely advertise "measured to be zero" and cause PnL attribution to under-attribute swap costs. Callers that want the in-token fee can derive it from result.swap_amounts.amount_in_decimal * fee_tier_bps / 1_000_000 using values already on result.swap_amounts.

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field.	required
`fee_tier_bps`	`int \| None`	Pool fee tier, forwarded from `bundle_metadata["selected_fee_tier"]` by the enricher.	`None`

Returns:

Type	Description
`ProtocolFees \| None`	`None` — USD conversion not available at this layer. A
`ProtocolFees \| None`	future iteration with price-oracle access will return a
`ProtocolFees \| None`	populated `ProtocolFees`. Also returns `None` when no
`ProtocolFees \| None`	Swap event is present OR when `fee_tier_bps` is missing.

is_uniswap_event ¶

is_uniswap_event(topic: str | bytes) -> bool

Check if a topic is a known Uniswap V3 event.

Parameters:

Name	Type	Description	Default
`topic`	`str \| bytes`	Event topic (supports bytes, hex string with/without 0x, any case)	required

Returns:

Type	Description
`bool`	True if topic is a known Uniswap V3 event

get_event_type ¶

get_event_type(topic: str | bytes) -> UniswapV3EventType

Get the event type for a topic.

Parameters:

Name	Type	Description	Default
`topic`	`str \| bytes`	Event topic (supports bytes, hex string with/without 0x, any case)	required

Returns:

Type	Description
`UniswapV3EventType`	Event type or UNKNOWN

InvalidFeeError ¶

InvalidFeeError(fee: int)

Bases: UniswapV3SDKError

Invalid fee tier provided.

InvalidTickError ¶

InvalidTickError(tick: int, reason: str)

Bases: UniswapV3SDKError

Invalid tick value provided.

PoolInfo `dataclass` ¶

PoolInfo(
    address: str,
    token0: str,
    token1: str,
    fee: int,
    tick_spacing: int,
)

Information about a Uniswap V3 pool.

Attributes:

Name	Type	Description
`address`	`str`	Computed pool address
`token0`	`str`	First token address (sorted)
`token1`	`str`	Second token address (sorted)
`fee`	`int`	Fee tier in hundredths of a bip
`tick_spacing`	`int`	Tick spacing for this fee tier

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PoolNotFoundError ¶

PoolNotFoundError(token0: str, token1: str, fee: int)

Bases: UniswapV3SDKError

Pool does not exist.

PoolState `dataclass` ¶

PoolState(
    sqrt_price_x96: int,
    tick: int,
    liquidity: int,
    fee_growth_global_0: int = 0,
    fee_growth_global_1: int = 0,
)

Current state of a Uniswap V3 pool.

Attributes:

Name	Type	Description
`sqrt_price_x96`	`int`	Current sqrt price (Q64.96 format)
`tick`	`int`	Current tick
`liquidity`	`int`	Current in-range liquidity
`fee_growth_global_0`	`int`	Fee growth for token0
`fee_growth_global_1`	`int`	Fee growth for token1

price `property` ¶

price: Decimal

Get current price (token1 per token0).

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

QuoteError ¶

QuoteError(message: str, token_in: str, token_out: str)

Bases: UniswapV3SDKError

Error fetching quote.

SwapTransaction `dataclass` ¶

SwapTransaction(
    to: str,
    value: int,
    data: str,
    gas_estimate: int,
    description: str,
)

Transaction data for a swap.

Attributes:

Name	Type	Description
`to`	`str`	Router address
`value`	`int`	ETH value to send (for native token swaps)
`data`	`str`	Encoded calldata
`gas_estimate`	`int`	Estimated gas
`description`	`str`	Human-readable description

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

UniswapV3SDK ¶

UniswapV3SDK(
    chain: str,
    rpc_url: str | None = None,
    web3: Any | None = None,
    gateway_client: GatewayClient | None = None,
)

SDK for Uniswap V3 operations.

This class provides methods for: - Computing pool addresses - Fetching quotes (requires RPC) - Building swap transactions - Tick math utilities

Example

sdk = UniswapV3SDK(chain="arbitrum", rpc_url="https://arb1.arbitrum.io/rpc")

Compute pool address (no RPC needed)¶

pool_info = sdk.get_pool_address( "0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # WETH "0xaf88d065e77c8cC2239327C5EDb3A432268e5831", # USDC fee_tier=3000, )

Get quote (requires RPC)¶

quote = await sdk.get_quote( token_in=weth_address, token_out=usdc_address, amount_in=10**18, # 1 WETH fee_tier=3000, )

Initialize the SDK.

Parameters:

Name	Type	Description	Default
`chain`	`str`	Target blockchain (ethereum, arbitrum, optimism, polygon, base)	required
`rpc_url`	`str \| None`	DEPRECATED — direct RPC URL. Prefer gateway_client.	`None`
`web3`	`Any \| None`	Existing Web3 instance (optional)	`None`
`gateway_client`	`GatewayClient \| None`	Gateway client for routing async eth_call through the gateway's RpcService. Preferred over rpc_url.	`None`

Raises:

Type	Description
`ValueError`	If chain is not supported

get_pool_address ¶

get_pool_address(
    token0: str, token1: str, fee_tier: int
) -> PoolInfo

Get the pool address for a token pair.

This computes the CREATE2 address deterministically without RPC calls.

Parameters:

Name	Type	Description	Default
`token0`	`str`	First token address	required
`token1`	`str`	Second token address	required
`fee_tier`	`int`	Fee tier (100, 500, 3000, 10000)	required

Returns:

Type	Description
`PoolInfo`	PoolInfo with computed address and token ordering

Raises:

Type	Description
`InvalidFeeError`	If fee_tier is not valid

Example

pool = sdk.get_pool_address(weth, usdc, fee_tier=3000) print(f"Pool address: {pool.address}")

get_quote `async` ¶

get_quote(
    token_in: str,
    token_out: str,
    amount_in: int,
    fee_tier: int,
) -> SwapQuote

Get a quote for a swap.

This fetches the expected output amount for a given input amount by calling the QuoterV2 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 wei	required
`fee_tier`	`int`	Fee tier	required

Returns:

Type	Description
`SwapQuote`	SwapQuote with expected output

Raises:

Type	Description
`QuoteError`	If quote fails
`InvalidFeeError`	If fee_tier is not valid

Note

Requires RPC connection. Use get_quote_local for offline estimation.

get_quote_local ¶

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

Get an estimated quote without RPC calls.

This provides an approximation based on fee tier only. For accurate quotes, use get_quote() with 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 wei	required
`fee_tier`	`int`	Fee tier	required
`price_ratio`	`Decimal \| None`	Optional token_out/token_in price ratio	`None`

Returns:

Type	Description
`SwapQuote`	SwapQuote with estimated output

build_swap_tx ¶

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

Build a swap transaction from a quote.

Parameters:

Name	Type	Description	Default
`quote`	`SwapQuote`	Quote from get_quote()	required
`recipient`	`str`	Address to receive output tokens	required
`slippage_bps`	`int`	Slippage tolerance in basis points	required
`deadline`	`int`	Unix timestamp deadline	required
`value`	`int`	ETH value to send (for native token swaps)	`0`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction with encoded calldata

Example

quote = await sdk.get_quote(weth, usdc, 10**18, 3000) tx = sdk.build_swap_tx( ... quote=quote, ... recipient="0x...", ... slippage_bps=50, ... deadline=int(time.time()) + 300, ... )

build_exact_output_swap_tx ¶

build_exact_output_swap_tx(
    token_in: str,
    token_out: str,
    fee: int,
    recipient: str,
    deadline: int,
    amount_out: int,
    amount_in_maximum: int,
    value: int = 0,
) -> SwapTransaction

Build an exact output swap transaction.

For swaps where you specify the exact output amount.

Parameters:

Name	Type	Description	Default
`token_in`	`str`	Input token address	required
`token_out`	`str`	Output token address	required
`fee`	`int`	Fee tier	required
`recipient`	`str`	Address to receive output tokens	required
`deadline`	`int`	Unix timestamp deadline	required
`amount_out`	`int`	Exact output amount desired	required
`amount_in_maximum`	`int`	Maximum input amount (with slippage)	required
`value`	`int`	ETH value to send (for native token swaps)	`0`

Returns:

Type	Description
`SwapTransaction`	SwapTransaction with encoded calldata

UniswapV3SDKError ¶

Bases: Exception

Base exception for Uniswap V3 SDK errors.

SDKSwapQuote `dataclass` ¶

SDKSwapQuote(
    token_in: str,
    token_out: str,
    amount_in: int,
    amount_out: int,
    fee: int,
    sqrt_price_x96_after: int = 0,
    initialized_ticks_crossed: int = 0,
    gas_estimate: int = 150000,
    quoted_at: datetime = (lambda: datetime.now(UTC))(),
)

Quote for a swap operation.

Attributes:

Name	Type	Description
`token_in`	`str`	Input token address
`token_out`	`str`	Output token address
`amount_in`	`int`	Input amount in wei
`amount_out`	`int`	Expected output amount in wei
`fee`	`int`	Fee tier
`sqrt_price_x96_after`	`int`	Price after swap
`initialized_ticks_crossed`	`int`	Number of initialized ticks crossed
`gas_estimate`	`int`	Estimated gas for the swap
`quoted_at`	`datetime`	Timestamp when quote was fetched

effective_price `property` ¶

effective_price: Decimal

Calculate effective price of the swap.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> SwapQuote

Create from dictionary.

compute_pool_address ¶

compute_pool_address(
    factory: str,
    token0: str,
    token1: str,
    fee: int,
    init_code_hash: str = POOL_INIT_CODE_HASH,
) -> str

Compute the CREATE2 address for a Uniswap V3 pool.

This deterministically computes the pool address without any RPC calls.

Parameters:

Name	Type	Description	Default
`factory`	`str`	Factory contract address	required
`token0`	`str`	First token address	required
`token1`	`str`	Second token address	required
`fee`	`int`	Fee tier	required

Returns:

Type	Description
`str`	Pool address

Raises:

Type	Description
`InvalidFeeError`	If fee is not a valid tier

Example

pool_addr = compute_pool_address( ... factory="0x1F98431c8aD98523631AE4a59f267346ea31F984", ... token0="0x82aF49447D8a07e3bd95BD0d56f35241523fBab1", # WETH ... token1="0xaf88d065e77c8cC2239327C5EDb3A432268e5831", # USDC ... fee=3000, ... )

get_max_tick ¶

get_max_tick(fee: int) -> int

Get the maximum valid tick for a fee tier.

Parameters:

Name	Type	Description	Default
`fee`	`int`	Fee tier	required

Returns:

Type	Description
`int`	Maximum valid tick

Raises:

Type	Description
`InvalidFeeError`	If fee is not a valid tier

get_min_tick ¶

get_min_tick(fee: int) -> int

Get the minimum valid tick for a fee tier.

Parameters:

Name	Type	Description	Default
`fee`	`int`	Fee tier	required

Returns:

Type	Description
`int`	Minimum valid tick

Raises:

Type	Description
`InvalidFeeError`	If fee is not a valid tier

get_nearest_tick ¶

get_nearest_tick(tick: int, fee: int) -> int

Get the nearest valid tick for a given fee tier.

Parameters:

Name	Type	Description	Default
`tick`	`int`	Raw tick value	required
`fee`	`int`	Fee tier	required

Returns:

Type	Description
`int`	Nearest valid tick

Raises:

Type	Description
`InvalidFeeError`	If fee is not a valid tier

price_to_sqrt_price_x96 ¶

price_to_sqrt_price_x96(price: Decimal | float) -> int

Convert a decimal price to sqrt price X96 format.

Parameters:

Name	Type	Description	Default
`price`	`Decimal \| float`	Price as decimal	required

Returns:

Type	Description
`int`	Sqrt price in Q64.96 format

price_to_tick ¶

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

Convert a price to the nearest tick.

Parameters:

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

Returns:

Type	Description
`int`	Tick value (may not be on a valid tick spacing boundary)

sort_tokens ¶

sort_tokens(token0: str, token1: str) -> tuple[str, str]

Sort two token addresses in ascending order.

Uniswap V3 pools always order tokens such that token0 < token1.

Parameters:

Name	Type	Description	Default
`token0`	`str`	First token address	required
`token1`	`str`	Second token address	required

Returns:

Type	Description
`tuple[str, str]`	Tuple of (token0, token1) sorted in ascending order

sqrt_price_x96_to_price ¶

sqrt_price_x96_to_price(sqrt_price_x96: int) -> Decimal

Convert sqrt price X96 to a decimal price.

Parameters:

Name	Type	Description	Default
`sqrt_price_x96`	`int`	Sqrt price in Q64.96 format	required

Returns:

Type	Description
`Decimal`	Price as Decimal

sqrt_price_x96_to_tick ¶

sqrt_price_x96_to_tick(sqrt_price_x96: int) -> int

Convert sqrt price in Q64.96 format to tick.

Parameters:

Name	Type	Description	Default
`sqrt_price_x96`	`int`	Sqrt price in Q64.96 format	required

Returns:

Type	Description
`int`	Tick value

Raises:

Type	Description
`ValueError`	If sqrt_price_x96 is invalid

tick_to_price ¶

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

Convert a tick to a human-readable price.

Parameters:

Name	Type	Description	Default
`tick`	`int`	Tick value	required
`decimals0`	`int`	Decimals of token0	`18`
`decimals1`	`int`	Decimals of token1	`18`

Returns:

Type	Description
`Decimal`	Price of token0 in terms of token1 (adjusted for decimals)

tick_to_sqrt_price_x96 ¶

tick_to_sqrt_price_x96(tick: int) -> int

Convert a tick to sqrt price in Q64.96 format.

Uses the formula: sqrt(1.0001^tick) * 2^96

Parameters:

Name	Type	Description	Default
`tick`	`int`	Tick value	required

Returns:

Type	Description
`int`	Sqrt price in Q64.96 format

Raises:

Type	Description
`InvalidTickError`	If tick is out of bounds

getattr ¶

__getattr__(name: str) -> Any

PEP 562 lazy attribute access.

Uniswap V3¶

almanak.connectors.uniswap_v3 ¶

Execute a swap¶

Use SDK for lower-level operations¶

SwapQuote dataclass ¶

to_dict ¶

SwapResult dataclass ¶

to_dict ¶

SwapType ¶

TransactionData dataclass ¶

to_dict ¶

UniswapV3Adapter ¶

Execute a swap¶

Compile a SwapIntent to ActionBundle¶

swap_exact_input ¶

swap_exact_output ¶

compile_swap_intent ¶

set_allowance ¶

clear_allowance_cache ¶

UniswapV3Config dataclass ¶

__post_init__ ¶

to_dict ¶

UniswapV3LPAdapter ¶

get_position_manager_address ¶

get_mint_calldata ¶

get_decrease_liquidity_calldata ¶

get_collect_calldata ¶

get_burn_calldata ¶

estimate_mint_gas ¶

estimate_close_gas ¶

ParsedSwapResult dataclass ¶

to_dict ¶

from_dict classmethod ¶

to_swap_result_payload ¶

ParseResult dataclass ¶

to_dict ¶

SwapEventData dataclass ¶

token0_is_input property ¶

token1_is_input property ¶

amount_in property ¶

amount_out property ¶

to_dict ¶

from_dict classmethod ¶

TransferEventData dataclass ¶

to_dict ¶

UniswapV3Event dataclass ¶

to_dict ¶

from_dict classmethod ¶

UniswapV3EventType ¶

UniswapV3ReceiptParser ¶

build_extract_kwargs ¶

parse_receipt ¶

parse_logs ¶

extract_position_id_result ¶

extract_swap_amounts_result ¶

extract_lp_close_data_result ¶

extract_lp_open_data_result ¶

extract_liquidity_result ¶

extract_position_id ¶

extract_position_id_from_logs staticmethod ¶

extract_swap_amounts ¶

extract_tick_lower ¶

extract_tick_upper ¶

extract_liquidity ¶

extract_lp_open_data ¶

extract_lp_close_data ¶

extract_registry_payload_open ¶

extract_registry_payload_close ¶

extract_protocol_fees ¶

is_uniswap_event ¶

get_event_type ¶

InvalidFeeError ¶

InvalidTickError ¶

PoolInfo dataclass ¶

to_dict ¶

PoolNotFoundError ¶

PoolState dataclass ¶

price property ¶

to_dict ¶

QuoteError ¶

SwapQuote `dataclass` ¶

SwapResult `dataclass` ¶

TransactionData `dataclass` ¶

UniswapV3Config `dataclass` ¶

ParsedSwapResult `dataclass` ¶

from_dict `classmethod` ¶

ParseResult `dataclass` ¶

SwapEventData `dataclass` ¶

token0_is_input `property` ¶

token1_is_input `property` ¶

amount_in `property` ¶

amount_out `property` ¶

from_dict `classmethod` ¶

TransferEventData `dataclass` ¶

UniswapV3Event `dataclass` ¶

from_dict `classmethod` ¶

extract_position_id_from_logs `staticmethod` ¶

PoolInfo `dataclass` ¶

PoolState `dataclass` ¶

price `property` ¶

SwapTransaction `dataclass` ¶

get_quote `async` ¶

SDKSwapQuote `dataclass` ¶

effective_price `property` ¶

from_dict `classmethod` ¶

getattr ¶