fix: add research input quality gates

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

Author: Pigbibi

.github/workflows/monthly_advisory_review.yml

@@ -67,11 +67,13 @@ jobs:
         with:
           path: advisor
       - name: Checkout political event repository
+        id: checkout_political_events
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/PoliticalEventTrackingResearch
           path: political-events
       - name: Checkout signal context repository
+        id: checkout_signal_context
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/ResearchSignalContextPipelines
@@ -111,6 +113,8 @@ jobs:
             --cadence monthly
             --political-events "../political-events/${POLITICAL_EVENTS_PATH}"
             --political-watchlist "../political-events/${POLITICAL_WATCHLIST_PATH}"
+            --upstream-repo-sha "QuantStrategyLab/PoliticalEventTrackingResearch=${{ steps.checkout_political_events.outputs.commit }}"
+            --upstream-repo-sha "QuantStrategyLab/ResearchSignalContextPipelines=${{ steps.checkout_signal_context.outputs.commit }}"
             --ai-signal "../research-signal-context/${AI_SIGNAL_PATH}"
             --output-dir data/output/monthly_advisory_review
             --monthly-review

.github/workflows/publish_advisory_site.yml

@@ -73,11 +73,13 @@ jobs:
         with:
           path: advisor
       - name: Checkout political event repository
+        id: checkout_political_events
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/PoliticalEventTrackingResearch
           path: political-events
       - name: Checkout signal context repository
+        id: checkout_signal_context
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/ResearchSignalContextPipelines
@@ -122,6 +124,8 @@ jobs:
             --cadence weekly
             --political-events "../political-events/${POLITICAL_EVENTS_PATH}"
             --political-watchlist "../political-events/${POLITICAL_WATCHLIST_PATH}"
+            --upstream-repo-sha "QuantStrategyLab/PoliticalEventTrackingResearch=${{ steps.checkout_political_events.outputs.commit }}"
+            --upstream-repo-sha "QuantStrategyLab/ResearchSignalContextPipelines=${{ steps.checkout_signal_context.outputs.commit }}"
             --ai-signal "../research-signal-context/${AI_SIGNAL_PATH}"
             --output-dir data/output/published
             --site-output-dir site

.github/workflows/weekly_advisory_review.yml

@@ -62,11 +62,13 @@ jobs:
         with:
           path: advisor
       - name: Checkout political event repository
+        id: checkout_political_events
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/PoliticalEventTrackingResearch
           path: political-events
       - name: Checkout signal context repository
+        id: checkout_signal_context
         uses: actions/checkout@v6
         with:
           repository: QuantStrategyLab/ResearchSignalContextPipelines
@@ -105,6 +107,8 @@ jobs:
             --cadence weekly
             --political-events "../political-events/${POLITICAL_EVENTS_PATH}"
             --political-watchlist "../political-events/${POLITICAL_WATCHLIST_PATH}"
+            --upstream-repo-sha "QuantStrategyLab/PoliticalEventTrackingResearch=${{ steps.checkout_political_events.outputs.commit }}"
+            --upstream-repo-sha "QuantStrategyLab/ResearchSignalContextPipelines=${{ steps.checkout_signal_context.outputs.commit }}"
             --ai-signal "../research-signal-context/${AI_SIGNAL_PATH}"
             --output-dir data/output/weekly_advisory_review
             --market-cache-dir .cache/market-data

docs/advisory_contract.md

