Fluid DEX¶

Connector for Fluid DEX protocol.

almanak.framework.connectors.fluid ¶

Fluid DEX Connector — Phase 1 (Arbitrum swaps + LP scaffolding).

Provides swap support for Fluid DEX T1 pools on Arbitrum via swapIn(). LP open/close intent routing is wired but LP deposit reverts on-chain (phase 2).

Scope (phase 1): - Arbitrum only - Swaps via swapIn() (fully functional) - LP deposit deferred (Liquidity-layer routing causes reverts)

Key contracts (Arbitrum): - DexFactory: 0x91716C4EDA1Fb55e84Bf8b4c7085f84285c19085 - DexResolver: 0x11D80CfF056Cef4F9E6d23da8672fE9873e5cC07

Example

from almanak.framework.connectors.fluid import FluidAdapter, FluidConfig

config = FluidConfig( chain="arbitrum", wallet_address="0x...", rpc_url="https://arb-mainnet.g.alchemy.com/v2/...", ) adapter = FluidAdapter(config)

FluidAdapter ¶

FluidAdapter(
    config: FluidConfig,
    token_resolver: TokenResolver | None = None,
)

High-level adapter for Fluid DEX LP operations.

Provides LP open/close with compile-time encumbrance guard. Phase 1 operates only on unencumbered pools (no smart-debt/collateral).

参数：

名称	类型	描述	默认
`config`	`FluidConfig`	FluidConfig with chain, wallet, and RPC settings	必需
`token_resolver`	`TokenResolver \| None`	Optional TokenResolver for symbol -> address resolution	`None`

resolve_token_address ¶

resolve_token_address(token: str) -> str

Resolve a token symbol or address to checksummed address.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol (e.g., "USDC") or address	必需

返回：

类型	描述
`str`	Checksummed address

get_token_decimals ¶

get_token_decimals(token: str) -> int

Get decimals for a token.

参数：

名称	类型	描述	默认
`token`	`str`	Token symbol or address	必需

返回：

类型	描述
`int`	Token decimals (never defaults to 18 — raises if unknown)

find_pool ¶

find_pool(token0: str, token1: str) -> str | None

Find a Fluid DEX pool for a token pair.

参数：

名称	类型	描述	默认
`token0`	`str`	First token symbol or address	必需
`token1`	`str`	Second token symbol or address	必需

返回：

类型	描述
`str \| None`	Pool address if found, None otherwise

get_pool_data ¶

get_pool_data(dex_address: str) -> DexPoolData

Get full pool data from the resolver.

参数：

名称	类型	描述	默认
`dex_address`	`str`	Pool contract address	必需

返回：

类型	描述
`DexPoolData`	DexPoolData with configs, reserves, and encumbrance flags

build_add_liquidity_transaction ¶

build_add_liquidity_transaction(
    dex_address: str,
    amount0: Decimal,
    amount1: Decimal,
    token0_decimals: int,
    token1_decimals: int,
) -> TransactionData

Build a transaction to open a new LP position.

Phase 1 limitation: Fluid DEX deposit() reverts on all pools due to complex Liquidity-layer routing. LP open is wired but not functional on-chain. Proper share calculation requires Liquidity-layer integration (follow-up).

引发：

类型	描述
`FluidSDKError`	Always — LP deposit is not yet supported.

build_remove_liquidity_transaction ¶

build_remove_liquidity_transaction(
    dex_address: str, nft_id: int
) -> TransactionData

Build a transaction to close an LP position (full withdrawal).

Calls operate(nftId, -MAX_INT256, 0, wallet) on the pool contract. The negative max collateral delta means "withdraw everything".

ENCUMBRANCE GUARD: This method refuses to build the transaction if the pool has smart-collateral or smart-debt enabled.

参数：

名称	类型	描述	默认
`dex_address`	`str`	Pool contract address	必需
`nft_id`	`int`	NFT position ID to close	必需

返回：

类型	描述
`TransactionData`	TransactionData with the operate() call

引发：

类型	描述
`FluidSDKError`	If the operation fails

get_position_details ¶

get_position_details(
    nft_id: int, dex_address: str
) -> FluidPositionDetails

Build FluidPositionDetails for a position.

Reads pool data from the resolver to populate APR fields.

参数：

名称	类型	描述	默认
`nft_id`	`int`	NFT position ID	必需
`dex_address`	`str`	Pool contract address	必需

返回：

类型	描述
`FluidPositionDetails`	FluidPositionDetails with pool data and APR info

build_approve_tx ¶

build_approve_tx(
    token_address: str,
    spender: str,
    amount: int | None = None,
) -> TransactionData

Build an ERC20 approval transaction.

参数：

名称	类型	描述	默认
`token_address`	`str`	Token contract address	必需
`spender`	`str`	Address to approve spending for	必需
`amount`	`int \| None`	Amount to approve (None = max uint256)	`None`

