build 059.6: map websocket OHLC close events

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

@@ -14,19 +14,23 @@ from src.market_data.acquisition.adapters.dzengi.models import (
     DzengiMinNotionalFilter,
     DzengiRawNumeric,
     DzengiTicker24hrResponse,
+    DzengiWebSocketOhlcEvent,
     DzengiWebSocketQuoteResponse,
 )
 from src.market_data.acquisition.exceptions import (
     CandleMappingError,
+    CandleWebSocketMappingError,
     InstrumentReferenceMappingError,
     QuoteMappingError,
 )
 from src.market_data.acquisition.models.candle import Candle
+from src.market_data.acquisition.models.candle_close import CandleCloseEvent
 from src.market_data.acquisition.models.instrument import Instrument
 from src.market_data.acquisition.models.quote import Quote
 
 
 _DZENGI_SOURCE_NAME = "dzengi"
+_DZENGI_WEBSOCKET_OHLC_SOURCE_NAME = "dzengi_websocket_ohlc"
 
 
 def map_dzengi_symbol_to_instrument(
@@ -318,6 +322,104 @@ def map_dzengi_websocket_quote_to_quote(
         source=_DZENGI_SOURCE_NAME,
     )
 
+
+def map_dzengi_websocket_ohlc_to_candle_close_event(
+    event: DzengiWebSocketOhlcEvent,
+    *,
+    received_at: datetime,
+) -> CandleCloseEvent:
+    """
+    Преобразовать проверенное Dzengi WebSocket OHLC-событие
+    во внутреннюю модель CandleCloseEvent.
+
+    Функция предполагает, что до mapper уже были выполнены:
+    schema validation, parsing и value validation.
+    """
+
+    normalized_received_at = _require_websocket_ohlc_aware_datetime(
+        received_at,
+        field_name="received_at",
+    )
+
+    return CandleCloseEvent(
+        symbol=event.symbol.strip(),
+        interval=event.interval,
+        candle_type=event.candle_type,
+        open_time=_websocket_ohlc_timestamp_ms_to_utc_datetime(
+            event.open_time,
+        ),
+        received_at=normalized_received_at,
+        open_price=_required_websocket_ohlc_decimal(
+            event.open_price,
+            field_name="open",
+        ),
+        high_price=_required_websocket_ohlc_decimal(
+            event.high_price,
+            field_name="high",
+        ),
+        low_price=_required_websocket_ohlc_decimal(
+            event.low_price,
+            field_name="low",
+        ),
+        close_price=_required_websocket_ohlc_decimal(
+            event.close_price,
+            field_name="close",
+        ),
+        source=_DZENGI_WEBSOCKET_OHLC_SOURCE_NAME,
+    )
+
+
+def _websocket_ohlc_timestamp_ms_to_utc_datetime(
+    value: int,
+) -> datetime:
+    try:
+        return datetime.fromtimestamp(
+            value / 1000,
+            tz=timezone.utc,
+        )
+    except (OverflowError, OSError, ValueError) as exc:
+        raise CandleWebSocketMappingError(
+            "Поле t WebSocket OHLC-события невозможно "
+            "преобразовать в UTC datetime."
+        ) from exc
+
+
+def _required_websocket_ohlc_decimal(
+    value: DzengiRawNumeric,
+    *,
+    field_name: str,
+) -> Decimal:
+    try:
+        result = Decimal(str(value))
+    except (InvalidOperation, ValueError) as exc:
+        raise CandleWebSocketMappingError(
+            f"Поле {field_name} WebSocket OHLC-события невозможно "
+            "преобразовать в Decimal."
+        ) from exc
+
+    if not result.is_finite():
+        raise CandleWebSocketMappingError(
+            f"Поле {field_name} WebSocket OHLC-события должно быть "
+            "конечным числом."
+        )
+
+    return result
+
+
+def _require_websocket_ohlc_aware_datetime(
+    value: datetime,
+    *,
+    field_name: str,
+) -> datetime:
+    if value.tzinfo is None or value.utcoffset() is None:
+        raise CandleWebSocketMappingError(
+            f"Поле {field_name} должно содержать "
+            "timezone-aware datetime."
+        )
+
+    return value
+
+
 def map_dzengi_klines_to_candles(
     response: DzengiKlinesResponse,
     *,

app/src/market_data/acquisition/exceptions.py

@@ -93,6 +93,12 @@ class CandleWebSocketValueError(MarketDataAcquisitionError):
     pass
 
 
+# Ошибка преобразования WebSocket OHLC-события
+# во внутреннюю модель уведомления о закрытии свечи.
+class CandleWebSocketMappingError(MarketDataAcquisitionError):
+    pass
+
+
 # Ошибка преобразования проверенного документа в raw-модели свечей.
 class CandleParseError(MarketDataAcquisitionError):
     pass

app/tests/unit/market_data/acquisition/adapters/dzengi/test_websocket_ohlc_mapper.py

@@ -0,0 +1,253 @@
+# app/tests/unit/market_data/acquisition/adapters/dzengi/test_websocket_ohlc_mapper.py
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+from typing import TypedDict
+
+import pytest
+
+from src.market_data.acquisition.adapters.dzengi.mapper import (
+    map_dzengi_websocket_ohlc_to_candle_close_event,
+)
+from src.market_data.acquisition.adapters.dzengi.models import (
+    DzengiWebSocketOhlcEvent,
+)
+from src.market_data.acquisition.exceptions import (
+    CandleWebSocketMappingError,
+)
+from src.market_data.acquisition.models.candle_close import (
+    CandleCloseEvent,
+)
+
+
+class OhlcValues(TypedDict):
+    open_price: str | int | float
+    high_price: str | int | float
+    low_price: str | int | float
+    close_price: str | int | float
+
+
+def _event(
+    *,
+    open_time: int = 1784227140000,
+    open_price: str | int | float = "63992.00",
+    high_price: str | int | float = "64032.55",
+    low_price: str | int | float = "63984.00",
+    close_price: str | int | float = "64032.55",
+) -> DzengiWebSocketOhlcEvent:
+    return DzengiWebSocketOhlcEvent(
+        symbol=" BTC/USD_LEVERAGE ",
+        interval="1m",
+        candle_type="classic",
+        open_time=open_time,
+        open_price=open_price,
+        high_price=high_price,
+        low_price=low_price,
+        close_price=close_price,
+    )
+
+
+def _received_at() -> datetime:
+    return datetime(
+        2026,
+        7,
+        16,
+        18,
+        40,
+        0,
+        125000,
+        tzinfo=timezone.utc,
+    )
+
+
+def test_map_websocket_ohlc_returns_candle_close_event() -> None:
+    result = map_dzengi_websocket_ohlc_to_candle_close_event(
+        _event(),
+        received_at=_received_at(),
+    )
+
+    assert isinstance(result, CandleCloseEvent)
+
+
+def test_map_websocket_ohlc_maps_all_fields() -> None:
+    result = map_dzengi_websocket_ohlc_to_candle_close_event(
+        _event(),
+        received_at=_received_at(),
+    )
+
+    assert result.symbol == "BTC/USD_LEVERAGE"
+    assert result.interval == "1m"
+    assert result.candle_type == "classic"
+
+    assert result.open_time == datetime(
+        2026,
+        7,
+        16,
+        18,
+        39,
+        tzinfo=timezone.utc,
+    )
+    assert result.received_at == _received_at()
+
+    assert result.open_price == Decimal("63992.00")
+    assert result.high_price == Decimal("64032.55")
+    assert result.low_price == Decimal("63984.00")
+    assert result.close_price == Decimal("64032.55")
+
+    assert result.source == "dzengi_websocket_ohlc"
+
+
+def test_map_websocket_ohlc_converts_numeric_variants_to_decimal() -> None:
+    result = map_dzengi_websocket_ohlc_to_candle_close_event(
+        _event(
+            open_price=63992,
+            high_price=64032.55,
+            low_price="63984.00",
+            close_price="64001.25",
+        ),
+        received_at=_received_at(),
+    )
+
+    assert result.open_price == Decimal("63992")
+    assert result.high_price == Decimal("64032.55")
+    assert result.low_price == Decimal("63984.00")
+    assert result.close_price == Decimal("64001.25")
+
+
+def test_map_websocket_ohlc_preserves_aware_received_at() -> None:
+    received_at = datetime(
+        2026,
+        7,
+        16,
+        21,
+        40,
+        tzinfo=timezone(timedelta(hours=3)),
+    )
+
+    result = map_dzengi_websocket_ohlc_to_candle_close_event(
+        _event(),
+        received_at=received_at,
+    )
+
+    assert result.received_at is received_at
+    assert result.received_at.utcoffset() == timedelta(hours=3)
+
+
+def test_map_websocket_ohlc_rejects_naive_received_at() -> None:
+    received_at = datetime(
+        2026,
+        7,
+        16,
+        18,
+        40,
+    )
+
+    with pytest.raises(
+        CandleWebSocketMappingError,
+        match="received_at.*timezone-aware",
+    ):
+        map_dzengi_websocket_ohlc_to_candle_close_event(
+            _event(),
+            received_at=received_at,
+        )
+
+
+def test_map_websocket_ohlc_rejects_invalid_timestamp() -> None:
+    with pytest.raises(
+        CandleWebSocketMappingError,
+        match="Поле t.*UTC datetime",
+    ):
+        map_dzengi_websocket_ohlc_to_candle_close_event(
+            _event(open_time=10**30),
+            received_at=_received_at(),
+        )
+
+
+@pytest.mark.parametrize(
+    ("field_name", "field_value"),
+    (
+        ("open_price", "invalid"),
+        ("high_price", "invalid"),
+        ("low_price", "invalid"),
+        ("close_price", "invalid"),
+    ),
+)
+def test_map_websocket_ohlc_rejects_invalid_decimal(
+    field_name: str,
+    field_value: str,
+) -> None:
+    values: OhlcValues = {
+        "open_price": "63992.00",
+        "high_price": "64032.55",
+        "low_price": "63984.00",
+        "close_price": "64032.55",
+    }
+    values[field_name] = field_value
+
+    with pytest.raises(CandleWebSocketMappingError):
+        map_dzengi_websocket_ohlc_to_candle_close_event(
+            _event(**values),
+            received_at=_received_at(),
+        )
+
+
+@pytest.mark.parametrize(
+    ("field_name", "field_value"),
+    (
+        ("open_price", "NaN"),
+        ("high_price", "Infinity"),
+        ("low_price", "-Infinity"),
+        ("close_price", "NaN"),
+    ),
+)
+def test_map_websocket_ohlc_rejects_non_finite_decimal(
+    field_name: str,
+    field_value: str,
+) -> None:
+    values: OhlcValues = {
+        "open_price": "63992.00",
+        "high_price": "64032.55",
+        "low_price": "63984.00",
+        "close_price": "64032.55",
+    }
+    values[field_name] = field_value
+
+    with pytest.raises(
+        CandleWebSocketMappingError,
+        match="конечным числом",
+    ):
+        map_dzengi_websocket_ohlc_to_candle_close_event(
+            _event(**values),
+            received_at=_received_at(),
+        )
+
+
+def test_map_websocket_ohlc_preserves_heikin_ashi_type() -> None:
+    event = DzengiWebSocketOhlcEvent(
+        symbol="BTC/USD_LEVERAGE",
+        interval="1m",
+        candle_type="heikin-ashi",
+        open_time=1784227140000,
+        open_price="64068.18",
+        high_price="64128.80",
+        low_price="64068.18",
+        close_price="64100.85",
+    )
+
+    result = map_dzengi_websocket_ohlc_to_candle_close_event(
+        event,
+        received_at=_received_at(),
+    )
+
+    assert result.candle_type == "heikin-ashi"
+
+
+def test_map_websocket_ohlc_does_not_create_canonical_volume() -> None:
+    result = map_dzengi_websocket_ohlc_to_candle_close_event(
+        _event(),
+        received_at=_received_at(),
+    )
+
+    assert not hasattr(result, "volume")