Add strict latest freshness validation

scripts/validate_latest_signal.py

@@ -9,7 +9,7 @@
 ROOT = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(ROOT / "src"))
 
-from research_signal_context_pipelines import SignalValidationError, validate_signal  # noqa: E402
+from research_signal_context_pipelines import SignalValidationError, validate_latest_signal, validate_signal  # noqa: E402
 
 
 DEFAULT_SIGNAL_PATH = Path("data/output/latest_signal.json")
@@ -19,6 +19,9 @@ def main() -> int:
     parser = argparse.ArgumentParser(description="Validate a long-horizon shadow signal artifact.")
     parser.add_argument("path", nargs="?", default=str(DEFAULT_SIGNAL_PATH), help="Signal JSON path")
     parser.add_argument("--allow-missing", action="store_true", help="Exit successfully when the path does not exist")
+    parser.add_argument("--strict-latest", action="store_true", help="Validate current signal and linked theme freshness/gate")
+    parser.add_argument("--theme-artifact", help="Explicit linked theme momentum artifact path for strict validation")
+    parser.add_argument("--allow-expired", action="store_true", help="Ignore expiry only; keep schema and gate checks")
     args = parser.parse_args()
 
     path = Path(args.path)
@@ -30,7 +33,14 @@ def main() -> int:
 
     payload = json.loads(path.read_text(encoding="utf-8"))
     try:
-        validate_signal(payload)
+        if args.strict_latest:
+            validate_latest_signal(
+                payload,
+                theme_artifact_path=args.theme_artifact,
+                allow_expired=args.allow_expired,
+            )
+        else:
+            validate_signal(payload)
     except SignalValidationError as exc:
         raise SystemExit(f"invalid signal artifact: {exc}") from exc
 

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,9 @@
 from __future__ import annotations
 
 import datetime as dt
+import json
 from collections.abc import Mapping, Sequence
+from pathlib import Path
 from typing import Any
 
 
@@ -130,6 +132,50 @@ def validate_signal(payload: Mapping[str, Any]) -> None:
     _require_string(policy.get("downstream_use"), "policy.downstream_use")
 
 
+def validate_latest_signal(
+    payload: Mapping[str, Any],
+    *,
+    reference_date: dt.date | None = None,
+    theme_artifact_path: str | Path | None = None,
+    allow_expired: bool = False,
+) -> None:
+    """Validate a current signal and its linked theme artifact fail-closed."""
+    validate_signal(payload)
+    if str(payload.get("schema_version")) != "2":
+        raise SignalValidationError("strict latest validation requires signal schema v2")
+    effective_date = reference_date or dt.datetime.now(dt.UTC).date()
+    if not allow_expired and dt.date.fromisoformat(str(payload["expires_at"])) < effective_date:
+        raise SignalValidationError(f"signal artifact expired on {payload['expires_at']}")
+
+    source_paths = list(payload.get("evidence", {}).get("sources", []))
+    artifact = Path(theme_artifact_path) if theme_artifact_path else next(
+        (Path(source) for source in source_paths if "theme_momentum_snapshot" in str(source)), None
+    )
+    if artifact is None:
+        raise SignalValidationError("latest signal requires a referenced theme momentum artifact")
+    if not artifact.exists():
+        raise SignalValidationError(f"referenced theme momentum artifact not found: {artifact}")
+    try:
+        theme_payload = json.loads(artifact.read_text(encoding="utf-8"))
+        from .theme_momentum import validate_theme_momentum_snapshot
+
+        validate_theme_momentum_snapshot(
+            theme_payload,
+            reference_date=effective_date,
+            compatibility=False,
+            check_freshness=True,
+            require_gate=True,
+            allow_expired=allow_expired,
+        )
+    except (OSError, json.JSONDecodeError, ValueError) as exc:
+        if isinstance(exc, SignalValidationError):
+            raise
+        raise SignalValidationError(f"invalid referenced theme momentum artifact: {exc}") from exc
+    gate = theme_payload["data_quality"]["gate"]
+    if gate["allow_downstream_recommendation"] is not True:
+        raise SignalValidationError("referenced theme momentum artifact is blocked by data_quality.gate")
+
+
 def _validate_bias_mapping(mapping: Mapping[str, Any], name: str) -> None:
     for key, bias in mapping.items():
         _require_string(key, f"{name} key")

