Repository Public
1"""Laboratory commands of a research job (v7): the ONLY evaluation entry points a job needs.23 python -m spq.job info what the job is and which data it has4 python -m spq.job trial --strategy <dir> [--params JSON] one configuration on train + validation (a counted trial)5 python -m spq.job evaluate --strategy <dir> --out r.json [--claims c.json]6 the standard report on the job's data: selection analyses,7 then THE look at the inner test split (sealed per job)8 python -m spq.job split --data-dir D ... (server side) where the test split starts on a data set910The job is described by a JSON file (`--job`, default $SPQ_JOB_FILE):11 {"kind": "spot"|"perp", "symbols": [...], "interval": "1h", "capital_usd": 1000, "data_dir": "...",12 "source": "coinbase"} # source: spot candle source (spot jobs); perp jobs read the hyperliquid store1314Hygiene, enforced here and not left to the caller:15- `trial` goes through Experiment.trial(): only train + validation bars, recorded in the job ledger ($SPQ_JOB_LEDGER).16- `evaluate` is the only command that reaches the inner test split, through Experiment.final_test(). The seal is kept17 ACROSS processes with the job ledger: once a test evaluation is recorded for one strategy.py hash, `evaluate` refuses18 any other hash (TestSplitError). Running it again on the same hash is allowed and is counted again.19- `evaluate` writes the claims file itself, from the report it just computed.2021What this cannot do: the job ledger is a file the job's own processes can write. The runner copies it out of reach22as it grows and reports any truncation, and the wallet's server re-runs everything on data the job never had.23"""24from __future__ import annotations2526import argparse27import json28import os29import sys30from pathlib import Path3132from .research import JOB_LEDGER_ENV, TestSplitError, read_ledger33from .strategy import StrategyError, load_strategy, sha256_file3435JOB_FILE_ENV = "SPQ_JOB_FILE"36CLAIM_KEYS = ("total_return", "sharpe", "max_drawdown", "trades")37#: resampling draws of the standard job evaluation (the metrics of the splits do not depend on it)38JOB_RESAMPLES = 30039JOB_FOLDS = 3404142def load_job(path=None) -> dict:43 path = path or os.environ.get(JOB_FILE_ENV)44 if not path:45 raise ValueError(f"no job file: pass --job or set {JOB_FILE_ENV}")46 job = json.loads(Path(path).read_text())47 if job.get("kind") not in ("spot", "perp"):48 raise ValueError("job file: kind must be spot or perp")49 syms = job.get("symbols")50 if not isinstance(syms, list) or not syms or not all(isinstance(s, str) for s in syms):51 raise ValueError("job file: symbols must be a non-empty list")52 for k in ("interval", "data_dir"):53 if not isinstance(job.get(k), str) or not job[k]:54 raise ValueError(f"job file: {k} is required")55 job.setdefault("capital_usd", 1000.0)56 return job575859def backtest_argv(job: dict, strategy_dir: str, out: str, *, sha256: str | None = None, ledger: str | None = None,60 job_ledger: str | None = None, resamples: int = JOB_RESAMPLES, folds: int = JOB_FOLDS) -> list[str]:61 """Arguments of spq.backtest / spq.perp_backtest for the standard evaluation of a job (same on both sides)."""62 argv = ["--strategy", str(strategy_dir), "--symbols", ",".join(job["symbols"]), "--interval", job["interval"],63 "--data-dir", job["data_dir"], "--initial-cash", str(float(job["capital_usd"])), "--out", str(out),64 "--resamples", str(resamples), "--folds", str(folds)]65 if job["kind"] == "spot" and job.get("source"):66 argv += ["--source", job["source"]]67 if sha256:68 argv += ["--sha256", sha256]69 if ledger:70 argv += ["--ledger", str(ledger)]71 if job_ledger:72 argv += ["--job-ledger", str(job_ledger)]73 return argv747576def _experiment(job: dict):77 """(experiment, compute_metrics) on the job's data — spot or perp."""78 if job["kind"] == "perp":79 from .data.hyperliquid import load_meta, load_perp_universe80 from .perp_backtest import make_config, parser81 from .perp_metrics import compute_perp_metrics82 from .perp_research import PerpExperiment8384 a = parser().parse_args(["--strategy", ".", "--data-dir", job["data_dir"], "--out", "-",85 "--initial-cash", str(float(job["capital_usd"]))])86 config = make_config(a, job["interval"], load_meta(job["data_dir"]), job["symbols"])87 data, _q, _fq, _raw = load_perp_universe(job["symbols"], job["interval"], None, None, root=job["data_dir"])88 return PerpExperiment(data, config), compute_perp_metrics89 from .backtest import parser as spot_parser90 from .data import load_universe91 from .engine import BacktestConfig92 from .metrics import compute_metrics93 from .research import Experiment9495 # the defaults of the spq.backtest command line (costs, rebalance threshold), so a trial and the report agree96 a = spot_parser().parse_args(["--strategy", ".", "--out", "-", "--initial-cash", str(float(job["capital_usd"]))])97 data, _q = load_universe(job["symbols"], job["interval"], None, None, source=job.get("source"),98 root=job["data_dir"])99 config = BacktestConfig(interval=job["interval"], fee_bps=a.fee_bps, slippage_bps=a.slippage_bps,100 initial_cash=a.initial_cash, min_order=a.min_order,101 rebalance_threshold=a.rebalance_threshold, decision_every=a.decision_every)102 return Experiment(data, config), compute_metrics103104105def _check_kind(strategy, job: dict):106 if strategy.kind != job["kind"]:107 raise StrategyError(f"this is a {strategy.kind} strategy; the job is a {job['kind']} job")108109110def cmd_info(job: dict, _a) -> int:111 exp, _ = _experiment(job)112 s = exp.splits113 print(json.dumps({"kind": job["kind"], "symbols": job["symbols"], "interval": job["interval"],114 "capital_usd": job["capital_usd"],115 "bars": int(len(next(iter(exp.lookahead_data().values())))),116 "splits": s.to_report(),117 "note": "these are the splits of the data this job was given; the wallet's server holds a "118 "later period that no job command can reach"}))119 return 0120121122def cmd_trial(job: dict, a) -> int:123 strategy = load_strategy(a.strategy)124 _check_kind(strategy, job)125 exp, metrics = _experiment(job)126 params = {**strategy.params, **json.loads(a.params)} if a.params else None127 exp.trial(strategy, params)128 out = {"strategy_sha256": strategy.sha256, "params": params or strategy.params}129 for split in ("train", "validation"):130 m = metrics(exp.run(strategy, split, params))131 out[split] = {k: m.get(k) for k in CLAIM_KEYS}132 print(json.dumps(out))133 return 0134135136def sealed_hash(ledger_path) -> str | None:137 """The strategy.py hash the job's inner test split was already evaluated for (None: not looked at yet)."""138 for rec in read_ledger(ledger_path) if ledger_path and Path(ledger_path).exists() else []:139 if isinstance(rec, dict) and rec.get("kind") == "test" and isinstance(rec.get("sha256"), str):140 return rec["sha256"]141 return None142143144def claims_from_report(report: dict, trials: int | None) -> dict:145 m = report.get("metrics", {})146 out = {name: {k: m.get(split, {}).get(k) for k in CLAIM_KEYS}147 for name, split in (("train", "train"), ("validation", "validation"), ("inner_test", "test"))}148 if trials is not None:149 out["trials"] = trials150 out["summary"] = ""151 return out152153154def cmd_evaluate(job: dict, a) -> int:155 digest = sha256_file(Path(a.strategy) / "strategy.py")156 ledger = os.environ.get(JOB_LEDGER_ENV)157 try:158 seal = sealed_hash(ledger)159 except ValueError:160 seal = None # an unreadable ledger seals nothing here; the runner's copy still counts every line161 if seal is not None and seal != digest:162 raise TestSplitError(163 f"the inner test split of this job was already evaluated for strategy.py {seal[:12]}...: a second look "164 "with another version is parameter selection on the test split. Submit the version that was evaluated.")165 if job["kind"] == "perp":166 from .perp_backtest import main as run167 else:168 from .backtest import main as run169 code = run(backtest_argv(job, a.strategy, a.out, sha256=digest))170 if code != 0:171 return code172 if a.claims:173 report = json.loads(Path(a.out).read_text())174 n = None175 if ledger and Path(ledger).exists():176 n = sum(1 for line in Path(ledger).read_text(errors="replace").splitlines() if line.strip())177 Path(a.claims).write_text(json.dumps(claims_from_report(report, n), indent=1) + "\n")178 return 0179180181def cmd_split(_job, a) -> int:182 """Server side: the boundaries the laboratory itself would use on this data set (default 0.6 / 0.2 / 0.2)."""183 job = {"kind": a.kind, "symbols": [s for s in a.symbols.split(",") if s], "interval": a.interval,184 "data_dir": a.data_dir, "source": a.source, "capital_usd": 1000.0}185 exp, _ = _experiment(job)186 s = exp.splits187 dev = len(next(iter(exp.dev_data().values())))188 full = len(next(iter(exp.lookahead_data().values())))189 print(json.dumps({"start": s.train[0].isoformat(), "test_start": s.test[0].isoformat(),190 "last": s.last["test"].isoformat(), "end": s.test[1].isoformat(),191 "bars": int(full), "dev_bars": int(dev), "test_bars": int(full - dev)}))192 return 0193194195def main(argv=None) -> int:196 p = argparse.ArgumentParser(prog="python -m spq.job", description=__doc__.split("\n")[0])197 p.add_argument("--job", default=None, help=f"job file (default: ${JOB_FILE_ENV})")198 sub = p.add_subparsers(dest="cmd", required=True)199 sub.add_parser("info")200 t = sub.add_parser("trial")201 t.add_argument("--strategy", required=True)202 t.add_argument("--params", default=None, help="JSON object merged over META['params'] for this trial")203 e = sub.add_parser("evaluate")204 e.add_argument("--strategy", required=True)205 e.add_argument("--out", required=True)206 e.add_argument("--claims", default=None, help="also write the claims file derived from the report")207 s = sub.add_parser("split")208 s.add_argument("--kind", choices=("spot", "perp"), required=True)209 s.add_argument("--symbols", required=True)210 s.add_argument("--interval", required=True)211 s.add_argument("--data-dir", required=True)212 s.add_argument("--source", default=None)213 a = p.parse_args(argv)214 try:215 job = None if a.cmd == "split" else load_job(a.job)216 return {"info": cmd_info, "trial": cmd_trial, "evaluate": cmd_evaluate, "split": cmd_split}[a.cmd](job, a)217 except TestSplitError as e:218 print(f"spq.job: TestSplitError: {e}", file=sys.stderr)219 return 7220 except Exception as e: # noqa: BLE001 - one line on stderr, non-zero exit221 print(f"spq.job: {type(e).__name__}: {e}", file=sys.stderr)222 return 2223224225if __name__ == "__main__":226 sys.exit(main())227