Build 060.24: implement Runtime Recovery Architecture

This commit is contained in:
2026-07-30 00:17:12 +03:00
parent ee8765b716
commit c142145361
21 changed files with 7475 additions and 5 deletions

View File

@@ -5,11 +5,11 @@ from __future__ import annotations
from collections import deque
from dataclasses import dataclass, field
from src.market_data.acquisition.models.trade import Trade
from src.market_data.acquisition.consistency.trade_stream_exceptions import (
TradeConsistencyError,
TradeOrderingError,
)
from src.market_data.acquisition.models.trade import Trade
DEFAULT_DEDUPLICATION_WINDOW_SIZE = 10_000
@@ -25,6 +25,7 @@ class TradeStreamState:
deduplication_window_size: int = DEFAULT_DEDUPLICATION_WINDOW_SIZE
last_trade_id: int | None = None
last_trade: Trade | None = None
_trade_window: deque[int] = field(init=False, repr=False)
_trades: dict[int, Trade] = field(init=False, repr=False)
@@ -88,6 +89,7 @@ class TradeStreamState:
self._append(trade)
self.last_trade_id = trade_id
self.last_trade = trade
return trade
@@ -100,4 +102,4 @@ class TradeStreamState:
self._trades.pop(oldest_trade_id, None)
self._trade_window.append(trade.trade_id)
self._trades[trade.trade_id] = trade
self._trades[trade.trade_id] = trade

View File

@@ -0,0 +1,51 @@
# app/src/market_data/acquisition/recovery/trade_recovery_window.py
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True, slots=True)
class TradeRecoveryWindow:
"""
Неизменяемое описание одного временного диапазона
восстановления Trade Stream.
Модель не выполняет Recovery и не содержит ограничений
конкретного REST API. Допустимый максимальный размер окна
контролируется TradeRecoveryWindowPlanner.
"""
symbol: str
start_time: int
end_time: int
def __post_init__(self) -> None:
if not isinstance(self.symbol, str):
raise TypeError("symbol must be a string")
if not self.symbol.strip():
raise ValueError("symbol must not be empty")
if not isinstance(self.start_time, int) or isinstance(
self.start_time,
bool,
):
raise TypeError("start_time must be an integer")
if not isinstance(self.end_time, int) or isinstance(
self.end_time,
bool,
):
raise TypeError("end_time must be an integer")
if self.start_time < 0:
raise ValueError("start_time must not be negative")
if self.end_time < 0:
raise ValueError("end_time must not be negative")
if self.start_time > self.end_time:
raise ValueError(
"start_time must not be greater than end_time"
)

View File

@@ -0,0 +1,147 @@
# app/src/market_data/acquisition/recovery/trade_recovery_window_planner.py
from __future__ import annotations
from src.market_data.acquisition.recovery.trade_recovery_window import (
TradeRecoveryWindow,
)
MAX_TRADE_RECOVERY_REQUEST_WINDOW_MS = 3_599_999
DEFAULT_TRADE_RECOVERY_WINDOW_MS = (
MAX_TRADE_RECOVERY_REQUEST_WINDOW_MS
)
class TradeRecoveryWindowPlanner:
"""
Stateless-планировщик временных окон восстановления Trade Stream.
Planner разбивает один непрерывный временной диапазон на
последовательность окон, каждое из которых совместимо с ограничением
TradeRecoveryRequest:
end_time - start_time < 3_600_000 ms
Planner не выполняет REST-запросы и не создаёт TradeRecoveryRequest.
"""
__slots__ = ("_max_window_ms",)
def __init__(
self,
*,
max_window_ms: int = DEFAULT_TRADE_RECOVERY_WINDOW_MS,
) -> None:
if not isinstance(max_window_ms, int) or isinstance(
max_window_ms,
bool,
):
raise TypeError("max_window_ms must be an integer")
if max_window_ms <= 0:
raise ValueError("max_window_ms must be positive")
if max_window_ms > MAX_TRADE_RECOVERY_REQUEST_WINDOW_MS:
raise ValueError(
"max_window_ms must not exceed "
f"{MAX_TRADE_RECOVERY_REQUEST_WINDOW_MS}"
)
self._max_window_ms = max_window_ms
@property
def max_window_ms(self) -> int:
"""
Максимальная длительность одного создаваемого окна.
"""
return self._max_window_ms
def build_windows(
self,
*,
symbol: str,
start_time: int,
end_time: int,
) -> tuple[TradeRecoveryWindow, ...]:
"""
Построить последовательность Recovery Window.
Если start_time равен end_time, восстановление не требуется
и возвращается пустой tuple.
Соседние окна имеют общую границу:
previous.end_time == next.start_time
Возможный повтор Trade на границе должен быть устранён
существующим Consistency Layer.
"""
self._validate_range(
symbol=symbol,
start_time=start_time,
end_time=end_time,
)
if start_time == end_time:
return ()
windows: list[TradeRecoveryWindow] = []
cursor = start_time
while cursor < end_time:
window_end = min(
cursor + self._max_window_ms,
end_time,
)
windows.append(
TradeRecoveryWindow(
symbol=symbol,
start_time=cursor,
end_time=window_end,
)
)
cursor = window_end
return tuple(windows)
@staticmethod
def _validate_range(
*,
symbol: str,
start_time: int,
end_time: int,
) -> None:
"""
Проверить исходный диапазон до начала разбиения.
"""
if not isinstance(symbol, str):
raise TypeError("symbol must be a string")
if not symbol.strip():
raise ValueError("symbol must not be empty")
if not isinstance(start_time, int) or isinstance(
start_time,
bool,
):
raise TypeError("start_time must be an integer")
if not isinstance(end_time, int) or isinstance(
end_time,
bool,
):
raise TypeError("end_time must be an integer")
if start_time < 0:
raise ValueError("start_time must not be negative")
if end_time < 0:
raise ValueError("end_time must not be negative")
if start_time > end_time:
raise ValueError(
"start_time must not be greater than end_time"
)

View File

@@ -0,0 +1,192 @@
# app/src/market_data/acquisition/runtime/heartbeat.py
from __future__ import annotations
import time
from enum import Enum
from typing import Callable, Protocol, runtime_checkable
from src.market_data.acquisition.runtime.runtime_events import (
HeartbeatTimeoutEvent,
)
from src.market_data.acquisition.runtime.websocket_protocol import (
AcquisitionRuntimeEventPublisherProtocol,
)
class HeartbeatState(str, Enum):
"""
Текущее состояние Heartbeat Monitor.
"""
IDLE = "idle"
MONITORING = "monitoring"
TIMED_OUT = "timed_out"
@runtime_checkable
class HeartbeatMonitorProtocol(Protocol):
"""
Контракт пассивного контроля активности Acquisition Runtime.
"""
@property
def state(self) -> HeartbeatState:
"""Вернуть текущее состояние Heartbeat Monitor."""
...
@property
def timeout_seconds(self) -> float:
"""Вернуть настроенный интервал timeout."""
...
@property
def last_activity_at(self) -> float | None:
"""Вернуть monotonic-время последней активности."""
...
def start(self) -> None:
"""Начать мониторинг активности."""
...
def stop(self) -> None:
"""Остановить мониторинг и очистить текущее состояние."""
...
def record_activity(self) -> None:
"""Зафиксировать текущий момент как последнюю активность."""
...
async def check_timeout(self) -> bool:
"""
Проверить наступление timeout.
Вернуть True только при первом обнаружении timeout.
"""
...
class HeartbeatMonitor:
"""
Пассивный монитор активности Acquisition Runtime.
Monitor:
- хранит monotonic-время последней активности;
- детерминированно проверяет превышение timeout;
- однократно публикует HeartbeatTimeoutEvent;
- не запускает собственные фоновые задачи;
- не выполняет reconnect;
- не управляет WebSocket lifecycle.
Периодический вызов check_timeout() выполняет Runtime Scheduler.
"""
__slots__ = (
"_event_publisher",
"_timeout_seconds",
"_clock",
"_state",
"_last_activity_at",
)
def __init__(
self,
event_publisher: AcquisitionRuntimeEventPublisherProtocol,
*,
timeout_seconds: float,
clock: Callable[[], float] = time.monotonic,
) -> None:
if not isinstance(timeout_seconds, (int, float)) or isinstance(
timeout_seconds,
bool,
):
raise TypeError(
"timeout_seconds must be an integer or float"
)
if timeout_seconds <= 0:
raise ValueError(
"timeout_seconds must be positive"
)
if not callable(clock):
raise TypeError("clock must be callable")
self._event_publisher = event_publisher
self._timeout_seconds = float(timeout_seconds)
self._clock = clock
self._state = HeartbeatState.IDLE
self._last_activity_at: float | None = None
@property
def state(self) -> HeartbeatState:
"""Вернуть текущее состояние Heartbeat Monitor."""
return self._state
@property
def timeout_seconds(self) -> float:
"""Вернуть настроенный интервал timeout."""
return self._timeout_seconds
@property
def last_activity_at(self) -> float | None:
"""Вернуть monotonic-время последней активности."""
return self._last_activity_at
def start(self) -> None:
"""
Начать мониторинг с текущего monotonic-времени.
"""
self._last_activity_at = self._clock()
self._state = HeartbeatState.MONITORING
def stop(self) -> None:
"""
Остановить мониторинг и удалить runtime-состояние Heartbeat.
"""
self._last_activity_at = None
self._state = HeartbeatState.IDLE
def record_activity(self) -> None:
"""
Зафиксировать текущий момент как последнюю активность.
Получение новой активности также переводит Monitor из состояния
TIMED_OUT обратно в MONITORING.
"""
self._last_activity_at = self._clock()
self._state = HeartbeatState.MONITORING
async def check_timeout(self) -> bool:
"""
Проверить, истёк ли настроенный интервал активности.
HeartbeatTimeoutEvent публикуется только один раз для каждого
периода отсутствия активности.
После новой активности Monitor снова может зафиксировать timeout.
"""
if self._state is not HeartbeatState.MONITORING:
return False
last_activity_at = self._last_activity_at
if last_activity_at is None:
return False
elapsed_seconds = self._clock() - last_activity_at
if elapsed_seconds < self._timeout_seconds:
return False
self._state = HeartbeatState.TIMED_OUT
await self._event_publisher.publish(
HeartbeatTimeoutEvent(
timeout_seconds=self._timeout_seconds,
)
)
return True

View File

@@ -1 +1,186 @@
# app/src/market_data/acquisition/runtime/reconnect.py
# app/src/market_data/acquisition/runtime/reconnect.py
from __future__ import annotations
from enum import Enum
from typing import Protocol, runtime_checkable
from src.market_data.acquisition.runtime.runtime_commands import (
ConnectCommand,
)
from src.market_data.acquisition.runtime.runtime_events import (
ReconnectCompletedEvent,
ReconnectFailedEvent,
ReconnectStartedEvent,
)
from src.market_data.acquisition.runtime.websocket_protocol import (
AcquisitionRuntimeCommandDispatcherProtocol,
AcquisitionRuntimeEventPublisherProtocol,
WebSocketSubscriptionManagerProtocol,
)
class ReconnectState(str, Enum):
"""
Текущее состояние ReconnectCoordinator.
Состояния описывают только lifecycle одной инфраструктурной
операции reconnect.
Состояния Heartbeat, Supervisor и Trade Recovery принадлежат
соответствующим специализированным компонентам.
"""
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
RESTORING_SUBSCRIPTIONS = "restoring_subscriptions"
CONNECTED = "connected"
FAILED = "failed"
@runtime_checkable
class ReconnectCoordinatorProtocol(Protocol):
"""
Контракт координатора повторного подключения Acquisition Runtime.
"""
@property
def state(self) -> ReconnectState:
"""
Вернуть текущее состояние reconnect lifecycle.
"""
...
@property
def attempt(self) -> int:
"""
Вернуть номер последней начатой попытки reconnect.
"""
...
async def reconnect(self) -> None:
"""
Выполнить одну попытку повторного подключения.
"""
...
class ReconnectCoordinator:
"""
Координатор одной попытки повторного подключения.
Coordinator:
- увеличивает номер попытки;
- публикует ReconnectStartedEvent;
- передаёт ConnectCommand в Runtime Dispatcher;
- восстанавливает зарегистрированные подписки;
- публикует ReconnectCompletedEvent;
- при ошибке публикует ReconnectFailedEvent;
- сохраняет текущее состояние lifecycle.
Coordinator не выполняет:
- retry loop;
- backoff;
- scheduling;
- heartbeat monitoring;
- Trade Recovery;
- обработку рыночных сообщений.
"""
__slots__ = (
"_command_dispatcher",
"_subscription_manager",
"_event_publisher",
"_state",
"_attempt",
)
def __init__(
self,
command_dispatcher: AcquisitionRuntimeCommandDispatcherProtocol,
subscription_manager: WebSocketSubscriptionManagerProtocol,
event_publisher: AcquisitionRuntimeEventPublisherProtocol,
) -> None:
self._command_dispatcher = command_dispatcher
self._subscription_manager = subscription_manager
self._event_publisher = event_publisher
self._state = ReconnectState.DISCONNECTED
self._attempt = 0
@property
def state(self) -> ReconnectState:
"""
Вернуть текущее состояние reconnect lifecycle.
"""
return self._state
@property
def attempt(self) -> int:
"""
Вернуть номер последней начатой попытки reconnect.
"""
return self._attempt
async def reconnect(self) -> None:
"""
Выполнить одну попытку повторного подключения.
Последовательность:
ReconnectStartedEvent
ConnectCommand
restore_subscriptions()
ReconnectCompletedEvent
Raises:
Exception:
Любая ошибка Runtime Dispatcher или Subscription Manager
распространяется вызывающему компоненту без обёртки.
"""
self._attempt += 1
current_attempt = self._attempt
self._state = ReconnectState.CONNECTING
await self._event_publisher.publish(
ReconnectStartedEvent(
attempt=current_attempt,
)
)
try:
await self._command_dispatcher.dispatch(
ConnectCommand()
)
self._state = (
ReconnectState.RESTORING_SUBSCRIPTIONS
)
await self._subscription_manager.restore_subscriptions()
except Exception as error:
self._state = ReconnectState.FAILED
await self._event_publisher.publish(
ReconnectFailedEvent(
attempt=current_attempt,
reason=str(error),
)
)
raise
self._state = ReconnectState.CONNECTED
await self._event_publisher.publish(
ReconnectCompletedEvent(
attempt=current_attempt,
)
)

View File

