Uniswap V3¶

Connector for Uniswap V3 concentrated liquidity DEX.

almanak.framework.connectors.uniswap_v3 ¶

Uniswap V3 Connector.

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

Supported chains: - Ethereum - Arbitrum - Optimism - Polygon - Base

Example

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

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

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`

parse_receipt ¶

parse_receipt(
    receipt: dict[str, Any],
    quoted_amount_out: int | 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`

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 ¶

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

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 LP mint transaction receipt.

Looks for Mint events from Uniswap V3 pools. Liquidity amount is in the data field.

Mint event data layout: - amount (uint128): offset 0 - amount0 (uint256): offset 16 - amount1 (uint256): offset 48

Parameters:

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

Returns:

Type	Description
`int \| None`	Liquidity amount if found, None otherwise

extract_lp_close_data ¶

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

Extract LP close data from transaction receipt.

Looks for Collect events from Uniswap V3 pools which indicate fees and principal being collected when closing/reducing a position.

Collect(address indexed owner, int24 indexed tickLower,

int24 indexed tickUpper, uint128 amount0, uint128 amount1)

Parameters:

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

Returns:

Type	Description
`LPCloseData \| None`	LPCloseData dataclass if Collect event found, None otherwise

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

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`	RPC URL for on-chain queries (optional)	`None`
`web3`	`Any \| None`	Existing Web3 instance (optional)	`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

Uniswap V3¶

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

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 ¶

parse_receipt ¶

parse_logs ¶

extract_position_id ¶

extract_position_id_from_logs staticmethod ¶

extract_swap_amounts ¶

extract_tick_lower ¶

extract_tick_upper ¶

extract_liquidity ¶

extract_lp_close_data ¶

is_uniswap_event ¶

get_event_type ¶

InvalidFeeError ¶

InvalidTickError ¶

PoolInfo dataclass ¶

to_dict ¶

PoolNotFoundError ¶

PoolState dataclass ¶

price property ¶

to_dict ¶

QuoteError ¶

SwapTransaction dataclass ¶

to_dict ¶

UniswapV3SDK ¶

Compute pool address (no RPC needed)¶

Get quote (requires RPC)¶

get_pool_address ¶

get_quote async ¶

get_quote_local ¶

build_swap_tx ¶

build_exact_output_swap_tx ¶

UniswapV3SDKError ¶

SDKSwapQuote dataclass ¶

effective_price property ¶

to_dict ¶

from_dict classmethod ¶

compute_pool_address ¶

get_max_tick ¶

get_min_tick ¶

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