[DCA][Staggered] init health check

Evaluator/Strategies/blank_strategy_evaluator/blank_strategy.py

@@ -28,7 +28,7 @@ def init_user_inputs(self, inputs: dict) -> None:
         """
         super().init_user_inputs(inputs)
         self.UI.user_input(common_constants.CONFIG_TENTACLES_REQUIRED_CANDLES_COUNT, common_enums.UserInputTypes.INT,
-                        200, inputs, min_val=200,
+                        200, inputs, min_val=1,
                         title="Initialization candles count: the number of historical candles to fetch from "
                               "exchanges when OctoBot is starting.")
 

Trading/Mode/dca_trading_mode/dca_trading.py

@@ -33,6 +33,7 @@
 import octobot_trading.util as trading_util
 import octobot_trading.errors as trading_errors
 import octobot_trading.personal_data as trading_personal_data
+import octobot_trading.exchanges as trading_exchanges
 import octobot_trading.modes.script_keywords as script_keywords
 
 
@@ -499,6 +500,9 @@ class DCATradingMode(trading_modes.AbstractTradingMode):
     MODE_PRODUCER_CLASSES = [DCATradingModeProducer]
     MODE_CONSUMER_CLASSES = [DCATradingModeConsumer]
     SUPPORTS_INITIAL_PORTFOLIO_OPTIMIZATION = True
+    SUPPORTS_HEALTH_CHECK = True
+    HEALTH_CHECK_SELL_ORPHAN_FUNDS_RATIO = decimal.Decimal("0.15")  # 15%
+    HEALTH_CHECK_FILL_ORDERS_TIMEOUT = 20
 
     def __init__(self, config, exchange_manager):
         super().__init__(config, exchange_manager)
@@ -734,3 +738,73 @@ async def single_exchange_process_optimize_initial_portfolio(
         return await trading_modes.convert_assets_to_target_asset(
             self, sellable_assets, target_asset, tickers
         )
+
+    async def single_exchange_process_health_check(self, chained_orders: list, tickers: dict) -> list:
+        created_orders = []
+        common_quote = trading_exchanges.get_common_traded_quote(self.exchange_manager)
+        for asset, amount in self._get_lost_funds_to_sell(common_quote, chained_orders):
+            # sell lost funds
+            self.logger.info(
+                f"Health check: selling {amount} {asset} into {common_quote}"
+            )
+            try:
+                created_orders += await trading_modes.convert_asset_to_target_asset(
+                    self, asset, common_quote, tickers, asset_amount=amount
+                )
+            except Exception as err:
+                self.logger.exception(
+                    err, True, f"Error when creating order to sell {asset} into {common_quote}: {err}"
+                )
+        if created_orders:
+            await asyncio.gather(
+                *[
+                    trading_personal_data.wait_for_order_fill(
+                        order, self.HEALTH_CHECK_FILL_ORDERS_TIMEOUT, True
+                    ) for order in created_orders
+                ]
+            )
+        return created_orders
+
+    def _get_lost_funds_to_sell(self, common_quote: str, chained_orders: list) -> list[(str, decimal.Decimal)]:
+        asset_and_amount = []
+        value_holder = self.exchange_manager.exchange_personal_data.portfolio_manager.portfolio_value_holder
+        traded_base_assets = set(
+            symbol.base
+            for symbol in self.exchange_manager.exchange_config.traded_symbols
+        )
+        sell_orders = [
+            order
+            for order in self.exchange_manager.exchange_personal_data.orders_manager.get_open_orders() + chained_orders
+            if order.side is trading_enums.TradeOrderSide.SELL
+        ]
+        orphan_asset_values_by_asset = {}
+        total_traded_assets_value = value_holder.value_converter.evaluate_value(
+            common_quote,
+            self.exchange_manager.exchange_personal_data.portfolio_manager.portfolio.get_currency_portfolio(
+                common_quote
+            ).total,
+            init_price_fetchers=False
+        )
+        for asset in traded_base_assets:
+            holdings = self.exchange_manager.exchange_personal_data.portfolio_manager.portfolio.get_currency_portfolio(
+                asset
+            ).total
+            holdings_value = value_holder.value_converter.evaluate_value(
+                asset, holdings, init_price_fetchers=False
+            )
+            total_traded_assets_value += holdings_value
+            holdings_in_sell_orders = sum(
+                order.origin_quantity
+                for order in sell_orders
+                if symbol_util.parse_symbol(order.symbol).base == asset
+            )
+            orphan_amount = holdings - holdings_in_sell_orders
+            if orphan_amount and orphan_amount > 0:
+                orphan_asset_values_by_asset[asset] = (holdings_value * orphan_amount / holdings, orphan_amount)
+
+        for asset, value_and_orphan_amount in orphan_asset_values_by_asset.items():
+            value, orphan_amount = value_and_orphan_amount
+            ratio = value / total_traded_assets_value
+            if ratio > self.HEALTH_CHECK_SELL_ORPHAN_FUNDS_RATIO:
+                asset_and_amount.append((asset, orphan_amount))
+        return asset_and_amount

Trading/Mode/grid_trading_mode/grid_trading.py

@@ -42,6 +42,7 @@ class GridTradingMode(staggered_orders_trading.StaggeredOrdersTradingMode):
     USER_COMMAND_PAUSE_ORDER_MIRRORING = "pause orders mirroring"
     USER_COMMAND_TRADING_PAIR = "trading pair"
     USER_COMMAND_PAUSE_TIME = "pause length in seconds"
+    SUPPORTS_HEALTH_CHECK = False   # WIP   # set True when self.health_check is implemented
 
     def init_user_inputs(self, inputs: dict) -> None:
         """

