diff --git a/src/quant_platform_kit/notifications/strategy_plugin_telegram.py b/src/quant_platform_kit/notifications/strategy_plugin_telegram.py index aaea086..93fe259 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_telegram.py +++ b/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") - - diff --git a/tests/test_strategy_plugin_alert_dispatcher.py b/tests/test_strategy_plugin_alert_dispatcher.py index c6a3ba6..4e47b74 100644 --- a/tests/test_strategy_plugin_alert_dispatcher.py +++ b/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() diff --git a/tests/test_strategy_plugin_telegram_notifications.py b/tests/test_strategy_plugin_telegram_notifications.py index 51c3fcd..0d2f7e4 100644 --- a/tests/test_strategy_plugin_telegram_notifications.py +++ b/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", "市场状态：抄底机会", "背景情况：机会观察：出现可能的反弹或低吸窗口，当前证据只够人工复核。",