File size: 2,959 Bytes
f86fb3b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | #include "../../include/market_data/gap_recovery.hpp"
#include "../../include/core/time_utils.hpp"
namespace quant::market_data {
GapRecoveryTracker::GapRecoveryTracker(GapCallback on_gap, InOrderTickCallback on_tick)
: on_gap_(std::move(on_gap)), on_tick_(std::move(on_tick)) {
history_cache_.resize(HISTORY_CAPACITY);
}
void GapRecoveryTracker::reset(uint64_t initial_seq) noexcept {
expected_seq_ = initial_seq;
total_gaps_detected_ = 0;
total_packets_processed_ = 0;
pending_buffer_.clear();
}
void GapRecoveryTracker::on_packet_received(
uint64_t seq_num,
uint64_t timestamp_ns,
const protocol::MarketDataTickPayload& payload
) {
if (seq_num < expected_seq_) {
// Duplicate or stale packet, ignore
return;
}
if (seq_num == expected_seq_) {
// In-order packet
++expected_seq_;
++total_packets_processed_;
if (on_tick_) {
on_tick_(payload, seq_num, timestamp_ns);
}
drain_buffered_packets();
} else {
// Sequence gap detected: seq_num > expected_seq_
++total_gaps_detected_;
SequenceGap gap{
expected_seq_,
seq_num - 1,
core::TimeUtils::now_ns()
};
if (on_gap_) {
on_gap_(gap);
}
// Buffer the ahead-of-time packet
pending_buffer_[seq_num] = BufferedPacket{seq_num, timestamp_ns, payload};
}
}
void GapRecoveryTracker::on_gap_packet_recovered(
uint64_t seq_num,
uint64_t timestamp_ns,
const protocol::MarketDataTickPayload& payload
) {
if (seq_num == expected_seq_) {
++expected_seq_;
++total_packets_processed_;
if (on_tick_) {
on_tick_(payload, seq_num, timestamp_ns);
}
drain_buffered_packets();
} else if (seq_num > expected_seq_) {
pending_buffer_[seq_num] = BufferedPacket{seq_num, timestamp_ns, payload};
}
}
void GapRecoveryTracker::drain_buffered_packets() {
while (!pending_buffer_.empty() && pending_buffer_.begin()->first == expected_seq_) {
auto it = pending_buffer_.begin();
const auto& pkt = it->second;
++expected_seq_;
++total_packets_processed_;
if (on_tick_) {
on_tick_(pkt.payload, pkt.seq_num, pkt.timestamp_ns);
}
pending_buffer_.erase(it);
}
}
void GapRecoveryTracker::store_historical_packet(uint64_t seq_num, const protocol::MarketDataTickPayload& payload) {
size_t idx = seq_num % HISTORY_CAPACITY;
history_cache_[idx].seq_num = seq_num;
history_cache_[idx].payload = payload;
}
std::optional<protocol::MarketDataTickPayload> GapRecoveryTracker::get_historical_packet(uint64_t seq_num) const {
size_t idx = seq_num % HISTORY_CAPACITY;
if (history_cache_[idx].seq_num == seq_num) {
return history_cache_[idx].payload;
}
return std::nullopt;
}
} // namespace quant::market_data
|