Curve¶

Connector for Curve Finance DEX.

almanak.framework.connectors.curve ¶

Curve Finance Connector.

This module provides the Curve Finance adapter for executing swaps and managing liquidity positions on Curve pools across multiple chains.

Supported chains: - Ethereum - Arbitrum

Supported operations: - SWAP: Token swaps via Curve pools (StableSwap, CryptoSwap, Tricrypto) - LP_OPEN: Add liquidity to Curve pools - LP_CLOSE: Remove liquidity from Curve pools

Example

from almanak.framework.connectors.curve import CurveAdapter, CurveConfig

config = CurveConfig( chain="ethereum", wallet_address="0x...", ) adapter = CurveAdapter(config)

Execute a swap¶

result = adapter.swap( pool_address="0xbEbc44782C7dB0a1A60Cb6fe97d0b483032FF1C7", # 3pool token_in="USDC", token_out="DAI", amount_in=Decimal("1000"), )

Add liquidity¶

lp_result = adapter.add_liquidity( pool_address="0xbEbc44782C7dB0a1A60Cb6fe97d0b483032FF1C7", amounts=[Decimal("1000"), Decimal("1000"), Decimal("1000")], # DAI, USDC, USDT )

CurveAdapter ¶

CurveAdapter(
    config: CurveConfig,
    token_resolver: TokenResolver | None = None,
)

Adapter for Curve Finance DEX protocol.

This adapter provides methods for: - Executing token swaps via Curve pools - Adding liquidity to pools (LP_OPEN) - Removing liquidity from pools (LP_CLOSE) - Handling ERC-20 approvals - Managing slippage protection

Example

config = CurveConfig( chain="ethereum", wallet_address="0x...", ) adapter = CurveAdapter(config)

Execute a swap on 3pool¶

result = adapter.swap( pool_address="0xbEbc44782C7dB0a1A60Cb6fe97d0b483032FF1C7", token_in="USDC", token_out="DAI", amount_in=Decimal("1000"), )

Initialize the adapter.

参数：

名称	类型	描述	默认
`config`	`CurveConfig`	Curve adapter configuration	必需
`token_resolver`	`TokenResolver \| None`	Optional TokenResolver instance. If None, uses singleton.	`None`

get_pool_info ¶

get_pool_info(pool_address: str) -> PoolInfo | None

Get information about a pool.

参数：

名称	类型	描述	默认
`pool_address`	`str`	Pool contract address	必需

返回：

类型	描述
`PoolInfo \| None`	PoolInfo if known, None otherwise

get_pool_by_name ¶

get_pool_by_name(name: str) -> PoolInfo | None

Get pool info by name.

参数：

名称	类型	描述	默认
`name`	`str`	Pool name (e.g., "3pool", "frax_usdc")	必需

返回：

类型	描述
`PoolInfo \| None`	PoolInfo if found, None otherwise

swap ¶

swap(
    pool_address: str,
    token_in: str,
    token_out: str,
    amount_in: Decimal,
    slippage_bps: int | None = None,
    recipient: str | None = None,
    price_ratio: Decimal | None = None,
) -> SwapResult

Build a swap transaction on a Curve pool.

参数：

名称	类型	描述	默认
`pool_address`	`str`	Pool contract address	必需
`token_in`	`str`	Input token symbol or address	必需
`token_out`	`str`	Output token symbol or address	必需
`amount_in`	`Decimal`	Amount of input token (in token units, not wei)	必需
`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`
`price_ratio`	`Decimal \| None`	Price of input token / price of output token (e.g., if swapping USDT at $1 for WETH at $2500, price_ratio = 1/2500 = 0.0004). Required for CryptoSwap/Tricrypto pools; StableSwap pools ignore it. When None and pool is CryptoSwap, the swap fails (fail-closed) rather than executing with inaccurate slippage protection.	`None`

返回：

类型	描述
`SwapResult`	SwapResult with transaction data

add_liquidity ¶

add_liquidity(
    pool_address: str,
    amounts: list[Decimal],
    slippage_bps: int | None = None,
    recipient: str | None = None,
) -> LiquidityResult

Build an add_liquidity transaction (LP_OPEN).

参数：

名称	类型	描述	默认
`pool_address`	`str`	Pool contract address	必需
`amounts`	`list[Decimal]`	List of token amounts to deposit (in token units)	必需
`slippage_bps`	`int \| None`	Slippage tolerance for min LP tokens (default from config)	`None`
`recipient`	`str \| None`	Address to receive LP tokens (default: wallet_address)	`None`

返回：

类型	描述
`LiquidityResult`	LiquidityResult with transaction data

remove_liquidity ¶

remove_liquidity(
    pool_address: str,
    lp_amount: Decimal,
    slippage_bps: int | None = None,
    recipient: str | None = None,
) -> LiquidityResult

Build a remove_liquidity transaction (LP_CLOSE, proportional).

参数：

名称	类型	描述	默认
`pool_address`	`str`	Pool contract address	必需
`lp_amount`	`Decimal`	Amount of LP tokens to burn	必需
`slippage_bps`	`int \| None`	Slippage tolerance for min output (default from config)	`None`
`recipient`	`str \| None`	Address to receive tokens (default: wallet_address)	`None`

返回：

类型	描述
`LiquidityResult`	LiquidityResult with transaction data

remove_liquidity_one_coin ¶

remove_liquidity_one_coin(
    pool_address: str,
    lp_amount: Decimal,
    coin_index: int,
    slippage_bps: int | None = None,
    recipient: str | None = None,
) -> LiquidityResult

Build a remove_liquidity_one_coin transaction (LP_CLOSE, single-sided).

参数：

名称	类型	描述	默认
`pool_address`	`str`	Pool contract address	必需
`lp_amount`	`Decimal`	Amount of LP tokens to burn	必需
`coin_index`	`int`	Index of the coin to receive	必需
`slippage_bps`	`int \| None`	Slippage tolerance (default from config)	`None`
`recipient`	`str \| None`	Address to receive tokens (default: wallet_address)	`None`

返回：

类型	描述
`LiquidityResult`	LiquidityResult with transaction data

set_allowance ¶

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

Set cached allowance (for testing).

参数：

名称	类型	描述	默认
`token`	`str`	Token address	必需
`spender`	`str`	Spender address	必需
`amount`	`int`	Allowance amount	必需

clear_allowance_cache ¶

clear_allowance_cache() -> None

Clear the allowance cache.

CurveConfig `dataclass` ¶

CurveConfig(
    chain: str,
    wallet_address: str,
    default_slippage_bps: int = 50,
    deadline_seconds: int = 300,
    rpc_url: str | None = None,
)

Configuration for CurveAdapter.

属性：

名称	类型	描述
`chain`	`str`	Target blockchain (ethereum, arbitrum)
`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)
`rpc_url`	`str \| None`	Optional JSON-RPC URL for on-chain state queries (e.g., pool balances for accurate remove_liquidity slippage estimates). When provided, the adapter queries pool.balances(i) and lp_token.totalSupply() to compute proportional min_amounts rather than returning zeros. When absent or on RPC failure, min_amounts fall back to [0, 0, ..., 0] with a warning.

__post_init__ ¶

__post_init__() -> None

Validate configuration.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

LiquidityResult `dataclass` ¶

LiquidityResult(
    success: bool,
    transactions: list[TransactionData] = list(),
    pool_address: str = "",
    operation: str = "",
    amounts: list[int] = list(),
    lp_amount: int = 0,
    error: str | None = None,
    gas_estimate: int = 0,
)

Result of a liquidity operation.

属性：

名称	类型	描述
`success`	`bool`	Whether the operation was built successfully
`transactions`	`list[TransactionData]`	List of transactions to execute
`pool_address`	`str`	Pool address
`operation`	`str`	Operation type (add_liquidity, remove_liquidity, remove_liquidity_one_coin)
`amounts`	`list[int]`	Token amounts for the operation
`lp_amount`	`int`	LP token amount (minted or burned)
`error`	`str \| None`	Error message if failed
`gas_estimate`	`int`	Total gas estimate

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PoolInfo `dataclass` ¶

PoolInfo(
    address: str,
    lp_token: str,
    coins: list[str],
    coin_addresses: list[str],
    pool_type: PoolType,
    n_coins: int,
    name: str = "",
    virtual_price: Decimal = (lambda: Decimal("1.0"))(),
)

Information about a Curve pool.

属性：

名称	类型	描述
`address`	`str`	Pool contract address
`lp_token`	`str`	LP token address
`coins`	`list[str]`	List of coin symbols
`coin_addresses`	`list[str]`	List of coin addresses
`pool_type`	`PoolType`	Type of pool (stableswap, cryptoswap, tricrypto)
`n_coins`	`int`	Number of coins in pool
`name`	`str`	Pool name
`virtual_price`	`Decimal`	Pool virtual price (LP token value relative to underlying). Mature pools accumulate fees so virtual_price > 1.0. Used to adjust LP token estimates to prevent over-estimation that causes add_liquidity reverts.

get_coin_index ¶

get_coin_index(coin: str) -> int

Get the index of a coin in the pool.

参数：

名称	类型	描述	默认
`coin`	`str`	Coin symbol or address	必需

返回：

类型	描述
`int`	Index of the coin

引发：

类型	描述
`ValueError`	If coin not found in pool

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

PoolType ¶

Bases: Enum

Curve pool type.

SwapResult `dataclass` ¶

SwapResult(
    success: bool,
    transactions: list[TransactionData] = list(),
    pool_address: str = "",
    amount_in: int = 0,
    amount_out_minimum: int = 0,
    token_in: str = "",
    token_out: str = "",
    error: str | None = None,
    gas_estimate: int = 0,
)

Result of a swap operation.

属性：

名称	类型	描述
`success`	`bool`	Whether the swap was built successfully
`transactions`	`list[TransactionData]`	List of transactions to execute
`pool_address`	`str`	Pool used for swap
`amount_in`	`int`	Input amount in wei
`amount_out_minimum`	`int`	Minimum output amount (with slippage)
`token_in`	`str`	Input token address
`token_out`	`str`	Output token address
`error`	`str \| None`	Error message if failed
`gas_estimate`	`int`	Total gas estimate

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

TransactionData `dataclass` ¶

TransactionData(
    to: str,
    value: int,
    data: str,
    gas_estimate: int,
    description: str,
    tx_type: str = "swap",
)

Transaction data for execution.

属性：

名称	类型	描述
`to`	`str`	Target contract address
`value`	`int`	Native token value to send
`data`	`str`	Encoded calldata
`gas_estimate`	`int`	Estimated gas
`description`	`str`	Human-readable description
`tx_type`	`str`	Type of transaction (approve, swap, add_liquidity, remove_liquidity)

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

AddLiquidityEventData `dataclass` ¶

AddLiquidityEventData(
    provider: str,
    token_amounts: list[int],
    fees: list[int],
    invariant: int,
    token_supply: int,
    pool_address: str,
)

Parsed data from AddLiquidity event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

CurveEvent `dataclass` ¶

CurveEvent(
    event_type: CurveEventType,
    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 Curve event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

CurveEventType ¶

Bases: Enum

Curve event types.

CurveReceiptParser ¶

CurveReceiptParser(chain: str = 'ethereum', **kwargs: Any)

Parser for Curve Finance transaction receipts.

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

Initialize the parser.

参数：

名称	类型	描述	默认
`chain`	`str`	Blockchain network	`'ethereum'`
`**kwargs`	`Any`	Additional arguments (ignored for compatibility)	`{}`

parse_receipt ¶

parse_receipt(receipt: dict[str, Any]) -> ParseResult

Parse a transaction receipt.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict	必需

返回：

类型	描述
`ParseResult`	ParseResult with extracted events

extract_swap_amounts ¶

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

Extract swap amounts from a transaction receipt.

Uses ERC-20 Transfer events to identify token addresses, then resolves actual decimals via TokenResolver for accurate decimal conversion. Falls back to returning None if decimals cannot be resolved (rather than returning wildly wrong amounts).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' and 'from' fields	必需

返回：

类型	描述
`SwapAmounts \| None`	SwapAmounts dataclass if swap event found, None otherwise

extract_position_id ¶

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

Extract position identifier from LP transaction receipt.

For Curve (pool-based LP, no NFT positions), returns the LP token contract address. Unlike V3 DEXes where position_id is an NFT tokenId, Curve LP tokens are fungible ERC-20s — the LP token address is the stable identifier for the position.

The minted LP token amount is available separately via extract_liquidity().

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	必需

返回：

类型	描述
`int \| str \| None`	LP token address as hex string, or None if not found

extract_liquidity ¶

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

Extract LP tokens minted from AddLiquidity transaction.

Returns the LP token amount in human-readable form (e.g., Decimal("98.133")) by dividing the raw wei value by 10^decimals. This matches the convention expected by the LP_CLOSE compiler, which treats the value as a human-readable amount and converts back to wei internally.

Curve LP tokens always have 18 decimals. If the LP token address is found in the receipt, decimals are resolved via the token resolver; otherwise falls back to 18.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	必需

返回：

类型	描述
`Decimal \| None`	LP token amount in human-readable Decimal, or None if not found

extract_lp_tokens_received ¶

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

Extract LP tokens received from AddLiquidity transaction.

Looks for Transfer events from the zero address (mint).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	必需

返回：

类型	描述
`Decimal \| None`	LP token amount in human-readable Decimal, or None if not found

extract_lp_close_data ¶

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

Extract LP close data from transaction receipt.

Looks for RemoveLiquidity, RemoveLiquidityOne, or RemoveLiquidityImbalance events.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' field	必需

返回：

类型	描述
`LPCloseData \| None`	LPCloseData dataclass if liquidity removal found, None otherwise

is_curve_event ¶

is_curve_event(topic: str | bytes) -> bool

Check if a topic is a known Curve event.

参数：

名称	类型	描述	默认
`topic`	`str \| bytes`	Event topic (supports bytes, hex string with/without 0x, any case)	必需

返回：

类型	描述
`bool`	True if topic is a known Curve event

get_event_type ¶

get_event_type(topic: str | bytes) -> CurveEventType

Get the event type for a topic.

参数：

名称	类型	描述	默认
`topic`	`str \| bytes`	Event topic (supports bytes, hex string with/without 0x, any case)	必需

返回：

类型	描述
`CurveEventType`	Event type or UNKNOWN

ParseResult `dataclass` ¶

ParseResult(
    success: bool,
    events: list[CurveEvent] = list(),
    swap_events: list[SwapEventData] = list(),
    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.

RemoveLiquidityEventData `dataclass` ¶

RemoveLiquidityEventData(
    provider: str,
    token_amounts: list[int],
    fees: list[int],
    token_supply: int,
    pool_address: str,
)

Parsed data from RemoveLiquidity event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapEventData `dataclass` ¶

SwapEventData(
    buyer: str,
    sold_id: int,
    tokens_sold: int,
    bought_id: int,
    tokens_bought: int,
    pool_address: str,
)

Parsed data from TokenExchange event.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to dictionary.

Curve¶

almanak.framework.connectors.curve ¶

Execute a swap¶

Add liquidity¶

CurveAdapter ¶

Execute a swap on 3pool¶

get_pool_info ¶

get_pool_by_name ¶

swap ¶

add_liquidity ¶

remove_liquidity ¶

remove_liquidity_one_coin ¶

set_allowance ¶

clear_allowance_cache ¶

CurveConfig dataclass ¶

__post_init__ ¶

to_dict ¶

LiquidityResult dataclass ¶

to_dict ¶

PoolInfo dataclass ¶

get_coin_index ¶

to_dict ¶

PoolType ¶

SwapResult dataclass ¶

to_dict ¶

TransactionData dataclass ¶

to_dict ¶

AddLiquidityEventData dataclass ¶

to_dict ¶

CurveEvent dataclass ¶

to_dict ¶

CurveEventType ¶

CurveReceiptParser ¶

parse_receipt ¶

extract_swap_amounts ¶

extract_position_id ¶

extract_liquidity ¶

extract_lp_tokens_received ¶

extract_lp_close_data ¶

is_curve_event ¶

get_event_type ¶

ParseResult dataclass ¶

to_dict ¶

RemoveLiquidityEventData dataclass ¶

to_dict ¶

SwapEventData dataclass ¶

to_dict ¶

CurveConfig `dataclass` ¶

LiquidityResult `dataclass` ¶

PoolInfo `dataclass` ¶

SwapResult `dataclass` ¶

TransactionData `dataclass` ¶

AddLiquidityEventData `dataclass` ¶

CurveEvent `dataclass` ¶

ParseResult `dataclass` ¶

RemoveLiquidityEventData `dataclass` ¶

SwapEventData `dataclass` ¶