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.

Parameters:

Name	Type	Description	Default
`config`	`AerodromeConfig`	Aerodrome 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,
    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.

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

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token symbol or address	required
`token_b`	`str`	Second token symbol or address	required
`amount_a`	`Decimal`	Amount of token A (in token units)	required
`amount_b`	`Decimal`	Amount of token B (in token units)	required
`stable`	`bool`	Pool type	`False`
`slippage_bps`	`int \| None`	Slippage tolerance in basis points	`None`
`recipient`	`str \| None`	Address to receive LP tokens	`None`

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token symbol or address	required
`token_b`	`str`	Second token symbol or address	required
`liquidity`	`Decimal`	LP token amount to burn (in LP token units)	required
`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`

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token symbol or address	required
`token_b`	`str`	Second token symbol or address	required
`tick_spacing`	`int`	Pool tick spacing	required
`tick_lower`	`int`	Lower tick bound (must be multiple of tick_spacing)	required
`tick_upper`	`int`	Upper tick bound (must be multiple of tick_spacing)	required
`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.	required
`amount_b`	`Decimal`	See `amount_a`.	required
`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`

Returns:

Type	Description
`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).

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT tokenId for the CL position	required
`slippage_bps`	`int \| None`	Slippage tolerance (unused, mins set to 0)	`None`
`recipient`	`str \| None`	Address to receive tokens (default: wallet_address)	`None`

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT tokenId for the CL position	required
`recipient`	`str \| None`	Address to receive collected tokens (default: wallet)	`None`

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`intent`	`SwapIntent`	The SwapIntent to compile	required
`stable`	`bool`	Pool type (True=stable, False=volatile)	`False`
`price_oracle`	`dict[str, Decimal] \| None`	Optional price oracle for USD conversions	`None`

Returns:

Type	Description
`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.

Attributes:

Name	Type	Description
`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.

Attributes:

Name	Type	Description
`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.

Attributes:

Name	Type	Description
`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.

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

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

Parameters:

Name	Type	Description	Default
`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.

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

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

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

Returns:

Type	Description
`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).

Parameters:

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

Returns:

Type	Description
`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.

Parameters:

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

Returns:

Type	Description
`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.

Parameters:

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

Returns:

Type	Description
`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.

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 Aerodrome event

get_event_type ¶

get_event_type(topic: str | bytes) -> AerodromeEventType

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

Parameters:

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

Returns:

Type	Description
`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.

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

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 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: decreaseLiquidity → collect. 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.

Parameters:

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

Returns:

Type	Description
`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).

Parameters:

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

Returns:

Type	Description
`dict[str, Any] \| None`	`dict` JSON-serializable with the 8 (or 9 with fee_tier) keys
`dict[str, Any] \| None`	ratified by 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:

Decode close-side amounts (extract_lp_close_data).
Decode the DecreaseLiquidity log (_decreaseliquidity_token_id).
Decode the Pool Burn log (_pool_address_from_burn).
Verify receipt-derived identity anchors are present and non-zero.
Cross-check against open_payload (v3_registry_payload.open_payload_disagrees) — refuse on disagreement.
Compose the receipt-only payload (v3_registry_payload.build_close_receipt_payload).
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).
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.

Parameters:

Name	Type	Description	Default
`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).

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token address	required
`token_b`	`str`	Second token address	required
`stable`	`bool`	Pool type	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token address	required
`token_b`	`str`	Second token address	required
`stable`	`bool`	Pool type	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token address	required
`token_b`	`str`	Second token address	required
`stable`	`bool`	Pool type	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`amount_in`	`int`	Input amount	required
`token_in`	`str`	Input token address	required
`token_out`	`str`	Output token address	required
`stable`	`bool`	Pool type	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`amount_in`	`int`	Input amount	required
`routes`	`list[SwapRoute]`	List of swap routes	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_address`	`str`	Token to approve	required
`spender`	`str`	Address to approve for spending	required
`amount`	`int`	Amount to approve (use MAX_UINT256 for unlimited)	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`amount_in`	`int`	Input token amount	required
`amount_out_min`	`int`	Minimum output amount (slippage protection)	required
`routes`	`list[SwapRoute]`	Swap routes	required
`recipient`	`str`	Recipient address	required
`deadline`	`int`	Unix timestamp deadline	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token address	required
`token_b`	`str`	Second token address	required
`stable`	`bool`	Pool type	required
`amount_a_desired`	`int`	Desired amount of token A	required
`amount_b_desired`	`int`	Desired amount of token B	required
`amount_a_min`	`int`	Minimum amount of token A	required
`amount_b_min`	`int`	Minimum amount of token B	required
`recipient`	`str`	LP token recipient	required
`deadline`	`int`	Unix timestamp deadline	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token address	required
`token_b`	`str`	Second token address	required
`stable`	`bool`	Pool type	required
`liquidity`	`int`	LP token amount to burn	required
`amount_a_min`	`int`	Minimum token A to receive	required
`amount_b_min`	`int`	Minimum token B to receive	required
`recipient`	`str`	Token recipient	required
`deadline`	`int`	Unix timestamp deadline	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`amount`	`int`	ETH amount to wrap	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`amount`	`int`	WETH amount to unwrap	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_a`	`str`	First token address	required
`token_b`	`str`	Second token address	required
`tick_spacing`	`int`	Pool tick spacing (int24)	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`pool_address`	`str`	CL pool contract address	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token0`	`str`	Token0 address (must be < token1 by address)	required
`token1`	`str`	Token1 address	required
`tick_spacing`	`int`	Pool tick spacing	required
`tick_lower`	`int`	Lower tick bound	required
`tick_upper`	`int`	Upper tick bound	required
`amount0_desired`	`int`	Desired token0 amount	required
`amount1_desired`	`int`	Desired token1 amount	required
`amount0_min`	`int`	Minimum token0 amount (slippage protection)	required
`amount1_min`	`int`	Minimum token1 amount (slippage protection)	required
`recipient`	`str`	NFT recipient address	required
`deadline`	`int`	Transaction deadline (unix timestamp)	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required
`sqrt_price_x96`	`int`	Initial sqrt price (0 for existing pool)	`0`

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID	required
`liquidity`	`int`	Amount of liquidity to remove	required
`amount0_min`	`int`	Minimum token0 to receive	required
`amount1_min`	`int`	Minimum token1 to receive	required
`deadline`	`int`	Transaction deadline	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`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.

