From 9efe268835cb0eebe4863c8d76bc9f196b5e9f2a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 7 Jun 2026 04:22:34 +0800 Subject: [PATCH] Improve vetoed opportunity notifications --- docs/market-regime-control-plan.md | 14 ++ docs/market-regime-control-plan.zh-CN.md | 9 + .../market_regime_control_plugin.py | 54 +++++- .../strategy_plugin_runner.py | 159 +++++++++++++----- tests/test_market_regime_control_plugin.py | 3 + tests/test_strategy_plugin_runner.py | 127 +++++++++++++- 6 files changed, 312 insertions(+), 54 deletions(-) diff --git a/docs/market-regime-control-plan.md b/docs/market-regime-control-plan.md index 6a5e241..55a148a 100644 --- a/docs/market-regime-control-plan.md +++ b/docs/market-regime-control-plan.md @@ -134,6 +134,20 @@ allowlist: - `since_version`: records the runner schema version where the permission became effective. +Permission boundaries live in documentation and machine-readable fields, not in +the human notification body: + +- The plugin repository only writes artifacts and notifications. It does not + call broker APIs or directly mutate account allocation. +- Automated position impact happens only when the strategy side explicitly + consumes `position_control`, and only when `position_control_allowed = true` + and `evidence_status = automation_approved`. +- `notification_only`, TACO, panic reversal, AI audit, and general notification + targets are for manual review only. +- Human notification copy should contain only the situation and suggested + action; it should not display internal governance fields such as + `position_control_allowed`, `execution_controls`, route codes, or veto codes. + SOXL/SOXX is not in the strategy-level `market_regime_control` consumption registry. It receives broad market-regime context through the general `notification_targets.market_regime_notification` artifact. That notification diff --git a/docs/market-regime-control-plan.zh-CN.md b/docs/market-regime-control-plan.zh-CN.md index 7959b03..d68b9bc 100644 --- a/docs/market-regime-control-plan.zh-CN.md +++ b/docs/market-regime-control-plan.zh-CN.md @@ -86,6 +86,15 @@ - `evidence_status`：记录该策略/插件组合是 `automation_approved`、`notification_only` 还是 `deprecated_compatibility`。 - `since_version`：记录该消费权限从哪个 runner schema 开始生效。 +权限边界写在文档和机器字段里，不重复写进人工通知正文： + +- 插件仓库只生成 artifact 和通知，不调用券商接口，也不直接改账户配置。 +- 自动仓位影响只发生在策略侧显式消费 `position_control` 时，并且必须同时满足 + `position_control_allowed = true` 和 `evidence_status = automation_approved`。 +- `notification_only`、TACO、panic reversal、AI audit 和通用通知只用于人工查看。 +- 人工通知正文只写“情况说明”和“建议操作”，不展示 `position_control_allowed`、 + `execution_controls`、route code 或 veto code 等内部治理字段。 + SOXL/SOXX 不出现在 `market_regime_control` 的策略级消费 registry 中；它通过 `notification_targets.market_regime_notification` 接收通用通知。通用通知不是 strategy，不允许进入策略 runtime metadata，也不能影响仓位，避免配置误用把通知 diff --git a/src/quant_strategy_plugins/market_regime_control_plugin.py b/src/quant_strategy_plugins/market_regime_control_plugin.py index 2f58553..195c17d 100644 --- a/src/quant_strategy_plugins/market_regime_control_plugin.py +++ b/src/quant_strategy_plugins/market_regime_control_plugin.py @@ -140,6 +140,21 @@ def _reason_codes(payload: Mapping[str, Any] | None) -> tuple[str, ...]: return () +def _opportunity_summary(component: str, payload: Mapping[str, Any] | None, veto: str) -> dict[str, Any] | None: + if not isinstance(payload, Mapping): + return None + return { + "component": component, + "profile": _optional_text(payload.get("plugin") or payload.get("profile")), + "as_of": _optional_text(payload.get("as_of")), + "canonical_route": _normalized_route(payload), + "suggested_action": _normalized_action(payload), + "reason_codes": _reason_codes(payload), + "veto": veto, + "manual_review_required": _as_bool(payload.get("manual_review_required"), default=False), + } + + def _blocked(payload: Mapping[str, Any] | None) -> bool: if not isinstance(payload, Mapping): return False @@ -274,6 +289,7 @@ def build_market_regime_control_signal( crisis_defense_required = False blocked_actions: tuple[str, ...] = () vetoes: list[str] = [] + vetoed_opportunities: list[dict[str, Any]] = [] reason_codes: list[str] = [] if crisis_active: @@ -288,9 +304,17 @@ def build_market_regime_control_signal( blocked_actions = ("increase_leverage", "increase_risk", "taco_rebound_veto", "panic_reversal_veto") reason_codes.extend(f"crisis:{code}" for code in _reason_codes(crisis) or ("true_crisis",)) if taco_active: - vetoes.append("crisis_blocks_taco") + veto = "crisis_blocks_taco" + vetoes.append(veto) + summary = _opportunity_summary(COMPONENT_TACO, taco, veto) + if summary: + vetoed_opportunities.append(summary) if panic_reversal_active: - vetoes.append("crisis_blocks_panic_reversal") + veto = "crisis_blocks_panic_reversal" + vetoes.append(veto) + summary = _opportunity_summary(COMPONENT_PANIC_REVERSAL, panic_reversal, veto) + if summary: + vetoed_opportunities.append(summary) elif macro_active and macro_route == "crisis": final_route = ROUTE_RISK_OFF suggested_action = ACTION_DEFEND @@ -302,9 +326,17 @@ def build_market_regime_control_signal( blocked_actions = ("increase_leverage", "increase_risk", "taco_rebound_veto", "panic_reversal_veto") reason_codes.extend(f"macro:{code}" for code in _reason_codes(macro) or ("crisis",)) if taco_active: - vetoes.append("macro_crisis_blocks_taco") + veto = "macro_crisis_blocks_taco" + vetoes.append(veto) + summary = _opportunity_summary(COMPONENT_TACO, taco, veto) + if summary: + vetoed_opportunities.append(summary) if panic_reversal_active: - vetoes.append("macro_crisis_blocks_panic_reversal") + veto = "macro_crisis_blocks_panic_reversal" + vetoes.append(veto) + summary = _opportunity_summary(COMPONENT_PANIC_REVERSAL, panic_reversal, veto) + if summary: + vetoed_opportunities.append(summary) elif macro_active: final_route = ROUTE_RISK_REDUCED suggested_action = ACTION_DELEVER @@ -316,9 +348,17 @@ def build_market_regime_control_signal( blocked_actions = ("increase_leverage", "taco_rebound_veto", "panic_reversal_veto") reason_codes.extend(f"macro:{code}" for code in _reason_codes(macro) or ("delever",)) if taco_active: - vetoes.append("macro_delever_blocks_taco") + veto = "macro_delever_blocks_taco" + vetoes.append(veto) + summary = _opportunity_summary(COMPONENT_TACO, taco, veto) + if summary: + vetoed_opportunities.append(summary) if panic_reversal_active: - vetoes.append("macro_delever_blocks_panic_reversal") + veto = "macro_delever_blocks_panic_reversal" + vetoes.append(veto) + summary = _opportunity_summary(COMPONENT_PANIC_REVERSAL, panic_reversal, veto) + if summary: + vetoed_opportunities.append(summary) elif blocked: final_route = ROUTE_BLOCKED suggested_action = ACTION_BLOCKED @@ -355,6 +395,8 @@ def build_market_regime_control_signal( "route_source": route_source, "reason_codes": tuple(dict.fromkeys(reason_codes)), "vetoes": tuple(vetoes), + "vetoed_opportunities": tuple(vetoed_opportunities), + "opportunity_vetoed_should_notify": bool(vetoed_opportunities), } position_control = { "allowed": True, diff --git a/src/quant_strategy_plugins/strategy_plugin_runner.py b/src/quant_strategy_plugins/strategy_plugin_runner.py index 758d183..b6b61bc 100644 --- a/src/quant_strategy_plugins/strategy_plugin_runner.py +++ b/src/quant_strategy_plugins/strategy_plugin_runner.py @@ -326,6 +326,47 @@ class PluginNotificationTargetPolicy: }, } +OPPORTUNITY_REVIEW_STATUS_LABELS: dict[str, dict[str, str]] = { + "blocked": {"en-US": "blocked", "zh-CN": "阻断状态"}, + "crisis": {"en-US": "crisis state", "zh-CN": "危机状态"}, + "delever": {"en-US": "de-risking state", "zh-CN": "降风险状态"}, + "no_action": {"en-US": "normal state", "zh-CN": "正常观察状态"}, + "opportunity_watch": {"en-US": "opportunity watch", "zh-CN": "机会观察状态"}, + "panic_reversal": {"en-US": "panic-reversal review", "zh-CN": "恐慌反转复核状态"}, + "risk_off": {"en-US": "defensive state", "zh-CN": "防守状态"}, + "risk_reduced": {"en-US": "de-risking state", "zh-CN": "降风险状态"}, + "taco_rebound": {"en-US": "event-rebound review", "zh-CN": "事件反弹复核状态"}, + "true_crisis": {"en-US": "crisis state", "zh-CN": "危机状态"}, + "watch": {"en-US": "watch state", "zh-CN": "观察状态"}, +} + +OPPORTUNITY_REVIEW_VETO_LABELS: dict[str, dict[str, str]] = { + "crisis_blocks_panic_reversal": { + "en-US": "crisis defense takes priority over the VIX panic-reversal signal", + "zh-CN": "危机防守信号优先于 VIX 恐慌反转", + }, + "crisis_blocks_taco": { + "en-US": "crisis defense takes priority over the TACO rebound signal", + "zh-CN": "危机防守信号优先于 TACO 事件反弹", + }, + "macro_crisis_blocks_panic_reversal": { + "en-US": "macro crisis signal takes priority over the VIX panic-reversal signal", + "zh-CN": "宏观危机信号优先于 VIX 恐慌反转", + }, + "macro_crisis_blocks_taco": { + "en-US": "macro crisis signal takes priority over the TACO rebound signal", + "zh-CN": "宏观危机信号优先于 TACO 事件反弹", + }, + "macro_delever_blocks_panic_reversal": { + "en-US": "macro de-risking signal takes priority over the VIX panic-reversal signal", + "zh-CN": "宏观降风险信号优先于 VIX 恐慌反转", + }, + "macro_delever_blocks_taco": { + "en-US": "macro de-risking signal takes priority over the TACO rebound signal", + "zh-CN": "宏观降风险信号优先于 TACO 事件反弹", + }, +} + PluginRunner = Callable[[Mapping[str, Any], str], PluginRunResult] PluginPayloadBuilder = Callable[[pd.DataFrame, Mapping[str, Any]], dict[str, Any]] @@ -808,6 +849,14 @@ def _localized_reason_labels(reason_codes: Sequence[str], locale: str) -> tuple[ return tuple(_localized_reason_label(reason_code, locale) for reason_code in reason_codes) +def _localized_opportunity_status(route: str, locale: str) -> str: + return _localized_label(OPPORTUNITY_REVIEW_STATUS_LABELS, route, locale) + + +def _localized_opportunity_veto_labels(vetoes: Sequence[str], locale: str) -> tuple[str, ...]: + return tuple(_localized_label(OPPORTUNITY_REVIEW_VETO_LABELS, veto, locale) for veto in vetoes) + + def _payload_should_notify(payload: Mapping[str, Any], route: str) -> bool: notification = _nested_mapping(payload, "notification") if "should_notify" in notification: @@ -858,6 +907,20 @@ def _active_panic_payload(payload: Mapping[str, Any], plugin: str) -> Mapping[st return {} +def _vetoed_opportunity_components(payload: Mapping[str, Any]) -> frozenset[str]: + notification = _nested_mapping(payload, "notification") + raw = notification.get("vetoed_opportunities") + if not isinstance(raw, Sequence) or isinstance(raw, (str, bytes, bytearray)): + return frozenset() + components: set[str] = set() + for item in raw: + if isinstance(item, Mapping): + component = str(item.get("component") or "").strip().lower() + if component: + components.add(component) + return frozenset(components) + + def _active_taco_payload(payload: Mapping[str, Any], plugin: str) -> Mapping[str, Any]: if plugin == PLUGIN_TACO_REBOUND_SHADOW or "rebound_confirmation" in payload: return payload @@ -917,7 +980,6 @@ def _append_panic_review_lines(lines: list[str], panic_payload: Mapping[str, Any if locale == "zh-CN": lines.extend( [ - "触发原因：", ( "- VIX 曾达到恐慌区间：" f"{int(_as_float_or_none(thresholds.get('vix_high_lookback_days')) or 5)} 日高点 " @@ -947,7 +1009,6 @@ def _append_panic_review_lines(lines: list[str], panic_payload: Mapping[str, Any return lines.extend( [ - "Trigger evidence:", ( "- VIX reached panic territory: " f"{int(_as_float_or_none(thresholds.get('vix_high_lookback_days')) or 5)}-day high " @@ -1025,77 +1086,87 @@ def _format_manual_review_notification_message( as_of: str, route: str, ) -> str | None: - if _payload_action(payload) != "notify_manual_review": + vetoed_components = _vetoed_opportunity_components(payload) + is_vetoed_opportunity_notice = bool(vetoed_components) + if _payload_action(payload) != "notify_manual_review" and not is_vetoed_opportunity_notice: return None panic_payload = _active_panic_payload(payload, plugin) taco_payload = _active_taco_payload(payload, plugin) + if "panic_reversal" not in vetoed_components and is_vetoed_opportunity_notice: + panic_payload = {} + if "taco" not in vetoed_components and is_vetoed_opportunity_notice: + taco_payload = {} if not panic_payload and not taco_payload: return None - position_control = _nested_mapping(payload, "position_control") - vetoes = _message_reason_codes(position_control.get("vetoes") or _nested_mapping(payload, "arbiter").get("vetoes")) + vetoes = _message_reason_codes(_nested_mapping(payload, "arbiter").get("vetoes")) attack_symbol = _review_attack_symbol(panic_payload, taco_payload) or target_label source_title = _manual_review_source_title(panic_payload=panic_payload, taco_payload=taco_payload, locale=locale) if locale == "zh-CN": + route_status = _localized_opportunity_status(route, locale) + veto_text = _message_join(_localized_opportunity_veto_labels(vetoes, locale), locale) + card_prefix = "机会被拦截" if is_vetoed_opportunity_notice else "机会复核" + situation_lines = ["- 机会信号已触发。"] + if is_vetoed_opportunity_notice: + situation_lines.append(f"- 当前仍处于{route_status}。") + if vetoes: + situation_lines.append(f"- {veto_text}。") + else: + situation_lines.append(f"- 当前状态：{route_status}。") + guidance_first = ( + "- 人工复核恐慌是否已缓和，以及策略侧是否已按自身风控处理。" + if is_vetoed_opportunity_notice + else "- 结合策略自身风控、持仓状态和最新基本面判断是否需要干预。" + ) lines = [ - f"【机会复核｜{attack_symbol}｜{source_title}】", + f"【{card_prefix}｜{attack_symbol}｜{source_title}】", f"日期：{as_of or '未知日期'}", - "结论：触发人工复核，不自动加仓。", - ( - f"仲裁：{plugin} = {route}；" - f"{'crisis/macro 未 veto' if not vetoes else '存在 veto：' + _message_join(vetoes, locale)}。" - ), - "执行权限：只通知；不下单；不修改仓位。", + "情况说明：", + *situation_lines, ] - if position_control: - scalar_bits = [] - if "taco_size_scalar" in position_control: - scalar_bits.append(f"taco_size_scalar = {_format_number(position_control.get('taco_size_scalar'))}") - if "panic_reversal_size_scalar" in position_control: - scalar_bits.append( - f"panic_reversal_size_scalar = {_format_number(position_control.get('panic_reversal_size_scalar'))}" - ) - if scalar_bits: - lines.append("仓位权限：" + "；".join(scalar_bits) + "。") if panic_payload: _append_panic_review_lines(lines, panic_payload, locale=locale) if taco_payload: _append_taco_review_lines(lines, taco_payload, locale=locale) lines.extend( [ - "人工复核建议：", - "- 只评估是否停止继续降风险或恢复观察，不作为自动买入信号。", - "- 若 crisis/macro 后续转为 risk_reduced 或 risk_off，本机会信号自动失效。", + "建议操作：", + guidance_first, + "- 若后续 VIX 或价格确认反转，本信号需要重新评估。", ] ) return "\n".join(lines) + route_status = _localized_opportunity_status(route, locale) + veto_text = _message_join(_localized_opportunity_veto_labels(vetoes, locale), locale) + card_prefix = "Opportunity Vetoed" if is_vetoed_opportunity_notice else "Opportunity Review" + situation_lines = ["- Opportunity signal triggered."] + if is_vetoed_opportunity_notice: + situation_lines.append(f"- Current state is still {route_status}.") + if vetoes: + situation_lines.append(f"- {veto_text}.") + else: + situation_lines.append(f"- Current state: {route_status}.") + guidance_first = ( + "- Manually review whether panic has eased and whether the strategy-side risk controls have already handled it." + if is_vetoed_opportunity_notice + else "- Consider strategy-side risk controls, current exposure, and latest fundamentals before intervening." + ) lines = [ - f"[Opportunity Review | {attack_symbol} | {source_title}]", + f"[{card_prefix} | {attack_symbol} | {source_title}]", f"Date: {as_of or 'unknown date'}", - "Conclusion: manual review triggered; no automatic position increase.", - f"Arbiter: {plugin} = {route}; {'crisis/macro did not veto' if not vetoes else 'vetoes: ' + _message_join(vetoes, locale)}.", - "Execution: notify only; no broker orders; no allocation mutation.", + "Situation:", + *situation_lines, ] - if position_control: - scalar_bits = [] - if "taco_size_scalar" in position_control: - scalar_bits.append(f"taco_size_scalar = {_format_number(position_control.get('taco_size_scalar'))}") - if "panic_reversal_size_scalar" in position_control: - scalar_bits.append( - f"panic_reversal_size_scalar = {_format_number(position_control.get('panic_reversal_size_scalar'))}" - ) - if scalar_bits: - lines.append("Position authority: " + "; ".join(scalar_bits) + ".") if panic_payload: _append_panic_review_lines(lines, panic_payload, locale=locale) if taco_payload: _append_taco_review_lines(lines, taco_payload, locale=locale) lines.extend( [ - "Manual review guidance:", - "- Review only whether to stop further de-risking or return to watch; this is not an automatic buy signal.", - "- If crisis/macro later moves to risk_reduced or risk_off, this opportunity signal is invalidated.", + "Suggested action:", + guidance_first, + "- Reassess this signal if VIX or price confirmation later reverses.", ] ) return "\n".join(lines) @@ -1394,7 +1465,9 @@ def _build_market_regime_control_payload(price_history: pd.DataFrame, plugin_con if _as_bool(plugin_config.get("taco_enabled"), default=True): components["taco"] = _build_taco_rebound_payload(price_history, plugin_config) if _as_bool(plugin_config.get("panic_reversal_enabled"), default=False): - components["panic_reversal"] = _build_panic_reversal_payload(price_history, plugin_config) + panic_config = dict(plugin_config) + panic_config.setdefault("suppress_when_price_crisis_guard_active", False) + components["panic_reversal"] = _build_panic_reversal_payload(price_history, panic_config) return build_market_regime_control_signal( components, strategy_policy=str(plugin_config.get("strategy_policy", "levered_growth_income_v1")).strip(), diff --git a/tests/test_market_regime_control_plugin.py b/tests/test_market_regime_control_plugin.py index ca9d03b..88b763e 100644 --- a/tests/test_market_regime_control_plugin.py +++ b/tests/test_market_regime_control_plugin.py @@ -147,6 +147,9 @@ def test_market_regime_control_macro_delever_blocks_panic_reversal() -> None: assert payload["suggested_action"] == "delever" assert payload["position_control"]["panic_reversal_allowed"] is False assert "macro_delever_blocks_panic_reversal" in payload["arbiter"]["vetoes"] + assert payload["notification"]["opportunity_vetoed_should_notify"] is True + assert payload["notification"]["vetoed_opportunities"][0]["component"] == "panic_reversal" + assert payload["notification"]["vetoed_opportunities"][0]["veto"] == "macro_delever_blocks_panic_reversal" def test_market_regime_control_blocked_component_blocks_taco_opportunity() -> None: diff --git a/tests/test_strategy_plugin_runner.py b/tests/test_strategy_plugin_runner.py index 42f3786..b3deab9 100644 --- a/tests/test_strategy_plugin_runner.py +++ b/tests/test_strategy_plugin_runner.py @@ -6,7 +6,9 @@ import pandas as pd import pytest +import quant_strategy_plugins.strategy_plugin_runner as strategy_plugin_runner_module from quant_strategy_plugins.crisis_response_research import ROUTE_TRUE_CRISIS +from quant_strategy_plugins.market_regime_control_plugin import build_market_regime_control_signal from quant_strategy_plugins.strategy_plugin_runner import ( EVIDENCE_AUTOMATION_APPROVED, EVIDENCE_NOTIFICATION_ONLY, @@ -24,6 +26,7 @@ PLUGIN_TACO_REBOUND_SHADOW, STRATEGY_PLUGIN_LOG_SCHEMA_VERSION, STRATEGY_PLUGIN_MESSAGE_SCHEMA_VERSION, + _apply_plugin_contract, load_plugin_config, main, run_configured_plugins, @@ -401,11 +404,123 @@ def test_strategy_plugin_runner_can_enable_panic_reversal_inside_market_regime_c assert "panic_reversal:panic_reversal" in payload["position_control"]["reason_codes"] zh_notification = payload["notification"]["localized_messages"]["zh-CN"] assert "【机会复核｜TQQQ｜VIX 恐慌反转】" in zh_notification - assert "结论：触发人工复核，不自动加仓。" in zh_notification + assert "情况说明：" in zh_notification + assert "- 机会信号已触发。" in zh_notification + assert "建议操作：" in zh_notification assert "VIX 曾达到恐慌区间" in zh_notification assert "VIX 已从高点回落" in zh_notification assert "QQQ 3 日收益" in zh_notification - assert "panic_reversal_size_scalar = 0.00" in zh_notification + assert "执行权限" not in zh_notification + assert "仓位权限" not in zh_notification + assert "panic_reversal_size_scalar" not in zh_notification + + +def test_market_regime_control_defaults_panic_crisis_suppression_to_arbiter(monkeypatch) -> None: + captured: list[bool | None] = [] + + def fake_build_panic_payload(_price_history: pd.DataFrame, plugin_config: dict[str, object]) -> dict[str, object]: + captured.append(plugin_config.get("suppress_when_price_crisis_guard_active")) + return { + "profile": "panic_reversal_shadow", + "as_of": "2025-04-09", + "canonical_route": "panic_reversal", + "suggested_action": "notify_manual_review", + "manual_review_required": True, + "panic_reversal_context_active": True, + "reason_codes": ["panic_reversal"], + } + + monkeypatch.setattr(strategy_plugin_runner_module, "_build_panic_reversal_payload", fake_build_panic_payload) + + payload = strategy_plugin_runner_module._build_market_regime_control_payload( + pd.DataFrame(), + { + "crisis_enabled": False, + "macro_enabled": False, + "taco_enabled": False, + "panic_reversal_enabled": True, + }, + ) + explicit_payload = strategy_plugin_runner_module._build_market_regime_control_payload( + pd.DataFrame(), + { + "crisis_enabled": False, + "macro_enabled": False, + "taco_enabled": False, + "panic_reversal_enabled": True, + "suppress_when_price_crisis_guard_active": True, + }, + ) + + assert captured == [False, True] + assert payload["canonical_route"] == "opportunity_watch" + assert explicit_payload["canonical_route"] == "opportunity_watch" + + +def test_market_regime_control_notification_surfaces_vetoed_panic_reversal() -> None: + payload = build_market_regime_control_signal( + { + "macro": { + "profile": "macro_risk_governor", + "as_of": "2025-04-09", + "canonical_route": "delever", + "suggested_action": "delever", + "leverage_scalar": 0.0, + "risk_asset_scalar": 1.0, + "reason_codes": ["vix_crisis_level"], + }, + "panic_reversal": { + "profile": "panic_reversal_shadow", + "as_of": "2025-04-09", + "canonical_route": "panic_reversal", + "suggested_action": "notify_manual_review", + "manual_review_required": True, + "panic_reversal_context_active": True, + "reason_codes": ["panic_reversal", "vix_panic_reversal", "price_rebound_confirmation"], + "metrics": { + "benchmark_symbol": "QQQ", + "attack_symbol": "TQQQ", + "vix": 33.62, + "vix_previous": 52.33, + "vix_lookback_high": 52.33, + "vix_pullback_from_high": 0.3575, + "vix_vix3m_ratio": 1.1237, + "benchmark_3d_return": 0.1025, + "benchmark_rebound_from_recent_low": 0.1200, + "attack_rebound_from_recent_low": 0.3524, + }, + "reversal_confirmation": { + "confirmed": True, + "thresholds": { + "vix_high_lookback_days": 5, + "min_vix_high": 50.0, + }, + }, + }, + } + ) + contracted = _apply_plugin_contract( + payload, + strategy=STRATEGY_NAME, + plugin=PLUGIN_MARKET_REGIME_CONTROL, + mode="shadow", + ) + + assert contracted["canonical_route"] == "risk_reduced" + assert contracted["position_control"]["panic_reversal_allowed"] is False + assert contracted["notification"]["opportunity_vetoed_should_notify"] is True + zh_notification = contracted["notification"]["localized_messages"]["zh-CN"] + assert "【机会被拦截｜TQQQ｜VIX 恐慌反转】" in zh_notification + assert "情况说明：" in zh_notification + assert "- 当前仍处于降风险状态。" in zh_notification + assert "- 宏观降风险信号优先于 VIX 恐慌反转。" in zh_notification + assert "建议操作：" in zh_notification + assert "TQQQ 从近 5 日低点反弹 +35.2%" in zh_notification + assert "market_regime_control" not in zh_notification + assert "veto" not in zh_notification + assert "macro_delever_blocks_panic_reversal" not in zh_notification + assert "执行权限" not in zh_notification + assert "仓位权限" not in zh_notification def test_strategy_plugin_runner_runs_general_market_regime_notification(tmp_path) -> None: @@ -713,10 +828,11 @@ def test_strategy_plugin_runner_runs_taco_rebound_notification_mount_for_tqqq(tm assert "sleeve_suggestion" not in latest zh_notification = latest["localized_messages"]["notification"]["zh-CN"] assert "【机会复核｜TQQQ｜TACO 事件反弹】" in zh_notification - assert "结论：触发人工复核，不自动加仓。" in zh_notification + assert "情况说明：" in zh_notification + assert "- 机会信号已触发。" in zh_notification assert "事件：" in zh_notification assert "价格确认：" in zh_notification - assert "人工复核建议：" in zh_notification + assert "建议操作：" in zh_notification def test_strategy_plugin_runner_can_enable_taco_ai_audit_without_api_key(tmp_path, monkeypatch) -> None: @@ -832,8 +948,9 @@ def test_strategy_plugin_runner_runs_panic_reversal_notification_mount_for_tqqq( assert latest["localized_messages"]["labels"]["canonical_route"]["zh-CN"] == "恐慌反转" zh_notification = latest["notification"]["localized_messages"]["zh-CN"] assert "【机会复核｜TQQQ｜VIX 恐慌反转】" in zh_notification - assert "执行权限：只通知；不下单；不修改仓位。" in zh_notification assert "TQQQ 从近 5 日低点反弹" in zh_notification + assert "执行权限" not in zh_notification + assert "仓位权限" not in zh_notification def test_strategy_plugin_runner_rejects_panic_reversal_for_soxl_strategy_mount(tmp_path) -> None: