feat: add bounded plugin lineage metadata

src/quant_strategy_plugins/plugin_lineage.py

@@ -0,0 +1,72 @@
+"""Small, deterministic provenance contract for strategy-plugin sidecars.
+
+The registry deliberately describes plugins as observers.  It does not grant
+position authority; that remains a strategy/runtime concern.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import date
+from typing import Any, Mapping
+
+import pandas as pd
+
+PLUGIN_LINEAGE_SCHEMA_VERSION = "strategy_plugin_lineage.v1"
+
+# Stable names are intentionally independent of implementation module names.
+PLUGIN_LINEAGE_REGISTRY: dict[str, dict[str, str]] = {
+    "crisis_response_shadow": {"lineage": "market_regime/crisis", "role": "shadow"},
+    "macro_risk_governor": {"lineage": "market_regime/macro", "role": "shadow"},
+    "market_regime_control": {"lineage": "market_regime/unified", "role": "shadow"},
+    "panic_reversal_shadow": {"lineage": "event_context/panic_reversal", "role": "notification"},
+    "taco_rebound_shadow": {"lineage": "event_context/taco_rebound", "role": "notification"},
+}
+
+
+def _json_digest(value: Any) -> str:
+    encoded = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def input_digest(frame: pd.DataFrame, config: Mapping[str, Any]) -> str:
+    """Hash the exact in-memory input and relevant configuration, not its path."""
+    values = pd.util.hash_pandas_object(frame, index=True).to_numpy().tobytes()
+    payload = {
+        "columns": [str(column) for column in frame.columns],
+        "dtypes": [str(dtype) for dtype in frame.dtypes],
+        "rows_sha256": hashlib.sha256(values).hexdigest(),
+        "config": {str(k): v for k, v in config.items() if k not in {"prices", "output_dir"}},
+    }
+    return _json_digest(payload)
+
+
+def build_plugin_lineage(
+    plugin: str,
+    *,
+    frame: pd.DataFrame,
+    config: Mapping[str, Any],
+) -> dict[str, Any]:
+    registration = PLUGIN_LINEAGE_REGISTRY.get(plugin, {"lineage": f"unregistered/{plugin}", "role": "shadow"})
+    valid_until = str(config.get("evidence_valid_until") or "").strip() or None
+    if valid_until:
+        try:
+            date.fromisoformat(valid_until)
+        except ValueError as exc:
+            raise ValueError(f"evidence_valid_until must be ISO date: {valid_until!r}") from exc
+    raw_budget = config.get("bounded_budget")
+    budget = raw_budget if isinstance(raw_budget, Mapping) else {}
+    return {
+        "schema_version": PLUGIN_LINEAGE_SCHEMA_VERSION,
+        "plugin": plugin,
+        "lineage": registration["lineage"],
+        "role": registration["role"],
+        "input_digest": input_digest(frame, config),
+        "evidence_valid_until": valid_until,
+        "evidence_expiry_status": "declared" if valid_until else "not_declared",
+        "bounded_budget": dict(budget),
+        "bounded_budget_status": "declared" if budget else "not_declared",
+        "position_mutation_allowed": False,
+        "broker_order_allowed": False,
+    }

src/quant_strategy_plugins/strategy_plugin_runner.py

@@ -59,6 +59,7 @@
     PluginLifecyclePolicy,
     PluginNotificationTargetPolicy,
 )
+from .plugin_lineage import build_plugin_lineage
 
 DEFAULT_RUNNER_OUTPUT_DIR = "data/output/strategy_plugins"
 SUPPORTED_PLUGIN_MODES = (SHADOW_MODE,)
@@ -1513,7 +1514,9 @@ def _run_table_strategy_plugin(
     prices_path = str(plugin_config.get("prices", "")).strip()
     if not prices_path:
         raise ValueError(f"{plugin} for strategy={strategy} requires a prices path")
-    payload = spec.build_payload(read_table(prices_path), plugin_config)
+    prices = read_table(prices_path)
+    payload = spec.build_payload(prices, plugin_config)
+    payload["plugin_lineage"] = build_plugin_lineage(plugin, frame=prices, config=plugin_config)
     payload = _apply_plugin_contract(
         payload,
         strategy=strategy,
@@ -1568,7 +1571,9 @@ def _run_table_notification_target_plugin(
     prices_path = str(plugin_config.get("prices", "")).strip()
     if not prices_path:
         raise ValueError(f"{plugin} for notification_target={notification_target} requires a prices path")
-    payload = spec.build_payload(read_table(prices_path), plugin_config)
+    prices = read_table(prices_path)
+    payload = spec.build_payload(prices, plugin_config)
+    payload["plugin_lineage"] = build_plugin_lineage(plugin, frame=prices, config=plugin_config)
     payload = _apply_plugin_contract(
         payload,
         notification_target=notification_target,

tests/test_strategy_plugin_runner.py

@@ -245,6 +245,41 @@ def test_strategy_plugin_runner_executes_strategy_scoped_shadow_plugin(tmp_path)
     assert "无需通知" in payload["localized_messages"]["notification"]["zh-CN"]
     assert payload["log_record"]["schema_version"] == STRATEGY_PLUGIN_LOG_SCHEMA_VERSION
     assert "策略=" in payload["log_record"]["localized_messages"]["zh-CN"]
+    lineage = payload["plugin_lineage"]
+    assert lineage["schema_version"] == "strategy_plugin_lineage.v1"
+    assert lineage["lineage"] == "market_regime/crisis"
+    assert lineage["role"] == "shadow"
+    assert len(lineage["input_digest"]) == 64
+    assert lineage["evidence_expiry_status"] == "not_declared"
+    assert lineage["bounded_budget_status"] == "not_declared"
+    assert lineage["position_mutation_allowed"] is False
+    assert lineage["broker_order_allowed"] is False
+
+
+def test_plugin_lineage_declares_expiry_and_bounded_budget(tmp_path) -> None:
+    prices_path = tmp_path / "lineage_prices.csv"
+    _quiet_prices().to_csv(prices_path, index=False)
+    output_dir = tmp_path / "lineage"
+    config = {
+        "output_dir": str(tmp_path / "runner"),
+        "default_mode": "shadow",
+        "strategy_plugins": [{
+            "strategy": STRATEGY_NAME,
+            "plugin": PLUGIN_CRISIS_RESPONSE_SHADOW,
+            "inputs": {
+                "prices": str(prices_path),
+                "evidence_valid_until": "2026-12-31",
+                "bounded_budget": {"max_runs": 1, "max_seconds": 30},
+            },
+            "outputs": {"output_dir": str(output_dir)},
+        }],
+    }
+    run_configured_plugins(config)
+    lineage = json.loads((output_dir / "latest_signal.json").read_text(encoding="utf-8"))["plugin_lineage"]
+    assert lineage["evidence_valid_until"] == "2026-12-31"
+    assert lineage["evidence_expiry_status"] == "declared"
+    assert lineage["bounded_budget"] == {"max_runs": 1, "max_seconds": 30}
+    assert lineage["bounded_budget_status"] == "declared"
 
 
 def test_strategy_plugin_runner_runs_macro_risk_governor_for_tqqq(tmp_path) -> None: