SushiSwap V3

Connector for SushiSwap V3 DEX.

almanak.connectors.sushiswap_v3

SushiSwap V3 Connector Package (concentrated liquidity AMM).

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.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 | None = None,
    token1_decimals: int | None = None,
    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 | 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[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" ... )

build_extract_kwargs

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

Return SushiSwapV3-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 _resolve_swap_decimals_with_hints resolve decimals when the TokenResolver misses or Transfer events cannot be classified, instead of dropping the whole SwapAmounts row.

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.

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.

Resolves token decimals independently from ERC-20 Transfer events in the receipt, so it produces correct human-readable amounts even when the parser was constructed without token metadata (the enrichment path).

Parameters:

Name	Type	Description	Default
receipt	dict[str, Any]	Transaction receipt dict with 'logs' and 'from' fields	required
expected_out	Decimal | None	VIB-3203 — pre-slippage-discount quote in human (Decimal) units from ActionBundle.metadata["expected_output_human"]. When provided, realized slippage_bps is computed as (expected_out - amount_out_decimal) / expected_out * 10_000.	None
swap_token_meta	dict[str, dict[str, Any]] | None	VIB-3164 — compiler-supplied token metadata threaded from build_extract_kwargs via the ResultEnricher hook. Shape: {"token_in": {"address": ..., "symbol": ..., "decimals": ...}, "token_out": {...}}. Hints win over the TokenResolver per address; never changes fail-closed semantics (None is still returned when decimals remain unresolved after hints).	None

Returns:

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

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_open_data

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

Extract LP open data from a SushiSwap V3 mint receipt.

Looks for IncreaseLiquidity events emitted by the SushiSwap V3 NonfungiblePositionManager when an LP position is opened or topped up. The event signature is::

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

SushiSwap V3 is a clean Uniswap V3 fork, so the surrounding receipt shape is identical: the V3 pool emits a Mint event right before the NPM IncreaseLiquidity; we track the most recent NPM-owned Pool Mint to recover tick bounds and pool address, then read the post-swap current tick from a Pool Swap event in the same receipt.

Behaviour contract (matches the Uniswap V3 baseline parser):

• Returns LPOpenData populated with the raw on-chain ints (position_id, liquidity, amount0, amount1). The accounting handler is responsible for decimal-scaling.
• Returns None (and emits a WARNING) when self.chain is not registered in SUSHISWAP_V3 — does NOT silently default.
• Returns None when no IncreaseLiquidity log is present in a receipt from a registered chain.
• No outer try/except — the fail-closed variant extract_lp_open_data_result distinguishes parser crash vs. missing event per VIB-3159 / Blueprint 19.

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.

Looks for Collect and/or Burn events from SushiSwap V3 pools which indicate fees and principal being collected when closing or reducing a position. The compiler emits LP_CLOSE as a 3-TX bundle on Sushi V3 (lp_decrease_liquidity + lp_collect + lp_burn), so a single receipt typically carries EITHER Burn (decrease tx) OR Collect (collect tx), not both — the runner's :meth:_extract_receipt_from_result picks one and asks the parser for whatever it can recover from it. Mirrors the Uniswap V3 baseline contract.

Collect(address indexed owner, int24 indexed tickLower,

int24 indexed tickUpper, uint128 amount0, uint128 amount1)

Burn(address indexed owner, int24 indexed tickLower,

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

Captures pool_address from the Burn event's emitter (the V3 pool) — required for the framework's slot0 fallback (VIB-3940) and for the registry-mode close payload (VIB-4198 / T12) whose physical_identity_hash / semantic_grouping_key derivation needs the pool address.

Parameters:

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

Returns:

Type	Description
LPCloseData | None	LPCloseData dataclass if at least one Collect or Burn event is
LPCloseData | None	present, None otherwise. amount0_collected / amount1_collected
LPCloseData | None	prefer Collect amounts when available; fall back to Burn's
LPCloseData | None	principal-only amounts when the receipt only carries a Burn (the
LPCloseData | None	lp_decrease_liquidity TX in a Sushi 3-TX close bundle).

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 for Sushi V3.

Reads :meth:extract_lp_open_data and composes the 8-key (or 11-key with fee_tier + token labels) shape ratified by PRD §Registry Data Shape and the T08 golden. 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. 3000 for 0.3%); forwarded from the intent's compile-time metadata. None when unknown — the payload key stays absent rather than substituting 0.	None

Returns:

Type	Description
dict[str, Any] | None	dict JSON-serializable with the canonical keys, OR
dict[str, Any] | None	None when the LP_OPEN data isn't extractable.

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 for Sushi V3.

Mirrors the Uniswap V3 baseline (the shape contract is identical — Sushi V3 is a clean fork). Reads :meth:extract_lp_close_data 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) contract: a real 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 malformed or fee-only close would produce a "successful" close payload with stale OPEN-side anchors, and the registry would mark a still-open NFT as closed (cutover spec D3.F6 silent-error class).

The flow is:

1. Decode close-side events (:meth:extract_lp_close_data) and the DecreaseLiquidity log (:meth:_decreaseliquidity_token_id).
2. Verify the receipt-derived identity anchors are present and non-zero.
3. Cross-check against open_payload if supplied — refuse on any disagreement (v3_registry_payload.open_payload_disagrees).
4. Compose the receipt-only payload (v3_registry_payload.build_close_receipt_payload).
5. Merge OPEN-time fields the close receipt cannot re-derive (v3_registry_payload.merge_open_payload_fields) — ticks, OPEN-time amounts, original mint liquidity, fee tier, token labels.
6. Apply the fee_tier argument if open_payload didn't carry one (setdefault semantics — OPEN-side wins).

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 INFO log (no zero substitution).

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.

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,
    gateway_client: GatewayClient | 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	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

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

__getattr__

__getattr__(name: str) -> Any

PEP 562 lazy attribute access.