#!/usr/bin/env python3
# bch.py - A Bitcoin Cash utility library
# Author: Simon Volpert <simon@simonvolpert.com>
# This program is free software, released under the Apache License, Version 2.0. See the LICENSE file for more information
import urllib.request
import json
import random
import sys
import datetime
import logging
#optional import pycoin.key
#optional import cashaddr # Local library file
MAX_ERRORS = 10
TIMEOUT = 5
exchanges = [
{
'url': 'https://api.coinmarketcap.com/v2/ticker/1831/?convert={cur}',
'price_key': 'data.quotes.{cur}.price',
},
{
'url': 'https://api.coinbase.com/v2/exchange-rates?currency=BCH',
'price_key': 'data.rates.{cur}',
},
{
'url': 'https://apiv2.bitcoinaverage.com/indices/global/ticker/short?crypto=BCH&fiat={cur}',
'price_key': 'BCH{cur}.last',
},
{
# Extremely limited ticker set
'url': 'https://api.kraken.com/0/public/Ticker?pair=BCH{cur}',
'price_key': 'result.BCH{cur}.c.0',
},
]
explorers = [
{
'url': 'https://cashexplorer.bitcoin.com/api/addr/{address}',
'tx_url': 'https://cashexplorer.bitcoin.com/api/tx/{txid}',
'balance_key': None,
'confirmed_key': 'balance',
'unconfirmed_key': 'unconfirmedBalance',
'last_tx_key': 'transactions.0',
'tx_time_key': 'time',
'tx_inputs_key': 'vin',
'tx_in_double_spend_key': 'doubleSpentTxID',
'tx_outputs_key': 'vout',
'tx_out_value_key': 'value',
'tx_out_address_key': 'scriptPubKey.addresses.0',
'tx_double_spend_key': None,
'tx_fee_key': 'fees',
'tx_size_key': 'size',
'tx_confirmations_key': 'confirmations',
'unit_satoshi': False,
'prefixes': '13',
},
{
'url': 'https://blockdozer.com/api/addr/{address}',
'tx_url': 'https://blockdozer.com/api/tx/{txid}',
'balance_key': None,
'confirmed_key': 'balance',
'unconfirmed_key': 'unconfirmedBalance',
'last_tx_key': 'transactions.-1',
'tx_time_key': 'time',
'tx_inputs_key': 'vin',
'tx_in_double_spend_key': 'doubleSpentTxID',
'tx_outputs_key': 'vout',
'tx_out_value_key': 'value',
'tx_out_address_key': 'scriptPubKey.addresses.0',
'tx_double_spend_key': None,
'tx_fee_key': 'fees',
'tx_size_key': 'size',
'tx_confirmations_key': 'confirmations',
'unit_satoshi': False,
'prefixes': 'qp13',
},
{
'url': 'https://bch-insight.bitpay.com/api/addr/{address}',
'tx_url': 'https://bch-insight.bitpay.com/api/tx/{txid}',
'balance_key': 'balance',
'confirmed_key': None,
'unconfirmed_key': 'unconfirmedBalance',
'last_tx_key': 'transactions.0',
'tx_time_key': 'time',
'tx_inputs_key': 'vin',
'tx_in_double_spend_key': 'doubleSpentTxID',
'tx_outputs_key': 'vout',
'tx_out_value_key': 'value',
'tx_out_address_key': 'scriptPubKey.addresses.0',
'tx_double_spend_key': None,
'tx_fee_key': 'fees',
'tx_size_key': 'size',
'tx_confirmations_key': 'confirmations',
'unit_satoshi': False,
'prefixes': 'qp',
},
{
'url': 'https://bch-chain.api.btc.com/v3/address/{address}',
'tx_url': 'https://bch-chain.api.btc.com/v3/tx/{txid}',
'balance_key': 'data.balance',
'confirmed_key': None,
'unconfirmed_key': 'data.unconfirmed_received',
'last_tx_key': 'data.last_tx',
'tx_time_key': 'data.created_at',
'tx_inputs_key': 'data.inputs',
'tx_in_double_spend_key': None,
'tx_outputs_key': 'data.outputs',
'tx_out_value_key': 'value',
'tx_out_address_key': 'addresses.0',
'tx_double_spend_key': 'data.is_double_spend',
'tx_fee_key': 'data.fee',
'tx_size_key': 'data.vsize',
'tx_confirmations_key': 'data.confirmations',
'unit_satoshi': True,
'prefixes': '13',
},
{
'url': 'https://bitcoincash.blockexplorer.com/api/addr/{address}',
'tx_url': 'https://bitcoincash.blockexplorer.com/api/tx/{txid}',
'balance_key': None,
'confirmed_key': 'balance',
'unconfirmed_key': 'unconfirmedBalance',
'last_tx_key': 'transactions.0',
'tx_time_key': 'time',
'tx_inputs_key': 'vin',
'tx_in_double_spend_key': 'doubleSpentTxID',
'tx_outputs_key': 'vout',
'tx_out_value_key': 'value',
'tx_out_address_key': 'scriptPubKey.addresses.0',
'tx_double_spend_key': None,
'tx_fee_key': 'fees',
'tx_size_key': 'size',
'tx_confirmations_key': 'confirmations',
'unit_satoshi': False,
'prefixes': '13',
},
{
'url': 'https://rest.bitbox.earth/v1/address/details/{address}',
'tx_url': 'https://rest.bitbox.earth/v1/transaction/details/{txid}',
'balance_key': None,
'confirmed_key': 'balance',
'unconfirmed_key': 'unconfirmedBalance',
'last_tx_key': 'transactions.0',
'tx_time_key': 'time',
'tx_inputs_key': 'vin',
'tx_in_double_spend_key': 'doubleSpentTxID',
'tx_outputs_key': 'vout',
'tx_out_value_key': 'value',
'tx_out_address_key': 'scriptPubKey.addresses.0',
'tx_double_spend_key': None,
'tx_fee_key': 'fees',
'tx_size_key': 'size',
'tx_confirmations_key': 'confirmations',
'unit_satoshi': False,
'prefixes': 'qp13',
},
]
# Initialize explorer and exchange list
random.seed()
random.shuffle(explorers)
for _server in explorers:
# Using urllib.parse here would require an instruction similar to this:
# _server['name'] = '.'.join(urllib.parse.urlparse(_server['url'])[1].split('.')[-2:])
# Which is neither any shorter, nor simpler, AND requires an additional import to work
_server['name'] = '.'.join(_server['url'].split('/')[2].split('.')[-2:])
for _server in exchanges:
_server['name'] = '.'.join(_server['url'].split('/')[2].split('.')[-2:])
def btc(amount):
'''Return a native bitcoin amount representation'''
result = ('%.8f' % float(amount)).rstrip('0.')
if result == '':
return '0'
return result
def bits(amount):
'''Return the amount represented in bits/cash'''
amount = fiat(float(amount) * 1000000)
if amount.endswith('.00'):
amount = amount[:-3]
return amount
def fiat(amount):
'''Return the amount represented in a dollar/cent notation'''
return ('%.2f' % float(amount))
def color(amount):
'''Return the amount as colorized HTML'''
# Inspired by Thomas Zander's proposal; Reference:
# https://twitter.com/FloweeTheHub/status/996341710027403265
# Only works for white or near-white background!
wholes, sats = '{:.8f}'.format(float(amount)).split('.')
mils = sats[0:3]
bits = sats[3:6]
sats = sats[6:8]
whole_color = 'gray' if wholes == '0' else 'black'
dot_color = 'lightgray' if mils == '000' and bits == '000' and sats == '00' else 'gray'
mil_color = 'lightgray' if mils == '000' and bits == '000' and sats == '00' else 'green'
bit_color = 'lightgray' if bits == '000' and sats == '00' else 'darkgreen'
if sats == '00':
sats = '—'
result = '''<span style="color: {}">{}</span><span style="color: {}; padding-left: 0">.</span><span style="color: {}; padding-left: 0">{}</span><span style="color: {}; padding-left: 0.25em">{}</span><span style="color: gray; padding-left: 0.25em">{}</span>'''.format(whole_color, wholes, dot_color, mil_color, mils, bit_color, bits, sats)
return result
def jsonload(url):
'''Load a web page and return the resulting JSON object'''
request = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(request, timeout=TIMEOUT) as webpage:
data = str(webpage.read(), 'UTF-8')
data = json.loads(data)
return data
def get_value(json_object, key_path):
'''Get the value at the end of a dot-separated key path'''
# Make sure the explorer did not return an error
if 'err_no' in json_object:
if json_object['err_no'] == 1:
raise urllib.error.HTTPError(None, 404, 'Resource Not Found', None, None)
elif json_object['err_no'] == 2:
raise urllib.error.HTTPError(None, 400, 'Parameter Error', None, None)
for k in key_path.split('.'):
# Process integer indices
try:
k = int(k)
except ValueError:
pass
# Expand the key
try:
json_object = json_object[k]
except (TypeError, IndexError):
# The key rightfully doesn't exist
return False
return json_object
def get_price(currency, exchange=exchanges[0]['name']):
'''Get the current Bitcoin Cash price in the desired currency'''
found = False
for server in exchanges:
if server['name'] == exchange:
found = True
break
if not found:
raise KeyError('{src} is not in list of exchanges'.format(src=exchange))
data = jsonload(server['url'].format(cur=currency.upper(), cur_lower=currency.lower()))
rate = float(get_value(data, server['price_key'].format(cur=currency.upper(), cur_lower=currency.lower())))
if rate == 0.0:
raise ValueError('Returned exchange rate is zero')
return round(rate, 2)
def pick_explorer(server_name=None, address_prefix=None, ignore_errors=False):
'''Advance the list of explorers until one that matches the requirements is found'''
for __ in explorers:
# Cycle to the next server
server = explorers.pop(0)
if server is None:
raise StopIteration('Server list depleted')
explorers.append(server)
# Populate server error count if necessary
if 'errors' not in server:
logging.debug('Adding control fields to {} definition'.format(server['name']))
server['errors'] = 0
server['last_error'] = None
server['last_data'] = None
# Filter by server name
if server_name is not None and server['name'] != server_name:
continue
# Filter by error rate
if not ignore_errors and server['errors'] > MAX_ERRORS:
logging.debug('Skipping {} based on error rates'.format(server['name']))
continue
# Filter by address prefix
if address_prefix is not None and address_prefix not in server['prefixes']:
logging.debug('Skipping {} due to unsupported address prefix'.format(server['name']))
continue
return server
raise KeyError('No servers match the requirements')
class AddressInfo(object):
'''A representation of a block explorer's idea of a bitcoin address state
Provided properties:
raw_data (dict) the raw explorer JSON response
address (str) the address in cash format
legacy_address (str) the address in legacy format
confirmed (float) the confirmed balance of the address
unconfirmed (float) the unconfirmed balance of the address
'''
def __init__(self, address, explorer=None, verify=False, ignore_errors=False):
'''Keyword arguments:
address (str) bitcoin_address or tuple(str xpub, int index)
explorer (str) the name of a specific explorer to query
verify (bool) the results should be verified with another explorer
ignore_errors (str) don't skip explorers disabled for excessive errors
'''
# Incompatible parameters
if verify and explorer is not None:
raise ValueError('The "verify" and "explorer" parameters are incompatible')
# Generated address request
xpub = None
idx = None
if type(address) is tuple:
xpub, idx = address
# Strip prefix
elif address[0].lower() == 'b':
address = address.split(':')[1]
# Normalize case
if address[0] in 'QP':
address = address.lower()
# Generate all address versions
if address[0] in 'qp':
self.address = address
self.legacy_address = None
elif address[0] in '13':
self.address = None
self.legacy_address = address
try:
if xpub is not None:
self.address = generate_address(xpub, idx)
self.legacy_address = generate_address(xpub, idx, False)
elif self.address is None:
self.address = convert_address(self.legacy_address)
elif self.legacy_address is None:
self.legacy_address = convert_address(self.address)
except ImportError:
if xpub is not None:
raise
# Add a temporary separator
explorers.append(None)
results = []
# Figure out specific address type availability
if self.address is not None and self.legacy_address is not None:
prefixes = 'qp13'
elif self.legacy_address is None:
prefixes = 'qp'
else:
prefixes = '13'
while explorers[0] is not None:
# Query the next explorer
if prefixes == 'qp13':
server = pick_explorer(explorer, ignore_errors=ignore_errors)
else:
server = pick_explorer(explorer, address_prefix=prefixes[0], ignore_errors=ignore_errors)
# Try to get balance
try:
# Get and cache the received data for possible future analysis
logging.debug('Querying {}'.format(server['name']))
if 'q' in server['prefixes']:
json = jsonload(server['url'].format(address=self.address))
else:
json = jsonload(server['url'].format(address=self.legacy_address))
server['last_data'] = self.raw_data = json
# Conditional balance processing
# TODO: This is a mighty convoluted way of doing it and needs rethinking
if server['confirmed_key'] is not None and server['unconfirmed_key'] is not None:
confirmed = float(get_value(json, server['confirmed_key']))
unconfirmed = float(get_value(json, server['unconfirmed_key']))
elif server['confirmed_key'] is not None and server['balance_key'] is not None:
confirmed = float(get_value(json, server['confirmed_key']))
balance = float(get_value(json, server['balance_key']))
unconfirmed = balance - confirmed
elif server['unconfirmed_key'] is not None and server['balance_key'] is not None:
balance = float(get_value(json, server['balance_key']))
unconfirmed = float(get_value(json, server['unconfirmed_key']))
confirmed = balance - unconfirmed
else:
raise RuntimeError('Cannot figure out address balance')
# Get the last txid
try:
txid = get_value(server['last_data'], server['last_tx_key'])
except (KeyError, IndexError):
txid = None
if not txid:
txid = None
except KeyboardInterrupt:
explorers.remove(None)
raise
except:
server['errors'] += 1
exception = sys.exc_info()[1]
try:
server['last_error'] = str(exception.reason)
except AttributeError:
server['last_error'] = str(exception)
if server['errors'] > MAX_ERRORS:
logging.error('Excessive errors from {server}, disabling. Last error: {error}'.format(server=server['name'], error=server['last_error']))
continue
# Convert balances to native units
if server['unit_satoshi']:
confirmed /= 100000000
unconfirmed /= 100000000
if server['errors'] > 0:
server['errors'] -= 1
data = (confirmed, unconfirmed, txid)
if verify:
if data not in results:
results.append(data)
continue
results.append(data)
break
# If the end of the server list was reached without a single success, assume a network error
explorers.remove(None)
if results == []:
for server in explorers:
if server['errors'] > 0:
server['errors'] -= 1
raise ConnectionError('No results from any known block explorer')
# Populate instance attributes
self.confirmed, self.unconfirmed, self.last_txid = results[-1]
def get_balance(address, explorer=None, verify=False, ignore_errors=False):
'''Get the current balance of an address from a block explorer
Takes the same arguments as AddressInfo()
Returns tuple(confirmed_balance, unconfirmed_balance)
'''
addr = AddressInfo(address, explorer, verify, ignore_errors)
return addr.confirmed, addr.unconfirmed
def get_last_txid(address, explorer=None, verify=False, ignore_errors=False):
'''Get the last tx associated with an address
Takes the same arguments as AddressInfo()
Returns str(txid)
'''
addr = AddressInfo(address, explorer, verify, ignore_errors)
return addr.last_txid
class TxNotFoundError(Exception):
'''Raised when a requested txid is not known to any block explorer'''
class TxInfo(object):
'''A representation of a block explorer's idea of a bitcoin transaction
Provided properties:
raw_data (dict) the raw explorer JSON response
time (datetime) the time this transaction was first seen or mined
outputs (dict) a mapping of receiving addresses to receiving values
both address formats are provided if possible
double_spend (bool) whether or not this transaction has a competing transaction
size (int) the size of the transaction in bytes
fee (float) the transaction fee
fee_per_byte (float) the transaction fee per byte
confirmations (int) the number of confirmations this transaction has
Will raise TxNotFoundError if the passed txid is not known to any explorer
'''
def __init__(self, txid, explorer=None, ignore_errors=None):
'''Keyword arguments:
txid (str) the txid to look for
explorer (str) the name of a specific explorer to query
ignore_errors (str) don't skip explorers disabled for excessive errors
'''
# Add a temporary separator
explorers.append(None)
#tx_size = 10
while explorers[0] is not None:
# Query the next explorer
try:
server = pick_explorer(explorer, ignore_errors=ignore_errors)
except StopIteration:
break
try:
# Get and cache the received data for possible future analysis
logging.debug('Querying {}'.format(server['name']))
json = jsonload(server['tx_url'].format(txid=txid))
server['last_data'] = self.raw_data = json
# Figure out if the tx is a double spend
if server['tx_double_spend_key'] is not None:
self.double_spend = get_value(json, server['tx_double_spend_key'])
else:
self.double_spend = False
for i, __ in enumerate(get_value(json, server['tx_inputs_key'])):
#tx_size += 148
try:
if get_value(json, '.'.join([server['tx_inputs_key'], str(i), server['tx_in_double_spend_key']])) is not None:
self.double_spend = True
# Workaround for explorers that don't provide empty double spend keys
except KeyError as k:
if str(k).strip('\'') != server['tx_in_double_spend_key']:
raise
# Assemble list of output values
self.outputs = {}
for i, __ in enumerate(get_value(json, server['tx_outputs_key'])):
#tx_size += 34
try:
addr = get_value(json, '.'.join([server['tx_outputs_key'], str(i), server['tx_out_address_key']]))
except KeyError:
# Most likely an OP_RETURN
continue
value = float(get_value(json, '.'.join([server['tx_outputs_key'], str(i), server['tx_out_value_key']])))
if server['unit_satoshi']:
value /= 100000000
if addr[0] in 'bB':
addr = addr.lower().split(':')[1]
self.outputs[addr] = value
# Provide both address formats if possible
try:
self.outputs[convert_address(addr)] = value
except ImportError:
pass
# Figure out the tx size and fee
self.fee = float(get_value(json, server['tx_fee_key']))
#self.size = tx_size
self.size = get_value(json, server['tx_size_key'])
if server['unit_satoshi']:
self.fee /= 100000000
self.fee_per_byte = self.fee / self.size * 100000000
self.time = datetime.datetime.fromtimestamp(get_value(json, server['tx_time_key']))
self.confirmations = get_value(json, server['tx_confirmations_key'])
break
except KeyboardInterrupt:
explorers.remove(None)
raise
except:
exception = sys.exc_info()[1]
if isinstance(exception, urllib.error.HTTPError) and exception.code == 404:
continue
server['errors'] += 1
try:
server['last_error'] = str(exception.reason)
except AttributeError:
server['last_error'] = str(exception)
if server['errors'] > MAX_ERRORS:
logging.error('Excessive errors from {server}, disabling. Last error: {error}'.format(server=server['name'], error=server['last_error']))
continue
try:
explorers.remove(None)
except ValueError:
pass
if self.__dict__ == {}:
raise TxNotFoundError('No results from any known block explorer')
def get_tx_propagation(txid, threshold=100, callback=None, stop_on_double_spend=False, ignore_errors=False):
'''Estimate a transaction's propagation across the Bitcoin Cash network
Returns a tuple consisting of:
* The percentage of explorers that are aware of the txid;
* The transaction's double spend status.
Keyword arguments:
txid The txid to query
threshold A percentage at which the propagation check is considered finished
callback A function which will be called after every explorer query
The function will be called with the perliminary results
stop_on_double_spend
The check will be aborted as soon as a double spend is detected
'''
sightings = 0
double_spend = False
num_servers = len(explorers)
propagation = 0
for server in explorers.copy():
if not ignore_errors and 'errors' in server and server['errors'] > MAX_ERRORS:
num_servers -= 1
continue
try:
tx = TxInfo(txid, explorer=server['name'], ignore_errors=ignore_errors)
except TxNotFoundError:
continue
except KeyboardInterrupt:
raise
except:
exception = sys.exc_info()[1]
try:
error = exception.reason
except AttributeError:
error = exception
logging.error('Could not fetch explorer data: {}'.format(error))
continue
if tx.double_spend:
double_spend = True
sightings += 1
propagation = 100 * sightings / num_servers
if callback is not None:
callback(propagation, double_spend)
if propagation >= threshold:
break
elif double_spend and stop_on_double_spend:
break
return propagation, double_spend
def generate_address(xpub, idx, cash=True):
'''Generate a bitcoin cash or bitcoin legacy address from the extended public key at the given index'''
# Optional dependencies if unused
import pycoin.key
import cashaddr
subkey = pycoin.key.Key.from_text(xpub).subkey(0).subkey(idx)
if cash:
return cashaddr.encode('bitcoincash', 0, subkey.hash160())
return subkey.address()
def validate_key(key):
'''Check the validity of a key or an address'''
# Optional dependencies if unused
import pycoin.key
import cashaddr
if ':' in key:
key = key.split(':')[1]
if key[0] in '13x':
try:
pycoin.key.Key.from_text(key)
except pycoin.encoding.EncodingError:
return False
elif key[0] in 'qpQP':
try:
cashaddr.decode('bitcoincash:' + key.lower())
except ValueError:
return False
else:
return False
return True
def convert_address(address):
'''Convert an address back and forth between cash and legacy formats'''
# Optional dependencies if unused
import pycoin.key
import cashaddr
if address[0] in 'bB':
address = address.split(':')[1]
if address[0] in '13':
subkey = pycoin.key.Key.from_text(address)
return cashaddr.encode('bitcoincash', 0, subkey.hash160())
elif address[0] in 'qpQP':
subkey = cashaddr.decode('bitcoincash:' + address.lower())[2]
return pycoin.key.Key(hash160=subkey).address()
else:
raise ValueError('Unsupported address format')
if __name__ == '__main__':
print('===== Known block explorers =====')
for server in explorers:
print(server['name'])
try:
cur = sys.argv[1].upper()
print('\n===== Known exchange rate sources with {cur} support ====='.format(cur=cur))
except IndexError:
print('\n===== Known exchange rate sources =====')
for server in exchanges:
support = True
try:
get_price(cur, exchange=server['name'])
except (KeyError, ValueError, urllib.error.HTTPError):
error = sys.exc_info()[1]
try:
error = error.reason
except AttributeError:
pass
if isinstance(error, KeyError):
error = 'Key error: ' + str(error)
print('{src} does not provide {cur} exchange rate: {error}'.format(src=server['name'], cur=cur, error=error))
support = False
except KeyboardInterrupt:
sys.exit()
except NameError:
pass
if support:
print(server['name'])