Skip to content

Instantly share code, notes, and snippets.

@DVN237294
Created April 2, 2022 07:44
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save DVN237294/4f40133b79926dd598333580a9f5d82f to your computer and use it in GitHub Desktop.
Save DVN237294/4f40133b79926dd598333580a9f5d82f to your computer and use it in GitHub Desktop.
Credits to @unrelentingfox for original script
#!/usr/local/bin/python3
import base64
import hashlib
import hmac
import json
import os
import requests
import time
import urllib.parse
from decimal import Decimal
# SAMPLE OUTPUT
# TICKER AVG COST LAST PRICE GAIN/LOSS
# XETHZUSD 4187.51 3911.61 -6.59%
# KAVAUSD 5.90 5.38 -8.88%
# DOTUSD 40.06 36.15 -9.78%
# XXBTZUSD 56039.83 49928.40 -10.91%
# XDGUSD 0.48 0.43 -9.75%
# Read Kraken API key and secret stored in environment variables
api_url = "https://api.kraken.com"
api_key = os.environ['api_key_KRAKEN']
api_sec = os.environ['api_sec_KRAKEN']
api_call_budget = 0
def get_kraken_signature(urlpath, data, secret):
postdata = urllib.parse.urlencode(data)
encoded = (str(data['nonce']) + postdata).encode()
message = urlpath.encode() + hashlib.sha256(encoded).digest()
mac = hmac.new(base64.b64decode(secret), message, hashlib.sha512)
sigdigest = base64.b64encode(mac.digest())
return sigdigest.decode()
# Attaches auth headers and returns results of a POST request
def kraken_request(uri_path, data, api_key, api_sec):
global api_call_budget
api_call_budget = api_call_budget + 2
if api_call_budget > 15:
print("Kraken requests are throttled..")
time.sleep(6.5)
headers = {}
headers['API-Key'] = api_key
# get_kraken_signature() as defined in the 'Authentication' section
headers['API-Sign'] = get_kraken_signature(uri_path, data, api_sec)
req = requests.post((api_url + uri_path), headers=headers, data=data)
return req
def check_errors(resp):
if resp["error"]:
print("There were errors with the request:")
print(resp["error"])
exit(1)
def get_average_cost():
request_offset = 0
average_cost = {}
min_buy_price = {}
max_buy_price = {}
total_vol = {}
total_cost = {}
total_fees = {}
seen_trades = []
while True:
resp = kraken_request('/0/private/TradesHistory', {
"nonce": str(int(1000*time.time())),
"trades": True,
"ofs": request_offset
}, api_key, api_sec).json()
check_errors(resp)
trade_history = resp["result"]["trades"]
total_trade_count = resp["result"]["count"]
resp_trade_count = len(trade_history)
print("Received trades %d-%d out of %d in total" % (request_offset, request_offset + resp_trade_count, total_trade_count))
for trade_key in filter(lambda tkey: tkey not in seen_trades, trade_history):
seen_trades.append(trade_key)
trade = trade_history[trade_key]
ttype = trade["type"]
currency_pair = trade["pair"]
volume = Decimal(trade["vol"])
cost = Decimal(trade["cost"])
price = Decimal(trade["price"])
fee = Decimal(trade["fee"])
if ttype == "buy":
if price > max_buy_price.get(currency_pair, Decimal(0)):
max_buy_price[currency_pair] = price
if price < min_buy_price.get(currency_pair, Decimal('inf')):
min_buy_price[currency_pair] = price
total_vol[currency_pair] = total_vol.get(currency_pair, Decimal(0)) + volume
total_cost[currency_pair] = total_cost.get(currency_pair, Decimal(0)) + (cost + fee)
elif ttype == "sell":
total_vol[currency_pair] = total_vol.get(currency_pair, Decimal(0)) - volume
total_cost[currency_pair] = total_cost.get(currency_pair, Decimal(0)) - (cost - fee)
total_fees[currency_pair] = total_fees.get(currency_pair, Decimal(0)) + fee
for currency_pair in total_vol:
if total_vol[currency_pair] > 0.0:
average_cost[currency_pair] = total_cost[currency_pair] / total_vol[currency_pair]
request_offset = request_offset + resp_trade_count
if request_offset >= total_trade_count:
break
return average_cost, total_vol, total_cost, total_fees, min_buy_price, max_buy_price
def get_last_price(currency_pairs):
last_price = {}
for currency_pair in currency_pairs:
resp = requests.get('https://api.kraken.com/0/public/Trades?pair=%s' % currency_pair).json()
check_errors(resp)
last_price[currency_pair] = Decimal(resp["result"][currency_pair][-1][0])
return last_price
def calc_gain_loss_percent(average_cost, last_price):
gain_loss_percent = {}
for currency_pair in average_cost:
gain_loss_percent[currency_pair] = (last_price[currency_pair] - average_cost[currency_pair]) / average_cost[currency_pair] * 100
return gain_loss_percent
def display_info(average_cost, last_price, gain_loss_percent, total_vol, total_cost, total_fees, min_buy_price, max_buy_price):
fcolumns = "{0:<12} {1:<12} {2:<12} {3:<12} {4:<12} {5:<12} {6:<12} {7:<12} {8:<12}"
ffloat = "{:.2f}"
fpercent = ffloat+"%"
print(fcolumns.format("TICKER", "AVG COST", "LAST PRICE", "GAIN/LOSS", "TOTAL VOL", "TOTAL COST", "TOTAL FEES", "MIN PRICE", "MAX PRICE"))
for currency_pair in average_cost:
print(fcolumns.format(
currency_pair,
ffloat.format(average_cost[currency_pair]),
ffloat.format(last_price[currency_pair]),
fpercent.format(gain_loss_percent[currency_pair]),
ffloat.format(total_vol[currency_pair]),
ffloat.format(total_cost[currency_pair]),
ffloat.format(total_fees[currency_pair]),
ffloat.format(min_buy_price[currency_pair]),
ffloat.format(max_buy_price[currency_pair])
))
average_cost, total_vol, total_cost, total_fees, min_buy_price, max_buy_price = get_average_cost()
last_price = get_last_price(average_cost.keys())
gain_loss_percent = calc_gain_loss_percent(average_cost, last_price)
display_info(average_cost, last_price, gain_loss_percent, total_vol, total_cost, total_fees, min_buy_price, max_buy_price)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment