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

Build 060.19: implement Trade Recovery subsystem

Browse Source
This commit is contained in:
Sergey
2026-07-21 22:10:40 +03:00
parent a891302eaf
commit cf58fc223c
18 changed files with 11340 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
app/pytest.ini Normal file
View File

@@ -0,0 +1,18 @@
[pytest]
minversion = 8.0
testpaths =
tests
python_files =
test_*.py
python_classes =
Test*
python_functions =
test_*
addopts =
-ra
--strict-markers

1664
app/scripts/check_trade_backfill_api.py Normal file
View File

File diff suppressed because it is too large Load Diff

1
app/src/market_data/acquisition/adapters/dzengi/auth.py
View File

@@ -0,0 +1 @@
# app/src/market_data/acquisition/adapters/dzengi/auth.py

1
app/src/market_data/acquisition/recovery/__init__.py Normal file
View File

@@ -0,0 +1 @@
# app/src/market_data/acquisition/recovery/__init__.py

118
app/src/market_data/acquisition/recovery/trade_recovery_controller.py Normal file
View File

@@ -0,0 +1,118 @@
# app/src/market_data/acquisition/recovery/trade_recovery_controller.py
from __future__ import annotations
from src.market_data.acquisition.adapters.dzengi.rest import (
DzengiTradesDocumentSource,
)
from src.market_data.acquisition.adapters.dzengi.rest_trade_adapter import (
adapt_rest_agg_trades_document,
)
from src.market_data.acquisition.consistency.trade_stream_protocol import (
TradeStreamConsistencyProtocol,
)
from src.market_data.acquisition.models.trade import Trade
from src.market_data.acquisition.recovery.trade_recovery_normalizer import (
normalize_recovered_trades,
)
from src.market_data.acquisition.recovery.trade_recovery_protocol import (
TradeRecoveryProtocol,
)
from src.market_data.acquisition.recovery.trade_recovery_request import (
TradeRecoveryRequest,
)
from src.market_data.acquisition.recovery.trade_recovery_result import (
TradeRecoveryResult,
)
from src.market_data.acquisition.validation.schema import (
validate_rest_agg_trades_schema,
)
class TradeRecoveryController(TradeRecoveryProtocol):
"""
Контроллер восстановления пропущенных сделок через Dzengi REST API.
Контроллер выполняет одну stateless-операцию восстановления:
1. получает сырой документ aggTrades;
2. выполняет schema validation;
3. преобразует документ в канонические Trade;
4. нормализует порядок сделок;
5. пропускает сделки через общий consistency-контроллер;
6. возвращает только принятые сделки.
Контроллер не владеет состоянием согласованности потока. Для Recovery
должен передаваться тот же экземпляр TradeStreamConsistencyProtocol,
который используется основным потоком сделок.
"""
def __init__(
self,
*,
document_source: DzengiTradesDocumentSource,
consistency_controller: TradeStreamConsistencyProtocol,
) -> None:
self._document_source = document_source
self._consistency_controller = consistency_controller
def recover(
self,
request: TradeRecoveryRequest,
) -> TradeRecoveryResult:
"""
Выполнить одну операцию восстановления сделок.
Идентичные дубликаты, возвращённые consistency-контроллером
как None, не включаются в результат.
Исключения transport, schema, parsing, value validation, mapping
и consistency не перехватываются и не оборачиваются.
"""
document = self._document_source.fetch_trades_document(
request.symbol,
start_time=request.start_time,
end_time=request.end_time,
limit=request.limit,
)
validated_document = validate_rest_agg_trades_schema(
document,
)
trades = adapt_rest_agg_trades_document(
validated_document,
symbol=request.symbol,
)
normalized_trades = normalize_recovered_trades(
trades,
)
recovered_trades = self._accept_trades(
normalized_trades,
)
return TradeRecoveryResult(
symbol=request.symbol,
requested_start_time=request.start_time,
requested_end_time=request.end_time,
recovered_trades=recovered_trades,
)
def _accept_trades(
self,
trades: tuple[Trade, ...],
) -> tuple[Trade, ...]:
accepted_trades: list[Trade] = []
for trade in trades:
accepted_trade = self._consistency_controller.accept(
trade,
)
if accepted_trade is not None:
accepted_trades.append(accepted_trade)
return tuple(accepted_trades)

37
app/src/market_data/acquisition/recovery/trade_recovery_exceptions.py Normal file
View File

@@ -0,0 +1,37 @@
# app/src/market_data/acquisition/recovery/trade_recovery_exceptions.py
from __future__ import annotations
from src.market_data.acquisition.exceptions import (
MarketDataAcquisitionError,
)
class TradeRecoveryError(MarketDataAcquisitionError):
"""
Базовое исключение подсистемы восстановления сделок.
"""
class TradeRecoveryWindowError(TradeRecoveryError):
"""
Некорректный диапазон восстановления сделок.
"""
class TradeRecoveryLimitError(TradeRecoveryError):
"""
Некорректное значение параметра limit.
"""
class TradeRecoveryNormalizationError(TradeRecoveryError):
"""
Ошибка нормализации восстановленных сделок.
"""
class TradeRecoveryControllerError(TradeRecoveryError):
"""
Ошибка контроллера восстановления сделок.
"""

40
app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py Normal file
View File

@@ -0,0 +1,40 @@
# app/src/market_data/acquisition/recovery/trade_recovery_normalizer.py
from __future__ import annotations
from collections.abc import Iterable
from src.market_data.acquisition.models.trade import Trade
def normalize_recovered_trades(
trades: Iterable[Trade],
) -> tuple[Trade, ...]:
"""
Нормализовать порядок восстановленных сделок.
Сделки возвращаются в возрастающем порядке по ``trade_id``.
Исходная последовательность не изменяется.
Нормализатор намеренно не выполняет дедупликацию и не проверяет
согласованность сделок. Идентичные и конфликтующие дубликаты должны
обрабатываться экземпляром ``TradeStreamConsistencyController``.
Parameters
----------
trades
Последовательность канонических сделок.
Returns
-------
tuple[Trade, ...]
Неизменяемая последовательность сделок, отсортированная
по возрастанию ``trade_id``.
"""
return tuple(
sorted(
trades,
key=lambda trade: trade.trade_id,
)
)

55
app/src/market_data/acquisition/recovery/trade_recovery_protocol.py Normal file
View File

@@ -0,0 +1,55 @@
# app/src/market_data/acquisition/recovery/trade_recovery_protocol.py
from __future__ import annotations
from abc import abstractmethod
from typing import Protocol
from src.market_data.acquisition.recovery.trade_recovery_request import (
TradeRecoveryRequest,
)
from src.market_data.acquisition.recovery.trade_recovery_result import (
TradeRecoveryResult,
)
class TradeRecoveryProtocol(Protocol):
"""
Контракт подсистемы восстановления сделок.
Реализация должна:
- получать сделки из внешнего источника;
- нормализовать их порядок;
- выполнять согласование через
TradeStreamConsistencyController;
- возвращать канонический результат восстановления.
Реализация не должна:
- выполнять повторные попытки;
- управлять WebSocket;
- управлять Runtime;
- выполнять кэширование;
- принимать решения о реконнекте.
"""
@abstractmethod
def recover(
self,
request: TradeRecoveryRequest,
) -> TradeRecoveryResult:
"""
Выполнить одну операцию восстановления сделок.
Parameters
----------
request
Параметры операции восстановления.
Returns
-------
TradeRecoveryResult
Канонический результат восстановления.
"""
...

78
app/src/market_data/acquisition/recovery/trade_recovery_request.py Normal file
View File

@@ -0,0 +1,78 @@
# app/src/market_data/acquisition/recovery/trade_recovery_request.py
from __future__ import annotations
from dataclasses import dataclass
_MAX_RECOVERY_WINDOW_MS = 60 * 60 * 1000
_MIN_RECOVERY_LIMIT = 1
_MAX_RECOVERY_LIMIT = 1000
# Неизменяемое описание одного REST-запроса восстановления сделок.
@dataclass(frozen=True, slots=True)
class TradeRecoveryRequest:
symbol: str
start_time: int
end_time: int
limit: int | None = None
def __post_init__(self) -> None:
"""
Проверить локальные инварианты запроса восстановления.
Временные границы задаются в миллисекундах Unix time и передаются
в Dzengi REST API как параметры startTime и endTime.
"""
if not isinstance(self.symbol, str):
raise TypeError("symbol должен иметь тип str.")
if not self.symbol.strip():
raise ValueError("symbol не должен быть пустым.")
if isinstance(self.start_time, bool) or not isinstance(
self.start_time,
int,
):
raise TypeError("start_time должен иметь тип int.")
if isinstance(self.end_time, bool) or not isinstance(
self.end_time,
int,
):
raise TypeError("end_time должен иметь тип int.")
if self.start_time < 0:
raise ValueError(
"start_time не должен быть отрицательным."
)
if self.end_time < 0:
raise ValueError(
"end_time не должен быть отрицательным."
)
if self.start_time > self.end_time:
raise ValueError(
"start_time не должен быть больше end_time."
)
if self.end_time - self.start_time >= _MAX_RECOVERY_WINDOW_MS:
raise ValueError(
"Диапазон восстановления должен быть меньше одного часа."
)
if self.limit is None:
return
if isinstance(self.limit, bool) or not isinstance(self.limit, int):
raise TypeError("limit должен иметь тип int или None.")
if not _MIN_RECOVERY_LIMIT <= self.limit <= _MAX_RECOVERY_LIMIT:
raise ValueError(
"limit должен находиться в диапазоне от 1 до 1000."
)

56
app/src/market_data/acquisition/recovery/trade_recovery_result.py Normal file
View File

@@ -0,0 +1,56 @@
# app/src/market_data/acquisition/recovery/trade_recovery_result.py
from __future__ import annotations
from dataclasses import dataclass
from src.market_data.acquisition.models.trade import Trade
# Неизменяемый результат одной операции восстановления сделок.
@dataclass(frozen=True, slots=True)
class TradeRecoveryResult:
symbol: str
requested_start_time: int
requested_end_time: int
recovered_trades: tuple[Trade, ...]
@property
def recovered_count(self) -> int:
"""
Количество успешно восстановленных сделок.
"""
return len(self.recovered_trades)
@property
def is_empty(self) -> bool:
"""
Признак отсутствия восстановленных сделок.
"""
return not self.recovered_trades
@property
def first_trade(self) -> Trade | None:
"""
Первая сделка после нормализации.
"""
if not self.recovered_trades:
return None
return self.recovered_trades[0]
@property
def last_trade(self) -> Trade | None:
"""
Последняя сделка после нормализации.
"""
if not self.recovered_trades:
return None
return self.recovered_trades[-1]

1
app/src/market_data/acquisition/runtime/reconnect.py
View File

@@ -0,0 +1 @@
# app/src/market_data/acquisition/runtime/reconnect.py

1
app/src/market_data/acquisition/runtime/supervisor.py
View File

@@ -0,0 +1 @@
# app/src/market_data/acquisition/runtime/supervisor.py

575
app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py Normal file
View File

