Skip to content

Instantly share code, notes, and snippets.

@rhettre
Last active May 31, 2024 02:49
Show Gist options
  • Save rhettre/7c17b2a22efcf5c7a9a01514e91566b4 to your computer and use it in GitHub Desktop.
Save rhettre/7c17b2a22efcf5c7a9a01514e91566b4 to your computer and use it in GitHub Desktop.
I built this script to populate trades made in any trading pair available on Coinbase Pro or Gemini into a Google Sheet so it would be easier to track trades, profitability, and cost basis. This could be extended to other exchanges. See https://www.youtube.com/watch?v=hutDJ-FVatw for full explanation.
import json
import gspread
import time
import sys
import base64
import hashlib
import hmac
import urllib.request
import requests
from datetime import datetime, timezone
import urllib.parse
from oauth2client.service_account import ServiceAccountCredentials
from coinbase_advanced_trader import coinbase_client
from coinbase_advanced_trader.config import set_api_credentials
import gemini
SPREADSHEET_LINK = "https://docs.google.com/spreadsheets/d/1VYuy5cSnZiQqF4yp6_sVFpLXXXszCGEiMh-Z37mKims/edit?usp=sharing"
COINBASE_PUBLIC_KEY = ''
COINBASE_PRIVATE_KEY = ''
set_api_credentials(COINBASE_PUBLIC_KEY, COINBASE_PRIVATE_KEY)
COINBASE_SYMBOLS = ["BTC-USD", "ETH-USD"]
GEMINI_PUBLIC_KEY = ''
GEMINI_PRIVATE_KEY = ''
GEMINI_SYMBOLS = ["BTCUSD", "ETHUSD"]
KRAKEN_API_URL = "https://api.kraken.com"
KRAKEN_KEY_PUBLIC = ''
KRAKEN_KEY_PRIVATE = ''
KRAKEN_SYMBOLS = ["XXBTZUSD", "ETHUSD", "ADAUSD", "LINKUSD", "RENUSD", ]
GOOGLE_SHEET_FILE_NAME = "The New Definitive Crypto Sheet"
AUDIT_FILE_SHEET_NAME = "Audit File"
SHEETS_CREDS_FILE_NAME = 'sheets_creds.json'
def authenticate_spreadsheet():
"""Set up access to the spreadsheet."""
scope = [
"https://spreadsheets.google.com/feeds",
'https://www.googleapis.com/auth/spreadsheets',
"https://www.googleapis.com/auth/drive.file",
"https://www.googleapis.com/auth/drive"
]
creds = ServiceAccountCredentials.from_json_keyfile_name(
SHEETS_CREDS_FILE_NAME, scope)
client = gspread.authorize(creds)
return client.open(GOOGLE_SHEET_FILE_NAME).worksheet(AUDIT_FILE_SHEET_NAME)
def add_coinbase_transaction(transaction, id):
"""Format a Coinbase transaction for the audit file."""
transaction_date = str(transaction['created_time'][0:10])
exchange = "Coinbase"
transaction_id = id
symbol = transaction['product_id'].replace('-', '')
side = transaction['side'].upper()
amount = float(transaction['filled_size'])
price = float(transaction['average_filled_price'])
fee = float(transaction['total_fees'])
sell_side_amount = float(transaction['filled_value'])
return [transaction_date, exchange, transaction_id, side, symbol, amount, price, sell_side_amount, fee]
def populate_coinbase(audit_file, symbol):
"""Populate the audit file with Coinbase transactions for a given symbol."""
sheet_transactions = audit_file.get_all_records()
coinbase_transactions = [
t for t in sheet_transactions if t['Exchange'] == 'Coinbase']
symbol_transactions = [
t for t in coinbase_transactions if t['Symbol'] == symbol.replace('-', '')]
sheet_transaction_ids = [t['Transaction ID'] for t in symbol_transactions]
all_orders = coinbase_client.listOrders(product_id=symbol)['orders']
valid_orders = [
order for order in all_orders if order['status'] != 'CANCELLED']
for order in reversed(valid_orders):
timestamp_str = order['created_time']
if timestamp_str:
timestamp_str = timestamp_str[:23] + 'Z'
created_time = int(datetime.strptime(
timestamp_str, '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc).timestamp() * 10000)
if created_time not in sheet_transaction_ids:
audit_file.append_row(add_coinbase_transaction(
order, created_time), value_input_option="USER_ENTERED")
def add_gemini_transaction(transaction):
transaction_date = str(datetime.datetime.fromtimestamp(
transaction['timestamp']).date())
transaction_id = float(transaction['tid'])
exchange = "Gemini"
symbol = transaction['symbol']
side = transaction['type'].capitalize()
amount = float(transaction['amount'])
price = float(transaction['price'])
fee = float(transaction['fee_amount'])
sell_side_amount = amount * price + fee
return [transaction_date, exchange, transaction_id, side, symbol, amount, price, sell_side_amount, fee]
def populate_gemini(audit_file, symbol):
trader = gemini.PrivateClient(GEMINI_PUBLIC_KEY, GEMINI_PRIVATE_KEY)
last_gemini_transaction = (list(filter(
lambda filterExchange: filterExchange['Exchange'] == 'Gemini', audit_file.get_all_records())))
last_symbol_transaction = (list(filter(
lambda filterSymbol: filterSymbol['Symbol'] == symbol, last_gemini_transaction)))
if (last_symbol_transaction):
transactions = trader.get_past_trades(symbol)[::-1]
for transaction in transactions:
if (transaction['tid'] > last_symbol_transaction[-1]['Transaction ID']):
audit_file.append_row(add_gemini_transaction(
transaction), value_input_option="USER_ENTERED")
print('added row')
def add_kraken_transaction(transaction):
transaction_date = str(
datetime.datetime.fromtimestamp(transaction['time']).date())
exchange = 'Kraken'
transaction_id = transaction['ordertxid']
side = transaction['type'].capitalize()
pair = transaction['pair']
amount = transaction['vol']
price = transaction['price']
fee = transaction['fee']
sell_side_amount = transaction['cost']
kraken_time = transaction['time']
if pair == "XXBTZUSD":
pair = "BTCUSD"
return [transaction_date, exchange, transaction_id, side, pair, amount, price, sell_side_amount, fee, kraken_time]
def get_kraken_trade_history():
try:
api_path = '/0/private/TradesHistory'
api_nonce = str(int(time.time()*1000))
api_trades = True
api_post = f'nonce={api_nonce}&trades={api_trades}' % {
'api_nonce': api_nonce, 'trades': api_trades}
api_sha256 = hashlib.sha256(
api_nonce.encode('utf8') + api_post.encode('utf8'))
api_hmac = hmac.new(base64.b64decode(KRAKEN_KEY_PRIVATE), api_path.encode(
'utf8') + api_sha256.digest(), hashlib.sha512)
api_signature = base64.b64encode(api_hmac.digest())
api_request = urllib.request.Request(
'https://api.kraken.com/'+api_path, api_post.encode('utf8'))
api_request.add_header('API-Key', KRAKEN_KEY_PUBLIC)
api_request.add_header('API-Sign', api_signature)
api_request.add_header('User-Agent', 'Kraken trading bot example')
api_response = urllib.request.urlopen(api_request).read().decode()
api_data = json.loads(api_response)
except Exception as error:
print('Failed (%s)' % error)
else:
return api_data['result']['trades']
def populate_kraken(audit_file, symbol):
transactions = (list(filter(
lambda filterPair: filterPair['pair'] == symbol, get_kraken_trade_history().values())))
if symbol == "XXBTZUSD":
symbol = "BTCUSD"
last_kraken_transaction = (list(filter(
lambda filterExchange: filterExchange['Exchange'] == 'Kraken', audit_file.get_all_records())))
last_symbol_transaction = (list(filter(
lambda filterSymbol: filterSymbol['Symbol'] == symbol, last_kraken_transaction)))
for transaction in transactions[::-1]:
if (int(transaction['time']) > int(last_symbol_transaction[-1]['Kraken_Time'])):
audit_file.append_row(add_kraken_transaction(
transaction), value_input_option="USER_ENTERED")
def lambda_handler(event, context):
audit_file = authenticate_spreadsheet()
for symbol in COINBASE_SYMBOLS:
populate_coinbase(audit_file, symbol)
for symbol in GEMINI_SYMBOLS:
populate_gemini(audit_file, symbol)
for symbol in KRAKEN_SYMBOLS:
populate_kraken(audit_file, symbol)
return {
'statusCode': 200,
'body': json.dumps('End of script')
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment