Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 12 additions & 32 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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 '<unknown>'}",
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 '<unknown>'}",
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 = []
Expand All @@ -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

Expand Down
18 changes: 17 additions & 1 deletion notifications/telegram.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,8 @@

from __future__ import annotations

from quant_platform_kit.common.operational_notification_localization import resolve_operational_notification_locale

import re

try:
Expand All @@ -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": "🚨 【策略异常】",
Expand Down Expand Up @@ -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】",
Expand Down Expand Up @@ -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

Expand Down
8 changes: 4 additions & 4 deletions tests/test_request_handling.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand All @@ -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):
Expand Down Expand Up @@ -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(
Expand Down
42 changes: 42 additions & 0 deletions tests/test_runtime_notification_i18n.py
Original file line number Diff line number Diff line change
@@ -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