@@ -0,0 +1,575 @@
# app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_controller.py
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal
import pytest
from src.market_data.acquisition.adapters.dzengi.rest import (
DzengiTradesDocumentSource,
)
from src.market_data.acquisition.consistency.trade_stream_exceptions import (
TradeConsistencyError,
)
from src.market_data.acquisition.models.trade import (
Trade,
TradeAggressorSide,
)
from src.market_data.acquisition.recovery.trade_recovery_controller import (
TradeRecoveryController,
)
from src.market_data.acquisition.recovery.trade_recovery_request import (
TradeRecoveryRequest,
)
class StubTradesDocumentSource(
DzengiTradesDocumentSource,
):
def __init__(
self,
document: object,
) -> None:
super().__init__()
self.document = document
self.calls: list[
tuple[
str,
int | None,
int | None,
int | None,
]
] = []
def fetch_trades_document(
self,
symbol: str,
*,
start_time: int | None = None,
end_time: int | None = None,
limit: int | None = None,
) -> object:
self.calls.append(
(
symbol,
start_time,
end_time,
limit,
)
)
return self.document
class StubConsistencyController:
def __init__(self) -> None:
self.received_trades: list[Trade] = []
def accept(
self,
trade: Trade,
) -> Trade | None:
self.received_trades.append(trade)
return trade
def _request(
*,
symbol: str = "BTCUSD",
start_time: int = 1_700_000_000_000,
end_time: int = 1_700_000_001_000,
limit: int | None = 500,
) -> TradeRecoveryRequest:
return TradeRecoveryRequest(
symbol=symbol,
start_time=start_time,
end_time=end_time,
limit=limit,
)
def _raw_trade(
*,
trade_id: int,
price: str = "50000.00",
quantity: str = "0.25",
timestamp: int = 1_700_000_000_000,
buyer_is_maker: bool = False,
) -> dict[str, object]:
return {
"a": trade_id,
"p": price,
"q": quantity,
"T": timestamp,
"m": buyer_is_maker,
}
def test_requests_rest_document_with_recovery_parameters() -> None:
source = StubTradesDocumentSource(document=[])
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
controller.recover(
_request(
symbol="BTCUSD",
start_time=100,
end_time=200,
limit=250,
)
)
assert source.calls == [
(
"BTCUSD",
100,
200,
250,
)
]
def test_returns_empty_result_for_empty_document() -> None:
source = StubTradesDocumentSource(document=[])
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(
_request(),
)
assert result.symbol == "BTCUSD"
assert result.requested_start_time == 1_700_000_000_000
assert result.requested_end_time == 1_700_000_001_000
assert result.recovered_trades == ()
assert result.recovered_count == 0
assert result.is_empty is True
def test_converts_rest_document_to_canonical_trades() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(
trade_id=100,
price="50000.50",
quantity="0.125",
timestamp=1_700_000_000_123,
buyer_is_maker=False,
),
]
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(
_request(),
)
assert result.recovered_count == 1
trade = result.recovered_trades[0]
assert trade.symbol == "BTCUSD"
assert trade.trade_id == 100
assert trade.price == Decimal("50000.50")
assert trade.quantity == Decimal("0.125")
assert trade.aggressor_side is TradeAggressorSide.BUY
assert trade.source == "dzengi"
assert trade.executed_at.tzinfo is not None
def test_maps_buyer_is_maker_to_sell_aggressor_side() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(
trade_id=100,
buyer_is_maker=True,
),
]
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(
_request(),
)
assert (
result.recovered_trades[0].aggressor_side
is TradeAggressorSide.SELL
)
def test_normalizes_trades_before_consistency_check() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(trade_id=103),
_raw_trade(trade_id=100),
_raw_trade(trade_id=102),
_raw_trade(trade_id=101),
]
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
controller.recover(
_request(),
)
assert [
trade.trade_id
for trade in consistency_controller.received_trades
] == [
100,
101,
102,
103,
]
def test_returns_trades_in_normalized_order() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(trade_id=102),
_raw_trade(trade_id=100),
_raw_trade(trade_id=101),
]
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(
_request(),
)
assert tuple(
trade.trade_id
for trade in result.recovered_trades
) == (
100,
101,
102,
)
def test_excludes_duplicate_rejected_by_consistency_controller() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(trade_id=100),
_raw_trade(trade_id=101),
_raw_trade(trade_id=102),
]
)
class DuplicateRejectingConsistencyController:
def __init__(self) -> None:
self.received_trade_ids: list[int] = []
def accept(
self,
trade: Trade,
) -> Trade | None:
self.received_trade_ids.append(trade.trade_id)
if trade.trade_id == 101:
return None
return trade
consistency_controller = DuplicateRejectingConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(
_request(),
)
assert consistency_controller.received_trade_ids == [
100,
101,
102,
]
assert tuple(
trade.trade_id
for trade in result.recovered_trades
) == (
100,
102,
)
def test_preserves_consistency_controller_returned_instance() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(trade_id=100),
]
)
replacement_trade = Trade(
symbol="BTCUSD",
trade_id=100,
price=Decimal("60000.00"),
quantity=Decimal("1.00"),
executed_at=datetime(
2026,
1,
1,
12,
0,
tzinfo=timezone.utc,
),
aggressor_side=TradeAggressorSide.SELL,
source="test",
)
class ReplacingConsistencyController:
def accept(
self,
trade: Trade,
) -> Trade | None:
return replacement_trade
controller = TradeRecoveryController(
document_source=source,
consistency_controller=ReplacingConsistencyController(),
)
result = controller.recover(
_request(),
)
assert result.recovered_trades == (replacement_trade,)
assert result.recovered_trades[0] is replacement_trade
def test_propagates_source_error() -> None:
expected_error = RuntimeError("source error")
class FailingSource(
DzengiTradesDocumentSource,
):
def fetch_trades_document(
self,
symbol: str,
*,
start_time: int | None = None,
end_time: int | None = None,
limit: int | None = None,
) -> object:
raise expected_error
controller = TradeRecoveryController(
document_source=FailingSource(),
consistency_controller=StubConsistencyController(),
)
with pytest.raises(RuntimeError) as exc_info:
controller.recover(
_request(),
)
assert exc_info.value is expected_error
def test_propagates_schema_validation_error() -> None:
controller = TradeRecoveryController(
document_source=StubTradesDocumentSource(
document={
"unexpected": "document",
}
),
consistency_controller=StubConsistencyController(),
)
with pytest.raises(Exception):
controller.recover(
_request(),
)
def test_propagates_consistency_error() -> None:
expected_error = TradeConsistencyError()
class FailingConsistencyController:
def accept(
self,
trade: Trade,
) -> Trade | None:
raise expected_error
controller = TradeRecoveryController(
document_source=StubTradesDocumentSource(
document=[
_raw_trade(trade_id=100),
]
),
consistency_controller=FailingConsistencyController(),
)
with pytest.raises(TradeConsistencyError) as exc_info:
controller.recover(
_request(),
)
assert exc_info.value is expected_error
def test_does_not_call_consistency_controller_for_empty_document() -> None:
source = StubTradesDocumentSource(document=[])
class FailingIfCalledConsistencyController:
def accept(
self,
trade: Trade,
) -> Trade | None:
raise AssertionError(
"Consistency controller must not be called."
)
controller = TradeRecoveryController(
document_source=source,
consistency_controller=FailingIfCalledConsistencyController(),
)
result = controller.recover(
_request(),
)
assert result.recovered_trades == ()
def test_uses_injected_consistency_controller_instance() -> None:
source = StubTradesDocumentSource(
document=[],
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
assert (
controller._consistency_controller # type: ignore[attr-defined]
is consistency_controller
)
def test_uses_same_consistency_controller_for_all_trades() -> None:
source = StubTradesDocumentSource(
document=[
_raw_trade(trade_id=102),
_raw_trade(trade_id=100),
_raw_trade(trade_id=101),
]
)
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=source,
consistency_controller=consistency_controller,
)
result = controller.recover(
_request(),
)
assert [
trade.trade_id
for trade in consistency_controller.received_trades
] == [
100,
101,
102,
]
assert tuple(
trade.trade_id
for trade in result.recovered_trades
) == (
100,
101,
102,
)
def test_returns_recovered_trades_as_tuple() -> None:
controller = TradeRecoveryController(
document_source=StubTradesDocumentSource(
document=[
_raw_trade(trade_id=100),
]
),
consistency_controller=StubConsistencyController(),
)
result = controller.recover(
_request(),
)
assert isinstance(result.recovered_trades, tuple)
def test_empty_recovery_returns_empty_tuple() -> None:
controller = TradeRecoveryController(
document_source=StubTradesDocumentSource(
document=[],
),
consistency_controller=StubConsistencyController(),
)
result = controller.recover(
_request(),
)
assert result.recovered_trades == ()
assert isinstance(result.recovered_trades, tuple)
def test_controller_does_not_create_additional_consistency_state() -> None:
consistency_controller = StubConsistencyController()
controller = TradeRecoveryController(
document_source=StubTradesDocumentSource(
document=[],
),
consistency_controller=consistency_controller,
)
assert controller.__dict__ == {
"_document_source": controller._document_source,
"_consistency_controller": consistency_controller,
}

