Repository Public
1// Pipeline ingest -> validate -> normalise -> store, as incremental updates (forward) and backfill (backward), plus the2// recomputation of the quality record. Requests to one exchange are sequential and paced by its adapter.3import { evaluate, type QualityReport } from './quality.ts';4import { earliestTs, latestTs, loadBySource, saveQuality, upsertCandles } from './store.ts';5import { normalise, validateBars, type RejectReason, type Rejected } from './validate.ts';6import { configuredSymbols } from './symbols.ts';7import { INTERVAL_MS, SourceError, expectedLastTs, type Interval, type MarketSource, type MarketSymbol, type SourceId } from './types.ts';89const DAY = 86_400_000;10export interface MarketConfig {11 symbols: readonly MarketSymbol[]; intervals: readonly Interval[];12 /** backfill depth per interval; null = everything the source has */13 depthMs: Record<Interval, number | null>;14 /** intervals ingested for a subset only (default: every symbol). 5m is kept to the base set: no agent decides on it. */15 only?: Partial<Record<Interval, readonly MarketSymbol[]>>;16 /** symbols served before the others in every pass (those an agent trades): their update and backfill never wait */17 first?: readonly MarketSymbol[];18 /** left out by the size limit (reported by the status route) */19 dropped?: readonly MarketSymbol[];20}21const days = (k: string, d: number | null): number | null => { const v = process.env[k]; return v === undefined || v === '' ? (d === null ? null : d * DAY) : v === 'all' ? null : Number(v) * DAY; };22/** Defaults: two years of 1h, all available 1d, 30 days of 5m. Override with SPW_MARKET_DEPTH_{1H,1D,5M}_DAYS (a number of days, or "all").23 * The symbol set is the registry-derived one (symbols.ts); the worker uses activeMarketConfig(), which puts the agents' symbols first. */24export const marketConfig = (agentSymbols: readonly string[] = []): MarketConfig => {25 const set = configuredSymbols(agentSymbols);26 return { symbols: set.symbols, intervals: ['1h', '1d', '5m'], only: { '5m': set.fiveMinute }, first: set.first, dropped: set.dropped,27 depthMs: { '1h': days('SPW_MARKET_DEPTH_1H_DAYS', 730), '1d': days('SPW_MARKET_DEPTH_1D_DAYS', null), '5m': days('SPW_MARKET_DEPTH_5M_DAYS', 30) } };28};29/** Series of a configuration in service order: every interval of the `first` symbols, then the rest, interval by interval. */30export function seriesOrder(cfg: MarketConfig): [Interval, MarketSymbol][] {31 const first = new Set(cfg.first ?? []);32 const all = cfg.intervals.flatMap((i) => (cfg.only?.[i] ?? cfg.symbols).map((s): [Interval, MarketSymbol] => [i, s]));33 return [...all.filter(([, s]) => first.has(s)), ...all.filter(([, s]) => !first.has(s))];34}3536/** bars refused by validation since the process started */37export const rejectedCount: Partial<Record<RejectReason, number>> = {};38/** 1h/1d/5m history starts no earlier than this (Bitstamp BTC/USD, the oldest pair, starts in August 2011) */39const EPOCH_FLOOR = Date.UTC(2010, 0, 1);40const key = (s: SourceId, symbol: string, interval: string) => `${s}:${symbol}:${interval}`;41const log = (m: string) => console.log(`[market] ${m}`);4243export interface IngestResult { fetched: number; valid: number; written: number; rejected: Rejected[] }44/** One request to one source for [start, end], through the whole pipeline. */45export async function ingest(src: MarketSource, symbol: MarketSymbol, interval: Interval, range: { start: number; end: number }, now: number): Promise<IngestResult> {46 const raw = await src.candles(symbol, interval, range, now);47 const { bars, rejected } = validateBars(raw, interval, now);48 for (const r of rejected) rejectedCount[r.reason] = (rejectedCount[r.reason] ?? 0) + 1;49 if (rejected.length) log(`${key(src.id, symbol, interval)}: ${rejected.length} bars rejected (${[...new Set(rejected.map((r) => r.reason))].join(', ')})`);50 const written = await upsertCandles(src.id, symbol, interval, normalise(bars));51 return { fetched: raw.length, valid: bars.length, written, rejected };52}5354export interface PassSummary { requests: number; written: number; rejected: number; errors: string[] }55const summary = (): PassSummary => ({ requests: 0, written: 0, rejected: 0, errors: [] });56const add = (s: PassSummary, r: IngestResult) => { s.requests++; s.written += r.written; s.rejected += r.rejected.length; };5758/** Runs `each` for every series of every source, sources in parallel. A source that answers "rate limited", or fails three59 * times in a row, is left alone for the rest of the pass so that it cannot hold the others back. */60async function forEachSeries(sources: MarketSource[], cfg: MarketConfig, s: PassSummary, each: (src: MarketSource, symbol: MarketSymbol, interval: Interval) => Promise<boolean | void>): Promise<void> {61 await Promise.all(sources.map(async (src) => {62 let failures = 0;63 for (const [interval, symbol] of seriesOrder(cfg)) {64 if (!src.pairs[symbol]) continue;65 try { if ((await each(src, symbol, interval)) === false) return; failures = 0; }66 catch (e) {67 s.errors.push(`${key(src.id, symbol, interval)}: ${(e as Error).message.slice(0, 160)}`);68 if (e instanceof SourceError && e.kind === 'rate_limited') { src.penalize(30_000); return; }69 if (++failures >= 3) return;70 }71 }72 }));73}7475/** Incremental update: for each series that lacks the latest closed bar, fetch forward from the last stored bar76 * (re-reading the two before it, so a late correction by the exchange is picked up). Nothing is requested when up to date. */77export async function updateAll(sources: MarketSource[], cfg: MarketConfig, clock: () => number = Date.now, maxPages = 4): Promise<PassSummary> {78 const s = summary();79 await forEachSeries(sources, cfg, s, async (src, symbol, interval) => {80 const iv = INTERVAL_MS[interval], expected = expectedLastTs(interval, clock());81 const latest = await latestTs(src.id, symbol, interval);82 if (latest !== null && latest >= expected) return undefined;83 const oldest = expected - (src.maxBars - 1) * iv;84 let start = latest === null ? oldest : latest - 2 * iv;85 if (src.historyBars !== null) start = Math.max(start, oldest);86 for (let page = 0; page < maxPages && start <= expected; page++) {87 const end = Math.min(start + (src.maxBars - 1) * iv, expected);88 add(s, await ingest(src, symbol, interval, { start, end }, clock()));89 start = end + iv;90 }91 return undefined;92 });93 return s;94}9596export interface BackfillBudget { /** series already complete (kept by the caller between passes) */ done: Set<string>; /** per source and per pass */ maxRequests: number; deadline?: number }97/** Backfill: for each series, page backward from the oldest stored bar down to the configured depth, or until the source98 * has nothing older (an empty page). Bounded by `maxRequests` per source and by `deadline`; call again to continue. */99export async function backfill(sources: MarketSource[], cfg: MarketConfig, budget: BackfillBudget, clock: () => number = Date.now): Promise<PassSummary & { complete: boolean }> {100 const s = summary();101 let pending = false;102 const left = new Map(sources.map((src) => [src.id, budget.maxRequests]));103 await forEachSeries(sources, cfg, s, async (src, symbol, interval) => {104 const k = key(src.id, symbol, interval);105 if (budget.done.has(k)) return undefined;106 const iv = INTERVAL_MS[interval], now = clock(), expected = expectedLastTs(interval, now), depth = cfg.depthMs[interval];107 let target = depth === null ? EPOCH_FLOOR : Math.max(EPOCH_FLOOR, Math.ceil((now - depth) / iv) * iv);108 if (src.historyBars !== null) target = Math.max(target, expected - (src.historyBars - 1) * iv);109 const earliest = await earliestTs(src.id, symbol, interval);110 let end = earliest === null ? expected : earliest - iv;111 while (end >= target) {112 if (left.get(src.id)! <= 0 || (budget.deadline !== undefined && clock() > budget.deadline)) { pending = true; return false; }113 const start = Math.max(target, end - (src.maxBars - 1) * iv);114 left.set(src.id, left.get(src.id)! - 1);115 const r = await ingest(src, symbol, interval, { start, end }, clock());116 add(s, r);117 if (r.valid === 0) break; // nothing older on this source118 end = start - iv;119 }120 budget.done.add(k);121 return undefined;122 });123 if (s.errors.length) pending = true;124 return { ...s, complete: !pending };125}126127export const QUALITY_WINDOW = 720;128/** Recomputes and stores the quality record of every (symbol, interval): the last 720 bars, history shorter than that tolerated. */129export async function recomputeQuality(cfg: MarketConfig, priority: readonly SourceId[], now: number = Date.now()): Promise<Record<string, QualityReport>> {130 const out: Record<string, QualityReport> = {};131 for (const [interval, symbol] of seriesOrder(cfg)) {132 const from = expectedLastTs(interval, now) - (QUALITY_WINDOW + 2) * INTERVAL_MS[interval];133 const report = evaluate(await loadBySource(symbol, interval, from), interval, now, { lookback: QUALITY_WINDOW, requireHistory: false, priority });134 await saveQuality(symbol, report);135 out[`${symbol}:${interval}`] = report;136 }137 return out;138}139