This Python script contains an AWS Lambda function designed to automate withdrawals from Coinbase accounts. It was created in response to the retirement of the Coinbase Pro API, which previously facilitated withdrawal automation. The script leverages the Coinbase Wallet API to list payment methods and initiate crypto withdrawals. Users can custo…

coinbase_withdrawals.py

import json
import hmac
import hashlib
import time
import requests
from requests.auth import AuthBase
import os

with open(os.path.join(os.path.dirname(__file__), 'config.json')) as config_file:
    config = json.load(config_file)

API_KEY = config['API_KEY']
API_SECRET = config['API_SECRET']
BTC_ADDRESS = config['BTC_ADDRESS']
WITHDRAW_CURRENCY = config['WITHDRAW_CURRENCY']
WITHDRAWAL_ID = config['WITHDRAWAL_ID']
FACTOR = config['WITHDRAWAL_FACTOR']

# Coinbase API Endpoints
WITHDRAW_ENDPOINT = f'https://api.coinbase.com/v2/accounts/{WITHDRAWAL_ID}/transactions'
ACCOUNTS_ENDPOINT = 'https://api.coinbase.com/v2/accounts'

# Authenticate with Coinbase API


class CoinbaseWalletAuth(AuthBase):
    def __init__(self, api_key, secret_key):
        self.api_key = api_key
        self.secret_key = secret_key

    def __call__(self, request):
        timestamp = str(int(time.time()))
        message = timestamp + request.method + \
            request.path_url + (request.body or '')
        signature = hmac.new(self.secret_key.encode(),
                             message.encode(), hashlib.sha256).hexdigest()

        request.headers.update({
            'CB-ACCESS-SIGN': signature,
            'CB-ACCESS-TIMESTAMP': timestamp,
            'CB-ACCESS-KEY': self.api_key,
        })
        return request


# Function to List Accounts - need this for getting WITHDRAWAL_ID

def list_accounts(starting_after=None):
    auth = CoinbaseWalletAuth(API_KEY, API_SECRET)
    url = f"{ACCOUNTS_ENDPOINT}?starting_after={starting_after}" if starting_after else ACCOUNTS_ENDPOINT

    response = requests.get(url, auth=auth)
    data = response.json()

    active_accounts = [account for account in data['data']
                       if float(account['balance']['amount']) > 0]

    print_custom_accounts(active_accounts)

    if data['pagination']['next_starting_after']:
        list_accounts(starting_after=data['pagination']['next_starting_after'])


def print_custom_accounts(accounts):
    for account in accounts:
        print(f"WITHDRAWAL_ID: {account['id']}")
        print(f"Name: {account['name']}")
        print(f"Amount: {account['balance']['amount']}")

        native_balance = account.get('native_balance', {'amount': 'N/A', 'currency': 'N/A'})
        print(
            f"Native Balance: {native_balance['amount']} {native_balance['currency']}\n")


def get_account_balance(account_id):
    auth = CoinbaseWalletAuth(API_KEY, API_SECRET)
    response = requests.get(f"{ACCOUNTS_ENDPOINT}/{account_id}", auth=auth)
    data = response.json()
    return float(data['data']['balance']['amount'])


def withdraw_crypto(amount, withdraw_all):
    auth = CoinbaseWalletAuth(API_KEY, API_SECRET)

    if withdraw_all:
        amount = round(get_account_balance(WITHDRAWAL_ID) * FACTOR, 8)
        print(
            f'Amount updated to {amount} because withdraw_all was set to True.')
    data = {
        'type': 'send',
        'to': BTC_ADDRESS,
        'amount': amount,
        'currency': WITHDRAW_CURRENCY
    }

    response = requests.post(WITHDRAW_ENDPOINT, data=data, auth=auth)
    if response.status_code == 201:
        print(
            f'You withdrew {amount} {WITHDRAW_CURRENCY} from your Coinbase account!')
        print(response.json())
        return response.json()
    else:
        print(response.text)
        raise Exception(f'Error: {response.text}')


def lambda_handler(event, context):
    list_accounts()
    withdraw_crypto(0, True)