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.

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

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

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

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

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],
    *,
    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).

Parameters:

Name	Type	Description	Default
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs' and 'from' fields	required
`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`

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

Parameters:

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

Returns:

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

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

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.

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.

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 ¶