Market data feeds (ITCH / MoldUDP64)

Deep dive

How it works

In: MoldUDP64 packets from lines A and B · Out: in-order ITCH messages, book updates · State: next expected sequence number per session, gap buffer · Cost: O(1) per message; a gap costs one request RTT or a snapshot load

MoldUDP64 packet (one UDP datagram)
+-----------+-------------+-------+------+---------+------+---------+
| session10 | seq u64 BE  | count | len  | msg 1   | len  | msg 2 ..|
+-----------+-------------+-------+------+---------+------+---------+
  message i has sequence number seq + i

ITCH 5.0 header (every message): type u8 | stock locate u16 | tracking u16 | timestamp u48 (ns since midnight)
'A' add order (36 B): header | order ref u64 | side 'B'/'S' | shares u32 | stock char[8] | price u32 (×10⁻⁴)
'E' executed (31 B):  header | order ref u64 | executed shares u32 | match number u64
'X' cancel (23 B):    header | order ref u64 | cancelled shares u32
'D' delete (19 B):    header | order ref u64
'U' replace (35 B):   header | original ref u64 | new ref u64 | shares u32 | price u32

Receiver loop per session:

1. Read a packet from line A or B. Packets entirely below next are duplicates (other line, retransmission): drop.
2. Packet starts at next (or overlaps it): deliver messages with sequence ≥ next, advance next.
3. Packet starts above next: a gap. Buffer the packet, send a MoldUDP64 request (session, next, count) to the re-request server, keep buffering both lines.
4. When the retransmission fills the gap, deliver it, then drain the buffer in order.
5. If the gap is older than the retransmission server holds, or recovery is slow: drop the book, load a snapshot (Nasdaq GLIMPSE over SoupBinTCP) which ends with the sequence number it reflects, then apply buffered live messages from there.

Code

use std::collections::BTreeMap;

pub struct Mold<'a> { pub session: &'a [u8], pub seq: u64, pub count: u16, pub body: &'a [u8] }

pub fn parse_mold(p: &[u8]) -> Option<Mold<'_>> {
    if p.len() < 20 { return None; }
    Some(Mold { session: &p[0..10], seq: u64::from_be_bytes(p[10..18].try_into().ok()?),
                count: u16::from_be_bytes(p[18..20].try_into().ok()?), body: &p[20..] })
}

/// Length-prefixed message blocks; stops early on a truncated packet.
pub fn messages(body: &[u8], count: u16) -> impl Iterator<Item = &[u8]> {
    let mut off = 0usize;
    (0..count).map_while(move |_| {
        let len = u16::from_be_bytes(body.get(off..off + 2)?.try_into().ok()?) as usize;
        let m = body.get(off + 2..off + 2 + len)?;
        off += 2 + len;
        Some(m)
    })
}

/// 20-byte MoldUDP64 retransmission request.
pub fn request(session: &[u8], seq: u64, count: u16) -> [u8; 20] {
    let mut r = [0u8; 20];
    r[0..10].copy_from_slice(&session[..10]);
    r[10..18].copy_from_slice(&seq.to_be_bytes());
    r[18..20].copy_from_slice(&count.to_be_bytes());
    r
}

#[derive(Debug)]
pub enum Itch { Add { r: u64, buy: bool, shares: u32, px: u32, locate: u16, ts: u64 },
                Exec { r: u64, shares: u32, match_no: u64 }, Cancel { r: u64, shares: u32 },
                Delete { r: u64 }, Replace { old: u64, new: u64, shares: u32, px: u32 }, Other(u8) }

pub fn decode(m: &[u8]) -> Option<Itch> {
    let need = match *m.first()? { b'A' => 36, b'E' => 31, b'X' => 23, b'D' => 19, b'U' => 35, _ => 1 };
    if m.len() < need { return None; }
    let u16_ = |a: usize| u16::from_be_bytes([m[a], m[a + 1]]);
    let u32_ = |a: usize| u32::from_be_bytes(m[a..a + 4].try_into().unwrap());
    let u64_ = |a: usize| u64::from_be_bytes(m[a..a + 8].try_into().unwrap());
    let ts = || { let mut t = [0u8; 8]; t[2..].copy_from_slice(&m[5..11]); u64::from_be_bytes(t) };
    Some(match m[0] {
        b'A' => Itch::Add { r: u64_(11), buy: m[19] == b'B', shares: u32_(20), px: u32_(32), locate: u16_(1), ts: ts() },
        b'E' => Itch::Exec { r: u64_(11), shares: u32_(19), match_no: u64_(23) },
        b'X' => Itch::Cancel { r: u64_(11), shares: u32_(19) },
        b'D' => Itch::Delete { r: u64_(11) },
        b'U' => Itch::Replace { old: u64_(11), new: u64_(19), shares: u32_(27), px: u32_(31) },
        t => Itch::Other(t),
    })
}

/// A/B arbitration + gap detection for one session.
pub struct Sequencer { pub next: u64, pending: BTreeMap<u64, Vec<u8>>, asked_until: u64 }

impl Sequencer {
    pub fn new(first: u64) -> Self { Sequencer { next: first, pending: BTreeMap::new(), asked_until: 0 } }

