28Reading Market Data

Goal. Decode a framed binary order-by-order feed, maintain the book, produce level 1 and a tape with aggressor side, and measure what receive-time jitter does to message order. End state: the three data figures of this chapter, from code/firm/feed/data/sample.itch (2 020 messages, 65 842 bytes).

1. Framing and header. Length prefix, type, then the fields every message shares. A length that does not fit the type is an error, not a hint.

   def decode(buf: bytes):
       """Yield messages from a framed buffer. Raises on a truncated frame or a length that does not
       match the message type: a decoder that guesses is worse than one that stops."""
       i = 0
       while i < len(buf):
           if i + 2 > len(buf):
               raise ValueError("truncated length prefix")
           (n,) = struct.unpack_from(">H", buf, i)
           body = buf[i + 2:i + 2 + n]
           if len(body) != n:
               raise ValueError("truncated message")
           kind = body[:1]
           if LENGTHS.get(kind) != n:
               raise ValueError(f"type {kind!r} with length {n}")
           locate, tracking = struct.unpack_from(">HH", body, 1)
           ts = int.from_bytes(body[5:11], "big")
           (ref,) = struct.unpack_from(">Q", body, 11)

2. The same in C++, over a span of bytes, without copies.

   inline void decode(std::span<const std::uint8_t> buf, const std::function<void(const Msg&)>& on) {
       std::size_t i = 0;
       while (i < buf.size()) {
           if (i + 2 > buf.size()) throw std::runtime_error("truncated length prefix");
           const std::size_t n = static_cast<std::size_t>(be(buf, i, 2));
           if (i + 2 + n > buf.size()) throw std::runtime_error("truncated message");
           const auto body = buf.subspan(i + 2, n);
           Msg m;
           m.kind = static_cast<char>(body[0]);
           if (expected_length(m.kind) != n) throw std::runtime_error("length does not match type");
           m.locate = static_cast<std::uint16_t>(be(body, 1, 2));
           m.ts = be(body, 5, 6);
           m.ref = be(body, 11, 8);

3. And in Rust, as an iterator that yields a result per message and stops after the first error.

       fn next(&mut self) -> Option<Self::Item> {
           if self.failed || self.buf.is_empty() {
               return None;
           }
           let fail = |s: &mut Self, e| {
               s.failed = true;
               Some(Err(e))
           };
           if self.buf.len() < 2 {
               return fail(self, DecodeError::TruncatedPrefix);
           }
           let n = be(&self.buf[..2]) as usize;
           if self.buf.len() < 2 + n {
               return fail(self, DecodeError::TruncatedMessage);
           }
           let body = &self.buf[2..2 + n];
           let kind = body.first().copied().unwrap_or(0);
           if expected_length(kind) != Some(n) {
               return fail(self, DecodeError::BadLength { kind, len: n });
           }
           let mut m = Msg {
               kind,
               locate: be(&body[1..3]) as u16,
               ts: be(&body[5..11]),
               reference: be(&body[11..19]),
               ..Msg::default()
           };

4. The book. An execution carries no price: it is looked up.

       def apply(self, m: Msg) -> None:
           if m.ts < self.last_ts:
               self.errors.append(f"timestamp went backwards at ref {m.ref}")
           self.last_ts = max(self.last_ts, m.ts)
           if m.kind == "A":
               if m.ref in self.orders:
                   self.errors.append(f"duplicate order {m.ref}")
                   return
               self.orders[m.ref] = Order(m.side, m.shares, m.price, m.locate)
               lv = self._level(m.locate, m.side)
               lv[m.price] = lv.get(m.price, 0) + m.shares
           elif m.kind == "E":
               before = self.orders.get(m.ref)
               price, side = (before.price, before.side) if before else (0, "")
               if self._reduce(m.ref, m.shares) is not None:
                   self.trades.append(Trade(m.ts, m.locate, price, m.shares, m.match, "S" if side == "B" else "B"))
           elif m.kind == "X":
               self._reduce(m.ref, m.shares)
           elif m.kind == "D":
               o = self.orders.get(m.ref)
               self._reduce(m.ref, o.shares if o else 0)
           elif m.kind == "P":
               self.trades.append(Trade(m.ts, m.locate, m.price, m.shares, m.match, ""))

What to change next. Drop one add-order message from the sample and watch the errors the book records, and how long it takes before an execution refers to the missing order. Then add a replace message (cancel and add with a new reference, losing priority) to all three decoders.