Merge pull request #1667 from Drakkar-Software/dev

Master update

Author: GuillaumeDSM

Automation/conditions/scripted_condition/__init__.py

@@ -0,0 +1 @@
+from .scripted_condition import ScriptedCondition

Automation/conditions/scripted_condition/metadata.json

@@ -0,0 +1,6 @@
+{
+  "version": "1.2.0",
+  "origin_package": "OctoBot-Default-Tentacles",
+  "tentacles": ["ScriptedCondition"],
+  "tentacles-requirements": []
+}

Automation/conditions/scripted_condition/scripted_condition.py

@@ -0,0 +1,103 @@
+#  This file is part of OctoBot (https://github.com/Drakkar-Software/OctoBot)
+#  Copyright (c) 2023 Drakkar-Software, All rights reserved.
+#
+#  OctoBot is free software; you can redistribute it and/or
+#  modify it under the terms of the GNU General Public License
+#  as published by the Free Software Foundation; either
+#  version 3.0 of the License, or (at your option) any later version.
+#
+#  OctoBot 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
+#  General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public
+#  License along with OctoBot. If not, see <https://www.gnu.org/licenses/>.
+import typing
+
+import octobot_commons.configuration as configuration
+import octobot_trading.api as trading_api
+import octobot_commons.enums as commons_enums
+import octobot.automation.bases.abstract_condition as abstract_condition
+import octobot_commons.dsl_interpreter as dsl_interpreter
+
+import tentacles.Meta.DSL_operators as dsl_operators
+
+
+class ScriptedCondition(abstract_condition.AbstractCondition):
+    SCRIPT = "script"
+    EXCHANGE = "exchange"
+
+    def __init__(self):
+        super().__init__()
+        self.script: str = ""
+        self.exchange_name: str = ""
+
+        self._dsl_interpreter: typing.Optional[dsl_interpreter.Interpreter] = None
+
+    async def evaluate(self) -> bool:
+        if self._dsl_interpreter:
+            script_result = await self._dsl_interpreter.interprete(self.script)
+            return bool(script_result)
+        raise ValueError("Scripted condition is not properly configured, the script is likely invalid.")
+
+    @staticmethod
+    def get_description() -> str:
+        return "Evaluates a scripted condition using the OctoBot DSL."
+
+    def get_user_inputs(self, UI: configuration.UserInputFactory, inputs: dict, step_name: str) -> dict:
+        exchanges = list(trading_api.get_exchange_names())
+        return {
+            self.SCRIPT: UI.user_input(
+                self.SCRIPT, commons_enums.UserInputTypes.TEXT, "", inputs,
+                title="Scripted condition: the OctoBot DSL expression to evaluate (more info in automation details). Its return value will be converted to a boolean using \"bool()\" to determine if the condition is met.",
+                parent_input_name=step_name,
+            ),
+            self.EXCHANGE: UI.user_input(
+                self.EXCHANGE, commons_enums.UserInputTypes.OPTIONS, exchanges[0], inputs,
+                options=exchanges,
+                title="Exchange: the name of the exchange to use for the condition.",
+                parent_input_name=step_name,
+            )
+        }
+
+    def apply_config(self, config):
+        self.script = config[self.SCRIPT]
+        self.exchange_name = config[self.EXCHANGE]
+        if self.script and self.exchange_name:
+            self._dsl_interpreter = self._create_dsl_interpreter()
+            self._validate_script()
+        else:
+            self._dsl_interpreter = None
+    
+    def _validate_script(self):
+        try:
+            self._dsl_interpreter.prepare(self.script)
+            self.logger.info(
+                f"Formula interpreter successfully prepared \"{self.script}\" condition"
+            )
+        except Exception as e:
+            self.logger.error(f"Error when parsing condition {self.script}: {e}")
+            raise e
+
+    def _create_dsl_interpreter(self):
+        exchange_manager = self._get_exchange_manager()
+        ohlcv_operators = []
+        portfolio_operators = []
+        if exchange_manager is not None:
+            ohlcv_operators = dsl_operators.exchange_operators.create_ohlcv_operators(
+                exchange_manager, None, None
+            )
+            portfolio_operators = dsl_operators.exchange_operators.create_portfolio_operators(
+                exchange_manager
+            )
+        return dsl_interpreter.Interpreter(
+            dsl_interpreter.get_all_operators() + ohlcv_operators + portfolio_operators
+        )
+    
+    def _get_exchange_manager(self):
+        for exchange_id in trading_api.get_exchange_ids():
+            exchange_manager = trading_api.get_exchange_manager_from_exchange_id(exchange_id)
+            if exchange_manager.exchange_name == self.exchange_name and exchange_manager.is_backtesting == False:
+                return exchange_manager
+        raise ValueError(f"No exchange manager found for exchange name: {self.exchange_name}")

