Sequence-Gap Detection and Recovery

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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201

/* Reference algorithms for D01-F05. This file intentionally uses
   JavaScript-compatible TypeScript so the parity runner can execute the same
   source as an ES module without a build dependency. */

const asInt = (name, value, minimum = null) => {
  if (!Number.isInteger(value)) throw new TypeError(`${name} must be an integer`);
  if (minimum !== null && value < minimum) throw new RangeError(`${name} must be >= ${minimum}`);
  return value;
};
const side = value => {
  if (value !== "bid" && value !== "ask") throw new RangeError("side must be bid or ask");
  return value;
};
const sortedLevels = book => ({
  bids: [...book.bid.entries()].filter(([,q])=>q>0).sort((a,b)=>b[0]-a[0]).map(([price_ticks,quantity])=>({price_ticks,quantity})),
  asks: [...book.ask.entries()].filter(([,q])=>q>0).sort((a,b)=>a[0]-b[0]).map(([price_ticks,quantity])=>({price_ticks,quantity})),
});
const bookFromLevels = levels => {
  const book={bid:new Map(),ask:new Map()};
  for (const [plural,s] of [["bids","bid"],["asks","ask"]]) {
    for (const [index,row] of (levels[plural]??[]).entries()) {
      const price=asInt(`${plural}[${index}].price_ticks`,row.price_ticks,1);
      const quantity=asInt(`${plural}[${index}].quantity`,row.quantity,0);
      if (book[s].has(price)) throw new RangeError(`duplicate ${s} price level ${price}`);
      if (quantity) book[s].set(price,quantity);
    }
  }
  return book;
};

export function normalizeEvents(events) {
  const output=[], lastSequence=new Map();
  for (const [index,raw] of events.entries()) {
    const venue=String(raw.venue??"").trim(), instrument=String(raw.instrument??"").trim();
    if (!venue||!instrument) throw new RangeError("venue and instrument are required");
    const sequence=asInt(`events[${index}].sequence`,raw.sequence,0);
    if (lastSequence.has(venue)&&sequence<=lastSequence.get(venue)) throw new RangeError(`sequence must increase within venue ${venue}`);
    lastSequence.set(venue,sequence);
    const event_time_ns=asInt("event_time_ns",raw.event_time_ns,0);
    const receive_time_ns=asInt("receive_time_ns",raw.receive_time_ns,0);
    if (receive_time_ns<event_time_ns) throw new RangeError("receive_time_ns must not precede event_time_ns in this contract");
    const scale=asInt("price_scale",raw.price_scale,1);
    const common={schema_version:1,venue,instrument,sequence,event_time_ns,receive_time_ns,latency_ns:receive_time_ns-event_time_ns,price_scale:scale};
    if (raw.kind==="trade") {
      const price_atoms=asInt("price_atoms",raw.price_atoms,1);
      output.push({...common,kind:"trade",trade_id:String(raw.trade_id??""),price_atoms,price:price_atoms/scale,quantity:asInt("quantity",raw.quantity,1),aggressor_side:raw.aggressor_side??null});
    } else if (raw.kind==="quote") {
      const bid=asInt("bid_price_atoms",raw.bid_price_atoms,1), ask=asInt("ask_price_atoms",raw.ask_price_atoms,1);
      if (bid>ask) throw new RangeError("crossed quote is outside the normalization contract");
      output.push({...common,kind:"quote",bid_price_atoms:bid,bid_price:bid/scale,bid_quantity:asInt("bid_quantity",raw.bid_quantity,0),ask_price_atoms:ask,ask_price:ask/scale,ask_quantity:asInt("ask_quantity",raw.ask_quantity,0),quote_state:bid===ask?"locked":"normal"});
    } else throw new RangeError("kind must be trade or quote");
  }
  return {schema_version:1,event_count:output.length,trade_count:output.filter(x=>x.kind==="trade").length,quote_count:output.filter(x=>x.kind==="quote").length,venue_count:lastSequence.size,max_latency_ns:output.reduce((n,x)=>Math.max(n,x.latency_ns),0),events:output,state:"normalized"};
}

