example code to do a swap on raydium

raydium_swap.py

import base64
import logging
import requests
from solana.rpc.api import Client
from solders.keypair import Keypair
from solders.transaction import Transaction
from solders.pubkey import Pubkey

from solana.rpc.types import TokenAccountOpts
from spl.token.constants import TOKEN_PROGRAM_ID
from spl.token._layouts import ACCOUNT_LAYOUT, MINT_LAYOUT  # Import the layout for decoding

class SolanaWallet:
    def __init__(self, private_key: str):
        self.client = Client("https://api.mainnet-beta.solana.com")
        self.public_key = Keypair.from_base58_string(private_key).pubkey()
        self.token_balances = self.load_token_balances()

    def get_sol_balance(self):
        balance = self.client.get_balance(self.public_key)
        return balance.value / 1e9  # Convert lamports to SOL

    def load_token_balances(self):
        token_balances = {}
        response = self.client.get_token_accounts_by_owner(
            self.public_key, TokenAccountOpts(program_id=TOKEN_PROGRAM_ID)
        )
        for token_account in response.value:
            parsed_data = ACCOUNT_LAYOUT.parse(token_account.account.data)  # Parse the data using the layout
            mint_address = Pubkey.from_bytes(parsed_data.mint)
            acc_info = self.client.get_account_info(mint_address)
            mint_info = MINT_LAYOUT.parse(acc_info.value.data)
            decimals = mint_info.decimals
            balance = parsed_data.amount / (10 ** decimals)
            token_balances[str(mint_address)] = (
                balance,
                decimals,
                token_account.pubkey,
            )

        print(token_balances)
        return token_balances

NATIVE_MINT = "So11111111111111111111111111111111111111112"

logging.basicConfig(
    level=logging.DEBUG, format="%(asctime)s %(levelname)s %(message)s"
)


def get_token_decimals(token_mint: str) -> int:
    # Connect to the Solana mainnet (or change the endpoint as needed)
    client = Client("https://api.mainnet-beta.solana.com")
    # Query the token supply which includes the decimals info.
    response = client.get_token_supply(token_mint)
    return response.value.decimals


def swap_tokens(
    wallet_private_key: str, input_mint: str, output_mint: str, amount: int, slippage: float
):
    """
    Swaps tokens using Raydium's API.

    Parameters:
      - wallet_private_key (str): Your wallet's private key in base58 format (64-byte secret key).
      - input_mint (str): The mint address of the token you want to swap from.
      - output_mint (str): The mint address of the token you want to swap to.
      - amount (int): Amount to swap in the smallest unit (e.g. if decimals=6, 1 token = 1_000_000).
      - slippage (float): Acceptable slippage percentage (e.g., 0.5 for 0.5%).

    Returns:
      - The response from sending the transaction.
    """
    sol = SolanaWallet(wallet_private_key)
    client = sol.client

    # Properly load your wallet keypair.
    wallet = Keypair.from_base58_string(wallet_private_key)

    fee_url = "https://api-v3.raydium.io/main/auto-fee"
    response = requests.get(fee_url)
    if response.status_code != 200:
        raise Exception(f"Failed to get fees: {response.status_code} {response.text}")

    fee_data = response.json()
    print(fee_data)

    tx_version = "LEGACY"

    # Set up the Raydium quote endpoint and parameters.
    quote_url = "https://transaction-v1.raydium.io/compute/swap-base-in"
    params = {
        "inputMint": input_mint,
        "outputMint": output_mint,
        "amount": amount,
        "slippageBps": int(slippage * 100),
        "txVersion": tx_version,
    }

    response = requests.get(quote_url, params=params)
    if response.status_code != 200:
        raise Exception(f"Failed to get quote: {response.status_code} {response.text}")

    data = response.json()
    print(str(wallet.pubkey()))

    tx_url = "https://transaction-v1.raydium.io/transaction/swap-base-in"
    params = {
        "computeUnitPriceMicroLamports": str(fee_data["data"]["default"]["m"]),
        "swapResponse": data,
        "txVersion": tx_version,
        "wallet": str(wallet.pubkey()),
        "wrapSol": input_mint == NATIVE_MINT,
        "unwrapSol": output_mint == NATIVE_MINT,
        "inputAccount": str(sol.token_balances[input_mint][2])
        if input_mint != NATIVE_MINT
        else None,
        "outputAccount": str(sol.token_balances[output_mint][2])
        if output_mint != NATIVE_MINT
        else None,
    }

    response = requests.post(tx_url, json=params)
    logging.info(response.request.body)
    if response.status_code != 200:
        raise Exception(f"Failed to get tx: {response.status_code} {response.text}")

    data = response.json()["data"]
    allTx = []
    print(data)

    for tx_data in data:
        tx = Transaction.from_bytes(base64.b64decode(tx_data["transaction"]))
        allTx.append(tx)

    for tx in allTx:
        recent_blockhash = client.get_latest_blockhash()
        tx.sign(keypairs=[wallet], recent_blockhash=recent_blockhash.value.blockhash)
        send_resp = client.send_transaction(txn=tx)
        print(send_resp)


if __name__ == "__main__":
    # Replace these values with your actual data.
    wallet_private_key = "<Pk HERE>"  # NEVER expose your private key publicly!
    sol = SolanaWallet(wallet_private_key)
    input_mint = "<Mint HERE>"  # e.g., the mint address for USDC
    output_mint = "So11111111111111111111111111111111111111112"  # e.g., the mint address for SOL
    amount = 10 ** sol.token_balances[input_mint][1]  # example: 1 USDC if USDC has 6 decimals.
    slippage = 0.5  # Acceptable slippage of 0.5%
    
    swap_tokens(wallet_private_key, input_mint, output_mint, amount, slippage)


    

requirements.txt

requests
base58
solders
solana 
aiohttp