How to Build a Raydium Sniper Bot with CoinMarketCap API

Introduction

On Solana, new tokens launch every minute.

Raydium is where most of them land. Via LaunchLab's bonding curve, tokens graduate to CPMM pools automatically when the SOL threshold is reached. The window between pool creation and widespread discovery is where sniper bots operate.

Execution speed matters. But signal quality matters more.

Most sniper bots fail not because they are slow, but because they enter every pool without filtering. The result is consistent losses to honeypots, rugs, and high-tax tokens.

CoinMarketCap API solves the signal problem. It gives you a structured data layer to discover new token launches, validate pool liquidity, detect contract risks, and assess momentum — before committing capital.

• CoinMarketCap API powers the signal engine
• Raydium SDK handles on-chain swap execution

Why Use CoinMarketCap API for a Raydium Sniper Bot?

Raydium gives you execution. CoinMarketCap gives you context.

Instead of blindly sniping every new pool, you can use CoinMarketCap to:

• discover tokens the moment they launch on Solana DEXs
• validate pool liquidity and LP burn/lock status
• detect honeypots, transfer restrictions, and hidden taxes
• measure transaction activity and whale entry signals
• track sub-minute price momentum on new pools
• backtest snipe thresholds against historical candle data

This turns your bot from a blind executor into a decision-making system.

System Architecture

CoinMarketCap DEX API (Signal Layer)

├─ New Token Discovery

├─ Pool Validation (liquidity, LP burn/lock)

├─ Security Checks (honeypot, taxes)

├─ Transaction Flow (whale detection)

└─ Sub-minute Candles (momentum)
↓

Signal Engine

↓

Raydium SDK (Execution Layer)

↓

Swap on Solana (CPMM / AMM V4)

Architecture Clarification

CoinMarketCap API acts strictly as an off-chain Signal Layer for new token discovery, liquidity validation, and risk filtering on Raydium/Solana. It is not an on-chain execution layer.

Pool creation events, LP burn status, real-time reserves, and transaction finality must be validated directly on-chain via Solana RPC before executing any snipe. CMC data has cache delays that make it unsuitable as the sole trigger for latency-sensitive sniper execution.

Project Setup

Python Dependencies

import os
import time
import requests
import pandas as pd
import numpy as np

Environment Variables

CMC_API_KEY = os.getenv("CMC_API_KEY")
CMC_BASE_URL = "https://pro-api.coinmarketcap.com"

Headers

HEADERS = {
"Accept": "application/json",
"Content-Type": "application/json",
"X-CMC_PRO_API_KEY": CMC_API_KEY,
}

Step 1: Resolve the Solana Platform ID

Before querying any DEX discovery endpoint, resolve Solana's internal platform ID. Never hardcode it.

Endpoint

GET /v1/dex/platform/list

def get_solana_platform_id():
url = f"{CMC_BASE_URL}/v1/dex/platform/list"

r = requests.get(url, headers=HEADERS)

r.raise_for_status()

for p in r.json()["data"]:
if p["n"].lower() == "solana":  # field is "n", not "name"
return str(p["id"])

The platform name field is "n", not "name". Using p["name"] raises a KeyError. Use the returned numeric ID as the platformIds value in all POST discovery calls.

Step 2: Get Raydium DEX Slugs

For a Solana sniper bot, the known Raydium slugs can be hardcoded. A latency-sensitive bot should not waste HTTP calls mapping static exchange names on every run.

# Known Raydium slugs on Solana — hardcode for production bots

SOLANA_DEX_SLUGS = ["raydium", "orca", "meteora"]

If you need to discover slugs dynamically without hitting paid or unstable endpoints, extract them from active pair data instead:

def discover_solana_dex_slugs():
# /v4/dex/spot-pairs/latest returns dex_slug per pair -

# build a set of active DEXs from the response at no extra cost
url = f"{CMC_BASE_URL}/v4/dex/spot-pairs/latest"
params = {"network_slug": "solana"}
r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()
return list({
pair["dex_slug"]
for pair in r.json()["data"]
if pair.get("dex_slug")
})

This reuses a call you are already making for discovery, extracts dex_slug from each DexSpotPairDTO, and builds a live directory of active DEXs on Solana with no additional credits spent.

Endpoint Notes

• /v4/dex/listings/quotes returns 500 in production.
• /v1/exchange/listings/latest?category=dex returns 403 on the Basic plan.
• Neither should be used.
• Use hardcoded slugs or the dynamic discovery above.