export function reconstructL2(snapshot,{snapshot_sequence,deltas}) {
  const anchor=asInt("snapshot_sequence",snapshot_sequence,0), book=bookFromLevels(snapshot);
  let expected=anchor+1; const applied_sequences=[],discarded_sequences=[];
  for (const [index,delta] of deltas.entries()) {
    const sequence=asInt(`deltas[${index}].sequence`,delta.sequence,0);
    if (sequence<=anchor) {discarded_sequences.push(sequence);continue;}
    if (sequence!==expected) throw new RangeError(`sequence gap: expected ${expected}, received ${sequence}`);
    const s=side(delta.side), price=asInt("price_ticks",delta.price_ticks,1), quantity=asInt("quantity",delta.quantity,0);
    if (quantity) book[s].set(price,quantity); else book[s].delete(price);
    applied_sequences.push(sequence); expected++;
  }
  const levels=sortedLevels(book), best_bid_ticks=levels.bids[0]?.price_ticks??null, best_ask_ticks=levels.asks[0]?.price_ticks??null;
  if (best_bid_ticks!==null&&best_ask_ticks!==null&&best_bid_ticks>best_ask_ticks) throw new RangeError("reconstruction produced a crossed book");
  return {quantity_semantics:"absolute-replacement",snapshot_sequence:anchor,last_sequence:expected-1,applied_sequences,applied_count:applied_sequences.length,discarded_sequences,discarded_count:discarded_sequences.length,...levels,best_bid_ticks,best_ask_ticks,state:"current"};
}

export function aggregatePriceLevels(orders,{depth_limit=null}={}) {
  const totals={bid:new Map(),ask:new Map()}; let order_count=0;
  for (const [index,order] of orders.entries()) {
    const s=side(order.side), price=asInt(`orders[${index}].price_ticks`,order.price_ticks,1), quantity=asInt("quantity",order.quantity,1);
    const current=totals[s].get(price)??[0,0]; current[0]+=quantity; current[1]++; totals[s].set(price,current); order_count++;
  }
  const limit=depth_limit===null?null:asInt("depth_limit",depth_limit,1);
  let bids=[...totals.bid.entries()].sort((a,b)=>b[0]-a[0]).map(([price_ticks,v])=>({price_ticks,quantity:v[0],order_count:v[1]}));
  let asks=[...totals.ask.entries()].sort((a,b)=>a[0]-b[0]).map(([price_ticks,v])=>({price_ticks,quantity:v[0],order_count:v[1]}));
  const full_bid_level_count=bids.length, full_ask_level_count=asks.length,full_bid_quantity=bids.reduce((n,x)=>n+x.quantity,0),full_ask_quantity=asks.reduce((n,x)=>n+x.quantity,0);
  if (limit!==null) {bids=bids.slice(0,limit);asks=asks.slice(0,limit);}
  const visible_bid_quantity=bids.reduce((n,x)=>n+x.quantity,0),visible_ask_quantity=asks.reduce((n,x)=>n+x.quantity,0);
  return {order_count,full_bid_level_count,full_ask_level_count,depth_limit:limit,bids,asks,visible_bid_quantity,visible_ask_quantity,full_bid_quantity,full_ask_quantity,hidden_by_depth_limit_bid_quantity:full_bid_quantity-visible_bid_quantity,hidden_by_depth_limit_ask_quantity:full_ask_quantity-visible_ask_quantity,state:"aggregated"};
}

export function reconstructL3(snapshot_orders,{snapshot_sequence,events}) {
  const anchor=asInt("snapshot_sequence",snapshot_sequence,0), orders=new Map(),action_counts={add:0,reduce:0,execute:0,delete:0,replace:0}; let rank=0, expected=anchor+1;
  for (const raw of snapshot_orders) {
    const order_id=String(raw.order_id??"");
    if (!order_id||orders.has(order_id)) throw new RangeError("snapshot order IDs must be non-empty and unique");
    rank++; orders.set(order_id,{order_id,side:side(raw.side),price_ticks:asInt("price_ticks",raw.price_ticks,1),quantity:asInt("quantity",raw.quantity,1),priority_rank:rank});
  }
  for (const [index,event] of events.entries()) {
    const sequence=asInt(`events[${index}].sequence`,event.sequence,0);
    if (sequence<=anchor) continue;
    if (sequence!==expected) throw new RangeError(`sequence gap: expected ${expected}, received ${sequence}`);
    const action=event.action, order_id=String(event.order_id??"");
    if (action==="add") {
      if (!order_id||orders.has(order_id)) throw new RangeError("add requires a new order_id");
      rank++; orders.set(order_id,{order_id,side:side(event.side),price_ticks:asInt("price_ticks",event.price_ticks,1),quantity:asInt("quantity",event.quantity,1),priority_rank:rank});
    } else if (action==="reduce"||action==="execute") {
      if (!orders.has(order_id)) throw new RangeError(`${action} references unknown order`);
      const reduction=asInt("quantity",event.quantity,1), row=orders.get(order_id);
      if (reduction>row.quantity) throw new RangeError("reduction exceeds resting quantity");
      row.quantity-=reduction; if (!row.quantity) orders.delete(order_id);
    } else if (action==="delete") {
      if (!orders.has(order_id)) throw new RangeError("delete references unknown order");
      orders.delete(order_id);
    } else if (action==="replace") {
      if (!orders.has(order_id)) throw new RangeError("replace references unknown order");
      const original=orders.get(order_id);orders.delete(order_id);
      const new_order_id=String(event.new_order_id??"");
      if (!new_order_id||orders.has(new_order_id)) throw new RangeError("replace requires a new unique new_order_id");
      rank++;orders.set(new_order_id,{order_id:new_order_id,side:original.side,price_ticks:asInt("price_ticks",event.price_ticks,1),quantity:asInt("quantity",event.quantity,1),priority_rank:rank});
    } else throw new RangeError("action must be add, reduce, execute, delete, or replace");
    action_counts[action]++;
    expected++;
  }
  const rows=[...orders.values()].sort((a,b)=>a.priority_rank-b.priority_rank), aggregation=aggregatePriceLevels(rows);
  return {snapshot_sequence:anchor,last_sequence:expected-1,order_count:rows.length,event_count:Object.values(action_counts).reduce((a,b)=>a+b,0),action_counts,orders:rows,bids:aggregation.bids,asks:aggregation.asks,state:"current"};
}

export function recoverSequenceStream(arrivals,{start_sequence}) {
  let expected=asInt("start_sequence",start_sequence,0),lastRequest=null; const buffer=new Map(), applied_sequences=[], duplicate_sequences=[], trace=[],recovery_requests=[];
  for (const [index,arrival] of arrivals.entries()) {
    const sequence=asInt(`arrivals[${index}].sequence`,arrival.sequence,0), source=String(arrival.source??"live");
    if (sequence<expected||buffer.has(sequence)) duplicate_sequences.push(sequence); else buffer.set(sequence,source);
    while (buffer.has(expected)) {buffer.delete(expected);applied_sequences.push(expected);expected++;}
    const buffered=[...buffer.keys()].sort((a,b)=>a-b), stop=buffered.length?buffered[0]:expected;
    const missing=Array.from({length:stop-expected},(_,i)=>expected+i);
    if (missing.length) {
      const key=`${missing[0]}:${missing.at(-1)}`;
      if (key!==lastRequest) {recovery_requests.push({from_sequence:missing[0],to_sequence:missing.at(-1)});lastRequest=key;}
    }
    trace.push({arrival_sequence:sequence,source,next_expected:expected,missing,buffered});
  }
  const buffered_sequences=[...buffer.keys()].sort((a,b)=>a-b), stop=buffered_sequences.length?buffered_sequences[0]:expected;
  return {start_sequence:asInt("start_sequence",start_sequence,0),applied_sequences,next_expected:expected,missing_sequences:Array.from({length:stop-expected},(_,i)=>expected+i),buffered_sequences,duplicate_sequences,recovery_requests,recovery_request_count:recovery_requests.length,replay_arrival_count:arrivals.filter(x=>(x.source??"live")==="replay").length,trace,state:buffer.size?"recovering":"current"};
}