返回：

类型	描述
`TransactionData`	TransactionData for the approval

extract_position_id ¶

extract_position_id(receipt: dict) -> int | None

Extract LP position NFT tokenId from operate() receipt.

Called by ResultEnricher for LP_OPEN intents.

参数：

名称	类型	描述	默认
`receipt`	`dict`	Transaction receipt dict	必需

返回：

类型	描述
`int \| None`	NFT position ID or None if not found

FluidConfig `dataclass` ¶

FluidConfig(
    chain: str,
    wallet_address: str,
    rpc_url: str,
    default_slippage_bps: int = 50,
)

Configuration for Fluid DEX adapter.

参数：

名称	类型	描述	默认
`chain`	`str`	Chain name (must be "arbitrum" for phase 1)	必需
`wallet_address`	`str`	Address of the wallet executing transactions	必需
`rpc_url`	`str`	RPC endpoint URL	必需
`default_slippage_bps`	`int`	Default slippage tolerance in basis points (default: 50 = 0.5%)	`50`

FluidPositionDetails `dataclass` ¶

FluidPositionDetails(
    fluid_nft_id: str,
    dex_address: str,
    token0: str,
    token1: str,
    swap_fee_apr: float = 0.0,
    lending_yield_apr: float = 0.0,
    combined_apr: float = 0.0,
    is_smart_collateral: bool = False,
    is_smart_debt: bool = False,
)

Typed position details for Fluid DEX LP positions.

Stored in PositionInfo.details as a dict (via asdict()). String NFT ID is consistent with Uniswap V3 / TraderJoe patterns.

属性：

名称	类型	描述
`fluid_nft_id`	`str`	NFT token ID (string for consistency with other connectors)
`dex_address`	`str`	Pool contract address
`token0`	`str`	Token0 address
`token1`	`str`	Token1 address
`swap_fee_apr`	`float`	Swap fee APR (from exchange price data)
`lending_yield_apr`	`float`	Lending yield APR (from liquidity resolver)
`combined_apr`	`float`	Combined APR (swap_fee_apr + lending_yield_apr)
`is_smart_collateral`	`bool`	Whether pool has smart collateral enabled
`is_smart_debt`	`bool`	Whether pool has smart debt enabled

FluidReceiptParser ¶

FluidReceiptParser(chain: str = 'arbitrum')

Receipt parser for Fluid DEX transactions (swaps and LP operations).

Extracts NFT position IDs and token amounts from LogOperate events. Supports both LP_OPEN (new position) and LP_CLOSE (withdrawal) receipts.

SUPPORTED_EXTRACTIONS declares which ResultEnricher fields this parser supports.

参数：

名称	类型	描述	默认
`chain`	`str`	Chain name for token resolution (default: "arbitrum")	`'arbitrum'`

parse_receipt ¶

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

Parse a Fluid DEX transaction receipt.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict with 'logs', 'transactionHash', etc.	必需

返回：

类型	描述
`FluidParseResult`	FluidParseResult with extracted events and data

extract_position_id ¶

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

Extract LP position NFT tokenId from receipt.

Called by ResultEnricher for LP_OPEN intents.

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict	必需

返回：

类型	描述
`int \| None`	NFT position ID or None if not found

extract_swap_amounts ¶

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

Extract swap amounts from receipt.

Called by ResultEnricher for SWAP intents. Resolves token decimals to produce human-readable decimal amounts (consistent with other parsers).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict	必需

返回：

类型	描述
`SwapAmounts \| None`	SwapAmounts with input/output amounts, or None

extract_lp_close_data ¶

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

Extract LP close data from receipt.

Called by ResultEnricher for LP_CLOSE intents. Returns amounts collected on withdrawal (negative amounts = withdrawn).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict	必需

返回：

类型	描述
`LPCloseData \| None`	LPCloseData with collected amounts, or None if not found

extract_liquidity ¶

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

Extract liquidity amount from LP_OPEN receipt.

For Fluid DEX, liquidity is represented by the collateral shares (token0_amt + token1_amt deposited).

参数：

名称	类型	描述	默认
`receipt`	`dict[str, Any]`	Transaction receipt dict	必需

返回：

类型	描述
`int \| None`	Combined deposit amount or None

FluidSDK ¶

FluidSDK(chain: str, rpc_url: str)

Low-level Fluid DEX protocol SDK.

Reads pool data directly from pool contracts using raw eth_call: - constantsView(): token addresses (words 9, 10 of 18-word response) - readFromStorage(bytes32(0)): dexVariables with smart-collateral/debt flags

参数：

名称	类型	描述	默认
`chain`	`str`	Chain name (must be "arbitrum" for phase 1)	必需
`rpc_url`	`str`	RPC endpoint URL	必需

get_all_dex_addresses ¶

