stable/rpc/telegram.py

import logging
from datetime import timedelta
from typing import Callable, Any

import arrow
from sqlalchemy import and_, func, text
from telegram.error import NetworkError
from telegram.ext import CommandHandler, Updater
from telegram import ParseMode, Bot, Update

from misc import get_state, State, update_state
from persistence import Trade

import exchange

# Remove noisy log messages
logging.getLogger('requests.packages.urllib3').setLevel(logging.INFO)
logging.getLogger('telegram').setLevel(logging.INFO)
logger = logging.getLogger(__name__)

_updater = None
_CONF = {}


def init(config: dict) -> None:
    """
    Initializes this module with the given config,
    registers all known command handlers
    and starts polling for message updates
    :param config: config to use
    :return: None
    """
    global _updater

    _CONF.update(config)
    if not _CONF['telegram']['enabled']:
        return

    _updater = Updater(token=config['telegram']['token'], workers=0)

    # Register command handler and start telegram message polling
    handles = [
        CommandHandler('status', _status),
        CommandHandler('profit', _profit),
        CommandHandler('start', _start),
        CommandHandler('stop', _stop),
        CommandHandler('forcesell', _forcesell),
        CommandHandler('performance', _performance),
    ]
    for handle in handles:
        _updater.dispatcher.add_handler(handle)
    _updater.start_polling(
        clean=True,
        bootstrap_retries=3,
        timeout=30,
        read_latency=60,
    )
    logger.info(
        'rpc.telegram is listening for following commands: %s',
        [h.command for h in handles]
    )


def authorized_only(command_handler: Callable[[Bot, Update], None]) -> Callable[..., Any]:
    """
    Decorator to check if the message comes from the correct chat_id
    :param command_handler: Telegram CommandHandler
    :return: decorated function
    """
    def wrapper(*args, **kwargs):
        bot, update = kwargs.get('bot') or args[0], kwargs.get('update') or args[1]

        if not isinstance(bot, Bot) or not isinstance(update, Update):
            raise ValueError('Received invalid Arguments: {}'.format(*args))

        chat_id = int(_CONF['telegram']['chat_id'])
        if int(update.message.chat_id) == chat_id:
            logger.info('Executing handler: %s for chat_id: %s', command_handler.__name__, chat_id)
            return command_handler(*args, **kwargs)
        else:
            logger.info('Rejected unauthorized message from: %s', update.message.chat_id)
    return wrapper


@authorized_only
def _status(bot: Bot, update: Update) -> None:
    """
    Handler for /status.
    Returns the current TradeThread status
    :param bot: telegram bot
    :param update: message update
    :return: None
    """
    # Fetch open trade
    trades = Trade.query.filter(Trade.is_open.is_(True)).all()
    if get_state() != State.RUNNING:
        send_msg('*Status:* `trader is not running`', bot=bot)
    elif not trades:
        send_msg('*Status:* `no active order`', bot=bot)
    else:
        for trade in trades:
            # calculate profit and send message to user
            current_rate = exchange.get_ticker(trade.pair)['bid']
            current_profit = 100 * ((current_rate - trade.open_rate) / trade.open_rate)
            orders = exchange.get_open_orders(trade.pair)
            orders = [o for o in orders if o['id'] == trade.open_order_id]
            order = orders[0] if orders else None

            fmt_close_profit = '{:.2f}%'.format(
                round(trade.close_profit, 2)
            ) if trade.close_profit else None
            message = """
*Trade ID:* `{trade_id}`
*Current Pair:* [{pair}]({market_url})
*Open Since:* `{date}`
*Amount:* `{amount}`
*Open Rate:* `{open_rate}`
*Close Rate:* `{close_rate}`
*Current Rate:* `{current_rate}`
*Close Profit:* `{close_profit}`
*Current Profit:* `{current_profit:.2f}%`
*Open Order:* `{open_order}`
            """.format(
                trade_id=trade.id,
                pair=trade.pair,
                market_url=exchange.get_pair_detail_url(trade.pair),
                date=arrow.get(trade.open_date).humanize(),
                open_rate=trade.open_rate,
                close_rate=trade.close_rate,
                current_rate=current_rate,
                amount=round(trade.amount, 8),
                close_profit=fmt_close_profit,
                current_profit=round(current_profit, 2),
                open_order='{} ({})'.format(order['remaining'], order['type']) if order else None,
            )
            send_msg(message, bot=bot)


