Coinbase Advanced API Python SDK

Welcome to the official Coinbase Advanced API Python SDK. This python project was created to allow coders to easily plug into the Coinbase Advanced API. This SDK also supports easy connection to the Coinbase Advanced Trade WebSocket API.

Coinbase Advanced Trade offers a comprehensive API for traders, providing access to real-time market data, order management, and execution. Elevate your trading strategies and develop sophisticated solutions using our powerful tools and features.

For thorough documentation of all available functions, refer to the following link: https://coinbase.github.io/coinbase-advanced-py

---

Installation

pip3 install coinbase-advanced-py

---

Coinbase Developer Platform (CDP) API Keys

This SDK uses Cloud Developer Platform (CDP) API keys. To use this SDK, you will need to create a CDP API key and secret by following the instructions here. Make sure to save your API key and secret in a safe place. You will not be able to retrieve your secret again.

Ed25519 is the recommended key type. The SDK also supports ECDSA for existing keys. The key type is auto-detected and the correct JWT signing algorithm (EdDSA or ES256) is selected automatically. Accepted formats:

• Ed25519 — PKCS8 PEM (-----BEGIN PRIVATE KEY-----), or raw base64 (32-byte seed or 64-byte seed||pubkey as downloaded from the CDP portal)
• ECDSA — SEC1 PEM (-----BEGIN EC PRIVATE KEY-----)

WARNING: We do not recommend that you save your API secrets directly in your code outside of testing purposes. Best practice is to use a secrets manager and access your secrets that way. You should be careful about exposing your secrets publicly if posting code that leverages this library.

Optional: Set your API key and secret in your environment (make sure to put these in quotation marks). For example, with an Ed25519 key:

export COINBASE_API_KEY="organizations/{org_id}/apiKeys/{key_id}"
export COINBASE_API_SECRET="YOUR_BASE64_ENCODED_ED25519_PRIVATE_KEY"

Or with an ECDSA key:

export COINBASE_API_KEY="organizations/{org_id}/apiKeys/{key_id}"
export COINBASE_API_SECRET="-----BEGIN EC PRIVATE KEY-----\nYOUR PRIVATE KEY\n-----END EC PRIVATE KEY-----\n"

---

REST API Client

In your code, import the RESTClient class and instantiate it:

from coinbase.rest import RESTClient

client = RESTClient() # Uses environment variables for API key and secret

If you did not set your API key and secret in your environment, you can pass them in as arguments:

from coinbase.rest import RESTClient

api_key = "organizations/{org_id}/apiKeys/{key_id}"
api_secret = "-----BEGIN EC PRIVATE KEY-----\nYOUR PRIVATE KEY\n-----END EC PRIVATE KEY-----\n"

client = RESTClient(api_key=api_key, api_secret=api_secret)

After creating your API key, a json file will be downloaded to your computer. It's possible to pass in the path to this file as an argument:

client = RESTClient(key_file="path/to/cdp_api_key.json")

We also support passing a file-like object as the key_file argument:

from io import StringIO
client = RESTClient(key_file=StringIO('{"name": "key-name", "privateKey": "private-key"}'))

You can also set a timeout in seconds for your REST requests like so:

client = RESTClient(api_key=api_key, api_secret=api_secret, timeout=5)

Using the REST Client

You are able to use any of the API hooks to make calls to the Coinbase API. For example:

from json import dumps

accounts = client.get_accounts()
print(dumps(accounts.to_dict(), indent=2))

order = client.market_order_buy(client_order_id="clientOrderId", product_id="BTC-USD", quote_size="1")
print(dumps(order.to_dict(), indent=2))

This code calls the get_accounts and market_order_buy endpoints.

TIP: Setting client_order_id to the empty string will auto generate a unique client_order_id per call. However, this will remove the intended safeguard of accidentally placing duplicate orders.

Refer to the Advanced API Reference for detailed information on each exposed endpoint. Look in the coinbase.rest module to see the API hooks that are exposed.

Custom Response Objects

Endpoints will return corresponding, custom class objects. This allows you to retrieve response object fields using dot-notation. Here is an example of how you can access a product's price via the Get Product endpoint:

product = client.get_product("BTC-USD")
print(product.price)

Dot-notation is only available for fields that are defined. Although all higher-level fields have been defined, not every nested field has. Fields that are not defined are still accessible using standard bracket notation.

For example, we make a call to List Accounts. We take the first account from the defined accounts field and access the defined available_balance field. Despite its nested fields not being explicitly defined and inaccessible via dot-notation, we can still access them manually using bracket notation, like:

accounts = client.get_accounts()
print(accounts.accounts[0].available_balance['value'])

Passing in additional parameters

Use kwargs to pass in any additional parameters. For example:

kwargs = {
    "param1": 10,
    "param2": "mock_param"
}
product = client.get_product(product_id="BTC-USD", **kwargs)

Generic REST Calls

You can make generic REST calls using the get, post, put, and delete methods. For example:

market_trades = client.get("/api/v3/brokerage/products/BTC-USD/ticker", params={"limit": 5})

portfolio = client.post("/api/v3/brokerage/portfolios", data={"name": "TestPortfolio"})

Here we are calling the GetMarketTrades and CreatePortfolio endpoints through the generic REST functions. Once again, the built-in way to query these through the SDK would be:

market_trades = client.get_market_trades(product_id="BTC-USD", limit=5)

portfolio = client.create_portfolio(name="TestPortfolio")

Rate Limit Response Headers

The Advanced API returns useful rate limit information in the response headers as detailed in our documentation. By initializing the RESTClient with the rate_limit_headers field set to True, as shown below, these headers will be appended as fields to the API response body:

client = RESTClient(api_key=api_key, api_secret=api_secret, rate_limit_headers=True)

---

WebSocket API Client

We offer a WebSocket API client that allows you to connect to the Coinbase Advanced Trade WebSocket API. Refer to the Advanced Trade WebSocket Channels page for detailed information on each offered channel.

In your code, import the WSClient class and instantiate it. The WSClient requires an API key and secret to be passed in as arguments. You can also use a key file or environment variables as described in the RESTClient instructions above.

You must specify an on_message function that will be called when a message is received from the WebSocket API. This function must take in a single argument, which will be the raw message received from the WebSocket API. For example:

from coinbase.websocket import WSClient

api_key = "organizations/{org_id}/apiKeys/{key_id}"
api_secret = "-----BEGIN EC PRIVATE KEY-----\nYOUR PRIVATE KEY\n-----END EC PRIVATE KEY-----\n"

def on_message(msg):
    print(msg)

client = WSClient(api_key=api_key, api_secret=api_secret, on_message=on_message)

In this example, the on_message function simply prints the message received from the WebSocket API.

You can also set a timeout in seconds for your WebSocket connection, as well as a max_size in bytes for the messages received from the WebSocket API.

client = WSClient(api_key=api_key, api_secret=api_secret, on_message=on_message, timeout=5, max_size=65536) # 64 KB max_size

Other configurable fields are the on_open and on_close functions. If provided, these are called when the WebSocket connection is opened or closed, respectively. For example:

def on_open():
    print("Connection opened!")

client = WSClient(api_key=api_key, api_secret=api_secret, on_message=on_message, on_open=on_open)

WebSocket User API Client

We offer a WebSocket User API client that allows you to connect to the Coinbase Advanced Trade WebSocket user channel and futures_balance_summary channel.

In your code, import the WSUserClient class instead of WSClient.

from coinbase.websocket import WSUserClient

api_key = "organizations/{org_id}/apiKeys/{key_id}"
api_secret = "-----BEGIN EC PRIVATE KEY-----\nYOUR PRIVATE KEY\n-----END EC PRIVATE KEY-----\n"

def on_message(msg):
    print(msg)

client = WSUserClient(api_key=api_key, api_secret=api_secret, on_message=on_message)

Using the WebSocket Client

Once you have instantiated the client, you can connect to the WebSocket API by calling the open method, and disconnect by calling the close method. The subscribe method allows you to subscribe to specific channels, for specific products. Similarly, the unsubscribe method allows you to unsubscribe from specific channels, for specific products. For example:

# open the connection and subscribe to the ticker and heartbeat channels for BTC-USD and ETH-USD
client.open()
client.subscribe(product_ids=["BTC-USD", "ETH-USD"], channels=["ticker", "heartbeats"])

# wait 10 seconds
time.sleep(10)

# unsubscribe from the ticker channel and heartbeat channels for BTC-USD and ETH-USD, and close the connection
client.unsubscribe(product_ids=["BTC-USD", "ETH-USD"], channels=["ticker", "heartbeats"])
client.close()

We also provide channel specific methods for subscribing and unsubscribing. For example, the below code is equivalent to the example from above:

client.open()
client.ticker(product_ids=["BTC-USD", "ETH-USD"])
client.heartbeats()

# wait 10 seconds
time.sleep(10)

client.ticker_unsubscribe(product_ids=["BTC-USD", "ETH-USD"])
client.heartbeats_unsubscribe()
client.close()

Automatic Reconnection to the WebSocket API

The WebSocket client will automatically attempt to reconnect the WebSocket API if the connection is lost, and will resubscribe to any channels that were previously subscribed to.

The client uses an exponential backoff algorithm to determine how long to wait before attempting to reconnect, with a maximum number of retries of 5.

If you do not want to automatically reconnect, you can set the retry argument to False when instantiating the client.

client = WSClient(api_key=api_key, api_secret=api_secret, on_message=on_message, retry=False)

Catching WebSocket Exceptions

The WebSocket API client will raise exceptions if it encounters an error. On forced disconnects it will raise a WSClientConnectionClosedException, otherwise it will raise a WSClientException.

NOTE: Errors on forced disconnects, or within logic in the message handler, will not be automatically raised since this will be running on its own thread.

We provide the sleep_with_exception_check and run_forever_with_exception_check methods to allow you to catch these exceptions. sleep_with_exception_check will sleep for the specified number of seconds, and will check for any exception raised during that time. run_forever_with_exception_check will run forever, checking for exceptions every second. For example:

from coinbase.websocket import (WSClient, WSClientConnectionClosedException,
                                WSClientException)

client = WSClient(api_key=api_key, api_secret=api_secret, on_message=on_message)

try:
    client.open()
    client.subscribe(product_ids=["BTC-USD", "ETH-USD"], channels=["ticker", "heartbeats"])
    client.run_forever_with_exception_check()
except WSClientConnectionClosedException as e:
    print("Connection closed! Retry attempts exhausted.")
except WSClientException as e:
    print("Error encountered!")

This code will open the connection, subscribe to the ticker and heartbeat channels for BTC-USD and ETH-USD, and will sleep forever, checking for exceptions every second. If an exception is raised, it will be caught and handled appropriately.

If you only want to run for 5 seconds, you can use sleep_with_exception_check:

client.sleep_with_exception_check(sleep=5)

Note that if the automatic reconnection fails after the retry limit is reached, a WSClientConnectionClosedException will be raised.

If you wish to implement your own reconnection logic, you can catch the WSClientConnectionClosedException and handle it appropriately. For example: ```python client = WSClient(api_key=api_key, api_secret=api_secret, on_message=on_message, retry=False)

def connect_and_subscribe(): tr