[DSL] POC2

.github/workflows/main.yml

@@ -65,6 +65,7 @@ jobs:
           - packages/commons
           - packages/evaluators
           - packages/node
+          - packages/flow
           - packages/services
           - packages/tentacles_manager
           - packages/trading
@@ -101,7 +102,7 @@ jobs:
         fi
 
     - name: Install tentacles
-      if:  matrix.package == 'octobot'
+      if:  matrix.package == 'octobot' || matrix.package == 'packages/node' || matrix.package == 'packages/flow'
       run: |
         mkdir -p output
         OctoBot tentacles -d packages/tentacles -p any_platform.zip
@@ -113,11 +114,16 @@ jobs:
           pytest tests -n auto --dist loadfile
           pytest --ignore=tentacles/Trading/Exchange tentacles -n auto --dist loadfile
         else
-          cd ${{ matrix.package }}
-          if [ "${{ matrix.package }}" = "packages/tentacles_manager" ] || [ "${{ matrix.package }}" = "packages/node" ]; then
-            pytest tests
+          if [ "${{ matrix.package }}" = "packages/node" ] || [ "${{ matrix.package }}" = "packages/flow" ]; then
+            echo "Running tests from root dir to allow tentacles import"
+            PYTHONPATH=.:$PYTHONPATH pytest ${{ matrix.package }}/tests -n auto --dist loadfile
           else
-            pytest tests -n auto --dist loadfile
+            cd ${{ matrix.package }}
+            if [ "${{ matrix.package }}" = "packages/tentacles_manager" ]; then
+              pytest tests
+            else
+              pytest tests -n auto --dist loadfile
+            fi
           fi
         fi
       env:

BUILD

@@ -32,6 +32,7 @@ PACKAGE_SOURCES = [
     "packages/commons:octobot_commons",
     "packages/evaluators:octobot_evaluators",
     "packages/node:octobot_node",
+    "packages/flow:octobot_flow",
     "packages/services:octobot_services",
     "packages/tentacles_manager:octobot_tentacles_manager",
     "packages/trading:octobot_trading",

additional_tests/exchanges_tests/abstract_authenticated_exchange_tester.py

@@ -35,7 +35,7 @@
 import octobot_trading.personal_data as personal_data
 import octobot_trading.personal_data.orders as personal_data_orders
 import octobot_trading.util.test_tools.exchanges_test_tools as exchanges_test_tools
-import octobot_trading.util.test_tools.exchange_data as exchange_data_import
+import octobot_trading.exchanges.util.exchange_data as exchange_data_import
 import trading_backend.enums
 import octobot_tentacles_manager.api as tentacles_manager_api
 from additional_tests.exchanges_tests import get_authenticated_exchange_manager, NoProvidedCredentialsError

octobot/backtesting/minimal_data_importer.py

@@ -7,7 +7,7 @@
 import octobot_backtesting.importers
 import octobot_backtesting.enums
 
-import octobot_trading.util.test_tools.exchange_data as exchange_data_import
+import octobot_trading.exchanges.util.exchange_data as exchange_data_import
 
 
 class MinimalDataImporter(octobot_backtesting.importers.ExchangeDataImporter):

packages/commons/octobot_commons/asyncio_tools.py

@@ -14,8 +14,11 @@
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
 import asyncio
+import contextlib
+import time
 import traceback
 import concurrent.futures
+import typing
 
 import octobot_commons.constants as constants
 import octobot_commons.logging as logging_util
@@ -117,6 +120,27 @@ async def gather_waiting_for_all_before_raising(*coros):
     return maybe_exceptions
 
 
+@contextlib.contextmanager
+def logged_waiter(self, name: str, sleep_time: float = 30) -> typing.Generator[None, None, None]:
+    """
+    Periodically log the time elapsed since the start of the waiter
+    """
+    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] ...")
+        except asyncio.CancelledError:
+            pass
+    task = None
+    try:
+        task = asyncio.create_task(_waiter())
+        yield
+    finally:
+        if task is not None and not task.done():
+            task.cancel()
+
+
 class RLock(asyncio.Lock):
     """
     Async Lock implementing reentrancy

packages/commons/octobot_commons/constants.py

@@ -249,6 +249,7 @@ def parse_boolean_environment_var(env_key: str, default_value: str) -> bool:
 # DSL interpreter
 BASE_OPERATORS_LIBRARY = "base"
 CONTEXTUAL_OPERATORS_LIBRARY = "contextual"
+UNRESOLVED_PARAMETER_PLACEHOLDER = "UNRESOLVED_PARAMETER"
 
 # Logging
 EXCEPTION_DESC = "exception_desc"

packages/commons/octobot_commons/dataclasses/flexible_dataclass.py

@@ -25,7 +25,7 @@
 
 @dataclasses.dataclass
 class FlexibleDataclass:
-    _class_field_cache: typing.ClassVar[dict] = {}
+    _class_field_cache: typing.ClassVar[dict] = dataclasses.field(default={}, repr=False)
     """
     Implements from_dict which can be called to instantiate a new instance of this class from a dict. Using from_dict 
     ignores any additional key from the given dict that is not defined as a dataclass field.

