[StopAutomation] convert to DSL

Author: GuillaumeDSM

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

@@ -38,7 +38,7 @@ 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 (
-            "reset_to_id" in result or "last_execution_result" in result
+            ReCallingOperatorResult.__name__ in result
         )
 
     def get_next_call_time(self) -> typing.Optional[float]:
@@ -88,7 +88,9 @@ def get_last_execution_result(
             (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).last_execution_result
+            return ReCallingOperatorResult.from_dict(result_dict[
+                ReCallingOperatorResult.__name__
+            ]).last_execution_result
         return None
 
     def build_re_callable_result(
@@ -101,11 +103,13 @@ def build_re_callable_result(
         """
         Builds a dict formatted re-callable result from the given parameters.
         """
-        return 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)
+        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/tests/dsl_interpreter/operators/test_re_callable_operator_mixin.py

@@ -23,12 +23,16 @@
 class TestReCallingOperatorResult:
     def test_is_re_calling_operator_result_with_reset_to_id(self):
         assert re_callable_operator_mixin.ReCallingOperatorResult.is_re_calling_operator_result(
-            {"reset_to_id": "some_id"}
+            {re_callable_operator_mixin.ReCallingOperatorResult.__name__: {"reset_to_id": "some_id"}}
         ) is True
 
     def test_is_re_calling_operator_result_with_last_execution_result(self):
         assert re_callable_operator_mixin.ReCallingOperatorResult.is_re_calling_operator_result(
-            {"last_execution_result": {"waiting_time": 5, "last_execution_time": 1000.0}}
+            {
+                re_callable_operator_mixin.ReCallingOperatorResult.__name__: {
+                    "last_execution_result": {"waiting_time": 5, "last_execution_time": 1000.0},
+                }
+            }
         ) is True
 
     def test_is_re_calling_operator_result_false_for_non_dict(self):
@@ -124,7 +128,9 @@ def test_get_last_execution_result_returns_inner_dict_for_valid_format(self):
         }
         result = operator.get_last_execution_result({
             dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY: {
-                "last_execution_result": inner,
+                re_callable_operator_mixin.ReCallingOperatorResult.__name__: {
+                    "last_execution_result": inner,
+                },
             },
         })
         assert result == inner
@@ -134,8 +140,10 @@ def test_get_last_execution_result_with_reset_to_id_format(self):
         inner = {"waiting_time": 3.0, "last_execution_time": 500.0}
         result = operator.get_last_execution_result({
             dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY: {
-                "reset_to_id": "abc",
-                "last_execution_result": inner,
+                re_callable_operator_mixin.ReCallingOperatorResult.__name__: {
+                    "reset_to_id": "abc",
+                    "last_execution_result": inner,
+                },
             },
         })
         assert result == inner
@@ -146,11 +154,12 @@ def test_build_re_callable_result(self):
             last_execution_time=1000.0,
             waiting_time=5.0,
         )
-        assert "last_execution_result" in result
-        assert result["last_execution_result"][
+        inner = result[re_callable_operator_mixin.ReCallingOperatorResult.__name__]
+        assert "last_execution_result" in inner
+        assert inner["last_execution_result"][
             re_callable_operator_mixin.ReCallingOperatorResultKeys.LAST_EXECUTION_TIME.value
         ] == 1000.0
-        assert result["last_execution_result"][
+        assert inner["last_execution_result"][
             re_callable_operator_mixin.ReCallingOperatorResultKeys.WAITING_TIME.value
         ] == 5.0
 
@@ -161,8 +170,9 @@ def test_build_re_callable_result_with_reset_to_id(self):
             last_execution_time=1000.0,
             waiting_time=5.0,
         )
-        assert result["reset_to_id"] == "target_123"
-        assert "last_execution_result" in result
+        inner = result[re_callable_operator_mixin.ReCallingOperatorResult.__name__]
+        assert inner["reset_to_id"] == "target_123"
+        assert "last_execution_result" in inner
 
     def test_build_re_callable_result_with_extra_kwargs(self):
         operator = _TestReCallableOperator()
@@ -171,4 +181,5 @@ def test_build_re_callable_result_with_extra_kwargs(self):
             waiting_time=5.0,
             extra_field=42,
         )
-        assert result["last_execution_result"]["extra_field"] == 42
+        inner = result[re_callable_operator_mixin.ReCallingOperatorResult.__name__]
+        assert inner["last_execution_result"]["extra_field"] == 42

packages/flow/octobot_flow/entities/actions/action_details.py

@@ -20,7 +20,9 @@ class AbstractActionDetails(octobot_commons.dataclasses.FlexibleDataclass):
     # unique id of the action
     id: str = dataclasses.field(repr=True)
     # result of the action. Set after the action is executed
