spb/sp-wallet

1// Market data worker: one loop, every minute. Incremental update first (cheap: only series missing their latest closed2// bar are requested), then the quality records, then a bounded slice of the backfill. The loop is asynchronous, so the3// API is listening long before the history is complete; the backfill resumes where it stopped, pass after pass.4import { loop } from '../worker.ts';5import { backfill, recomputeQuality, updateAll } from './pipeline.ts';6import { activeMarketConfig } from './symbols.ts';7import { defaultSources } from './sources.ts';89const log = (m: string) => console.log(`[market] ${m}`);1011export function startMarketWorker(): void {12  const sources = defaultSources(), priority = sources.map((s) => s.id), done = new Set<string>();13  let complete = false, set = '';14  loop('market', async () => {15    // the symbol set follows the registry and the agents (their symbols first): recomputed at every pass16    const cfg = await activeMarketConfig();17    const key = `${cfg.symbols.join(',')}|${(cfg.first ?? []).join(',')}`;18    if (key !== set) { log(`symbols: ${cfg.symbols.length} (${(cfg.first ?? []).length} used by agents)${cfg.dropped?.length ? `, ${cfg.dropped.length} left out by SPW_MARKET_MAX_SYMBOLS` : ''}`); set = key; complete = false; }19    const u = await updateAll(sources, cfg);20    if (u.requests || u.errors.length) log(`update: ${u.requests} requests, ${u.written} bars written${u.errors.length ? `, ${u.errors.length} errors (${u.errors[0]})` : ''}`);21    await recomputeQuality(cfg, priority);22    if (complete) return;23    const b = await backfill(sources, cfg, { done, maxRequests: 80, deadline: Date.now() + 30_000 });24    complete = b.complete;25    if (b.requests) { log(`backfill: ${b.requests} requests, ${b.written} bars written${b.complete ? ', complete' : ''}${b.errors.length ? `, ${b.errors.length} errors (${b.errors[0]})` : ''}`); await recomputeQuality(cfg, priority); }26  }, 60_000);27}28