224
app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py Normal file
View File

@@ -0,0 +1,224 @@
# app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_normalizer.py
from __future__ import annotations
from datetime import datetime, timezone
from decimal import Decimal
from src.market_data.acquisition.models.trade import (
Trade,
TradeAggressorSide,
)
from src.market_data.acquisition.recovery.trade_recovery_normalizer import (
normalize_recovered_trades,
)
def _trade(
*,
trade_id: int,
price: Decimal = Decimal("50000.00"),
) -> Trade:
return Trade(
symbol="BTCUSD",
trade_id=trade_id,
price=price,
quantity=Decimal("0.25"),
executed_at=datetime(
2026,
1,
1,
12,
0,
tzinfo=timezone.utc,
),
aggressor_side=TradeAggressorSide.BUY,
source="dzengi",
)
def test_returns_empty_tuple_for_empty_input() -> None:
result = normalize_recovered_trades(())
assert result == ()
assert isinstance(result, tuple)
def test_keeps_already_sorted_trades() -> None:
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
third_trade = _trade(trade_id=102)
result = normalize_recovered_trades(
(
first_trade,
second_trade,
third_trade,
)
)
assert result == (
first_trade,
second_trade,
third_trade,
)
def test_sorts_reverse_order_by_trade_id() -> None:
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
third_trade = _trade(trade_id=102)
result = normalize_recovered_trades(
(
third_trade,
second_trade,
first_trade,
)
)
assert result == (
first_trade,
second_trade,
third_trade,
)
def test_sorts_arbitrary_order_by_trade_id() -> None:
trade_100 = _trade(trade_id=100)
trade_101 = _trade(trade_id=101)
trade_102 = _trade(trade_id=102)
trade_103 = _trade(trade_id=103)
result = normalize_recovered_trades(
(
trade_102,
trade_100,
trade_103,
trade_101,
)
)
assert result == (
trade_100,
trade_101,
trade_102,
trade_103,
)
def test_preserves_stable_order_for_equal_trade_ids() -> None:
first_duplicate = _trade(
trade_id=100,
price=Decimal("50000.00"),
)
second_duplicate = _trade(
trade_id=100,
price=Decimal("50001.00"),
)
result = normalize_recovered_trades(
(
first_duplicate,
second_duplicate,
)
)
assert result == (
first_duplicate,
second_duplicate,
)
def test_does_not_remove_identical_duplicates() -> None:
trade = _trade(trade_id=100)
result = normalize_recovered_trades(
(
trade,
trade,
)
)
assert result == (
trade,
trade,
)
def test_accepts_list_input() -> None:
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
result = normalize_recovered_trades(
[
second_trade,
first_trade,
]
)
assert result == (
first_trade,
second_trade,
)
def test_accepts_generator_input() -> None:
trades = (
_trade(trade_id=trade_id)
for trade_id in (
102,
100,
101,
)
)
result = normalize_recovered_trades(trades)
assert tuple(
trade.trade_id
for trade in result
) == (
100,
101,
102,
)
def test_does_not_modify_source_list() -> None:
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
source = [
second_trade,
first_trade,
]
normalize_recovered_trades(source)
assert source == [
second_trade,
first_trade,
]
def test_returns_tuple_for_non_tuple_input() -> None:
result = normalize_recovered_trades(
[
_trade(trade_id=100),
]
)
assert isinstance(result, tuple)
def test_returns_new_tuple_for_tuple_input() -> None:
source = (
_trade(trade_id=100),
_trade(trade_id=101),
)
result = normalize_recovered_trades(source)
assert result == source
assert result is not source

