[Flow] add signals pull & push

Author: GuillaumeDSM

octobot/community/authentication.py

@@ -656,6 +656,12 @@ def decrypt_node_wallet(self, passphrase: str):
 
     def verify_node_passphrase(self, passphrase: str) -> bool:
         return self._wallet_backend.verify_node_passphrase(passphrase)
+        
+    @property
+    def sync_client(self) -> sync_client.StarfishClient:
+        if self._sync_client is None:
+            raise errors.SyncClientNotInitializedError("Sync client is not initialized")
+        return self._sync_client
 
     def init_sync_client(self):
         if self._sync_client is not None:

octobot/community/errors.py

@@ -32,6 +32,10 @@ class JWTExpiredError(commons_authentication.AuthenticationError):
     pass
 
 
+class SyncClientNotInitializedError(commons_authentication.AuthenticationError):
+    pass
+
+
 class BotError(commons_authentication.UnavailableError):
     pass
 

octobot/community/local_authenticator.py

@@ -16,39 +16,45 @@ def get_stateless_configuration() -> octobot_commons.configuration.Configuration
 
 @contextlib.asynccontextmanager
 async def local_user_authenticator(
-    email: str,
-    hidden: bool,
+    email: typing.Optional[str] = None,
+    hidden: typing.Optional[bool] = None,
     backend_url: typing.Optional[str] = None,
     password: typing.Optional[str] = None,
     auth_key: typing.Optional[str] = None,
 ) -> typing.AsyncGenerator["community.CommunityAuthentication", None]:
-    if not email:
-        raise ValueError("email is required")
     community.IdentifiersProvider.use_production()
     local_instance = None
     configuration = get_stateless_configuration()
+    authenticate = password or auth_key
+    if authenticate and not email:
+        raise ValueError("email is required when authenticating with password or auth_key")
     try:
         local_instance = community.CommunityAuthentication(
             config=configuration, backend_url=backend_url, use_as_singleton=False
         )
         local_instance.supabase_client.is_admin = False
-        local_instance.silent_auth = hidden
+        local_instance.silent_auth = False if hidden is None else hidden
         if auth_key:
             password_value = None
             auth_key_value = auth_key
         else:
             password_value = password
             auth_key_value = None
-        await local_instance.login(
-            email, password_value, password_token=None, auth_key=auth_key_value, minimal=True
-        )
-        common_logging.get_logger("local_community_user_authenticator").info(
-            f"Authenticated as {email[:3]}[...]{email[-4:]}"
-        )
+        if authenticate:
+            email = typing.cast(str, email) # email is always str here
+            await local_instance.login(
+                email, password_value, password_token=None, auth_key=auth_key_value, minimal=True
+            )
+            auth_logger = common_logging.get_logger("local_community_user_authenticator")
+            if len(email) > 7:
+                auth_logger.info(f"Authenticated as {email[:3]}[...]{email[-4:]}")
+            else:
+                auth_logger.info("Authenticated as local community user")
         yield local_instance
     finally:
         if local_instance is not None:
-            await local_instance.logout()
+            if authenticate:
+                await local_instance.logout()
             await local_instance.stop()
 
 

packages/flow/octobot_flow/entities/__init__.py

@@ -33,6 +33,7 @@
 from octobot_flow.entities.community import (
     UserAuthentication,
     TradingSignal,
+    register_passphrase,
 )
 __all__ = [
     "AccountElements",
@@ -63,4 +64,5 @@
     "AdditionalActions",
     "UserAuthentication",
     "TradingSignal",
+    "register_passphrase",
 ]

packages/flow/octobot_flow/entities/community/__init__.py

@@ -1,7 +1,11 @@
-from octobot_flow.entities.community.user_authentication import UserAuthentication
+from octobot_flow.entities.community.user_authentication import (
+    UserAuthentication,
+    register_passphrase,
+)
 from octobot_flow.entities.community.trading_signal import TradingSignal
 
 __all__ = [
     "UserAuthentication",
     "TradingSignal",
+    "register_passphrase",
 ]

packages/flow/octobot_flow/entities/community/user_authentication.py

@@ -3,6 +3,13 @@
 import octobot_commons.dataclasses
 
 
+_PASSPHRASES_BY_PUBLIC_KEY: dict[str, str] = {}
+
+
+def register_passphrase(public_key: str, passphrase: str) -> None:
+    _PASSPHRASES_BY_PUBLIC_KEY[public_key] = passphrase
+
+
 @dataclasses.dataclass
 class UserAuthentication(octobot_commons.dataclasses.FlexibleDataclass):
     email: typing.Optional[str] = None
@@ -11,6 +18,13 @@ class UserAuthentication(octobot_commons.dataclasses.FlexibleDataclass):
     user_id: typing.Optional[str] = None
     auth_key: typing.Optional[str] = None
     encrypted_keys_by_exchange: dict[str, str] = dataclasses.field(default_factory=dict)
+    public_key: typing.Optional[str] = None
 
     def has_auth_details(self) -> bool:
-        return bool(self.password or self.auth_key)
+        return bool(self.password or self.auth_key or self.public_key)
+
+    @property
+    def passphrase(self) -> typing.Optional[str]:
+        if self.public_key is None:
+            return None
+        return _PASSPHRASES_BY_PUBLIC_KEY.get(self.public_key)

packages/flow/octobot_flow/jobs/automation_job.py

@@ -64,7 +64,10 @@ async def run(self) -> list[octobot_flow.entities.AbstractActionDetails]:
         executed_actions = []
         async with self._maybe_authenticator() as maybe_authenticator:
             maybe_community_repository = (
-                octobot_flow.repositories.community.CommunityRepository(maybe_authenticator)
+                octobot_flow.repositories.community.CommunityRepository(
+                    maybe_authenticator,
+                    passphrase=self.auth_details.passphrase,
+                )
                 if maybe_authenticator else None
             )
             with octobot_flow.encryption.decrypted_bots_configurations(self.automation_state):
@@ -311,9 +314,7 @@ async def _init_all_required_copy_trading_data(
                 raise octobot_flow.errors.CommunityTradingSignalError(
                     "Community authentication is required to fetch copy trading signals"
                 )
-            trading_signals_repository = octobot_flow.repositories.community.TradingSignalsRepository(
-                maybe_community_repository.authenticator
-            )
+            trading_signals_repository = octobot_flow.repositories.community.TradingSignalsRepository.from_community_repository(maybe_community_repository)
             self._logger.info(f"Fetching copy trading signals for {to_fetch_signals} strategies")
             trading_signals = await trading_signals_repository.fetch_trading_signals(
                 to_fetch_signals,
@@ -387,8 +388,8 @@ async def _emit_trading_signals(
             fetched_dependencies.fetched_exchange_data,
             reference_market
         )
-        trading_signals_repository = octobot_flow.repositories.community.TradingSignalsRepository(
-            maybe_community_repository.authenticator
+        trading_signals_repository = octobot_flow.repositories.community.TradingSignalsRepository.from_community_repository(
+            maybe_community_repository
         )
         await trading_signals_repository.insert_trading_signal(
             octobot_flow.entities.TradingSignal(

packages/flow/octobot_flow/repositories/community/authenticator_factory.py

@@ -22,8 +22,6 @@ def enable_community_authentication(self) -> bool:
 
     @contextlib.asynccontextmanager
     async def local_authenticator(self) -> typing.AsyncGenerator[community.CommunityAuthentication, None]:
-        if not self.auth_details.email:
-            raise ValueError("auth_details.email is required")
         async with local_community_auth.local_user_authenticator(
             email=self.auth_details.email,
             hidden=self.auth_details.hidden,

packages/flow/octobot_flow/repositories/community/community_repository.py

@@ -1,14 +1,25 @@
 import contextlib
 import asyncio
+import typing
+import starfish_sdk
+
+import octobot_commons.logging
 
 import octobot.community
+import octobot.community.errors as community_errors
 
 import octobot_flow.entities
+import octobot_flow.errors
 
 
 class CommunityRepository:
-    def __init__(self, authenticator: octobot.community.CommunityAuthentication):
+    def __init__(self, authenticator: octobot.community.CommunityAuthentication, passphrase: typing.Optional[str] = None):
         self.authenticator: octobot.community.CommunityAuthentication = authenticator
+        self.passphrase: typing.Optional[str] = passphrase
+
+    @classmethod
+    def from_community_repository(cls, other_repository: "CommunityRepository") -> typing.Self:
+        return cls(other_repository.authenticator, other_repository.passphrase)
 
     async def insert_bot_logs(self, log_data: list[octobot.community.BotLogData]):
         await asyncio.gather(
@@ -30,3 +41,15 @@ def automation_context(self, automation: octobot_flow.entities.AutomationDetails
             yield
         finally:
             self.authenticator.user_account.bot_id = previous_bot_id # type: ignore
+
+    def _get_sync_client(self) -> starfish_sdk.StarfishClient:
+        try:
+            return self.authenticator.sync_client
+        except community_errors.SyncClientNotInitializedError as e:
+            raise octobot_flow.errors.CommunityTradingSignalError(
+                "Starfish sync client is unavailable (no sync server configured or wallet passphrase required)"
+            ) from e
+
+    @classmethod
+    def _logger(cls) -> octobot_commons.logging.BotLogger:
+        return octobot_commons.logging.get_logger(cls.__name__)

packages/flow/octobot_flow/repositories/community/trading_signals_repository.py

@@ -1,15 +1,86 @@
-import octobot.community
+import starfish_sdk
+
 
 import octobot_flow.entities
 import octobot_flow.repositories.community.trading_signals_channel as trading_signals_channel
+import octobot_flow.repositories.community.community_repository as community_repository
+
+
+FETCH_SIGNALS_SEGMENT = "latest"
+
+
+def _sync_signals_path(sync_kind: str, strategy_id: str, segment: str) -> str:
+    return f"/v1/{sync_kind}/strategies/{strategy_id}/signals/{segment}"
+
+
+def _signals_pull_and_push_paths(strategy_id: str, segment: str) -> tuple[str, str]:
+    return (
+        _sync_signals_path("pull", strategy_id, segment),
+        _sync_signals_path("push", strategy_id, segment),
+    )
 
 
-class TradingSignalsRepository:
-    def __init__(self, authenticator: octobot.community.CommunityAuthentication):
-        self.authenticator: octobot.community.CommunityAuthentication = authenticator
+def _signals_paths_for_upload(strategy_id: str, account_updated_at: float) -> tuple[str, str]:
+    return _signals_pull_and_push_paths(strategy_id, str(account_updated_at))
 
+
+def _pull_path_for_fetch(strategy_id: str) -> str:
+    return _sync_signals_path("pull", strategy_id, FETCH_SIGNALS_SEGMENT)
+
+
+def _trim_historical_snapshots_if_needed(
+    trading_signal: octobot_flow.entities.TradingSignal,
+    history_size: int,
+) -> None:
+    account = trading_signal.account
+    if not account.historical_snapshots or len(account.historical_snapshots) <= history_size:
+        return
+    account.historical_snapshots = account.historical_snapshots[:history_size]
+
+
+class TradingSignalsRepository(community_repository.CommunityRepository):
     async def insert_trading_signal(self, trading_signal: octobot_flow.entities.TradingSignal):
         await trading_signals_channel.send_internal_trading_signal(trading_signal)
+        await self._upload_trading_signal(trading_signal)
+
+    async def fetch_trading_signals(
+        self,
+        strategy_ids: list[str],
+        history_size: int,
+    ) -> list[octobot_flow.entities.TradingSignal]:
+        client = self._get_sync_client()
+        trading_signals: list[octobot_flow.entities.TradingSignal] = []
+        for strategy_identifier in strategy_ids:
+            try:
+                pull_path = _pull_path_for_fetch(strategy_identifier)
+                pull_result = await client.pull(pull_path, checkpoint=None)
+                trading_signal = octobot_flow.entities.TradingSignal.from_dict(pull_result.data)
+                _trim_historical_snapshots_if_needed(trading_signal, history_size)
+                trading_signals.append(trading_signal)
+            except Exception as strategy_error:
+                self._logger().exception(
+                    strategy_error,
+                    True,
+                    f"Failed to fetch trading signals for strategy {strategy_identifier!r}: {strategy_error}",
+                )
+        return trading_signals
 
-    async def fetch_trading_signals(self, strategy_ids: list[str], history_size: int) -> list[octobot_flow.entities.TradingSignal]:
-        raise NotImplementedError("TODO: fetch_trading_signals")
+    async def _upload_trading_signal(
+        self,
+        trading_signal: octobot_flow.entities.TradingSignal,
+    ):
+        pull_path, push_path = _signals_paths_for_upload(
+            trading_signal.strategy_id,
+            trading_signal.account.updated_at,
+        )
+        manager = starfish_sdk.SyncManager(
+            client=self._get_sync_client(),
+            pull_path=pull_path,
+            push_path=push_path,
+            sign_data=self.authenticator._sync_data_signer,
+        )
+        payload = trading_signal.to_dict()
+        try:
+            await manager.push(payload)
+        except Exception as upload_error:
+            self._logger().exception(upload_error, True, f"Failed to upload trading signal: {upload_error}")