Intégration Python

Construire des bots et des services pour Raydium en Python via solders, solana-py et anchorpy — configuration de l’environnement, lecture de l’état du pool, envoi de swaps et architecture type d’un bot.

Cette page est traduite automatiquement par IA. La version anglaise fait foi.

Voir la version anglaise →

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

Versions qui fonctionnent ensemble actuellement :

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"])

Pour les bots qui nécessitent une latence minimale, décodez directement les octets on-chain :

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)

Le layout complet se trouve dans 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"]

C’est le chemin le plus rapide vers un bot fonctionnel. Le quote du serveur expire rapidement (≈30s) ; ne le mettez pas en cache.

​

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,
        },
    ),
)

Les dérivations PDA (observation state, pool authority) suivent les mêmes formules que dans le chapitre CPMM. 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
└──────────────────┘

Décisions clés pour la production :

• 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)

Le .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

• sdk-api/rest-api — les endpoints HTTP utilisés ci-dessus.
• sdk-api/anchor-idl — où obtenir l’IDL pour anchorpy.
• integration-guides/routing-and-mev — patterns de bundles Jito.

Sources :

• solders
• solana-py
• anchorpy