fix: keep theme artifact readers version compatible

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

src/research_signal_context_pipelines/__init__.py

@@ -4,7 +4,11 @@
 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 .theme_momentum import build_theme_momentum_snapshot, write_theme_momentum_snapshot
+from .theme_momentum import (
+    build_theme_momentum_snapshot,
+    validate_theme_momentum_snapshot,
+    write_theme_momentum_snapshot,
+)
 from .theme_universe import build_theme_context, load_symbol_theme_exposure, load_theme_taxonomy
 
 __all__ = [
@@ -20,6 +24,7 @@
     "load_symbol_theme_exposure",
     "load_theme_taxonomy",
     "validate_signal",
+    "validate_theme_momentum_snapshot",
     "write_filtered_price_history",
     "write_theme_momentum_snapshot",
 ]

src/research_signal_context_pipelines/theme_momentum.py

@@ -271,6 +271,29 @@ def build_theme_momentum_snapshot(
     }
 
 
+def validate_theme_momentum_snapshot(snapshot: Mapping[str, Any]) -> None:
+    """Validate the stable metadata and core shape of v1/v2 theme artifacts."""
+    required = ("schema_version", "as_of", "generated_at", "mode", "artifact_type", "theme_ranks", "data_quality", "policy")
+    missing = [key for key in required if key not in snapshot]
+    if missing:
+        raise ValueError(f"theme momentum snapshot missing required keys: {', '.join(missing)}")
+    schema_version = str(snapshot["schema_version"])
+    if schema_version not in {"1", "2"}:
+        raise ValueError("theme momentum snapshot schema_version must be '1' or '2'")
+    parse_price_date(snapshot["as_of"])
+    generated_at = snapshot["generated_at"]
+    if not isinstance(generated_at, str) or not generated_at.strip():
+        raise ValueError("theme momentum snapshot generated_at must be a non-empty string")
+    if schema_version == "2":
+        for key in ("expires_at", "model_version", "scoring_version"):
+            if not isinstance(snapshot.get(key), str) or not snapshot[key].strip():
+                raise ValueError(f"theme momentum snapshot {key} must be a non-empty string")
+        if parse_price_date(snapshot["expires_at"]) < parse_price_date(snapshot["as_of"]):
+            raise ValueError("theme momentum snapshot expires_at must not be before as_of")
+    if not isinstance(snapshot["theme_ranks"], list) or not isinstance(snapshot["data_quality"], Mapping):
+        raise ValueError("theme momentum snapshot core shape is invalid")
+
+
 def write_theme_momentum_snapshot(snapshot: Mapping[str, Any], path: str | Path) -> Path:
     output_path = Path(path)
     output_path.parent.mkdir(parents=True, exist_ok=True)

tests/test_theme_momentum.py

@@ -3,7 +3,10 @@
 import datetime as dt
 
 from research_signal_context_pipelines.price_history import PriceRow
-from research_signal_context_pipelines.theme_momentum import build_theme_momentum_snapshot
+from research_signal_context_pipelines.theme_momentum import (
+    build_theme_momentum_snapshot,
+    validate_theme_momentum_snapshot,
+)
 from research_signal_context_pipelines.theme_universe import SymbolThemeExposure, ThemeDefinition
 
 
@@ -100,3 +103,31 @@ def test_theme_momentum_records_missing_price_coverage() -> None:
     assert snapshot["data_quality"]["coverage"]["price_coverage_ratio"] == 0.5
     assert snapshot["theme_ranks"][0]["component_count"] == 2
     assert snapshot["theme_ranks"][0]["priced_symbol_count"] == 1
+
+
+def test_theme_momentum_validator_reads_v1_and_v2_metadata() -> None:
+    themes = {
+        "hbm_memory": ThemeDefinition(
+            taxonomy_version="test-v1",
+            theme_id="hbm_memory",
+            theme_name="HBM and memory",
+            sector="technology",
+            horizon="6-24 months",
+            description="memory theme",
+            source_policy="primary evidence required",
+        )
+    }
+    exposures = {"MU": SymbolThemeExposure("MU", ("hbm_memory",), "high", "memory exposure")}
+    snapshot = build_theme_momentum_snapshot(
+        _trend_rows("MU", start_close=50, daily_step=0.22),
+        themes=themes,
+        exposures=exposures,
+        generated_at=dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc),
+    )
+
+    validate_theme_momentum_snapshot(snapshot)
+    legacy = dict(snapshot)
+    legacy["schema_version"] = "1"
+    for key in ("expires_at", "model_version", "scoring_version"):
+        legacy.pop(key)
+    validate_theme_momentum_snapshot(legacy)