Crypto.com 交易所 API 请求 Python 身份验证失败

我正在尝试在 Crypto.com 的交易所测试一个基本的交易脚本，但我一直收到此消息。

API 响应：{'code'：40101，'message'：'身份验证失败'} 购买订单响应：{'code'：40101，'message'：'身份验证失败'}

我尝试创建几个新的 API 密钥并确保我完全复制了它们，但仍然出现相同的错误。我认为我的数字签名有问题，但我不确定是什么问题。

我发现这篇文章中有一个类似的问题，但我无法通过阅读它找出我的问题。

www.crypto.com 身份验证，python HMAC-SHA256

以下是删除了我的 API 密钥后的代码。

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

我更新了生成签名函数如下，并解决了该问题。

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