fix: Clarify shared risk-block attention notifications

docs/operational-attention-v1.zh-CN.md

@@ -18,6 +18,10 @@
 - `publish_attention_telegram_transition` — ACTION/HALT 跃迁才发 TG
 - `build_drift_alert` / `publish_drift_alerts`：WATCH 不建页；Telegram 走真实 `send_telegram_message`；缺配置记 `skipped` 并打日志（不再吞异常）
 
+Attention Telegram 发送异常或返回 False 时，日志键的账户段固定显示为 `redacted`。
+发送、成功记录和重复通知判断仍使用原始去重键；失败或跳过不记成功 marker。
+消费方须通过后续协调的 QPK pin 更新采用此修复，旧 pin 不会自动获得日志脱敏。
+
 ## 分级
 
 | Level | 例 | TG |

docs/platform-notification-outcomes.md

@@ -166,3 +166,28 @@ including data branches. Findings contain file paths and rule names only; no
 scan report is uploaded or committed. A failed scheduled run uses the existing
 GitHub Actions failure notification mechanism. Repository and organization
 notification settings are unchanged.
+
+## Risk-block attention messages
+
+`risk.attention.render_attention_compact` and the shared Telegram publisher render
+broker, a masked account suffix (at most two characters), strategy, status, every
+reason code with a localized explanation, optional financial evidence, SGT run
+time, a caller-supplied HTTPS console run link, and one action line. Chinese
+variants and English use the same fields. Missing or invalid evidence is shown
+as unknown; a missing link is explicit. Links are never built from account IDs,
+and URL credentials, query strings and fragments are not displayed.
+
+The optional display arguments are `daily_loss_context`, `occurred_at` (an aware
+datetime or ISO timestamp), and `run_url`. Existing callers remain valid.
+`daily_loss_context` accepts `daily_loss_fact_status`, `daily_loss_usd`,
+`max_daily_loss_usd`, and `daily_loss_baseline_as_of`. A loss is displayed only
+when its fact status is `verified`; it is described as baseline-relative loss,
+not assumed to be intraday trading P&L. Limits must be explicit, never inferred
+from account equity. These fields do not change admission or notification dedup.
+
+Consumers need a coordinated QPK pin update before the new fields take effect.
+Schwab's adapter uses `render_attention_notification` as a feature marker and
+retains its legacy rendering until that update. Its existing notification report
+context supplies `started_at` and, when available, `console_run_url`. A console
+producer must supply the actual run URL; the renderer does not invent a route.
+Rollout and pin changes are separate from this presentation patch.

src/quant_platform_kit/common/operational_notification_localization.py

@@ -89,8 +89,8 @@
         "attention_next_console_optional": "下一步：可在管理站查看详情",
         "attention_next_open_console_confirm": "下一步：打开管理站确认意图（确认意图不等于放行实盘）",
         "attention_next_open_console_resume": "下一步：打开管理站处理恢复 / 复位（需授权）",
-        "attention_next_review_daily_loss": "下一步：在管理站核对当日净值、资金进出和亏损上限；保持禁止新增风险，确认意图不会解除保护",
-        "attention_next_inspect_guard_reason": "下一步：在管理站查看该次运行的风控原因码和账户状态；保持保护，确认意图不会解除保护",
+        "attention_next_review_daily_loss": "下一步：在管理站核对当日净值、资金进出和亏损上限；保持禁止新增风险",
+        "attention_next_inspect_guard_reason": "下一步：在管理站查看该次运行的风控原因码和账户状态；保持禁止新增风险",
     },
     "en": {
         "runtime_guard_title": "[Runtime Guard] {name}",
@@ -164,8 +164,8 @@
         "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 (confirming intent does not enable live)",
         "attention_next_open_console_resume": "Next: open the management console for recovery / reset (authorization required)",
-        "attention_next_review_daily_loss": "Next: review today's equity, cash flows and daily-loss limit in the console; keep new risk blocked. Confirming intent does not release the guard",
-        "attention_next_inspect_guard_reason": "Next: inspect this run's guard reason codes and account state; keep protections active. Confirming intent does not release the guard",
+        "attention_next_review_daily_loss": "Next: review today's equity, cash flows and daily-loss limit in the console; keep new risk blocked",
+        "attention_next_inspect_guard_reason": "Next: inspect this run's guard reason codes and account state; keep protections active",
     },
 }
 

