spb/sp-wallet

1// In-process consumer API of the market data engine. The trading engine calls assertFresh before every decision.2import { DEFAULT_PRIORITY, DEFAULT_QUALITY, consolidate, evaluate, type QualityOpts, type QualityReport } from './quality.ts';3import { loadBySource, loadLastBySource } from './store.ts';4import { findGaps } from './validate.ts';5import { INTERVAL_MS, MarketDataError, expectedLastTs, isInterval, isSymbol, type Interval, type MarketSymbol, type SeriesBar, type SourceId } from './types.ts';67export interface Series {8  symbol: MarketSymbol; interval: Interval; priority: readonly SourceId[];9  /** closed bars, oldest first; each one names the single source it comes from */10  bars: SeriesBar[];11  /** missing bar-open times between the first and the last bar returned */12  gaps: number;13  /** how many of the bars each source provided */14  provided: Partial<Record<SourceId, number>>;15}1617function check(symbol: string, interval: string): asserts symbol is MarketSymbol {18  if (!isSymbol(symbol) || !isInterval(interval)) throw new MarketDataError('unsupported', `no market data for ${symbol} ${interval}`);19}2021/** The consolidated series (rule in quality.ts: whole bar from the highest-ranked source that has it, never a blend):22 *  the `limit` most recent closed bars whose bar-open time is at or before `until` (default: no bound).23 *  This does NOT say the series is usable for a decision: call assertFresh for that. */24export async function getSeries(symbol: MarketSymbol, interval: Interval, o: { limit: number; until?: number; priority?: readonly SourceId[] }): Promise<Series> {25  check(symbol, interval);26  const limit = Math.max(1, Math.min(Math.floor(o.limit), 50_000)), priority = o.priority ?? DEFAULT_PRIORITY;27  const bars = consolidate(await loadLastBySource(symbol, interval, limit, o.until), priority).slice(-limit);28  const provided: Series['provided'] = {};29  for (const b of bars) provided[b.source] = (provided[b.source] ?? 0) + 1;30  return { symbol, interval, priority, bars, gaps: findGaps(bars.map((b) => b.ts), interval).gaps, provided };31}3233/** The verdict at `now`, computed from the stored bars (never from the cached market_quality row). */34export async function checkQuality(symbol: MarketSymbol, interval: Interval, now: number = Date.now(), o: Partial<QualityOpts> = {}): Promise<QualityReport> {35  check(symbol, interval);36  const opts = { ...DEFAULT_QUALITY, ...o };37  const from = expectedLastTs(interval, now) - (opts.lookback + Math.max(opts.maxLagBars, 0) + 1) * INTERVAL_MS[interval];38  return evaluate(await loadBySource(symbol, interval, from), interval, now, opts);39}4041/** Fails closed. Resolves (with the report) only if, at `now`:42 *  - the latest consolidated bar is the one that should be the latest closed bar, or the one before (`maxLagBars`, default 1);43 *  - the `lookback` bars (default 200) ending at that bar are all there: no hole, and the history is that long;44 *  - each of the last `agreeBars` bars (default 3) exists on at least `minSources` sources (default 2), and the bar the45 *    series uses is confirmed by `minSources - 1` other sources within `maxDeviationBps` (default 50); a further source46 *    that disagrees is recorded, not blocking.47 *  Otherwise throws MarketDataError (code stale | gap | insufficient_history | insufficient_sources | disagreement | no_data |48 *  unsupported) carrying the full report. A database failure also throws: no answer is never a yes. */49export async function assertFresh(symbol: MarketSymbol, interval: Interval, now: number = Date.now(), o: Partial<QualityOpts> = {}): Promise<QualityReport> {50  const report = await checkQuality(symbol, interval, now, o);51  const f = report.failures[0];52  if (f || report.status !== 'ok') throw new MarketDataError(f?.code ?? 'no_data', `${symbol} ${interval}: ${report.failures.map((x) => x.message).join('; ') || 'not usable'}`, report);53  return report;54}55