diff --git a/docs/capital_envelope_target_scale_v1.zh-CN.md b/docs/capital_envelope_target_scale_v1.zh-CN.md new file mode 100644 index 0000000..12f43ca --- /dev/null +++ b/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) diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index 1e6cdd7..61b257f 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/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", ] diff --git a/src/quant_platform_kit/risk/attention.py b/src/quant_platform_kit/risk/attention.py index 87b9103..dcd770f 100644 --- a/src/quant_platform_kit/risk/attention.py +++ b/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_`` → + ``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 + diff --git a/src/quant_platform_kit/risk/attention_notify.py b/src/quant_platform_kit/risk/attention_notify.py new file mode 100644 index 0000000..7b8a3cf --- /dev/null +++ b/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"] diff --git a/tests/test_attention.py b/tests/test_attention.py index 1a21d2c..fc46eea 100644 --- a/tests/test_attention.py +++ b/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()