1

app/integrations/feishu_push.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import os
+from copy import deepcopy
 
 import requests
 
@@ -25,6 +26,44 @@ def _feishu_settings():
     }
 
 
+def _runtime_env() -> str:
+    return str(os.getenv("ALPHAX_ENV") or "dev").strip().lower() or "dev"
+
+
+def _env_label(env: str) -> str:
+    labels = {
+        "dev": "DEV",
+        "development": "DEV",
+        "local": "DEV",
+        "test": "TEST",
+        "testing": "TEST",
+        "stage": "STAGING",
+        "staging": "STAGING",
+    }
+    return labels.get(env, env.upper())
+
+
+def _mark_card_environment(card_content):
+    env = _runtime_env()
+    card = deepcopy(card_content)
+    metadata = dict(card.get("metadata") or {})
+    metadata["alphax_env"] = env
+    card["metadata"] = metadata
+
+    if env in ("prod", "production"):
+        return card
+
+    label = _env_label(env)
+    header = card.setdefault("header", {})
+    title = header.setdefault("title", {"tag": "plain_text", "content": ""})
+    if isinstance(title, dict):
+        content = str(title.get("content") or "")
+        prefix = f"[{label}] "
+        if not content.startswith(prefix):
+            title["content"] = prefix + content
+    return card
+
+
 def push_card(card_content):
     """Paper trading cards only.
 
@@ -37,6 +76,7 @@ def push_card(card_content):
     if str(metadata.get("source") or "").strip().lower() != "paper_trading":
         return False, {"skipped": True, "reason": "paper_trading_only"}
 
+    card_content = _mark_card_environment(card_content)
     settings = _feishu_settings()
     if not settings["enabled"]:
         return False, "feishu notification disabled"
@@ -53,13 +93,4 @@ def push_card(card_content):
         return False, str(exc)
 
 
-def push_altcoin_tp_sl_alert(*args, **kwargs):
+__all__ = ["push_card"]
-    """Backward-compatible alias for legacy imports.
-
-    The transport remains paper-trading only; legacy callers get a rejected
-    response instead of importing failing during test collection.
-    """
-    return False, {"skipped": True, "reason": "deprecated_alias"}
-
-
-__all__ = ["push_card", "push_altcoin_tp_sl_alert"]

tests/test_runtime_config.py

@@ -258,6 +258,7 @@ def test_paper_trading_system_config_controls_account_model(monkeypatch):
 
 def test_notification_system_config_controls_feishu_webhook(monkeypatch):
     calls = []
+    monkeypatch.setenv("ALPHAX_ENV", "dev")
     monkeypatch.setenv("TEST_FEISHU_WEBHOOK", "https://open.feishu.test/hook")
     set_config("system", "notification", {
         "enabled": True,
@@ -275,20 +276,58 @@ def test_notification_system_config_controls_feishu_webhook(monkeypatch):
         return FakeResponse()
 
     monkeypatch.setattr(feishu_push.requests, "post", fake_post)
-    ok, result = feishu_push.push_card({"elements": []})
+    ok, result = feishu_push.push_card({
+        "metadata": {"source": "paper_trading"},
+        "header": {"title": {"tag": "plain_text", "content": "模拟交易开仓 — BTC"}},
+        "elements": [],
+    })
 
     assert ok is True
     assert result["StatusCode"] == 0
     assert calls[0]["url"] == "https://open.feishu.test/hook"
     assert calls[0]["timeout"] == 3
+    assert calls[0]["json"]["card"]["header"]["title"]["content"].startswith("[DEV] ")
+    assert calls[0]["json"]["card"]["metadata"]["alphax_env"] == "dev"
     assert notification_config()["feishu"]["webhook_env"] == "TEST_FEISHU_WEBHOOK"
 
     set_config("system", "notification", {"enabled": False, "feishu": {"enabled": True, "webhook_env": "TEST_FEISHU_WEBHOOK"}})
-    ok, reason = feishu_push.push_card({"elements": []})
+    ok, reason = feishu_push.push_card({"metadata": {"source": "paper_trading"}, "elements": []})
     assert ok is False
     assert "disabled" in reason
 
 
+def test_notification_production_does_not_prefix_feishu_title(monkeypatch):
+    calls = []
+    monkeypatch.setenv("ALPHAX_ENV", "production")
+    monkeypatch.setenv("TEST_FEISHU_WEBHOOK", "https://open.feishu.test/hook")
+    set_config("system", "notification", {
+        "enabled": True,
+        "feishu": {"enabled": True, "webhook_env": "TEST_FEISHU_WEBHOOK", "timeout": 3},
+    })
+
+    class FakeResponse:
+        status_code = 200
+
+        def json(self):
+            return {"StatusCode": 0}
+
+    def fake_post(url, json=None, timeout=None):
+        calls.append({"url": url, "json": json, "timeout": timeout})
+        return FakeResponse()
+
+    monkeypatch.setattr(feishu_push.requests, "post", fake_post)
+    ok, result = feishu_push.push_card({
+        "metadata": {"source": "paper_trading"},
+        "header": {"title": {"tag": "plain_text", "content": "模拟交易开仓 — BTC"}},
+        "elements": [],
+    })
+
+    assert ok is True
+    assert result["StatusCode"] == 0
+    assert calls[0]["json"]["card"]["header"]["title"]["content"] == "模拟交易开仓 — BTC"
+    assert calls[0]["json"]["card"]["metadata"]["alphax_env"] == "production"
+
+
 def test_email_system_config_uses_env_pointers(monkeypatch):
     sent = []
     monkeypatch.setenv("SMTP_USER_ENV", "noreply@example.com")