From a10a760166a6ef5c5d3f2bd746890ea764a6599b Mon Sep 17 00:00:00 2001 From: Sergey Date: Wed, 15 Jul 2026 13:23:35 +0300 Subject: [PATCH] build 046: integrate canonical candles feed into acquisition service --- app/src/market_data/acquisition/service.py | 46 +- .../market_data/acquisition/test_service.py | 386 ++++++++++++- docs/migrations/build_046.md | 515 ++++++++++++++++++ 3 files changed, 928 insertions(+), 19 deletions(-) create mode 100644 docs/migrations/build_046.md diff --git a/app/src/market_data/acquisition/service.py b/app/src/market_data/acquisition/service.py index b14cb39..012156e 100644 --- a/app/src/market_data/acquisition/service.py +++ b/app/src/market_data/acquisition/service.py @@ -2,15 +2,16 @@ from __future__ import annotations +from src.market_data.acquisition.models.candle import Candle from src.market_data.acquisition.models.instrument import Instrument from src.market_data.acquisition.models.quote import Quote from src.market_data.acquisition.registry import ( + CandlesFeedRegistry, InstrumentFeedRegistry, QuoteFeedRegistry, ) -# Application-level сервис получения справочника инструментов. class InstrumentAcquisitionService: def __init__( self, @@ -23,19 +24,11 @@ class InstrumentAcquisitionService: self, source_name: str, ) -> tuple[Instrument, ...]: - """ - Получить Instrument Feed из Registry и загрузить справочник инструментов. - - Service не создаёт Feed, не выполняет transport, parsing, validation, - mapping, retry, кэширование или преобразование результата. - """ - feed = self._registry.get(source_name) return feed.load_instruments() -# Application-level сервис получения текущих котировок. class QuoteAcquisitionService: def __init__( self, @@ -49,14 +42,33 @@ class QuoteAcquisitionService: source_name: str, symbol: str, ) -> Quote: - """ - Получить Quotes Feed из Registry и загрузить текущую котировку. - - Service не создаёт Feed, не выполняет transport, parsing, validation, - mapping, нормализацию symbol, retry, кэширование или преобразование - результата. - """ - feed = self._registry.get(source_name) return feed.load_quote(symbol) + + +class CandlesAcquisitionService: + def __init__( + self, + *, + registry: CandlesFeedRegistry, + ) -> None: + self._registry = registry + + def load_candles( + self, + source_name: str, + symbol: str, + *, + interval: str, + limit: int, + price_type: str, + ) -> tuple[Candle, ...]: + feed = self._registry.get(source_name) + + return feed.load_candles( + symbol, + interval=interval, + limit=limit, + price_type=price_type, + ) \ No newline at end of file diff --git a/app/tests/unit/market_data/acquisition/test_service.py b/app/tests/unit/market_data/acquisition/test_service.py index b1d5a3d..f529b2e 100644 --- a/app/tests/unit/market_data/acquisition/test_service.py +++ b/app/tests/unit/market_data/acquisition/test_service.py @@ -2,6 +2,7 @@ from __future__ import annotations +from datetime import datetime, timezone from decimal import Decimal import pytest @@ -11,12 +12,16 @@ from src.market_data.acquisition.exceptions import ( InstrumentReferenceMappingError, InstrumentReferenceTransportError, InstrumentReferenceValueError, + CandleFeedRegistryError, + CandleMappingError, + CandleTransportError, + CandleValueError, ) +from src.market_data.acquisition.models.candle import Candle from src.market_data.acquisition.models.instrument import Instrument from src.market_data.acquisition.protocol import InstrumentFeedProtocol from src.market_data.acquisition.registry import InstrumentFeedRegistry -from src.market_data.acquisition.service import InstrumentAcquisitionService - +from src.market_data.acquisition.service import InstrumentAcquisitionService, CandlesAcquisitionService def _instrument( *, @@ -473,3 +478,380 @@ def test_quote_service_does_not_retry_after_error() -> None: service.load_quote("dzengi", "BTC/USD_LEVERAGE") assert feed.symbols == ["BTC/USD_LEVERAGE"] + + +# --------------------------------------------------------------------------- +# Candles Acquisition Service tests +# --------------------------------------------------------------------------- + + +def _make_candle( + *, + symbol: str = "BTC/USD", + interval: str = "1m", + open_time: datetime | None = None, + open_price: str = "100", + high_price: str = "110", + low_price: str = "90", + close_price: str = "105", + volume: str = "12.5", + source: str = "dzengi", +) -> Candle: + return Candle( + symbol=symbol, + interval=interval, + open_time=open_time or datetime(2026, 1, 1, tzinfo=timezone.utc), + open_price=Decimal(open_price), + high_price=Decimal(high_price), + low_price=Decimal(low_price), + close_price=Decimal(close_price), + volume=Decimal(volume), + source=source, + ) + + +class _FakeCandlesFeed: + def __init__( + self, + *, + result: tuple[Candle, ...] = (), + error: Exception | None = None, + ) -> None: + self.result = result + self.error = error + self.calls: list[tuple[str, str, int, str]] = [] + + def load_candles( + self, + symbol: str, + *, + interval: str, + limit: int, + price_type: str, + ) -> tuple[Candle, ...]: + self.calls.append( + ( + symbol, + interval, + limit, + price_type, + ) + ) + + if self.error is not None: + raise self.error + + return self.result + + +class _FakeCandlesRegistry: + def __init__( + self, + *, + feed: _FakeCandlesFeed | None = None, + error: Exception | None = None, + ) -> None: + self.feed = feed + self.error = error + self.calls: list[str] = [] + + def get( + self, + source_name: str, + ) -> _FakeCandlesFeed: + self.calls.append(source_name) + + if self.error is not None: + raise self.error + + if self.feed is None: + raise AssertionError("Candles Feed не настроен.") + + return self.feed + + +def test_candles_acquisition_service_returns_registered_feed_result() -> None: + candles = ( + _make_candle(), + ) + feed = _FakeCandlesFeed(result=candles) + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + result = service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert result == candles + + +def test_candles_acquisition_service_preserves_result_identity() -> None: + candles = ( + _make_candle(), + ) + feed = _FakeCandlesFeed(result=candles) + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + result = service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert result is candles + + +def test_candles_acquisition_service_passes_source_name_unchanged() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + " DzEnGi ", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert registry.calls == [" DzEnGi "] + + +def test_candles_acquisition_service_calls_registry_once() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert len(registry.calls) == 1 + + +def test_candles_acquisition_service_passes_symbol_unchanged() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + "dzengi", + " btc/usd ", + interval="1m", + limit=100, + price_type="bid", + ) + + assert feed.calls == [ + ( + " btc/usd ", + "1m", + 100, + "bid", + ) + ] + + +def test_candles_acquisition_service_passes_interval_unchanged() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + "dzengi", + "BTC/USD", + interval=" 1M ", + limit=100, + price_type="bid", + ) + + assert feed.calls[0][1] == " 1M " + + +def test_candles_acquisition_service_passes_limit_unchanged() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=123, + price_type="bid", + ) + + assert feed.calls[0][2] == 123 + + +def test_candles_acquisition_service_passes_price_type_unchanged() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type=" BiD ", + ) + + assert feed.calls[0][3] == " BiD " + + +def test_candles_acquisition_service_calls_feed_once() -> None: + feed = _FakeCandlesFeed() + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert len(feed.calls) == 1 + + +def test_candles_acquisition_service_preserves_empty_tuple_identity() -> None: + candles: tuple[Candle, ...] = () + feed = _FakeCandlesFeed(result=candles) + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + result = service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert result is candles + + +def test_candles_acquisition_service_preserves_candle_order() -> None: + first = _make_candle( + open_time=datetime(2026, 1, 1, 0, 1, tzinfo=timezone.utc), + ) + second = _make_candle( + open_time=datetime(2026, 1, 1, 0, 2, tzinfo=timezone.utc), + ) + candles = ( + second, + first, + ) + feed = _FakeCandlesFeed(result=candles) + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + result = service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert result == ( + second, + first, + ) + + +def test_candles_acquisition_service_propagates_registry_error() -> None: + error = CandleFeedRegistryError("Registry error.") + registry = _FakeCandlesRegistry(error=error) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + with pytest.raises(CandleFeedRegistryError) as exc_info: + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert exc_info.value is error + + +def test_candles_acquisition_service_does_not_call_feed_after_registry_error() -> None: + feed = _FakeCandlesFeed() + error = CandleFeedRegistryError("Registry error.") + registry = _FakeCandlesRegistry( + feed=feed, + error=error, + ) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + with pytest.raises(CandleFeedRegistryError): + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert feed.calls == [] + + +@pytest.mark.parametrize( + "error", + [ + CandleTransportError("Transport error."), + CandleValueError("Value error."), + CandleMappingError("Mapping error."), + ], +) +def test_candles_acquisition_service_propagates_feed_error( + error: Exception, +) -> None: + feed = _FakeCandlesFeed(error=error) + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + with pytest.raises(type(error)) as exc_info: + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert exc_info.value is error + + +def test_candles_acquisition_service_does_not_retry_after_feed_error() -> None: + error = CandleTransportError("Transport error.") + feed = _FakeCandlesFeed(error=error) + registry = _FakeCandlesRegistry(feed=feed) + service = CandlesAcquisitionService(registry=registry) # type: ignore[arg-type] + + with pytest.raises(CandleTransportError): + service.load_candles( + "dzengi", + "BTC/USD", + interval="1m", + limit=100, + price_type="bid", + ) + + assert len(registry.calls) == 1 + assert len(feed.calls) == 1 \ No newline at end of file diff --git a/docs/migrations/build_046.md b/docs/migrations/build_046.md new file mode 100644 index 0000000..8126713 --- /dev/null +++ b/docs/migrations/build_046.md @@ -0,0 +1,515 @@ +# Build 046 — Интеграция канонического Candles Feed в сервисный слой Market Data Acquisition + +## Статус + +**Завершён** + +--- + +## Цель Build + +Интегрировать канонический `Candles Feed` в существующий сервисный слой подсистемы: + +```text +src/market_data/acquisition/ +``` + +без переключения legacy-потребителей `ExchangeService.get_klines()` и без изменения существующего поведения работающего торгового бота. + +Build продолжает поэтапную миграцию получения OHLCV-данных из legacy-подсистемы: + +```text +src/integrations/exchange/ +``` + +в каноническую подсистему: + +```text +src/market_data/acquisition/ +``` + +--- + +## Исходное состояние + +До начала Build 046 канонический OHLCV-контур уже включал: + +```text +Candle + ↓ +DzengiCandlesDocumentSource + ↓ +schema validation + ↓ +parser + ↓ +value validation + ↓ +mapper + ↓ +DzengiCandlesDocumentHandler + ↓ +CandlesFeed + ↓ +CandlesFeedRegistry +``` + +Однако сервисный слой `Market Data Acquisition` ещё не предоставлял application-level сервис для получения канонического набора свечей. + +В файле: + +```text +src/market_data/acquisition/service.py +``` + +существовали отдельные сервисы для: + +```text +InstrumentAcquisitionService +QuoteAcquisitionService +``` + +При этом `CandlesAcquisitionService` отсутствовал. + +--- + +## Принятое архитектурное решение + +В рамках Build 046 не создавался новый объединяющий класс `MarketDataAcquisitionService`. + +Существующая архитектура сервисного слоя использует отдельный application-level сервис для каждого типа рыночных данных: + +```text +InstrumentAcquisitionService +QuoteAcquisitionService +``` + +Поэтому для OHLCV-данных добавлен отдельный: + +```text +CandlesAcquisitionService +``` + +Это сохраняет существующий архитектурный паттерн и не требует изменения уже работающих контрактов Instrument Reference Data и Quotes Feed. + +--- + +## Границы Build + +В рамках Build 046 изменены только: + +```text +src/market_data/acquisition/service.py +tests/unit/market_data/acquisition/test_service.py +docs/migrations/build_046.md +``` + +Не изменялись: + +```text +src/market_data/acquisition/registry.py +src/market_data/acquisition/protocol.py +src/market_data/acquisition/feeds/candles_feed.py +src/market_data/acquisition/handlers/candles_handler.py +src/market_data/acquisition/adapters/dzengi/ +src/integrations/exchange/service.py +src/integrations/exchange/models.py +src/trading/market_analysis/ +``` + +--- + +## Реализованные изменения + +### 1. Добавлен `CandlesAcquisitionService` + +В файл: + +```text +src/market_data/acquisition/service.py +``` + +добавлен класс: + +```python +class CandlesAcquisitionService: + def __init__( + self, + *, + registry: CandlesFeedRegistry, + ) -> None: + self._registry = registry + + def load_candles( + self, + source_name: str, + symbol: str, + *, + interval: str, + limit: int, + price_type: str, + ) -> tuple[Candle, ...]: + feed = self._registry.get(source_name) + + return feed.load_candles( + symbol, + interval=interval, + limit=limit, + price_type=price_type, + ) +``` + +--- + +### 2. Добавлены зависимости сервисного слоя + +В `service.py` добавлены импорты: + +```python +from src.market_data.acquisition.models.candle import Candle +``` + +и: + +```python +from src.market_data.acquisition.registry import CandlesFeedRegistry +``` + +--- + +### 3. Зафиксирована ответственность `CandlesAcquisitionService` + +`CandlesAcquisitionService` выполняет только две операции: + +1. получает зарегистрированный `Candles Feed` по имени источника; +2. делегирует ему загрузку свечей. + +Каноническая цепочка: + +```text +source_name + ↓ +CandlesFeedRegistry.get() + ↓ +CandlesFeedProtocol + ↓ +load_candles() + ↓ +tuple[Candle, ...] +``` + +Сервис не выполняет: + +```text +нормализацию source_name +нормализацию symbol +валидацию interval +ограничение limit +нормализацию price_type +schema validation +parsing +value validation +mapping +REST-запросы +retry +кэширование +сортировку свечей +копирование результата +оборачивание исключений +``` + +Это сохраняет строгие границы ответственности между слоями. + +--- + +## Контракт `CandlesAcquisitionService` + +Метод: + +```python +load_candles( + source_name: str, + symbol: str, + *, + interval: str, + limit: int, + price_type: str, +) -> tuple[Candle, ...] +``` + +гарантирует следующую последовательность: + +```text +1. Получить Feed: + registry.get(source_name) + +2. Передать Feed исходные параметры: + symbol + interval + limit + price_type + +3. Вернуть результат Feed без преобразований. +``` + +Сервис не изменяет входные параметры. + +Сервис не изменяет порядок свечей. + +Сервис сохраняет identity возвращённого immutable `tuple`. + +Сервис не выполняет автоматический retry при ошибке. + +Исключения нижележащих слоёв пробрасываются без оборачивания. + +--- + +## Добавленные тесты + +В: + +```text +tests/unit/market_data/acquisition/test_service.py +``` + +добавлено тестовое покрытие `CandlesAcquisitionService`. + +Проверяются следующие свойства: + +1. сервис возвращает результат зарегистрированного Feed; +2. сохраняется identity возвращённого `tuple`; +3. `source_name` передаётся в registry без изменений; +4. registry вызывается ровно один раз; +5. `symbol` передаётся в Feed без изменений; +6. `interval` передаётся без изменений; +7. `limit` передаётся без изменений; +8. `price_type` передаётся без изменений; +9. Feed вызывается ровно один раз; +10. пустой `tuple` возвращается без изменения; +11. порядок свечей сохраняется; +12. `CandleFeedRegistryError` пробрасывается без оборачивания; +13. после ошибки registry Feed не вызывается; +14. ошибки Feed пробрасываются без оборачивания; +15. после ошибки Feed автоматический retry не выполняется. + +Для проверки ошибок Feed используются: + +```text +CandleTransportError +CandleValueError +CandleMappingError +``` + +--- + +## Результаты проверок + +### Компиляция + +Выполнена команда: + +```bash +python -m compileall \ + src/market_data/acquisition/service.py \ + tests/unit/market_data/acquisition/test_service.py +``` + +Результат: + +```text +Compiling 'src/market_data/acquisition/service.py'... +Compiling 'tests/unit/market_data/acquisition/test_service.py'... +``` + +Ошибок компиляции нет. + +--- + +### Targeted tests + +Выполнена команда: + +```bash +python -m pytest -q \ + tests/unit/market_data/acquisition/test_service.py +``` + +Результат: + +```text +40 passed in 0.03s +``` + +--- + +### Полный test suite + +Выполнена команда: + +```bash +python -m pytest -q +``` + +Результат: + +```text +724 passed in 2.43s +``` + +Регрессий не обнаружено. + +--- + +### Проверка Git diff + +Выполнена команда: + +```bash +git diff --check +``` + +Результат: + +```text +пустой вывод +``` + +Whitespace errors отсутствуют. + +--- + +## Архитектурный результат + +После Build 046 канонический OHLCV-контур имеет следующую структуру: + +```text +External Dzengi API + ↓ +DzengiCandlesDocumentSource + ↓ +raw transport document + ↓ +validate_candles_schema() + ↓ +parse_candles() + ↓ +validate_candles_values() + ↓ +map_candles() + ↓ +DzengiCandlesDocumentHandler + ↓ +CandlesFeed + ↓ +CandlesFeedRegistry + ↓ +CandlesAcquisitionService + ↓ +tuple[Candle, ...] +``` + +Таким образом, канонический Candles Feed теперь имеет полный application-level путь от внешнего REST API до сервисного слоя `Market Data Acquisition`. + +--- + +## Состояние legacy-контура + +В рамках Build 046 legacy-метод: + +```text +ExchangeService.get_klines() +``` + +не изменялся и не удалялся. + +Legacy-потребители: + +```text +src/trading/market_analysis/service.py +src/trading/market_analysis/htf.py +``` + +продолжают использовать существующий API. + +Это соответствует основному правилу миграции Dzentra: + +> Сначала строится и проверяется новый канонический путь, затем потребители переключаются на него поэтапно, и только после подтверждения полной совместимости удаляется legacy-код. + +--- + +## Что намеренно не выполнялось + +Build 046 не включает: + +```text +переключение ExchangeService.get_klines() +переключение trading/market_analysis +удаление legacy KlineBatch +удаление legacy Kline +удаление legacy parser +удаление legacy endpoint /api/v1/klines +регистрацию production Candles Feed на composition root +изменение runtime +изменение scheduler +кэширование свечей +хранение свечей +проверку последовательности свечей +``` + +Эти задачи должны выполняться отдельными Build с собственными границами и проверками. + +--- + +## Критерии завершения + +Build 046 считается завершённым, поскольку: + +- `CandlesAcquisitionService` реализован; +- сервис использует `CandlesFeedRegistry`; +- сервис возвращает канонический `tuple[Candle, ...]`; +- входные параметры передаются без изменений; +- результат не копируется и не сортируется; +- ошибки не оборачиваются; +- retry отсутствует; +- существующие Instrument и Quote сервисы не нарушены; +- targeted tests успешно проходят; +- полный test suite успешно проходит; +- `git diff --check` не обнаруживает ошибок; +- legacy-потребители не переключались; +- работающий торговый бот сохраняет обратную совместимость. + +--- + +## Итог + +**Build 046 завершён успешно.** + +Канонический `Candles Feed` интегрирован в сервисный слой `Market Data Acquisition`. + +Текущий завершённый путь: + +```text +Dzengi REST API + ↓ +DzengiCandlesDocumentSource + ↓ +Schema Validation + ↓ +Parser + ↓ +Value Validation + ↓ +Mapper + ↓ +DzengiCandlesDocumentHandler + ↓ +CandlesFeed + ↓ +CandlesFeedRegistry + ↓ +CandlesAcquisitionService + ↓ +tuple[Candle, ...] +``` + +Следующий Build должен продолжить миграцию OHLCV-контура без преждевременного удаления legacy-реализации и без нарушения работы существующего торгового бота. \ No newline at end of file