Documentation Index
Fetch the complete documentation index at: https://docs.raydium.io/llms.txt
Use this file to discover all available pages before exploring further.
Raydium ne publie pas de SDK Python officiel. Les patterns présentés ici composent trois bibliothèques communautaires bien maintenues : solders (primitives Solana liées à Rust), solana-py (client RPC) et anchorpy (constructeurs d’instructions de style Anchor à partir des IDL). Cette combinaison couvre tout ce que fait le SDK TS, juste un peu moins polie.
Environnement
python -m venv .venv
source .venv/bin/activate
pip install solders solana anchorpy construct base58
solders == 0.21.*
solana == 0.34.*
anchorpy == 0.20.*
anchorpy accuse parfois du retard par rapport à anchor-lang ; pour un programme Raydium récemment déployé, vérifiez que l’IDL compile sous votre version épinglée d’anchorpy avant de vous engager.
Connexion et keypair
from solana.rpc.async_api import AsyncClient
from solders.keypair import Keypair
client = AsyncClient("https://api.mainnet-beta.solana.com", commitment="confirmed")
owner = Keypair.from_bytes(bytes(open("keypair.json", "rb").read()))
AsyncClient est la variante async ; le Client sync est disponible pour les scripts rapides mais async est préféré pour tout ce qui envoie plusieurs requêtes.
Lecture de l’état du pool
La plupart des usages de production lisent l’état du pool décodé depuis l’API REST de Raydium (voir sdk-api/rest-api) plutôt que de décoder manuellement les données on-chain — c’est plus simple et la latence est acceptable pour la plupart des cas d’usage.
import httpx
async def get_pool(pool_id: str) -> dict:
async with httpx.AsyncClient() as http:
r = await http.get(
"https://api-v3.raydium.io/pools/info/ids",
params={"ids": pool_id},
)
r.raise_for_status()
data = r.json()
if not data["success"]:
raise RuntimeError(data["error"]["message"])
return data["data"][0]
pool = await get_pool("58oQChx4yWmvKdwLLZzBi4ChoCc2fqCUWBkwMihLYQo2")
print(pool["price"], pool["day"]["volume"])
from construct import Struct, Int64ul, Int128ul, Bytes, this
# Layout partiel de CPMM PoolState (premiers champs)
POOL_STATE_LAYOUT = Struct(
"discriminator" / Bytes(8),
"amm_config" / Bytes(32),
"pool_creator" / Bytes(32),
"token_0_vault" / Bytes(32),
"token_1_vault" / Bytes(32),
"lp_mint" / Bytes(32),
"token_0_mint" / Bytes(32),
"token_1_mint" / Bytes(32),
# ...
)
from solders.pubkey import Pubkey
async def decode_pool(pool_id: Pubkey) -> dict:
resp = await client.get_account_info(pool_id)
data = resp.value.data
return POOL_STATE_LAYOUT.parse(data)
src/raydium/cpmm/layout.ts (source TS) ; portez-le vers construct selon vos besoins. anchorpy peut faire cela automatiquement à partir de l’IDL — voir ci-dessous.
Construire et envoyer un swap
Pour la simplicité, utilisez le endpoint server-built-transaction de Raydium. Le serveur retourne une transaction prête à signer ; vous devez seulement ajouter votre signature :
import httpx
import base64
from solders.transaction import VersionedTransaction
from solana.rpc.types import TxOpts
async def swap(pool_id: str, amount_in: int, slippage_bps: int):
async with httpx.AsyncClient() as http:
r = await http.get(
"https://api-v3.raydium.io/transaction/swap-base-in",
params={
"poolId": pool_id,
"amount": amount_in,
"inputMint": "So11111111111111111111111111111111111111112", # WSOL
"outputMint": "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v", # USDC
"slippageBps": slippage_bps,
"wallet": str(owner.pubkey()),
"txVersion": "V0",
"computeUnitPriceMicroLamports": 50_000,
},
)
r.raise_for_status()
data = r.json()["data"]
# Décodez la tx pré-construite, signez avec notre keypair, envoyez.
raw = base64.b64decode(data["tx"]["transaction"])
tx = VersionedTransaction.from_bytes(raw)
tx.sign([owner])
sig = await client.send_transaction(tx, opts=TxOpts(skip_preflight=False))
await client.confirm_transaction(sig.value, commitment="confirmed")
return sig.value, data["swapResponse"]
Construire un swap côté client (via anchorpy)
Pour une latence plus faible ou quand vous ne pouvez pas accéder à l’API de Raydium (régions sanctionnées, configurations isolées) :
from anchorpy import Program, Provider, Wallet, Context
from solana.rpc.async_api import AsyncClient
from solders.pubkey import Pubkey
import json
idl = json.load(open("cpmm.json")) # from raydium-sdk-v2
provider = Provider(client, Wallet(owner))
program = Program(idl, Pubkey.from_string(CPMM_PROGRAM_ID), provider)
# Invoquez swap_base_input :
tx_sig = await program.rpc["swap_base_input"](
amount_in,
minimum_amount_out,
ctx=Context(
accounts={
"payer": owner.pubkey(),
"authority": owner.pubkey(),
"amm_config": amm_config_pk,
"pool_state": pool_state_pk,
"input_token_account": user_input_ata,
"output_token_account": user_output_ata,
"input_vault": input_vault_pk,
"output_vault": output_vault_pk,
"input_token_program": TOKEN_PROGRAM_ID,
"output_token_program": TOKEN_PROGRAM_ID,
"input_token_mint": input_mint,
"output_token_mint": output_mint,
"observation_state": observation_state_pk,
},
),
)
anchorpy ne les dérive pas automatiquement.
Architecture type d’un bot
Une structure courante de bot Python Raydium :
┌──────────────────┐
│ Scheduler │ cron / asyncio / redis queue
└──────────┬───────┘
│
▼
┌──────────────────┐
│ Price poller │ httpx + Raydium REST API
│ (per pool) │ or WebSocket RPC sub
└──────────┬───────┘
│ event
▼
┌──────────────────┐
│ Strategy engine │ compute signal, decide trade params
└──────────┬───────┘
│ trade params
▼
┌──────────────────┐
│ TX builder │ Raydium REST server-built-tx or anchorpy
│ + signer │ solders.Keypair
└──────────┬───────┘
│ VersionedTransaction
▼
┌──────────────────┐
│ RPC sender │ solana-py AsyncClient + Jito RPC
│ (retry + monitor)│ priority-fee logic
└──────────┬───────┘
│ sig
▼
┌──────────────────┐
│ Ledger store │ Postgres for positions, pending txs, PnL
└──────────────────┘
- Fournisseur RPC. Les RPC publiques mainnet rate-limitent agressivement. Utilisez un fournisseur dédié (Helius, QuickNode, Triton) pour un trafic soutenu.
- WebSocket pour l’état du pool.
client.account_subscribe(pool_id) pousse les mises à jour à chaque changement d’état. Beaucoup plus serré que du polling.
- Fournisseur de frais de priorité. Helius a un endpoint
getPriorityFeeEstimate ; Triton a le sien. Dimensionnez votre frais selon le 75e percentile des frais récents sur le programme cible.
- Bundles pour les trades sensibles au MEV. Routez par le moteur de bloc de Jito si vous ne pouvez pas tolérer le risque de sandwich. Libs Python :
jito-sdk-python (tiers, qualité variable).
Lecture de l’état du farm
FARM_V6_ID = Pubkey.from_string("...")
async def get_farm_v6(farm_id: Pubkey):
resp = await client.get_account_info(farm_id)
return farm_v6_idl_program.account["FarmState"].decode(resp.value.data)
farm = await get_farm_v6(farm_id)
print(farm.total_staked, farm.reward_info_count)
for r in farm.reward_infos[:farm.reward_info_count]:
print(r.reward_mint, r.emission_per_second_x64)
.account["X"].decode(bytes) d’anchorpy donne un objet Python natif correspondant à la struct IDL.
Pièges
1. Gestion des décimales
Le float natif de Python est IEEE-754 double ; les montants dans les mints à 9 décimales (1 SOL = 1e9 unités) restent précis mais les ratios et les produits perdent en précision. Utilisez int (solders retourne int pour tous les champs de montant) et routez par decimal.Decimal pour tout calcul arithmétique de prix.
2. Raisonnement basé sur les slots vs basé sur les timestamps
Certaines versions de farm utilisent des compteurs de slots ; LaunchLab utilise des timestamps. solana-py retourne slot dans les réponses RPC, mais convertir slot → timestamp n’est pas sans perte (varie selon le planning des leaders). Si vous avez besoin de l’heure murale, appelez get_block_time(slot) explicitement.
3. Épuisement du pool de connexions
AsyncClient ouvre une connexion HTTP par requête par défaut. Sous charge élevée, réutilisez les sessions httpx.AsyncClient et définissez une limits=httpx.Limits(max_connections=100) appropriée.
4. Limites de taille de transaction
Les transactions construites en Python ne sont pas plus petites que celles construites en TS — la limite de 1232 octets s’applique équitablement. Utilisez les transactions V0 (address lookup tables) pour tout ce qui route par plus de ~2 pools.
Ressources
Sources :
