[Python] Solana transaction to Raydium Token with Python

So, this is my python script to buy a Token on Raydium with Solana. Im working with PyQt5 and getting everytime the same ERRORS. I dont know what i did wrong.

Working with solana-py and solders. solana-py for the AsyncClient solders for all other stuff, like public keys and private key of my wallet adresses on Solana, transactions like sending solana to another wallet.

[Error after trying the transaction] https://i.sstatic.net/BwcDbBzu.png [Showcase of the UI] https://i.sstatic.net/k3Dw2Kb8.png

import sys
import time
import base64
import requests
import traceback
import asyncio
from PyQt5.QtWidgets import (
QApplication, QDialog, QVBoxLayout, QLabel, QLineEdit,
QPushButton, QTableWidget, QTableWidgetItem, QMessageBox, QHeaderView
)
from PyQt5.QtCore import Qt, QThread, pyqtSignal
from solders.pubkey import Pubkey
from solders.transaction import Transaction
from solders.instruction import Instruction, AccountMeta
from solders.keypair import Keypair
from solders.system_program import ID as SYS_PROGRAM_ID
from solders.sysvar import RENT
from solders.message import Message
from solders.hash import Hash
from solders.system_program import transfer, TransferParams
from auth import supabase, get_current_user
from typing import Tuple
import base58

# -----------------------------------------------------------------------------
# Async Setup (für Fee-Abfragen)
# -----------------------------------------------------------------------------
global_loop = asyncio.new_event_loop()
asyncio.set_event_loop(global_loop)
from solana.rpc.async_api import AsyncClient
solana_client = AsyncClient("https://api.mainnet-beta.solana.com")
def run_async(coro):
return global_loop.run_until_complete(coro)

# -----------------------------------------------------------------------------
# Konstanten
# -----------------------------------------------------------------------------
RPC_URL = "https://api.mainnet-beta.solana.com"
RAYDIUM_API_BASE = "https://transaction-v1.raydium.io"
QUOTE_ENDPOINT = "/compute/swap-base-in"
TX_ENDPOINT = "/transaction/swap-base-in"
WSOL_MINT = "So11111111111111111111111111111111111111112"
USDC_MINT = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v"
TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
ASSOCIATED_TOKEN_PROGRAM_ID = Pubkey.from_string("ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL")

# -----------------------------------------------------------------------------
# Hilfsfunktionen
# -----------------------------------------------------------------------------
def get_priority_fee() -> str:
"""Gibt 0.05 SOL in Mikro-Lamports zurück."""
return "50000"

def validate_mint(address: str) -> bool:
"""Erlaubt explizit den USDC-Mint; ansonsten wird die Adresslänge geprüft."""
if address == USDC_MINT:
return True
try:
Pubkey.from_string(address)
return len(address) == 44
except:
return False

def get_balance(pubkey_str: str) -> float:
headers = {"Content-Type": "application/json"}
payload = {"jsonrpc": "2.0", "id": 1, "method": "getBalance", "params": [pubkey_str]}
try:
resp = requests.post(RPC_URL, json=payload, headers=headers)
data = resp.json()
if "error" in data:
print(f"RPC-Fehler für {pubkey_str}: {data['error']}")
return 0.0
print(f"RPC-Antwort für {pubkey_str}: {data}")
return data["result"]["value"] / 1e9
except Exception as e:
print(f"Fehler in get_balance für {pubkey_str}: {e}")
return 0.0

def send_transaction(tx: Transaction, keypair: Keypair) -> str:
try:
tx_bytes = bytes(tx)
tx_b64 = base64.b64encode(tx_bytes).decode("utf-8")
payload = {
"jsonrpc": "2.0", "id": 1, "method": "sendTransaction",
"params": [tx_b64, {"encoding": "base64", "preflightCommitment": "confirmed", "skipPreflight": False}]
}
resp = requests.post(RPC_URL, json=payload, timeout=10)
return resp.json().get("result", "")
except Exception as e:
print(f"Transaktion fehlgeschlagen: {str(e)}")
return ""

