Dollar Cost Averaging with Coinbase

This article intends to get you set up using a Dollar-Cost Averaging (DCA) technique to buy crypto through Coinbase via a Python script. This method results in much lower fees than their standard Recurring Buy method.

If you just want code, skip to it here.

I have written this for people like me: familiar with basic scripting, but definitely not professional software developers. Also note, I am certainly not a financial advisor either.

What Does DCA Stand For

Dollar-Cost Averaging (DCA) means making investments at regular intervals. The belief being that DCA returns are more reliable than trying to time the market with lump sums.

Crypto Exchange-Traded Funds and Fees

If you aren’t able to get this code up and running, auto-buying bitcoin ETFs (exchange traded funds) every week through a brokerage may be the best option for you. As of this article, fund management fees actually make for a competitive alternative to regularly buying crypto directly.

TierTaker FeeMaker Fee
\$0K-$10K0.60%0.40%
\$10K-$50K0.40%0.25%
\$50K-$100K0.25%0.15%
\$100K-$1M0.20%0.10%
\$1M-$15M0.18%0.08%
\$15M-$75M0.16%0.06%
\$75M-$250M0.12%0.03%
\$250M-$400M0.08%0.00%
\$400M+0.05%0.00%
Coinbase fee schedule as of 3/2/2024

Fund NameFund SymbolManagement Fee
Ark 21Shares Bitcoin ETFARKB0.25%
iShares Bitcoin TrustIBIT0.25%
Fidelity Wise Origin Bitcoin FundFBTC0.25%
Example Bitcoin ETF management fees as of 3/2/2024

Why DCA Crypto Directly

You may just want to buy some form of crypto directly for the fun of it or to say you have it. Or, you may be buying large amounts, to the point where standard maker/taker fees in the table above become much lower than ETF management fees.

Why Not Use Coinbase Recurring Buy?

From the picture below, you can see that the recurring buy fee there when setting up through Coinbase is nearly 50 times higher than when I run a custom “buy” script through Coinbase’s API.

Coinbase recurring buy (left) vs scripted recurring buy (right)

Put another way, recurring buys set up through Coinbase have fees over 28% depending on the amount you buy. Whereas the example on the right has the typically sub-1% fees you would expect from a typical limit order placed manually.

I cannot find out a clear explanation of the fees Coinbase charges for recurring buys of popular coins like Bitcoin and Coinbase. The only fee-free recurring buy option is USDC.

The Two-Step Custom-Scripted Coinbase DCA Process

  1. Recur Buy USDC
    • Set this up with Coinbase. No script or fancy setup needed.
    • This is the ONLY cryptocurrency I support buying on a recurring basis with Coinbase.
    • There are no fees if buying through a linked bank account.
    • They give you a whopping 5.1% APR for just holding USDC! (as of 3/2/2024)
  2. Use the USDC to buy crypto regularly using a script
    • This is where the script in this article comes in.
    • You can run this as much as you like. I recommend at least weekly.
    • I buy at least 3 times a day. I run the script as a cron job on a Linux machine (specifically, a raspberry pi). I don’t have any studies or even anecdotal evidence saying this is the right buying interval to use. It’s just fun.

1. Recur Buy USDC

The article assumes you already have a Coinbase account.

Once you have an account, follow the instructions here to set up recurring buys. You should be buying USDC. Note that you can’t be in the Advanced trading section of Coinbase for this step, but we will switch over to that side of the website soon.

To start, buy weekly. Your amount should be enough to cover you for a week (if you plan on buying \$10 worth every day, transfer in \$70 a week).

There are no fees for USDC recurring buys on Coinbase when using bank transfers.

2a. Use the USDC to buy Crypto Regularly – Pre-Req 1

You will need to set up an API key that the script can use. It will be in the form of a downloadable JSON file. See below for step-by-step guide of generating this file.

Navigate to cloud.coinbase.com/access/api

Create a “Trading key”

Verify the key has View and Trade permissions.

If applicable, enter the code in your authenticator app.

A download window should appear, and you need to save the JSON file it gives to you.

2b. Use the USDC to buy Crypto Regularly – Pre-Req 2

You need to have Python installed to run this script along with some libraries:

  • requests
  • json
  • jwt
  • time
  • logging
  • uuid
  • secrets
  • cryptography

If you prefer to use Docker to test things out (like me), here is a great video showing you a way to use Docker for Python development. I had some hiccups getting the libraries above loaded on my Windows machine, so Docker was a good choice for me.

2c. Use the USDC to buy Crypto Regularly – The Script

This script is written in Python and leverages the Coinbase Cloud API to execute buying strategies for cryptocurrencies like Bitcoin and Ethereum. It was made with ChatGPT and careful editing until it worked.

How It Works / What it Does:

  1. Initialization: The bot starts by loading your Coinbase Cloud API credentials from a JSON file which you should have downloaded as described here. This includes a key name and a secret private key used for authenticating API requests.
  2. Setting Constants: It defines several constants, such as the API URL, cryptocurrency IDs (for BTC and ETH), the amount in USDC to use for transactions, and the maximum order time. You can change these to your liking.
  3. Authentication: To interact with the Coinbase API, the bot generates a JSON Web Token (JWT) for each request. This involves creating a payload with your credentials, the request details, and a short expiry time. If you are just looking to pull some tips from this guide to write your own script, how authentication is handled is probably what you want to copy most.
  4. Price Analysis: Before placing an order, the bot fetches the best bid and ask prices for the selected cryptocurrency. This ensures that you’re making informed decisions based on current market conditions.
  5. Order Management: With the price data, the bot can place, edit, or cancel orders. The logic of the code is essentially to evaluate the current price, throw a low-ball out there, wait a bit, then increase the offer price slightly. The bot will keep walking up slowly to try capturing a good deal.

Things you need to change in this file:

  • ****The name and location of your JSON file containing the API key.****
  • The type of cryptocurrency pairs you want to use
  • The amounts you want to buy
  • The order placing strategy (see the goofy adjustement_factor variable in main)
  • Set up how often you want this to run in your operating system. The method of doing this is highly dependent on your machine. I run this as a cron job in Linux several times a day.
#!/usr/bin/env python3
"""
Purpose of the Script:
This script is designed to automate the process of buying cryptocurrencies using Dollar-Cost Averaging (DCA) through Coinbase, leveraging the Coinbase Cloud API. It aims to optimize investment by utilizing lower maker/taker fees instead of higher recurring buy fees, making cryptocurrency investment more cost-effective. The script handles authentication, price analysis, and order management to execute buy orders based on the best bid/ask prices.

The detailed guide and explanation of this script, including step-by-step instructions on setting it up and customizing it for individual needs, can be found at TheVoltageDrop.com. This resource is intended for those with a technical background interested in automating their cryptocurrency investments efficiently.

Note: Users should customize the script according to their requirements, including API key file location, adjusting cryptocurrency pairs, and defining investment amounts.
"""


import requests
import json
import jwt
import time
import logging
import uuid
import secrets
from cryptography.hazmat.primitives import serialization

# Load Coinbase Cloud API credentials from a JSON file
# ****CHANGE THE DIRECTORY BELOW TO YOUR ACTUAL DIRECTORY****
with open('/home/cointrader/scripts/coinbase_cloud_api_key.json', 'r') as file:
    credentials = json.load(file)
KEY_NAME = credentials['name']
KEY_SECRET = credentials['privateKey']

# Define constants for Coinbase API
API_URL = "https://api.coinbase.com/api/v3/"
BTC_ID = "BTC-USDC"  # ID for Bitcoin in the Coinbase API
ETH_ID = "ETH-USDC"  # ID for Ethereum in the Coinbase API
BTC_AMOUNT = 3.5  # Amount in USDC to use for buying Bitcoin
ETH_AMOUNT = 3.5  # Amount in USDC to use for buying Ethereum
MAX_ORDER_TIME = 5 * 60  # Maximum time an order can be active (5 minutes)

# Setup logging to help with debugging and tracking the bot's activity
logger = logging.getLogger('')
logger.setLevel(logging.DEBUG)  # Capture all levels of log messages
log_formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.DEBUG)
console_handler.setFormatter(log_formatter)
logger.addHandler(console_handler)

def build_jwt(request_method, request_path, service_name):
    """Generate a JWT for authentication with the Coinbase API."""
    request_host = "api.coinbase.com"
    private_key_bytes = KEY_SECRET.encode('utf-8')
    private_key = serialization.load_pem_private_key(private_key_bytes, password=None)
    # JWT payload as per Coinbase Cloud API documentation
    jwt_payload = {
        'sub': KEY_NAME,
        'iss': "coinbase-cloud",
        'nbf': int(time.time()),
        'exp': int(time.time()) + 120,  # Token expiry set to 2 minutes
        'aud': [service_name],
        'uri': f"{request_method} {request_host}{request_path}",
    }
    jwt_token = jwt.encode(
        jwt_payload,
        private_key,
        algorithm='ES256',
        headers={'kid': KEY_NAME, 'nonce': secrets.token_hex()},
    )
    return jwt_token

def get_best_bid_ask(coin_id):
    """Retrieve the best bid and ask prices for a specified cryptocurrency."""
    request_method = "GET"
    request_host = "api.coinbase.com"
    request_path = "/api/v3/brokerage/best_bid_ask"
    service_name = "retail_rest_api_proxy"

    jwt_token = build_jwt(request_method, request_path, service_name)
    headers = {
        "Authorization": f"Bearer {jwt_token.decode('utf-8')}",
        "Content-Type": "application/json"
    }

    try:
        full_url = f"https://{request_host}{request_path}"
        response = requests.get(full_url, headers=headers)
        if response.status_code == 200:
            response_data = response.json()
            # Find the coin_id in the response and return its best bid and ask prices
            for pricebook in response_data['pricebooks']:
                if pricebook['product_id'] == coin_id:
                    best_bid = float(pricebook['bids'][0]['price']) if pricebook['bids'] else None
                    best_ask = float(pricebook['asks'][0]['price']) if pricebook['asks'] else None
                    return best_bid, best_ask
            logger.error(f"No pricebook data found for {coin_id}")
        else:
            logger.error(f"Error retrieving best bid and ask data: {response.text}")
    except Exception as e:
        logger.error(f"Error in get_best_bid_ask: {e}")
    return None, None

def place_order(coin_id, amount, price):
    """Place a buy order for a specified cryptocurrency at a given price."""
    request_method = "POST"
    service_name = "retail_rest_api_proxy"

    jwt_token = build_jwt(request_method, "/api/v3/brokerage/orders", service_name)
    headers = {
        "Authorization": f"Bearer {jwt_token.decode('utf-8')}"
    }

    # Generate a unique client order ID for tracking
    client_order_id = str(uuid.uuid4())
    # Prepare the data payload for the order
    data = {
        "client_order_id": client_order_id,
        "product_id": coin_id,
        "side": "BUY",
        "order_configuration": {
            "limit_limit_gtc": {
                "base_size": str(round(amount / float(price), 8)),  # Calculate the quantity to buy based on amount and price
                "limit_price": str(round(float(price), 2)),  # Set the limit price for the order
                "post_only": False
            }
        }
    }

    try:
       response = requests.post(f"{API_URL}brokerage/orders", headers=headers, json=data)
        response_data = response.json()
        if response.status_code == 200 and 'success' in response_data:
            if response_data['success']:
                logger.info(f"Order placement succeeded for {coin_id}: {response_data}")
                return response_data['success_response']['order_id']
            else:
                failure_reason = response_data.get('failure_reason', 'Unknown reason')
                logger.warning(f"Order placement had issues for {coin_id}: {failure_reason}")
        else:
            logger.error(f"Error placing order for {coin_id}: {response.text}")
    except Exception as e:
        logger.error(f"Exception in place_order for {coin_id}: {e}")
    return None

def cancel_order(order_id):
    """Cancel an existing order using its order ID."""
    request_method = "POST"
    service_name = "retail_rest_api_proxy"

    jwt_token = build_jwt(request_method, "/api/v3/brokerage/orders/batch_cancel", service_name)
    headers = {
        "Authorization": f"Bearer {jwt_token.decode('utf-8')}",
        "Content-Type": "application/json"
    }

    payload = json.dumps({"order_ids": [order_id]})
    try:
        response = requests.post(f"{API_URL}brokerage/orders/batch_cancel", headers=headers, data=payload)
        if response.status_code == 200:
            logger.info(f"Successfully canceled order {order_id}")
            return True
        else:
            logger.error(f"Error canceling order {order_id}: {response.text}")
    except Exception as e:
        logger.error(f"Exception in cancel_order for {order_id}: {e}")
    return False

