[TradingMode] support dsl calls

.github/workflows/main.yml

@@ -29,7 +29,7 @@ jobs:
           architecture: ${{ matrix.arch }}
 
       - name: Install Pants
-        uses: pantsbuild/actions/init-pants@v10
+        uses: pantsbuild/actions/init-pants@v11
         with:
           gha-cache-key: ${{ runner.os }}-pants-build
           named-caches-hash: ${{ hashFiles('pants.toml') }}
@@ -82,14 +82,15 @@ jobs:
           - packages/evaluators
           - packages/node
           - packages/flow
+          - packages/copy
           - packages/services
           - packages/sync
           - packages/tentacles_manager
           - packages/trading
           - packages/trading_backend
 
     env:
-      USES_TENTACLES: ${{ matrix.package == 'octobot' || matrix.package == 'packages/node' || matrix.package == 'packages/flow' }}
+      USES_TENTACLES: ${{ matrix.package == 'octobot' || matrix.package == 'packages/node' || matrix.package == 'packages/flow' || matrix.package == 'packages/copy' }}
 
     steps:
     - uses: actions/checkout@v6
@@ -159,7 +160,7 @@ jobs:
           pytest tests -n auto --dist loadfile
           pytest --ignore=tentacles/Trading/Exchange tentacles -n auto --dist loadfile
         else
-          if [ "${{ matrix.package }}" = "packages/node" ] || [ "${{ matrix.package }}" = "packages/flow" ]; then
+          if [ "${{ matrix.package }}" = "packages/node" ] || [ "${{ matrix.package }}" = "packages/flow" ] || [ "${{ matrix.package }}" = "packages/copy" ]; then
             echo "Running tests from root dir to allow tentacles import"
             PYTHONPATH=.:$PYTHONPATH pytest ${{ matrix.package }}/tests -n auto --dist loadfile
           else

BUILD

