[DCA][Staggered] add optimize portfolio tests

Trading/Mode/dca_trading_mode/dca_trading.py

@@ -459,6 +459,7 @@ async def _send_alert_notification(self, symbol, state, step):
 class DCATradingMode(trading_modes.AbstractTradingMode):
     MODE_PRODUCER_CLASSES = [DCATradingModeProducer]
     MODE_CONSUMER_CLASSES = [DCATradingModeConsumer]
+    SUPPORTS_INITIAL_PORTFOLIO_OPTIMIZATION = True
 
     def __init__(self, config, exchange_manager):
         super().__init__(config, exchange_manager)
@@ -664,28 +665,10 @@ def get_current_state(self) -> (str, float):
             else self.producers[0].final_eval
         )
 
-    async def optimize_initial_portfolio(self, sellable_assets: list, tickers: dict) -> list:
-        if not self.producers:
-            # nothing to do
-            return []
-        target_asset = exchange_util.get_common_traded_quote(self.exchange_manager)
-        if target_asset is None:
-            self.logger.error(f"Impossible to optimize initial portfolio with different quotes in traded pairs")
-            return []
-        async with self.producers[0].trading_mode_trigger():
-            if self.producers[0].producer_exchange_wide_lock(self.exchange_manager).locked():
-                # already locked by another trading mode instance: this other trading mode will do the rebalancing
-                self.logger.info(
-                    f"Skipping portfolio optimization for trading mode with symbol {self.symbol}: "
-                    f"portfolio optimization already in progress"
-                )
-                return []
-            async with self.producers[0].producer_exchange_wide_lock(self.exchange_manager):
-                self.logger.info(f"Optimizing portfolio: selling {sellable_assets} to buy {target_asset}")
-                created_orders = await trading_modes.convert_assets_to_target_asset(
-                    self, sellable_assets, target_asset, tickers
-                )
-                if not created_orders:
-                    self.logger.info("Optimizing portfolio: no order to create")
-            await trading_modes.notify_portfolio_optimization_complete()
-            return created_orders
+    async def single_exchange_process_optimize_initial_portfolio(
+        self, sellable_assets, target_asset: str, tickers: dict
+    ) -> list:
+        self.logger.info(f"Optimizing portfolio: selling {sellable_assets} to buy {target_asset}")
+        return await trading_modes.convert_assets_to_target_asset(
+            self, sellable_assets, target_asset, tickers
+        )

Trading/Mode/dca_trading_mode/tests/test_dca_trading_mode.py

@@ -37,6 +37,7 @@
 import octobot_trading.personal_data as trading_personal_data
 import octobot_trading.enums as trading_enums
 import octobot_trading.constants as trading_constants
+import octobot_trading.modes
 
 import tentacles.Evaluator.TA as TA
 import tentacles.Evaluator.Strategies as Strategies
@@ -694,6 +695,18 @@ async def _create_entry_order(_, __, ___, ____, _____, created_orders, ______):
         _create_entry_order_mock.reset_mock()
 
 
+async def test_single_exchange_process_optimize_initial_portfolio(tools):
+    update = {}
+    mode, producer, consumer, trader = await _init_mode(tools, _get_config(tools, update))
+
+    with mock.patch.object(
+        octobot_trading.modes, "convert_assets_to_target_asset", mock.AsyncMock(return_value=["order_1"])
+    ) as convert_assets_to_target_asset_mock:
+        orders = await mode.single_exchange_process_optimize_initial_portfolio(["BTC", "ETH"], "USDT", {})
+        convert_assets_to_target_asset_mock.assert_called_once_with(mode, ["BTC", "ETH"], "USDT", {})
+        assert orders == ["order_1"]
+
+
 async def _check_open_orders_count(trader, count):
     assert len(trading_api.get_open_orders(trader.exchange_manager)) == count
 

Trading/Mode/staggered_orders_trading_mode/staggered_orders_trading.py

