Aller au contenu

Aerodrome

Connector for Aerodrome (Solidly-based) DEX on Base, and on Optimism via the Velodrome V2 alias. Some capabilities are Base-only: pool-history and volume data endpoints, and the concentrated-liquidity (Slipstream) router/factory — Optimism uses classic Solidly routing.

almanak.connectors.aerodrome

Aerodrome Finance Connector (Base L2).

This package provides integration with Aerodrome Finance, a Solidly-based AMM on Base chain. Aerodrome supports dual pool types: - Volatile pools: xy=k formula (0.3% fee) - Stable pools: x^3y + y^3*x formula (0.05% fee)

Key Features: - Token swaps (exact input) - Liquidity provision (add/remove) - Fungible LP tokens (not NFT positions like Uniswap V3)

Example

from almanak.connectors.aerodrome import AerodromeAdapter, AerodromeConfig

config = AerodromeConfig( chain="base", wallet_address="0x...", ) adapter = AerodromeAdapter(config)

Execute a volatile pool swap

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

Execute a stable pool swap

result = adapter.swap_exact_input( token_in="USDC", token_out="USDbC", amount_in=Decimal("1000"), stable=True, )

AerodromeAdapter

AerodromeAdapter(
    config: AerodromeConfig,
    token_resolver: TokenResolver | None = None,
)

Adapter for Aerodrome Finance DEX protocol.

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

Example

config = AerodromeConfig( chain="base", wallet_address="0x...", ) adapter = AerodromeAdapter(config)

Execute a volatile pool swap

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

Execute a stable pool swap

result = adapter.swap_exact_input( token_in="USDC", token_out="USDbC", amount_in=Decimal("1000"), stable=True, slippage_bps=10, )

Initialize the adapter.

参数:

名称 类型 描述 默认
config AerodromeConfig

Aerodrome adapter configuration

必需
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,
    stable: bool = False,
    slippage_bps: int | None = None,
    recipient: str | None = None,
    tick_spacing: int = 100,
    use_classic: bool = False,
) -> SwapResult

Build a swap transaction with exact input amount.

By default, routes through the Slipstream CL (concentrated liquidity) pool. Use use_classic=True to opt into the Classic (v1) volatile/stable router.

参数:

名称 类型 描述 默认
token_in str

Input token symbol or address

必需
token_out str

Output token symbol or address

必需
amount_in Decimal

Amount of input token (in token units, not wei)

必需
stable bool

Pool type for Classic routing (True=stable, False=volatile)

False
slippage_bps int | None

Slippage tolerance in basis points (default from config)

None
recipient str | None

Address to receive output tokens (default: wallet_address)

None
tick_spacing int

Slipstream CL tick spacing (default 100)

100
use_classic bool

If True, route through Classic router instead of CL

False

返回:

类型 描述
SwapResult

SwapResult with transaction data

add_liquidity

add_liquidity(
    token_a: str,
    token_b: str,
    amount_a: Decimal,
    amount_b: Decimal,
    stable: bool = False,
    slippage_bps: int | None = None,
    recipient: str | None = None,
) -> LiquidityResult

Build an add liquidity transaction.

参数:

名称 类型 描述 默认
token_a str

First token symbol or address

必需
token_b str

Second token symbol or address

必需
amount_a Decimal

Amount of token A (in token units)

必需
amount_b Decimal

Amount of token B (in token units)

必需
stable bool

Pool type

False
slippage_bps int | None

Slippage tolerance in basis points

None
recipient str | None

Address to receive LP tokens

None

返回:

类型 描述
LiquidityResult

LiquidityResult with transaction data

remove_liquidity

remove_liquidity(
    token_a: str,
    token_b: str,
    liquidity: Decimal,
    stable: bool = False,
    slippage_bps: int | None = None,
    recipient: str | None = None,
    pool_address: str | None = None,
) -> LiquidityResult

Build a remove liquidity transaction.

参数:

名称 类型 描述 默认
token_a str

First token symbol or address

必需
token_b str

Second token symbol or address

必需
liquidity Decimal

LP token amount to burn (in LP token units)

必需
stable bool

Pool type

False
slippage_bps int | None

Slippage tolerance in basis points

None
recipient str | None

Address to receive tokens

None
pool_address str | None

Pre-resolved pool address. If provided, skips the direct RPC lookup (required for deployed mode where the strategy container has no outbound network access).

None

返回:

类型 描述
LiquidityResult

LiquidityResult with transaction data

add_cl_liquidity

