nxy/bot/coins.py

# -*- coding: utf-8 -*-
import requests
from docopt import Dict
from irc3.plugins.command import command
from irc3.utils import IrcString

from . import Plugin

def _currency(name, prefix=None, suffix=None):
    def tostr(value):
        return '{prefix}{value:,.2f}{suffix}'.format(
                prefix=prefix or '',
                suffix=suffix or ('' if (prefix or suffix) else (' ' + name)),
                value=value)
    return tostr

class Coins(Plugin):
    API_URL = 'https://api.coingecko.com/api/v3/simple/price'

    CURRENCIES = {
        'usd': _currency('usd', prefix='$'),
        'eur': _currency('eur', suffix='€'),
        'eth': _currency('eth', suffix='Ξ'),
        'btc': _currency('btc', suffix='฿'),
        'xmr': _currency('xmr'),
        'xrp': _currency('xrp')
    }

    CRYPTO_IDS = {
        'btc': 'bitcoin',
        'eth': 'ethereum',
        'xmr': 'monero',
        'xrp': 'ripple'
    }

    @command
    def btc(self, mask: IrcString, target: IrcString, args: Dict):
        """Gets the Bitcoin values from coingecko.

        %%btc [<currency>]
        """
        return self.coingecko_summary('btc', args)

    @command
    def eth(self, mask: IrcString, target: IrcString, args: Dict):
        """Gets the Ethereum values from coingecko.

        %%eth [<currency>]
        """
        return self.coingecko_summary('eth', args)

    @command
    def xmr(self, mask: IrcString, target: IrcString, args: Dict):
        """Gets the Monero values from coingecko.

        %%xmr [<currency>]
        """
        return self.coingecko_summary('xmr', args)

    @command
    def xrp(self, mask: IrcString, target: IrcString, args: Dict):
        """Gets the Ripple values from coingecko.

        %%xrp [<currency>]
        """
        return self.coingecko_summary('xrp', args)

    def coingecko_summary(self, crypto: str, args: Dict):
        crypto_id = self.CRYPTO_IDS.get(crypto)
        if crypto_id is None:
            raise ValueError

        currency = args.get('<currency>', 'eur').lower()

        def format_response(response: str) -> str:
            return f'\x02[{crypto.upper()}]\x02 {response}'

        if currency not in self.CURRENCIES or crypto == currency:
            return format_response(f'Can\'t convert or invalid currency: {currency}')

        res = requests.get(self.API_URL, params={
            'ids': crypto_id,
            'vs_currencies': currency,
            'include_market_cap': 'true',
            'include_24hr_vol': 'true',
            'include_24hr_change': 'true'
        })

        if res.status_code == 429:
            return format_response('Rate Limit exceeded')

        if res.status_code != 200:
            return format_response('API Error')

        data = res.json()[crypto_id]

        if currency not in data:
            return format_response('No data received')

        to_currency = self.CURRENCIES[currency]

        return format_response(
               'Price: \x02\x0307{price}\x0F - ' \
               'Change: \x02\x0307{change:,.2f}%\x0F - ' \
               'Volume: \x02\x0307{volume}\x0F - ' \
               'Market Cap: \x02\x0307{market_cap}\x0F' \
            .format(crypto=crypto.upper(),
                    price=to_currency(data[currency]),
                    change=data[f'{currency}_24h_change'] * 100,
                    volume=data[f'{currency}_24h_vol'],
                    market_cap=to_currency(data[f'{currency}_market_cap'])))