src/quant_platform_kit/risk/attention.py

@@ -180,8 +180,11 @@ def render_attention_compact(
     strategy_profile: str,
     decision: AttentionDecision,
     guard_reason_codes: tuple[str, ...] = (),
+    daily_loss_context: Mapping[str, Any] | None = None,
+    occurred_at: object | None = None,
+    run_url: str | None = None,
 ) -> str:
-    """Four-line operator message: identity / result / reason / next step."""
+    """Operator message with display-only evidence; missing facts remain unknown."""
 
     from quant_platform_kit.common.display_names import display_name
     from quant_platform_kit.common.operational_notification_localization import (
@@ -190,6 +193,8 @@ def render_attention_compact(
         resolve_operational_notification_locale,
     )
 
+    from quant_platform_kit.risk.attention_display import attention_details, masked_account
+
     lang = resolve_operational_notification_locale(locale)
     level = decision.level
     # Presentation-only evidence from the guard; never change the decision,
@@ -208,14 +213,17 @@ def render_attention_compact(
         lang,
         "attention_title",
         platform=display_name("platforms", platform, lang),
-        account=display_name("accounts", account_alias, lang, platform=platform),
-        strategy=display_name("strategies", strategy_profile, lang),
+        account=masked_account(account_alias, lang),
+        strategy=display_name("strategies", strategy_profile, lang, fallback="策略" if lang == "zh" else "Strategy"),
     )
     result_key = _result_text_key(level, decision.reason_codes)
     result = operational_notification_text(lang, result_key)
     descriptions = []
     for code in reasons:
         explanation = localize_attention_reason_code(lang, code)
+        if str(platform).lower() == "schwab" and code == "DAILY_LOSS_LIMIT_EXCEEDED":
+            explanation = ("相对基准净值的下降达到配置上限" if lang == "zh" else
+                           "The decline relative to the equity baseline reached the configured limit")
         if explanation == code:
             explanation = operational_notification_text(lang, "attention_reason_unknown")
         descriptions.append(f"{code}: {explanation}")
@@ -228,12 +236,21 @@ def render_attention_compact(
     if level is AttentionLevel.ACTION:
         if "DAILY_LOSS_LIMIT_EXCEEDED" in reasons:
             next_key = "review_daily_loss"
-        elif "new_risk_prohibited" in reasons:
+        elif "new_risk_prohibited" in decision.reason_codes:
             next_key = "inspect_guard_reason"
     next_step = operational_notification_text(lang, f"attention_next_{next_key}")
-    return f"{emoji} {title}\n{result}\n{reason_line}\n{next_step}"
+    context = daily_loss_context
+    if "DAILY_LOSS_LIMIT_EXCEEDED" in reasons and not context:
+        context = {"daily_loss_fact_status": "unknown"}
+    details = attention_details(
+        locale=lang, daily_loss_context=context,
+        occurred_at=occurred_at, run_url=run_url,
+    )
+    return "\n".join((f"{emoji} {title}", result, reason_line, *details, next_step))
 
 
+# Feature marker for pinned consumers that retain a legacy renderer until rollout.
+render_attention_notification = render_attention_compact
 format_attention_compact_message = render_attention_compact
 
 

src/quant_platform_kit/risk/attention_display.py

@@ -0,0 +1,68 @@
+"""Display-only attention evidence. Never infer financial facts or run links."""
+from __future__ import annotations
+
+import math
+from collections.abc import Mapping
+from datetime import datetime
+from urllib.parse import urlsplit, urlunsplit
+from zoneinfo import ZoneInfo
+
+
+def masked_account(value: object, locale: str) -> str:
+    raw = str(value or "").strip()
+    label = "账户未知" if locale == "zh" else "Account unknown"
+    if raw.lower() in {"", "unknown", "none", "account", "账户"}:
+        return label
+    # Mask all aliases too: a display mapping may itself contain a raw account ID.
+    return "••" + raw[-2:] if len(raw) >= 2 else label
+
+
+def _stamp(value: object, zone: str, pattern: str) -> str | None:
+    try:
+        stamp = value if isinstance(value, datetime) else datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+        if stamp.tzinfo is None or stamp.utcoffset() is None:
+            return None
+        return stamp.astimezone(ZoneInfo(zone)).strftime(pattern)
+    except (ValueError, TypeError, OverflowError):
+        return None
+
+
+def _money(value: object, unknown: str) -> str:
+    if value is None or isinstance(value, bool):
+        return unknown
+    try:
+        number = float(value)
+        return f"USD {number:,.2f}" if math.isfinite(number) else unknown
+    except (TypeError, ValueError, OverflowError):
+        return unknown
+
+
+def attention_details(*, locale: str, daily_loss_context: Mapping | None,
+                      occurred_at: object, run_url: str | None) -> list[str]:
+    zh = locale == "zh"
+    unknown = "未知" if zh else "Unknown"
+    context = daily_loss_context if isinstance(daily_loss_context, Mapping) else {}
+    lines = []
+    # This field is a baseline-relative loss, not necessarily intraday trading P&L.
+    if context:
+        verified = context.get("daily_loss_fact_status") == "verified"
+        loss = _money(context.get("daily_loss_usd") if verified else None, unknown)
+        limit = _money(context.get("max_daily_loss_usd"), unknown)
+        lines.append(f"基准亏损 / 上限：{loss} / {limit}" if zh else f"Baseline loss / limit: {loss} / {limit}")
+        baseline = _stamp(context.get("daily_loss_baseline_as_of"), "America/New_York", "%Y-%m-%d %H:%M") if verified else None
+        if baseline:
+            lines.append(f"基准：{baseline}（纽约时间）的账户运行快照" if zh else f"Baseline: account runtime snapshot at {baseline} New York time")
+        else:
+            lines.append("基准时间和来源待核对" if zh else "Baseline time and source require review")
+    stamp = _stamp(occurred_at, "Asia/Singapore", "%Y-%m-%d %H:%M:%S SGT")
+    lines.append(("运行时间：" if zh else "Run time: ") + (stamp or unknown))
+    safe_url = None
+    try:
+        parsed = urlsplit(run_url or "")
+        if parsed.scheme == "https" and parsed.hostname and not parsed.username and not parsed.password and not any(c.isspace() for c in run_url or ""):
+            # Strip query credentials/fragments; never derive a link from account IDs.
+            safe_url = urlunsplit((parsed.scheme, parsed.netloc, parsed.path, "", ""))
+    except ValueError:
+        pass
+    lines.append(("运行详情：" if zh else "Run details: ") + (safe_url or ("未提供运行链接" if zh else "Run link not provided")))
+    return lines

src/quant_platform_kit/risk/attention_notify.py

@@ -26,6 +26,9 @@ def publish_attention_telegram_transition(
     strategy_profile: str,
     locale: object | None = None,
     guard_reason_codes: tuple[str, ...] = (),
+    daily_loss_context: Mapping[str, Any] | None = None,
+    occurred_at: object | None = None,
+    run_url: str | None = None,
     previous_level: AttentionLevel | str | None = None,
     previous_reason_codes: Sequence[str] | None = None,
     already_sent_keys: Sequence[str] | None = None,
@@ -68,16 +71,27 @@ def publish_attention_telegram_transition(
         strategy_profile=strategy_profile,
         decision=decision,
         guard_reason_codes=guard_reason_codes,
+        daily_loss_context=daily_loss_context,
+        occurred_at=occurred_at,
+        run_url=run_url,
     )
     sender = telegram_sender or _default_telegram_sender
+    # Logging must never expose the account-bearing deduplication key.
+    log_key = attention_transition_key(
+        platform=platform,
+        account_alias="redacted",
+        strategy_profile=strategy_profile,
+        level=decision.level,
+        primary_reason=primary,
+    )
     try:
         ok = bool(sender(text=text, alert_key=alert_key))
     except Exception as exc:  # noqa: BLE001
-        log_message(f"attention_telegram_failed key={alert_key} error={type(exc).__name__}")
+        log_message(f"attention_telegram_failed key={log_key} error={type(exc).__name__}")
         counts["failed"] += 1
         return counts
     if not ok:
-        log_message(f"attention_telegram_skipped key={alert_key} reason=telegram_not_configured_or_false")
+        log_message(f"attention_telegram_skipped key={log_key} reason=telegram_not_configured_or_false")
         counts["skipped"] += 1
         return counts
     counts["sent"] += 1

tests/test_attention.py

@@ -92,7 +92,8 @@ def test_compact_zh_cn_locale_normalizes(self) -> None:
         self.assertNotIn("原因：已禁止新增风险", text)
         self.assertIn("下一步", text)
         self.assertIn("管理站", text)
-        self.assertIn("确认意图不会解除保护", text)
+        self.assertIn("保持禁止新增风险", text)
+        self.assertNotIn("确认意图", text)
         self.assertNotIn("accept", text)
 
     def test_compact_localizes_drift_critical_reason_zh_and_en(self) -> None:

tests/test_attention_display.py

@@ -0,0 +1,61 @@
+"""Synthetic display evidence; no broker calls or real notifications."""
+from datetime import datetime, timezone
+
+import pytest
+
+from quant_platform_kit.risk.attention import AttentionAxes, evaluate_attention, render_attention_compact
+from quant_platform_kit.risk.attention_notify import publish_attention_telegram_transition
+
+
+@pytest.mark.parametrize("platform", ["schwab", "ibkr", "longbridge", "binance", "alpaca", "firstrade", "qmt"])
+@pytest.mark.parametrize("locale", ["zh-CN", "en"])
+def test_brokers_share_masked_identity_codes_numbers_and_run_evidence(platform, locale):
+    decision = evaluate_attention(AttentionAxes(new_risk_prohibited=True))
+    original = decision.to_dict()
+    sent = []
+    result = publish_attention_telegram_transition(
+        decision=decision, platform=platform, account_alias="fake-account-42",
+        strategy_profile="soxl_soxx_trend_income", locale=locale,
+        guard_reason_codes=("DAILY_LOSS_UNKNOWN_FAIL_CLOSED",),
+        daily_loss_context={"daily_loss_fact_status": "verified", "daily_loss_usd": 12.5, "max_daily_loss_usd": 100},
+        occurred_at=datetime(2026, 11, 2, 14, 0, tzinfo=timezone.utc),
+        run_url="https://console.example.invalid/runs/fake-run?token=fake-secret#fragment",
+        telegram_sender=lambda **kwargs: sent.append(kwargs) or True,
+    )
+    text = sent[0]["text"]
+    assert result == {"sent": 1, "skipped": 0, "failed": 0}
+    assert "••42" in text and "fake-account" not in text
+    assert "DAILY_LOSS_UNKNOWN_FAIL_CLOSED" in text
+    assert "USD 12.50 / USD 100.00" in text
+    assert "2026-11-02 22:00:00 SGT" in text
+    assert "https://console.example.invalid/runs/fake-run" in text
+    assert "fake-secret" not in text and "fragment" not in text
+    assert "确认意图" not in text and "Confirming intent" not in text
+    assert sent[0]["alert_key"].endswith("/new_risk_prohibited")
+    assert decision.to_dict() == original
+
+
+@pytest.mark.parametrize("bad", [None, "bad", "2026-11-02T14:00:00", float("nan"), True])
+def test_missing_or_invalid_evidence_stays_unknown(bad):
+    text = render_attention_compact(
+        locale="en", platform="schwab", account_alias="unknown", strategy_profile="fake",
+        decision=evaluate_attention(AttentionAxes(new_risk_prohibited=True)),
+        occurred_at=bad, run_url="https://fake:fake@example.invalid/run",
+        daily_loss_context={"daily_loss_fact_status": "unverified", "daily_loss_usd": 40, "max_daily_loss_usd": bad},
+    )
+    assert "Account unknown" in text
+    assert "Run time: Unknown" in text
+    assert "Run link not provided" in text
+    assert "Baseline loss / limit: Unknown / Unknown" in text
+    assert "USD 40" not in text
+    assert "fake:fake" not in text
+
+
+@pytest.mark.parametrize("url", ["http://example.invalid", "javascript:fake", "https://[", "https://example.invalid/\nfake"])
+def test_bad_links_do_not_break_or_inject_message(url):
+    text = render_attention_compact(
+        locale="en", platform="ibkr", account_alias="fake42", strategy_profile="fake",
+        decision=evaluate_attention(AttentionAxes(new_risk_prohibited=True)), run_url=url,
+    )
+    assert "Run link not provided" in text
+    assert url not in text

tests/test_attention_log_redaction.py

@@ -0,0 +1,70 @@
+"""Publisher logs mask synthetic account IDs without changing deduplication."""
+
+import pytest
+
+from quant_platform_kit.risk.attention import (
+    attention_transition_key,
+    evaluate_attention,
+)
+from quant_platform_kit.risk.attention_notify import publish_attention_telegram_transition
+
+
+@pytest.mark.parametrize("failure", ["exception", "false"])
+@pytest.mark.parametrize("account", ["fake-account-12345678", "FAKE/ACCOUNT-12345678", "x"])
+def test_failure_logs_redact_account_and_preserve_retry_dedup(failure, account):
+    decision = evaluate_attention(new_risk_prohibited=True)
+    key = attention_transition_key(
+        platform="demo", account_alias=account, strategy_profile="demo-strategy",
+        level=decision.level, primary_reason=decision.reason_codes[0],
+    )
+    logs, calls, recorded = [], [], []
+
+    def fail(**kwargs):
+        calls.append(kwargs)
+        if failure == "exception":
+            raise RuntimeError(account)
+        return False
+
+    kwargs = dict(
+        decision=decision, platform="demo", account_alias=account,
+        strategy_profile="demo-strategy", log_message=logs.append,
+        record_sent_key=recorded.append,
+    )
+    result = publish_attention_telegram_transition(**kwargs, telegram_sender=fail)
+    assert result == {"sent": 0, "skipped": int(failure == "false"), "failed": int(failure == "exception")}
+    assert calls[0]["alert_key"] == key
+    assert recorded == []
+    assert len(logs) == 1
+    assert account not in logs[0]
+    assert key.split("/")[2] not in logs[0]
+    assert "key=attention/demo/redacted/demo-strategy/action/new_risk_prohibited" in logs[0]
+    assert ("error=RuntimeError" if failure == "exception" else "reason=telegram_not_configured_or_false") in logs[0]
+
+    def succeed(**kwargs):
+        calls.append(kwargs)
+        return True
+
+    # A failed/skipped delivery remains retryable and records the original key.
+    assert publish_attention_telegram_transition(**kwargs, telegram_sender=succeed)["sent"] == 1
+    assert calls[-1]["alert_key"] == key
+    assert recorded == [key]
+    # The original key still suppresses the next delivery before invoking sender.
+    assert publish_attention_telegram_transition(
+        **kwargs, telegram_sender=succeed, already_sent_keys=recorded,
+    ) == {"sent": 0, "skipped": 1, "failed": 0}
+    assert len(calls) == 2
+    assert len(logs) == 1
+
+
+def test_accounts_with_same_suffix_keep_distinct_dedup_keys():
+    recorded, calls = [], []
+    for account in ("fake-first-78", "fake-second-78"):
+        result = publish_attention_telegram_transition(
+            decision=evaluate_attention(new_risk_prohibited=True),
+            platform="demo", account_alias=account, strategy_profile="demo-strategy",
+            already_sent_keys=recorded, record_sent_key=recorded.append,
+            telegram_sender=lambda **kwargs: calls.append(kwargs) or True,
+        )
+        assert result == {"sent": 1, "skipped": 0, "failed": 0}
+    assert len(set(recorded)) == 2
+    assert recorded == [call["alert_key"] for call in calls]