-    result: typing.Optional[dict] = dataclasses.field(default=None, repr=True)
+    result: typing.Optional[
+        octobot_commons.dsl_interpreter.ComputedOperatorParameterType
+    ] = dataclasses.field(default=None, repr=True)
     # error status of the action. Set after the action is executed, in case an error occured
     error_status: typing.Optional[str] = dataclasses.field(default=None, repr=True)       # ActionErrorStatus
     # time at which the action was executed

packages/flow/octobot_flow/entities/automations/post_iteration_actions_details.py

@@ -1,7 +1,8 @@
 import dataclasses
-import octobot_commons.dataclasses
 import typing
 
+import octobot_commons.dataclasses
+
 
 @dataclasses.dataclass
 class RefreshExchangeBotsAuthenticatedDataDetails:
@@ -21,7 +22,7 @@ class NextIterationDetails(octobot_commons.dataclasses.FlexibleDataclass):
 
 
 @dataclasses.dataclass
-class PostIterationActionsDetails(octobot_commons.dataclasses.FlexibleDataclass):
+class PostIterationActionsDetails(octobot_commons.dataclasses.MinimizableDataclass):
     stop_automation: bool = False
     postpone_execution: bool = False
     postpone_reason: typing.Optional[str] = None

packages/flow/octobot_flow/enums.py

@@ -26,7 +26,6 @@ class ChangedElements(enum.Enum):
 
 class ActionType(enum.Enum):
     APPLY_CONFIGURATION = "apply_configuration"
-    STOP_AUTOMATION = "stop_automation"
     UNKNOWN = "unknown"
 
 

packages/flow/octobot_flow/jobs/automation_job.py

@@ -59,12 +59,14 @@ async def run(self) -> list[octobot_flow.entities.AbstractActionDetails]:
                 to_execute_actions, are_priority_actions = self._get_actions_to_execute()
                 if are_priority_actions:
                     self._logger.info(f"Running {len(to_execute_actions)} priority actions: {to_execute_actions}")
+                    self._resolve_dsl_scripts(to_execute_actions, True)
                 else:
                     # fetch the actions and signals if any
                     await self._fetch_actions(maybe_authenticator)
                     # resolve the DSL scripts in case it has dependencies on other actions
                     self._resolve_dsl_scripts(
-                        self.automation_state.automation.actions_dag.get_executable_actions()
+                        self.automation_state.automation.actions_dag.get_executable_actions(),
+                        True
                     )
                 # fetch the dependencies of the automation environment
                 fetched_dependencies = await self._fetch_dependencies(maybe_community_repository, to_execute_actions)
@@ -285,10 +287,17 @@ def _get_pending_priority_actions(self) -> list[octobot_flow.entities.AbstractAc
             action for action in self.automation_state.priority_actions if not action.is_completed()
         ]
 
-    def _resolve_dsl_scripts(self, actions: list[octobot_flow.entities.AbstractActionDetails]):
-        self.automation_state.automation.actions_dag.resolve_dsl_scripts(
-            actions
-        )
+    def _resolve_dsl_scripts(
+        self, actions: list[octobot_flow.entities.AbstractActionDetails],
+        from_actions_dag: bool
+    ):
+        if from_actions_dag:
+            self.automation_state.automation.actions_dag.resolve_dsl_scripts(
+                actions
+            )
+        else:
+            local_dag = octobot_flow.entities.ActionsDAG(actions=actions)
+            local_dag.resolve_dsl_scripts(actions)
 
     def _clear_resolved_dsl_scripts(self, actions: list[octobot_flow.entities.AbstractActionDetails]):
         for action in actions:

packages/flow/octobot_flow/logic/actions/actions_executor.py

@@ -10,7 +10,6 @@
 import octobot_flow.entities
 import octobot_flow.repositories.community
 import octobot_flow.logic.dsl
-import octobot_flow.logic.configured_actions
 import octobot_flow.enums
 import octobot_flow.errors
 
@@ -48,30 +47,9 @@ async def execute(self):
         async with dsl_executor.dependencies_context(self._actions):
             for index, action in enumerate(self._actions):
                 await self._execute_action(dsl_executor, action)
-                if octobot_commons.dsl_interpreter.ReCallingOperatorResult.is_re_calling_operator_result(
-                    action.result
-                ):
-                    recall_dag_details = octobot_commons.dsl_interpreter.ReCallingOperatorResult.from_dict(
-                        action.result
-                    )
-                    if not recall_dag_details.reset_to_id:
-                        # reset to the current action if no specific id is provided (loop on this action)
-                        recall_dag_details.reset_to_id = action.id
-                    if recall_dag_details.reset_to_id == action.id:
-                        # Keep executing other selected actions if any: those are not affected by the reset
-                        # as they don't depend on the reset action
-                        continue
-                    # Reset to a past action: interrupt execution of the following actions 
-                    # as they might depend on the reset action
-                    if index < len(self._actions) - 1:
-                        interrupted_action = self._actions[index + 1: ]
-                        self._get_logger().info(
-                            f"DAG reset required. Interrupting execution of "
-                            f"{len(interrupted_action)} actions: "
-                            f"{', '.join([action.id for action in interrupted_action])}"
-                        )
-                        break
-
+                recall_dag_details, should_stop_processing = self._handle_execution_result(action, index)
+                if should_stop_processing:
+                    break
         self._sync_after_execution()
         await self._update_actions_history()
         await self._insert_execution_bot_logs(dsl_executor.pending_bot_logs)
