How to Build a Pump.fun Bot with CoinMarketCap API (Python Guide)

pip install requests pandas numpy websocket-client

# .env, or your shell profile. Never commit the key.
CMC_API_KEY=your_key_here
CMC_BASE_URL=https://pro-api.coinmarketcap.com

import json
import os
import time
from typing import Any, Dict, List

import numpy as np
import pandas as pd
import requests

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

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

{
  "timestamp": 1790750998137,
  "path": "/v1/dex/new/list",
  "status": 405,
  "error": "Method Not Allowed",
  "requestId": "e11461cd-20"
}

class CMCError(RuntimeError):
    """Any CoinMarketCap failure, whichever envelope it arrived in."""


def cmc_payload(response: requests.Response) -> Dict[str, Any]:
    try:
        payload = response.json()
    except ValueError:
        raise CMCError(
            f"HTTP {response.status_code}, non-JSON body: {response.text[:200]}"
        )

    status = payload.get("status")

    # Gateway-level errors (405, some 5xx) have no CMC status object at all.
    if not isinstance(status, dict):
        detail = payload.get("error") or response.text[:200]
        path = payload.get("path") or response.url
        raise CMCError(f"HTTP {response.status_code} {detail} on {path}")

    # error_code is "0" as a string on some routes and 0 as a number on others.
    if str(status.get("error_code", "0")) != "0":
        raise CMCError(
            f"CMC {status.get('error_code')}: {status.get('error_message')}"
        )

    return payload

def platform_ids() -> Dict[str, int]:
    """Chain name to platform ID. Keyless."""
    response = requests.get(
        f"{CMC_KEYLESS}/v1/dex/platform/list",
        headers={"Accept": "application/json"},
        timeout=30,
    )
    rows = cmc_payload(response)["data"]
    # DEX responses abbreviate field names: "n" is the chain name.
    return {row["n"]: row["id"] for row in rows}


SOLANA_PLATFORM_ID = 16

def show_fields(rows: List[Dict[str, Any]], limit: int = 3) -> None:
    """Print the field names and a sample row, so the mapping is not guesswork."""
    if not rows:
        print("no rows returned")
        return
    print("fields:", ", ".join(sorted(rows[0])))
    print(json.dumps(rows[:limit], indent=2)[:2000])

def to_frame(rows: List[Dict[str, Any]], mapping: Dict[str, str]) -> pd.DataFrame:
    """
    Build a DataFrame from {our_column: api_field}.

    Any mapped field missing from the response raises, rather than producing a
    column of None that silently filters every candidate out.
    """
    if not rows:
        return pd.DataFrame(columns=list(mapping))

    missing = [api for api in mapping.values() if api not in rows[0]]
    if missing:
        raise CMCError(
            "response is missing mapped fields: " + ", ".join(missing)
            + " | fields present: " + ", ".join(sorted(rows[0]))
        )

    return pd.DataFrame(
        [{col: row.get(api) for col, api in mapping.items()} for row in rows]
    )

def fetch_new_tokens(body: Dict[str, Any]) -> List[Dict[str, Any]]:
    """
    POST /v1/dex/new/list. Requires a key on a tier that includes DEX discovery.

    Build body from the request schema for this endpoint in the CoinMarketCap
    endpoint reference. The DEX family is not uniform about snake_case versus
    camelCase between routes, so take the key names from the reference for your
    API version rather than from any guide, this one included.
    """
    response = requests.post(
        f"{CMC_BASE_URL}/v1/dex/new/list",
        headers={**HEADERS, "Content-Type": "application/json"},
        json=body,
        timeout=30,
    )
    return cmc_payload(response)["data"]

raw = fetch_new_tokens({"limit": 20, "platformId": SOLANA_PLATFORM_ID})
show_fields(raw)

CANDIDATE_FIELDS = {
    "token_address": "addr",
    "symbol": "s",
    "name": "n",
    "platform_id": "pltId",
    "price_usd": "pu",
    "volume_24h": "v24h",
    "market_cap": "mc",
    "liquidity_usd": "liq",
    "created_at": "pt",
}

candidates = to_frame(raw, CANDIDATE_FIELDS)

def prepare_candidates(df: pd.DataFrame) -> pd.DataFrame:
    now = pd.Timestamp.now(tz="UTC")
    df = df.copy()

    # Several DEX routes return epoch milliseconds as a STRING. to_numeric
    # first, then to_datetime, so a string timestamp does not become NaT.
    created = pd.to_numeric(df["created_at"], errors="coerce")
    df["created_at"] = pd.to_datetime(created, unit="ms", utc=True, errors="coerce")

    df["age_minutes"] = (now - df["created_at"]).dt.total_seconds() / 60

    for column in ("price_usd", "volume_24h", "market_cap", "liquidity_usd"):
        df[column] = pd.to_numeric(df[column], errors="coerce")

    df["liquidity_to_mcap"] = df["liquidity_usd"] / df["market_cap"].replace(0, np.nan)
    return df