def edit_order(order_id, new_price, amount):
    """Edit an existing order, updating its price and amount."""
    request_method = "POST"
    service_name = "retail_rest_api_proxy"

    jwt_token = build_jwt(request_method, "/api/v3/brokerage/orders/edit", service_name)
    headers = {
        "Authorization": f"Bearer {jwt_token.decode('utf-8')}"
    }

    # Calculate the new size (amount of cryptocurrency to buy) based on the new price
    size = round(amount / float(new_price), 8)
    data = {
        "order_id": order_id,
        "price": "{:.2f}".format(new_price),
        "size": "{:.8f}".format(size)
    }

    try:
        response = requests.post(f"{API_URL}brokerage/orders/edit", headers=headers, json=data)
        if response.status_code == 200 and response.json().get('success', False):
            logger.info("Order edit successful.")
            return True
        else:
            logger.error(f"Error editing order {order_id}: {response.text}")
    except Exception as e:
        logger.error(f"Exception in edit_order for {order_id}: {e}")
    return False

def check_order_status(order_id):
    """Check the status of an order to determine if it has been filled."""
    request_method = "GET"
    service_name = "retail_rest_api_proxy"

    jwt_token = build_jwt(request_method, f"/api/v3/brokerage/orders/historical/{order_id}", service_name)
    headers = {
        "Authorization": f"Bearer {jwt_token.decode('utf-8')}"
    }

    try:
        response = requests.get(f"{API_URL}brokerage/orders/historical/{order_id}", headers=headers)
        if response.status_code == 200:
            order_status = response.json().get('status', '')
            return order_status == 'FILLED'
        else:
            logger.error(f"Error retrieving order status for {order_id}: {response.text}")
    except Exception as e:
        logger.error(f"Exception in check_order_status for {order_id}: {e}")
    return False

def main(coin_id, amount):
    """Main function to execute buy orders based on the best bid/ask prices."""
    order_id = None
    try:
        while True:
            # Get the current best bid and ask prices for the cryptocurrency
            best_bid, best_ask = get_best_bid_ask(coin_id)
            if best_bid is None or best_ask is None:
                break

            # Calculate the average price and set a limit price slightly below it
            average_price = (best_bid + best_ask) / 2
            limit_price = average_price * 0.98

            # Place an initial order if one hasn't been placed yet
            if order_id is None:
                order_id = place_order(coin_id, amount, limit_price)

            time.sleep(30)  # Wait a bit before checking the order status
            if check_order_status(order_id):
                break  # Exit the loop if the order has been filled

            # Attempt to adjust the order price by applying an adjustment factor and re-checking the order status
            for adjustment_factor in [0.985, 0.990, 0.991, 0.992, 0.993, 0.994, 0.995, 0.996, 0.997, 0.998, 0.999, 1.0]:
                best_bid, best_ask = get_best_bid_ask(coin_id)
                if best_bid is None or best_ask is None:
                    break

                new_average_price = (best_bid + best_ask) / 2
                new_limit_price = new_average_price * adjustment_factor
                if edit_order(order_id, new_limit_price, amount):
                    time.sleep(60)  # Give some time for the order to potentially fill
                    if check_order_status(order_id):
                        return  # Exit if the order has been filled

            time.sleep(5 * 60)  # Wait for a longer period before the next cycle
            # If the order hasn't filled, attempt to cancel it before ending the loop
            if not check_order_status(order_id):
                cancel_order(order_id)
                logger.info("Order not filled. No purchase made.")
                break
    except Exception as e:
        logger.error(f"Exception in main function: {e}")

if __name__ == "__main__":
    # Execute the trading strategy for both Ethereum and Bitcoin
    main(ETH_ID, ETH_AMOUNT)
    main(BTC_ID, BTC_AMOUNT)

Summary

This article provides a comprehensive guide for individuals interested in leveraging Dollar-Cost Averaging (DCA) to invest in cryptocurrencies through Coinbase, focusing on a cost-efficient approach that avoids high recurring buy fees via the standard Coinbase method. The core of the strategy involves using a custom script to automate purchases, contrasting the standard recurring buy fees with the much lower maker/taker fees available through this method.

In the article we:

  • Began with an overview of DCA as an investment strategy
  • Discussed how Bitcoin ETFs are actually a viable alternative for some people.
  • Discussed how you should buy USDC first, then your other crypto with that.
  • Discussed what was needed in order to run the script.
  • The script itself.

I wanted to share this because I got it working and haven’t found anywhere else with a single-file scripting solution like this (really two files if you count the API JSON). Please leave comments if you find something to improve.

Leave a Comment