Meta/DSL_operators/exchange_operators/__init__.py

@@ -21,11 +21,17 @@
     ExchangeDataDependency,
     create_ohlcv_operators,
 )
-
+import tentacles.Meta.DSL_operators.exchange_operators.exchange_private_data_operators
+from tentacles.Meta.DSL_operators.exchange_operators.exchange_private_data_operators import (
+    PortfolioOperator,
+    create_portfolio_operators,
+)
 
 
 __all__ = [
     "OHLCVOperator",
     "ExchangeDataDependency",
     "create_ohlcv_operators",
+    "PortfolioOperator",
+    "create_portfolio_operators",
 ]

Meta/DSL_operators/exchange_operators/exchange_private_data_operators/__init__.py

@@ -0,0 +1,26 @@
+# pylint: disable=R0801
+#  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 tentacles.Meta.DSL_operators.exchange_operators.exchange_private_data_operators.portfolio_operators
+from tentacles.Meta.DSL_operators.exchange_operators.exchange_private_data_operators.portfolio_operators import (
+    PortfolioOperator,
+    create_portfolio_operators,
+)
+__all__ = [
+    "PortfolioOperator",
+    "create_portfolio_operators",
+]

Meta/DSL_operators/exchange_operators/exchange_private_data_operators/portfolio_operators.py

@@ -0,0 +1,85 @@
+# pylint: disable=missing-class-docstring,missing-function-docstring
+#  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 typing
+
+import octobot_commons.constants
+import octobot_commons.errors
+import octobot_commons.dsl_interpreter as dsl_interpreter
+import octobot_trading.personal_data
+import octobot_trading.exchanges
+import octobot_trading.api
+
+import tentacles.Meta.DSL_operators.exchange_operators.exchange_operator as exchange_operator
+
+
+class PortfolioOperator(exchange_operator.ExchangeOperator):
+    def __init__(self, *parameters: dsl_interpreter.OperatorParameterType, **kwargs: typing.Any):
+        super().__init__(*parameters, **kwargs)
+        self.value: dsl_interpreter_operator.ComputedOperatorParameterType = exchange_operator.UNINITIALIZED_VALUE # type: ignore
+
+    @staticmethod
+    def get_library() -> str:
+        # this is a contextual operator, so it should not be included by default in the get_all_operators function return values
+        return octobot_commons.constants.CONTEXTUAL_OPERATORS_LIBRARY
+
+    @staticmethod
+    def get_parameters() -> list[dsl_interpreter.OperatorParameter]:
+        return [
+            dsl_interpreter.OperatorParameter(name="asset", description="the asset to get the value for", required=False, type=str),
+        ]
+
+    def compute(self) -> dsl_interpreter.ComputedOperatorParameterType:
+        if self.value is exchange_operator.UNINITIALIZED_VALUE:
+            raise octobot_commons.errors.DSLInterpreterError("{self.__class__.__name__} has not been initialized")
+        return self.value
+
+
+def create_portfolio_operators(
+    exchange_manager: typing.Optional[octobot_trading.exchanges.ExchangeManager],
+) -> typing.List[type[PortfolioOperator]]:
+
+    def _get_asset_holdings(asset: str) -> octobot_trading.personal_data.Asset:
+        return octobot_trading.api.get_portfolio_currency(exchange_manager, asset)
+
+    class _TotalOperator(PortfolioOperator):
+        DESCRIPTION = "Returns the total holdings of the asset in the portfolio"
+        EXAMPLE = "total('BTC')"
+
+        @staticmethod
+        def get_name() -> str:
+            return "total"
+
+        async def pre_compute(self) -> None:
+            await super().pre_compute()
+            asset = self.get_computed_parameters()[0]
+            self.value = float(_get_asset_holdings(asset).total)
+
+    class _AvailableOperator(PortfolioOperator):
+        DESCRIPTION = "Returns the available holdings of the asset in the portfolio"
+        EXAMPLE = "available('BTC')"
+
+        @staticmethod
+        def get_name() -> str:
+            return "available"
+
+        async def pre_compute(self) -> None:
+            await super().pre_compute()
+            asset = self.get_computed_parameters()[0]
+            self.value = float(_get_asset_holdings(asset).available)
+        
+
+    return [_TotalOperator, _AvailableOperator]

