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

build 046: integrate canonical candles feed into acquisition service

Browse Source
This commit is contained in:
Sergey
2026-07-15 13:23:35 +03:00
parent 0e86cca19d
commit a10a760166
3 changed files with 928 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
app/src/market_data/acquisition/service.py
View File

@@ -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,
)

386
app/tests/unit/market_data/acquisition/test_service.py
View File

@@ -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

515
docs/migrations/build_046.md Normal file
View File

@@ -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-реализации и без нарушения работы существующего торгового бота.