test(robustness): 55-test battery — primitives, contract, suites, gates, no-interference

Coverage matrix:
- test_robustness_primitives.py  (18) — CPCV shape/embargo/purge, PBO
  bounds on pure-noise vs signal families, PSR high/zero/degenerate,
  null audit shape/determinism/validation, jitter anchor-recovery,
  jitter name-in-anchor + negative-fraction error paths.
- test_kuramoto_contract.py      (6)  — 28-hash verification, missing
  manifest fail-closed, sha256 mismatch fail-closed, missing-file
  fail-closed, schema-consistency assertions, daily_returns shape.
- test_kuramoto_candidate_set.py (5)  — legit names accepted, each
  forbidden prefix rejected, multi-offender listing, anchor-cover.
- test_kuramoto_suites.py        (10) — CPCV pbo bounds + fold count,
  null two-family shape + determinism + invalid-bootstrap error,
  jitter mode + anchor + forbidden-rejection + monotonicity.
- test_kuramoto_gate_runner.py   (12) — decision-layer PASS/FAIL/
  INSUFFICIENT truth table + end-to-end pipeline on frozen bundle.
- test_kuramoto_no_interference.py (4) — AST + regex scan asserting
  no writes under shadow_validation/, demo/, core/cross_asset_kuramoto/,
  etc.; all result-path literals route to robustness_v1/ or a frozen
  read-only input; no imports from execution/strategies/paper_trader.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: neuron7xLab

tests/research/robustness/__init__.py

@@ -0,0 +1,47 @@
+# Copyright (c) 2023-2026 Yaroslav Vasylenko (neuron7xLab)
+# SPDX-License-Identifier: MIT
+"""Candidate-set anti-inflation guard tests."""
+
+from __future__ import annotations
+
+import pytest
+
+from research.robustness.protocols.kuramoto_candidate_set import (
+    CandidateSetInflationError,
+    assert_anchor_covers_candidates,
+    validate_candidate_parameter_names,
+)
+
+
+def test_legit_names_accepted() -> None:
+    validate_candidate_parameter_names({"cost_bps": 0.1, "vol_target_annualised": 0.05})
+
+
+@pytest.mark.parametrize(
+    "bad",
+    [
+        "seed_value",
+        "random_noise_scale",
+        "jitter_extra",
+    ],
+)
+def test_forbidden_prefixes_rejected(bad: str) -> None:
+    with pytest.raises(CandidateSetInflationError):
+        validate_candidate_parameter_names({bad: 0.1})
+
+
+def test_multiple_offenders_listed_together() -> None:
+    with pytest.raises(CandidateSetInflationError) as exc:
+        validate_candidate_parameter_names({"seed_a": 0.1, "random_b": 0.1, "cost_bps": 0.2})
+    msg = str(exc.value)
+    assert "seed_a" in msg
+    assert "random_b" in msg
+    assert "cost_bps" not in msg
+
+
+def test_missing_anchor_key_rejected() -> None:
+    with pytest.raises(CandidateSetInflationError):
+        assert_anchor_covers_candidates(
+            {"cost_bps": 1.0},
+            {"nonexistent": 0.1},
+        )

tests/research/robustness/test_kuramoto_candidate_set.py

