Repository Public
1// Consolidation rule, cross-source deviation and the quality verdict. Pure functions over bars already validated and stored.2//3// CONSOLIDATION RULE (the only one): sources are ranked (default coinbase > bitstamp > kraken). For each bar-open time,4// the consolidated bar is the WHOLE bar (open, high, low, close, volume) of the highest-ranked source that has one, and5// it carries that source's name. Fields of two sources are never combined inside one bar, and nothing is interpolated:6// a time no source has is a gap. Volume is therefore the volume of one venue, not a market total, and it changes scale7// where the series falls back to another source (Coinbase omits intervals without a trade; the others print them with volume 0).8import { findGaps, type GapRange } from './validate.ts';9import { INTERVAL_MS, SOURCE_IDS, expectedLastTs, type BySource, type Candle, type Interval, type MarketErrorCode, type SeriesBar, type SourceId } from './types.ts';1011export const DEFAULT_PRIORITY: readonly SourceId[] = SOURCE_IDS;1213export function consolidate(bySource: BySource, priority: readonly SourceId[] = DEFAULT_PRIORITY): SeriesBar[] {14 const out = new Map<number, SeriesBar>();15 for (const source of priority) for (const c of bySource[source] ?? []) if (!out.has(c.ts)) out.set(c.ts, { ...c, source });16 return [...out.values()].sort((a, b) => a.ts - b.ts);17}1819/** Spread of the closes of one bar across sources, in basis points of their midpoint: (max - min) / ((max + min) / 2) * 10^4. */20export function spreadBps(closes: number[]): number | null {21 if (closes.length < 2) return null;22 const lo = Math.min(...closes), hi = Math.max(...closes);23 return ((hi - lo) / ((hi + lo) / 2)) * 1e4;24}2526export interface Deviation { ts: number; bps: number; closes: Partial<Record<SourceId, number>> }27/** Close-price deviation for every bar that at least two sources have, oldest first. */28export function deviations(bySource: BySource, from = -Infinity): Deviation[] {29 const at = new Map<number, Partial<Record<SourceId, number>>>();30 for (const s of SOURCE_IDS) for (const c of bySource[s] ?? []) if (c.ts >= from) at.set(c.ts, { ...at.get(c.ts), [s]: c.close });31 const out: Deviation[] = [];32 for (const [ts, closes] of at) { const bps = spreadBps(Object.values(closes)); if (bps !== null) out.push({ ts, bps, closes }); }33 return out.sort((a, b) => a.ts - b.ts);34}3536export interface QualityOpts {37 /** bars, ending at the latest one, that must form an unbroken series */38 lookback: number;39 /** when false, a history shorter than `lookback` is not a failure (status display); assertFresh requires it */40 requireHistory: boolean;41 /** number of most recent bars on which the sources must agree */42 agreeBars: number;43 maxDeviationBps: number;44 /** sources that must have each of those bars */45 minSources: number;46 /** bars of ingestion delay tolerated behind the bar that should be the latest closed one */47 maxLagBars: number;48 priority: readonly SourceId[];49}50export const DEFAULT_QUALITY: QualityOpts = { lookback: 200, requireHistory: true, agreeBars: 3, maxDeviationBps: 50, minSources: 2, maxLagBars: 1, priority: DEFAULT_PRIORITY };5152export interface QualityReport {53 interval: Interval; checkedAt: number;54 /** bar-open time of the bar that should be the latest closed one */55 expectedTs: number; lastTs: number | null; lagBars: number | null; fresh: boolean;56 /** consolidated bars inside the window, and missing slots between the first of them and the last */57 bars: number; gaps: number; gapRanges: GapRange[];58 /** slots missing before the first bar of the window (history shorter than the lookback) */59 missingHistory: number;60 /** largest close deviation over the last `agreeBars` bars; null when no bar had two sources */61 maxDeviationBps: number | null;62 agreement: { ts: number; closes: Partial<Record<SourceId, number>>; bps: number | null }[];63 /** deviation statistics over the whole window */64 window: { compared: number; p50: number | null; p99: number | null; max: number | null };65 /** consolidated bars of the window by providing source, and state of each source */66 provided: Partial<Record<SourceId, number>>;67 sources: Partial<Record<SourceId, { lastTs: number | null; bars: number; gaps: number }>>;68 status: 'ok' | MarketErrorCode; failures: { code: MarketErrorCode; message: string }[];69}7071const iso = (ts: number) => new Date(ts).toISOString();72const quantile = (sorted: number[], q: number) => (sorted.length ? sorted[Math.min(sorted.length - 1, Math.floor(q * sorted.length))]! : null);7374/** The verdict for one (symbol, interval) at `now`. `status` is 'ok' only when `failures` is empty. */75export function evaluate(bySource: BySource, interval: Interval, now: number, o: Partial<QualityOpts> = {}): QualityReport {76 const opts = { ...DEFAULT_QUALITY, ...o }, iv = INTERVAL_MS[interval], expectedTs = expectedLastTs(interval, now);77 const known: BySource = {};78 for (const s of opts.priority) known[s] = (bySource[s] ?? []).filter((c) => c.ts <= expectedTs);79 const cons = consolidate(known, opts.priority);80 const failures: QualityReport['failures'] = [];81 const fail = (code: MarketErrorCode, message: string) => failures.push({ code, message });82 const lastTs = cons.at(-1)?.ts ?? null;83 const report: QualityReport = { interval, checkedAt: now, expectedTs, lastTs, lagBars: null, fresh: false, bars: 0, gaps: 0, gapRanges: [], missingHistory: 0,84 maxDeviationBps: null, agreement: [], window: { compared: 0, p50: null, p99: null, max: null }, provided: {}, sources: {}, status: 'no_data', failures };85 if (lastTs === null) { fail('no_data', 'no closed bar is stored'); return report; }8687 report.lagBars = Math.round((expectedTs - lastTs) / iv);88 report.fresh = report.lagBars <= opts.maxLagBars;89 if (!report.fresh) fail('stale', `latest bar ${iso(lastTs)} is ${report.lagBars} bars behind ${iso(expectedTs)} (allowed: ${opts.maxLagBars})`);9091 const from = lastTs - (opts.lookback - 1) * iv;92 const win = cons.filter((c) => c.ts >= from);93 const g = findGaps(win.map((c) => c.ts), interval);94 // slots missing at the start of the window are a hole if an older bar is known, a short history otherwise95 const head = Math.round((win[0]!.ts - from) / iv);96 if (head > 0 && cons[0]!.ts < from) { g.gaps += head; g.ranges.unshift({ from, to: win[0]!.ts - iv, missing: head }); } else report.missingHistory = head;97 report.bars = win.length; report.gaps = g.gaps; report.gapRanges = g.ranges.slice(-20);98 if (g.gaps > 0) fail('gap', `${g.gaps} missing bars in the last ${opts.lookback} (latest hole ${iso(g.ranges.at(-1)!.from)})`);99 if (opts.requireHistory && report.missingHistory > 0) fail('insufficient_history', `only ${win.length} of ${opts.lookback} bars are available`);100 for (const c of win) report.provided[c.source] = (report.provided[c.source] ?? 0) + 1;101 for (const s of opts.priority) {102 const own = known[s]!.filter((c) => c.ts >= from);103 report.sources[s] = { lastTs: known[s]!.at(-1)?.ts ?? null, bars: own.length, gaps: findGaps(own.map((c) => c.ts), interval).gaps };104 }105106 const devs = deviations(known, from), byTs = new Map(devs.map((d) => [d.ts, d]));107 const sorted = devs.map((d) => d.bps).sort((a, b) => a - b);108 report.window = { compared: devs.length, p50: quantile(sorted, 0.5), p99: quantile(sorted, 0.99), max: sorted.at(-1) ?? null };109 let alone = 0;110 for (let k = opts.agreeBars - 1; k >= 0; k--) {111 const ts = lastTs - k * iv, closes: Partial<Record<SourceId, number>> = {};112 for (const s of opts.priority) { const c = bar(known[s]!, ts); if (c) closes[s] = c.close; }113 // The bar the series actually uses is the one of the highest-ranked source that has it. It must be corroborated: at least114 // `minSources - 1` OTHER sources within the tolerance of it. A further source that disagrees (typically a thin venue whose115 // last trade is old) is recorded but does not block; the deviation that gates is the one to the nearest corroborators.116 const used = opts.priority.find((s) => closes[s] !== undefined);117 const others = used === undefined ? [] : opts.priority.filter((s) => s !== used && closes[s] !== undefined)118 .map((s) => (Math.abs(closes[s]! - closes[used]!) / ((closes[s]! + closes[used]!) / 2)) * 1e4).sort((x, y) => x - y);119 const need = opts.minSources - 1, bps = others.length >= need && need > 0 ? others[need - 1]! : byTs.get(ts)?.bps ?? null;120 report.agreement.push({ ts, closes, bps });121 if (Object.keys(closes).length < opts.minSources) alone++;122 if (bps !== null) report.maxDeviationBps = Math.max(report.maxDeviationBps ?? 0, bps);123 }124 if (alone > 0) fail('insufficient_sources', `${alone} of the last ${opts.agreeBars} bars are confirmed by fewer than ${opts.minSources} sources`);125 if (report.maxDeviationBps !== null && report.maxDeviationBps > opts.maxDeviationBps)126 fail('disagreement', `the bar used is not confirmed: the nearest confirming source differs by ${report.maxDeviationBps.toFixed(1)} bps on the last ${opts.agreeBars} bars (allowed: ${opts.maxDeviationBps})`);127 report.status = failures[0]?.code ?? 'ok';128 return report;129}130131/** Bars are sorted: binary search by bar-open time. */132function bar(bars: Candle[], ts: number): Candle | undefined {133 let lo = 0, hi = bars.length - 1;134 while (lo <= hi) { const mid = (lo + hi) >> 1, t = bars[mid]!.ts; if (t === ts) return bars[mid]; if (t < ts) lo = mid + 1; else hi = mid - 1; }135 return undefined;136}137