require auditable plugin position control (#190)

Co-authored-by: Codex <noreply@openai.com>

Author: Pigbibi

docs/strategy_plugin_runtime_contract.md

@@ -204,6 +204,10 @@ the matching `notification_targets` artifact. The plugin-alert stream remains
 for non-delegated manual-review or notification-only cases, including
 `notification_targets`, `blocked`, `watch_only`, and `notify_manual_review`
 routes.
+Automatic position control also requires auditable evidence: an
+`evidence_package_id`, a validity field such as `evidence_valid_until`, and a
+`bounded_budget`; if any of those are missing, treat the signal as review-only
+or loaded-not-applied rather than a real auto-position-control signal.
 
 Platforms may still choose their delivery sinks, but shared escalation helpers
 are available for email, SMS, push, and Telegram:

docs/strategy_plugin_runtime_contract.zh-CN.md

@@ -182,6 +182,9 @@ notification target；这类委托告警只从对应 `notification_targets` arti
 发送一次。插件告警流只保留给未委托的人工复核或 notification-only 场景，
 包括 `notification_targets`、`blocked`、`watch_only` 和
 `notify_manual_review` 路线。
+自动仓位控制还必须带可审计证据：`evidence_package_id`、有效期字段
+（如 `evidence_valid_until`）和 `bounded_budget`；缺少任一项时，只能按
+普通复核/加载态处理，不能当作真正的自动仓位信号。
 
 平台仍可选择自己的投递 sink；共享 helper 已提供 email、SMS、push 和
 Telegram 的聚合入口：

src/quant_platform_kit/common/strategy_plugins.py

@@ -925,6 +925,72 @@ def _strategy_plugin_consumption_policy(signal: StrategyPluginSignal) -> Mapping
     return policy if isinstance(policy, Mapping) else {}
 
 
+def _strategy_plugin_value_is_present(value: Any) -> bool:
+    if value is None:
+        return False
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, (int, float)):
+        return value != 0
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        return bool(normalized) and normalized not in {"none", "null", "unbounded", "unlimited", "infinite", "inf"}
+    if isinstance(value, Mapping):
+        return any(_strategy_plugin_value_is_present(item) for item in value.values())
+    if isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        return any(_strategy_plugin_value_is_present(item) for item in value)
+    return bool(value)
+
+
+def _strategy_plugin_find_field_value(value: Any, field_names: frozenset[str]) -> Any | None:
+    if isinstance(value, Mapping):
+        for key, item in value.items():
+            if key in field_names and _strategy_plugin_value_is_present(item):
+                return item
+            found = _strategy_plugin_find_field_value(item, field_names)
+            if found is not None:
+                return found
+    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+        for item in value:
+            found = _strategy_plugin_find_field_value(item, field_names)
+            if found is not None:
+                return found
+    return None
+
+
+def _strategy_plugin_has_auditable_position_control(signal: StrategyPluginSignal) -> bool:
+    controls = _strategy_plugin_execution_controls(signal)
+    policy = _strategy_plugin_consumption_policy(signal)
+    if not (
+        _as_bool(controls.get("position_control_allowed"), default=False)
+        or _as_bool(policy.get("position_control_allowed"), default=False)
+    ):
+        return False
+    if _strategy_plugin_consumption_status(signal) != "automation_approved":
+        return False
+    payload = getattr(signal, "payload", {}) or {}
+    if not isinstance(payload, Mapping):
+        return False
+    evidence_sources: tuple[Any, ...] = (controls, payload)
+    return all(
+        _strategy_plugin_find_field_value(evidence_sources, field_names) is not None
+        for field_names in (
+            frozenset({"evidence_package_id", "evidence_package", "package_id"}),
+            frozenset(
+                {
+                    "evidence_valid_until",
+                    "evidence_valid_until_at",
+                    "evidence_expires_at",
+                    "valid_until",
+                    "valid_until_at",
+                    "expires_at",
+                }
+            ),
+            frozenset({"bounded_budget", "budget", "budget_limit"}),
+        )
+    )
+
+
 def _strategy_plugin_consumption_status(signal: StrategyPluginSignal) -> str:
     controls = _strategy_plugin_execution_controls(signal)
     status = _optional_string(controls.get("consumption_evidence_status"))
@@ -965,6 +1031,7 @@ def _strategy_plugin_auto_consumption_allowed(signal: StrategyPluginSignal, *, a
         action in STRATEGY_PLUGIN_AUTOMATED_POSITION_ACTIONS
         and _strategy_plugin_policy_allows_position_control(signal)
         and _strategy_plugin_consumption_status(signal) == "automation_approved"
+        and _strategy_plugin_has_auditable_position_control(signal)
     )
 
 
