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.

Parameters:

Name	Type	Description	Default
`config`	`CurveConfig`	Curve adapter configuration	required
`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.

Parameters:

Name	Type	Description	Default
`pool_address`	`str`	Pool contract address	required

Returns:

Type	Description
`PoolInfo \| None`	PoolInfo if known, None otherwise

get_pool_by_name ¶

get_pool_by_name(name: str) -> PoolInfo | None

Get pool info by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	Pool name (e.g., "3pool", "frax_usdc")	required

Returns:

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

Parameters:

Name	Type	Description	Default
`pool_address`	`str`	Pool contract address	required
`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
`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`

Returns:

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

Parameters:

Name	Type	Description	Default
`pool_address`	`str`	Pool contract address	required
`amounts`	`list[Decimal]`	List of token amounts to deposit (in token units)	required
`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`

Returns:

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

Parameters:

Name	Type	Description	Default
`pool_address`	`str`	Pool contract address	required
`lp_amount`	`Decimal`	Amount of LP tokens to burn	required
`slippage_bps`	`int \| None`	Slippage tolerance for min output (default from config)	`None`
`recipient`	`str \| None`	Address to receive tokens (default: wallet_address)	`None`

Returns:

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

Parameters:

Name	Type	Description	Default
`pool_address`	`str`	Pool contract address	required
`lp_amount`	`Decimal`	Amount of LP tokens to burn	required
`coin_index`	`int`	Index of the coin to receive	required
`slippage_bps`	`int \| None`	Slippage tolerance (default from config)	`None`
`recipient`	`str \| None`	Address to receive tokens (default: wallet_address)	`None`

Returns:

Type	Description
`LiquidityResult`	LiquidityResult with transaction data

set_allowance ¶

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

Set cached allowance (for testing).

Parameters:

Name	Type	Description	Default
`token`	`str`	Token address	required
`spender`	`str`	Spender address	required
`amount`	`int`	Allowance amount	required

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.

Attributes:

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

Attributes:

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

Attributes:

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

Parameters:

Name	Type	Description	Default
`coin`	`str`	Coin symbol or address	required

Returns:

Type	Description
`int`	Index of the coin

Raises:

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

Attributes:

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

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.

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.

Parameters:

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

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict	required

Returns:

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

Parameters:

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

Returns:

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

Parameters:

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

Returns:

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

Parameters:

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

Returns:

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

Parameters:

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

Returns:

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

Parameters:

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

Returns:

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

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

get_event_type ¶

get_event_type(topic: str | bytes) -> CurveEventType

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