Add shared notification localization helpers

src/quant_platform_kit/common/__init__.py

@@ -1,5 +1,10 @@
 """Shared domain models, ports, strategy contracts, and plugin helpers."""
 
+from .notification_localization import (
+    COMMON_ZH_NOTIFICATION_REPLACEMENTS,
+    localize_notification_text,
+    translator_uses_zh,
+)
 from .strategy_plugins import (
     PLUGIN_MODE_ADVISORY,
     PLUGIN_MODE_LIVE,
@@ -17,17 +22,20 @@
 )
 
 __all__ = [
+    "COMMON_ZH_NOTIFICATION_REPLACEMENTS",
     "PLUGIN_MODE_ADVISORY",
     "PLUGIN_MODE_LIVE",
     "PLUGIN_MODE_PAPER",
     "PLUGIN_MODE_SHADOW",
     "SUPPORTED_STRATEGY_PLUGIN_MODES",
+    "localize_notification_text",
     "StrategyPluginMountConfig",
     "StrategyPluginSignal",
     "build_strategy_plugin_report_payload",
     "load_configured_strategy_plugin_signals",
     "load_strategy_plugin_signal",
     "normalize_strategy_plugin_mode",
     "parse_strategy_plugin_mounts",
+    "translator_uses_zh",
     "validate_strategy_plugin_signal_payload",
 ]

src/quant_platform_kit/common/notification_localization.py

@@ -0,0 +1,101 @@
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+
+NotificationTranslator = Callable[..., str]
+NotificationReplacement = tuple[str, str]
+
+
+COMMON_ZH_NOTIFICATION_REPLACEMENTS: tuple[NotificationReplacement, ...] = (
+    ("feature snapshot guard blocked execution", "特征快照校验阻止执行"),
+    ("feature snapshot required", "需要特征快照"),
+    ("feature snapshot compute failed", "特征快照计算失败"),
+    ("feature_snapshot_download_failed", "特征快照下载失败"),
+    ("feature_snapshot_compute_failed", "特征快照计算失败"),
+    ("feature_snapshot_path_missing", "缺少特征快照路径"),
+    ("feature_snapshot_missing", "特征快照不存在"),
+    ("feature_snapshot_stale", "特征快照过旧"),
+    ("feature_snapshot_manifest_missing", "缺少快照清单"),
+    ("feature_snapshot_profile_mismatch", "快照策略名不匹配"),
+    ("feature_snapshot_config_name_mismatch", "快照配置名不匹配"),
+    ("feature_snapshot_config_path_mismatch", "快照配置路径不匹配"),
+    ("feature_snapshot_contract_version_mismatch", "快照契约版本不匹配"),
+    ("soxl_soxx_trend_income", "SOXL/SOXX 半导体趋势收益"),
+    ("tqqq_growth_income", "TQQQ 增长收益"),
+    ("global_etf_rotation", "全球 ETF 轮动"),
+    ("russell_1000_multi_factor_defensive", "罗素1000多因子"),
+    ("tech_communication_pullback_enhancement", "科技通信回调增强"),
+    ("qqq_tech_enhancement", "科技通信回调增强"),
+    ("mega_cap_leader_rotation_aggressive", "Mega Cap 激进龙头轮动"),
+    ("mega_cap_leader_rotation_dynamic_top20", "Mega Cap 动态 Top20 龙头轮动"),
+    ("mega_cap_leader_rotation_top50_balanced", "Mega Cap Top50 平衡龙头轮动"),
+    ("dynamic_mega_leveraged_pullback", "Mega Cap 2x 回调策略"),
+    ("outside_monthly_execution_window", "当前不在月度执行窗口"),
+    ("no_execution_window_after_snapshot", "快照后没有可用执行窗口"),
+    ("no-op", "不执行"),
+    ("monthly snapshot cadence", "月度快照节奏"),
+    ("waiting inside execution window", "等待进入执行窗口"),
+    ("small_account_warning=true", "小账户提示=是"),
+    ("portfolio_equity=", "净值="),
+    ("min_recommended_equity=", "建议最低净值="),
+    (
+        "integer_shares_min_position_value_may_prevent_backtest_replication",
+        "整数股和最小仓位限制可能导致实盘无法完全复现回测",
+    ),
+    (
+        "integer-share minimum position sizing may prevent backtest replication",
+        "整数股和最小仓位限制可能导致实盘无法完全复现回测",
+    ),
+    ("small account warning: portfolio equity", "小账户提示：净值"),
+    ("small account warning", "小账户提示"),
+    ("is below the recommended", "低于建议"),
+    ("is below recommended", "低于建议"),
+    ("snapshot_as_of=", "快照日期="),
+    ("snapshot=", "快照日期="),
+    ("allowed=", "允许日期="),
+    ("<unknown>", "未知"),
+    ("<none>", "无"),
+    ("RISK-ON", "风险开启"),
+    ("DE-LEVER", "降杠杆"),
+    ("regime=hard_defense", "市场阶段=强防御"),
+    ("regime=soft_defense", "市场阶段=软防御"),
+    ("regime=risk_on", "市场阶段=进攻"),
+    ("benchmark_trend=down", "基准趋势=向下"),
+    ("benchmark_trend=up", "基准趋势=向上"),
+    ("benchmark=down", "基准趋势=向下"),
+    ("benchmark=up", "基准趋势=向上"),
+    ("breadth=", "市场宽度="),
+    ("target_stock=", "目标股票仓位="),
+    ("realized_stock=", "实际股票仓位="),
+    ("stock_exposure=", "股票目标仓位="),
+    ("safe_haven=", "避险仓位="),
+    ("selected=", "入选标的数="),
+    ("top=", "前排标的="),
+    ("no_selection", "无入选标的"),
+    ("outside_execution_window", "当前不在执行窗口"),
+    ("insufficient_buying_power", "购买力不足"),
+    ("missing_price", "缺少报价"),
+    ("no_equity", "无净值"),
+    ("fail_closed", "关闭执行"),
+    ("reason=", "原因="),
+)
+
+
+def translator_uses_zh(translator: NotificationTranslator) -> bool:
+    sample = str(translator("no_trades"))
+    return any("\u4e00" <= ch <= "\u9fff" for ch in sample)
+
+
+def localize_notification_text(
+    text: object,
+    *,
+    translator: NotificationTranslator,
+    extra_replacements: Sequence[NotificationReplacement] = (),
+) -> str:
+    value = str(text or "").strip()
+    if not value or not translator_uses_zh(translator):
+        return value
+    localized = value
+    for source, target in (*tuple(extra_replacements), *COMMON_ZH_NOTIFICATION_REPLACEMENTS):
+        localized = localized.replace(source, target)
+    return localized

