diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cf49b9..e773ac7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ All notable changes to this project will be documented in this file. - Grouped surrogate validation holds out whole designs or regimes alongside separate row-validation metrics. Prediction and recommendation expose observed-support diagnostics and warn on extrapolation for GP and RF (#152). - `run_sequential()` allocates bounded extra replication to unresolved comparisons and feasibility boundaries, preserves raw replicate ids, and reports budgets, stopping reasons and simultaneous finite-horizon mean intervals under explicit bounded-score assumptions (#153). - Adaptive sessions queue known configurations and import compatible completed observations with persistent identity/provenance and duplicate protection. Versioned session schemas validate reopened journals and refuse unverifiable legacy storage (#154). +- Opt-in `EvaluationCache` reuses grid evaluations by typed configuration, replicate namespace/id, model/scorer revision, fidelity, objective and annotation definitions; includes provenance, bypass, invalidation and conflicting-evidence checks (#154). ## [0.3.0] — 2026-10-02 diff --git a/docs/api/io.md b/docs/api/io.md index 3b6fe53..96b0850 100644 --- a/docs/api/io.md +++ b/docs/api/io.md @@ -4,4 +4,6 @@ Serialization and deserialization of study results. ::: trade_study.load_results +::: trade_study.EvaluationCache + ::: trade_study.save_results diff --git a/docs/api/runner.md b/docs/api/runner.md index 4c6a54e..131519e 100644 --- a/docs/api/runner.md +++ b/docs/api/runner.md @@ -24,6 +24,44 @@ between an external side effect and saving its result may require repeating that evaluation. Make side effects safe to repeat. Retry limits do not bound the number of explicit resume invocations. +## Reusing identified evaluations + +Reuse is opt-in through an explicit cache: + +```python +cache = EvaluationCache( + "evaluations.sqlite", + revision="simulator-v2-scorer-v1-data-v3-annotations-v1", + replicate_namespace="experiment-2026-seed-17", + fidelity="high", +) +results = run_grid(world, scorer, grid, observables, n_reps=3, cache=cache) +``` + +The key includes a typed canonical configuration, replicate id, caller revision, +replicate namespace, fidelity, inspectable simulator/scorer class code, +objective definitions and annotation semantics. Grid position and total replicate +count are excluded: reordering designs or requesting more replicates can reuse +identical existing evaluations. Raw result rows still retain the current grid's +design-point ids. Different replicate ids and independent experiment namespaces +remain distinct even when their configurations match. + +The caller must update the revision when instance settings, external data, +globals or opaque callable behavior changes. The replicate namespace must +identify the actual randomness/seed convention. Use a fresh namespace for an +independent experiment; matching configuration alone never establishes reuse. +Supported scalar/container configurations preserve types; opaque values are +refused. Cached evaluators must not mutate their configuration. + +`cache_bypass=True` skips reads and writes, leaving stored evidence untouched. +`cache.clear()` invalidates all contexts in the cache file. Changing identity +creates a new context. Conflicting scores for one identity are refused. +Metadata exposes `cache_hit`, `cache_key`, and the revision/namespace/fidelity +context. Cached wall time is the original evaluation's time. Completed grid +checkpoints can populate the cache without re-evaluation; bypassing the cache +does not bypass checkpoints. Concurrent workers may evaluate an absent identity +more than once before either saves; reuse does not promise exactly-once effects. + ::: trade_study.run_grid ::: trade_study.run_adaptive diff --git a/src/trade_study/__init__.py b/src/trade_study/__init__.py index 15367f7..f2f1a11 100644 --- a/src/trade_study/__init__.py +++ b/src/trade_study/__init__.py @@ -6,6 +6,7 @@ from ._pareto import extract_front, hypervolume, igd_plus, pareto_rank from ._scoring import coverage_curve, score from ._version import __version__ +from .cache import EvaluationCache from .design import ( Factor, FactorConstraint, @@ -57,6 +58,7 @@ "Annotation", "Constraint", "Direction", + "EvaluationCache", "Factor", "FactorConstraint", "FactorType", diff --git a/src/trade_study/cache.py b/src/trade_study/cache.py new file mode 100644 index 0000000..07f92dc --- /dev/null +++ b/src/trade_study/cache.py @@ -0,0 +1,211 @@ +"""Opt-in evaluation reuse with explicit revision and replicate identity.""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from contextlib import closing +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from typing import TYPE_CHECKING, Any + +import numpy as np + +from ._checkpoint import _type_identity +from ._recovery import _grid_identity +from .protocols import TrialResult + +if TYPE_CHECKING: + from .protocols import Annotation, Observable, Scorer, Simulator + + +class EvaluationCache: + """Persistent immutable evaluation evidence for explicitly identified runs.""" + + def __init__( + self, + path: str | Path, + *, + revision: str, + replicate_namespace: str, + fidelity: str, + ) -> None: + """Create or reopen an evaluation cache. + + Args: + path: SQLite cache file; compatible contexts may share one file. + revision: Caller-managed simulator/scorer/data/annotation revision. + Include every behavior input not identifiable from class code. + replicate_namespace: Explicit randomness/seed experiment identity. + Use a new namespace for genuinely independent replication. + fidelity: Explicit simulation/evaluation fidelity identity. + + Raises: + ValueError: If an identity is empty or the cache format is incompatible. + """ + if any( + not value.strip() for value in (revision, replicate_namespace, fidelity) + ): + msg = "revision, replicate_namespace and fidelity must be nonempty" + raise ValueError(msg) + self.path = Path(path) + self.revision = revision + self.replicate_namespace = replicate_namespace + self.fidelity = fidelity + self.path.parent.mkdir(parents=True, exist_ok=True) + with closing(sqlite3.connect(self.path, timeout=30)) as connection, connection: + connection.execute( + "CREATE TABLE IF NOT EXISTS cache_format (version INTEGER)" + ) + stored = connection.execute("SELECT version FROM cache_format").fetchone() + if stored is None: + connection.execute("INSERT INTO cache_format VALUES (1)") + elif stored[0] != 1: + msg = "Incompatible evaluation cache format" + raise ValueError(msg) + connection.execute( + "CREATE TABLE IF NOT EXISTS evaluations " + "(key TEXT PRIMARY KEY, payload TEXT NOT NULL)" + ) + + def clear(self) -> None: + """Invalidate all stored evaluations across every context in this file.""" + with closing(sqlite3.connect(self.path, timeout=30)) as connection, connection: + connection.execute("DELETE FROM evaluations") + + +def _canonical(value: object) -> object: + """Canonicalize configurations without collapsing opaque or container types. + + Returns: + A typed JSON representation with stable dictionary ordering. + + Raises: + ValueError: If a value has unsupported opaque behavior. + """ + if isinstance(value, Enum): + return {"enum": _type_identity(value), "value": _canonical(value.value)} + if isinstance(value, np.generic): + return {"numpy": str(value.dtype), "value": _canonical(value.item())} + if type(value) in {type(None), bool, int, float, str}: + return {"type": type(value).__name__, "value": value} + if isinstance(value, (list, tuple)) and type(value) in {list, tuple}: + return {"type": type(value).__name__, "value": [_canonical(v) for v in value]} + if isinstance(value, dict) and type(value) is dict: + return { + "dict": sorted( + ( + json.dumps(_canonical(k), sort_keys=True, allow_nan=False), + _canonical(v), + ) + for k, v in value.items() + ) + } + msg = "Cached configurations require supported JSON/scalar values" + raise ValueError(msg) + + +@dataclass(frozen=True) +class _BoundCache: + cache: EvaluationCache + identity: str + context: dict[str, str] + + def key(self, config: dict[str, Any], rep: int) -> str: + value = json.dumps( + {"context": self.identity, "config": _canonical(config), "rep": rep}, + sort_keys=True, + allow_nan=False, + ) + return hashlib.sha256(value.encode()).hexdigest() + + def load(self, config: dict[str, Any], rep: int) -> TrialResult | None: + key = self.key(config, rep) + with closing(sqlite3.connect(self.cache.path, timeout=30)) as connection: + row = connection.execute( + "SELECT payload FROM evaluations WHERE key = ?", (key,) + ).fetchone() + if row is None: + return None + payload = json.loads(row[0]) + return TrialResult( + config, + payload["scores"], + payload["wall_seconds"], + { + "rep": rep, + "cache_hit": True, + "cache_key": key, + "cache_context": dict(self.context), + }, + ) + + def save(self, config: dict[str, Any], rep: int, result: TrialResult) -> None: + key = self.key(config, rep) + payload = json.dumps( + {"scores": result.scores, "wall_seconds": result.wall_seconds}, + sort_keys=True, + ) + with ( + closing(sqlite3.connect(self.cache.path, timeout=30)) as connection, + connection, + ): + connection.execute( + "INSERT OR IGNORE INTO evaluations VALUES (?, ?)", (key, payload) + ) + row = connection.execute( + "SELECT payload FROM evaluations WHERE key = ?", (key,) + ).fetchone() + if json.dumps(json.loads(row[0])["scores"], sort_keys=True) != json.dumps( + result.scores, sort_keys=True + ): + msg = ( + "Conflicting scores for one cached evaluation identity; " + "change the replicate namespace/revision" + ) + raise ValueError(msg) + result.metadata.update({ + "cache_hit": False, + "cache_key": key, + "cache_context": dict(self.context), + }) + + +def _bind_cache( + cache: EvaluationCache, + world: Simulator, + scorer: Scorer, + observables: list[Observable], + annotations: list[Annotation] | None, +) -> _BoundCache: + identity = json.dumps( + { + "definition": _grid_identity( + world, scorer, [], observables, annotations, 1, cache.revision + ), + "replicate_namespace": cache.replicate_namespace, + "fidelity": cache.fidelity, + "annotation_lookups": [ + _canonical(a.lookup) + if isinstance(a.lookup, dict) + else { + "module": getattr(a.lookup, "__module__", None), + "name": getattr(a.lookup, "__qualname__", None), + } + for a in (annotations or []) + ], + }, + sort_keys=True, + ) + return _BoundCache( + cache, + identity, + { + "revision": cache.revision, + "replicate_namespace": cache.replicate_namespace, + "fidelity": cache.fidelity, + "schema_digest": hashlib.sha256(identity.encode()).hexdigest(), + }, + ) diff --git a/src/trade_study/runner.py b/src/trade_study/runner.py index 732d8ff..283b29c 100644 --- a/src/trade_study/runner.py +++ b/src/trade_study/runner.py @@ -8,11 +8,13 @@ import inspect import time +from copy import deepcopy from typing import TYPE_CHECKING, Any import numpy as np from ._recovery import _grid_identity, _GridLedger +from .cache import _bind_cache from .protocols import ( Annotation, Observable, @@ -27,6 +29,7 @@ from collections.abc import Callable from pathlib import Path + from .cache import EvaluationCache, _BoundCache from .design import Factor ProgressCallback = Callable[[int, int, TrialResult], None] @@ -87,21 +90,53 @@ def _attempt_grid_task( *, supports_rep: bool, ledger: _GridLedger | None, + cache: _BoundCache | None, ) -> TrialResult | Exception: design_point, config, rep = task + cache_key = cache.key(config, rep) if cache is not None else None + evaluation_config = deepcopy(config) if cache is not None else config try: - result = _run_single(world, scorer, config, rep=rep, supports_rep=supports_rep) + result = _run_single( + world, scorer, evaluation_config, rep=rep, supports_rep=supports_rep + ) # User evaluators may raise any Exception; the caller retries or re-raises it. except Exception as error: # ruff: ignore[blind-except] if ledger is not None: ledger.record(design_point, rep, None, f"{type(error).__name__}: {error}") return error result.metadata["design_point"] = design_point + if cache is not None: + if cache.key(result.config, rep) != cache_key: + msg = "Cached evaluations must not mutate their configurations" + raise ValueError(msg) + cache.save(config, rep, result) if ledger is not None: ledger.record(design_point, rep, result, None) return result +def _recover_grid_task( + task: _GridTask, + ledger: _GridLedger | None, + cache: _BoundCache | None, +) -> TrialResult | None: + design_point, config, rep = task + if ledger is not None: + recovered = ledger.load(design_point, rep, config) + if recovered is not None: + if cache is not None: + cache.save(config, rep, recovered) + return recovered + if cache is not None: + cached = cache.load(config, rep) + if cached is not None: + cached.metadata["design_point"] = design_point + if ledger is not None: + ledger.record(design_point, rep, cached, None) + return cached + return None + + def _run_grid_task( world: Simulator, scorer: Scorer, @@ -110,15 +145,14 @@ def _run_grid_task( supports_rep: bool, ledger: _GridLedger | None, max_retries: int, + cache: _BoundCache | None, ) -> TrialResult: - design_point, config, rep = task - if ledger is not None: - recovered = ledger.load(design_point, rep, config) - if recovered is not None: - return recovered + recovered = _recover_grid_task(task, ledger, cache) + if recovered is not None: + return recovered for attempt in range(max_retries + 1): result = _attempt_grid_task( - world, scorer, task, supports_rep=supports_rep, ledger=ledger + world, scorer, task, supports_rep=supports_rep, ledger=ledger, cache=cache ) if isinstance(result, Exception): if attempt == max_retries: @@ -144,6 +178,8 @@ def run_grid( # ruff: ignore[too-many-arguments] checkpoint_path: str | Path | None = None, checkpoint_key: str | None = None, max_retries: int = 0, + cache: EvaluationCache | None = None, + cache_bypass: bool = False, ) -> ResultsTable: """Run all configurations in a grid. @@ -177,6 +213,10 @@ def run_grid( # ruff: ignore[too-many-arguments] max_retries: Additional attempts per unfinished task per invocation. Zero (default) propagates the first evaluation exception. Retries retain the same config and replicate id. + cache: Opt-in revision/replicate/fidelity/schema-aware evaluation reuse. + Independent experiments must use different replicate namespaces. + cache_bypass: Skip cache reads and writes. Completed checkpoint tasks + still resume; use a new checkpoint path to re-evaluate them. Returns: ResultsTable with scored results. @@ -203,6 +243,11 @@ def run_grid( # ruff: ignore[too-many-arguments] ) supports_rep = _generate_accepts_rep(world) + bound_cache = ( + _bind_cache(cache, world, scorer, observables, annotations) + if cache is not None and not cache_bypass + else None + ) tasks = [ (design_point, cfg, rep) for design_point, cfg in enumerate(grid) @@ -220,6 +265,7 @@ def run_grid( # ruff: ignore[too-many-arguments] supports_rep=supports_rep, ledger=ledger, max_retries=max_retries, + cache=bound_cache, ) results.append(r) if callback is not None: @@ -235,6 +281,7 @@ def run_grid( # ruff: ignore[too-many-arguments] supports_rep=supports_rep, ledger=ledger, max_retries=max_retries, + cache=bound_cache, ) for task in tasks ) diff --git a/tests/test_cache.py b/tests/test_cache.py new file mode 100644 index 0000000..98353ca --- /dev/null +++ b/tests/test_cache.py @@ -0,0 +1,211 @@ +"""Evaluation reuse identities, invalidation, bypass, and independent replicates.""" + +from __future__ import annotations + +import sqlite3 +from contextlib import closing +from enum import Enum +from typing import TYPE_CHECKING, Any + +import numpy as np +import pytest + +from trade_study import Annotation, Direction, EvaluationCache, Observable, run_grid + +if TYPE_CHECKING: + from pathlib import Path + +_OBS = [Observable("loss", Direction.MINIMIZE)] +_GRID = [{"x": 0.5, "mode": "a"}, {"x": 1.0, "mode": "b"}] + + +class _World: + def __init__(self, offset: float = 0) -> None: + self.offset = offset + self.calls: list[tuple[float, int]] = [] + + def generate(self, config: dict[str, Any], *, rep: int = 0) -> tuple[float, float]: + self.calls.append((config["x"], rep)) + return config["x"], config["x"] + rep + self.offset + + +class _Scorer: + @staticmethod + def score(truth: float, observations: float, config: dict[str, Any]) -> dict: + del truth, config + return {"loss": observations} + + +def _cache(path: Path, **changes: str) -> EvaluationCache: + options = { + "revision": "model-v1-scorer-v1", + "replicate_namespace": "seed-17", + "fidelity": "high", + } + options.update(changes) + return EvaluationCache(path, **options) + + +def test_reordered_grid_and_increased_replication_reuse_exact_identities( + tmp_path: Path, +) -> None: + path = tmp_path / "cache.sqlite" + world = _World() + first = run_grid(world, _Scorer(), _GRID, _OBS, n_reps=2, cache=_cache(path)) + assert len(world.calls) == 4 + assert all(not m["cache_hit"] for m in first.metadata) + world.calls.clear() + reversed_grid = [dict(reversed(list(cfg.items()))) for cfg in _GRID[::-1]] + second = run_grid( + world, _Scorer(), reversed_grid, _OBS, n_reps=3, cache=_cache(path) + ) + assert world.calls == [(1.0, 2), (0.5, 2)] + assert [m["cache_hit"] for m in second.metadata] == [True, True, False] * 2 + assert [m["design_point"] for m in second.metadata] == [0, 0, 0, 1, 1, 1] + assert second.metadata[0]["cache_context"]["replicate_namespace"] == "seed-17" + np.testing.assert_array_equal(second.scores[:, 0], [1, 2, 3, 0.5, 1.5, 2.5]) + + +@pytest.mark.parametrize("change", ["revision", "replicate_namespace", "fidelity"]) +def test_behavior_or_independent_replication_changes_never_hit_old_entries( + tmp_path: Path, change: str +) -> None: + path = tmp_path / "context.sqlite" + world = _World() + run_grid(world, _Scorer(), _GRID, _OBS, cache=_cache(path)) + world.calls.clear() + result = run_grid( + world, _Scorer(), _GRID, _OBS, cache=_cache(path, **{change: "different"}) + ) + assert len(world.calls) == 2 + assert all(not m["cache_hit"] for m in result.metadata) + + +def test_objective_and_annotation_semantics_are_part_of_the_key(tmp_path: Path) -> None: + cache = _cache(tmp_path / "schema.sqlite") + world = _World() + run_grid(world, _Scorer(), _GRID, _OBS, cache=cache) + altered_obs = [Observable("loss", Direction.MAXIMIZE, weight=2)] + result = run_grid(world, _Scorer(), _GRID, altered_obs, cache=cache) + assert all(not m["cache_hit"] for m in result.metadata) + costs = [Annotation("cost", {"a": 1, "b": 2}, "mode")] + first = run_grid(world, _Scorer(), _GRID, _OBS, cache=cache, annotations=costs) + costs[0].lookup["a"] = 3 + second = run_grid(world, _Scorer(), _GRID, _OBS, cache=cache, annotations=costs) + assert first.metadata[0]["cache_key"] != second.metadata[0]["cache_key"] + assert all(not m["cache_hit"] for m in second.metadata) + assert second.annotations is not None + assert second.annotations[0, 0] == 3 + + +def test_bypass_does_not_replace_evidence_and_clear_invalidates(tmp_path: Path) -> None: + cache = _cache(tmp_path / "bypass.sqlite") + world = _World() + first = run_grid(world, _Scorer(), _GRID, _OBS, cache=cache) + world.offset = 10 + fresh = run_grid(world, _Scorer(), _GRID, _OBS, cache=cache, cache_bypass=True) + np.testing.assert_array_equal(fresh.scores, first.scores + 10) + assert all("cache_hit" not in m for m in fresh.metadata) + cached = run_grid(world, _Scorer(), _GRID, _OBS, cache=cache) + np.testing.assert_array_equal(cached.scores, first.scores) + cache.clear() + new = run_grid(world, _Scorer(), _GRID, _OBS, cache=cache) + np.testing.assert_array_equal(new.scores, fresh.scores) + + +def test_completed_checkpoint_can_populate_cache_and_retains_identity( + tmp_path: Path, +) -> None: + path = tmp_path / "run.sqlite" + cache = _cache(tmp_path / "evidence.sqlite") + world = _World() + run_grid(world, _Scorer(), _GRID, _OBS, checkpoint_path=path) + world.calls.clear() + recovered = run_grid( + world, _Scorer(), _GRID, _OBS, checkpoint_path=path, cache=cache + ) + assert world.calls == [] + assert all(m["recovered"] for m in recovered.metadata) + result = run_grid(world, _Scorer(), _GRID[::-1], _OBS, cache=cache) + assert world.calls == [] + assert all(m["cache_hit"] for m in result.metadata) + + +def test_parallel_workers_reuse_persistent_entries(tmp_path: Path) -> None: + cache = _cache(tmp_path / "parallel.sqlite") + first = run_grid(_World(), _Scorer(), _GRID, _OBS, cache=cache, n_jobs=2) + second = run_grid(_World(), _Scorer(), _GRID, _OBS, cache=cache, n_jobs=2) + np.testing.assert_array_equal(first.scores, second.scores) + assert all(m["cache_hit"] for m in second.metadata) + + +class _Mode(Enum): + A = "a" + + +def test_typed_canonicalization_does_not_collapse_container_or_key_types( + tmp_path: Path, +) -> None: + cache = _cache(tmp_path / "types.sqlite") + configs = [ + {"x": 0.5, "extra": {1: "value"}}, + {"x": 0.5, "extra": {"1": "value"}}, + {"x": 0.5, "extra": [1]}, + {"x": 0.5, "extra": (1,)}, + {"x": 0.5, "extra": np.int64(1)}, + {"x": 0.5, "extra": _Mode.A}, + ] + world = _World() + first = run_grid(world, _Scorer(), configs, _OBS, cache=cache) + assert len({m["cache_key"] for m in first.metadata}) == len(configs) + world.calls.clear() + run_grid(world, _Scorer(), configs, _OBS, cache=cache) + assert world.calls == [] + with pytest.raises(ValueError, match="supported JSON/scalar"): + run_grid(world, _Scorer(), [{"x": 0.5, "opaque": object()}], _OBS, cache=cache) + + +class _MutatingWorld(_World): + def generate(self, config: dict[str, Any], *, rep: int = 0) -> tuple[float, float]: + config["x"] += 1 + return super().generate(config, rep=rep) + + +def test_mutating_evaluator_cannot_cache_under_the_wrong_configuration( + tmp_path: Path, +) -> None: + grid = [{"x": 0.5}] + with pytest.raises(ValueError, match="must not mutate"): + run_grid( + _MutatingWorld(), + _Scorer(), + grid, + _OBS, + cache=_cache(tmp_path / "mutation.sqlite"), + ) + assert grid == [{"x": 0.5}] + + +def test_invalid_identities_and_cache_format_are_rejected(tmp_path: Path) -> None: + path = tmp_path / "invalid.sqlite" + with pytest.raises(ValueError, match="nonempty"): + _cache(path, revision=" ") + _cache(path) + with closing(sqlite3.connect(path)) as connection, connection: + connection.execute("UPDATE cache_format SET version = 999") + with pytest.raises(ValueError, match="Incompatible evaluation cache format"): + _cache(path) + + +def test_conflicting_checkpoint_evidence_never_overwrites_cache(tmp_path: Path) -> None: + checkpoint = tmp_path / "completed.sqlite" + cache = _cache(tmp_path / "conflict.sqlite") + run_grid(_World(), _Scorer(), _GRID, _OBS, checkpoint_path=checkpoint) + # Intentionally lie about the revision to exercise conflict detection. + newer = run_grid(_World(offset=10), _Scorer(), _GRID, _OBS, cache=cache) + with pytest.raises(ValueError, match="Conflicting scores"): + run_grid( + _World(), _Scorer(), _GRID, _OBS, checkpoint_path=checkpoint, cache=cache + ) + retained = run_grid(_World(), _Scorer(), _GRID, _OBS, cache=cache) + np.testing.assert_array_equal(retained.scores, newer.scores)