@@ -0,0 +1,295 @@
# app/src/market_data/acquisition/runtime/
# runtime_recovery_coordinator.py
from __future__ import annotations
from datetime import datetime, timezone
from src.market_data.acquisition.consistency.trade_stream_state_store_exceptions import (
TradeStreamStateNotFoundError,
)
from src.market_data.acquisition.consistency.trade_stream_state_store_protocol import (
TradeStreamStateStoreProtocol,
)
from src.market_data.acquisition.models.trade import Trade
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.recovery.trade_recovery_window_planner import (
TradeRecoveryWindowPlanner,
)
class RuntimeRecoveryCoordinator:
"""
Runtime-координатор восстановления Trade Stream.
Coordinator связывает существующие компоненты:
- TradeStreamStateStoreProtocol;
- checkpoint в TradeStreamState.last_trade;
- TradeRecoveryWindowPlanner;
- TradeRecoveryProtocol.
Последовательность восстановления:
state_store.get(symbol)
TradeStreamState.last_trade
executed_at → Unix milliseconds
TradeRecoveryWindowPlanner.build_windows()
TradeRecoveryRequest для каждого окна
TradeRecoveryProtocol.recover()
объединённый TradeRecoveryResult
Coordinator не выполняет:
- WebSocket connect или disconnect;
- reconnect;
- восстановление подписок;
- heartbeat monitoring;
- scheduling;
- retry или backoff;
- публикацию Runtime Events;
- создание нового TradeStreamState;
- обработку и проверку согласованности Trade.
Отсутствие состояния или последней принятой сделки является
штатной ситуацией: восстановление пропускается и возвращается
пустой TradeRecoveryResult.
Все остальные ошибки State Store, Window Planner и Trade Recovery
распространяются вызывающему коду без обёртки.
"""
__slots__ = (
"_state_store",
"_window_planner",
"_recovery_controller",
)
def __init__(
self,
*,
state_store: TradeStreamStateStoreProtocol,
window_planner: TradeRecoveryWindowPlanner,
recovery_controller: TradeRecoveryProtocol,
) -> None:
"""
Создать Runtime Recovery Coordinator.
Args:
state_store:
Общее хранилище состояния Trade Stream Consistency.
Должен передаваться тот же экземпляр хранилища,
который используется основным WebSocket-потоком.
window_planner:
Планировщик допустимых временных окон восстановления.
recovery_controller:
Существующий stateless-контроллер Trade Recovery.
"""
self._state_store = state_store
self._window_planner = window_planner
self._recovery_controller = recovery_controller
def recover(
self,
*,
symbol: str,
recovery_end_time: int,
) -> TradeRecoveryResult:
"""
Восстановить пропущенные сделки одного инструмента.
Начальная граница определяется по времени последней сделки,
принятой общим Trade Stream Consistency Layer.
Конечная граница передаётся вызывающим Runtime-компонентом.
Coordinator не получает системное время самостоятельно.
Возможный повтор последней принятой сделки на левой границе
безопасно устраняется существующим Consistency Layer.
Args:
symbol:
Символ торгового инструмента.
recovery_end_time:
Конечная граница восстановления в миллисекундах
Unix time.
Returns:
Объединённый TradeRecoveryResult для полного диапазона.
Если состояние или checkpoint отсутствуют, возвращается
пустой результат с границами, равными recovery_end_time.
Raises:
TypeError:
Если symbol или recovery_end_time имеют неверный тип.
ValueError:
Если symbol пустой, recovery_end_time отрицательный,
checkpoint не содержит timezone-aware datetime либо
конечная граница расположена раньше checkpoint.
Exception:
Ошибки Window Planner и Trade Recovery Controller
распространяются без изменения.
"""
self._validate_input(
symbol=symbol,
recovery_end_time=recovery_end_time,
)
try:
state = self._state_store.get(symbol)
except TradeStreamStateNotFoundError:
return self._build_empty_result(
symbol=symbol,
boundary_time=recovery_end_time,
)
checkpoint_trade = state.last_trade
if checkpoint_trade is None:
return self._build_empty_result(
symbol=symbol,
boundary_time=recovery_end_time,
)
recovery_start_time = self._datetime_to_unix_ms(
checkpoint_trade.executed_at,
)
windows = self._window_planner.build_windows(
symbol=symbol,
start_time=recovery_start_time,
end_time=recovery_end_time,
)
recovered_trades: list[Trade] = []
for window in windows:
window_result = self._recovery_controller.recover(
TradeRecoveryRequest(
symbol=window.symbol,
start_time=window.start_time,
end_time=window.end_time,
)
)
recovered_trades.extend(
window_result.recovered_trades,
)
return TradeRecoveryResult(
symbol=symbol,
requested_start_time=recovery_start_time,
requested_end_time=recovery_end_time,
recovered_trades=tuple(recovered_trades),
)
@staticmethod
def _validate_input(
*,
symbol: str,
recovery_end_time: int,
) -> None:
"""
Проверить входные параметры до обращения к State Store.
"""
if not isinstance(symbol, str):
raise TypeError(
"symbol должен иметь тип str."
)
if not symbol.strip():
raise ValueError(
"symbol не должен быть пустым."
)
if isinstance(recovery_end_time, bool) or not isinstance(
recovery_end_time,
int,
):
raise TypeError(
"recovery_end_time должен иметь тип int."
)
if recovery_end_time < 0:
raise ValueError(
"recovery_end_time не должен быть отрицательным."
)
@staticmethod
def _datetime_to_unix_ms(
value: datetime,
) -> int:
"""
Преобразовать timezone-aware datetime в Unix milliseconds.
Расчёт выполняется относительно UTC epoch без использования
float, чтобы не вносить погрешность преобразования timestamp.
"""
if not isinstance(value, datetime):
raise TypeError(
"checkpoint executed_at должен иметь тип datetime."
)
if value.tzinfo is None or value.utcoffset() is None:
raise ValueError(
"checkpoint executed_at должен содержать timezone."
)
utc_value = value.astimezone(
timezone.utc,
)
epoch = datetime(
1970,
1,
1,
tzinfo=timezone.utc,
)
delta = utc_value - epoch
return (
delta.days * 86_400_000
+ delta.seconds * 1_000
+ delta.microseconds // 1_000
)
@staticmethod
def _build_empty_result(
*,
symbol: str,
boundary_time: int,
) -> TradeRecoveryResult:
"""
Создать пустой результат при отсутствии Recovery Checkpoint.
Равные границы явно обозначают отсутствие диапазона,
который можно восстановить на основе существующего состояния.
"""
return TradeRecoveryResult(
symbol=symbol,
requested_start_time=boundary_time,
requested_end_time=boundary_time,
recovered_trades=(),
)

View File

@@ -0,0 +1,72 @@
# app/src/market_data/acquisition/runtime/runtime_recovery_protocol.py
"""
Публичный контракт Runtime Recovery Coordinator.
Runtime Recovery Coordinator связывает Runtime Layer с существующим
Trade Recovery Layer, но не выполняет reconnect, WebSocket-операции,
heartbeat, scheduling или восстановление подписок.
"""
from __future__ import annotations
from typing import Protocol, runtime_checkable
from src.market_data.acquisition.recovery.trade_recovery_result import (
TradeRecoveryResult,
)
@runtime_checkable
class RuntimeRecoveryProtocol(Protocol):
"""
Контракт Runtime-координатора восстановления Trade Stream.
Реализация должна:
- получить существующий checkpoint торгового инструмента;
- определить полный временной диапазон восстановления;
- разбить диапазон на допустимые Recovery Window;
- последовательно выполнить Trade Recovery для каждого окна;
- вернуть объединённый TradeRecoveryResult.
Реализация не должна:
- создавать новое состояние Trade Stream;
- выполнять WebSocket connect или disconnect;
- восстанавливать WebSocket-подписки;
- управлять Heartbeat Monitor;
- управлять Runtime Supervisor;
- запускать Scheduler;
- выполнять retry или backoff;
- скрывать ошибки нижележащих компонентов.
"""
def recover(
self,
*,
symbol: str,
recovery_end_time: int,
) -> TradeRecoveryResult:
"""
Восстановить пропущенные сделки одного торгового инструмента.
Args:
symbol:
Символ торгового инструмента.
recovery_end_time:
Правая граница восстанавливаемого диапазона
в миллисекундах Unix time.
Значение определяется вызывающим Runtime-компонентом.
Coordinator самостоятельно текущее время не вычисляет.
Returns:
Объединённый результат восстановления всего диапазона.
Если существующее состояние или checkpoint отсутствуют,
реализация возвращает пустой TradeRecoveryResult и не создаёт
новое состояние Trade Stream.
"""
...

View File

@@ -0,0 +1,199 @@
# app/src/market_data/acquisition/runtime/scheduler.py
from __future__ import annotations
import asyncio
from collections.abc import Awaitable, Callable
from typing import Protocol, runtime_checkable
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatMonitorProtocol,
)
from src.market_data.acquisition.runtime.supervisor import (
RuntimeSupervisorProtocol,
)
RuntimeSleep = Callable[[float], Awaitable[None]]
@runtime_checkable
class RuntimeSchedulerProtocol(Protocol):
"""
Контракт периодического планировщика Acquisition Runtime.
"""
@property
def running(self) -> bool:
"""
Вернуть признак активного scheduler loop.
"""
...
@property
def interval_seconds(self) -> float:
"""
Вернуть интервал между периодическими проверками.
"""
...
async def start(self) -> None:
"""
Запустить scheduler loop.
Метод выполняется до вызова stop() либо возникновения ошибки.
"""
...
def stop(self) -> None:
"""
Запросить остановку scheduler loop.
"""
...
async def run_once(self) -> bool:
"""
Выполнить одну проверку Heartbeat.
Вернуть True, если был обнаружен timeout.
"""
...
class RuntimeScheduler:
"""
Периодический планировщик Acquisition Runtime.
Scheduler отвечает только за время выполнения проверок:
HeartbeatMonitor.check_timeout()
├── False
│ └── ожидание следующего интервала
└── True
└── RuntimeSupervisor
.handle_heartbeat_timeout()
Scheduler не выполняет:
- reconnect;
- Heartbeat calculations;
- управление WebSocket;
- Trade Recovery;
- backoff;
- retry policy;
- публикацию Runtime Events.
Ошибки Heartbeat Monitor, Runtime Supervisor и sleep-функции
распространяются вызывающему коду без обёртки.
"""
__slots__ = (
"_heartbeat_monitor",
"_runtime_supervisor",
"_interval_seconds",
"_sleep",
"_running",
)
def __init__(
self,
heartbeat_monitor: HeartbeatMonitorProtocol,
runtime_supervisor: RuntimeSupervisorProtocol,
*,
interval_seconds: float,
sleep: RuntimeSleep = asyncio.sleep,
) -> None:
if not isinstance(interval_seconds, (int, float)) or isinstance(
interval_seconds,
bool,
):
raise TypeError(
"interval_seconds must be an integer or float"
)
if interval_seconds <= 0:
raise ValueError(
"interval_seconds must be positive"
)
if not callable(sleep):
raise TypeError("sleep must be callable")
self._heartbeat_monitor = heartbeat_monitor
self._runtime_supervisor = runtime_supervisor
self._interval_seconds = float(interval_seconds)
self._sleep = sleep
self._running = False
@property
def running(self) -> bool:
"""
Вернуть признак активного scheduler loop.
"""
return self._running
@property
def interval_seconds(self) -> float:
"""
Вернуть интервал между проверками.
"""
return self._interval_seconds
async def start(self) -> None:
"""
Запустить периодический scheduler loop.
Если Scheduler уже запущен, повторный вызов немедленно
завершается и не создаёт второй параллельный цикл.
Scheduler всегда сбрасывает running в False:
- после stop();
- после ошибки;
- после отмены внешней asyncio-задачи.
"""
if self._running:
return
self._running = True
try:
while self._running:
await self.run_once()
if not self._running:
break
await self._sleep(
self._interval_seconds,
)
finally:
self._running = False
def stop(self) -> None:
"""
Запросить завершение scheduler loop.
Метод идемпотентен. Он не отменяет внешнюю asyncio-задачу,
а останавливает цикл на ближайшей управляемой границе.
"""
self._running = False
async def run_once(self) -> bool:
"""
Выполнить одну проверку Heartbeat.
Если Heartbeat Monitor подтверждает timeout, Scheduler
передаёт его Runtime Supervisor.
Возвращаемое значение отражает только результат проверки
Heartbeat Monitor.
"""
timed_out = await self._heartbeat_monitor.check_timeout()
if timed_out:
await self._runtime_supervisor.handle_heartbeat_timeout()
return timed_out

View File

@@ -1 +1,203 @@
# app/src/market_data/acquisition/runtime/supervisor.py
# app/src/market_data/acquisition/runtime/supervisor.py
from __future__ import annotations
from enum import Enum
from typing import Protocol, runtime_checkable
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatMonitorProtocol,
)
from src.market_data.acquisition.runtime.reconnect import (
ReconnectCoordinatorProtocol,
)
class RuntimeSupervisorState(str, Enum):
"""
Текущее состояние Runtime Supervisor.
"""
STOPPED = "stopped"
RUNNING = "running"
RECONNECTING = "reconnecting"
FAILED = "failed"
@runtime_checkable
class RuntimeSupervisorProtocol(Protocol):
"""
Контракт координатора lifecycle Acquisition Runtime.
"""
@property
def state(self) -> RuntimeSupervisorState:
"""
Вернуть текущее состояние Runtime Supervisor.
"""
...
def start(self) -> None:
"""
Запустить supervision и Heartbeat Monitor.
"""
...
def stop(self) -> None:
"""
Остановить supervision и Heartbeat Monitor.
"""
...
def notify_activity(self) -> None:
"""
Передать Heartbeat Monitor информацию о Runtime activity.
"""
...
async def handle_heartbeat_timeout(self) -> bool:
"""
Обработать подтверждённый Heartbeat timeout.
Вернуть True, если была выполнена попытка reconnect.
"""
...
class RuntimeSupervisor:
"""
Координатор lifecycle Acquisition Runtime.
Supervisor объединяет:
- HeartbeatMonitor;
- ReconnectCoordinator;
- общее состояние Runtime Session.
Supervisor:
- запускает и останавливает Heartbeat Monitor;
- принимает уведомления о Runtime activity;
- запускает одну попытку reconnect после Heartbeat timeout;
- после успешного reconnect начинает новый период Heartbeat;
- фиксирует состояния STOPPED, RUNNING, RECONNECTING и FAILED;
- предотвращает параллельный reconnect.
Supervisor не выполняет:
- периодические проверки Heartbeat;
- retry loop;
- backoff;
- scheduling;
- Trade Recovery;
- управление WebSocket Transport;
- публикацию Runtime Events.
Периодический вызов Heartbeat и передача timeout в Supervisor
будут ответственностью Runtime Scheduler.
"""
__slots__ = (
"_heartbeat_monitor",
"_reconnect_coordinator",
"_state",
)
def __init__(
self,
heartbeat_monitor: HeartbeatMonitorProtocol,
reconnect_coordinator: ReconnectCoordinatorProtocol,
) -> None:
self._heartbeat_monitor = heartbeat_monitor
self._reconnect_coordinator = reconnect_coordinator
self._state = RuntimeSupervisorState.STOPPED
@property
def state(self) -> RuntimeSupervisorState:
"""
Вернуть текущее состояние Runtime Supervisor.
"""
return self._state
def start(self) -> None:
"""
Запустить supervision.
Повторный вызов start() начинает новый Heartbeat monitoring
period и сохраняет Supervisor в состоянии RUNNING.
"""
self._heartbeat_monitor.start()
self._state = RuntimeSupervisorState.RUNNING
def stop(self) -> None:
"""
Остановить supervision.
Повторный вызов stop() безопасен и сохраняет состояние STOPPED.
"""
self._heartbeat_monitor.stop()
self._state = RuntimeSupervisorState.STOPPED
def notify_activity(self) -> None:
"""
Передать информацию о Runtime activity в Heartbeat Monitor.
Активность учитывается только во время RUNNING lifecycle.
События, полученные после stop() либо во время reconnect,
не должны неявно запускать Heartbeat Monitor.
"""
if self._state is not RuntimeSupervisorState.RUNNING:
return
self._heartbeat_monitor.record_activity()
async def handle_heartbeat_timeout(self) -> bool:
"""
Обработать подтверждённый Heartbeat timeout.
Последовательность:
RUNNING
stop Heartbeat
RECONNECTING
ReconnectCoordinator.reconnect()
├── success:
│ start Heartbeat
│ RUNNING
└── failure:
FAILED
exception propagates
Если Supervisor не находится в состоянии RUNNING,
новая попытка reconnect не запускается.
Returns:
bool:
True, если reconnect был выполнен успешно.
False, если Supervisor не находился в RUNNING.
Raises:
Exception:
Исходная ошибка ReconnectCoordinator распространяется
вызывающему компоненту без обёртки.
"""
if self._state is not RuntimeSupervisorState.RUNNING:
return False
self._heartbeat_monitor.stop()
self._state = RuntimeSupervisorState.RECONNECTING
try:
await self._reconnect_coordinator.reconnect()
except Exception:
self._state = RuntimeSupervisorState.FAILED
raise
self._heartbeat_monitor.start()
self._state = RuntimeSupervisorState.RUNNING
return True

