Python SDK#

The Coalesce Finance Python SDK for building applications and scripts that interact with the permissioned lending protocol.

Installation#

pip install coalescefi-sdk

Or with development dependencies:

pip install coalescefi-sdk[dev]

Requirements#

  • Python 3.10+
  • solana-py >= 0.35.0
  • solders >= 0.21.0

Quick Start — CoalesceClient#

The high-level CoalesceClient handles PDA derivation, ATA resolution, and account fetching automatically. You only need to provide signers, market addresses, and amounts:

Python (19 lines)
from coalescefi_sdk import CoalesceClient

client = CoalesceClient.mainnet("https://api.mainnet-beta.solana.com")

lender = Pubkey.from_string("Lender...")
market = Pubkey.from_string("Market...")

# Build a deposit instruction (resolves vault, ATA, blacklist check, etc.)
ixs = await client.deposit(lender, market, 100_000_000_000)  # 100,000 USDC (6 decimals)

# Withdraw all shares and close the position in one transaction
ixs = await client.withdraw_and_close(lender, market)

# Claim haircut recovery after a settlement shortfall
ixs = await client.claim_haircut(lender, market)

# Claim haircut and close the position in one transaction
ixs = await client.claim_haircut_and_close(lender, market)

Other network constructors: CoalesceClient.devnet(url), CoalesceClient.localnet(url), or CoalesceClient(connection, program_id) for a custom program ID.

Quick Start — Low-Level#

Python (32 lines)
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey

from coalescefi_sdk import (
    configure_sdk,
    find_market_pda,
    find_lender_position_pda,
    fetch_market,
    create_deposit_instruction,
)

# Configure the SDK
configure_sdk(network="devnet")

# Or with explicit program ID
configure_sdk(program_id=Pubkey.from_string("CoaLfi..."))

# Derive PDAs
borrower = Pubkey.from_string("Borrower...")
market_pda, market_bump = find_market_pda(borrower, market_nonce=1)
lender_pda, lender_bump = find_lender_position_pda(market_pda, lender_pubkey)

# Fetch account data
async def main():
    connection = AsyncClient("https://api.devnet.solana.com")
    market = await fetch_market(connection, market_pda)
    if market:
        print(f"Total deposited: {market.total_deposited}")
        print(f"Interest rate: {market.annual_interest_bps} bps")

asyncio.run(main())

Configuration#

Network-based Configuration#

from coalescefi_sdk import configure_sdk, NetworkName

# Use predefined network
configure_sdk(network="mainnet")  # or "devnet", "localnet"

Explicit Program ID#

from solders.pubkey import Pubkey
from coalescefi_sdk import configure_sdk

configure_sdk(
    program_id=Pubkey.from_string("CoaLfi...")
)

Environment Variables#

The SDK also reads from environment variables:

export COALESCEFI_PROGRAM_ID="CoaLfi..."
export COALESCEFI_NETWORK="devnet"

PDA Derivation#

All PDA functions return a tuple of (Pubkey, bump):

Python (35 lines)
from coalescefi_sdk import (
    find_protocol_config_pda,
    find_market_pda,
    find_market_authority_pda,
    find_vault_pda,
    find_lender_position_pda,
    find_borrower_whitelist_pda,
    find_blacklist_check_pda,
    find_haircut_state_pda,
    derive_market_pdas,
)

# Protocol config (singleton)
config_pda, config_bump = find_protocol_config_pda()

# Market PDAs
market_pda, market_bump = find_market_pda(borrower, market_nonce=1)
authority_pda, authority_bump = find_market_authority_pda(market_pda)
vault_pda, vault_bump = find_vault_pda(market_pda)

# Lender position
lender_pda, lender_bump = find_lender_position_pda(market_pda, lender)

# Borrower whitelist
whitelist_pda, whitelist_bump = find_borrower_whitelist_pda(borrower)

# Haircut state (per-market, tracks recovery distribution)
haircut_state_pda, haircut_bump = find_haircut_state_pda(market_pda)

