Time Bars

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

/** Reference implementation of session-anchored, left-closed Time Bars. */
export type Trade = {
  tradeId: string;
  timestamp: string;
  session: string;
  symbol: string;
  price: number;
  volume: number;
  currency: string;
  sequence?: number;
};

export type Config = {
  intervalSeconds: number;
  sessionStarts: Record<string, string>;
  closePartial?: boolean;
  emptyBarPolicy?: "omit";
};

export type Bar = {
  barIndex: number;
  session: string;
  intervalIndex: number;
  startTime: string;
  endTime: string;
  firstTradeTime: string;
  lastTradeTime: string;
  open: number;
  high: number;
  low: number;
  close: number;
  volume: number;
  dollarValue: number;
  tickCount: number;
  firstTradeId: string;
  lastTradeId: string;
  emptyIntervalsBefore: number;
  closeReason: "interval" | "session_end" | "stream_end";
};

const UTC_MILLISECOND = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}Z$/;

function milliseconds(value: unknown): number {
  if (typeof value !== "string" || !UTC_MILLISECOND.test(value)) {
    throw new Error("timestamp must use YYYY-MM-DDTHH:mm:ss.sssZ");
  }
  const parsed = Date.parse(value);
  if (!Number.isFinite(parsed)) throw new Error("timestamp must be a valid UTC instant");
  return parsed;
}

function positive(value: unknown, name: string): number {
  if (typeof value !== "number" || !Number.isFinite(value) || value <= 0) {
    throw new Error(`${name} must be a finite positive number`);
  }
  return value;
}

function rounded(value: number): number {
  return Number(value.toFixed(8));
}

function validate(trades: unknown, config: Config): asserts trades is Trade[] {
  if (!Array.isArray(trades)) throw new Error("trades must be an array");
  if (!config || typeof config !== "object") throw new Error("config must be an object");
  positive(config.intervalSeconds, "intervalSeconds");
  if (!config.sessionStarts || typeof config.sessionStarts !== "object" || !Object.keys(config.sessionStarts).length) {
    throw new Error("sessionStarts must be a non-empty mapping");
  }
  for (const [session, anchor] of Object.entries(config.sessionStarts)) {
    if (!session || typeof anchor !== "string") throw new Error("sessionStarts contains an invalid entry");
    milliseconds(anchor);
  }
  if ((config.emptyBarPolicy ?? "omit") !== "omit") {
    throw new Error("this reference implementation supports emptyBarPolicy='omit' only");
  }
  if (config.closePartial !== undefined && typeof config.closePartial !== "boolean") {
    throw new Error("closePartial must be boolean");
  }

  const required = ["tradeId", "timestamp", "session", "symbol", "price", "volume", "currency"] as const;
  const ids = new Set<string>();
  const completedSessions = new Set<string>();
  let activeSession: string | null = null;
  let previousTime = -Infinity;
  let previousSequence: number | undefined;
  let symbol: string | undefined;
  let currency: string | undefined;

  for (const raw of trades) {
    if (!raw || typeof raw !== "object") throw new Error("trade must be an object");
    const trade = raw as Record<string, unknown>;
    if (required.some((field) => !(field in trade))) throw new Error("trade is missing a required field");
    for (const field of ["tradeId", "timestamp", "session", "symbol", "currency"] as const) {
      if (typeof trade[field] !== "string" || !trade[field]) throw new Error(`${field} must be a non-empty string`);
    }
    const now = milliseconds(trade.timestamp);
    if (now < previousTime) throw new Error("trades must be globally ordered by execution event time");
    if (now === previousTime) {
      if (!Number.isInteger(trade.sequence) || previousSequence === undefined || Number(trade.sequence) <= previousSequence) {
        throw new Error("equal timestamps require a strictly increasing integer sequence");
      }
    } else if (trade.sequence !== undefined && !Number.isInteger(trade.sequence)) {
      throw new Error("sequence must be an integer when supplied");
    }
    previousTime = now;
    previousSequence = Number.isInteger(trade.sequence) ? Number(trade.sequence) : undefined;

    const id = trade.tradeId as string;
    if (ids.has(id)) throw new Error("tradeId must be unique after corrections and cancels are resolved");
    ids.add(id);
    positive(trade.price, "price");
    positive(trade.volume, "volume");

    const nextSession = trade.session as string;
    if (activeSession !== nextSession) {
      if (activeSession !== null) completedSessions.add(activeSession);
      if (completedSessions.has(nextSession)) throw new Error("a session cannot reappear after a later session begins");
      activeSession = nextSession;
    }
    if (!(activeSession in config.sessionStarts)) throw new Error("sessionStarts must contain every session");
    if (now < milliseconds(config.sessionStarts[activeSession])) throw new Error("trade precedes its declared session anchor");

    symbol ??= trade.symbol as string;
    currency ??= trade.currency as string;
    if (trade.symbol !== symbol || trade.currency !== currency) {
      throw new Error("one constructBars call must contain one symbol and one currency");
    }
  }
}

export function constructBars(trades: Trade[], config: Config): Bar[] {
  validate(trades, config);
  if (!trades.length) return [];

  const bars: Bar[] = [];
  let current: Trade[] = [];
  let session: string | null = null;
  let bucket: number | null = null;
  let previousEmittedSession: string | null = null;
  let previousEmittedBucket: number | null = null;

  const emit = (reason: Bar["closeReason"]): void => {
    if (!current.length || session === null || bucket === null) return;
    const intervalMs = config.intervalSeconds * 1000;
    const startMs = milliseconds(config.sessionStarts[session]) + bucket * intervalMs;
    const prices = current.map((trade) => trade.price);
    const volumes = current.map((trade) => trade.volume);
    const emptyIntervalsBefore = previousEmittedSession === session && previousEmittedBucket !== null
      ? Math.max(0, bucket - previousEmittedBucket - 1)
      : bucket;
    bars.push({
      barIndex: bars.length,
      session,
      intervalIndex: bucket,
      startTime: new Date(startMs).toISOString(),
      endTime: new Date(startMs + intervalMs).toISOString(),
      firstTradeTime: current[0].timestamp,
      lastTradeTime: current.at(-1)!.timestamp,
      open: rounded(prices[0]),
      high: rounded(Math.max(...prices)),
      low: rounded(Math.min(...prices)),
      close: rounded(prices.at(-1)!),
      volume: rounded(volumes.reduce((sum, value) => sum + value, 0)),
      dollarValue: rounded(current.reduce((sum, trade) => sum + trade.price * trade.volume, 0)),
      tickCount: current.length,
      firstTradeId: current[0].tradeId,
      lastTradeId: current.at(-1)!.tradeId,
      emptyIntervalsBefore,
      closeReason: reason,
    });
    previousEmittedSession = session;
    previousEmittedBucket = bucket;
    current = [];
  };

  for (const trade of trades) {
    if (session !== null && trade.session !== session) {
      if (current.length && config.closePartial !== false) emit("session_end");
      else current = [];
      session = trade.session;
      bucket = null;
    } else if (session === null) {
      session = trade.session;
    }
    const offset = milliseconds(trade.timestamp) - milliseconds(config.sessionStarts[session]);
    const nextBucket = Math.floor(offset / (config.intervalSeconds * 1000));
    if (bucket !== null && nextBucket !== bucket) emit("interval");
    bucket = nextBucket;
    current.push(trade);
  }
  if (current.length && config.closePartial !== false) emit("stream_end");
  return bars;
}