Aller au contenu

Orca

Connector for Orca protocol.

almanak.framework.connectors.orca

Orca Whirlpools concentrated liquidity connector.

Provides LP operations on Orca Whirlpool pools on Solana: - Open concentrated liquidity positions - Close positions (decrease liquidity + burn NFT)

Uses the same Q64.64 tick math as Raydium CLMM (reused from connectors/raydium/math.py) and Anchor-style instruction encoding.

Reference: https://github.com/orca-so/whirlpools

OrcaAdapter

OrcaAdapter(
    config: OrcaConfig,
    token_resolver: TokenResolver | None = None,
)

Adapter for Orca Whirlpools integration with the Intent system.

Converts LP intents to ActionBundles containing serialized Solana VersionedTransactions built from Orca Whirlpool instructions.

compile_lp_open_intent

compile_lp_open_intent(
    intent: LPOpenIntent,
) -> ActionBundle

Compile an LPOpenIntent to an ActionBundle.

参数:

名称 类型 描述 默认
intent LPOpenIntent

The LPOpenIntent to compile.

必需

返回:

类型 描述
ActionBundle

ActionBundle containing serialized Solana transaction(s).

compile_lp_close_intent

compile_lp_close_intent(
    intent: LPCloseIntent,
) -> ActionBundle

Compile an LPCloseIntent to an ActionBundle.

参数:

名称 类型 描述 默认
intent LPCloseIntent

The LPCloseIntent to compile.

必需

返回:

类型 描述
ActionBundle

ActionBundle containing serialized Solana transaction(s).

OrcaConfig

OrcaConfig(wallet_address: str, rpc_url: str = '')

Configuration for Orca adapter.

属性:

名称 类型 描述
wallet_address

Solana wallet public key (Base58).

rpc_url

Solana RPC endpoint URL (for position queries).

OrcaAPIError

OrcaAPIError(
    message: str, status_code: int = 0, endpoint: str = ""
)

Bases: OrcaError

Error communicating with the Orca API.

OrcaConfigError

OrcaConfigError(message: str, parameter: str = '')

Bases: OrcaError

Invalid Orca configuration.

OrcaError

Bases: Exception

Base exception for Orca operations.

OrcaPoolError

Bases: OrcaError

Error with pool state or operations.

OrcaPool dataclass

OrcaPool(
    address: str,
    mint_a: str,
    mint_b: str,
    symbol_a: str = "",
    symbol_b: str = "",
    decimals_a: int = 9,
    decimals_b: int = 6,
    tick_spacing: int = 64,
    current_price: float = 0.0,
    tvl: float = 0.0,
    vault_a: str = "",
    vault_b: str = "",
    fee_rate: int = 3000,
    tick_current_index: int = 0,
    sqrt_price: str = "0",
    oracle_address: str = "",
    raw_response: dict[str, Any] = dict(),
)

Orca Whirlpool pool information.

Constructed from the Orca API response or on-chain data.

属性:

名称 类型 描述
address str

Whirlpool account address (Base58).

mint_a str

Token A mint address.

mint_b str

Token B mint address.

symbol_a str

Token A symbol (e.g., "SOL").

symbol_b str

Token B symbol (e.g., "USDC").

decimals_a int

Token A decimals.

decimals_b int

Token B decimals.

tick_spacing int

Tick spacing for this pool.

current_price float

Current price of token A in terms of token B.

tvl float

Total value locked in USD.

vault_a str

Token A vault address.

vault_b str

Token B vault address.

fee_rate int

Fee rate in basis points.

tick_current_index int

Current tick index.

sqrt_price str

Current sqrt price as string (u128).

oracle_address str

Oracle account address.

from_api_response classmethod

from_api_response(data: dict[str, Any]) -> OrcaPool

Create from Orca API /pools/{address} response.

OrcaPosition dataclass

OrcaPosition(
    nft_mint: str,
    pool_address: str,
    tick_lower: int,
    tick_upper: int,
    liquidity: int = 0,
    position_address: str = "",
)

Orca Whirlpool position (owned by the user).

属性:

名称 类型 描述
nft_mint str

Position NFT mint address.

pool_address str

Whirlpool account address.

tick_lower int

Lower tick boundary.

tick_upper int

Upper tick boundary.

liquidity int

Current liquidity in the position.

position_address str

Position PDA address.

OrcaTransactionBundle dataclass

OrcaTransactionBundle(
    transactions: list[str],
    action: str,
    position_nft_mint: str = "",
    metadata: dict[str, Any] = dict(),
)

Bundle of serialized transactions for an Orca operation.

属性:

名称 类型 描述
transactions list[str]

List of base64-encoded VersionedTransactions.

action str

Action type ("open_position", "close_position", etc.).

position_nft_mint str

NFT mint address (for open_position).

metadata dict[str, Any]

Additional metadata.

OrcaReceiptParser

OrcaReceiptParser(**kwargs: Any)

Parser for Orca Whirlpool transaction receipts.

Supports the extraction methods required by ResultEnricher: - extract_position_id(receipt) -> str | None - extract_liquidity(receipt) -> dict | None - extract_lp_close_data(receipt) -> dict | None

parse_receipt

parse_receipt(receipt: dict[str, Any]) -> dict[str, Any]

Parse a receipt for ReceiptParser protocol compatibility.

extract_position_id

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

Extract the position NFT mint from an Orca LP open receipt.

The position NFT is minted during open_position. We find it by looking for a new token account with amount=1 in postTokenBalances that wasn't in preTokenBalances.

extract_liquidity

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

Extract liquidity data from an LP open/increase receipt.

extract_lp_close_data

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

Extract LP close data from a receipt.

OrcaWhirlpoolSDK

