Prevent Telegram market symbol auto-links

Author: Pigbibi

src/quant_platform_kit/notifications/telegram.py

@@ -3,13 +3,24 @@
 from __future__ import annotations
 
 import json
+import re
 import urllib.parse
 import urllib.request
 from collections.abc import Sequence
 from typing import Any
 
 
 DEFAULT_TELEGRAM_BOT_API_BASE_URL = "https://api.telegram.org"
+_TELEGRAM_MARKET_SYMBOL_LINK_RE = re.compile(r"(?<![A-Za-z0-9_])([A-Z0-9]{1,12})\.([A-Z]{2,4})(?![A-Za-z0-9_])")
+_TELEGRAM_MARKET_SYMBOL_LINK_JOINER = "\u2060"
+
+
+def _break_telegram_market_symbol_auto_links(value: object) -> 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()

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 = []