Skip to content
Cerebrus Pulse

Python with x402

Python with x402

Section titled “Python with x402”

Call Cerebrus Pulse from Python with the official x402 client, handling the payment yourself.

Installation

Section titled “Installation”
Terminal window
pip install "x402[evm]" httpx eth-account

Setup

Section titled “Setup”

Build an x402Client, register a payment scheme onto it with your signer, then hand that client to the HTTP wrapper. The registration goes on the x402 client, not on the HTTP client. This is the x402 v2 client: it pays with the PAYMENT-SIGNATURE header, the only one this API accepts. The client refuses any payment above its spend cap, $1 by default; the setup below lowers it to $0.10.

import os
from eth_account import Account
from x402 import x402Client
from x402.http.clients import x402HttpxClient
from x402.mechanisms.evm.exact import register_exact_evm_client


# Load your private key securely (env var, keyfile, etc.)
PRIVATE_KEY = os.environ["BASE_WALLET_PRIVATE_KEY"]


signer = Account.from_key(PRIVATE_KEY)


x402 = x402Client()
register_exact_evm_client(x402, signer)
x402.set_spend_controls({"max_amount_per_payment": "$0.10"})  # refuse any payment above $0.10
Section titled “Async Usage (Recommended)”

x402HttpxClient takes the configured x402Client and handles the 402 flow for you: it sends the request, reads the payment terms, signs, and retries.

import asyncio


BASE_URL = "https://api.cerebruspulse.xyz"


async def get_pulse(coin: str, timeframes: str = "1h,4h") -> dict:
    async with x402HttpxClient(x402) as client:
        resp = await client.get(
            f"{BASE_URL}/pulse/{coin}",
            params={"timeframes": timeframes}
        )
        resp.raise_for_status()
        return resp.json()


async def main():
    data = await get_pulse("BTC")
    print(f"BTC: {data['confluence']['bias']} (score: {data['confluence']['score']})")


asyncio.run(main())

Helper Functions

Section titled “Helper Functions”
async def get_sentiment() -> dict:
    async with x402HttpxClient(x402) as client:
        resp = await client.get(f"{BASE_URL}/sentiment")
        resp.raise_for_status()
        return resp.json()


async def get_funding(coin: str, lookback_hours: int = 24) -> dict:
    async with x402HttpxClient(x402) as client:
        resp = await client.get(
            f"{BASE_URL}/funding/{coin}",
            params={"lookback_hours": lookback_hours}
        )
        resp.raise_for_status()
        return resp.json()


async def get_bundle(coin: str, timeframes: str = "1h,4h") -> dict:
    async with x402HttpxClient(x402) as client:
        resp = await client.get(
            f"{BASE_URL}/bundle/{coin}",
            params={"timeframes": timeframes}
        )
        resp.raise_for_status()
        return resp.json()

Working with the Response

Section titled “Working with the Response”

Field meanings are on the /pulse reference. Three to get right: funding rates are hourly, open interest is in contracts (BTC for BTC, not USD), and confluence.score is a bullish vote share (1 = all bullish, 0 = all bearish), not a confidence.

async def analyze_market(coin: str):
    data = await get_pulse(coin, timeframes="1h,4h")


    price = data["price"].get("current")
    confluence = data["confluence"]
    regime = data["regime"]["current"]


    print(f"\n{'='*50}")
    print(f"{coin} @ ${price:,.2f}  |  Regime: {regime}")
    print(f"Confluence: {confluence['bias']} (bullish share {confluence['score']})")
    print(f"{'='*50}")


    for tf, analysis in data["timeframes"].items():
        if "error" in analysis:
            print(f"\n  [{tf}] {analysis['error']}")
            continue
        ind = analysis["indicators"]
        print(f"\n  [{tf}] closed at {analysis['last_closed_candle_time']}")
        print(f"    RSI:     {ind['rsi_14']} ({ind['rsi_zone']})")
        print(f"    Trend:   {ind['trend']['label']} ({ind['trend']['ema_stack']})")
        print(f"    BB Pos:  {ind['bollinger']['position_pct']:.0%}")
        print(f"    Z-Score: {ind['zscore_100']}")


    deriv = data["derivatives"]
    oi_usd = deriv["open_interest"] * price  # open interest is in contracts
    print(f"\n  Funding: {deriv['funding_rate']:.7f}/h = {deriv['funding_annualized_pct']:.2f}% annualized")
    print(f"  OI: {deriv['open_interest']:,.0f} {data['price_unit'].split()[-1]} (${oi_usd:,.0f})")
    print(f"  Impact spread: {deriv['impact_spread_bps']:.2f} bps")

Error Handling

Section titled “Error Handling”
import httpx


async def safe_pulse(coin: str) -> dict | None:
    try:
        async with x402HttpxClient(x402) as client:
            resp = await client.get(f"{BASE_URL}/pulse/{coin}")
            resp.raise_for_status()
            return resp.json()
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 400:
            print(f"Invalid input (not charged): {e.response.json().get('detail')}")
        elif e.response.status_code == 402:
            print("Payment not accepted, e.g. not enough USDC (nothing settled)")
        elif e.response.status_code == 404:
            print(f"Unknown coin: {coin} (GET /coins lists them)")
        elif e.response.status_code == 429:
            print("Rate limited — wait and retry")
        elif e.response.status_code == 503:
            print(f"Unavailable; retry after {e.response.headers.get('Retry-After')} s")
        elif e.response.status_code == 504:
            print("Engine timeout (not charged) — try again")
        else:
            print(f"Error {e.response.status_code}: {e.response.text}")
        return None

Multi-Coin Scanning

Section titled “Multi-Coin Scanning”
async def scan_all_coins():
    # First, get the coin list (free)
    async with x402HttpxClient(x402) as client:
        resp = await client.get(f"{BASE_URL}/coins")
        coins = resp.json()["coins"]


    print(f"Scanning {len(coins)} coins...")


    # Then fetch pulse for each (paid)
    for coin in coins[:5]:  # Limit to 5 for this example
        data = await safe_pulse(coin)
        if data:
            c = data["confluence"]
            print(f"  {coin}: {c['bias']} (bullish share {c['score']})")

Tips

Section titled “Tips”
  • Reuse the client within a session to avoid reconnection overhead
  • Use /bundle to get /pulse, 24h /funding and /sentiment in one request (one rate-limit slot instead of three); its page shows how its price compares
  • Check /health before batch operations to avoid wasting payments on a degraded gateway
  • Handle stale data — meta.warning is "stale_data" when the price feed is more than 15 minutes old, and meta.warnings lists stalled inputs such as "stale_candles"
Section titled “Related”