@authorized_only
def _profit(bot: Bot, update: Update) -> None:
    """
    Handler for /profit.
    Returns a cumulative profit statistics.
    :param bot: telegram bot
    :param update: message update
    :return: None
    """
    trades = Trade.query.order_by(Trade.id).all()

    profit_amounts = []
    profits = []
    durations = []
    for trade in trades:
        if trade.close_date:
            durations.append((trade.close_date - trade.open_date).total_seconds())
        if trade.close_profit:
            profit = trade.close_profit
        else:
            # Get current rate
            current_rate = exchange.get_ticker(trade.pair)['bid']
            profit = 100 * ((current_rate - trade.open_rate) / trade.open_rate)

        profit_amounts.append((profit / 100) * trade.stake_amount)
        profits.append(profit)

    best_pair = Trade.session.query(Trade.pair, func.sum(Trade.close_profit).label('profit_sum')) \
        .filter(Trade.is_open.is_(False)) \
        .group_by(Trade.pair) \
        .order_by(text('profit_sum DESC')) \
        .first()

    if not best_pair:
        send_msg('*Status:* `no closed trade`', bot=bot)
        return

    bp_pair, bp_rate = best_pair
    markdown_msg = """
*ROI:* `{profit_btc:.2f} ({profit:.2f}%)`
*Trade Count:* `{trade_count}`
*First Trade opened:* `{first_trade_date}`
*Latest Trade opened:* `{latest_trade_date}`
*Avg. Duration:* `{avg_duration}`
*Best Performing:* `{best_pair}: {best_rate:.2f}%`
    """.format(
        profit_btc=round(sum(profit_amounts), 8),
        profit=round(sum(profits), 2),
        trade_count=len(trades),
        first_trade_date=arrow.get(trades[0].open_date).humanize(),
        latest_trade_date=arrow.get(trades[-1].open_date).humanize(),
        avg_duration=str(timedelta(seconds=sum(durations) / float(len(durations)))).split('.')[0],
        best_pair=bp_pair,
        best_rate=round(bp_rate, 2),
    )
    send_msg(markdown_msg, bot=bot)


@authorized_only
def _start(bot: Bot, update: Update) -> None:
    """
    Handler for /start.
    Starts TradeThread
    :param bot: telegram bot
    :param update: message update
    :return: None
    """
    if get_state() == State.RUNNING:
        send_msg('*Status:* `already running`', bot=bot)
    else:
        update_state(State.RUNNING)


@authorized_only
def _stop(bot: Bot, update: Update) -> None:
    """
    Handler for /stop.
    Stops TradeThread
    :param bot: telegram bot
    :param update: message update
    :return: None
    """
    if get_state() == State.RUNNING:
        send_msg('`Stopping trader ...`', bot=bot)
        update_state(State.STOPPED)
    else:
        send_msg('*Status:* `already stopped`', bot=bot)


@authorized_only
def _forcesell(bot: Bot, update: Update) -> None:
    """
    Handler for /forcesell <id>.
    Sells the given trade at current price
    :param bot: telegram bot
    :param update: message update
    :return: None
    """
    if get_state() != State.RUNNING:
        send_msg('`trader is not running`', bot=bot)
        return

    try:
        trade_id = int(update.message.text
                       .replace('/forcesell', '')
                       .strip())
        # Query for trade
        trade = Trade.query.filter(and_(
            Trade.id == trade_id,
            Trade.is_open.is_(True)
        )).first()
        if not trade:
            send_msg('There is no open trade with ID: `{}`'.format(trade_id))
            return
        # Get current rate
        current_rate = exchange.get_ticker(trade.pair)['bid']
        # Get available balance
        currency = trade.pair.split('_')[1]
        balance = exchange.get_balance(currency)
        # Execute sell
        profit = trade.exec_sell_order(current_rate, balance)
        message = '*{}:* Selling [{}]({}) at rate `{:f} (profit: {}%)`'.format(
            trade.exchange.name,
            trade.pair.replace('_', '/'),
            exchange.get_pair_detail_url(trade.pair),
            trade.close_rate,
            round(profit, 2)
        )
        logger.info(message)
        send_msg(message)

    except ValueError:
        send_msg('Invalid argument. Usage: `/forcesell <trade_id>`')
        logger.warning('/forcesell: Invalid argument received')


@authorized_only
def _performance(bot: Bot, update: Update) -> None:
    """
    Handler for /performance.
    Shows a performance statistic from finished trades
    :param bot: telegram bot
    :param update: message update
    :return: None
    """
    if get_state() != State.RUNNING:
        send_msg('`trader is not running`', bot=bot)
        return

    pair_rates = Trade.session.query(Trade.pair, func.sum(Trade.close_profit).label('profit_sum')) \
        .filter(Trade.is_open.is_(False)) \
        .group_by(Trade.pair) \
        .order_by(text('profit_sum DESC')) \
        .all()

    stats = '\n'.join('{index}. <code>{pair}\t{profit:.2f}%</code>'.format(
        index=i + 1,
        pair=pair,
        profit=round(rate, 2)
    ) for i, (pair, rate) in enumerate(pair_rates))

    message = '<b>Performance:</b>\n{}\n'.format(stats)
    logger.debug(message)
    send_msg(message, parse_mode=ParseMode.HTML)


def send_msg(msg: str, bot: Bot = None, parse_mode: ParseMode = ParseMode.MARKDOWN) -> None:
    """
    Send given markdown message
    :param msg: message
    :param bot: alternative bot
    :param parse_mode: telegram parse mode
    :return: None
    """
    if _CONF['telegram'].get('enabled', False):
        try:
            bot = bot or _updater.bot
            try:
                bot.send_message(_CONF['telegram']['chat_id'], msg, parse_mode=parse_mode)
            except NetworkError as error:
                # Sometimes the telegram server resets the current connection,
                # if this is the case we send the message again.
                logger.warning(
                    'Got Telegram NetworkError: %s! Trying one more time.',
                    error.message
                )
                bot.send_message(_CONF['telegram']['chat_id'], msg, parse_mode=parse_mode)
        except Exception:
            logger.exception('Exception occurred within Telegram API')