packages/commons/octobot_commons/dsl_interpreter/__init__.py

@@ -38,15 +38,23 @@
     NameOperator,
     ExpressionOperator,
     PreComputingCallOperator,
+    ReCallableOperatorMixin,
+    ReCallingOperatorResult,
+    ReCallingOperatorResultKeys,
 )
 from octobot_commons.dsl_interpreter.interpreter_dependency import (
     InterpreterDependency,
 )
 from octobot_commons.dsl_interpreter.parameters_util import (
     format_parameter_value,
     resove_operator_params,
+    apply_resolved_parameter_value,
+    add_resolved_parameter_value,
+    has_unresolved_parameters,
+)
+from octobot_commons.dsl_interpreter.dsl_call_result import (
+    DSLCallResult,
 )
-from octobot_commons.dsl_interpreter.dsl_call_result import DSLCallResult
 
 __all__ = [
     "get_all_operators",
@@ -64,8 +72,14 @@
     "NameOperator",
     "ExpressionOperator",
     "PreComputingCallOperator",
+    "ReCallableOperatorMixin",
     "InterpreterDependency",
     "format_parameter_value",
     "resove_operator_params",
+    "apply_resolved_parameter_value",
+    "add_resolved_parameter_value",
     "DSLCallResult",
+    "has_unresolved_parameters",
+    "ReCallingOperatorResult",
+    "ReCallingOperatorResultKeys",
 ]

packages/commons/octobot_commons/dsl_interpreter/dsl_call_result.py

@@ -16,11 +16,22 @@
 
 import dataclasses
 import typing
+
 import octobot_commons.dataclasses
-import octobot_commons.errors
+
 
 @dataclasses.dataclass
 class DSLCallResult(octobot_commons.dataclasses.FlexibleDataclass):
+    """
+    Stores a DSL call result alongside its statement (and error if any)
+    """
     statement: str
     result: typing.Optional[typing.Any] = None
     error: typing.Optional[str] = None
+
+    def succeeded(self) -> bool:
+        """
+        Check if the DSL call succeeded
+        :return: True if the DSL call succeeded, False otherwise
+        """
+        return self.error is None

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

@@ -45,6 +45,11 @@
 from octobot_commons.dsl_interpreter.operators.pre_computing_call_operator import (
     PreComputingCallOperator,
 )
+from octobot_commons.dsl_interpreter.operators.re_callable_operator_mixin import (
+    ReCallableOperatorMixin,
+    ReCallingOperatorResult,
+    ReCallingOperatorResultKeys,
+)
 
 __all__ = [
     "BinaryOperator",
@@ -57,4 +62,7 @@
     "SubscriptingOperator",
     "IterableOperator",
     "PreComputingCallOperator",
+    "ReCallableOperatorMixin",
+    "ReCallingOperatorResult",
+    "ReCallingOperatorResultKeys",
 ]

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

@@ -0,0 +1,115 @@
+#  Drakkar-Software OctoBot-Commons
+#  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.
+import dataclasses
+import typing
+import time
+import enum
+
+import octobot_commons.dataclasses
+import octobot_commons.dsl_interpreter.operator_parameter as operator_parameter
+
+
+class ReCallingOperatorResultKeys(str, enum.Enum):
+    WAITING_TIME = "waiting_time"
+    LAST_EXECUTION_TIME = "last_execution_time"
+
+
+@dataclasses.dataclass
+class ReCallingOperatorResult(octobot_commons.dataclasses.MinimizableDataclass):
+    reset_to_id: typing.Optional[str] = None
+    last_execution_result: typing.Optional[dict] = None
+
+    @staticmethod
+    def is_re_calling_operator_result(result: typing.Any) -> bool:
+        """
+        Check if the result is a re-calling operator result.
+        """
+        return isinstance(result, dict) and (
+            ReCallingOperatorResult.__name__ in result
+        )
+
+    def get_next_call_time(self) -> typing.Optional[float]:
+        """
+        Returns the next call time based on the last execution result's 
+        waiting time and last execution time.
+        """
+        if (
+            self.last_execution_result
+            and (waiting_time := self.last_execution_result.get(ReCallingOperatorResultKeys.WAITING_TIME.value))
+        ):
+            last_execution_time = self.last_execution_result.get(
+                ReCallingOperatorResultKeys.LAST_EXECUTION_TIME.value
+            ) or time.time()
+            return last_execution_time + waiting_time
+        return None
+
+
+class ReCallableOperatorMixin:
+    """
+    Mixin for re-callable operators.
+    """
+    LAST_EXECUTION_RESULT_KEY = "last_execution_result"
+
+    @classmethod
+    def get_re_callable_parameters(cls) -> list[operator_parameter.OperatorParameter]:
+        """
+        Returns the parameters for the re-callable operator.
+        """
+        return [
+            operator_parameter.OperatorParameter(
+                name=cls.LAST_EXECUTION_RESULT_KEY,
+                description="the return value of the previous call",
+                required=False,
+                type=dict,
+                default=None,
+            ),
+        ]
+
+    def get_last_execution_result(
+        self, param_by_name: dict[str, typing.Any]
+    ) -> typing.Optional[dict]:
+        """
+        Returns the potential last execution result from param_by_name.
+        """
+        if (
+            (result_dict := param_by_name.get(self.LAST_EXECUTION_RESULT_KEY, None))
+            and ReCallingOperatorResult.is_re_calling_operator_result(result_dict)
+        ):
+            return ReCallingOperatorResult.from_dict(result_dict[
+                ReCallingOperatorResult.__name__
+            ]).last_execution_result
+        return None
+
+    def build_re_callable_result(
+        self,
+        reset_to_id: typing.Optional[str] = None,
+        waiting_time: typing.Optional[float] = None,
+        last_execution_time: typing.Optional[float] = None,
+        **kwargs: typing.Any,
+    ) -> dict:
+        """
+        Builds a dict formatted re-callable result from the given parameters.
+        """
+        return {
+            ReCallingOperatorResult.__name__: ReCallingOperatorResult(
+                reset_to_id=reset_to_id,
+                last_execution_result={
+                    ReCallingOperatorResultKeys.WAITING_TIME.value: waiting_time,
+                    ReCallingOperatorResultKeys.LAST_EXECUTION_TIME.value: last_execution_time,
+                    **kwargs,
+                },
+            ).to_dict(include_default_values=False)
+        }

