4131167254

Build Type

push

github-actions

Committed by GitHub

Commit Message

Merge pull request #7983 from stash86/bt-metrics

Run Details

16866 of 17748 relevant lines covered (95.03%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

92.86

/freqtrade/rpc/webhook.py

"""
This module manages webhook communication
"""
import logging
import time
from typing import Any, Dict, Optional

from requests import RequestException, post

from freqtrade.constants import Config
from freqtrade.enums import RPCMessageType
from freqtrade.rpc import RPC, RPCHandler


logger = logging.getLogger(__name__)

logger.debug('Included module rpc.webhook ...')


class Webhook(RPCHandler):
    """  This class handles all webhook communication """

    def __init__(self, rpc: RPC, config: Config) -> None:
        """
        Init the Webhook class, and init the super class RPCHandler
        :param rpc: instance of RPC Helper class
        :param config: Configuration object
        :return: None
        """
        super().__init__(rpc, config)

        self._url = self._config['webhook']['url']
        self._format = self._config['webhook'].get('format', 'form')
        self._retries = self._config['webhook'].get('retries', 0)
        self._retry_delay = self._config['webhook'].get('retry_delay', 0.1)

    def cleanup(self) -> None:
        """
        Cleanup pending module resources.
        This will do nothing for webhooks, they will simply not be called anymore
        """
        pass

    def _get_value_dict(self, msg: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        whconfig = self._config['webhook']
        # Deprecated 2022.10 - only keep generic method.
        if msg['type'] in [RPCMessageType.ENTRY]:
            valuedict = whconfig.get('webhookentry')
        elif msg['type'] in [RPCMessageType.ENTRY_CANCEL]:
            valuedict = whconfig.get('webhookentrycancel')
        elif msg['type'] in [RPCMessageType.ENTRY_FILL]:
            valuedict = whconfig.get('webhookentryfill')
        elif msg['type'] == RPCMessageType.EXIT:
            valuedict = whconfig.get('webhookexit')
        elif msg['type'] == RPCMessageType.EXIT_FILL:
            valuedict = whconfig.get('webhookexitfill')
        elif msg['type'] == RPCMessageType.EXIT_CANCEL:
            valuedict = whconfig.get('webhookexitcancel')
        elif msg['type'] in (RPCMessageType.STATUS,
                             RPCMessageType.STARTUP,
                             RPCMessageType.WARNING):
            valuedict = whconfig.get('webhookstatus')
        elif msg['type'].value in whconfig:
            # Allow all types ...
            valuedict = whconfig.get(msg['type'].value)
        elif msg['type'] in (
                RPCMessageType.PROTECTION_TRIGGER,
                RPCMessageType.PROTECTION_TRIGGER_GLOBAL,
                RPCMessageType.WHITELIST,
                RPCMessageType.ANALYZED_DF,
                RPCMessageType.NEW_CANDLE,
                RPCMessageType.STRATEGY_MSG):
            # Don't fail for non-implemented types
            return None
        return valuedict

    def send_msg(self, msg: Dict[str, Any]) -> None:
        """ Send a message to telegram channel """
        try:

            valuedict = self._get_value_dict(msg)

            if not valuedict:
                logger.info("Message type '%s' not configured for webhooks", msg['type'])
                return

            payload = {key: value.format(**msg) for (key, value) in valuedict.items()}
            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:
        """do the actual call to the webhook"""

        success = False
        attempts = 0
        while not success and attempts <= self._retries:
            if attempts:
                if self._retry_delay:
                    time.sleep(self._retry_delay)
                logger.info("Retrying webhook...")

            attempts += 1

            try:
                if self._format == 'form':
                    response = post(self._url, data=payload)
                elif self._format == 'json':
                    response = post(self._url, json=payload)
                elif self._format == 'raw':
                    response = post(self._url, data=payload['data'],
                                    headers={'Content-Type': 'text/plain'})
                else:
                    raise NotImplementedError('Unknown format: {}'.format(self._format))

                # Throw a RequestException if the post was not successful
                response.raise_for_status()
                success = True

            except RequestException as exc:
                logger.warning("Could not call webhook url. Exception: %s", exc)

1	"""
2	This module manages webhook communication
3	"""
4	import logging	1✔
5	import time	1✔
6	from typing import Any, Dict, Optional	1✔
7
8	from requests import RequestException, post	1✔
9
10	from freqtrade.constants import Config	1✔
11	from freqtrade.enums import RPCMessageType	1✔
12	from freqtrade.rpc import RPC, RPCHandler	1✔
13
14
15	logger = logging.getLogger(__name__)	1✔
16
17	logger.debug('Included module rpc.webhook ...')	1✔
18
19
20	class Webhook(RPCHandler):	1✔
21	""" This class handles all webhook communication """
22
23	def __init__(self, rpc: RPC, config: Config) -> None:	1✔
24	"""
25	Init the Webhook class, and init the super class RPCHandler
26	:param rpc: instance of RPC Helper class
27	:param config: Configuration object
28	:return: None
29	"""
30	super().__init__(rpc, config)	1✔
31
32	self._url = self._config['webhook']['url']	1✔
33	self._format = self._config['webhook'].get('format', 'form')	1✔
34	self._retries = self._config['webhook'].get('retries', 0)	1✔
35	self._retry_delay = self._config['webhook'].get('retry_delay', 0.1)	1✔
36
37	def cleanup(self) -> None:	1✔
38	"""
39	Cleanup pending module resources.
40	This will do nothing for webhooks, they will simply not be called anymore
41	"""
42	pass	×
43
44	def _get_value_dict(self, msg: Dict[str, Any]) -> Optional[Dict[str, Any]]:	1✔
45	whconfig = self._config['webhook']	1✔
46	# Deprecated 2022.10 - only keep generic method.
47	if msg['type'] in [RPCMessageType.ENTRY]:	1✔
48	valuedict = whconfig.get('webhookentry')	1✔
49	elif msg['type'] in [RPCMessageType.ENTRY_CANCEL]:	1✔
50	valuedict = whconfig.get('webhookentrycancel')	1✔
51	elif msg['type'] in [RPCMessageType.ENTRY_FILL]:	1✔
52	valuedict = whconfig.get('webhookentryfill')	1✔
53	elif msg['type'] == RPCMessageType.EXIT:	1✔
54	valuedict = whconfig.get('webhookexit')	1✔
55	elif msg['type'] == RPCMessageType.EXIT_FILL:	1✔
56	valuedict = whconfig.get('webhookexitfill')	1✔
57	elif msg['type'] == RPCMessageType.EXIT_CANCEL:	1✔
58	valuedict = whconfig.get('webhookexitcancel')	1✔
59	elif msg['type'] in (RPCMessageType.STATUS,	1✔
60	RPCMessageType.STARTUP,
61	RPCMessageType.WARNING):
62	valuedict = whconfig.get('webhookstatus')	1✔
63	elif msg['type'].value in whconfig:	1✔
64	# Allow all types ...
65	valuedict = whconfig.get(msg['type'].value)	1✔
66	elif msg['type'] in (	1✔
67	RPCMessageType.PROTECTION_TRIGGER,
68	RPCMessageType.PROTECTION_TRIGGER_GLOBAL,
69	RPCMessageType.WHITELIST,
70	RPCMessageType.ANALYZED_DF,
71	RPCMessageType.NEW_CANDLE,
72	RPCMessageType.STRATEGY_MSG):
73	# Don't fail for non-implemented types
74	return None	1✔
75	return valuedict	1✔
76
77	def send_msg(self, msg: Dict[str, Any]) -> None:	1✔
78	""" Send a message to telegram channel """
79	try:	1✔
80
81	valuedict = self._get_value_dict(msg)	1✔
82
83	if not valuedict:	1✔
84	logger.info("Message type '%s' not configured for webhooks", msg['type'])	1✔
85	return	1✔
86
87	payload = {key: value.format(**msg) for (key, value) in valuedict.items()}	1✔
88	self._send_msg(payload)	1✔
89	except KeyError as exc:	1✔
90	logger.exception("Problem calling Webhook. Please check your webhook configuration. "	1✔
91	"Exception: %s", exc)
92
93	def _send_msg(self, payload: dict) -> None:	1✔
94	"""do the actual call to the webhook"""
95
96	success = False	1✔
97	attempts = 0	1✔
98	while not success and attempts <= self._retries:	1✔
99	if attempts:	1✔
100	if self._retry_delay:	×
101	time.sleep(self._retry_delay)	×
102	logger.info("Retrying webhook...")	×
103
104	attempts += 1	1✔
105
106	try:	1✔
107	if self._format == 'form':	1✔
108	response = post(self._url, data=payload)	1✔
109	elif self._format == 'json':	1✔
110	response = post(self._url, json=payload)	1✔
111	elif self._format == 'raw':	1✔
112	response = post(self._url, data=payload['data'],	1✔
113	headers={'Content-Type': 'text/plain'})
114	else:
115	raise NotImplementedError('Unknown format: {}'.format(self._format))	×
116
117	# Throw a RequestException if the post was not successful
118	response.raise_for_status()	1✔
119	success = True	1✔
120
121	except RequestException as exc:	1✔
122	logger.warning("Could not call webhook url. Exception: %s", exc)	1✔

freqtrade / freqtrade / 4131167254

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous