Aller au contenu

Pendle

Connector for Pendle yield trading protocol.

almanak.framework.connectors.pendle

Pendle Protocol Connector

This module provides integration with Pendle Finance, a permissionless yield-trading protocol that enables: - Tokenizing yield-bearing assets into PT (Principal) and YT (Yield) tokens - Trading PT and YT on Pendle's AMM - Providing liquidity to PT/SY pools - Redeeming PT at maturity

Components: - PendleSDK: Low-level protocol interactions - PendleAdapter: ActionType to SDK mapping - PendleReceiptParser: Transaction receipt parsing

Supported Chains: - Arbitrum (primary) - Ethereum

Example

from almanak.framework.connectors.pendle import ( PendleSDK, PendleAdapter, PendleReceiptParser, )

Create SDK

sdk = PendleSDK(rpc_url="https://arb1.arbitrum.io/rpc", chain="arbitrum")

Build swap transaction

tx = sdk.build_swap_exact_token_for_pt( receiver="0x...", market="0x...", token_in="0x...", amount_in=1018, min_pt_out=1018, )

PendleAdapter

PendleAdapter(
    rpc_url: str,
    chain: str = "arbitrum",
    wallet_address: str | None = None,
    api_client: PendleAPIClient | None = None,
    on_chain_reader: PendleOnChainReader | None = None,
)

Adapter for Pendle Protocol operations.

This adapter translates between the framework's ActionType enum and Pendle's specific operations. It handles: - Token swaps to/from PT (Principal Token) - Token swaps to/from YT (Yield Token) - Liquidity provision (adding/removing) - PT/YT redemption at maturity

Example

adapter = PendleAdapter(rpc_url="https://arb1.arbitrum.io/rpc", chain="arbitrum")

Build a swap transaction

tx = adapter.build_swap( params=PendleSwapParams( market="0x...", token_in="0x...", token_out="0x...", amount_in=1018, min_amount_out=1018, receiver="0x...", swap_type="token_to_pt", ) )

Initialize the Pendle adapter.

参数:

名称 类型 描述 默认
rpc_url str

RPC endpoint URL

必需
chain str

Target chain (arbitrum, ethereum)

'arbitrum'
wallet_address str | None

Optional default wallet address for transactions

None
api_client PendleAPIClient | None

Optional PendleAPIClient for REST API quotes

None
on_chain_reader PendleOnChainReader | None

Optional PendleOnChainReader for on-chain fallback

None

supports_action

supports_action(action_type: ActionType) -> bool

Check if this adapter supports the given action type.

get_supported_actions

get_supported_actions() -> list[ActionType]

Get list of supported action types.

build_swap

build_swap(
    params: PendleSwapParams,
) -> PendleTransactionData

Build a swap transaction based on the swap type.

参数:

名称 类型 描述 默认
params PendleSwapParams

Swap parameters including market, tokens, and amounts

必需

返回:

类型 描述
PendleTransactionData

Transaction data ready for execution

build_swap_token_to_pt

