调用Coinbase Advanced API获取K线报400错误，疑日期格式问题

• 调用get_product_candles方法时触发400错误，请求URL中start参数为01-01-2021、end参数为01-01-2022
• 已确认参数传递逻辑无误，timestamp为1690483742
• 怀疑是日期格式输入错误导致该问题

错误栈 #

Traceback (most recent call last):
File "D:\user\coinbase_neural_network.py", line 47, in collect_data
candles = api.API.get_product_candles(product_id, start, end, granularity)
File "D:\user\Coinbase_neural_network\api.py", line 76, in get_product_candles
response.raise_for_status()
File "C:\Users\micha\AppData\Local\Programs\Python\Python310\lib\site-packages\requests\models.py", line 1021, in raise_for_status
raise HTTPError(http_error_msg, response=self)
requests.exceptions.HTTPError: 400 Client Error: Bad Request for url: https://api.coinbase.com/api/v3/brokerage/products/LTC-USD/candles?start=01-01-2021&end=01-01-2022&granularity=ONE_HOUR

自定义API类代码 #

import configparser
import http.client
import json
import hashlib
import hmac
import time
from typing import Tuple, List, Dict, Any
import requests

class API:

    config = configparser.ConfigParser()
    config.read(r"D:\user\config.ini")
    api_key = config.get('coinbase', 'API_KEY')
secret_key = config.get('coinbase', 'SECRET_KEY')

@staticmethod
def generate_signature(timestamp: str, method: str, request_path: str, body: str) -> str:
    signature_string = timestamp + method.upper() + request_path + body
    hmac_object = hmac.new(API.secret_key.encode(), signature_string.encode(), hashlib.sha256)
    return hmac_object.hexdigest()

@staticmethod
def get_best_bid_ask() -> str:
    payload: str = ''
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "GET", "/api/v3/brokerage/best_bid_ask", payload),
        'CB-ACCESS-TIMESTAMP': timestamp,
    }
    with http.client.HTTPSConnection("api.coinbase.com") as conn:
        conn.request("GET", "/api/v3/brokerage/best_bid_ask", payload, headers)
    res: http.client.HTTPResponse = conn.getresponse()
    data: bytes = res.read()
    return data.decode("utf-8")

@staticmethod
def get_brokerage_accounts(limit: int, cursor: str) -> str:
    payload: str = ''
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "GET", f"/api/v3/brokerage/accounts?limit={limit}&cursor={cursor}", payload),
        'CB-ACCESS-TIMESTAMP': timestamp,
    }
    with http.client.HTTPSConnection("api.coinbase.com") as conn:
        conn.request("GET", f"/api/v3/brokerage/accounts?limit={limit}&cursor={cursor}", payload, headers)
    res: http.client.HTTPResponse = conn.getresponse()
    data: bytes = res.read()
    return data.decode("utf-8")

@staticmethod
def get_product_candles(product_id: str, start: str, end: str, granularity: str) -> List[Dict[str, str]]:
    payload: str = ''
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "GET",
                                                 f"/api/v3/brokerage/products/{product_id}/candles", payload),
        'CB-ACCESS-TIMESTAMP': timestamp}

    start = start.replace(' ', 'T')
    end = end.replace(' ', 'T')
    url = f"https://api.coinbase.com/api/v3/brokerage/products/{product_id}/candles"
    params = {
        "start": start,
        "end": end,
        "granularity": granularity}
    print(timestamp)
    response = requests.get(url, params=params, headers=headers)
    response.raise_for_status()
    candles = response.json()
    return [dict(zip(["start", "low", "high", "open", "close", "volume"], c)) for c in candles]

