[Trading] Add ProfileCopyTradingMode

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

Author: Herklos

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,153 @@
+#  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 octobot_trading.exchanges as exchanges
+
+class ExchangeServiceFeedChannel(services_channel.AbstractServiceFeedChannel):
+    pass
+
+
+class ExchangeProfile:
+    DEFAULT_EXCHANGE_PROFILE_REFRESH_TIME = 60
+
+    def __init__(self, profile_id: str, exchange_name: str, refresh_time: float = DEFAULT_EXCHANGE_PROFILE_REFRESH_TIME, next_refresh: typing.Optional[float] = None):
+        self.profile_id: str = profile_id
+        self.exchange_name: str = exchange_name
+        self.refresh_time: float = refresh_time
+        self.next_refresh: float = next_refresh if next_refresh is not None else time.time() + refresh_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 update_next_refresh(self):
+        self.next_refresh = time.time() + self.refresh_time
+
+    def __str__(self):
+        return f"{self.exchange_name} {self.profile_id}"
+
+
+class ExchangeServiceFeed(service_feeds.AbstractServiceFeed):
+    FEED_CHANNEL = ExchangeServiceFeedChannel
+    REQUIRED_SERVICES = False
+
+    DEFAULT_REFRESH_TIME = 10
+
+    def __init__(self, config, main_async_loop, bot_id: str):
+        super().__init__(config, main_async_loop, bot_id)
+        self.exchange_profiles: typing.Dict[str, ExchangeProfile] = {}
+
+    def _initialize(self):
+        pass
+
+    # merge new config into existing config
+    def update_feed_config(self, config):
+        for profile in config[services_constants.CONFIG_EXCHANGE_PROFILES]:
+            if profile[services_constants.CONFIG_EXCHANGE_PROFILE_ID] not in self.exchange_profiles.keys():
+                self.exchange_profiles[profile[services_constants.CONFIG_EXCHANGE_PROFILE_ID]] = ExchangeProfile(
+                    profile_id=profile[services_constants.CONFIG_EXCHANGE_PROFILE_ID],
+                    exchange_name=profile[services_constants.CONFIG_EXCHANGE_EXCHANGE_NAME]
+                )
+            else:
+                self.exchange_profiles[profile[services_constants.CONFIG_EXCHANGE_PROFILE_ID]].update(profile)
+
+    def _get_profile_exchange_manager(self, exchange_name: str) -> typing.Optional[exchanges.ExchangeManager]:
+        exchange_configurations = list(trading_api.get_exchange_configurations_from_exchange_name(exchange_name).values())
+        if exchange_configurations:
+            return exchange_configurations[0].exchange_manager # TODO: is there a better way to get the exchange manager ?
+        return None
+
+    def _something_to_watch(self):
+        service_config = self.config[services_constants.CONFIG_CATEGORY_SERVICES][services_constants.CONFIG_EXCHANGE]
+        # update feed config before checking if there is something to watch
+        self.update_feed_config(service_config)
+        # TODO later: check if exchange supports profiles fetching before considering profiles as to be watched
+        return bool(self.exchange_profiles.keys())
+
+    def _get_sleep_time_before_next_wakeup(self):
+        if not self.exchange_profiles.keys():
+            return self.DEFAULT_REFRESH_TIME
+        closest_wakeup = min(profile.next_refresh for profile in self.exchange_profiles.values())
+        return max(0, closest_wakeup - time.time())
+
+    async def _fetch_exchange_profile(self, profile_id: str) -> bool:
+        updated = False
+        current_profile = self.exchange_profiles.get(profile_id, None)
+        if current_profile is None:
+            self.logger.error(f"Exchange profile {profile_id} not found")
+            return False
+
+        self.logger.debug(f"Fetching exchange profile on {current_profile.exchange_name} {current_profile.profile_id}")
+        exchange_manager = self._get_profile_exchange_manager(current_profile.exchange_name)
+        current_profile.update_next_refresh()
+        exchange_has_positions = True
+        if exchange_has_positions:
+            previous_positions = copy.deepcopy(current_profile.positions) if current_profile is not None else []
+            current_profile.positions = await exchange_manager.exchange.get_user_positions(current_profile.profile_id)
+            if previous_positions != current_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.values():
+            if time.time() >= profile.next_refresh:
+                profile_updated = await self._fetch_exchange_profile(profile.profile_id)
+                if profile_updated:
+                    await self._async_notify_consumers(
+                        {
+                            services_constants.FEED_METADATA: profile
+                        }
+                    )
+
+    async def _push_update_and_wait(self):
+        if self.exchange_profiles.keys():
+            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"]
+}