From 8cb33e133693565168c748698ca1105a654d21f8 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 18 Sep 2026 07:57:53 +0800 Subject: [PATCH] feat(risk): operational attention levels and fix drift Telegram dispatch Add mandate-relative AttentionLevel evaluation so leveraged-path drawdowns do not page operators, and wire drift REVIEW/CRITICAL alerts to the real Telegram sender with observable skip/fail instead of a missing import. Co-authored-by: Cursor --- docs/operational_attention_v1.zh-CN.md | 39 +++ .../operational_notification_localization.py | 24 +- src/quant_platform_kit/risk/__init__.py | 22 ++ src/quant_platform_kit/risk/attention.py | 302 +++++++++++++++++ .../strategy_lifecycle/drift_alerts.py | 307 ++++++++---------- tests/test_attention.py | 95 ++++++ tests/test_drift_alerts.py | 104 ++++++ 7 files changed, 727 insertions(+), 166 deletions(-) create mode 100644 docs/operational_attention_v1.zh-CN.md create mode 100644 src/quant_platform_kit/risk/attention.py create mode 100644 tests/test_attention.py create mode 100644 tests/test_drift_alerts.py diff --git a/docs/operational_attention_v1.zh-CN.md b/docs/operational_attention_v1.zh-CN.md new file mode 100644 index 0000000..086d029 --- /dev/null +++ b/docs/operational_attention_v1.zh-CN.md @@ -0,0 +1,39 @@ +# 可行动注意力门槛 V1（AttentionLevel） + +> 状态：库侧纯函数 + drift Telegram 分发修复（本 PR）；平台周期接线与 mandate 预算表另票。 +> 范围：通知分级 / 人工恢复路由；不授权 live、不抬 RRL、不做 Kelly。 + +## 原则 + +1. **不**用全局绝对回撤（如 10%）对 TQQQ/SOXL 发 Telegram。 +2. 回撤仅相对显式 `mandate_dd_budget`；缺预算则 **omit** 该轴。 +3. 仅 `ACTION` / `HALT` 可通知；键为状态跃迁，发送成功后才记 marker。 +4. 自动刹车仍在 NEW_RISK / RiskEngine；本模块只分类注意力与短文案。 + +## API + +- `evaluate_attention(AttentionAxes | mapping) → AttentionDecision` +- `attention_transition_key(...)` 去重键 +- `render_attention_compact(locale=...)`（`zh-CN`→`zh`） +- `build_drift_alert` / `publish_drift_alerts`：WATCH 不建页；Telegram 走真实 `send_telegram_message`；缺配置记 `skipped` 并打日志（不再吞异常） + +## 分级 + +| Level | 例 | TG | +|---|---|---| +| OK | 预算内路径回撤 | 否 | +| WATCH | drift watch；dd_ratio∈[0.7,1) | 否 | +| ACTION | NEW_RISK 禁买；drift review；dd 打穿 mandate | 是（跃迁） | +| HALT | drift critical；对账不确定；hard PARK | 是（跃迁） | + +## 复利风控审计对照（同批） + +生存层（拒单 / 禁新买 / Policy A / RRL）已能支撑「先活下来」的几何复利。 +连续最优仓位仍缺：日损事实生产者、`combined_scale` live 缩仓、`assess_with_evidence`。 +**默认后置** Kelly / 抬 RRL；下一有界工程仅「日损事实→禁买」（材料齐时）。 + +## 后续（非本 PR） + +- 平台在禁买/CRITICAL 首次跃迁时调用 publish + marker store +- 为 SOXL/TQQQ 配置 mandate_dd_budget（研究/mandate 口径，非 10%） +- 信封 `combined_scale`→目标缩仓另票 diff --git a/src/quant_platform_kit/common/operational_notification_localization.py b/src/quant_platform_kit/common/operational_notification_localization.py index 1f479a6..2c7d403 100644 --- a/src/quant_platform_kit/common/operational_notification_localization.py +++ b/src/quant_platform_kit/common/operational_notification_localization.py @@ -48,7 +48,18 @@ "workflow_heartbeat_latest_failed": "最近一次运行未成功完成（结论：{conclusion}）", "workflow_heartbeat_no_success": "GitHub Actions 查询未返回成功的运行记录", "workflow_heartbeat_query_failed": "GitHub Actions 运行记录查询失败", - "workflow_heartbeat_missing_dispatches": "已连续 {count} 个预期周期未发现运行（阈值：{threshold}）", + "workflow_heartbeat_missing_dispatches": "已连续 {count} 个预期周期未发现运行（频率：{threshold}）", + "attention_title": "{platform} · {account} · {strategy}", + "attention_result_ok": "结果：运行正常", + "attention_result_watch": "结果：观察中（无需立即处理）", + "attention_result_action": "结果：需确认（已限制新风险或偏离待审）", + "attention_result_halt": "结果：硬停 / 运行异常，需人工恢复", + "attention_result_new_risk_prohibited": "结果：已禁止新增风险", + "attention_reason": "原因：{code}", + "attention_next_none": "下一步：无需操作", + "attention_next_console_optional": "下一步：可在管理站查看详情", + "attention_next_open_console_confirm": "下一步：打开管理站确认意图（accept ≠ live）", + "attention_next_open_console_resume": "下一步：打开管理站处理恢复 / 复位（需授权）", }, "en": { "runtime_guard_title": "[Runtime Guard] {name}", @@ -85,6 +96,17 @@ "workflow_heartbeat_no_success": "The GitHub Actions query returned no successful runtime run", "workflow_heartbeat_query_failed": "The GitHub Actions runtime-run query failed", "workflow_heartbeat_missing_dispatches": "No runtime dispatch was found for {count} expected interval(s) (threshold: {threshold})", + "attention_title": "{platform} · {account} · {strategy}", + "attention_result_ok": "Result: normal", + "attention_result_watch": "Result: watch (no immediate action)", + "attention_result_action": "Result: action needed (new risk limited or drift review)", + "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_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_resume": "Next: open the management console for recovery / reset (authorization required)", }, } diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index 128f77c..3337bbe 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/src/quant_platform_kit/risk/__init__.py @@ -84,6 +84,18 @@ resolve_risk_profile_scale, size_target_weight, ) +from quant_platform_kit.risk.attention import ( + AttentionAxes, + AttentionDecision, + AttentionLevel, + attention_alert_key, + attention_level_from_drift_status, + attention_transition_key, + evaluate_attention, + format_attention_compact_message, + render_attention_compact, + should_notify_attention_transition, +) from quant_platform_kit.risk.synthetic_combo_evidence import ( DEFAULT_CORRELATED_GROUP_CAP, DEFAULT_CORRELATION_THRESHOLD, @@ -159,4 +171,14 @@ "SyntheticComboMember", "SyntheticComboMemberEvidence", "evaluate_synthetic_combo_evidence", + "AttentionAxes", + "AttentionDecision", + "AttentionLevel", + "attention_alert_key", + "attention_level_from_drift_status", + "attention_transition_key", + "evaluate_attention", + "format_attention_compact_message", + "render_attention_compact", + "should_notify_attention_transition", ] diff --git a/src/quant_platform_kit/risk/attention.py b/src/quant_platform_kit/risk/attention.py new file mode 100644 index 0000000..87b9103 --- /dev/null +++ b/src/quant_platform_kit/risk/attention.py @@ -0,0 +1,302 @@ +"""Operational attention levels for notify / HITL routing (not a sizing engine). + +Synthesizes capital (new-risk), operational fault, and research-drift axes into +``OK | WATCH | ACTION | HALT``. Absolute drawdown percentages are never enough: +drawdown only contributes when a mandate budget is supplied +(``dd_ratio = drawdown_from_peak / mandate_dd_budget``). + +This module does not send Telegram, grant live, raise RRL, or compute Kelly. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from typing import Any, Mapping + + +class AttentionLevel(str, Enum): + OK = "ok" + WATCH = "watch" + ACTION = "action" + HALT = "halt" + + @property + def severity_order(self) -> int: + return { + AttentionLevel.OK: 0, + AttentionLevel.WATCH: 1, + AttentionLevel.ACTION: 2, + AttentionLevel.HALT: 3, + }[self] + + +_NOTIFY_LEVELS = frozenset({AttentionLevel.ACTION, AttentionLevel.HALT}) + +_DRIFT_STATUS_LEVEL = { + "critical": AttentionLevel.HALT, + "review": AttentionLevel.ACTION, + "watch": AttentionLevel.WATCH, + "healthy": AttentionLevel.OK, + "ok": AttentionLevel.OK, +} + + +@dataclass(frozen=True) +class AttentionAxes: + """Explicit attention inputs; missing fields omit that axis (fail-soft).""" + + new_risk_prohibited: bool | None = None + drift_status: str | None = None + production_drift_status: str | None = None + operational_uncertain: bool | None = None + operational_fault: bool | None = None + hard_park: bool | None = None + drawdown_from_peak: float | None = None + mandate_dd_budget: float | None = None + + +@dataclass(frozen=True) +class AttentionDecision: + level: AttentionLevel + reason_codes: tuple[str, ...] + dd_ratio: float | None + should_notify: bool + next_step_key: str + + def to_dict(self) -> dict[str, Any]: + return { + "level": self.level.value, + "reason_codes": list(self.reason_codes), + "dd_ratio": self.dd_ratio, + "should_notify": self.should_notify, + "next_step_key": self.next_step_key, + } + + +def evaluate_attention(axes: AttentionAxes | Mapping[str, Any] | None = None, **kwargs: Any) -> AttentionDecision: + """Combine explicit axes into one attention level. + + Accepts ``AttentionAxes``, a mapping, or keyword overrides. Drawdown without + ``mandate_dd_budget`` does not raise level. + """ + + resolved = _coerce_axes(axes, kwargs) + candidates: list[tuple[AttentionLevel, str]] = [] + + if resolved.hard_park is True: + candidates.append((AttentionLevel.HALT, "hard_park")) + if resolved.operational_uncertain is True or resolved.operational_fault is True: + candidates.append((AttentionLevel.HALT, "operational_uncertain")) + + drift_key = str( + resolved.drift_status or resolved.production_drift_status or "" + ).strip().lower() + if drift_key: + drift_level = _DRIFT_STATUS_LEVEL.get(drift_key) + if drift_level is None: + candidates.append((AttentionLevel.WATCH, "drift_status_unknown")) + elif drift_level is AttentionLevel.HALT: + candidates.append((AttentionLevel.HALT, "drift_critical")) + elif drift_level is AttentionLevel.ACTION: + candidates.append((AttentionLevel.ACTION, "drift_review")) + elif drift_level is AttentionLevel.WATCH: + candidates.append((AttentionLevel.WATCH, "drift_watch")) + + if resolved.new_risk_prohibited is True: + candidates.append((AttentionLevel.ACTION, "new_risk_prohibited")) + + dd_ratio = _resolve_dd_ratio(resolved.drawdown_from_peak, resolved.mandate_dd_budget) + if dd_ratio is not None: + if dd_ratio >= 1.0: + candidates.append((AttentionLevel.ACTION, "mandate_dd_exhausted")) + elif dd_ratio >= 0.7: + candidates.append((AttentionLevel.WATCH, "mandate_dd_elevated")) + + if not candidates: + level = AttentionLevel.OK + reasons: tuple[str, ...] = () + else: + level = max(candidates, key=lambda item: item[0].severity_order)[0] + # Keep every contributing reason so operators see the full stack. + reasons = tuple(dict.fromkeys(code for _, code in candidates)) + + return AttentionDecision( + level=level, + reason_codes=reasons, + dd_ratio=dd_ratio, + should_notify=level in _NOTIFY_LEVELS, + next_step_key=_next_step_key(level, reasons), + ) + + +def attention_transition_key( + *, + platform: str, + account_alias: str, + strategy_profile: str, + level: AttentionLevel | str, + primary_reason: str, +) -> str: + """Stable dedup key for attention notifications (not price/time based).""" + + return "/".join( + [ + "attention", + _clean_segment(platform), + _clean_segment(account_alias), + _clean_segment(strategy_profile), + _coerce_level(level).value, + _clean_segment(primary_reason), + ] + ) + + +# Back-compat alias used by early drafts / exports. +attention_alert_key = attention_transition_key + + +def should_notify_attention_transition( + *, + previous_level: AttentionLevel | str | None, + new_level: AttentionLevel | str, + previous_reason_codes: tuple[str, ...] | list[str] | None = None, + new_reason_codes: tuple[str, ...] | list[str] | None = None, +) -> bool: + current = _coerce_level(new_level) + if current not in _NOTIFY_LEVELS: + return False + prior = _coerce_level(previous_level) if previous_level is not None else None + if prior != current: + return True + return tuple(previous_reason_codes or ()) != tuple(new_reason_codes or ()) + + +def render_attention_compact( + *, + locale: object | None, + platform: str, + account_alias: str, + strategy_profile: str, + decision: AttentionDecision, +) -> str: + """Four-line operator message: identity / result / reason / next step.""" + + from quant_platform_kit.common.operational_notification_localization import ( + operational_notification_text, + resolve_operational_notification_locale, + ) + + lang = resolve_operational_notification_locale(locale) + level = decision.level + reason = decision.reason_codes[0] if decision.reason_codes else "none" + emoji = { + AttentionLevel.OK: "✅", + AttentionLevel.WATCH: "👀", + AttentionLevel.ACTION: "⚠️", + AttentionLevel.HALT: "🚨", + }[level] + title = operational_notification_text( + lang, + "attention_title", + platform=platform, + account=account_alias, + strategy=strategy_profile, + ) + 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) + 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}" + + +format_attention_compact_message = render_attention_compact + + +def attention_level_from_drift_status(status: object) -> AttentionLevel: + value = getattr(status, "value", status) + return _DRIFT_STATUS_LEVEL.get(str(value or "").strip().lower(), AttentionLevel.WATCH) + + +def _coerce_axes( + axes: AttentionAxes | Mapping[str, Any] | None, + kwargs: Mapping[str, Any], +) -> AttentionAxes: + if isinstance(axes, AttentionAxes) and not kwargs: + return axes + raw: dict[str, Any] = {} + if isinstance(axes, AttentionAxes): + raw.update( + { + "new_risk_prohibited": axes.new_risk_prohibited, + "drift_status": axes.drift_status, + "production_drift_status": axes.production_drift_status, + "operational_uncertain": axes.operational_uncertain, + "operational_fault": axes.operational_fault, + "hard_park": axes.hard_park, + "drawdown_from_peak": axes.drawdown_from_peak, + "mandate_dd_budget": axes.mandate_dd_budget, + } + ) + elif isinstance(axes, Mapping): + raw.update(dict(axes)) + raw.update(dict(kwargs)) + return AttentionAxes( + new_risk_prohibited=raw.get("new_risk_prohibited"), + drift_status=raw.get("drift_status"), + production_drift_status=raw.get("production_drift_status"), + operational_uncertain=raw.get("operational_uncertain"), + operational_fault=raw.get("operational_fault"), + hard_park=raw.get("hard_park"), + drawdown_from_peak=raw.get("drawdown_from_peak"), + mandate_dd_budget=raw.get("mandate_dd_budget"), + ) + + +def _resolve_dd_ratio( + drawdown_from_peak: float | None, + mandate_dd_budget: float | None, +) -> float | None: + if drawdown_from_peak is None or mandate_dd_budget is None: + return None + try: + dd = float(drawdown_from_peak) + budget = float(mandate_dd_budget) + except (TypeError, ValueError): + return None + if budget <= 0.0 or dd < 0.0: + return None + return dd / budget + + +def _next_step_key(level: AttentionLevel, reasons: tuple[str, ...]) -> str: + if "operational_uncertain" in reasons or "hard_park" in reasons: + return "open_console_resume" + if level is AttentionLevel.HALT: + return "open_console_resume" + if level is AttentionLevel.ACTION: + return "open_console_confirm" + if level is AttentionLevel.WATCH: + return "console_optional" + return "none" + + +def _result_text_key(level: AttentionLevel, reasons: tuple[str, ...]) -> str: + if "new_risk_prohibited" in reasons and level in _NOTIFY_LEVELS: + return "attention_result_new_risk_prohibited" + return f"attention_result_{level.value}" + + +def _coerce_level(value: AttentionLevel | str) -> AttentionLevel: + if isinstance(value, AttentionLevel): + return value + key = str(value or "").strip().lower() + try: + return AttentionLevel(key) + except ValueError: + return AttentionLevel.WATCH + + +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] diff --git a/src/quant_platform_kit/strategy_lifecycle/drift_alerts.py b/src/quant_platform_kit/strategy_lifecycle/drift_alerts.py index 0761c46..b574c5e 100644 --- a/src/quant_platform_kit/strategy_lifecycle/drift_alerts.py +++ b/src/quant_platform_kit/strategy_lifecycle/drift_alerts.py @@ -1,36 +1,41 @@ -"""Drift alert signal builder — integrates with existing notification channels.""" +"""Drift alert signal builder — Telegram via real send path + attention keys.""" from __future__ import annotations -from collections.abc import Mapping +import os +from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from datetime import date -from typing import Any, Sequence - +from typing import Any + +from quant_platform_kit.risk.attention import ( + AttentionAxes, + AttentionDecision, + AttentionLevel, + attention_level_from_drift_status, + attention_transition_key, + evaluate_attention, + render_attention_compact, +) from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus from quant_platform_kit.strategy_lifecycle.drift_policy import DriftPolicy +_PAGEABLE = frozenset({DriftStatus.REVIEW, DriftStatus.CRITICAL}) + @dataclass(frozen=True) class DriftAlertEvent: - """A drift alert ready for dispatch through notification channels.""" - strategy_profile: str domain: str as_of: date drift_score: float status: DriftStatus escalated: bool - - # Alert content subject: str body: str - - # Metadata for dedup and routing alert_key: str channels: tuple[str, ...] - severity: str # info, warning, critical - + severity: str metadata: Mapping[str, Any] = field(default_factory=dict) @@ -39,84 +44,50 @@ def build_drift_alert( *, policy: DriftPolicy | None = None, previous_alerts_sent: int = 0, + locale: object | None = None, + platform: str | None = None, + account_alias: str | None = None, ) -> DriftAlertEvent | None: - """Build a drift alert event from a DriftResult. + """Build a pageable drift alert. WATCH/HEALTHY return None (no Telegram page).""" - Returns None if the drift status is HEALTHY or if the alert should be suppressed - (e.g., cooldown active, weekly limit reached). - - Args: - drift: The drift analysis result. - policy: Drift policy for cooldown/limit checks. - previous_alerts_sent: Count of alerts already sent for this strategy this week. - - Returns: - DriftAlertEvent if an alert should be sent, None otherwise. - """ policy = policy or DriftPolicy.load_default() - if drift.as_of is None or drift.status == DriftStatus.HEALTHY: return None - + if drift.status not in _PAGEABLE: + return None if drift.alert_suppressed: return None - if previous_alerts_sent >= policy.max_alerts_per_strategy_per_week: return None - # Determine severity - severity_map = { - DriftStatus.WATCH: "info", - DriftStatus.REVIEW: "warning", - DriftStatus.CRITICAL: "critical", - } - severity = severity_map.get(drift.status, "info") - - # Build subject - status_labels = { - DriftStatus.WATCH: "⚠️ WATCH", - DriftStatus.REVIEW: "🔴 REVIEW", - DriftStatus.CRITICAL: "🚨 CRITICAL", - } - label = status_labels.get(drift.status, "UNKNOWN") - escalated_tag = " [ESCALATED]" if drift.escalated else "" - subject = f"[{drift.domain}] {label}{escalated_tag}: {drift.strategy_profile} — drift={drift.drift_score:.2f}" - - # Build body - lines = [ - "Strategy Lifecycle Drift Alert", - "", - f"Strategy: {drift.strategy_profile}", - f"Domain: {drift.domain}", - f"As-of: {drift.as_of}", - f"Drift Score: {drift.drift_score:.3f}", - f"Status: {drift.status.value.upper()}", - f"Escalated: {'Yes' if drift.escalated else 'No'}", - "", - "Breached Dimensions:", - ] - - breached = drift.breached_dimensions - if breached: - for dim in breached: - lines.append( - f" - {dim.metric_name}: actual={dim.actual:.4f}, " - f"expected={dim.expected:.4f}, " - f"deviation={dim.deviation_pct:.1%} (threshold: {dim.threshold:.1%})" - ) - else: - lines.append(" (no individual dimensions breached — composite score triggered)") - - lines.extend( - [ - "", - "Action required:", - _action_for_status(drift.status), - ] + attention = evaluate_attention( + AttentionAxes(drift_status=drift.status.value) + ) + platform_id = str(platform or drift.domain or "platform").strip() or "platform" + account = str(account_alias or "-").strip() or "-" + primary_reason = attention.reason_codes[0] if attention.reason_codes else f"drift_{drift.status.value}" + alert_key = attention_transition_key( + platform=platform_id, + account_alias=account, + strategy_profile=drift.strategy_profile, + level=attention.level, + primary_reason=primary_reason, ) - # Build dedup key - alert_key = f"drift/{drift.domain}/{drift.strategy_profile}/{drift.as_of.isoformat()}/{drift.status.value}" + severity = "critical" if drift.status == DriftStatus.CRITICAL else "warning" + label = "🚨 CRITICAL" if drift.status == DriftStatus.CRITICAL else "🔴 REVIEW" + escalated_tag = " [ESCALATED]" if drift.escalated else "" + subject = ( + f"[{drift.domain}] {label}{escalated_tag}: " + f"{drift.strategy_profile} — drift={drift.drift_score:.2f}" + ) + body = render_attention_compact( + locale=locale, + platform=platform_id, + account_alias=account, + strategy_profile=drift.strategy_profile, + decision=attention, + ) return DriftAlertEvent( strategy_profile=drift.strategy_profile, @@ -126,15 +97,15 @@ def build_drift_alert( status=drift.status, escalated=drift.escalated, subject=subject, - body="\n".join(lines), + body=body, alert_key=alert_key, channels=policy.notification_channels, severity=severity, metadata={ "alert_type": "drift", + "attention_level": attention.level.value, + "reason_codes": list(attention.reason_codes), "drift_score": drift.drift_score, - "breached_count": len(breached), - "dimensions": {k: v.to_dict() for k, v in drift.dimensions.items()}, }, ) @@ -143,98 +114,104 @@ def publish_drift_alerts( events: Sequence[DriftAlertEvent], *, dry_run: bool = False, + telegram_sender: Callable[..., bool] | None = None, + already_sent_keys: Sequence[str] | None = None, + record_sent_key: Callable[[str], Any] | None = None, + log_message: Callable[..., Any] = print, ) -> dict[str, int]: - """Publish drift alerts through configured notification channels. + """Publish pageable drift alerts. Returns counts: sent/skipped/failed. - Integrates with QuantPlatformKit notification system (telegram, email, etc.). - - Args: - events: Alert events to publish. - dry_run: If True, log but don't actually send. - - Returns: - Dict of channel → count of alerts published. + Compatible with lifecycle CLI ``sum(counts.values())``. """ - counts: dict[str, int] = {} - for event in events: - for channel in event.channels: - try: - if not dry_run: - _dispatch_to_channel(event, channel) - counts[channel] = counts.get(channel, 0) + 1 - except Exception: - # Don't let one failed channel block others - pass + counts = {"sent": 0, "skipped": 0, "failed": 0} + seen = {str(key) for key in (already_sent_keys or ())} + for event in events: + if "telegram" not in event.channels: + _log(log_message, f"drift_alert_skipped_no_telegram_channel key={event.alert_key}") + counts["skipped"] += 1 + continue + if event.alert_key in seen: + counts["skipped"] += 1 + continue + if dry_run: + _log(log_message, f"drift_alert_dry_run key={event.alert_key}") + counts["sent"] += 1 + seen.add(event.alert_key) + if record_sent_key is not None: + record_sent_key(event.alert_key) + continue + + sender = telegram_sender or _default_telegram_sender + try: + ok = bool( + sender( + subject=event.subject, + body=event.body, + text=f"{event.subject}\n\n{event.body}", + alert_key=event.alert_key, + ) + ) + except Exception as exc: # noqa: BLE001 + _log( + log_message, + f"drift_alert_send_failed key={event.alert_key} error={type(exc).__name__}", + ) + counts["failed"] += 1 + continue + if not ok: + _log(log_message, f"telegram_not_configured_or_send_false key={event.alert_key}") + # Missing config is skipped (observable), hard false from sender after config is failed. + # Default sender returns False only when unconfigured → skipped. + if telegram_sender is None: + counts["skipped"] += 1 + _log(log_message, f"telegram_not_configured key={event.alert_key}") + else: + counts["failed"] += 1 + continue + counts["sent"] += 1 + seen.add(event.alert_key) + if record_sent_key is not None: + record_sent_key(event.alert_key) return counts -def _dispatch_to_channel(event: DriftAlertEvent, channel: str) -> None: - """Dispatch a single alert event to a specific notification channel. +def _default_telegram_sender(**kwargs: Any) -> bool: + from quant_platform_kit.notifications.telegram import send_telegram_message - Tries to use QuantPlatformKit notification adapters; falls back to printing. - """ - # Try importing and using the platform notification system - try: - if channel == "telegram": - from quant_platform_kit.notifications.strategy_plugin_telegram import ( - send_strategy_plugin_telegram_alert, - ) - send_strategy_plugin_telegram_alert( - subject=event.subject, - body=event.body, - alert_key=event.alert_key, - ) - elif channel == "email": - from quant_platform_kit.notifications.strategy_plugin_email import ( - send_strategy_plugin_email_alert, - ) - send_strategy_plugin_email_alert( - subject=event.subject, - body=event.body, - alert_key=event.alert_key, - ) - elif channel == "push": - from quant_platform_kit.notifications.strategy_plugin_push import ( - send_strategy_plugin_push_alert, - ) - send_strategy_plugin_push_alert( - subject=event.subject, - body=event.body, - alert_key=event.alert_key, - ) - elif channel == "webhook": - from quant_platform_kit.notifications.strategy_plugin_webhook import ( - send_strategy_plugin_webhook_alert, - ) - send_strategy_plugin_webhook_alert( - subject=event.subject, - body=event.body, - alert_key=event.alert_key, - ) - else: - # Fallback: just print - print(f"[drift_alert][{channel}] {event.subject}") - except ImportError: - print(f"[drift_alert][{channel}] {event.subject}") - - -def _action_for_status(status: DriftStatus) -> str: - if status == DriftStatus.CRITICAL: - return ( - "CRITICAL: Review immediately. Consider pausing new risk additions, " - "reducing position size, or switching to defensive allocation. " - "The strategy may need parameter re-optimization or retirement review." - ) - if status == DriftStatus.REVIEW: - return ( - "REVIEW: Schedule a manual review within the next 1-2 trading days. " - "Compare live performance against backtest on multiple dimensions. " - "Consider triggering a parameter re-optimization run." + 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: + subject = str(kwargs.get("subject") or "").strip() + body = str(kwargs.get("body") or "").strip() + text = f"{subject}\n\n{body}".strip() + if not text: + return False + return bool( + send_telegram_message( + bot_token=token, + chat_ids=chats, + text=text, + parse_mode=None, ) - return ( - "WATCH: Monitor for further deterioration. No immediate action required, " - "but track the trend over the next week. If drift persists or worsens, " - "escalate to REVIEW." ) + + +def _log(log_message: Callable[..., Any], line: str) -> None: + try: + log_message(line) + except TypeError: + log_message(line, flush=True) diff --git a/tests/test_attention.py b/tests/test_attention.py new file mode 100644 index 0000000..1a21d2c --- /dev/null +++ b/tests/test_attention.py @@ -0,0 +1,95 @@ +"""Operational attention classification tests.""" + +from __future__ import annotations + +import unittest + +from quant_platform_kit.risk.attention import ( + AttentionAxes, + AttentionLevel, + attention_transition_key, + evaluate_attention, + render_attention_compact, +) + + +class AttentionEvaluationTest(unittest.TestCase): + def test_ok_when_axes_omitted(self) -> None: + decision = evaluate_attention(AttentionAxes()) + self.assertEqual(decision.level, AttentionLevel.OK) + self.assertFalse(decision.should_notify) + self.assertIsNone(decision.dd_ratio) + + def test_leveraged_path_drawdown_below_mandate_stays_ok(self) -> None: + decision = evaluate_attention( + AttentionAxes(drawdown_from_peak=0.12, mandate_dd_budget=0.35) + ) + self.assertEqual(decision.level, AttentionLevel.OK) + self.assertFalse(decision.should_notify) + + def test_dd_without_budget_is_omitted(self) -> None: + decision = evaluate_attention(AttentionAxes(drawdown_from_peak=0.25)) + self.assertEqual(decision.level, AttentionLevel.OK) + self.assertIsNone(decision.dd_ratio) + + def test_dd_watch_and_action_bands(self) -> None: + watch = evaluate_attention( + AttentionAxes(drawdown_from_peak=0.28, mandate_dd_budget=0.35) + ) + self.assertEqual(watch.level, AttentionLevel.WATCH) + self.assertFalse(watch.should_notify) + self.assertIn("mandate_dd_elevated", watch.reason_codes) + + action = evaluate_attention( + AttentionAxes(drawdown_from_peak=0.36, mandate_dd_budget=0.35) + ) + self.assertEqual(action.level, AttentionLevel.ACTION) + self.assertTrue(action.should_notify) + self.assertIn("mandate_dd_exhausted", action.reason_codes) + + def test_new_risk_and_drift_compose_to_highest(self) -> None: + decision = evaluate_attention( + { + "new_risk_prohibited": True, + "drift_status": "critical", + } + ) + self.assertEqual(decision.level, AttentionLevel.HALT) + self.assertTrue(decision.should_notify) + self.assertIn("drift_critical", decision.reason_codes) + self.assertIn("new_risk_prohibited", decision.reason_codes) + + def test_operational_uncertain_is_halt(self) -> None: + decision = evaluate_attention(AttentionAxes(operational_uncertain=True)) + self.assertEqual(decision.level, AttentionLevel.HALT) + self.assertEqual(decision.next_step_key, "open_console_resume") + + def test_transition_key_stable_for_dedup(self) -> None: + key = attention_transition_key( + platform="ibkr", + account_alias="U159", + strategy_profile="soxl_soxx_trend_income", + level=AttentionLevel.ACTION, + primary_reason="new_risk_prohibited", + ) + self.assertEqual( + key, + "attention/ibkr/u159/soxl_soxx_trend_income/action/new_risk_prohibited", + ) + + def test_compact_zh_cn_locale_normalizes(self) -> None: + decision = evaluate_attention(AttentionAxes(new_risk_prohibited=True)) + text = render_attention_compact( + locale="zh-CN", + platform="Schwab", + account_alias="00682", + strategy_profile="soxl", + decision=decision, + ) + self.assertIn("已禁止新增风险", text) + self.assertIn("下一步", text) + self.assertIn("管理站", text) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_drift_alerts.py b/tests/test_drift_alerts.py new file mode 100644 index 0000000..b785487 --- /dev/null +++ b/tests/test_drift_alerts.py @@ -0,0 +1,104 @@ +"""Drift alert builder / publisher tests.""" + +from __future__ import annotations + +from datetime import date +import os +import unittest +from unittest import mock + +from quant_platform_kit.strategy_lifecycle.contracts import DriftResult, DriftStatus +from quant_platform_kit.strategy_lifecycle.drift_alerts import ( + build_drift_alert, + publish_drift_alerts, +) +from quant_platform_kit.strategy_lifecycle.drift_policy import DriftPolicy + + +def _policy() -> DriftPolicy: + return DriftPolicy(notification_channels=("telegram",)) + + +def _drift(status: DriftStatus, *, score: float = 0.8) -> DriftResult: + return DriftResult( + strategy_profile="soxl_soxx_trend_income", + domain="us_equity", + as_of=date(2026, 9, 18), + drift_score=score, + status=status, + escalated=status == DriftStatus.CRITICAL, + alert_suppressed=False, + dimensions={}, + ) + + +class DriftAlertTest(unittest.TestCase): + def test_watch_does_not_build_pageable_alert(self) -> None: + event = build_drift_alert(_drift(DriftStatus.WATCH), policy=_policy(), locale="zh") + self.assertIsNone(event) + + def test_review_builds_attention_compact_and_transition_key(self) -> None: + event = build_drift_alert( + _drift(DriftStatus.REVIEW), + policy=_policy(), + locale="zh-CN", + platform="ibkr", + account_alias="U159", + ) + assert event is not None + self.assertEqual(event.severity, "warning") + self.assertIn("管理站", event.body) + self.assertEqual( + event.alert_key, + "attention/ibkr/u159/soxl_soxx_trend_income/action/drift_review", + ) + + def test_publish_telegram_records_only_after_send(self) -> None: + event = build_drift_alert( + _drift(DriftStatus.CRITICAL), + policy=_policy(), + locale="en", + platform="ibkr", + ) + assert event is not None + recorded: list[str] = [] + + def _send(**_kwargs: object) -> bool: + return True + + counts = publish_drift_alerts( + [event], + telegram_sender=_send, + record_sent_key=recorded.append, + log_message=lambda *_a, **_k: None, + ) + self.assertEqual(counts.get("sent"), 1) + self.assertEqual(recorded, [event.alert_key]) + + counts2 = publish_drift_alerts( + [event], + already_sent_keys=recorded, + telegram_sender=_send, + record_sent_key=recorded.append, + log_message=lambda *_a, **_k: None, + ) + self.assertEqual(counts2.get("skipped"), 1) + self.assertEqual(recorded, [event.alert_key]) + + def test_publish_telegram_missing_config_is_skipped_not_silent_pass(self) -> None: + event = build_drift_alert(_drift(DriftStatus.CRITICAL), policy=_policy(), platform="lb") + assert event is not None + logs: list[str] = [] + cleared = { + key: value + for key, value in os.environ.items() + if "TELEGRAM" not in key and "CHAT_ID" not in key + } + with mock.patch.dict(os.environ, cleared, clear=True): + counts = publish_drift_alerts([event], log_message=logs.append) + self.assertGreaterEqual(counts.get("skipped", 0), 1) + self.assertTrue(any("telegram_not_configured" in line for line in logs)) + + +if __name__ == "__main__": + unittest.main()