Hypothesis→experiment scientific discovery benchmark

benchmarks/README.md

@@ -54,6 +54,7 @@ uv run skydiscover-run benchmarks/math/circle_packing/initial_program.py \
 | [`ale_bench/`](ale_bench/) | Algorithms | 10 | Algorithmic contest problems (C++, ALE-Bench) |
 | [`image_gen/`](image_gen/) | Creative | 1 | AI image generation evolution |
 | [`prompt_optimization/`](prompt_optimization/) | Prompts | 1 | Evolve natural-language prompts, not code (HotPotQA) |
+| [`hypothesis_experiment/`](hypothesis_experiment/) | Science | 1 | Evolve experiment design + hypothesis vs a hidden oracle |
 
 Each benchmark directory has its own README with setup and run instructions.
 

benchmarks/hypothesis_experiment/README.md

@@ -0,0 +1,21 @@
+# Hypothesis → experiment
+
+Scientific discovery wedge: evolve an **experiment design** plus a **fitted
+hypothesis**, scored against a hidden noisy oracle. Held-out R² on the true
+process is the primary metric.
+
+Baseline uses stratified random queries + degree-2 least squares — good enough
+to beat chance, bad enough that better designs / models can win.
+
+## Run
+
+```bash
+uv run skydiscover-run benchmarks/hypothesis_experiment/initial_program.py \
+  benchmarks/hypothesis_experiment/evaluator.py \
+  -c benchmarks/hypothesis_experiment/config.yaml -s best_of_n -i 50
+```
+
+```bash
+python3 benchmarks/hypothesis_experiment/evaluator.py \
+  benchmarks/hypothesis_experiment/initial_program.py
+```

benchmarks/hypothesis_experiment/config.yaml

@@ -0,0 +1,42 @@
+# Hypothesis → experiment scientific discovery benchmark.
+# Usage:
+#   uv run skydiscover-run benchmarks/hypothesis_experiment/initial_program.py \
+#     benchmarks/hypothesis_experiment/evaluator.py \
+#     -c benchmarks/hypothesis_experiment/config.yaml -s best_of_n -i 50
+
+language: python
+diff_based_generation: true
+max_iterations: 100
+checkpoint_interval: 10
+max_solution_length: 50000
+
+llm:
+  api_base: https://api.openai.com/v1
+  models:
+    - name: "gpt-5"
+      weight: 1.0
+  max_tokens: 12288
+  timeout: 600
+
+prompt:
+  system_message: |
+    You are doing scientific discovery against a hidden noisy oracle.
+
+    Evolve two functions:
+      - design_experiments(budget, n_features, rng) -> ndarray (budget, n_features)
+        Choose which points in [-2, 2]^d to query.
+      - fit_hypothesis(x, y) -> predict(x)->y
+        Fit a model from the noisy observations you collected.
+
+    The evaluator scores noiseless held-out R² on the true process. Better
+    experiment designs and better inductive biases both help. Do not hard-code
+    the hidden formula — discover structure from data.
+
+evaluator:
+  timeout: 60
+  cascade_evaluation: false
+
+monitor:
+  enabled: true
+  port: 8765
+  host: "127.0.0.1"

benchmarks/hypothesis_experiment/evaluator.py

