Use shared plugin email alert publisher

Author: Pigbibi

README.md

@@ -216,6 +216,7 @@ Current behavior is fail-fast:
 - missing key fields in the selected group (`ib_gateway_instance_name`, `ib_gateway_mode`, `ib_client_id`) → startup error
 
 When `IBKR_STRATEGY_PLUGIN_MOUNTS_JSON` includes the `crisis_response_shadow` plugin, the normal strategy-cycle Telegram message still includes the compact plugin line. If the plugin signal escalates beyond `no_action` (for example `canonical_route=true_crisis`, `suggested_action=defend`/`blocked`, or `would_trade_if_enabled=true`), the service also sends an independent crisis email when the `CRISIS_ALERT_*` SMTP settings are complete.
+Email alert results are written into the runtime report. Duplicate suppression uses stable plugin alert keys and stores markers under `STRATEGY_PLUGIN_ALERT_STATE_GCS_URI` when set, otherwise `EXECUTION_REPORT_GCS_URI`, with a local `/tmp` marker fallback.
 
 ### GitHub-managed Cloud Run env sync
 
@@ -433,6 +434,7 @@ IB_GATEWAY_IP_MODE=internal
 - 选中的账号组缺少关键字段（`ib_gateway_instance_name`、`ib_gateway_mode`、`ib_client_id`）→ 启动直接报错
 
 如果 `IBKR_STRATEGY_PLUGIN_MOUNTS_JSON` 挂载了 `crisis_response_shadow` 插件，常规策略周期 Telegram 仍会包含插件摘要行。当插件信号升级到非 `no_action`（例如 `canonical_route=true_crisis`、`suggested_action=defend`/`blocked`，或 `would_trade_if_enabled=true`）时，只要 `CRISIS_ALERT_*` SMTP 配置完整，服务还会额外发一封独立危机邮件。
+邮件告警结果会写入 runtime report。重复发送抑制使用稳定的插件告警 key；如配置了 `STRATEGY_PLUGIN_ALERT_STATE_GCS_URI` 则写入该前缀，否则复用 `EXECUTION_REPORT_GCS_URI`，并有本地 `/tmp` marker fallback。
 
 ### GitHub 统一管理 Cloud Run 环境变量
 

main.py

@@ -25,7 +25,11 @@
 from decision_mapper import map_strategy_decision
 from entrypoints.cloud_run import is_market_open_today
 from notifications.telegram import build_strategy_display_name, build_translator, send_telegram_message
