Source code for i3pystatus.bitcoin

import urllib.request
import json
from datetime import datetime

from i3pystatus import IntervalModule
from i3pystatus.core.util import internet, require, user_open

import locale
import threading
from contextlib import contextmanager

LOCALE_LOCK = threading.Lock()


@contextmanager
def setlocale(name):
    # To deal with locales only in this module and keep it thread save
    with LOCALE_LOCK:
        saved = locale.setlocale(locale.LC_ALL)
        try:
            yield locale.setlocale(locale.LC_ALL, name)
        finally:
            locale.setlocale(locale.LC_ALL, saved)


[docs]class Bitcoin(IntervalModule): """ This module fetches and displays current Bitcoin market prices and optionally monitors transactions to and from a list of user-specified wallet addresses. Market data is pulled from the BitcoinAverage Price Index API <https://bitcoinaverage.com> and it is possible to specify the exchange to be monitored. Transaction data is pulled from blockchain.info <https://blockchain.info/api/blockchain_api>. .. rubric:: Available formatters * {last_price} * {ask_price} * {bid_price} * {daily_average} * {volume} * {volume_thousand} * {volume_percent} * {age} * {status} * {last_tx_type} * {last_tx_addr} * {last_tx_value} * {balance_btc} * {balance_fiat} * {symbol} """ settings = ( ("format", "Format string used for output."), ("currency", "Base fiat currency used for pricing."), ("wallet_addresses", "List of wallet address(es) to monitor."), ("color", "Standard color"), ("exchange", "Get ticker from a custom exchange instead"), ("colorize", "Enable color change on price increase/decrease"), ("color_up", "Color for price increases"), ("color_down", "Color for price decreases"), ("interval", "Update interval."), ("symbol", "Symbol for bitcoin sign"), "status" ) format = "{symbol} {status}{last_price}" currency = "USD" exchange = None symbol = "฿" wallet_addresses = "" color = "#FFFFFF" colorize = False color_up = "#00FF00" color_down = "#FF0000" interval = 600 status = { "price_up": "▲", "price_down": "▼", } on_leftclick = "electrum" on_rightclick = ["open_something", "https://bitcoinaverage.com/"] _price_prev = 0 def _get_age(self, bitcoinaverage_timestamp): with setlocale('C'): # Deal with locales (months name differ) # Assume format is always utc, to avoid import pytz diff = datetime.utcnow() - \ datetime.fromtimestamp(bitcoinaverage_timestamp) return int(diff.total_seconds()) def _query_api(self, api_url): url = "{}BTC{}".format(api_url, self.currency.upper()) response = urllib.request.urlopen(url).read().decode("utf-8") return json.loads(response) def _fetch_price_data(self): if self.exchange is None: api_url = "https://apiv2.bitcoinaverage.com/indices/global/ticker/" return self._query_api(api_url) else: api_url = "https://api.bitcoinaverage.com/exchanges/" ret = self._query_api(api_url) exchange = ret[self.exchange.lower()] # Adapt values to global ticker format exchange['ask'] = exchange['rates']['ask'] exchange['bid'] = exchange['rates']['bid'] exchange['last'] = exchange['rates']['last'] exchange['24h_avg'] = None exchange['timestamp'] = ret['timestamp'] return exchange def _fetch_blockchain_data(self): api = "https://blockchain.info/multiaddr?active=" addresses = "|".join(self.wallet_addresses) url = "{}{}".format(api, addresses) return json.loads(urllib.request.urlopen(url).read().decode("utf-8")) @require(internet) def run(self): price_data = self._fetch_price_data() fdict = { "symbol": self.symbol, "daily_average": price_data["averages"]["day"], "ask_price": price_data["ask"], "bid_price": price_data["bid"], "last_price": price_data["last"], "volume": price_data["volume"], "volume_thousand": float(price_data["volume"]) / 1000, "volume_percent": price_data["volume_percent"], "age": self._get_age(price_data['timestamp']) } if self._price_prev and fdict["last_price"] > self._price_prev: color = self.color_up fdict["status"] = self.status["price_up"] elif self._price_prev and fdict["last_price"] < self._price_prev: color = self.color_down fdict["status"] = self.status["price_down"] else: color = self.color fdict["status"] = "" self._price_prev = fdict["last_price"] if not self.colorize: color = self.color if self.wallet_addresses: blockchain_data = self._fetch_blockchain_data() wallet_data = blockchain_data["wallet"] balance_btc = wallet_data["final_balance"] / 100000000 fdict["balance_btc"] = round(balance_btc, 2) balance_fiat = fdict["balance_btc"] * fdict["last_price"] fdict["balance_fiat"] = round(balance_fiat, 2) fdict["total_sent"] = wallet_data["total_sent"] fdict["total_received"] = wallet_data["total_received"] fdict["transactions"] = wallet_data["n_tx"] if fdict["transactions"]: last_tx = blockchain_data["txs"][0] fdict["last_tx_addr"] = last_tx["out"][0]["addr"] fdict["last_tx_value"] = last_tx["out"][0]["value"] / 100000000 if fdict["last_tx_addr"] in self.wallet_addresses: fdict["last_tx_type"] = "recv" else: fdict["last_tx_type"] = "sent" self.data = fdict self.output = { "full_text": self.format.format(**fdict), "color": color, }
[docs] def open_something(self, url_or_command): """ Wrapper function, to pass the arguments to user_open """ user_open(url_or_command)