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	1日あたり 50K クレジット	月額 $49 以上	拡張 RPC、DAS API
QuickNode	制限あり	月額 $49 以上	マルチチェーン、WebSocket
Triton	無料枠なし	月額約 $300 以上	Yellowstone gRPC が付属
Shyft	制限あり	月額 $49 以上	Yellowstone gRPC が付属
Alchemy	月間 300M CU	従量制	無料枠が充実
Public (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])

ブロックとスロット

# 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

コミットメントレベル

レベル	説明	使用する場面
processed	単一ノードによる確認	安全性より速度を優先する場合
confirmed	超多数（2/3 以上）	取引でのデフォルト
finalized	最大の超多数 + 31 スロット	重要な操作

コミットメントは必ず明示的に指定してください。デフォルト値はプロバイダーによって異なります。

よく使うパターン

ウォレットが保有するすべてのトークンを取得する

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
ウォレットの損益追跡	SolanaTracker
トークンのリスク評価	SolanaTracker
クロスチェーンデータ	DexScreener または CoinGecko

ファイル

リファレンス

• references/methods.md — パラメーターとレスポンススキーマを含む、RPC メソッドの完全なリファレンス
• references/error_handling.md — エラーコード、レート制限、タイムアウト処理、再試行戦略
• references/providers.md — 価格と機能を比較した RPC プロバイダーの一覧

スクリプト

• scripts/wallet_scanner.py — ウォレットをスキャンし、残高を含むすべてのトークン保有状況を取得します
• scripts/token_holders.py — 任意のトークンの上位保有者と保有集中度を取得します