packages/commons/octobot_commons/dsl_interpreter/parameters_util.py

@@ -13,11 +13,13 @@
 #
 #  You should have received a copy of the GNU Lesser General Public
 #  License along with this library.
+import re
 import typing
 import json
 
 import octobot_commons.dsl_interpreter.operator as dsl_interpreter_operator
 import octobot_commons.errors
+import octobot_commons.constants
 
 
 def format_parameter_value(value: typing.Any) -> str: # pylint: disable=too-many-return-statements
@@ -111,3 +113,51 @@ def resolve_operator_args_and_kwargs(
         )
 
     return merged_args, remaining_kwargs
+
+
+def apply_resolved_parameter_value(script: str, parameter: str, value: typing.Any):
+    """
+    Apply a resolved parameter value to a DSL script.
+    """
+    to_replace = f"{parameter}={octobot_commons.constants.UNRESOLVED_PARAMETER_PLACEHOLDER}"
+    if to_replace not in script:
+        raise octobot_commons.errors.ResolvedParameterNotFoundError(
+            f"Parameter {parameter} not found in script: {script}"
+        )
+    new_value = f"{parameter}={format_parameter_value(value)}"
+    return script.replace(to_replace, new_value)
+
+
+def add_resolved_parameter_value(script: str, parameter: str, value: typing.Any):
+    """
+    Append a resolved parameter value to the end of a DSL script.
+    Supports:
+    - Calls with no parenthesis (e.g. op -> op(x='a'))
+    - Calls with no existing params (e.g. op() -> op(x='a'))
+    - Calls with existing params (e.g. op(1) -> op(1, x='a'))
+    Raises InvalidParametersError if the parameter is already in the operator keyword args.
+    """
+    param_str = f"{parameter}={format_parameter_value(value)}"
+    if script[-1] == ")":
+        # Script ends with ) - append to existing call
+        if re.search(rf"(?:\(|,)\s*{re.escape(parameter)}\s*=", script):
+            raise octobot_commons.errors.InvalidParametersError(
+                f"Parameter {parameter} is already in operator keyword args: {script}"
+            )
+        inner = script[:-1]
+        has_existing_params = inner.rstrip().endswith("(")
+        if has_existing_params:
+            return f"{inner}{param_str})"
+        return f"{inner}, {param_str})"
+    if "(" in script:
+        raise octobot_commons.errors.InvalidParametersError(
+            f"Script {script} has unclosed parenthesis"
+        )
+    return f"{script}({param_str})"
+
+
+def has_unresolved_parameters(script: str) -> bool:
+    """
+    Check if a DSL script has unresolved parameters.
+    """
+    return octobot_commons.constants.UNRESOLVED_PARAMETER_PLACEHOLDER in script

packages/commons/octobot_commons/errors.py

@@ -165,6 +165,12 @@ class InvalidParameterFormatError(InvalidParametersError):
     """
 
 
+class ResolvedParameterNotFoundError(DSLInterpreterError):
+    """
+    Raised when a resolved parameter is not found in the script
+    """
+
+
 class ErrorStatementEncountered(DSLInterpreterError):
     """
     Raised when a error statement is encountered when executing a script