@staticmethod
def get_market_trades(product_id: str, limit: int) -> Tuple[List[Dict[str, str]], str, str]:
    payload: str = ''
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "GET", f"/api/v3/brokerage/products/{product_id}/ticker?limit={limit}", payload),
        'CB-ACCESS-TIMESTAMP': timestamp,
    }
    with http.client.HTTPSConnection("api.coinbase.com") as conn:
        conn.request("GET", f"/api/v3/brokerage/products/{product_id}/ticker?limit={limit}", payload, headers)
    res: http.client.HTTPResponse = conn.getresponse()
    data: bytes = res.read()
    trades: Dict[str, Any] = json.loads(data.decode("utf-8"))
    trade_data: List[Dict[str, str]] = [dict(zip(["trade_id", "product_id", "price", "size", "time", "side", "bid",
                                                  "ask"], t)) for t in trades["trades"]]
    best_bid: str = trades["best_bid"]
    best_ask: str = trades["best_ask"]
    return trade_data, best_bid, best_ask

@staticmethod
def create_buy_market_order(client_order_id: str, product_id: str, side: str, order_configuration: str) -> str:
    payload: str = json.dumps({
        "client_order_id": client_order_id,
        "product_id": product_id,
        "side": side,
        "order_configuration": order_configuration
    })
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "POST", "/api/v3/brokerage/orders", payload),
        'CB-ACCESS-TIMESTAMP': timestamp,
    }
    with http.client.HTTPSConnection("api.coinbase.com") as conn:
        conn.request("POST", "/api/v3/brokerage/orders", payload, headers)
    res: http.client.HTTPResponse = conn.getresponse()
    data: bytes = res.read()
    return data.decode("utf-8")

@staticmethod
def create_sell_market_order(client_order_id: str, product_id: str, side: str, order_configuration: str) -> str:
    payload: str = json.dumps({
        "client_order_id": client_order_id,
        "product_id": product_id,
        "side": side,
        "order_configuration": order_configuration
    })
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "POST", "/api/v3/brokerage/orders", payload),
        'CB-ACCESS-TIMESTAMP': timestamp,
    }
    with http.client.HTTPSConnection("api.coinbase.com") as conn:
        conn.request("POST", "/api/v3/brokerage/orders", payload, headers)
    res: http.client.HTTPResponse = conn.getresponse()
    data: bytes = res.read()
    return data.decode("utf-8")

是的，问题根源就是日期格式不符合Coinbase Advanced API的要求。

Coinbase Advanced API的get_product_candles接口要求start和end参数必须是RFC3339格式的时间字符串，比如2021-01-01T00:00:00Z，而你当前传递的01-01-2021这种MM-DD-YYYY格式完全不被API识别，直接触发了400 Bad Request错误。

修复方案 #

修改get_product_candles方法中处理日期的逻辑，将输入的日期转换为RFC3339格式：

from datetime import datetime

@staticmethod
def get_product_candles(product_id: str, start: str, end: str, granularity: str) -> List[Dict[str, str]]:
    payload: str = ''
    timestamp = str(int(time.time()))
    headers: dict[str, str] = {
        'Content-Type': 'application/json',
        'CB-ACCESS-KEY': API.api_key,
        'CB-ACCESS-SIGN': API.generate_signature(timestamp, "GET",
                                                 f"/api/v3/brokerage/products/{product_id}/candles", payload),
        'CB-ACCESS-TIMESTAMP': timestamp}

    # 将MM-DD-YYYY格式转换为RFC3339标准格式
    def format_date(date_str):
        dt = datetime.strptime(date_str, "%m-%d-%Y")
        return dt.strftime("%Y-%m-%dT00:00:00Z")
    
    start = format_date(start)
    end = format_date(end)
    url = f"https://api.coinbase.com/api/v3/brokerage/products/{product_id}/candles"
    params = {
        "start": start,
        "end": end,
        "granularity": granularity}
    print(timestamp)
    response = requests.get(url, params=params, headers=headers)
    response.raise_for_status()
    candles = response.json()
    return [dict(zip(["start", "low", "high", "open", "close", "volume"], c)) for c in candles]

额外注意事项：

• 确认granularity参数的值是API允许的选项（比如ONE_HOUR是合法值，需保证拼写完全正确）
• 签名生成时的请求路径不要包含查询参数，你当前代码的签名逻辑是正确的，符合Coinbase API要求

内容的提问来源于stack exchange，提问作者michael king