def filter_candidates(df: pd.DataFrame) -> pd.DataFrame:
    return df[
        (df["age_minutes"] <= 10)
        & (df["liquidity_usd"] >= 5000)
        & (df["market_cap"] >= 2500)
        & (df["liquidity_to_mcap"] >= 0.05)
    ]

import websocket

CMC_STREAM_URL = "wss://pro-stream.coinmarketcap.com/v1"


def stream_tokens(watchlist: List[Dict[str, Any]], on_event) -> None:
    """
    Subscribe to per-swap price events for every token in the watchlist.

    watchlist entries are {"platform_id": int, "address": str}.
    on_event(channel, payload) is called for each push.
    """

    def on_open(ws: "websocket.WebSocketApp") -> None:
        for index, token in enumerate(watchlist, start=1):
            ws.send(json.dumps({
                "id": index,
                "method": "subscribe",
                "channel": "onchain@token_agg_event",
                "params": {
                    "platform_id": token["platform_id"],
                    "address": token["address"],
                },
            }))

    def on_message(ws: "websocket.WebSocketApp", raw: str) -> None:
        message = json.loads(raw)
        on_event(message.get("channel"), message)

    def on_error(ws: "websocket.WebSocketApp", error: Exception) -> None:
        print("stream error:", error)

    app = websocket.WebSocketApp(
        CMC_STREAM_URL,
        header=[f"X-CMC_PRO_API_KEY: {CMC_API_KEY}"],
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
    )
    # reconnect_delay: the stream is in beta and excluded from the SLA, so
    # assume disconnects and let the client re-establish rather than exiting.
    app.run_forever(reconnect=5, ping_interval=20)

def fetch_token_price(platform_id: int, address: str) -> Dict[str, Any]:
    """
    GET /v1/dex/token/price. Confirm the exact parameter names against the
    endpoint reference for your API version before relying on this in
    production; the DEX routes differ between versions.
    """
    response = requests.get(
        f"{CMC_BASE_URL}/v1/dex/token/price",
        headers=HEADERS,
        params={"platform_id": platform_id, "address": address},
        timeout=15,
    )
    return cmc_payload(response)["data"]


def poll_token(platform_id: int, address: str, entry_price: float,
               take_profit: float = 0.20, stop_loss: float = -0.08,
               poll_seconds: int = 30, max_failures: int = 6) -> str:
    """Poll one token until a rule fires or the failure budget runs out."""
    failures = 0

    while failures < max_failures:
        try:
            data = fetch_token_price(platform_id, address)
        except CMCError as error:
            failures += 1
            backoff = min(poll_seconds * (2 ** failures), 900)
            print(f"{error} | backing off {backoff}s")
            time.sleep(backoff)
            continue

        # data may be a dict or a single-item list depending on the route.
        record = data[0] if isinstance(data, list) and data else data
        price = record.get("pu") if isinstance(record, dict) else None
        if price is None:
            failures += 1
            time.sleep(poll_seconds)
            continue

        failures = 0
        change = (float(price) - entry_price) / entry_price

        if change >= take_profit:
            return "take_profit"
        if change <= stop_loss:
            return "stop_loss"

        time.sleep(poll_seconds)

    return "gave_up"

def record_outcome(signals: List[Dict[str, Any]], row: pd.Series,
                   outcome: str, exit_price: float) -> None:
    signals.append({
        "observed_at": pd.Timestamp.now(tz="UTC").isoformat(),
        "token_address": row["token_address"],
        "symbol": row["symbol"],
        "age_minutes_at_entry": row["age_minutes"],
        "liquidity_usd_at_entry": row["liquidity_usd"],
        "liquidity_to_mcap_at_entry": row["liquidity_to_mcap"],
        "entry_price": row["price_usd"],
        "exit_price": exit_price,
        "outcome": outcome,
    })

def run_once(discovery_body: Dict[str, Any], top_n: int = 5) -> pd.DataFrame:
    raw = fetch_new_tokens(discovery_body)
    candidates = to_frame(raw, CANDIDATE_FIELDS)
    prepared = prepare_candidates(candidates)
    shortlist = filter_candidates(prepared)
    return shortlist.sort_values("liquidity_usd", ascending=False).head(top_n)


if __name__ == "__main__":
    watchlist = run_once({"limit": 50, "platformId": SOLANA_PLATFORM_ID})
    print(watchlist[["symbol", "age_minutes", "liquidity_usd", "liquidity_to_mcap"]])

    signals: List[Dict[str, Any]] = []
    for _, row in watchlist.iterrows():
        outcome = poll_token(
            platform_id=int(row["platform_id"]),
            address=row["token_address"],
            entry_price=float(row["price_usd"]),
        )
        record_outcome(signals, row, outcome, exit_price=float(row["price_usd"]))

    pd.DataFrame(signals).to_csv("signals.csv", index=False)