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.
Esta página fue traducida automáticamente por IA. La versión en inglés es la fuente autorizada.Ver versión en inglés → Raydium no publica un SDK oficial para Python. Los patrones aquí componen tres bibliotecas mantenidas por la comunidad: solders (primitivas de Solana vinculadas a Rust), solana-py (cliente RPC) y anchorpy (constructores de instrucciones al estilo Anchor a partir de IDLs). La combinación cubre todo lo que hace el SDK de TS; simplemente es menos pulida.
Entorno
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 periódicamente se queda atrás de la versión de anchor-lang; para un programa Raydium recientemente desplegado, verifica que el IDL se compile bajo tu anchorpy fijada antes de hacer commit.
Conexión y 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 es la variante async; el Client síncrono está disponible para scripts rápidos pero async es preferible para cualquier cosa que envíe múltiples solicitudes.
Lectura del estado del pool
La mayoría del uso en producción lee el estado del pool decodificado desde la API REST de Raydium (ver sdk-api/rest-api) en lugar de decodificar manualmente los datos en cadena — es más simple y la latencia es aceptable para la mayoría de casos de uso.
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
# Esquema parcial de PoolState de CPMM (primeros campos)
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 (fuente de TS); transpórtalo a construct según sea necesario. anchorpy puede hacer esto automáticamente dado el IDL — ve abajo.
Construcción y envío de un swap
Para simplificar, usa el endpoint server-built-transaction de Raydium. El servidor devuelve una transacción lista para firmar; solo necesitas agregar tu firma:
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"]
# Decodifica la tx pre-construida, firma con nuestra keypair, envía.
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"]
Construcción de un swap del lado del cliente (vía anchorpy)
Para menor latencia o cuando no puedes acceder a la API de Raydium (regiones sancionadas, configuraciones aisladas):
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)
# Invoca 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 no las deriva automáticamente.
Arquitectura típica de bot
Una estructura común de bot Raydium en Python:
┌──────────────────┐
│ Scheduler │ cron / asyncio / redis queue
└──────────┬───────┘
│
▼
┌──────────────────┐
│ Price poller │ httpx + Raydium REST API
│ (per pool) │ o WebSocket RPC sub
└──────────┬───────┘
│ event
▼
┌──────────────────┐
│ Strategy engine │ calcula signal, decide parámetros de trade
└──────────┬───────┘
│ parámetros de trade
▼
┌──────────────────┐
│ TX builder │ server-built-tx REST de Raydium o anchorpy
│ + signer │ solders.Keypair
└──────────┬───────┘
│ VersionedTransaction
▼
┌──────────────────┐
│ RPC sender │ solana-py AsyncClient + Jito RPC
│ (retry + monitor)│ priority-fee logic
└──────────┬───────┘
│ sig
▼
┌──────────────────┐
│ Ledger store │ Postgres para positions, txs pendientes, PnL
└──────────────────┘
- Proveedor RPC. Los RPC públicos mainnet limitan la velocidad agresivamente. Usa un proveedor dedicado (Helius, QuickNode, Triton) para tráfico sostenido.
- WebSocket para estado del pool.
client.account_subscribe(pool_id) envía actualizaciones en cada cambio de estado. Mucho más ajustado que polling.
- Proveedor de priority fee. Helius tiene un endpoint
getPriorityFeeEstimate; Triton tiene el suyo. Dimensiona tu fee basándose en el percentil 75 de fees recientes en el programa objetivo.
- Bundles para trades sensibles a MEV. Rutea a través del block engine de Jito si no puedes tolerar riesgo de sandwich. Librerías Python:
jito-sdk-python (tercera parte, calidad varía).
Lectura del estado de la granja
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) de anchorpy da un objeto nativo de Python que coincide con el struct de IDL.
Trampas
1. Manejo de decimales
El float nativo de Python es IEEE-754 double; los montos en mints de 9 decimales (1 SOL = 1e9 unidades) se mantienen precisos pero las ratios y productos pierden precisión. Usa int (solders devuelve int para todos los campos de cantidad) y encamina a través de decimal.Decimal para cualquier aritmética de precios.
2. Razonamiento basado en slot vs basado en timestamp
Algunas versiones de granja usan contadores de slot; LaunchLab usa timestamps. solana-py devuelve slot en respuestas RPC, pero convertir slot → timestamp es con pérdida (varía según el schedule de leader). Si necesitas tiempo de pared, llama a get_block_time(slot) explícitamente.
3. Agotamiento del pool de conexión
AsyncClient abre una conexión HTTP por solicitud por defecto. Bajo carga alta, reutiliza sesiones de httpx.AsyncClient y establece un limits=httpx.Limits(max_connections=100) apropiado.
4. Límites de tamaño de transacción
Las transacciones construidas en Python no son más pequeñas que las construidas en TS — el límite de 1232 bytes se aplica por igual. Usa transacciones V0 (tablas de búsqueda de direcciones) para cualquier cosa que se enrute a través de más de ~2 pools.
Referencias
Fuentes:
