diff --git a/src/research_signal_context_pipelines/__init__.py b/src/research_signal_context_pipelines/__init__.py index 99b98fe..029f735 100644 --- a/src/research_signal_context_pipelines/__init__.py +++ b/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", diff --git a/src/research_signal_context_pipelines/schema.py b/src/research_signal_context_pipelines/schema.py index 4076cdc..2ef0bdf 100644 --- a/src/research_signal_context_pipelines/schema.py +++ b/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')}" + ) diff --git a/tests/test_signal_validation.py b/tests/test_signal_validation.py index b2c33fe..d0dad91 100644 --- a/tests/test_signal_validation.py +++ b/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"