How to Get Real-Time Crypto Prices in Python with REST & WebSockets

TL;DR

CoinGecko API supports multiple delivery methods for both aggregated coin prices and onchain DEX token prices through REST, and WebSocket APIs.
REST is best for scheduled, bulk, or historical price retrieval, while WebSocket is best for continuous, lower-latency updates. REST /simple/price costs 1 credit per call and is cached for up to 20s, while WebSocket costs 0.1 credit per price update.

Most real-time crypto price trackers start with a REST API call at a set interval. Each request returns the latest available price, and the next request is made when the interval ends. This works well when periodic updates are sufficient. If you need prices as they update, WebSocket provides a different approach. You open one connection and receive price updates as they arrive, without repeatedly requesting the same data.

REST and WebSocket serve different purposes. This guide shows you how to build both approaches in Python using CoinGecko price data and explains when to use each one.

CoinGecko API offers three data delivery methods: REST polling, WebSocket streaming, and webhook.

Prerequisites & Setup

You’ll need a CoinGecko API key. A free Demo API key is sufficient to access the REST API endpoints. If you don’t have one, follow the guide to get a free Demo API key. The WebSocket implementation requires a Basic plan or higher.

Your API key type must match the base URL and authentication header used in your requests:

Key type	Base URL	Header
Demo	https://api.coingecko.com/api/v3	x-cg-demo-api-key
Paid	https://pro-api.coingecko.com/api/v3	x-cg-pro-api-key

Choose the Price Data You Need

CoinGecko provides two types of price data, with different endpoints for each asset type.

Coins: Aggregated market data for established assets such as Bitcoin, Ethereum, and Solana, identified by Coin API ID (e.g. bitcoin). This is the market price shown on CoinGecko’s site
Onchain DEX tokens: Token prices from decentralized exchanges, identified by network ID, and contract address (e.g. eth:0xc02a…cc2). This is suited to new or long-tail tokens and is powered by GeckoTerminal’s data.

Note: CoinGecko unified its API with the GeckoTerminal API, so standard market data and onchain DEX data are now accessible through a single CoinGecko API.

Both asset types are available through REST and WebSocket. The sections below cover both delivery methods for each asset type.

Python setup

You'll also need Python 3.8 or later and two libraries:

pip install requests websockets

💡 Prefer less setup? You can call these endpoints with CoinGecko's official Python SDK or TypeScript SDK instead of raw HTTP requests. The SDK handles the base URL, headers, and authentication for you, so there is less boilerplate to write.

How to Get Real-Time Crypto Prices in Python with a REST API

The REST approach repeatedly requests data from the CoinGecko API. Use the /simple/price endpoint with the coins and quote currencies to track, then repeat the request at your chosen interval.

Two parameters are relevant here. include_last_updated_at=true returns the UNIX timestamp of the latest price update, which you can use to check data freshness, while ids supports up to 515 coins in a single request.

import os
import time
from datetime import datetime, timezone

import requests

API_KEY = os.environ.get("COINGECKO_API_KEY", "")
BASE_URL = "https://api.coingecko.com/api/v3"
HEADERS = {"x-cg-demo-api-key": API_KEY} if API_KEY else {}

COINS = ["bitcoin", "ethereum", "solana"]  # up to 515 ids in one call
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", 60))  # Demo cache 60s; paid 20s


