fix: limit plugin telegram alerts to manual review

src/quant_platform_kit/notifications/strategy_plugin_telegram.py

@@ -21,6 +21,8 @@
 
 _DEFAULT_TELEGRAM_BODY_MAX_CHARS = 3900
 _MARKET_REGIME_CONTROL_PLUGIN = "market_regime_control"
+_AUTOMATED_POSITION_ACTIONS = frozenset({"defend", "delever"})
+_MANUAL_REVIEW_ACTIONS = frozenset({"notify_manual_review"})
 _ZH_MARKET_REGIME_ROUTE_LABELS = {
     "true_crisis": "黑天鹅",
     "crisis": "黑天鹅",
@@ -164,11 +166,15 @@ def publish_strategy_plugin_telegram_alerts(
     log_message: Callable[..., Any] = print,
 ) -> StrategyPluginTelegramAlertPublishResult:
     settings = StrategyPluginTelegramSettings.from_object(telegram_settings)
+    del context_label  # The plugin Telegram bot is a unified manual-review channel.
+    manual_review_signals = tuple(
+        signal for signal in signals if _is_manual_review_telegram_signal(signal)
+    )
     messages = build_strategy_plugin_alert_messages(
-        signals,
+        manual_review_signals,
         translator=translator,
         strategy_label=strategy_label,
-        context_label=context_label,
+        context_label=None,
         alert_namespace="strategy_plugin_telegram_alert",
     )
     deliveries: list[StrategyPluginTelegramAlertDelivery] = []
@@ -210,6 +216,63 @@ def publish_strategy_plugin_telegram_alerts(
     return result
 
 
+def _is_manual_review_telegram_signal(signal: object) -> bool:
+    if _is_auto_consumable_signal(signal):
+        return False
+    target_type = str(getattr(signal, "target_type", None) or "strategy").strip().lower()
+    if target_type == "notification_target":
+        return True
+    action = str(getattr(signal, "suggested_action", None) or "").strip().lower()
+    if action in _MANUAL_REVIEW_ACTIONS:
+        return True
+    controls = _signal_execution_controls(signal)
+    policy = _signal_consumption_policy(signal)
+    status = str(
+        controls.get("consumption_evidence_status")
+        or policy.get("evidence_status")
+        or ""
+    ).strip().lower()
+    if status == "notification_only":
+        return True
+    for key in ("capital_impact", "notification_profile"):
+        value = str(controls.get(key) or policy.get(key) or "").strip().lower()
+        if value in {"notification_only", "shadow_only"}:
+            return True
+    return False
+
+
+def _is_auto_consumable_signal(signal: object) -> bool:
+    action = str(getattr(signal, "suggested_action", None) or "").strip().lower()
+    if action not in _AUTOMATED_POSITION_ACTIONS:
+        return False
+    controls = _signal_execution_controls(signal)
+    policy = _signal_consumption_policy(signal)
+    status = str(
+        controls.get("consumption_evidence_status")
+        or policy.get("evidence_status")
+        or ""
+    ).strip().lower()
+    if status != "automation_approved":
+        return False
+    return (
+        _coerce_bool(controls.get("position_control_allowed"), default=False)
+        or _coerce_bool(policy.get("position_control_allowed"), default=False)
+    )
+
+
+def _signal_execution_controls(signal: object) -> Mapping[str, Any]:
+    controls = getattr(signal, "execution_controls", {}) or {}
+    return controls if isinstance(controls, Mapping) else {}
+
+
+def _signal_consumption_policy(signal: object) -> 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 _delivery(
     message: StrategyPluginAlertMessage,
     *,
@@ -277,6 +340,7 @@ def _build_compact_market_regime_body(message: StrategyPluginAlertMessage) -> st
     if use_zh:
         return "\n".join(
             (
+                "统一人工复核信号",
                 f"日期：{as_of}",
                 f"市场状态：{_market_regime_route_label(route, use_zh=True)}",
                 f"背景情况：{background}",
@@ -285,6 +349,7 @@ def _build_compact_market_regime_body(message: StrategyPluginAlertMessage) -> st
         )
     return "\n".join(
         (
+            "Unified manual-review signal",
             f"Date: {as_of}",
             f"Market state: {_market_regime_route_label(route, use_zh=False)}",
             f"Background: {background}",
@@ -420,5 +485,3 @@ def _coerce_int(value: Any, default: int) -> int:
 
 def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str:
     return "strategy_plugin_telegram_alert/" + _clean_relative_key(message.subject or "unknown")
-
-

tests/test_strategy_plugin_alert_dispatcher.py

@@ -15,12 +15,13 @@
 def _alert_signal():
     return SimpleNamespace(
         strategy="tqqq_growth_income",
-        plugin="crisis_response_shadow",
+        plugin="market_regime_control",
         effective_mode="shadow",
         as_of="2026-05-24",
-        canonical_route="true_crisis",
-        suggested_action="defend",
-        would_trade_if_enabled=True,
+        canonical_route="opportunity_watch",
+        suggested_action="notify_manual_review",
+        would_trade_if_enabled=False,
+        execution_controls={"notification_profile": "shadow_only"},
     )
 
 
@@ -199,6 +200,8 @@ def test_publish_strategy_plugin_alerts_can_target_telegram_channel(self):
         self.assertIsNotNone(result.telegram_result)
         self.assertEqual(result.sent_count, 1)
         self.assertEqual(telegram_messages[0]["chat_ids"], ("123456",))
+        self.assertIn("Unified manual-review signal", telegram_messages[0]["body"])
+        self.assertNotIn("ibkr / live-slot-a", telegram_messages[0]["body"])
 
     def test_publish_strategy_plugin_alerts_reads_channels_from_settings(self):
         settings = _NotificationSettings()

tests/test_strategy_plugin_telegram_notifications.py

@@ -18,12 +18,30 @@
 def _alert_signal():
     return SimpleNamespace(
         strategy="tqqq_growth_income",
-        plugin="crisis_response_shadow",
+        plugin="market_regime_control",
         effective_mode="shadow",
         as_of="2026-05-24",
-        canonical_route="true_crisis",
+        canonical_route="opportunity_watch",
+        suggested_action="notify_manual_review",
+        would_trade_if_enabled=False,
+        execution_controls={"notification_profile": "shadow_only"},
+    )
+
+
+def _auto_consumable_signal():
+    return SimpleNamespace(
+        strategy="tqqq_growth_income",
+        plugin="market_regime_control",
+        effective_mode="shadow",
+        as_of="2026-05-24",
+        canonical_route="risk_off",
         suggested_action="defend",
         would_trade_if_enabled=True,
+        execution_controls={
+            "strategy_runtime_metadata_allowed": True,
+            "position_control_allowed": True,
+            "consumption_evidence_status": "automation_approved",
+        },
     )
 
 
@@ -172,9 +190,27 @@ def test_publish_strategy_plugin_telegram_alerts_sends_and_dedupes(self):
         self.assertEqual(second.skipped_count, 1)
         self.assertEqual(second.deliveries[0].reason, "duplicate_alert")
         self.assertEqual(calls[0]["chat_ids"], ("123",))
-        self.assertIn("ibkr / live-slot-a", calls[0]["body"])
+        self.assertIn("Unified manual-review signal", calls[0]["body"])
+        self.assertNotIn("ibkr / live-slot-a", calls[0]["body"])
         self.assertTrue(tmp_dir)
 
+    def test_publish_strategy_plugin_telegram_alerts_skips_auto_consumable_signals(self):
+        calls = []
+
+        result = publish_strategy_plugin_telegram_alerts(
+            [_auto_consumable_signal()],
+            telegram_settings=StrategyPluginTelegramSettings(
+                chat_ids=("123",),
+                bot_token="bot-token",
+            ),
+            send_notification=lambda **kwargs: calls.append(kwargs) or True,
+            log_message=lambda *_args, **_kwargs: None,
+        )
+
+        self.assertEqual(result.attempted_count, 0)
+        self.assertEqual(result.sent_count, 0)
+        self.assertEqual(calls, [])
+
     def test_publish_strategy_plugin_telegram_alerts_compacts_market_regime_body(self):
         calls = []
         translations = STRATEGY_PLUGIN_I18N["zh"]
@@ -218,6 +254,7 @@ def test_publish_strategy_plugin_telegram_alerts_compacts_market_regime_body(sel
             calls[0]["body"],
             "\n".join(
                 (
+                    "统一人工复核信号",
                     "日期：2026-06-19",
                     "市场状态：抄底机会",
                     "背景情况：机会观察：出现可能的反弹或低吸窗口，当前证据只够人工复核。",