dzentra_bot/app/src/bootstrap/trade_stream_runtime.py

from __future__ import annotations

from src.core.config import Settings
from src.integrations.exchange.rest_client import ExchangeRestClient
from src.market_data.acquisition.adapters.dzengi.rest import (
    DzengiTradesDocumentSource,
)
from src.market_data.acquisition.adapters.dzengi.websocket import (
    DzengiUnifiedWebSocketAdapter,
)
from src.market_data.acquisition.adapters.dzengi.websocket_control_message_handler import (
    DzengiWebSocketControlMessageHandler,
)
from src.market_data.acquisition.adapters.dzengi.websocket_inbound_message_classifier import (
    DzengiWebSocketInboundMessageClassifier,
)
from src.market_data.acquisition.adapters.dzengi.websocket_transport import (
    DzengiWebSocketTransport,
)
from src.market_data.acquisition.consistency.trade_observation_sink_protocol import (
    TradeObservationSinkProtocol,
)
from src.market_data.acquisition.runtime.acquisition_runtime_event_logging_consumer import (
    AcquisitionRuntimeEventLoggingConsumer,
)
from src.market_data.acquisition.runtime.acquisition_runtime_event_publisher import (
    AcquisitionRuntimeEventPublisher,
)
from src.market_data.acquisition.runtime.trade_stream_production_runtime import (
    TradeStreamProductionRuntime,
)
from src.market_data.acquisition.runtime.websocket_session import (
    WebSocketSession,
)
from src.market_data.acquisition.runtime.websocket_subscription_manager import (
    WebSocketSubscriptionManager,
)
from src.market_data.acquisition.trade_stream_runtime_composition import (
    build_trade_stream_runtime_composition,
)
from src.market_data.storage.contracts import (
    TradeCheckpointStorageProtocol,
)


def build_trade_stream_production_runtime(
    settings: Settings,
    *,
    trade_observation_sink: TradeObservationSinkProtocol | None = None,
    checkpoint_storage: TradeCheckpointStorageProtocol | None = None,
) -> TradeStreamProductionRuntime | None:
    """
    Собрать Production Trade Stream Runtime без запуска lifecycle.

    Выключенный feature flag возвращает ``None`` до создания
    WebSocket/REST-зависимостей. Все stateful-компоненты включённого
    Runtime создаются один раз и затем передаются по identity.
    """
    trade_stream = settings.trade_stream
    storage_enabled = settings.market_data_storage.enabled

    if not trade_stream.enabled:
        if storage_enabled:
            raise RuntimeError(
                "Trade Stream must be enabled when Market Data "
                "Storage is enabled."
            )

        return None

    if storage_enabled and (
        trade_observation_sink is None
        or checkpoint_storage is None
    ):
        raise RuntimeError(
            "Enabled Market Data Storage requires both Trade "
            "observation sink and checkpoint storage."
        )

    if not storage_enabled and checkpoint_storage is not None:
        raise RuntimeError(
            "Checkpoint storage requires enabled Market Data Storage."
        )

    headers = {
        "Origin": settings.exchange_base_url.rstrip("/"),
        "Content-Type": "application/json",
    }

    if settings.exchange_api_key:
        headers["X-MBX-APIKEY"] = settings.exchange_api_key

    transport = DzengiWebSocketTransport(
        url=trade_stream.websocket_url,
        headers=headers,
        open_timeout=trade_stream.open_timeout_seconds,
        ping_interval=None,
        ping_timeout=None,
        probe_timeout=trade_stream.probe_timeout_seconds,
        close_timeout=trade_stream.close_timeout_seconds,
    )
    session = WebSocketSession(transport)
    subscription_manager = WebSocketSubscriptionManager(
        transport,
        supports_unsubscribe=False,
    )
    event_publisher = AcquisitionRuntimeEventPublisher(
        (
            AcquisitionRuntimeEventLoggingConsumer(),
        )
    )
    recovery_document_source = DzengiTradesDocumentSource(
        ExchangeRestClient(settings=settings),
    )
    message_adapter = DzengiUnifiedWebSocketAdapter()

    composition = build_trade_stream_runtime_composition(
        session=session,
        transport=transport,
        subscription_manager=subscription_manager,
        event_publisher=event_publisher,
        message_adapter=message_adapter,
        recovery_document_source=recovery_document_source,
        symbols=trade_stream.symbols,
        heartbeat_timeout_seconds=(
            trade_stream.heartbeat_timeout_seconds
        ),
        scheduler_interval_seconds=(
            trade_stream.scheduler_interval_seconds
        ),
        trade_observation_sink=trade_observation_sink,
        max_recovery_window_ms=trade_stream.recovery_window_ms,
        checkpoint_storage=checkpoint_storage,
        checkpoint_venue=(
            settings.exchange_name
            if checkpoint_storage is not None
            else None
        ),
    )

    return TradeStreamProductionRuntime(
        session=session,
        transport=transport,
        subscription_manager=subscription_manager,
        event_publisher=event_publisher,
        trade_stream_service=(
            composition.trade_stream_acquisition_service
        ),
        message_classifier=(
            DzengiWebSocketInboundMessageClassifier()
        ),
        control_message_handler=(
            DzengiWebSocketControlMessageHandler()
        ),
        live_processing_gate=composition.live_processing_gate,
        reconnect_recovery_coordinator=(
            composition.runtime_reconnect_recovery_coordinator
        ),
        runtime_supervisor=composition.runtime_supervisor,
        runtime_scheduler=composition.runtime_scheduler,
        symbols=trade_stream.symbols,
        startup_recovery_coordinator=(
            composition.runtime_startup_recovery_coordinator
        ),
        subscription_ack_timeout_seconds=(
            trade_stream.subscription_ack_timeout_seconds
        ),
        startup_market_buffer_capacity=(
            trade_stream.startup_market_buffer_capacity
        ),
    )