@@ -0,0 +1,116 @@
+# Copyright (c) 2023-2026 Yaroslav Vasylenko (neuron7xLab)
+# SPDX-License-Identifier: MIT
+"""Contract loader tests — hash verification is fail-closed."""
+
+from __future__ import annotations
+
+import json
+import shutil
+from pathlib import Path
+
+import pytest
+
+from research.robustness.protocols.kuramoto_contract import (
+    FrozenArtifactManifest,
+    FrozenArtifactMismatch,
+    KuramotoRobustnessContract,
+)
+
+REPO = Path(__file__).resolve().parents[3]
+
+
+class TestFrozenArtifactManifest:
+    def test_loads_real_manifest(self) -> None:
+        manifest = FrozenArtifactManifest.load()
+        assert manifest.generated_utc
+        assert len(manifest.hashes) == 28
+
+    def test_missing_manifest_raises(self, tmp_path: Path) -> None:
+        with pytest.raises(FrozenArtifactMismatch):
+            FrozenArtifactManifest.load(tmp_path / "nope.json")
+
+    def test_hash_mismatch_raises_fail_closed(self, tmp_path: Path) -> None:
+        # Snapshot one real artifact into tmp_path, bump its hash in the
+        # manifest copy, point the manifest's repo_root at tmp_path and
+        # verify_all() must raise.
+        src_file = REPO / "results" / "cross_asset_kuramoto" / "demo" / "DEMO_BRIEF.md"
+        rel = "results/cross_asset_kuramoto/demo/DEMO_BRIEF.md"
+        dst_file = tmp_path / rel
+        dst_file.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy(src_file, dst_file)
+
+        manifest = FrozenArtifactManifest(
+            generated_utc="2026-01-01T00:00:00Z",
+            regenerated_utc=None,
+            hashes={rel: "0" * 64},
+            repo_root=tmp_path,
+        )
+        with pytest.raises(FrozenArtifactMismatch) as exc:
+            manifest.verify_all()
+        assert "sha256 mismatch" in str(exc.value)
+
+    def test_missing_file_reports_missing(self, tmp_path: Path) -> None:
+        manifest = FrozenArtifactManifest(
+            generated_utc="2026-01-01T00:00:00Z",
+            regenerated_utc=None,
+            hashes={"does/not/exist.txt": "0" * 64},
+            repo_root=tmp_path,
+        )
+        with pytest.raises(FrozenArtifactMismatch) as exc:
+            manifest.verify_all()
+        assert "missing" in str(exc.value)
+
+
+class TestKuramotoRobustnessContract:
+    def test_from_frozen_artifacts_loads_real_bundle(self) -> None:
+        c = KuramotoRobustnessContract.from_frozen_artifacts()
+        assert len(c.equity_curve) > 1000
+        assert len(c.fold_metrics) >= 2
+        assert {"sharpe", "ann_vol"} <= set(c.risk_metrics.columns)
+        assert c.parameter_lock["seed"] == 42
+
+    def test_daily_returns_have_expected_length(self) -> None:
+        c = KuramotoRobustnessContract.from_frozen_artifacts()
+        r = c.daily_strategy_returns()
+        assert len(r) == len(c.equity_curve) - 1
+
+    def test_fold_metrics_contract_violation_is_caught(self, tmp_path: Path) -> None:
+        # Build a manifest whose frozen fold_metrics.csv is missing the
+        # 'sharpe' column — assert_frozen_consistency must refuse it.
+        rel_equity = "results/cross_asset_kuramoto/demo/equity_curve.csv"
+        rel_folds = "results/cross_asset_kuramoto/demo/fold_metrics.csv"
+        rel_risk = "results/cross_asset_kuramoto/demo/risk_metrics.csv"
+        rel_lock = "results/cross_asset_kuramoto/PARAMETER_LOCK.json"
+
+        (tmp_path / "results" / "cross_asset_kuramoto" / "demo").mkdir(parents=True, exist_ok=True)
+        (tmp_path / rel_equity).write_text(
+            "date,strategy_cumret,benchmark_cumret,drawdown\n"
+            "2020-01-01,1.0,1.0,-0.0\n2020-01-02,1.01,1.0,-0.0\n"
+            "2020-01-03,1.02,1.0,-0.0\n"
+        )
+        (tmp_path / rel_folds).write_text(
+            "fold_id,is_start,os_start,os_end,NOT_SHARPE,max_dd,pass_fail\n"
+            "1,2020-01-01,2020-02-01,2020-03-01,1.0,-0.1,PASS\n"
+            "2,2020-01-01,2020-03-01,2020-04-01,0.5,-0.1,PASS\n"
+        )
+        (tmp_path / rel_risk).write_text("sharpe,ann_return,ann_vol,max_dd\n1.0,0.1,0.1,-0.1\n")
+        (tmp_path / rel_lock).write_text(json.dumps({"seed": 42}))
+        manifest_path = tmp_path / "manifest.json"
+        import hashlib
+
+        hashes = {
+            rel: hashlib.sha256((tmp_path / rel).read_bytes()).hexdigest()
+            for rel in (rel_equity, rel_folds, rel_risk, rel_lock)
+        }
+        manifest_path.write_text(
+            json.dumps(
+                {"generated_utc": "x", "hashes": hashes},
+                indent=2,
+            )
+        )
+        with pytest.raises(FrozenArtifactMismatch) as exc:
+            KuramotoRobustnessContract.from_frozen_artifacts(
+                manifest_path=manifest_path,
+                repo_root=tmp_path,
+            )
+        assert "fold_metrics" in str(exc.value)

tests/research/robustness/test_kuramoto_contract.py

