Curve¶

Connector for Curve Finance DEX.

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

quote_swap_output ¶

quote_swap_output(
    *,
    pool_address: str,
    token_in: str,
    token_out: str,
    amount_in_wei: int,
) -> int

Quote a Curve exact-input swap with the pool's on-chain quote method.

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,
    gateway_client: GatewayClient | 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"))(),
    use_underlying: bool = False,
    is_ng: bool = False,
)

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,
    amount_out_estimate: int = 0,
    token_out_decimals: int = 18,
    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],
    *,
    expected_out: Decimal | None = None,
) -> 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	必需
`expected_out`	`Decimal \| None`	VIB-3203 Phase B — pre-slippage-discount quote in human (Decimal) units, sourced from `ActionBundle.metadata["expected_output_human"]` by the ResultEnricher. When provided and positive, realized `slippage_bps` is computed as `(expected_out - amount_out_decimal) / expected_out * 10_000`. When absent, `slippage_bps` stays `None` (legacy behaviour).	`None`

返回：

类型	描述
`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_open_data ¶

extract_lp_open_data(
    receipt: dict[str, Any],
) -> LPOpenData | None

Extract LP open data from an AddLiquidity transaction receipt (VIB-4968).

Curve is a fungible-LP (ERC20 LP-token) venue — there is no NFT position, no tick bracket, and no per-position id. The single field the LP accounting handler genuinely needs from the open receipt is the canonical pool_address so handle_lp can book the LP_OPEN event (_resolve_lp_pool_address step 1). Pre-VIB-4968 the parser had no extract_lp_open_data at all, so the receipt-extraction priority yielded nothing and — combined with the bare-label position-key tail — the handler dropped the event entirely (zero accounting_events rows for every Curve LP_OPEN).

Directional null-contract (Empty ≠ Zero ≠ None, blueprint 27):

pool_address = the AddLiquidity event emitter (the Curve pool contract). Real 0x address — chain data, most reliable.
position_id = 0 — fungible LP has no NFT id. The handler's _resolve_lp_open_discriminator treats 0 as "no discriminator" and persists position_id = None (the faithful fungible-LP value).
tick_lower / tick_upper / liquidity / current_tick / position_hash stay None — Curve has no tick model and fabricating a bracket would be a correctness regression.
amount0 / amount1 carry the raw measured token_amounts for the first two coins so the handler can scale them by token decimals. None (not 0) when the leg is genuinely absent.

参数：

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

返回：

类型	描述
`LPOpenData \| None`	LPOpenData if an AddLiquidity event is found, None otherwise.

extract_primitive_money_legs ¶

extract_primitive_money_legs(
    receipt: dict[str, Any],
) -> PrimitiveMoneyLegs | None

VIB-3587 — declare the LP_OPEN money legs as a typed PrimitiveMoneyLegs the ledger dispatcher consumes directly (the Lido / TJ V2 US-009 pattern).

Inverts the legacy control flow (blueprint 27 §6.6, 05 §7): instead of the ledger reverse-engineering an LP_OPEN's legs from LPOpenData.amount0 / amount1 (which it maps positionally onto token_in /token_out of the pool's FIRST TWO coins), the connector DECLARES the coin(s) it actually funded on-chain. Curve is a multi-coin (2/3/4) venue with single-sided and non-leading-coin deposits, so the two-slot legacy guess is structurally wrong:

a single-sided deposit of coin 0 left coin 1 carrying a fabricated zero amount_out leg (and vice-versa) — a measured-zero where the coin was simply UNFUNDED (Empty ≠ Zero violation);
a deposit of coin index 2+ (e.g. USDT in 3pool, crvUSD in 4pool) was dropped entirely — amount0 / amount1 only ever carry coins 0/1.

The declared legs are built FROM the AddLiquidity event's pool-coin-ordered token_amounts (chain truth) joined to the pool's coin_addresses (same index order), emitting one INPUT leg per FUNDED coin and NOTHING for an unfunded coin. The dispatcher (_extract_from_declared_legs) projects leg0 → token_in / amount_in and leg1 → token_out / amount_out — lane-symmetric with _extract_from_lp_open for a 2-coin deposit, and a single funded leg lands on token_in only (token_out stays empty — no fabricated zero). A 3rd+ funded coin surfaces the dispatcher's documented "dropped leg" WARN rather than silently corrupting the trade tape.

Returns None (→ legacy LP_OPEN fallback, byte-identical rows) when the receipt carries no AddLiquidity event, the pool's coin metadata is unknown, or no coin is funded — so non-Curve-resolvable receipts degrade unchanged. Never raises: any failure degrades to None rather than halting the live accounting writer.

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

extract_protocol_fees ¶

extract_protocol_fees(
    receipt: dict[str, Any],
) -> ProtocolFees

VIB-3495: Curve Finance LP protocol fee coverage audit.

Curve NG pools encode fees arrays in AddLiquidity/RemoveLiquidity events, but these are LP-accrued fee amounts in token units — NOT a USD-denominated protocol fee. Additionally, Curve charges an admin fee (a cut of the LP fee) that is retained by the DAO, but this is not emitted in any receipt event. Converting token amounts to USD requires a price oracle unavailable at the receipt-parser layer.

Returns a ProtocolFees with unavailable_reason so downstream attribution records "known-unknown" rather than "parser absent" (returning None was the pre-VIB-3495 behaviour).

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.

getattr ¶

__getattr__(name: str) -> Any

PEP 562 lazy attribute access.

Curve¶

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

quote_swap_output ¶

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

extract_primitive_money_legs ¶

extract_lp_close_data ¶

extract_protocol_fees ¶

is_curve_event ¶

get_event_type ¶

ParseResult dataclass ¶

to_dict ¶

RemoveLiquidityEventData dataclass ¶

to_dict ¶

SwapEventData dataclass ¶

to_dict ¶

__getattr__ ¶

CurveConfig `dataclass` ¶

LiquidityResult `dataclass` ¶

PoolInfo `dataclass` ¶

SwapResult `dataclass` ¶

TransactionData `dataclass` ¶

AddLiquidityEventData `dataclass` ¶

CurveEvent `dataclass` ¶

ParseResult `dataclass` ¶

RemoveLiquidityEventData `dataclass` ¶

SwapEventData `dataclass` ¶

getattr ¶