277
app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_request.py Normal file
View File

@@ -0,0 +1,277 @@
# app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_request.py
from __future__ import annotations
import pytest
from src.market_data.acquisition.recovery.trade_recovery_request import (
TradeRecoveryRequest,
)
def test_creates_valid_request() -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
limit=500,
)
assert request.symbol == "BTCUSD"
assert request.start_time == 1_700_000_000_000
assert request.end_time == 1_700_000_001_000
assert request.limit == 500
def test_allows_equal_start_and_end_time() -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_000_000,
)
assert request.start_time == request.end_time
def test_allows_none_limit() -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
limit=None,
)
assert request.limit is None
@pytest.mark.parametrize(
"limit",
[
1,
1000,
],
)
def test_allows_limit_boundaries(
limit: int,
) -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
limit=limit,
)
assert request.limit == limit
def test_rejects_non_string_symbol() -> None:
with pytest.raises(
TypeError,
match="symbol должен иметь тип str",
):
TradeRecoveryRequest(
symbol=123, # type: ignore[arg-type]
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
)
@pytest.mark.parametrize(
"symbol",
[
"",
" ",
],
)
def test_rejects_empty_symbol(
symbol: str,
) -> None:
with pytest.raises(
ValueError,
match="symbol не должен быть пустым",
):
TradeRecoveryRequest(
symbol=symbol,
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
)
@pytest.mark.parametrize(
"start_time",
[
1.5,
"1700000000000",
None,
True,
],
)
def test_rejects_invalid_start_time_type(
start_time: object,
) -> None:
with pytest.raises(
TypeError,
match="start_time должен иметь тип int",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=start_time, # type: ignore[arg-type]
end_time=1_700_000_001_000,
)
@pytest.mark.parametrize(
"end_time",
[
1.5,
"1700000001000",
None,
True,
],
)
def test_rejects_invalid_end_time_type(
end_time: object,
) -> None:
with pytest.raises(
TypeError,
match="end_time должен иметь тип int",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=end_time, # type: ignore[arg-type]
)
def test_rejects_negative_start_time() -> None:
with pytest.raises(
ValueError,
match="start_time не должен быть отрицательным",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=-1,
end_time=1_000,
)
def test_rejects_negative_end_time() -> None:
with pytest.raises(
ValueError,
match="end_time не должен быть отрицательным",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=0,
end_time=-1,
)
def test_rejects_start_time_greater_than_end_time() -> None:
with pytest.raises(
ValueError,
match="start_time не должен быть больше end_time",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=2_000,
end_time=1_000,
)
def test_allows_window_shorter_than_one_hour() -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=0,
end_time=3_599_999,
)
assert request.end_time - request.start_time == 3_599_999
@pytest.mark.parametrize(
"end_time",
[
3_600_000,
3_600_001,
],
)
def test_rejects_window_of_one_hour_or_more(
end_time: int,
) -> None:
with pytest.raises(
ValueError,
match="Диапазон восстановления должен быть меньше одного часа",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=0,
end_time=end_time,
)
@pytest.mark.parametrize(
"limit",
[
1.5,
"100",
True,
],
)
def test_rejects_invalid_limit_type(
limit: object,
) -> None:
with pytest.raises(
TypeError,
match="limit должен иметь тип int или None",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
limit=limit, # type: ignore[arg-type]
)
@pytest.mark.parametrize(
"limit",
[
0,
-1,
1001,
],
)
def test_rejects_limit_outside_allowed_range(
limit: int,
) -> None:
with pytest.raises(
ValueError,
match="limit должен находиться в диапазоне от 1 до 1000",
):
TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
limit=limit,
)
def test_request_is_immutable() -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
)
with pytest.raises(AttributeError):
request.symbol = "ETHUSD" # type: ignore[misc]
def test_request_uses_slots() -> None:
request = TradeRecoveryRequest(
symbol="BTCUSD",
start_time=1_700_000_000_000,
end_time=1_700_000_001_000,
)
assert not hasattr(request, "__dict__")