View File

@@ -0,0 +1,184 @@
# app/src/market_data/acquisition/
# trade_stream_runtime_composition.py
from __future__ import annotations
import asyncio
import time
from collections.abc import Callable
from dataclasses import dataclass
from src.market_data.acquisition.adapters.dzengi.rest import (
DzengiTradesDocumentSource,
)
from src.market_data.acquisition.consistency.trade_stream_consistency_controller import (
TradeStreamConsistencyController,
)
from src.market_data.acquisition.consistency.trade_stream_state_store import (
TradeStreamStateStore,
)
from src.market_data.acquisition.recovery.trade_recovery_controller import (
TradeRecoveryController,
)
from src.market_data.acquisition.recovery.trade_recovery_window_planner import (
DEFAULT_TRADE_RECOVERY_WINDOW_MS,
TradeRecoveryWindowPlanner,
)
from src.market_data.acquisition.runtime.acquisition_runtime_service import (
AcquisitionRuntimeService,
)
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatMonitor,
)
from src.market_data.acquisition.runtime.reconnect import (
ReconnectCoordinator,
)
from src.market_data.acquisition.runtime.runtime_recovery_coordinator import (
RuntimeRecoveryCoordinator,
)
from src.market_data.acquisition.runtime.scheduler import (
RuntimeScheduler,
RuntimeSleep,
)
from src.market_data.acquisition.runtime.supervisor import (
RuntimeSupervisor,
)
from src.market_data.acquisition.runtime.websocket_protocol import (
AcquisitionRuntimeEventPublisherProtocol,
WebSocketSessionProtocol,
WebSocketSubscriptionManagerProtocol,
WebSocketTransportProtocol,
)
from src.market_data.acquisition.trade_stream_acquisition_service import (
TradeStreamAcquisitionService,
)
from src.market_data.acquisition.trade_stream_message_adapter_protocol import (
TradeStreamMessageAdapterProtocol,
)
@dataclass(frozen=True, slots=True)
class TradeStreamRuntimeComposition:
"""
Неизменяемый результат композиции Trade Stream Runtime.
Объект предоставляет явно построенный граф зависимостей и не является
Registry либо Service Locator. Все компоненты создаются factory-функцией
ровно один раз и повторно используют общие stateful-зависимости.
Создание Composition не запускает WebSocket, Heartbeat, Supervisor
или Scheduler и не выполняет Recovery.
"""
state_store: TradeStreamStateStore
consistency_controller: TradeStreamConsistencyController
recovery_controller: TradeRecoveryController
recovery_window_planner: TradeRecoveryWindowPlanner
runtime_recovery_coordinator: RuntimeRecoveryCoordinator
acquisition_runtime_service: AcquisitionRuntimeService
trade_stream_acquisition_service: TradeStreamAcquisitionService
reconnect_coordinator: ReconnectCoordinator
heartbeat_monitor: HeartbeatMonitor
runtime_supervisor: RuntimeSupervisor
runtime_scheduler: RuntimeScheduler
def build_trade_stream_runtime_composition(
*,
session: WebSocketSessionProtocol,
transport: WebSocketTransportProtocol,
subscription_manager: WebSocketSubscriptionManagerProtocol,
event_publisher: AcquisitionRuntimeEventPublisherProtocol,
message_adapter: TradeStreamMessageAdapterProtocol,
recovery_document_source: DzengiTradesDocumentSource,
heartbeat_timeout_seconds: float,
scheduler_interval_seconds: float,
max_recovery_window_ms: int = DEFAULT_TRADE_RECOVERY_WINDOW_MS,
heartbeat_clock: Callable[[], float] = time.monotonic,
scheduler_sleep: RuntimeSleep = asyncio.sleep,
) -> TradeStreamRuntimeComposition:
"""
Построить изолированный граф зависимостей Trade Stream Runtime.
Factory создаёт единые экземпляры TradeStreamStateStore и
TradeStreamConsistencyController. Благодаря этому Live Stream
и Recovery используют один checkpoint и одинаковые правила
согласованности сделок.
Внешние WebSocket-зависимости передаются через Protocol-контракты.
Factory не создаёт production transport, не читает Settings,
не запускает lifecycle и не создаёт фоновые asyncio-задачи.
"""
state_store = TradeStreamStateStore()
consistency_controller = TradeStreamConsistencyController(
state_store,
)
recovery_controller = TradeRecoveryController(
document_source=recovery_document_source,
consistency_controller=consistency_controller,
)
recovery_window_planner = TradeRecoveryWindowPlanner(
max_window_ms=max_recovery_window_ms,
)
runtime_recovery_coordinator = RuntimeRecoveryCoordinator(
state_store=state_store,
window_planner=recovery_window_planner,
recovery_controller=recovery_controller,
)
acquisition_runtime_service = AcquisitionRuntimeService(
session=session,
transport=transport,
subscription_manager=subscription_manager,
event_publisher=event_publisher,
)
trade_stream_acquisition_service = TradeStreamAcquisitionService(
runtime_service=acquisition_runtime_service,
adapter=message_adapter,
consistency_controller=consistency_controller,
)
reconnect_coordinator = ReconnectCoordinator(
command_dispatcher=acquisition_runtime_service,
subscription_manager=subscription_manager,
event_publisher=event_publisher,
)
heartbeat_monitor = HeartbeatMonitor(
event_publisher=event_publisher,
timeout_seconds=heartbeat_timeout_seconds,
clock=heartbeat_clock,
)
runtime_supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat_monitor,
reconnect_coordinator=reconnect_coordinator,
)
runtime_scheduler = RuntimeScheduler(
heartbeat_monitor=heartbeat_monitor,
runtime_supervisor=runtime_supervisor,
interval_seconds=scheduler_interval_seconds,
sleep=scheduler_sleep,
)
return TradeStreamRuntimeComposition(
state_store=state_store,
consistency_controller=consistency_controller,
recovery_controller=recovery_controller,
recovery_window_planner=recovery_window_planner,
runtime_recovery_coordinator=runtime_recovery_coordinator,
acquisition_runtime_service=acquisition_runtime_service,
trade_stream_acquisition_service=trade_stream_acquisition_service,
reconnect_coordinator=reconnect_coordinator,
heartbeat_monitor=heartbeat_monitor,
runtime_supervisor=runtime_supervisor,
runtime_scheduler=runtime_scheduler,
)

View File

@@ -1,3 +1,5 @@
# app/tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py
from __future__ import annotations
from datetime import datetime, timezone
@@ -53,6 +55,13 @@ def test_trade_stream_state_uses_slots() -> None:
assert not hasattr(state, "__dict__")
def test_new_state_has_no_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
assert state.last_trade_id is None
assert state.last_trade is None
def test_accepts_first_trade() -> None:
state = TradeStreamState(symbol="BTCUSD")
trade = _trade()
@@ -61,6 +70,17 @@ def test_accepts_first_trade() -> None:
assert result == trade
assert state.last_trade_id == trade.trade_id
assert state.last_trade is trade
def test_first_accepted_trade_becomes_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
trade = _trade()
state.accept(trade)
assert state.last_trade is trade
assert state.last_trade_id == trade.trade_id
def test_accepts_trade_with_greater_trade_id() -> None:
@@ -73,6 +93,19 @@ def test_accepts_trade_with_greater_trade_id() -> None:
assert result == second_trade
assert state.last_trade_id == second_trade.trade_id
assert state.last_trade is second_trade
def test_next_accepted_trade_replaces_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
state.accept(first_trade)
state.accept(second_trade)
assert state.last_trade is second_trade
assert state.last_trade_id == second_trade.trade_id
def test_accepts_trade_with_gap() -> None:
@@ -85,6 +118,17 @@ def test_accepts_trade_with_gap() -> None:
assert result == trade_after_gap
assert state.last_trade_id == trade_after_gap.trade_id
assert state.last_trade is trade_after_gap
def test_checkpoint_preserves_trade_identity() -> None:
state = TradeStreamState(symbol="BTCUSD")
trade = _trade()
result = state.accept(trade)
assert result is trade
assert state.last_trade is trade
def test_returns_none_for_identical_duplicate() -> None:
@@ -96,6 +140,22 @@ def test_returns_none_for_identical_duplicate() -> None:
assert result is None
assert state.last_trade_id == trade.trade_id
assert state.last_trade is trade
def test_identical_duplicate_does_not_change_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
original_trade = _trade()
state.accept(original_trade)
duplicate_trade = _trade()
result = state.accept(duplicate_trade)
assert result is None
assert state.last_trade is original_trade
assert state.last_trade_id == original_trade.trade_id
def test_raises_consistency_error_for_conflicting_duplicate() -> None:
@@ -115,6 +175,26 @@ def test_raises_consistency_error_for_conflicting_duplicate() -> None:
state.accept(conflicting_trade)
def test_conflicting_duplicate_does_not_change_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
original_trade = _trade(
trade_id=100,
price=Decimal("50000.00"),
)
conflicting_trade = _trade(
trade_id=100,
price=Decimal("50001.00"),
)
state.accept(original_trade)
with pytest.raises(TradeConsistencyError):
state.accept(conflicting_trade)
assert state.last_trade is original_trade
assert state.last_trade_id == original_trade.trade_id
def test_raises_ordering_error_for_older_trade() -> None:
state = TradeStreamState(symbol="BTCUSD")
current_trade = _trade(trade_id=100)
@@ -126,6 +206,20 @@ def test_raises_ordering_error_for_older_trade() -> None:
state.accept(older_trade)
def test_older_trade_does_not_change_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
current_trade = _trade(trade_id=100)
older_trade = _trade(trade_id=99)
state.accept(current_trade)
with pytest.raises(TradeOrderingError):
state.accept(older_trade)
assert state.last_trade is current_trade
assert state.last_trade_id == current_trade.trade_id
def test_returns_none_for_duplicate_still_inside_window() -> None:
state = TradeStreamState(
symbol="BTCUSD",
@@ -143,6 +237,29 @@ def test_returns_none_for_duplicate_still_inside_window() -> None:
assert result is None
assert state.last_trade_id == third_trade.trade_id
assert state.last_trade is third_trade
def test_duplicate_inside_window_does_not_change_checkpoint() -> None:
state = TradeStreamState(
symbol="BTCUSD",
deduplication_window_size=3,
)
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
third_trade = _trade(trade_id=102)
state.accept(first_trade)
state.accept(second_trade)
state.accept(third_trade)
duplicate_trade = _trade(trade_id=100)
result = state.accept(duplicate_trade)
assert result is None
assert state.last_trade is third_trade
assert state.last_trade_id == third_trade.trade_id
def test_raises_ordering_error_after_trade_leaves_window() -> None:
@@ -162,6 +279,26 @@ def test_raises_ordering_error_after_trade_leaves_window() -> None:
state.accept(first_trade)
def test_ordering_error_after_window_does_not_change_checkpoint() -> None:
state = TradeStreamState(
symbol="BTCUSD",
deduplication_window_size=2,
)
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=101)
third_trade = _trade(trade_id=102)
state.accept(first_trade)
state.accept(second_trade)
state.accept(third_trade)
with pytest.raises(TradeOrderingError):
state.accept(first_trade)
assert state.last_trade is third_trade
assert state.last_trade_id == third_trade.trade_id
def test_rejects_unexpected_symbol() -> None:
state = TradeStreamState(symbol="BTCUSD")
trade = _trade(symbol="ETHUSD")
@@ -170,6 +307,36 @@ def test_rejects_unexpected_symbol() -> None:
state.accept(trade)
def test_unexpected_symbol_does_not_change_checkpoint() -> None:
state = TradeStreamState(symbol="BTCUSD")
accepted_trade = _trade(trade_id=100)
state.accept(accepted_trade)
unexpected_trade = _trade(
symbol="ETHUSD",
trade_id=101,
)
with pytest.raises(ValueError):
state.accept(unexpected_trade)
assert state.last_trade is accepted_trade
assert state.last_trade_id == accepted_trade.trade_id
def test_checkpoint_trade_id_matches_last_trade_id() -> None:
state = TradeStreamState(symbol="BTCUSD")
first_trade = _trade(trade_id=100)
second_trade = _trade(trade_id=105)
state.accept(first_trade)
state.accept(second_trade)
assert state.last_trade is not None
assert state.last_trade.trade_id == state.last_trade_id
def test_rejects_empty_symbol() -> None:
with pytest.raises(ValueError):
TradeStreamState(symbol="")
@@ -189,4 +356,4 @@ def test_rejects_non_positive_window_size(
TradeStreamState(
symbol="BTCUSD",
deduplication_window_size=window_size,
)
)

View File

@@ -0,0 +1,158 @@
# app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_window.py
from __future__ import annotations
from dataclasses import FrozenInstanceError
import pytest
from src.market_data.acquisition.recovery.trade_recovery_window import (
TradeRecoveryWindow,
)
def test_creates_valid_window() -> None:
window = TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1_700_000_000_000,
end_time=1_700_000_100_000,
)
assert window.symbol == "BTCUSDT"
assert window.start_time == 1_700_000_000_000
assert window.end_time == 1_700_000_100_000
def test_allows_equal_start_and_end_time() -> None:
window = TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1_700_000_000_000,
end_time=1_700_000_000_000,
)
assert window.start_time == window.end_time
@pytest.mark.parametrize(
"symbol",
[
"",
" ",
],
)
def test_rejects_empty_symbol(
symbol: str,
) -> None:
with pytest.raises(ValueError):
TradeRecoveryWindow(
symbol=symbol,
start_time=1,
end_time=2,
)
@pytest.mark.parametrize(
"symbol",
[
None,
123,
True,
],
)
def test_rejects_non_string_symbol(
symbol: object,
) -> None:
with pytest.raises(TypeError):
TradeRecoveryWindow(
symbol=symbol, # type: ignore[arg-type]
start_time=1,
end_time=2,
)
@pytest.mark.parametrize(
"start_time",
[
1.5,
"1000",
None,
True,
],
)
def test_rejects_invalid_start_time_type(
start_time: object,
) -> None:
with pytest.raises(TypeError):
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=start_time, # type: ignore[arg-type]
end_time=2,
)
@pytest.mark.parametrize(
"end_time",
[
1.5,
"1000",
None,
True,
],
)
def test_rejects_invalid_end_time_type(
end_time: object,
) -> None:
with pytest.raises(TypeError):
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1,
end_time=end_time, # type: ignore[arg-type]
)
def test_rejects_negative_start_time() -> None:
with pytest.raises(ValueError):
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=-1,
end_time=2,
)
def test_rejects_negative_end_time() -> None:
with pytest.raises(ValueError):
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1,
end_time=-1,
)
def test_rejects_start_time_greater_than_end_time() -> None:
with pytest.raises(ValueError):
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=2,
end_time=1,
)
def test_window_is_immutable() -> None:
window = TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1,
end_time=2,
)
with pytest.raises(FrozenInstanceError):
window.end_time = 3 # type: ignore[misc]
def test_window_uses_slots() -> None:
window = TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1,
end_time=2,
)
assert not hasattr(window, "__dict__")

View File

@@ -0,0 +1,380 @@
# app/tests/unit/market_data/acquisition/recovery/test_trade_recovery_window_planner.py
from __future__ import annotations
import pytest
from src.market_data.acquisition.recovery.trade_recovery_window import (
TradeRecoveryWindow,
)
from src.market_data.acquisition.recovery.trade_recovery_window_planner import (
DEFAULT_TRADE_RECOVERY_WINDOW_MS,
MAX_TRADE_RECOVERY_REQUEST_WINDOW_MS,
TradeRecoveryWindowPlanner,
)
def test_planner_uses_slots() -> None:
planner = TradeRecoveryWindowPlanner()
assert not hasattr(planner, "__dict__")
def test_uses_default_max_window() -> None:
planner = TradeRecoveryWindowPlanner()
assert (
planner.max_window_ms
== DEFAULT_TRADE_RECOVERY_WINDOW_MS
)
def test_accepts_custom_max_window() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
assert planner.max_window_ms == 1_000
@pytest.mark.parametrize(
"max_window_ms",
[
1.5,
"1000",
None,
True,
],
)
def test_rejects_invalid_max_window_type(
max_window_ms: object,
) -> None:
with pytest.raises(TypeError):
TradeRecoveryWindowPlanner(
max_window_ms=max_window_ms, # type: ignore[arg-type]
)
@pytest.mark.parametrize(
"max_window_ms",
[
0,
-1,
],
)
def test_rejects_non_positive_max_window(
max_window_ms: int,
) -> None:
with pytest.raises(ValueError):
TradeRecoveryWindowPlanner(
max_window_ms=max_window_ms,
)
def test_rejects_window_larger_than_request_limit() -> None:
with pytest.raises(ValueError):
TradeRecoveryWindowPlanner(
max_window_ms=(
MAX_TRADE_RECOVERY_REQUEST_WINDOW_MS + 1
),
)
def test_returns_empty_tuple_for_equal_boundaries() -> None:
planner = TradeRecoveryWindowPlanner()
result = planner.build_windows(
symbol="BTCUSDT",
start_time=1_000,
end_time=1_000,
)
assert result == ()
assert isinstance(result, tuple)
def test_builds_single_window_for_short_range() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=1_000,
end_time=1_500,
)
assert result == (
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1_000,
end_time=1_500,
),
)
def test_builds_single_window_at_exact_configured_maximum() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=1_000,
end_time=2_000,
)
assert result == (
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1_000,
end_time=2_000,
),
)
def test_splits_range_into_multiple_windows() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=1_000,
end_time=3_500,
)
assert result == (
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=1_000,
end_time=2_000,
),
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=2_000,
end_time=3_000,
),
TradeRecoveryWindow(
symbol="BTCUSDT",
start_time=3_000,
end_time=3_500,
),
)
def test_returns_tuple_for_multiple_windows() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=0,
end_time=2_500,
)
assert isinstance(result, tuple)
def test_preserves_symbol_in_every_window() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="ETHUSDT",
start_time=0,
end_time=2_500,
)
assert result
assert all(
window.symbol == "ETHUSDT"
for window in result
)
def test_windows_are_contiguous() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=0,
end_time=3_500,
)
for previous, current in zip(
result,
result[1:],
strict=False,
):
assert previous.end_time == current.start_time
def test_windows_cover_complete_range() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=500,
end_time=3_750,
)
assert result[0].start_time == 500
assert result[-1].end_time == 3_750
def test_no_window_exceeds_configured_size() -> None:
planner = TradeRecoveryWindowPlanner(
max_window_ms=1_000,
)
result = planner.build_windows(
symbol="BTCUSDT",
start_time=0,
end_time=4_500,
)
assert all(
window.end_time - window.start_time <= 1_000
for window in result
)
def test_default_windows_satisfy_recovery_request_limit() -> None:
planner = TradeRecoveryWindowPlanner()
result = planner.build_windows(
symbol="BTCUSDT",
start_time=0,
end_time=7_500_000,
)
assert result
assert all(
window.end_time - window.start_time < 3_600_000
for window in result
)
@pytest.mark.parametrize(
"symbol",
[
"",
" ",
],
)
def test_rejects_empty_symbol(
symbol: str,
) -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(ValueError):
planner.build_windows(
symbol=symbol,
start_time=1,
end_time=2,
)
@pytest.mark.parametrize(
"symbol",
[
None,
123,
True,
],
)
def test_rejects_non_string_symbol(
symbol: object,
) -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(TypeError):
planner.build_windows(
symbol=symbol, # type: ignore[arg-type]
start_time=1,
end_time=2,
)
@pytest.mark.parametrize(
"start_time",
[
1.5,
"1000",
None,
True,
],
)
def test_rejects_invalid_start_time_type(
start_time: object,
) -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(TypeError):
planner.build_windows(
symbol="BTCUSDT",
start_time=start_time, # type: ignore[arg-type]
end_time=2,
)
@pytest.mark.parametrize(
"end_time",
[
1.5,
"1000",
None,
True,
],
)
def test_rejects_invalid_end_time_type(
end_time: object,
) -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(TypeError):
planner.build_windows(
symbol="BTCUSDT",
start_time=1,
end_time=end_time, # type: ignore[arg-type]
)
def test_rejects_negative_start_time() -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(ValueError):
planner.build_windows(
symbol="BTCUSDT",
start_time=-1,
end_time=2,
)
def test_rejects_negative_end_time() -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(ValueError):
planner.build_windows(
symbol="BTCUSDT",
start_time=1,
end_time=-1,
)
def test_rejects_start_time_greater_than_end_time() -> None:
planner = TradeRecoveryWindowPlanner()
with pytest.raises(ValueError):
planner.build_windows(
symbol="BTCUSDT",
start_time=2,
end_time=1,
)

View File

@@ -0,0 +1,466 @@
# app/tests/unit/market_data/acquisition/runtime/test_heartbeat_monitor.py
from __future__ import annotations
import asyncio
from typing import Any
import pytest
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatMonitor,
HeartbeatMonitorProtocol,
HeartbeatState,
)
from src.market_data.acquisition.runtime.runtime_events import (
HeartbeatTimeoutEvent,
)
class FakeClock:
def __init__(
self,
initial_value: float = 0.0,
) -> None:
self.value = initial_value
def __call__(self) -> float:
return self.value
def advance(
self,
seconds: float,
) -> None:
self.value += seconds
class FakeEventPublisher:
def __init__(self) -> None:
self.events: list[Any] = []
async def publish(
self,
event: Any,
) -> None:
self.events.append(event)
def create_monitor(
*,
timeout_seconds: float = 10.0,
initial_clock_value: float = 0.0,
) -> tuple[
HeartbeatMonitor,
FakeClock,
FakeEventPublisher,
]:
clock = FakeClock(
initial_value=initial_clock_value,
)
publisher = FakeEventPublisher()
monitor = HeartbeatMonitor(
event_publisher=publisher,
timeout_seconds=timeout_seconds,
clock=clock,
)
return (
monitor,
clock,
publisher,
)
def test_monitor_implements_protocol() -> None:
monitor, *_ = create_monitor()
assert isinstance(
monitor,
HeartbeatMonitorProtocol,
)
def test_monitor_uses_slots() -> None:
monitor, *_ = create_monitor()
assert not hasattr(monitor, "__dict__")
def test_initial_state_is_idle() -> None:
monitor, *_ = create_monitor()
assert monitor.state is HeartbeatState.IDLE
assert monitor.last_activity_at is None
def test_exposes_timeout_seconds() -> None:
monitor, *_ = create_monitor(
timeout_seconds=15.5,
)
assert monitor.timeout_seconds == 15.5
@pytest.mark.parametrize(
"timeout_seconds",
[
1.5,
10,
],
)
def test_accepts_positive_timeout(
timeout_seconds: float,
) -> None:
monitor, *_ = create_monitor(
timeout_seconds=timeout_seconds,
)
assert monitor.timeout_seconds == float(timeout_seconds)
@pytest.mark.parametrize(
"timeout_seconds",
[
"10",
None,
True,
],
)
def test_rejects_invalid_timeout_type(
timeout_seconds: object,
) -> None:
with pytest.raises(TypeError):
HeartbeatMonitor(
event_publisher=FakeEventPublisher(),
timeout_seconds=timeout_seconds, # type: ignore[arg-type]
clock=FakeClock(),
)
@pytest.mark.parametrize(
"timeout_seconds",
[
0,
-1,
-0.5,
],
)
def test_rejects_non_positive_timeout(
timeout_seconds: float,
) -> None:
with pytest.raises(ValueError):
HeartbeatMonitor(
event_publisher=FakeEventPublisher(),
timeout_seconds=timeout_seconds,
clock=FakeClock(),
)
def test_rejects_non_callable_clock() -> None:
with pytest.raises(TypeError):
HeartbeatMonitor(
event_publisher=FakeEventPublisher(),
timeout_seconds=10.0,
clock=object(), # type: ignore[arg-type]
)
def test_start_begins_monitoring() -> None:
monitor, clock, _ = create_monitor(
initial_clock_value=100.0,
)
monitor.start()
assert monitor.state is HeartbeatState.MONITORING
assert monitor.last_activity_at == 100.0
def test_stop_returns_monitor_to_idle() -> None:
monitor, *_ = create_monitor()
monitor.start()
monitor.stop()
assert monitor.state is HeartbeatState.IDLE
assert monitor.last_activity_at is None
def test_record_activity_starts_monitoring() -> None:
monitor, clock, _ = create_monitor(
initial_clock_value=50.0,
)
monitor.record_activity()
assert monitor.state is HeartbeatState.MONITORING
assert monitor.last_activity_at == 50.0
def test_record_activity_updates_last_activity_time() -> None:
monitor, clock, _ = create_monitor()
monitor.record_activity()
clock.advance(3.5)
monitor.record_activity()
assert monitor.last_activity_at == 3.5
def test_check_timeout_returns_false_while_idle() -> None:
monitor, _, publisher = create_monitor()
result = asyncio.run(
monitor.check_timeout()
)
assert result is False
assert publisher.events == []
assert monitor.state is HeartbeatState.IDLE
def test_check_timeout_returns_false_before_threshold() -> None:
monitor, clock, publisher = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(9.999)
result = asyncio.run(
monitor.check_timeout()
)
assert result is False
assert publisher.events == []
assert monitor.state is HeartbeatState.MONITORING
def test_check_timeout_triggers_at_exact_threshold() -> None:
monitor, clock, publisher = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(10.0)
result = asyncio.run(
monitor.check_timeout()
)
assert result is True
assert monitor.state is HeartbeatState.TIMED_OUT
assert publisher.events == [
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
]
def test_check_timeout_triggers_after_threshold() -> None:
monitor, clock, publisher = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(15.0)
result = asyncio.run(
monitor.check_timeout()
)
assert result is True
assert publisher.events == [
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
]
def test_timeout_event_is_published_only_once_per_inactivity_period() -> None:
monitor, clock, publisher = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(10.0)
first_result = asyncio.run(
monitor.check_timeout()
)
clock.advance(5.0)
second_result = asyncio.run(
monitor.check_timeout()
)
assert first_result is True
assert second_result is False
assert publisher.events == [
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
]
def test_record_activity_resets_timed_out_state() -> None:
monitor, clock, _ = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(10.0)
asyncio.run(
monitor.check_timeout()
)
clock.advance(1.0)
monitor.record_activity()
assert monitor.state is HeartbeatState.MONITORING
assert monitor.last_activity_at == 11.0
def test_new_activity_allows_future_timeout_event() -> None:
monitor, clock, publisher = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(10.0)
first_result = asyncio.run(
monitor.check_timeout()
)
clock.advance(1.0)
monitor.record_activity()
clock.advance(10.0)
second_result = asyncio.run(
monitor.check_timeout()
)
assert first_result is True
assert second_result is True
assert publisher.events == [
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
]
def test_stop_after_timeout_clears_runtime_state() -> None:
monitor, clock, _ = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(10.0)
asyncio.run(
monitor.check_timeout()
)
monitor.stop()
assert monitor.state is HeartbeatState.IDLE
assert monitor.last_activity_at is None
def test_start_after_timeout_begins_new_monitoring_period() -> None:
monitor, clock, publisher = create_monitor(
timeout_seconds=10.0,
)
monitor.start()
clock.advance(10.0)
asyncio.run(
monitor.check_timeout()
)
clock.advance(5.0)
monitor.start()
assert monitor.state is HeartbeatState.MONITORING
assert monitor.last_activity_at == 15.0
clock.advance(10.0)
result = asyncio.run(
monitor.check_timeout()
)
assert result is True
assert publisher.events == [
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
HeartbeatTimeoutEvent(
timeout_seconds=10.0,
),
]
def test_event_publisher_error_is_propagated() -> None:
class BrokenEventPublisher(FakeEventPublisher):
async def publish(
self,
event: Any,
) -> None:
raise RuntimeError("publish failed")
clock = FakeClock()
monitor = HeartbeatMonitor(
event_publisher=BrokenEventPublisher(),
timeout_seconds=10.0,
clock=clock,
)
monitor.start()
clock.advance(10.0)
with pytest.raises(
RuntimeError,
match="publish failed",
):
asyncio.run(
monitor.check_timeout()
)
def test_publisher_error_leaves_monitor_timed_out() -> None:
class BrokenEventPublisher(FakeEventPublisher):
async def publish(
self,
event: Any,
) -> None:
raise RuntimeError("publish failed")
clock = FakeClock()
monitor = HeartbeatMonitor(
event_publisher=BrokenEventPublisher(),
timeout_seconds=10.0,
clock=clock,
)
monitor.start()
clock.advance(10.0)
with pytest.raises(RuntimeError):
asyncio.run(
monitor.check_timeout()
)
assert monitor.state is HeartbeatState.TIMED_OUT

View File

