diff --git a/src/quant_platform_kit/notifications/telegram.py b/src/quant_platform_kit/notifications/telegram.py index 0a5b4d9..e3bbac3 100644 --- a/src/quant_platform_kit/notifications/telegram.py +++ b/src/quant_platform_kit/notifications/telegram.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import re import urllib.parse import urllib.request from collections.abc import Sequence @@ -10,6 +11,16 @@ DEFAULT_TELEGRAM_BOT_API_BASE_URL = "https://api.telegram.org" +_TELEGRAM_MARKET_SYMBOL_LINK_RE = re.compile(r"(? str: + text = str(value or "") + return _TELEGRAM_MARKET_SYMBOL_LINK_RE.sub( + lambda match: f"{match.group(1)}.{_TELEGRAM_MARKET_SYMBOL_LINK_JOINER}{match.group(2)}", + text, + ) def parse_telegram_chat_ids(raw_value: str | Sequence[str] | None) -> tuple[str, ...]: @@ -85,7 +96,7 @@ def send_telegram_message( for chat_id in resolved_chat_ids: payload: dict[str, object] = { "chat_id": chat_id, - "text": message, + "text": _break_telegram_market_symbol_auto_links(message), "disable_web_page_preview": bool(disable_web_page_preview), } text_parse_mode = str(parse_mode or "").strip() diff --git a/tests/test_strategy_plugin_telegram_notifications.py b/tests/test_strategy_plugin_telegram_notifications.py index 23837e8..d892353 100644 --- a/tests/test_strategy_plugin_telegram_notifications.py +++ b/tests/test_strategy_plugin_telegram_notifications.py @@ -72,6 +72,28 @@ def opener(request, timeout): self.assertEqual(payload["text"], "危机通知") self.assertTrue(payload["disable_web_page_preview"]) + def test_send_telegram_message_breaks_market_symbol_auto_links(self): + requests = [] + + def opener(request, timeout): + requests.append((request, timeout)) + return _FakeResponse() + + sent = send_telegram_message( + text="SOXL.US 预计;00700.HK 持仓;https://example.com 保持原样", + chat_ids=("123",), + bot_token="123456:ABC", + api_base_url="https://telegram.example.test", + opener=opener, + ) + + self.assertTrue(sent) + payload = json.loads(requests[0][0].data.decode("utf-8")) + self.assertEqual( + payload["text"], + "SOXL.\u2060US 预计;00700.\u2060HK 持仓;https://example.com 保持原样", + ) + def test_send_strategy_plugin_telegram_combines_title_and_body(self): requests = []