diff --git a/main.py b/main.py index d963caf..637c604 100644 --- a/main.py +++ b/main.py @@ -825,39 +825,20 @@ def _runtime_error_notification_targets() -> tuple[tuple[str, str], ...]: def _runtime_error_notification_message(exc: Exception, *, route_label: str | None = None) -> str: - error_text = f"{type(exc).__name__}: {exc}" - if len(error_text) > 1200: - error_text = error_text[:1197] + "..." - route = route_label or f"{request.method} {request.path}" - if str(NOTIFY_LANG or "").strip().lower().startswith("zh"): - return "\n".join( - ( - "IBKR 策略运行失败", - f"服务: {SERVICE_NAME or os.getenv('K_SERVICE', 'interactive-brokers-platform')}", - f"版本: {os.getenv('K_REVISION') or ''}", - f"路由: {route}", - f"策略: {STRATEGY_PROFILE}", - f"账户组: {ACCOUNT_GROUP}", - f"错误: {error_text}", - ) - ) - return "\n".join( - ( - "IBKR strategy run failed", - f"service: {SERVICE_NAME or os.getenv('K_SERVICE', 'interactive-brokers-platform')}", - f"revision: {os.getenv('K_REVISION') or ''}", - f"route: {route}", - f"strategy: {STRATEGY_PROFILE}", - f"account_group: {ACCOUNT_GROUP}", - f"error: {error_text}", - ) - ) + title = "runtime_failure_title" + return "\n".join(( + t(title), + t("strategy_label", name=strategy_display_name or STRATEGY_PROFILE), + t("runtime_failure_context", context=ACCOUNT_GROUP), + t("runtime_failure_result"), + t("runtime_failure_action"), + )) def _notify_runtime_error(exc: Exception, *, route_label: str | None = None) -> bool: targets = _runtime_error_notification_targets() if not targets: - print("IBKR runtime error notification skipped: no Telegram target configured.", flush=True) + print(t("runtime_notification_missing_target"), flush=True) return False message = _runtime_error_notification_message(exc, route_label=route_label) outcomes = [] @@ -876,14 +857,13 @@ def _publish_runtime_failure_notification(*, detailed_text: str, compact_text: s if publish_notification(detailed_text=detailed_text, compact_text=compact_text): return True return _notify_runtime_error(exc) - except Exception as notification_exc: - print(f"IBKR runtime error notification fallback: {notification_exc}", flush=True) + except Exception: + print(t("runtime_notification_delivery_failed"), flush=True) return _notify_runtime_error(exc) def _handle_route_runtime_error(exc: Exception, *, route_label: str | None = None): - print(f"IBKR route failed before strategy-cycle handling: {type(exc).__name__}: {exc}", flush=True) - traceback.print_exc() + print(t("runtime_failure_log"), flush=True) _notify_runtime_error(exc, route_label=route_label) return "Error", 500 diff --git a/notifications/telegram.py b/notifications/telegram.py index 8e6bf2a..cc0e7e3 100644 --- a/notifications/telegram.py +++ b/notifications/telegram.py @@ -2,6 +2,8 @@ from __future__ import annotations +from quant_platform_kit.common.operational_notification_localization import resolve_operational_notification_locale + import re try: @@ -26,6 +28,13 @@ def _break_telegram_market_symbol_auto_links(value) -> str: I18N = { "zh": { + "runtime_failure_title": "⚠️ IBKR 策略运行失败", + "runtime_failure_context": "运行目标：{context}", + "runtime_failure_result": "本次运行未正常结束，请查看最新执行报告。", + "runtime_failure_action": "下一步：检查 Gateway 连接及最新执行报告。", + "runtime_notification_missing_target": "异常通知未发送：未配置 Telegram 接收目标（notification_target_missing）", + "runtime_notification_delivery_failed": "异常通知发送失败（notification_delivery_failed）", + "runtime_failure_log": "策略运行失败（runtime_setup_failed）", "rebalance_title": "🔔 【调仓指令】", "heartbeat_title": "💓 【心跳检测】", "error_title": "🚨 【策略异常】", @@ -245,6 +254,13 @@ def _break_telegram_market_symbol_auto_links(value) -> str: "strategy_name_tecl_xlk_trend_income": "TECL/XLK 科技趋势收益", }, "en": { + "runtime_failure_title": "⚠️ IBKR strategy run failed", + "runtime_failure_context": "Target: {context}", + "runtime_failure_result": "The run did not finish successfully; check the latest execution report.", + "runtime_failure_action": "Next: Check Gateway connectivity and the latest execution report.", + "runtime_notification_missing_target": "Runtime alert not sent: Telegram target is not configured (notification_target_missing)", + "runtime_notification_delivery_failed": "Runtime alert delivery failed (notification_delivery_failed)", + "runtime_failure_log": "Strategy run failed (runtime_setup_failed)", "rebalance_title": "🔔 【Trade Execution Report】", "heartbeat_title": "💓 【Heartbeat】", "error_title": "🚨 【Strategy Error】", @@ -476,8 +492,8 @@ def _break_telegram_market_symbol_auto_links(value) -> str: def build_translator(lang): + active_lang = resolve_operational_notification_locale(lang) def translate(key, **kwargs): - active_lang = lang if lang in I18N else "en" template = I18N[active_lang].get(key, key) return template.format(**kwargs) if kwargs else template diff --git a/tests/test_request_handling.py b/tests/test_request_handling.py index e3add0f..9213010 100644 --- a/tests/test_request_handling.py +++ b/tests/test_request_handling.py @@ -1323,7 +1323,7 @@ def fake_post(_url, *, json, timeout): assert len(sent_payloads) == 1 assert sent_payloads[0][0]["chat_id"] == "chat-1" assert "IBKR strategy run failed" in sent_payloads[0][0]["text"] - assert "RuntimeError: boom" in sent_payloads[0][0]["text"] + assert "boom" not in sent_payloads[0][0]["text"] def test_handle_request_runtime_error_fallback_uses_chinese_copy(strategy_module_factory, monkeypatch): @@ -1350,8 +1350,8 @@ def fake_post(_url, *, json, timeout): assert body == "Error" text = sent_payloads[0][0]["text"] assert "IBKR 策略运行失败" in text - assert "账户组:" in text - assert "错误: RuntimeError: boom" in text + assert "运行目标：" in text + assert "boom" not in text def test_handle_request_runtime_error_fallback_prefixes_account_scope(strategy_module_factory, monkeypatch): @@ -1393,7 +1393,7 @@ def fake_post(_url, *, json, timeout): assert status == 500 assert body == "Error" - assert sent_payloads[0][0]["text"].startswith("[U1234567] IBKR strategy run failed") + assert sent_payloads[0][0]["text"].startswith("[U1234567] ⚠️ IBKR strategy run failed") def test_handle_reconciliation_persists_and_logs_only_redacted_failure_fields( diff --git a/tests/test_runtime_notification_i18n.py b/tests/test_runtime_notification_i18n.py new file mode 100644 index 0000000..7002715 --- /dev/null +++ b/tests/test_runtime_notification_i18n.py @@ -0,0 +1,42 @@ +import ast +import os +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from notifications.telegram import build_translator + + +@pytest.mark.parametrize("locale", ["zh-CN", "zh_TW", "ZH-hans", " zh "]) +def test_chinese_locale_variants_use_chinese(locale): + assert build_translator(locale)("strategy_label", name="test") == build_translator("zh")("strategy_label", name="test") + + +def runtime_function(name, **overrides): + tree = ast.parse((Path(__file__).resolve().parents[1] / "main.py").read_text()) + function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name) + namespace = dict(os=os, request=SimpleNamespace(method="GET", path="/main"), + NOTIFY_LANG="zh-CN", SERVICE_NAME="Test service", SECRET_NAME="Test secret alias", + STRATEGY_PROFILE="test_strategy", ACCOUNT_REGION="SG", ACCOUNT_GROUP="Test account", + strategy_display_name="Test strategy", build_translator=build_translator, + t=build_translator("zh-CN")) + namespace.update(overrides) + exec(compile(ast.Module(body=[function], type_ignores=[]), "main.py", "exec"), namespace) + return namespace[name] + + +@pytest.mark.parametrize("locale", ["zh-CN", "en"]) +def test_startup_alert_is_compact_localized_and_does_not_embed_provider_text(locale, monkeypatch): + monkeypatch.setenv("NOTIFY_LANG", locale) + monkeypatch.delenv("QSL_NOTIFY_LANG", raising=False) + monkeypatch.setenv("STRATEGY_PROFILE", "test_strategy") + fn = runtime_function("_runtime_error_notification_message", NOTIFY_LANG=locale, t=build_translator(locale)) + message = fn(RuntimeError("PRIVATE_PROVIDER_SENTINEL")) + assert len(message.splitlines()) <= 5 + assert "PRIVATE_PROVIDER_SENTINEL" not in message + assert "未提交订单" not in message + assert "no order" not in message.lower() + assert ("未正常结束" in message) is locale.startswith("zh") + assert ("did not finish successfully" in message) is (locale == "en") + assert "test_strategy" in message or "Test strategy" in message