@@ -0,0 +1,349 @@
# app/tests/unit/market_data/acquisition/runtime/test_reconnect_coordinator.py
from __future__ import annotations
import asyncio
from typing import Any
import pytest
from src.market_data.acquisition.runtime.reconnect import (
ReconnectCoordinator,
ReconnectCoordinatorProtocol,
ReconnectState,
)
from src.market_data.acquisition.runtime.runtime_commands import (
ConnectCommand,
)
from src.market_data.acquisition.runtime.runtime_events import (
ReconnectCompletedEvent,
ReconnectFailedEvent,
ReconnectStartedEvent,
)
class FakeCommandDispatcher:
def __init__(self) -> None:
self.commands: list[Any] = []
async def dispatch(
self,
command: Any,
) -> None:
self.commands.append(command)
class FakeSubscriptionManager:
def __init__(self) -> None:
self.restore_calls = 0
async def subscribe(
self,
subscription_key: str,
message: Any,
) -> None:
return None
async def unsubscribe(
self,
subscription_key: str,
message: Any,
) -> None:
return None
async def restore_subscriptions(self) -> None:
self.restore_calls += 1
async def clear_subscriptions(self) -> None:
return None
class FakeEventPublisher:
def __init__(self) -> None:
self.events: list[Any] = []
async def publish(
self,
event: Any,
) -> None:
self.events.append(event)
def create_coordinator() -> tuple[
ReconnectCoordinator,
FakeCommandDispatcher,
FakeSubscriptionManager,
FakeEventPublisher,
]:
dispatcher = FakeCommandDispatcher()
subscriptions = FakeSubscriptionManager()
publisher = FakeEventPublisher()
coordinator = ReconnectCoordinator(
command_dispatcher=dispatcher,
subscription_manager=subscriptions,
event_publisher=publisher,
)
return (
coordinator,
dispatcher,
subscriptions,
publisher,
)
def test_coordinator_implements_protocol() -> None:
coordinator, *_ = create_coordinator()
assert isinstance(
coordinator,
ReconnectCoordinatorProtocol,
)
def test_coordinator_uses_slots() -> None:
coordinator, *_ = create_coordinator()
assert not hasattr(coordinator, "__dict__")
def test_initial_state_is_disconnected() -> None:
coordinator, *_ = create_coordinator()
assert coordinator.state is ReconnectState.DISCONNECTED
assert coordinator.attempt == 0
def test_reconnect_dispatches_connect_command() -> None:
coordinator, dispatcher, *_ = create_coordinator()
asyncio.run(coordinator.reconnect())
assert len(dispatcher.commands) == 1
assert isinstance(
dispatcher.commands[0],
ConnectCommand,
)
def test_reconnect_restores_subscriptions() -> None:
coordinator, _, subscriptions, _ = create_coordinator()
asyncio.run(coordinator.reconnect())
assert subscriptions.restore_calls == 1
def test_successful_reconnect_publishes_lifecycle_events() -> None:
coordinator, _, _, publisher = create_coordinator()
asyncio.run(coordinator.reconnect())
assert publisher.events == [
ReconnectStartedEvent(attempt=1),
ReconnectCompletedEvent(attempt=1),
]
def test_successful_reconnect_sets_connected_state() -> None:
coordinator, *_ = create_coordinator()
asyncio.run(coordinator.reconnect())
assert coordinator.state is ReconnectState.CONNECTED
assert coordinator.attempt == 1
def test_attempt_increments_for_each_reconnect() -> None:
coordinator, *_ = create_coordinator()
asyncio.run(coordinator.reconnect())
asyncio.run(coordinator.reconnect())
assert coordinator.attempt == 2
def test_second_reconnect_uses_next_attempt_number() -> None:
coordinator, _, _, publisher = create_coordinator()
asyncio.run(coordinator.reconnect())
asyncio.run(coordinator.reconnect())
assert publisher.events == [
ReconnectStartedEvent(attempt=1),
ReconnectCompletedEvent(attempt=1),
ReconnectStartedEvent(attempt=2),
ReconnectCompletedEvent(attempt=2),
]
def test_connect_error_publishes_failed_event() -> None:
class BrokenCommandDispatcher(FakeCommandDispatcher):
async def dispatch(
self,
command: Any,
) -> None:
self.commands.append(command)
raise RuntimeError("connection failed")
dispatcher = BrokenCommandDispatcher()
subscriptions = FakeSubscriptionManager()
publisher = FakeEventPublisher()
coordinator = ReconnectCoordinator(
command_dispatcher=dispatcher,
subscription_manager=subscriptions,
event_publisher=publisher,
)
with pytest.raises(
RuntimeError,
match="connection failed",
):
asyncio.run(coordinator.reconnect())
assert publisher.events == [
ReconnectStartedEvent(attempt=1),
ReconnectFailedEvent(
attempt=1,
reason="connection failed",
),
]
def test_connect_error_sets_failed_state() -> None:
class BrokenCommandDispatcher(FakeCommandDispatcher):
async def dispatch(
self,
command: Any,
) -> None:
raise RuntimeError("connection failed")
coordinator = ReconnectCoordinator(
command_dispatcher=BrokenCommandDispatcher(),
subscription_manager=FakeSubscriptionManager(),
event_publisher=FakeEventPublisher(),
)
with pytest.raises(RuntimeError):
asyncio.run(coordinator.reconnect())
assert coordinator.state is ReconnectState.FAILED
assert coordinator.attempt == 1
def test_connect_error_does_not_restore_subscriptions() -> None:
class BrokenCommandDispatcher(FakeCommandDispatcher):
async def dispatch(
self,
command: Any,
) -> None:
raise RuntimeError("connection failed")
subscriptions = FakeSubscriptionManager()
coordinator = ReconnectCoordinator(
command_dispatcher=BrokenCommandDispatcher(),
subscription_manager=subscriptions,
event_publisher=FakeEventPublisher(),
)
with pytest.raises(RuntimeError):
asyncio.run(coordinator.reconnect())
assert subscriptions.restore_calls == 0
def test_subscription_restore_error_publishes_failed_event() -> None:
class BrokenSubscriptionManager(FakeSubscriptionManager):
async def restore_subscriptions(self) -> None:
self.restore_calls += 1
raise RuntimeError("restore failed")
dispatcher = FakeCommandDispatcher()
subscriptions = BrokenSubscriptionManager()
publisher = FakeEventPublisher()
coordinator = ReconnectCoordinator(
command_dispatcher=dispatcher,
subscription_manager=subscriptions,
event_publisher=publisher,
)
with pytest.raises(
RuntimeError,
match="restore failed",
):
asyncio.run(coordinator.reconnect())
assert publisher.events == [
ReconnectStartedEvent(attempt=1),
ReconnectFailedEvent(
attempt=1,
reason="restore failed",
),
]
def test_subscription_restore_error_sets_failed_state() -> None:
class BrokenSubscriptionManager(FakeSubscriptionManager):
async def restore_subscriptions(self) -> None:
raise RuntimeError("restore failed")
coordinator = ReconnectCoordinator(
command_dispatcher=FakeCommandDispatcher(),
subscription_manager=BrokenSubscriptionManager(),
event_publisher=FakeEventPublisher(),
)
with pytest.raises(RuntimeError):
asyncio.run(coordinator.reconnect())
assert coordinator.state is ReconnectState.FAILED
def test_failed_attempt_can_be_retried() -> None:
class FailOnceCommandDispatcher(FakeCommandDispatcher):
def __init__(self) -> None:
super().__init__()
self.calls = 0
async def dispatch(
self,
command: Any,
) -> None:
self.calls += 1
self.commands.append(command)
if self.calls == 1:
raise RuntimeError("temporary failure")
dispatcher = FailOnceCommandDispatcher()
subscriptions = FakeSubscriptionManager()
publisher = FakeEventPublisher()
coordinator = ReconnectCoordinator(
command_dispatcher=dispatcher,
subscription_manager=subscriptions,
event_publisher=publisher,
)
with pytest.raises(RuntimeError):
asyncio.run(coordinator.reconnect())
asyncio.run(coordinator.reconnect())
assert coordinator.attempt == 2
assert coordinator.state is ReconnectState.CONNECTED
assert subscriptions.restore_calls == 1
assert publisher.events == [
ReconnectStartedEvent(attempt=1),
ReconnectFailedEvent(
attempt=1,
reason="temporary failure",
),
ReconnectStartedEvent(attempt=2),
ReconnectCompletedEvent(attempt=2),
]

View File

@@ -0,0 +1,989 @@
# app/tests/unit/market_data/acquisition/runtime/
# test_runtime_recovery_coordinator.py
from __future__ import annotations
from datetime import datetime, timedelta, timezone
from decimal import Decimal
from typing import Any
import pytest
from src.market_data.acquisition.consistency.trade_stream_state import (
TradeStreamState,
)
from src.market_data.acquisition.consistency.trade_stream_state_store_exceptions import (
TradeStreamStateNotFoundError,
)
from src.market_data.acquisition.models.trade import (
Trade,
TradeAggressorSide,
)
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.recovery.trade_recovery_window import (
TradeRecoveryWindow,
)
from src.market_data.acquisition.recovery.trade_recovery_window_planner import (
TradeRecoveryWindowPlanner,
)
from src.market_data.acquisition.runtime.runtime_recovery_coordinator import (
RuntimeRecoveryCoordinator,
)
from src.market_data.acquisition.runtime.runtime_recovery_protocol import (
RuntimeRecoveryProtocol,
)
SYMBOL = "BTC/USD_LEVERAGE"
CHECKPOINT_TIME = datetime(
2026,
1,
1,
0,
0,
0,
123000,
tzinfo=timezone.utc,
)
CHECKPOINT_TIME_MS = 1_767_225_600_123
RECOVERY_END_TIME_MS = CHECKPOINT_TIME_MS + 5_000
def make_trade(
*,
trade_id: int = 1,
symbol: str = SYMBOL,
executed_at: datetime = CHECKPOINT_TIME,
) -> Trade:
"""
Создать каноническую Trade для Runtime Recovery tests.
"""
return Trade(
symbol=symbol,
trade_id=trade_id,
price=Decimal("64555.55"),
quantity=Decimal("0.002"),
executed_at=executed_at,
aggressor_side=TradeAggressorSide.BUY,
source="test",
)
def make_state_with_checkpoint(
*,
checkpoint: Trade | None = None,
) -> TradeStreamState:
"""
Создать TradeStreamState с последней принятой сделкой.
"""
checkpoint_trade = checkpoint or make_trade()
state = TradeStreamState(
symbol=checkpoint_trade.symbol,
)
accepted_trade = state.accept(
checkpoint_trade,
)
assert accepted_trade is checkpoint_trade
assert state.last_trade is checkpoint_trade
return state
class FakeStateStore:
"""
Управляемая реализация TradeStreamStateStoreProtocol.
"""
def __init__(
self,
*,
states: dict[str, TradeStreamState] | None = None,
) -> None:
self._states = dict(
states or {},
)
self.get_calls: list[str] = []
self.get_or_create_calls: list[str] = []
self.contains_calls: list[str] = []
self.remove_calls: list[str] = []
self.clear_calls = 0
def get_or_create(
self,
symbol: str,
) -> TradeStreamState:
self.get_or_create_calls.append(
symbol,
)
state = self._states.get(
symbol,
)
if state is None:
state = TradeStreamState(
symbol=symbol,
)
self._states[symbol] = state
return state
def get(
self,
symbol: str,
) -> TradeStreamState:
self.get_calls.append(
symbol,
)
try:
return self._states[symbol]
except KeyError as error:
raise TradeStreamStateNotFoundError(
f"Trade Stream state для {symbol!r} не найден."
) from error
def contains(
self,
symbol: str,
) -> bool:
self.contains_calls.append(
symbol,
)
return symbol in self._states
def remove(
self,
symbol: str,
) -> None:
self.remove_calls.append(
symbol,
)
try:
del self._states[symbol]
except KeyError as error:
raise TradeStreamStateNotFoundError(
f"Trade Stream state для {symbol!r} не найден."
) from error
def clear(self) -> None:
self.clear_calls += 1
self._states.clear()
class RecordingWindowPlanner:
"""
Planner с заранее заданным результатом.
"""
def __init__(
self,
*,
windows: tuple[TradeRecoveryWindow, ...] = (),
) -> None:
self._windows = windows
self.calls: list[
dict[str, Any]
] = []
@property
def max_window_ms(self) -> int:
return 3_599_999
def build_windows(
self,
*,
symbol: str,
start_time: int,
end_time: int,
) -> tuple[TradeRecoveryWindow, ...]:
self.calls.append(
{
"symbol": symbol,
"start_time": start_time,
"end_time": end_time,
}
)
return self._windows
class RecordingRecoveryController:
"""
Recovery Controller с записью полученных запросов.
"""
def __init__(
self,
*,
results: tuple[TradeRecoveryResult, ...] = (),
) -> None:
self._results = list(
results,
)
self.requests: list[
TradeRecoveryRequest
] = []
def recover(
self,
request: TradeRecoveryRequest,
) -> TradeRecoveryResult:
self.requests.append(
request,
)
if self._results:
return self._results.pop(0)
return TradeRecoveryResult(
symbol=request.symbol,
requested_start_time=request.start_time,
requested_end_time=request.end_time,
recovered_trades=(),
)
def create_coordinator(
*,
state_store: FakeStateStore | None = None,
window_planner: RecordingWindowPlanner | None = None,
recovery_controller: RecordingRecoveryController | None = None,
) -> tuple[
RuntimeRecoveryCoordinator,
FakeStateStore,
RecordingWindowPlanner,
RecordingRecoveryController,
]:
"""
Создать Coordinator и управляемые зависимости.
"""
resolved_state_store = (
state_store
or FakeStateStore()
)
resolved_window_planner = (
window_planner
or RecordingWindowPlanner()
)
resolved_recovery_controller = (
recovery_controller
or RecordingRecoveryController()
)
coordinator = RuntimeRecoveryCoordinator(
state_store=resolved_state_store,
window_planner=resolved_window_planner, # type: ignore[arg-type]
recovery_controller=resolved_recovery_controller,
)
return (
coordinator,
resolved_state_store,
resolved_window_planner,
resolved_recovery_controller,
)
def test_coordinator_implements_protocol() -> None:
coordinator, *_ = create_coordinator()
assert isinstance(
coordinator,
RuntimeRecoveryProtocol,
)
def test_coordinator_uses_slots() -> None:
coordinator, *_ = create_coordinator()
assert not hasattr(
coordinator,
"__dict__",
)
@pytest.mark.parametrize(
"symbol",
[
None,
123,
True,
],
)
def test_rejects_non_string_symbol(
symbol: object,
) -> None:
coordinator, state_store, planner, recovery = (
create_coordinator()
)
with pytest.raises(
TypeError,
match="symbol",
):
coordinator.recover(
symbol=symbol, # type: ignore[arg-type]
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert state_store.get_calls == []
assert planner.calls == []
assert recovery.requests == []
@pytest.mark.parametrize(
"symbol",
[
"",
" ",
],
)
def test_rejects_empty_symbol(
symbol: str,
) -> None:
coordinator, state_store, planner, recovery = (
create_coordinator()
)
with pytest.raises(
ValueError,
match="symbol",
):
coordinator.recover(
symbol=symbol,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert state_store.get_calls == []
assert planner.calls == []
assert recovery.requests == []
@pytest.mark.parametrize(
"recovery_end_time",
[
1.5,
"1000",
None,
True,
],
)
def test_rejects_invalid_recovery_end_time_type(
recovery_end_time: object,
) -> None:
coordinator, state_store, planner, recovery = (
create_coordinator()
)
with pytest.raises(
TypeError,
match="recovery_end_time",
):
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=recovery_end_time, # type: ignore[arg-type]
)
assert state_store.get_calls == []
assert planner.calls == []
assert recovery.requests == []
def test_rejects_negative_recovery_end_time() -> None:
coordinator, state_store, planner, recovery = (
create_coordinator()
)
with pytest.raises(
ValueError,
match="recovery_end_time",
):
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=-1,
)
assert state_store.get_calls == []
assert planner.calls == []
assert recovery.requests == []
def test_reads_state_by_symbol() -> None:
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
)
coordinator, _, _, _ = create_coordinator(
state_store=state_store,
)
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert state_store.get_calls == [
SYMBOL,
]
def test_does_not_create_state_during_recovery() -> None:
coordinator, state_store, _, _ = (
create_coordinator()
)
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert state_store.get_or_create_calls == []
def test_missing_state_returns_empty_result() -> None:
coordinator, _, planner, recovery = (
create_coordinator()
)
result = coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert result == TradeRecoveryResult(
symbol=SYMBOL,
requested_start_time=RECOVERY_END_TIME_MS,
requested_end_time=RECOVERY_END_TIME_MS,
recovered_trades=(),
)
assert planner.calls == []
assert recovery.requests == []
def test_state_without_checkpoint_returns_empty_result() -> None:
state_store = FakeStateStore(
states={
SYMBOL: TradeStreamState(
symbol=SYMBOL,
),
}
)
coordinator, _, planner, recovery = (
create_coordinator(
state_store=state_store,
)
)
result = coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert result.is_empty is True
assert (
result.requested_start_time
== RECOVERY_END_TIME_MS
)
assert (
result.requested_end_time
== RECOVERY_END_TIME_MS
)
assert planner.calls == []
assert recovery.requests == []
def test_checkpoint_time_is_passed_to_planner_as_unix_ms() -> None:
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
)
planner = RecordingWindowPlanner()
coordinator, _, _, _ = create_coordinator(
state_store=state_store,
window_planner=planner,
)
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert planner.calls == [
{
"symbol": SYMBOL,
"start_time": CHECKPOINT_TIME_MS,
"end_time": RECOVERY_END_TIME_MS,
}
]
def test_checkpoint_timezone_is_normalized_to_utc() -> None:
offset_timezone = timezone(
timedelta(hours=3),
)
checkpoint = make_trade(
executed_at=datetime(
2026,
1,
1,
3,
0,
0,
123000,
tzinfo=offset_timezone,
),
)
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(
checkpoint=checkpoint,
),
}
)
planner = RecordingWindowPlanner()
coordinator, _, _, _ = create_coordinator(
state_store=state_store,
window_planner=planner,
)
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert planner.calls[0]["start_time"] == (
CHECKPOINT_TIME_MS
)
def test_rejects_naive_checkpoint_datetime() -> None:
checkpoint = make_trade(
executed_at=datetime(
2026,
1,
1,
0,
0,
0,
123000,
),
)
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(
checkpoint=checkpoint,
),
}
)
coordinator, _, planner, recovery = (
create_coordinator(
state_store=state_store,
)
)
with pytest.raises(
ValueError,
match="timezone",
):
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert planner.calls == []
assert recovery.requests == []
def test_empty_planner_result_returns_empty_recovery() -> None:
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
)
planner = RecordingWindowPlanner(
windows=(),
)
coordinator, _, _, recovery = create_coordinator(
state_store=state_store,
window_planner=planner,
)
result = coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert result.is_empty is True
assert recovery.requests == []
def test_single_window_is_forwarded_to_controller() -> None:
window = TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS,
end_time=RECOVERY_END_TIME_MS,
)
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
)
planner = RecordingWindowPlanner(
windows=(window,),
)
recovery = RecordingRecoveryController()
coordinator, *_ = create_coordinator(
state_store=state_store,
window_planner=planner,
recovery_controller=recovery,
)
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert len(recovery.requests) == 1
request = recovery.requests[0]
assert request.symbol == SYMBOL
assert request.start_time == window.start_time
assert request.end_time == window.end_time
def test_multiple_windows_are_processed_sequentially() -> None:
windows = (
TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS,
end_time=CHECKPOINT_TIME_MS + 1_000,
),
TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS + 1_000,
end_time=CHECKPOINT_TIME_MS + 3_000,
),
TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS + 3_000,
end_time=RECOVERY_END_TIME_MS,
),
)
state_store = FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
)
planner = RecordingWindowPlanner(
windows=windows,
)
recovery = RecordingRecoveryController()
coordinator, *_ = create_coordinator(
state_store=state_store,
window_planner=planner,
recovery_controller=recovery,
)
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert [
(r.start_time, r.end_time)
for r in recovery.requests
] == [
(
CHECKPOINT_TIME_MS,
CHECKPOINT_TIME_MS + 1_000,
),
(
CHECKPOINT_TIME_MS + 1_000,
CHECKPOINT_TIME_MS + 3_000,
),
(
CHECKPOINT_TIME_MS + 3_000,
RECOVERY_END_TIME_MS,
),
]
def test_multiple_window_results_are_aggregated() -> None:
middle_time = CHECKPOINT_TIME_MS + 2_500
window_one = TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS,
end_time=middle_time,
)
window_two = TradeRecoveryWindow(
symbol=SYMBOL,
start_time=middle_time,
end_time=RECOVERY_END_TIME_MS,
)
first_trade = make_trade(
trade_id=100,
)
second_trade = make_trade(
trade_id=200,
)
result_one = TradeRecoveryResult(
symbol=SYMBOL,
requested_start_time=CHECKPOINT_TIME_MS,
requested_end_time=middle_time,
recovered_trades=(
first_trade,
),
)
result_two = TradeRecoveryResult(
symbol=SYMBOL,
requested_start_time=middle_time,
requested_end_time=RECOVERY_END_TIME_MS,
recovered_trades=(
second_trade,
),
)
coordinator, *_ = create_coordinator(
state_store=FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
),
window_planner=RecordingWindowPlanner(
windows=(
window_one,
window_two,
),
),
recovery_controller=RecordingRecoveryController(
results=(
result_one,
result_two,
),
),
)
result = coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert result.recovered_trades == (
first_trade,
second_trade,
)
assert (
result.requested_start_time
== CHECKPOINT_TIME_MS
)
assert (
result.requested_end_time
== RECOVERY_END_TIME_MS
)
assert result.first_trade is first_trade
assert result.last_trade is second_trade
def test_single_window_result_is_aggregated_into_full_result() -> None:
recovered_trade = make_trade(
trade_id=777,
)
expected = TradeRecoveryResult(
symbol=SYMBOL,
requested_start_time=CHECKPOINT_TIME_MS,
requested_end_time=RECOVERY_END_TIME_MS,
recovered_trades=(
recovered_trade,
),
)
coordinator, *_ = create_coordinator(
state_store=FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
),
window_planner=RecordingWindowPlanner(
windows=(
TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS,
end_time=RECOVERY_END_TIME_MS,
),
),
),
recovery_controller=RecordingRecoveryController(
results=(expected,),
),
)
result = coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert result is not expected
assert result == expected
assert result.recovered_trades == (
recovered_trade,
)
assert result.first_trade is recovered_trade
assert result.last_trade is recovered_trade
def test_controller_error_is_not_swallowed() -> None:
class BrokenRecoveryController(
RecordingRecoveryController,
):
def recover(
self,
request: TradeRecoveryRequest,
) -> TradeRecoveryResult:
raise RuntimeError(
"controller failed",
)
coordinator, *_ = create_coordinator(
state_store=FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
),
window_planner=RecordingWindowPlanner(
windows=(
TradeRecoveryWindow(
symbol=SYMBOL,
start_time=CHECKPOINT_TIME_MS,
end_time=RECOVERY_END_TIME_MS,
),
),
),
recovery_controller=BrokenRecoveryController(),
)
with pytest.raises(
RuntimeError,
match="controller failed",
):
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
def test_planner_error_is_not_swallowed() -> None:
class BrokenPlanner(
TradeRecoveryWindowPlanner,
):
def __init__(self) -> None:
pass
@property
def max_window_ms(self) -> int:
return 1
def build_windows(
self,
*,
symbol: str,
start_time: int,
end_time: int,
) -> tuple[TradeRecoveryWindow, ...]:
raise RuntimeError(
"planner failed",
)
coordinator, *_ = create_coordinator(
state_store=FakeStateStore(
states={
SYMBOL: make_state_with_checkpoint(),
}
),
window_planner=BrokenPlanner(),
)
with pytest.raises(
RuntimeError,
match="planner failed",
):
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
def test_store_error_is_not_swallowed() -> None:
class BrokenStore(
FakeStateStore,
):
def get(
self,
symbol: str,
) -> TradeStreamState:
raise RuntimeError(
"store failed",
)
coordinator, *_ = create_coordinator(
state_store=BrokenStore(),
)
with pytest.raises(
RuntimeError,
match="store failed",
):
coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)