def get_current_fee() -> float:
try:
dummy_signer = Keypair()
dummy_instruction = transfer(
TransferParams(
from_pubkey=dummy_signer.pubkey(),
to_pubkey=Pubkey.from_string("11111111111111111111111111111111"),
lamports=0
)
)
message = Message([dummy_instruction])
recent_blockhash_resp = run_async(solana_client.get_latest_blockhash())
recent_blockhash = Hash.from_string(recent_blockhash_resp.value.blockhash)
fee_resp = run_async(solana_client.get_fee_for_message(message))
lamports_fee = fee_resp.value if fee_resp.value is not None else 5000
fee_sol = lamports_fee / 1_000_000_000
print(f"Normale Gebühr: {fee_sol} SOL")
return fee_sol
except Exception as e:
print("Fehler beim Abrufen der aktuellen Gebühr:", str(e))
return 0.0

def get_ata_fee() -> float:
try:
account_size = 165
resp = run_async(solana_client.get_minimum_balance_for_rent_exemption(account_size))
lamports = resp.value
ata_fee_sol = lamports / 1_000_000_000
print(f"ATA-Gebühr (Rent-Exemption): {ata_fee_sol} SOL")
return ata_fee_sol
except Exception as e:
print("Fehler beim Abrufen der ATA-Gebühr:", str(e))
return 0.0

# -----------------------------------------------------------------------------
# Worker-Thread für Swap-Operationen
# -----------------------------------------------------------------------------
class SwapWorker(QThread):
status_update = pyqtSignal(int, str, object)
summary_result = pyqtSignal(int, list)

def __init__(self, dialog, wallets, distributed_amounts, token_mint):
super().__init__()
self.dialog = dialog
self.wallets = wallets
self.distributed_amounts = distributed_amounts
self.token_mint = token_mint

def run(self):
success_count = 0
error_messages = []
for idx, (wallet, amount) in enumerate(zip(self.wallets, self.distributed_amounts)):
if amount <= 0:
self.status_update.emit(idx, " Niedriger Kontostand", Qt.darkYellow)
continue
self.status_update.emit(idx, " Verarbeitung...", Qt.blue)
success, error_msg = self.dialog.perform_raydium_swap(wallet, self.token_mint, amount)
if success:
self.status_update.emit(idx, "✅ Erfolg", Qt.darkGreen)
success_count += 1
else:
self.status_update.emit(idx, f"❌ Fehlgeschlagen: {error_msg[:60]}", Qt.red)
error_messages.append(f"Wallet {wallet['public_key'][:5]}...: {error_msg}")
time.sleep(1)
self.summary_result.emit(success_count, error_messages)

# -----------------------------------------------------------------------------
# Hauptdialog-Klasse für Token-Swaps (Kauf)
# -----------------------------------------------------------------------------
class BuyTokenDialog(QDialog):
def __init__(self, parent=None):
super().__init__(parent)
self.wallets = []
self.distributed_amounts = []
self.swap_worker = None
self.setWindowTitle("Raydium Token Swapper")
self.setup_ui()

def setup_ui(self):
layout = QVBoxLayout(self)
layout.addWidget(QLabel("Token Mint Adresse:"))
self.token_address_input = QLineEdit()
layout.addWidget(self.token_address_input)
self.fee_info_label = QLabel("Gebühren: -")
self.fee_info_label.setStyleSheet("color: gray; font-style: italic;")
layout.addWidget(self.fee_info_label)
self.wallet_table = QTableWidget()
self.wallet_table.setColumnCount(3)
self.wallet_table.setHorizontalHeaderLabels(["Wallet", "Verfügbares SOL", "Status"])
self.wallet_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch)
layout.addWidget(self.wallet_table)
load_btn = QPushButton("Wallets laden")
load_btn.clicked.connect(self.load_wallets)
layout.addWidget(load_btn)
self.buy_btn = QPushButton("Swaps ausführen")
self.buy_btn.clicked.connect(self.execute_purchases)
self.buy_btn.setEnabled(False)
layout.addWidget(self.buy_btn)

def load_wallets(self):
user = get_current_user()
if not user:
QMessageBox.critical(self, "Fehler", "Nicht authentifiziert")
return
response = supabase.table("wallets").select("*").eq("user_id", user["user_id"]).execute()
if response.data:
self.wallets = response.data
self.update_wallet_table()
self.buy_btn.setEnabled(True)
else:
QMessageBox.warning(self, "Warnung", "Keine Wallets gefunden")

def update_wallet_table(self):
self.wallet_table.setRowCount(len(self.wallets))
self.distributed_amounts = []
token_mint = self.token_address_input.text().strip()
if token_mint and validate_mint(token_mint):
ata_fee = get_ata_fee()
else:
ata_fee = 0.0
normal_fee = get_current_fee()
fee_reserve = normal_fee + ata_fee
self.fee_info_label.setText(
f"Gebühren: Normal: {normal_fee:.9f} SOL | ATA: {ata_fee:.9f} SOL | Total: {fee_reserve:.9f} SOL"
)
print(f"Verwendeter Fee-Reserve pro Wallet: {fee_reserve} SOL (Normal: {normal_fee} + ATA: {ata_fee})")
for row, wallet in enumerate(self.wallets):
balance = get_balance(wallet["public_key"])
investable = max(balance - fee_reserve, 0)
self.distributed_amounts.append(investable)
wallet_item = QTableWidgetItem(wallet["public_key"])
self.wallet_table.setItem(row, 0, wallet_item)
available_sol_item = QTableWidgetItem(f"{investable:.3f} SOL")
available_sol_item.setTextAlignment(Qt.AlignRight | Qt.AlignVCenter)
self.wallet_table.setItem(row, 1, available_sol_item)
status_text = " Niedriger Kontostand" if investable <= 0 else " Bereit"
status_item = QTableWidgetItem(status_text)
status_item.setTextAlignment(Qt.AlignCenter)
self.wallet_table.setItem(row, 2, status_item)

def get_ata_for_token(self, wallet_address: str, token_mint: str) -> Pubkey:
try:
wallet_pubkey = Pubkey.from_string(wallet_address)
mint_pubkey = Pubkey.from_string(token_mint)
seeds = [bytes(wallet_pubkey), bytes(TOKEN_PROGRAM_ID), bytes(mint_pubkey)]
ata, _ = Pubkey.find_program_address(seeds, ASSOCIATED_TOKEN_PROGRAM_ID)
return ata
except Exception as e:
print(f"ATA-Erstellung fehlgeschlagen: {str(e)}")
return None

def check_ata_exists(self, ata: Pubkey) -> bool:
key_str = base58.b58encode(bytes(ata)).decode("utf-8")
payload = {"jsonrpc": "2.0", "id": 1, "method": "getAccountInfo", "params": [key_str]}
resp = requests.post(RPC_URL, json=payload)
return resp.json().get("result", {}).get("value") is not None

def create_ata(self, wallet_address: str, token_mint: str) -> bool:
try:
response = supabase.table("wallets").select("private_key").eq("public_key", wallet_address).execute()
if not response.data:
print(f"❌ Keine Wallet gefunden: {wallet_address}")
return False
private_key = response.data[0]["private_key"]
keypair = Keypair.from_bytes(bytes.fromhex(private_key))
blockhash_resp = requests.post(RPC_URL, json={"jsonrpc": "2.0", "id": 1, "method": "getLatestBlockhash"})
recent_blockhash = Hash.from_string(blockhash_resp.json()["result"]["value"]["blockhash"])
ix = Instruction(
program_id=ASSOCIATED_TOKEN_PROGRAM_ID,
data=b"",
accounts=[
AccountMeta(keypair.pubkey(), True, True),
AccountMeta(self.get_ata_for_token(wallet_address, token_mint), False, True),
AccountMeta(keypair.pubkey(), False, False),
AccountMeta(Pubkey.from_string(token_mint), False, False),
AccountMeta(SYS_PROGRAM_ID, False, False),
AccountMeta(TOKEN_PROGRAM_ID, False, False),
AccountMeta(RENT, False, False)
]
)
message = Message([ix], keypair.pubkey())
tx = Transaction(message=message, recent_blockhash=recent_blockhash, from_keypairs=[keypair])
tx_sig = send_transaction(tx, keypair)
print(f" ATA erstellt für {token_mint}: {tx_sig}")
return True
except Exception as e:
print(f"❌ ATA-Erstellung fehlgeschlagen: {str(e)}")
return False