# Derive all market-related PDAs at once
market_pdas = derive_market_pdas(borrower, market_nonce=1)
print(market_pdas.market)
print(market_pdas.market_authority)
print(market_pdas.vault)

Account Fetching#

The SDK provides async fetchers with built-in retry logic:

Python (32 lines)
from solana.rpc.async_api import AsyncClient
from coalescefi_sdk import (
    fetch_protocol_config,
    fetch_market,
    fetch_lender_position,
    fetch_borrower_whitelist,
    RetryConfig,
)

async def fetch_accounts():
    connection = AsyncClient("https://api.devnet.solana.com")

    # With default retry config (3 retries, exponential backoff)
    market = await fetch_market(connection, market_pda)

    # With custom retry config
    custom_retry = RetryConfig(
        max_retries=5,
        base_delay_ms=500,
        max_delay_ms=5000
    )
    config = await fetch_protocol_config(
        connection,
        config_pda,
        retry_config=custom_retry
    )

    # Returns None if account doesn't exist
    position = await fetch_lender_position(connection, lender_pda)
    if position is None:
        print("Position not found")

Account Decoding#

For manual decoding of raw account data:

Python (19 lines)
from coalescefi_sdk import (
    decode_protocol_config,
    decode_market,
    decode_lender_position,
    decode_borrower_whitelist,
    decode_account,  # Auto-detects type
)

# Decode specific account type
market = decode_market(account_data)
print(f"Borrower: {market.borrower}")
print(f"Annual interest: {market.annual_interest_bps} bps")
print(f"Maturity: {market.maturity_timestamp}")

# Auto-detect and decode
account = decode_account(raw_data)
if isinstance(account, Market):
    print("It's a market!")

Instruction Building#

Build instructions for all protocol operations:

Python (24 lines)
from coalescefi_sdk import (
    create_deposit_instruction,
    create_borrow_instruction,
    create_repay_instruction,
    create_withdraw_instruction,
)

# Deposit instruction
deposit_ix = create_deposit_instruction(
    accounts={
        "market": market_pda,
        "lender": lender_pubkey,
        "lender_token_account": lender_ata,
        "vault": vault_pda,
        "lender_position": lender_position_pda,
        "blacklist_check": blacklist_check_pda,
        "protocol_config": config_pda,
        "mint": usdc_mint,
        "token_program": TOKEN_PROGRAM_ID,
        "system_program": SYSTEM_PROGRAM_ID,
    },
    args={"amount": 1_000_000},  # 1 USDC (6 decimals)
)

Available Instructions#

FunctionDescription
create_initialize_protocol_instructionInitialize protocol config
create_set_fee_config_instructionUpdate fee settings
create_create_market_instructionCreate a new lending market
create_deposit_instructionDeposit tokens to a market
create_borrow_instructionBorrow from a market
create_repay_instructionRepay borrowed amount
create_repay_interest_instructionRepay accrued interest
create_withdraw_instructionWithdraw deposited tokens
create_collect_fees_instructionCollect protocol fees
create_re_settle_instructionRe-settle after maturity
create_close_lender_position_instructionClose empty position
create_withdraw_excess_instructionWithdraw excess funds
create_set_borrower_whitelist_instructionManage whitelist
create_set_pause_instructionPause/unpause protocol
create_set_blacklist_mode_instructionConfigure blacklist
create_set_admin_instructionTransfer admin role
create_set_whitelist_manager_instructionSet whitelist manager
create_force_close_position_instructionBorrower force-closes a lender position after grace period
create_claim_haircut_instructionLender claims haircut recovery tokens
create_force_claim_haircut_instructionBorrower force-claims haircut recovery on behalf of a lender

Combined Builders (CoalesceClient)#

The CoalesceClient also provides combined async builders that bundle multiple instructions into a single transaction:

