Refactor webhook method

freqtrade/rpc/webhook.py

@@ -3,7 +3,7 @@ This module manages webhook communication
 """
 import logging
 import time
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 from requests import RequestException, post
 
@@ -41,38 +41,43 @@ class Webhook(RPCHandler):
         """
         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.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:
-            whconfig = self._config['webhook']
+
-            # Deprecated 2022.10 - only keep generic method.
+            valuedict = self._get_value_dict(msg)
-            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.STRATEGY_MSG):
-                # Don't fail for non-implemented types
-                return
 
             if not valuedict:
                 logger.info("Message type '%s' not configured for webhooks", msg['type'])

tests/rpc/test_rpc_webhook.py

@@ -336,20 +336,13 @@ def test_exception_send_msg(default_conf, mocker, caplog):
                    caplog)
 
     default_conf["webhook"] = get_webhook_dict()
-    default_conf["webhook"]["webhookentry"]["value1"] = "{DEADBEEF:8f}"
+    default_conf["webhook"]["strategy_msg"] = {"value1": "{DEADBEEF:8f}"}
     msg_mock = MagicMock()
     mocker.patch("freqtrade.rpc.webhook.Webhook._send_msg", msg_mock)
     webhook = Webhook(RPC(get_patched_freqtradebot(mocker, default_conf)), default_conf)
     msg = {
-        'type': RPCMessageType.ENTRY,
+        'type': RPCMessageType.STRATEGY_MSG,
-        'exchange': 'Binance',
+        'msg': 'hello world',
-        'pair': 'ETH/BTC',
-        'limit': 0.005,
-        'order_type': 'limit',
-        'stake_amount': 0.8,
-        'stake_amount_fiat': 500,
-        'stake_currency': 'BTC',
-        'fiat_currency': 'EUR'
     }
     webhook.send_msg(msg)
     assert log_has("Problem calling Webhook. Please check your webhook configuration. "