Repository Public
1"""v7 research jobs: the job ledger hook, the job's commands (trial / evaluate with a cross-process seal / split),2the static lint that never executes, and the deflated Sharpe ratio that counts a job's trials."""3from __future__ import annotations45import json67import numpy as np8import pytest910from conftest import make_data, write_strategy11from spq import job as jobmod12from spq import lint13from spq.dsr import deflate_for_job, deflated_sharpe_from_returns14from spq.engine import BacktestConfig15from spq.report import build_report16from spq.research import JOB_LEDGER_ENV, Experiment, read_job_trials, read_ledger17from spq.strategy import load_strategy1819META = {"name": "rj_test", "version": "1", "universe": ["AAA", "BBB"], "interval": "1h", "lookback": 60,20 "params": {"fast": 8, "slow": 40}}21BODY = """22fast, slow = int(params["fast"]), int(params["slow"])23out = {}24for s, df in data.items():25 c = df["close"].to_numpy()26 out[s] = 0.5 if len(c) >= slow and c[-fast:].mean() > c[-slow:].mean() else 0.027return out28"""293031@pytest.fixture32def job(tmp_path, monkeypatch):33 """A job directory: data files, a job file, a ledger path in the environment."""34 data = make_data(900, seed=3)35 root = tmp_path / "data" / "coinbase"36 root.mkdir(parents=True)37 for s, df in data.items():38 out = df.reset_index()39 out["ts"] = (out["ts"].astype("int64") // 10**9).astype("int64") if str(out["ts"].dtype).endswith("ns, UTC]") \40 else out["ts"].map(lambda t: int(t.timestamp()))41 out.to_csv(root / f"{s}_1h.csv", index=False)42 jf = tmp_path / "JOB.json"43 jf.write_text(json.dumps({"kind": "spot", "symbols": ["AAA", "BBB"], "interval": "1h", "capital_usd": 1000,44 "data_dir": str(tmp_path / "data"), "source": "coinbase"}))45 ledger = tmp_path / "ledger" / "trials.jsonl"46 monkeypatch.setenv(JOB_LEDGER_ENV, str(ledger))47 monkeypatch.setenv(jobmod.JOB_FILE_ENV, str(jf))48 return {"dir": tmp_path, "ledger": ledger, "data": data, "file": jf}495051def lines(path):52 return [json.loads(x) for x in path.read_text().splitlines() if x.strip()] if path.exists() else []535455# ------------------------------------------------------------------------------------------------- the ledger hook5657def test_every_experiment_appends_to_the_job_ledger_whatever_ledger_the_caller_passes(tmp_path, monkeypatch):58 data = make_data(400, seed=1)59 s = load_strategy(write_strategy(tmp_path / "s", BODY, META))60 job_ledger = tmp_path / "job.jsonl"61 monkeypatch.setenv(JOB_LEDGER_ENV, str(job_ledger))62 exp = Experiment(data, BacktestConfig("1h")) # no ledger asked for63 exp.trial(s)64 exp.trial(s, {"fast": 4, "slow": 40})65 exp.trial(s) # cached: not a new trial66 assert [r["kind"] for r in lines(job_ledger)] == ["trial", "trial"]67 own = tmp_path / "own.jsonl"68 exp2 = Experiment(data, BacktestConfig("1h"), ledger=own) # a private ledger does not hide the trial69 exp2.trial(s)70 exp2.final_test(s)71 assert [r["kind"] for r in lines(own)] == ["trial", "test"]72 assert [r["kind"] for r in lines(job_ledger)] == ["trial", "trial", "trial", "test"]73 exp3 = Experiment(data, BacktestConfig("1h"), ledger=job_ledger) # the same file twice: one line, not two74 exp3.trial(s)75 assert len(lines(job_ledger)) == 576 monkeypatch.delenv(JOB_LEDGER_ENV)77 Experiment(data, BacktestConfig("1h")).trial(s) # without the variable: as before v778 assert len(lines(job_ledger)) == 5798081def test_the_job_ledger_is_read_defensively(tmp_path):82 p = tmp_path / "l.jsonl"83 p.write_text("\n".join([84 json.dumps({"kind": "trial", "validation_sharpe_per_bar": 0.01}),85 json.dumps({"kind": "trial", "validation_sharpe_per_bar": 1e9}), # absurd: counted, number dropped86 json.dumps({"kind": "trial", "validation_sharpe_per_bar": "0.5"}), # not a number87 json.dumps({"kind": "trial", "validation_sharpe_per_bar": float("nan")}),88 json.dumps({"kind": "test"}),89 json.dumps({"kind": "something else"}),90 "not json at all",91 json.dumps([1, 2, 3]),92 "",93 ]) + "\n")94 r = read_job_trials(p)95 assert r["count"] == 8 # every non-empty line is something the job did96 assert r["sharpes"] == [0.01]97 assert r["unreadable"] == 398 assert read_job_trials(tmp_path / "missing") == {"count": 0, "sharpes": [], "unreadable": 0}99100101# --------------------------------------------------------------------------------------------- deflated Sharpe ratio102103def test_a_jobs_trials_raise_the_bar_and_its_numbers_can_never_lower_it():104 rng = np.random.default_rng(0)105 r = rng.normal(0.0008, 0.01, 2000)106 own = [0.01, -0.02, 0.03, 0.0, 0.015]107 base = deflated_sharpe_from_returns(r, own)108 # 300 more trials with dispersed Sharpe ratios: more trials and at least the same variance -> a lower DSR109 wide = {"count": 300, "sharpes": list(rng.normal(0, 0.03, 300)), "unreadable": 0}110 d = deflate_for_job(base, r, own, wide)111 assert d["n_trials"] == 305 and d["n_trials_job"] == 300 and d["n_trials_this_run"] == 5112 assert d["var_trial_sharpes"] >= base["var_trial_sharpes"]113 assert d["deflated_sharpe"] < base["deflated_sharpe"]114 assert d["expected_max_sharpe_per_bar"] > base["expected_max_sharpe_per_bar"]115 # an agent that writes 300 identical Sharpe ratios to shrink the variance: the variance of the server's own116 # trials is the floor, the count still rises117 flat = {"count": 300, "sharpes": [0.0] * 300, "unreadable": 0}118 f = deflate_for_job(base, r, own, flat)119 assert f["var_trial_sharpes"] == pytest.approx(base["var_trial_sharpes"])120 assert f["n_trials"] == 305 and f["deflated_sharpe"] < base["deflated_sharpe"]121 # a job that recorded nothing changes nothing122 z = deflate_for_job(base, r, own, {"count": 0, "sharpes": [], "unreadable": 0})123 assert z["deflated_sharpe"] == pytest.approx(base["deflated_sharpe"]) and z["n_trials"] == 5124 assert base["n_trials"] == 5 # the input was not modified125126127def test_the_report_counts_the_jobs_trials(tmp_path):128 data = make_data(700, seed=5)129 s = load_strategy(write_strategy(tmp_path / "s", BODY, META))130 job_ledger = tmp_path / "job.jsonl"131 job_ledger.write_text("".join(json.dumps({"kind": "trial", "validation_sharpe_per_bar": 0.02 * (i % 5 - 2)}) + "\n"132 for i in range(40)))133 plain, _ = build_report(s, data, BacktestConfig("1h"), n_resamples=50, n_folds=2)134 counted, _ = build_report(s, data, BacktestConfig("1h"), n_resamples=50, n_folds=2, job_ledger=job_ledger)135 n = plain["robustness"]["trials_tested"]136 assert counted["robustness"]["trials_tested"] == n + 40137 d = counted["robustness"]["deflated_sharpe_detail"]138 assert d["n_trials_job"] == 40 and d["n_trials_this_run"] == n139 assert counted["metrics"] == plain["metrics"] # only the statistical bar moved140141142# ------------------------------------------------------------------------------------------------------- spq.lint143144def test_lint_parses_and_hashes_but_never_executes(tmp_path, capsys):145 ok = write_strategy(tmp_path / "ok", "", raw="X = 1 / 0\nMETA = {}\n\n\ndef target_weights(data, params):\n return {}\n")146 assert lint.main(["--strategy", str(ok)]) == 0 # importing it would raise ZeroDivisionError147 out = json.loads(capsys.readouterr().out)148 assert out["problems"] == [] and len(out["sha256"]) == 64149 bad = write_strategy(tmp_path / "bad", "", raw="import socket\nimport os\n\n\ndef target_weights(data, params):\n"150 " return {'x': open('/etc/passwd').read()}\n")151 assert lint.main(["--strategy", str(bad)]) == 3152 problems = json.loads(capsys.readouterr().out)["problems"]153 assert any("socket" in p for p in problems) and any("'open'" in p for p in problems)154 assert lint.main(["--strategy", str(tmp_path / "nothing")]) == 2155 big = write_strategy(tmp_path / "big", "", raw="# " + "x" * (lint.MAX_BYTES + 1) + "\n")156 assert lint.main(["--strategy", str(big)]) == 2157 link = tmp_path / "link"158 link.mkdir()159 (link / "strategy.py").symlink_to(ok / "strategy.py")160 assert lint.main(["--strategy", str(link)]) == 2 # a symbolic link is not a strategy file161162163# -------------------------------------------------------------------------------------------------------- spq.job164165def test_trial_runs_on_train_and_validation_only_and_is_counted(job, capsys):166 d = write_strategy(job["dir"] / "cand", BODY, META)167 assert jobmod.main(["trial", "--strategy", str(d)]) == 0168 out = json.loads(capsys.readouterr().out)169 assert set(out) == {"strategy_sha256", "params", "train", "validation"}170 assert jobmod.main(["trial", "--strategy", str(d), "--params", '{"fast": 4}']) == 0171 assert json.loads(capsys.readouterr().out)["params"] == {"fast": 4, "slow": 40}172 recs = lines(job["ledger"])173 assert [r["kind"] for r in recs] == ["trial", "trial"]174 assert recs[1]["params"] == {"fast": 4, "slow": 40}175176177def test_evaluate_writes_the_report_and_the_claims_and_seals_the_inner_test_split_across_processes(job, capsys):178 a = write_strategy(job["dir"] / "a", BODY, META)179 out, claims = job["dir"] / "report.json", job["dir"] / "claims.json"180 assert jobmod.main(["evaluate", "--strategy", str(a), "--out", str(out), "--claims", str(claims)]) == 0181 capsys.readouterr()182 report, c = json.loads(out.read_text()), json.loads(claims.read_text())183 assert c["train"]["sharpe"] == report["metrics"]["train"]["sharpe"]184 assert c["inner_test"]["total_return"] == report["metrics"]["test"]["total_return"]185 assert c["summary"] == "" and c["trials"] == len(lines(job["ledger"]))186 assert report["config"]["initial_cash"] == 1000.0187 tests = [r for r in lines(job["ledger"]) if r["kind"] == "test"]188 assert len(tests) == 1 and tests[0]["sha256"] == report["strategy_sha256"]189 assert jobmod.sealed_hash(job["ledger"]) == report["strategy_sha256"]190 # another version after the look: refused, in a NEW process state (nothing in memory links the two calls)191 b = write_strategy(job["dir"] / "b", BODY, {**META, "params": {"fast": 12, "slow": 40}})192 assert jobmod.main(["evaluate", "--strategy", str(b), "--out", str(job["dir"] / "r2.json")]) == 7193 assert "already evaluated" in capsys.readouterr().err194 assert not (job["dir"] / "r2.json").exists()195 assert len([r for r in lines(job["ledger"]) if r["kind"] == "test"]) == 1196 # the same version again is allowed (and counted again)197 assert jobmod.main(["evaluate", "--strategy", str(a), "--out", str(out)]) == 0198 assert len([r for r in lines(job["ledger"]) if r["kind"] == "test"]) == 2199200201def test_split_reports_the_boundaries_the_laboratory_itself_uses(job, capsys):202 assert jobmod.main(["split", "--kind", "spot", "--symbols", "AAA,BBB", "--interval", "1h",203 "--data-dir", str(job["dir"] / "data"), "--source", "coinbase"]) == 0204 s = json.loads(capsys.readouterr().out)205 idx = job["data"]["AAA"].index206 assert (s["bars"], s["dev_bars"], s["test_bars"]) == (900, 720, 180)207 assert s["start"] == idx[0].isoformat() and s["test_start"] == idx[720].isoformat() and s["last"] == idx[-1].isoformat()208 assert jobmod.main(["info"]) == 0209 assert json.loads(capsys.readouterr().out)["bars"] == 900210211212def test_wrong_kind_and_missing_job_file_fail_with_one_line(job, capsys, monkeypatch):213 perp = write_strategy(job["dir"] / "p", "return {}\n", {**META, "kind": "perp", "max_leverage": 2.0,214 "max_gross_exposure": 1.0})215 assert jobmod.main(["trial", "--strategy", str(perp)]) == 2216 assert "the job is a spot job" in capsys.readouterr().err217 monkeypatch.delenv(jobmod.JOB_FILE_ENV)218 assert jobmod.main(["info"]) == 2219 assert "no job file" in capsys.readouterr().err220221222def test_the_standard_arguments_are_the_same_on_both_sides():223 j = {"kind": "spot", "symbols": ["BTC", "ETH"], "interval": "1h", "capital_usd": 1000, "data_dir": "/d",224 "source": "coinbase"}225 assert jobmod.backtest_argv(j, "/s", "/o.json", sha256="ab") == [226 "--strategy", "/s", "--symbols", "BTC,ETH", "--interval", "1h", "--data-dir", "/d", "--initial-cash", "1000.0",227 "--out", "/o.json", "--resamples", "300", "--folds", "3", "--source", "coinbase", "--sha256", "ab"]228 p = jobmod.backtest_argv({**j, "kind": "perp"}, "/s", "/o.json", ledger="/l", job_ledger="/j")229 assert "--source" not in p and p[-4:] == ["--ledger", "/l", "--job-ledger", "/j"]230 assert read_ledger("/nonexistent/ledger") == []231