noob here.
TL;DR can anyone please provide a working example with library versions in python to execute a simple swap using jupyter api or at least point me to a working documentation??
Longer version: I'm a bit desperate in finding a working example for a simple python integration for executing a simple swap on SOL mainnet.
The used libraries (i.e. solders, solana-py, anchorpy etc.) are a big mess (cross dependancies, not up to date versions, imported functions do not work or have been moved). I'm constantly reinstalling to random versions in order to find out which method works with which version.
The furthest i was getting was to actually receive a txid, but it does not appear on the solana explorer, so i assume something went wrong. But again, can not pinpoint on what exactly went wrong. I will provide the code which worked "best" so far, maybe someone with more knowledge sees, what the problem is, i would be grateful!
solana==0.16.0 solders==0.18.1 anchorpy==0.18.0 anchorpy-core==0.1.3
private_key = >YOUR_PRIVATE_KEY_HERE<
keypair = >YOUR_KEYPAIR<
# Jupiter API base URL
JUPITER_API_BASE_URL = "https://quote-api.jup.ag/v6/quote"
# Jupiter API endpoint for performing the swap
swap_url = "https://quote-api.jup.ag/v6/swap"
# Function to get the best quote from Jupiter API
def get_best_quote(input_mint, output_mint, amount, slippageBps, onlyDirectRoutes='true'):
params = {
"inputMint": So11111111111111111111111111111111111111112,
"outputMint": EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v,
"amount": 1000000,
"slippageBps": 50,
"onlyDirectRoutes": onlyDirectRoutes # Only direct routes
}
headers = {
'Accept': 'application/json'
}
response = requests.get(JUPITER_API_BASE_URL, params=params, headers=headers)
if response.status_code == 200:
return response.json()
else:
print("Error fetching quote:", response.text)
return None
async def swap_tokens(
input_mint,
output_mint,
amount,
slippageBps,
solana_client
):
quote = get_best_quote(
input_mint['token_address'],
output_mint['token_address'],
amount,
slippageBps=slippageBps
)
sender = Keypair.from_base58_string(os.getenv('TRADING_BOT_PRIVATE_KEY'))
if quote:
# Payload for the swap
payload = {
"userPublicKey": str(sender.pubkey()), # Convert to string, should not be null
"wrapAndUnwrapSol": True,
"quoteResponse": quote # Ensure this dictionary doesn't contain any unexpected nulls
}
swap_response = requests.post(swap_url, json=payload)
if swap_response.status_code == 200:
swap_result = swap_response.json()
print("Swap response:", swap_result)
try:
# Deserialize the transaction
swap_transaction = swap_result['swapTransaction']
raw_tx = solders.transaction.VersionedTransaction.from_bytes(
base64.b64decode(swap_transaction)
)
signed_txn = sender.sign_message(bytes(raw_tx.message))
s_signed_txn = solders.transaction.VersionedTransaction.populate(
raw_tx.message, [signed_txn]
)
encoded_tx = base64.b64encode(bytes(s_signed_txn)).decode("utf-8")
headers = {"Content-Type": "application/json"}
data = {
"jsonrpc": "2.0",
"id": 1,
"method": "sendTransaction",
"params": [
encoded_tx,
{
"skipPreflight": True,
"preflightCommitment": "finalized",
"encoding": "base64",
"maxRetries": None,
"minContextSlot": None,
},
],
}
async with httpx.AsyncClient() as client:
response = await client.post(
"https://api.mainnet-beta.solana.com",
headers=headers,
json=data,
)
return response.json()