@@ -253,58 +253,41 @@ 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 optimize_initial_portfolio(self, sellable_assets: list, tickers: dict) -> list:
-        if not self.producers:
-            # nothing to do
-            return []
-        producer = self.producers[0]
-        common_quote = exchange_util.get_common_traded_quote(self.exchange_manager)
-        if common_quote is None:
-            self.logger.error(f"Impossible to optimize initial portfolio with different quotes in traded pairs")
-            return []
+    async def single_exchange_process_optimize_initial_portfolio(
+        self, sellable_assets, target_asset: str, tickers: dict
+    ) -> list:
         portfolio = self.exchange_manager.exchange_personal_data.portfolio_manager.portfolio
-        # first acquire trading mode lock to be sure we are not in during init phase
-        async with producer.trading_mode_trigger():
-            if producer.producer_exchange_wide_lock(self.exchange_manager).locked():
-                # already locked by another trading mode instance: this other trading mode will do the rebalancing
-                self.logger.info(
-                    f"Skipping portfolio optimization for trading mode with symbol {self.symbol}: "
-                    f"portfolio optimization already initialized"
-                )
-                return []
-            async with producer.producer_exchange_wide_lock(self.exchange_manager):
-                self.logger.info(f"Starting portfolio optimization using trading mode with symbol {self.symbol}")
-                pair_bases = set()
-                # 1. cancel open orders
-                try:
-                    cancelled_orders = await self._cancel_associated_orders(producer, pair_bases)
-                except Exception as err:
-                    self.logger.exception(err, True, f"Error during portfolio optimization cancel orders step: {err}")
-                    cancelled_orders = []
-
-                # 2. convert assets to sell funds into target assets
-                try:
-                    part_1_orders = await self._convert_assets_into_target(
-                        producer, pair_bases, common_quote, set(sellable_assets), tickers
-                    )
-                except Exception as err:
-                    self.logger.exception(
-                        err, True, f"Error during portfolio optimization convert into target step: {err}"
-                    )
-                    part_1_orders = []
+        producer = self.producers[0]
+        pair_bases = set()
+        # 1. cancel open orders
+        try:
+            cancelled_orders = await self._cancel_associated_orders(producer, pair_bases)
+        except Exception as err:
+            self.logger.exception(err, True, f"Error during portfolio optimization cancel orders step: {err}")
+            cancelled_orders = []
 
-                # 3. compute necessary funds for each configured_pairs
-                converted_quote_amount_per_symbol = self._get_converted_quote_amount_per_symbol(
-                    portfolio, pair_bases, common_quote
-                )
+        # 2. convert assets to sell funds into target assets
+        try:
+            part_1_orders = await self._convert_assets_into_target(
+                producer, pair_bases, target_asset, set(sellable_assets), tickers
+            )
+        except Exception as err:
+            self.logger.exception(
+                err, True, f"Error during portfolio optimization convert into target step: {err}"
+            )
+            part_1_orders = []
 
-                # 4. buy assets
-                part_2_orders = await self._buy_assets(
-                    producer, pair_bases, common_quote, converted_quote_amount_per_symbol, tickers
-                )
+        # 3. compute necessary funds for each configured_pairs
+        converted_quote_amount_per_symbol = self._get_converted_quote_amount_per_symbol(
+            portfolio, pair_bases, target_asset
+        )
+
+        # 4. buy assets
+        part_2_orders = await self._buy_assets(
+            producer, pair_bases, target_asset, converted_quote_amount_per_symbol, tickers
+        )
 
-                await trading_modes.notify_portfolio_optimization_complete()
-                return [cancelled_orders, part_1_orders, part_2_orders]
+        return [cancelled_orders, part_1_orders, part_2_orders]
 
     async def _cancel_associated_orders(self, producer, pair_bases) -> list:
         cancelled_orders = []
@@ -314,7 +297,7 @@ async def _cancel_associated_orders(self, producer, pair_bases) -> list:
             if producer.get_symbol_trading_config(symbol) is not None:
                 pair_bases.add(symbol_util.parse_symbol(symbol).base)
                 for order in self.exchange_manager.exchange_personal_data.orders_manager.get_open_orders(
-                        symbol=symbol
+                    symbol=symbol
                 ):
                     if not (order.is_cancelled() or order.is_closed()):
                         await self.cancel_order(order)

Trading/Mode/staggered_orders_trading_mode/tests/test_staggered_orders_trading_mode.py

