Memecoin markets move in seconds. By the time a human sees a new pair on a Telegram channel or DexScreener, the first wave of buyers has already entered. That’s why developers build sniper bots — automated systems that detect new liquidity, apply filters, and execute buys faster than manual trading allows.
This guide walks you through building a practical, production-oriented memecoin sniper bot on Solana using Python. We’ll cover:
- Reliable RPC and WebSocket connections
- Detecting new pairs and liquidity events
- On-chain safety filters (mint authority, freeze authority, liquidity locks, holder distribution)
- Execution via Jupiter Aggregator (recommended) or direct Raydium
- Position sizing and hard risk limits
- Error handling, logging, and circuit breakers
- Deploying on a VPS with monitoring
This is not a “get rich quick” script. Sniping is competitive, capital-intensive, and full of rugs. The goal here is a clean, maintainable foundation you can harden and extend.
Strategy note: Infrastructure alone doesn’t make a profitable system. Entry filters, exit rules, sizing, and risk management matter more than raw speed. If you want a ready-made memecoin trading framework with tested logic, risk rules, and practical filters, check the resource here: Memecoin Trading Strategy. This article focuses on the bot architecture that supports such strategies.
Why Solana for Memecoin Sniping?
Solana currently hosts the majority of high-velocity memecoin activity because of:
- Extremely low transaction fees
- Fast block times
- Rich ecosystem of AMMs (Raydium, Pump.fun, Meteora, etc.)
- Good aggregator support (Jupiter)
The trade-offs are real: RPC rate limits, network congestion during hype, and a constant stream of low-quality or malicious tokens. A production bot must handle all of that gracefully.
Project Setup
mkdir solana-memecoin-sniper
cd solana-memecoin-sniper
python -m venv venv
source venv/bin/activate
pip install solana solders httpx python-dotenv loguru tenacity pandas
Optional but useful:
pip install anchorpy base58
Recommended structure:
solana-memecoin-sniper/
├── .env
├── config.py
├── bot.py
├── listener.py
├── filters.py
├── executor.py
├── risk.py
├── utils/
│ ├── logger.py
│ └── helpers.py
└── requirements.txt
.env example:
RPC_URL=https://mainnet.helius-rpc.com/?api-key=YOUR_KEY
WS_URL=wss://mainnet.helius-rpc.com/?api-key=YOUR_KEY
PRIVATE_KEY=your_base58_private_key
JUPITER_API=https://quote-api.jup.ag/v6
TELEGRAM_BOT_TOKEN=...
TELEGRAM_CHAT_ID=...
MAX_SOL_PER_TRADE=0.5
MAX_OPEN_POSITIONS=3
SLIPPAGE_BPS=800
Security warning: Never commit your private key. Use a dedicated hot wallet with limited funds. Prefer hardware wallet + separate signing service for larger capital.
Connecting to Solana
Use a high-quality RPC. Public endpoints will rate-limit you instantly during volume spikes. Popular paid options include Helius, QuickNode, Triton, and GenesysGo.
# config.py
import os
from dotenv import load_dotenv
from solders.keypair import Keypair
from solana.rpc.async_api import AsyncClient
import base58
load_dotenv()
RPC_URL = os.getenv("RPC_URL")
WS_URL = os.getenv("WS_URL")
PRIVATE_KEY = os.getenv("PRIVATE_KEY")
keypair = Keypair.from_bytes(base58.b58decode(PRIVATE_KEY))
client = AsyncClient(RPC_URL)
Test the connection:
import asyncio
from solana.rpc.async_api import AsyncClient
async def test():
client = AsyncClient(RPC_URL)
balance = await client.get_balance(keypair.pubkey())
print(balance)
await client.close()
asyncio.run(test())
Detecting New Pairs & Liquidity Events
There are several common approaches:
- Listen to Raydium or Pump.fun program logs via WebSocket
- Poll new token listings from Birdeye, DexScreener, or Helius enhanced APIs
- Monitor specific pool creation instructions
For a practical starting point, many bots combine:
- WebSocket logs for speed
- REST confirmation + metadata for safety
Here’s a simplified listener pattern using logs:
# listener.py
import asyncio
import json
from solana.rpc.websocket_api import connect
from loguru import logger
RAYDIUM_AMM_PROGRAM = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8" # example
async def listen_for_new_pools():
async with connect(WS_URL) as websocket:
await websocket.logs_subscribe(
filter_={"mentions": [RAYDIUM_AMM_PROGRAM]},
commitment="confirmed"
)
logger.info("Listening for new pool events...")
while True:
try:
msg = await websocket.recv()
# Parse logs for initialize2 / pool creation patterns
# This part requires careful log decoding
process_log_message(msg)
except Exception as e:
logger.error(f"WebSocket error: {e}")
await asyncio.sleep(5)
In practice you will also want:
- Helius
transactionSubscribeor enhanced websockets - Birdeye / DexScreener new-pairs endpoints as a secondary signal
- A short confirmation delay (1–3 blocks) before acting
Raw speed without filters is a fast way to buy rugs.
Safety Filters (The Real Edge)
Most sniper bots lose money because they buy everything. Production systems apply strict pre-trade filters.
Common high-value checks:
# filters.py
from solders.pubkey import Pubkey
async def check_mint_authority(client, mint: str) -> bool:
"""Return True if mint authority is revoked (safer)."""
info = await client.get_account_info(Pubkey.from_string(mint))
# Parse mint account data – authority should be None
# Implementation depends on token program layout
return True # placeholder
async def check_freeze_authority(client, mint: str) -> bool:
"""Return True if freeze authority is revoked."""
return True # placeholder
async def check_liquidity(client, pool_address: str, min_sol: float = 10.0) -> bool:
"""Ensure minimum SOL (or USD) liquidity exists."""
return True
async def check_top_holders(client, mint: str, max_top10_pct: float = 40.0) -> bool:
"""Reject if top 10 holders own too much supply."""
return True
async def is_safe_token(client, mint: str, pool: str) -> bool:
checks = [
await check_mint_authority(client, mint),
await check_freeze_authority(client, mint),
await check_liquidity(client, pool),
await check_top_holders(client, mint),
]
return all(checks)
Additional filters worth adding later:
- LP locked or burned
- No high buy/sell tax (via simulation)
- Social presence / website (optional, slower)
- Contract age or renounced ownership patterns
- Blacklist of known deployer wallets
These filters are where most of the edge lives. Blind sniping is usually negative EV.
Execution: Jupiter Aggregator (Recommended)
Jupiter gives you best-price routing across many Solana DEXs with a clean API.
Basic quote + swap flow:
# executor.py
import httpx
from solders.transaction import VersionedTransaction
from solana.rpc.types import TxOpts
from loguru import logger
JUPITER_QUOTE = "https://quote-api.jup.ag/v6/quote"
JUPITER_SWAP = "https://quote-api.jup.ag/v6/swap"
async def get_quote(input_mint: str, output_mint: str, amount: int, slippage_bps: int = 800):
params = {
"inputMint": input_mint,
"outputMint": output_mint,
"amount": amount,
"slippageBps": slippage_bps,
}
async with httpx.AsyncClient() as client:
resp = await client.get(JUPITER_QUOTE, params=params)
resp.raise_for_status()
return resp.json()
async def execute_swap(quote: dict, user_public_key: str):
payload = {
"quoteResponse": quote,
"userPublicKey": user_public_key,
"wrapAndUnwrapSol": True,
}
async with httpx.AsyncClient() as client:
resp = await client.post(JUPITER_SWAP, json=payload)
resp.raise_for_status()
swap_data = resp.json()
# Deserialize and sign
tx = VersionedTransaction.from_bytes(
bytes(swap_data["swapTransaction"]) # base64 decode first in real code
)
# Sign with keypair and send
# ... full signing + send_raw_transaction logic here
logger.success("Swap submitted")
return tx
Important production details:
- Always simulate the transaction first when possible
- Use priority fees (compute unit price) during congestion
- Handle partial fills and failed transactions cleanly
- Track transaction signatures and confirm finality
Direct Raydium instruction building is possible but more complex and usually worse priced than Jupiter.
Position Sizing & Risk Controls
Never risk more than a small fixed percentage of your hot wallet per trade.
# risk.py
MAX_SOL_PER_TRADE = float(os.getenv("MAX_SOL_PER_TRADE", 0.5))
MAX_OPEN_POSITIONS = int(os.getenv("MAX_OPEN_POSITIONS", 3))
MAX_DAILY_LOSS_SOL = 2.0
class RiskManager:
def __init__(self):
self.open_positions = 0
self.daily_pnl = 0.0
def can_open_trade(self, sol_amount: float) -> bool:
if sol_amount > MAX_SOL_PER_TRADE:
return False
if self.open_positions >= MAX_OPEN_POSITIONS:
return False
if self.daily_pnl < -MAX_DAILY_LOSS_SOL:
return False
return True
def record_open(self):
self.open_positions += 1
def record_close(self, pnl: float):
self.open_positions = max(0, self.open_positions - 1)
self.daily_pnl += pnl
Additional hard rules many serious bots use:
- Maximum trades per hour
- Cooldown after a loss streak
- Automatic pause if RPC latency exceeds threshold
- Kill switch via Telegram command
Main Bot Loop
# bot.py
import asyncio
from loguru import logger
from risk import RiskManager
class SniperBot:
def __init__(self):
self.risk = RiskManager()
self.running = True
async def run(self):
logger.info("Sniper bot starting...")
# Start listener in background
listener_task = asyncio.create_task(listen_for_new_pools())
while self.running:
try:
# In real implementation the listener pushes candidates
# into an asyncio.Queue that we consume here
candidate = await self.get_next_candidate()
if candidate is None:
await asyncio.sleep(0.5)
continue
if not await is_safe_token(client, candidate["mint"], candidate["pool"]):
logger.info(f"Rejected unsafe token: {candidate['mint']}")
continue
sol_amount = MAX_SOL_PER_TRADE
if not self.risk.can_open_trade(sol_amount):
logger.warning("Risk limits reached — skipping")
continue
# Get quote and execute
quote = await get_quote(
input_mint="So11111111111111111111111111111111111111112", # SOL
output_mint=candidate["mint"],
amount=int(sol_amount * 1e9),
)
await execute_swap(quote, str(keypair.pubkey()))
self.risk.record_open()
logger.success(f"Entered {candidate['mint']}")
except Exception as e:
logger.exception(e)
await asyncio.sleep(2)
listener_task.cancel()
Logging, Alerts & Monitoring
Use structured logging and push critical events to Telegram:
from loguru import logger
import sys
logger.remove()
logger.add(sys.stdout, level="INFO")
logger.add("logs/sniper_{time}.log", rotation="50 MB", retention="7 days")
Telegram helper for fills, rejects, and errors is essential. You should also track:
- RPC latency
- Success vs failure rate of swaps
- Average entry slippage
- Daily realized P&L
Deployment on a VPS
Same principles as any production trading bot:
- Dedicated VPS close to your RPC provider when possible
- Docker or systemd
- Automatic restart
- Log rotation
- Separate hot wallet with limited SOL
Docker example is almost identical to the previous CCXT article — just change the entrypoint and environment variables.
During major memecoin launches, expect RPC and network congestion. Build in graceful degradation (pause new entries when latency spikes).
Realistic Expectations & Next Steps
A basic sniper that only checks mint/freeze authority and minimum liquidity will still buy many losers. The difference between break-even and profitable usually comes from:
- Better filters (holder distribution, LP lock, tax simulation, deployer history)
- Position sizing and exit logic (trailing stops, time-based exits, partial takes)
- Capital discipline
- Continuous monitoring
Speed helps, but filters and risk management matter more.
The bot architecture in this article gives you a solid, extensible foundation. For a complete memecoin strategy layer — including entry rules, exit frameworks, and practical risk parameters — see the dedicated resource: https://selar.com/60lw5u0623.
Final Checklist Before Going Live
- [ ] Hot wallet only, limited funds
- [ ] All safety filters implemented and tested
- [ ] Simulation / dry-run mode works
- [ ] Telegram alerts for entries, exits, and errors
- [ ] Circuit breaker and daily loss limit active
- [ ] RPC and WebSocket reconnection logic tested
- [ ] Logs are being written and rotated
- [ ] You understand you can (and probably will) lose the entire hot-wallet balance
Start extremely small. Measure everything. Iterate on filters before increasing size.
Happy building — and stay careful out there.
Useful resources
Top comments (0)