feat(risk): envelope target scale + mandate DD budget

docs/capital_envelope_target_scale_v1.zh-CN.md

@@ -0,0 +1,19 @@
+# 资金信封目标缩仓与 Attention 预算（2026-09-18）
+
+> 状态：库侧本 PR；平台接线随后 pin 本 SHA。
+
+## 本 PR
+
+1. `apply_combined_scale_to_targets(targets, combined_scale)`  
+   - `None`/非法 scale → **omit**（原样返回有限非负目标）  
+   - 合法 scale ∈ [0,1] → 只缩不扩  
+2. `resolve_mandate_dd_budget(profile)`  
+   - SOXL/TQQQ 默认 `0.35`（非 10%）  
+   - 可被 `override` / `QSL_MANDATE_DD_BUDGET[_PROFILE]` 覆盖  
+3. `publish_attention_telegram_transition` — ACTION/HALT 跃迁才发 TG
+
+## 平台接线（后续 PR）
+
+- admission 后：`allocation["targets"] = apply_combined_scale_to_targets(..., admission.combined_scale)`  
+- 禁买/CRITICAL：调用 `publish_attention_telegram_transition` + marker  
+- 日损事实生产者：仍阻塞（见 DAILY_LOSS_FACT_PRODUCER_GAP）

src/quant_platform_kit/risk/__init__.py

@@ -95,8 +95,10 @@
     evaluate_attention,
     format_attention_compact_message,
     render_attention_compact,
+    resolve_mandate_dd_budget,
     should_notify_attention_transition,
 )
+from quant_platform_kit.risk.attention_notify import publish_attention_telegram_transition
 from quant_platform_kit.risk.synthetic_combo_evidence import (
     DEFAULT_CORRELATED_GROUP_CAP,
     DEFAULT_CORRELATION_THRESHOLD,
@@ -182,5 +184,7 @@
     "evaluate_attention",
     "format_attention_compact_message",
     "render_attention_compact",
+    "resolve_mandate_dd_budget",
     "should_notify_attention_transition",
+    "publish_attention_telegram_transition",
 ]

src/quant_platform_kit/risk/attention.py

@@ -300,3 +300,59 @@ def _coerce_level(value: AttentionLevel | str) -> AttentionLevel:
 def _clean_segment(value: object) -> str:
     text = str(value or "").strip().lower() or "unknown"
     return "".join(ch if ch.isalnum() or ch in {"-", "_", "."} else "-" for ch in text)[:80]
+
+
+# Leveraged live profiles: path DD commonly exceeds 10%; use mandate budgets, not 0.10.
+_DEFAULT_MANDATE_DD_BUDGET_BY_PROFILE: dict[str, float] = {
+    "soxl_soxx_trend_income": 0.35,
+    "tqqq_growth_income": 0.35,
+}
+
+
+def resolve_mandate_dd_budget(
+    strategy_profile: object | None,
+    *,
+    override: float | None = None,
+    environ: Mapping[str, str] | None = None,
+) -> float | None:
+    """Resolve mandate drawdown budget for attention (never invents 10% default).
+
+    Precedence: explicit ``override`` → ``QSL_MANDATE_DD_BUDGET_<PROFILE>`` →
+    ``QSL_MANDATE_DD_BUDGET`` → built-in leveraged profile map → ``None`` (omit DD axis).
+    """
+
+    import os
+
+    env = environ if environ is not None else os.environ
+    for candidate in (override, _env_budget(env, strategy_profile), _env_budget(env, None)):
+        parsed = _optional_positive_unit(candidate)
+        if parsed is not None:
+            return parsed
+    profile = str(strategy_profile or "").strip().lower()
+    if profile in _DEFAULT_MANDATE_DD_BUDGET_BY_PROFILE:
+        return _DEFAULT_MANDATE_DD_BUDGET_BY_PROFILE[profile]
+    return None
+
+
+def _env_budget(env: Mapping[str, str], strategy_profile: object | None) -> float | None:
+    if strategy_profile is None:
+        raw = env.get("QSL_MANDATE_DD_BUDGET")
+    else:
+        profile = str(strategy_profile or "").strip().upper().replace("-", "_")
+        raw = env.get(f"QSL_MANDATE_DD_BUDGET_{profile}") if profile else None
+    return _optional_positive_unit(raw)
+
+
+def _optional_positive_unit(value: object) -> float | None:
+    if value is None or value == "":
+        return None
+    try:
+        if isinstance(value, bool):
+            return None
+        number = float(value)
+    except (TypeError, ValueError):
+        return None
+    if number != number or number <= 0.0 or number > 1.0:
+        return None
+    return number
+

src/quant_platform_kit/risk/attention_notify.py