Meta/DSL_operators/exchange_operators/exchange_public_data_operators/ohlcv_operators.py

@@ -77,8 +77,8 @@ def compute(self) -> dsl_interpreter.ComputedOperatorParameterType:
 
 def create_ohlcv_operators(
     exchange_manager: typing.Optional[octobot_trading.exchanges.ExchangeManager],
-    symbol: str,
-    time_frame: str,
+    symbol: typing.Optional[str],
+    time_frame: typing.Optional[str],
     candle_manager_by_time_frame_by_symbol: typing.Optional[
         typing.Dict[str, typing.Dict[str, octobot_trading.exchange_data.CandlesManager]]
     ] = None
@@ -120,7 +120,7 @@ def _get_candles_values_with_latest_kline_if_available(
                 else:
                     octobot_commons.logging.get_logger(OHLCVOperator.__name__).error(
                         f"{exchange_manager.exchange_name + '' if exchange_manager is not None else ''}{_symbol} {_time_frame} "
-                        f"kline time ({kline_time}) is not equal to last candle time not the last time + {time_frame} "
+                        f"kline time ({kline_time}) is not equal to last candle time not the last time + {_time_frame} "
                         f"({last_candle_time} + {tf_seconds}) seconds. Kline has been ignored."
                     )
         return candles_values
@@ -145,41 +145,59 @@ async def pre_compute(self) -> None:
             self.value = _get_candles_values_with_latest_kline_if_available(*self.get_symbol_and_time_frame(), self.PRICE_INDEX, -1)
 
     class _OpenPriceOperator(_LocalOHLCVOperator):
+        DESCRIPTION = "Returns the candle's open price as array of floats"
+        EXAMPLE = "open('BTC/USDT', '1h')"
+
         PRICE_INDEX = commons_enums.PriceIndexes.IND_PRICE_OPEN
 
         @staticmethod
         def get_name() -> str:
             return "open"
 
     class _HighPriceOperator(_LocalOHLCVOperator):
+        DESCRIPTION = "Returns the candle's high price as array of floats"
+        EXAMPLE = "high('BTC/USDT', '1h')"
+
         PRICE_INDEX = commons_enums.PriceIndexes.IND_PRICE_HIGH
 
         @staticmethod
         def get_name() -> str:
             return "high"
 
     class _LowPriceOperator(_LocalOHLCVOperator):
+        DESCRIPTION = "Returns the candle's low price as array of floats"
+        EXAMPLE = "low('BTC/USDT', '1h')"
+
         PRICE_INDEX = commons_enums.PriceIndexes.IND_PRICE_LOW
 
         @staticmethod
         def get_name() -> str:
             return "low"
 
     class _ClosePriceOperator(_LocalOHLCVOperator):
+        DESCRIPTION = "Returns the candle's close price as array of floats"
+        EXAMPLE = "close('BTC/USDT', '1h')"
+
         PRICE_INDEX = commons_enums.PriceIndexes.IND_PRICE_CLOSE
 
         @staticmethod
         def get_name() -> str:
             return "close"
 
     class _VolumePriceOperator(_LocalOHLCVOperator):
+        DESCRIPTION = "Returns the candle's volume as array of floats"
+        EXAMPLE = "volume('BTC/USDT', '1h')"
+
         PRICE_INDEX = commons_enums.PriceIndexes.IND_PRICE_VOL
 
         @staticmethod
         def get_name() -> str:
             return "volume"
 
     class _TimePriceOperator(_LocalOHLCVOperator):
+        DESCRIPTION = "Returns the candle's time as array of floats"
+        EXAMPLE = "time('BTC/USDT', '1h')"
+
         PRICE_INDEX = commons_enums.PriceIndexes.IND_PRICE_TIME
 
         @staticmethod