Duplicate-Trade Resolver: Rebuild One Effective Trade Without Losing Its History

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

/** Auditable duplicate-trade lifecycle resolver; source messages are never mutated. */

export type Action = "NEW" | "CORRECT" | "CANCEL" | "ERROR";
export type TradeEvent = {
  source_id: string; session_id: string; instrument_id: string;
  event_id: string; trade_id: string; action: string;
  receive_ts: string; sequence: number; ref_event_id?: string;
  price?: number; size?: number; [key: string]: unknown;
};

export const POLICY_VERSION = "synthetic-lifecycle-v2";
const ACTIONS = new Set(["NEW", "CORRECT", "CANCEL", "ERROR"]);
const textFields = ["source_id", "session_id", "instrument_id", "event_id", "trade_id", "action", "receive_ts"] as const;
const eventKey = (e: TradeEvent) => `${e.source_id}|${e.session_id}|${e.event_id}`;
const tradeKey = (e: TradeEvent) => `${e.source_id}|${e.session_id}|${e.instrument_id}|${e.trade_id}`;
const publicEvent = (e: Record<string, unknown>) => Object.fromEntries(Object.entries(e).filter(([k]) => !k.startsWith("_")));
const fingerprint = (e: Record<string, unknown>) => JSON.stringify(Object.fromEntries(Object.entries(publicEvent(e)).sort(([a], [b]) => a.localeCompare(b))));

function normalize(raw: TradeEvent): {event: TradeEvent & {_receive_sort?: number}; errors: string[]} {
  const event = {...raw, action: String(raw.action ?? "").toUpperCase()} as TradeEvent & {_receive_sort?: number};
  const errors: string[] = [];
  for (const field of textFields) if (typeof event[field] !== "string" || !(event[field] as string).trim()) errors.push(`${field} must be a non-empty string`);
  if (!ACTIONS.has(event.action)) errors.push("action must be NEW, CORRECT, CANCEL, or ERROR");
  if (!Number.isInteger(event.sequence) || event.sequence < 0) errors.push("sequence must be a non-negative integer");
  const parsed = Date.parse(event.receive_ts);
  if (!Number.isFinite(parsed) || !/(Z|[+-]\d\d:\d\d)$/.test(event.receive_ts)) errors.push("receive_ts must be a valid timezone-bearing ISO timestamp");
  else event._receive_sort = parsed;
  if (event.action === "NEW" && event.ref_event_id) errors.push("NEW must not carry ref_event_id");
  if (["CORRECT", "CANCEL", "ERROR"].includes(event.action) && (typeof event.ref_event_id !== "string" || !event.ref_event_id.trim())) errors.push(`${event.action} requires ref_event_id`);
  if (["NEW", "CORRECT"].includes(event.action)) {
    if (!Number.isFinite(event.price) || event.price! <= 0 || !Number.isInteger(event.size) || event.size! <= 0) errors.push(`${event.action} requires finite positive price and integer size`);
  } else if (event.price !== undefined || event.size !== undefined) errors.push(`${event.action} must not replace price or size`);
  return {event, errors};
}