Parameters:

Name	Type	Description	Default
`token_id`	`int`	NFT token ID	required
`recipient`	`str`	Token recipient address	required
`amount0_max`	`int`	Maximum token0 to collect (use MAX_UINT128 for all)	required
`amount1_max`	`int`	Maximum token1 to collect (use MAX_UINT128 for all)	required
`sender`	`str`	Transaction sender	required
`web3`	`Any`	Web3 instance	required

Returns:

Type	Description
`dict[str, Any]`	Transaction dictionary

resolve_token ¶

resolve_token(token: str) -> str

Resolve token symbol to address using TokenResolver.

Parameters:

Name	Type	Description	Default
`token`	`str`	Token symbol or address	required

Returns:

Type	Description
`str`	Token address

Raises:

Type	Description
`TokenResolutionError`	If the token cannot be resolved

get_token_symbol ¶

get_token_symbol(address: str) -> str

Get token symbol from address using TokenResolver.

Parameters:

Name	Type	Description	Default
`address`	`str`	Token address	required

Returns:

Type	Description
`str`	Token symbol

get_token_decimals ¶

get_token_decimals(symbol: str) -> int

Get token decimals from symbol using TokenResolver.

Parameters:

Name	Type	Description	Default
`symbol`	`str`	Token symbol	required

Returns:

Type	Description
`int`	Token decimals

Raises:

Type	Description
`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.

Attributes:

Name	Type	Description
`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.

Attributes:

Name	Type	Description
`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.

Attributes:

Name	Type	Description
`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.

Attributes:

Name	Type	Description
`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.

Aerodrome¶

almanak.connectors.aerodrome ¶

Execute a volatile pool swap¶

Execute a stable pool swap¶

AerodromeAdapter ¶

Execute a volatile pool swap¶

Execute a stable pool swap¶

swap_exact_input ¶

add_liquidity ¶

remove_liquidity ¶

add_cl_liquidity ¶

remove_cl_liquidity ¶

collect_cl_fees ¶

compile_swap_intent ¶

quote_swap_output ¶

set_allowance ¶

clear_allowance_cache ¶

AerodromeConfig dataclass ¶

__post_init__ ¶

to_dict ¶

CLLiquidityResult dataclass ¶

to_dict ¶

LiquidityResult dataclass ¶

to_dict ¶

PoolType ¶

SwapQuote dataclass ¶

to_dict ¶

SwapResult dataclass ¶

to_dict ¶

SwapType ¶

TransactionData dataclass ¶

to_dict ¶

AerodromeCompiler ¶

AerodromeEvent dataclass ¶

to_dict ¶

from_dict classmethod ¶

AerodromeEventType ¶

AerodromeReceiptParser ¶

parse_receipt ¶

build_extract_kwargs ¶

extract_swap_amounts_result ¶

extract_lp_close_data_result ¶

extract_position_id_result ¶

extract_liquidity_result ¶

extract_swap_amounts ¶

extract_lp_close_data ¶

extract_position_id ¶

extract_liquidity ¶

extract_protocol_fees ¶

is_aerodrome_event ¶

get_event_type ¶

AerodromeSlipstreamReceiptParser ¶

extract_position_id ¶

extract_liquidity ¶

extract_lp_open_data ¶

extract_lp_close_data ¶

extract_fees0 ¶

extract_fees1 ¶

extract_lp_close_data_result ¶

extract_lp_open_data_result ¶

extract_position_id_result ¶

extract_liquidity_result ¶

extract_registry_payload_open ¶

extract_registry_payload_close ¶

BurnEventData dataclass ¶

to_dict ¶

MintEventData dataclass ¶

to_dict ¶

ParsedLiquidityResult dataclass ¶

to_dict ¶

ParsedSwapResult dataclass ¶

to_dict ¶

to_swap_result_payload ¶

ParseResult dataclass ¶

to_dict ¶

SwapEventData dataclass ¶

token0_is_input property ¶

token1_is_input property ¶

amount_in property ¶

amount_out property ¶

AerodromeConfig `dataclass` ¶

CLLiquidityResult `dataclass` ¶

LiquidityResult `dataclass` ¶

SwapQuote `dataclass` ¶

SwapResult `dataclass` ¶

TransactionData `dataclass` ¶

AerodromeEvent `dataclass` ¶

from_dict `classmethod` ¶

BurnEventData `dataclass` ¶

MintEventData `dataclass` ¶

ParsedLiquidityResult `dataclass` ¶

ParsedSwapResult `dataclass` ¶

ParseResult `dataclass` ¶

SwapEventData `dataclass` ¶

token0_is_input `property` ¶

token1_is_input `property` ¶

amount_in `property` ¶

amount_out `property` ¶

TransferEventData `dataclass` ¶

CLPositionInfo `dataclass` ¶

PoolInfo `dataclass` ¶

SwapRoute `dataclass` ¶

SDKSwapQuote `dataclass` ¶

getattr ¶