SushiSwap V3¶

Connector for SushiSwap V3 DEX.

almanak.framework.connectors.sushiswap_v3 ¶

SushiSwap V3 Connector Package.

This package provides the SushiSwap V3 protocol integration for the Almanak Strategy Framework, including SDK, adapter, and receipt parser.

SushiSwap V3 is a concentrated liquidity AMM forked from Uniswap V3, deployed across multiple chains including Arbitrum, Ethereum, Base, Polygon, Avalanche, BSC, and Optimism.

Key Components: - SushiSwapV3SDK: Low-level SDK for direct protocol interaction - SushiSwapV3Adapter: High-level adapter for framework integration - SushiSwapV3ReceiptParser: Transaction receipt parsing

Supported Operations: - Token swaps (exact input and exact output) - LP position management (mint, increase, decrease, collect) - Quote fetching - Pool address computation

Example

from almanak.framework.connectors.sushiswap_v3 import ( SushiSwapV3SDK, SushiSwapV3Adapter, SushiSwapV3Config, SushiSwapV3ReceiptParser, )

Using the SDK directly¶

sdk = SushiSwapV3SDK(chain="arbitrum") pool = sdk.get_pool_address(weth_address, usdc_address, fee_tier=3000)

Using the adapter¶

config = SushiSwapV3Config( chain="arbitrum", wallet_address="0x...", price_provider={"ETH": Decimal("3400"), "USDC": Decimal("1")}, ) adapter = SushiSwapV3Adapter(config) result = adapter.swap_exact_input("USDC", "WETH", Decimal("1000"))

Parsing receipts¶

parser = SushiSwapV3ReceiptParser(chain="arbitrum") parse_result = parser.parse_receipt(receipt)

LPResult `dataclass` ¶

LPResult(
    success: bool,
    transactions: list[TransactionData] = list(),
    error: str | None = None,
    gas_estimate: int = 0,
    position_info: dict[str, Any] = dict(),
)

Result of an LP operation.

Attributes:

Name	Type	Description
`success`	`bool`	Whether the operation was built successfully
`transactions`	`list[TransactionData]`	List of transactions to execute
`error`	`str \| None`	Error message if failed
`gas_estimate`	`int`	Total gas estimate for all transactions
`position_info`	`dict[str, Any]`	Additional info about the LP position

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

SushiSwapV3Adapter ¶

SushiSwapV3Adapter(
    config: SushiSwapV3Config,
    token_resolver: TokenResolver | None = None,
)

Adapter for SushiSwap V3 DEX protocol.

This adapter provides methods for: - Executing token swaps (exact input and exact output) - Building swap transactions - Managing LP positions (mint, increase, decrease, collect) - Handling ERC-20 approvals - Managing slippage protection

Example

config = SushiSwapV3Config( chain="arbitrum", wallet_address="0x...", price_provider={"ETH": Decimal("3400"), "USDC": Decimal("1")}, ) adapter = SushiSwapV3Adapter(config)

Execute a swap¶

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

Open LP position¶

lp_result = adapter.open_lp_position( token0="USDC", token1="WETH", amount0=Decimal("1000"), amount1=Decimal("0.5"), fee_tier=3000, tick_lower=-887220, tick_upper=887220, )

Initialize the adapter.

Parameters:

Name	Type	Description	Default
`config`	`SushiSwapV3Config`	SushiSwap 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

open_lp_position ¶

open_lp_position(
    token0: str,
    token1: str,
    amount0: Decimal,
    amount1: Decimal,
    fee_tier: int | None = None,
    tick_lower: int = -887220,
    tick_upper: int = 887220,
    slippage_bps: int | None = None,
    recipient: str | None = None,
) -> LPResult

Build transactions to open a new LP position.

Parameters:

