gitadmin/dzentra_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

build 059.1-059.2: add websocket OHLC transport and schema validation

Browse Source
This commit is contained in:
Sergey
2026-07-16 22:14:25 +03:00
parent ebb578db35
commit e0aa04b32a
6 changed files with 1016 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
app/src/market_data/acquisition/adapters/dzengi/models.py
View File

@@ -147,4 +147,23 @@ class DzengiKline:
# Транспортное представление ответа Dzengi GET /api/v1/klines.
@dataclass(frozen=True, slots=True)
class DzengiKlinesResponse:
items: tuple[DzengiKline, ...]
items: tuple[DzengiKline, ...]
# Транспортное представление одного события Dzengi WebSocket ohlc.event.
#
# Событие содержит завершённую OHLC-свечу без объёма и поэтому не является
# канонической моделью Candle. Поле candle_type принимает значения,
# подтверждённые runtime-контрактом Dzengi: classic или heikin-ashi.
@dataclass(frozen=True, slots=True)
class DzengiWebSocketOhlcEvent:
symbol: str
interval: str
candle_type: str
open_time: int
open_price: DzengiRawNumeric
high_price: DzengiRawNumeric
low_price: DzengiRawNumeric
close_price: DzengiRawNumeric

5
app/src/market_data/acquisition/exceptions.py
View File

@@ -78,6 +78,11 @@ class CandleSchemaError(MarketDataAcquisitionError):
pass
# Ошибка структуры события Candles Feed из WebSocket.
class CandleWebSocketSchemaError(MarketDataAcquisitionError):
pass
# Ошибка преобразования проверенного документа в raw-модели свечей.
class CandleParseError(MarketDataAcquisitionError):
pass

131
app/src/market_data/acquisition/validation/schema.py
View File

@@ -8,6 +8,7 @@ from typing import Mapping
from src.market_data.acquisition.exceptions import (
CandleSchemaError,
CandleWebSocketSchemaError,
InstrumentReferenceSchemaError,
QuoteSchemaError,
)
@@ -591,4 +592,134 @@ def _require_candle_mapping(
f"типа {type(key).__name__}."
)
return value
# Структурно проверенное представление события Dzengi WebSocket ohlc.event.
@dataclass(frozen=True, slots=True)
class ValidatedWebSocketOhlcDocument:
payload: Mapping[str, object]
status: object
destination: object
correlation_id: object | None
def validate_dzengi_websocket_ohlc_schema(
document: object,
) -> ValidatedWebSocketOhlcDocument:
"""
Проверить структуру события Dzengi WebSocket OHLC без проверки значений.
Ожидаемый runtime-контракт:
{
"status": "OK",
"destination": "ohlc.event",
"payload": {
"symbol": "BTC/USD_LEVERAGE",
"interval": "1m",
"type": "classic",
"t": 1784224740000,
"o": 63992.0,
"h": 64032.55,
"l": 63984.0,
"c": 64032.55
}
}
Функция проверяет только структуру сообщения:
- корневой JSON-объект;
- наличие status;
- наличие destination;
- наличие payload;
- обязательный набор полей OHLC;
- строковые ключи объектов.
Функция не проверяет предметные значения, не преобразует timestamp,
не преобразует цены и не создаёт transport-модель адаптера.
"""
root = _require_websocket_ohlc_mapping(
document,
path="$",
)
if "status" not in root:
raise CandleWebSocketSchemaError(
"$.status отсутствует в событии WebSocket OHLC."
)
if "destination" not in root:
raise CandleWebSocketSchemaError(
"$.destination отсутствует в событии WebSocket OHLC."
)
if "payload" not in root:
raise CandleWebSocketSchemaError(
"$.payload отсутствует в событии WebSocket OHLC."
)
payload = _require_websocket_ohlc_mapping(
root.get("payload"),
path="$.payload",
)
_validate_websocket_ohlc_payload(payload)
return ValidatedWebSocketOhlcDocument(
payload=MappingProxyType(dict(payload)),
status=root.get("status"),
destination=root.get("destination"),
correlation_id=root.get("correlationId"),
)
def _validate_websocket_ohlc_payload(
payload: Mapping[str, object],
) -> None:
required_fields = (
"symbol",
"interval",
"type",
"t",
"o",
"h",
"l",
"c",
)
missing_fields = tuple(
field_name
for field_name in required_fields
if field_name not in payload
)
if missing_fields:
formatted_fields = ", ".join(missing_fields)
raise CandleWebSocketSchemaError(
"$.payload не содержит обязательные поля WebSocket OHLC: "
f"{formatted_fields}."
)
def _require_websocket_ohlc_mapping(
value: object,
*,
path: str,
) -> Mapping[str, object]:
if not isinstance(value, dict):
raise CandleWebSocketSchemaError(
f"{path} должен быть JSON-объектом, "
f"получен {type(value).__name__}."
)
for key in value:
if not isinstance(key, str):
raise CandleWebSocketSchemaError(
f"{path} содержит нестроковый ключ "
f"типа {type(key).__name__}."
)
return value

217
app/tests/unit/market_data/acquisition/validation/test_websocket_ohlc_schema.py Normal file
View File

