Repository Public
1// Strategy library and backtests. The API never writes strategy code: it lists what the research zone put under2// research/strategies, pins a version by the SHA-256 of its strategy.py, and runs the laboratory's backtest on stored candles.3import { createHash, randomUUID } from 'node:crypto';4import { appendFileSync, existsSync, mkdirSync, readdirSync, readFileSync, writeFileSync } from 'node:fs';5import path from 'node:path';6import type { FastifyInstance } from 'fastify';7import postgres from 'postgres';8import { z } from 'zod';9import { JOB_STRATEGY_RE, listingSources, MAX_AGENT_SYMBOLS } from '@spw/core';10import { config, UserError } from '../config.ts';11import { audit, sql } from '../db.ts';12import { dropScratch, labExec, RESEARCH_DIR, scratchDir } from './lab.ts';13import { agentSymbols, type AgentRow } from './routes.ts';1415const ROOT = path.join(RESEARCH_DIR, 'strategies');16export interface StrategyInfo { path: string; sha256: string;17 /** universe-generic: META['universe'] == ['*'], the symbols are the agent's (between minSymbols and maxSymbols of them) */18 generic: boolean; minSymbols: number; maxSymbols: number;19 meta: { name: string; version: string; universe: string[]; interval: string; lookback: number; params: Record<string, unknown>; min_symbols?: number; max_symbols?: number } }2021/** Run a laboratory command (v8: always inside the OS sandbox of lab.ts — no network, NO database URL, read-only laboratory,22 * one scratch directory). `strategyDir` is the only strategy directory the process can read. Never throws: when the sandbox23 * is unavailable the answer is `code: null` with the reason, exactly like a process that could not start. */24export function py(args: string[], o: { timeoutMs: number; strategyDir?: string; readable?: string[]; scratch?: string } = { timeoutMs: 60_000 }): Promise<{ code: number | null; out: string; err: string }> {25 return labExec(args, o).then((r) => ({ code: r.killed ? null : r.code, out: r.out, err: r.killed ? `${r.err}\nkilled: ${r.killed}` : r.err }),26 (e: Error) => ({ code: null, out: '', err: `could not start: ${e.message}` }));27}2829const Rel = z.string().regex(/^[a-z0-9_]{1,40}\/[a-z0-9_]{1,12}$/);30const Info = z.object({ sha256: z.string().regex(/^[0-9a-f]{64}$/), meta: z.object({ name: z.string(), version: z.string(), universe: z.array(z.string()).min(1), interval: z.string(), lookback: z.number().int().positive(), params: z.record(z.string(), z.unknown()),31 min_symbols: z.number().int().min(1).max(MAX_AGENT_SYMBOLS).optional(), max_symbols: z.number().int().min(1).max(MAX_AGENT_SYMBOLS).optional() }) });32/** Generic = exactly ['*'] (decided here from META, not from what the laboratory says about it). */33export const isGeneric = (universe: string[]): boolean => universe.length === 1 && universe[0] === '*';34const cache = new Map<string, StrategyInfo>();35export async function describe(rel: string): Promise<StrategyInfo> {36 Rel.parse(rel);37 const dir = path.join(ROOT, rel);38 if (!existsSync(path.join(dir, 'strategy.py'))) throw new UserError('not_found', 'strategy version not found', 404);39 const r = await py(['spq.describe', '--strategy', dir], { timeoutMs: 60_000, strategyDir: dir });40 if (r.code !== 0) throw new UserError('bad_strategy', `the laboratory refused this strategy: ${r.err.trim().split('\n').pop()?.slice(0, 160) ?? ''}`, 422);41 const p = Info.parse(JSON.parse(r.out));42 const generic = isGeneric(p.meta.universe);43 if (!generic && p.meta.universe.includes('*')) throw new UserError('bad_strategy', 'a universe is a list of symbols or exactly ["*"]', 422);44 const info: StrategyInfo = { path: `strategies/${rel}`, ...p, generic, minSymbols: generic ? p.meta.min_symbols ?? 1 : p.meta.universe.length, maxSymbols: generic ? p.meta.max_symbols ?? MAX_AGENT_SYMBOLS : p.meta.universe.length };45 cache.set(`${rel}:${p.sha256}`, info);46 return info;47}48export async function library(): Promise<(StrategyInfo & { readme: string })[]> {49 if (!existsSync(ROOT)) return [];50 const rels: [string, string][] = [];51 for (const name of readdirSync(ROOT).sort()) { if (!/^[a-z0-9_]{1,40}$/.test(name)) continue;52 if (JOB_STRATEGY_RE.test(name)) continue; // v7: a research job's strategy is listed by the research routes, with its server report53 for (const ver of readdirSync(path.join(ROOT, name)).sort()) if (/^[a-z0-9_]{1,12}$/.test(ver)) rels.push([name, ver]); }54 // LISTING ONLY (the page waited ~20 s: one laboratory process per version, one after the other). A version already described55 // under the same code hash is served from memory, the others are described in parallel. Registration still calls describe().56 const found = await Promise.all(rels.map(async ([name, ver]) => {57 try { let hit: StrategyInfo | undefined;58 try { hit = cache.get(`${name}/${ver}:${createHash('sha256').update(readFileSync(path.join(ROOT, name, ver, 'strategy.py'))).digest('hex')}`); } catch { /* unreadable: describe() decides */ }59 const i = hit ?? await describe(`${name}/${ver}`); let readme = '';60 for (const f of [path.join(ROOT, name, ver, 'README.md'), path.join(ROOT, name, '1', 'README.md')]) { try { readme = readFileSync(f, 'utf8').slice(0, 1500); if (readme.length > 200) break; } catch { /* none */ } }61 return [{ ...i, readme }]; } catch { return []; /* not loadable: not offered */ }62 }));63 return found.flat();64}65const sameSet = (a: string[], b: string[]) => a.length === b.length && [...a].sort().join() === [...b].sort().join();6667/** Can this library version be attached to an agent trading these symbols at this interval? Pure. A fixed universe must68 * equal the agent's symbols; a generic one fits any agent whose number of symbols is within its bounds. */69export function fitsAgent(info: Pick<StrategyInfo, 'generic' | 'minSymbols' | 'maxSymbols' | 'meta'>, symbols: string[], interval: string): { ok: true } | { ok: false; code: string; message: string } {70 if (info.meta.interval !== interval) return { ok: false, code: 'interval_mismatch', message: `this strategy decides every ${info.meta.interval}; the agent is set to ${interval}` };71 if (info.generic) {72 if (symbols.length < info.minSymbols || symbols.length > info.maxSymbols) return { ok: false, code: 'universe_size', message: `this strategy needs between ${info.minSymbols} and ${info.maxSymbols} symbols; the agent trades ${symbols.length} (${symbols.join(', ')})` };73 return { ok: true };74 }75 if (!sameSet(info.meta.universe, symbols)) return { ok: false, code: 'universe_mismatch', message: `this strategy trades ${info.meta.universe.join(', ')}; the agent's assets are ${symbols.join(', ')}` };76 return { ok: true };77}78/** The laboratory backtests on ONE candle source (it never mixes them): the first, in the consolidation order, that lists every symbol. */79export function backtestSource(symbols: string[]): string | null {80 return (['coinbase', 'bitstamp', 'kraken'] as const).find((src) => symbols.every((x) => listingSources(x).includes(src))) ?? null;81}8283const BAR: Record<string, string> = { '1h': '1 hour', '4h': '4 hours', '1d': '1 day' };84/** v8: the candles of a backtest as FILES prepared by the API (the laboratory process has no database URL and no network):85 * `<root>/<source>/<SYMBOL>_<interval>.csv`, every closed bar of ONE source, oldest first — the rows the laboratory's own86 * PostgreSQL loader used to read (`spq.data.pg.read_pg`), in the layout its file loader reads. Numbers are written in their87 * shortest round-trip form and pandas parses them back to the exact float64 (measured; lab-identity.test.ts): same values. */88export async function exportCandles(root: string, source: string, symbols: string[], interval: string, db: typeof sql = sql): Promise<void> {89 const step = BAR[interval];90 if (!step) throw new Error(`unsupported interval ${interval}`);91 for (const sym of symbols) {92 const rows = await db<{ t: string; o: number; h: number; l: number; c: number; v: number }[]>`select extract(epoch from ts)::bigint::text as t, open as o, high as h, low as l, close as c, volume as v93 from candles where source = ${source} and symbol = ${sym} and interval = ${interval} and ts + ${step}::interval <= now() order by ts`;94 if (!rows.length) throw new Error(`no ${interval} candles of ${sym} from ${source} in the market store`);95 const f = path.join(root, source, `${sym}_${interval}.csv`); mkdirSync(path.dirname(f), { recursive: true });96 writeFileSync(f, `ts,open,high,low,close,volume\n${rows.map((r) => `${r.t},${r.o},${r.h},${r.l},${r.c},${r.v}`).join('\n')}\n`);97 }98}99100/** Where the candles of a backtest are read from: RESEARCH_DATABASE_URL when it is set (the read-only role of the research101 * zone — before v8 this URL was handed to the laboratory process itself), else the API's own connection. Read-only session. */102export async function withCandleDb<T>(fn: (db: typeof sql) => Promise<T>): Promise<T> {103 const url = process.env.RESEARCH_DATABASE_URL;104 if (!url) return fn(sql);105 const db = postgres(url, { max: 1, idle_timeout: 5, onnotice: () => {}, connection: { default_transaction_read_only: true } });106 try { return await fn(db as unknown as typeof sql); } finally { await db.end({ timeout: 5 }).catch(() => {}); }107}108109/** The cross-run trial ledger (research/results/ledger.jsonl) is written by the API, never by the sandboxed process: the110 * process works on a copy in its scratch directory; the lines it APPENDED are then added to the real file. A copy whose111 * beginning no longer matches (rewritten history) is not merged. */112export const LEDGER = (): string => path.join(RESEARCH_DIR, 'results', 'ledger.jsonl');113export function stageLedger(scratch: string): { copy: string; merge(): boolean } {114 const real = LEDGER(), copy = path.join(scratch, 'ledger.jsonl'), before = existsSync(real) ? readFileSync(real, 'utf8') : '';115 if (before) writeFileSync(copy, before); // (no file at all when there is no ledger yet, as before)116 return { copy, merge: () => { let after = ''; try { after = readFileSync(copy, 'utf8'); } catch { return false; }117 if (!after.startsWith(before)) return false;118 const added = after.slice(before.length);119 if (added) { mkdirSync(path.dirname(real), { recursive: true }); appendFileSync(real, added); }120 return true; } };121}122123const running = new Set<string>();124/** Backtest a registered version on the stored candles and attach the report. Never throws; the outcome is stored on the row. */125export async function backtest(versionId: string, info: StrategyInfo, symbols: string[]): Promise<void> {126 if (running.has(versionId)) return; running.add(versionId);127 const tmp = scratchDir(), out = path.join(tmp, 'report.json'), data = path.join(tmp, 'data');128 try {129 await sql`update strategy_versions set status = 'backtesting', notes = '' where id = ${versionId} and status in ('draft', 'backtesting')`;130 const source = backtestSource(symbols);131 if (!source || !symbols.every((x) => /^[A-Z0-9]{2,10}$/.test(x))) { await sql`update strategy_versions set status = 'draft', notes = ${`backtest failed: no single exchange lists every symbol (${symbols.join(', ')}); the laboratory does not mix candle sources`} where id = ${versionId} and status = 'backtesting'`; return; }132 // v8: the candles are handed over as files by THIS process; the laboratory process is offline and has no database URL133 await withCandleDb((db) => exportCandles(data, source, symbols, info.meta.interval, db));134 const ledger = stageLedger(tmp), dir = path.join(RESEARCH_DIR, info.path);135 // the agent's symbols are always passed: a generic strategy has no list of its own, a fixed one equals them (checked at registration)136 const r = await py(['spq.backtest', '--strategy', dir, '--sha256', info.sha256, '--symbols', symbols.join(','), '--data-dir', data, '--source', source, '--out', out, '--ledger', ledger.copy],137 { timeoutMs: 20 * 60_000, strategyDir: dir, scratch: tmp });138 if (r.code !== 0 || !existsSync(out)) { await sql`update strategy_versions set status = 'draft', notes = ${`backtest failed: ${r.err.trim().split('\n').pop()?.slice(0, 300) ?? 'no report'}`} where id = ${versionId} and status = 'backtesting'`; return; }139 ledger.merge();140 const report = JSON.parse(readFileSync(out, 'utf8')) as { strategy_sha256?: string; config?: { symbols?: string[] } };141 if (report.strategy_sha256 !== info.sha256) { await sql`update strategy_versions set status = 'draft', notes = 'backtest report is for another code hash' where id = ${versionId}`; return; }142 // the report must be about the agent's own symbols (it records them in config.symbols)143 if (!Array.isArray(report.config?.symbols) || !sameSet(report.config.symbols, symbols)) { await sql`update strategy_versions set status = 'draft', notes = 'backtest report is for other symbols than the agent\'s' where id = ${versionId}`; return; }144 await sql`update strategy_versions set status = 'draft', report = ${sql.json(report as never)}, notes = '' where id = ${versionId} and status = 'backtesting'`;145 await audit('strategy.backtested', 'engine', { version: versionId, sha256: info.sha256 });146 } catch (e) { await sql`update strategy_versions set status = 'draft', notes = ${`backtest failed: ${(e as Error).message.slice(0, 200)}`} where id = ${versionId} and status = 'backtesting'`.catch(() => {}); }147 finally { running.delete(versionId); dropScratch(tmp); }148}149150export function strategyRoutes(app: FastifyInstance): void {151 // v6: `?kind=perp` lists the contract-2 library (perp agents); without it the spot library, which can never contain a perp strategy152 app.get('/api/v1/strategies', async (req) => {153 const q = z.object({ kind: z.enum(['spot', 'perp']).default('spot') }).parse(req.query);154 if (q.kind === 'perp') return { kind: 'perp', strategies: await (await import('./perp-routes.ts')).perpLibrary() };155 return { strategies: await library() };156 });157158 // pin a library version to an agent and backtest it; approval stays a separate owner action and needs the report159 app.post('/api/v1/agents/:id/strategies', async (req) => {160 const id = z.uuid().parse((req.params as { id: string }).id);161 const b = z.strictObject({ path: Rel }).parse(req.body);162 const [a] = await sql<AgentRow[]>`select * from agents where id = ${id}`;163 if (!a) throw new UserError('not_found', 'agent not found', 404);164 // v7: a research job's strategy (rj_*) is attached with the server report of its job — never re-described or re-backtested without the job's trial count165 if (JOB_STRATEGY_RE.test(b.path.split('/')[0]!)) return (await import('../research/evaluate.ts')).attachJobStrategy(a, b.path);166 // v6: a perp agent takes only contract-2 strategies (described with --contract 2, backtested by spq.perp_backtest)167 if (a.kind === 'perp') { const p = await import('./perp-routes.ts'); return p.registerPerp(a as never, b.path); }168 const info = await describe(b.path);169 const symbols = agentSymbols(a), fit = fitsAgent(info, symbols, a.interval);170 if (!fit.ok) throw new UserError(fit.code, fit.message, 422);171 const [row] = await sql<{ id: string; status: string; has_report: boolean }[]>`insert into strategy_versions (id, agent_id, name, version, code_sha256, path, meta)172 values (${randomUUID()}, ${a.id}, ${info.meta.name}, ${info.meta.version}, ${info.sha256}, ${info.path}, ${sql.json({ ...info.meta, generic: info.generic, symbols } as never)})173 on conflict (agent_id, code_sha256) do update set path = excluded.path returning id, status, report is not null as has_report`;174 await audit('strategy.registered', 'owner', { agent: a.id, version: row!.id, path: info.path, sha256: info.sha256, generic: info.generic, symbols });175 if (!row!.has_report && row!.status !== 'approved') void backtest(row!.id, info, symbols);176 return { id: row!.id, sha256: info.sha256, generic: info.generic, symbols, status: row!.has_report ? row!.status : 'backtesting' };177 });178}179