Processing a major exchange feed as a consistency problem.

Why This Matters

• It is the first result where the consistency layer changes operational behavior, not just labeling.
• It shows why persistent moderate drift deserves its own trigger logic.
• It suggests rebuild policy can be driven by structural exposure, not only by crashes.
• It gives a concrete path from theory to feed-quality monitoring and recovery policy.

• c_pkt: packet-sequence defect on the multicast channel.
• c_msg: per-product message-sequence defect.
• c_id: modify/delete/execute references a missing visible order key.
• c_qty: execution exceeds remaining displayed quantity.
• c_px: best bid rises above best ask after processing.
• c_state: message is inconsistent with product or instrument trading state.

def order_key(msg):
    return (msg.security_id, msg.side, msg.priority_ts)

def process_datagram(raw, state, structure):
    packet = parse_packet_header(raw)      # little-endian EOBI packet header
    defect = {"pkt": 0, "msg": 0, "id": 0, "qty": 0, "px": 0, "state": 0}

    if packet.seq_num != state.next_packet_seq:
        defect["pkt"] += abs(packet.seq_num - state.next_packet_seq)

    offset = packet.header_len
    while offset < packet.packet_len:
        header = parse_message_header(raw, offset)
        body = raw[offset:offset + header.body_len]
        msg = decode_message(header.template_id, body)
        apply_eobi_message(state, msg, defect)
        state.next_msg_seq[msg.product_id] = header.msg_seq_num + 1
        offset += header.body_len

    defect["px"] += max(0.0, best_bid(state) - best_ask(state))
    return structure.residual_from_defect(defect)

def apply_eobi_message(state, msg, defect):
    key = order_key(msg) if hasattr(msg, "priority_ts") else None

    if msg.template_id == 13100:  # Order Add
        state.orders[key] = msg
    elif msg.template_id in (13101, 13106):  # Modify
        prev_key = (msg.security_id, msg.side, msg.prev_priority_ts)
        if prev_key not in state.orders:
            defect["id"] += 1
            return
        state.orders.pop(prev_key)
        state.orders[key] = msg
    elif msg.template_id == 13102:  # Delete
        if state.orders.pop(key, None) is None:
            defect["id"] += 1
    elif msg.template_id == 13105:  # Partial execution
        order = state.orders.get(key)
        if order is None:
            defect["id"] += 1
            return
        if msg.last_qty > order.display_qty:
            defect["qty"] += msg.last_qty - order.display_qty
        order.display_qty = max(0, order.display_qty - msg.last_qty)
    elif msg.template_id == 13104:  # Full execution
        if state.orders.pop(key, None) is None:
            defect["id"] += 1