def fetch_prices(coin_ids):
    """Fetch current prices for several coins in a single request."""
    response = requests.get(
        f"{BASE_URL}/simple/price",
        headers=HEADERS,
        params={
            "ids": ",".join(coin_ids),
            "vs_currencies": "usd",
            "include_24hr_change": "true",
            "include_last_updated_at": "true",
        },
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

    # Official CoinGecko Python SDK equivalent:
    #   pip install coingecko-sdk
    #
    #   from coingecko_sdk import Coingecko
    #   client = Coingecko(demo_api_key=API_KEY, environment="demo")
    #   response = client.simple.price.get(
    #       ids=",".join(coin_ids),
    #       vs_currencies="usd",
    #       include_24hr_change=True,
    #       include_last_updated_at=True,
    #   )


def poll_forever(coin_ids, interval=POLL_INTERVAL):
    """Print a timestamped price line for each coin on every poll."""
    while True:
        prices = fetch_prices(coin_ids)
        polled_at = datetime.now(timezone.utc).strftime("%H:%M:%S")

        for coin_id in coin_ids:
            data = prices.get(coin_id)
            if not data:
                print(f"[{polled_at}] {coin_id:<10} no data returned")
                continue

            price = data.get("usd")
            change = data.get("usd_24h_change")
            updated_at = data.get("last_updated_at")
            age = int(time.time()) - updated_at if updated_at else None

            print(
                f"[{polled_at}] {coin_id:<10} ${price:>12,.2f}  "
                f"{change:+6.2f}% 24h   data age: {age}s"
            )

        print("-" * 68)
        time.sleep(interval)


if __name__ == "__main__":
    poll_forever(COINS)

Here’s what the response looks like:

CoinGecko's /simple/price REST endpoint returns bitcoin, ethereum, and solana prices with 24h change in one call.

Get Real-Time DEX Token Prices via REST

For DEX-traded tokens, use the Token Price by Token Addresses endpoint. It supports price lookups using a network ID and token contract address, including tokens that don't have a CoinGecko coin ID.

You can query up to 100 token addresses per call on the same network, with real-time, cacheless responses. If a token’s contract address is already known, pair it with the correct network ID from the Networks List endpoint. To discover new tokens, see How to Track New Tokens Onchain.

import os
import time
from datetime import datetime, timezone

import requests

API_KEY = os.environ.get("COINGECKO_API_KEY", "")
BASE_URL = "https://api.coingecko.com/api/v3"
HEADERS = {"x-cg-demo-api-key": API_KEY} if API_KEY else {}

NETWORK = "eth"
TOKEN_ADDRESSES = ["0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2"]  # WETH
POLL_INTERVAL = int(os.environ.get("POLL_INTERVAL", 30))


def fetch_token_prices(network, token_addresses):
    """Fetch current onchain token prices for one or more contract addresses."""
    response = requests.get(
        f"{BASE_URL}/onchain/simple/networks/{network}/token_price/{','.join(token_addresses)}",
        headers=HEADERS,
        params={"include_24hr_price_change": "true"},
        timeout=10,
    )
    response.raise_for_status()
    return response.json()

    # Official CoinGecko Python SDK equivalent:
    #   client.onchain.simple.networks.token_price.get_addresses(
    #       ",".join(token_addresses), network=network, include_24hr_price_change=True
    #   )


def poll_forever(network, token_addresses, interval=POLL_INTERVAL):
    """Print a timestamped price line for each token on every poll."""
    while True:
        payload = fetch_token_prices(network, token_addresses)
        attrs = payload["data"]["attributes"]
        polled_at = datetime.now(timezone.utc).strftime("%H:%M:%S")

        for address in token_addresses:
            price = attrs["token_prices"].get(address)
            change = attrs["h24_price_change_percentage"].get(address)
            if price is None:
                print(f"[{polled_at}] {address:<44} no data returned")
                continue

            print(
                f"[{polled_at}] {address:<44} ${float(price):>12,.2f}  "
                f"{float(change):+6.2f}% 24h"
            )

        print("-" * 90)
        time.sleep(interval)


if __name__ == "__main__":
    poll_forever(NETWORK, TOKEN_ADDRESSES)

Here’s what the response looks like:

GeckoTerminal's onchain simple token price endpoint returns a DEX token's price and 24h change by contract address.

How to Stream Crypto Prices in Python with WebSockets

With CoinGecko’s WebSocket, a persistent connection remains open to receive price updates as they become available, without continuously polling for new data. The CGSimplePrice channel streams the same aggregated coin prices available through /simple/price. CoinGecko sends an update only when the price changes, so each message contains a new price rather than a repeated value. For large-cap, actively traded coins, updates can arrive as frequently as every ~10 seconds.

The CGSimplePrice channel can be tested directly in the CoinGecko WebSocket docs. For other WebSocket URLs, WebSocket King provides a general-purpose WebSocket client for testing connections.

Example in the CoinGecko docs:

CGSimplePrice channel can be tested directly in the CoinGecko WebSocket docs

Subscribing to the CGSimplePrice channel

Connecting to the WebSocket involves three steps, with the subscription message requiring the correct format and parameters:

Connect to wss://stream.coingecko.com/v1?x_cg_pro_api_key=YOUR_KEY. The server sends two greeting messages, a connection acknowledgement and a welcome message. You can also pass the API key in the x-cg-pro-api-key header to keep it out of connection logs.
Subscribe to the channel. The identifier value must be a JSON string nested inside the JSON message, not a nested object: "identifier": "{\"channel\":\"CGSimplePrice\"}". Sending it as an object will prevent the subscription from working as expected.
Send a set_tokens message specifying the coins to track.

import asyncio
import json
import os
from datetime import datetime, timezone

import websockets

API_KEY = os.environ.get("COINGECKO_API_KEY", "")
STREAM_URL = f"wss://stream.coingecko.com/v1?x_cg_pro_api_key={API_KEY}"

COINS = ["bitcoin", "ethereum", "solana"]
VS_CURRENCIES = ["usd"]

# The identifier is a JSON string nested inside the JSON message, not an object.
CHANNEL = json.dumps({"channel": "CGSimplePrice"})

# CGSimplePrice payloads use abbreviated keys.
FIELDS = {
    "i": "coin_id",
    "vs": "vs_currency",
    "p": "price",
    "pp": "price_24h_change_percentage",
    "t": "last_updated_at",
}


def format_update(payload):
    """Turn an abbreviated payload into a readable line."""
    coin_id = payload.get("i", "unknown")
    price = payload.get("p")
    change = payload.get("pp")
    received_at = datetime.now(timezone.utc).strftime("%H:%M:%S")

    # Any field can be null when data is unavailable.
    price_text = f"${price:>12,.2f}" if price is not None else f"{'no price':>13}"
    change_text = f"{change:+6.2f}%" if change is not None else "     -"

    return f"[{received_at}] {coin_id:<10} {price_text}  {change_text} 24h"


async def stream_prices():
    async with websockets.connect(STREAM_URL) as socket:
        # 1. The server greets us before we subscribe to anything.
        print(await socket.recv())
        print(await socket.recv())

        # 2. Subscribe to the channel.
        await socket.send(json.dumps({"command": "subscribe", "identifier": CHANNEL}))
        print(await socket.recv())

        # 3. Tell the channel which coins to stream.
        await socket.send(
            json.dumps(
                {
                    "command": "message",
                    "identifier": CHANNEL,
                    "data": json.dumps(
                        {
                            "coin_id": COINS,
                            "vs_currencies": VS_CURRENCIES,
                            "action": "set_tokens",
                        }
                    ),
                }
            )
        )

        # 4. Read updates as the server pushes them.
        async for raw_message in socket:
            message = json.loads(raw_message)

            if message.get("type") == "ping":
                continue  # informational heartbeat; safe to ignore
            if message.get("c") == "C1":
                print(format_update(message))
            elif "message" in message:
                print(f"[server] {message['message']}")


if __name__ == "__main__":
    asyncio.run(stream_prices())

Running this script produces output like the following:

Running ws_stream.py opens a terminal ready to stream live CoinGecko WebSocket price updates in Python.

Stream Real-Time DEX Token Prices via WebSocket

OnchainSimpleTokenPrice uses the same connection setup, with only the lookup and identifier differing.

The lookup uses network_id:token_address instead of a coin ID, with n (network) and ta (token address) replacing i (coin ID). OnchainSimpleTokenPrice and OnchainOHLCV both provide ~1-second updates for actively traded tokens and pools. OnchainTrade streams individual pool trades at ~0.1-second intervals.

import asyncio
import json
import os
from datetime import datetime, timezone

import websockets

API_KEY = os.environ.get("COINGECKO_API_KEY", "")
STREAM_URL = f"wss://stream.coingecko.com/v1?x_cg_pro_api_key={API_KEY}"

NETWORK = "eth"
TOKEN_ADDRESSES = {
    "0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2": "WETH",
    "0x6982508145454ce325ddbe47a25d4ec3d2311933": "PEPE",
    "0x95ad61b0a150d79219dcf64e1e6cc01f0b64c4ce": "SHIB",
    "0x514910771af9ca656af840dff83e8264ecf986ca": "LINK",
}

CHANNEL = json.dumps({"channel": "OnchainSimpleTokenPrice"})


def format_update(payload):
    """Turn an abbreviated payload into a readable line."""
    address = payload.get("ta", "unknown")
    symbol = TOKEN_ADDRESSES.get(address, address)
    price = payload.get("p")
    change = payload.get("pp")
    received_at = datetime.now(timezone.utc).strftime("%H:%M:%S")

    # Any field can be null when data is unavailable.
    price_text = f"${price:>12,.6f}" if price is not None else f"{'no price':>13}"
    change_text = f"{change:+6.2f}%" if change is not None else "     -"

    return f"[{received_at}] {symbol:<10} {price_text}  {change_text} 24h"


async def stream_dex_prices():
    async with websockets.connect(STREAM_URL) as socket:
        # 1. The server greets us before we subscribe to anything.
        print(await socket.recv())
        print(await socket.recv())

        # 2. Subscribe to the channel.
        await socket.send(json.dumps({"command": "subscribe", "identifier": CHANNEL}))
        print(await socket.recv())

        # 3. Tell the channel which tokens to stream.
        await socket.send(
            json.dumps(
                {
                    "command": "message",
                    "identifier": CHANNEL,
                    "data": json.dumps(
                        {
                            "network_id:token_addresses": [
                                f"{NETWORK}:{address}" for address in TOKEN_ADDRESSES
                            ],
                            "action": "set_tokens",
                        }
                    ),
                }
            )
        )

        # 4. Read updates as the server pushes them.
        async for raw_message in socket:
            message = json.loads(raw_message)

            if message.get("type") == "ping":
                continue  # informational heartbeat; safe to ignore
            if message.get("c") == "C1":
                print(format_update(message))
            elif "message" in message:
                print(f"[server] {message['message']}")


if __name__ == "__main__":
    asyncio.run(stream_dex_prices())

Running this script produces output like the following:

Running ws_dex_stream.py opens a terminal ready to stream real-time DEX token prices via WebSocket.

How to Build a Live Candlestick Chart with Onchain OHLCV Streams

The CoinGecko OnchainOHLCV channel streams OHLCV data for a pool to power live candlestick charts. Updates arrive at ~1-second intervals for actively traded pools. Each message updates the current candle until the interval closes, then a new candle starts at the next timestamp.

The connection and subscription steps follow the same pattern as the other onchain channels, with two additional parameters: interval (for example, 1m or 1h) and token (base or quote) to specify which side of the pool to chart.

import asyncio
import json
import os
from collections import deque
from datetime import datetime, timezone

import websockets

API_KEY = os.environ.get("COINGECKO_API_KEY", "")  # a Basic-plan (paid) key
STREAM_URL = f"wss://stream.coingecko.com/v1?x_cg_pro_api_key={API_KEY}"

NETWORK = "eth"
POOL_ADDRESS = "0x88e6a0c2ddd26feeb64f039a2c41296fcb3f5640"  # WETH/USDC
INTERVAL = "1m"
CHANNEL = json.dumps({"channel": "OnchainOHLCV"})

MAX_CANDLES = 60
candles = deque(maxlen=MAX_CANDLES)  # ring buffer feeding the chart


def render(candle):
    """Print one candle as a colored line; swap this for a real chart library."""
    color = "up" if candle["c"] >= candle["o"] else "down"
    candle_time = datetime.fromtimestamp(candle["t"], tz=timezone.utc).strftime("%H:%M:%S")
    print(
        f"[{color}] {candle_time} O:{candle['o']:.4f} H:{candle['h']:.4f} "
        f"L:{candle['l']:.4f} C:{candle['c']:.4f} V:{candle['v']:.2f}"
    )


async def stream_candles():
    async with websockets.connect(STREAM_URL) as socket:
        # 1. The server greets us before we subscribe to anything.
        print(await socket.recv())
        print(await socket.recv())

        # 2. Subscribe to the channel.
        await socket.send(json.dumps({"command": "subscribe", "identifier": CHANNEL}))
        print(await socket.recv())

        # 3. Tell the channel which pool to stream.
        await socket.send(
            json.dumps(
                {
                    "command": "message",
                    "identifier": CHANNEL,
                    "data": json.dumps(
                        {
                            "network_id:pool_addresses": [f"{NETWORK}:{POOL_ADDRESS}"],
                            "interval": INTERVAL,
                            "token": "base",
                            "action": "set_pools",
                        }
                    ),
                }
            )
        )

        # 4. Read candle updates as the server pushes them.
        async for raw_message in socket:
            message = json.loads(raw_message)

            if message.get("ch") != "G3":
                continue
            # Replace the in-progress candle, or start a new one when the timestamp changes.
            if candles and candles[-1]["t"] == message["t"]:
                candles[-1] = message
            else:
                candles.append(message)
            render(message)


if __name__ == "__main__":
    asyncio.run(stream_candles())

Here’s what the response looks like:

A successful CoinGecko WebSocket connection returns a code 3000 message confirming the session is established.

Replace render() with a charting library such as Lightweight Charts or mplfinance to display the candlestick chart and update it with each message.

A live OnchainOHLCV candlestick chart streams real-time WETH/USDC 1-minute price data via WebSocket.

How to Keep a WebSocket Connection Open in Python

For a stream that runs continuously, the connection needs to handle interruptions such as deployments, reboots, and network issues.

Ping/pong is handled automatically: CoinGecko’s server sends a ping every 10 seconds and closes the connection if no pong is received within 20 seconds. The websockets library responds automatically, so no manual heartbeat loop is needed.
Reconnect with exponential backoff and jitter: When multiple connections close at the same time, backoff spreads reconnection attempts, while jitter prevents clients from retrying simultaneously.
Re-subscribe after reconnecting: Each new WebSocket connection starts without the subscriptions from the previous connection, so the subscription must be sent again after reconnecting.
Handle clean connection closes: A WebSocket can close without raising an error. For example, CoinGecko may close the connection during a planned deployment. In this case, Python exits the async for loop normally instead of raising an exception, so reconnect logic inside except will not run. Handle the connection closing explicitly so the client reconnects with the same backoff used for other disconnections.

To keep the CoinGecko WebSocket price stream running across disconnections, handle reconnection and re-subscription as follows:

import asyncio
import json
import os
import random
import time
from datetime import datetime, timezone

import websockets

API_KEY = os.environ.get("COINGECKO_API_KEY", "")
STREAM_URL = f"wss://stream.coingecko.com/v1?x_cg_pro_api_key={API_KEY}"

COINS = ["bitcoin", "ethereum", "solana"]
CHANNEL = json.dumps({"channel": "CGSimplePrice"})

BASE_DELAY = 1  # seconds
MAX_DELAY = 60  # cap, so backoff never grows unbounded
STABLE_AFTER = 30  # a connection lasting this long counts as healthy


def log(message):
    stamp = datetime.now(timezone.utc).strftime("%H:%M:%S")
    print(f"[{stamp}] {message}")


async def subscribe(socket):
    """Re-establish the subscription. A new connection has none."""
    await socket.send(json.dumps({"command": "subscribe", "identifier": CHANNEL}))
    await socket.send(
        json.dumps(
            {
                "command": "message",
                "identifier": CHANNEL,
                "data": json.dumps(
                    {
                        "coin_id": COINS,
                        "vs_currencies": ["usd"],
                        "action": "set_tokens",
                    }
                ),
            }
        )
    )


async def consume(socket):
    """Print updates until the server stops sending them."""
    async for raw_message in socket:
        message = json.loads(raw_message)

        if message.get("type") == "ping":
            continue  # informational heartbeat; safe to ignore
        if message.get("c") == "C1":
            price = message.get("p")
            price_text = f"${price:,.2f}" if price is not None else "no price"
            log(f"{message.get('i', 'unknown'):<10} {price_text}")
        elif message.get("message"):
            log(f"server: {message['message']}")


async def stream_with_reconnect():
    attempt = 0

    while True:
        connected_at = time.monotonic()

        try:
            # websockets answers the server's pings automatically, so there is
            # no heartbeat code to write here.
            async with websockets.connect(STREAM_URL) as socket:
                log("connected")
                await subscribe(socket)
                await consume(socket)

            # Reaching here means the server closed the connection cleanly.
            # That is not an exception, so it must be handled explicitly -
            # otherwise the loop reconnects instantly and spins.
            reason = "closed by server"

        except (websockets.exceptions.WebSocketException, OSError) as error:
            reason = type(error).__name__

        uptime = time.monotonic() - connected_at

        # Only treat the connection as healthy if it actually stayed up.
        # A socket that closes immediately every time is still failing.
        attempt = 1 if uptime >= STABLE_AFTER else attempt + 1

        delay = min(BASE_DELAY * 2 ** (attempt - 1), MAX_DELAY)
        delay += random.uniform(0, delay * 0.1)  # jitter

        log(f"disconnected after {uptime:.1f}s ({reason}) - retry {attempt} in {delay:.1f}s")
        await asyncio.sleep(delay)


if __name__ == "__main__":
    try:
        asyncio.run(stream_with_reconnect())
    except KeyboardInterrupt:
        log("stopped")

Here’s what the response looks like:

Running ws_resilient.py opens a terminal demonstrating automatic WebSocket reconnection handling in Python.

WebSocket vs REST API: How to Choose for Crypto Prices

REST polling works well when your application needs the latest price at intervals you control, while WebSocket is better suited for continuous price updates as they happen. Neither is better than the other. The right choice depends on how your application needs to use the price data.

The table below shows which option is better suited to common use cases:

Use case	Recommended delivery method	Rationale
Page loads and one-off snapshots	REST	A request to `/simple/price` returns the current price in one response without maintaining a persistent connection.
Reports, cron jobs, and scheduled snapshots	REST	A scheduled job calls `/simple/price` for coin prices or `/onchain/simple/networks/{network}/token_price/{addresses}` for onchain token prices each time it runs.
Serverless and stateless functions	REST	A single HTTP request works well with short-lived functions such as AWS Lambda, which run, return a response and then stop.
Tracking many coins or tokens	REST	`/simple/price` returns prices for up to 515 coins per call for 1 credit. The onchain Token Price by Token Addresses endpoint (`/onchain/simple/networks/{network}/token_price/{addresses}`) returns up to 100 contract addresses per call.
Historical or backfill data	REST	Endpoints such as /coins/{id}/market_chart/range and /onchain/networks/{network}/pools/{pool_address}/ohlcv/{timeframe} provide historical price and OHLCV data for a specified date range.
Live tickers, trading interfaces, and dashboards	WebSocket	`CGSimplePrice` and `OnchainSimpleTokenPrice` push price updates instead of requiring repeated requests. Updates occur every ~10 seconds for large-cap coins and ~1 second for actively traded onchain pools.
Automated and algorithmic execution	WebSocket	A subscribed bot receives updates over one open connection with lower latency than polling, letting it react to new price data as soon as it arrives.
Interactive candlestick charts	WebSocket	`OnchainOHLCV` streams OHLCV data for a pool, with updates at ~1-second intervals for actively traded pools. Charts can update with each new data point.

Need to know only when something changes? CoinGecko’s webhooks provide an alternative delivery method for event-driven updates. Instead of polling or maintaining a persistent connection, your server receives a callback when a subscribed event occurs. The cg.coin.info.updated webhook triggers when coin metadata changes, including links, categories, contract addresses, or images. Price Alerts (cg.coin.price.updated) trigger when a price target is reached, while New Listings (cg.coin.listed) notify you of new listings — both are currently in private beta. Submit this form to request early access.

Conclusion

CoinGecko supports multiple delivery methods through one API for aggregated coin and onchain DEX token prices. Use REST API when the cache interval is sufficient, or WebSocket for lower-latency updates and powering live interfaces, with reconnect logic to keep the stream running.

A price feed can be integrated into different applications and workflows. The crypto price alerts for trending coins and categories guide shows how price updates can trigger notifications, while the Solana Sniper Bot guide shows how lower-latency updates can support automated trading. For visualization, the crypto portfolio dashboard in Python guide shows how to display crypto prices. For a market-risk use case, the stablecoin depeg risk detection guide shows how different delivery methods can be combined to monitor market risk.

Ready to start building? Sign up for a free Demo API plan to fetch crypto prices with the REST API. When you need lower-latency price updates, upgrade to the Basic plan for WebSocket and webhook access, higher API credit and rate limits, with a commercial license included.