Repository Public
1// V7 research runtime — what the API does with a submission. Nothing in it is trusted:2// 1. the archive is decoded and bounded (only the expected files, sizes, UTF-8);3// 2. strategy.py is checked STATICALLY (spq.lint: parsed and hashed, never imported) before anything runs;4// 3. it is installed under research/strategies/rj_<job>/<version>/ — the API is the only writer there, the path is built from5// the job row, never from the archive;6// 4. the laboratory describes it (META, hash) in the OS sandbox; name, version, kind, interval and universe must be the job's;7// 5. the SERVER backtests it, twice, in the sandbox, from files, with no network and no database URL:8// - replication: the standard evaluation on the data the agent had → compared with what the agent claims;9// - holdout: the report on the full frozen data set, whose test split the job never had, with the deflated Sharpe ratio10// computed over this run's trials PLUS every trial the job consumed (the runner's append-only ledger);11// 6. the server's report — never the agent's numbers — is attached to a strategy version left in `draft` for the owner.12// A failure at any step leaves no report: a version without a server report cannot be approved, paper-traded or mandated.13import { randomUUID } from 'node:crypto';14import { existsSync, mkdirSync, readFileSync, realpathSync, rmSync, writeFileSync } from 'node:fs';15import path from 'node:path';16import { z } from 'zod';17import { compareClaim, decodeSubmission, JOB_STRATEGY_RE, parseClaims, type AgentClaims, type RunnerResult, type SubmissionArchive } from '@spw/core';18import { UserError } from '../config.ts';19import { audit, sql } from '../db.ts';20import { RESEARCH_DIR } from '../agents/live.ts';21import { perpReportProblem } from '../agents/perp-strategies.ts';22import type { AgentRow } from '../agents/routes.ts';23import { agentSymbols } from '../agents/routes.ts';24import { dataDirs } from './dataset.ts';25import { dropScratch, labRun, lastLine, scratchDir } from './lab.ts';26import { loadJob, move, sysLog, type JobRow } from './store.ts';2728const STRATEGIES = path.join(existsSync(RESEARCH_DIR) ? realpathSync(RESEARCH_DIR) : RESEARCH_DIR, 'strategies');29const sameSet = (a: string[], b: string[]): boolean => a.length === b.length && [...a].sort().join() === [...b].sort().join();30const Meta = z.object({ name: z.string().max(40), version: z.string().max(12), kind: z.string().optional(), universe: z.array(z.string().max(16)).min(1).max(24), interval: z.string().max(4), lookback: z.number().int().positive().max(100_000),31 max_leverage: z.number().positive().optional(), max_gross_exposure: z.number().positive().optional(), min_symbols: z.number().int().optional(), max_symbols: z.number().int().optional(), params: z.record(z.string(), z.unknown()) });32const Described = z.object({ sha256: z.string().regex(/^[0-9a-f]{64}$/), meta: Meta, generic: z.boolean().optional(), kind: z.enum(['spot', 'perp']).optional(), contract: z.number().int().optional() });33type Described = z.infer<typeof Described>;34const SPLIT_KEYS = ['total_return', 'sharpe', 'max_drawdown', 'trades'] as const;35type SplitNums = Partial<Record<(typeof SPLIT_KEYS)[number], number | null>>;36const pick = (m: unknown): SplitNums => Object.fromEntries(SPLIT_KEYS.map((k) => [k, typeof (m as Record<string, unknown> | undefined)?.[k] === 'number' ? (m as Record<string, number>)[k]! : null]));3738class Refused extends Error {}39const refuse = (why: string): never => { throw new Refused(why); };4041/** job-scoped directory of a job's strategy inside the laboratory's library; built from the job row only */42export function installDir(job: Pick<JobRow, 'strategy_name' | 'strategy_version'>): { rel: string; dir: string } {43 if (!JOB_STRATEGY_RE.test(job.strategy_name) || !/^[a-z0-9_]{1,12}$/.test(job.strategy_version)) throw new Error('bad job strategy name');44 const dir = path.resolve(STRATEGIES, job.strategy_name, job.strategy_version);45 if (!dir.startsWith(STRATEGIES + path.sep)) throw new Error('strategy path escapes the library');46 return { rel: `${job.strategy_name}/${job.strategy_version}`, dir };47}4849/** The arguments of the standard evaluation of a job — the same as spq.job.backtest_argv on the laboratory side. */50export function backtestArgs(job: Pick<JobRow, 'kind' | 'universe' | 'interval' | 'capital_usd' | 'dataset'>, strategyDir: string, sha256: string, dataDir: string, out: string, o: { resamples?: number; folds?: number; ledger?: string; jobLedger?: string }): string[] {51 return [job.kind === 'perp' ? 'spq.perp_backtest' : 'spq.backtest', '--strategy', strategyDir, '--sha256', sha256, '--symbols', job.universe.join(','), '--interval', job.interval, '--data-dir', dataDir,52 '--initial-cash', String(job.capital_usd), '--out', out, ...(job.kind === 'spot' && job.dataset?.source ? ['--source', job.dataset.source] : []),53 ...(o.resamples ? ['--resamples', String(o.resamples)] : []), ...(o.folds ? ['--folds', String(o.folds)] : []), ...(o.ledger ? ['--ledger', o.ledger] : []), ...(o.jobLedger ? ['--job-ledger', o.jobLedger] : [])];54}5556export interface ServerSummary {57 status: 'ok' | 'failed'; reason?: string; sha256?: string; sandbox?: 'sandbox-exec' | 'none'; meta?: Record<string, unknown>;58 holdout?: { testStart: string; testEnd: string; bars: number };59 trials?: { job: number; server: number; total: number; ledgerTampered: boolean };60 deflatedSharpe?: number | null; sharpeCi?: [number | null, number | null] | null;61 metrics?: Record<'train' | 'validation' | 'test', SplitNums>; benchmark?: SplitNums; checks?: Record<string, unknown>;62 /** the standard evaluation re-run by the server on the data the agent had */63 replication?: Record<'train' | 'validation' | 'inner_test', SplitNums>;64 /** what the agent claims next to what the server measured on the same data */65 comparison?: { split: string; metric: string; claimed: number | null; measured: number | null; differs: boolean }[];66 claimsDiffer?: boolean; claimsProblem?: string | null;67}6869async function describeSandboxed(dir: string, kind: 'spot' | 'perp', scratch: string): Promise<{ d: Described; sandbox: 'sandbox-exec' | 'none' }> {70 const r = await labRun(['spq.describe', '--strategy', dir, ...(kind === 'perp' ? ['--contract', '2'] : [])], { timeoutMs: 60_000, scratch, strategyDir: dir });71 if (r.code !== 0) refuse(`the laboratory refused this strategy: ${lastLine(r.err, 200)}`);72 const p = Described.safeParse((() => { try { return JSON.parse(r.out); } catch { return null; } })());73 if (!p.success) return refuse('the laboratory\'s description of the strategy is not readable');74 return { d: p.data, sandbox: r.sandbox };75}7677/** Steps 1–3: decode, static check, install. Returns the code hash. Throws Refused with a reason the owner can read. */78export async function installSubmission(job: JobRow, archive: SubmissionArchive, scratch: string): Promise<{ sha256: string; files: Record<string, string>; claimsText: string | null }> {79 let sub; try { sub = decodeSubmission(archive); } catch (e) { return refuse(`malformed submission: ${(e as Error).message}`); }80 const text = (n: 'strategy.py' | 'README.md' | 'NOTES.md' | 'claims.json'): string | null => (sub.files[n] ? new TextDecoder().decode(sub.files[n]) : null);81 const stage = path.join(scratch, 'stage'); mkdirSync(stage, { recursive: true });82 writeFileSync(path.join(stage, 'strategy.py'), sub.files['strategy.py']!);83 const lint = await labRun(['spq.lint', '--strategy', stage], { timeoutMs: 30_000, scratch });84 let parsed: { sha256?: string; problems?: string[] } = {};85 try { parsed = JSON.parse(lint.out) as typeof parsed; } catch { /* handled below */ }86 if (lint.code === 3 && parsed.problems?.length) refuse(`strategy.py fails the purity check: ${parsed.problems.slice(0, 6).join('; ').slice(0, 500)}`);87 if (lint.code !== 0 || !parsed.sha256 || !/^[0-9a-f]{64}$/.test(parsed.sha256)) refuse(`strategy.py could not be checked: ${lastLine(lint.err, 200)}`);88 const { dir } = installDir(job);89 if (existsSync(dir)) refuse('a strategy is already installed for this job');90 mkdirSync(dir, { recursive: true, mode: 0o755 });91 for (const n of ['strategy.py', 'README.md', 'NOTES.md'] as const) if (sub.files[n]) writeFileSync(path.join(dir, n), sub.files[n], { flag: 'wx', mode: 0o644 });92 return { sha256: parsed.sha256!, files: { 'strategy.py': text('strategy.py')!, 'README.md': text('README.md')!, ...(text('NOTES.md') ? { 'NOTES.md': text('NOTES.md')! } : {}) }, claimsText: text('claims.json') };93}94const uninstall = (job: JobRow): void => { try { const { dir } = installDir(job); rmSync(path.dirname(dir), { recursive: true, force: true }); } catch { /* nothing installed */ } };9596function compare(claims: AgentClaims | null, rep: NonNullable<ServerSummary['replication']>): NonNullable<ServerSummary['comparison']> {97 const out: NonNullable<ServerSummary['comparison']> = [];98 for (const split of ['train', 'validation', 'inner_test'] as const) for (const metric of SPLIT_KEYS) {99 const c = compareClaim(claims?.[split]?.[metric], rep[split][metric]);100 out.push({ split, metric, ...c });101 }102 return out;103}104105/** The whole pipeline for one submission. Never throws; the job ends `submitted` (server report attached) or `failed`. */106export async function evaluateSubmission(jobId: string, result: Extract<RunnerResult, { outcome: 'submission' }>): Promise<void> {107 const job = await loadJob(jobId);108 if (job.state !== 'evaluating') return;109 const scratch = scratchDir();110 let versionId: string | null = null, installed = false;111 const summary: ServerSummary = { status: 'failed' };112 try {113 const inst = await installSubmission(job, result.archive, scratch); installed = true;114 const { claims, problem: claimsProblem } = parseClaims(inst.claimsText);115 const { rel, dir } = installDir(job);116 await sql`update research_jobs set submission = ${sql.json({ sha256: inst.sha256, path: `strategies/${rel}`, files: inst.files, claims, claimsProblem, bytes: Object.fromEntries(Object.entries(inst.files).map(([k, v]) => [k, v.length])) } as never)} where id = ${job.id}`;117 await sysLog(job.id, `submission accepted for evaluation: strategy.py ${inst.sha256.slice(0, 16)}… passed the static purity check and was installed as strategies/${rel}`);118119 // 4 — describe (the first time any of the submitted code runs: module level only, in the sandbox)120 const { d, sandbox } = await describeSandboxed(dir, job.kind, scratch);121 summary.sha256 = d.sha256; summary.sandbox = sandbox; summary.meta = d.meta;122 if (d.sha256 !== inst.sha256) refuse('the installed file does not have the hash that was checked');123 if (d.meta.name !== job.strategy_name || d.meta.version !== job.strategy_version) refuse(`META must carry the job's name and version ("${job.strategy_name}" / "${job.strategy_version}"), got "${d.meta.name.slice(0, 40)}" / "${d.meta.version.slice(0, 12)}"`);124 const kind = d.meta.kind === 'perp' ? 'perp' : 'spot';125 if (kind !== job.kind) refuse(`this is a ${kind} strategy; the job asked for a ${job.kind} strategy`);126 if (d.meta.interval !== job.interval) refuse(`the strategy decides every ${d.meta.interval}; the job is at ${job.interval}`);127 const generic = d.meta.universe.length === 1 && d.meta.universe[0] === '*';128 if (job.kind === 'perp') {129 if (d.meta.max_leverage === undefined || d.meta.max_gross_exposure === undefined) refuse('a perp strategy declares max_leverage and max_gross_exposure');130 if (d.meta.universe.some((s) => s.endsWith('.SPOT'))) refuse('spot legs (X.SPOT) are not supported by perp agents');131 if (!sameSet(d.meta.universe, job.universe)) refuse(`the strategy trades ${d.meta.universe.join(', ')}; the job's instruments are ${job.universe.join(', ')}`);132 } else if (generic) {133 const lo = d.meta.min_symbols ?? 1, hi = d.meta.max_symbols ?? 12;134 if (job.universe.length < lo || job.universe.length > hi) refuse(`the strategy takes between ${lo} and ${hi} symbols; the job has ${job.universe.length}`);135 } else if (!sameSet(d.meta.universe, job.universe)) refuse(`the strategy trades ${d.meta.universe.join(', ')}; the job's symbols are ${job.universe.join(', ')}`);136137 // a version row for the agent (status backtesting: not approvable until the server's report is attached)138 if (job.agent_id) {139 const [a] = await sql<AgentRow[]>`select * from agents where id = ${job.agent_id}`;140 if (!a) refuse('the agent of this job no longer exists');141 const symbols = a!.kind === 'perp' ? a!.universe : agentSymbols(a!);142 if ((a!.kind ?? 'spot') !== job.kind || a!.interval !== job.interval || !sameSet(symbols, job.universe)) refuse('the agent changed since the job was created (kind, interval or symbols): the strategy was not attached');143 const [row] = await sql<{ id: string; status: string }[]>`insert into strategy_versions (id, agent_id, name, version, code_sha256, path, meta, status, notes)144 values (${randomUUID()}, ${a!.id}, ${d.meta.name}, ${d.meta.version}, ${d.sha256}, ${`strategies/${rel}`}, ${sql.json({ ...d.meta, ...(job.kind === 'spot' ? { generic } : {}), symbols: job.universe, researchJob: job.id } as never)}, 'backtesting', '')145 on conflict (agent_id, code_sha256) do update set path = excluded.path returning id, status`;146 if (row!.status === 'approved') refuse('this exact code is already approved for the agent');147 versionId = row!.id;148 await sql`update research_jobs set version_id = ${versionId} where id = ${job.id}`;149 await audit('strategy.registered', 'research', { agent: a!.id, version: versionId, path: `strategies/${rel}`, sha256: d.sha256, job: job.id });150 }151152 // 5 — the server's own backtests, from files, sandboxed153 const dirs = dataDirs(job), rep = path.join(scratch, 'replication.json'), out = path.join(scratch, 'report.json'), jobLedger = path.join(scratch, 'job-ledger.jsonl');154 writeFileSync(jobLedger, result.archive.ledger);155 const r1 = await labRun(backtestArgs(job, dir, d.sha256, dirs.dev, rep, { resamples: 300, folds: 3 }), { timeoutMs: 20 * 60_000, scratch, readable: [dirs.dev], strategyDir: dir });156 if (r1.code !== 0 || !existsSync(rep)) refuse(`the server could not reproduce the evaluation on the agent's data: ${lastLine(r1.err)}`);157 const repReport = JSON.parse(readFileSync(rep, 'utf8')) as { metrics?: Record<string, unknown> };158 summary.replication = { train: pick(repReport.metrics?.train), validation: pick(repReport.metrics?.validation), inner_test: pick(repReport.metrics?.test) };159 const r2 = await labRun(backtestArgs(job, dir, d.sha256, dirs.full, out, { ledger: path.join(scratch, 'server-ledger.jsonl'), jobLedger }), { timeoutMs: 30 * 60_000, scratch, readable: [dirs.full], strategyDir: dir });160 if (r2.code !== 0 || !existsSync(out)) refuse(`the server-side backtest failed: ${lastLine(r2.err)}`);161 const report = JSON.parse(readFileSync(out, 'utf8')) as Record<string, any>;162 if (report.strategy_sha256 !== d.sha256) refuse('the server report is for another code hash');163 if (!Array.isArray(report.config?.symbols) || !sameSet(report.config.symbols, job.universe)) refuse('the server report is for other symbols than the job\'s');164 if (job.kind === 'perp') { const why = perpReportProblem(report, d.sha256, job.universe); if (why) refuse(why); }165 // the test split of the server's report must start exactly where the agent's data ended166 const testStart = Date.parse(String(report.splits?.test?.[0])), cut = Date.parse(job.dataset!.holdout!.devEnd);167 if (!(testStart >= cut)) refuse(`the server's test split starts at ${String(report.splits?.test?.[0])}, inside the data the agent had (it ended at ${job.dataset!.holdout!.devEnd}): this is not an out-of-sample test`);168 const detail = report.robustness?.deflated_sharpe_detail ?? {};169 if (!(Number(detail.n_trials_job) >= result.usage.trials && Number(report.robustness?.trials_tested) >= Number(detail.n_trials_job))) refuse('the report does not count the trials the job consumed');170 report.research_job = { id: job.id, trials_job: detail.n_trials_job, ledger_tampered: result.usage.ledgerTampered, holdout_start: job.dataset!.holdout!.devEnd, agent_data_end: job.dataset!.holdout!.devEnd };171172 summary.holdout = { testStart: String(report.splits.test[0]), testEnd: String(report.splits.test[1]), bars: job.dataset!.holdout!.testBars };173 summary.trials = { job: Number(detail.n_trials_job), server: Number(detail.n_trials_this_run), total: Number(report.robustness.trials_tested), ledgerTampered: result.usage.ledgerTampered };174 summary.deflatedSharpe = typeof report.robustness.deflated_sharpe === 'number' ? report.robustness.deflated_sharpe : null;175 summary.sharpeCi = Array.isArray(report.robustness.bootstrap_sharpe_ci) ? report.robustness.bootstrap_sharpe_ci : null;176 summary.metrics = { train: pick(report.metrics?.train), validation: pick(report.metrics?.validation), test: pick(report.metrics?.test) };177 summary.benchmark = pick(report.benchmarks?.[job.kind === 'perp' ? 'perp_long_1x' : 'buy_and_hold']);178 summary.checks = report.checks ?? {};179 summary.comparison = compare(claims, summary.replication); summary.claimsDiffer = summary.comparison.some((c) => c.differs) || (claims?.trials !== undefined && claims.trials < result.usage.trials);180 summary.claimsProblem = claimsProblem;181 summary.status = 'ok';182183 // 6 — attach the SERVER's report; the version stays a draft for the owner184 if (versionId) await sql`update strategy_versions set status = 'draft', report = ${sql.json(report as never)}, notes = '' where id = ${versionId} and status = 'backtesting'`;185 await sql`update research_jobs set server = ${sql.json(summary as never)}, report = ${sql.json(report as never)}, updated_at = now() where id = ${job.id}`;186 const dsr = typeof summary.deflatedSharpe === 'number' ? summary.deflatedSharpe.toFixed(3) : 'n/a';187 await sql`insert into research_log (agent_id, kind, title, body, data) values (${job.agent_id}, 'research', 'Research job submitted', ${`${job.strategy_name}/${job.strategy_version} (${d.sha256.slice(0, 12)}…): server-side backtest on held-out data done; ${summary.trials.total} trials counted (${summary.trials.job} by the job), deflated Sharpe ${dsr}${summary.claimsDiffer ? '; the agent\'s reported numbers differ from the server\'s' : ''}. Awaiting your approval.`},188 ${sql.json({ job: job.id, version: versionId, sha256: d.sha256, trials: summary.trials, deflatedSharpe: summary.deflatedSharpe, claimsDiffer: summary.claimsDiffer } as never)})`;189 if (versionId) await audit('strategy.backtested', 'engine', { version: versionId, sha256: d.sha256, job: job.id, ...(job.kind === 'perp' ? { kind: 'perp' } : {}) });190 await move(job.id, 'evaluating', 'submitted');191 await audit('research.submitted', 'research', { job: job.id, agent: job.agent_id, version: versionId, name: job.strategy_name, sha256: d.sha256, trials: summary.trials.total, claimsDiffer: summary.claimsDiffer });192 await sysLog(job.id, `server evaluation done: ${summary.trials.total} trials counted, deflated Sharpe ${dsr}; the version is a draft awaiting the owner's approval`);193 } catch (e) {194 const reason = e instanceof Refused ? e.message : e instanceof UserError ? e.message : `evaluation error: ${(e as Error).message.slice(0, 200)}`;195 summary.status = 'failed'; summary.reason = reason.slice(0, 600);196 // a version that got no server report stays an un-approvable draft with the reason; a strategy nobody references is removed197 if (versionId) await sql`update strategy_versions set status = 'draft', notes = ${`research job refused: ${reason}`.slice(0, 400)} where id = ${versionId} and status = 'backtesting'`.catch(() => {});198 else if (installed) uninstall(job);199 await sql`update research_jobs set server = ${sql.json(summary as never)}, updated_at = now() where id = ${job.id}`.catch(() => {});200 await move(job.id, 'evaluating', 'failed', { reason: summary.reason }).catch(() => {});201 await sql`insert into research_log (agent_id, kind, title, body, data) values (${job.agent_id}, 'research', 'Research job refused', ${summary.reason}, ${sql.json({ job: job.id } as never)})`.catch(() => {});202 await audit('research.failed', 'research', { job: job.id, agent: job.agent_id, name: job.strategy_name, state: 'failed', reason: summary.reason.slice(0, 200) }).catch(() => {});203 await sysLog(job.id, `submission refused: ${summary.reason}`);204 } finally { dropScratch(scratch); }205}206207/** Attach a job's evaluated strategy to an agent through the ordinary route (POST /agents/:id/strategies with an rj_* path):208 * no code runs — the server report of the job is reused, so it is valid only for exactly the job's symbols and interval. */209export async function attachJobStrategy(a: AgentRow, rel: string): Promise<{ id: string; sha256: string; symbols: string[]; status: string }> {210 const [name, version] = rel.split('/');211 const [job] = await sql<JobRow[]>`select * from research_jobs where strategy_name = ${name ?? ''} and strategy_version = ${version ?? ''}`;212 if (!job || job.state !== 'submitted' || !job.report || job.server?.status !== 'ok') throw new UserError('not_found', 'this research strategy has no server-side report: it cannot be attached', 404);213 const symbols = a.kind === 'perp' ? a.universe : agentSymbols(a);214 if ((a.kind ?? 'spot') !== job.kind) throw new UserError('wrong_kind', `this research strategy is a ${job.kind} strategy`, 422);215 if (a.interval !== job.interval || !sameSet(symbols, job.universe)) throw new UserError('universe_mismatch', `the server evaluated this strategy on ${job.universe.join(', ')} at ${job.interval}; this agent trades ${symbols.join(', ')} at ${a.interval}. Start a research job for this agent instead.`, 422);216 const sha = String(job.server.sha256), meta = job.server.meta as Record<string, unknown>;217 const [row] = await sql<{ id: string; status: string }[]>`insert into strategy_versions (id, agent_id, name, version, code_sha256, path, meta, status, report)218 values (${randomUUID()}, ${a.id}, ${job.strategy_name}, ${job.strategy_version}, ${sha}, ${`strategies/${rel}`}, ${sql.json({ ...meta, symbols, researchJob: job.id } as never)}, 'draft', ${sql.json(job.report as never)})219 on conflict (agent_id, code_sha256) do update set path = excluded.path returning id, status`;220 await audit('strategy.registered', 'owner', { agent: a.id, version: row!.id, path: `strategies/${rel}`, sha256: sha, job: job.id });221 return { id: row!.id, sha256: sha, symbols, status: row!.status };222}223224/** Put back the strategy files of evaluated jobs that are missing from the laboratory's library (a fresh checkout, a225 * redeployment): the API stored what it installed, and only writes a file whose SHA-256 is the one it evaluated. */226export async function restoreJobStrategies(): Promise<number> {227 const rows = await sql<{ strategy_name: string; strategy_version: string; submission: { files?: Record<string, string> } | null; server: { sha256?: string } | null }[]>`228 select strategy_name, strategy_version, submission, server from research_jobs where state = 'submitted' and server->>'status' = 'ok'`;229 let n = 0;230 for (const r of rows) {231 try {232 const { dir } = installDir(r), code = r.submission?.files?.['strategy.py'];233 if (existsSync(path.join(dir, 'strategy.py')) || typeof code !== 'string') continue;234 const { createHash } = await import('node:crypto');235 if (createHash('sha256').update(Buffer.from(code, 'utf8')).digest('hex') !== r.server?.sha256) continue; // not byte-identical: leave it missing (the engine then holds)236 mkdirSync(dir, { recursive: true, mode: 0o755 });237 for (const f of ['strategy.py', 'README.md', 'NOTES.md']) if (typeof r.submission?.files?.[f] === 'string') writeFileSync(path.join(dir, f), r.submission.files[f]!, { flag: 'wx', mode: 0o644 });238 n += 1;239 } catch { /* one job's files never block the others */ }240 }241 return n;242}243