ChangeNOW API Integration

ChangeNOW API Integration

ChangeNOW is one of the largest non-custodial crypto-to-crypto exchange providers. Integration with their API allows you to add exchange functionality to your wallet, exchanger, or platform without building your own liquidity from scratch.

Basic Integration

Getting a Rate

import httpx
from decimal import Decimal

class ChangeNOWClient:
    BASE_URL = "https://api.changenow.io/v2"

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.headers = {"x-changenow-api-key": api_key}

    async def get_estimated_amount(
        self,
        from_currency: str,  # 'btc'
        to_currency: str,    # 'eth'
        from_amount: float,
        flow: str = 'standard'  # 'standard' or 'fixed-rate'
    ) -> dict:
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.BASE_URL}/exchange/estimated-amount",
                params={
                    "fromCurrency": from_currency.lower(),
                    "toCurrency": to_currency.lower(),
                    "fromAmount": str(from_amount),
                    "flow": flow,
                    "type": "direct"
                },
                headers=self.headers
            )
        data = response.json()

        if "error" in data:
            raise ChangeNOWError(f"{data['error']}: {data.get('message', '')}")

        return {
            "estimated_amount": data["toAmount"],
            "rate": data["toAmount"] / from_amount,
            "min_amount": data.get("minAmount"),
            "max_amount": data.get("maxAmount"),
            "network_fee": data.get("networkFee")
        }

Creating an Exchange

async def create_exchange(
    self,
    from_currency: str,
    to_currency: str,
    from_amount: float,
    to_address: str,
    refund_address: str = None,
    flow: str = 'standard',
    user_id: str = None
) -> dict:
    payload = {
        "fromCurrency": from_currency.lower(),
        "toCurrency": to_currency.lower(),
        "fromAmount": str(from_amount),
        "address": to_address,
        "flow": flow,
        "type": "direct",
        "extraId": "",  # memo/tag for XRP, EOS, etc.
    }

    if refund_address:
        payload["refundAddress"] = refund_address

    # userId—for tracking affiliate conversions
    if user_id:
        payload["userId"] = user_id

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{self.BASE_URL}/exchange",
            json=payload,
            headers=self.headers
        )
    data = response.json()

    return {
        "order_id": data["id"],
        "deposit_address": data["payinAddress"],
        "deposit_amount": data["fromAmount"],
        "receive_amount": data["toAmount"],
        "payin_extra_id": data.get("payinExtraId"),  # memo for some networks
        "status": data["status"]
    }

Monitoring Status

async def get_exchange_status(self, order_id: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}/exchange/by-id",
            params={"id": order_id},
            headers=self.headers
        )
    data = response.json()

    # Map ChangeNOW statuses to internal
    status_map = {
        "waiting": "awaiting_deposit",
        "confirming": "confirming",
        "exchanging": "processing",
        "sending": "sending",
        "finished": "completed",
        "failed": "failed",
        "refunded": "refunded",
        "verifying": "kyc_required"
    }

    return {
        "status": status_map.get(data["status"], data["status"]),
        "payin_hash": data.get("payinHash"),
        "payout_hash": data.get("payoutHash"),
        "amount_received": data.get("amountReceived"),
        "amount_sent": data.get("amountSent"),
        "updated_at": data.get("updatedAt")
    }

List of Supported Currencies

async def get_currencies(self, active: bool = True) -> list[dict]:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}/exchange/currencies",
            params={"active": str(active).lower(), "flow": "standard"},
            headers=self.headers
        )
    currencies = response.json()
    return [
        {
            "ticker": c["ticker"],
            "name": c["name"],
            "network": c.get("network"),
            "image": c.get("image"),
            "is_stable": c.get("isStable", False)
        }
        for c in currencies
    ]

Minimum Amounts and Validation

async def validate_exchange_params(
    self,
    from_currency: str,
    to_currency: str,
    from_amount: float
) -> ValidationResult:
    range_data = await self.get_range(from_currency, to_currency)

    if from_amount < range_data["min_amount"]:
        return ValidationResult(
            valid=False,
            error=f"Amount too small. Min: {range_data['min_amount']} {from_currency.upper()}"
        )

    if range_data.get("max_amount") and from_amount > range_data["max_amount"]:
        return ValidationResult(
            valid=False,
            error=f"Amount too large. Max: {range_data['max_amount']} {from_currency.upper()}"
        )

    return ValidationResult(valid=True)

async def get_range(self, from_currency: str, to_currency: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{self.BASE_URL}/exchange/range",
            params={"fromCurrency": from_currency, "toCurrency": to_currency, "flow": "standard"},
            headers=self.headers
        )
    data = response.json()
    return {"min_amount": data["minAmount"], "max_amount": data.get("maxAmount")}

ChangeNOW supports 850+ coins, Standard and Fixed Rate modes, referral program (35-40% of margin), and has well-documented API. For quick start of an exchanger—optimal first liquidity provider.