@@ -38,6 +38,7 @@ PACKAGE_SOURCES = [
     "packages/evaluators:octobot_evaluators",
     "packages/node:octobot_node",
     "packages/flow:octobot_flow",
+    "packages/copy:octobot_copy",
     "packages/services:octobot_services",
     "packages/sync:octobot_sync",
     "packages/tentacles_manager:octobot_tentacles_manager",

additional_tests/exchanges_tests/__init__.py

@@ -112,7 +112,7 @@ async def get_authenticated_exchange_manager(
         .is_exchange_only()
     if http_proxy_callback_factory:
         proxy_callback = http_proxy_callback_factory(exchange_builder.exchange_manager)
-        exchange_builder.set_proxy_config(exchanges.ProxyConfig(http_proxy_callback=proxy_callback))
+        exchange_builder.set_proxy_config(exchanges.ExchangeProxyConfig(http_proxy_callback=proxy_callback))
     exchange_manager_instance = await exchange_builder.build()
     # create trader afterwards to init exchange personal data
     exchange_manager_instance.trader.is_enabled = True

packages/async_channel/async_channel/util/__init__.py

@@ -18,6 +18,7 @@
 """
 from async_channel.util import channel_creator
 from async_channel.util import logging_util
+from async_channel.util import synchronization_util
 
 from async_channel.util.channel_creator import (
     create_all_subclasses_channel,
@@ -28,8 +29,13 @@
     get_logger,
 )
 
+from async_channel.util.synchronization_util import (
+    trigger_and_bypass_consumers_queue,
+)
+
 __all__ = [
     "create_all_subclasses_channel",
     "create_channel_instance",
     "get_logger",
+    "trigger_and_bypass_consumers_queue",
 ]

packages/async_channel/async_channel/util/synchronization_util.py

@@ -0,0 +1,37 @@
+#  Drakkar-Software Async-Channel
+#  Copyright (c) Drakkar-Software, All rights reserved.
+#
+#  This library is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU Lesser General Public
+#  License as published by the Free Software Foundation; either
+#  version 3.0 of the License, or (at your option) any later version.
+#
+#  This library is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+#  Lesser General Public License for more details.
+#
+#  You should have received a copy of the GNU Lesser General Public
+#  License along with this library.
+"""
+Define async_channel synchronization utilities
+"""
+import asyncio
+import typing
+
+if typing.TYPE_CHECKING:
+    import async_channel.consumer
+
+
+async def trigger_and_bypass_consumers_queue(
+    consumers: list["async_channel.consumer.Consumer"], kwargs: dict
+):
+    """
+    Triggers the consumers queue and bypasses the consumers callback.
+    Warning: this can cause concurrent async executions of the consumer callback
+    as the queue is bypassed.
+    """
+    await asyncio.gather(*[
+        consumer.callback(**kwargs)
+        for consumer in consumers
+    ])

packages/commons/octobot_commons/asyncio_tools.py

@@ -128,8 +128,9 @@ def logged_waiter(self, name: str, sleep_time: float = 30) -> typing.Generator[N
     async def _waiter() -> None:
         t0 = time.time()
         try:
-            await asyncio.sleep(sleep_time)
-            self.logger.info(f"{name} is still processing [{time.time() - t0:.2f} seconds] ...")
+            while True:
+                await asyncio.sleep(sleep_time)
+                self.logger.info(f"{name} is still processing [{time.time() - t0:.2f} seconds] ...")
         except asyncio.CancelledError:
             pass
     task = None

packages/commons/octobot_commons/configuration/__init__.py

@@ -46,6 +46,8 @@
     get_password_hash,
 )
 from octobot_commons.configuration.user_inputs import (
+    USER_INPUT_TYPE_TO_PYTHON_TYPE,
+    MAX_USER_INPUT_ORDER,
     UserInput,
     UserInputFactory,
     sanitize_user_input_name,
@@ -85,6 +87,8 @@
     "decrypt",
     "decrypt_element_if_possible",
     "get_password_hash",
+    "USER_INPUT_TYPE_TO_PYTHON_TYPE",
+    "MAX_USER_INPUT_ORDER",
     "UserInput",
     "UserInputFactory",
     "sanitize_user_input_name",

packages/commons/octobot_commons/configuration/fields_utils.py

@@ -42,7 +42,7 @@ def encrypt(data):
     try:
         return fernet.Fernet(octobot_commons.OCTOBOT_KEY).encrypt(data.encode())
     except Exception as global_exception:
-        logging.getLogger().error(f"Failed to encrypt : {data}")
+        logging.getLogger("fields_utils").error(f"Failed to encrypt : {data}")
         raise global_exception
 
 
@@ -59,12 +59,12 @@ def decrypt(data, silent_on_invalid_token=False):
         )
     except fernet.InvalidToken as invalid_token_error:
         if not silent_on_invalid_token:
-            logging.getLogger().error(
+            logging.getLogger("fields_utils").error(
                 f"Failed to decrypt : {data} ({invalid_token_error})"
             )
         raise invalid_token_error
     except Exception as global_exception:
-        logging.getLogger().error(f"Failed to decrypt : {data} ({global_exception})")
+        logging.getLogger("fields_utils").error(f"Failed to decrypt : {data} ({global_exception})")
         raise global_exception
 
 

packages/commons/octobot_commons/configuration/user_inputs.py

@@ -20,8 +20,23 @@
 import octobot_commons.dict_util as dict_util
 
 
+USER_INPUT_TYPE_TO_PYTHON_TYPE = {
+    enums.UserInputTypes.INT.value: int,
+    enums.UserInputTypes.FLOAT.value: float,
+    enums.UserInputTypes.BOOLEAN.value: bool,
+    enums.UserInputTypes.TEXT.value: str,
+    enums.UserInputTypes.OBJECT.value: dict,
+    enums.UserInputTypes.OBJECT_ARRAY.value: list,
+    enums.UserInputTypes.STRING_ARRAY.value: list,
+    enums.UserInputTypes.OPTIONS.value: str,
+    enums.UserInputTypes.MULTIPLE_OPTIONS.value: list,
+}
+
+
+MAX_USER_INPUT_ORDER = 9999
+
+
 class UserInput:
-    MAX_ORDER = 9999
 
     def __init__(
         self,

packages/commons/octobot_commons/dsl_interpreter/interpreter.py

@@ -370,7 +370,7 @@ def _visit_node(self, node: typing.Optional[ast.AST]) -> typing.Union[
             )
 
         raise octobot_commons.errors.UnsupportedOperatorError(
-            f"Unsupported AST node type: {type(node).__name__}"
+            f"Unsupported AST node type: {type(node).__name__}. Expression: {self._parsed_expression}"
         )
 
     def _get_name_from_node(self, node: ast.AST) -> str:

packages/commons/octobot_commons/dsl_interpreter/operator.py

@@ -93,7 +93,9 @@ def _validate_parameters(
                 )
             if max_params is not None and total_params > max_params:
                 raise octobot_commons.errors.InvalidParametersError(
-                    f"{self.get_name()} supports up to {max_params} "
+                    f"{self.get_name()} got {total_params} parameters "
+                    f"({', '.join([str(p) for p in tuple(parameters) + tuple(kwargs.values())])}) "
+                    f"but supports up to {max_params} "
                     f"parameters: {self.get_parameters_description()}"
                 )
 

packages/commons/octobot_commons/dsl_interpreter/operators/re_callable_operator_mixin.py

@@ -20,15 +20,18 @@
 
 import octobot_commons.dataclasses
 import octobot_commons.dsl_interpreter.operator_parameter as operator_parameter
+import octobot_commons.dsl_interpreter.parameters_util as parameters_util
 
 
 class ReCallingOperatorResultKeys(str, enum.Enum):
     WAITING_TIME = "waiting_time"
     LAST_EXECUTION_TIME = "last_execution_time"
+    SCRIPT_OVERRIDE = "script_override"
 
 
 @dataclasses.dataclass
 class ReCallingOperatorResult(octobot_commons.dataclasses.MinimizableDataclass):
+    keyword: typing.Optional[str] = None
     reset_to_id: typing.Optional[str] = None
     last_execution_result: typing.Optional[dict] = None
 
@@ -56,6 +59,24 @@ def get_next_call_time(self) -> typing.Optional[float]:
             return last_execution_time + waiting_time
         return None
 
+    @staticmethod
+    def get_script_override(result: typing.Any) -> typing.Optional[str]:
+        """
+        Returns the script override from the last execution result.
+        """
+        if not ReCallingOperatorResult.is_re_calling_operator_result(result):
+            return None
+        return result[ReCallingOperatorResult.__name__].get("last_execution_result", {}).get(
+            ReCallingOperatorResultKeys.SCRIPT_OVERRIDE.value
+        )
+
+    @staticmethod
+    def get_keyword(result: typing.Any) -> typing.Optional[str]:
+        """
+        Returns the keyword from the re-calling operator result.
+        """
+        return result[ReCallingOperatorResult.__name__]["keyword"]
+
 
 class ReCallableOperatorMixin:
     """
@@ -93,23 +114,58 @@ def get_last_execution_result(
             ]).last_execution_result
         return None
 
-    def build_re_callable_result(
+    def create_re_callable_result( # pylint: disable=too-many-arguments
+        self,
+        keyword: str,
+        reset_to_id: typing.Optional[str] = None,
+        waiting_time: typing.Optional[float] = None,
+        last_execution_time: typing.Optional[float] = None,
+        script_override: typing.Optional[str] = None,
+        **kwargs: typing.Any,
+    ) -> ReCallingOperatorResult:
+        """
+        Builds a re-callable result from the given parameters.
+        """
+        return ReCallingOperatorResult(
+            keyword=keyword,
+            reset_to_id=reset_to_id,
+            last_execution_result={
+                ReCallingOperatorResultKeys.WAITING_TIME.value: waiting_time,
+                ReCallingOperatorResultKeys.LAST_EXECUTION_TIME.value: last_execution_time,
+                ReCallingOperatorResultKeys.SCRIPT_OVERRIDE.value: script_override,
+                **kwargs,
+            },
+        )
+
+    def create_re_callable_result_dict( # pylint: disable=too-many-arguments
         self,
+        keyword: str,
         reset_to_id: typing.Optional[str] = None,
         waiting_time: typing.Optional[float] = None,
         last_execution_time: typing.Optional[float] = None,
+        script_override: typing.Optional[str] = None,
         **kwargs: typing.Any,
     ) -> dict:
         """
         Builds a dict formatted re-callable result from the given parameters.
         """
         return {
-            ReCallingOperatorResult.__name__: ReCallingOperatorResult(
+            ReCallingOperatorResult.__name__: self.create_re_callable_result(
+                keyword=keyword,
                 reset_to_id=reset_to_id,
-                last_execution_result={
-                    ReCallingOperatorResultKeys.WAITING_TIME.value: waiting_time,
-                    ReCallingOperatorResultKeys.LAST_EXECUTION_TIME.value: last_execution_time,
-                    **kwargs,
-                },
+                waiting_time=waiting_time,
+                last_execution_time=last_execution_time,
+                script_override=script_override,
+                **kwargs,
             ).to_dict(include_default_values=False)
         }
+
+    def re_create_script(self, param_by_name: dict[str, typing.Any]):
+        """
+        Returns the re-created script from the given parameters.
+        """
+        param_without_re_callable_operator_params = {
+            k: v for k, v in param_by_name.items() if k != self.LAST_EXECUTION_RESULT_KEY
+        }
+        params = parameters_util.resove_operator_params(self, param_without_re_callable_operator_params)
+        return f"{self.get_name()}({', '.join(params)})" # type: ignore