export function resolveTrades(input: TradeEvent[]) {
  const rows = input.map(e => ({...e}));
  const auditByIndex = new Map<number, any>();
  const raw_history: any[] = [];
  const valid: Array<[number, TradeEvent & {_receive_sort?: number}]> = [];
  rows.forEach((raw, arrival_index) => {
    const {event, errors} = normalize(raw);
    raw_history.push({arrival_index, event: {...raw}, validation_errors: errors});
    if (errors.length) auditByIndex.set(arrival_index, {arrival_index, event_id: raw.event_id, trade_id: raw.trade_id, decision: "REJECT_INVALID", reason: errors.join("; ")});
    else valid.push([arrival_index, event]);
  });

  const grouped = new Map<string, Array<[number, TradeEvent & {_receive_sort?: number}]>>();
  for (const item of valid) grouped.set(eventKey(item[1]), [...(grouped.get(eventKey(item[1])) ?? []), item]);
  const canonical: Array<[number, TradeEvent & {_receive_sort?: number}]> = [];
  for (const group of grouped.values()) {
    if (new Set(group.map(([, e]) => fingerprint(e))).size > 1) {
      for (const [i, e] of group) auditByIndex.set(i, {arrival_index:i,event_id:e.event_id,trade_id:e.trade_id,decision:"CONFLICT_EVENT_ID",reason:"same scoped event_id arrived with different payloads; all variants quarantined"});
      continue;
    }
    const first = [...group].sort((a,b)=>a[0]-b[0])[0]; canonical.push(first);
    for (const [i,e] of group) if (i !== first[0]) auditByIndex.set(i,{arrival_index:i,event_id:e.event_id,trade_id:e.trade_id,decision:"DROP_REPLAY",reason:"byte-equivalent normalized event_id already retained"});
  }

  const sequenceGroups = new Map<string, typeof canonical>();
  for (const item of canonical) { const e=item[1], k=`${e.source_id}|${e.session_id}|${e.sequence}`; sequenceGroups.set(k,[...(sequenceGroups.get(k)??[]),item]); }
  const eligible: typeof canonical = [];
  for (const group of sequenceGroups.values()) {
    if (group.length > 1) for (const [i,e] of group) auditByIndex.set(i,{arrival_index:i,event_id:e.event_id,trade_id:e.trade_id,decision:"CONFLICT_SEQUENCE",reason:"two event IDs share one source/session sequence; ordering is ambiguous"});
    else eligible.push(group[0]);
  }
  eligible.sort((a,b)=>a[1].source_id.localeCompare(b[1].source_id)||a[1].session_id.localeCompare(b[1].session_id)||a[1].sequence-b[1].sequence||a[1]._receive_sort!-b[1]._receive_sort!||a[1].event_id.localeCompare(b[1].event_id));
  const eligibleByKey = new Map(eligible.map(([,e])=>[eventKey(e),e]));
  const eventTradeKeys = new Map<string,string>();
  const states: Record<string,any> = {}; const versions:any[]=[]; const pending:any[]=[];
  const record=(i:number,e:TradeEvent,decision:string,reason:string,before:string|null,after:string|null)=>{
    auditByIndex.set(i,{arrival_index:i,event_id:e.event_id,trade_id:e.trade_id,decision,reason});
    versions.push({event_key:eventKey(e),trade_key:tradeKey(e),event_id:e.event_id,action:e.action,ref_event_id:e.ref_event_id,sequence:e.sequence,price:e.price,size:e.size,decision,head_before:before,head_after:after});
  };
  for (const [i,e] of eligible) {
    const ek=eventKey(e), tk=tradeKey(e), action=e.action as Action; eventTradeKeys.set(ek,tk);
    const current=states[tk], before=current?.head_event_id??null;
    if(action==="NEW"){
      if(current){record(i,e,"DROP_DUPLICATE_KEY","trade key already has an active or terminal chain",before,before);continue;}
      states[tk]={trade_key:tk,source_id:e.source_id,session_id:e.session_id,instrument_id:e.instrument_id,trade_id:e.trade_id,status:"ACTIVE",head_event_id:e.event_id,head_event_key:ek,price:e.price,size:e.size,chain:[e.event_id]};
      record(i,e,"APPLY_NEW","first valid NEW for this scoped trade key",null,e.event_id);continue;
    }
    const refKey=`${e.source_id}|${e.session_id}|${e.ref_event_id}`;
    if(!eligibleByKey.has(refKey)||!eventTradeKeys.has(refKey)){
      pending.push({event_key:ek,trade_key:tk,ref_event_id:e.ref_event_id,action});record(i,e,"PENDING_REFERENCE","referenced event is missing, quarantined, or not yet valid in source sequence",before,before);continue;
    }
    if(eventTradeKeys.get(refKey)!==tk){record(i,e,"REJECT_CROSS_KEY_REFERENCE","reference crosses instrument, trade, source, or session boundary",before,before);continue;}
    if(!current){pending.push({event_key:ek,trade_key:tk,ref_event_id:e.ref_event_id,action});record(i,e,"PENDING_REFERENCE","referenced event did not produce an effective trade",null,null);continue;}
    if(current.status!=="ACTIVE"){record(i,e,"REJECT_TERMINAL","package policy does not reinstate canceled or erroneous trades",before,before);continue;}
    if(current.head_event_key!==refKey){record(i,e,"REJECT_STALE_REFERENCE","reference is not the current applied version",before,before);continue;}
    const next={...current,head_event_id:e.event_id,head_event_key:ek,chain:[...current.chain,e.event_id]};
    if(action==="CORRECT"){next.price=e.price;next.size=e.size;states[tk]=next;record(i,e,"APPLY_CORRECTION","correction references the current active version",before,e.event_id);}
    else {next.status=action==="CANCEL"?"CANCELED":"ERRORED";next.tombstone={action,event_id:e.event_id,ref_event_id:e.ref_event_id};states[tk]=next;record(i,e,action==="CANCEL"?"APPLY_CANCEL":"APPLY_ERROR",`${action} creates a terminal tombstone under package policy`,before,e.event_id);}
  }
  const orderedStates=Object.fromEntries(Object.entries(states).sort(([a],[b])=>a.localeCompare(b)));
  const values=Object.values(orderedStates);
  return {policy_version:POLICY_VERSION,active:values.filter((s:any)=>s.status==="ACTIVE"),states:orderedStates,tombstones:values.filter((s:any)=>s.status!=="ACTIVE"),pending,versions,raw_history,audit:rows.map((_,i)=>auditByIndex.get(i))};
}