From 8af26fa3ff576855412620de2eda266186283f41 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:49:33 +0800 Subject: [PATCH 1/2] feat: add strict latest artifact validation Co-Authored-By: Codex --- .github/workflows/dispatch_shadow_signal.yml | 2 +- scripts/validate_latest_signal.py | 14 +++++- .../__init__.py | 3 +- .../schema.py | 44 ++++++++++++++++++ .../theme_momentum.py | 46 +++++++++++++++++-- tests/test_signal_validation.py | 35 +++++++++++++- tests/test_theme_momentum.py | 45 ++++++++++++++++++ 7 files changed, 181 insertions(+), 8 deletions(-) diff --git a/.github/workflows/dispatch_shadow_signal.yml b/.github/workflows/dispatch_shadow_signal.yml index 1f09ecb..4ba11df 100644 --- a/.github/workflows/dispatch_shadow_signal.yml +++ b/.github/workflows/dispatch_shadow_signal.yml @@ -50,7 +50,7 @@ jobs: python-version: "3.11" - name: Validate existing latest signal if present - run: python scripts/validate_latest_signal.py --allow-missing + run: python scripts/validate_latest_signal.py --allow-missing --strict-latest - name: Build point-in-time context bundle run: | diff --git a/scripts/validate_latest_signal.py b/scripts/validate_latest_signal.py index bfd98cc..7d505bf 100644 --- a/scripts/validate_latest_signal.py +++ b/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 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..8b8e561 100644 --- a/src/research_signal_context_pipelines/schema.py +++ b/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,48 @@ 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) + 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") diff --git a/src/research_signal_context_pipelines/theme_momentum.py b/src/research_signal_context_pipelines/theme_momentum.py index 0edd39c..6aa8f5e 100644 --- a/src/research_signal_context_pipelines/theme_momentum.py +++ b/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: diff --git a/tests/test_signal_validation.py b/tests/test_signal_validation.py index b2c33fe..f4c0d7b 100644 --- a/tests/test_signal_validation.py +++ b/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,38 @@ 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_signal_requires_long_horizon_contract() -> None: payload = load_example() payload["horizon"] = "1-3 months" diff --git a/tests/test_theme_momentum.py b/tests/test_theme_momentum.py index 2fbf055..db1763c 100644 --- a/tests/test_theme_momentum.py +++ b/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, + ) From 7363fdadf3d9aef35e28f71de86d75d2038f9d31 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 13 Jul 2026 16:51:47 +0800 Subject: [PATCH 2/2] fix: keep strict validation fail-closed Co-Authored-By: Codex --- .github/workflows/dispatch_shadow_signal.yml | 2 +- src/research_signal_context_pipelines/schema.py | 2 ++ tests/test_signal_validation.py | 6 ++++++ 3 files changed, 9 insertions(+), 1 deletion(-) diff --git a/.github/workflows/dispatch_shadow_signal.yml b/.github/workflows/dispatch_shadow_signal.yml index 4ba11df..1f09ecb 100644 --- a/.github/workflows/dispatch_shadow_signal.yml +++ b/.github/workflows/dispatch_shadow_signal.yml @@ -50,7 +50,7 @@ jobs: python-version: "3.11" - name: Validate existing latest signal if present - run: python scripts/validate_latest_signal.py --allow-missing --strict-latest + run: python scripts/validate_latest_signal.py --allow-missing - name: Build point-in-time context bundle run: | diff --git a/src/research_signal_context_pipelines/schema.py b/src/research_signal_context_pipelines/schema.py index 8b8e561..4caf735 100644 --- a/src/research_signal_context_pipelines/schema.py +++ b/src/research_signal_context_pipelines/schema.py @@ -141,6 +141,8 @@ def validate_latest_signal( ) -> 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']}") diff --git a/tests/test_signal_validation.py b/tests/test_signal_validation.py index f4c0d7b..924ab9c 100644 --- a/tests/test_signal_validation.py +++ b/tests/test_signal_validation.py @@ -81,6 +81,12 @@ def test_latest_signal_validation_requires_fresh_linked_theme(tmp_path) -> None: ) +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"