Step 3: Discover New Token Launches

Poll for tokens that recently graduated from LaunchLab or launched directly as new Raydium pools.

Endpoints

POST /v1/dex/new/list

POST /v1/dex/meme/list

Paid Endpoint Warning

Both endpoints return HTTP 403 on the Basic plan.

Error Code: 1006 [API_KEY_PLAN_NOT_AUTHORIZED]
Message: "Your API Key subscription plan doesn't support this endpoint."

def fetch_new_solana_tokens(platform_id):
url = f"{CMC_BASE_URL}/v1/dex/new/list

body = {
"platformIds": platform_id,
"limit": 50
}

r = requests.post(url, headers=HEADERS, json=body)
r.raise_for_status()

return r.json()["data"]

Key fields to extract:

• addr — SPL token mint address
• name / symbol
• liqUsd — liquidity in USD
• v24 — 24h volume
• pubAt — pool publish timestamp

Filter by pubAt to target tokens launched within your recency window, such as the last 10 minutes.

Basic Plan Fallback

Both /v1/dex/new/list and /v1/dex/tokens/trending/list require a paid plan. On the Basic plan, use /v4/dex/spot-pairs/latest filtered by network and sorted by volume:

def fetch_new_pairs_fallback(dex_slug="raydium"):
url = f"{CMC_BASE_URL}/v4/dex/spot-pairs/latest"

params = {
"network_slug": "solana",
"dex_slug":     dex_slug,
"sort":         "volume_24h",
"sort_dir":     "desc"
}

r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()

return r.json()["data"]

This surfaces active pairs but does not isolate brand-new launches. For genuine new-launch detection, a paid plan is required.

Step 4: Validate Pool Liquidity and LP Status

Not every new pool is worth sniping. Low liquidity and unlocked LPs are the primary rug vectors.

Endpoint

GET /v1/dex/token/pools

def fetch_token_pools(token_address):
url = f"{CMC_BASE_URL}/v1/dex/token/pools"

params = {
"address":  token_address,
"platform": "solana"
}

r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()

return r.json()["data"]

Key fields per pool:

• exn — DEX name
• liqUsd — liquidity in USD
• v24 — 24h volume
• addr — pool address
• pubAt — pool creation timestamp
• lr — locked rate, percentage of LP locked
• br — burned rate, percentage of LP burned

A pool where both lr and br equal zero means LP is fully withdrawable at any time.

def is_pool_valid(pool):
liq    = pool.get("liqUsd", 0) or 0
burned = pool.get("br", 0)    or 0
locked = pool.get("lr", 0)    or 0
return liq > 10_000 and (burned + locked) > 50

Step 5: Run Security and Rug Checks

Validate the token contract against known exploit patterns before any capital commitment.

Endpoint

GET /v1/dex/security/detail

def check_token_security(token_address):
url = f"{CMC_BASE_URL}/v1/dex/security/detail"

params = {
"platformName": "solana",
"address":      token_address
}

r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()

results = r.json()["data"]        # data is a list

return results[0] if results else {}

The response object has two separate structures:

• securityItems — a list of risk objects, each with riskCode and isHit. Boolean flags like honeypot and transfer_pausable live here.
• extraInfo, or root-level fields — numeric tax values buyTax and sellTax live here as strings, not inside securityItems.

For newly launched tokens not yet scanned by Go+, securityItems may be absent or empty. Always parse defensively:

def parse_security_flags(sec):

# Boolean flags — iterate securityItems, match by riskCode
items = sec.get("securityItems") or []
risk_map = {
item["riskCode"]: item.get("isHit", False)

for item in items
if item.get("riskCode")
}

honeypot          = risk_map.get("honeypot", False)
transfer_pausable = risk_map.get("transfer_pausable", False)

# Numeric taxes — live at root level, not inside securityItems
buy_tax  = 0
sell_tax = 0

try:
buy_tax  = float(sec.get("buyTax",  0) or 0)
sell_tax = float(sec.get("sellTax", 0) or 0)

except (TypeError, ValueError):
pass

return {
"honeypot":          honeypot,
"transfer_pausable": transfer_pausable,
"buy_tax":           buy_tax,
"sell_tax":          sell_tax,
"combined_tax":      buy_tax + sell_tax,
}

