[Trading] Add ProfileCopyTradingMode

Signed-off-by: Herklos <[email protected]>

Author: Herklos

Services/Services_bases/exchange_service/__init__.py

@@ -0,0 +1 @@
+from .exchange import ExchangeService

Services/Services_bases/exchange_service/exchange.py

@@ -0,0 +1,85 @@
+#  Drakkar-Software OctoBot-Tentacles
+#  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_services.constants as services_constants
+import octobot_services.services as services
+import octobot_services.errors as errors
+
+import octobot_commons.constants as commons_constants
+
+import octobot.constants as constants
+
+
+class ExchangeService(services.AbstractService):
+    def get_fields_description(self) -> typing.Dict[str, str]:
+        return {
+            services_constants.CONFIG_EXCHANGE_INTERNAL_NAMES: "Your exchange names",
+            services_constants.CONFIG_EXCHANGE_PROFILE_IDS: "Your exchange profile ids",
+        }
+
+    def get_default_value(self) -> typing.Dict[str, typing.Any]:
+        return {
+            services_constants.CONFIG_EXCHANGE_INTERNAL_NAMES: [],
+            services_constants.CONFIG_EXCHANGE_PROFILE_IDS: [],
+        }
+
+    @staticmethod
+    def is_setup_correctly(config) -> bool:
+        return True
+
+    def check_required_config(self, config) -> bool:
+        try:
+            config_key = config[services_constants.CONFIG_EXCHANGE_INTERNAL_NAMES]
+            return bool(config_key) and config_key not in commons_constants.DEFAULT_CONFIG_VALUES
+        except KeyError:
+            return False
+
+    def has_required_configuration(self) -> bool:
+        try:
+            return self.check_required_config(
+                self.config[services_constants.CONFIG_CATEGORY_SERVICES].get(services_constants.CONFIG_EXCHANGE, {})
+            )
+        except KeyError:
+            return False
+
+    def get_required_config(self):
+        return [services_constants.CONFIG_EXCHANGE_INTERNAL_NAMES]
+
+    def get_internal_names(self) -> typing.List[str]:
+        try:
+            return self.config[services_constants.CONFIG_EXCHANGE_INTERNAL_NAMES]
+        except KeyError:
+            return []
+
+    def get_profile_ids(self) -> typing.List[str]:
+        try:
+            return self.config[services_constants.CONFIG_EXCHANGE_PROFILE_IDS]
+        except KeyError:
+            return []
+
+    @classmethod
+    def get_help_page(cls) -> str:
+        return f"{constants.OCTOBOT_DOCS_URL}/octobot-interfaces/exchange"
+
+    def get_type(self) -> str:
+        return services_constants.CONFIG_EXCHANGE
+
+    def get_successful_startup_message(self) -> str:
+        return f"{', '.join(self._get_internal_names())} exchange service configured and ready."
+
+    async def stop(self):
+        pass

Services/Services_bases/exchange_service/metadata.json

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

Services/Services_feeds/exchange_service_feed/__init__.py

@@ -0,0 +1,2 @@
+from .exchange_feed import ExchangeServiceFeed
+from .exchange_feed import ExchangeProfile

Services/Services_feeds/exchange_service_feed/exchange_feed.py

@@ -0,0 +1,127 @@
+#  Drakkar-Software OctoBot-Tentacles
+#  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 asyncio
+import time
+import typing
+import copy
+
+import octobot_services.channel as services_channel
+import octobot_services.constants as services_constants
+import octobot_services.service_feeds as service_feeds
+
+import octobot_trading.api as trading_api
+import octobot_trading.personal_data as personal_data
+
+import tentacles.Services.Services_bases as Services_bases
+
+
+class ExchangeServiceFeedChannel(services_channel.AbstractServiceFeedChannel):
+    pass
+
+
+class ExchangeProfile:
+    def __init__(self, exchange_name: str, profile_id: str, refresh_time: float):
+        self.exchange_name: str = exchange_name
+        self.profile_id: str = profile_id
+        self.refresh_time: float = refresh_time
+        self.next_refresh: float = time.time()
+
+        self.positions: typing.List[personal_data.Position] = []
+        self.closed_positions: typing.List[personal_data.Position] = []
+        self.trades: typing.List[personal_data.Trade] = []
+        self.portfolio: typing.Optional[personal_data.Portfolio] = None
+        self.orders: typing.List[personal_data.Order] = []
+
+    def __str__(self):
+        return f"{self.exchange_name} {self.profile_id}"
+
+
+class ExchangeServiceFeed(service_feeds.AbstractServiceFeed):
+    FEED_CHANNEL = ExchangeServiceFeedChannel
+    REQUIRED_SERVICES = [Services_bases.ExchangeService]
+
+    def __init__(self, config, main_async_loop, bot_id: str):
+        super().__init__(config, main_async_loop, bot_id)
+        self.exchange_profiles: typing.List[ExchangeProfile] = []
+
+    # merge new config into existing config
+    def update_feed_config(self, config):
+        self.exchange_profiles.extend(profile
+                                  for profile in config[services_constants.CONFIG_EXCHANGE_PROFILE_IDS]
+                                  if profile not in self.exchange_profiles)
+
+    def _get_profile_exchange_manager(self, profile: ExchangeProfile):
+        return trading_api.get_exchange_manager_from_exchange_name(profile.exchange_name)
+
+    def _something_to_watch(self):
+        # TODO later: check if exchange supports profiles fetching before considering profiles as to be watched
+        return bool(self.exchange_profiles)
+
+    def _get_sleep_time_before_next_wakeup(self):
+        closest_wakeup = min(profile.next_refresh for profile in self.exchange_profiles)
+        return max(0, closest_wakeup - time.time())
+
+    async def _get_exchange_profile(self, profile: ExchangeProfile) -> bool:
+        updated = False
+        self.logger.debug(f"Fetching exchange profile on {profile.exchange_name} {profile.profile_id}")
+        exchange_manager = self._get_profile_exchange_manager(profile)
+        profile.next_refresh = time.time() + profile.refresh_time
+        exchange_has_positions = True
+        if exchange_has_positions:
+            previous_positions = copy.deepcopy(profile.positions)
+            profile.positions = await exchange_manager.exchange.get_user_positions(profile.profile_id)
+            if previous_positions != profile.positions:
+                updated = True
+        else:
+            # Update portfolio
+            pass
+
+        # TODO: fetch orders ?
+        # TODO: fetch trades ?
+        return updated
+
+    async def _push_update_and_wait_exchange_profiles(self):
+        for profile in self.exchange_profiles:
+            if time.time() >= profile.next_refresh:
+                profile_updated = await self._get_exchange_profile(profile)
+                if profile_updated:
+                    await self._async_notify_consumers(
+                        {
+                            services_constants.FEED_METADATA: profile
+                        }
+                    )
+
+    async def _push_update_and_wait(self):
+        if self.exchange_profiles:
+            await self._push_update_and_wait_exchange_profiles()
+        await asyncio.sleep(self._get_sleep_time_before_next_wakeup())
+
+    async def _update_loop(self):
+        while not self.should_stop:
+            try:
+                await self._push_update_and_wait()
+            except Exception as e:
+                self.logger.exception(e, True, f"Error when receiving exchange feed: ({e})")
+                self.should_stop = True
+        return False
+
+    async def _start_service_feed(self):
+        try:
+            asyncio.create_task(self._update_loop())
+        except Exception as e:
+            self.logger.exception(e, True, f"Error when initializing exchange feed: {e}")
+            return False
+        return True

Services/Services_feeds/exchange_service_feed/metadata.json

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