Improve strategy plugin consumption notifications

Author: Pigbibi

src/quant_platform_kit/common/notification_localization.py

@@ -29,7 +29,15 @@
         "strategy_plugin_alert_locale": "zh-CN",
         "strategy_plugin_alert_reason_joiner": "、",
         "strategy_plugin_alert_reason_none": "未提供明确原因",
-        "strategy_plugin_line": "🧩 插件：{plugin} | 状态：{route} | 提醒：{action}",
+        "strategy_plugin_line": "🧩 插件：{plugin} | 启用：{enabled} | 状态：{route} | 提醒：{action}",
+        "strategy_plugin_enabled_true": "是",
+        "strategy_plugin_enabled_false": "否",
+        "strategy_plugin_consumption_auto": "🧩 插件消费：已按策略规则参与本轮仓位计算",
+        "strategy_plugin_consumption_auto_defend": "🧩 插件消费：已按策略规则参与本轮仓位计算；风险仓位按防守规则处理",
+        "strategy_plugin_consumption_auto_delever": "🧩 插件消费：已按策略规则参与本轮仓位计算；杠杆仓位按降档规则缩放",
+        "strategy_plugin_consumption_loaded_not_applied": "🧩 插件消费：已加载但未改写仓位；当前策略未启用该状态的自动消费",
+        "strategy_plugin_consumption_review_only": "🧩 插件消费：仅通知复核，未参与自动仓位计算",
+        "strategy_plugin_consumption_unavailable": "🧩 插件消费：未消费插件信号",
         "strategy_plugin_error_line": "⚠️ 插件信号未加载：{reason}；本轮使用策略内置规则",
         "strategy_plugin_error_reason_JSONDecodeError": "插件配置格式错误",
         "strategy_plugin_error_reason_FileNotFoundError": "插件信号文件不存在",
@@ -119,7 +127,15 @@
         "strategy_plugin_alert_locale": "en-US",
         "strategy_plugin_alert_reason_joiner": ", ",
         "strategy_plugin_alert_reason_none": "no explicit reason provided",
-        "strategy_plugin_line": "🧩 Plugin: {plugin} | status: {route} | notice: {action}",
+        "strategy_plugin_line": "🧩 Plugin: {plugin} | enabled: {enabled} | status: {route} | notice: {action}",
+        "strategy_plugin_enabled_true": "yes",
+        "strategy_plugin_enabled_false": "no",
+        "strategy_plugin_consumption_auto": "🧩 Plugin consumption: included in this cycle's position calculation under strategy rules",
+        "strategy_plugin_consumption_auto_defend": "🧩 Plugin consumption: included in this cycle's position calculation; risk exposure follows defensive rules",
+        "strategy_plugin_consumption_auto_delever": "🧩 Plugin consumption: included in this cycle's position calculation; leveraged exposure follows de-risking rules",
+        "strategy_plugin_consumption_loaded_not_applied": "🧩 Plugin consumption: loaded but did not rewrite positions; this strategy does not enable automatic consumption for this state",
+        "strategy_plugin_consumption_review_only": "🧩 Plugin consumption: review-only notice, not used for automatic position calculation",
+        "strategy_plugin_consumption_unavailable": "🧩 Plugin consumption: no plugin signal consumed",
         "strategy_plugin_error_line": "⚠️ Plugin signal failed to load: {reason}; this run falls back to built-in strategy rules",
         "strategy_plugin_error_reason_JSONDecodeError": "invalid plugin mount JSON",
         "strategy_plugin_error_reason_FileNotFoundError": "plugin signal file not found",
@@ -283,7 +299,7 @@
 def merge_strategy_plugin_i18n(
     i18n: dict[str, dict[str, str]],
     *,
-    shared_wins: bool = True,
+    shared_wins: bool = False,
 ) -> dict[str, dict[str, str]]:
     merged = {str(locale): dict(values) for locale, values in i18n.items()}
     for locale, shared_values in STRATEGY_PLUGIN_I18N.items():

src/quant_platform_kit/common/strategy_plugins.py

@@ -892,8 +892,110 @@ def extract_strategy_plugin_localized_message(
                 default_locale = _optional_string(notification.get("default_locale"))
                 if default_locale:
                     localized = _optional_string(notification_messages.get(default_locale))
-                    if localized:
-                        return localized
+            if localized:
+                return localized
+    return None
+
+
+def _strategy_plugin_enabled_label(
+    signal: StrategyPluginSignal,
+    *,
+    translator: Callable[..., str] | None = None,
+) -> str:
+    enabled = _as_bool(getattr(signal, "enabled", True), default=True)
+    return _translate(
+        translator,
+        f"strategy_plugin_enabled_{str(enabled).lower()}",
+        fallback="yes" if enabled else "no",
+    )
+
+
+def _strategy_plugin_execution_controls(signal: StrategyPluginSignal) -> Mapping[str, Any]:
+    controls = getattr(signal, "execution_controls", {}) or {}
+    return controls if isinstance(controls, Mapping) else {}
+
+
+def _strategy_plugin_consumption_policy(signal: StrategyPluginSignal) -> Mapping[str, Any]:
+    payload = getattr(signal, "payload", {}) or {}
+    if not isinstance(payload, Mapping):
+        return {}
+    policy = payload.get("consumption_policy") or {}
+    return policy if isinstance(policy, Mapping) else {}
+
+
+def _strategy_plugin_consumption_status(signal: StrategyPluginSignal) -> str:
+    controls = _strategy_plugin_execution_controls(signal)
+    status = _optional_string(controls.get("consumption_evidence_status"))
+    if status is None:
+        status = _optional_string(_strategy_plugin_consumption_policy(signal).get("evidence_status"))
+    return _normalize_strategy_plugin_field(status)
+
+
+def _strategy_plugin_policy_allows_position_control(signal: StrategyPluginSignal) -> bool:
+    controls = _strategy_plugin_execution_controls(signal)
+    if "position_control_allowed" in controls:
+        return _as_bool(controls.get("position_control_allowed"), default=False)
+    return _as_bool(
+        _strategy_plugin_consumption_policy(signal).get("position_control_allowed"),
+        default=False,
+    )
+
+
+def _strategy_plugin_is_notification_only(signal: StrategyPluginSignal) -> bool:
+    controls = _strategy_plugin_execution_controls(signal)
+    policy = _strategy_plugin_consumption_policy(signal)
+    if _strategy_plugin_consumption_status(signal) == "notification_only":
+        return True
+    if str(controls.get("capital_impact") or policy.get("capital_impact") or "").strip().lower() == "notification_only":
+        return True
+    if str(controls.get("notification_profile") or policy.get("notification_profile") or "").strip().lower() == "shadow_only":
+        return True
+    return False
+
+
+def _strategy_plugin_auto_consumption_allowed(signal: StrategyPluginSignal, *, action: str) -> bool:
+    if _is_strategy_position_control_notice(signal, action=action):
+        return True
+    controls = _strategy_plugin_execution_controls(signal)
+    if controls:
+        return False
+    return bool(
+        action in STRATEGY_PLUGIN_AUTOMATED_POSITION_ACTIONS
+        and _strategy_plugin_policy_allows_position_control(signal)
+        and _strategy_plugin_consumption_status(signal) == "automation_approved"
+    )
+
+
+def _strategy_plugin_consumption_notification_line(
+    signal: StrategyPluginSignal,
+    *,
+    translator: Callable[..., str] | None = None,
+) -> str | None:
+    route = _normalize_strategy_plugin_field(getattr(signal, "canonical_route", None))
+    action = _normalize_strategy_plugin_field(getattr(signal, "suggested_action", None))
+    if _strategy_plugin_auto_consumption_allowed(signal, action=action):
+        if action == "defend" or route == "risk_off":
+            key = "strategy_plugin_consumption_auto_defend"
+        elif action == "delever" or route == "risk_reduced":
+            key = "strategy_plugin_consumption_auto_delever"
+        else:
+            key = "strategy_plugin_consumption_auto"
+        return _translate(translator, key, fallback="Plugin consumption: included in this cycle's position calculation")
+    if _strategy_plugin_is_notification_only(signal):
+        return _translate(
+            translator,
+            "strategy_plugin_consumption_review_only",
+            fallback="Plugin consumption: review-only notice, not used for automatic position calculation",
+        )
+    if action in STRATEGY_PLUGIN_AUTOMATED_POSITION_ACTIONS:
+        return _translate(
+            translator,
+            "strategy_plugin_consumption_loaded_not_applied",
+            fallback=(
+                "Plugin consumption: loaded but did not rewrite positions; this strategy does not enable "
+                "automatic consumption for this state"
+            ),
+        )
     return None
 
 
@@ -912,20 +1014,28 @@ def build_strategy_plugin_notification_lines(
         )
         if localized_line:
             lines.append(localized_line)
+            if translator is not None:
+                consumption_line = _strategy_plugin_consumption_notification_line(signal, translator=translator)
+                if consumption_line:
+                    lines.append(consumption_line)
             continue
         route = getattr(signal, "canonical_route", None) or "unknown_route"
         action = getattr(signal, "suggested_action", None) or "unknown_action"
         lines.append(
             _translate(
                 translator,
                 "strategy_plugin_line",
-                fallback="Plugin: {plugin} | status: {route} | notice: {action}",
+                fallback="Plugin: {plugin} | enabled: {enabled} | status: {route} | notice: {action}",
                 plugin=translate_strategy_plugin_value("name", getattr(signal, "plugin", None), translator=translator),
+                enabled=_strategy_plugin_enabled_label(signal, translator=translator),
                 mode=translate_strategy_plugin_value("mode", getattr(signal, "effective_mode", None), translator=translator),
                 route=translate_strategy_plugin_value("route", route, translator=translator),
                 action=translate_strategy_plugin_value("action", action, translator=translator),
             )
         )
+        consumption_line = _strategy_plugin_consumption_notification_line(signal, translator=translator)
+        if consumption_line:
+            lines.append(consumption_line)
     return tuple(lines)
 
 
@@ -985,6 +1095,11 @@ def build_strategy_plugin_error_notification_lines(
             fallback="Plugin signal failed to load: {reason}; this run falls back to built-in strategy rules",
             reason=reason,
         ),
+        _translate(
+            translator,
+            "strategy_plugin_consumption_unavailable",
+            fallback="Plugin consumption: no plugin signal consumed",
+        ),
     )
 
 