@@ -0,0 +1,163 @@
+# Copyright (c) 2023-2026 Yaroslav Vasylenko (neuron7xLab)
+# SPDX-License-Identifier: MIT
+"""End-to-end gate-runner + decision-layer tests."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+from backtest.robustness_gates import DecisionLabel, evaluate_robustness_gates
+from research.robustness.protocols.kuramoto_contract import KuramotoRobustnessContract
+from research.robustness.protocols.kuramoto_gate_runner import run_kuramoto_gate_runner
+
+
+@dataclass(frozen=True)
+class _FakeCPCV:
+    pbo_pass: bool
+    psr_pass: bool
+    annualised_sharpe: float
+    n_folds: int
+
+
+@dataclass(frozen=True)
+class _FakeNull:
+    all_families_pass: bool
+
+
+@dataclass(frozen=True)
+class _FakeJitter:
+    evaluator_mode: str
+    fraction_within_tol_pass: bool
+
+
+@dataclass(frozen=True)
+class _FakeEvidence:
+    cpcv: _FakeCPCV
+    null: _FakeNull
+    jitter: _FakeJitter
+
+
+class TestDecisionLayer:
+    def _ev(
+        self,
+        *,
+        pbo_pass: bool = True,
+        psr_pass: bool = True,
+        annualised_sharpe: float = 1.2,
+        n_folds: int = 5,
+        null_pass: bool = True,
+        jitter_mode: str = "LIVE",
+        jitter_pass: bool = True,
+    ) -> _FakeEvidence:
+        return _FakeEvidence(
+            cpcv=_FakeCPCV(
+                pbo_pass=pbo_pass,
+                psr_pass=psr_pass,
+                annualised_sharpe=annualised_sharpe,
+                n_folds=n_folds,
+            ),
+            null=_FakeNull(all_families_pass=null_pass),
+            jitter=_FakeJitter(
+                evaluator_mode=jitter_mode,
+                fraction_within_tol_pass=jitter_pass,
+            ),
+        )
+
+    def test_all_green_gives_pass(self) -> None:
+        r = evaluate_robustness_gates(self._ev())
+        assert r.label is DecisionLabel.PASS
+        assert r.reasons == ()
+
+    def test_pbo_red_gives_fail(self) -> None:
+        r = evaluate_robustness_gates(self._ev(pbo_pass=False))
+        assert r.label is DecisionLabel.FAIL
+        assert any("PBO" in reason for reason in r.reasons)
+
+    def test_psr_red_gives_fail(self) -> None:
+        r = evaluate_robustness_gates(self._ev(psr_pass=False))
+        assert r.label is DecisionLabel.FAIL
+        assert any("PSR" in reason for reason in r.reasons)
+
+    def test_null_red_gives_fail(self) -> None:
+        r = evaluate_robustness_gates(self._ev(null_pass=False))
+        assert r.label is DecisionLabel.FAIL
+
+    def test_placeholder_jitter_with_require_live_demotes_to_insufficient(
+        self,
+    ) -> None:
+        r = evaluate_robustness_gates(
+            self._ev(jitter_mode="PLACEHOLDER_APPROXIMATION"),
+            require_live_jitter=True,
+        )
+        assert r.label is DecisionLabel.INSUFFICIENT_EVIDENCE
+        assert r.jitter_is_placeholder
+
+    def test_placeholder_jitter_without_require_live_can_pass(self) -> None:
+        r = evaluate_robustness_gates(
+            self._ev(jitter_mode="PLACEHOLDER_APPROXIMATION"),
+            require_live_jitter=False,
+        )
+        assert r.label is DecisionLabel.PASS
+        assert r.jitter_is_placeholder
+
+    def test_single_fold_gives_insufficient(self) -> None:
+        r = evaluate_robustness_gates(self._ev(n_folds=1))
+        assert r.label is DecisionLabel.INSUFFICIENT_EVIDENCE
+
+    def test_live_jitter_failure_is_fail(self) -> None:
+        r = evaluate_robustness_gates(self._ev(jitter_mode="LIVE", jitter_pass=False))
+        assert r.label is DecisionLabel.FAIL
+
+
+class TestGateRunnerEndToEnd:
+    def test_pipeline_produces_all_three_suites(self) -> None:
+        contract = KuramotoRobustnessContract.from_frozen_artifacts()
+        evidence = run_kuramoto_gate_runner(
+            contract,
+            null_kwargs={"n_bootstrap": 32},
+            jitter_kwargs={"n_candidates": 8},
+        )
+        assert evidence.cpcv.n_folds >= 2
+        assert len(evidence.null.families) == 2
+        assert evidence.jitter.evaluator_mode == "PLACEHOLDER_APPROXIMATION"
+
+    def test_decision_matches_evidence(self) -> None:
+        contract = KuramotoRobustnessContract.from_frozen_artifacts()
+        evidence = run_kuramoto_gate_runner(
+            contract,
+            null_kwargs={"n_bootstrap": 32, "seed": 42},
+            jitter_kwargs={"n_candidates": 8},
+        )
+        decision = evaluate_robustness_gates(evidence)
+        # With only 32 bootstraps and proxy returns the null suite
+        # almost certainly fails at the 5 % threshold, so the runner
+        # must report a FAIL verdict on the frozen demo evidence.
+        assert decision.label is DecisionLabel.FAIL
+        assert not decision.null_pass
+        assert any("null" in reason for reason in decision.reasons)
+
+
+def test_cli_writes_expected_artifacts(tmp_path: Path) -> None:
+    from scripts import run_kuramoto_robustness_v1 as cli  # noqa: PLC0415
+
+    cwd_out = tmp_path / "robustness_v1"
+    rc = cli.main(
+        [
+            "--n-bootstrap",
+            "32",
+            "--n-jitter-candidates",
+            "8",
+            "--out-dir",
+            str(cwd_out),
+        ]
+    )
+    assert rc in (0, 1)
+    for name in (
+        "verdict.json",
+        "cpcv_summary.json",
+        "null_summary.json",
+        "jitter_summary.json",
+        "ROBUSTNESS_v1.md",
+    ):
+        assert (cwd_out / name).is_file()