freqtrade_origin/freqtrade/rpc/webhook.py

86 lines
3.2 KiB
Python
Raw Normal View History

2018-07-05 19:32:53 +00:00
"""
This module manages webhook communication
"""
import logging
2020-09-28 17:39:41 +00:00
from typing import Any, Dict
2018-07-05 19:32:53 +00:00
2020-09-28 17:39:41 +00:00
from requests import RequestException, post
2018-07-05 19:32:53 +00:00
from freqtrade.rpc import RPC, RPCHandler, RPCMessageType
2018-07-05 19:32:53 +00:00
logger = logging.getLogger(__name__)
logger.debug('Included module rpc.webhook ...')
class Webhook(RPCHandler):
2018-07-05 19:32:53 +00:00
""" This class handles all webhook communication """
def __init__(self, rpc: RPC, config: Dict[str, Any]) -> None:
2018-07-05 19:32:53 +00:00
"""
Init the Webhook class, and init the super class RPCHandler
:param rpc: instance of RPC Helper class
:param config: Configuration object
2018-07-05 19:32:53 +00:00
:return: None
"""
super().__init__(rpc, config)
2018-07-05 19:32:53 +00:00
2018-07-14 11:29:34 +00:00
self._url = self._config['webhook']['url']
2018-07-05 19:32:53 +00:00
2021-02-26 14:46:23 +00:00
self._format = self._config['webhook'].get('format', 'form')
if self._format != 'form' and self._format != 'json':
2021-02-26 18:31:33 +00:00
raise NotImplementedError('Unknown webhook format `{}`, possible values are '
'`form` (default) and `json`'.format(self._format))
2021-02-26 14:46:23 +00:00
2018-07-05 19:32:53 +00:00
def cleanup(self) -> None:
"""
Cleanup pending module resources.
This will do nothing for webhooks, they will simply not be called anymore
"""
pass
def send_msg(self, msg: Dict[str, Any]) -> None:
""" Send a message to telegram channel """
try:
if msg['type'] == RPCMessageType.BUY_NOTIFICATION:
valuedict = self._config['webhook'].get('webhookbuy', None)
2020-02-08 20:02:52 +00:00
elif msg['type'] == RPCMessageType.BUY_CANCEL_NOTIFICATION:
valuedict = self._config['webhook'].get('webhookbuycancel', None)
2018-07-05 19:32:53 +00:00
elif msg['type'] == RPCMessageType.SELL_NOTIFICATION:
valuedict = self._config['webhook'].get('webhooksell', None)
2020-02-08 20:02:52 +00:00
elif msg['type'] == RPCMessageType.SELL_CANCEL_NOTIFICATION:
valuedict = self._config['webhook'].get('webhooksellcancel', None)
elif msg['type'] in (RPCMessageType.STATUS_NOTIFICATION,
RPCMessageType.STARTUP_NOTIFICATION,
RPCMessageType.WARNING_NOTIFICATION):
2018-07-05 19:32:53 +00:00
valuedict = self._config['webhook'].get('webhookstatus', None)
else:
raise NotImplementedError('Unknown message type: {}'.format(msg['type']))
if not valuedict:
2020-09-19 18:04:12 +00:00
logger.info("Message type '%s' not configured for webhooks", msg['type'])
2018-07-05 19:32:53 +00:00
return
2018-07-12 19:54:31 +00:00
payload = {key: value.format(**msg) for (key, value) in valuedict.items()}
2018-07-05 19:32:53 +00:00
self._send_msg(payload)
except KeyError as exc:
logger.exception("Problem calling Webhook. Please check your webhook configuration. "
"Exception: %s", exc)
def _send_msg(self, payload: dict) -> None:
2018-07-14 11:29:34 +00:00
"""do the actual call to the webhook"""
2018-07-05 19:32:53 +00:00
2021-02-26 14:46:23 +00:00
if self._format == 'form':
kwargs = {'data': payload}
elif self._format == 'json':
kwargs = {'json': payload}
else:
raise NotImplementedError('Unknown format: {}'.format(self._format))
2018-07-05 19:32:53 +00:00
try:
2021-02-26 14:46:23 +00:00
post(self._url, **kwargs)
2018-07-05 19:32:53 +00:00
except RequestException as exc:
logger.warning("Could not call webhook url. Exception: %s", exc)