Asynchronous Return Alignment: An Interval-Overlap Diagnostic

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

/** Event-time interval-overlap diagnostics for asynchronous returns.
 *
 * This module does not estimate covariance, allocate returns, or claim to
 * remove nonsynchronous-sampling bias.
 */

export type ReturnInterval = {
  return_id: string;
  instrument: string;
  partition: string;
  event_start: string;
  event_end: string;
  available_at: string;
  return_value?: number;
};

type ValidatedInterval = ReturnInterval & {
  _start_ms: number;
  _end_ms: number;
  _available_ms: number;
};

const milliseconds = (value: string, field: string): number => {
  if (typeof value !== "string" || !/^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/.test(value)) {
    throw new Error(`${field} must be an RFC 3339 UTC timestamp at second or millisecond precision`);
  }
  const parsed = Date.parse(value);
  if (!Number.isFinite(parsed)) throw new Error(`${field} is not a valid timestamp`);
  return parsed;
};

const timestamp = (value: number): string => new Date(value).toISOString();

function validateStream(rows: ReturnInterval[], side: string): ValidatedInterval[] {
  if (!Array.isArray(rows) || rows.length === 0) {
    throw new Error(`${side} stream must contain at least one return interval`);
  }
  const required: (keyof ReturnInterval)[] = [
    "return_id", "instrument", "partition", "event_start", "event_end", "available_at",
  ];
  const seen = new Set<string>();
  const validated = rows.map((row) => {
    const missing = required.filter((key) => !(key in row));
    if (missing.length) throw new Error(`${side} interval is missing ${missing.join(",")}`);
    for (const key of ["return_id", "instrument", "partition"] as const) {
      if (typeof row[key] !== "string" || !row[key]) throw new Error(`${key} must be a non-empty string`);
    }
    if (seen.has(row.return_id)) throw new Error("return_id values must be unique after correction resolution");
    seen.add(row.return_id);
    const start = milliseconds(row.event_start, "event_start");
    const end = milliseconds(row.event_end, "event_end");
    const available = milliseconds(row.available_at, "available_at");
    if (start >= end) throw new Error("event_start must be strictly earlier than event_end");
    if (available < end) throw new Error("available_at cannot precede event_end");
    if (row.return_value !== undefined && !Number.isFinite(row.return_value)) {
      throw new Error("return_value must be finite when supplied");
    }
    return { ...row, _start_ms: start, _end_ms: end, _available_ms: available };
  }).sort((a, b) => a.partition.localeCompare(b.partition) || a._start_ms - b._start_ms || a._end_ms - b._end_ms);

  if (new Set(validated.map((row) => row.instrument)).size !== 1) {
    throw new Error(`${side} stream must contain exactly one instrument`);
  }
  const previous = new Map<string, ValidatedInterval>();
  for (const row of validated) {
    const prior = previous.get(row.partition);
    if (prior && prior._end_ms > row._start_ms) {
      throw new Error("return intervals within a stream cannot overlap");
    }
    previous.set(row.partition, row);
  }
  return validated;
}

const publicInterval = (row: ValidatedInterval): ReturnInterval => ({
  return_id: row.return_id,
  instrument: row.instrument,
  partition: row.partition,
  event_start: row.event_start,
  event_end: row.event_end,
  available_at: row.available_at,
  ...(row.return_value === undefined ? {} : { return_value: row.return_value }),
});

function classifyValidatedPair(left: ValidatedInterval, right: ValidatedInterval, evaluationMs: number | null) {
  const availableMs = Math.max(left._available_ms, right._available_ms);
  let geometry: string;
  let overlapStart: string | null = null;
  let overlapEnd: string | null = null;
  let overlapMs = 0;

  if (left.partition !== right.partition) {
    geometry = "partition_mismatch";
  } else {
    const low = Math.max(left._start_ms, right._start_ms);
    const high = Math.min(left._end_ms, right._end_ms);
    overlapMs = Math.max(0, high - low);
    if (overlapMs > 0) {
      overlapStart = timestamp(low);
      overlapEnd = timestamp(high);
    }
    if (high < low) geometry = "disjoint";
    else if (high === low) geometry = "touching";
    else if (left._start_ms === right._start_ms && left._end_ms === right._end_ms) geometry = "exact";
    else if (left._start_ms >= right._start_ms && left._end_ms <= right._end_ms) geometry = "left_within_right";
    else if (right._start_ms >= left._start_ms && right._end_ms <= left._end_ms) geometry = "right_within_left";
    else geometry = "partial";
  }

  return {
    left_return_id: left.return_id,
    right_return_id: right.return_id,
    partition: left.partition === right.partition ? left.partition : null,
    geometry,
    event_overlap_start: overlapStart,
    event_overlap_end: overlapEnd,
    event_overlap_ms: overlapMs,
    left_overlap_fraction: overlapMs / (left._end_ms - left._start_ms),
    right_overlap_fraction: overlapMs / (right._end_ms - right._start_ms),
    pair_available_at: timestamp(availableMs),
    jointly_available_at_evaluation: evaluationMs === null ? null : availableMs <= evaluationMs,
  };
}

export function classifyIntervalPair(left: ReturnInterval, right: ReturnInterval, evaluatedAt?: string) {
  const a = validateStream([left], "left")[0];
  const b = validateStream([right], "right")[0];
  return classifyValidatedPair(a, b, evaluatedAt === undefined ? null : milliseconds(evaluatedAt, "evaluated_at"));
}

export function diagnoseAsynchronousReturnAlignment(left: ReturnInterval[], right: ReturnInterval[], evaluatedAt?: string) {
  const a = validateStream(left, "left");
  const b = validateStream(right, "right");
  const evaluationMs = evaluatedAt === undefined ? null : milliseconds(evaluatedAt, "evaluated_at");
  const pairs = a.flatMap((leftRow) => b
    .filter((rightRow) => leftRow.partition === rightRow.partition)
    .map((rightRow) => classifyValidatedPair(leftRow, rightRow, evaluationMs)));
  const positive = pairs.filter((pair) => pair.event_overlap_ms > 0);
  const counts: Record<string, number> = {};
  for (const pair of pairs) counts[pair.geometry] = (counts[pair.geometry] ?? 0) + 1;
  const geometryCounts = Object.fromEntries(Object.entries(counts).sort(([x], [y]) => x.localeCompare(y)));
  const totalOverlapMs = positive.reduce((sum, pair) => sum + pair.event_overlap_ms, 0);
  const leftTotalMs = a.reduce((sum, row) => sum + row._end_ms - row._start_ms, 0);
  const rightTotalMs = b.reduce((sum, row) => sum + row._end_ms - row._start_ms, 0);
  return {
    diagnostic_only: true,
    estimator: null,
    weighting_or_return_allocation: "none",
    left_intervals: a.map(publicInterval),
    right_intervals: b.map(publicInterval),
    pairs,
    positive_overlap_pairs: positive,
    geometry_counts: geometryCounts,
    total_event_overlap_ms: totalOverlapMs,
    left_event_time_coverage: totalOverlapMs / leftTotalMs,
    right_event_time_coverage: totalOverlapMs / rightTotalMs,
    all_pairs_available_at_evaluation: evaluationMs === null || pairs.length === 0
      ? null
      : pairs.every((pair) => pair.jointly_available_at_evaluation),
  };
}

export const alignAsynchronousReturns = diagnoseAsynchronousReturnAlignment;