Build 060.22: implement Acquisition Runtime Service

2026-07-27 07:48:57 +03:00
parent 433ac5d375
commit 9953bab993
7 changed files with 3233 additions and 4 deletions
--- a/app/src/market_data/acquisition/runtime/acquisition_runtime_service.py
+++ b/app/src/market_data/acquisition/runtime/acquisition_runtime_service.py
@@ -0,0 +1,136 @@
+# app/src/market_data/acquisition/runtime/acquisition_runtime_service.py
+
+from __future__ import annotations
+
+"""
+Сервис исполнения команд Acquisition Runtime.
+
+Build 060.22 вводит первую производственную реализацию
+сервисного слоя Runtime подсистемы Market Data Acquisition.
+
+Сервис принимает типизированные Runtime-команды и делегирует
+их выполнение соответствующим инфраструктурным зависимостям.
+
+Сервис не содержит знаний о Trades Feed, Consistency или Recovery.
+"""
+
+from src.market_data.acquisition.runtime.acquisition_runtime_service_protocol import (
+    AcquisitionRuntimeServiceProtocol,
+)
+from src.market_data.acquisition.runtime.runtime_commands import (
+    ConnectCommand,
+    DisconnectCommand,
+    SendBinaryCommand,
+    SendTextCommand,
+    SubscribeCommand,
+    UnsubscribeCommand,
+)
+from src.market_data.acquisition.runtime.websocket_protocol import (
+    AcquisitionRuntimeCommand,
+    AcquisitionRuntimeCommandDispatcherProtocol,
+    AcquisitionRuntimeEventPublisherProtocol,
+    WebSocketSessionProtocol,
+    WebSocketSubscriptionManagerProtocol,
+    WebSocketTransportProtocol,
+)
+
+
+class AcquisitionRuntimeService(
+    AcquisitionRuntimeServiceProtocol,
+    AcquisitionRuntimeCommandDispatcherProtocol,
+):
+    """
+    Сервис маршрутизации Acquisition Runtime Commands.
+
+    Каждая команда делегируется ровно одной инфраструктурной
+    зависимости.
+
+    Сервис не управляет reconnect, heartbeat, scheduler,
+    Recovery или обработкой рыночных данных.
+    """
+
+    def __init__(
+        self,
+        session: WebSocketSessionProtocol,
+        transport: WebSocketTransportProtocol,
+        subscription_manager: WebSocketSubscriptionManagerProtocol,
+        event_publisher: AcquisitionRuntimeEventPublisherProtocol,
+    ) -> None:
+        """
+        Создать Acquisition Runtime Service.
+
+        Args:
+            session:
+                Компонент жизненного цикла WebSocket-сессии.
+
+            transport:
+                Низкоуровневый WebSocket-транспорт.
+
+            subscription_manager:
+                Компонент управления активными подписками.
+
+            event_publisher:
+                Компонент публикации инфраструктурных Runtime-событий.
+
+                В Build 060.22 зависимость сохраняется как часть
+                утверждённой архитектуры, но публикация событий
+                будет подключена в последующих Build.
+        """
+        self._session = session
+        self._transport = transport
+        self._subscription_manager = subscription_manager
+        self._event_publisher = event_publisher
+
+    async def dispatch(
+        self,
+        command: AcquisitionRuntimeCommand,
+    ) -> None:
+        """
+        Выполнить одну инфраструктурную команду Acquisition Runtime.
+
+        Args:
+            command:
+                Типизированная команда Runtime Layer.
+
+        Raises:
+            TypeError:
+                Если передан неподдерживаемый тип команды.
+
+            Exception:
+                Любое исключение инфраструктурной зависимости
+                распространяется вызывающему компоненту без изменения.
+        """
+        if isinstance(command, ConnectCommand):
+            await self._session.start()
+            return
+
+        if isinstance(command, DisconnectCommand):
+            await self._session.stop()
+            return
+
+        if isinstance(command, SubscribeCommand):
+            await self._subscription_manager.subscribe(
+                command.subscription_key,
+                command.message,
+            )
+            return
+
+        if isinstance(command, UnsubscribeCommand):
+            await self._subscription_manager.unsubscribe(
+                command.subscription_key,
+                command.message,
+            )
+            return
+
+        if isinstance(command, SendTextCommand):
+            await self._transport.send(command.message.payload)
+            return
+
+        if isinstance(command, SendBinaryCommand):
+            await self._transport.send(command.message.payload)
+            return
+
+        raise TypeError(
+            "Неподдерживаемый тип Acquisition Runtime Command: "
+            f"{type(command).__name__}."
+        )
--- a/app/src/market_data/acquisition/runtime/acquisition_runtime_service_protocol.py
+++ b/app/src/market_data/acquisition/runtime/acquisition_runtime_service_protocol.py
@@ -0,0 +1,44 @@
+# app/src/market_data/acquisition/runtime/acquisition_runtime_service_protocol.py
+
+from __future__ import annotations
+
+"""
+Публичный контракт Acquisition Runtime Service.
+
+Build 060.22 вводит сервисный уровень исполнения типизированных
+команд Runtime подсистемы Market Data Acquisition.
+
+Контракт не определяет устройство WebSocket Session, Transport,
+Subscription Manager или механизма публикации событий.
+"""
+
+from typing import Protocol, runtime_checkable
+
+from src.market_data.acquisition.runtime.websocket_protocol import (
+    AcquisitionRuntimeCommand,
+)
+
+
+@runtime_checkable
+class AcquisitionRuntimeServiceProtocol(Protocol):
+    """
+    Контракт сервиса исполнения Acquisition Runtime Commands.
+
+    Реализация принимает одну типизированную команду и передаёт её
+    соответствующей инфраструктурной зависимости.
+
+    Сервис не содержит знаний о Trades Feed, Consistency и Recovery.
+    """
+
+    async def dispatch(
+        self,
+        command: AcquisitionRuntimeCommand,
+    ) -> None:
+        """
+        Выполнить одну инфраструктурную команду Acquisition Runtime.
+
+        Args:
+            command:
+                Типизированная команда Runtime Layer.
+        """
+        ...
--- a/app/src/market_data/acquisition/runtime/websocket_protocol.py
+++ b/app/src/market_data/acquisition/runtime/websocket_protocol.py
@@ -23,6 +23,10 @@ from src.market_data.acquisition.runtime.runtime_events import (
    ReconnectFailedEvent,
    ReconnectStartedEvent,
 )