def get_raydium_quote(self, token_mint: str, sol_amount: float):
params = {
"inputMint": WSOL_MINT,
"outputMint": token_mint,
"amount": int(sol_amount * 1e9),
"slippageBps": 200,
"txVersion": "V0" # Ursprünglicher Platzhalter; wird durch die Antwort dynamisch ergänzt
}
try:
print("[DEBUG] Anfrage an Raydium für Quote mit Parametern:", params)
resp = requests.get(RAYDIUM_API_BASE + QUOTE_ENDPOINT, params=params, timeout=10)
data = resp.json()
print("[DEBUG] Raydium Quote Antwort:", data)
# Rückgabe: success, swap_response und Versionsnummer (Standard "V1", falls nicht vorhanden)
return data.get("success", False), data.get("data"), data.get("version", "V1")
except Exception as e:
print("[DEBUG] Raydium Quote Anfrage fehlgeschlagen:", str(e))
return False, None, "V1"

def perform_raydium_swap(self, wallet, token_mint: str, sol_amount: float) -> Tuple[bool, str]:
try:
if not validate_mint(token_mint):
return False, "Ungültige Token Mint Adresse"
quote_ok, swap_response, quote_version = self.get_raydium_quote(token_mint, sol_amount)
if not quote_ok or not swap_response:
return False, "Fehler beim Abrufen eines gültigen Quotes von Raydium"
# Sicherstellen, dass das finale Token-ATA existiert
ata_final = self.get_ata_for_token(wallet["public_key"], token_mint)
if not self.check_ata_exists(ata_final):
if not self.create_ata(wallet["public_key"], token_mint):
return False, "Erstellung des finalen Token ATA fehlgeschlagen"
time.sleep(2)
# Erstellen der ATAs für Zwischen-Tokens aus dem Routenplan
route_plan = swap_response.get("routePlan", [])
required_tokens = set()
for step in route_plan:
for key in ["inputMint", "outputMint"]:
token = step.get(key)
if token and token not in [WSOL_MINT, token_mint]:
required_tokens.add(token)
for intermediate_token in required_tokens:
ata_intermediate = self.get_ata_for_token(wallet["public_key"], intermediate_token)
if not self.check_ata_exists(ata_intermediate):
print(f"[DEBUG] Erstelle ATA für Zwischen-Token: {intermediate_token}")
if not self.create_ata(wallet["public_key"], intermediate_token):
return False, f"Erstellung des ATA für Zwischen-Token {intermediate_token} fehlgeschlagen"
time.sleep(2)
# Aufbau des finalen Swap-Requests
post_data = {
"computeUnitPriceMicroLamports": get_priority_fee(),
"swapResponse": swap_response,
"txVersion": quote_version, # Version aus der Quote (normalerweise "V1")
"wallet": wallet["public_key"],
"wrapSol": True,
"unwrapSol": False,
"outputTokenAccount": str(ata_final),
"clientOrderId": str(int(time.time())),
"referralAccount": ""
}
print("[DEBUG] Sende finalen Swap-Request an Raydium mit post_data:", post_data)
try:
response = supabase.table("wallets").select("private_key").eq("public_key", wallet["public_key"]).execute()
private_key = response.data[0]["private_key"]
keypair = Keypair.from_bytes(bytes.fromhex(private_key))
except Exception as e:
return False, f"Wallet Schlüssel Fehler: {str(e)}"
try:
resp = requests.post(RAYDIUM_API_BASE + TX_ENDPOINT, json=post_data, timeout=15)
resp.raise_for_status()
resp_json = resp.json()
print("[DEBUG] Raydium finaler TX Response:", resp_json)
except requests.exceptions.RequestException as e:
return False, f"Verbindungsfehler zur Raydium API: {str(e)}"
# Falls der Fehler REQ_TX_VERSION_ERROR zurückkommt, wird ein Fallback versucht
if not resp_json.get("success"):
error_msg = resp_json.get("msg", "Unbekannter Raydium Fehler")
if "REQ_TX_VERSION_ERROR" in error_msg:
fallback_version = "V0" if post_data["txVersion"] == "V1" else "V1"
print(f"[DEBUG] Fallback: Versuch mit txVersion {fallback_version} statt {post_data['txVersion']}")
post_data["txVersion"] = fallback_version
try:
resp = requests.post(RAYDIUM_API_BASE + TX_ENDPOINT, json=post_data, timeout=15)
resp.raise_for_status()
resp_json = resp.json()
print("[DEBUG] Fallback Raydium finaler TX Response:", resp_json)
except requests.exceptions.RequestException as e:
return False, f"Fallback Verbindungsfehler zur Raydium API: {str(e)}"
if not resp_json.get("success"):
return False, f"Raydium Fehler (Fallback): {resp_json.get('msg', 'Unbekannter Fehler')}"
else:
return False, f"Raydium Fehler: {error_msg}"
success = False
error_messages = []
for tx_data in resp_json["data"]:
try:
tx = Transaction.from_bytes(base64.b64decode(tx_data["transaction"]))
tx.sign([keypair])
sig = send_transaction(tx, keypair)
if sig:
print(f"Swap Erfolg: {sig}")
success = True
else:
error_messages.append("Transaktion konnte nicht abgeschickt werden")
except Exception as e:
error_messages.append(str(e))
if not success and error_messages:
return False, " | ".join(error_messages[:3])
return success, ""
except Exception as e:
tb = traceback.format_exc()
print(f"Kritischer Fehler: {tb}")
return False, f"Unerwarteter Fehler: {str(e)}"

