From 04e423406f4e7e96c0815658f47a412bc9ed4b01 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 29 May 2026 04:01:53 +0800 Subject: [PATCH] Localize signal snapshot source in notifications --- notifications/renderers.py | 60 ++++++++++++++++++++++++++++++++----- tests/test_notifications.py | 25 ++++++++++++++++ 2 files changed, 78 insertions(+), 7 deletions(-) diff --git a/notifications/renderers.py b/notifications/renderers.py index 147e013..c5d0b3b 100644 --- a/notifications/renderers.py +++ b/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}") diff --git a/tests/test_notifications.py b/tests/test_notifications.py index d6e74ee..2be0484 100644 --- a/tests/test_notifications.py +++ b/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")