@@ -22,18 +22,25 @@
 import contextlib
 
 import async_channel.util as channel_util
+
 import octobot_tentacles_manager.api as tentacles_manager_api
+
 import octobot_backtesting.api as backtesting_api
+
 import octobot_commons.asyncio_tools as asyncio_tools
 import octobot_commons.constants as commons_constants
 import octobot_commons.tests.test_config as test_config
+
 import octobot_trading.api as trading_api
 import octobot_trading.exchange_channel as exchanges_channel
 import octobot_trading.enums as trading_enums
 import octobot_trading.exchanges as exchanges
 import octobot_trading.personal_data as trading_personal_data
 import octobot_trading.constants as trading_constants
+import octobot_trading.modes
+
 import tentacles.Trading.Mode.staggered_orders_trading_mode.staggered_orders_trading as staggered_orders_trading
+
 import tests.test_utils.config as test_utils_config
 import tests.test_utils.memory_check_util as memory_check_util
 import tests.test_utils.test_exchanges as test_exchanges
@@ -1339,6 +1346,53 @@ async def test_ensure_current_price_in_limit_parameters():
             assert producer.already_errored_on_out_of_window_price is True
 
 
+async def test_single_exchange_process_optimize_initial_portfolio():
+    async with _get_tools("BTC/USD") as tools:
+        producer, _, exchange_manager = tools
+        mode = producer.trading_mode
+        exchange_manager.exchange_config.traded_symbol_pairs = ["BTC/USD"]
+        exchange_manager.client_symbols = ["BTC/USD"]
+
+        initial_portfolio = exchange_manager.exchange_personal_data.portfolio_manager.portfolio.portfolio
+        assert initial_portfolio["BTC"].available == decimal.Decimal("10")
+        assert initial_portfolio["USD"].available == decimal.Decimal("1000")
+
+        limit_buy = trading_personal_data.BuyLimitOrder(exchange_manager.trader)
+        limit_buy.update(order_type=trading_enums.TraderOrderType.BUY_LIMIT,
+                         symbol="BTC/USD",
+                         current_price=decimal.Decimal(str(50)),
+                         quantity=decimal.Decimal(str(2)),
+                         price=decimal.Decimal(str(50)))
+        await exchange_manager.exchange_personal_data.orders_manager.upsert_order_instance(limit_buy)
+
+        orders = await mode.single_exchange_process_optimize_initial_portfolio(
+            ["BTC", "ETH"], "USD", {"BTC/USD": {trading_enums.ExchangeConstantsTickersColumns.CLOSE.value: 1000}}
+        )
+        cancelled_orders, part_1_orders, part_2_orders = [orders[0], orders[1], orders[2]]
+
+        assert len(cancelled_orders) == 1
+        assert cancelled_orders[0] is limit_buy
+
+        assert len(part_1_orders) == 1
+        part_1_order = part_1_orders[0]
+        assert isinstance(part_1_order, trading_personal_data.SellMarketOrder)
+        assert part_1_order.created_last_price == decimal.Decimal("1000")
+        assert part_1_order.origin_quantity == decimal.Decimal("10")    # 10 BTC to sell into 10 000 USD
+        assert part_1_order.status == trading_enums.OrderStatus.FILLED
+
+        assert part_2_orders
+        part_2_order = part_2_orders[0]
+        assert isinstance(part_2_order, trading_personal_data.BuyMarketOrder)
+        assert part_2_order.created_last_price == decimal.Decimal("1000")
+        assert part_2_order.origin_quantity == decimal.Decimal("5.545")    # 50% of funds
+        assert part_2_order.status == trading_enums.OrderStatus.FILLED
+
+        # check portfolio is rebalanced
+        final_portfolio = exchange_manager.exchange_personal_data.portfolio_manager.portfolio.portfolio
+        assert final_portfolio["BTC"].available == decimal.Decimal('5.539455')  # 5.545 - fees
+        assert final_portfolio["USD"].available == decimal.Decimal("5545")
+
+
 async def _wait_for_orders_creation(orders_count=1):
     for _ in range(orders_count):
         await asyncio_tools.wait_asyncio_next_cycle()