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:

Decode close-side events (extract_lp_close_data) and the DecreaseLiquidity log (_decreaseliquidity_token_id).
Verify the receipt-derived identity anchors are present and non-zero.
Cross-check against open_payload if supplied — refuse on any disagreement (v3_registry_payload.open_payload_disagrees).
Compose the receipt-only payload (v3_registry_payload.build_close_receipt_payload).
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.
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.

PancakeSwap V3¶

almanak.connectors.pancakeswap_v3 ¶

Swap exact input¶

PancakeSwapV3Adapter ¶

Swap 100 USDT for WBNB¶

swap_exact_input ¶

swap_exact_output ¶

PancakeSwapV3Config dataclass ¶

__post_init__ ¶

TransactionResult dataclass ¶

PancakeSwapV3EventType ¶

PancakeSwapV3ReceiptParser ¶

build_extract_kwargs ¶

extract_swap_amounts ¶

extract_position_id ¶

extract_tick_lower ¶

extract_tick_upper ¶

extract_liquidity ¶

extract_lp_close_data ¶

extract_lp_open_data ¶

extract_lp_open_data_result ¶

extract_registry_payload_open ¶

extract_registry_payload_close ¶

extract_protocol_fees ¶

parse_swap ¶

is_pancakeswap_event ¶

get_event_type ¶

ParseResult dataclass ¶

to_dict ¶

SwapEventData dataclass ¶

token0_in property ¶

token1_in property ¶

to_dict ¶

__getattr__ ¶

PancakeSwapV3Config `dataclass` ¶

TransactionResult `dataclass` ¶

ParseResult `dataclass` ¶

SwapEventData `dataclass` ¶

token0_in `property` ¶

token1_in `property` ¶

getattr ¶