Name	Type	Description	Default
`token0`	`str`	First token symbol or address	required
`token1`	`str`	Second token symbol or address	required
`amount0`	`Decimal`	Amount of token0 to provide	required
`amount1`	`Decimal`	Amount of token1 to provide	required
`fee_tier`	`int \| None`	Pool fee tier (default from config)	`None`
`tick_lower`	`int`	Lower tick bound (default: full range)	`-887220`
`tick_upper`	`int`	Upper tick bound (default: full range)	`887220`
`slippage_bps`	`int \| None`	Slippage tolerance in basis points (default from config)	`None`
`recipient`	`str \| None`	Address to receive the NFT (default: wallet_address)	`None`

Returns:

Type	Description
`LPResult`	LPResult with transaction data

close_lp_position ¶

close_lp_position(
    token_id: int,
    liquidity: int,
    amount0_min: int = 0,
    amount1_min: int = 0,
    recipient: str | None = None,
) -> LPResult

Build transactions to close an LP position.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the position	required
`liquidity`	`int`	Amount of liquidity to remove (use position's full liquidity to close)	required
`amount0_min`	`int`	Minimum amount of token0 to receive	`0`
`amount1_min`	`int`	Minimum amount of token1 to receive	`0`
`recipient`	`str \| None`	Address to receive tokens (default: wallet_address)	`None`

Returns:

Type	Description
`LPResult`	LPResult with transaction data

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.

SushiSwapV3Config `dataclass` ¶

SushiSwapV3Config(
    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 SushiSwapV3Adapter.

Attributes:

Name	Type	Description
`chain`	`str`	Target blockchain (ethereum, arbitrum, base, polygon, avalanche, bsc, optimism)
`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.

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 = SUSHISWAP_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, mint, etc.)

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[SushiSwapV3Event] = 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.

SushiSwapV3Event `dataclass` ¶

SushiSwapV3Event(
    event_type: SushiSwapV3EventType,
    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 SushiSwap V3 event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

from_dict `classmethod` ¶

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

Create from dictionary.

SushiSwapV3EventType ¶

Bases: Enum

SushiSwap V3 event types.

SushiSwapV3ReceiptParser ¶

SushiSwapV3ReceiptParser(
    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 = 18,
    token1_decimals: int = 18,
    quoted_price: Decimal | None = None,
    **kwargs: Any,
)

Parser for SushiSwap V3 transaction receipts.

Note: This parser uses HexDecoder and EventRegistry from the base module. A future refactor could inherit from BaseReceiptParser for full standardization.

This parser handles: - Swap events (exact input and exact output) - Mint events (new LP positions) - Burn events (decrease liquidity) - Collect events (claim fees/tokens) - Transfer events (ERC-20 and ERC-721)

Example

parser = SushiSwapV3ReceiptParser( chain="arbitrum", token0_address="", # e.g., WETH token1_address="", # e.g., USDC ) result = parser.parse_receipt(receipt)

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`	Decimals for token0	`18`
`token1_decimals`	`int`	Decimals for token1	`18`
`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[SushiSwapV3Event]

Parse a list of logs.

Parameters:

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

Returns:

Type	Description
`list[SushiSwapV3Event]`	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 = SushiSwapV3ReceiptParser(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 = SushiSwapV3ReceiptParser.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 = SushiSwapV3ReceiptParser(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 SushiSwap 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 SushiSwap 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 SushiSwap V3 pools. Liquidity amount is in the data field.

Mint event data layout (non-indexed fields, 32-byte padded): - sender (address): offset 0 - amount (uint128): offset 32 - amount0 (uint256): offset 64 - amount1 (uint256): offset 96

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 SushiSwap 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_sushiswap_event ¶

is_sushiswap_event(topic: str | bytes) -> bool

Check if a topic is a known SushiSwap 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 SushiSwap V3 event

get_event_type ¶

get_event_type(topic: str | bytes) -> SushiSwapV3EventType

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
`SushiSwapV3EventType`	Event type or UNKNOWN

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.

In V3 swaps: - Positive amount0/amount1 = tokens going INTO the pool (user spent) - Negative amount0/amount1 = tokens going OUT of the pool (user received)

token0_is_input `property` ¶

token0_is_input: bool

Check if token0 is the input token (user paid token0).

token1_is_input `property` ¶

token1_is_input: bool

Check if token1 is the input token (user paid token1).

amount_in `property` ¶

amount_in: int

Get the absolute input amount (what user paid).

amount_out `property` ¶

amount_out: int

Get the absolute output amount (what user received).

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.

InvalidFeeError ¶

InvalidFeeError(fee: int)

Bases: SushiSwapV3SDKError

Invalid fee tier provided.

InvalidTickError ¶

InvalidTickError(tick: int, reason: str)

Bases: SushiSwapV3SDKError

Invalid tick value provided.

LPTransaction `dataclass` ¶

LPTransaction(
    to: str,
    value: int,
    data: str,
    gas_estimate: int,
    description: str,
    operation: str,
)

Transaction data for LP operations.

Attributes:

Name	Type	Description
`to`	`str`	Position manager address
`value`	`int`	ETH value to send
`data`	`str`	Encoded calldata
`gas_estimate`	`int`	Estimated gas
`description`	`str`	Human-readable description
`operation`	`str`	LP operation type (mint, increase, decrease, collect)

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

MintParams `dataclass` ¶

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

Parameters for minting a new LP position.

Attributes:

Name	Type	Description
`token0`	`str`	First token address
`token1`	`str`	Second token address
`fee`	`int`	Fee tier
`tick_lower`	`int`	Lower tick bound
`tick_upper`	`int`	Upper tick bound
`amount0_desired`	`int`	Desired amount of token0
`amount1_desired`	`int`	Desired amount of token1
`amount0_min`	`int`	Minimum amount of token0 (slippage protection)
`amount1_min`	`int`	Minimum amount of token1 (slippage protection)
`recipient`	`str`	Address to receive the NFT
`deadline`	`int`	Transaction deadline timestamp

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PoolInfo `dataclass` ¶

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

Information about a SushiSwap 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: SushiSwapV3SDKError

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 SushiSwap 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: SushiSwapV3SDKError

Error fetching quote.

SushiSwapV3SDK ¶

SushiSwapV3SDK(
    chain: str,
    rpc_url: str | None = None,
    web3: Any | None = None,
)

SDK for SushiSwap V3 operations.

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

Example

sdk = SushiSwapV3SDK(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, base, polygon, avalanche, bsc, optimism)	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

build_mint_tx ¶

build_mint_tx(
    params: MintParams, value: int = 0
) -> LPTransaction

Build a transaction to mint a new LP position.

Parameters:

Name	Type	Description	Default
`params`	`MintParams`	Mint parameters	required
`value`	`int`	ETH value to send (if one token is WETH and user wants to use ETH)	`0`

Returns:

Type	Description
`LPTransaction`	LPTransaction with encoded calldata

build_increase_liquidity_tx ¶

build_increase_liquidity_tx(
    token_id: int,
    amount0_desired: int,
    amount1_desired: int,
    amount0_min: int,
    amount1_min: int,
    deadline: int,
    value: int = 0,
) -> LPTransaction

Build a transaction to increase liquidity in an existing position.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the position	required
`amount0_desired`	`int`	Desired amount of token0 to add	required
`amount1_desired`	`int`	Desired amount of token1 to add	required
`amount0_min`	`int`	Minimum amount of token0 (slippage protection)	required
`amount1_min`	`int`	Minimum amount of token1 (slippage protection)	required
`deadline`	`int`	Transaction deadline timestamp	required
`value`	`int`	ETH value to send	`0`

Returns:

Type	Description
`LPTransaction`	LPTransaction with encoded calldata

build_decrease_liquidity_tx ¶

build_decrease_liquidity_tx(
    token_id: int,
    liquidity: int,
    amount0_min: int,
    amount1_min: int,
    deadline: int,
) -> LPTransaction

Build a transaction to decrease liquidity in a position.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the position	required
`liquidity`	`int`	Amount of liquidity to remove	required
`amount0_min`	`int`	Minimum amount of token0 to receive	required
`amount1_min`	`int`	Minimum amount of token1 to receive	required
`deadline`	`int`	Transaction deadline timestamp	required

Returns:

Type	Description
`LPTransaction`	LPTransaction with encoded calldata

build_collect_tx ¶

build_collect_tx(
    token_id: int,
    recipient: str,
    amount0_max: int = MAX_UINT128,
    amount1_max: int = MAX_UINT128,
) -> LPTransaction

Build a transaction to collect fees/tokens from a position.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID of the position	required
`recipient`	`str`	Address to receive collected tokens	required
`amount0_max`	`int`	Maximum amount of token0 to collect (default: all)	`MAX_UINT128`
`amount1_max`	`int`	Maximum amount of token1 to collect (default: all)	`MAX_UINT128`

Returns:

Type	Description
`LPTransaction`	LPTransaction with encoded calldata

SushiSwapV3SDKError ¶

Bases: Exception

Base exception for SushiSwap V3 SDK errors.

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.

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 SushiSwap 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
`init_code_hash`	`str`	Pool init code hash (default for SushiSwap V3)	`POOL_INIT_CODE_HASH`

Returns:

Type	Description
`str`	Pool address

Raises:

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

Example

pool_addr = compute_pool_address( ... factory="0x1af415a1EbA07a4986a52B6f2e7dE7003D82231e", ... 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.

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

SushiSwap V3¶

almanak.framework.connectors.sushiswap_v3 ¶

Using the SDK directly¶

Using the adapter¶

Parsing receipts¶

LPResult dataclass ¶

to_dict ¶

SushiSwapV3Adapter ¶

Execute a swap¶

Open LP position¶

swap_exact_input ¶

swap_exact_output ¶

open_lp_position ¶

close_lp_position ¶

set_allowance ¶

clear_allowance_cache ¶

SushiSwapV3Config dataclass ¶

__post_init__ ¶

to_dict ¶

SwapQuote dataclass ¶

to_dict ¶

SwapResult dataclass ¶

to_dict ¶

SwapType ¶

TransactionData dataclass ¶

to_dict ¶

ParsedSwapResult dataclass ¶

to_dict ¶

from_dict classmethod ¶

to_swap_result_payload ¶

ParseResult dataclass ¶

to_dict ¶

SushiSwapV3Event dataclass ¶

to_dict ¶

from_dict classmethod ¶

SushiSwapV3EventType ¶

SushiSwapV3ReceiptParser ¶

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

get_event_type ¶

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 ¶

InvalidFeeError ¶

InvalidTickError ¶

LPTransaction dataclass ¶

to_dict ¶

MintParams dataclass ¶

to_dict ¶

PoolInfo dataclass ¶

to_dict ¶

PoolNotFoundError ¶

PoolState dataclass ¶

price property ¶

to_dict ¶

QuoteError ¶

SushiSwapV3SDK ¶

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 ¶

build_mint_tx ¶

build_increase_liquidity_tx ¶

LPResult `dataclass` ¶

SushiSwapV3Config `dataclass` ¶

SwapQuote `dataclass` ¶

SwapResult `dataclass` ¶

TransactionData `dataclass` ¶

ParsedSwapResult `dataclass` ¶

from_dict `classmethod` ¶

ParseResult `dataclass` ¶

SushiSwapV3Event `dataclass` ¶

from_dict `classmethod` ¶

extract_position_id_from_logs `staticmethod` ¶

SwapEventData `dataclass` ¶

token0_is_input `property` ¶

token1_is_input `property` ¶

amount_in `property` ¶

amount_out `property` ¶

from_dict `classmethod` ¶

TransferEventData `dataclass` ¶

LPTransaction `dataclass` ¶

MintParams `dataclass` ¶

PoolInfo `dataclass` ¶

PoolState `dataclass` ¶

price `property` ¶

get_quote `async` ¶

SwapTransaction `dataclass` ¶