Рыночные данные: бинарный фид и стакан из потока

Главное недоразумение новичка в биржевых данных звучит так: «пришлите мне текущий стакан». Биржа его не пришлёт. Она пришлёт поток событий: заявка встала, заявка исполнилась, заявку сняли. Стакан существует только у тебя в памяти, и он ровно настолько правильный, насколько правильно ты применил каждое событие по порядку. Этот урок про то, как разобрать такой поток и как собрать из него стакан, за который не стыдно торговать.

Почему поток событий, а не снимки

Причина арифметическая. На ликвидном инструменте стакан меняется десятки тысяч раз в секунду, и в каждом изменении участвует один уровень из сотни. Слать целиком то, что изменилось на процент, значит слать в сто раз больше байт, чем нужно.

Поэтому фид устроен как журнал изменений, а полный снимок присылают редко: при подключении и после сбоя. Дальше ты сам поддерживаешь состояние. Плата за экономию прямая и жёсткая: пропустил одно событие, и твой стакан больше не равен биржевому, причём молча. Отсюда весь дальнейший разговор про порядковые номера.

Формат сообщения

Спроектируем свой фид по образцу семейства ITCH. Правила формата:

Фиксированная ширина полей. Ни разделителей, ни длин, ни текста. Позиция каждого поля известна заранее, разбор это чтение по смещению.

Порядок байт от старшего к младшему. Это не дань традиции: в таком порядке числа сравниваются побайтово прямо в буфере, не собираясь в число.

Целые цены. Цена это номер тика, количество это лот. Дробей на бирже нет, и это не упрощение: расхождение округлений между участниками это расхождение в том, кто кому должен.

Заголовок одинаков у всех сообщений и занимает 17 байт: тип (1), порядковый номер (8), метка времени биржи (8). Дальше тело:

Тип	Байт	Тело	Что значит
A	17	номер заявки, сторона, цена, количество	новая лимитная заявка встала в стакан
E	12	номер заявки, количество	стоящая заявка частично или полностью исполнилась
X	12	номер заявки, количество	заявку уменьшили, но не сняли
D	8	номер заявки	заявку сняли целиком
R	24	старый номер, новый номер, цена, количество	заявку заменили
T	9	цена, количество, сторона агрессора	сделка мимо видимого стакана

Обрати внимание на разницу между E и X. Оба уменьшают объём на уровне, но E это сделка (цена последней сделки меняется, кто-то получил исполнение), а X это отмена части заявки (сделки не было). Смешивать их нельзя: половина сигналов стратегии строится на потоке сделок, а не на потоке изменений стакана.

Тип T нужен для сделок, прошедших мимо видимого стакана: скрытая ликвидность, аукцион открытия, кросс-сделка. Стакан от них не меняется, а цена последней сделки меняется.

Тела в коде:

/// Тело сообщения. Цены целые: это номер тика, а не рубли с копейками.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Body {
    /// Новая лимитная заявка встала в стакан.
    Add {
        order_id: u64,
        side: Side,
        price: u32,
        quantity: u32,
    },
    /// Стоящая заявка частично или полностью исполнилась.
    Execute { order_id: u64, quantity: u32 },
    /// Стоящую заявку уменьшили, но не сняли.
    Reduce { order_id: u64, quantity: u32 },
    /// Заявку сняли целиком.
    Delete { order_id: u64 },
    /// Заявку заменили: старый номер умер, родился новый.
    Replace {
        old_id: u64,
        new_id: u64,
        price: u32,
        quantity: u32,
    },
    /// Сделка мимо видимого стакана (скрытая ликвидность, аукцион).
    Trade {
        price: u32,
        quantity: u32,
        aggressor: Side,
    },
}

/// Сообщение фида целиком.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Message {
    /// Порядковый номер. Растёт на единицу, разрыв значит потерю.
    pub sequence: u64,
    /// Метка времени биржи в наносекундах. Наша собственная метка ставится
    /// отдельно, при получении: разница между ними это задержка канала.
    pub timestamp: u64,
    pub body: Body,
}

Две метки времени это важно и часто упускается. Метка биржи говорит, когда событие случилось у неё. Наша метка говорит, когда мы его увидели. Разница между ними это задержка канала, и следить за ней надо постоянно: её рост означает, что нас либо перегрузили, либо мы отстали от мультикаст-группы.

Разбор

Заводим таблицу длин и читалки целых. Всё остальное это чтение по смещениям.

/// Заголовок: тип, номер, метка времени.
const HEADER: usize = 1 + 8 + 8;

fn body_len(kind: u8) -> Option<usize> {
    match kind {
        b'A' => Some(8 + 1 + 4 + 4),
        b'E' | b'X' => Some(8 + 4),
        b'D' => Some(8),
        b'R' => Some(8 + 8 + 4 + 4),
        b'T' => Some(4 + 4 + 1),
        _ => None,
    }
}

fn read_u32(bytes: &[u8], at: usize) -> u32 {
    u32::from_be_bytes([bytes[at], bytes[at + 1], bytes[at + 2], bytes[at + 3]])
}

Сам разбор возвращает сообщение и его длину, чтобы вызывающий знал, где начинается следующее:

/// Что пошло не так при разборе.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FeedError {
    /// Байтов меньше, чем нужно телу. Не ошибка при чтении из сокета:
    /// значит, пакет ещё не дочитан.
    Truncated { need: usize, got: usize },
    /// Неизвестный тип. Ошибка серьёзная: длину тела мы не знаем, а значит
    /// не знаем, где начинается следующее сообщение.
    UnknownType(u8),
    /// Байт стороны не `B` и не `S`.
    BadSide(u8),
}

