Repository Public
1// Market data for perp agents (V6): the VENUE's own closed bars and hourly funding settlements, stored in PostgreSQL, and the2// gate that decides whether a signal may be computed on them right now.3//4// - Source: the derivatives data service's venue registry (public read endpoints only — the read-only view has no trading5// object). Nothing here signs, calls the signer or reaches a trading endpoint.6// - Storage: `perp_candles` / `perp_funding` (exact numerics). A series is fetched once (the lookback) and then extended by7// the bars that closed since: an hourly signal costs one small candle request and one funding request per instrument and8// hour, and nothing at all between two closes.9// - Rate budget: every request is accounted with the venue's documented weights against OUR share of the venue's limit10// (`weightPerMinute`, default 500 of Hyperliquid's 1200 per minute; the rest is for the terminal and the paper worker). When11// the share is spent the store refuses BEFORE sending (the engine holds, a backtest waits).12// - Gate (`gate`): fail closed. Latest closed bar present; no gap in the lookback; sane OHLC; a funding settlement for the13// hours of the window (stated tolerance) and the one of the decision bar's open; nothing dated after the decision bar; a14// closed bar the venue later serves with other values holds that pass; live mark within a band of the venue's spot index15// (its oracle), or of the spot market data when the venue publishes no index; last close within a band of the live mark.16import { D, hlInfoExtraWeight, hlInfoWeight, VenueError, type DecStr, type Instrument, type Ticker, type VenueId } from '@spw/derivatives';17import { sql } from '../db.ts';18import type { DerivData } from '../derivatives/data.ts';19import type { ReadAdapter } from '../derivatives/venues.ts';2021export const PERP_INTERVALS = ['1h', '4h'] as const;22export type PerpInterval = (typeof PERP_INTERVALS)[number];23export const PERP_STEP_MS: Record<PerpInterval, number> = { '1h': 3_600_000, '4h': 14_400_000 };24const HOUR = 3_600_000;25/** the venue keeps only its most recent 5000 candles of an interval */26export const VENUE_MAX_BARS = 5000;27const FUNDING_PAGE = 500;28/** a settlement instant further than this from the hour cannot be attributed to a settlement (the lab's rule) */29const FUNDING_TOLERANCE_MS = 60_000;3031/** bands and tolerances of the gate — every one of them is a refusal to trade, never an adjustment */32export const GATE = {33 /** live mark vs the venue's spot index (oracle), basis points */34 markIndexBps: 100,35 /** live mark vs the last hourly close of the spot market data — used only when the venue publishes no index */36 markSpotBps: 300,37 /** last closed bar's close vs the live mark: the stored series is the live market's */38 lastCloseBps: 1000,39 /** the ticker must have been read at most this long ago */40 tickerMaxAgeMs: 15_000,41 /** funding settlements missing in the lookback: at most max(1, 0.5 % of its hours), never two hours in a row, never the decision bar's */42 fundingMissingFraction: 0.005, fundingMaxRun: 1,43} as const;4445export interface PerpBar { ts: number; open: number; high: number; low: number; close: number; volume: number }46export interface FundingRow { ts: number; rate: number; premium: number | null }47export interface PerpSeries { bars: PerpBar[]; funding: FundingRow[]; /** exact close of the last bar */ lastClose: DecStr | null }4849export class PerpDataError extends Error {50 readonly code: string; readonly retryAfterMs: number | null;51 constructor(code: string, message: string, retryAfterMs: number | null = null) { super(message); this.name = 'PerpDataError'; this.code = code; this.retryAfterMs = retryAfterMs; }52}5354/** sliding one-minute window of request weight spent by this store */55export class WeightBudget {56 readonly limit: number;57 private readonly now: () => number;58 private spent: { at: number; w: number }[] = [];59 constructor(limit: number, now: () => number) { this.limit = limit; this.now = now; }60 used(): number { const cut = this.now() - 60_000; this.spent = this.spent.filter((s) => s.at > cut); return this.spent.reduce((n, s) => n + s.w, 0); }61 /** ms until `w` fits (0 = now) */62 waitMs(w: number): number {63 let used = this.used();64 if (used + w <= this.limit) return 0;65 for (const s of this.spent) { used -= s.w; if (used + w <= this.limit) return Math.max(1, s.at + 60_000 - this.now()); }66 return 60_000;67 }68 take(w: number, what: string): void {69 const wait = this.waitMs(w);70 if (wait > 0) throw new PerpDataError('rate_budget', `perp data budget spent (${this.used()}/${this.limit} weight per minute) before ${what}`, wait);71 this.spent.push({ at: this.now(), w });72 }73 charge(w: number): void { if (w > 0) this.spent.push({ at: this.now(), w }); }74}7576const bps = (a: DecStr, ref: DecStr): number => {77 const r = D.from(ref);78 if (!D.isPos(r)) return Infinity;79 return D.toNumber(D.div(D.mul(D.abs(D.sub(D.from(a), r)), D.from(10000n)), r, 4, 'up')); // a comparison figure, not an amount80};8182/** The checks of the gate that need no network. Throws PerpDataError; returns the decision bar (open time of the last closed bar). */83export function checkSeries(symbol: string, s: PerpSeries, o: { interval: PerpInterval; lookback: number; now: number }): number {84 const step = PERP_STEP_MS[o.interval], expected = Math.floor(o.now / step) * step - step;85 const bars = s.bars.slice(-o.lookback);86 if (bars.length < o.lookback) throw new PerpDataError('history', `${symbol}: ${bars.length} closed bars stored, the strategy needs ${o.lookback}`);87 const last = bars[bars.length - 1]!;88 if (last.ts > expected) throw new PerpDataError('unclosed', `${symbol}: a bar that is not closed yet is in the series`);89 if (last.ts < expected) throw new PerpDataError('stale', `${symbol}: the latest closed bar (${new Date(expected).toISOString()}) is missing; last stored ${new Date(last.ts).toISOString()}`);90 for (let i = 0; i < bars.length; i++) {91 const b = bars[i]!;92 if (i > 0 && b.ts - bars[i - 1]!.ts !== step) throw new PerpDataError('gap', `${symbol}: gap in the lookback after ${new Date(bars[i - 1]!.ts).toISOString()}`);93 if (![b.open, b.high, b.low, b.close, b.volume].every(Number.isFinite) || !(b.low > 0) || b.high < Math.max(b.open, b.close) || b.low > Math.min(b.open, b.close) || b.volume < 0) {94 throw new PerpDataError('bad_bar', `${symbol}: impossible OHLCV values at ${new Date(b.ts).toISOString()}`);95 }96 }97 // funding: one settlement per hour of the window (first bar's open … decision bar's open), aligned on the hour98 const first = bars[0]!.ts, have = new Set<number>();99 for (const f of s.funding) {100 if (f.ts > last.ts) throw new PerpDataError('future_funding', `${symbol}: a funding settlement dated after the decision bar is in the input`);101 if (f.ts % HOUR !== 0 || !Number.isFinite(f.rate)) throw new PerpDataError('bad_funding', `${symbol}: funding row off the hour grid or not finite`);102 have.add(f.ts);103 }104 if (!have.has(last.ts)) throw new PerpDataError('funding_stale', `${symbol}: the funding settlement of ${new Date(last.ts).toISOString()} is not published yet`);105 let missing = 0, run = 0, longest = 0;106 for (let t = first; t <= last.ts; t += HOUR) { if (have.has(t)) run = 0; else { missing += 1; run += 1; longest = Math.max(longest, run); } }107 const hours = (last.ts - first) / HOUR + 1, allowed = Math.max(1, Math.floor(hours * GATE.fundingMissingFraction));108 if (missing > allowed || longest > GATE.fundingMaxRun) throw new PerpDataError('funding_gap', `${symbol}: ${missing} funding settlements missing in the lookback (longest run ${longest}; allowed ${allowed}, run ${GATE.fundingMaxRun})`);109 return last.ts;110}111112/** The live cross-check of the gate. `spot` is the last close of the spot market data, when the caller has one. Throws PerpDataError. */113export function checkMark(symbol: string, ticker: Ticker, lastClose: DecStr | null, spot: DecStr | null, now: number): void {114 if (!Number.isFinite(ticker.ts) || Math.abs(now - ticker.ts) > GATE.tickerMaxAgeMs) throw new PerpDataError('stale_ticker', `${symbol}: the venue ticker is ${Math.round((now - ticker.ts) / 1000)} s old`);115 if (!D.isValid(ticker.mark) || !D.isPos(D.from(ticker.mark))) throw new PerpDataError('bad_mark', `${symbol}: mark price is not positive`);116 if (ticker.index !== null) {117 const d = bps(ticker.mark, ticker.index);118 if (!(d <= GATE.markIndexBps)) throw new PerpDataError('mark_index', `${symbol}: mark ${ticker.mark} is ${d.toFixed(1)} bps away from the venue's spot index ${ticker.index} (band ${GATE.markIndexBps} bps)`);119 } else {120 if (spot === null) throw new PerpDataError('no_reference', `${symbol}: the venue publishes no index and no spot market data is available to cross-check the mark`);121 const d = bps(ticker.mark, spot);122 if (!(d <= GATE.markSpotBps)) throw new PerpDataError('mark_spot', `${symbol}: mark ${ticker.mark} is ${d.toFixed(1)} bps away from the spot market ${spot} (band ${GATE.markSpotBps} bps)`);123 }124 if (lastClose !== null) {125 const d = bps(lastClose, ticker.mark);126 if (!(d <= GATE.lastCloseBps)) throw new PerpDataError('last_close', `${symbol}: the last closed bar (${lastClose}) is ${d.toFixed(0)} bps away from the live mark ${ticker.mark} (band ${GATE.lastCloseBps} bps)`);127 }128}129130export interface PerpDataOptions {131 now?: () => number;132 /** our share of the venue's weight limit per minute */133 weightPerMinute?: number;134 /** last close of the spot market data for a base symbol (fallback reference when the venue publishes no index); null = none */135 spot?: (symbol: string) => Promise<DecStr | null>;136}137138const same = (a: string, b: string): boolean => D.eq(D.from(a), D.from(b));139const key = (inst: Instrument, what: string): string => `${inst.id}:${what}`;140141export class PerpDataStore {142 readonly data: DerivData;143 readonly now: () => number;144 readonly budget: WeightBudget;145 private readonly spot: NonNullable<PerpDataOptions['spot']>;146 /** a series whose window could not be completed is not asked again before this time (a venue gap, a settlement not published yet) */147 private readonly retryAt = new Map<string, number>();148 private metaChargedAt = 0;149 private readonly locks = new Map<string, Promise<void>>();150 /** requests sent to the venue since start (tests, diagnostics) */151 readonly stats = { candles: 0, funding: 0 };152153 constructor(data: DerivData, o: PerpDataOptions = {}) {154 this.data = data; this.now = o.now ?? data.now;155 this.budget = new WeightBudget(o.weightPerMinute ?? Number(process.env.SPW_PERP_DATA_WEIGHT ?? 500), this.now);156 this.spot = o.spot ?? (async () => null);157 }158159 private async call<T>(venue: VenueId, type: 'candleSnapshot' | 'fundingHistory', expectedItems: number, fn: (a: ReadAdapter) => Promise<T>): Promise<T> {160 // the adapter checks the symbol against the venue's asset table, which it re-reads at most once a minute (weight 20)161 const meta = this.now() - this.metaChargedAt >= 60_000 ? hlInfoWeight('meta') : 0;162 this.budget.take(hlInfoWeight(type) + hlInfoExtraWeight(type, expectedItems) + meta, `${type}`);163 if (meta) this.metaChargedAt = this.now();164 try { return await this.data.registry.call(venue, fn); }165 catch (e) {166 if (e instanceof VenueError && e.code === 'rate_limited') throw new PerpDataError('rate_budget', `venue rate limit: ${e.message.slice(0, 120)}`, 15_000);167 throw new PerpDataError('venue', `${venue}: ${(e as Error).message.slice(0, 160)}`);168 }169 }170171 /**172 * Bring the stored series of one instrument up to the last closed bar, for a window of `bars` bars. Sends nothing when the173 * window is already complete. Throws PerpDataError (`rate_budget` carries `retryAfterMs`; progress made is kept).174 */175 async sync(inst: Instrument, interval: PerpInterval, bars: number): Promise<void> {176 // one sync of a series at a time (an engine pass and a backtest may want the same instrument): the second waits, then finds it done177 const k = key(inst, interval), prev = this.locks.get(k) ?? Promise.resolve();178 const run = prev.catch(() => {}).then(async () => {179 const step = PERP_STEP_MS[interval], now = this.now(), last = Math.floor(now / step) * step - step;180 const n = Math.min(Math.max(1, bars), VENUE_MAX_BARS - 1), from = last - (n - 1) * step;181 await this.syncCandles(inst, interval, from, last, n);182 await this.syncFunding(inst, from, last);183 });184 this.locks.set(k, run);185 try { await run; } finally { if (this.locks.get(k) === run) this.locks.delete(k); }186 }187188 private async syncCandles(inst: Instrument, interval: PerpInterval, from: number, last: number, n: number): Promise<void> {189 const step = PERP_STEP_MS[interval], now = this.now(), k = key(inst, `c:${interval}`);190 const [c] = await sql<{ n: number; hi: Date | null }[]>`select count(*)::int as n, max(ts) as hi from perp_candles where venue = ${inst.venue} and symbol = ${inst.symbol} and interval = ${interval} and ts >= ${new Date(from)} and ts <= ${new Date(last)}`;191 if (c!.n === n) return; // complete: nothing is asked192 if ((this.retryAt.get(k) ?? 0) > now) return; // asked a moment ago and still incomplete: the gate will say why193 // contiguous from the start of the window → only the tail is missing (one bar of overlap re-checks what is stored)194 const hi = c!.hi?.getTime() ?? null, contiguous = hi !== null && c!.n === (hi - from) / step + 1;195 const start = contiguous ? hi - step : from, want = Math.round((now - start) / step) + 1;196 const got = await this.call<{ ts: number; open: DecStr; high: DecStr; low: DecStr; close: DecStr; volume: DecStr }[]>(inst.venue, 'candleSnapshot', want,197 (a) => a.candles(inst.symbol, interval, { start, end: now, limit: Math.min(VENUE_MAX_BARS, want + 1) }));198 this.stats.candles += 1;199 const closed = got.filter((b) => b.ts % step === 0 && b.ts + step <= now && b.ts >= start); // the bar in progress is never stored200 let revised = 0;201 if (closed.length) {202 const old = await sql<{ ts: Date; open: string; high: string; low: string; close: string; volume: string }[]>`select ts, open::text, high::text, low::text, close::text, volume::text from perp_candles203 where venue = ${inst.venue} and symbol = ${inst.symbol} and interval = ${interval} and ts >= ${new Date(closed[0]!.ts)} and ts <= ${new Date(closed[closed.length - 1]!.ts)}`;204 const byTs = new Map(old.map((r) => [r.ts.getTime(), r]));205 for (const b of closed) { const o = byTs.get(b.ts); if (o && !(same(o.open, b.open) && same(o.high, b.high) && same(o.low, b.low) && same(o.close, b.close) && same(o.volume, b.volume))) revised += 1; }206 for (let i = 0; i < closed.length; i += 1000) {207 const rows = closed.slice(i, i + 1000).map((b) => ({ venue: inst.venue, symbol: inst.symbol, interval, ts: new Date(b.ts), open: b.open, high: b.high, low: b.low, close: b.close, volume: b.volume }));208 await sql`insert into perp_candles ${sql(rows)} on conflict (venue, symbol, interval, ts) do update set open = excluded.open, high = excluded.high, low = excluded.low, close = excluded.close, volume = excluded.volume`;209 }210 }211 const [after] = await sql<{ n: number }[]>`select count(*)::int as n from perp_candles where venue = ${inst.venue} and symbol = ${inst.symbol} and interval = ${interval} and ts >= ${new Date(from)} and ts <= ${new Date(last)}`;212 if (after!.n !== n) this.retryAt.set(k, now + (contiguous ? 30_000 : 10 * 60_000)); else this.retryAt.delete(k);213 // the venue's value is kept, but this pass must not trade on a series that just changed under it214 if (revised) throw new PerpDataError('revised', `${inst.symbol}: the venue now serves ${revised} closed ${interval} bar(s) with other values than the stored ones; stored values replaced, holding this pass`);215 }216217 private async syncFunding(inst: Instrument, from: number, last: number): Promise<void> {218 const now = this.now(), k = key(inst, 'f'), hours = (last - from) / HOUR + 1;219 const stored = async (): Promise<{ n: number; hi: number | null; contiguous: boolean }> => {220 const [c] = await sql<{ n: number; hi: Date | null }[]>`select count(*)::int as n, max(ts) as hi from perp_funding where venue = ${inst.venue} and symbol = ${inst.symbol} and ts >= ${new Date(from)} and ts <= ${new Date(last)}`;221 const hi = c!.hi?.getTime() ?? null;222 return { n: c!.n, hi, contiguous: hi !== null && c!.n === (hi - from) / HOUR + 1 };223 };224 const before = await stored();225 if (before.n === hours) { this.retryAt.delete(k); return; } // complete: nothing is asked226 if ((this.retryAt.get(k) ?? 0) > now) return;227 // contiguous from the start of the window → only the tail is asked; otherwise the whole window, page after page (500 rows each)228 let cursor = before.contiguous ? before.hi! + FUNDING_TOLERANCE_MS + 1 : from - FUNDING_TOLERANCE_MS;229 for (let page = 0; page < 40; page++) {230 const expect = Math.min(FUNDING_PAGE, Math.ceil((now - cursor) / HOUR) + 1);231 const start = cursor;232 const got = await this.call<{ ts: number; rate: DecStr; premium: DecStr | null }[]>(inst.venue, 'fundingHistory', expect, (a) => a.fundingHistory(inst.symbol, { start }));233 this.stats.funding += 1;234 const rows = new Map<number, { venue: string; symbol: string; ts: Date; time_ms: number; rate: string; premium: string | null }>();235 for (const f of got) {236 const hour = Math.round(f.ts / HOUR) * HOUR;237 if (Math.abs(f.ts - hour) > FUNDING_TOLERANCE_MS || hour > now) continue; // off the hour grid: attributed to no settlement238 rows.set(hour, { venue: inst.venue, symbol: inst.symbol, ts: new Date(hour), time_ms: f.ts, rate: f.rate, premium: f.premium });239 }240 if (rows.size) await sql`insert into perp_funding ${sql([...rows.values()])} on conflict (venue, symbol, ts) do update set time_ms = excluded.time_ms, rate = excluded.rate, premium = excluded.premium`;241 const newest = got.length ? Math.max(...got.map((f) => f.ts)) : 0;242 if (got.length < FUNDING_PAGE || newest >= last || newest < cursor) break;243 cursor = newest + 1; // a full page that ends before the window does244 }245 const after = await stored();246 if (after.n === hours) { this.retryAt.delete(k); return; }247 // still incomplete: only the tail is missing (a settlement not published yet: asked again soon), or the venue never248 // recorded some settlement (kept as a gap — the gate counts it — and not re-read before ten minutes)249 this.retryAt.set(k, now + (after.contiguous ? 30_000 : 10 * 60_000));250 }251252 /** stored closed bars (oldest first, at most `limit`) and the settlements of their hours — nothing after the last bar's open */253 async series(inst: Instrument, interval: PerpInterval, limit: number): Promise<PerpSeries> {254 const rows = await sql<{ ts: Date; open: string; high: string; low: string; close: string; volume: string }[]>`select ts, open::text, high::text, low::text, close::text, volume::text from perp_candles255 where venue = ${inst.venue} and symbol = ${inst.symbol} and interval = ${interval} order by ts desc limit ${limit}`;256 rows.reverse();257 const bars = rows.map((r) => ({ ts: r.ts.getTime(), open: Number(r.open), high: Number(r.high), low: Number(r.low), close: Number(r.close), volume: Number(r.volume) }));258 if (!bars.length) return { bars, funding: [], lastClose: null };259 const f = await sql<{ ts: Date; rate: string; premium: string | null }[]>`select ts, rate::text, premium::text from perp_funding where venue = ${inst.venue} and symbol = ${inst.symbol}260 and ts >= ${new Date(bars[0]!.ts)} and ts <= ${new Date(bars[bars.length - 1]!.ts)} order by ts`;261 return { bars, funding: f.map((r) => ({ ts: r.ts.getTime(), rate: Number(r.rate), premium: r.premium === null ? null : Number(r.premium) })), lastClose: D.str(D.from(rows[rows.length - 1]!.close)) };262 }263264 /** exact stored rows, for the laboratory's backtest files */265 async exact(inst: Instrument, interval: PerpInterval, limit: number): Promise<{ candles: [number, string, string, string, string, string][]; funding: [number, string, string | null][] }> {266 const c = await sql<{ ts: Date; open: string; high: string; low: string; close: string; volume: string }[]>`select ts, open::text, high::text, low::text, close::text, volume::text from perp_candles267 where venue = ${inst.venue} and symbol = ${inst.symbol} and interval = ${interval} order by ts desc limit ${limit}`;268 c.reverse();269 if (!c.length) return { candles: [], funding: [] };270 const f = await sql<{ time_ms: string; rate: string; premium: string | null }[]>`select time_ms::text, rate::text, premium::text from perp_funding where venue = ${inst.venue} and symbol = ${inst.symbol} and ts >= ${c[0]!.ts} order by ts`;271 return { candles: c.map((r) => [r.ts.getTime(), r.open, r.high, r.low, r.close, r.volume]), funding: f.map((r) => [Number(r.time_ms), r.rate, r.premium]) };272 }273274 /**275 * The gate of one instrument: sync, then every check. Returns the series the signal may use, or throws PerpDataError — the276 * engine then holds. `lookback` is the strategy's own (META['lookback']).277 */278 async gate(inst: Instrument, interval: PerpInterval, lookback: number): Promise<{ bar: number; series: PerpSeries; ticker: Ticker }> {279 if (inst.status !== 'active') throw new PerpDataError('instrument', `${inst.id} is ${inst.status} on the venue`);280 await this.sync(inst, interval, lookback);281 const series = await this.series(inst, interval, lookback), now = this.now();282 const bar = checkSeries(inst.symbol, series, { interval, lookback, now });283 let ticker: Ticker;284 try { ticker = await this.data.ticker(inst, { within: 5_000 }); } catch (e) { throw new PerpDataError('venue', `${inst.symbol}: no live ticker (${(e as Error).message.slice(0, 120)})`); }285 const spot = ticker.index === null ? await this.spot(inst.base).catch(() => null) : null;286 checkMark(inst.symbol, ticker, series.lastClose, spot, now);287 return { bar, series, ticker };288 }289290 /** fill the store for a backtest (as much history as the venue serves), waiting for the budget instead of failing */291 async backfill(insts: Instrument[], interval: PerpInterval, bars: number, o: { deadlineMs: number; sleep?: (ms: number) => Promise<void> }): Promise<void> {292 const sleep = o.sleep ?? ((ms: number) => new Promise<void>((r) => setTimeout(r, ms)));293 const until = Date.now() + o.deadlineMs;294 for (const inst of insts) {295 for (;;) {296 try { await this.sync(inst, interval, bars); break; }297 catch (e) {298 if (e instanceof PerpDataError && e.code === 'revised') continue; // stored values were replaced: read again299 if (!(e instanceof PerpDataError) || e.code !== 'rate_budget' || Date.now() > until) throw e;300 await sleep(Math.min(Math.max(1_000, e.retryAfterMs ?? 5_000), 20_000));301 }302 }303 }304 }305}306