spb/sp-wallet

1// Perp agents: the engine's real-world side. Wires the engine (perp-engine.ts) to the derivatives module — venue data through2// the data service, execution through the paper account — and runs one pass per minute. A PAPER agent never reaches the signer3// or a trading endpoint. A LIVE agent (V7) goes through perp-live-cycle.ts: a venue account under a signer-enforced mandate.4import { engineLoop } from '../loop-stats.ts'; // v85import { D, type DecStr } from '@spw/derivatives';6import { sql } from '../db.ts';7import { derivatives, type Derivatives } from '../derivatives/index.ts';8import { getSeries } from '../market/series.ts';9import type { MarketSymbol } from '../market/types.ts';10import type { LiveService } from '../derivatives/exec.ts';11import { live as sharedLive } from '../derivatives/live.ts';12import { PerpDataError, PerpDataStore, type PerpSeries } from './perp-data.ts';13import { tickLivePerpAgent } from './perp-live-cycle.ts';14import { instrumentId, tickPerpAgent, type PerpAgentRow, type PerpEngineDeps } from './perp-engine.ts';15import { PaperPerpExecutor } from './perp-exec.ts';16import { runPerpSignal } from './perp-signal.ts';17import type { PerpLabDeps } from './perp-strategies.ts';1819export interface PerpRuntime { deriv: Derivatives; store: PerpDataStore; deps: PerpEngineDeps; lab: PerpLabDeps;20  /** V7: the live execution engine a LIVE agent trades through (the process-wide one unless a test or a script injects its own) */21  live: () => LiveService }2223/** last hourly close of the spot market data, if it is recent: the reference when the venue publishes no index */24async function spotClose(symbol: string): Promise<DecStr | null> {25  try {26    const bar = (await getSeries(symbol as MarketSymbol, '1h', { limit: 1 })).bars.at(-1);27    return bar && Date.now() - bar.ts < 2 * 3_600_000 && bar.close > 0 ? D.str(D.from(bar.close.toFixed(8))) : null;28  } catch { return null; }29}30async function spotCandles(symbol: string, interval: string, fromMs: number): Promise<[number, number, number, number, number, number][]> {31  const rows = await sql<{ ts: Date; open: number; high: number; low: number; close: number; volume: number }[]>`select ts, open, high, low, close, volume from candles32    where source = 'coinbase' and symbol = ${symbol} and interval = ${interval} and ts >= ${new Date(fromMs)} order by ts`;33  return rows.map((r) => [r.ts.getTime(), r.open, r.high, r.low, r.close, r.volume]);34}3536export function createPerpRuntime(deriv: Derivatives, o: { signal?: PerpEngineDeps['signal']; weightPerMinute?: number; live?: LiveService; gate?: PerpEngineDeps['gate'] } = {}): PerpRuntime {37  const store = new PerpDataStore(deriv.data, { now: deriv.data.now, spot: spotClose, ...(o.weightPerMinute === undefined ? {} : { weightPerMinute: o.weightPerMinute }) });38  const deps: PerpEngineDeps = {39    now: deriv.data.now,40    gate: o.gate ?? (async (symbols, interval, lookback) => {41      const data: Record<string, PerpSeries> = {}; let bar: number | null = null;42      for (const s of symbols) {43        const g = await store.gate(await deriv.data.instrument(instrumentId(s)), interval, lookback);44        if (bar !== null && g.bar !== bar) throw new PerpDataError('misaligned', `${s}: its last closed bar is not the other instruments'`);45        bar = g.bar; data[s] = g.series;46      }47      if (bar === null) throw new PerpDataError('empty', 'no instrument');48      return { bar, data };49    }),50    signal: o.signal ?? runPerpSignal,51    market: async (symbol) => {52      const inst = await deriv.data.instrument(instrumentId(symbol));53      const [ticker, book] = await Promise.all([deriv.data.ticker(inst, { within: 1_000 }), deriv.data.book(inst, { within: 1_000 })]);54      if (!book) throw new Error(`${inst.venue} has no order book`);55      return { inst, ticker, book };56    },57    executor: (a) => new PaperPerpExecutor(deriv.paper, a.deriv_account_id),58  };59  return { deriv, store, deps, lab: { data: deriv.data, store, spotCandles }, live: () => o.live ?? sharedLive() };60}6162let shared: PerpRuntime | null = null;63/** the process-wide runtime, on the derivatives instance the routes and the paper worker share */64export function perpRuntime(): PerpRuntime { return (shared ??= createPerpRuntime(derivatives())); }65/** server.ts (an injected derivatives instance), tests and the simulation script */66export function setPerpRuntime(r: PerpRuntime | null): void { shared = r; }6768const backoff = new Map<string, number>();69/** One pass over every perp agent in paper trading or on a venue account. Errors never cross agents; an agent that failed waits two minutes. */70export async function perpEnginePass(rt: PerpRuntime = perpRuntime(), now = Date.now()): Promise<Record<string, string>> {71  const out: Record<string, string> = {};72  const agents = await sql<PerpAgentRow[]>`select * from agents where kind = 'perp' and ((state = 'PAPER_TRADING' and mode <> 'LIVE') or (mode = 'LIVE' and state in ('LIVE', 'PAUSED', 'STOPPING'))) order by created_at`;73  for (const a of agents) {74    try {75      if ((backoff.get(a.id) ?? 0) > now) { out[a.id] = 'backing off after an error'; continue; }76      out[a.id] = a.mode === 'LIVE' ? await tickLivePerpAgent(a, rt) : await tickPerpAgent(a, rt.deps);77    } catch (e) {78      backoff.set(a.id, now + 2 * 60_000); out[a.id] = `error: ${(e as Error).message.slice(0, 160)}`;79      await sql`insert into research_log (agent_id, kind, title, body) values (${a.id}, 'hold', 'Engine pass interrupted', ${(e as Error).message.slice(0, 300)})`.catch(() => {});80    }81  }82  return out;83}8485export function startPerpAgentEngine(everyMs = 60_000): () => void {86  return engineLoop('perp-agent-engine', everyMs, () => perpEnginePass()); // v8: heartbeat in loopStats87}88