Build 060.16: add Trade Subscription Layer and subscription builders

app/scripts/check_trades_unsubscribe.py

@@ -0,0 +1,214 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+from typing import Any
+from uuid import uuid4
+
+import websockets
+from websockets.typing import Subprotocol
+
+
+SYMBOL = os.getenv(
+    "TRADE_TEST_SYMBOL",
+    "BTC/USD_LEVERAGE",
+)
+
+OBSERVE_BEFORE_UNSUBSCRIBE_SECONDS = 30
+OBSERVE_AFTER_UNSUBSCRIBE_SECONDS = 60
+RECEIVE_TIMEOUT_SECONDS = 10
+
+
+def build_ws_url() -> str:
+    raw_url = os.getenv(
+        "EXCHANGE_WS_URL",
+        "wss://api-adapter.dzengi.com",
+    ).rstrip("/")
+
+    if raw_url.startswith("https://"):
+        raw_url = raw_url.replace("https://", "wss://", 1)
+    elif raw_url.startswith("http://"):
+        raw_url = raw_url.replace("http://", "ws://", 1)
+
+    if raw_url.endswith("/connect"):
+        return raw_url
+
+    return f"{raw_url}/connect"
+
+
+def build_headers() -> dict[str, str]:
+    base_url = os.getenv(
+        "EXCHANGE_BASE_URL",
+        "https://api-adapter.dzengi.com",
+    ).rstrip("/")
+
+    headers = {
+        "Origin": base_url,
+        "Content-Type": "application/json",
+    }
+
+    api_key = os.getenv("EXCHANGE_API_KEY", "").strip()
+
+    if api_key:
+        headers["X-MBX-APIKEY"] = api_key
+
+    return headers
+
+
+def build_request(destination: str) -> dict[str, Any]:
+    return {
+        "correlationId": str(uuid4()),
+        "destination": destination,
+        "payload": {
+            "symbols": [SYMBOL],
+        },
+    }
+
+
+def decode_message(raw_message: str | bytes) -> Any:
+    if isinstance(raw_message, bytes):
+        raw_message = raw_message.decode("utf-8")
+
+    try:
+        return json.loads(raw_message)
+    except json.JSONDecodeError:
+        return raw_message
+
+
+def print_message(label: str, message: Any) -> None:
+    print()
+    print("=" * 80)
+    print(label)
+    print("=" * 80)
+
+    if isinstance(message, (dict, list)):
+        print(json.dumps(message, indent=2, ensure_ascii=False))
+    else:
+        print(message)
+
+
+def is_trade_event(message: Any) -> bool:
+    return (
+        isinstance(message, dict)
+        and message.get("destination") == "internal.trade"
+    )
+
+
+async def receive_until(
+    websocket: Any,
+    *,
+    duration_seconds: int,
+    phase: str,
+) -> int:
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + duration_seconds
+    trade_count = 0
+
+    while loop.time() < deadline:
+        remaining = deadline - loop.time()
+        timeout = min(RECEIVE_TIMEOUT_SECONDS, remaining)
+
+        try:
+            raw_message = await asyncio.wait_for(
+                websocket.recv(),
+                timeout=timeout,
+            )
+        except asyncio.TimeoutError:
+            print(
+                f"[{phase}] За последние {timeout:.0f} секунд "
+                "сообщений не получено; соединение остаётся открытым."
+            )
+            continue
+
+        message = decode_message(raw_message)
+        print_message(f"[{phase}] Получено сообщение", message)
+
+        if is_trade_event(message):
+            trade_count += 1
+
+    return trade_count
+
+
+async def main() -> None:
+    ws_url = build_ws_url()
+    headers = build_headers()
+
+    subscribe_request = build_request("trades.subscribe")
+    unsubscribe_request = build_request("trades.unsubscribe")
+
+    print(f"WebSocket URL: {ws_url}")
+    print(f"Symbol: {SYMBOL}")
+
+    async with websockets.connect(
+        ws_url,
+        extra_headers=headers,
+        subprotocols=[Subprotocol("json")],
+        ping_interval=20,
+        ping_timeout=20,
+        open_timeout=20,
+    ) as websocket:
+        print_message(
+            "Отправляется trades.subscribe",
+            subscribe_request,
+        )
+
+        await websocket.send(
+            json.dumps(subscribe_request),
+        )
+
+        before_count = await receive_until(
+            websocket,
+            duration_seconds=OBSERVE_BEFORE_UNSUBSCRIBE_SECONDS,
+            phase="BEFORE UNSUBSCRIBE",
+        )
+
+        print()
+        print(
+            "Количество Trade Event до unsubscribe: "
+            f"{before_count}"
+        )
+
+        print_message(
+            "Отправляется trades.unsubscribe",
+            unsubscribe_request,
+        )
+
+        await websocket.send(
+            json.dumps(unsubscribe_request),
+        )
+
+        after_count = await receive_until(
+            websocket,
+            duration_seconds=OBSERVE_AFTER_UNSUBSCRIBE_SECONDS,
+            phase="AFTER UNSUBSCRIBE",
+        )
+
+        print()
+        print("=" * 80)
+        print("РЕЗУЛЬТАТ НАБЛЮДЕНИЯ")
+        print("=" * 80)
+        print(f"Trade Event до unsubscribe: {before_count}")
+        print(f"Trade Event после unsubscribe: {after_count}")
+
+        if before_count == 0:
+            print(
+                "До unsubscribe не было получено ни одной сделки. "
+                "Эксперимент нельзя считать доказательным."
+            )
+        elif after_count == 0:
+            print(
+                "После trades.unsubscribe новые сделки не поступили. "
+                "Команда, вероятно, поддерживается, но необходимо "
+                "проверить ACK или повторить тест при высокой активности."
+            )
+        else:
+            print(
+                "После trades.unsubscribe сделки продолжили поступать. "
+                "Команда либо не поддерживается, либо была отклонена, "
+                "либо имеет другой формат."
+            )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

app/src/market_data/acquisition/adapters/dzengi/__init__.py

@@ -0,0 +1 @@
+# app/src/market_data/acquisition/adapters/dzengi/__init__.py

app/src/market_data/acquisition/subscriptions/__init__.py

@@ -0,0 +1,6 @@
+"""
+Формирование транспортных подписок Market Data Acquisition.
+
+Модули пакета преобразуют параметры предметной области в универсальные
+Runtime-команды и не выполняют сетевые операции самостоятельно.
+"""

app/src/market_data/acquisition/subscriptions/trades.py

@@ -0,0 +1,111 @@
+# app/src/market_data/acquisition/subscriptions/trades.py
+
+from __future__ import annotations
+
+import json
+from collections.abc import Sequence
+from uuid import uuid4
+
+from src.market_data.acquisition.runtime.runtime_commands import (
+    SubscribeCommand,
+)
+from src.market_data.acquisition.runtime.transport_messages import (
+    TransportTextMessage,
+)
+
+
+TRADE_SUBSCRIPTION_DESTINATION = "trades.subscribe"
+TRADE_SUBSCRIPTION_KEY_PREFIX = "trades"
+
+
+def _normalize_trade_subscription_symbols(
+    symbols: Sequence[str],
+) -> tuple[str, ...]:
+    """
+    Нормализовать символы Trade-подписки.
+
+    Удаляет внешние пробелы, пустые значения и дубликаты.
+    Возвращает символы в стабильном лексикографическом порядке.
+    """
+
+    normalized_symbols = {
+        symbol.strip()
+        for symbol in symbols
+        if symbol.strip()
+    }
+
+    if not normalized_symbols:
+        raise ValueError(
+            "Trade subscription requires at least one non-empty symbol."
+        )
+
+    return tuple(sorted(normalized_symbols))
+
+
+def build_trade_subscription_key(
+    symbols: Sequence[str],
+) -> str:
+    """
+    Построить стабильный идентификатор Trade-подписки.
+
+    Идентификатор не зависит от порядка входных символов и используется
+    Runtime для регистрации и последующего восстановления подписки.
+    """
+
+    normalized_symbols = _normalize_trade_subscription_symbols(symbols)
+
+    return (
+        f"{TRADE_SUBSCRIPTION_KEY_PREFIX}:"
+        f"{','.join(normalized_symbols)}"
+    )
+
+
+def build_trade_subscribe_message(
+    symbols: Sequence[str],
+    *,
+    correlation_id: str | None = None,
+) -> TransportTextMessage:
+    """
+    Построить текстовое сообщение подписки Dzengi Trade WebSocket.
+
+    correlation_id может быть передан вызывающим кодом для детерминированных
+    тестов и сопоставления ACK. Если значение не передано, создаётся новый UUID.
+    """
+
+    normalized_symbols = _normalize_trade_subscription_symbols(symbols)
+
+    document = {
+        "correlationId": correlation_id or str(uuid4()),
+        "destination": TRADE_SUBSCRIPTION_DESTINATION,
+        "payload": {
+            "symbols": list(normalized_symbols),
+        },
+    }
+
+    return TransportTextMessage(
+        payload=json.dumps(
+            document,
+            separators=(",", ":"),
+        )
+    )
+
+
+def build_trade_subscribe_command(
+    symbols: Sequence[str],
+    *,
+    correlation_id: str | None = None,
+) -> SubscribeCommand:
+    """
+    Построить универсальную Runtime-команду подписки на Trade Feed.
+
+    Runtime получает стабильный ключ и готовое транспортное сообщение,
+    не интерпретируя Dzengi-specific JSON.
+    """
+
+    return SubscribeCommand(
+        subscription_key=build_trade_subscription_key(symbols),
+        message=build_trade_subscribe_message(
+            symbols,
+            correlation_id=correlation_id,
+        ),
+    )

app/tests/unit/market_data/acquisition/subscriptions/test_trades.py

@@ -0,0 +1,178 @@
+# app/tests/unit/market_data/acquisition/subscriptions/test_trades.py
+
+from __future__ import annotations
+
+import json
+from uuid import UUID
+
+import pytest
+
+from src.market_data.acquisition.runtime.runtime_commands import (
+    SubscribeCommand,
+)
+from src.market_data.acquisition.runtime.transport_messages import (
+    TransportTextMessage,
+)
+from src.market_data.acquisition.subscriptions.trades import (
+    TRADE_SUBSCRIPTION_DESTINATION,
+    build_trade_subscribe_command,
+    build_trade_subscribe_message,
+    build_trade_subscription_key,
+)
+
+
+def test_build_trade_subscription_key_for_single_symbol() -> None:
+    result = build_trade_subscription_key(
+        ["BTC/USD_LEVERAGE"],
+    )
+
+    assert result == "trades:BTC/USD_LEVERAGE"
+
+
+def test_build_trade_subscription_key_normalizes_symbols() -> None:
+    result = build_trade_subscription_key(
+        [
+            " ETH/USD_LEVERAGE ",
+            "BTC/USD_LEVERAGE",
+            "ETH/USD_LEVERAGE",
+            "",
+            "   ",
+        ],
+    )
+
+    assert result == (
+        "trades:BTC/USD_LEVERAGE,ETH/USD_LEVERAGE"
+    )
+
+
+def test_build_trade_subscription_key_rejects_empty_symbols() -> None:
+    with pytest.raises(
+        ValueError,
+        match="at least one non-empty symbol",
+    ):
+        build_trade_subscription_key(
+            ["", "   "],
+        )
+
+
+def test_build_trade_subscribe_message_returns_transport_message() -> None:
+    result = build_trade_subscribe_message(
+        ["BTC/USD_LEVERAGE"],
+        correlation_id="trade-subscription-1",
+    )
+
+    assert isinstance(result, TransportTextMessage)
+
+
+def test_build_trade_subscribe_message_builds_dzengi_document() -> None:
+    result = build_trade_subscribe_message(
+        [
+            "ETH/USD_LEVERAGE",
+            "BTC/USD_LEVERAGE",
+        ],
+        correlation_id="trade-subscription-1",
+    )
+
+    document = json.loads(result.payload)
+
+    assert document == {
+        "correlationId": "trade-subscription-1",
+        "destination": TRADE_SUBSCRIPTION_DESTINATION,
+        "payload": {
+            "symbols": [
+                "BTC/USD_LEVERAGE",
+                "ETH/USD_LEVERAGE",
+            ],
+        },
+    }
+
+
+def test_build_trade_subscribe_message_generates_uuid() -> None:
+    result = build_trade_subscribe_message(
+        ["BTC/USD_LEVERAGE"],
+    )
+
+    document = json.loads(result.payload)
+
+    generated_id = UUID(document["correlationId"])
+
+    assert str(generated_id) == document["correlationId"]
+
+
+def test_build_trade_subscribe_message_normalizes_symbols() -> None:
+    result = build_trade_subscribe_message(
+        [
+            " BTC/USD_LEVERAGE ",
+            "BTC/USD_LEVERAGE",
+            "",
+        ],
+        correlation_id="trade-subscription-1",
+    )
+
+    document = json.loads(result.payload)
+
+    assert document["payload"]["symbols"] == [
+        "BTC/USD_LEVERAGE",
+    ]
+
+
+def test_build_trade_subscribe_message_rejects_empty_symbols() -> None:
+    with pytest.raises(
+        ValueError,
+        match="at least one non-empty symbol",
+    ):
+        build_trade_subscribe_message(
+            [],
+            correlation_id="trade-subscription-1",
+        )
+
+
+def test_build_trade_subscribe_command_returns_runtime_command() -> None:
+    result = build_trade_subscribe_command(
+        ["BTC/USD_LEVERAGE"],
+        correlation_id="trade-subscription-1",
+    )
+
+    assert isinstance(result, SubscribeCommand)
+    assert result.subscription_key == "trades:BTC/USD_LEVERAGE"
+    assert isinstance(result.message, TransportTextMessage)
+
+
+def test_build_trade_subscribe_command_contains_dzengi_message() -> None:
+    result = build_trade_subscribe_command(
+        ["BTC/USD_LEVERAGE"],
+        correlation_id="trade-subscription-1",
+    )
+
+    document = json.loads(result.message.payload)
+
+    assert document == {
+        "correlationId": "trade-subscription-1",
+        "destination": "trades.subscribe",
+        "payload": {
+            "symbols": [
+                "BTC/USD_LEVERAGE",
+            ],
+        },
+    }
+
+
+def test_build_trade_subscribe_command_uses_same_normalized_symbols() -> None:
+    result = build_trade_subscribe_command(
+        [
+            "ETH/USD_LEVERAGE",
+            " BTC/USD_LEVERAGE ",
+            "ETH/USD_LEVERAGE",
+        ],
+        correlation_id="trade-subscription-1",
+    )
+
+    document = json.loads(result.message.payload)
+
+    assert result.subscription_key == (
+        "trades:BTC/USD_LEVERAGE,ETH/USD_LEVERAGE"
+    )
+    assert document["payload"]["symbols"] == [
+        "BTC/USD_LEVERAGE",
+        "ETH/USD_LEVERAGE",
+    ]