Add shared strategy plugin alert helpers

Add shared strategy-plugin alert policy, alert message builders, and SMTP email helper for platform-neutral crisis plugin notifications.

Author: Pigbibi

docs/strategy_plugin_runtime_contract.md

@@ -63,6 +63,8 @@ Use `quant_platform_kit.common.strategy_plugins`:
 
 ```python
 from quant_platform_kit.common.strategy_plugins import (
+    build_strategy_plugin_alert_messages,
+    build_strategy_plugin_notification_lines,
     build_strategy_plugin_report_payload,
     load_configured_strategy_plugin_signals,
     parse_strategy_plugin_mounts,
@@ -74,6 +76,8 @@ signals = load_configured_strategy_plugin_signals(
     strategy_profile=current_strategy_profile,
 )
 report_section = build_strategy_plugin_report_payload(signals)
+notification_lines = build_strategy_plugin_notification_lines(signals)
+alert_messages = build_strategy_plugin_alert_messages(signals)
 ```
 
 The loader validates:
@@ -93,3 +97,18 @@ notification context.
 `paper`, `advisory`, and `live` plugin modes are not supported by the shared
 contract. Platforms should not maintain plugin ledgers or execute plugin-driven
 allocation changes from this sidecar path.
+
+## Escalated Alerts
+
+The shared kit owns the platform-neutral alert policy. A plugin signal escalates
+when any of the following is true:
+
+- `canonical_route` is not `no_action`
+- `suggested_action` is `defend` or `blocked`
+- `would_trade_if_enabled` is `true`
+
+Platforms may still choose their delivery sinks, but should use
+`build_strategy_plugin_alert_messages()` for the subject/body and
+`quant_platform_kit.notifications.email.send_smtp_email()` when SMTP email is
+configured. This keeps the Crisis Response plugin behavior consistent across
+IBKR, Schwab, LongBridge, Firstrade, and future platform runtimes.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "quant-platform-kit"
-version = "0.7.23"
+version = "0.7.24"
 description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies."
 readme = "README.md"
 requires-python = ">=3.9"

src/quant_platform_kit/common/__init__.py

@@ -40,14 +40,21 @@
 )
 from .strategy_plugins import (
     PLUGIN_MODE_SHADOW,
+    STRATEGY_PLUGIN_ALERT_ACTIONS,
+    STRATEGY_PLUGIN_NON_ALERT_ROUTES,
     SUPPORTED_STRATEGY_PLUGIN_MODES,
+    StrategyPluginAlertMessage,
     StrategyPluginMountConfig,
     StrategyPluginSignal,
+    build_strategy_plugin_alert_messages,
+    build_strategy_plugin_notification_lines,
     build_strategy_plugin_report_payload,
     load_configured_strategy_plugin_signals,
     load_strategy_plugin_signal,
     normalize_strategy_plugin_mode,
     parse_strategy_plugin_mounts,
+    should_alert_strategy_plugin_signal,
+    translate_strategy_plugin_value,
     validate_strategy_plugin_signal_payload,
 )
 
@@ -66,6 +73,8 @@
     "STAGE_PARTIAL_SUBMITTED",
     "STAGE_RECONCILED",
     "STAGE_SUBMITTED",
+    "STRATEGY_PLUGIN_ALERT_ACTIONS",
+    "STRATEGY_PLUGIN_NON_ALERT_ROUTES",
     "SUPPORTED_STRATEGY_PLUGIN_MODES",
     "filter_execution_blocking_skips",
     "is_terminal_funding_block",
@@ -82,15 +91,20 @@
     "RuntimeLogContext",
     "RuntimeAssembly",
     "build_runtime_assembly",
+    "StrategyPluginAlertMessage",
     "StrategyPluginMountConfig",
     "StrategyPluginSignal",
+    "build_strategy_plugin_alert_messages",
+    "build_strategy_plugin_notification_lines",
     "build_strategy_plugin_report_payload",
     "build_runtime_target",
     "load_configured_strategy_plugin_signals",
     "load_strategy_plugin_signal",
     "normalize_strategy_plugin_mode",
     "parse_strategy_plugin_mounts",
     "resolve_runtime_target_from_env",
+    "should_alert_strategy_plugin_signal",
+    "translate_strategy_plugin_value",
     "translator_uses_zh",
     "validate_strategy_plugin_signal_payload",
 ]

src/quant_platform_kit/common/strategy_plugins.py

