Trading & Bots

import asyncio
from solana.rpc.async_api import AsyncClient
 
async def bot_loop(rpc: str):
    async with AsyncClient(rpc) as client:
        while True:
            slot = (await client.get_slot()).value
            # 1. read signals  2. build ix  3. simulate  4. send  5. confirm
            await asyncio.sleep(0.4)  # respect rate limits

import asyncio
import logging
from solders.compute_budget import set_compute_unit_limit, set_compute_unit_price
from solders.keypair import Keypair
from solders.pubkey import Pubkey
from solders.hash import Hash
from solders.message import Message
from solders.transaction import VersionedTransaction
from solana.rpc.async_api import AsyncClient
from solana.rpc.types import TxOpts
 
log = logging.getLogger("mm-bot")
RPC = "https://api.mainnet-beta.solana.com"  # use private RPC in production
 
class SimpleSender:
    def __init__(self, client: AsyncClient, payer: Keypair):
        self.client = client
        self.payer = payer
 
    async def send(self, instructions, micro_lamports: int):
        bh = await self.client.get_latest_blockhash()
        blockhash = Hash.from_string(str(bh.value.blockhash))
        ixs = [
            set_compute_unit_limit(200_000),
            set_compute_unit_price(micro_lamports),
            *instructions,
        ]
        msg = Message.new_with_blockhash(ixs, self.payer.pubkey(), blockhash)
        tx = VersionedTransaction(msg, [self.payer])
 
        sim = await self.client.simulate_transaction(tx)
        if sim.value.err:
            log.warning("sim_fail err=%s logs=%s", sim.value.err, sim.value.logs)
            return None
 
        resp = await self.client.send_transaction(
            tx, opts=TxOpts(skip_preflight=True, preflight_commitment="processed")
        )
        await self.client.confirm_transaction(resp.value, commitment="confirmed")
        return resp.value
 
async def main():
    payer = Keypair()  # load securely
    async with AsyncClient(RPC) as client:
        sender = SimpleSender(client, payer)
        # instructions = [swap_ix_from_jupiter_api(...)]
        # await sender.send(instructions, micro_lamports=10_000)
 
asyncio.run(main())

# Never block the event loop on CPU work
await asyncio.to_thread(heavy_analytics, frame)
 
# Jittered backoff on 429 / blockhash errors
import random
await asyncio.sleep(0.2 + random.random() * 0.3)