diff --git a/src/quant_strategy_plugins/plugin_lineage.py b/src/quant_strategy_plugins/plugin_lineage.py new file mode 100644 index 0000000..5a8158f --- /dev/null +++ b/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, + } diff --git a/src/quant_strategy_plugins/strategy_plugin_runner.py b/src/quant_strategy_plugins/strategy_plugin_runner.py index 0a35297..00b06ce 100644 --- a/src/quant_strategy_plugins/strategy_plugin_runner.py +++ b/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, diff --git a/tests/test_strategy_plugin_runner.py b/tests/test_strategy_plugin_runner.py index a2fa0ef..bbf246f 100644 --- a/tests/test_strategy_plugin_runner.py +++ b/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: