Search code examples
pythoncrypto.com-exchange-api

Authentication Failure on Crypto.com exchange API request Python


I am trying to test a basic trading script on Crypto.com's exchange and I keep getting this message.

API Response: {'code': 40101, 'message': 'Authentication failure'} Buy Order Response: {'code': 40101, 'message': 'Authentication failure'}

I have tried creating a few new API Keys and ensuring I am copying them over exactly and I am still getting the same error. I think there's something wrong with my digital signature, but I am not sure what it is.

I found this post where there was a somewhat similar issue, but I was not able to figure out my issue from reading it.

www.crypto.com authenticating, python HMAC-SHA256

Here is my code with my API Keys removed.

import requests
import hmac
import hashlib
import json
from datetime import datetime, timedelta

# Replace with your Crypto.com API credentials
API_KEY = "MY_API_KEY"
API_SECRET = "MY_API_SECRET"
BASE_URL = "https://api.crypto.com/exchange/v1"
SYMBOL = "BTC_USDT"
BUY_AMOUNT = 0.00001  # Adjust according to your trading strategy, 0.00002 min order qty
SELL_THRESHOLD = 1.005  # 0.5% increase
last_buy_price = None  # Store the last buy price


def generate_signature(payload: dict) -> str:
    """Generates HMAC signature for authentication."""
    message = json.dumps(payload, separators=(',', ':'), sort_keys=True)
    return hmac.new(API_SECRET.encode(), message.encode(), hashlib.sha256).hexdigest()


def get_ticker():
    """Fetches the latest ticker information."""
    endpoint = f"{BASE_URL}/public/get-tickers"
    params = {"instrument_name": SYMBOL}
    response = requests.get(endpoint, params=params)
    data = response.json()

    if "result" in data and "data" in data["result"]:
        ticker_data = data["result"]["data"]
        if isinstance(ticker_data, list):
            return ticker_data[0]  # Extract the first item if it's a list
        return ticker_data
    else:
        raise ValueError("Unexpected API response structure: " + str(data))


def place_buy_order():
    """Places a buy order for BTC."""
    global last_buy_price
    endpoint = f"{BASE_URL}/private/create-order"
    timestamp = int(time.time() * 1000)
    ticker_data = get_ticker()
    
     # Ensure price precision (Crypto.com may require specific decimal places)
    buy_price = str(round(float(ticker_data["a"]), 2))  # Convert price to string with 2 decimal places
    buy_amount = str(format(BUY_AMOUNT,'f'))  # Ensure minimum order quantity and convert to type string

    params = {
        "instrument_name": SYMBOL,
        "price": buy_price,
        "quantity": buy_amount,
        "side": "BUY",
        "type": "LIMIT"
    }

    payload = {
        "api_key": API_KEY,        
        "id": timestamp,
        "method": "private/create-order",
        "params": params,
        "nonce": timestamp
    }
    payload["sig"] = generate_signature(payload)
    headers = {"Content-Type": "application/json"}

    # Debugging: Print request payload
    print("Payload being sent:", json.dumps(payload, indent=2))
    
    response = requests.post(endpoint, json=payload, headers=headers)
    response_data = response.json()
    
    print("API Response:", response.json())

    if response.status_code == 200 and response_data.get("code") == 0:
        last_buy_price = float(buy_price)  # Update last buy price  

    return response_data


def place_sell_order():
    """Places a sell order for BTC."""
    endpoint = f"{BASE_URL}/private/create-order"
    timestamp = int(time.time() * 1000)
    ticker_data = get_ticker()
    
    # Ensure price precision (Crypto.com may require specific decimal places)
    sell_price = str(round(float(ticker_data["a"]), 2))  # Adjust precision if needed
    sell_amount = str(format(BUY_AMOUNT,'f'))  # Ensure minimum order quantity and convert to type string
    
    params = {
        "instrument_name": SYMBOL,
        "price": sell_price,
        "quantity": sell_amount,
"side": "SELL",
        "type": "LIMIT"
    }

    payload = {
        "api_key": API_KEY,
        "id": timestamp,
        "method": "private/create-order",
        "params": params,
        "nonce": timestamp
    }
    payload["sig"] = generate_signature(payload)
    headers = {"Content-Type": "application/json"}

    # Debugging: Print request payload
    print("Payload being sent:", json.dumps(payload, indent=2))

    response = requests.post(endpoint, json=payload, headers=headers)
    response_data = response.json()

    print("API Response:", response.json())

    return response_data


def check_price_drop():
    """Checks if BTC price dropped more than 0.25% in the last 24 hours."""
    price_data = get_ticker()

    if isinstance(price_data, list):
        price_data = price_data[0]  # If it's a list, take the first element

    current_price = float(price_data["a"])
    high_price_24h = float(price_data["h"])
    drop_percentage = (high_price_24h - current_price) / high_price_24h * 100
    return drop_percentage >= 0.25


def check_price_rise():
    """Checks if BTC price increased more than 0.5% above the last buy price."""
    global last_buy_price
    if last_buy_price is None:
        return False
    
    price_data = get_ticker()
    if isinstance(price_data, list):
        price_data = price_data[0]
    
    current_price = float(price_data["a"])
    return current_price >= last_buy_price * SELL_THRESHOLD


if __name__ == "__main__":
    while True:
        try:
            if check_price_drop():
                print("Price dropped more than 0.25%! Placing buy order...")
                order_response = place_buy_order()
                print("Buy Order Response:", order_response)
            elif check_price_rise():
                print("Price increased 0.5% above last buy price! Placing sell order...")
                order_response = place_sell_order()
                print("Sell Order Response:", order_response)
            else:
                print("No significant price movement detected. Checking again in 5 minutes.")
        except Exception as e:
            print("Error:", e)
        
        time.sleep(300)  # Check every 5 minutes

I updated the generate signature function as follows and that fixed the issue.

def generate_signature(request_body: dict, secret_key: str) -> str:
    """Generates a valid HMAC SHA256 signature according to Crypto.com API documentation."""
    
    # Extract necessary fields
    method = request_body["method"]
    req_id = str(request_body["id"])
    api_key = request_body["api_key"]
    nonce = str(request_body["nonce"])
    
    # Convert 'params' to a correctly formatted string
    def params_to_str(params, level=0):
        if params is None:
            return ""
        if isinstance(params, dict):
            return "".join(
                key + params_to_str(params[key], level + 1)
                for key in sorted(params.keys())
            )
        if isinstance(params, list):
            return "".join(params_to_str(item, level + 1) for item in params)
        return str(params)

    param_string = params_to_str(request_body.get("params", {}))
    
    # Construct the final signature payload
    sig_payload = method + req_id + api_key + param_string + nonce

    # Generate HMAC-SHA256 signature
    signature = hmac.new(
        secret_key.encode(), sig_payload.encode(), hashlib.sha256
    ).hexdigest()

    return signature

Solution

  • It looks like you're creating the message via json.dumps. This doesn't appear to work how the crypto.com docs indicate you should build it.

    https://exchange-docs.crypto.com/exchange/v1/rest-ws/index.html#digital-signature

    From here, it says you should create the message like this:

    Next, do the following: method + id + api_key + parameter string + nonce

    Your code probably won't produce this exact order. I'm guessing the confusion came from the instructions about the parameter string, which does have you simply alphabetically order the parameters. That can work for the parameter string but not the message which does define a specific order.