@@ -8,11 +8,13 @@
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any
+from typing import Any, Callable
 
 PLUGIN_MODE_SHADOW = "shadow"
 SUPPORTED_STRATEGY_PLUGIN_MODES = frozenset({PLUGIN_MODE_SHADOW})
 DEFAULT_PLUGIN_ARTIFACT_CACHE_DIR = Path(tempfile.gettempdir()) / "quant_strategy_plugin_artifacts"
+STRATEGY_PLUGIN_NON_ALERT_ROUTES = frozenset({"no_action"})
+STRATEGY_PLUGIN_ALERT_ACTIONS = frozenset({"defend", "blocked"})
 
 
 @dataclass(frozen=True)
@@ -59,6 +61,12 @@ def report_summary(self) -> dict[str, Any]:
         }
 
 
+@dataclass(frozen=True)
+class StrategyPluginAlertMessage:
+    subject: str
+    body: str
+
+
 def normalize_strategy_plugin_mode(value: Any, *, field_name: str = "mode") -> str:
     mode = str(value or "").strip().lower()
     if mode not in SUPPORTED_STRATEGY_PLUGIN_MODES:
@@ -229,6 +237,121 @@ def build_strategy_plugin_report_payload(signals: Sequence[StrategyPluginSignal]
     }
 
 
+def translate_strategy_plugin_value(
+    category: str,
+    raw_value: str | None,
+    *,
+    translator: Callable[..., str] | None = None,
+) -> str:
+    value = str(raw_value or "").strip() or "unknown"
+    if translator is None:
+        return value
+    key = f"strategy_plugin_{category}_{value}"
+    translated = translator(key)
+    return translated if translated != key else value
+
+
+def build_strategy_plugin_notification_lines(
+    signals: Sequence[StrategyPluginSignal],
+    *,
+    translator: Callable[..., str] | None = None,
+) -> tuple[str, ...]:
+    lines: list[str] = []
+    for signal in signals:
+        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}",
+                plugin=translate_strategy_plugin_value("name", getattr(signal, "plugin", None), 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),
+            )
+        )
+    return tuple(lines)
+
+
+def should_alert_strategy_plugin_signal(signal: StrategyPluginSignal) -> bool:
+    route = _normalize_strategy_plugin_field(getattr(signal, "canonical_route", None))
+    action = _normalize_strategy_plugin_field(getattr(signal, "suggested_action", None))
+    return (
+        bool(getattr(signal, "would_trade_if_enabled", False))
+        or route not in STRATEGY_PLUGIN_NON_ALERT_ROUTES
+        or action in STRATEGY_PLUGIN_ALERT_ACTIONS
+    )
+
+
+def build_strategy_plugin_alert_messages(
+    signals: Sequence[StrategyPluginSignal],
+    *,
+    translator: Callable[..., str] | None = None,
+    strategy_label: str | None = None,
+) -> tuple[StrategyPluginAlertMessage, ...]:
+    messages: list[StrategyPluginAlertMessage] = []
+    for signal in signals:
+        if not should_alert_strategy_plugin_signal(signal):
+            continue
+        route = getattr(signal, "canonical_route", None) or "unknown_route"
+        action = getattr(signal, "suggested_action", None) or "unknown_action"
+        plugin = translate_strategy_plugin_value("name", getattr(signal, "plugin", None), translator=translator)
+        translated_route = translate_strategy_plugin_value("route", route, translator=translator)
+        translated_action = translate_strategy_plugin_value("action", action, translator=translator)
+        strategy = str(strategy_label or getattr(signal, "strategy", None) or "").strip() or "unknown"
+        subject = _translate(
+            translator,
+            "strategy_plugin_alert_subject",
+            fallback="Strategy plugin alert: {plugin} | {route}",
+            strategy=strategy,
+            plugin=plugin,
+            route=translated_route,
+        )
+        body_lines = [
+            _translate(translator, "strategy_plugin_alert_title", fallback="Strategy Plugin Alert"),
+            _translate(
+                translator,
+                "strategy_plugin_line",
+                fallback="Plugin: {plugin} | status: {route} | notice: {action}",
+                plugin=plugin,
+                mode=translate_strategy_plugin_value("mode", getattr(signal, "effective_mode", None), translator=translator),
+                route=translated_route,
+                action=translated_action,
+            ),
+            _translate(
+                translator,
+                "strategy_plugin_alert_strategy",
+                fallback="Strategy: {strategy}",
+                strategy=strategy,
+            ),
+            _translate(
+                translator,
+                "strategy_plugin_alert_as_of",
+                fallback="Signal as-of: {as_of}",
+                as_of=getattr(signal, "as_of", None) or "unknown",
+            ),
+            _translate(
+                translator,
+                "strategy_plugin_alert_would_trade",
+                fallback="Would trade if enabled: {value}",
+                value=str(bool(getattr(signal, "would_trade_if_enabled", False))).lower(),
+            ),
+        ]
+        source = getattr(signal, "source_uri", None) or getattr(signal, "local_path", None)
+        if source:
+            body_lines.append(
+                _translate(
+                    translator,
+                    "strategy_plugin_alert_source",
+                    fallback="Source: {source}",
+                    source=source,
+                )
+            )
+        messages.append(StrategyPluginAlertMessage(subject=subject, body="\n".join(body_lines)))
+    return tuple(messages)
+
+
 def _materialize_artifact_path(reference: str, *, client_factory: Any = None) -> tuple[Path, dict[str, str | None]]:
     raw_reference = _required_string(reference, field_name="reference")
     if not raw_reference.startswith("gs://"):