@@ -1141,10 +1208,7 @@ def _is_strategy_position_control_notice(signal: StrategyPluginSignal, *, action
         return False
     if not _as_bool(controls.get("strategy_runtime_metadata_allowed"), default=False):
         return False
-    if not _as_bool(controls.get("position_control_allowed"), default=False):
-        return False
-    evidence_status = _normalize_strategy_plugin_field(controls.get("consumption_evidence_status"))
-    return evidence_status == "automation_approved"
+    return _strategy_plugin_has_auditable_position_control(signal)
 
 
 def _is_strategy_manual_review_notification_delegated(signal: StrategyPluginSignal) -> bool:

tests/test_strategy_plugin_telegram_notifications.py

@@ -41,6 +41,13 @@ def _auto_consumable_signal():
             "strategy_runtime_metadata_allowed": True,
             "position_control_allowed": True,
             "consumption_evidence_status": "automation_approved",
+            "evidence_package_id": "pkg_001",
+            "evidence_valid_until": "2026-08-01T00:00:00Z",
+            "bounded_budget": {
+                "name": "position_control",
+                "amount": 0.4,
+                "unit": "fraction",
+            },
         },
     )
 

tests/test_strategy_plugins.py

@@ -72,6 +72,22 @@ def _signal_payload(*, strategy="tqqq_growth_income", plugin="crisis_response_sh
     }
 
 
+def _auditable_position_control_controls():
+    return {
+        **_signal_payload()["execution_controls"],
+        "strategy_runtime_metadata_allowed": True,
+        "position_control_allowed": True,
+        "consumption_evidence_status": "automation_approved",
+        "evidence_package_id": "pkg_001",
+        "evidence_valid_until": "2026-08-01T00:00:00Z",
+        "bounded_budget": {
+            "name": "position_control",
+            "amount": 0.4,
+            "unit": "fraction",
+        },
+    }
+
+
 class StrategyPluginsTests(unittest.TestCase):
     def test_parse_strategy_plugin_mounts_uses_artifact_mode_not_platform_mode(self):
         with tempfile.TemporaryDirectory() as tmp_dir:
@@ -625,12 +641,7 @@ def test_strategy_plugin_notification_lines_include_auto_consumption_line(self):
                 "canonical_route": "risk_off",
                 "suggested_action": "defend",
                 "would_trade_if_enabled": True,
-                "execution_controls": {
-                    **_signal_payload()["execution_controls"],
-                    "strategy_runtime_metadata_allowed": True,
-                    "position_control_allowed": True,
-                    "consumption_evidence_status": "automation_approved",
-                },
+                "execution_controls": _auditable_position_control_controls(),
             }
         )
         translations = {
@@ -657,6 +668,44 @@ def test_strategy_plugin_notification_lines_include_auto_consumption_line(self):
             ),
         )
 
+    def test_strategy_plugin_auto_position_control_without_auditable_evidence_is_not_auto_consumed(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
+                "canonical_route": "risk_off",
+                "suggested_action": "defend",
+                "would_trade_if_enabled": True,
+                "execution_controls": {
+                    **_signal_payload()["execution_controls"],
+                    "strategy_runtime_metadata_allowed": True,
+                    "position_control_allowed": True,
+                    "consumption_evidence_status": "automation_approved",
+                },
+            }
+        )
+        translations = {
+            "strategy_plugin_line": "plugin={plugin}|enabled={enabled}|route={route}|action={action}",
+            "strategy_plugin_enabled_true": "yes",
+            "strategy_plugin_name_market_regime_control": "Market Regime",
+            "strategy_plugin_route_risk_off": "risk off",
+            "strategy_plugin_action_defend": "defend",
+            "strategy_plugin_consumption_loaded_not_applied": "consumption=loaded-not-applied",
+        }
+
+        self.assertTrue(should_alert_strategy_plugin_signal(signal))
+        self.assertEqual(
+            build_strategy_plugin_notification_lines(
+                [signal],
+                translator=lambda key, **kwargs: translations.get(key, key).format(**kwargs)
+                if kwargs
+                else translations.get(key, key),
+            ),
+            (
+                "plugin=Market Regime|enabled=yes|route=risk off|action=defend",
+                "consumption=loaded-not-applied",
+            ),
+        )
+
     def test_strategy_plugin_notification_lines_include_review_only_consumption_line(self):
         signal = validate_strategy_plugin_signal_payload(
             {
@@ -820,12 +869,7 @@ def test_strategy_plugin_auto_position_control_signal_stays_with_strategy_notifi
                 "canonical_route": "risk_off",
                 "suggested_action": "defend",
                 "would_trade_if_enabled": True,
-                "execution_controls": {
-                    **_signal_payload()["execution_controls"],
-                    "strategy_runtime_metadata_allowed": True,
-                    "position_control_allowed": True,
-                    "consumption_evidence_status": "automation_approved",
-                },
+                "execution_controls": _auditable_position_control_controls(),
             }
         )
 
@@ -890,6 +934,13 @@ def test_delegated_manual_review_strategy_signal_stays_with_notification_target(
                     "strategy_runtime_metadata_allowed": True,
                     "position_control_allowed": True,
                     "consumption_evidence_status": "automation_approved",
+                    "evidence_package_id": "pkg_001",
+                    "evidence_valid_until": "2026-08-01T00:00:00Z",
+                    "bounded_budget": {
+                        "name": "position_control",
+                        "amount": 0.4,
+                        "unit": "fraction",
+                    },
                     "manual_review_notification_delegated": True,
                     "manual_review_notification_target": GENERAL_MARKET_REGIME_NOTIFICATION_TARGET,
                     "manual_review_notification_delegate": (