/// Разобрать одно сообщение из начала среза.
///
/// Ничего не копирует и ничего не выделяет: все поля фиксированной ширины,
/// читаются на месте.
pub fn decode(bytes: &[u8]) -> Result<(Message, usize), FeedError> {
    if bytes.len() < HEADER {
        return Err(FeedError::Truncated {
            need: HEADER,
            got: bytes.len(),
        });
    }
    let kind = bytes[0];
    let payload = body_len(kind).ok_or(FeedError::UnknownType(kind))?;
    let total = HEADER + payload;
    if bytes.len() < total {
        return Err(FeedError::Truncated {
            need: total,
            got: bytes.len(),
        });
    }

    let sequence = read_u64(bytes, 1);
    let timestamp = read_u64(bytes, 9);
    let at = HEADER;

    let body = match kind {
        b'A' => Body::Add {
            order_id: read_u64(bytes, at),
            side: Side::from_byte(bytes[at + 8]).ok_or(FeedError::BadSide(bytes[at + 8]))?,
            price: read_u32(bytes, at + 9),
            quantity: read_u32(bytes, at + 13),
        },
        b'E' => Body::Execute {
            order_id: read_u64(bytes, at),
            quantity: read_u32(bytes, at + 8),
        },
        b'X' => Body::Reduce {
            order_id: read_u64(bytes, at),
            quantity: read_u32(bytes, at + 8),
        },
        b'D' => Body::Delete {
            order_id: read_u64(bytes, at),
        },
        b'R' => Body::Replace {
            old_id: read_u64(bytes, at),
            new_id: read_u64(bytes, at + 8),
            price: read_u32(bytes, at + 16),
            quantity: read_u32(bytes, at + 20),
        },
        b'T' => Body::Trade {
            price: read_u32(bytes, at),
            quantity: read_u32(bytes, at + 4),
            aggressor: Side::from_byte(bytes[at + 8]).ok_or(FeedError::BadSide(bytes[at + 8]))?,
        },
        other => return Err(FeedError::UnknownType(other)),
    };

    Ok((
        Message {
            sequence,
            timestamp,
            body,
        },
        total,
    ))
}

Различие между тремя ошибками принципиальное, и оно про то, что делать дальше.

Truncated это вообще не ошибка при чтении из сокета: байты приходят кусками, и граница куска не совпадает с границей сообщения. Правильная реакция это подождать остаток.

BadSide это испорченное сообщение, но длина его известна, значит можно пропустить его и читать дальше.

UnknownType это конец разбора. Мы не знаем длину тела, значит не знаем, где начинается следующее сообщение, и весь дальнейший буфер для нас мусор. Реакция: переподключиться и запросить снимок.

Разборщик потока держит позицию и останавливается на первом неполном сообщении, а не пытается его додумать:

#[derive(Debug, Default)]
pub struct Decoder {
    offset: usize,
}

impl Decoder {
    /// Следующее сообщение из буфера. `Ok(None)` значит «данные кончились,
    /// приноси ещё», а не «поток закончился».
    pub fn next(&mut self, buffer: &[u8]) -> Result<Option<Message>, FeedError> {
        if self.offset >= buffer.len() {
            return Ok(None);
        }
        match decode(&buffer[self.offset..]) {
            Ok((message, len)) => {
                self.offset += len;
                Ok(Some(message))
            }
            Err(FeedError::Truncated { .. }) => Ok(None),
            Err(other) => Err(other),
        }
    }
}

Хвост после offset надо перенести в начало следующего буфера. Это скучная бухгалтерия, и именно на ней ломаются самодельные парсеры: при большой нагрузке недочитанный хвост случается постоянно, а на тестовых данных из файла не случается никогда.

Номера: единственная защита от тихого расхождения

Биржевой фид почти всегда идёт по UDP, обычно мультикастом. Причина та же, что и везде в этом уроке: подтверждения доставки стоят времени, а мультикаст позволяет отдать один пакет тысяче подписчиков.

Плата за это: пакет может потеряться, продублироваться или прийти не в том порядке. Единственная защита это порядковые номера.

/// Что делать с сообщением.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Verdict {
    /// Применять к стакану.
    Apply,
    /// Уже видели, пропустить.
    Duplicate,
    /// Разрыв: стакан больше не достоверен.
    Gap(Gap),
}

impl Sequencer {
    pub fn check(&mut self, sequence: u64) -> Verdict {
        if sequence == self.expected {
            self.expected += 1;
            Verdict::Apply
        } else if sequence < self.expected {
            self.duplicates += 1;
            Verdict::Duplicate
        } else {
            let gap = Gap {
                expected: self.expected,
                got: sequence,
            };
            self.gaps += 1;
            // После разрыва продолжаем считать от пришедшего номера: поток
            // живой, недостоверен только стакан.
            self.expected = sequence + 1;
            Verdict::Gap(gap)
        }
    }
}

Дубль отбрасывается молча, это норма. А вот разрыв это событие, требующее решения, и решение всегда одно: остановить торговлю. Продолжать считать по дырявому потоку значит торговать по стакану, которого нет. Именно поэтому в прошлом уроке про риск среди причин остановки была FeedGap.

Промышленный ответ на потери называется арбитраж каналов A и B: биржа шлёт две копии потока разными путями, а получатель склеивает их по номерам. Разрыв тогда означает, что потеряли оба канала сразу, и это уже настоящая авария. Плюс к этому есть отдельный сервис повторной передачи для небольших дыр и периодический полный снимок для больших.