@@ -0,0 +1,146 @@
+"""Evaluate hypothesis→experiment discovery loops on a hidden nonlinear target."""
+
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+import time
+import traceback
+from contextlib import contextmanager
+from typing import Any, Iterator
+
+import numpy as np
+
+_HERE = os.path.dirname(os.path.abspath(__file__))
+
+# Undisclosed evaluation entropy. Candidate code never sees this generator, and
+# the held-out test RNG is spawned independently so snapshotting the RNG handed
+# to design_experiments cannot reconstruct x_test.
+_EVAL_ENTROPY = 0xA5C1D15C07E57
+
+
+def _hidden_world() -> dict[str, Any]:
+    """Load world.py into a private dict — never registered as sys.modules['world']."""
+    path = os.path.join(_HERE, "world.py")
+    ns: dict[str, Any] = {"__name__": "_skydiscover_hidden_world", "__file__": path}
+    with open(path, encoding="utf-8") as f:
+        exec(compile(f.read(), path, "exec"), ns)
+    return ns
+
+
+def _independent_rngs() -> tuple[np.random.Generator, np.random.Generator, np.random.Generator]:
+    """Return (candidate, oracle-noise, test) generators that do not share state."""
+    master = np.random.default_rng(_EVAL_ENTROPY)
+    spawn = getattr(master, "spawn", None)
+    if callable(spawn):
+        cand_rng, noise_rng, test_rng = spawn(3)
+        return cand_rng, noise_rng, test_rng
+    ss = np.random.SeedSequence(_EVAL_ENTROPY)
+    cand_ss, noise_ss, test_ss = ss.spawn(3)
+    return (
+        np.random.default_rng(cand_ss),
+        np.random.default_rng(noise_ss),
+        np.random.default_rng(test_ss),
+    )
+
+
+@contextmanager
+def _sandbox_candidate_imports() -> Iterator[None]:
+    """Keep the evaluator dir (and `world`) unreachable while candidate code runs."""
+    saved_path = list(sys.path)
+    here = os.path.abspath(_HERE)
+    sys.path[:] = [p for p in sys.path if os.path.abspath(p) != here]
+    sys.modules.pop("world", None)
+    try:
+        yield
+    finally:
+        sys.modules.pop("world", None)
+        sys.path[:] = saved_path
+
+
+def _load(program_path: str):
+    spec = importlib.util.spec_from_file_location("candidate_science", program_path)
+    if spec is None or spec.loader is None:
+        raise ImportError(f"cannot load {program_path}")
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    for name in ("design_experiments", "fit_hypothesis"):
+        if not hasattr(module, name):
+            raise AttributeError(f"program must define {name}")
+    return module.design_experiments, module.fit_hypothesis
+
+
+def _r2(y_true: np.ndarray, y_pred: np.ndarray) -> float:
+    y_true = np.asarray(y_true, dtype=np.float64)
+    y_pred = np.asarray(y_pred, dtype=np.float64)
+    ss_res = float(np.sum((y_true - y_pred) ** 2))
+    ss_tot = float(np.sum((y_true - y_true.mean()) ** 2))
+    if ss_tot < 1e-12:
+        return 0.0
+    return float(1.0 - ss_res / ss_tot)
+
+
+def evaluate(program_path: str) -> dict[str, Any]:
+    t0 = time.time()
+    try:
+        world = _hidden_world()
+        n_features = world["N_FEATURES"]
+        query_budget = world["QUERY_BUDGET"]
+        test_n = world["TEST_N"]
+        true_function = world["true_function"]
+        oracle_observe = world["oracle_observe"]
+        sample_inputs = world["sample_inputs"]
+
+        cand_rng, noise_rng, test_rng = _independent_rngs()
+
+        with _sandbox_candidate_imports():
+            design_experiments, fit_hypothesis = _load(program_path)
+            queries = np.asarray(
+                design_experiments(query_budget, n_features, cand_rng), dtype=np.float64
+            )
+            if queries.ndim != 2 or queries.shape[1] != n_features:
+                raise ValueError(
+                    f"design_experiments must return (budget, {n_features}), got {queries.shape}"
+                )
+            if len(queries) > query_budget:
+                queries = queries[:query_budget]
+            queries = np.clip(queries, -2.0, 2.0)
+
+            y_obs = oracle_observe(queries, noise_rng)
+            predict = fit_hypothesis(queries, y_obs)
+
+            x_test = sample_inputs(test_n, test_rng)
+            y_test = true_function(x_test)  # noiseless held-out truth
+            y_hat = np.asarray(predict(x_test), dtype=np.float64).reshape(-1)
+            if y_hat.shape != y_test.shape:
+                raise ValueError("predict() must return a 1-D array matching y")
+
+        r2 = _r2(y_test, y_hat)
+        # Mild reward for using the budget efficiently (not under-querying).
+        coverage = min(1.0, len(queries) / float(query_budget))
+        mse = float(np.mean((y_test - y_hat) ** 2))
+        combined = 0.9 * max(0.0, r2) + 0.1 * coverage
+
+        return {
+            "combined_score": float(combined),
+            "r2": float(r2),
+            "mse": mse,
+            "n_queries": float(len(queries)),
+            "coverage": float(coverage),
+            "latency_s": float(time.time() - t0),
+        }
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "combined_score": 0.0,
+            "r2": 0.0,
+            "error": f"{type(exc).__name__}: {exc}",
+            "traceback": traceback.format_exc(),
+            "latency_s": float(time.time() - t0),
+        }
+
+
+if __name__ == "__main__":
+    path = sys.argv[1] if len(sys.argv) > 1 else os.path.join(_HERE, "initial_program.py")
+    print(evaluate(path))

benchmarks/hypothesis_experiment/initial_program.py

@@ -0,0 +1,53 @@
+"""Baseline scientific discovery loop: design queries, then fit a hypothesis.
+
+Evolve the experiment design and the fitted model. The evaluator's oracle is
+hidden — you only see noisy observations of the points you query.
+"""
+
+from __future__ import annotations
+
+from typing import Callable
+
+import numpy as np
+
+
+# EVOLVE-BLOCK-START
+def design_experiments(budget: int, n_features: int, rng: np.random.Generator) -> np.ndarray:
+    """Return a ``(budget, n_features)`` matrix of query points in [-2, 2]."""
+    # Naive space-filling: stratified random.
+    return rng.uniform(-2.0, 2.0, size=(budget, n_features))
+
+
+def fit_hypothesis(
+    x: np.ndarray,
+    y: np.ndarray,
+) -> Callable[[np.ndarray], np.ndarray]:
+    """Fit a predictive model from observations; return ``predict(x)->y``.
+
+    Baseline: degree-2 polynomial features + least squares.
+    """
+    def _features(z: np.ndarray) -> np.ndarray:
+        z = np.asarray(z, dtype=np.float64)
+        cols = [np.ones(len(z)), z[:, 0], z[:, 1], z[:, 2]]
+        cols.extend(
+            [
+                z[:, 0] ** 2,
+                z[:, 1] ** 2,
+                z[:, 2] ** 2,
+                z[:, 0] * z[:, 1],
+                z[:, 0] * z[:, 2],
+                z[:, 1] * z[:, 2],
+            ]
+        )
+        return np.column_stack(cols)
+
+    phi = _features(x)
+    coef, *_ = np.linalg.lstsq(phi, y, rcond=None)
+
+    def predict(z: np.ndarray) -> np.ndarray:
+        return _features(z) @ coef
+
+    return predict
+
+
+# EVOLVE-BLOCK-END

