Search code examples
pythonneural-networktracebackcoinbase-api

how do i properly access the coinbase advanced api


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


Solution

  • 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.