[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/community/user_authentication.py

@@ -11,6 +11,7 @@ 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)
+    passphrase: 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.passphrase)

packages/flow/octobot_flow/jobs/automation_job.py

@@ -311,9 +311,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 +385,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: str | None = 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}")

packages/flow/tests/functionnal_tests/__init__.py

@@ -43,7 +43,7 @@ def set_emit_signals_metadata(automation_state: dict, emit_signals: bool) -> Non
 
 
 @contextlib.contextmanager
-def trading_signal_emission_patches(emit_signals: bool):
+def trading_signal_emission_patches(emit_signals: bool, *, mock_authenticator: bool = True):
     with contextlib.ExitStack() as stack:
         insert_mock = stack.enter_context(
             mock.patch.object(
@@ -52,7 +52,7 @@ def trading_signal_emission_patches(emit_signals: bool):
                 mock.AsyncMock(),
             )
         )
-        if emit_signals:
+        if emit_signals and mock_authenticator:
 
             @contextlib.asynccontextmanager
             async def _fake_maybe_authenticator(self):

packages/flow/tests/functionnal_tests/trading_modes_actions/simulator/test_grid_trading_mode_action.py

@@ -46,6 +46,9 @@
 _FIXED_BTC_USDC_CLOSE = 100000.0
 GRID_REFERENCE_LOWEST_BUY = _FIXED_BTC_USDC_CLOSE - (spread / 2) - increment * 2 + 12.12
 
+# Placeholder passphrase for community `UserAuthentication` (not a production secret; tests mock Starfish paths).
+SIMULATOR_COPY_GRID_FUNCTIONAL_TEST_COMMUNITY_PASSPHRASE = "simulator-copy-grid-functional-test"
+
 grid_pair_settings = [
     grid_trading.GridTradingMode.get_default_pair_config(
         "BTC/USDC",
@@ -726,6 +729,9 @@ async def test_simulator_copy_grid(
     reference_account until a fetched trading signal fills the DSL; otherwise the reference account
     is embedded in the action from the start.
     """
+    community_auth_details = octobot_flow.entities.UserAuthentication(
+        passphrase=SIMULATOR_COPY_GRID_FUNCTIONAL_TEST_COMMUNITY_PASSPHRASE,
+    )
     patched_fetch_tickers = tickers_repository_fetch_tickers_btc_usdc_close_override(
         lambda: _FIXED_BTC_USDC_CLOSE
     )
@@ -762,20 +768,9 @@ async def test_simulator_copy_grid(
                 fetch_trading_signals_mock,
             )
         )
-        if start_as_uninitialized_copy_keyword and not emit_signals:
-
-            @contextlib.asynccontextmanager
-            async def _fake_maybe_authenticator(self):
-                yield mock.MagicMock()
-
-            patch_stack.enter_context(
-                mock.patch.object(
-                    octobot_flow.AutomationJob,
-                    "_maybe_authenticator",
-                    _fake_maybe_authenticator,
-                )
-            )
-        insert_trading_signal_mock = patch_stack.enter_context(trading_signal_emission_patches(emit_signals))
+        insert_trading_signal_mock = patch_stack.enter_context(
+            trading_signal_emission_patches(emit_signals, mock_authenticator=False)
+        )
 
         reference_market = init_action["config"]["exchange_account_details"]["portfolio"]["unit"]
         if start_as_uninitialized_copy_keyword:
@@ -790,7 +785,9 @@ async def _fake_maybe_authenticator(self):
         set_emit_signals_metadata(automation_state, emit_signals)
 
         # 1. run init action
-        async with octobot_flow.AutomationJob(automation_state, [], [], {}) as automation_job:
+        async with octobot_flow.AutomationJob(
+            automation_state, [], [], community_auth_details
+        ) as automation_job:
             await automation_job.run()
         after_init_execution_dump = automation_job.dump()
 
@@ -808,7 +805,9 @@ async def _fake_maybe_authenticator(self):
                 assert action.previous_execution_result is None
 
         # 2. run copy exchange account action (rebalance + mirror reference grid orders)
-        async with octobot_flow.AutomationJob(after_init_execution_dump, [], [], {}) as automation_job:
+        async with octobot_flow.AutomationJob(
+            after_init_execution_dump, [], [], community_auth_details
+        ) as automation_job:
             await automation_job.run()
         after_initial_copy_execution_dump = automation_job.dump()
         assert len(automation_job.automation_state.automation.actions_dag.actions) == len(all_actions)
@@ -902,7 +901,9 @@ async def _fake_maybe_authenticator(self):
         assert d_order_price(sell_orders[1][price_col]) == lowest_buy_price + D_INCREMENT + D_SPREAD + D_INCREMENT
 
         # 3. trigger again: portfolio and mirrored grid should be unchanged
-        async with octobot_flow.AutomationJob(after_initial_copy_execution_dump, [], [], {}) as automation_job:
+        async with octobot_flow.AutomationJob(
+            after_initial_copy_execution_dump, [], [], community_auth_details
+        ) as automation_job:
             await automation_job.run()
         after_second_call_execution_dump = automation_job.dump()
         assert len(automation_job.automation_state.automation.actions_dag.actions) == len(all_actions)