fix(notify): localize attention reason codes for operators

src/quant_platform_kit/common/operational_notification_localization.py

@@ -56,9 +56,23 @@
         "attention_result_halt": "结果：硬停 / 运行异常，需人工恢复",
         "attention_result_new_risk_prohibited": "结果：已禁止新增风险",
         "attention_reason": "原因：{code}",
+        "attention_reason_none": "无",
+        "attention_reason_new_risk_prohibited": "已禁止新增风险",
+        "attention_reason_drift_critical": "生产偏离严重（CRITICAL）",
+        "attention_reason_drift_review": "生产偏离待审（REVIEW）",
+        "attention_reason_drift_watch": "生产偏离观察（WATCH）",
+        "attention_reason_drift_status_unknown": "生产偏离状态未知",
+        "attention_reason_operational_uncertain": "运行状态不确定 / 故障",
+        "attention_reason_hard_park": "硬停（需授权恢复）",
+        "attention_reason_mandate_dd_exhausted": "授权回撤额度已用尽",
+        "attention_reason_mandate_dd_elevated": "授权回撤接近上限",
+        "attention_reason_PRODUCTION_DRIFT_CRITICAL": "生产偏离严重（CRITICAL）",
+        "attention_reason_PRODUCTION_DRIFT_REVIEW": "生产偏离待审（REVIEW）",
+        "attention_reason_PRODUCTION_DRIFT_WATCH": "生产偏离观察（WATCH）",
+        "attention_reason_PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED": "生产偏离状态非法（失败关闭）",
         "attention_next_none": "下一步：无需操作",
         "attention_next_console_optional": "下一步：可在管理站查看详情",
-        "attention_next_open_console_confirm": "下一步：打开管理站确认意图（accept ≠ live）",
+        "attention_next_open_console_confirm": "下一步：打开管理站确认意图（确认意图不等于放行实盘）",
         "attention_next_open_console_resume": "下一步：打开管理站处理恢复 / 复位（需授权）",
     },
     "en": {
@@ -103,9 +117,23 @@
         "attention_result_halt": "Result: halt / operational fault — human recovery required",
         "attention_result_new_risk_prohibited": "Result: new risk additions prohibited",
         "attention_reason": "Reason: {code}",
+        "attention_reason_none": "none",
+        "attention_reason_new_risk_prohibited": "new risk additions prohibited",
+        "attention_reason_drift_critical": "production drift critical",
+        "attention_reason_drift_review": "production drift review",
+        "attention_reason_drift_watch": "production drift watch",
+        "attention_reason_drift_status_unknown": "production drift status unknown",
+        "attention_reason_operational_uncertain": "operational uncertainty / fault",
+        "attention_reason_hard_park": "hard park (authorized recovery required)",
+        "attention_reason_mandate_dd_exhausted": "mandate drawdown budget exhausted",
+        "attention_reason_mandate_dd_elevated": "mandate drawdown elevated",
+        "attention_reason_PRODUCTION_DRIFT_CRITICAL": "production drift critical",
+        "attention_reason_PRODUCTION_DRIFT_REVIEW": "production drift review",
+        "attention_reason_PRODUCTION_DRIFT_WATCH": "production drift watch",
+        "attention_reason_PRODUCTION_DRIFT_STATUS_INVALID_FAIL_CLOSED": "production drift status invalid (fail closed)",
         "attention_next_none": "Next: none",
         "attention_next_console_optional": "Next: review details in the management console if needed",
-        "attention_next_open_console_confirm": "Next: open the management console to confirm intent (accept ≠ live)",
+        "attention_next_open_console_confirm": "Next: open the management console to confirm intent (confirming intent does not enable live)",
         "attention_next_open_console_resume": "Next: open the management console for recovery / reset (authorization required)",
     },
 }
@@ -134,6 +162,19 @@ def operational_notification_text(
     return template.format(**values)
 
 
+def localize_attention_reason_code(locale: object | None, code: object) -> str:
+    """Translate a known attention / production-drift reason code for operators.
+
+    Unknown codes stay as the original machine token so diagnostics remain intact.
+    """
+
+    raw = str(code or "").strip() or "none"
+    key = f"attention_reason_{raw}"
+    normalized = resolve_operational_notification_locale(locale)
+    template = _TEXTS[normalized].get(key) or _TEXTS["en"].get(key)
+    return template if template is not None else raw
+
+
 def localize_operational_activity(locale: object | None, detail: object) -> str:
     """Translate known activity labels while keeping diagnostic values intact."""
 

src/quant_platform_kit/risk/attention.py

@@ -183,6 +183,7 @@ def render_attention_compact(
     """Four-line operator message: identity / result / reason / next step."""
 
     from quant_platform_kit.common.operational_notification_localization import (
+        localize_attention_reason_code,
         operational_notification_text,
         resolve_operational_notification_locale,
     )
@@ -205,7 +206,11 @@ def render_attention_compact(
     )
     result_key = _result_text_key(level, decision.reason_codes)
     result = operational_notification_text(lang, result_key)
-    reason_line = operational_notification_text(lang, "attention_reason", code=reason)
+    reason_line = operational_notification_text(
+        lang,
+        "attention_reason",
+        code=localize_attention_reason_code(lang, reason),
+    )
     next_step = operational_notification_text(lang, f"attention_next_{decision.next_step_key}")
     return f"{emoji} {title}\n{result}\n{reason_line}\n{next_step}"
 

tests/test_attention.py

@@ -87,8 +87,37 @@ def test_compact_zh_cn_locale_normalizes(self) -> None:
             decision=decision,
         )
         self.assertIn("已禁止新增风险", text)
+        self.assertIn("原因：已禁止新增风险", text)
+        self.assertNotIn("new_risk_prohibited", text)
         self.assertIn("下一步", text)
         self.assertIn("管理站", text)
+        self.assertIn("确认意图不等于放行实盘", text)
+        self.assertNotIn("accept", text)
+
+    def test_compact_localizes_drift_critical_reason_zh_and_en(self) -> None:
+        decision = evaluate_attention(
+            AttentionAxes(new_risk_prohibited=True, production_drift_status="critical")
+        )
+        zh = render_attention_compact(
+            locale="zh",
+            platform="schwab",
+            account_alias="00682",
+            strategy_profile="soxl_soxx_trend_income",
+            decision=decision,
+        )
+        en = render_attention_compact(
+            locale="en",
+            platform="schwab",
+            account_alias="00682",
+            strategy_profile="soxl_soxx_trend_income",
+            decision=decision,
+        )
+        self.assertIn("原因：生产偏离严重（CRITICAL）", zh)
+        self.assertIn("打开管理站处理恢复", zh)
+        self.assertNotIn("accept ≠ live", zh)
+        self.assertIn("Reason: production drift critical", en)
+        self.assertIn("recovery / reset", en)
+        self.assertNotIn("accept ≠ live", en)
 
     def test_resolve_mandate_dd_budget_for_leveraged_profiles(self) -> None:
         from quant_platform_kit.risk.attention import resolve_mandate_dd_budget

tests/test_notification_localization.py

@@ -13,6 +13,7 @@
 from quant_platform_kit.common.operational_notification_localization import (
     format_operational_alert,
     format_operational_heartbeat_status,
+    localize_attention_reason_code,
     localize_operational_activity,
     operational_notification_text,
     resolve_operational_notification_locale,
@@ -161,6 +162,32 @@ def test_operational_locale_and_normal_heartbeat_are_bilingual(self):
             "[执行回执心跳] LongBridge SG\n状态：正常\n运行目标已停用；未提交订单。",
         )
 
+    def test_attention_reason_codes_are_localized(self):
+        self.assertEqual(
+            localize_attention_reason_code("zh", "new_risk_prohibited"),
+            "已禁止新增风险",
+        )
+        self.assertEqual(
+            localize_attention_reason_code("zh", "drift_critical"),
+            "生产偏离严重（CRITICAL）",
+        )
+        self.assertEqual(
+            localize_attention_reason_code("en", "PRODUCTION_DRIFT_REVIEW"),
+            "production drift review",
+        )
+        self.assertEqual(
+            localize_attention_reason_code("zh", "unknown_future_code"),
+            "unknown_future_code",
+        )
+        self.assertIn(
+            "确认意图不等于放行实盘",
+            operational_notification_text("zh", "attention_next_open_console_confirm"),
+        )
+        self.assertNotIn(
+            "accept",
+            operational_notification_text("zh", "attention_next_open_console_confirm"),
+        )
+
 
 if __name__ == "__main__":
     unittest.main()