Repository Public
1// Storage of candles and of the quality record. Every write is an idempotent upsert.2import { sql } from '../db.ts';3import type { QualityReport } from './quality.ts';4import { SOURCE_IDS, type BySource, type Candle, type Interval, type MarketSymbol, type SourceId } from './types.ts';56/** Inserts or corrects bars; returns how many rows were actually written (0 when the same bars are stored again). */7export async function upsertCandles(source: SourceId, symbol: MarketSymbol, interval: Interval, bars: Candle[]): Promise<number> {8 let written = 0;9 for (let i = 0; i < bars.length; i += 1000) {10 const rows = bars.slice(i, i + 1000).map((c) => ({ source, symbol, interval, ts: new Date(c.ts), open: c.open, high: c.high, low: c.low, close: c.close, volume: c.volume }));11 const r = await sql`insert into candles ${sql(rows, 'source', 'symbol', 'interval', 'ts', 'open', 'high', 'low', 'close', 'volume')}12 on conflict (source, symbol, interval, ts) do update set open = excluded.open, high = excluded.high, low = excluded.low, close = excluded.close, volume = excluded.volume13 where (candles.open, candles.high, candles.low, candles.close, candles.volume) is distinct from (excluded.open, excluded.high, excluded.low, excluded.close, excluded.volume)14 returning 1`;15 written += r.length;16 }17 return written;18}1920/** upper bound PostgreSQL accepts for a timestamp parameter */21const FOREVER = Date.UTC(9999, 0, 1);22const ms = (d: Date | null | undefined) => (d ? d.getTime() : null);23export async function latestTs(source: SourceId, symbol: MarketSymbol, interval: Interval): Promise<number | null> {24 const [r] = await sql<{ ts: Date }[]>`select ts from candles where source = ${source} and symbol = ${symbol} and interval = ${interval} order by ts desc limit 1`;25 return ms(r?.ts);26}27export async function earliestTs(source: SourceId, symbol: MarketSymbol, interval: Interval): Promise<number | null> {28 const [r] = await sql<{ ts: Date }[]>`select ts from candles where source = ${source} and symbol = ${symbol} and interval = ${interval} order by ts asc limit 1`;29 return ms(r?.ts);30}3132type Row = { source: SourceId; ts: Date; open: number; high: number; low: number; close: number; volume: number };33function group(rows: Row[]): BySource {34 const out: BySource = {};35 for (const r of rows) if ((SOURCE_IDS as readonly string[]).includes(r.source)) (out[r.source] ??= []).push({ ts: r.ts.getTime(), open: r.open, high: r.high, low: r.low, close: r.close, volume: r.volume });36 return out;37}38/** Bars of every source with bar-open time in [from, to], oldest first. */39export async function loadBySource(symbol: MarketSymbol, interval: Interval, from: number, to: number = FOREVER): Promise<BySource> {40 return group(await sql<Row[]>`select source, ts, open, high, low, close, volume from candles41 where symbol = ${symbol} and interval = ${interval} and ts >= ${new Date(from)} and ts <= ${new Date(Math.min(to, FOREVER))} order by ts asc`);42}43/** Bars of every source for the `limit` most recent distinct bar-open times at or before `until`. */44export async function loadLastBySource(symbol: MarketSymbol, interval: Interval, limit: number, until: number = FOREVER): Promise<BySource> {45 const to = new Date(Math.min(until, FOREVER));46 return group(await sql<Row[]>`select source, ts, open, high, low, close, volume from candles47 where symbol = ${symbol} and interval = ${interval} and ts <= ${to} and ts >= (select min(ts) from (48 select distinct ts from candles where symbol = ${symbol} and interval = ${interval} and ts <= ${to} order by ts desc limit ${limit}) t)49 order by ts asc`);50}5152export async function saveQuality(symbol: MarketSymbol, r: QualityReport): Promise<void> {53 await sql`insert into market_quality (symbol, interval, checked_at, last_ts, fresh, gaps, max_deviation_bps, status, detail)54 values (${symbol}, ${r.interval}, ${new Date(r.checkedAt)}, ${r.lastTs === null ? null : new Date(r.lastTs)}, ${r.fresh}, ${r.gaps}, ${r.maxDeviationBps}, ${r.status}, ${sql.json(r as never)})55 on conflict (symbol, interval) do update set checked_at = excluded.checked_at, last_ts = excluded.last_ts, fresh = excluded.fresh, gaps = excluded.gaps,56 max_deviation_bps = excluded.max_deviation_bps, status = excluded.status, detail = excluded.detail`;57}58export interface QualityRow { symbol: string; interval: string; checked_at: Date; last_ts: Date | null; fresh: boolean; gaps: number; max_deviation_bps: number | null; status: string; detail: QualityReport }59export const listQuality = () => sql<QualityRow[]>`select * from market_quality order by symbol, interval`;6061export interface CoverageRow { source: string; symbol: string; interval: string; bars: number; first_ts: Date; last_ts: Date }62export const coverage = () => sql<CoverageRow[]>`select source, symbol, interval, count(*)::int as bars, min(ts) as first_ts, max(ts) as last_ts63 from candles group by source, symbol, interval order by symbol, interval, source`;64