-from quant_platform_kit.notifications.email import send_smtp_email
+from quant_platform_kit.notifications.strategy_plugin_email import (
+    StrategyPluginEmailAlertMarkerStore,
+    build_strategy_plugin_alert_context_label as build_email_alert_context_label,
+    publish_strategy_plugin_email_alerts,
+)
 from quant_platform_kit.common.runtime_assembly import build_runtime_assembly
 from quant_platform_kit.common.runtime_reports import (
     append_runtime_report_error,
@@ -233,14 +237,6 @@ def _env_flag(name: str) -> bool:
 TG_TOKEN = RUNTIME_SETTINGS.tg_token
 TG_CHAT_ID = RUNTIME_SETTINGS.tg_chat_id
 NOTIFY_LANG = RUNTIME_SETTINGS.notify_lang
-CRISIS_ALERT_EMAIL_TO = getattr(RUNTIME_SETTINGS, "crisis_alert_email_to", ())
-CRISIS_ALERT_EMAIL_FROM = getattr(RUNTIME_SETTINGS, "crisis_alert_email_from", None)
-CRISIS_ALERT_SMTP_HOST = getattr(RUNTIME_SETTINGS, "crisis_alert_smtp_host", None)
-CRISIS_ALERT_SMTP_PORT = getattr(RUNTIME_SETTINGS, "crisis_alert_smtp_port", 587)
-CRISIS_ALERT_SMTP_USERNAME = getattr(RUNTIME_SETTINGS, "crisis_alert_smtp_username", None)
-CRISIS_ALERT_SMTP_PASSWORD = getattr(RUNTIME_SETTINGS, "crisis_alert_smtp_password", None)
-CRISIS_ALERT_SMTP_STARTTLS = getattr(RUNTIME_SETTINGS, "crisis_alert_smtp_starttls", True)
-CRISIS_ALERT_SMTP_SSL = getattr(RUNTIME_SETTINGS, "crisis_alert_smtp_ssl", False)
 
 CASH_RESERVE_RATIO = STRATEGY_RUNTIME.cash_reserve_ratio
 CASH_RESERVE_FLOOR_USD = getattr(STRATEGY_RUNTIME, "cash_reserve_floor_usd", 0.0)
@@ -481,29 +477,42 @@ def build_strategy_plugin_alert_messages(signals):
     return build_strategy_adapters().build_strategy_plugin_alert_messages(signals)
 
 
-def send_crisis_alert_email(alert_message) -> bool:
-    return send_smtp_email(
-        subject=alert_message.subject,
-        body=alert_message.body,
-        smtp_host=CRISIS_ALERT_SMTP_HOST,
-        smtp_port=CRISIS_ALERT_SMTP_PORT,
-        sender=CRISIS_ALERT_EMAIL_FROM,
-        recipients=CRISIS_ALERT_EMAIL_TO,
-        username=CRISIS_ALERT_SMTP_USERNAME,
-        password=CRISIS_ALERT_SMTP_PASSWORD,
-        use_starttls=CRISIS_ALERT_SMTP_STARTTLS,
-        use_ssl=CRISIS_ALERT_SMTP_SSL,
+def build_strategy_plugin_alert_store():
+    return StrategyPluginEmailAlertMarkerStore(
+        local_dir=os.getenv("STRATEGY_PLUGIN_ALERT_STATE_DIR") or "/tmp/quant_strategy_plugin_alerts",
+        gcs_prefix_uri=os.getenv("STRATEGY_PLUGIN_ALERT_STATE_GCS_URI") or os.getenv("EXECUTION_REPORT_GCS_URI"),
+        gcp_project_id=PROJECT_ID,
     )
 
 
-def publish_strategy_plugin_alerts(signals) -> int:
-    sent_count = 0
-    for alert_message in build_strategy_plugin_alert_messages(signals):
-        if send_crisis_alert_email(alert_message):
-            sent_count += 1
-    if sent_count:
-        print(f"strategy_plugin_alert_email_sent count={sent_count}", flush=True)
-    return sent_count
+def build_strategy_plugin_alert_context_label() -> str:
+    return build_email_alert_context_label(
+        platform_id="ibkr",
+        strategy_profile=STRATEGY_PROFILE,
+        account_scope=ACCOUNT_GROUP,
+        service_name=SERVICE_NAME,
+        runtime_target=RUNTIME_SETTINGS.runtime_target,
+    )
+
+
+def attach_strategy_plugin_alert_email_result(report, result) -> None:
+    report.setdefault("summary", {})["strategy_plugin_alert_email_sent_count"] = result.sent_count
+    report.setdefault("diagnostics", {}).update(result.to_report_fields())
+
+
+def publish_strategy_plugin_alerts(signals, *, report=None):
+    result = publish_strategy_plugin_email_alerts(
+        signals,
+        email_settings=RUNTIME_SETTINGS,
+        translator=t,
+        strategy_label=STRATEGY_PROFILE,
+        context_label=build_strategy_plugin_alert_context_label(),
+        alert_store=build_strategy_plugin_alert_store(),
+        log_message=print,
+    )
+    if report is not None:
+        attach_strategy_plugin_alert_email_result(report, result)
+    return result
 
 
 def build_account_notification_lines() -> tuple[str, ...]:
@@ -640,7 +649,7 @@ def _handle_request(*, dry_run_only_override: bool | None = None, response_body:
             execution_window="precheck" if dry_run_only_override else "execution",
         )
         if dry_run_only_override is None:
-            publish_strategy_plugin_alerts(strategy_plugin_signals)
+            publish_strategy_plugin_alerts(strategy_plugin_signals, report=report)
         cycle_result = coerce_strategy_cycle_result(
             run_strategy_core(
                 strategy_plugin_signals=strategy_plugin_signals,

requirements.txt

@@ -1,6 +1,6 @@
 flask
 gunicorn
-quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@1b6febbba7df81179ad7579f430c26a811c0e1a8
+quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@ba67541711228f5a72a294def0e5cc24cc5479f3
 us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@1636271a3e0c17fc0c5da363f67eabe114eeff48
 pandas
 numpy

tests/test_request_handling.py

@@ -48,11 +48,21 @@ def test_handle_request_sends_escalated_strategy_plugin_alert(strategy_module, m
     monkeypatch.setattr(strategy_module, "is_market_open_today", lambda: True)
     monkeypatch.setattr(strategy_module, "load_strategy_plugin_signals", lambda: ((signal,), None))
     monkeypatch.setattr(strategy_module, "attach_strategy_plugin_report", lambda *args, **kwargs: None)
-    monkeypatch.setattr(
-        strategy_module,
-        "send_crisis_alert_email",
-        lambda alert_message: observed["alerts"].append(alert_message) or True,
-    )
+
+    def fake_publish(signals, **kwargs):
+        observed["alerts"].append((tuple(signals), kwargs))
+        return types.SimpleNamespace(
+            sent_count=1,
+            to_report_fields=lambda: {
+                "strategy_plugin_alert_email_attempted_count": 1,
+                "strategy_plugin_alert_email_sent_count": 1,
+                "strategy_plugin_alert_email_skipped_count": 0,
+                "strategy_plugin_alert_email_failed_count": 0,
+                "strategy_plugin_alert_email_deliveries": [],
+            },
+        )
+
+    monkeypatch.setattr(strategy_module, "publish_strategy_plugin_email_alerts", fake_publish)
     monkeypatch.setattr(strategy_module, "run_strategy_core", lambda **_kwargs: "OK - executed")
 
     with strategy_module.app.test_request_context("/", method="POST"):
@@ -61,7 +71,8 @@ def test_handle_request_sends_escalated_strategy_plugin_alert(strategy_module, m
     assert status == 200
     assert body == "OK - executed"
     assert len(observed["alerts"]) == 1
-    assert "Crisis" in observed["alerts"][0].subject
+    assert observed["alerts"][0][0] == (signal,)
+    assert "ibkr" in observed["alerts"][0][1]["context_label"]
 
 
 def test_handle_precheck_post_uses_dry_run_override(strategy_module, monkeypatch):