get_all_dex_addresses() -> list[str]

Get all Fluid DEX pool addresses from the resolver.

get_total_dexes ¶

get_total_dexes() -> int

Get the total number of Fluid DEX pools.

get_dex_data ¶

get_dex_data(dex_address: str) -> DexPoolData

Get pool data by calling constantsView() and readFromStorage() directly.

Uses raw eth_call to avoid ABI decoding issues with the complex DexEntireData struct. constantsView() returns 18 words: - word[9]: token0 address - word[10]: token1 address

readFromStorage(bytes32(0)) returns dexVariables: - bit 1: isSmartCollateralEnabled - bit 2: isSmartDebtEnabled

参数：

名称	类型	描述	默认
`dex_address`	`str`	Pool contract address	必需

返回：

类型	描述
`DexPoolData`	DexPoolData with token addresses and encumbrance flags

is_position_encumbered ¶

is_position_encumbered(
    dex_address: str, nft_id: int = 0
) -> bool

Check if a specific position has outstanding debt.

In Fluid DEX, ALL pools have smart-debt capability (that's the design). The encumbrance check is at the POSITION level, not pool level: - Positions we create with newDebt=0 have no debt, so they're safe to close. - For safety, we verify the pool's smart-debt flag but don't block on it for positions we know were opened without debt.

For phase 1, this always returns False for nft_id=0 (new position check) since we enforce newDebt=0 in build_operate_tx().

参数：

名称	类型	描述	默认
`dex_address`	`str`	Pool contract address	必需
`nft_id`	`int`	NFT position ID (0 = checking for new position)	`0`

返回：

类型	描述
`bool`	True if the position has outstanding debt

find_dex_by_tokens ¶

find_dex_by_tokens(token0: str, token1: str) -> str | None

Find a Fluid DEX pool for a given token pair.

Token order is automatically handled (tries both orderings).

get_swap_quote ¶

get_swap_quote(
    dex_address: str,
    swap0to1: bool,
    amount_in: int,
    to: str,
) -> int

Get a swap quote (estimate) from a Fluid DEX pool.

Calls swapIn via eth_call with state overrides to simulate token approval and balance. Without overrides, swapIn() reverts because it internally calls transferFrom() which requires prior approval.

参数：

名称	类型	描述	默认
`dex_address`	`str`	Pool contract address	必需
`swap0to1`	`bool`	True to swap token0->token1, False for token1->token0	必需
`amount_in`	`int`	Input amount in token's smallest unit	必需
`to`	`str`	Recipient address	必需

返回：

类型	描述
`int`	Expected output amount in token's smallest unit

build_swap_tx ¶

build_swap_tx(
    dex_address: str,
    swap0to1: bool,
    amount_in: int,
    amount_out_min: int,
    to: str,
    value: int = 0,
) -> dict[str, Any]

Build a swapIn transaction for a Fluid DEX pool.

参数：

名称	类型	描述	默认
`dex_address`	`str`	Pool contract address	必需
`swap0to1`	`bool`	True to swap token0->token1, False for token1->token0	必需
`amount_in`	`int`	Input amount in token's smallest unit	必需
`amount_out_min`	`int`	Minimum acceptable output (slippage protection)	必需
`to`	`str`	Recipient address	必需
`value`	`int`	Native token value (for ETH-paired swaps)	`0`

返回：

类型	描述
`dict[str, Any]`	Transaction dict with 'to', 'data', 'value', 'gas'

build_operate_tx ¶

build_operate_tx(
    dex_address: str,
    nft_id: int,
    new_col: int,
    new_debt: int,
    to: str,
) -> dict[str, Any]

Build an operate() transaction for a Fluid DEX pool.

operate() is the main entry point for LP operations: - Open position: nft_id=0, new_col>0, new_debt=0 - Close position: nft_id=X, new_col<0 (negative = withdraw), new_debt=0

Fluid DEX¶

almanak.framework.connectors.fluid ¶

FluidAdapter ¶

resolve_token_address ¶

get_token_decimals ¶

find_pool ¶

get_pool_data ¶

build_add_liquidity_transaction ¶

build_remove_liquidity_transaction ¶

get_position_details ¶

build_approve_tx ¶

extract_position_id ¶

FluidConfig dataclass ¶

FluidPositionDetails dataclass ¶

FluidReceiptParser ¶

parse_receipt ¶

extract_position_id ¶

extract_swap_amounts ¶

extract_lp_close_data ¶

extract_liquidity ¶

FluidSDK ¶

get_all_dex_addresses ¶

get_total_dexes ¶

get_dex_data ¶

is_position_encumbered ¶

find_dex_by_tokens ¶

get_swap_quote ¶

build_swap_tx ¶

build_operate_tx ¶

FluidConfig `dataclass` ¶

FluidPositionDetails `dataclass` ¶