MethodDescription
withdraw_and_closeFull withdrawal + close lender position
claim_haircut_and_closeClaim haircut recovery + close lender position

Error Handling#

Python (21 lines)
from coalescefi_sdk import (
    CoalescefiError,
    CoalescefiErrorCode,
    parse_coalescefi_error,
    is_user_recoverable_error,
    get_error_recovery_action,
)

try:
    # ... transaction execution
    pass
except Exception as e:
    error = parse_coalescefi_error(e)
    if error:
        print(f"Error code: {error.code.name}")
        print(f"Message: {error.message}")

        if is_user_recoverable_error(error.code):
            action = get_error_recovery_action(error.code)
            print(f"Recovery action: {action}")

Error Categories#

from coalescefi_sdk import (
    get_error_category,
    get_error_severity,
    ErrorCategory,
    ErrorSeverity,
)

category = get_error_category(CoalescefiErrorCode.InsufficientBalance)
# ErrorCategory.Balance

severity = get_error_severity(CoalescefiErrorCode.Unauthorized)
# ErrorSeverity.Error

Idempotency Support#

Prevent duplicate transactions with the idempotency manager:

Python (23 lines)
from coalescefi_sdk import (
    IdempotencyManager,
    generate_idempotency_key,
)

manager = IdempotencyManager()

# Generate deterministic key for operation
key = generate_idempotency_key("deposit", {
    "market": str(market_pda),
    "amount": str(amount),
    "lender": str(lender),
})

# Execute with idempotency protection
async def safe_deposit():
    result = await manager.execute_once(
        connection,
        key,
        lambda: execute_deposit_transaction(),
    )
    return result

HaircutState Account#

The HaircutState account (88 bytes) tracks per-market haircut recovery distribution state. It is created alongside the market and updated during settlement and claims.

from coalescefi_sdk import decode_haircut_state, HaircutState

haircut_state: HaircutState = decode_haircut_state(account_data)

print(f"Market: {haircut_state.market}")
print(f"Claim weight sum: {haircut_state.claim_weight_sum}")
print(f"Claim offset sum: {haircut_state.claim_offset_sum}")
FieldTypeOffsetDescription
discriminatorbytes0Account discriminator ("COALHCST")
versionint8Schema version
marketPubkey9Market this state belongs to
claim_weight_sumint41Sum of claim weights (u128 LE)
claim_offset_sumint57Sum of claim offsets (u128 LE)
bumpint73PDA bump

Type Annotations#

The SDK is fully typed for IDE support:

from coalescefi_sdk import (
    ProtocolConfig,
    Market,
    LenderPosition,
    BorrowerWhitelist,
    HaircutState,
)

def process_market(market: Market) -> None:
    # Full type hints available
    borrower: Pubkey = market.borrower
    rate: int = market.annual_interest_bps
    maturity: int = market.maturity_timestamp

Math Reference#

The math module provides WAD-precision (10^18) fixed-point arithmetic matching the on-chain Rust implementation exactly.

Python (37 lines)
from coalescefi_sdk import (
    # Core WAD arithmetic
    mul_wad,              # (a * b) / WAD
    pow_wad,              # base^exp in WAD
    growth_factor_wad,    # Daily compound + linear sub-day growth

    # Deposit/Withdraw conversions
    calculate_scaled_amount,      # tokens → sTokens (shares)
    calculate_normalized_amount,  # sTokens → tokens

    # Position analysis
    calculate_position_value,     # Current value of lender shares
    estimate_value_at_maturity,   # Projected value at maturity
    calculate_settlement_payout,  # Actual payout after settlement

    # Market analytics
    estimate_interest_accrual,    # Preview interest growth
    calculate_total_supply,       # Normalized total supply
    calculate_available_vault_balance,  # Deprecated approximation; prefer vault RPC balance
    calculate_utilization_rate,   # Outstanding principal / supply (bps + decimal)
    calculate_apr,                # BPS to decimal (e.g., 1000 → 0.10)

    # Utilities
    safe_divide,          # Returns 0 on division by zero
    would_overflow_u64,
    would_overflow_u128,

    # Constants
    SECONDS_PER_DAY,      # 86,400
    DAYS_PER_YEAR,        # 365

    # Types
    InterestAccrualResult,
    UtilizationRateResult,
    MathOverflowError,
)

Deposit Preview#

from coalescefi_sdk import calculate_scaled_amount, WAD

# How many shares does depositing 1 USDC yield?
scale_factor = WAD * 11 // 10  # 1.10 (10% interest accrued)
shares = calculate_scaled_amount(1_000_000, scale_factor)
# shares < 1_000_000 (fewer shares because each is worth more)

Interest Accrual Preview#

import time
from coalescefi_sdk import estimate_interest_accrual, fetch_market

market = await fetch_market(connection, market_pda)
now = int(time.time())

result = estimate_interest_accrual(market, now)
print(f"Estimated scale factor: {result.new_scale_factor}")
print(f"Interest delta: {result.interest_delta}")

Market Analytics#

For exact available liquidity, read the market vault SPL Token account balance via RPC. calculate_available_vault_balance is a deprecated counter-based approximation and can diverge after fee collection, withdraw_excess, force_close_position, or withdrawal payouts.

Python (21 lines)
from coalescefi_sdk import (
    calculate_apr,
    calculate_total_supply,
    calculate_utilization_rate,
    fetch_market,
)

market = await fetch_market(connection, market_pda)

apr = calculate_apr(market.annual_interest_bps)
total_supply = calculate_total_supply(market.scaled_total_supply, market.scale_factor)
utilization = calculate_utilization_rate(market)

print(f"APR: {apr * 100:.2f}%")
print(f"Total supply: {total_supply}")
print(f"Utilization: {utilization.decimal * 100:.2f}% ({utilization.bps} bps)")

# Exact available vault balance: read the SPL Token account balance for market.vault via RPC.
# Deprecated approximation if you only have market counters:
# approx_available = calculate_available_vault_balance(market)

Settlement Payout#

Python (16 lines)
from coalescefi_sdk import (
    calculate_settlement_payout,
    calculate_normalized_amount,
)

# After maturity — what does a lender actually receive?
# Two-step: normalize by scale_factor, then apply settlement_factor
payout = calculate_settlement_payout(
    position.scaled_balance, market.scale_factor, market.settlement_factor_wad
)
full_value = calculate_normalized_amount(
    position.scaled_balance, market.scale_factor
)
haircut_bps = (full_value - payout) * 10_000 // full_value if full_value > 0 else 0
print(f"Payout: {payout}, haircut: {haircut_bps / 100:.2f}%")

Constants#

Python (25 lines)
from coalescefi_sdk import (
    # Mathematical constants
    WAD,  # 10^18 - precision factor
    BPS,  # 10000 - basis points denominator
    SECONDS_PER_YEAR,

    # Protocol limits
    MAX_ANNUAL_INTEREST_BPS,  # 10000 (100%)
    MAX_FEE_RATE_BPS,  # 10000 (100%)
    USDC_DECIMALS,  # 6

    # Account sizes
    PROTOCOL_CONFIG_SIZE,
    MARKET_SIZE,
    LENDER_POSITION_SIZE,
    BORROWER_WHITELIST_SIZE,
    HAIRCUT_STATE_SIZE,      # 88 bytes

    # Seeds
    SEED_PROTOCOL_CONFIG,
    SEED_MARKET,
    SEED_VAULT,
    SEED_HAIRCUT_STATE,
)

Testing#

import pytest
from coalescefi_sdk import configure_sdk, reset_sdk_config

@pytest.fixture(autouse=True)
def reset_config():
    """Reset SDK config between tests."""
    yield
    reset_sdk_config()

def test_pda_derivation():
    configure_sdk(network="localnet")
    pda, bump = find_market_pda(borrower, 1)
    assert bump > 0

Next Steps#