Skip to content

Instantly share code, notes, and snippets.

@rhettre
Last active December 1, 2023 05:43
Show Gist options
  • Save rhettre/a86b11bf1916a2b9bf73cbd2aa65fbdd to your computer and use it in GitHub Desktop.
Save rhettre/a86b11bf1916a2b9bf73cbd2aa65fbdd to your computer and use it in GitHub Desktop.
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…
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)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment