> ## 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.
# 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 →](/sdk-api/python-integration)
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
```bash theme={null}
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
```python theme={null}
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`](/fr/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.
```python theme={null}
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 :
```python theme={null}
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 :
```python theme={null}
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) :
```python theme={null}
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
```python theme={null}
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`](/fr/sdk-api/rest-api) — les endpoints HTTP utilisés ci-dessus.
* [`sdk-api/anchor-idl`](/fr/sdk-api/anchor-idl) — où obtenir l'IDL pour `anchorpy`.
* [`integration-guides/routing-and-mev`](/fr/integration-guides/routing-and-mev) — patterns de bundles Jito.
Sources :
* [solders](https://github.com/kevinheavey/solders)
* [solana-py](https://github.com/michaelhly/solana-py)
* [anchorpy](https://github.com/kevinheavey/anchorpy)