OrcaWhirlpoolSDK(
    wallet_address: str,
    base_url: str = ORCA_API_BASE_URL,
    timeout: int = 30,
)

SDK for building Orca Whirlpool instructions.

Provides methods to: - Fetch pool data from the Orca API - Build Solana instructions for LP operations - Compute PDAs for positions, tick arrays, etc.

Example

sdk = OrcaWhirlpoolSDK(wallet_address="your-pubkey") pool = sdk.get_pool_info("pool-address") ixs, nft_mint = sdk.build_open_position_ix( pool=pool, tick_lower=-100, tick_upper=100, amount_a_max=1_000_000, amount_b_max=500_000_000, liquidity=1000000, )

get_pool_info

get_pool_info(pool_address: str) -> OrcaPool

Fetch pool information from the Orca API.

参数:

名称 类型 描述 默认
pool_address str

Whirlpool account address (Base58).

必需

返回:

类型 描述
OrcaPool

OrcaPool with current pool data.

引发:

类型 描述
OrcaPoolError

If pool not found.

OrcaAPIError

If API request fails.

find_pool_by_tokens

find_pool_by_tokens(
    token_a: str, token_b: str, tick_spacing: int = 64
) -> OrcaPool | None

Find a Whirlpool by token pair.

参数:

名称 类型 描述 默认
token_a str

Token A mint address.

必需
token_b str

Token B mint address.

必需
tick_spacing int

Preferred tick spacing (default: 64 = 0.30% fee).

64

返回:

类型 描述
OrcaPool | None

Best matching OrcaPool, or None if not found.

get_position_state

get_position_state(
    nft_mint: str, rpc_url: str
) -> OrcaPosition

Query on-chain Position account for an Orca Whirlpool position.

Position layout (Anchor, 8-byte discriminator): [0:8] discriminator [8:40] whirlpool (Pubkey) [40:72] position_mint (Pubkey) [72:88] liquidity (u128 LE) [88:92] tick_lower_index (i32 LE) [92:96] tick_upper_index (i32 LE)

参数:

名称 类型 描述 默认
nft_mint str

Position NFT mint address (Base58).

必需
rpc_url str

Solana RPC endpoint URL.

必需

返回:

类型 描述
OrcaPosition

OrcaPosition with on-chain tick range and liquidity.

引发:

类型 描述
OrcaPoolError

If position account not found or data invalid.

build_open_position_ix

build_open_position_ix(
    pool: OrcaPool,
    tick_lower: int,
    tick_upper: int,
    amount_a_max: int,
    amount_b_max: int,
    liquidity: int,
) -> tuple[list[Instruction], Keypair]

Build instructions for opening a new Whirlpool position.

Creates: open_position_with_metadata + increase_liquidity. Orca separates position creation (open_position) from liquidity deposit (increase_liquidity).

参数:

名称 类型 描述 默认
pool OrcaPool

Pool information.

必需
tick_lower int

Lower tick boundary (aligned to tick spacing).

必需
tick_upper int

Upper tick boundary (aligned to tick spacing).

必需
amount_a_max int

Maximum amount of token A in smallest units.

必需
amount_b_max int

Maximum amount of token B in smallest units.

必需
liquidity int

Target liquidity amount (u128).

必需

返回:

类型 描述
tuple[list[Instruction], Keypair]

Tuple of (instructions, nft_mint_keypair).

build_decrease_liquidity_ix

build_decrease_liquidity_ix(
    pool: OrcaPool,
    position: OrcaPosition,
    liquidity: int,
    amount_a_min: int = 0,
    amount_b_min: int = 0,
) -> list[Instruction]

Build instructions for removing liquidity from a position.

参数:

名称 类型 描述 默认
pool OrcaPool

Pool information.

必需
position OrcaPosition

Position to decrease.

必需
liquidity int

Amount of liquidity to remove.

必需
amount_a_min int

Minimum acceptable token A out.

0
amount_b_min int

Minimum acceptable token B out.

0

返回:

类型 描述
list[Instruction]

List of Solana instructions.

build_close_position_ix

build_close_position_ix(
    position: OrcaPosition,
) -> list[Instruction]

Build instructions for closing a position (burn NFT, recover rent).

参数:

名称 类型 描述 默认
position OrcaPosition

Position to close.

必需

返回:

类型 描述
list[Instruction]

List of Solana instructions.

build_open_position_transaction

build_open_position_transaction(
    pool: OrcaPool,
    price_lower: float,
    price_upper: float,
    amount_a: int,
    amount_b: int,
    slippage_bps: int = 100,
) -> tuple[list[Instruction], Keypair, dict[str, Any]]

Build a complete open position transaction from price bounds.

参数:

名称 类型 描述 默认
pool OrcaPool

Pool information.

必需
price_lower float

Lower price bound (token_b per token_a).

必需
price_upper float

Upper price bound.

必需
amount_a int

Amount of token A in smallest units.

必需
amount_b int

Amount of token B in smallest units.

必需
slippage_bps int

Slippage tolerance in basis points (default: 100 = 1%).

100

返回:

类型 描述
tuple[list[Instruction], Keypair, dict[str, Any]]

Tuple of (instructions, nft_mint_keypair, metadata).

build_close_position_transaction

build_close_position_transaction(
    pool: OrcaPool,
    position: OrcaPosition,
    slippage_bps: int = 100,
) -> tuple[list[Instruction], dict[str, Any]]

Build instructions to fully close a position.

Decreases all liquidity, then closes the position account.

参数:

名称 类型 描述 默认
pool OrcaPool

Pool information.

必需
position OrcaPosition

Position to close.

必需
slippage_bps int

Slippage tolerance for decrease liquidity.

100

返回:

类型 描述
tuple[list[Instruction], dict[str, Any]]

Tuple of (all_instructions, metadata).