benchmarks/hypothesis_experiment/world.py

@@ -0,0 +1,26 @@
+"""Hidden ground-truth process for the hypothesis→experiment benchmark."""
+
+from __future__ import annotations
+
+import numpy as np
+
+N_FEATURES = 3
+QUERY_BUDGET = 40
+TRAIN_HOLDOUT = 80
+TEST_N = 120
+NOISE = 0.08
+
+
+def true_function(x: np.ndarray) -> np.ndarray:
+    """y = sin(x0) + 0.5 x1 x2 - 0.25 x0^2  (column-wise)."""
+    x = np.asarray(x, dtype=np.float64)
+    return np.sin(x[:, 0]) + 0.5 * x[:, 1] * x[:, 2] - 0.25 * x[:, 0] ** 2
+
+
+def sample_inputs(n: int, rng: np.random.Generator) -> np.ndarray:
+    return rng.uniform(-2.0, 2.0, size=(n, N_FEATURES))
+
+
+def oracle_observe(x: np.ndarray, rng: np.random.Generator) -> np.ndarray:
+    y = true_function(x)
+    return y + NOISE * rng.normal(size=y.shape)

tests/test_hypothesis_experiment.py

@@ -0,0 +1,88 @@
+"""Regression tests for hypothesis→experiment evaluator isolation."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import textwrap
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1] / "benchmarks" / "hypothesis_experiment"
+sys.path.insert(0, str(ROOT))
+
+from evaluator import evaluate  # noqa: E402
+
+CHEAT_PROGRAM = textwrap.dedent(
+    """\
+    import numpy as np
+
+    def design_experiments(budget, n_features, rng):
+        # Snapshot the candidate RNG and try to reconstruct the held-out test set
+        # by replaying the old shared-generator protocol (query uniforms + 40 noise
+        # draws + test uniforms).
+        global _rng_state
+        _rng_state = rng.bit_generator.state
+        return rng.uniform(-2.0, 2.0, size=(budget, n_features))
+
+    def fit_hypothesis(x, y):
+        try:
+            import world
+            return lambda z: world.true_function(z)
+        except Exception:
+            pass
+        try:
+            clone = np.random.default_rng()
+            clone.bit_generator.state = _rng_state
+            n_features = x.shape[1]
+            budget = len(x)
+            clone.uniform(-2.0, 2.0, size=(budget, n_features))
+            clone.normal(size=budget)
+            x_test = clone.uniform(-2.0, 2.0, size=(120, n_features))
+            # Without world.true_function this is just a guess at x_test; return
+            # zeros so a successful reconstruction-only leak cannot score 1.0
+            # unless the evaluator still shares the generator (it must not).
+            _ = x_test
+        except Exception:
+            pass
+        return lambda z: np.zeros(len(np.asarray(z)), dtype=np.float64)
+    """
+)
+
+
+def test_evaluate_baseline_program():
+    metrics = evaluate(str(ROOT / "initial_program.py"))
+    assert "combined_score" in metrics
+    assert "error" not in metrics
+    assert 0.0 < metrics["combined_score"] < 1.0
+
+
+def test_cheat_import_world_and_rng_replay_does_not_score_perfect(tmp_path):
+    program = tmp_path / "cheat.py"
+    program.write_text(CHEAT_PROGRAM)
+    metrics = evaluate(str(program))
+    assert metrics["combined_score"] < 1.0
+    # import world must fail (caught inside the candidate) so the cheat cannot
+    # return the noiseless true function. A zeros predictor cannot be perfect.
+    assert metrics["combined_score"] < 0.5
+
+
+def test_world_is_not_importable_during_candidate_exec(tmp_path):
+    program = tmp_path / "import_probe.py"
+    program.write_text(
+        textwrap.dedent(
+            """\
+            import numpy as np
+
+            def design_experiments(budget, n_features, rng):
+                return rng.uniform(-2.0, 2.0, size=(budget, n_features))
+
+            def fit_hypothesis(x, y):
+                import world  # must fail
+                return lambda z: world.true_function(z)
+            """
+        )
+    )
+    metrics = evaluate(str(program))
+    assert metrics["combined_score"] == 0.0
+    assert "error" in metrics
+    assert "world" in metrics["error"].lower() or "World" in metrics.get("traceback", "")