199
app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_result.py Normal file
View File

@@ -0,0 +1,199 @@
# app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_result.py
from __future__ import annotations
from dataclasses import FrozenInstanceError
from datetime import datetime, timezone
from decimal import Decimal
import pytest
from src.market_data.acquisition.models.trade import (
Trade,
TradeAggressorSide,
)
from src.market_data.acquisition.recovery.trade_recovery_result import (
TradeRecoveryResult,
)
def _trade(
*,
trade_id: int,
symbol: str = "BTCUSD",
) -> Trade:
return Trade(
symbol=symbol,
trade_id=trade_id,
price=Decimal("50000.00"),
quantity=Decimal("0.25"),
executed_at=datetime(
2026,
1,
1,
12,
0,
tzinfo=timezone.utc,
),
aggressor_side=TradeAggressorSide.BUY,
source="dzengi",
)
def test_creates_empty_result() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
assert result.symbol == "BTCUSD"
assert result.requested_start_time == 100
assert result.requested_end_time == 200
assert result.recovered_trades == ()
def test_empty_result_has_zero_recovered_count() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
assert result.recovered_count == 0
def test_empty_result_is_empty() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
assert result.is_empty is True
def test_empty_result_has_no_first_trade() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
assert result.first_trade is None
def test_empty_result_has_no_last_trade() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
assert result.last_trade is None
def test_non_empty_result_reports_recovered_count() -> None:
trades = (
_trade(trade_id=100),
_trade(trade_id=101),
_trade(trade_id=102),
)
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=trades,
)
assert result.recovered_count == 3
def test_non_empty_result_is_not_empty() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(
_trade(trade_id=100),
),
)
assert result.is_empty is False
def test_returns_first_recovered_trade() -> None:
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(
first_trade,
second_trade,
),
)
assert result.first_trade is first_trade
def test_returns_last_recovered_trade() -> None:
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(
first_trade,
second_trade,
),
)
assert result.last_trade is second_trade
def test_single_trade_is_both_first_and_last() -> None:
trade = _trade(trade_id=100)
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(trade,),
)
assert result.first_trade is trade
assert result.last_trade is trade
def test_result_is_frozen() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
with pytest.raises(FrozenInstanceError):
result.symbol = "ETHUSD" # type: ignore[misc]
def test_result_uses_slots() -> None:
result = TradeRecoveryResult(
symbol="BTCUSD",
requested_start_time=100,
requested_end_time=200,
recovered_trades=(),
)
assert not hasattr(result, "__dict__")