add_cl_liquidity(
    token_a: str,
    token_b: str,
    tick_spacing: int,
    tick_lower: int,
    tick_upper: int,
    amount_a: Decimal,
    amount_b: Decimal,
    slippage_bps: int | None = None,
    recipient: str | None = None,
    *,
    amount_a_wei: int | None = None,
    amount_b_wei: int | None = None,
    amount_a_min_wei: int | None = None,
    amount_b_min_wei: int | None = None,
) -> CLLiquidityResult

Build Slipstream CL add liquidity transactions.

Builds approve + mint transactions for an Aerodrome Slipstream CL position. Tokens are sorted by address (token0 < token1) as required by V3-style pools.

参数:

名称 类型 描述 默认
token_a str

First token symbol or address

必需
token_b str

Second token symbol or address

必需
tick_spacing int

Pool tick spacing

必需
tick_lower int

Lower tick bound (must be multiple of tick_spacing)

必需
tick_upper int

Upper tick bound (must be multiple of tick_spacing)

必需
amount_a Decimal

Desired amount of token_a (in token units). The wei conversion uses this value ONLY when the wei-overload kwargs are NOT supplied; in the wei-overload path this value is unused, but the parameter remains required by the type signature — pass Decimal(0) if you have no Decimal value.

必需
amount_b Decimal

See amount_a.

必需
slippage_bps int | None

Slippage tolerance in basis points. Used for the fallback mins formula only when amount_*_min_wei are not supplied. Defaults to config.default_slippage_bps.

None
recipient str | None

Address to receive the NFT position (default: wallet_address)

None
amount_a_wei int | None

Already-converted token_a amount in wei. Wei-overload path skips Decimal→wei conversion. Must be paired with amount_b_wei, amount_a_min_wei and amount_b_min_wei.

None
amount_b_wei int | None

See amount_a_wei.

None
amount_a_min_wei int | None

Pre-computed minimum token_a amount in wei. Bypasses the broken raw-amount × (1 - slippage) formula. Used by the IntentCompiler after slot0 recomputation so that mins are derived from pool-aligned amounts, not the oracle inputs. Must be paired with the other three wei overload kwargs.

None
amount_b_min_wei int | None

See amount_a_min_wei.

None

返回:

类型 描述
CLLiquidityResult

CLLiquidityResult with transaction data

remove_cl_liquidity

remove_cl_liquidity(
    token_id: int,
    slippage_bps: int | None = None,
    recipient: str | None = None,
) -> LiquidityResult

Build Slipstream CL remove liquidity transactions (decreaseLiquidity + collect).

参数:

名称 类型 描述 默认
token_id int

NFT tokenId for the CL position

必需
slippage_bps int | None

Slippage tolerance (unused, mins set to 0)

None
recipient str | None

Address to receive tokens (default: wallet_address)

None

返回:

类型 描述
LiquidityResult

LiquidityResult with transaction data

collect_cl_fees

collect_cl_fees(
    token_id: int, recipient: str | None = None
) -> LiquidityResult

Build a Slipstream CL collect() transaction without burning the position.

The NonfungiblePositionManager.collect() entry point harvests any owed token0/token1 (accrued fees plus principal previously unlocked by decreaseLiquidity) into recipient and leaves the position's liquidity intact. Used by LP_COLLECT_FEES intents that want to compound fees in-position rather than fully exit.

Calling on a position with zero owed tokens is a contract-level no-op (the tx still succeeds and emits Collect with zero amounts). We always emit the tx so the runner sees a deterministic outcome rather than client-side guessing the zero case.

参数:

名称 类型 描述 默认
token_id int

NFT tokenId for the CL position

必需
recipient str | None

Address to receive collected tokens (default: wallet)

None

返回:

类型 描述
LiquidityResult

LiquidityResult with a single collect transaction

compile_swap_intent

compile_swap_intent(
    intent: SwapIntent,
    stable: bool = False,
    price_oracle: dict[str, Decimal] | None = None,
) -> ActionBundle

Compile a SwapIntent to an ActionBundle.

参数:

名称 类型 描述 默认
intent SwapIntent

The SwapIntent to compile

必需
stable bool

Pool type (True=stable, False=volatile)

False
price_oracle dict[str, Decimal] | None

Optional price oracle for USD conversions

None

返回:

类型 描述
ActionBundle

ActionBundle containing transactions for execution

quote_swap_output

quote_swap_output(
    *,
    token_in: str,
    token_out: str,
    amount_in_wei: int,
    stable: bool = False,
    tick_spacing: int = 100,
    use_cl: bool = True,
    require_onchain: bool = False,
) -> int

Quote an exact-input swap output in token base units.

set_allowance

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

Set cached allowance (for testing).

clear_allowance_cache

clear_allowance_cache() -> None

Clear the allowance cache.

AerodromeConfig dataclass