@@ -283,6 +406,23 @@ def _optional_string(value: Any) -> str | None:
     return text or None
 
 
+def _normalize_strategy_plugin_field(value: str | None) -> str:
+    return str(value or "").strip().lower() or "unknown"
+
+
+def _translate(
+    translator: Callable[..., str] | None,
+    key: str,
+    *,
+    fallback: str,
+    **kwargs: Any,
+) -> str:
+    if translator is None:
+        return fallback.format(**kwargs)
+    translated = translator(key, **kwargs)
+    return translated if translated != key else fallback.format(**kwargs)
+
+
 def _required_string(value: Any, *, field_name: str) -> str:
     text = _optional_string(value)
     if text is None:

src/quant_platform_kit/notifications/__init__.py

@@ -1,9 +1,12 @@
 """Notification integrations."""
 
+from .email import parse_email_recipients, send_smtp_email
 from .events import NotificationPublisher, RenderedNotification, publish_rendered_notification
 
 __all__ = [
     "NotificationPublisher",
     "RenderedNotification",
+    "parse_email_recipients",
     "publish_rendered_notification",
+    "send_smtp_email",
 ]

src/quant_platform_kit/notifications/email.py

@@ -0,0 +1,67 @@
+"""SMTP email notification helpers."""
+
+from __future__ import annotations
+
+import smtplib
+from collections.abc import Sequence
+from email.message import EmailMessage
+
+
+def parse_email_recipients(raw_value: str | Sequence[str] | None) -> tuple[str, ...]:
+    if raw_value is None:
+        return ()
+    if isinstance(raw_value, str):
+        values = raw_value.replace(";", ",").replace("\n", ",").split(",")
+    else:
+        values = raw_value
+    recipients = []
+    seen = set()
+    for value in values:
+        recipient = str(value or "").strip()
+        if not recipient or recipient in seen:
+            continue
+        recipients.append(recipient)
+        seen.add(recipient)
+    return tuple(recipients)
+
+
+def send_smtp_email(
+    *,
+    subject: str,
+    body: str,
+    smtp_host: str | None,
+    smtp_port: int,
+    sender: str | None,
+    recipients: Sequence[str],
+    username: str | None = None,
+    password: str | None = None,
+    use_starttls: bool = True,
+    use_ssl: bool = False,
+    timeout: float = 10.0,
+    smtp_module=smtplib,
+    printer=print,
+) -> bool:
+    resolved_recipients = parse_email_recipients(recipients)
+    host = str(smtp_host or "").strip()
+    from_addr = str(sender or "").strip()
+    if not host or not from_addr or not resolved_recipients:
+        return False
+
+    message = EmailMessage()
+    message["From"] = from_addr
+    message["To"] = ", ".join(resolved_recipients)
+    message["Subject"] = str(subject or "").strip() or "strategy alert"
+    message.set_content(str(body or "").strip())
+
+    try:
+        smtp_cls = smtp_module.SMTP_SSL if use_ssl else smtp_module.SMTP
+        with smtp_cls(host, int(smtp_port), timeout=timeout) as smtp:
+            if use_starttls and not use_ssl:
+                smtp.starttls()
+            if username:
+                smtp.login(str(username), str(password or ""))
+            smtp.send_message(message)
+        return True
+    except Exception as exc:
+        printer(f"Email send failed: {exc}", flush=True)
+        return False