build_swap_token_to_pt(
    market: str,
    token_in: str,
    amount_in: int,
    min_pt_out: int,
    receiver: str,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a token -> PT swap transaction.

This is a convenience method for the most common swap type.

参数:

名称 类型 描述 默认
market str

Pendle market address

必需
token_in str

Input token address

必需
amount_in int

Amount of input token

必需
min_pt_out int

Minimum PT to receive

必需
receiver str

Address to receive PT

必需
slippage_bps int

Slippage tolerance in basis points

50

返回:

类型 描述
PendleTransactionData

Transaction data

build_swap_pt_to_token

build_swap_pt_to_token(
    market: str,
    pt_amount: int,
    token_out: str,
    min_token_out: int,
    receiver: str,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a PT -> token swap transaction.

参数:

名称 类型 描述 默认
market str

Pendle market address

必需
pt_amount int

Amount of PT to swap

必需
token_out str

Output token address

必需
min_token_out int

Minimum token to receive

必需
receiver str

Address to receive token

必需
slippage_bps int

Slippage tolerance

50

返回:

类型 描述
PendleTransactionData

Transaction data

build_add_liquidity

build_add_liquidity(
    params: PendleLPParams,
) -> PendleTransactionData

Build an add liquidity transaction.

参数:

名称 类型 描述 默认
params PendleLPParams

Liquidity parameters

必需

返回:

类型 描述
PendleTransactionData

Transaction data

build_remove_liquidity

build_remove_liquidity(
    params: PendleLPParams,
) -> PendleTransactionData

Build a remove liquidity transaction.

参数:

名称 类型 描述 默认
params PendleLPParams

Liquidity parameters

必需

返回:

类型 描述
PendleTransactionData

Transaction data

build_redeem

build_redeem(
    params: PendleRedeemParams,
) -> PendleTransactionData

Build a PT+YT redemption transaction.

参数:

名称 类型 描述 默认
params PendleRedeemParams

Redemption parameters

必需

返回:

类型 描述
PendleTransactionData

Transaction data

build_approve

build_approve(
    token_address: str, amount: int | None = None
) -> PendleTransactionData

Build an approval transaction for the Pendle Router.

参数:

名称 类型 描述 默认
token_address str

Token to approve

必需
amount int | None

Amount to approve (defaults to max)

None

返回:

类型 描述
PendleTransactionData

Transaction data

get_router_address

get_router_address() -> str

Get the Pendle Router address for this chain.

get_gas_estimate

get_gas_estimate(action: PendleActionType) -> int

Get gas estimate for an action.

estimate_output

estimate_output(
    market: str,
    token_in: str,
    amount_in: int,
    swap_type: str,
    slippage_bps: int = 50,
) -> int

Estimate output amount for a swap using a 3-tier cascade: 1. Pendle REST API quote (most accurate) 2. On-chain RouterStatic rate (good fallback) 3. Conservative 1% haircut estimate (last resort, always logged as WARNING)

参数:

名称 类型 描述 默认
market str

Market address

必需
token_in str

Input token address

必需
amount_in int

Input amount in wei

必需
swap_type str

Type of swap ("token_to_pt", "pt_to_token", etc.)

必需
slippage_bps int

Slippage tolerance in basis points

50

返回:

类型 描述
int

Estimated output amount in wei

PendleLPParams dataclass

PendleLPParams(
    market: str,
    token: str,
    amount: int,
    min_amount: int,
    receiver: str,
    operation: str,
    slippage_bps: int = 50,
)

Parameters for Pendle liquidity operations.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PendleRedeemParams dataclass

PendleRedeemParams(
    yt_address: str,
    py_amount: int,
    token_out: str,
    min_token_out: int,
    receiver: str,
    slippage_bps: int = 50,
)

Parameters for Pendle redemption operations.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PendleSwapParams dataclass

PendleSwapParams(
    market: str,
    token_in: str,
    token_out: str,
    amount_in: int,
    min_amount_out: int,
    receiver: str,
    swap_type: str,
    slippage_bps: int = 50,
    token_mint_sy: str | None = None,
)

Parameters for Pendle swap operations.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

BurnEventData dataclass

BurnEventData(
    receiver: str,
    net_lp_burned: int,
    net_sy_out: int,
    net_pt_out: int,
    market_address: str,
)

Parsed data from Pendle Burn (LP removal) event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

MintEventData dataclass

MintEventData(
    receiver: str,
    net_lp_minted: int,
    net_sy_used: int,
    net_pt_used: int,
    market_address: str,
)

Parsed data from Pendle Mint (LP) event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

ParsedSwapResult dataclass

ParsedSwapResult(
    token_in: str,
    token_out: str,
    amount_in: int,
    amount_out: int,
    amount_in_decimal: Decimal,
    amount_out_decimal: Decimal,
    effective_price: Decimal,
    slippage_bps: int,
    market_address: str,
    swap_type: str,
)

High-level swap result from Pendle.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

ParseResult dataclass

ParseResult(
    success: bool,
    events: list[PendleEvent] = list(),
    swap_events: list[SwapEventData] = list(),
    mint_events: list[MintEventData] = list(),
    burn_events: list[BurnEventData] = list(),
    redeem_events: list[RedeemPYEventData] = list(),
    transfer_events: list[TransferEventData] = list(),
    swap_result: ParsedSwapResult | None = None,
    error: str | None = None,
    transaction_hash: str = "",
    block_number: int = 0,
    transaction_success: bool = True,
)

Result of parsing a Pendle receipt.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PendleEvent dataclass

PendleEvent(
    event_type: PendleEventType,
    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 Pendle event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PendleEventType

Bases: Enum

Pendle event types.

PendleReceiptParser

PendleReceiptParser(
    chain: str = "arbitrum",
    token_in_decimals: int = 18,
    token_out_decimals: int = 18,
    quoted_price: Decimal | None = None,
    **kwargs: Any,
)

Parser for Pendle Protocol transaction receipts.

Uses the base infrastructure (EventRegistry, HexDecoder) for standardized event parsing while handling Pendle-specific event structures.

Example

parser = PendleReceiptParser(chain="arbitrum") result = parser.parse_receipt(receipt)

if result.success and result.swap_events: swap = result.swap_events[0] print(f"Swapped {swap.sy_amount} SY for {swap.pt_amount} PT")

Initialize the Pendle receipt parser.

参数:

名称 类型 描述 默认
chain str

Chain name for address resolution

'arbitrum'
token_in_decimals int

Decimals for input token

18
token_out_decimals int

Decimals for output token

18
quoted_price Decimal | None

Expected price for slippage calculation

None

parse_receipt

parse_receipt(
    receipt: dict[str, Any],
    quoted_amount_out: int | None = None,
) -> ParseResult

Parse a Pendle transaction receipt.

参数:

名称 类型 描述 默认
receipt dict[str, Any]

Transaction receipt dictionary

必需
quoted_amount_out int | None

Expected output for slippage calculation

None

返回:

类型 描述
ParseResult

ParseResult with extracted events and swap data

extract_swap_amounts

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

Extract swap amounts from receipt for Result Enrichment.

Called by the framework after SWAP execution to populate ExecutionResult.swap_amounts.

返回:

类型 描述
dict[str, Any] | None

Dictionary with amount_in, amount_out, effective_price, slippage_bps

extract_lp_minted

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

Extract LP tokens minted from receipt.

Called by the framework after LP_OPEN execution.

extract_lp_burned

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

Extract LP tokens burned from receipt.

Called by the framework after LP_CLOSE execution.

extract_redemption_amounts

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

Extract redemption amounts from receipt.

Called by the framework after WITHDRAW/REDEEM execution.

RedeemPYEventData dataclass

RedeemPYEventData(
    caller: str,
    receiver: str,
    net_py_redeemed: int,
    net_sy_redeemed: int,
    yt_address: str,
)

Parsed data from Pendle RedeemPY event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapEventData dataclass

SwapEventData(
    caller: str,
    receiver: str,
    pt_to_account: int,
    sy_to_account: int,
    market_address: str,
)

Parsed data from Pendle Swap event.

is_buy_pt property

is_buy_pt: bool

Check if this is a buy PT operation (SY -> PT).

is_sell_pt property

is_sell_pt: bool

Check if this is a sell PT operation (PT -> SY).

pt_amount property

pt_amount: int

Get absolute PT amount.

sy_amount property

sy_amount: int

Get absolute SY amount.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

TransferEventData dataclass

TransferEventData(
    from_addr: str,
    to_addr: str,
    value: int,
    token_address: str,
)

Parsed data from ERC20 Transfer event.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

LiquidityParams dataclass

LiquidityParams(
    receiver: str,
    market: str,
    token_in: str,
    amount_in: int,
    min_lp_out: int,
    slippage_bps: int = 50,
)

Parameters for liquidity operations.

MarketInfo dataclass

MarketInfo(
    market_address: str,
    sy_address: str,
    pt_address: str,
    yt_address: str,
    expiry: int,
    underlying_token: str,
    underlying_symbol: str,
)

Information about a Pendle market.

is_expired

is_expired(current_timestamp: int) -> bool

Check if the market has expired.

PendleActionType

Bases: Enum

Pendle action types.

PendleQuote dataclass

PendleQuote(
    token_in: str,
    token_out: str,
    amount_in: int,
    amount_out: int,
    price_impact_bps: int,
    gas_estimate: int,
    effective_price: Decimal,
)

Quote for a Pendle operation.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

PendleSDK

PendleSDK(
    rpc_url: str,
    chain: str = "arbitrum",
    token_resolver: TokenResolver | None = None,
)

SDK for interacting with Pendle Protocol.

Pendle enables yield tokenization and trading through its AMM. This SDK builds transactions for: - Swapping tokens to/from PT (Principal Token) - Swapping tokens to/from YT (Yield Token) - Adding/removing liquidity - Minting/redeeming SY and PY tokens

Example

sdk = PendleSDK(rpc_url="https://arb1.arbitrum.io/rpc", chain="arbitrum")

Build swap transaction (WETH -> PT-wstETH)

tx = sdk.build_swap_exact_token_for_pt( receiver="0x...", market="0x...", token_in="0x...", # WETH amount_in=1018, # 1 WETH min_pt_out=1018, # Minimum PT to receive )

Initialize Pendle SDK.

参数:

名称 类型 描述 默认
rpc_url str

RPC endpoint URL

必需
chain str

Target chain (arbitrum, ethereum)

'arbitrum'
token_resolver TokenResolver | None

Optional TokenResolver instance. If None, uses singleton.

None

router_abi property

router_abi: list[dict]

Load router ABI lazily.

erc20_abi property

erc20_abi: list[dict]

Load ERC20 ABI lazily.

get_router

get_router() -> Contract

Get the Pendle Router contract.

build_swap_exact_token_for_pt

build_swap_exact_token_for_pt(
    receiver: str,
    market: str,
    token_in: str,
    amount_in: int,
    min_pt_out: int,
    slippage_bps: int = 50,
    token_mint_sy: str | None = None,
) -> PendleTransactionData

Build a swap transaction from token to PT using swapExactTokenForPtSimple.

This uses the simplified Pendle V4 function that doesn't require ApproxParams or LimitOrderData, making encoding more reliable.

参数:

名称 类型 描述 默认
receiver str

Address to receive the PT

必需
market str

Market address

必需
token_in str

Input token address

必需
amount_in int

Amount of input token (in wei)

必需
min_pt_out int

Minimum PT to receive

必需
slippage_bps int

Slippage tolerance in basis points

50
token_mint_sy str | None

Token that mints SY (defaults to token_in if not specified). For yield-bearing token markets (like fUSDT0), this should be the yield-bearing token address, not the underlying.

None

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_swap_exact_pt_for_token

build_swap_exact_pt_for_token(
    receiver: str,
    market: str,
    pt_amount: int,
    token_out: str,
    min_token_out: int,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a swap transaction from PT to token using swapExactPtForTokenSimple.

This uses the simplified Pendle V4 function that doesn't require LimitOrderData, making encoding more reliable.

参数:

名称 类型 描述 默认
receiver str

Address to receive the token

必需
market str

Market address

必需
pt_amount int

Amount of PT to swap

必需
token_out str

Output token address

必需
min_token_out int

Minimum output token to receive

必需
slippage_bps int

Slippage tolerance in basis points

50

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_swap_exact_token_for_yt

build_swap_exact_token_for_yt(
    receiver: str,
    market: str,
    token_in: str,
    amount_in: int,
    min_yt_out: int,
    slippage_bps: int = 50,
    token_mint_sy: str | None = None,
) -> PendleTransactionData

Build a swap transaction from token to YT using swapExactTokenForYt.

Unlike PT swaps which use the Simple variant, YT swaps require ApproxParams for binary search of optimal flash swap size, plus LimitOrderData.

参数:

名称 类型 描述 默认
receiver str

Address to receive the YT

必需
market str

Market address

必需
token_in str

Input token address

必需
amount_in int

Amount of input token (in wei)

必需
min_yt_out int

Minimum YT to receive

必需
slippage_bps int

Slippage tolerance in basis points

50
token_mint_sy str | None

Token that mints SY (defaults to token_in)

None

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_swap_exact_yt_for_token

build_swap_exact_yt_for_token(
    receiver: str,
    market: str,
    yt_amount: int,
    token_out: str,
    min_token_out: int,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a swap transaction from YT to token using swapExactYtForToken.

参数:

名称 类型 描述 默认
receiver str

Address to receive the token

必需
market str

Market address

必需
yt_amount int

Amount of YT to swap

必需
token_out str

Output token address

必需
min_token_out int

Minimum output token to receive

必需
slippage_bps int

Slippage tolerance in basis points

50

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_add_liquidity_single_token

build_add_liquidity_single_token(
    receiver: str,
    market: str,
    token_in: str,
    amount_in: int,
    min_lp_out: int,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a transaction to add liquidity with a single token.

This adds liquidity to a Pendle market using a single input token. The router handles conversion to the proper ratio of SY and PT.

参数:

名称 类型 描述 默认
receiver str

Address to receive LP tokens

必需
market str

Market address

必需
token_in str

Input token address

必需
amount_in int

Amount of input token

必需
min_lp_out int

Minimum LP tokens to receive

必需
slippage_bps int

Slippage tolerance in basis points

50

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_remove_liquidity_single_token

build_remove_liquidity_single_token(
    receiver: str,
    market: str,
    lp_amount: int,
    token_out: str,
    min_token_out: int,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a transaction to remove liquidity to a single token.

参数:

名称 类型 描述 默认
receiver str

Address to receive output token

必需
market str

Market address

必需
lp_amount int

Amount of LP tokens to burn

必需
token_out str

Output token address

必需
min_token_out int

Minimum output token to receive

必需
slippage_bps int

Slippage tolerance in basis points

50

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_redeem_py_to_token

build_redeem_py_to_token(
    receiver: str,
    yt_address: str,
    py_amount: int,
    token_out: str,
    min_token_out: int,
    slippage_bps: int = 50,
) -> PendleTransactionData

Build a transaction to redeem PT+YT to token.

After maturity, PT can be redeemed 1:1 for the underlying. Before maturity, you need equal amounts of PT and YT to redeem.

参数:

名称 类型 描述 默认
receiver str

Address to receive output token

必需
yt_address str

YT contract address

必需
py_amount int

Amount of PT+YT to redeem

必需
token_out str

Output token address

必需
min_token_out int

Minimum output token

必需
slippage_bps int

Slippage tolerance

50

返回:

类型 描述
PendleTransactionData

Transaction data for execution

build_approve_tx

build_approve_tx(
    token_address: str,
    spender: str | None = None,
    amount: int = MAX_UINT256,
) -> PendleTransactionData

Build an ERC-20 approval transaction.

参数:

名称 类型 描述 默认
token_address str

Token to approve

必需
spender str | None

Spender address (defaults to router)

None
amount int

Amount to approve (defaults to max)

MAX_UINT256

返回:

类型 描述
PendleTransactionData

Transaction data for execution

PendleTransactionData dataclass

PendleTransactionData(
    to: str,
    value: int,
    data: str,
    gas_estimate: int,
    description: str,
    action_type: PendleActionType,
)

Transaction data for Pendle operations.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapParams dataclass

SwapParams(
    receiver: str,
    market: str,
    min_out: int,
    token_in: str,
    amount_in: int,
    slippage_bps: int = 50,
)

Parameters for a swap operation.

amount_out_minimum property

amount_out_minimum: int

Calculate minimum output with slippage.

get_pendle_adapter

get_pendle_adapter(
    rpc_url: str,
    chain: str = "arbitrum",
    wallet_address: str | None = None,
) -> PendleAdapter

Factory function to create a PendleAdapter instance.

get_pendle_sdk

get_pendle_sdk(
    rpc_url: str, chain: str = "arbitrum"
) -> PendleSDK

Factory function to create a PendleSDK instance.