Localize signal snapshot source in notifications

notifications/renderers.py

@@ -11,6 +11,54 @@
     translator_uses_zh as _base_translator_uses_zh,
 )
 
+try:
+    from quant_platform_kit.common.notification_localization import (
+        localize_price_source_label as _localize_price_source_label,
+        localize_quote_overlay_state as _localize_quote_overlay_state,
+    )
+except ImportError:  # pragma: no cover - compatibility with older pinned shared wheels
+    _PRICE_SOURCE_LABELS = {
+        "longbridge_candlesticks": ("LongBridge 日线K线", "LongBridge daily candlesticks"),
+        "schwab_daily_history_with_live_quote_overlay": (
+            "Schwab 日线历史 + 实时报价覆盖",
+            "Schwab daily history + live quote overlay",
+        ),
+        "firstrade_ohlc_with_live_quote_overlay": (
+            "Firstrade OHLC + 实时报价覆盖",
+            "Firstrade OHLC + live quote overlay",
+        ),
+        "market_quote": ("实时行情报价", "market quote"),
+        "mixed_market_quote_snapshot_close": (
+            "实时行情报价 + 快照收盘价回补",
+            "market quote + snapshot close fallback",
+        ),
+        "mixed_market_quote_historical_close": (
+            "实时行情报价 + 历史收盘价回补",
+            "market quote + historical close fallback",
+        ),
+        "snapshot_close": ("快照收盘价", "snapshot close"),
+        "historical_close": ("历史收盘价", "historical close"),
+        "market_data": ("市场数据", "market data"),
+    }
+
+    def _localize_price_source_label(value, *, translator=None, locale=None):
+        source = str(value or "").strip()
+        use_zh = _base_translator_uses_zh(translator) if translator is not None else str(locale or "").startswith("zh")
+        if not source:
+            return "未知" if use_zh else "unknown"
+        label = _PRICE_SOURCE_LABELS.get(source)
+        if label is not None:
+            return label[0] if use_zh else label[1]
+        return source.replace("_", " ")
+
+    def _localize_quote_overlay_state(value, *, translator=None, locale=None):
+        use_zh = _base_translator_uses_zh(translator) if translator is not None else str(locale or "").startswith("zh")
+        if value is True:
+            return "是" if use_zh else "yes"
+        if value is False:
+            return "否" if use_zh else "no"
+        return "未知" if use_zh else "unknown"
+
 _DETAIL_FIELD_SPLIT_RE = re.compile(r"\s+(?=[^\s=:：]+[=:：])")
 
 
@@ -134,20 +182,18 @@ def _format_signal_snapshot_line(snapshot, *, translator) -> str:
     if not market_date and not source and overlay is None and warning in (None, "", False):
         return ""
     if _translator_uses_zh(translator):
-        overlay_text = "是" if overlay is True else "否" if overlay is False else "未知"
         parts = [
             f"日期 {market_date or '未知'}",
-            f"数据源 {source or '未知'}",
-            f"报价覆盖 {overlay_text}",
+            f"数据源 {_localize_price_source_label(source, translator=translator)}",
+            f"报价覆盖 {_localize_quote_overlay_state(overlay, translator=translator)}",
         ]
         if warning not in (None, "", False):
-            parts.append(f"提示 {warning}")
+            parts.append(f"提示 {_localize_notification_text(warning, translator=translator)}")
         return "🧾 信号快照: " + " | ".join(parts)
-    overlay_text = "yes" if overlay is True else "no" if overlay is False else "unknown"
     parts = [
         f"date {market_date or 'unknown'}",
-        f"source {source or 'unknown'}",
-        f"quote overlay {overlay_text}",
+        f"source {_localize_price_source_label(source, translator=translator)}",
+        f"quote overlay {_localize_quote_overlay_state(overlay, translator=translator)}",
     ]
     if warning not in (None, "", False):
         parts.append(f"warning {warning}")

tests/test_notifications.py

@@ -14,6 +14,7 @@
     build_strategy_display_name,
     build_translator,
 )
+from notifications.renderers import render_heartbeat_notification
 from strategy_registry import SUPPORTED_STRATEGY_PROFILES
 
 
@@ -112,6 +113,30 @@ def test_supported_strategy_profiles_have_translated_names(self):
             self.assertNotEqual(zh_name(profile), profile)
             self.assertNotEqual(en_name(profile), profile)
 
+    def test_heartbeat_signal_snapshot_localizes_price_source(self):
+        rendered = render_heartbeat_notification(
+            execution={
+                "signal_snapshot": {
+                    "market_date": "2026-05-28",
+                    "latest_price_source": "longbridge_candlesticks",
+                    "quote_overlay_used": None,
+                },
+                "status_display": "🚀 风险开启（SOXX+SOXL）",
+                "signal_display": "SOXX 站上 140 日门槛线，持有 SOXL 70.0% + SOXX 20.0%",
+            },
+            skip_logs=(),
+            note_logs=(),
+            translator=build_translator("zh"),
+            separator="━━━━━━━━━━━━━━━━━━",
+            strategy_display_name="SOXL/SOXX 半导体趋势收益",
+            dry_run_only=False,
+        )
+
+        self.assertIn("数据源 LongBridge 日线K线", rendered.compact_text)
+        self.assertIn("报价覆盖 未知", rendered.compact_text)
+        self.assertIn("📊 市场状态: 🚀 风险开启（SOXX+SOXL）", rendered.compact_text)
+        self.assertNotIn("longbridge_candlesticks", rendered.compact_text)
+
     def test_build_prefixer_prefers_account_prefix_only(self):
         with_prefix = build_prefixer("HK", "longbridge-quant-semiconductor-rotation-income-hk")
         self.assertEqual(with_prefix("hello"), "[HK] hello")