solana-rpc

Solana RPC — 直接与区块链交互

Solana JSON-RPC API 提供对区块链的直接读写访问。可用于查询账户状态、查询代币余额、构建和提交交易，以及枚举程序账户。当更高级的 API（Birdeye、Helius、SolanaTracker）没有所需数据时，这是底层基础。

快速开始

import httpx

RPC = os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com")

def rpc_call(method: str, params: list = None) -> dict:
    resp = httpx.post(RPC, json={
        "jsonrpc": "2.0", "id": 1,
        "method": method, "params": params or [],
    }, timeout=30.0)
    return resp.json()

# Get SOL balance
result = rpc_call("getBalance", ["WALLET_PUBKEY"])
sol_balance = result["result"]["value"] / 1e9

# Get latest blockhash
result = rpc_call("getLatestBlockhash")
blockhash = result["result"]["value"]["blockhash"]

RPC 服务提供商

服务提供商	免费套餐	付费套餐	备注
Helius	每天 50K credits	$49+/mo	增强型 RPC、DAS API
QuickNode	有限	$49+/mo	多链、WebSocket
Triton	无免费套餐	~$300+/mo	捆绑 Yellowstone gRPC
Shyft	有限	$49+/mo	捆绑 Yellowstone gRPC
Alchemy	300M CU/mo	按量扩展	免费套餐不错
公共（mainnet-beta）	免费	—	有速率限制，不可靠

建议：开发时使用 Helius 或 QuickNode。生产交易绝不要使用公共 RPC。

核心读取方法

账户与余额

# SOL balance (in lamports, divide by 1e9 for SOL)
getBalance(pubkey, {commitment: "confirmed"})

# Full account info (data, owner, lamports, executable)
getAccountInfo(pubkey, {encoding: "jsonParsed"})

# Multiple accounts in one call
getMultipleAccounts([pubkey1, pubkey2], {encoding: "jsonParsed"})

代币账户

# All SPL token accounts owned by a wallet
getTokenAccountsByOwner(wallet_pubkey, {
    "programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
}, {"encoding": "jsonParsed"})

# Token balance for a specific token account
getTokenAccountBalance(token_account_pubkey)

# Largest token accounts (top holders)
getTokenLargestAccounts(mint_pubkey)

# Total supply of a token
getTokenSupply(mint_pubkey)

交易数据

# Get parsed transaction by signature
getTransaction(signature, {
    "encoding": "jsonParsed",
    "maxSupportedTransactionVersion": 0,
})

# Recent transaction signatures for an address
getSignaturesForAddress(pubkey, {
    "limit": 20,
    "before": "optional_signature",  # pagination cursor
})

# Transaction status
getSignatureStatuses([sig1, sig2])

区块与 Slot

# Current slot
getSlot({commitment: "confirmed"})

# Block data
getBlock(slot, {
    "encoding": "jsonParsed",
    "transactionDetails": "full",
    "maxSupportedTransactionVersion": 0,
})

# Latest blockhash (needed for tx building)
getLatestBlockhash({commitment: "confirmed"})

# Slot leader schedule
getLeaderSchedule()

程序账户

# All accounts owned by a program (with filters)
getProgramAccounts(program_pubkey, {
    "encoding": "jsonParsed",
    "filters": [
        {"dataSize": 165},  # Filter by account data size
        {"memcmp": {         # Filter by data content
            "offset": 32,
            "bytes": "base58_encoded_value",
        }},
    ],
})

警告：不带筛选条件的 getProgramAccounts 可能返回数百万条结果并超时。务必使用 dataSize 和／或 memcmp 筛选条件。

优先费用

# Recent priority fee estimates
getRecentPrioritizationFees([account_pubkey])
# Returns array of { slot, prioritizationFee } for recent slots

# Minimum rent for account
getMinimumBalanceForRentExemption(data_length)

写入方法

发送交易

# Send a signed, serialized transaction
sendTransaction(base64_tx, {
    "encoding": "base64",
    "skipPreflight": False,
    "preflightCommitment": "confirmed",
    "maxRetries": 3,
})

# Simulate before sending
simulateTransaction(base64_tx, {
    "encoding": "base64",
    "sigVerify": False,
    "commitment": "confirmed",
})

交易确认

import time

def confirm_transaction(rpc_url: str, signature: str, timeout: float = 30.0) -> bool:
    """Poll for transaction confirmation."""
    start = time.time()
    while time.time() - start < timeout:
        result = rpc_call("getSignatureStatuses", [[signature]])
        statuses = result.get("result", {}).get("value", [None])
        if statuses[0] is not None:
            status = statuses[0]
            if status.get("err"):
                return False
            if status.get("confirmationStatus") in ("confirmed", "finalized"):
                return True
        time.sleep(0.5)
    return False

Commitment 级别

级别	说明	适用场景
processed	单节点确认	优先速度而非安全性
confirmed	超级多数（2/3+）	交易的默认级别
finalized	最大超级多数 + 31 个 Slot	关键操作

务必明确指定 commitment。默认值因服务提供商而异。

常见模式

获取钱包的所有代币持仓

def get_wallet_tokens(wallet: str) -> list[dict]:
    """Get all SPL token holdings with metadata."""
    result = rpc_call("getTokenAccountsByOwner", [
        wallet,
        {"programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"},
        {"encoding": "jsonParsed"},
    ])
    tokens = []
    for acct in result["result"]["value"]:
        info = acct["account"]["data"]["parsed"]["info"]
        tokens.append({
            "mint": info["mint"],
            "amount": int(info["tokenAmount"]["amount"]),
            "decimals": info["tokenAmount"]["decimals"],
            "ui_amount": info["tokenAmount"]["uiAmount"],
        })
    return [t for t in tokens if t["amount"] > 0]

获取某种代币的顶级持有者

def get_top_holders(mint: str) -> list[dict]:
    """Get the 20 largest holders of a token."""
    result = rpc_call("getTokenLargestAccounts", [mint])
    supply_result = rpc_call("getTokenSupply", [mint])
    total_supply = int(supply_result["result"]["value"]["amount"])

    holders = []
    for acct in result["result"]["value"]:
        amount = int(acct["amount"])
        holders.append({
            "address": acct["address"],
            "amount": amount,
            "decimals": acct["decimals"],
            "ui_amount": acct["uiAmount"],
            "percentage": amount / total_supply * 100 if total_supply > 0 else 0,
        })
    return holders

批量 RPC 调用

def rpc_batch(calls: list[tuple[str, list]]) -> list[dict]:
    """Execute multiple RPC calls in a single HTTP request."""
    payload = [
        {"jsonrpc": "2.0", "id": i, "method": method, "params": params}
        for i, (method, params) in enumerate(calls)
    ]
    resp = httpx.post(RPC, json=payload, timeout=30.0)
    results = resp.json()
    results.sort(key=lambda r: r["id"])
    return [r.get("result") for r in results]

重要程序 ID

程序	ID
System	11111111111111111111111111111111
SPL Token	TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA
Token-2022	TokenzQdBNbLqP5VEhdkAS6EPFLC1PHnBqCXEpPxuEb
Associated Token	ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL
Raydium AMM	675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8
Raydium CLMM	CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK
Orca Whirlpool	whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc
Meteora DLMM	LBUZKhRxPF3XUpBCjp4YzTKgLccjZhTSDM9YuVaPwxo
PumpFun	6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P
Jupiter v6	JUP6LkbZbjS1jKKwapdHNy74zcZ3tLUZoi5QNyVTaV4
Compute Budget	ComputeBudget111111111111111111111111111111

何时使用直接 RPC，何时使用更高级的 API

需求	使用方式
检查代币余额	直接 RPC（getTokenAccountsByOwner）
顶级 20 位持有者	直接 RPC（getTokenLargestAccounts）
历史 OHLCV	Birdeye 或 SolanaTracker
已解析的交易历史	Helius Enhanced Transactions
代币元数据（名称、图片）	Helius DAS API
实时流	Yellowstone gRPC
钱包 PnL 跟踪	SolanaTracker
代币风险评分	SolanaTracker
跨链数据	DexScreener 或 CoinGecko

文件

参考资料

• references/methods.md — 完整的 RPC 方法参考，包含参数和响应架构
• references/error_handling.md — 错误代码、速率限制、超时处理和重试策略
• references/providers.md — RPC 服务提供商比较，包含定价和功能

脚本

• scripts/wallet_scanner.py — 扫描钱包中的所有代币持仓及余额
• scripts/token_holders.py — 获取任意代币的顶级持有者和持仓集中度指标