Improve EMC config validations

freqtrade/configuration/config_validation.py

@@ -1,4 +1,5 @@
 import logging
+from collections import Counter
 from copy import deepcopy
 from typing import Any, Dict
 
@@ -85,6 +86,7 @@ def validate_config_consistency(conf: Dict[str, Any], preliminary: bool = False)
     _validate_unlimited_amount(conf)
     _validate_ask_orderbook(conf)
     _validate_freqai_hyperopt(conf)
+    _validate_consumers(conf)
     validate_migrated_strategy_settings(conf)
 
     # validate configuration before returning
@@ -332,6 +334,23 @@ def _validate_freqai_hyperopt(conf: Dict[str, Any]) -> None:
             'Using analyze-per-epoch parameter is not supported with a FreqAI strategy.')
 
 
+def _validate_consumers(conf: Dict[str, Any]) -> None:
+    emc_conf = conf.get('external_message_consumer', {})
+    if emc_conf.get('enabled', False):
+        if len(emc_conf.get('producers', [])) < 1:
+            raise OperationalException("You must specify at least 1 Producer to connect to.")
+
+        producer_names = [p['name'] for p in emc_conf.get('producers', [])]
+        duplicates = [item for item, count in Counter(producer_names).items() if count > 1]
+        if duplicates:
+            raise OperationalException(
+                f"Producer names must be unique. Duplicate: {', '.join(duplicates)}")
+        if conf.get('process_only_new_candles', True):
+            # Warning here or require it?
+            logger.warning("To receive best performance with external data, "
+                           "please set `process_only_new_candles` to False")
+
+
 def _strategy_settings(conf: Dict[str, Any]) -> None:
 
     process_deprecated_setting(conf, None, 'use_sell_signal', None, 'use_exit_signal')

freqtrade/rpc/external_message_consumer.py

@@ -15,7 +15,6 @@ from pydantic import ValidationError
 
 from freqtrade.data.dataprovider import DataProvider
 from freqtrade.enums import RPCMessageType
-from freqtrade.exceptions import OperationalException
 from freqtrade.misc import remove_entry_exit_signals
 from freqtrade.rpc.api_server.ws import WebSocketChannel
 from freqtrade.rpc.api_server.ws_schemas import (WSAnalyzedDFMessage, WSAnalyzedDFRequest,
@@ -74,8 +73,6 @@ class ExternalMessageConsumer:
         # as the websockets client expects bytes.
         self.message_size_limit = (self._emc_config.get('message_size_limit', 8) << 20)
 
-        self.validate_config()
-
         # Setting these explicitly as they probably shouldn't be changed by a user
         # Unless we somehow integrate this with the strategy to allow creating
         # callbacks for the messages
@@ -96,18 +93,6 @@ class ExternalMessageConsumer:
 
         self.start()
 
-    def validate_config(self):
-        """
-        Make sure values are what they are supposed to be
-        """
-        if self.enabled and len(self.producers) < 1:
-            raise OperationalException("You must specify at least 1 Producer to connect to.")
-
-        if self.enabled and self._config.get('process_only_new_candles', True):
-            # Warning here or require it?
-            logger.warning("To receive best performance with external data, "
-                           "please set `process_only_new_candles` to False")
-
     def start(self):
         """
         Start the main internal loop in another thread to run coroutines

tests/rpc/test_rpc_emc.py

@@ -11,7 +11,6 @@ import pytest
 import websockets
 
 from freqtrade.data.dataprovider import DataProvider
-from freqtrade.exceptions import OperationalException
 from freqtrade.rpc.external_message_consumer import ExternalMessageConsumer
 from tests.conftest import log_has, log_has_re, log_has_when
 
@@ -73,23 +72,12 @@ def test_emc_shutdown(patched_emc, caplog):
     assert not log_has("Stopping ExternalMessageConsumer", caplog)
 
 
-def test_emc_init(patched_emc, default_conf):
+def test_emc_init(patched_emc):
     # Test the settings were set correctly
     assert patched_emc.initial_candle_limit <= 1500
     assert patched_emc.wait_timeout > 0
     assert patched_emc.sleep_time > 0
 
-    default_conf.update({
-        "external_message_consumer": {
-            "enabled": True,
-            "producers": []
-        }
-    })
-    dataprovider = DataProvider(default_conf, None, None, None)
-    with pytest.raises(OperationalException,
-                       match="You must specify at least 1 Producer to connect to."):
-        ExternalMessageConsumer(default_conf, dataprovider)
-
 
 # Parametrize this?
 def test_emc_handle_producer_message(patched_emc, caplog, ohlcv_history):

tests/test_configuration.py

@@ -1089,6 +1089,58 @@ def test__validate_pricing_rules(default_conf, caplog) -> None:
         validate_config_consistency(conf)
 
 
+def test__validate_consumers(default_conf, caplog) -> None:
+    conf = deepcopy(default_conf)
+    conf.update({
+            "external_message_consumer": {
+                "enabled": True,
+                "producers": []
+                }
+            })
+    with pytest.raises(OperationalException,
+                       match="You must specify at least 1 Producer to connect to."):
+        validate_config_consistency(conf)
+
+    conf = deepcopy(default_conf)
+    conf.update({
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": "127.0.0.1",
+                    "port": 8081,
+                    "ws_token": "secret_ws_t0ken."
+                }, {
+                    "name": "default",
+                    "host": "127.0.0.1",
+                    "port": 8080,
+                    "ws_token": "secret_ws_t0ken."
+                }
+            ]}
+        })
+    with pytest.raises(OperationalException,
+                       match="Producer names must be unique. Duplicate: default"):
+        validate_config_consistency(conf)
+
+    conf = deepcopy(default_conf)
+    conf.update({
+        "process_only_new_candles": True,
+        "external_message_consumer": {
+            "enabled": True,
+            "producers": [
+                {
+                    "name": "default",
+                    "host": "127.0.0.1",
+                    "port": 8081,
+                    "ws_token": "secret_ws_t0ken."
+                }
+            ]}
+        })
+    validate_config_consistency(conf)
+    assert log_has_re("To receive best performance with external data.*", caplog)
+
+
 def test_load_config_test_comments() -> None:
     """
     Load config with comments