Crypto balances retrieval script in Python

Script Documentation: Crypto Balances Retrieval

Overview

This script retrieves and calculates the total USD value of crypto balances from Binance, KuCoin, and Coinbase exchanges.

Script Code

import ccxt
from coinbase.wallet.client import Client
import json
import sqlite3
from datetime import datetime
import numpy as np
import matplotlib.pyplot as plt

# Load API keys from configuration file
def load_api_keys():
    with open('keys.json') as f:
        keys = json.load(f)
    return keys

# Load API keys configuration
keys = load_api_keys()

# Initialize exchanges and Coinbase client
binance_exchanges = []
for binance_key in keys['binance']:
    exchange = ccxt.binance({
        'apiKey': binance_key['api_key'],
        'secret': binance_key['api_secret'],
    })
    binance_exchanges.append(exchange)

kucoin_exchange = ccxt.kucoin({
    'apiKey': keys['kucoin']['api_key'],
    'secret': keys['kucoin']['api_secret'],
    'password': keys['kucoin']['passphrase'],
})

coinbase_client = Client(keys['coinbase']['api_key'], keys['coinbase']['api_secret'])

# SQLite database initialization
def init_db():
    conn = sqlite3.connect('balances.db')
    c = conn.cursor()
    c.execute('''CREATE TABLE IF NOT EXISTS balances
                 (id INTEGER PRIMARY KEY, exchange TEXT, balance_usd REAL, timestamp DATETIME)''')
    conn.commit()
    return conn

conn = init_db()

# Function to retrieve current USD balance from Binance
def get_binance_usd_balance():
    try:
        total_usd_value = 0.0
        for exchange in binance_exchanges:
            balances = exchange.fetch_balance()
            for symbol in balances['total']:
                if balances['total'][symbol] > 0:
                    try:
                        if symbol == 'USDT':
                            usd_value = balances['total'][symbol]
                        else:
                            ticker = exchange.fetch_ticker(f"{symbol}/USDT")
                            if ticker and 'last' in ticker and ticker['last'] is not None:
                                usd_value = balances['total'][symbol] * ticker['last']
                            else:
                                raise ValueError("Ticker data not available")
                        total_usd_value += usd_value
                    except Exception:
                        continue
        return total_usd_value
    except Exception as e:
        print(f"Error fetching balances from Binance: {str(e)}")
        return None

# Function to retrieve current USD balance from KuCoin
def get_kucoin_usd_balance():
    try:
        total_usd_value = 0.0
        balances = kucoin_exchange.fetch_balance()
        for symbol in balances['total']:
            if balances['total'][symbol] > 0:
                try:
                    if symbol == 'USDT':
                        usd_value = balances['total'][symbol]
                    else:
                        ticker = kucoin_exchange.fetch_ticker(f"{symbol}/USDT")
                        if ticker and 'last' in ticker and ticker['last'] is not None:
                            usd_value = balances['total'][symbol] * ticker['last']
                        else:
                            raise ValueError("Ticker data not available")
                    total_usd_value += usd_value
                except Exception:
                    continue
        return total_usd_value
    except Exception as e:
        print(f"Error fetching balances from KuCoin: {str(e)}")
        return None

# Function to retrieve current USD balance from Coinbase
def get_coinbase_usd_balance():
    try:
        total_usd_value = 0.0
        accounts = coinbase_client.get_accounts()
        for account in accounts['data']:
            balance = float(account['balance']['amount'])
            currency = account['balance']['currency']
            if balance > 0:
                try:
                    if currency == 'USD' or currency == 'USDT':
                        usd_value = balance
                    else:
                        ticker = coinbase_client.get_spot_price(currency_pair=f"{currency}-USD")
                        if ticker and ticker['amount'] is not None:
                            usd_value = balance * float(ticker['amount'])
                        else:
                            raise ValueError("Ticker data not available")
                    total_usd_value += usd_value
                except Exception:
                    continue
        return total_usd_value
    except Exception as e:
        print(f"Error fetching balances from Coinbase: {str(e)}")
        return None

# Function to calculate total USD value of all tokens
def get_total_usd_value():
    binance_usd_balance = get_binance_usd_balance() or 0.0
    kucoin_usd_balance = get_kucoin_usd_balance() or 0.0
    coinbase_usd_balance = get_coinbase_usd_balance() or 0.0
    total_usd_value = binance_usd_balance + kucoin_usd_balance + coinbase_usd_balance
    return binance_usd_balance, kucoin_usd_balance, coinbase_usd_balance, total_usd_value

# Function to save balances into the database
def save_balances(binance_usd, kucoin_usd, coinbase_usd, total_usd):
    timestamp = datetime.now()
    conn.execute("INSERT INTO balances (exchange, balance_usd, timestamp) VALUES (?, ?, ?)", 
                 ('Binance', binance_usd, timestamp))
    conn.execute("INSERT INTO balances (exchange, balance_usd, timestamp) VALUES (?, ?, ?)", 
                 ('KuCoin', kucoin_usd, timestamp))
    conn.execute("INSERT INTO balances (exchange, balance_usd, timestamp) VALUES (?, ?, ?)", 
                 ('Coinbase', coinbase_usd, timestamp))
    conn.execute("INSERT INTO balances (exchange, balance_usd, timestamp) VALUES (?, ?, ?)", 
                 ('Total', total_usd, timestamp))
    conn.commit()

# Function to retrieve daily averages from the database
def get_daily_averages():
    c = conn.cursor()
    c.execute("SELECT date(timestamp) as day, AVG(balance_usd) as avg_balance FROM balances WHERE exchange = 'Total' GROUP BY day")
    rows = c.fetchall()
    return rows

# Function to create ASCII daily portfolio graph
def create_ascii_graph(data):
    days = [row[0] for row in data]
    balances = [row[1] for row in data]
    max_balance = max(balances)
    min_balance = min(balances)
    range_balance = max_balance - min_balance
    height = 10
    width = len(days)
    
    for i in range(height, 0, -1):
        line = ""
        for balance in balances:
            if balance >= min_balance + i * range_balance / height:
                line += "█"
            else:
                line += " "
        print(line)
    print("".join(["-" for _ in range(width)]))
    print("".join([str(i % 10) for i in range(width)]))  # X-axis labels (0-9 repeated)

# Run calculations, save to DB, and print total USD value
if __name__ == "__main__":
    binance_usd_balance, kucoin_usd_balance, coinbase_usd_balance, total_usd_value = get_total_usd_value()
    save_balances(binance_usd_balance, kucoin_usd_balance, coinbase_usd_balance, total_usd_value)
    
    print(f"Total USD value on Binance: {binance_usd_balance}")
    print(f"Total USD value on KuCoin: {kucoin_usd_balance}")
    print(f"Total USD value on Coinbase: {coinbase_usd_balance}")
    print(f"\nTotal USD value of all tokens: {total_usd_value}")
    
    daily_averages = get_daily_averages()
    create_ascii_graph(daily_averages)

Example JSON Configuration

{
  "binance": [
    {
      "api_key": "your_binance_api_key_here",
      "api_secret": "your_binance_api_secret_here"
    }
  ],
  "kucoin": {
    "api_key": "your_kucoin_api_key_here",
    "api_secret": "your_kucoin_api_secret_here",
    "passphrase": "your_kucoin_passphrase_here"
  },
  "coinbase": {
    "api_key": "your_coinbase_api_key_here",
    "api_secret": "your_coinbase_api_secret_here"
 

}
}