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.0solders>= 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#
| Function | Description |
|---|---|
create_initialize_protocol_instruction | Initialize protocol config |
create_set_fee_config_instruction | Update fee settings |
create_create_market_instruction | Create a new lending market |
create_deposit_instruction | Deposit tokens to a market |
create_borrow_instruction | Borrow from a market |
create_repay_instruction | Repay borrowed amount |
create_repay_interest_instruction | Repay accrued interest |
create_withdraw_instruction | Withdraw deposited tokens |
create_collect_fees_instruction | Collect protocol fees |
create_re_settle_instruction | Re-settle after maturity |
create_close_lender_position_instruction | Close empty position |
create_withdraw_excess_instruction | Withdraw excess funds |
create_set_borrower_whitelist_instruction | Manage whitelist |
create_set_pause_instruction | Pause/unpause protocol |
create_set_blacklist_mode_instruction | Configure blacklist |
create_set_admin_instruction | Transfer admin role |
create_set_whitelist_manager_instruction | Set whitelist manager |
create_force_close_position_instruction | Borrower force-closes a lender position after grace period |
create_claim_haircut_instruction | Lender claims haircut recovery tokens |
create_force_claim_haircut_instruction | Borrower 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:
| Method | Description |
|---|---|
withdraw_and_close | Full withdrawal + close lender position |
claim_haircut_and_close | Claim 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}")
| Field | Type | Offset | Description |
|---|---|---|---|
discriminator | bytes | 0 | Account discriminator ("COALHCST") |
version | int | 8 | Schema version |
market | Pubkey | 9 | Market this state belongs to |
claim_weight_sum | int | 41 | Sum of claim weights (u128 LE) |
claim_offset_sum | int | 57 | Sum of claim offsets (u128 LE) |
bump | int | 73 | PDA 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#
- TypeScript SDK - If you need browser support
- Rust SDK - For on-chain programs
- SDK Overview - Protocol reference and shared-core
- Error Codes - Error codes reference