Skip to content

PancakeSwap V3

Connector for PancakeSwap V3 DEX.

almanak.framework.connectors.pancakeswap_v3

PancakeSwap V3 Connector.

This module provides an adapter for interacting with PancakeSwap V3, which is a Uniswap V3 fork with different fee tiers and addresses.

PancakeSwap V3 is a decentralized exchange supporting: - Exact input swaps (swap specific amount of input token) - Exact output swaps (receive specific amount of output token) - Multiple fee tiers (100, 500, 2500, 10000 bps)

Supported chains: - BNB Smart Chain (BSC) - Ethereum - Arbitrum

Example

from almanak.framework.connectors.pancakeswap_v3 import ( PancakeSwapV3Adapter, PancakeSwapV3Config, )

config = PancakeSwapV3Config( chain="bnb", wallet_address="0x...", ) adapter = PancakeSwapV3Adapter(config)

Swap exact input

result = adapter.swap_exact_input( token_in="USDT", token_out="WBNB", amount_in=Decimal("100"), )

PancakeSwapV3Adapter

PancakeSwapV3Adapter(
    config: PancakeSwapV3Config,
    token_resolver: TokenResolver | None = None,
)

Adapter for PancakeSwap V3 decentralized exchange.

This adapter provides methods for swapping tokens on PancakeSwap V3: - Exact input swaps (specify input amount, receive variable output) - Exact output swaps (specify output amount, send variable input)

PancakeSwap V3 is a Uniswap V3 fork with different fee tiers (100, 500, 2500, 10000 bps).

Example

config = PancakeSwapV3Config( chain="bnb", wallet_address="0x...", ) adapter = PancakeSwapV3Adapter(config)

Swap 100 USDT for WBNB

result = adapter.swap_exact_input("USDT", "WBNB", Decimal("100"))

Initialize the adapter.

Parameters:

Name Type Description Default
config PancakeSwapV3Config

Adapter configuration

required
token_resolver TokenResolver | None

Optional TokenResolver instance. If None, uses singleton from get_token_resolver().

None

swap_exact_input

swap_exact_input(
    token_in: str,
    token_out: str,
    amount_in: Decimal,
    amount_out_min: Decimal | None = None,
    fee_tier: int | None = None,
    recipient: str | None = None,
    deadline: int | None = None,
) -> TransactionResult

Build an exact input swap transaction.

Swaps a specific amount of input token for a variable amount of output token.

Parameters:

Name Type Description Default
token_in str

Input token symbol or address

required
token_out str

Output token symbol or address

required
amount_in Decimal

Exact amount of input token to swap

required
amount_out_min Decimal | None

Minimum output amount (uses slippage if None)

None
fee_tier int | None

Pool fee tier in bps (default from config)

None
recipient str | None

Address to receive output (default: wallet_address)

None
deadline int | None

Transaction deadline timestamp (default: 20 min from now)

None

Returns:

Type Description
TransactionResult

TransactionResult with transaction data

swap_exact_output

swap_exact_output(
    token_in: str,
    token_out: str,
    amount_out: Decimal,
    amount_in_max: Decimal | None = None,
    fee_tier: int | None = None,
    recipient: str | None = None,
    deadline: int | None = None,
) -> TransactionResult

Build an exact output swap transaction.

Swaps a variable amount of input token for a specific amount of output token.

Parameters:

Name Type Description Default
token_in str

Input token symbol or address

required
token_out str

Output token symbol or address

required
amount_out Decimal

Exact amount of output token to receive

required
amount_in_max Decimal | None

Maximum input amount (uses slippage if None)

None
fee_tier int | None

Pool fee tier in bps (default from config)

None
recipient str | None

Address to receive output (default: wallet_address)

None
deadline int | None

Transaction deadline timestamp (default: 20 min from now)

None

Returns:

Type Description
TransactionResult

TransactionResult with transaction data

PancakeSwapV3Config dataclass

PancakeSwapV3Config(
    chain: str,
    wallet_address: str,
    default_slippage_bps: int = 50,
    default_fee_tier: int = 100,
    price_provider: dict[str, Decimal] | None = None,
    allow_placeholder_prices: bool = False,
)

Configuration for PancakeSwap V3 adapter.

Attributes:

Name Type Description
chain str

Blockchain network (bnb, ethereum, arbitrum)

wallet_address str

User wallet address

default_slippage_bps int

Default slippage tolerance in basis points

default_fee_tier int

Default fee tier in basis points