For Solana tokens, buyTax and sellTax may not be populated at the root level depending on scan coverage. When this happens, both values return 0 silently. The try/except prevents a crash, but tax filtering will not activate. Treat tax scores as best-effort on Solana and rely on honeypot and transfer_pausable flags as the primary hard rejection criteria.

Never call sec.get("honeypot") at the top level. The field does not exist there and returns None silently, allowing honeypot tokens to pass the filter undetected.

Step 6: Analyse Transaction Flow and Detect Whales

Understand who is trading before entering a position.

Endpoint

GET /v1/dex/tokens/transactions

def fetch_transactions(token_address, min_volume=10_000):
url = f"{CMC_BASE_URL}/v1/dex/tokens/transactions"

params = {
"address":   token_address,
"platform":  "solana",   # "solana" only — "sol" returns 400
"minVolume": min_volume
}

r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()

return r.json()["data"]

This endpoint returns raw swap records, not aggregated counts. Use minVolume in USD to isolate whale-sized entries.

Key fields per record:

• tx — transaction hash
• v — volume in USD
• t0a — base asset address
• t1a — quote asset address

Step 7: Get Real Pool Price and Reserves

For execution decisions, always use pool-level pricing. Never use global VWAP for sniper decisions.

Endpoint

GET /v4/dex/pairs/quotes/latest

def fetch_pool_quote(pool_address, network_slug="solana"):
url = f"{CMC_BASE_URL}/v4/dex/pairs/quotes/latest"

params = {
"network_slug":     network_slug,    # required — disambiguates contract across chains
"contract_address": pool_address,
"aux": "pool_base_asset,pool_quote_asset,buy_tax,sell_tax"
}

r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()

return r.json()["data"]

network_slug is required alongside contract_address. Contract addresses can collide across chains, so the API uses the network slug to disambiguate. Passing contract_address alone returns a 400 error.

Cache Warning

This endpoint has an approximately 60-second cache. Do not use it as a millisecond price oracle for execution timing. Validate reserves directly on-chain via Solana RPC immediately before submitting your swap transaction.

Step 8: Detect Momentum with Sub-Minute Candles

Sub-minute candle data is available for fast breakout detection on newly launched tokens. This is also the recommended endpoint for historical backtesting on Solana pairs.

Endpoint

GET /v1/k-line/candles

def fetch_candles(token_address, interval="1min"):
url = f"{CMC_BASE_URL}/v1/k-line/candles"

params = {
"platform": "solana",
"address":  token_address,
"interval": interval  # 1s, 5s, 30s, 1min, 3min
}

r = requests.get(url, headers=HEADERS, params=params)
r.raise_for_status()

return r.json()["data"]

Each candle is a positional array of 7 elements — not a dict.

The exact index order is:

Index

Field

[0]

open

[1]

high

[2]

low

[3]

close

[4]

volume

[5]

timestamp, UNIX seconds

[6]

traders, unique trader count

Do not call .get() on a candle. Use parse_candle() to convert to a named dict:

def parse_candle(c):
return {
"open":      c[0],
"high":      c[1],
"low":       c[2],
"close":     c[3],
"volume":    c[4],
"timestamp": c[5],
"traders":   c[6],
}

Use "1s" or "5s" intervals for the fastest breakout detection.

Note: /v4/dex/pairs/ohlcv/historical returns 500 in production. Use /v1/k-line/candles for all candle data on Solana.

Step 9: Build the Snipe Signal Score

Combine all signals into a structured score before deciding to snipe.

def compute_snipe_score(pool, security_flags, candles):
score = 0

# Liquidity score
liq = pool.get("liqUsd", 0) or 0

if liq > 50_000:   score += 30
elif liq > 10_000: score += 15

# LP safety score
burned = pool.get("br", 0) or 0
locked = pool.get("lr", 0) or 0

if (burned + locked) > 80:   score += 25
elif (burned + locked) > 50: score += 10

# Tax penalty
combined_tax = security_flags.get("combined_tax", 0) or 0

if combined_tax == 0:   score += 25
elif combined_tax < 5: score += 10
else:                   score -= 30

# Momentum score — candles are positional arrays, traders at index [6]

if candles:
latest  = candles[-1]
traders = latest[6] if len(latest) > 6 else 0

if traders > 20:   score += 20
elif traders > 10: score += 10
return score

def should_snipe(score, threshold=60):
return score >= threshold

Step 10: Minimal End-to-End Flow

def run_sniper_scanner(platform_id):
tokens  = fetch_new_solana_tokens(platform_id)

