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

返回：

类型	描述
`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,
)

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)

__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 = "",
)

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

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.

Note: Decimal conversions assume 18 decimals. Curve pools often use tokens with different decimals (e.g., USDC/USDT with 6, WBTC with 8). The raw amount_in/amount_out fields are always accurate; use those with your own decimal scaling for precise calculations.

参数：

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

返回：

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

extract_lp_tokens_received ¶

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

Extract LP tokens received from AddLiquidity transaction.

Looks for Transfer events from the pool address to the user.

参数：

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

返回：

类型	描述
`int \| None`	LP token amount if found, None otherwise

extract_lp_close_data ¶

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

Extract LP close data from transaction receipt.

Looks for 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_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` ¶