Repository Public
1"""Chronological splits and the Experiment object that enforces their use.23The rule "the test split is never used for parameter selection" is enforced by the API:45- Every routine that compares configurations (grid search, sensitivity, walk-forward) goes through6 `Experiment.trial()` / `Experiment.run()`, which only ever see `Experiment.dev_data()`: a copy of the data7 truncated before the first test bar. Asking them for the test split raises `TestSplitError`.8- The test split is reachable through `Experiment.final_test()` only. It can be evaluated once per strategy name;9 from that moment the experiment is sealed for that strategy: any further trial, or a second `final_test` with10 another version or other parameters, raises `TestSplitError`.11- Every trial and every test evaluation is counted (and appended to an optional JSONL ledger that survives the12 process), so the report can state how many configurations were tried and how often the test split was looked at.1314What it cannot do: stop someone from creating a new Experiment and starting again. The persistent ledger exists15to make that visible (`test_evaluations` in the report), not impossible.16"""17from __future__ import annotations1819import itertools20import json21import math22import os23from dataclasses import dataclass24from pathlib import Path2526import numpy as np27import pandas as pd2829from .data.core import DataError30from .engine import BacktestConfig, BacktestResult, buy_and_hold, check_aligned, equal_weight, run_backtest31from .metrics import sharpe_ratio32from .strategy import Strategy3334SELECTION_SPLITS = ("train", "validation", "development")35JOB_LEDGER_ENV = "SPQ_JOB_LEDGER"363738class TestSplitError(PermissionError):39 """An attempt to use the test split for anything other than the single final evaluation."""4041 __test__ = False # not a pytest class424344@dataclass(frozen=True)45class Splits:46 """Half-open [start, end) windows on bar-open time; `*_last` are the last bars inside each window."""4748 train: tuple[pd.Timestamp, pd.Timestamp]49 validation: tuple[pd.Timestamp, pd.Timestamp]50 test: tuple[pd.Timestamp, pd.Timestamp]51 last: dict5253 @classmethod54 def chronological(cls, index: pd.DatetimeIndex, fractions=(0.6, 0.2, 0.2)) -> "Splits":55 f = tuple(float(x) for x in fractions)56 if len(f) != 3 or any(x <= 0 for x in f) or abs(sum(f) - 1.0) > 1e-9:57 raise ValueError("fractions must be three positive numbers summing to 1 (train, validation, test)")58 n = len(index)59 i1, i2 = int(n * f[0]), int(n * (f[0] + f[1]))60 if min(i1, i2 - i1, n - i2) < 3:61 raise DataError("not enough bars for three splits")62 step = index[-1] - index[-2]63 end = index[-1] + step64 return cls(65 train=(index[0], index[i1]), validation=(index[i1], index[i2]), test=(index[i2], end),66 last={"train": index[i1 - 1], "validation": index[i2 - 1], "test": index[-1]},67 )6869 def window(self, split: str) -> tuple[pd.Timestamp, pd.Timestamp]:70 if split == "development":71 return (self.train[0], self.validation[1])72 if split not in ("train", "validation", "test"):73 raise ValueError(f"unknown split {split!r}")74 return getattr(self, split)7576 def to_report(self) -> dict:77 return {s: [self.window(s)[0].isoformat(), self.last[s].isoformat()] for s in ("train", "validation", "test")}787980def params_key(params: dict) -> str:81 return json.dumps(params, sort_keys=True, default=str)828384class Experiment:85 #: backtest function used for every run; spq.perp_research.PerpExperiment replaces it with the perp backtester86 _runner = staticmethod(run_backtest)8788 def __init__(self, data: dict[str, pd.DataFrame], config: BacktestConfig, *, fractions=(0.6, 0.2, 0.2),89 ledger: str | Path | None = None):90 index = check_aligned(data)91 self.config = config92 self.splits = Splits.chronological(index, fractions)93 self.symbols = list(data.keys())94 self.__full = data95 test_start = self.splits.test[0]96 self._dev = {s: df.loc[df.index < test_start] for s, df in data.items()}97 self.ledger_path = Path(ledger) if ledger else None98 self.trials: list[dict] = []99 self.test_evaluations: list[dict] = []100 self._cache: dict[tuple, BacktestResult] = {}101 self._sealed: dict[str, tuple[str, str]] = {}102103 # ----------------------------------------------------------------------------------------------- selection side104 def dev_data(self) -> dict[str, pd.DataFrame]:105 """Train + validation bars only. This is all that selection code can obtain."""106 return dict(self._dev)107108 def _check_open(self, strategy: Strategy):109 if strategy.name in self._sealed:110 raise TestSplitError(111 f"the test split was already evaluated for strategy {strategy.name!r}: selection is closed. "112 "Tuning after looking at the test split would turn it into a second validation split."113 )114115 def trial(self, strategy: Strategy, params: dict | None = None) -> BacktestResult:116 """Run one configuration over the whole development range (train + validation) and record it as a trial."""117 self._check_open(strategy)118 p = strategy.params if params is None else dict(params)119 key = (strategy.sha256, params_key(p))120 if key not in self._cache:121 start, end = self.splits.window("development")122 res = self._runner(strategy, self._dev, self.config, start=start, end=end, params=p)123 self._cache[key] = res124 rec = {"kind": "trial", "strategy": strategy.name, "sha256": strategy.sha256, "params": p,125 "symbols": self.symbols, "interval": self.config.interval}126 for split in ("train", "validation"):127 r = self.split_returns(res, split)128 rec[f"{split}_sharpe_per_bar"] = _finite(sharpe_ratio(r))129 rec[f"{split}_bars"] = int(len(r))130 rec[f"{split}_return"] = _finite(float(np.prod(1.0 + r) - 1.0))131 self.trials.append(rec)132 self._append_ledger(rec)133 return self._cache[key]134135 def split_returns(self, result: BacktestResult, split: str) -> np.ndarray:136 """Bar returns of a development-range run restricted to `split` (train or validation)."""137 if split not in SELECTION_SPLITS:138 raise TestSplitError("only train, validation and development are available for selection")139 start, end = self.splits.window(split)140 r = result.returns141 return r[(r.index >= start) & (r.index < end)].to_numpy()142143 def run(self, strategy: Strategy, split: str, params: dict | None = None) -> BacktestResult:144 """Stand-alone backtest of one split (starting from cash at the split start). Never the test split."""145 if split not in SELECTION_SPLITS:146 raise TestSplitError(147 f"split {split!r} is not available here: the test split can only be evaluated once, "148 "through Experiment.final_test()"149 )150 self._check_open(strategy)151 start, end = self.splits.window(split)152 return self._runner(strategy, self._dev, self.config, start=start, end=end, params=params)153154 def run_window(self, strategy: Strategy, start, end, params: dict | None = None) -> BacktestResult:155 """Backtest an arbitrary window that lies entirely inside the development range (walk-forward folds)."""156 self._check_open(strategy)157 if pd.Timestamp(end) > self.splits.test[0]:158 raise TestSplitError("the requested window reaches into the test split")159 return self._runner(strategy, self._dev, self.config, start=start, end=end, params=params)160161 def trial_sharpes(self, split: str = "validation", strategy_name: str | None = None) -> list[float]:162 return [t[f"{split}_sharpe_per_bar"] for t in self.trials163 if t.get(f"{split}_sharpe_per_bar") is not None164 and (strategy_name is None or t["strategy"] == strategy_name)]165166 # ---------------------------------------------------------------------------------------------- reporting side167 def final_test(self, strategy: Strategy, params: dict | None = None) -> BacktestResult:168 """The single out-of-sample evaluation of a strategy. Seals the experiment for that strategy name."""169 p = strategy.params if params is None else dict(params)170 ident = (strategy.sha256, params_key(p))171 if strategy.name in self._sealed and self._sealed[strategy.name] != ident:172 raise TestSplitError(173 f"strategy {strategy.name!r} was already evaluated on the test split with another version or "174 "other parameters; a second look is parameter selection on the test split"175 )176 start, end = self.splits.window("test")177 res = self._runner(strategy, self.__full, self.config, start=start, end=end, params=p)178 if strategy.name not in self._sealed:179 self._sealed[strategy.name] = ident180 rec = {"kind": "test", "strategy": strategy.name, "sha256": strategy.sha256, "params": p,181 "symbols": self.symbols, "interval": self.config.interval,182 "window": [start.isoformat(), self.splits.last["test"].isoformat()]}183 self.test_evaluations.append(rec)184 self._append_ledger(rec)185 return res186187 def benchmarks(self, split: str) -> dict[str, BacktestResult]:188 """Buy-and-hold (per symbol and equal-weight, never rebalanced) and rebalanced equal-weight on a split.189190 Benchmarks have no parameters, so they may be computed on any split, including test.191 """192 start, end = self.splits.window(split)193 data = self.__full if split == "test" else self._dev194 out = {"buy_and_hold": buy_and_hold(data, self.config, start=start, end=end),195 "equal_weight": equal_weight(data, self.config, start=start, end=end)}196 if len(self.symbols) > 1:197 for s in self.symbols:198 out[f"buy_and_hold_{s}"] = buy_and_hold({s: data[s]}, self.config, start=start, end=end)199 return out200201 def lookahead_data(self) -> dict[str, pd.DataFrame]:202 """Full data for the look-ahead check only (it compares a run with itself; nothing is selected)."""203 return dict(self.__full)204205 # ------------------------------------------------------------------------------------------------------ ledger206 def _append_ledger(self, rec: dict):207 # v7 research jobs: when SPQ_JOB_LEDGER is set (by the job runner, outside the agent), EVERY trial and test208 # evaluation of every Experiment of the process is appended there, whatever `ledger=` the caller passed.209 # The runner copies the lines out of the agent's reach as they appear and counts them against the budget.210 job = os.environ.get(JOB_LEDGER_ENV)211 seen = set()212 for path in (self.ledger_path, Path(job) if job else None):213 if path is None or str(path.absolute()) in seen:214 continue215 seen.add(str(path.absolute()))216 path.parent.mkdir(parents=True, exist_ok=True)217 with path.open("a") as fh:218 fh.write(json.dumps(rec, default=str) + "\n")219220221def read_ledger(path) -> list[dict]:222 path = Path(path)223 if not path.exists():224 return []225 return [json.loads(line) for line in path.read_text().splitlines() if line.strip()]226227228# --- v7 research jobs ---------------------------------------------------------------------------------------------229MAX_SANE_SHARPE_PER_BAR = 1.0230231232def read_job_trials(path) -> dict:233 """Trials consumed by a research job, from the runner's copy of the job ledger. The file was written by an234 UNTRUSTED process: it is read defensively. Every non-empty line counts as one trial (a line is something the235 job did, so garbage can only raise the count); only finite validation Sharpe ratios of plausible size, on236 well-formed "trial" records, are kept as numbers."""237 out = {"count": 0, "sharpes": [], "unreadable": 0}238 path = Path(path)239 if not path.exists():240 return out241 for line in path.read_text(errors="replace").splitlines():242 if not line.strip():243 continue244 out["count"] += 1245 try:246 rec = json.loads(line)247 except (ValueError, TypeError):248 out["unreadable"] += 1249 continue250 if not isinstance(rec, dict) or rec.get("kind") not in ("trial", "test"):251 out["unreadable"] += 1252 continue253 v = rec.get("validation_sharpe_per_bar")254 if (isinstance(v, (int, float)) and not isinstance(v, bool) and math.isfinite(v)255 and abs(v) <= MAX_SANE_SHARPE_PER_BAR):256 out["sharpes"].append(float(v))257 return out258# --- end v7 -------------------------------------------------------------------------------------------------------259260261def _finite(x):262 return float(x) if x is not None and math.isfinite(x) else None263264265# ------------------------------------------------------------------------------------------------------- grids266267def one_at_a_time_grid(params: dict, factors=(0.5, 2.0)) -> list[dict]:268 """Default sensitivity grid: the base parameters, then each numeric parameter scaled by each factor in turn."""269 grid = [dict(params)]270 for k, v in params.items():271 if isinstance(v, bool) or not isinstance(v, (int, float)):272 continue273 for f in factors:274 nv = max(1, int(round(v * f))) if isinstance(v, int) else float(v * f)275 if nv != v:276 cand = {**params, k: nv}277 if cand not in grid:278 grid.append(cand)279 return grid280281282def cartesian_grid(base: dict, axes: dict[str, list]) -> list[dict]:283 """Full grid over `axes` (parameter -> list of values), other parameters taken from `base`."""284 unknown = set(axes) - set(base)285 if unknown:286 raise ValueError(f"grid parameters not in META['params']: {sorted(unknown)}")287 keys = list(axes)288 grid = [dict(base)]289 for combo in itertools.product(*(axes[k] for k in keys)):290 cand = {**base, **dict(zip(keys, combo))}291 if cand not in grid:292 grid.append(cand)293 return grid294