AerodromeConfig(
    chain: str,
    wallet_address: str,
    default_slippage_bps: int = 50,
    deadline_seconds: int = DEFAULT_DEADLINE_SECONDS,
    price_provider: dict[str, Decimal] | None = None,
    allow_placeholder_prices: bool = False,
    rpc_url: str | None = None,
    gateway_client: GatewayClient | None = None,
)

Configuration for AerodromeAdapter.

属性:

名称 类型 描述
chain str

Target blockchain ("base" for Aerodrome, "optimism" for Velodrome V2)

wallet_address str

Address executing transactions

default_slippage_bps int

Default slippage tolerance in basis points (default 50 = 0.5%)

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.

CLLiquidityResult dataclass

CLLiquidityResult(
    success: bool,
    transactions: list[TransactionData] = list(),
    token0: str = "",
    token1: str = "",
    token_id: int | None = None,
    amount0: int = 0,
    amount1: int = 0,
    tick_lower: int = 0,
    tick_upper: int = 0,
    tick_spacing: int = 100,
    error: str | None = None,
    gas_estimate: int = 0,
)

Result of a Slipstream CL liquidity operation.

属性:

名称 类型 描述
success bool

Whether operation was built successfully

transactions list[TransactionData]

List of transactions to execute

token0 str

Token0 address

token1 str

Token1 address

token_id int | None

NFT tokenId (None before TX executes on open)

amount0 int

Amount of token0

amount1 int

Amount of token1

tick_lower int

Lower tick bound

tick_upper int

Upper tick bound

tick_spacing int

Pool tick spacing

error str | None

Error message if failed

gas_estimate int

Total gas estimate

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

LiquidityResult dataclass

LiquidityResult(
    success: bool,
    transactions: list[TransactionData] = list(),
    token_a: str = "",
    token_b: str = "",
    amount_a: int = 0,
    amount_b: int = 0,
    liquidity: int = 0,
    stable: bool = False,
    error: str | None = None,
    gas_estimate: int = 0,
)

Result of a liquidity operation.

属性:

名称 类型 描述
success bool

Whether operation was built successfully

transactions list[TransactionData]

List of transactions to execute

token_a str

First token address

token_b str

Second token address

amount_a int

Amount of token A

amount_b int

Amount of token B

liquidity int

LP tokens (minted or burned)

stable bool

Pool type

error str | None

Error message if failed

gas_estimate int

Total gas estimate

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PoolType

Bases: Enum

Pool type for Aerodrome.

SwapQuote dataclass

SwapQuote(
    token_in: str,
    token_out: str,
    amount_in: int,
    amount_out: int,
    stable: bool,
    gas_estimate: int = AERODROME_GAS_ESTIMATES["swap"],
    price_impact_bps: int = 0,
    effective_price: Decimal = Decimal("0"),
    quoted_at: datetime = (lambda: datetime.now(UTC))(),
)

Quote for a swap operation.

属性:

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

stable bool

Pool type (True=stable, False=volatile)

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.

属性:

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

属性:

名称 类型 描述
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, add_liquidity, remove_liquidity)

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

AerodromeCompiler

Bases: BaseConcentratedLiquidityCompiler

Compiler for Aerodrome classic and Slipstream routes.

AerodromeEvent dataclass

