如何正确访问Coinbase高级API

huangapple go评论90阅读模式
英文:

how do i properly access the coinbase advanced api

问题

我正在尝试访问Coinbase高级API,但我遇到了这个错误:

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

当我尝试在这个类中使用collect_data函数时:

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:
        # 签名生成的部分代码

    @staticmethod
    def get_best_bid_ask() -> str:
        # 获取最佳买卖价的部分代码

    # 其他方法的定义...

一切都正常通过,这是时间戳:1690483742,以及URL:在收集数据时发生错误: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

我输入日期有问题吗?

英文:

im trying to access the coinbase adveanced api ut im getting this error:
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

when i try to use the collect data function in this class:
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")

everything is passed right this is the timestamp: 1690483742
and url: Error occurred while collecting data: 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
am i entering the date wrong?your text

答案1

得分: 0

你需要将时间发送为Unix时间戳。因此,请将此部分替换为:

    start = int(datetime.datetime.strptime(start, '%m-%d-%Y').timestamp())
    end = int(datetime.datetime.strptime(end, '%m-%d-%Y').timestamp())

假设(1)你在顶部导入了import datetime,并且(2)你始终以MM-DD-YYYY格式提供日期。

英文:

You need to send the times as Unix timestamps. So replace this:

    start = start.replace(' ', 'T')
    end = end.replace(' ', 'T')

With this:

    start = int(datetime.datetime.strptime(start, '%m-%d-%Y').timestamp())
    end = int(datetime.datetime.strptime(end, '%m-%d-%Y').timestamp())

That assumes (1) you have import datetime at the top, and (2) you always give your dates in MM-DD-YYYY format.

huangapple
  • 本文由 发表于 2023年7月28日 03:08:13
  • 转载请务必保留本文链接:https://go.coder-hub.com/76782749.html
匿名

发表评论

匿名网友

:?: :razz: :sad: :evil: :!: :smile: :oops: :grin: :eek: :shock: :???: :cool: :lol: :mad: :twisted: :roll: :wink: :idea: :arrow: :neutral: :cry: :mrgreen:

确定