View File

@@ -0,0 +1,590 @@
# app/tests/unit/market_data/acquisition/runtime/test_runtime_scheduler.py
from __future__ import annotations
import asyncio
from collections.abc import Awaitable
from typing import Callable
import pytest
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatState,
)
from src.market_data.acquisition.runtime.scheduler import (
RuntimeScheduler,
RuntimeSchedulerProtocol,
)
from src.market_data.acquisition.runtime.supervisor import (
RuntimeSupervisorState,
)
class FakeHeartbeatMonitor:
def __init__(
self,
results: list[bool] | None = None,
) -> None:
self._results = list(results or [])
self.check_timeout_calls = 0
@property
def state(self) -> HeartbeatState:
return HeartbeatState.MONITORING
@property
def timeout_seconds(self) -> float:
return 10.0
@property
def last_activity_at(self) -> float | None:
return 0.0
def start(self) -> None:
return None
def stop(self) -> None:
return None
def record_activity(self) -> None:
return None
async def check_timeout(self) -> bool:
self.check_timeout_calls += 1
if not self._results:
return False
return self._results.pop(0)
class FakeRuntimeSupervisor:
def __init__(self) -> None:
self.handle_timeout_calls = 0
@property
def state(self) -> RuntimeSupervisorState:
return RuntimeSupervisorState.RUNNING
def start(self) -> None:
return None
def stop(self) -> None:
return None
def notify_activity(self) -> None:
return None
async def handle_heartbeat_timeout(self) -> bool:
self.handle_timeout_calls += 1
return True
class RecordingSleep:
def __init__(
self,
*,
on_call: Callable[[], None] | None = None,
) -> None:
self.calls: list[float] = []
self._on_call = on_call
async def __call__(
self,
seconds: float,
) -> None:
self.calls.append(seconds)
if self._on_call is not None:
self._on_call()
def create_scheduler(
*,
heartbeat_results: list[bool] | None = None,
interval_seconds: float = 1.0,
sleep: Callable[[float], Awaitable[None]] | None = None,
) -> tuple[
RuntimeScheduler,
FakeHeartbeatMonitor,
FakeRuntimeSupervisor,
]:
heartbeat = FakeHeartbeatMonitor(
results=heartbeat_results,
)
supervisor = FakeRuntimeSupervisor()
scheduler = RuntimeScheduler(
heartbeat_monitor=heartbeat,
runtime_supervisor=supervisor,
interval_seconds=interval_seconds,
sleep=sleep or asyncio.sleep,
)
return (
scheduler,
heartbeat,
supervisor,
)
def test_scheduler_implements_protocol() -> None:
scheduler, *_ = create_scheduler()
assert isinstance(
scheduler,
RuntimeSchedulerProtocol,
)
def test_scheduler_uses_slots() -> None:
scheduler, *_ = create_scheduler()
assert not hasattr(scheduler, "__dict__")
def test_initial_state_is_not_running() -> None:
scheduler, *_ = create_scheduler()
assert scheduler.running is False
def test_exposes_interval_seconds() -> None:
scheduler, *_ = create_scheduler(
interval_seconds=2.5,
)
assert scheduler.interval_seconds == 2.5
@pytest.mark.parametrize(
"interval_seconds",
[
1,
1.5,
],
)
def test_accepts_positive_interval(
interval_seconds: float,
) -> None:
scheduler, *_ = create_scheduler(
interval_seconds=interval_seconds,
)
assert scheduler.interval_seconds == float(interval_seconds)
@pytest.mark.parametrize(
"interval_seconds",
[
"1",
None,
True,
],
)
def test_rejects_invalid_interval_type(
interval_seconds: object,
) -> None:
with pytest.raises(TypeError):
RuntimeScheduler(
heartbeat_monitor=FakeHeartbeatMonitor(),
runtime_supervisor=FakeRuntimeSupervisor(),
interval_seconds=interval_seconds, # type: ignore[arg-type]
)
@pytest.mark.parametrize(
"interval_seconds",
[
0,
-1,
-0.5,
],
)
def test_rejects_non_positive_interval(
interval_seconds: float,
) -> None:
with pytest.raises(ValueError):
RuntimeScheduler(
heartbeat_monitor=FakeHeartbeatMonitor(),
runtime_supervisor=FakeRuntimeSupervisor(),
interval_seconds=interval_seconds,
)
def test_rejects_non_callable_sleep() -> None:
with pytest.raises(TypeError):
RuntimeScheduler(
heartbeat_monitor=FakeHeartbeatMonitor(),
runtime_supervisor=FakeRuntimeSupervisor(),
interval_seconds=1.0,
sleep=object(), # type: ignore[arg-type]
)
def test_run_once_checks_heartbeat() -> None:
scheduler, heartbeat, supervisor = create_scheduler(
heartbeat_results=[False],
)
result = asyncio.run(
scheduler.run_once()
)
assert result is False
assert heartbeat.check_timeout_calls == 1
assert supervisor.handle_timeout_calls == 0
def test_run_once_calls_supervisor_on_timeout() -> None:
scheduler, heartbeat, supervisor = create_scheduler(
heartbeat_results=[True],
)
result = asyncio.run(
scheduler.run_once()
)
assert result is True
assert heartbeat.check_timeout_calls == 1
assert supervisor.handle_timeout_calls == 1
def test_run_once_does_not_call_supervisor_without_timeout() -> None:
scheduler, _, supervisor = create_scheduler(
heartbeat_results=[False],
)
asyncio.run(
scheduler.run_once()
)
assert supervisor.handle_timeout_calls == 0
def test_stop_is_safe_before_start() -> None:
scheduler, *_ = create_scheduler()
scheduler.stop()
assert scheduler.running is False
def test_repeated_stop_is_safe() -> None:
scheduler, *_ = create_scheduler()
scheduler.stop()
scheduler.stop()
assert scheduler.running is False
def test_start_runs_periodic_check() -> None:
scheduler: RuntimeScheduler
def stop_scheduler() -> None:
scheduler.stop()
sleep = RecordingSleep(
on_call=stop_scheduler,
)
scheduler, heartbeat, supervisor = create_scheduler(
heartbeat_results=[False],
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
assert heartbeat.check_timeout_calls == 1
assert supervisor.handle_timeout_calls == 0
assert sleep.calls == [1.0]
assert scheduler.running is False
def test_start_uses_configured_interval() -> None:
scheduler: RuntimeScheduler
def stop_scheduler() -> None:
scheduler.stop()
sleep = RecordingSleep(
on_call=stop_scheduler,
)
scheduler, _, _ = create_scheduler(
heartbeat_results=[False],
interval_seconds=2.5,
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
assert sleep.calls == [2.5]
def test_scheduler_handles_timeout_during_loop() -> None:
scheduler: RuntimeScheduler
def stop_scheduler() -> None:
scheduler.stop()
sleep = RecordingSleep(
on_call=stop_scheduler,
)
scheduler, heartbeat, supervisor = create_scheduler(
heartbeat_results=[True],
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
assert heartbeat.check_timeout_calls == 1
assert supervisor.handle_timeout_calls == 1
assert scheduler.running is False
def test_scheduler_can_run_multiple_iterations() -> None:
scheduler: RuntimeScheduler
sleep_calls = 0
async def sleep(
seconds: float,
) -> None:
nonlocal sleep_calls
assert seconds == 1.0
sleep_calls += 1
if sleep_calls == 2:
scheduler.stop()
scheduler, heartbeat, supervisor = create_scheduler(
heartbeat_results=[
False,
True,
],
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
assert heartbeat.check_timeout_calls == 2
assert supervisor.handle_timeout_calls == 1
assert sleep_calls == 2
def test_stop_during_run_once_prevents_sleep() -> None:
heartbeat = FakeHeartbeatMonitor(
results=[False],
)
supervisor = FakeRuntimeSupervisor()
sleep = RecordingSleep()
scheduler: RuntimeScheduler
class StoppingHeartbeatMonitor(FakeHeartbeatMonitor):
async def check_timeout(self) -> bool:
self.check_timeout_calls += 1
scheduler.stop()
return False
heartbeat = StoppingHeartbeatMonitor()
scheduler = RuntimeScheduler(
heartbeat_monitor=heartbeat,
runtime_supervisor=supervisor,
interval_seconds=1.0,
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
assert heartbeat.check_timeout_calls == 1
assert sleep.calls == []
assert scheduler.running is False
def test_repeated_start_while_running_does_not_create_second_loop() -> None:
heartbeat = FakeHeartbeatMonitor(
results=[False],
)
supervisor = FakeRuntimeSupervisor()
scheduler: RuntimeScheduler
nested_start_completed = False
async def sleep(
seconds: float,
) -> None:
nonlocal nested_start_completed
assert seconds == 1.0
await scheduler.start()
nested_start_completed = True
scheduler.stop()
scheduler = RuntimeScheduler(
heartbeat_monitor=heartbeat,
runtime_supervisor=supervisor,
interval_seconds=1.0,
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
assert nested_start_completed is True
assert heartbeat.check_timeout_calls == 1
assert scheduler.running is False
def test_scheduler_can_restart_after_stop() -> None:
scheduler: RuntimeScheduler
sleep_calls = 0
async def sleep(
seconds: float,
) -> None:
nonlocal sleep_calls
sleep_calls += 1
scheduler.stop()
scheduler, heartbeat, _ = create_scheduler(
heartbeat_results=[
False,
False,
],
sleep=sleep,
)
asyncio.run(
scheduler.start()
)
asyncio.run(
scheduler.start()
)
assert heartbeat.check_timeout_calls == 2
assert sleep_calls == 2
assert scheduler.running is False
def test_heartbeat_error_is_propagated() -> None:
class BrokenHeartbeatMonitor(FakeHeartbeatMonitor):
async def check_timeout(self) -> bool:
raise RuntimeError("heartbeat failed")
scheduler = RuntimeScheduler(
heartbeat_monitor=BrokenHeartbeatMonitor(),
runtime_supervisor=FakeRuntimeSupervisor(),
interval_seconds=1.0,
)
with pytest.raises(
RuntimeError,
match="heartbeat failed",
):
asyncio.run(
scheduler.run_once()
)
def test_supervisor_error_is_propagated() -> None:
class BrokenRuntimeSupervisor(FakeRuntimeSupervisor):
async def handle_heartbeat_timeout(self) -> bool:
raise RuntimeError("supervisor failed")
scheduler = RuntimeScheduler(
heartbeat_monitor=FakeHeartbeatMonitor(
results=[True],
),
runtime_supervisor=BrokenRuntimeSupervisor(),
interval_seconds=1.0,
)
with pytest.raises(
RuntimeError,
match="supervisor failed",
):
asyncio.run(
scheduler.run_once()
)
def test_start_resets_running_after_heartbeat_error() -> None:
class BrokenHeartbeatMonitor(FakeHeartbeatMonitor):
async def check_timeout(self) -> bool:
raise RuntimeError("heartbeat failed")
scheduler = RuntimeScheduler(
heartbeat_monitor=BrokenHeartbeatMonitor(),
runtime_supervisor=FakeRuntimeSupervisor(),
interval_seconds=1.0,
)
with pytest.raises(RuntimeError):
asyncio.run(
scheduler.start()
)
assert scheduler.running is False
def test_start_resets_running_after_supervisor_error() -> None:
class BrokenRuntimeSupervisor(FakeRuntimeSupervisor):
async def handle_heartbeat_timeout(self) -> bool:
raise RuntimeError("supervisor failed")
scheduler = RuntimeScheduler(
heartbeat_monitor=FakeHeartbeatMonitor(
results=[True],
),
runtime_supervisor=BrokenRuntimeSupervisor(),
interval_seconds=1.0,
)
with pytest.raises(RuntimeError):
asyncio.run(
scheduler.start()
)
assert scheduler.running is False
def test_sleep_error_is_propagated_and_resets_running() -> None:
async def broken_sleep(
seconds: float,
) -> None:
raise RuntimeError("sleep failed")
scheduler, heartbeat, _ = create_scheduler(
heartbeat_results=[False],
sleep=broken_sleep,
)
with pytest.raises(
RuntimeError,
match="sleep failed",
):
asyncio.run(
scheduler.start()
)
assert heartbeat.check_timeout_calls == 1
assert scheduler.running is False

View File

@@ -0,0 +1,528 @@
# app/tests/unit/market_data/acquisition/runtime/test_runtime_supervisor.py
from __future__ import annotations
import asyncio
import pytest
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatState,
)
from src.market_data.acquisition.runtime.reconnect import (
ReconnectState,
)
from src.market_data.acquisition.runtime.supervisor import (
RuntimeSupervisor,
RuntimeSupervisorProtocol,
RuntimeSupervisorState,
)
class FakeHeartbeatMonitor:
def __init__(self) -> None:
self.start_calls = 0
self.stop_calls = 0
self.record_activity_calls = 0
self._state = HeartbeatState.IDLE
self._last_activity_at: float | None = None
@property
def state(self) -> HeartbeatState:
return self._state
@property
def timeout_seconds(self) -> float:
return 10.0
@property
def last_activity_at(self) -> float | None:
return self._last_activity_at
def start(self) -> None:
self.start_calls += 1
self._last_activity_at = float(self.start_calls)
self._state = HeartbeatState.MONITORING
def stop(self) -> None:
self.stop_calls += 1
self._last_activity_at = None
self._state = HeartbeatState.IDLE
def record_activity(self) -> None:
self.record_activity_calls += 1
self._last_activity_at = float(
self.record_activity_calls
)
self._state = HeartbeatState.MONITORING
async def check_timeout(self) -> bool:
return False
class FakeReconnectCoordinator:
def __init__(self) -> None:
self.reconnect_calls = 0
self._attempt = 0
self._state = ReconnectState.DISCONNECTED
@property
def state(self) -> ReconnectState:
return self._state
@property
def attempt(self) -> int:
return self._attempt
async def reconnect(self) -> None:
self.reconnect_calls += 1
self._attempt += 1
self._state = ReconnectState.CONNECTED
def create_supervisor() -> tuple[
RuntimeSupervisor,
FakeHeartbeatMonitor,
FakeReconnectCoordinator,
]:
heartbeat = FakeHeartbeatMonitor()
reconnect = FakeReconnectCoordinator()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=reconnect,
)
return (
supervisor,
heartbeat,
reconnect,
)
def test_supervisor_implements_protocol() -> None:
supervisor, *_ = create_supervisor()
assert isinstance(
supervisor,
RuntimeSupervisorProtocol,
)
def test_supervisor_uses_slots() -> None:
supervisor, *_ = create_supervisor()
assert not hasattr(supervisor, "__dict__")
def test_initial_state_is_stopped() -> None:
supervisor, heartbeat, reconnect = create_supervisor()
assert supervisor.state is RuntimeSupervisorState.STOPPED
assert heartbeat.start_calls == 0
assert heartbeat.stop_calls == 0
assert reconnect.reconnect_calls == 0
def test_start_starts_heartbeat() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.start()
assert heartbeat.start_calls == 1
assert heartbeat.state is HeartbeatState.MONITORING
def test_start_sets_running_state() -> None:
supervisor, *_ = create_supervisor()
supervisor.start()
assert supervisor.state is RuntimeSupervisorState.RUNNING
def test_repeated_start_begins_new_monitoring_period() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.start()
supervisor.start()
assert heartbeat.start_calls == 2
assert supervisor.state is RuntimeSupervisorState.RUNNING
def test_stop_stops_heartbeat() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.start()
supervisor.stop()
assert heartbeat.stop_calls == 1
assert heartbeat.state is HeartbeatState.IDLE
def test_stop_sets_stopped_state() -> None:
supervisor, *_ = create_supervisor()
supervisor.start()
supervisor.stop()
assert supervisor.state is RuntimeSupervisorState.STOPPED
def test_stop_is_safe_before_start() -> None:
supervisor, heartbeat, reconnect = create_supervisor()
supervisor.stop()
assert supervisor.state is RuntimeSupervisorState.STOPPED
assert heartbeat.stop_calls == 1
assert reconnect.reconnect_calls == 0
def test_repeated_stop_remains_stopped() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.stop()
supervisor.stop()
assert heartbeat.stop_calls == 2
assert supervisor.state is RuntimeSupervisorState.STOPPED
def test_notify_activity_delegates_while_running() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.start()
supervisor.notify_activity()
assert heartbeat.record_activity_calls == 1
def test_notify_activity_does_nothing_while_stopped() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.notify_activity()
assert heartbeat.record_activity_calls == 0
assert supervisor.state is RuntimeSupervisorState.STOPPED
def test_timeout_does_nothing_while_stopped() -> None:
supervisor, heartbeat, reconnect = create_supervisor()
result = asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert result is False
assert reconnect.reconnect_calls == 0
assert heartbeat.start_calls == 0
assert heartbeat.stop_calls == 0
assert supervisor.state is RuntimeSupervisorState.STOPPED
def test_timeout_stops_heartbeat_before_reconnect() -> None:
call_order: list[str] = []
class OrderedHeartbeatMonitor(FakeHeartbeatMonitor):
def stop(self) -> None:
call_order.append("heartbeat.stop")
super().stop()
class OrderedReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
call_order.append("reconnect")
await super().reconnect()
heartbeat = OrderedHeartbeatMonitor()
reconnect = OrderedReconnectCoordinator()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=reconnect,
)
supervisor.start()
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert call_order == [
"heartbeat.stop",
"reconnect",
]
def test_timeout_runs_single_reconnect_attempt() -> None:
supervisor, _, reconnect = create_supervisor()
supervisor.start()
result = asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert result is True
assert reconnect.reconnect_calls == 1
def test_successful_reconnect_restarts_heartbeat() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.start()
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert heartbeat.stop_calls == 1
assert heartbeat.start_calls == 2
assert heartbeat.state is HeartbeatState.MONITORING
def test_successful_reconnect_returns_to_running() -> None:
supervisor, _, reconnect = create_supervisor()
supervisor.start()
result = asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert result is True
assert supervisor.state is RuntimeSupervisorState.RUNNING
assert reconnect.state is ReconnectState.CONNECTED
def test_notify_activity_works_after_successful_reconnect() -> None:
supervisor, heartbeat, _ = create_supervisor()
supervisor.start()
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
supervisor.notify_activity()
assert heartbeat.record_activity_calls == 1
def test_reconnect_error_is_propagated() -> None:
class BrokenReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
self.reconnect_calls += 1
self._attempt += 1
self._state = ReconnectState.FAILED
raise RuntimeError("reconnect failed")
heartbeat = FakeHeartbeatMonitor()
reconnect = BrokenReconnectCoordinator()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=reconnect,
)
supervisor.start()
with pytest.raises(
RuntimeError,
match="reconnect failed",
):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
def test_reconnect_error_sets_failed_state() -> None:
class BrokenReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
self.reconnect_calls += 1
self._state = ReconnectState.FAILED
raise RuntimeError("reconnect failed")
heartbeat = FakeHeartbeatMonitor()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=(
BrokenReconnectCoordinator()
),
)
supervisor.start()
with pytest.raises(RuntimeError):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert supervisor.state is RuntimeSupervisorState.FAILED
def test_reconnect_error_leaves_heartbeat_stopped() -> None:
class BrokenReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
raise RuntimeError("reconnect failed")
heartbeat = FakeHeartbeatMonitor()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=(
BrokenReconnectCoordinator()
),
)
supervisor.start()
with pytest.raises(RuntimeError):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert heartbeat.start_calls == 1
assert heartbeat.stop_calls == 1
assert heartbeat.state is HeartbeatState.IDLE
def test_timeout_does_not_start_second_reconnect_after_failure() -> None:
class BrokenReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
self.reconnect_calls += 1
raise RuntimeError("reconnect failed")
heartbeat = FakeHeartbeatMonitor()
reconnect = BrokenReconnectCoordinator()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=reconnect,
)
supervisor.start()
with pytest.raises(RuntimeError):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
second_result = asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert second_result is False
assert reconnect.reconnect_calls == 1
assert supervisor.state is RuntimeSupervisorState.FAILED
def test_start_can_restart_supervision_after_failure() -> None:
class FailOnceReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
self.reconnect_calls += 1
self._attempt += 1
if self.reconnect_calls == 1:
self._state = ReconnectState.FAILED
raise RuntimeError("temporary failure")
self._state = ReconnectState.CONNECTED
heartbeat = FakeHeartbeatMonitor()
reconnect = FailOnceReconnectCoordinator()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=reconnect,
)
supervisor.start()
with pytest.raises(RuntimeError):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
supervisor.start()
result = asyncio.run(
supervisor.handle_heartbeat_timeout()
)
assert result is True
assert reconnect.reconnect_calls == 2
assert supervisor.state is RuntimeSupervisorState.RUNNING
def test_notify_activity_does_nothing_after_failure() -> None:
class BrokenReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
raise RuntimeError("reconnect failed")
heartbeat = FakeHeartbeatMonitor()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=(
BrokenReconnectCoordinator()
),
)
supervisor.start()
with pytest.raises(RuntimeError):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
supervisor.notify_activity()
assert heartbeat.record_activity_calls == 0
def test_stop_can_reset_failed_supervisor() -> None:
class BrokenReconnectCoordinator(
FakeReconnectCoordinator
):
async def reconnect(self) -> None:
raise RuntimeError("reconnect failed")
heartbeat = FakeHeartbeatMonitor()
supervisor = RuntimeSupervisor(
heartbeat_monitor=heartbeat,
reconnect_coordinator=(
BrokenReconnectCoordinator()
),
)
supervisor.start()
with pytest.raises(RuntimeError):
asyncio.run(
supervisor.handle_heartbeat_timeout()
)
supervisor.stop()
assert supervisor.state is RuntimeSupervisorState.STOPPED
assert heartbeat.state is HeartbeatState.IDLE

