spb/market-atlas

1import type { Bar, BarResolution, CanonicalQuote } from "@market-atlas/market-model";2import { CANONICAL_VERSIONS, RESOLUTION_MS, floorTo } from "@market-atlas/market-model";3import { pool } from "../db/pool.js";4import { logger } from "../logger.js";5import { telemetry } from "./telemetry.js";6import { barsMutex } from "./mutex.js";78/**9 * Live 1-minute bar aggregation from canonical prices; higher resolutions are rolled up in SQL10 * by the scheduler (reproducible from the 1m series).11 */12export class BarAggregator {13  private open = new Map<string, Bar>(); // instrumentId -> current 1m bar14  private pending: Bar[] = [];15  private timer: NodeJS.Timeout | null = null;1617  onQuote(q: CanonicalQuote): void {18    if (q.price == null || q.realtimeStatus === "STALE" || q.realtimeStatus === "END_OF_DAY") return;19    const ts = floorTo(q.sourceTimestamp ?? q.updatedAt, RESOLUTION_MS["1m"]);20    const cur = this.open.get(q.instrumentId);21    if (cur && cur.ts !== ts) {22      this.pending.push(cur);23      this.open.delete(q.instrumentId);24    }25    if (!this.open.has(q.instrumentId)) {26      this.open.set(q.instrumentId, {27        instrumentId: q.instrumentId,28        resolution: "1m",29        ts,30        open: q.price,31        high: q.price,32        low: q.price,33        close: q.price,34        volume: null,35        sourceCount: q.sourceCount,36        producer: "consensus",37        version: CANONICAL_VERSIONS.consensus,38      });39    } else {40      const b = this.open.get(q.instrumentId)!;41      b.high = Math.max(b.high, q.price);42      b.low = Math.min(b.low, q.price);43      b.close = q.price;44      b.sourceCount = Math.max(b.sourceCount, q.sourceCount);45    }46    if (!this.timer) this.timer = setTimeout(() => void this.flush(), 5000);47  }4849  /** Flush completed bars plus a snapshot of open bars (upserted again when they close). */50  async flush(): Promise<void> {51    this.timer = null;52    const batch = [...this.pending, ...this.open.values()];53    this.pending = [];54    if (!batch.length) return;55    await upsertBars(batch);56    if (this.open.size) this.timer = setTimeout(() => void this.flush(), 5000);57  }58}5960export async function upsertBars(input: Bar[]): Promise<void> {61  if (!input.length) return;62  // One row per key (a late tick can re-open an already flushed minute) and a deterministic lock order.63  const byKey = new Map<string, Bar>();64  for (const b of input) byKey.set(`${b.instrumentId}|${b.resolution}|${b.ts}`, b);65  const bars = [...byKey.values()].sort((a, b) => a.instrumentId.localeCompare(b.instrumentId) || a.resolution.localeCompare(b.resolution) || a.ts - b.ts);66  await barsMutex.run(() => upsertSorted(bars));67}6869async function upsertSorted(bars: Bar[]): Promise<void> {70  const cols = 11;71  for (let i = 0; i < bars.length; i += 1000) {72    const chunk = bars.slice(i, i + 1000);73    const tuples: string[] = [];74    const values: unknown[] = [];75    chunk.forEach((b, j) => {76      const base = j * cols;77      tuples.push(`(${Array.from({ length: cols }, (_, k) => `$${base + k + 1}`).join(",")})`);78      values.push(b.instrumentId, b.resolution, new Date(b.ts).toISOString(), b.open, b.high, b.low, b.close, b.volume, b.sourceCount, b.producer, b.version);79    });80    try {81      await pool.query(82        `insert into bars (instrument_id, resolution, ts, open, high, low, close, volume, source_count, producer, version) values ${tuples.join(",")}83         on conflict (instrument_id, resolution, ts) do update set open = excluded.open,84           high = excluded.high, low = excluded.low, close = excluded.close, volume = coalesce(excluded.volume, bars.volume), source_count = excluded.source_count, producer = excluded.producer, version = excluded.version85         where bars.producer = 'consensus' or excluded.producer <> 'consensus'`,86        values,87      );88      telemetry.inc("bars_written_total", chunk.length);89    } catch (err) {90      logger.error({ err }, "bar upsert failed");91    }92  }93}9495/** Roll 1m bars into 5m/15m/1h and 1m→1d (UTC days) for the last `hours` hours. */96export async function rollupBars(hours = 3): Promise<void> {97  await barsMutex.run(() => rollupInner(hours));98}99100async function rollupInner(hours: number): Promise<void> {101  const targets: Array<[BarResolution, string]> = [102    ["5m", "5 minutes"],103    ["15m", "15 minutes"],104    ["1h", "1 hour"],105    ["1d", "1 day"],106  ];107  for (const [res, interval] of targets) {108    await pool.query(109      `insert into bars (instrument_id, resolution, ts, open, high, low, close, volume, source_count, producer, version)110       select instrument_id, $1, bucket, (array_agg(open order by ts))[1], max(high), min(low), (array_agg(close order by ts desc))[1], sum(volume), max(source_count), 'consensus', $3111       from (select *, date_bin($2::interval, ts, timestamptz '2000-01-01') as bucket from bars where resolution = '1m' and producer = 'consensus' and ts > now() - ($4 || ' hours')::interval) b112       group by instrument_id, bucket113       order by instrument_id, bucket114       on conflict (instrument_id, resolution, ts) do update set open = excluded.open, high = excluded.high, low = excluded.low, close = excluded.close, volume = excluded.volume, source_count = excluded.source_count115       where bars.producer = 'consensus'`,116      [res, interval, CANONICAL_VERSIONS.consensus, String(hours)],117    );118  }119}120121export const barAggregator = new BarAggregator();122