stable/rpc/telegram.py

319 lines
12 KiB
Python
Raw Normal View History

2017-05-14 12:14:16 +00:00
import logging
2017-05-12 17:11:56 +00:00
from datetime import timedelta
2017-05-14 12:14:16 +00:00
import arrow
2017-06-08 20:52:38 +00:00
from sqlalchemy import and_, func
2017-05-21 14:52:36 +00:00
from telegram.error import NetworkError
2017-05-12 17:11:56 +00:00
from telegram.ext import CommandHandler, Updater
2017-05-14 12:14:16 +00:00
from telegram import ParseMode, Bot, Update
from wrapt import synchronized
2017-05-12 17:11:56 +00:00
2017-05-14 12:14:16 +00:00
from persistence import Trade, Session
2017-05-12 17:11:56 +00:00
from exchange import get_exchange_api
from utils import get_conf
2017-05-14 12:14:16 +00:00
# Remove noisy log messages
logging.getLogger('requests.packages.urllib3').setLevel(logging.INFO)
logging.getLogger('telegram').setLevel(logging.INFO)
2017-05-12 17:11:56 +00:00
logger = logging.getLogger(__name__)
_updater = None
conf = get_conf()
api_wrapper = get_exchange_api(conf)
2017-05-14 12:14:16 +00:00
def authorized_only(command_handler):
"""
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 = args[0], 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)
2017-05-14 12:14:16 +00:00
return command_handler(*args, **kwargs)
else:
logger.info('Rejected unauthorized message from: %s', update.message.chat_id)
2017-05-14 12:14:16 +00:00
return wrapper
2017-05-12 17:11:56 +00:00
2017-05-14 12:14:16 +00:00
class TelegramHandler(object):
2017-05-12 17:11:56 +00:00
@staticmethod
2017-05-14 12:14:16 +00:00
@authorized_only
2017-05-12 17:11:56 +00:00
def _status(bot, update):
"""
2017-05-14 12:14:16 +00:00
Handler for /status.
Returns the current TradeThread status
2017-05-12 17:11:56 +00:00
:param bot: telegram bot
:param update: message update
:return: None
"""
# Fetch open trade
trades = Trade.query.filter(Trade.is_open.is_(True)).all()
2017-05-14 12:14:16 +00:00
from main import get_instance
if not get_instance().is_alive():
2017-06-08 18:01:01 +00:00
TelegramHandler.send_msg('*Status:* `trader is not running`', bot=bot)
elif not trades:
TelegramHandler.send_msg('*Status:* `no active order`', bot=bot)
2017-05-12 17:11:56 +00:00
else:
for trade in trades:
# calculate profit and send message to user
current_rate = api_wrapper.get_ticker(trade.pair)['bid']
current_profit = 100 * ((current_rate - trade.open_rate) / trade.open_rate)
orders = api_wrapper.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
message = """
*Trade ID:* `{trade_id}`
2017-06-05 19:17:10 +00:00
*Current Pair:* [{pair}]({market_url})
2017-05-12 17:11:56 +00:00
*Open Since:* `{date}`
*Amount:* `{amount}`
*Open Rate:* `{open_rate}`
*Close Rate:* `{close_rate}`
*Current Rate:* `{current_rate}`
*Close Profit:* `{close_profit}`
2017-05-12 17:11:56 +00:00
*Current Profit:* `{current_profit}%`
*Open Order:* `{open_order}`
2017-08-27 19:17:29 +00:00
""".format(
trade_id=trade.id,
2017-06-05 19:17:10 +00:00
pair=trade.pair,
market_url=api_wrapper.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='{}%'.format(round(trade.close_profit, 2)) if trade.close_profit else None,
current_profit=round(current_profit, 2),
open_order='{} ({})'.format(order['remaining'], order['type']) if order else None,
)
TelegramHandler.send_msg(message, bot=bot)
2017-05-12 17:11:56 +00:00
@staticmethod
2017-05-14 12:14:16 +00:00
@authorized_only
2017-05-12 17:11:56 +00:00
def _profit(bot, update):
"""
2017-05-14 12:14:16 +00:00
Handler for /profit.
Returns a cumulative profit statistics.
2017-05-12 17:11:56 +00:00
:param bot: telegram bot
:param update: message update
:return: None
"""
2017-06-08 20:52:38 +00:00
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 = api_wrapper.get_ticker(trade.pair)['bid']
profit = 100 * ((current_rate - trade.open_rate) / trade.open_rate)
profit_amounts.append((profit / 100) * trade.btc_amount)
profits.append(profit)
bp_pair, bp_rate = Session.query(Trade.pair, func.sum(Trade.close_profit).label('profit_sum')) \
.filter(Trade.is_open.is_(False)) \
.group_by(Trade.pair) \
.order_by('profit_sum DESC') \
.first()
2017-05-12 17:11:56 +00:00
markdown_msg = """
2017-06-05 19:17:10 +00:00
*ROI:* `{profit_btc} ({profit}%)`
2017-05-12 17:11:56 +00:00
*Trade Count:* `{trade_count}`
2017-06-08 20:52:38 +00:00
*First Trade opened:* `{first_trade_date}`
*Latest Trade opened:* `{latest_trade_date}`
*Avg. Duration:* `{avg_duration}`
*Best Performing:* `{best_pair}: {best_rate}%`
2017-08-27 19:17:29 +00:00
""".format(
2017-06-08 20:52:38 +00:00
profit_btc=round(sum(profit_amounts), 8),
profit=round(sum(profits), 2),
trade_count=len(trades),
2017-05-12 17:11:56 +00:00
first_trade_date=arrow.get(trades[0].open_date).humanize(),
latest_trade_date=arrow.get(trades[-1].open_date).humanize(),
2017-06-08 20:52:38 +00:00
avg_duration=str(timedelta(seconds=sum(durations) / float(len(durations)))).split('.')[0],
best_pair=bp_pair,
best_rate=round(bp_rate, 2),
2017-05-12 17:11:56 +00:00
)
TelegramHandler.send_msg(markdown_msg, bot=bot)
@staticmethod
2017-05-14 12:14:16 +00:00
@authorized_only
2017-05-12 17:11:56 +00:00
def _start(bot, update):
"""
2017-05-14 12:14:16 +00:00
Handler for /start.
Starts TradeThread
2017-05-12 17:11:56 +00:00
:param bot: telegram bot
:param update: message update
:return: None
"""
2017-05-14 12:14:16 +00:00
from main import get_instance
if get_instance().is_alive():
2017-05-12 17:11:56 +00:00
TelegramHandler.send_msg('*Status:* `already running`', bot=bot)
else:
2017-05-14 12:14:16 +00:00
get_instance(recreate=True).start()
2017-05-12 17:11:56 +00:00
@staticmethod
2017-05-14 12:14:16 +00:00
@authorized_only
2017-05-12 17:11:56 +00:00
def _stop(bot, update):
"""
2017-05-14 12:14:16 +00:00
Handler for /stop.
Stops TradeThread
2017-05-12 17:11:56 +00:00
:param bot: telegram bot
:param update: message update
:return: None
"""
from main import get_instance
2017-05-14 12:14:16 +00:00
if get_instance().is_alive():
TelegramHandler.send_msg('`Stopping trader ...`', bot=bot)
get_instance().stop()
2017-05-12 17:11:56 +00:00
else:
TelegramHandler.send_msg('*Status:* `already stopped`', bot=bot)
2017-05-14 12:14:16 +00:00
2017-06-08 18:01:01 +00:00
@staticmethod
@authorized_only
def _forcesell(bot, update):
"""
Handler for /forcesell <id>.
Sells the given trade at current price
:param bot: telegram bot
:param update: message update
:return: None
"""
from main import get_instance
if not get_instance().is_alive():
TelegramHandler.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:
TelegramHandler.send_msg('There is no open trade with ID: `{}`'.format(trade_id))
return
# Get current rate
current_rate = api_wrapper.get_ticker(trade.pair)['bid']
# Get available balance
currency = trade.pair.split('_')[1]
balance = api_wrapper.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('_', '/'),
api_wrapper.get_pair_detail_url(trade.pair),
trade.close_rate,
round(profit, 2)
)
logger.info(message)
TelegramHandler.send_msg(message)
except ValueError:
TelegramHandler.send_msg('Invalid argument. Usage: `/forcesell <trade_id>`')
logger.warning('/forcesell: Invalid argument received')
2017-06-08 20:52:38 +00:00
@staticmethod
@authorized_only
def _performance(bot, update):
"""
Handler for /performance.
Shows a performance statistic from finished trades
:param bot: telegram bot
:param update: message update
:return: None
"""
from main import get_instance
if not get_instance().is_alive():
TelegramHandler.send_msg('`trader is not running`', bot=bot)
return
pair_rates = Session.query(Trade.pair, func.sum(Trade.close_profit).label('profit_sum')) \
.filter(Trade.is_open.is_(False)) \
.group_by(Trade.pair) \
.order_by('profit_sum DESC') \
.all()
stats = '\n'.join('{}. <code>{}\t{}%</code>'.format(i + 1, pair, round(rate, 2)) for i, (pair, rate) in enumerate(pair_rates))
message = '<b>Performance:</b>\n{}\n'.format(stats)
logger.debug(message)
TelegramHandler.send_msg(message, parse_mode=ParseMode.HTML)
2017-05-14 12:14:16 +00:00
@staticmethod
@synchronized
def get_updater(conf):
"""
Returns the current telegram updater instantiates a new one
:param conf:
:return: telegram.ext.Updater
"""
global _updater
if not _updater:
_updater = Updater(token=conf['telegram']['token'], workers=0)
return _updater
@staticmethod
def listen():
"""
Registers all known command handlers and starts polling for message updates
:return: None
"""
# Register command handler and start telegram message polling
handles = [
CommandHandler('status', TelegramHandler._status),
CommandHandler('profit', TelegramHandler._profit),
CommandHandler('start', TelegramHandler._start),
CommandHandler('stop', TelegramHandler._stop),
2017-06-08 18:01:01 +00:00
CommandHandler('forcesell', TelegramHandler._forcesell),
2017-06-08 20:52:38 +00:00
CommandHandler('performance', TelegramHandler._performance),
2017-05-14 12:14:16 +00:00
]
for handle in handles:
TelegramHandler.get_updater(conf).dispatcher.add_handler(handle)
2017-06-08 23:06:29 +00:00
TelegramHandler.get_updater(conf).start_polling(
clean=True,
bootstrap_retries=3,
timeout=30,
read_latency=60,
)
2017-05-14 12:14:16 +00:00
logger.info('TelegramHandler is listening for following commands: {}'
.format([h.command for h in handles]))
@staticmethod
def send_msg(msg, bot=None, parse_mode=ParseMode.MARKDOWN):
2017-05-14 12:14:16 +00:00
"""
Send given markdown message
:param msg: message
:param bot: alternative bot
:param parse_mode: telegram parse mode
2017-05-14 12:14:16 +00:00
:return: None
"""
if conf['telegram'].get('enabled', False):
try:
2017-05-20 19:30:42 +00:00
bot = bot or TelegramHandler.get_updater(conf).bot
try:
bot.send_message(conf['telegram']['chat_id'], msg, parse_mode=parse_mode)
2017-08-27 13:40:27 +00:00
except NetworkError as error:
2017-05-20 19:30:42 +00:00
# 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)
2017-05-20 19:30:42 +00:00
bot.send_message(conf['telegram']['chat_id'], msg, parse_mode=parse_mode)
2017-05-14 12:14:16 +00:00
except Exception:
logger.exception('Exception occurred within Telegram API')