@@ -0,0 +1,217 @@
# app/tests/unit/market_data/acquisition/validation/test_websocket_ohlc_schema.py
from __future__ import annotations
from types import MappingProxyType
import pytest
from src.market_data.acquisition.exceptions import (
CandleWebSocketSchemaError,
)
from src.market_data.acquisition.validation.schema import (
ValidatedWebSocketOhlcDocument,
validate_dzengi_websocket_ohlc_schema,
)
def _valid_document() -> dict[str, object]:
return {
"status": "OK",
"destination": "ohlc.event",
"payload": {
"symbol": "BTC/USD_LEVERAGE",
"interval": "1m",
"type": "classic",
"t": 1784224740000,
"o": 63992.0,
"h": 64032.55,
"l": 63984.0,
"c": 64032.55,
},
}
def test_validate_websocket_ohlc_schema_returns_immutable_document() -> None:
result = validate_dzengi_websocket_ohlc_schema(
_valid_document()
)
assert isinstance(result, ValidatedWebSocketOhlcDocument)
assert isinstance(result.payload, MappingProxyType)
assert result.status == "OK"
assert result.destination == "ohlc.event"
assert result.correlation_id is None
assert result.payload == {
"symbol": "BTC/USD_LEVERAGE",
"interval": "1m",
"type": "classic",
"t": 1784224740000,
"o": 63992.0,
"h": 64032.55,
"l": 63984.0,
"c": 64032.55,
}
def test_validate_websocket_ohlc_schema_preserves_correlation_id() -> None:
document = _valid_document()
document["correlationId"] = "correlation-1"
result = validate_dzengi_websocket_ohlc_schema(document)
assert result.correlation_id == "correlation-1"
def test_validate_websocket_ohlc_schema_copies_payload() -> None:
document = _valid_document()
payload = document["payload"]
assert isinstance(payload, dict)
result = validate_dzengi_websocket_ohlc_schema(document)
payload["c"] = 1.0
assert result.payload["c"] == 64032.55
def test_validate_websocket_ohlc_schema_rejects_non_mapping_root() -> None:
with pytest.raises(
CandleWebSocketSchemaError,
match=r"\$ должен быть JSON-объектом",
):
validate_dzengi_websocket_ohlc_schema([])
@pytest.mark.parametrize(
"field_name",
[
"status",
"destination",
"payload",
],
)
def test_validate_websocket_ohlc_schema_rejects_missing_root_field(
field_name: str,
) -> None:
document = _valid_document()
del document[field_name]
with pytest.raises(
CandleWebSocketSchemaError,
match=rf"\$\.{field_name} отсутствует",
):
validate_dzengi_websocket_ohlc_schema(document)
def test_validate_websocket_ohlc_schema_rejects_non_mapping_payload() -> None:
document = _valid_document()
document["payload"] = []
with pytest.raises(
CandleWebSocketSchemaError,
match=r"\$\.payload должен быть JSON-объектом",
):
validate_dzengi_websocket_ohlc_schema(document)
@pytest.mark.parametrize(
"field_name",
[
"symbol",
"interval",
"type",
"t",
"o",
"h",
"l",
"c",
],
)
def test_validate_websocket_ohlc_schema_rejects_missing_payload_field(
field_name: str,
) -> None:
document = _valid_document()
payload = document["payload"]
assert isinstance(payload, dict)
del payload[field_name]
with pytest.raises(
CandleWebSocketSchemaError,
match=rf"обязательные поля WebSocket OHLC: {field_name}",
):
validate_dzengi_websocket_ohlc_schema(document)
def test_validate_websocket_ohlc_schema_reports_all_missing_fields() -> None:
document = _valid_document()
payload = document["payload"]
assert isinstance(payload, dict)
del payload["symbol"]
del payload["t"]
del payload["c"]
with pytest.raises(
CandleWebSocketSchemaError,
match=r"symbol, t, c",
):
validate_dzengi_websocket_ohlc_schema(document)
def test_validate_websocket_ohlc_schema_does_not_validate_values() -> None:
document = _valid_document()
payload = document["payload"]
assert isinstance(payload, dict)
document["status"] = 123
document["destination"] = None
payload["symbol"] = None
payload["interval"] = []
payload["type"] = "unknown"
payload["t"] = -1
payload["o"] = "NaN"
payload["h"] = object()
payload["l"] = False
payload["c"] = None
result = validate_dzengi_websocket_ohlc_schema(document)
assert result.status == 123
assert result.destination is None
assert result.payload["type"] == "unknown"
assert result.payload["t"] == -1
def test_validate_websocket_ohlc_schema_rejects_non_string_root_key() -> None:
document = _valid_document()
document[1] = "invalid" # type: ignore[index]
with pytest.raises(
CandleWebSocketSchemaError,
match=r"\$ содержит нестроковый ключ",
):
validate_dzengi_websocket_ohlc_schema(document)
def test_validate_websocket_ohlc_schema_rejects_non_string_payload_key() -> None:
document = _valid_document()
payload = document["payload"]
assert isinstance(payload, dict)
payload[1] = "invalid" # type: ignore[index]
with pytest.raises(
CandleWebSocketSchemaError,
match=r"\$\.payload содержит нестроковый ключ",
):
validate_dzengi_websocket_ohlc_schema(document)