View File

@@ -0,0 +1,752 @@
# app/tests/unit/market_data/acquisition/
# test_trade_stream_runtime_composition.py
from __future__ import annotations
import asyncio
from dataclasses import FrozenInstanceError, dataclass
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.models.trade import (
Trade,
TradeAggressorSide,
)
from src.market_data.acquisition.runtime.acquisition_runtime_service_protocol import (
AcquisitionRuntimeServiceProtocol,
)
from src.market_data.acquisition.runtime.heartbeat import (
HeartbeatMonitorProtocol,
HeartbeatState,
)
from src.market_data.acquisition.runtime.reconnect import (
ReconnectCoordinatorProtocol,
ReconnectState,
)
from src.market_data.acquisition.runtime.runtime_events import (
ReconnectCompletedEvent,
ReconnectStartedEvent,
)
from src.market_data.acquisition.runtime.runtime_recovery_protocol import (
RuntimeRecoveryProtocol,
)
from src.market_data.acquisition.runtime.scheduler import (
RuntimeSchedulerProtocol,
)
from src.market_data.acquisition.runtime.supervisor import (
RuntimeSupervisorProtocol,
RuntimeSupervisorState,
)
from src.market_data.acquisition.runtime.websocket_protocol import (
AcquisitionRuntimeCommandDispatcherProtocol,
AcquisitionRuntimeEvent,
AcquisitionSubscriptionMessage,
)
from src.market_data.acquisition.trade_stream_acquisition_protocol import (
TradeStreamAcquisitionServiceProtocol,
)
from src.market_data.acquisition.trade_stream_message_adapter_protocol import (
TradeStreamMappedMessage,
)
from src.market_data.acquisition.trade_stream_runtime_composition import (
TradeStreamRuntimeComposition,
build_trade_stream_runtime_composition,
)
SYMBOL = "BTC/USD_LEVERAGE"
CHECKPOINT_TIME = datetime(
2026,
7,
29,
12,
0,
0,
123000,
tzinfo=timezone.utc,
)
CHECKPOINT_TIME_MS = 1_785_326_400_123
RECOVERY_END_TIME_MS = CHECKPOINT_TIME_MS + 5_000
def make_trade(
*,
trade_id: int = 100,
executed_at: datetime = CHECKPOINT_TIME,
) -> Trade:
return Trade(
symbol=SYMBOL,
trade_id=trade_id,
price=Decimal("64555.55"),
quantity=Decimal("0.002"),
executed_at=executed_at,
aggressor_side=TradeAggressorSide.BUY,
source="test",
)
def make_raw_trade(
*,
trade_id: int,
timestamp: int,
) -> dict[str, object]:
return {
"a": trade_id,
"p": "64556.00",
"q": "0.003",
"T": timestamp,
"m": False,
}
class FakeSession:
def __init__(self) -> None:
self.start_calls = 0
self.stop_calls = 0
@property
def is_connected(self) -> bool:
return self.start_calls > self.stop_calls
async def start(self) -> None:
self.start_calls += 1
async def stop(self) -> None:
self.stop_calls += 1
class FakeTransport:
def __init__(self) -> None:
self.connect_calls = 0
self.disconnect_calls = 0
self.sent_messages: list[str | bytes] = []
self.receive_calls = 0
async def connect(self) -> None:
self.connect_calls += 1
async def disconnect(self) -> None:
self.disconnect_calls += 1
async def send(
self,
message: str | bytes,
) -> None:
self.sent_messages.append(message)
async def receive(self) -> str | bytes:
self.receive_calls += 1
return ""
class FakeSubscriptionManager:
def __init__(self) -> None:
self.subscriptions: list[
tuple[str, AcquisitionSubscriptionMessage]
] = []
self.unsubscriptions: list[
tuple[str, AcquisitionSubscriptionMessage]
] = []
self.restore_calls = 0
self.clear_calls = 0
async def subscribe(
self,
subscription_key: str,
message: AcquisitionSubscriptionMessage,
) -> None:
self.subscriptions.append(
(
subscription_key,
message,
)
)
async def unsubscribe(
self,
subscription_key: str,
message: AcquisitionSubscriptionMessage,
) -> None:
self.unsubscriptions.append(
(
subscription_key,
message,
)
)
async def restore_subscriptions(self) -> None:
self.restore_calls += 1
async def clear_subscriptions(self) -> None:
self.clear_calls += 1
class FakeEventPublisher:
def __init__(self) -> None:
self.events: list[AcquisitionRuntimeEvent] = []
async def publish(
self,
event: AcquisitionRuntimeEvent,
) -> None:
self.events.append(event)
class FakeMessageAdapter:
def __init__(
self,
result: TradeStreamMappedMessage,
) -> None:
self._result = result
self.documents: list[object] = []
def map_message(
self,
document: object,
) -> TradeStreamMappedMessage:
self.documents.append(document)
return self._result
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 FakeClock:
def __init__(
self,
value: float = 100.0,
) -> None:
self.value = value
def __call__(self) -> float:
return self.value
class RecordingSleep:
def __init__(self) -> None:
self.calls: list[float] = []
async def __call__(
self,
seconds: float,
) -> None:
self.calls.append(seconds)
@dataclass(slots=True)
class CompositionDependencies:
session: FakeSession
transport: FakeTransport
subscription_manager: FakeSubscriptionManager
event_publisher: FakeEventPublisher
message_adapter: FakeMessageAdapter
recovery_document_source: StubTradesDocumentSource
heartbeat_clock: FakeClock
scheduler_sleep: RecordingSleep
def create_composition(
*,
trade: Trade | None = None,
recovery_document: object = (),
heartbeat_timeout_seconds: float = 10.0,
scheduler_interval_seconds: float = 1.0,
max_recovery_window_ms: int = 3_599_999,
) -> tuple[
TradeStreamRuntimeComposition,
CompositionDependencies,
]:
dependencies = CompositionDependencies(
session=FakeSession(),
transport=FakeTransport(),
subscription_manager=FakeSubscriptionManager(),
event_publisher=FakeEventPublisher(),
message_adapter=FakeMessageAdapter(
trade or make_trade(),
),
recovery_document_source=StubTradesDocumentSource(
recovery_document,
),
heartbeat_clock=FakeClock(),
scheduler_sleep=RecordingSleep(),
)
composition = build_trade_stream_runtime_composition(
session=dependencies.session,
transport=dependencies.transport,
subscription_manager=dependencies.subscription_manager,
event_publisher=dependencies.event_publisher,
message_adapter=dependencies.message_adapter,
recovery_document_source=(
dependencies.recovery_document_source
),
heartbeat_timeout_seconds=heartbeat_timeout_seconds,
scheduler_interval_seconds=scheduler_interval_seconds,
max_recovery_window_ms=max_recovery_window_ms,
heartbeat_clock=dependencies.heartbeat_clock,
scheduler_sleep=dependencies.scheduler_sleep,
)
return (
composition,
dependencies,
)
def test_composition_uses_slots_and_is_frozen() -> None:
composition, *_ = create_composition()
assert not hasattr(
composition,
"__dict__",
)
with pytest.raises(FrozenInstanceError):
composition.runtime_scheduler = ( # type: ignore[misc]
composition.runtime_scheduler
)
def test_components_implement_public_protocols() -> None:
composition, *_ = create_composition()
assert isinstance(
composition.acquisition_runtime_service,
AcquisitionRuntimeServiceProtocol,
)
assert isinstance(
composition.acquisition_runtime_service,
AcquisitionRuntimeCommandDispatcherProtocol,
)
assert isinstance(
composition.trade_stream_acquisition_service,
TradeStreamAcquisitionServiceProtocol,
)
assert isinstance(
composition.runtime_recovery_coordinator,
RuntimeRecoveryProtocol,
)
assert isinstance(
composition.reconnect_coordinator,
ReconnectCoordinatorProtocol,
)
assert isinstance(
composition.heartbeat_monitor,
HeartbeatMonitorProtocol,
)
assert isinstance(
composition.runtime_supervisor,
RuntimeSupervisorProtocol,
)
assert isinstance(
composition.runtime_scheduler,
RuntimeSchedulerProtocol,
)
def test_external_dependencies_are_reused() -> None:
composition, dependencies = create_composition()
runtime_service = composition.acquisition_runtime_service
assert runtime_service._session is dependencies.session
assert runtime_service._transport is dependencies.transport
assert (
runtime_service._subscription_manager
is dependencies.subscription_manager
)
assert (
runtime_service._event_publisher
is dependencies.event_publisher
)
assert (
composition.trade_stream_acquisition_service._adapter
is dependencies.message_adapter
)
assert (
composition.recovery_controller._document_source
is dependencies.recovery_document_source
)
def test_live_stream_and_recovery_share_consistency_state() -> None:
composition, *_ = create_composition()
assert (
composition.consistency_controller._state_store
is composition.state_store
)
assert (
composition.trade_stream_acquisition_service
._consistency_controller
is composition.consistency_controller
)
assert (
composition.recovery_controller._consistency_controller
is composition.consistency_controller
)
assert (
composition.runtime_recovery_coordinator._state_store
is composition.state_store
)
def test_runtime_components_share_lifecycle_dependencies() -> None:
composition, dependencies = create_composition()
assert (
composition.trade_stream_acquisition_service._runtime_service
is composition.acquisition_runtime_service
)
assert (
composition.reconnect_coordinator._command_dispatcher
is composition.acquisition_runtime_service
)
assert (
composition.reconnect_coordinator._subscription_manager
is dependencies.subscription_manager
)
assert (
composition.reconnect_coordinator._event_publisher
is dependencies.event_publisher
)
assert (
composition.runtime_supervisor._heartbeat_monitor
is composition.heartbeat_monitor
)
assert (
composition.runtime_supervisor._reconnect_coordinator
is composition.reconnect_coordinator
)
assert (
composition.runtime_scheduler._heartbeat_monitor
is composition.heartbeat_monitor
)
assert (
composition.runtime_scheduler._runtime_supervisor
is composition.runtime_supervisor
)
def test_configuration_is_forwarded() -> None:
composition, *_ = create_composition(
heartbeat_timeout_seconds=15.5,
scheduler_interval_seconds=2.5,
max_recovery_window_ms=2_000,
)
assert composition.heartbeat_monitor.timeout_seconds == 15.5
assert composition.runtime_scheduler.interval_seconds == 2.5
assert composition.recovery_window_planner.max_window_ms == 2_000
def test_clock_and_sleep_are_forwarded() -> None:
composition, dependencies = create_composition()
assert (
composition.heartbeat_monitor._clock
is dependencies.heartbeat_clock
)
assert (
composition.runtime_scheduler._sleep
is dependencies.scheduler_sleep
)
def test_creation_has_no_runtime_side_effects() -> None:
composition, dependencies = create_composition()
assert dependencies.session.start_calls == 0
assert dependencies.session.stop_calls == 0
assert dependencies.transport.connect_calls == 0
assert dependencies.transport.disconnect_calls == 0
assert dependencies.transport.sent_messages == []
assert dependencies.transport.receive_calls == 0
assert dependencies.subscription_manager.subscriptions == []
assert dependencies.subscription_manager.unsubscriptions == []
assert dependencies.subscription_manager.restore_calls == 0
assert dependencies.subscription_manager.clear_calls == 0
assert dependencies.event_publisher.events == []
assert dependencies.recovery_document_source.calls == []
assert composition.heartbeat_monitor.state is HeartbeatState.IDLE
assert (
composition.runtime_supervisor.state
is RuntimeSupervisorState.STOPPED
)
assert (
composition.reconnect_coordinator.state
is ReconnectState.DISCONNECTED
)
assert composition.runtime_scheduler.running is False
def test_live_checkpoint_is_visible_to_runtime_recovery() -> None:
checkpoint_trade = make_trade()
composition, dependencies = create_composition(
trade=checkpoint_trade,
recovery_document=[],
)
accepted_trade = (
composition.trade_stream_acquisition_service.handle_message(
{
"destination": "internal.trade",
}
)
)
result = composition.runtime_recovery_coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
assert accepted_trade is checkpoint_trade
assert result.requested_start_time == CHECKPOINT_TIME_MS
assert result.requested_end_time == RECOVERY_END_TIME_MS
assert result.is_empty is True
assert dependencies.recovery_document_source.calls == [
(
SYMBOL,
CHECKPOINT_TIME_MS,
RECOVERY_END_TIME_MS,
None,
)
]
def test_recovery_advances_shared_live_checkpoint() -> None:
recovered_trade_id = 101
composition, *_ = create_composition(
recovery_document=[
make_raw_trade(
trade_id=recovered_trade_id,
timestamp=CHECKPOINT_TIME_MS + 1_000,
),
],
)
composition.trade_stream_acquisition_service.handle_message(
{
"destination": "internal.trade",
}
)
result = composition.runtime_recovery_coordinator.recover(
symbol=SYMBOL,
recovery_end_time=RECOVERY_END_TIME_MS,
)
state = composition.state_store.get(
SYMBOL,
)
assert result.recovered_count == 1
assert result.last_trade is state.last_trade
assert state.last_trade_id == recovered_trade_id
assert state.last_trade is not None
assert state.last_trade.trade_id == recovered_trade_id
def test_subscription_uses_composed_runtime_service() -> None:
composition, dependencies = create_composition()
asyncio.run(
composition.trade_stream_acquisition_service.subscribe(
(
SYMBOL,
),
correlation_id="composition-test",
)
)
assert len(
dependencies.subscription_manager.subscriptions
) == 1
subscription_key, _ = (
dependencies.subscription_manager.subscriptions[0]
)
assert SYMBOL in subscription_key
assert dependencies.session.start_calls == 0
def test_reconnect_uses_composed_runtime_dependencies() -> None:
composition, dependencies = create_composition()
asyncio.run(
composition.reconnect_coordinator.reconnect()
)
assert dependencies.session.start_calls == 1
assert dependencies.subscription_manager.restore_calls == 1
assert len(dependencies.event_publisher.events) == 2
assert isinstance(
dependencies.event_publisher.events[0],
ReconnectStartedEvent,
)
assert isinstance(
dependencies.event_publisher.events[1],
ReconnectCompletedEvent,
)
def test_separate_compositions_have_independent_state() -> None:
first, _ = create_composition()
second, _ = create_composition()
assert first is not second
assert first.state_store is not second.state_store
assert (
first.consistency_controller
is not second.consistency_controller
)
assert (
first.runtime_recovery_coordinator
is not second.runtime_recovery_coordinator
)
assert (
first.runtime_supervisor
is not second.runtime_supervisor
)
assert (
first.runtime_scheduler
is not second.runtime_scheduler
)
@pytest.mark.parametrize(
"heartbeat_timeout_seconds",
[
0,
-1,
True,
"10",
],
)
def test_invalid_heartbeat_configuration_is_not_hidden(
heartbeat_timeout_seconds: object,
) -> None:
dependencies = CompositionDependencies(
session=FakeSession(),
transport=FakeTransport(),
subscription_manager=FakeSubscriptionManager(),
event_publisher=FakeEventPublisher(),
message_adapter=FakeMessageAdapter(
make_trade(),
),
recovery_document_source=StubTradesDocumentSource(
[],
),
heartbeat_clock=FakeClock(),
scheduler_sleep=RecordingSleep(),
)
with pytest.raises(
(TypeError, ValueError),
):
build_trade_stream_runtime_composition(
session=dependencies.session,
transport=dependencies.transport,
subscription_manager=dependencies.subscription_manager,
event_publisher=dependencies.event_publisher,
message_adapter=dependencies.message_adapter,
recovery_document_source=(
dependencies.recovery_document_source
),
heartbeat_timeout_seconds=heartbeat_timeout_seconds, # type: ignore[arg-type]
scheduler_interval_seconds=1.0,
heartbeat_clock=dependencies.heartbeat_clock,
scheduler_sleep=dependencies.scheduler_sleep,
)
@pytest.mark.parametrize(
"scheduler_interval_seconds",
[
0,
-1,
True,
"1",
],
)
def test_invalid_scheduler_configuration_is_not_hidden(
scheduler_interval_seconds: object,
) -> None:
with pytest.raises(
(TypeError, ValueError),
):
create_composition(
scheduler_interval_seconds=scheduler_interval_seconds, # type: ignore[arg-type]
)
@pytest.mark.parametrize(
"max_recovery_window_ms",
[
0,
-1,
True,
3_600_000,
],
)
def test_invalid_recovery_window_configuration_is_not_hidden(
max_recovery_window_ms: object,
) -> None:
with pytest.raises(
(TypeError, ValueError),
):
create_composition(
max_recovery_window_ms=max_recovery_window_ms, # type: ignore[arg-type]
)