Market Data Processing in C++

Exchange matching engine
        │
        │  UDP multicast (ITCH, PITCH, OPRA, XDP...)
        ▼
  [Market data feed]
   ├── Primary feed (primary port)
   └── Backup feed (redundant port)
        │
        │  kernel-bypass NIC (DPDK / Solarflare ef_vi / Mellanox VMA)
        ▼
  [Feed handler process]   ← pinned to isolated CPU core
   ├── Sequence gap detection
   ├── Multicast recovery (via TCP retransmit channel)
   └── Order book reconstruction
        │
        │  shared memory (huge pages, lock-free ring)
        ▼
  [Strategy process]        ← alpha generation

#pragma pack(push, 1)
// ITCH Add Order (message type 'A')
struct ItchAddOrder {
    char     msg_type;      // 'A'
    uint16_t stock_locate;
    uint16_t tracking_num;
    uint8_t  timestamp[6];  // 48-bit nanoseconds since midnight
    uint64_t order_ref;
    char     side;          // 'B' or 'S'
    uint32_t shares;
    char     stock[8];      // space-padded
    uint32_t price;         // in 1/10000 dollars (4 implied decimals)
};

struct ItchDeleteOrder {
    char     msg_type;      // 'D'
    uint16_t stock_locate;
    uint16_t tracking_num;
    uint8_t  timestamp[6];
    uint64_t order_ref;
};

struct ItchExecuteOrder {
    char     msg_type;      // 'E'
    uint16_t stock_locate;
    uint16_t tracking_num;
    uint8_t  timestamp[6];
    uint64_t order_ref;
    uint32_t executed_shares;
    uint64_t match_num;
};
#pragma pack(pop)

class ItchFeedHandler {
    OrderBook& m_book;

public:
    // Parse one message from a UDP packet payload
    void OnMessage(const uint8_t* data, int len) {
        char type = static_cast<char>(data[0]);
        switch (type) {
            case 'A': return OnAddOrder(reinterpret_cast<const ItchAddOrder*>(data));
            case 'D': return OnDeleteOrder(reinterpret_cast<const ItchDeleteOrder*>(data));
            case 'E': return OnExecuteOrder(reinterpret_cast<const ItchExecuteOrder*>(data));
            case 'U': return OnReplaceOrder(data);
            case 'P': return OnTrade(data);
            case 'H': return OnHalt(data);
        }
    }

private:
    void OnAddOrder(const ItchAddOrder* msg) {
        // Byte-swap (ITCH is big-endian)
        uint64_t ref   = __builtin_bswap64(msg->order_ref);
        uint32_t qty   = __builtin_bswap32(msg->shares);
        uint32_t price = __builtin_bswap32(msg->price);
        char side      = msg->side;

        m_book.Add(ref, side == 'B', qty, price);
    }
};

class SequencedFeedHandler {
    uint64_t m_expected_seq = 1;

    struct GapRequest {
        uint64_t start, end;
        std::chrono::steady_clock::time_point requested_at;
    };
    std::vector<GapRequest> m_pending_gaps;

public:
    void OnPacket(uint64_t seq, const uint8_t* data, int len) {
        if (seq < m_expected_seq) return; // duplicate, ignore

        if (seq > m_expected_seq) {
            // Gap detected!
            RequestRetransmit(m_expected_seq, seq - 1);
            m_pending_gaps.push_back({ m_expected_seq, seq - 1,
                std::chrono::steady_clock::now() });
            // Queue this packet for later processing
            m_reorder_buffer[seq] = std::vector<uint8_t>(data, data + len);
            return;
        }

        // In-order packet
        ProcessMessage(data, len);
        m_expected_seq = seq + 1;

        // Drain reorder buffer
        while (true) {
            auto it = m_reorder_buffer.find(m_expected_seq);
            if (it == m_reorder_buffer.end()) break;
            ProcessMessage(it->second.data(), it->second.size());
            m_reorder_buffer.erase(it);
            ++m_expected_seq;
        }
    }

    void RequestRetransmit(uint64_t start, uint64_t end) {
        // Connect to exchange retransmit server (TCP) and request range
        // Most exchanges provide this as a separate TCP service
    }
};

class DualFeedMerger {
    uint64_t m_last_seq_a = 0;
    uint64_t m_last_seq_b = 0;

    // Set of sequence numbers already processed
    // Using a sliding window bitmask instead of unordered_set
    static constexpr int WINDOW = 64;
    uint64_t m_seen_bitmask = 0;
    uint64_t m_window_base = 0;

public:
    // Returns true if this message is new (should be processed)
    bool ShouldProcess(uint64_t seq) {
        if (seq < m_window_base) return false; // too old
        if (seq >= m_window_base + WINDOW) {
            // Advance window
            uint64_t advance = seq - (m_window_base + WINDOW) + 1;
            m_seen_bitmask >>= advance;
            m_window_base += advance;
        }
        uint64_t bit = seq - m_window_base;
        if (m_seen_bitmask & (1ULL << bit)) return false; // already seen from other feed
        m_seen_bitmask |= (1ULL << bit);
        return true;
    }
};

// Hardware timestamp from NIC (SOLARFLARE, MELLANOX)
// ef_vi and DPDK support hardware timestamping

struct MarketDataPacket {
    uint64_t exchange_timestamp_ns;  // from ITCH header (ns since midnight)
    uint64_t recv_timestamp_ns;      // NIC hardware timestamp
    uint64_t process_timestamp_ns;   // TSC when handler ran
};

void MeasureLatency(const MarketDataPacket& pkt) {
    uint64_t wire_latency = pkt.recv_timestamp_ns - pkt.exchange_timestamp_ns;
    uint64_t processing   = pkt.process_timestamp_ns - pkt.recv_timestamp_ns;

    // Track percentiles (no allocations — reservoir sampling or HDR histogram)
    m_wire_latency_hist.record(wire_latency);
    m_processing_hist.record(processing);
}

// SPSC (single-producer, single-consumer) ring buffer
// Feed handler writes, strategy reads
template<typename T, int N>
class MarketDataRing {
    static_assert((N & (N-1)) == 0, "N must be power of 2");
    alignas(64) T m_data[N];
    alignas(64) std::atomic<uint64_t> m_write_idx{0};
    alignas(64) std::atomic<uint64_t> m_read_idx{0};

public:
    // Called by feed handler (producer)
    bool TryWrite(const T& item) {
        uint64_t w = m_write_idx.load(std::memory_order_relaxed);
        uint64_t r = m_read_idx.load(std::memory_order_acquire);
        if (w - r >= N) return false; // full
        m_data[w & (N-1)] = item;
        m_write_idx.store(w + 1, std::memory_order_release);
        return true;
    }

    // Called by strategy (consumer)
    bool TryRead(T& out) {
        uint64_t r = m_read_idx.load(std::memory_order_relaxed);
        uint64_t w = m_write_idx.load(std::memory_order_acquire);
        if (r == w) return false; // empty
        out = m_data[r & (N-1)];
        m_read_idx.store(r + 1, std::memory_order_release);
        return true;
    }
};