price_provider dict[str, Decimal] | None

Price oracle dict (token symbol -> USD price). Required for production use to calculate accurate slippage amounts.

allow_placeholder_prices bool

If False (default), raises ValueError when no price_provider is given. Set to True ONLY for unit tests.

__post_init__

__post_init__() -> None

Validate configuration.

TransactionResult dataclass

TransactionResult(
    success: bool,
    tx_data: dict[str, Any] | None = None,
    gas_estimate: int = 0,
    description: str = "",
    error: str | None = None,
)

Result of a transaction build operation.

Attributes:

Name Type Description
success bool

Whether operation succeeded

tx_data dict[str, Any] | None

Transaction data (to, value, data)

gas_estimate int

Estimated gas

description str

Human-readable description

error str | None

Error message if failed

PancakeSwapV3EventType

Bases: Enum

PancakeSwap V3 event types.

PancakeSwapV3ReceiptParser

PancakeSwapV3ReceiptParser(
    chain: str = "bsc", **kwargs: Any
)

Bases: BaseReceiptParser[SwapEventData, ParseResult]

Parser for PancakeSwap V3 transaction receipts.

Uses base infrastructure for common parsing logic while handling PancakeSwap V3-specific event decoding.

Initialize the parser.

Parameters:

Name Type Description Default
chain str

Blockchain network (for position manager address lookup)

'bsc'

extract_swap_amounts

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

Extract swap amounts from a transaction receipt.

Note: Decimal conversions assume 18 decimals. PancakeSwap pools often include tokens with different decimals (e.g., USDC with 6, WBTC with 8). The raw amount_in/amount_out fields are always accurate; use those with your own decimal scaling for precise calculations.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

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 | None

Extract LP position ID (NFT tokenId) from a transaction receipt.

Looks for ERC-721 Transfer events from the NonfungiblePositionManager where from=address(0), indicating a mint.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

required

Returns:

Type Description
int | None

Position ID (tokenId) if found, None otherwise

extract_tick_lower

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

Extract tick lower from LP mint transaction receipt.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

required

Returns:

Type Description
int | None

Tick lower value if found, None otherwise

extract_tick_upper

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

Extract tick upper from LP mint transaction receipt.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

required

Returns:

Type Description
int | None

Tick upper value if found, None otherwise

extract_liquidity

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

Extract liquidity from LP mint transaction receipt.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

required

Returns:

Type Description
int | None

Liquidity amount if found, None otherwise

extract_lp_close_data

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

Extract LP close data from transaction receipt.

Looks for Collect events which indicate fees and principal being collected.

Parameters:

Name Type Description Default
receipt dict[str, Any]

Transaction receipt dict with 'logs' field

required

Returns:

Type Description
LPCloseData | None

LPCloseData dataclass if Collect event found, None otherwise

parse_swap

parse_swap(log: dict[str, Any]) -> SwapEventData | None

Parse a Swap event from a single log entry.

Backward compatibility method.

is_pancakeswap_event

is_pancakeswap_event(topic: str | bytes) -> bool

Check if a topic is a known PancakeSwap V3 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 PancakeSwap V3 event

get_event_type

get_event_type(
    topic: str | bytes,
) -> PancakeSwapV3EventType

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
PancakeSwapV3EventType

Event type or UNKNOWN

ParseResult dataclass

ParseResult(
    success: bool,
    swaps: list[SwapEventData] = list(),
    error: str | None = None,
    transaction_hash: str = "",
    block_number: int = 0,
)

Result of parsing a receipt.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.

SwapEventData dataclass

SwapEventData(
    pool: str,
    sender: str,
    recipient: str,
    amount0: Decimal,
    amount1: Decimal,
    sqrt_price_x96: int = 0,
    liquidity: int = 0,
    tick: int = 0,
    protocol_fees_token0: int = 0,
    protocol_fees_token1: int = 0,
)

Parsed data from PancakeSwap V3 Swap event.

Note: PancakeSwap V3 Swap event has 9 parameters (vs 7 for Uniswap V3): - sender, recipient (indexed) - amount0, amount1, sqrtPriceX96, liquidity, tick (same as UniV3) - protocolFeesToken0, protocolFeesToken1 (PancakeSwap V3 specific)

token0_in property

token0_in: bool

Check if token0 is the input token.

token1_in property

token1_in: bool

Check if token1 is the input token.

to_dict

to_dict() -> dict[str, Any]

Convert to dictionary.