Enforce strict latest theme linkage

src/research_signal_context_pipelines/__init__.py

@@ -3,7 +3,7 @@
 from .overlay_backtest import OverlayPolicy, backtest_overlay
 from .context_bundle import DEFAULT_UNIVERSE, build_context_bundle, build_context_from_source
 from .price_history import PriceExtractionSummary, write_filtered_price_history
-from .schema import SignalValidationError, validate_signal
+from .schema import SignalValidationError, validate_latest_signal, validate_signal
 from .theme_momentum import (
     build_theme_momentum_snapshot,
     validate_theme_momentum_snapshot,
@@ -24,6 +24,7 @@
     "load_symbol_theme_exposure",
     "load_theme_taxonomy",
     "validate_signal",
+    "validate_latest_signal",
     "validate_theme_momentum_snapshot",
     "write_filtered_price_history",
     "write_theme_momentum_snapshot",

src/research_signal_context_pipelines/schema.py

@@ -1,7 +1,11 @@
 from __future__ import annotations
 
 import datetime as dt
+import hashlib
+import json
+import re
 from collections.abc import Mapping, Sequence
+from pathlib import Path
 from typing import Any
 
 
@@ -152,3 +156,74 @@ def _validate_bias_value(value: Any, name: str) -> None:
                 _require_string_list(raw[optional_list_key], f"{name}.{optional_list_key}", allow_empty=True)
     if bias not in ALLOWED_BIAS_VALUES:
         raise SignalValidationError(f"{name} must be one of: {', '.join(sorted(ALLOWED_BIAS_VALUES))}")
+
+
+def validate_latest_signal(
+    payload: Mapping[str, Any],
+    *,
+    theme_artifact_path: str | Path | None = None,
+    signal_base_dir: str | Path | None = None,
+) -> None:
+    """Validate a v2 signal against its declared theme artifact linkage."""
+    validate_signal(payload)
+    if str(payload.get("schema_version")) != "2":
+        raise SignalValidationError("strict latest validation requires signal schema v2")
+
+    evidence = _require_mapping(payload["evidence"], "evidence")
+    sources = _require_string_list(evidence.get("sources"), "evidence.sources")
+    theme_sources = [source for source in sources if "theme_momentum_snapshot" in Path(source).name]
+    if len(theme_sources) != 1:
+        raise SignalValidationError("latest signal must declare exactly one theme source")
+    declared_source = theme_sources[0]
+    base_dir = Path(signal_base_dir).resolve() if signal_base_dir else None
+
+    def resolve_source(value: str | Path) -> Path:
+        path = Path(value)
+        if path.is_absolute():
+            return path.resolve()
+        if base_dir is None:
+            raise SignalValidationError("relative theme source requires signal_base_dir")
+        return (base_dir / path).resolve()
+
+    declared_path = resolve_source(declared_source)
+    artifact_path = resolve_source(theme_artifact_path) if theme_artifact_path else declared_path
+    if artifact_path != declared_path:
+        raise SignalValidationError("theme artifact override must match the declared theme source")
+    if not artifact_path.exists():
+        raise SignalValidationError(f"declared theme artifact not found: {artifact_path}")
+
+    source_hashes = evidence.get("source_hashes", {})
+    if source_hashes is None:
+        source_hashes = {}
+    if not isinstance(source_hashes, Mapping):
+        raise SignalValidationError("evidence.source_hashes must be an object")
+    if declared_source in source_hashes:
+        expected_hash = source_hashes[declared_source]
+    elif str(declared_path) in source_hashes:
+        expected_hash = source_hashes[str(declared_path)]
+    else:
+        expected_hash = None
+    if expected_hash is not None or declared_source in source_hashes or str(declared_path) in source_hashes:
+        expected_hash = _require_string(expected_hash, "evidence.source_hashes value")
+        if not re.fullmatch(r"[0-9a-fA-F]{64}", expected_hash):
+            raise SignalValidationError("evidence.source_hashes value must be a SHA-256 hex digest")
+        try:
+            digest = hashlib.sha256(artifact_path.read_bytes()).hexdigest()
+        except OSError as exc:
+            raise SignalValidationError(f"declared theme artifact unreadable: {artifact_path}") from exc
+        if digest != expected_hash.lower():
+            raise SignalValidationError("declared theme artifact sha256 does not match")
+
+    try:
+        theme_payload = json.loads(artifact_path.read_text(encoding="utf-8"))
+        from .theme_momentum import validate_theme_momentum_snapshot
+
+        validate_theme_momentum_snapshot(theme_payload)
+    except (OSError, json.JSONDecodeError, ValueError) as exc:
+        raise SignalValidationError(f"invalid declared theme artifact: {exc}") from exc
+    if str(theme_payload.get("schema_version")) != "2":
+        raise SignalValidationError("strict latest validation requires theme schema v2")
+    if str(theme_payload.get("as_of")) != str(payload["as_of"]):
+        raise SignalValidationError(
+            f"signal/theme as_of mismatch: signal={payload['as_of']} theme={theme_payload.get('as_of')}"
+        )

tests/test_signal_validation.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 import json
+import hashlib
 from pathlib import Path
 
 import pytest
 
-from research_signal_context_pipelines import SignalValidationError, validate_signal
+from research_signal_context_pipelines import SignalValidationError, validate_latest_signal, validate_signal
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -48,6 +49,87 @@ def test_v2_signal_requires_model_and_scoring_versions() -> None:
         validate_signal(payload)
 
 
+def _strict_signal(*, source: str, as_of: str = "2026-01-01", source_sha256: str | None = None) -> dict:
+    payload = load_example()
+    payload.update(
+        {
+            "schema_version": "2",
+            "as_of": as_of,
+            "model_version": "shadow-v2",
+            "scoring_version": "rules-v2",
+        }
+    )
+    payload["evidence"]["sources"] = [source]
+    if source_sha256:
+        payload["evidence"]["source_hashes"] = {source: source_sha256}
+    return payload
+
+
+def _theme_snapshot(*, as_of: str = "2026-01-01") -> dict:
+    return {
+        "schema_version": "2",
+        "as_of": as_of,
+        "generated_at": f"{as_of}T00:00:00Z",
+        "expires_at": "2026-12-31",
+        "model_version": "theme-v1",
+        "scoring_version": "rules-v1",
+        "mode": "theme_momentum_snapshot",
+        "artifact_type": "medium_horizon_theme_context",
+        "theme_ranks": [],
+        "data_quality": {},
+        "policy": {},
+    }
+
+
+def test_strict_latest_rejects_override_not_declared_by_signal(tmp_path) -> None:
+    declared = tmp_path / "declared-theme_momentum_snapshot.json"
+    override = tmp_path / "other-theme_momentum_snapshot.json"
+    declared.write_text(json.dumps(_theme_snapshot()), encoding="utf-8")
+    override.write_text(json.dumps(_theme_snapshot()), encoding="utf-8")
+    payload = _strict_signal(
+        source=str(declared),
+        source_sha256=hashlib.sha256(declared.read_bytes()).hexdigest(),
+    )
+
+    with pytest.raises(SignalValidationError, match="declared theme source"):
+        validate_latest_signal(payload, theme_artifact_path=override)
+
+
+def test_strict_latest_rejects_linked_theme_as_of_mismatch(tmp_path) -> None:
+    theme_path = tmp_path / "theme_momentum_snapshot.json"
+    theme_path.write_text(json.dumps(_theme_snapshot(as_of="2025-12-31")), encoding="utf-8")
+    payload = _strict_signal(source=str(theme_path))
+
+    with pytest.raises(SignalValidationError, match="as_of"):
+        validate_latest_signal(payload, theme_artifact_path=theme_path)
+
+
+def test_strict_latest_rejects_present_but_invalid_source_hash(tmp_path) -> None:
+    theme_path = tmp_path / "theme_momentum_snapshot.json"
+    theme_path.write_text(json.dumps(_theme_snapshot()), encoding="utf-8")
+    payload = _strict_signal(source=str(theme_path))
+    payload["evidence"]["source_hashes"] = {str(theme_path): ""}
+
+    with pytest.raises(SignalValidationError, match="source_hashes"):
+        validate_latest_signal(payload, theme_artifact_path=theme_path)
+
+
+def test_strict_latest_resolves_relative_source_from_signal_base_dir(tmp_path) -> None:
+    theme_path = tmp_path / "theme_momentum_snapshot.json"
+    theme_path.write_text(json.dumps(_theme_snapshot()), encoding="utf-8")
+    relative_source = "theme_momentum_snapshot.json"
+    payload = _strict_signal(
+        source=relative_source,
+        source_sha256=hashlib.sha256(theme_path.read_bytes()).hexdigest(),
+    )
+
+    validate_latest_signal(
+        payload,
+        theme_artifact_path=relative_source,
+        signal_base_dir=tmp_path,
+    )
+
+
 def test_signal_requires_long_horizon_contract() -> None:
     payload = load_example()
     payload["horizon"] = "1-3 months"