packages/core/src/features/viz/lib/financial-aggregation.ts
This is the source snapshot used to build these API details. View this revision on GitHub.
1 import type { LineAccessor } from "./types.js";
2 import { finiteCoordinate, pointAccessor, pointData } from "./point-data.js";
3
4 /** One half-open time interval used to group complete OHLCV records. See {@link AggregateOhlcvOptions}. */
5 export interface OhlcvBucket {
6 /** Inclusive finite start coordinate. */
7 readonly start: number;
8 /** Exclusive finite end coordinate after the start. */
9 readonly end: number;
10 }
11
12 /** Original source record that contributed to an aggregated OHLCV bar. See {@link AggregatedOhlcvPoint}. */
13 export interface OhlcvSource<D> {
14 /** Caller-owned complete contributing record. */
15 readonly datum: D;
16 /** Original source index. */
17 readonly sourceIndex: number;
18 }
19
20 /** One frozen OHLCV bar aggregated without sorting caller-owned records. See {@link aggregateOhlcv}. */
21 export interface AggregatedOhlcvPoint<D> extends OhlcvBucket {
22 /** Open from the first contributing record. */
23 readonly open: number;
24 /** Maximum high from contributing records. */
25 readonly high: number;
26 /** Minimum low from contributing records. */
27 readonly low: number;
28 /** Close from the last contributing record. */
29 readonly close: number;
30 /** Finite sum of non-negative contributing volumes. */
31 readonly volume: number;
32 /** Frozen references to contributing caller-owned records. */
33 readonly sources: readonly OhlcvSource<D>[];
34 }
35
36 /** Inputs for chronological OHLCV aggregation. See {@link aggregateOhlcv}. */
37 export interface AggregateOhlcvOptions<D> {
38 /** Numeric property or callback that supplies each record's time coordinate. */
39 readonly time: LineAccessor<D>;
40 /** Numeric property or callback that supplies open. */
41 readonly open: LineAccessor<D>;
42 /** Numeric property or callback that supplies high. */
43 readonly high: LineAccessor<D>;
44 /** Numeric property or callback that supplies low. */
45 readonly low: LineAccessor<D>;
46 /** Numeric property or callback that supplies close. */
47 readonly close: LineAccessor<D>;
48 /** Numeric property or callback that supplies non-negative volume. */
49 readonly volume: LineAccessor<D>;
50 /**
51 * Returns the half-open aggregate bucket for one eligible finite timestamp, or null to skip it.
52 * @param time - Eligible finite source timestamp.
53 * @returns A containing half-open bucket, or null when the record is not aggregated.
54 * @see {@link OhlcvBucket}
55 */
56 readonly bucket: (time: number) => OhlcvBucket | null;
57 /**
58 * Excludes a record before its OHLCV and bucket accessors run.
59 * @param datum - Original source observation.
60 * @param index - Zero-based source index.
61 * @param data - Complete caller-supplied source array.
62 * @returns Whether this observation may contribute.
63 */
64 readonly defined?: (datum: D, index: number, data: readonly D[]) => boolean;
65 }
66
67 /** A valid session identifier supplied to {@link sessionVwap}. */
68 export type SessionIdentifier = string | number;
69
70 /**
71 * A property or callback that supplies the session containing a source record.
72 * @param datum - Source observation under consideration.
73 * @param index - Zero-based source index.
74 * @param data - Complete caller-supplied source array.
75 * @returns The record's string or finite numeric session identifier.
76 * @see {@link SessionVwapOptions}
77 */
78 export type SessionAccessor<D> =
79 | { [K in keyof D]-?: D[K] extends SessionIdentifier ? K : never }[keyof D]
80 | ((datum: D, index: number, data: readonly D[]) => SessionIdentifier);
81
82 /** Inputs for source-aligned session VWAP. See {@link sessionVwap}. */
83 export interface SessionVwapOptions<D> {
84 /** Numeric property or callback that supplies each record's price. */
85 readonly price: LineAccessor<D>;
86 /** Numeric property or callback that supplies non-negative volume. */
87 readonly volume: LineAccessor<D>;
88 /** Property or callback that supplies the current string or finite numeric session identifier. */
89 readonly session: SessionAccessor<D>;
90 /**
91 * Excludes a row before price and volume accessors run; its session still resets state.
92 * @param datum - Original source observation.
93 * @param index - Zero-based source index.
94 * @param data - Complete caller-supplied source array.
95 * @returns Whether this observation participates in VWAP calculation.
96 */
97 readonly defined?: (datum: D, index: number, data: readonly D[]) => boolean;
98 }
99
100 /** One frozen source-aligned session VWAP point. See {@link sessionVwap}. */
101 export interface SessionVwapPoint<D> {
102 /** Caller-owned original record. */
103 readonly datum: D;
104 /** Original source index, including excluded and incomplete records. */
105 readonly sourceIndex: number;
106 /** Evaluated session, including for an excluded source record. */
107 readonly session: SessionIdentifier;
108 /** Finite volume accumulated for this session before or including this record. */
109 readonly cumulativeVolume: number;
110 /** Session VWAP after this valid record, null for incomplete or excluded records. */
111 readonly vwap: number | null;
112 }
113
114 function validBucket(value: OhlcvBucket | null, time: number): OhlcvBucket | null {
115 if (value === null) return null;
116 if (value === undefined || typeof value !== "object" || Array.isArray(value))
117 throw new TypeError("aggregateOhlcv bucket must return an object or null.");
118 const { start, end } = value;
119 if (!finiteCoordinate(start) || !finiteCoordinate(end) || start >= end)
120 throw new RangeError("aggregateOhlcv bucket bounds must be finite and increasing.");
121 if (time < start || time >= end)
122 throw new RangeError("aggregateOhlcv bucket must contain its source time.");
123 return value;
124 }
125
126 function validateCandle(
127 open: number, high: number, low: number, close: number, volume: number,
128 ): void {
129 if (volume < 0) throw new RangeError("aggregateOhlcv volume must be non-negative.");
130 if (low > high || open < low || open > high || close < low || close > high)
131 throw new RangeError("aggregateOhlcv high and low must contain open and close.");
132 }
133
134 function sessionAccessor<D>(input: SessionAccessor<D>): (datum: D, index: number, data: readonly D[]) => SessionIdentifier {
135 return typeof input === "function"
136 ? input
137 : (datum) => datum[input] as SessionIdentifier;
138 }
139
140 function validSession(value: unknown): value is SessionIdentifier {
141 if (typeof value === "string") return true;
142 if (typeof value === "number" && Number.isFinite(value)) return true;
143 throw new TypeError("sessionVwap session must be a string or finite number.");
144 }
145
146 function weightedAverage(previous: number, previousVolume: number, price: number, volume: number): number {
147 const total = previousVolume + volume;
148 if (!Number.isFinite(total)) throw new RangeError("sessionVwap cumulative volume must be finite.");
149 const scale = Math.max(Math.abs(previous), Math.abs(price));
150 if (scale === 0) return 0;
151 const result = ((previous / scale) * (previousVolume / total) + (price / scale) * (volume / total)) * scale;
152 if (!Number.isFinite(result)) throw new RangeError("sessionVwap derived value must be finite.");
153 return result;
154 }
155
156 /**
157 * Aggregates source-ordered complete OHLCV records into caller-defined chronological buckets.
158 *
159 * It never sorts or mutates source records. Missing, excluded, and null-bucket records are skipped;
160 * finite malformed candles, negative volume, noncontiguous bucket bounds, and time disorder reject.
161 * @param data - Caller-owned source observations in chronological order.
162 * @param options - OHLCV accessors, bucket definition, and optional participation callback.
163 * @returns Frozen anchored aggregate bars with frozen original source references.
164 * @see {@link AggregateOhlcvOptions}
165 * @see {@link AggregatedOhlcvPoint}
166 */
167 export function aggregateOhlcv<D>(data: readonly D[], options: AggregateOhlcvOptions<D>): readonly AggregatedOhlcvPoint<D>[] {
168 const rows = pointData(data, "aggregateOhlcv");
169 const time = pointAccessor(options.time), open = pointAccessor(options.open), high = pointAccessor(options.high);
170 const low = pointAccessor(options.low), close = pointAccessor(options.close), volume = pointAccessor(options.volume);
171 const aggregates: AggregatedOhlcvPoint<D>[] = [];
172 let previousTime: number | undefined;
173 let current: { start: number; end: number; open: number; high: number; low: number; close: number; volume: number; sources: OhlcvSource<D>[] } | undefined;
174 for (let index = 0; index < rows.length; index++) {
175 const datum = rows[index]!;
176 if (!(options.defined?.(datum, index, rows) ?? true)) continue;
177 const timestamp = time(datum, index, rows);
178 if (!finiteCoordinate(timestamp)) continue;
179 if (previousTime !== undefined && timestamp < previousTime)
180 throw new RangeError("aggregateOhlcv data must be chronological.");
181 previousTime = timestamp;
182 const interval = validBucket(options.bucket(timestamp), timestamp);
183 if (interval === null) continue;
184 const o = open(datum, index, rows), h = high(datum, index, rows), l = low(datum, index, rows);
185 const c = close(datum, index, rows), v = volume(datum, index, rows);
186 if (!finiteCoordinate(o) || !finiteCoordinate(h) || !finiteCoordinate(l)
187 || !finiteCoordinate(c) || !finiteCoordinate(v)) continue;
188 validateCandle(o, h, l, c, v);
189 if (current === undefined || current.start !== interval.start || current.end !== interval.end) {
190 if (current !== undefined) {
191 if (interval.start < current.end)
192 throw new RangeError("aggregateOhlcv buckets must be non-overlapping and source ordered.");
193 aggregates.push(Object.freeze({ ...current, sources: Object.freeze(current.sources) }));
194 }
195 current = { start: interval.start, end: interval.end, open: o, high: h, low: l, close: c, volume: v, sources: [Object.freeze({ datum, sourceIndex: index })] };
196 continue;
197 }
198 const nextVolume = current.volume + v;
199 if (!Number.isFinite(nextVolume)) throw new RangeError("aggregateOhlcv volume must remain finite.");
200 current.high = Math.max(current.high, h);
201 current.low = Math.min(current.low, l);
202 current.close = c;
203 current.volume = nextVolume;
204 current.sources.push(Object.freeze({ datum, sourceIndex: index }));
205 }
206 if (current !== undefined)
207 aggregates.push(Object.freeze({ ...current, sources: Object.freeze(current.sources) }));
208 return Object.freeze(aggregates);
209 }
210
211 /**
212 * Derives source-aligned volume-weighted prices that reset whenever the session identifier changes.
213 *
214 * Missing prices or volumes and excluded rows produce null without changing the current accumulated
215 * state. Zero-volume complete records preserve the preceding VWAP; negative volumes reject.
216 * @param data - Caller-owned source observations in presentation order.
217 * @param options - Price, volume, session accessors, and optional participation callback.
218 * @returns Frozen source-aligned points with current cumulative volume and VWAP or null.
219 * @see {@link SessionVwapOptions}
220 * @see {@link SessionVwapPoint}
221 */
222 export function sessionVwap<D>(data: readonly D[], options: SessionVwapOptions<D>): readonly SessionVwapPoint<D>[] {
223 const rows = pointData(data, "sessionVwap");
224 const price = pointAccessor(options.price), volume = pointAccessor(options.volume), session = sessionAccessor(options.session);
225 const results: SessionVwapPoint<D>[] = [];
226 let activeSession: SessionIdentifier | undefined;
227 let cumulativeVolume = 0;
228 let average: number | undefined;
229 for (let index = 0; index < rows.length; index++) {
230 const datum = rows[index]!;
231 const currentSession = session(datum, index, rows);
232 validSession(currentSession);
233 if (activeSession !== currentSession) {
234 activeSession = currentSession;
235 cumulativeVolume = 0;
236 average = undefined;
237 }
238 if (!(options.defined?.(datum, index, rows) ?? true)) {
239 results.push(Object.freeze({ datum, sourceIndex: index, session: currentSession, cumulativeVolume, vwap: null }));
240 continue;
241 }
242 const currentPrice = price(datum, index, rows), currentVolume = volume(datum, index, rows);
243 if (!finiteCoordinate(currentPrice) || !finiteCoordinate(currentVolume)) {
244 results.push(Object.freeze({ datum, sourceIndex: index, session: currentSession, cumulativeVolume, vwap: null }));
245 continue;
246 }
247 if (currentVolume < 0) throw new RangeError("sessionVwap volume must be non-negative.");
248 if (currentVolume > 0) {
249 average = average === undefined ? currentPrice : weightedAverage(average, cumulativeVolume, currentPrice, currentVolume);
250 cumulativeVolume += currentVolume;
251 if (!Number.isFinite(cumulativeVolume)) throw new RangeError("sessionVwap cumulative volume must be finite.");
252 }
253 results.push(Object.freeze({ datum, sourceIndex: index, session: currentSession, cumulativeVolume, vwap: average ?? null }));
254 }
255 return Object.freeze(results);
256 }
257 Documentation version
Section titled “Documentation version”Documentation built with @pibbl/core 0.0.2, revision dd39ce2. ALPHA — NOT FOR PRODUCTION USE.