tests/test_notification_localization.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import unittest
+
+from quant_platform_kit.common.notification_localization import (
+    COMMON_ZH_NOTIFICATION_REPLACEMENTS,
+    localize_notification_text,
+    translator_uses_zh,
+)
+
+
+def _translator_factory(no_trades_text: str):
+    def _translator(key: str, **_kwargs) -> str:
+        if key == "no_trades":
+            return no_trades_text
+        return key
+
+    return _translator
+
+
+class NotificationLocalizationTests(unittest.TestCase):
+    def test_translator_uses_zh_detects_chinese_output(self):
+        self.assertTrue(translator_uses_zh(_translator_factory("无需交易")))
+        self.assertFalse(translator_uses_zh(_translator_factory("No trades")))
+
+    def test_localize_notification_text_applies_common_replacements(self):
+        localized = localize_notification_text(
+            "no-op | reason=outside_monthly_execution_window | snapshot=2026-04-16",
+            translator=_translator_factory("无需交易"),
+        )
+
+        self.assertEqual(
+            localized,
+            "不执行 | 原因=当前不在月度执行窗口 | 快照日期=2026-04-16",
+        )
+
+    def test_localize_notification_text_keeps_english_for_non_zh_translator(self):
+        text = "no-op | reason=outside_monthly_execution_window"
+        self.assertEqual(
+            localize_notification_text(text, translator=_translator_factory("No trades")),
+            text,
+        )
+
+    def test_localize_notification_text_applies_extra_replacements_after_common_set(self):
+        localized = localize_notification_text(
+            "fail_reason=same_day_execution_locked",
+            translator=_translator_factory("无需交易"),
+            extra_replacements=(
+                ("fail_reason=", "失败原因="),
+                ("same_day_execution_locked", "当日执行锁已存在"),
+            ),
+        )
+
+        self.assertEqual(localized, "失败原因=当日执行锁已存在")
+
+    def test_common_replacements_include_reason_label(self):
+        self.assertIn(("reason=", "原因="), COMMON_ZH_NOTIFICATION_REPLACEMENTS)
+
+
+if __name__ == "__main__":
+    unittest.main()