Trading/Mode/staggered_orders_trading_mode/staggered_orders_trading.py

@@ -32,6 +32,7 @@
 import octobot_trading.enums as trading_enums
 import octobot_trading.personal_data as trading_personal_data
 import octobot_trading.errors as trading_errors
+import octobot_trading.exchanges.util as exchange_util
 
 
 class StrategyModes(enum.Enum):
@@ -119,11 +120,13 @@ class StaggeredOrdersTradingMode(trading_modes.AbstractTradingMode):
     CONFIG_SELL_VOLUME_PER_ORDER = "sell_volume_per_order"
     CONFIG_BUY_VOLUME_PER_ORDER = "buy_volume_per_order"
     CONFIG_IGNORE_EXCHANGE_FEES = "ignore_exchange_fees"
+    ENABLE_UPWARDS_PRICE_FOLLOW = "enable_upwards_price_follow"
     CONFIG_USE_FIXED_VOLUMES_FOR_MIRROR_ORDERS = "use_fixed_volume_for_mirror_orders"
     CONFIG_DEFAULT_SPREAD_PERCENT = 1.5
     CONFIG_DEFAULT_INCREMENT_PERCENT = 0.5
     REQUIRE_TRADES_HISTORY = True   # set True when this trading mode needs the trade history to operate
     SUPPORTS_INITIAL_PORTFOLIO_OPTIMIZATION = True  # set True when self._optimize_initial_portfolio is implemented
+    SUPPORTS_HEALTH_CHECK = False   # set True when self.health_check is implemented
 
     def init_user_inputs(self, inputs: dict) -> None:
         """
@@ -251,6 +254,23 @@ def get_is_symbol_wildcard(cls) -> bool:
     def set_default_config(self):
         raise RuntimeError(f"Impossible to start {self.get_name()} without a valid configuration file.")
 
+    async def single_exchange_process_health_check(self, chained_orders: list, tickers: dict) -> list:
+        created_orders = []
+        if await self._should_rebalance_orders():
+            target_asset = exchange_util.get_common_traded_quote(self.exchange_manager)
+            created_orders += await self.single_exchange_process_optimize_initial_portfolio([], target_asset, tickers)
+            for producer in self.producers:
+                await producer.trigger_staggered_orders_creation()
+        return created_orders
+
+    async def _should_rebalance_orders(self):
+        for producer in self.producers:
+            if producer.enable_upwards_price_follow:
+                # trigger rebalance when current price is beyond the highest sell order
+                if await producer.is_price_beyond_boundaries():
+                    return True
+        return False
+
     async def single_exchange_process_optimize_initial_portfolio(
         self, sellable_assets, target_asset: str, tickers: dict
     ) -> list:
@@ -515,6 +535,7 @@ def __init__(self, channel, config, trading_mode, exchange_manager):
 
         self.use_existing_orders_only = self.limit_orders_count_if_necessary = \
             self.ignore_exchange_fees = self.use_fixed_volume_for_mirror_orders = False
+        self.enable_upwards_price_follow = True
         self.mode = self.spread \
             = self.increment = self.operational_depth \
             = self.lowest_buy = self.highest_sell \
@@ -591,6 +612,9 @@ def read_config(self):
         # end tmp
         self.ignore_exchange_fees = self.symbol_trading_config.get(self.trading_mode.CONFIG_IGNORE_EXCHANGE_FEES,
                                                                    self.ignore_exchange_fees)
+        self.enable_upwards_price_follow = self.symbol_trading_config.get(
+            self.trading_mode.ENABLE_UPWARDS_PRICE_FOLLOW, self.enable_upwards_price_follow
+        )
 
     async def start(self) -> None:
         await super().start()
@@ -617,6 +641,18 @@ async def set_final_eval(self, matrix_id: str, cryptocurrency: str, symbol: str,
         # nothing to do: this is not a strategy related trading mode
         pass
 
+    async def is_price_beyond_boundaries(self):
+        open_orders = self.exchange_manager.exchange_personal_data.orders_manager.get_open_orders(self.symbol)
+        price = await trading_personal_data.get_up_to_date_price(
+            self.exchange_manager, self.symbol, timeout=self.PRICE_FETCHING_TIMEOUT
+        )
+        max_order_price = max(
+            order.origin_price for order in open_orders
+        )
+        # price is above max order price
+        if max_order_price < price and self.enable_upwards_price_follow:
+            return True
+
     def _schedule_order_refresh(self):
         # schedule order creation / health check
         asyncio.create_task(self._ensure_staggered_orders_and_reschedule())