PancakeSwap V3

Connector for PancakeSwap V3 DEX.

almanak.connectors.pancakeswap_v3

PancakeSwap V3 Connector (concentrated liquidity AMM).

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

Example

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

build_extract_kwargs

build_extract_kwargs(
    *, field: str, bundle_metadata: dict[str, Any]
) -> dict[str, Any]

Return PancakeSwapV3-owned kwargs for ResultEnricher extraction calls.

VIB-3164: the compiler records full token identity (from_token / to_token dicts with address, symbol, decimals — see UniswapV3Compiler.compile_swap) in ActionBundle.metadata. Threading it here lets _resolve_swap_decimals resolve decimals when the TokenResolver misses, instead of dropping the row or emitting an unresolved-input row.

Native-token entries are skipped: the receipt's Transfer events carry the wrapped token's address, so a native entry can never match by address, and its decimals (18) equal the fallback anyway.

extract_swap_amounts

extract_swap_amounts(
    receipt: dict[str, Any],
    *,
    expected_out: Decimal | None = None,
    swap_token_meta: dict[str, dict[str, Any]]
    | 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 the token resolver for accurate decimal conversion.

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. When absent, slippage_bps stays None.	None
swap_token_meta	dict[str, dict[str, Any]] | None	VIB-3164 — compiler-supplied token metadata threaded from build_extract_kwargs via the ResultEnricher hook. Shape: {"token_in": {"address": ..., "symbol": ..., "decimals": ...}, "token_out": {...}}. Hints win over the TokenResolver per address (address-keyed only — no direction fallback, because when _pick_swap_raw_amounts returns None there are no raw amounts to scale, so hints have nothing to rescue in that path).	None

Returns:

Type	Description
SwapAmounts | None	SwapAmounts dataclass if swap event found, None otherwise

Implementation note

Phase 8.4 — this is a thin orchestrator over per-phase helpers. Each helper is individually testable and keeps CC bounded. Preserves: (1) wallet-level "first transfer out / last transfer in" semantics for multi-hop disambiguation, (2) SwapAmounts field surface, (3) amount0/amount1 sign conventions (the parser is wallet-transfer based and does not inspect Swap-event signs here — that path runs on parse_receipt).

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.

Mint event data layout (non-indexed fields, 32-byte padded): - sender (address): offset 0 - amount (uint128): offset 32 - amount0 (uint256): offset 64 - amount1 (uint256): offset 96

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.

Decodes the PancakeSwap V3 LP close pattern: decreaseLiquidity emits Burn (which carries the principal amounts), and collect emits Collect (which carries principal PLUS earned fees). The accrued fees are the difference between the two. Mirrors the Uniswap V3 implementation since PancakeSwap V3 is a direct UV3 fork at the pool contract level.

Burn(address indexed owner, int24 indexed tickLower,

int24 indexed tickUpper, uint128 amount, uint256 amount0, uint256 amount1)

• data layout: amount (uint128, left-padded to 32B) ‖ amount0 (uint256) ‖ amount1 (uint256)

Collect(address indexed owner, address recipient,

int24 indexed tickLower, int24 indexed tickUpper, uint128 amount0, uint128 amount1)

• owner, tickLower, tickUpper are indexed (3 topics + topic0). recipient is non-indexed — it occupies the first 32-byte data slot, so amount0/amount1 start at offsets 32 and 64.

For a fee-only collect() (no decreaseLiquidity in the same TX), there is no Burn event. We treat the full Collect amounts as fees, principal=0.

VIB-4305: captures pool_address from the Burn event emitter so the registry-payload builder can stamp the LP_CLOSE position_registry row's pool_address field (semantic_grouping_key anchor) without an off-chain RPC. Mirrors the Uniswap V3 close-side capture from VIB-3940.

Parameters:

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

Returns:

Type	Description
LPCloseData | None	LPCloseData if Burn or Collect event found, None otherwise.

extract_lp_open_data

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

Extract LP open data from a PancakeSwap V3 mint receipt.

Looks for IncreaseLiquidity events emitted by the PancakeSwap V3 NonfungiblePositionManager when an LP position is opened or topped up. The event signature is identical to Uniswap V3's NPM::

IncreaseLiquidity(
    uint256 indexed tokenId,
    uint128 liquidity,
    uint256 amount0,
    uint256 amount1,
)

Note: PancakeSwap V3's Swap event has 9 parameters (vs 7 for Uniswap V3 — the extra two are protocolFeesToken{0,1}), so current_tick is recovered from the Swap event at the same slot offset as UV3 (offset 128 = 5th 32-byte slot) — the layout amount0 | amount1 | sqrtPriceX96 | liquidity | tick is identical for the first 5 slots; the protocol-fee slots come after and don't shift the tick offset.

Behaviour contract (matches the Uniswap V3 baseline):

• Returns LPOpenData populated with the raw on-chain ints (position_id, liquidity, amount0, amount1). The accounting handler is responsible for decimal-scaling.
• Returns None when no IncreaseLiquidity log is present.
• No outer try/except — the fail-closed variant extract_lp_open_data_result distinguishes parser crash vs. missing event per VIB-3159 / Blueprint 19.
• Fail-loud on unsupported chains (warn + return None). NEVER silently default to a known-chain NPM — that would mis-attribute logs the moment PCS deploys with a different address.

Parameters:

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

Returns:

Type	Description
LPOpenData | None	LPOpenData if an IncreaseLiquidity event is present,
LPOpenData | None	None otherwise.

extract_lp_open_data_result

extract_lp_open_data_result(
    receipt: dict[str, Any],
) -> ExtractResult[LPOpenData]

Fail-closed variant of :meth:extract_lp_open_data — see VIB-3159.

Distinguishes "no IncreaseLiquidity event in receipt" (benign — e.g. an LP_OPEN that failed mid-bundle, or a non-NPM contract path) from "parser crashed on a malformed receipt". Both are returned as None by the legacy method, which forces the enricher to treat genuine parse failures as missing data — the same ghost-position class of bug VIB-3159 addresses for Uniswap V3.

extract_registry_payload_open

extract_registry_payload_open(
    receipt: dict[str, Any], *, fee_tier: int | None = None
) -> dict[str, Any] | None

Build the LP_OPEN position_registry.payload dict.

Wraps :meth:extract_lp_open_data and composes the canonical 8-key shape (plus optional fee_tier and the per-chain nft_manager_addr). Returns None when any of the load-bearing identity fields are missing — the caller treats that as "fall back to accounting_only", per CLAUDE.md "Empty != Zero" (a zero-substituted token_id would silently corrupt the physical_identity_hash).

Parameters:

Name	Type	Description	Default
receipt	dict[str, Any]	Transaction receipt dict with logs field.	required
fee_tier	int | None	Optional pool fee tier (e.g. 500 for 0.05%); forwarded from the intent's compile-time metadata. None when unknown — the payload key stays absent rather than substituting 0 (Empty != Zero).	None

Returns:

Type	Description
dict[str, Any] | None	dict JSON-serializable with the 8 (or 9 with fee_tier) keys
dict[str, Any] | None	ratified by PRD §Registry Data Shape and the T08 golden, OR
dict[str, Any] | None	None when the LP_OPEN data isn't extractable from the
dict[str, Any] | None	receipt.

extract_registry_payload_close

extract_registry_payload_close(
    receipt: dict[str, Any],
    *,
    open_payload: dict[str, Any] | None = None,
    fee_tier: int | None = None,
) -> dict[str, Any] | None

Build the LP_CLOSE position_registry.payload dict.

Reads the existing :meth:extract_lp_close_data output (Burn / Collect amounts) and the close-side DecreaseLiquidity event for the NFT token_id, then composes the 13-key shape that the T08 lp_close/expected_registry_row.json golden specifies for Uniswap V3 (PancakeSwap V3 is a direct UV3 fork at the NPM contract level, so the shape is identical).

Audit M1 (CodeRabbit): a real LP_CLOSE proves itself with DecreaseLiquidity on the receipt AND a Burn log carrying the pool address. A Collect-only receipt is NOT a close — it's a fee harvest. If we silently synthesized token_id / pool_address from open_payload here, a Collect-only receipt or a malformed close would produce a "successful" close payload with stale OPEN-side anchors, and the registry would mark a still-open NFT as closed (the cutover spec D3.F6 silent-error class).

The flow is:

1. Decode close-side events (extract_lp_close_data) and the DecreaseLiquidity log (_decreaseliquidity_token_id).
2. Verify the receipt-derived identity anchors are present and non-zero.
3. Cross-check against open_payload if supplied — refuse on any disagreement (v3_registry_payload.open_payload_disagrees).
4. Compose the receipt-only payload (v3_registry_payload.build_close_receipt_payload).
5. Merge OPEN-time fields the close receipt cannot re-derive (v3_registry_payload.merge_open_payload_fields) — ticks, OPEN-time amounts, original mint liquidity, fee tier, token labels.
6. Apply the fee_tier argument if open_payload didn't carry one.

Returns None when the close-side identity anchors (token_id + pool_address) cannot be derived OR cross-checks fail. The caller treats that as "fall back to accounting_only" with an INFO log (no zero substitution).

extract_protocol_fees

extract_protocol_fees(
    receipt: dict[str, Any],
    *,
    fee_tier_bps: int | None = None,
) -> ProtocolFees | None

Extract DEX protocol fees from a PancakeSwap V3 swap receipt.

PancakeSwap V3 is a Uniswap V3 fork with the same pip-based fee tiers. The fee tier is resolved at compile time and forwarded by the ResultEnricher via bundle_metadata["selected_fee_tier"] (signature-introspection opt-in — see VIB-3203 / VIB-3204).

VIB-3204 audit fix (Codex P1, pr-auditor Blocker #2): do NOT return ProtocolFees(total_usd=Decimal(0)) when the fee is known-non-zero but USD conversion isn't available. That would systematically under-attribute swap costs — a silent accounting bug. Until a price oracle is plumbed through to this layer, this parser returns None (ExtractMissing semantic). Callers that want the in-token fee can derive it from amount_in_decimal * fee_tier_bps / 1_000_000 using values already on result.swap_amounts.

Parameters:

Name	Type	Description	Default
receipt	dict[str, Any]	Transaction receipt dict with 'logs' field.	required
fee_tier_bps	int | None	Pool fee tier in pips (e.g. 500 = 5 bps), forwarded by the enricher.	None

Returns:

Type	Description
ProtocolFees | None	None — USD conversion not available at this layer. A
ProtocolFees | None	future iteration with price-oracle access will return a
ProtocolFees | None	populated ProtocolFees. Also returns None when no
ProtocolFees | None	Swap event is present OR when fee_tier_bps is missing.

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.

__getattr__

__getattr__(name: str) -> Any

PEP 562 lazy attribute access.