tests/test_notification_localization.py

@@ -80,7 +80,7 @@ def test_strategy_plugin_i18n_has_matching_locale_keys(self):
         self.assertIn("strategy_plugin_name_taco_rebound_shadow", STRATEGY_PLUGIN_I18N["zh"])
         self.assertEqual(STRATEGY_PLUGIN_I18N["zh"]["strategy_plugin_name_taco_rebound_shadow"], "TACO 反弹观察通知")
 
-    def test_merge_strategy_plugin_i18n_fills_and_overrides_shared_keys(self):
+    def test_merge_strategy_plugin_i18n_fills_missing_keys_without_overriding_callers(self):
         merged = merge_strategy_plugin_i18n(
             {
                 "zh": {
@@ -92,6 +92,20 @@ def test_merge_strategy_plugin_i18n_fills_and_overrides_shared_keys(self):
         )
 
         self.assertEqual(merged["zh"]["no_trades"], "无需调仓")
+        self.assertEqual(merged["zh"]["strategy_plugin_name_taco_rebound_shadow"], "TACO 旧观察通知")
+        self.assertEqual(merged["en"]["strategy_plugin_name_taco_rebound_shadow"], "TACO Rebound Watch Notice")
+
+    def test_merge_strategy_plugin_i18n_can_prefer_shared_keys(self):
+        merged = merge_strategy_plugin_i18n(
+            {
+                "zh": {
+                    "strategy_plugin_name_taco_rebound_shadow": "TACO 旧观察通知",
+                },
+                "en": {},
+            },
+            shared_wins=True,
+        )
+
         self.assertEqual(merged["zh"]["strategy_plugin_name_taco_rebound_shadow"], "TACO 反弹观察通知")
         self.assertEqual(merged["en"]["strategy_plugin_route_watch"], "watch")
 

tests/test_strategy_plugins.py

@@ -590,11 +590,12 @@ def test_attach_strategy_plugin_metadata_ignores_shadow_artifact_without_runtime
     def test_strategy_plugin_notification_lines_use_translator_when_available(self):
         signal = validate_strategy_plugin_signal_payload(_signal_payload())
         translations = {
-            "strategy_plugin_line": "plugin={plugin}|mode={mode}|route={route}|action={action}",
+            "strategy_plugin_line": "plugin={plugin}|enabled={enabled}|route={route}|action={action}",
             "strategy_plugin_error_line": "plugin-error={reason}|fallback=built-in",
             "strategy_plugin_error_reason_ValueError": "config validation failed",
+            "strategy_plugin_consumption_unavailable": "consumption=none",
+            "strategy_plugin_enabled_true": "yes",
             "strategy_plugin_name_crisis_response_shadow": "Crisis",
-            "strategy_plugin_mode_shadow": "shadow",
             "strategy_plugin_route_no_action": "no action",
             "strategy_plugin_action_watch_only": "watch only",
         }
@@ -606,15 +607,134 @@ def test_strategy_plugin_notification_lines_use_translator_when_available(self):
             else translations.get(key, key),
         )
 
-        self.assertEqual(lines, ("plugin=Crisis|mode=shadow|route=no action|action=watch only",))
+        self.assertEqual(lines, ("plugin=Crisis|enabled=yes|route=no action|action=watch only",))
         self.assertEqual(
             build_strategy_plugin_error_notification_lines(
                 "ValueError: bad config\ntrace line",
                 translator=lambda key, **kwargs: translations.get(key, key).format(**kwargs)
                 if kwargs
                 else translations.get(key, key),
             ),
-            ("plugin-error=config validation failed|fallback=built-in",),
+            ("plugin-error=config validation failed|fallback=built-in", "consumption=none"),
+        )
+
+    def test_strategy_plugin_notification_lines_include_auto_consumption_line(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_auto_defend": "consumption=auto-defend",
+        }
+
+        lines = build_strategy_plugin_notification_lines(
+            [signal],
+            translator=lambda key, **kwargs: translations.get(key, key).format(**kwargs)
+            if kwargs
+            else translations.get(key, key),
+        )
+
+        self.assertEqual(
+            lines,
+            (
+                "plugin=Market Regime|enabled=yes|route=risk off|action=defend",
+                "consumption=auto-defend",
+            ),
+        )
+
+    def test_strategy_plugin_notification_lines_include_review_only_consumption_line(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
+                "canonical_route": "opportunity_watch",
+                "suggested_action": "notify_manual_review",
+                "would_trade_if_enabled": False,
+                "execution_controls": {
+                    **_signal_payload()["execution_controls"],
+                    "strategy_runtime_metadata_allowed": False,
+                    "position_control_allowed": False,
+                    "consumption_evidence_status": "notification_only",
+                    "capital_impact": "notification_only",
+                },
+            }
+        )
+        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_opportunity_watch": "opportunity watch",
+            "strategy_plugin_action_notify_manual_review": "manual review",
+            "strategy_plugin_consumption_review_only": "consumption=review-only",
+        }
+
+        lines = build_strategy_plugin_notification_lines(
+            [signal],
+            translator=lambda key, **kwargs: translations.get(key, key).format(**kwargs)
+            if kwargs
+            else translations.get(key, key),
+        )
+
+        self.assertEqual(
+            lines,
+            (
+                "plugin=Market Regime|enabled=yes|route=opportunity watch|action=manual review",
+                "consumption=review-only",
+            ),
+        )
+
+    def test_strategy_plugin_notification_lines_include_loaded_not_applied_consumption_line(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_MARKET_REGIME_CONTROL),
+                "canonical_route": "risk_reduced",
+                "suggested_action": "delever",
+                "would_trade_if_enabled": True,
+                "execution_controls": {
+                    **_signal_payload()["execution_controls"],
+                    "strategy_runtime_metadata_allowed": False,
+                    "position_control_allowed": False,
+                    "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_reduced": "risk reduced",
+            "strategy_plugin_action_delever": "de-lever",
+            "strategy_plugin_consumption_review_only": "consumption=review-only",
+            "strategy_plugin_consumption_loaded_not_applied": "consumption=loaded-not-applied",
+        }
+
+        lines = build_strategy_plugin_notification_lines(
+            [signal],
+            translator=lambda key, **kwargs: translations.get(key, key).format(**kwargs)
+            if kwargs
+            else translations.get(key, key),
+        )
+
+        self.assertEqual(
+            lines,
+            (
+                "plugin=Market Regime|enabled=yes|route=risk reduced|action=de-lever",
+                "consumption=loaded-not-applied",
+            ),
         )
 
     def test_strategy_plugin_notification_lines_can_use_artifact_localized_message(self):