export function reconcileSnapshotIncrementals(snapshot,{snapshot_sequence,incrementals,checkpoint,checkpoint_sequence}) {
  const anchor=asInt("snapshot_sequence",snapshot_sequence,0), check=asInt("checkpoint_sequence",checkpoint_sequence,anchor);
  const eligible=incrementals.filter(row=>asInt("sequence",row.sequence,0)<=check),pending=incrementals.filter(row=>asInt("sequence",row.sequence,0)>check), reconstructed=reconstructL2(snapshot,{snapshot_sequence:anchor,deltas:eligible});
  if (reconstructed.last_sequence!==check) throw new RangeError("incrementals do not reach checkpoint_sequence contiguously");
  const observed=bookFromLevels(checkpoint), rebuilt={bid:new Map(reconstructed.bids.map(x=>[x.price_ticks,x.quantity])),ask:new Map(reconstructed.asks.map(x=>[x.price_ticks,x.quantity]))}, differences=[];
  for (const s of ["bid","ask"]) {
    const prices=[...new Set([...rebuilt[s].keys(),...observed[s].keys()])].sort((a,b)=>s==="bid"?b-a:a-b);
    for (const price_ticks of prices) {
      const left=rebuilt[s].get(price_ticks)??0,right=observed[s].get(price_ticks)??0;
      if (left!==right) differences.push({side:s,price_ticks,reconstructed_quantity:left,snapshot_quantity:right,difference:left-right,kind:left===0?"missing-reconstructed-level":right===0?"missing-checkpoint-level":"quantity-mismatch"});
    }
  }
  const mismatch_counts={"quantity-mismatch":0,"missing-reconstructed-level":0,"missing-checkpoint-level":0};for(const row of differences)mismatch_counts[row.kind]++;
  return {snapshot_sequence:anchor,checkpoint_sequence:check,compared_level_count:new Set([...rebuilt.bid.keys(),...observed.bid.keys()]).size+new Set([...rebuilt.ask.keys(),...observed.ask.keys()]).size,difference_count:differences.length,mismatch_counts,replayed_incremental_count:eligible.length,pending_after_checkpoint_count:pending.length,differences,reconstructed:{bids:reconstructed.bids,asks:reconstructed.asks},checkpoint:sortedLevels(observed),state:differences.length?"drift-detected":"reconciled"};
}