AerodromeEvent(
    event_type: AerodromeEventType,
    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 Aerodrome event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

from_dict classmethod

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

Create from dictionary.

AerodromeEventType

Bases: Enum

Aerodrome event types.

AerodromeReceiptParser

AerodromeReceiptParser(
    chain: str = "base",
    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 Aerodrome transaction receipts.

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

Initialize the parser.

参数:

名称 类型 描述 默认
chain str

Blockchain network (for token symbol resolution)

'base'
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.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict

必需
quoted_amount_out int | None

Expected output amount for slippage calculation

None

返回:

类型 描述
ParseResult

ParseResult with extracted events and swap data

build_extract_kwargs

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

Return Aerodrome-owned kwargs for ResultEnricher extraction calls.

VIB-3164: the compiler records full token identity (from_token / to_token dicts with address, symbol, decimals — see compile_swap_aerodrome) in ActionBundle.metadata. Threading it here lets _resolve_swap_decimals 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_result

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

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

VIB-3203: expected_out is forwarded to :meth:extract_swap_amounts for realized slippage_bps computation.

VIB-3164: swap_token_meta is forwarded to :meth:extract_swap_amounts so compiler-supplied token hints can resolve decimals on resolver misses. This is the method the ResultEnricher calls — the kwarg MUST be here; omitting it would cause the enricher's TypeError fallback to silently drop expected_out too (see result_enricher.py §_invoke_extract).

extract_lp_close_data_result

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

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

extract_position_id_result

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

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

extract_liquidity_result

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

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

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.

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

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' and 'from' fields

必需
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; a single-swap-gated direction fallback fills still-empty addresses; fail-closed semantics (return None on unresolved decimals) preserved.

None

返回:

类型 描述
SwapAmounts | None

SwapAmounts dataclass if swap event found, None otherwise

Implementation note

This method is intentionally thin — each logical phase is a dedicated helper so the control flow stays CC-bounded and individually testable (Phase 7.3 refactor). Preserves: (1) first-swap-event-wins multi-hop semantics, (2) sign conventions for amount0/amount1 (V1 + CL), (3) SwapAmounts field surface.

extract_lp_close_data

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

Extract LP close data from transaction receipt.

Primary path: extracts from Burn events (amount0, amount1). Fallback path: extracts from Transfer events when Burn event is not detected (some Aerodrome pool variants may not emit a standard Burn event, but always emit Transfer events for the returned tokens).

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

必需

返回:

类型 描述
LPCloseData | None

LPCloseData dataclass if token amounts found, None otherwise

extract_position_id

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

Extract position ID from LP mint transaction receipt.

For Aerodrome (Solidly fork), LP tokens are fungible ERC-20s (not NFTs). The position ID is the pool address that emitted the Mint event, which is also the LP token contract address.

Called by ResultEnricher for LP_OPEN intents.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

必需

返回:

类型 描述
str | None

Pool address (LP token address) as hex string, or None if not found

extract_liquidity

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

Extract liquidity from LP mint transaction receipt.

For Aerodrome V1, this extracts the LP tokens minted from Transfer events.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

必需

返回:

类型 描述
int | None

LP token amount if found, None otherwise

extract_protocol_fees

extract_protocol_fees(
    _receipt: dict[str, Any],
) -> ProtocolFees

VIB-3495: Aerodrome LP protocol fee coverage audit.

Aerodrome V1 (Solidly fork) charges stable vs volatile pool-level fees (0.05% / 0.3%) on swaps. The fee is charged inside the pool's swap() function and is NOT emitted as a distinct on-chain event — neither the Swap event nor the Transfer events carry the fee amount separately from the net amounts. Resolving the USD fee amount would require (a) reading the pool's fee-rate slot and (b) a price oracle, neither of which is available at the receipt-parser layer.

Returns a ProtocolFees with unavailable_reason so downstream attribution records "known-unknown" rather than "parser absent" (which was the old behaviour when this method returned None).

is_aerodrome_event

is_aerodrome_event(topic: str | bytes) -> bool

Check if a topic is a known Aerodrome event.

参数:

名称 类型 描述 默认
topic str | bytes

Event topic (supports bytes, hex string with/without 0x, any case)

必需

返回:

类型 描述
bool

True if topic is a known Aerodrome event

get_event_type

get_event_type(topic: str | bytes) -> AerodromeEventType

Get the event type for a topic.

参数:

名称 类型 描述 默认
topic str | bytes

Event topic (supports bytes, hex string with/without 0x, any case)

必需

返回:

类型 描述
AerodromeEventType

Event type or UNKNOWN

AerodromeSlipstreamReceiptParser

AerodromeSlipstreamReceiptParser(
    chain: str = "base",
    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,
)

Bases: AerodromeReceiptParser

Receipt parser for Aerodrome Slipstream CL (concentrated liquidity) transactions.

Extends AerodromeReceiptParser to handle NFT position events from the NonfungiblePositionManager contract: - IncreaseLiquidity: emitted on mint (LP_OPEN) - DecreaseLiquidity: emitted on decreaseLiquidity (LP_CLOSE) - ERC-721 Transfer: used to extract tokenId on mint

Position ID is the NFT tokenId extracted from the ERC-721 Transfer (mint) event where from == zero_address.

extract_position_id

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

Extract NFT tokenId from LP mint receipt.

For Slipstream CL, minting creates an ERC-721 NFT. The tokenId is found in the Transfer event emitted by the NonfungiblePositionManager where from == zero_address (mint). ERC-721 Transfer events have 4 topics: [Transfer_sig, from (indexed), to (indexed), tokenId (indexed)].

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

必需

返回:

类型 描述
str | None

NFT tokenId as string (e.g. "12345"), or None if not found

extract_liquidity

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

Extract liquidity from CL mint receipt via IncreaseLiquidity event.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

必需

返回:

类型 描述
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 Slipstream CL mint receipt.

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

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

Slipstream is a Uniswap V3 fork, so the surrounding receipt shape is identical: the CL pool emits a Uniswap-V3-style Mint event right before the NPM IncreaseLiquidity; we track the most recent NPM-owned Pool Mint to recover tick bounds and pool address.

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 when no IncreaseLiquidity log is present.
  • No outer try/except — the fail-closed variant extract_lp_open_data_result distinguishes parser crash vs. missing event per VIB-3159 / Blueprint 19.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field.

必需

返回:

类型 描述
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 Slipstream CL receipt.

Reads the Collect event (actual amounts transferred to recipient) in preference to DecreaseLiquidity (amounts unlocked but not yet transferred). This ensures fees already owed before the close are included in the reported amounts, not just the principal removed in this transaction.

Slipstream's close is a two-transaction sequence: decreaseLiquiditycollect. The two events land in different receipts. Each call to this method sees ONE receipt, so it returns either:

  • a Collect-sourced LPCloseData (source="collect") — the principal + pre-existing fees actually transferred to the recipient;
  • a DecreaseLiquidity-sourced fallback (source="decrease_liquidity") — the principal unlocked into tokensOwed but not yet transferred.

ResultEnricher aggregates across receipts and prefers the Collect-sourced variant when both are present (VIB-4310). The source tag is the explicit signal it uses; do not strip it.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

必需

返回:

类型 描述
LPCloseData | None

LPCloseData if found, None otherwise

extract_fees0

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

Extract token0 fees collected from a Slipstream Collect event.

Used to enrich LP_COLLECT_FEES intents. The Collect event amount equals the realized fee in token0 only when liquidity is unchanged; if a DecreaseLiquidity event is also present we return None rather than misreport principal as fees (see _extract_collect_amounts).

extract_fees1

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

Extract token1 fees collected from a Slipstream Collect event.

See extract_fees0 for semantics.

extract_lp_close_data_result

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

Fail-closed variant of extract_lp_close_data for Slipstream CL.

extract_lp_open_data_result

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

Fail-closed variant of extract_lp_open_data for Slipstream CL.

Per VIB-3159 / Blueprint 19: callers that need to distinguish "parser crashed" from "no IncreaseLiquidity event present" use this variant. The bare extract_lp_open_data returns None on missing event and propagates exceptions unchanged.

extract_position_id_result

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

Fail-closed variant of extract_position_id for Slipstream CL.

extract_liquidity_result

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

Fail-closed variant of extract_liquidity for Slipstream CL.

extract_registry_payload_open

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

Build the LP_OPEN position_registry.payload dict.

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

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dict with logs field.

必需
fee_tier int | None

Optional pool fee tier (e.g. 500 for 0.05%); forwarded from the intent's compile-time metadata. None when unknown — the payload key stays absent rather than substituting 0 (Empty ≠ zero).

None

返回:

类型 描述
dict[str, Any] | None

dict JSON-serializable with the 8 (or 9 with fee_tier) keys

dict[str, Any] | None

ratified by the PRD §Registry Data Shape and the T08 golden, OR

dict[str, Any] | None

None when the LP_OPEN data isn't extractable from the

dict[str, Any] | None

receipt.

extract_registry_payload_close

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

Build the LP_CLOSE position_registry.payload dict.

Reads :meth:extract_lp_close_data (Burn / Collect amounts) and decodes the close-side DecreaseLiquidity event for the NFT token_id, plus the Slipstream Pool Burn log for the pool emitter address. Then composes the close payload shape ratified by the T08 lp_close/expected_registry_row.json golden.

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

The flow:

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

Helpers open_payload_disagrees / build_close_receipt_payload / merge_open_payload_fields are shared via almanak.connectors._strategy_base.v3_registry_payload — they operate on plain dicts with no UV3-specific assumptions and are the single source of truth for the merge / cross-check semantics.

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

BurnEventData dataclass

BurnEventData(
    sender: str,
    amount0: int,
    amount1: int,
    to: str,
    pool_address: str,
)

Parsed data from Burn event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

MintEventData dataclass

MintEventData(
    sender: str,
    amount0: int,
    amount1: int,
    pool_address: str,
)

Parsed data from Mint event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

ParsedLiquidityResult dataclass

ParsedLiquidityResult(
    operation: str,
    token0: str,
    token1: str,
    token0_symbol: str,
    token1_symbol: str,
    amount0: int,
    amount1: int,
    amount0_decimal: Decimal,
    amount1_decimal: Decimal,
    pool_address: str,
)

High-level liquidity result extracted from receipt.

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

High-level swap result extracted from receipt.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

to_swap_result_payload

to_swap_result_payload() -> SwapResultPayload

Convert to SwapResultPayload for event emission.

ParseResult dataclass

ParseResult(
    success: bool,
    events: list[AerodromeEvent] = list(),
    swap_events: list[SwapEventData] = list(),
    mint_events: list[MintEventData] = list(),
    burn_events: list[BurnEventData] = list(),
    transfer_events: list[TransferEventData] = list(),
    swap_result: ParsedSwapResult | None = None,
    liquidity_result: ParsedLiquidityResult | 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,
    to: str,
    amount0_in: int,
    amount1_in: int,
    amount0_out: int,
    amount1_out: 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 input amount.

amount_out property

amount_out: int

Get the output amount.

to_dict

to_dict() -> dict[str, Any]

Convert to 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.

AerodromeSDK

AerodromeSDK(
    chain: str = "base",
    rpc_url: str | None = None,
    token_resolver: TokenResolver | None = None,
    gateway_client: GatewayClient | None = None,
)

Low-level SDK for Aerodrome/Velodrome Finance (Solidly forks).

This SDK provides direct interaction with Solidly-fork contracts (Aerodrome on Base, Velodrome V2 on Optimism): - Pool queries (reserves, amounts) - Transaction building (swaps, liquidity) - ABI encoding for all operations

Example

sdk = AerodromeSDK(chain="base")

Get quote for swap

quote = sdk.get_swap_quote( token_in="0x...", token_out="0x...", amount_in=1000000, stable=False, )

Build swap transaction

tx = sdk.build_swap_exact_tokens_tx( amount_in=1000000, amount_out_min=990000, routes=[SwapRoute(token_in, token_out, stable=False)], recipient="0x...", deadline=int(time.time()) + 300, sender="0x...", )

Initialize the SDK.

参数:

名称 类型 描述 默认
chain str

Target chain ("base" for Aerodrome, "optimism" for Velodrome V2)

'base'
rpc_url str | None

DEPRECATED — direct RPC URL. Bypasses the gateway and is only used for ad-hoc scripts. Prefer gateway_client for any code path that runs in a strategy container.

None
token_resolver TokenResolver | None

Optional TokenResolver instance. If None, uses singleton.

None
gateway_client GatewayClient | None

Gateway client used to route on-chain queries (eth_call) through the gateway's RpcService. Preferred over rpc_url for all production code paths.

None

get_pool_address

get_pool_address(
    token_a: str, token_b: str, stable: bool
) -> str | None

Get pool address from factory.

Routes eth_call through the gateway when gateway_client is set on the SDK. Falls back to direct RPC only for ad-hoc script usage (deprecated).

参数:

名称 类型 描述 默认
token_a str

First token address

必需
token_b str

Second token address

必需
stable bool

Pool type

必需

返回:

类型 描述
str | None

Pool address if exists, None otherwise

get_pool_address_from_factory

get_pool_address_from_factory(
    token_a: str, token_b: str, stable: bool, web3: Any
) -> str | None

Get pool address from factory contract.

参数:

名称 类型 描述 默认
token_a str

First token address

必需
token_b str

Second token address

必需
stable bool

Pool type

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
str | None

Pool address if exists, None otherwise

get_pool_info

get_pool_info(
    token_a: str, token_b: str, stable: bool, web3: Any
) -> PoolInfo | None

Get full pool information.

参数:

名称 类型 描述 默认
token_a str

First token address

必需
token_b str

Second token address

必需
stable bool

Pool type

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
PoolInfo | None

PoolInfo if pool exists, None otherwise

get_amount_out

get_amount_out(
    amount_in: int,
    token_in: str,
    token_out: str,
    stable: bool,
    web3: Any,
) -> int | None

Get expected output amount for a swap.

参数:

名称 类型 描述 默认
amount_in int

Input amount

必需
token_in str

Input token address

必需
token_out str

Output token address

必需
stable bool

Pool type

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
int | None

Output amount or None if pool doesn't exist

get_amounts_out

get_amounts_out(
    amount_in: int, routes: list[SwapRoute], web3: Any
) -> list[int] | None

Get expected output amounts for multi-hop swap.

参数:

名称 类型 描述 默认
amount_in int

Input amount

必需
routes list[SwapRoute]

List of swap routes

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
list[int] | None

List of amounts for each hop, or None on error

build_approve_tx

build_approve_tx(
    token_address: str,
    spender: str,
    amount: int,
    sender: str,
    web3: Any,
) -> dict[str, Any]

Build ERC-20 approve transaction.

参数:

名称 类型 描述 默认
token_address str

Token to approve

必需
spender str

Address to approve for spending

必需
amount int

Amount to approve (use MAX_UINT256 for unlimited)

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_swap_exact_tokens_tx

build_swap_exact_tokens_tx(
    amount_in: int,
    amount_out_min: int,
    routes: list[SwapRoute],
    recipient: str,
    deadline: int,
    sender: str,
    web3: Any,
) -> dict[str, Any]

Build swapExactTokensForTokens transaction.

参数:

名称 类型 描述 默认
amount_in int

Input token amount

必需
amount_out_min int

Minimum output amount (slippage protection)

必需
routes list[SwapRoute]

Swap routes

必需
recipient str

Recipient address

必需
deadline int

Unix timestamp deadline

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_add_liquidity_tx

build_add_liquidity_tx(
    token_a: str,
    token_b: str,
    stable: bool,
    amount_a_desired: int,
    amount_b_desired: int,
    amount_a_min: int,
    amount_b_min: int,
    recipient: str,
    deadline: int,
    sender: str,
    web3: Any,
) -> dict[str, Any]

Build addLiquidity transaction.

参数:

名称 类型 描述 默认
token_a str

First token address

必需
token_b str

Second token address

必需
stable bool

Pool type

必需
amount_a_desired int

Desired amount of token A

必需
amount_b_desired int

Desired amount of token B

必需
amount_a_min int

Minimum amount of token A

必需
amount_b_min int

Minimum amount of token B

必需
recipient str

LP token recipient

必需
deadline int

Unix timestamp deadline

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_remove_liquidity_tx

build_remove_liquidity_tx(
    token_a: str,
    token_b: str,
    stable: bool,
    liquidity: int,
    amount_a_min: int,
    amount_b_min: int,
    recipient: str,
    deadline: int,
    sender: str,
    web3: Any,
) -> dict[str, Any]

Build removeLiquidity transaction.

参数:

名称 类型 描述 默认
token_a str

First token address

必需
token_b str

Second token address

必需
stable bool

Pool type

必需
liquidity int

LP token amount to burn

必需
amount_a_min int

Minimum token A to receive

必需
amount_b_min int

Minimum token B to receive

必需
recipient str

Token recipient

必需
deadline int

Unix timestamp deadline

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_wrap_eth_tx

build_wrap_eth_tx(
    amount: int, sender: str, web3: Any
) -> dict[str, Any]

Build WETH wrap (deposit) transaction.

参数:

名称 类型 描述 默认
amount int

ETH amount to wrap

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_unwrap_eth_tx

build_unwrap_eth_tx(
    amount: int, sender: str, web3: Any
) -> dict[str, Any]

Build WETH unwrap (withdraw) transaction.

参数:

名称 类型 描述 默认
amount int

WETH amount to unwrap

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

get_cl_pool_address

get_cl_pool_address(
    token_a: str, token_b: str, tick_spacing: int, web3: Any
) -> str | None

Get Slipstream CL pool address from cl_factory.

参数:

名称 类型 描述 默认
token_a str

First token address

必需
token_b str

Second token address

必需
tick_spacing int

Pool tick spacing (int24)

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
str | None

Pool address if exists, None otherwise

get_cl_pool_slot0

get_cl_pool_slot0(
    pool_address: str, web3: Any
) -> tuple[int, int] | None

Get sqrtPriceX96 and current tick from Slipstream CL pool slot0.

参数:

名称 类型 描述 默认
pool_address str

CL pool contract address

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
tuple[int, int] | None

Tuple of (sqrtPriceX96, current_tick) or None if query failed

get_cl_position

get_cl_position(
    token_id: int, web3: Any
) -> CLPositionInfo | None

Get Slipstream CL position info by NFT token ID.

参数:

名称 类型 描述 默认
token_id int

NFT token ID

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
CLPositionInfo | None

CLPositionInfo if found, None otherwise

build_cl_mint_tx

build_cl_mint_tx(
    token0: str,
    token1: str,
    tick_spacing: int,
    tick_lower: int,
    tick_upper: int,
    amount0_desired: int,
    amount1_desired: int,
    amount0_min: int,
    amount1_min: int,
    recipient: str,
    deadline: int,
    sender: str,
    web3: Any,
    sqrt_price_x96: int = 0,
) -> dict[str, Any]

Build Slipstream CL NonfungiblePositionManager mint transaction.

参数:

名称 类型 描述 默认
token0 str

Token0 address (must be < token1 by address)

必需
token1 str

Token1 address

必需
tick_spacing int

Pool tick spacing

必需
tick_lower int

Lower tick bound

必需
tick_upper int

Upper tick bound

必需
amount0_desired int

Desired token0 amount

必需
amount1_desired int

Desired token1 amount

必需
amount0_min int

Minimum token0 amount (slippage protection)

必需
amount1_min int

Minimum token1 amount (slippage protection)

必需
recipient str

NFT recipient address

必需
deadline int

Transaction deadline (unix timestamp)

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需
sqrt_price_x96 int

Initial sqrt price (0 for existing pool)

0

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_cl_decrease_liquidity_tx

build_cl_decrease_liquidity_tx(
    token_id: int,
    liquidity: int,
    amount0_min: int,
    amount1_min: int,
    deadline: int,
    sender: str,
    web3: Any,
) -> dict[str, Any]

Build Slipstream CL decreaseLiquidity transaction.

参数:

名称 类型 描述 默认
token_id int

NFT token ID

必需
liquidity int

Amount of liquidity to remove

必需
amount0_min int

Minimum token0 to receive

必需
amount1_min int

Minimum token1 to receive

必需
deadline int

Transaction deadline

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

build_cl_collect_tx

build_cl_collect_tx(
    token_id: int,
    recipient: str,
    amount0_max: int,
    amount1_max: int,
    sender: str,
    web3: Any,
) -> dict[str, Any]

Build Slipstream CL collect transaction.

参数:

名称 类型 描述 默认
token_id int

NFT token ID

必需
recipient str

Token recipient address

必需
amount0_max int

Maximum token0 to collect (use MAX_UINT128 for all)

必需
amount1_max int

Maximum token1 to collect (use MAX_UINT128 for all)

必需
sender str

Transaction sender

必需
web3 Any

Web3 instance

必需

返回:

类型 描述
dict[str, Any]

Transaction dictionary

resolve_token

resolve_token(token: str) -> str

Resolve token symbol to address using TokenResolver.

参数:

名称 类型 描述 默认
token str

Token symbol or address

必需

返回:

类型 描述
str

Token address

引发:

类型 描述
TokenResolutionError

If the token cannot be resolved

get_token_symbol

get_token_symbol(address: str) -> str

Get token symbol from address using TokenResolver.

参数:

名称 类型 描述 默认
address str

Token address

必需

返回:

类型 描述
str

Token symbol

get_token_decimals

get_token_decimals(symbol: str) -> int

Get token decimals from symbol using TokenResolver.

参数:

名称 类型 描述 默认
symbol str

Token symbol

必需

返回:

类型 描述
int

Token decimals

引发:

类型 描述
TokenResolutionError

If decimals cannot be determined

AerodromeSDKError

Bases: Exception

Base exception for Aerodrome SDK errors.

CLPositionInfo dataclass

CLPositionInfo(
    token_id: int,
    token0: str,
    token1: str,
    tick_spacing: int,
    tick_lower: int,
    tick_upper: int,
    liquidity: int,
    tokens_owed0: int,
    tokens_owed1: int,
)

Information about a Slipstream CL NFT position.

属性:

名称 类型 描述
token_id int

NFT token ID

token0 str

Address of token0

token1 str

Address of token1

tick_spacing int

Pool tick spacing

tick_lower int

Lower tick of the position range

tick_upper int

Upper tick of the position range

liquidity int

Current liquidity in the position

tokens_owed0 int

Uncollected fees for token0

tokens_owed1 int

Uncollected fees for token1

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

InsufficientLiquidityError

Bases: AerodromeSDKError

Raised when pool has insufficient liquidity.

PoolInfo dataclass

PoolInfo(
    address: str,
    token0: str,
    token1: str,
    stable: bool,
    reserve0: int = 0,
    reserve1: int = 0,
    decimals0: int = 18,
    decimals1: int = 18,
)

Information about an Aerodrome pool.

属性:

名称 类型 描述
address str

Pool contract address

token0 str

First token address

token1 str

Second token address

stable bool

True for stable pool, False for volatile

reserve0 int

Current reserve of token0

reserve1 int

Current reserve of token1

decimals0 int

Decimals of token0

decimals1 int

Decimals of token1

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PoolNotFoundError

Bases: AerodromeSDKError

Raised when a pool doesn't exist.

SwapRoute dataclass

SwapRoute(
    from_token: str,
    to_token: str,
    stable: bool,
    factory: str | None = None,
)

A single hop in a swap route.

属性:

名称 类型 描述
from_token str

Input token address

to_token str

Output token address

stable bool

Pool type (True=stable, False=volatile)

factory str | None

Factory address (optional, uses default)

to_tuple

to_tuple(default_factory: str) -> tuple

Convert to tuple format for contract call.

All addresses are checksummed to prevent web3.py rejection which would silently fall back to zero slippage protection.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

SDKSwapQuote dataclass

SDKSwapQuote(
    amount_in: int,
    amount_out: int,
    routes: list[SwapRoute],
    price_impact_bps: int = 0,
    gas_estimate: int = AERODROME_GAS_ESTIMATES["swap"],
)

Quote for a swap operation.

属性:

名称 类型 描述
amount_in int

Input amount

amount_out int

Expected output amount

routes list[SwapRoute]

List of swap routes

price_impact_bps int

Estimated price impact in basis points

gas_estimate int

Estimated gas for the swap

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

__getattr__

__getattr__(name: str) -> Any

PEP 562 lazy attribute access.