@@ -85,18 +63,51 @@ async def execute(self):
             # no reset: schedule immediately
             self.next_execution_scheduled_to = 0
 
+    def _handle_execution_result(
+        self, action: octobot_flow.entities.AbstractActionDetails, index: int
+    ) -> tuple[typing.Optional[octobot_commons.dsl_interpreter.ReCallingOperatorResult], bool]:
+        if not isinstance(action.result, dict):
+            return None, False
+        if octobot_flow.entities.PostIterationActionsDetails.__name__ in action.result:
+            post_iteration_actions_details = octobot_flow.entities.PostIterationActionsDetails.from_dict(
+                action.result[octobot_flow.entities.PostIterationActionsDetails.__name__]
+            )
+            if post_iteration_actions_details.stop_automation:
+                self._get_logger().info(f"Stopping automation: {self._automation.metadata.automation_id}")
+                self._automation.post_actions.stop_automation = True
+                # todo cancel open orders and sell assets if required in action config
+                return None, True
+            return None, False
+        if octobot_commons.dsl_interpreter.ReCallingOperatorResult.is_re_calling_operator_result(action.result):
+            recall_dag_details = octobot_commons.dsl_interpreter.ReCallingOperatorResult.from_dict(
+                action.result[octobot_commons.dsl_interpreter.ReCallingOperatorResult.__name__]
+            )
+            if not recall_dag_details.reset_to_id:
+                # reset to the current action if no specific id is provided (loop on this action)
+                recall_dag_details.reset_to_id = action.id
+            if recall_dag_details.reset_to_id == action.id:
+                # Keep executing other selected actions if any: those are not affected by the reset
+                # as they don't depend on the reset action
+                return recall_dag_details, False
+            # Reset to a past action: interrupt execution of the following actions 
+            # as they might depend on the reset action
+            if index < len(self._actions) - 1:
+                interrupted_action = self._actions[index + 1: ]
+                self._get_logger().info(
+                    f"DAG reset required. Interrupting execution of "
+                    f"{len(interrupted_action)} actions: "
+                    f"{', '.join([action.id for action in interrupted_action])}"
+                )
+                return recall_dag_details, True
+        return None, False
+
     async def _execute_action(
         self,
         dsl_executor: "octobot_flow.logic.dsl.DSLExecutor",
         action: octobot_flow.entities.AbstractActionDetails
     ):
         if isinstance(action, octobot_flow.entities.DSLScriptActionDetails):
             return await dsl_executor.execute_action(action)
-        elif isinstance(action, octobot_flow.entities.ConfiguredActionDetails):
-            configured_actions_executor = octobot_flow.logic.configured_actions.ConfiguredActionsExecutor(
-                self._exchange_manager, self._automation
-            )
-            return await configured_actions_executor.execute_action(action)
         raise octobot_flow.errors.UnsupportedActionTypeError(
             f"{self.__class__.__name__} does not support action type: {type(action)}"
         ) from None

packages/flow/octobot_flow/logic/configured_actions/__init__.py

@@ -103,7 +103,9 @@ async def test_exchange_actions_creating_and_waiting_and_cancelling_limit(
         rescheduled_parameters = wait_action.get_rescheduled_parameters()
         assert dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY in rescheduled_parameters
         last_execution_result = dsl_interpreter.ReCallingOperatorResult.from_dict(
-            rescheduled_parameters[dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY]
+            rescheduled_parameters[dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY][
+                dsl_interpreter.ReCallingOperatorResult.__name__
+            ]
         )
         assert isinstance(last_execution_result.last_execution_result, dict)
         waiting_time_1 = last_execution_result.last_execution_result[dsl_interpreter.ReCallingOperatorResultKeys.WAITING_TIME.value]
@@ -135,7 +137,9 @@ async def test_exchange_actions_creating_and_waiting_and_cancelling_limit(
         rescheduled_parameters = wait_action.get_rescheduled_parameters()
         assert dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY in rescheduled_parameters
         last_execution_result = dsl_interpreter.ReCallingOperatorResult.from_dict(
-            rescheduled_parameters[dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY]
+            rescheduled_parameters[dsl_interpreter.ReCallableOperatorMixin.LAST_EXECUTION_RESULT_KEY][
+                dsl_interpreter.ReCallingOperatorResult.__name__
+            ]
         )
         assert isinstance(last_execution_result.last_execution_result, dict)
         waiting_time_2 = last_execution_result.last_execution_result[dsl_interpreter.ReCallingOperatorResultKeys.WAITING_TIME.value]