1492 lines
60 KiB
Markdown
1492 lines
60 KiB
Markdown
# Build 060.18 — Trade Stream Consistency Controller
|
||
|
||
**Engineering Migration Report**
|
||
|
||
---
|
||
|
||
# Контроль документа
|
||
|
||
| Свойство | Значение |
|
||
|----------|----------|
|
||
| Build | 060.18 |
|
||
| Название | Trade Stream Consistency Controller |
|
||
| Статус | Completed |
|
||
| Проект | Dzentra |
|
||
| Подсистема | Market Data Acquisition |
|
||
| Компонент | Trade Stream Consistency |
|
||
| Версия | 1.0 |
|
||
|
||
---
|
||
|
||
# Связанные документы
|
||
|
||
- build_060_18_architecture.md — архитектурная спецификация Build.
|
||
- build_060_17.md — Engineering Migration Report предыдущего Build.
|
||
|
||
---
|
||
|
||
# Цель Build
|
||
|
||
После завершения Build 060.17 подсистема Market Data Acquisition получила полноценную инфраструктуру формирования канонической модели сделки (`Trade`).
|
||
|
||
К этому моменту архитектура уже обеспечивала:
|
||
|
||
- получение транспортных сообщений из различных источников;
|
||
- преобразование транспортных моделей в канонический объект `Trade`;
|
||
- единый механизм Parser;
|
||
- Value Validation;
|
||
- Mapper;
|
||
- Trade Adapter;
|
||
- инфраструктуру Feed.
|
||
|
||
Таким образом система уже умела получать отдельные корректные сделки независимо от источника их происхождения.
|
||
|
||
Однако корректность отдельной сделки ещё не означает корректность последовательности сделок.
|
||
|
||
Поток данных, поступающий от биржи, может содержать:
|
||
|
||
- повторную доставку уже опубликованных сделок;
|
||
- нарушение порядка поступления сообщений;
|
||
- конфликтующие дубликаты;
|
||
- повторную передачу одной и той же сделки через различные транспортные каналы.
|
||
|
||
До настоящего Build подобные ситуации никак не контролировались.
|
||
|
||
Каждый компонент, получающий объект `Trade`, был вынужден самостоятельно предполагать, что поток уже является корректным.
|
||
|
||
Подобная архитектура противоречила фундаментальному принципу Dzentra.
|
||
|
||
Согласованность потока должна обеспечиваться централизованно.
|
||
|
||
Потребители рыночных данных не должны повторно выполнять проверку порядка или дедупликацию.
|
||
|
||
Главной задачей Build 060.18 становится построение специализированной подсистемы **Trade Stream Consistency**, обеспечивающей формирование единственного канонического потока сделок внутри Acquisition Layer.
|
||
|
||
После завершения Build система получает:
|
||
|
||
- специализированный `TradeStreamConsistencyProtocol`;
|
||
- специализированный `TradeStreamConsistencyController`;
|
||
- внутреннюю модель состояния `TradeStreamState`;
|
||
- специализированные доменные исключения;
|
||
- полноценное unit-тестирование новой подсистемы.
|
||
|
||
При этом Build принципиально не затрагивает:
|
||
|
||
- Parser;
|
||
- Mapper;
|
||
- Value Validation;
|
||
- Trade Adapter;
|
||
- Handler;
|
||
- Feed;
|
||
- Runtime;
|
||
- Subscription Layer;
|
||
- REST Backfill;
|
||
- Gap Recovery;
|
||
- Reconnect;
|
||
- интеграцию с Runtime.
|
||
|
||
Все перечисленные задачи относятся к следующим этапам развития подсистемы Trades Feed.
|
||
|
||
---
|
||
|
||
# Предпосылки
|
||
|
||
К началу Build архитектура обработки сделок уже обеспечивала формирование единой канонической модели `Trade`.
|
||
|
||
Полный конвейер обработки выглядел следующим образом.
|
||
|
||
```text
|
||
Transport Message
|
||
│
|
||
▼
|
||
Schema Validation
|
||
│
|
||
▼
|
||
Parser
|
||
│
|
||
▼
|
||
Value Validation
|
||
│
|
||
▼
|
||
Mapper
|
||
│
|
||
▼
|
||
Trade
|
||
```
|
||
|
||
Каждый уровень обладал строго определённой областью ответственности.
|
||
|
||
Schema Validation отвечала за корректность транспортного документа.
|
||
|
||
Parser извлекал необходимые поля.
|
||
|
||
Value Validation проверяла корректность отдельных значений.
|
||
|
||
Mapper строил каноническую модель предметной области.
|
||
|
||
Полученный объект `Trade` уже являлся полностью независимым от транспортного формата и мог использоваться всеми последующими компонентами системы.
|
||
|
||
Однако существовал один принципиальный архитектурный пробел.
|
||
|
||
Система гарантировала корректность каждой отдельной сделки, но не гарантировала корректность последовательности этих сделок.
|
||
|
||
Например, следующий поток состоял исключительно из корректных объектов `Trade`.
|
||
|
||
```text
|
||
Trade #100
|
||
|
||
Trade #101
|
||
|
||
Trade #100
|
||
```
|
||
|
||
Каждая сделка по отдельности являлась полностью корректной.
|
||
|
||
Однако сам поток нарушал инвариант уникальности публикации.
|
||
|
||
Аналогично поток
|
||
|
||
```text
|
||
Trade #100
|
||
|
||
Trade #101
|
||
|
||
Trade #99
|
||
```
|
||
|
||
также состоял из корректных объектов `Trade`, но нарушал инвариант порядка.
|
||
|
||
Следовательно, между построением объекта `Trade` и его публикацией должен существовать самостоятельный уровень контроля согласованности потока.
|
||
|
||
Именно эту архитектурную задачу решает Build 060.18.
|
||
|
||
---
|
||
|
||
# Результаты архитектурного аудита
|
||
|
||
Перед началом реализации Build был выполнен полный аудит существующей подсистемы **Market Data Acquisition**.
|
||
|
||
Целью аудита являлась проверка соответствия фактической реализации архитектурной модели, утверждённой в `build_060_18_architecture.md`, а также определение точек интеграции новой подсистемы Stream Consistency.
|
||
|
||
Особое внимание уделялось существующей инфраструктуре получения сделок.
|
||
|
||
Первоначально предполагалось, что после реализации `TradeStreamConsistencyController` он будет интегрирован непосредственно в существующий `TradesFeed`.
|
||
|
||
Именно такое решение рассматривалось на этапе архитектурного проектирования.
|
||
|
||
Однако проведённый аудит показал, что фактическая структура подсистемы отличается от первоначальных предположений.
|
||
|
||
---
|
||
|
||
## Анализ существующего Trades Feed
|
||
|
||
В ходе проверки был полностью проанализирован существующий компонент:
|
||
|
||
```text
|
||
feeds/trades_feed.py
|
||
```
|
||
|
||
Аудит показал, что данный компонент реализует исключительно сценарий получения **одиночной сделки** посредством REST API.
|
||
|
||
Его публичный контракт имеет следующий вид.
|
||
|
||
```text
|
||
load_trade(symbol)
|
||
│
|
||
▼
|
||
Trade
|
||
```
|
||
|
||
Feed не содержит:
|
||
|
||
- непрерывного потока сообщений;
|
||
- подписок WebSocket;
|
||
- обработки последовательности сделок;
|
||
- собственного внутреннего состояния;
|
||
- механизма публикации Trade Consumer;
|
||
- инфраструктуры обработки событий.
|
||
|
||
Каждый вызов `load_trade()` полностью независим от предыдущих вызовов.
|
||
|
||
Таким образом существующий `TradesFeed` представляет собой stateless-компонент, предназначенный исключительно для получения отдельных канонических объектов `Trade`.
|
||
|
||
---
|
||
|
||
## Отсутствие потокового Feed
|
||
|
||
Дополнительно был проведён аудит инфраструктуры WebSocket.
|
||
|
||
Проверка показала, что к моменту реализации Build 060.18 в проекте отсутствует специализированный компонент, отвечающий за обработку непрерывного потока сделок.
|
||
|
||
Иными словами, архитектура уже содержит:
|
||
|
||
- WebSocket Runtime;
|
||
- Subscription Layer;
|
||
- Router;
|
||
- Handler;
|
||
- Adapter;
|
||
- каноническую модель `Trade`;
|
||
|
||
но ещё не содержит специализированного **WebSocket Trades Feed**, который бы организовывал непрерывное получение и публикацию последовательности сделок.
|
||
|
||
Это стало ключевым результатом архитектурного аудита.
|
||
|
||
---
|
||
|
||
## Последствия для реализации Build
|
||
|
||
Полученные результаты существенно повлияли на окончательную реализацию Build.
|
||
|
||
Интеграция `TradeStreamConsistencyController` в существующий REST Feed привела бы к смешению двух различных архитектурных уровней.
|
||
|
||
REST Feed отвечает за получение отдельной сделки.
|
||
|
||
Trade Stream Consistency отвечает за обработку непрерывного потока сделок.
|
||
|
||
Эти два сценария обладают различной природой и различными требованиями к состоянию системы.
|
||
|
||
Попытка объединить их в рамках одного компонента нарушила бы принцип единственной ответственности и создала бы искусственную зависимость между REST и потоковой обработкой.
|
||
|
||
Поэтому было принято решение отказаться от подобной интеграции.
|
||
|
||
---
|
||
|
||
# Архитектурное решение
|
||
|
||
По результатам проведённого аудита было принято решение оставить новую подсистему **Trade Stream Consistency** полностью самостоятельной.
|
||
|
||
В рамках Build реализуются только компоненты, непосредственно отвечающие за проверку согласованности потока.
|
||
|
||
Интеграция с инфраструктурой получения данных сознательно переносится на последующие этапы развития Trades Feed.
|
||
|
||
После завершения Build архитектура принимает следующий вид.
|
||
|
||
```text
|
||
Trade
|
||
│
|
||
▼
|
||
TradeStreamConsistencyController
|
||
│
|
||
▼
|
||
Canonical Trade Stream
|
||
```
|
||
|
||
Таким образом Build 060.18 завершает построение самостоятельного слоя согласованности потока, не изменяя существующую инфраструктуру получения сделок.
|
||
|
||
Это решение обеспечивает слабую связанность компонентов и позволяет независимо развивать:
|
||
|
||
- инфраструктуру получения данных;
|
||
- механизмы восстановления потока;
|
||
- обработку разрывов последовательности;
|
||
- механизмы повторной синхронизации.
|
||
|
||
Все перечисленные возможности смогут использовать уже готовый `TradeStreamConsistencyController` без изменения его внутренней реализации.
|
||
|
||
---
|
||
|
||
# Почему Controller не интегрирован в существующий Feed
|
||
|
||
На этапе архитектурного проектирования предполагалось, что новой подсистеме потребуется непосредственная интеграция в `TradesFeed`.
|
||
|
||
Однако инженерный аудит показал, что такая интеграция является преждевременной.
|
||
|
||
Причина заключается в различии ответственности компонентов.
|
||
|
||
`TradesFeed` в текущей реализации получает отдельную сделку.
|
||
|
||
`TradeStreamConsistencyController` принимает решения исключительно относительно непрерывного потока сделок.
|
||
|
||
Следовательно, контроллер не может эффективно использоваться до появления полноценного потокового Feed.
|
||
|
||
В результате было принято следующее окончательное решение Build.
|
||
|
||
Настоящий Build завершает реализацию самостоятельной подсистемы Stream Consistency, полностью готовой к использованию.
|
||
|
||
Её интеграция будет выполнена после появления специализированного **WebSocket Trades Feed**, который станет источником непрерывного потока сделок.
|
||
|
||
Подобное решение позволило сохранить архитектурную чистоту проекта и избежать появления технического долга на раннем этапе развития подсистемы.
|
||
|
||
---
|
||
|
||
# Новая подсистема Trade Stream Consistency
|
||
|
||
Главным результатом настоящего Build становится появление в подсистеме **Market Data Acquisition** нового архитектурного уровня — **Trade Stream Consistency**.
|
||
|
||
До начала Build система завершала обработку сделки сразу после построения канонической модели `Trade`.
|
||
|
||
Конвейер обработки имел следующий вид.
|
||
|
||
```text
|
||
Transport Message
|
||
│
|
||
▼
|
||
Schema Validation
|
||
│
|
||
▼
|
||
Parser
|
||
│
|
||
▼
|
||
Value Validation
|
||
│
|
||
▼
|
||
Mapper
|
||
│
|
||
▼
|
||
Trade
|
||
```
|
||
|
||
После завершения Build между построением канонической модели и её публикацией появляется дополнительный уровень.
|
||
|
||
```text
|
||
Transport Message
|
||
│
|
||
▼
|
||
Schema Validation
|
||
│
|
||
▼
|
||
Parser
|
||
│
|
||
▼
|
||
Value Validation
|
||
│
|
||
▼
|
||
Mapper
|
||
│
|
||
▼
|
||
Trade
|
||
│
|
||
▼
|
||
Trade Stream Consistency
|
||
│
|
||
▼
|
||
Canonical Trade Stream
|
||
```
|
||
|
||
Появление данного уровня является принципиальным изменением архитектуры Acquisition Layer.
|
||
|
||
Если ранее система гарантировала корректность отдельных объектов `Trade`, то теперь она гарантирует корректность всей последовательности опубликованных сделок.
|
||
|
||
Именно последовательность становится новой доменной сущностью.
|
||
|
||
---
|
||
|
||
# Архитектурное решение
|
||
|
||
Во время проектирования рассматривались несколько вариантов реализации проверки согласованности.
|
||
|
||
Первый вариант предполагал распределение логики между различными компонентами Acquisition Layer.
|
||
|
||
Например:
|
||
|
||
- часть проверки выполнять внутри Feed;
|
||
- часть — внутри Handler;
|
||
- часть — внутри Runtime.
|
||
|
||
После анализа архитектуры данный подход был отклонён.
|
||
|
||
Подобное распределение приводило к нескольким серьёзным недостаткам.
|
||
|
||
Во-первых, логика проверки потока оказывалась размазанной между различными уровнями системы.
|
||
|
||
Во-вторых, различные источники данных могли реализовывать разные правила проверки.
|
||
|
||
В-третьих, последующее развитие механизмов Recovery и REST Backfill существенно усложнялось.
|
||
|
||
Поэтому было принято другое решение.
|
||
|
||
Вся логика проверки согласованности концентрируется внутри специализированной подсистемы.
|
||
|
||
Она становится единственной точкой формирования канонического потока сделок.
|
||
|
||
---
|
||
|
||
# Архитектура новой подсистемы
|
||
|
||
В рамках Build реализованы четыре новых компонента.
|
||
|
||
```text
|
||
TradeStreamConsistencyProtocol
|
||
|
||
TradeStreamConsistencyController
|
||
|
||
TradeStreamState
|
||
|
||
Trade Stream Exceptions
|
||
```
|
||
|
||
Каждый компонент обладает собственной областью ответственности.
|
||
|
||
Ни один компонент не выполняет обязанности другого.
|
||
|
||
Подобное разделение полностью соответствует принципу **Single Responsibility**, принятому в архитектуре Dzentra.
|
||
|
||
---
|
||
|
||
# TradeStreamConsistencyProtocol
|
||
|
||
Одной из целей Build являлось формирование полноценного контрактного уровня новой подсистемы.
|
||
|
||
До начала Build соответствующий Protocol отсутствовал.
|
||
|
||
В рамках реализации был добавлен новый контракт.
|
||
|
||
```text
|
||
TradeStreamConsistencyProtocol
|
||
```
|
||
|
||
Protocol определяет единственную публичную операцию.
|
||
|
||
```text
|
||
accept(trade)
|
||
│
|
||
▼
|
||
Trade | None
|
||
```
|
||
|
||
Никаких других обязанностей Protocol не содержит.
|
||
|
||
Он не определяет:
|
||
|
||
- внутреннее состояние;
|
||
- способы хранения данных;
|
||
- размер окна дедупликации;
|
||
- алгоритмы проверки;
|
||
- механизмы публикации.
|
||
|
||
Все перечисленные детали относятся исключительно к реализации Controller.
|
||
|
||
Благодаря подобному разделению любые последующие компоненты системы смогут зависеть только от контракта, а не от конкретной реализации.
|
||
|
||
Это полностью соответствует принципу **Dependency Inversion**, принятому в проекте Dzentra.
|
||
|
||
---
|
||
|
||
# Почему выбран минимальный Protocol
|
||
|
||
Во время проектирования рассматривались различные варианты публичного API.
|
||
|
||
В частности анализировались варианты:
|
||
|
||
- возврата логического значения (`bool`);
|
||
- использования специализированного объекта результата;
|
||
- публикации событий вместо возврата значения.
|
||
|
||
После анализа было принято решение оставить контракт максимально простым.
|
||
|
||
Контроллер принимает объект `Trade` и возвращает либо этот же объект, либо `None`.
|
||
|
||
Нарушения архитектурных инвариантов выражаются специализированными исключениями.
|
||
|
||
Такой контракт оказался наиболее устойчивым к дальнейшему развитию системы.
|
||
|
||
Он одинаково хорошо подходит для:
|
||
|
||
- WebSocket Feed;
|
||
- REST Backfill;
|
||
- Recovery Pipeline;
|
||
- Integration Tests;
|
||
- Unit Tests.
|
||
|
||
При этом публичный интерфейс остаётся минимальным и легко читаемым.
|
||
|
||
---
|
||
|
||
# Новые доменные исключения
|
||
|
||
Следующим результатом Build становится появление специализированных исключений подсистемы Stream Consistency.
|
||
|
||
До настоящего Build подобные ошибки отсутствовали.
|
||
|
||
В рамках реализации добавлены два новых класса.
|
||
|
||
```text
|
||
TradeOrderingError
|
||
|
||
TradeConsistencyError
|
||
```
|
||
|
||
Каждый тип ошибки отражает отдельное нарушение архитектурных инвариантов канонического потока.
|
||
|
||
Разделение исключений позволяет вызывающему компоненту принимать различные решения в зависимости от характера проблемы.
|
||
|
||
Например, нарушение порядка и конфликтующий дубликат имеют различную природу и требуют различной стратегии обработки.
|
||
|
||
Поэтому использование единственного общего исключения было признано нецелесообразным.
|
||
|
||
Оба класса наследуются от общей иерархии исключений подсистемы Market Data Acquisition и полностью соответствуют существующей архитектуре проекта.
|
||
|
||
---
|
||
|
||
# TradeStreamConsistencyController
|
||
|
||
Центральным компонентом настоящего Build становится
|
||
|
||
```text
|
||
TradeStreamConsistencyController
|
||
```
|
||
|
||
Именно он завершает формирование новой подсистемы **Trade Stream Consistency** внутри Acquisition Layer.
|
||
|
||
До начала Build все компоненты системы были stateless.
|
||
|
||
Parser не хранил состояние.
|
||
|
||
Value Validation не хранила состояние.
|
||
|
||
Mapper не хранил состояние.
|
||
|
||
Trade Adapter не хранил состояние.
|
||
|
||
Handler не хранил состояние.
|
||
|
||
Feed также не содержал собственного состояния.
|
||
|
||
Появление Trade Stream Consistency впервые вводит в подсистему компонент, принимающий решения на основании ранее обработанных сделок.
|
||
|
||
Именно поэтому Controller становится первой stateful-службой внутри Acquisition Layer.
|
||
|
||
---
|
||
|
||
## Архитектура Controller
|
||
|
||
Конструкция Controller намеренно сделана максимально простой.
|
||
|
||
Он содержит только одно внутреннее хранилище.
|
||
|
||
```text
|
||
symbol
|
||
│
|
||
▼
|
||
TradeStreamState
|
||
```
|
||
|
||
Для каждого торгового символа существует собственный экземпляр состояния.
|
||
|
||
Например,
|
||
|
||
```text
|
||
BTCUSDT
|
||
│
|
||
▼
|
||
TradeStreamState
|
||
```
|
||
|
||
и
|
||
|
||
```text
|
||
ETHUSDT
|
||
│
|
||
▼
|
||
TradeStreamState
|
||
```
|
||
|
||
обслуживаются полностью независимо друг от друга.
|
||
|
||
Controller не хранит информацию о самих сделках.
|
||
|
||
Он лишь определяет, какому состоянию необходимо передать очередную сделку для проверки.
|
||
|
||
---
|
||
|
||
## Ответственность Controller
|
||
|
||
Во время проектирования особое внимание уделялось разделению ответственности между компонентами новой подсистемы.
|
||
|
||
В результате Controller получил исключительно координационные обязанности.
|
||
|
||
Он отвечает за:
|
||
|
||
- выбор состояния по symbol;
|
||
- ленивое создание нового состояния;
|
||
- маршрутизацию сделки;
|
||
- возврат результата проверки вызывающему компоненту.
|
||
|
||
При этом Controller сознательно не реализует:
|
||
|
||
- алгоритм дедупликации;
|
||
- проверку порядка;
|
||
- хранение окна сделок;
|
||
- сравнение Trade;
|
||
- управление FIFO.
|
||
|
||
Все перечисленные задачи полностью принадлежат TradeStreamState.
|
||
|
||
Подобное разделение значительно упрощает дальнейшее развитие системы.
|
||
|
||
---
|
||
|
||
# TradeStreamState
|
||
|
||
Вторым ключевым компонентом новой подсистемы становится
|
||
|
||
```text
|
||
TradeStreamState
|
||
```
|
||
|
||
Если Controller представляет собой уровень координации, то State представляет собой уровень хранения состояния и проверки инвариантов потока.
|
||
|
||
Каждый экземпляр TradeStreamState обслуживает только один торговый символ.
|
||
|
||
Это является одним из фундаментальных архитектурных принципов Build.
|
||
|
||
Благодаря подобному решению состояние различных инструментов никогда не смешивается между собой.
|
||
|
||
Поток BTCUSDT полностью независим от потока ETHUSDT.
|
||
|
||
Даже совпадающие значения `trade_id` не создают никаких конфликтов.
|
||
|
||
---
|
||
|
||
## Внутреннее состояние
|
||
|
||
Каждый экземпляр TradeStreamState хранит минимальный объём информации, необходимый для проверки согласованности потока.
|
||
|
||
Внутреннее состояние включает:
|
||
|
||
```text
|
||
symbol
|
||
|
||
last_trade_id
|
||
|
||
FIFO Window
|
||
|
||
Dictionary Trade Cache
|
||
```
|
||
|
||
Этого набора данных достаточно для реализации всех архитектурных инвариантов Build 060.18.
|
||
|
||
---
|
||
|
||
## last_trade_id
|
||
|
||
Поле
|
||
|
||
```text
|
||
last_trade_id
|
||
```
|
||
|
||
хранит максимальный идентификатор сделки, успешно опубликованной системой для данного символа.
|
||
|
||
Следует подчеркнуть, что речь идёт именно об опубликованной сделке.
|
||
|
||
Если поступает сделка, нарушающая порядок,
|
||
|
||
```text
|
||
100
|
||
|
||
101
|
||
|
||
95
|
||
```
|
||
|
||
то состояние не изменяется.
|
||
|
||
После возникновения `TradeOrderingError`
|
||
|
||
значение
|
||
|
||
```text
|
||
last_trade_id
|
||
```
|
||
|
||
остаётся равным
|
||
|
||
```text
|
||
101
|
||
```
|
||
|
||
Подобное поведение обеспечивает атомарность всех операций Controller.
|
||
|
||
Ошибочная сделка никогда не влияет на состояние потока.
|
||
|
||
---
|
||
|
||
## FIFO Window
|
||
|
||
Для проверки повторов TradeStreamState использует ограниченное окно ранее опубликованных сделок.
|
||
|
||
Во время проектирования рассматривались различные варианты хранения.
|
||
|
||
В частности анализировались:
|
||
|
||
- полная история сделок;
|
||
- HashSet;
|
||
- OrderedDict;
|
||
- LRU Cache;
|
||
- Ring Buffer.
|
||
|
||
После анализа было принято решение использовать ограниченное FIFO-окно.
|
||
|
||
Такое решение наиболее точно соответствует природе непрерывного потока сделок.
|
||
|
||
По мере поступления новых сделок самые старые записи постепенно удаляются из памяти.
|
||
|
||
Это позволяет ограничить использование памяти независимо от времени работы процесса.
|
||
|
||
---
|
||
|
||
## Используемые структуры данных
|
||
|
||
Внутренняя реализация TradeStreamState построена на совместном использовании двух стандартных структур данных Python.
|
||
|
||
Для хранения порядка поступления используется
|
||
|
||
```text
|
||
deque
|
||
```
|
||
|
||
Для быстрого поиска зарегистрированных сделок используется
|
||
|
||
```text
|
||
dict
|
||
```
|
||
|
||
Совместное использование этих структур обеспечивает выполнение всех основных операций со средней сложностью
|
||
|
||
```text
|
||
O(1)
|
||
```
|
||
|
||
В частности:
|
||
|
||
- поиск зарегистрированной сделки;
|
||
- регистрация новой сделки;
|
||
- удаление самой старой записи;
|
||
- поддержание фиксированного размера окна.
|
||
|
||
Подобная комбинация оказалась наиболее простой, эффективной и полностью удовлетворяющей требованиям настоящего Build.
|
||
|
||
---
|
||
|
||
## Размер окна дедупликации
|
||
|
||
По умолчанию TradeStreamState использует окно размером
|
||
|
||
```text
|
||
10 000
|
||
```
|
||
|
||
сделок.
|
||
|
||
Данное значение было выбрано как разумный компромисс между объёмом используемой памяти и вероятностью повторной доставки одной и той же сделки.
|
||
|
||
При этом размер окна не является архитектурным ограничением.
|
||
|
||
Он задаётся отдельным параметром при создании состояния и может быть изменён без каких-либо изменений внутреннего алгоритма Controller.
|
||
|
||
Таким образом Build фиксирует только принцип использования ограниченного окна, но не навязывает конкретный объём хранения для всех последующих реализаций.
|
||
|
||
---
|
||
|
||
# Алгоритм обработки сделки
|
||
|
||
После завершения реализации Build поведение новой подсистемы становится полностью детерминированным.
|
||
|
||
Каждая входящая сделка проходит одну и ту же последовательность проверок.
|
||
|
||
Результат обработки зависит исключительно от текущего состояния потока и содержимого входящей сделки.
|
||
|
||
Никакие внешние факторы не влияют на принятие решения.
|
||
|
||
Полный алгоритм обработки выглядит следующим образом.
|
||
|
||
```text
|
||
Trade
|
||
│
|
||
▼
|
||
Получение состояния symbol
|
||
│
|
||
▼
|
||
Поиск trade_id
|
||
│
|
||
├─────────────── Найден ───────────────┐
|
||
│ │
|
||
▼ ▼
|
||
Сравнение объектов Trade Новый trade_id
|
||
│ │
|
||
├─────────────── Совпадают ─────────────┤
|
||
│ │
|
||
▼ ▼
|
||
Duplicate Проверка порядка
|
||
│ │
|
||
▼ ▼
|
||
return None trade_id < last_trade_id ?
|
||
│
|
||
┌───────────────┴───────────────┐
|
||
▼ ▼
|
||
TradeOrderingError Регистрация сделки
|
||
│
|
||
▼
|
||
Обновление состояния
|
||
│
|
||
▼
|
||
return Trade
|
||
```
|
||
|
||
Подобная последовательность является обязательной.
|
||
|
||
Изменение порядка выполнения проверок приведёт к нарушению архитектурных инвариантов новой подсистемы.
|
||
|
||
---
|
||
|
||
# Почему сначала проверяется дедупликация
|
||
|
||
Во время проектирования отдельно анализировалась последовательность выполнения проверок.
|
||
|
||
На первый взгляд могло показаться естественным сначала проверять порядок, а уже затем выполнять поиск повторов.
|
||
|
||
Однако данный вариант оказался ошибочным.
|
||
|
||
Рассмотрим следующий поток.
|
||
|
||
```text
|
||
100
|
||
|
||
101
|
||
|
||
100
|
||
```
|
||
|
||
Если первой выполняется проверка порядка, система обнаруживает уменьшение `trade_id` и немедленно генерирует `TradeOrderingError`.
|
||
|
||
Однако в действительности последняя сделка не является нарушением порядка.
|
||
|
||
Она представляет собой корректную повторную доставку уже опубликованной сделки.
|
||
|
||
Следовательно, подобная ситуация должна обрабатываться как обычный Duplicate.
|
||
|
||
Именно поэтому поиск зарегистрированной сделки всегда выполняется раньше проверки порядка.
|
||
|
||
Данное решение было окончательно закреплено в процессе реализации и подтверждено соответствующими unit-тестами.
|
||
|
||
---
|
||
|
||
# Проверка порядка
|
||
|
||
Если входящая сделка отсутствует в окне дедупликации, она рассматривается как новая.
|
||
|
||
После этого выполняется проверка монотонности последовательности.
|
||
|
||
Единственным критерием является значение
|
||
|
||
```text
|
||
trade_id
|
||
```
|
||
|
||
Для каждого символа должно выполняться условие.
|
||
|
||
```text
|
||
trade_id(new) >= last_trade_id
|
||
```
|
||
|
||
При этом система сознательно допускает наличие разрывов последовательности.
|
||
|
||
Например,
|
||
|
||
```text
|
||
100
|
||
|
||
101
|
||
|
||
150
|
||
```
|
||
|
||
является полностью корректным потоком.
|
||
|
||
Настоящий Build не занимается анализом причин возникновения подобных разрывов.
|
||
|
||
Он лишь фиксирует факт отсутствия нарушения порядка.
|
||
|
||
Задача обнаружения пропусков относится к следующему этапу развития подсистемы.
|
||
|
||
---
|
||
|
||
# Обработка повторов
|
||
|
||
После определения идентичности сделки возможны два различных варианта повторной доставки.
|
||
|
||
Первый вариант представляет собой полный повтор ранее опубликованной сделки.
|
||
|
||
Во втором случае идентичность совпадает, однако содержимое сделки отличается.
|
||
|
||
Эти ситуации обладают различной природой и требуют различной реакции системы.
|
||
|
||
---
|
||
|
||
## Полный дубликат
|
||
|
||
Если ранее зарегистрированная сделка полностью совпадает с новой по всем каноническим полям, повтор считается корректным.
|
||
|
||
Контроллер не публикует такую сделку повторно.
|
||
|
||
Ошибки при этом не возникает.
|
||
|
||
Метод
|
||
|
||
```python
|
||
accept()
|
||
```
|
||
|
||
возвращает
|
||
|
||
```python
|
||
None
|
||
```
|
||
|
||
Подобное поведение рассматривается как нормальная рабочая ситуация при повторной доставке сообщений транспортным уровнем.
|
||
|
||
---
|
||
|
||
## Конфликтующий дубликат
|
||
|
||
Совершенно иной характер имеет ситуация, при которой идентичность сделки совпадает, но содержимое отличается.
|
||
|
||
Например,
|
||
|
||
```text
|
||
trade_id = 500
|
||
|
||
price = 100
|
||
```
|
||
|
||
позже
|
||
|
||
```text
|
||
trade_id = 500
|
||
|
||
price = 101
|
||
```
|
||
|
||
Подобная ситуация означает внутреннее противоречие данных.
|
||
|
||
С точки зрения канонической модели две различные сделки не могут обладать одинаковой идентичностью.
|
||
|
||
В этом случае Controller немедленно прекращает обработку и генерирует
|
||
|
||
```text
|
||
TradeConsistencyError
|
||
```
|
||
|
||
Состояние потока при этом остаётся неизменным.
|
||
|
||
---
|
||
|
||
# Атомарность операций
|
||
|
||
Одним из фундаментальных требований настоящего Build являлось обеспечение атомарности обработки каждой сделки.
|
||
|
||
Любая ошибка должна приводить к полному откату текущей операции.
|
||
|
||
Если в процессе проверки возникает:
|
||
|
||
- `TradeOrderingError`;
|
||
- `TradeConsistencyError`;
|
||
|
||
никакие изменения внутреннего состояния не выполняются.
|
||
|
||
Не изменяются:
|
||
|
||
- `last_trade_id`;
|
||
- окно дедупликации;
|
||
- словарь зарегистрированных сделок.
|
||
|
||
Таким образом после возникновения ошибки Controller остаётся в том же состоянии, в котором находился до начала обработки.
|
||
|
||
Подобное решение существенно упрощает последующую реализацию Recovery Pipeline и гарантирует внутреннюю согласованность состояния независимо от количества ошибок транспортного уровня.
|
||
|
||
---
|
||
|
||
# Производительность
|
||
|
||
Во время проектирования новой подсистемы одним из обязательных требований являлось сохранение постоянной сложности основных операций.
|
||
|
||
Поскольку Controller будет использоваться при обработке непрерывного потока сделок, любые алгоритмы с линейной сложностью быстро стали бы узким местом всей Acquisition Layer.
|
||
|
||
Именно поэтому внутренняя реализация была построена таким образом, чтобы обеспечить среднюю сложность
|
||
|
||
```text
|
||
O(1)
|
||
```
|
||
|
||
для всех наиболее часто выполняемых операций.
|
||
|
||
В частности:
|
||
|
||
| Операция | Средняя сложность |
|
||
|----------|-------------------|
|
||
| Поиск зарегистрированной сделки | O(1) |
|
||
| Регистрация новой сделки | O(1) |
|
||
| Проверка дубликата | O(1) |
|
||
| Удаление самой старой записи | O(1) |
|
||
| Получение состояния symbol | O(1) |
|
||
|
||
Использование памяти ограничивается исключительно размером окна дедупликации.
|
||
|
||
В результате объём памяти остаётся постоянным независимо от продолжительности работы процесса.
|
||
|
||
Это позволяет использовать Controller в составе долгоживущих потоковых сервисов без риска неограниченного роста потребления памяти.
|
||
|
||
---
|
||
|
||
# Изменённые файлы
|
||
|
||
В рамках Build были добавлены четыре новых компонента подсистемы **Trade Stream Consistency**.
|
||
|
||
Все изменения были сознательно локализованы внутри нового каталога
|
||
|
||
```text
|
||
src/market_data/acquisition/consistency/
|
||
```
|
||
|
||
Подобное решение позволило полностью изолировать новую функциональность от уже существующих компонентов Acquisition Layer.
|
||
|
||
Ни один ранее реализованный Parser, Mapper, Handler или Adapter не потребовал изменений.
|
||
|
||
---
|
||
|
||
## Trade Stream Protocol
|
||
|
||
```text
|
||
src/market_data/acquisition/consistency/trade_stream_protocol.py
|
||
```
|
||
|
||
Добавлен новый Protocol.
|
||
|
||
```text
|
||
TradeStreamConsistencyProtocol
|
||
```
|
||
|
||
Protocol определяет единственный публичный контракт новой подсистемы.
|
||
|
||
```python
|
||
accept(trade: Trade) -> Trade | None
|
||
```
|
||
|
||
Благодаря этому все последующие компоненты системы смогут зависеть исключительно от абстракции, а не от конкретной реализации Controller.
|
||
|
||
---
|
||
|
||
## Trade Stream Exceptions
|
||
|
||
```text
|
||
src/market_data/acquisition/consistency/trade_stream_exceptions.py
|
||
```
|
||
|
||
Добавлены два специализированных доменных исключения.
|
||
|
||
```text
|
||
TradeOrderingError
|
||
|
||
TradeConsistencyError
|
||
```
|
||
|
||
Оба класса наследуются от общей иерархии исключений Acquisition Layer и используются исключительно новой подсистемой Stream Consistency.
|
||
|
||
Разделение ошибок позволяет вызывающему компоненту различать нарушение порядка и внутреннюю противоречивость потока.
|
||
|
||
---
|
||
|
||
## Trade Stream State
|
||
|
||
```text
|
||
src/market_data/acquisition/consistency/trade_stream_state.py
|
||
```
|
||
|
||
Реализована внутренняя модель состояния одного торгового символа.
|
||
|
||
TradeStreamState отвечает за:
|
||
|
||
- хранение последнего опубликованного `trade_id`;
|
||
- проверку принадлежности символа;
|
||
- дедупликацию сделок;
|
||
- обнаружение конфликтующих повторов;
|
||
- поддержку ограниченного FIFO-окна;
|
||
- обновление внутреннего состояния после успешной публикации сделки.
|
||
|
||
Вся бизнес-логика проверки согласованности сосредоточена именно внутри данного компонента.
|
||
|
||
---
|
||
|
||
## Trade Stream Consistency Controller
|
||
|
||
```text
|
||
src/market_data/acquisition/consistency/trade_stream_consistency_controller.py
|
||
```
|
||
|
||
Реализован компонент
|
||
|
||
```text
|
||
TradeStreamConsistencyController
|
||
```
|
||
|
||
Controller отвечает исключительно за:
|
||
|
||
- выбор состояния по symbol;
|
||
- ленивое создание новых состояний;
|
||
- маршрутизацию сделки;
|
||
- возврат результата вызывающему компоненту.
|
||
|
||
Внутренняя логика проверки полностью делегируется соответствующему экземпляру `TradeStreamState`.
|
||
|
||
Благодаря подобному разделению Controller остаётся компактным координационным компонентом и не зависит от деталей хранения состояния.
|
||
|
||
---
|
||
|
||
# Добавленные unit-тесты
|
||
|
||
Настоящий Build сопровождается полноценным покрытием новой подсистемы unit-тестами.
|
||
|
||
Все тесты написаны исключительно через публичный API компонентов.
|
||
|
||
Внутренние структуры данных не используются напрямую.
|
||
|
||
Подобный подход позволяет свободно изменять внутреннюю реализацию без изменения тестового набора.
|
||
|
||
---
|
||
|
||
## TradeStreamState
|
||
|
||
Добавлен новый файл.
|
||
|
||
```text
|
||
tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py
|
||
```
|
||
|
||
Проверяются следующие сценарии.
|
||
|
||
---
|
||
|
||
### Создание состояния
|
||
|
||
Подтверждается корректная инициализация нового состояния.
|
||
|
||
Проверяются:
|
||
|
||
- symbol;
|
||
- размер окна;
|
||
- отсутствие зарегистрированных сделок.
|
||
|
||
---
|
||
|
||
### Первая сделка
|
||
|
||
Подтверждается успешная публикация первой сделки нового символа.
|
||
|
||
---
|
||
|
||
### Возрастающий trade_id
|
||
|
||
Проверяется корректная обработка монотонно возрастающей последовательности сделок.
|
||
|
||
---
|
||
|
||
### Разрыв последовательности
|
||
|
||
Подтверждается, что пропуски идентификаторов не рассматриваются как ошибка.
|
||
|
||
---
|
||
|
||
### Полный дубликат
|
||
|
||
Подтверждается возврат
|
||
|
||
```python
|
||
None
|
||
```
|
||
|
||
при повторной доставке идентичной сделки.
|
||
|
||
---
|
||
|
||
### Конфликтующий дубликат
|
||
|
||
Проверяется генерация
|
||
|
||
```text
|
||
TradeConsistencyError
|
||
```
|
||
|
||
при несовпадении содержимого сделки.
|
||
|
||
---
|
||
|
||
### Нарушение порядка
|
||
|
||
Проверяется генерация
|
||
|
||
```text
|
||
TradeOrderingError
|
||
```
|
||
|
||
при уменьшении `trade_id`.
|
||
|
||
---
|
||
|
||
### Работа FIFO
|
||
|
||
Проверяется корректное удаление наиболее старой записи после переполнения окна.
|
||
|
||
---
|
||
|
||
### Повтор после выхода из окна
|
||
|
||
Подтверждается, что сделка, удалённая из окна дедупликации, больше не рассматривается как известная системе.
|
||
|
||
---
|
||
|
||
### Проверка symbol
|
||
|
||
Подтверждается невозможность передачи сделки другого торгового символа.
|
||
|
||
---
|
||
|
||
### Проверка конфигурации
|
||
|
||
Проверяется корректная обработка недопустимого размера окна дедупликации.
|
||
|
||
---
|
||
|
||
## TradeStreamConsistencyController
|
||
|
||
Добавлен новый файл.
|
||
|
||
```text
|
||
tests/unit/market_data/acquisition/consistency/test_trade_stream_consistency_controller.py
|
||
```
|
||
|
||
Проверяются следующие сценарии.
|
||
|
||
---
|
||
|
||
### Ленивое создание состояния
|
||
|
||
Подтверждается автоматическое создание нового `TradeStreamState` при первом появлении символа.
|
||
|
||
---
|
||
|
||
### Повторное использование состояния
|
||
|
||
Проверяется, что для всех последующих сделок одного символа используется уже существующее состояние.
|
||
|
||
---
|
||
|
||
### Независимость символов
|
||
|
||
Подтверждается полная независимость состояний различных торговых инструментов.
|
||
|
||
---
|
||
|
||
### Обработка полного дубликата
|
||
|
||
Подтверждается корректная передача результата
|
||
|
||
```python
|
||
None
|
||
```
|
||
|
||
вызывающему компоненту.
|
||
|
||
---
|
||
|
||
### Передача исключений
|
||
|
||
Проверяется, что Controller не подавляет:
|
||
|
||
- `TradeOrderingError`;
|
||
- `TradeConsistencyError`;
|
||
|
||
а корректно передаёт их вызывающему компоненту.
|
||
|
||
---
|
||
|
||
# Результаты тестирования
|
||
|
||
После завершения реализации был выполнен запуск полного набора unit-тестов новой подсистемы.
|
||
|
||
Использовались команды.
|
||
|
||
```bash
|
||
python -m pytest -q \
|
||
tests/unit/market_data/acquisition/consistency/test_trade_stream_state.py
|
||
```
|
||
|
||
Результат.
|
||
|
||
```text
|
||
13 passed
|
||
```
|
||
|
||
После этого была выполнена проверка Controller.
|
||
|
||
```bash
|
||
python -m pytest -q \
|
||
tests/unit/market_data/acquisition/consistency/test_trade_stream_consistency_controller.py
|
||
```
|
||
|
||
Результат.
|
||
|
||
```text
|
||
6 passed
|
||
```
|
||
|
||
Итоговый результат настоящего Build.
|
||
|
||
```text
|
||
19 passed
|
||
```
|
||
|
||
Все предусмотренные сценарии успешно пройдены.
|
||
|
||
Тестирование подтвердило:
|
||
|
||
- корректность проверки порядка;
|
||
- корректность дедупликации;
|
||
- независимость символов;
|
||
- атомарность операций;
|
||
- корректную работу FIFO-окна;
|
||
- соответствие Controller утверждённой архитектуре.
|
||
|
||
Ни одного отклонения от архитектурной спецификации обнаружено не было.
|
||
|
||
---
|
||
|
||
# Итоги Build
|
||
|
||
Build 060.18 полностью завершает построение слоя **Trade Stream Consistency** внутри подсистемы Market Data Acquisition.
|
||
|
||
До начала настоящего Build система гарантировала корректность отдельных объектов `Trade`.
|
||
|
||
После завершения Build система дополнительно гарантирует корректность последовательности публикуемых сделок.
|
||
|
||
Таким образом ответственность Acquisition Layer расширяется.
|
||
|
||
Теперь подсистема обеспечивает:
|
||
|
||
- построение канонической модели сделки;
|
||
- проверку корректности потока;
|
||
- обнаружение повторной доставки;
|
||
- обнаружение конфликтующих дубликатов;
|
||
- контроль монотонности последовательности;
|
||
- формирование единственного канонического потока сделок.
|
||
|
||
При этом все существующие компоненты системы продолжают работать без изменений.
|
||
|
||
Build не нарушил обратную совместимость и не потребовал модификации ранее реализованных Parser, Mapper, Adapter или Handler.
|
||
|
||
---
|
||
|
||
# Архитектурный результат
|
||
|
||
Главным архитектурным результатом настоящего Build становится появление нового самостоятельного уровня Acquisition Layer.
|
||
|
||
Теперь общая архитектура обработки сделок принимает следующий вид.
|
||
|
||
```text
|
||
Transport Message
|
||
│
|
||
▼
|
||
Schema Validation
|
||
│
|
||
▼
|
||
Parser
|
||
│
|
||
▼
|
||
Value Validation
|
||
│
|
||
▼
|
||
Mapper
|
||
│
|
||
▼
|
||
Trade
|
||
│
|
||
▼
|
||
Trade Stream Consistency
|
||
│
|
||
▼
|
||
Canonical Trade Stream
|
||
```
|
||
|
||
Данный уровень полностью изолирован от транспортной реализации и работает исключительно с канонической моделью предметной области.
|
||
|
||
Это означает, что независимо от источника данных — WebSocket, REST Backfill или Recovery Pipeline — все сделки будут проходить через единый механизм проверки согласованности.
|
||
|
||
Подобное решение исключает дублирование логики и гарантирует единообразное поведение системы.
|
||
|
||
---
|
||
|
||
# Что сознательно НЕ реализовано
|
||
|
||
В соответствии с утверждёнными границами Build 060.18 ряд задач был сознательно оставлен за пределами реализации.
|
||
|
||
Настоящий Build **не включает**:
|
||
|
||
- интеграцию с существующим REST `TradesFeed`;
|
||
- реализацию WebSocket Trades Feed;
|
||
- обнаружение пропусков последовательности (`Gap Detection`);
|
||
- автоматический REST Backfill;
|
||
- механизм восстановления потока (`Recovery Pipeline`);
|
||
- повторную синхронизацию после разрыва соединения;
|
||
- публикацию событий подписчикам;
|
||
- буферизацию или повторную доставку сообщений;
|
||
- управление жизненным циклом WebSocket-соединений.
|
||
|
||
Все перечисленные возможности относятся к последующим этапам развития подсистемы и будут использовать уже реализованный слой Trade Stream Consistency в качестве готового архитектурного фундамента.
|
||
|
||
---
|
||
|
||
# Влияние на последующие Build
|
||
|
||
Реализация настоящего Build существенно упрощает дальнейшее развитие подсистемы Trades Feed.
|
||
|
||
Следующие этапы смогут опираться на уже готовый механизм проверки согласованности и сосредоточиться исключительно на задачах получения и восстановления потока данных.
|
||
|
||
В частности, слой Trade Stream Consistency станет обязательной частью конвейера обработки:
|
||
|
||
```text
|
||
WebSocket
|
||
│
|
||
▼
|
||
Trade Adapter
|
||
│
|
||
▼
|
||
WebSocket Trades Feed
|
||
│
|
||
▼
|
||
Trade Stream Consistency
|
||
│
|
||
▼
|
||
Gap Detection
|
||
│
|
||
▼
|
||
REST Backfill
|
||
│
|
||
▼
|
||
Canonical Trade Stream
|
||
```
|
||
|
||
Такое разделение обязанностей позволяет развивать каждый уровень независимо, не изменяя уже реализованные компоненты.
|
||
|
||
---
|
||
|
||
# Заключение
|
||
|
||
Build 060.18 успешно достиг всех поставленных целей.
|
||
|
||
В рамках реализации:
|
||
|
||
- сформирован самостоятельный слой **Trade Stream Consistency**;
|
||
- реализован контракт `TradeStreamConsistencyProtocol`;
|
||
- реализован координирующий `TradeStreamConsistencyController`;
|
||
- реализована модель состояния `TradeStreamState`;
|
||
- введены специализированные исключения предметной области;
|
||
- выполнено полное покрытие новой подсистемы unit-тестами;
|
||
- подтверждено соответствие реализации утверждённой архитектурной спецификации;
|
||
- проведён архитектурный аудит существующей инфраструктуры Trades Feed, результаты которого зафиксированы в настоящем документе.
|
||
|
||
Полученные результаты формируют прочную основу для следующего этапа развития подсистемы — построения полноценного **WebSocket Trades Feed**, который станет первым источником непрерывного канонического потока сделок и позволит интегрировать реализованный механизм проверки согласованности в общий конвейер обработки рыночных данных.
|
||
|
||
---
|
||
|
||
# Следующий Build
|
||
|
||
Build 060.19 — Trade Gap Detection & REST Backfill |