@@ -270,6 +270,12 @@ policy
 generated_at
 ```
 
+`source_artifacts` also records each existing input's SHA-256, point-in-time
+`as_of`/`generated_at`/`expires_at`, and schema/header metadata. Scheduled
+workflows pass the checked-out commit SHA for each upstream repository in
+`upstream_repositories`; missing or stale context is excluded from scoring and
+reported in `summary.data_quality_warnings`.
+
 ## Source Mode
 
 `summary.source_mode` is:

src/quant_advisor_research/advisory_report.py

@@ -53,6 +53,8 @@
 }
 SHORT_PRIMARY_MIN_SCORE_EDGE = 0.05
 THEME_MOMENTUM_ARTIFACT_TYPE = "medium_horizon_theme_context"
+AI_SIGNAL_MAX_AGE_DAYS = 31
+THEME_MOMENTUM_MAX_AGE_DAYS = 7
 
 CADENCE_LABELS_ZH = {
     "daily": "日度",
@@ -207,6 +209,8 @@ class MarketConfirmation:
     market_score: float | None = None
     data_source: str = ""
     price_observation_count: int = 0
+    confirmation_quality: str = ""
+    warnings: str = ""
 
 
 def parse_date(value: str) -> dt.date:
@@ -265,6 +269,62 @@ def load_ai_signal(path: str | Path | None) -> dict[str, Any] | None:
     return payload
 
 
+def input_freshness(payload: dict[str, Any], *, name: str, as_of: dt.date, max_age_days: int) -> tuple[bool, str | None]:
+    try:
+        artifact_as_of = parse_date(str(payload.get("as_of", "")))
+    except ValueError:
+        return False, f"{name}_missing_or_invalid_as_of"
+    if artifact_as_of > as_of:
+        return False, f"{name}_as_of_in_future"
+    if (as_of - artifact_as_of).days > max_age_days:
+        return False, f"{name}_stale"
+    generated_at = str(payload.get("generated_at", "")).strip()
+    if generated_at:
+        try:
+            dt.datetime.fromisoformat(generated_at.replace("Z", "+00:00"))
+        except ValueError:
+            return False, f"{name}_invalid_generated_at"
+    expires_at = str(payload.get("expires_at", "")).strip()
+    if expires_at:
+        try:
+            if parse_date(expires_at[:10]) < as_of:
+                return False, f"{name}_expired"
+        except ValueError:
+            return False, f"{name}_invalid_expires_at"
+    return True, None
+
+
+def apply_input_freshness(
+    *, ai_signal: dict[str, Any] | None, theme_momentum: dict[str, Any] | None, as_of: dt.date
+) -> tuple[dict[str, Any] | None, dict[str, Any] | None, list[str]]:
+    warnings: list[str] = []
+    if ai_signal:
+        valid, warning = input_freshness(ai_signal, name="ai_signal", as_of=as_of, max_age_days=AI_SIGNAL_MAX_AGE_DAYS)
+        if not valid:
+            warnings.append(warning or "ai_signal_not_fresh")
+            ai_signal = None
+    if theme_momentum:
+        valid, warning = input_freshness(
+            theme_momentum, name="theme_momentum", as_of=as_of, max_age_days=THEME_MOMENTUM_MAX_AGE_DAYS
+        )
+        if not valid:
+            warnings.append(warning or "theme_momentum_not_fresh")
+            theme_momentum = None
+            return ai_signal, theme_momentum, warnings
+        extreme = any(
+            abs(as_float(item.get("return_3m"))) > 2.0
+            for theme in theme_momentum.get("theme_ranks", [])
+            if isinstance(theme, dict)
+            for item in theme.get("top_symbols", [])
+            if isinstance(item, dict)
+        )
+        if extreme:
+            warnings.append("theme_momentum_extreme_return_3m")
+            theme_momentum = None
+            return ai_signal, theme_momentum, warnings
+    return ai_signal, theme_momentum, warnings
+
+
 
 def load_theme_momentum(path: str | Path | None) -> dict[str, Any] | None:
     if path is None:
@@ -364,6 +424,8 @@ def load_market_confirmation(path: str | Path | None, as_of: dt.date) -> dict[st
             market_score=as_float(row.get("market_score")) if str(row.get("market_score", "")).strip() else None,
             data_source=str(row.get("data_source", "")),
             price_observation_count=int(as_float(row.get("price_observation_count"))),
+            confirmation_quality=str(row.get("confirmation_quality", "")),
+            warnings=str(row.get("warnings", "")),
         )
     return confirmations
 
@@ -959,6 +1021,8 @@ def market_confirmation_score(market: MarketConfirmation | None) -> float | None
         return None
     if market.data_source == "theme_momentum_fallback":
         return None
+    if market.confirmation_quality == "anomalous" or "extreme_return_63d" in market.warnings or "stale_price" in market.warnings:
+        return None
     if market.market_score is not None:
         return round(clamp(market.market_score, 0, 1), 3)
     relative_20d = clamp(market.relative_return_20d / 0.20, -1, 1)
@@ -1437,6 +1501,9 @@ def build_advisory_report(
     events = load_events(political_events_path, as_of_date)
     ai_signal = load_ai_signal(ai_signal_path)
     theme_momentum = load_theme_momentum(theme_momentum_path)
+    ai_signal, theme_momentum, freshness_warnings = apply_input_freshness(
+        ai_signal=ai_signal, theme_momentum=theme_momentum, as_of=as_of_date
+    )
     market_confirmations = load_market_confirmation(market_confirmation_path, as_of_date)
     theme_momentum_summary = summarize_theme_momentum(theme_momentum)
     source_mode, data_quality_warnings = source_mode_for_paths(
@@ -1446,6 +1513,7 @@ def build_advisory_report(
         theme_momentum_path,
         market_confirmation_path,
     )
+    data_quality_warnings.extend(freshness_warnings)
 
     events_by_symbol: dict[str, list[Event]] = defaultdict(list)
     for event in events:

src/quant_advisor_research/artifacts.py

@@ -2,6 +2,7 @@
 
 import hashlib
 import json
+import csv
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import Any, Mapping
@@ -37,6 +38,8 @@ def write_report_manifest(
     git_sha: str | None = None,
     run_id: str | None = None,
     run_attempt: str | None = None,
+    upstream_repo_shas: Mapping[str, str] | None = None,
+    input_paths: Mapping[str, str | Path] | None = None,
 ) -> Path:
     resolved_report = Path(report_path)
     resolved_markdown = Path(markdown_path)
@@ -59,6 +62,36 @@ def write_report_manifest(
     if run_attempt:
         version_parts.append(f"attempt-{run_attempt}")
 
+    source_artifacts = {}
+    for name, raw_path in (input_paths or {}).items():
+        if raw_path is None:
+            continue
+        path = Path(raw_path)
+        if not path.exists():
+            continue
+        metadata: dict[str, Any] = {"path": str(path), "sha256": sha256_file(path)}
+        if path.suffix.lower() == ".csv":
+            with path.open(encoding="utf-8", newline="") as handle:
+                reader = csv.DictReader(handle)
+                metadata["schema"] = ",".join(reader.fieldnames or [])
+                as_of_values = sorted({str(row.get("as_of", "")).strip() for row in reader if row.get("as_of")})
+                if as_of_values:
+                    metadata["as_of"] = as_of_values[-1]
+        elif path.suffix.lower() == ".json":
+            try:
+                source = json.loads(path.read_text(encoding="utf-8"))
+                metadata.update(
+                    {
+                        "as_of": source.get("as_of", ""),
+                        "generated_at": source.get("generated_at", ""),
+                        "expires_at": source.get("expires_at", ""),
+                        "schema": str(source.get("schema_version", "")),
+                    }
+                )
+            except (OSError, json.JSONDecodeError):
+                metadata["schema"] = "invalid_json"
+        source_artifacts[name] = metadata
+
     payload = {
         "manifest_type": "model_recommendation_report",
         "artifact_type": "model_recommendations",
@@ -76,7 +109,8 @@ def write_report_manifest(
             "github_run_id": run_id or "",
             "github_run_attempt": run_attempt or "",
         },
-        "source_artifacts": dict(report.get("source_artifacts") or {}),
+        "source_artifacts": source_artifacts or dict(report.get("source_artifacts") or {}),
+        "upstream_repositories": dict(upstream_repo_shas or {}),
         "summary": dict(report.get("summary") or {}),
         "artifacts": {
             "json": {

src/quant_advisor_research/build_pipeline.py

@@ -203,6 +203,7 @@ def build_advisory_artifacts(
     site_url: str = DEFAULT_SITE_URL,
     feed_title: str = DEFAULT_FEED_TITLE,
     recover_site_archive: bool = False,
+    upstream_repo_shas: dict[str, str] | None = None,
 ) -> BuildPipelineResult:
     output = Path(output_dir)
     output.mkdir(parents=True, exist_ok=True)
@@ -253,6 +254,14 @@ def build_advisory_artifacts(
         git_sha=os.environ.get("GITHUB_SHA"),
         run_id=os.environ.get("GITHUB_RUN_ID"),
         run_attempt=os.environ.get("GITHUB_RUN_ATTEMPT"),
+        upstream_repo_shas=upstream_repo_shas,
+        input_paths={
+            "political_events": political_events_path,
+            "political_watchlist": political_watchlist_path,
+            "ai_signal": resolved_ai_signal,
+            "theme_momentum": resolved_theme_momentum,
+            "market_confirmation": market_path,
+        },
     )
 
     monthly_review_json: Path | None = None
@@ -357,6 +366,7 @@ def build_arg_parser() -> argparse.ArgumentParser:
     parser.add_argument("--site-url", default=DEFAULT_SITE_URL)
     parser.add_argument("--feed-title", default=DEFAULT_FEED_TITLE)
     parser.add_argument("--recover-site-archive", action="store_true", help="Recover prior report JSONs from published site index.")
+    parser.add_argument("--upstream-repo-sha", action="append", default=[], metavar="REPO=SHA")
     return parser
 
 
@@ -388,6 +398,7 @@ def main(argv: list[str] | None = None) -> None:
         site_url=args.site_url,
         feed_title=args.feed_title,
         recover_site_archive=args.recover_site_archive,
+        upstream_repo_shas={key: value for item in args.upstream_repo_sha for key, value in [item.split("=", 1)] if key and value},
     )
     print(
         "advisory_artifacts_built "

src/quant_advisor_research/market_confirmation.py

@@ -50,6 +50,7 @@
     "XLE",
     "XLK",
 }
+EXTREME_RETURN_63D_THRESHOLD = 2.0
 
 
 @dataclass(frozen=True)
@@ -295,7 +296,12 @@ def compute_market_confirmation(
         volatility_21d=volatility_21d,
     )
     price_age_days = max(((requested_as_of or bars[-1].date) - bars[-1].date).days, 0)
-    confirmation_quality = "price_observed" if price_age_days <= 7 else "stale_price"
+    warnings = [item for item in (warning.split(";") if warning else []) if item]
+    if abs(return_63d) > EXTREME_RETURN_63D_THRESHOLD:
+        warnings.append("extreme_return_63d")
+    if price_age_days > 7:
+        warnings.append("stale_price")
+    confirmation_quality = "anomalous" if "extreme_return_63d" in warnings else "price_observed" if price_age_days <= 7 else "stale_price"
     return MarketConfirmationRow(
         symbol=symbol.upper(),
         as_of=bars[-1].date,
@@ -312,7 +318,7 @@ def compute_market_confirmation(
         price_observation_count=len(bars),
         price_age_days=price_age_days,
         confirmation_quality=confirmation_quality,
-        warnings=warning,
+        warnings=";".join(dict.fromkeys(warnings)),
     )
 
 

tests/test_advisory_report.py

@@ -34,6 +34,47 @@ def test_build_advisory_report_blocks_execution_and_allocation() -> None:
     assert report["recommendations"]
 
 
+def test_expired_ai_signal_is_excluded_from_long_context_and_warns(tmp_path: Path) -> None:
+    signal = json.loads((ROOT / "examples/research_signal_context.example.json").read_text(encoding="utf-8"))
+    signal["expires_at"] = "2026-05-01"
+    signal_path = tmp_path / "expired_signal.json"
+    signal_path.write_text(json.dumps(signal), encoding="utf-8")
+
+    report = build_advisory_report(
+        as_of="2026-05-30",
+        cadence="weekly",
+        political_events_path=ROOT / "examples/political_events.example.csv",
+        political_watchlist_path=ROOT / "examples/political_watchlist.example.csv",
+        ai_signal_path=signal_path,
+    )
+
+    assert any("ai_signal_expired" in warning for warning in report["summary"]["data_quality_warnings"])
+    assert report["summary"]["long_context_available"] is False
+
+
+def test_manifest_records_input_hash_and_upstream_metadata(tmp_path: Path) -> None:
+    report_path = tmp_path / "report.json"
+    markdown_path = tmp_path / "report.md"
+    report_path.write_text("{}\n", encoding="utf-8")
+    markdown_path.write_text("# report\n", encoding="utf-8")
+    input_path = ROOT / "examples/political_events.example.csv"
+    report = {"as_of": "2026-05-30", "cadence": "weekly", "schema_version": "5", "mode": "model_recommendations"}
+
+    manifest_path = write_report_manifest(
+        report=report,
+        report_path=report_path,
+        markdown_path=markdown_path,
+        manifest_path=tmp_path / "report.manifest.json",
+        upstream_repo_shas={"QuantStrategyLab/PoliticalEventTrackingResearch": "abc123"},
+        input_paths={"political_events": input_path},
+    )
+
+    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    assert manifest["upstream_repositories"]["QuantStrategyLab/PoliticalEventTrackingResearch"] == "abc123"
+    assert manifest["source_artifacts"]["political_events"]["sha256"]
+    assert manifest["source_artifacts"]["political_events"]["schema"].startswith("event_id,")
+
+
 def test_low_confidence_events_remain_verify_source_until_verified() -> None:
     report = build_advisory_report(
         as_of="2026-05-30",

tests/test_market_confirmation.py

@@ -58,6 +58,18 @@ def test_compute_market_confirmation_uses_relative_strength_volume_drawdown_and_
     assert row.confirmation_quality == "price_observed"
 
 
+def test_extreme_return_is_marked_as_data_quality_anomaly() -> None:
+    start = dt.date(2026, 1, 1)
+    symbol_bars = make_bars(start, [100.0] * 69 + [350.0])
+    benchmark_bars = make_bars(start, [100.0] * 70)
+
+    row = compute_market_confirmation("MU", symbol_bars, benchmark_bars, data_source="unit_test")
+
+    assert row is not None
+    assert row.confirmation_quality == "anomalous"
+    assert "extreme_return_63d" in row.warnings
+
+
 def test_market_confirmation_falls_back_to_theme_momentum_without_network(tmp_path: Path) -> None:
     theme_payload = {
         "as_of": "2026-05-29",