Prevent Telegram market symbol auto-links

Author: Pigbibi

notifications/telegram.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import re
+
 try:
     from quant_platform_kit.common.notification_localization import (
         merge_strategy_plugin_i18n as _merge_strategy_plugin_i18n,
@@ -10,6 +12,18 @@
     _merge_strategy_plugin_i18n = None
 
 
+_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) -> 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,
+    )
+
+
 I18N = {
     "zh": {
         "rebalance_title": "🔔 【调仓指令】",
@@ -451,7 +465,7 @@ def send_telegram_message(
         printer(f"TG:\n{message}", flush=True)
         response = requests_module.post(
             url,
-            json={"chat_id": chat_id, "text": message},
+            json={"chat_id": chat_id, "text": _break_telegram_market_symbol_auto_links(message)},
             timeout=10,
         )
         if not 200 <= response.status_code < 300:

tests/test_notifications.py

@@ -262,6 +262,30 @@ def post(*args, **kwargs):
     assert "Telegram send failed with status 401: unauthorized" in captured.out
 
 
+def test_send_telegram_message_breaks_market_symbol_auto_links():
+    calls = []
+
+    class FakeResponse:
+        status_code = 200
+        text = ""
+
+    class FakeRequests:
+        @staticmethod
+        def post(*args, **kwargs):
+            calls.append((args, kwargs))
+            return FakeResponse()
+
+    send_telegram_message(
+        "SOXL.US 预计；00700.HK 持仓；https://example.com 保持原样",
+        token="token",
+        chat_id="chat-id",
+        requests_module=FakeRequests,
+    )
+
+    payload = calls[0][1]["json"]
+    assert payload["text"] == "SOXL.\u2060US 预计；00700.\u2060HK 持仓；https://example.com 保持原样"
+
+
 def test_send_telegram_message_redacts_exception_detail(capsys):
     class FakeRequests:
         @staticmethod