results = []

for token in tokens:
addr = token.get("addr")
if not addr:
continue

# Pool validation
try:
pools = fetch_token_pools(addr)
except Exception:
continue
valid_pools = [p for p in pools if is_pool_valid(p)]

if not valid_pools:
continue
pool = valid_pools[0]

# Security — unwrap list, parse securityItems by riskCode
try:
sec_raw   = check_token_security(addr)
sec_flags = parse_security_flags(sec_raw)
except Exception:
continue

if sec_flags["honeypot"] or sec_flags["transfer_pausable"]:
continue

if sec_flags["combined_tax"] > 10:
continue

# Momentum
try:
candles = fetch_candles(addr, interval="1min")
except Exception:
candles = []
score = compute_snipe_score(pool, sec_flags, candles)

if should_snipe(score):
results.append({
"address":      addr,
"symbol":       token.get("symbol"),
"score":        score,
"pool":         pool.get("addr"),
"liqUsd":       pool.get("liqUsd"),
"combined_tax": sec_flags["combined_tax"],
})

return sorted(results, key=lambda x: -x["score"])

Tokens that pass the score threshold are passed to your Raydium SDK execution layer for the on-chain swap.

Rate Limits and Polling

CoinMarketCap API is REST-only. There is no WebSocket streaming.

Cache Intervals

Endpoint Group

Cache Interval

Discovery endpoints, /v1/dex/new/list, /v1/dex/meme/list

60 seconds

Pool data, /v1/dex/token/pools

60 seconds

Quotes, /v4/dex/pairs/quotes/latest

60 seconds

Candles, /v1/k-line/candles

60 seconds

Best Practices

• poll every 60 seconds for all discovery and quote endpoints
• cache responses locally to avoid redundant calls
• batch token addresses in a single request where possible
• use exponential backoff for HTTP 429 errors
• treat the rate reset as 60 seconds

def request_with_backoff(fn, retries=3, base_delay=2):

for attempt in range(retries):
try:
return fn()
except requests.exceptions.HTTPError as e:

if e.response.status_code == 429:
time.sleep(base_delay ** attempt)
else:
raise

raise Exception("Max retries exceeded")

Common Mistakes

Using p["name"] on platform list

The platform name field is "n", not "name". Using p["name"] raises a KeyError on every platform object.

Calling sec.get("honeypot") at the top level

The honeypot and transfer_pausable flags live inside securityItems[], identified by riskCode. Calling .get() at the root returns None silently, which lets honeypot tokens pass the filter undetected.

Always iterate securityItems and build a risk_map by riskCode.

Calling .get() on candle data

Each candle is a positional array. Calling candle.get("traders") raises an AttributeError.

Use index access instead:

candle[6]

Omitting network_slug from pool quotes

/v4/dex/pairs/quotes/latest requires both network_slug and contract_address. Passing contract_address alone returns a 400 error.

Using "sol" as the platform value

The /v1/dex/tokens/transactions endpoint only accepts "solana". The abbreviated "sol" returns 400 despite appearing in some documentation examples.

Using /v4/dex/listings/quotes or /v4/dex/pairs/ohlcv/historical

Both return 500 in production. Use hardcoded slugs or the dynamic discovery pattern from /v4/dex/spot-pairs/latest, and use /v1/k-line/candles for all candle data.

Using the Core API for new tokens

For tokens launched via LaunchLab in the last few hours, /v1/cryptocurrency/map is too slow. It requires minimum metrics for indexing. Always use the DEX API family for new token discovery on Solana.

No LP check

Sniping a pool with br=0 and lr=0 means LP is fully withdrawable at any time. Always check burned and locked rate before entering.

Final Thoughts

Raydium sniper bots fail for one of two reasons: no filters, or slow filters.

CoinMarketCap API solves the signal problem. It gives you a structured data layer covering new token discovery, liquidity validation, rug detection, transaction flow, and momentum — all before you write a single swap instruction.

The key separation:

• CoinMarketCap identifies what is worth sniping
• Raydium SDK executes the snipe on-chain

Better signals lead to better snipes.

Next Steps

• add minimum trader count thresholds per pool age bucket
• track snipe PnL per signal threshold to refine score weights
• implement auto-sell at configurable take-profit and stop-loss levels
• model Jito bundle priority fees based on pool liquidity depth
• backtest LP burn thresholds against historical rug data using /v1/k-line/candles
• store local snapshots to build rolling signal history over time