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

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.

Parameters:

Name	Type	Description	Default
rpc_url	str	RPC endpoint URL	required
chain	str	Target chain (arbitrum, ethereum)	'arbitrum'
wallet_address	str | None	Optional default wallet address for transactions	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.

Parameters:

Name	Type	Description	Default
params	PendleSwapParams	Swap parameters including market, tokens, and amounts	required

Returns:

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

Parameters:

Name	Type	Description	Default
market	str	Pendle market address	required
token_in	str	Input token address	required
amount_in	int	Amount of input token	required
min_pt_out	int	Minimum PT to receive	required
receiver	str	Address to receive PT	required
slippage_bps	int	Slippage tolerance in basis points	50

Returns:

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

Parameters:

Name	Type	Description	Default
market	str	Pendle market address	required
pt_amount	int	Amount of PT to swap	required
token_out	str	Output token address	required
min_token_out	int	Minimum token to receive	required
receiver	str	Address to receive token	required
slippage_bps	int	Slippage tolerance	50

Returns:

Type	Description
PendleTransactionData	Transaction data

build_add_liquidity

build_add_liquidity(
    params: PendleLPParams,
) -> PendleTransactionData

Build an add liquidity transaction.

Parameters:

Name	Type	Description	Default
params	PendleLPParams	Liquidity parameters	required

Returns:

Type	Description
PendleTransactionData	Transaction data

build_remove_liquidity

build_remove_liquidity(
    params: PendleLPParams,
) -> PendleTransactionData

Build a remove liquidity transaction.

Parameters:

Name	Type	Description	Default
params	PendleLPParams	Liquidity parameters	required

Returns:

Type	Description
PendleTransactionData	Transaction data

build_redeem

build_redeem(
    params: PendleRedeemParams,
) -> PendleTransactionData

Build a PT+YT redemption transaction.

Parameters:

Name	Type	Description	Default
params	PendleRedeemParams	Redemption parameters	required

Returns:

Type	Description
PendleTransactionData	Transaction data

build_approve

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

Build an approval transaction for the Pendle Router.

Parameters:

Name	Type	Description	Default
token_address	str	Token to approve	required
amount	int | None	Amount to approve (defaults to max)	None

Returns:

Type	Description
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,
) -> int

Estimate output amount for a swap.

Note: This is a simplified estimation. In production, use the Pendle API or RouterStatic contract for accurate quotes.

Parameters:

Name	Type	Description	Default
market	str	Market address	required
token_in	str	Input token	required
amount_in	int	Input amount	required
swap_type	str	Type of swap	required

Returns:

Type	Description
int	Estimated output amount

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.

Parameters:

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

Parameters:

Name	Type	Description	Default
receipt	dict[str, Any]	Transaction receipt dictionary	required
quoted_amount_out	int | None	Expected output for slippage calculation	None

Returns:

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

Returns:

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

Parameters:

Name	Type	Description	Default
rpc_url	str	RPC endpoint URL	required
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.

Parameters:

Name	Type	Description	Default
receiver	str	Address to receive the PT	required
market	str	Market address	required
token_in	str	Input token address	required
amount_in	int	Amount of input token (in wei)	required
min_pt_out	int	Minimum PT to receive	required
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

Returns:

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

Parameters:

Name	Type	Description	Default
receiver	str	Address to receive the token	required
market	str	Market address	required
pt_amount	int	Amount of PT to swap	required
token_out	str	Output token address	required
min_token_out	int	Minimum output token to receive	required
slippage_bps	int	Slippage tolerance in basis points	50

Returns:

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

Parameters:

Name	Type	Description	Default
receiver	str	Address to receive LP tokens	required
market	str	Market address	required
token_in	str	Input token address	required
amount_in	int	Amount of input token	required
min_lp_out	int	Minimum LP tokens to receive	required
slippage_bps	int	Slippage tolerance in basis points	50

Returns:

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

Parameters:

Name	Type	Description	Default
receiver	str	Address to receive output token	required
market	str	Market address	required
lp_amount	int	Amount of LP tokens to burn	required
token_out	str	Output token address	required
min_token_out	int	Minimum output token to receive	required
slippage_bps	int	Slippage tolerance in basis points	50

Returns:

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

Parameters:

Name	Type	Description	Default
receiver	str	Address to receive output token	required
yt_address	str	YT contract address	required
py_amount	int	Amount of PT+YT to redeem	required
token_out	str	Output token address	required
min_token_out	int	Minimum output token	required
slippage_bps	int	Slippage tolerance	50

Returns:

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

Parameters:

Name	Type	Description	Default
token_address	str	Token to approve	required
spender	str | None	Spender address (defaults to router)	None
amount	int	Amount to approve (defaults to max)	MAX_UINT256

Returns:

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