src/research_signal_context_pipelines/theme_momentum.py

@@ -250,6 +250,11 @@ def build_theme_momentum_snapshot(
         },
         "theme_ranks": theme_ranks,
         "data_quality": {
+            "gate": {
+                "status": "pass",
+                "allow_downstream_recommendation": True,
+                "reasons": [],
+            },
             "coverage": {
                 "configured_symbol_count": len(exposure_symbols),
                 "priced_symbol_count": len(priced_exposure_symbols),
@@ -271,7 +276,15 @@ def build_theme_momentum_snapshot(
     }
 
 
-def validate_theme_momentum_snapshot(snapshot: Mapping[str, Any]) -> None:
+def validate_theme_momentum_snapshot(
+    snapshot: Mapping[str, Any],
+    *,
+    reference_date: dt.date | None = None,
+    compatibility: bool = True,
+    check_freshness: bool = False,
+    require_gate: bool = False,
+    allow_expired: bool = False,
+) -> 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]
@@ -280,18 +293,45 @@ def validate_theme_momentum_snapshot(snapshot: Mapping[str, Any]) -> None:
     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'")
+    if schema_version == "1" and not compatibility:
+        raise ValueError("schema 1 theme momentum snapshots require explicit compatibility=True")
+    if snapshot["mode"] != "theme_momentum_snapshot":
+        raise ValueError("theme momentum snapshot mode is invalid")
+    if snapshot["artifact_type"] != THEME_MOMENTUM_ARTIFACT_TYPE:
+        raise ValueError("theme momentum snapshot artifact_type is invalid")
     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")
+        raise ValueError("theme momentum snapshot generated_at must be an ISO datetime")
+    normalized_generated_at = generated_at[:-1] + "+00:00" if generated_at.endswith("Z") else generated_at
+    try:
+        dt.datetime.fromisoformat(normalized_generated_at)
+    except ValueError as exc:
+        raise ValueError("theme momentum snapshot generated_at must be an ISO datetime") from exc
     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"]):
+        expires_at = parse_price_date(snapshot["expires_at"])
+        as_of = parse_price_date(snapshot["as_of"])
+        if expires_at < as_of:
             raise ValueError("theme momentum snapshot expires_at must not be before as_of")
+        if check_freshness and reference_date and expires_at < reference_date and not allow_expired:
+            raise ValueError(f"theme momentum snapshot expired on {expires_at.isoformat()}")
     if not isinstance(snapshot["theme_ranks"], list) or not isinstance(snapshot["data_quality"], Mapping):
         raise ValueError("theme momentum snapshot core shape is invalid")
+    if require_gate:
+        gate = snapshot["data_quality"].get("gate")
+        if not isinstance(gate, Mapping):
+            raise ValueError("theme momentum snapshot data_quality.gate is required")
+        if gate.get("status") not in {"pass", "blocked"}:
+            raise ValueError("theme momentum snapshot data_quality.gate.status is invalid")
+        if not isinstance(gate.get("allow_downstream_recommendation"), bool):
+            raise ValueError("theme momentum snapshot data_quality.gate.allow_downstream_recommendation must be boolean")
+        if not isinstance(gate.get("reasons"), list):
+            raise ValueError("theme momentum snapshot data_quality.gate.reasons must be a list")
+        if gate["status"] != ("pass" if gate["allow_downstream_recommendation"] else "blocked"):
+            raise ValueError("theme momentum snapshot data_quality.gate is inconsistent")
 
 
 def write_theme_momentum_snapshot(snapshot: Mapping[str, Any], path: str | Path) -> Path:

tests/test_signal_validation.py

@@ -1,11 +1,12 @@
 from __future__ import annotations
 
 import json
+import datetime as dt
 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,44 @@ def test_v2_signal_requires_model_and_scoring_versions() -> None:
         validate_signal(payload)
 
 
+def test_latest_signal_validation_requires_fresh_linked_theme(tmp_path) -> None:
+    payload = load_example()
+    payload.update({"schema_version": "2", "model_version": "shadow-v2", "scoring_version": "rules-v2"})
+    theme = {
+        "schema_version": "2",
+        "as_of": "2026-01-01",
+        "generated_at": "2026-01-01T00:00:00Z",
+        "expires_at": "2026-01-31",
+        "model_version": "theme-v1",
+        "scoring_version": "rules-v1",
+        "mode": "theme_momentum_snapshot",
+        "artifact_type": "medium_horizon_theme_context",
+        "theme_ranks": [],
+        "data_quality": {"gate": {"status": "pass", "allow_downstream_recommendation": True, "reasons": []}},
+        "policy": {},
+    }
+    theme_path = tmp_path / "theme_momentum_snapshot.json"
+    theme_path.write_text(json.dumps(theme), encoding="utf-8")
+
+    validate_latest_signal(payload, theme_artifact_path=theme_path, reference_date=dt.date(2026, 1, 15))
+
+    with pytest.raises(SignalValidationError, match="expired"):
+        validate_latest_signal(payload, theme_artifact_path=theme_path, reference_date=dt.date(2026, 2, 1))
+
+    validate_latest_signal(
+        payload,
+        theme_artifact_path=theme_path,
+        reference_date=dt.date(2026, 2, 1),
+        allow_expired=True,
+    )
+
+
+def test_latest_signal_rejects_v1_with_controlled_error(tmp_path) -> None:
+    payload = load_example()
+    with pytest.raises(SignalValidationError, match="schema v2"):
+        validate_latest_signal(payload, theme_artifact_path=tmp_path / "missing.json")
+
+
 def test_signal_requires_long_horizon_contract() -> None:
     payload = load_example()
     payload["horizon"] = "1-3 months"

tests/test_theme_momentum.py

@@ -2,6 +2,8 @@
 
 import datetime as dt
 
+import pytest
+
 from research_signal_context_pipelines.price_history import PriceRow
 from research_signal_context_pipelines.theme_momentum import (
     build_theme_momentum_snapshot,
@@ -74,6 +76,11 @@ def test_theme_momentum_ranks_strong_broad_theme_first() -> None:
     assert [item["symbol"] for item in ranked[0]["top_symbols"]] == ["MU", "HBM2"]
     assert ranked[0]["momentum_score"] > ranked[1]["momentum_score"]
     assert snapshot["policy"]["execution_allowed"] is False
+    assert snapshot["data_quality"]["gate"] == {
+        "status": "pass",
+        "allow_downstream_recommendation": True,
+        "reasons": [],
+    }
     assert snapshot["data_quality"]["coverage"]["price_coverage_ratio"] == 1.0
 
 
@@ -131,3 +138,41 @@ def test_theme_momentum_validator_reads_v1_and_v2_metadata() -> None:
     for key in ("expires_at", "model_version", "scoring_version"):
         legacy.pop(key)
     validate_theme_momentum_snapshot(legacy)
+
+
+def test_theme_momentum_validator_enforces_strict_schema_freshness_and_gate() -> 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",
+        )
+    }
+    snapshot = build_theme_momentum_snapshot(
+        _trend_rows("MU", start_close=50, daily_step=0.22),
+        themes=themes,
+        exposures={"MU": SymbolThemeExposure("MU", ("hbm_memory",), "high", "memory exposure")},
+        as_of="2025-10-07",
+        generated_at=dt.datetime(2025, 10, 7, tzinfo=dt.timezone.utc),
+    )
+
+    validate_theme_momentum_snapshot(
+        snapshot,
+        reference_date=dt.date(2025, 10, 8),
+        compatibility=False,
+        check_freshness=True,
+        require_gate=True,
+    )
+    snapshot["data_quality"]["gate"]["allow_downstream_recommendation"] = False
+    with pytest.raises(ValueError, match="gate"):
+        validate_theme_momentum_snapshot(
+            snapshot,
+            reference_date=dt.date(2025, 10, 8),
+            compatibility=False,
+            check_freshness=True,
+            require_gate=True,
+        )