export function consolidateVenues(quotes,{as_of_receive_time_ns,max_staleness_ns}) {
  const as_of=asInt("as_of_receive_time_ns",as_of_receive_time_ns,0), tolerance=asInt("max_staleness_ns",max_staleness_ns,0), latest=new Map();
  for (const [index,quote] of quotes.entries()) {
    const venue=String(quote.venue??"").trim(); if (!venue) throw new RangeError("venue is required");
    const receive=asInt(`quotes[${index}].receive_time_ns`,quote.receive_time_ns,0); if (receive>as_of) continue;
    if (!latest.has(venue)||receive>latest.get(venue).receive_time_ns) latest.set(venue,{...quote});
  }
  const eligible=[],excluded=[];
  for (const [venue,quote] of [...latest.entries()].sort()) {
    const age=as_of-asInt("receive_time_ns",quote.receive_time_ns,0);
    if (quote.eligible===false) excluded.push({venue,reason:"ineligible"});
    else if ((quote.status??"open")!=="open") excluded.push({venue,reason:"not-open"});
    else if (age>tolerance) excluded.push({venue,reason:"stale"});
    else {
      const levels=("bids" in quote||"asks" in quote)?sortedLevels(bookFromLevels(quote)):{bids:[{price_ticks:asInt("bid_price_ticks",quote.bid_price_ticks,1),quantity:asInt("bid_quantity",quote.bid_quantity,0)}],asks:[{price_ticks:asInt("ask_price_ticks",quote.ask_price_ticks,1),quantity:asInt("ask_quantity",quote.ask_quantity,0)}]};
      if (!levels.bids.length||!levels.asks.length) {excluded.push({venue,reason:"empty-local-side"});continue;}
      if (levels.bids[0].price_ticks>levels.asks[0].price_ticks) {excluded.push({venue,reason:"crossed-local-book"});continue;}
      eligible.push({venue,bids:levels.bids,asks:levels.asks,receive_time_ns:quote.receive_time_ns});
    }
  }
  if (!eligible.length) return {state:"no-eligible-quote",eligible_venue_count:0,excluded};
  const depth={bid:new Map(),ask:new Map()};
  for (const row of eligible) {
    for (const [plural,s] of [["bids","bid"],["asks","ask"]]) for (const level of row[plural]) {
      if (!depth[s].has(level.price_ticks)) depth[s].set(level.price_ticks,{quantity:0,venues:[]});
      const cell=depth[s].get(level.price_ticks);cell.quantity+=level.quantity;cell.venues.push(row.venue);
    }
  }
  const bids=[...depth.bid.entries()].sort((a,b)=>b[0]-a[0]).map(([price_ticks,v])=>({price_ticks,quantity:v.quantity,venues:v.venues.sort()})),asks=[...depth.ask.entries()].sort((a,b)=>a[0]-b[0]).map(([price_ticks,v])=>({price_ticks,quantity:v.quantity,venues:v.venues.sort()}));
  const best_bid_ticks=bids[0].price_ticks,best_ask_ticks=asks[0].price_ticks;
  return {as_of_receive_time_ns:as_of,max_staleness_ns:tolerance,eligible_venue_count:eligible.length,excluded,best_bid_ticks,best_bid_quantity:bids[0].quantity,best_bid_venues:bids[0].venues,best_ask_ticks,best_ask_quantity:asks[0].quantity,best_ask_venues:asks[0].venues,consolidated_book:{bids,asks},consolidated_levels:[...bids.map(x=>({side:"bid",...x})),...asks.map(x=>({side:"ask",...x}))],state:best_bid_ticks>best_ask_ticks?"crossed-consolidated":best_bid_ticks===best_ask_ticks?"locked-consolidated":"normal"};
}

export function calculate(topicId, inputs) {
  if (topicId==="D01-F05-A01") return normalizeEvents(inputs.events);
  if (topicId==="D01-F05-A02") return reconstructL2(inputs.snapshot,{snapshot_sequence:inputs.snapshot_sequence,deltas:inputs.deltas});
  if (topicId==="D01-F05-A03") return reconstructL3(inputs.snapshot_orders,{snapshot_sequence:inputs.snapshot_sequence,events:inputs.events});
  if (topicId==="D01-F05-A04") return recoverSequenceStream(inputs.arrivals,{start_sequence:inputs.start_sequence});
  if (topicId==="D01-F05-A05") return aggregatePriceLevels(inputs.orders,{depth_limit:inputs.depth_limit??null});
  if (topicId==="D01-F05-A06") return reconcileSnapshotIncrementals(inputs.snapshot,{snapshot_sequence:inputs.snapshot_sequence,incrementals:inputs.incrementals,checkpoint:inputs.checkpoint,checkpoint_sequence:inputs.checkpoint_sequence});
  if (topicId==="D01-F05-A07") return consolidateVenues(inputs.quotes,{as_of_receive_time_ns:inputs.as_of_receive_time_ns,max_staleness_ns:inputs.max_staleness_ns});
  throw new RangeError(`unsupported topic_id: ${topicId}`);
}