@@ -0,0 +1,116 @@
+"""Publish AttentionLevel ACTION/HALT transitions to Telegram (operator page).
+
+Does not grant live, raise RRL, or auto-resume. Marker recording only after send.
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Callable, Mapping, Sequence
+from typing import Any
+
+from quant_platform_kit.risk.attention import (
+    AttentionDecision,
+    AttentionLevel,
+    attention_transition_key,
+    render_attention_compact,
+    should_notify_attention_transition,
+)
+
+
+def publish_attention_telegram_transition(
+    *,
+    decision: AttentionDecision,
+    platform: str,
+    account_alias: str,
+    strategy_profile: str,
+    locale: object | None = None,
+    previous_level: AttentionLevel | str | None = None,
+    previous_reason_codes: Sequence[str] | None = None,
+    already_sent_keys: Sequence[str] | None = None,
+    record_sent_key: Callable[[str], Any] | None = None,
+    telegram_sender: Callable[..., bool] | None = None,
+    log_message: Callable[..., Any] = print,
+) -> dict[str, int]:
+    """Send one compact Telegram page on ACTION/HALT transition.
+
+    Returns counts ``sent`` / ``skipped`` / ``failed`` (sum-friendly for CLI).
+    """
+
+    counts = {"sent": 0, "skipped": 0, "failed": 0}
+    if not should_notify_attention_transition(
+        previous_level=previous_level,
+        new_level=decision.level,
+        previous_reason_codes=previous_reason_codes,
+        new_reason_codes=decision.reason_codes,
+    ):
+        counts["skipped"] += 1
+        return counts
+
+    primary = decision.reason_codes[0] if decision.reason_codes else decision.level.value
+    alert_key = attention_transition_key(
+        platform=platform,
+        account_alias=account_alias,
+        strategy_profile=strategy_profile,
+        level=decision.level,
+        primary_reason=primary,
+    )
+    seen = {str(key) for key in (already_sent_keys or ())}
+    if alert_key in seen:
+        counts["skipped"] += 1
+        return counts
+
+    text = render_attention_compact(
+        locale=locale or os.environ.get("QSL_NOTIFY_LANG") or os.environ.get("NOTIFY_LANG"),
+        platform=platform,
+        account_alias=account_alias,
+        strategy_profile=strategy_profile,
+        decision=decision,
+    )
+    sender = telegram_sender or _default_telegram_sender
+    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__}")
+        counts["failed"] += 1
+        return counts
+    if not ok:
+        log_message(f"attention_telegram_skipped key={alert_key} reason=telegram_not_configured_or_false")
+        counts["skipped"] += 1
+        return counts
+    counts["sent"] += 1
+    if record_sent_key is not None:
+        record_sent_key(alert_key)
+    return counts
+
+
+def _default_telegram_sender(**kwargs: Any) -> bool:
+    from quant_platform_kit.notifications.telegram import send_telegram_message
+
+    token = str(
+        os.environ.get("STRATEGY_PLUGIN_ALERT_TELEGRAM_BOT_TOKEN")
+        or os.environ.get("TELEGRAM_TOKEN")
+        or ""
+    ).strip()
+    chats = (
+        os.environ.get("QSL_GLOBAL_TELEGRAM_CHAT_ID")
+        or os.environ.get("STRATEGY_PLUGIN_ALERT_TELEGRAM_CHAT_IDS")
+        or os.environ.get("GLOBAL_TELEGRAM_CHAT_ID")
+        or ""
+    )
+    if not token or not str(chats).strip():
+        return False
+    text = str(kwargs.get("text") or "").strip()
+    if not text:
+        return False
+    return bool(
+        send_telegram_message(
+            bot_token=token,
+            chat_ids=chats,
+            text=text,
+            parse_mode=None,
+        )
+    )
+
+
+__all__ = ["publish_attention_telegram_transition"]

tests/test_attention.py

@@ -90,6 +90,53 @@ def test_compact_zh_cn_locale_normalizes(self) -> None:
         self.assertIn("下一步", text)
         self.assertIn("管理站", text)
 
+    def test_resolve_mandate_dd_budget_for_leveraged_profiles(self) -> None:
+        from quant_platform_kit.risk.attention import resolve_mandate_dd_budget
+
+        self.assertEqual(resolve_mandate_dd_budget("soxl_soxx_trend_income"), 0.35)
+        self.assertEqual(resolve_mandate_dd_budget("tqqq_growth_income"), 0.35)
+        self.assertIsNone(resolve_mandate_dd_budget("unknown_profile"))
+        self.assertEqual(
+            resolve_mandate_dd_budget("soxl_soxx_trend_income", override=0.40),
+            0.40,
+        )
+        self.assertEqual(
+            resolve_mandate_dd_budget(
+                "other",
+                environ={"QSL_MANDATE_DD_BUDGET": "0.25"},
+            ),
+            0.25,
+        )
+
+    def test_publish_attention_transition_dedups(self) -> None:
+        from quant_platform_kit.risk.attention_notify import publish_attention_telegram_transition
+
+        decision = evaluate_attention(AttentionAxes(new_risk_prohibited=True))
+        sent: list[str] = []
+        recorded: list[str] = []
+        counts = publish_attention_telegram_transition(
+            decision=decision,
+            platform="schwab",
+            account_alias="00682",
+            strategy_profile="soxl_soxx_trend_income",
+            locale="zh",
+            telegram_sender=lambda **kwargs: sent.append(str(kwargs.get("text") or "")) or True,
+            record_sent_key=recorded.append,
+        )
+        self.assertEqual(counts["sent"], 1)
+        self.assertEqual(len(sent), 1)
+        counts2 = publish_attention_telegram_transition(
+            decision=decision,
+            platform="schwab",
+            account_alias="00682",
+            strategy_profile="soxl_soxx_trend_income",
+            already_sent_keys=recorded,
+            telegram_sender=lambda **kwargs: sent.append(str(kwargs.get("text") or "")) or True,
+            record_sent_key=recorded.append,
+        )
+        self.assertEqual(counts2["skipped"], 1)
+        self.assertEqual(len(sent), 1)
+
 
 if __name__ == "__main__":
     unittest.main()