    /// Feed a packet from any line. Calls `f(seq, msg)` in order; returns a (seq, count) to re-request.
    pub fn on_packet<F: FnMut(u64, &[u8])>(&mut self, pkt: &[u8], f: &mut F) -> Option<(u64, u16)> {
        let h = parse_mold(pkt)?;
        if h.count == 0xFFFF { return None; }                    // end of session
        if h.seq + h.count as u64 <= self.next { return None; }         // duplicate / heartbeat
        if h.seq > self.next {                                  // gap: slow path, copies
            self.pending.entry(h.seq).or_insert_with(|| pkt.to_vec());
            return self.ask(h.seq);
        }
        self.deliver(&h, f);
        while let Some(e) = self.pending.first_entry() {
            if *e.key() > self.next { break; }
            let p = e.remove();
            if let Some(h) = parse_mold(&p) { self.deliver(&h, f); }
        }
        let first = self.pending.keys().next().copied();
        first.and_then(|s| self.ask(s))
    }

    fn deliver<F: FnMut(u64, &[u8])>(&mut self, h: &Mold, f: &mut F) {
        for (i, m) in messages(h.body, h.count).enumerate() {
            let s = h.seq + i as u64;
            if s == self.next { f(s, m); self.next += 1; }      // skip already-seen prefix
        }
    }

    fn ask(&mut self, upto: u64) -> Option<(u64, u16)> {
        if self.asked_until >= upto { return None; }            // already outstanding; a timer re-asks
        self.asked_until = upto;
        Some((self.next, (upto - self.next).min(u16::MAX as u64 - 1) as u16))
    }
}

Complexity and constants

• Per message: a type switch and a few big-endian loads (bswap / movbe), order of 10 ns to decode; the book update behind it costs more.
• Message sizes 19-50 bytes, so a 1,500-byte MTU packet carries dozens of messages; packet rate, not byte rate, is the receive bottleneck.
• Recovery: a retransmission request costs one RTT to the re-request server (sub-ms in colocation); a snapshot reload costs seconds and blanks that book meanwhile.

Efficiency

Cost: the bound is packet rate and cache misses, not bytes. Decoding a message is order of 10 ns (a type switch and big-endian loads); messages are 19-50 bytes, so a 1,500-byte packet carries dozens. A gap costs one retransmission RTT (sub-ms in colocation) or a snapshot reload of seconds with a blank book.

Levers

• Large SO_RCVBUF, pinned receive thread, kernel bypass (kernel bypass) — gaps stay rare — high packet rates.
• Flat array of books indexed by stock locate, integer prices, decode in place — fewer misses — every update.
• recvmmsg — many datagrams per syscall — Linux sockets.
• A/B lines on one thread, first copy of each sequence wins — no locks — redundant feeds.

Naive → standard → state of the art

• Naive: TCP or one UDP line, no sequence check · silent gaps
• Standard: sequence check, A/B arbitration, retransmission request, snapshot fallback · one RTT per gap
• State of the art: kernel-bypass receive with in-place decode · order of 10 ns per message Cheaper when queue position and sub-millisecond latency are not needed: a vendor-normalized or market-by-price feed.

Implementation notes

• Index by locate: ITCH stock locate codes are small per-day integers; use them as array indexes for per-symbol books instead of hashing the 8-byte ticker.
• Order references: unique for the day per feed; Replace gives a new reference and loses time priority; an E/X reduces, D removes.
• Prices: keep the raw u32 (×10⁻⁴) or convert to book ticks with integer math; never to f64.
• Socket setup: IP_ADD_MEMBERSHIP per group; large SO_RCVBUF (raise net.core.rmem_max first or the kernel silently caps it); watch /proc/net/snmp UDP RcvbufErrors to catch drops before they become gaps. Production handlers use kernel bypass and hardware timestamps.
• A/B arbitration is just "first copy of each sequence wins"; run both lines into the same sequencer on one thread so no locking is needed.
• Buffer bound: cap the pending map; past the cap, give up on retransmission and go to snapshot recovery.
• Session rollover: a new session id resets sequence numbers; key all state by (session, seq).
• Zero copy: decode directly from the receive buffer; the Vec copy above is only taken on the gap path.

Pitfalls

• Treating a heartbeat (count 0) as data, or ignoring it: its sequence number is how you detect a gap when the market is quiet.
• Applying messages out of order "because they're for different symbols": an execute can precede the add it refers to; the order reference then doesn't exist.
• Requesting the whole gap in one request when it exceeds what the server serves per request; chunk and pace requests.
• Re-requesting on every out-of-order packet (often just A/B skew of a few µs): wait for a short reorder window before asking.
• Applying snapshot then live messages without filtering by the snapshot's sequence number: double-applied adds.
• Endianness: ITCH and MoldUDP64 are big-endian; x86 and ARM are little-endian.

Variants and alternatives

In production

Nasdaq TotalView-ITCH over MoldUDP64, Nasdaq Nordic and other Nasdaq-technology venues, Cboe PITCH, CME MDP 3.0; feed handlers at every trading firm and market data vendor.

Implementations

Sources

• Nasdaq, Nasdaq TotalView-ITCH 5.0 specification. PDF
• Nasdaq, MoldUDP64 Protocol Specification, version 1.00. PDF
• Nasdaq, SoupBinTCP specification. PDF
• CME Group, MDP 3.0 client documentation.