+from src.market_data.acquisition.runtime.transport_messages import (
+    TransportBinaryMessage,
+    TransportTextMessage,
+)


 AcquisitionRuntimeCommand = (
@@ -46,6 +50,11 @@ AcquisitionRuntimeEvent = (
    | HeartbeatTimeoutEvent
 )

+AcquisitionSubscriptionMessage = (
+    TransportTextMessage
+    | TransportBinaryMessage
+)
+

@runtime_checkable
 class WebSocketTransportProtocol(Protocol):
@@ -108,10 +117,50 @@ class WebSocketSubscriptionManagerProtocol(Protocol):
    """
    Контракт управления активными WebSocket-подписками.

-    Конкретные модели подписок и формат сообщений будут добавлены
-    в последующих Build'ах.
+    Subscription Manager отвечает за регистрацию и удаление
+    активных подписок, а также за восстановление зарегистрированных
+    подписок после повторного подключения.
+
+    Менеджер работает только с универсальными транспортными
+    сообщениями и не содержит exchange-specific логики.
    """

+    async def subscribe(
+        self,
+        subscription_key: str,
+        message: AcquisitionSubscriptionMessage,
+    ) -> None:
+        """
+        Зарегистрировать и выполнить WebSocket-подписку.
+
+        Args:
+            subscription_key:
+                Уникальный инфраструктурный ключ подписки.
+
+            message:
+                Полностью сформированное транспортное сообщение
+                подписки.
+        """
+        ...
+
+    async def unsubscribe(
+        self,
+        subscription_key: str,
+        message: AcquisitionSubscriptionMessage,
+    ) -> None:
+        """
+        Отменить и удалить существующую WebSocket-подписку.
+
+        Args:
+            subscription_key:
+                Уникальный инфраструктурный ключ подписки.
+
+            message:
+                Полностью сформированное транспортное сообщение
+                отмены подписки.
+        """
+        ...
+
    async def restore_subscriptions(self) -> None:
        """Восстановить активные подписки после подключения."""
        ...
@@ -162,4 +211,4 @@ class AcquisitionRuntimeEventPublisherProtocol(Protocol):
        """
        Опубликовать одно инфраструктурное событие Runtime.
        """
-        ...
+        ...
--- a/app/tests/unit/market_data/acquisition/runtime/test_acquisition_runtime_service.py
+++ b/app/tests/unit/market_data/acquisition/runtime/test_acquisition_runtime_service.py
@@ -0,0 +1,323 @@
+# app/tests/unit/market_data/acquisition/runtime/test_acquisition_runtime_service.py
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from src.market_data.acquisition.runtime.acquisition_runtime_service import (
+    AcquisitionRuntimeService,
+)
+from src.market_data.acquisition.runtime.acquisition_runtime_service_protocol import (
+    AcquisitionRuntimeServiceProtocol,
+)
+from src.market_data.acquisition.runtime.runtime_commands import (
+    ConnectCommand,
+    DisconnectCommand,
+    SendBinaryCommand,
+    SendTextCommand,
+    SubscribeCommand,
+    UnsubscribeCommand,
+)
+from src.market_data.acquisition.runtime.transport_messages import (
+    TransportBinaryMessage,
+    TransportTextMessage,
+)
+from src.market_data.acquisition.runtime.websocket_protocol import (
+    AcquisitionRuntimeCommandDispatcherProtocol,
+    AcquisitionRuntimeEvent,
+    AcquisitionSubscriptionMessage,
+)
+
+
+class FakeSession:
+    def __init__(self) -> None:
+        self.started = 0
+        self.stopped = 0
+
+    @property
+    def is_connected(self) -> bool:
+        return False
+
+    async def start(self) -> None:
+        self.started += 1
+
+    async def stop(self) -> None:
+        self.stopped += 1
+
+
+class FakeTransport:
+    def __init__(self) -> None:
+        self.messages: list[str | bytes] = []
+
+    async def connect(self) -> None:
+        return None
+
+    async def disconnect(self) -> None:
+        return None
+
+    async def send(
+        self,
+        message: str | bytes,
+    ) -> None:
+        self.messages.append(message)
+
+    async def receive(self) -> str | bytes:
+        raise AssertionError("receive() should not be called.")
+
+
+class FakeSubscriptionManager:
+    def __init__(self) -> None:
+        self.subscriptions: list[
+            tuple[str, AcquisitionSubscriptionMessage]
+        ] = []
+        self.unsubscriptions: list[
+            tuple[str, AcquisitionSubscriptionMessage]
+        ] = []
+
+    async def subscribe(
+        self,
+        subscription_key: str,
+        message: AcquisitionSubscriptionMessage,
+    ) -> None:
+        self.subscriptions.append(
+            (
+                subscription_key,
+                message,
+            )
+        )
+
+    async def unsubscribe(
+        self,
+        subscription_key: str,
+        message: AcquisitionSubscriptionMessage,
+    ) -> None:
+        self.unsubscriptions.append(
+            (
+                subscription_key,
+                message,
+            )
+        )
+
+    async def restore_subscriptions(self) -> None:
+        return None
+
+    async def clear_subscriptions(self) -> None:
+        return None
+
+
+class FakeEventPublisher:
+    def __init__(self) -> None:
+        self.events: list[AcquisitionRuntimeEvent] = []
+
+    async def publish(
+        self,
+        event: AcquisitionRuntimeEvent,
+    ) -> None:
+        self.events.append(event)
+
+
+def create_service() -> tuple[
+    AcquisitionRuntimeService,
+    FakeSession,
+    FakeTransport,
+    FakeSubscriptionManager,
+    FakeEventPublisher,
+]:
+    session = FakeSession()
+    transport = FakeTransport()
+    subscriptions = FakeSubscriptionManager()
+    publisher = FakeEventPublisher()
+
+    service = AcquisitionRuntimeService(
+        session=session,
+        transport=transport,
+        subscription_manager=subscriptions,
+        event_publisher=publisher,
+    )
+
+    return (
+        service,
+        session,
+        transport,
+        subscriptions,
+        publisher,
+    )
+
+
+def test_service_implements_protocol() -> None:
+    service, *_ = create_service()
+
+    assert isinstance(
+        service,
+        AcquisitionRuntimeServiceProtocol,
+    )
+
+
+def test_service_is_command_dispatcher() -> None:
+    service, *_ = create_service()
+
+    assert isinstance(
+        service,
+        AcquisitionRuntimeCommandDispatcherProtocol,
+    )
+
+
+def test_dispatch_connect_command() -> None:
+    service, session, *_ = create_service()
+
+    asyncio.run(
+        service.dispatch(
+            ConnectCommand(),
+        )
+    )
+
+    assert session.started == 1
+    assert session.stopped == 0
+
+
+def test_dispatch_disconnect_command() -> None:
+    service, session, *_ = create_service()
+
+    asyncio.run(
+        service.dispatch(
+            DisconnectCommand(),
+        )
+    )
+
+    assert session.started == 0
+    assert session.stopped == 1
+
+
+def test_dispatch_subscribe_command() -> None:
+    service, _, _, subscriptions, _ = create_service()
+
+    message = TransportTextMessage(
+        payload='{"subscribe":true}',
+    )
+
+    asyncio.run(
+        service.dispatch(
+            SubscribeCommand(
+                subscription_key="BTCUSDT",
+                message=message,
+            )
+        )
+    )
+
+    assert subscriptions.subscriptions == [
+        (
+            "BTCUSDT",
+            message,
+        )
+    ]
+
+
+def test_dispatch_unsubscribe_command() -> None:
+    service, _, _, subscriptions, _ = create_service()
+
+    message = TransportTextMessage(
+        payload='{"unsubscribe":true}',
+    )
+
+    asyncio.run(
+        service.dispatch(
+            UnsubscribeCommand(
+                subscription_key="BTCUSDT",
+                message=message,
+            )
+        )
+    )
+
+    assert subscriptions.unsubscriptions == [
+        (
+            "BTCUSDT",
+            message,
+        )
+    ]
+
+
+def test_dispatch_send_text_command() -> None:
+    service, _, transport, _, _ = create_service()
+
+    asyncio.run(
+        service.dispatch(
+            SendTextCommand(
+                message=TransportTextMessage(
+                    payload="hello",
+                )
+            )
+        )
+    )
+
+    assert transport.messages == [
+        "hello",
+    ]
+
+
+def test_dispatch_send_binary_command() -> None:
+    service, _, transport, _, _ = create_service()
+
+    asyncio.run(
+        service.dispatch(
+            SendBinaryCommand(
+                message=TransportBinaryMessage(
+                    payload=b"\x01\x02",
+                )
+            )
+        )
+    )
+
+    assert transport.messages == [
+        b"\x01\x02",
+    ]
+
+
+def test_event_publisher_is_not_used_yet() -> None:
+    service, _, _, _, publisher = create_service()
+
+    asyncio.run(
+        service.dispatch(
+            ConnectCommand(),
+        )
+    )
+
+    assert publisher.events == []
+
+
+def test_unsupported_command_raises_type_error() -> None:
+    class UnsupportedCommand:
+        pass
+
+    service, *_ = create_service()
+
+    with pytest.raises(
+        TypeError,
+        match="Неподдерживаемый тип Acquisition Runtime Command",
+    ):
+        asyncio.run(
+            service.dispatch(
+                UnsupportedCommand(),  # type: ignore[arg-type]
+            )
+        )
+
+
+def test_session_exception_is_propagated() -> None:
+    class BrokenSession(FakeSession):
+        async def start(self) -> None:
+            raise RuntimeError("boom")
+
+    service = AcquisitionRuntimeService(
+        session=BrokenSession(),
+        transport=FakeTransport(),
+        subscription_manager=FakeSubscriptionManager(),
+        event_publisher=FakeEventPublisher(),
+    )
+
+    with pytest.raises(RuntimeError, match="boom"):
+        asyncio.run(
+            service.dispatch(
+                ConnectCommand(),
+            )
+        )
--- a/app/tests/unit/market_data/acquisition/runtime/test_websocket_protocol.py
+++ b/app/tests/unit/market_data/acquisition/runtime/test_websocket_protocol.py
@@ -7,6 +7,7 @@ from src.market_data.acquisition.runtime.websocket_protocol import (
    AcquisitionRuntimeCommandDispatcherProtocol,
    AcquisitionRuntimeEvent,
    AcquisitionRuntimeEventPublisherProtocol,
+    AcquisitionSubscriptionMessage,
    WebSocketSessionProtocol,
    WebSocketSubscriptionManagerProtocol,
    WebSocketTransportProtocol,
@@ -40,6 +41,20 @@ class FakeWebSocketSession:


 class FakeWebSocketSubscriptionManager:
+    async def subscribe(
+        self,
+        subscription_key: str,
+        message: AcquisitionSubscriptionMessage,
+    ) -> None:
+        return None
+
+    async def unsubscribe(
+        self,
+        subscription_key: str,
+        message: AcquisitionSubscriptionMessage,
+    ) -> None:
+        return None
+
    async def restore_subscriptions(self) -> None:
        return None

@@ -100,6 +115,20 @@ def test_incomplete_transport_does_not_satisfy_protocol() -> None:
    assert not isinstance(IncompleteTransport(), WebSocketTransportProtocol)


+def test_incomplete_subscription_manager_does_not_satisfy_protocol() -> None:
+    class IncompleteSubscriptionManager:
+        async def restore_subscriptions(self) -> None:
+            return None
+
+        async def clear_subscriptions(self) -> None:
+            return None
+
+    assert not isinstance(
+        IncompleteSubscriptionManager(),
+        WebSocketSubscriptionManagerProtocol,
+    )
+
+
 def test_incomplete_command_dispatcher_does_not_satisfy_protocol() -> None:
    class IncompleteCommandDispatcher:
        pass
@@ -117,4 +146,4 @@ def test_incomplete_event_publisher_does_not_satisfy_protocol() -> None:
    assert not isinstance(
        IncompleteEventPublisher(),
        AcquisitionRuntimeEventPublisherProtocol,
-    )
+    )