Разработка хранилища исторических данных ордеров

Разработка хранилища исторических данных ордеров

История ордеров — это не просто архив. Это первичный источник данных для анализа эффективности исполнения, бэктестинга, расчёта комиссий, налоговой отчётности и аудита торговых стратегий. Правильно спроектированное хранилище должно обеспечивать высокую скорость записи в real-time и аналитические запросы по многомерным срезам на годах данных.

Структура данных ордера

Ордер в торговой системе — это не просто запись "купил 1 BTC по 50000". Полная модель включает несколько типов событий:

Order Lifecycle Events:

• ORDER_CREATED — первичное создание с параметрами
• ORDER_ACCEPTED — биржа подтвердила получение
• ORDER_PARTIALLY_FILLED — частичное исполнение
• ORDER_FILLED — полное исполнение
• ORDER_CANCELLED — отмена (с причиной)
• ORDER_REJECTED — отказ биржи (с кодом ошибки)
• ORDER_EXPIRED — истёк таймаут

Хранение этих событий отдельно (event sourcing) даёт полную воспроизводимость: вы всегда можете восстановить состояние любого ордера на любой момент времени.

Схема базы данных

Для временных рядов ордеров оптимально использовать TimescaleDB (расширение PostgreSQL) или ClickHouse.

TimescaleDB — хороший выбор если уже используете PostgreSQL. Автоматически партиционирует таблицы по времени (hypertables), поддерживает непрерывные агрегации (continuous aggregates) и политики компрессии.

CREATE TABLE order_events (
    event_id        UUID DEFAULT gen_random_uuid(),
    event_time      TIMESTAMPTZ NOT NULL,
    order_id        UUID NOT NULL,
    exchange        VARCHAR(32) NOT NULL,
    symbol          VARCHAR(32) NOT NULL,
    event_type      VARCHAR(32) NOT NULL,
    side            VARCHAR(8),
    order_type      VARCHAR(16),
    price           NUMERIC(24, 8),
    quantity        NUMERIC(24, 8),
    filled_qty      NUMERIC(24, 8),
    avg_fill_price  NUMERIC(24, 8),
    commission      NUMERIC(24, 8),
    commission_asset VARCHAR(16),
    client_order_id VARCHAR(64),
    strategy_id     VARCHAR(64),
    metadata        JSONB
);

SELECT create_hypertable('order_events', 'event_time',
    chunk_time_interval => INTERVAL '1 day');

Индексы для типичных паттернов запросов:

CREATE INDEX ON order_events (order_id, event_time DESC);
CREATE INDEX ON order_events (exchange, symbol, event_time DESC);
CREATE INDEX ON order_events (strategy_id, event_time DESC);
CREATE INDEX ON order_events USING GIN (metadata);

Паттерны запросов и оптимизация

Восстановление состояния ордера — частая операция. Вместо повторного вычисления из событий каждый раз, поддерживаем материализованную таблицу orders с текущим состоянием:

CREATE TABLE orders (
    order_id        UUID PRIMARY KEY,
    exchange        VARCHAR(32) NOT NULL,
    symbol          VARCHAR(32) NOT NULL,
    status          VARCHAR(32) NOT NULL,
    side            VARCHAR(8) NOT NULL,
    order_type      VARCHAR(16) NOT NULL,
    price           NUMERIC(24, 8),
    quantity        NUMERIC(24, 8),
    filled_qty      NUMERIC(24, 8) DEFAULT 0,
    avg_fill_price  NUMERIC(24, 8),
    total_commission NUMERIC(24, 8) DEFAULT 0,
    strategy_id     VARCHAR(64),
    created_at      TIMESTAMPTZ NOT NULL,
    updated_at      TIMESTAMPTZ NOT NULL
);

Обновление этой таблицы происходит при каждом новом событии через триггер или application-side логику.

Аналитические запросы — типично это агрегации по стратегии, инструменту, периоду:

-- P&L по стратегиям за период
SELECT
    strategy_id,
    symbol,
    SUM(CASE WHEN side = 'BUY' THEN -filled_qty * avg_fill_price ELSE filled_qty * avg_fill_price END) as realized_pnl,
    SUM(total_commission) as total_fees,
    COUNT(*) as order_count
FROM orders
WHERE created_at BETWEEN '2024-01-01' AND '2024-12-31'
  AND status = 'FILLED'
GROUP BY strategy_id, symbol
ORDER BY realized_pnl DESC;

TimescaleDB continuous aggregates позволяют предвычислить эти агрегации и обновлять их инкрементально:

CREATE MATERIALIZED VIEW orders_daily_summary
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('1 day', created_at) AS day,
    exchange,
    symbol,
    strategy_id,
    COUNT(*) AS order_count,
    SUM(filled_qty * avg_fill_price) AS volume,
    SUM(total_commission) AS fees
FROM orders
WHERE status = 'FILLED'
GROUP BY 1, 2, 3, 4;

Хранение fills отдельно

Для детального анализа execution quality критично хранить individual fills (частичные исполнения) отдельно от ордеров:

CREATE TABLE order_fills (
    fill_id         UUID DEFAULT gen_random_uuid(),
    fill_time       TIMESTAMPTZ NOT NULL,
    order_id        UUID NOT NULL REFERENCES orders(order_id),
    exchange        VARCHAR(32) NOT NULL,
    symbol          VARCHAR(32) NOT NULL,
    price           NUMERIC(24, 8) NOT NULL,
    quantity        NUMERIC(24, 8) NOT NULL,
    commission      NUMERIC(24, 8),
    commission_asset VARCHAR(16),
    is_maker        BOOLEAN,
    trade_id        VARCHAR(64)
);

SELECT create_hypertable('order_fills', 'fill_time',
    chunk_time_interval => INTERVAL '1 day');

Это позволяет рассчитывать VWAP исполнения, сравнивать с mid-price в момент исполнения (market impact), анализировать maker/taker ratio по стратегии.

Retention политики и архивация

Горячие данные (последние 30 дней) хранятся без компрессии для максимальной скорости записи и чтения. Более старые данные компрессируются:

ALTER TABLE order_events SET (
    timescaledb.compress,
    timescaledb.compress_segmentby = 'exchange, symbol',
    timescaledb.compress_orderby = 'event_time DESC'
);

SELECT add_compression_policy('order_events',
    INTERVAL '30 days');

Компрессия TimescaleDB даёт 10–20x уменьшение размера для временных рядов с повторяющимися значениями. На практике таблица order_events объёмом 100GB сжимается до 5–10GB.

Данные старше 2 лет можно экспортировать в Parquet-файлы на S3 с помощью pg_parquet или custom ETL, сохраняя возможность исторического анализа через Athena или ClickHouse.

Ingestion pipeline

Высокочастотная запись требует batching. Вместо INSERT на каждый event используем:

COPY для bulk inserts — в 10–50x быстрее индивидуальных INSERT. Накапливаем события в памяти (100ms или 1000 событий, что наступит раньше) и записываем одним COPY.

Unlogged tables для промежуточного буфера — WAL не пишется, скорость записи значительно выше. После агрегации данные переносятся в основную logged таблицу.

Connection pooling через PgBouncer — торговые системы открывают много коротких соединений. PgBouncer в transaction mode позволяет обслуживать тысячи клиентов через небольшой пул реальных соединений PostgreSQL.

Мониторинг и алерты

Ключевые метрики для мониторинга хранилища:

Потеря ордеров — это всегда критический инцидент. Система должна иметь механизм reconciliation: периодически сравнивать локальную историю с данными биржи через REST API и заполнять пробелы.

Репликация и отказоустойчивость

Production-хранилище работает в режиме streaming replication PostgreSQL: primary для записи, replica для аналитических запросов. При падении primary — failover через Patroni или PgBouncer с автоматическим переключением. RPO (Recovery Point Objective) при правильной настройке synchronous_commit — нулевой.