def execute_purchases(self):
token_mint = self.token_address_input.text().strip()
if not validate_mint(token_mint):
QMessageBox.warning(self, "Fehler", "Ungültige Token Adresse")
return
self.buy_btn.setEnabled(False)
self.swap_worker = SwapWorker(self, self.wallets, self.distributed_amounts, token_mint)
self.swap_worker.status_update.connect(self.update_status)
self.swap_worker.summary_result.connect(self.handle_summary)
self.swap_worker.start()

def handle_summary(self, success_count, error_messages):
summary = []
if success_count > 0:
summary.append(f"Erfolgreich: {success_count}")
if error_messages:
summary.append(f"Fehlgeschlagen: {len(error_messages)}")
details = "\n".join(error_messages[:5])
if len(error_messages) > 5:
details += f"\n...und {len(error_messages)-5} weitere"
QMessageBox.warning(self, "Swap Zusammenfassung", f"{' | '.join(summary)}\n\nFehlerdetails:\n{details}")
else:
QMessageBox.information(self, "Swap Zusammenfassung", f"Alle Swaps erfolgreich abgeschlossen!\nErfolgreich: {success_count}")
self.buy_btn.setEnabled(True)

def update_status(self, row: int, text: str, color=Qt.black):
item = QTableWidgetItem(text)
item.setForeground(color)
self.wallet_table.setItem(row, 2, item)
QApplication.processEvents()

if __name__ == "__main__":
app = QApplication(sys.argv)
dialog = BuyTokenDialog()
dialog.show()
sys.exit(app.exec_())

Continue reading...