Aerodrome

Connector for Aerodrome (Solidly-based) DEX on Base.

almanak.framework.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.framework.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,
) -> 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)	required
amount_b	Decimal	Desired amount of token_b (in token units)	required
slippage_bps	int | None	Slippage tolerance in basis points (unused for LP, kept for API compat)	None
recipient	str | None	Address to receive the NFT position (default: wallet_address)	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

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

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.

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

extract_swap_amounts_result

extract_swap_amounts_result(
    receipt: dict[str, Any],
    *,
    expected_out: Decimal | 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.

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

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

Placeholder for Aerodrome protocol-fee extraction (VIB-3204).

Aerodrome uses stable vs volatile pool-level fees (0.05% / 0.3%) that are not exposed in the standard Swap event — they require a second call into the pool metadata. Resolving the pool -> fee mapping at the receipt-parser layer is deferred to a follow-up; this method returns None so the enricher skips the field cleanly without polluting extraction_warnings.

Follow-up ticket: "Aerodrome pool-fee resolution in receipt parser — follow-up to VIB-3204".

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

The Slipstream close is a two-tx flow (decreaseLiquidity → collect). The Collect event appears in the second tx receipt; the DecreaseLiquidity event appears in the first. ResultEnricher iterates all receipts and takes the first successful extraction, so the correct receipt will win.

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

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.