Skip to content

Clarify opportunity notification copy #18

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Pigbibi merged 1 commit into from
Jun 7, 2026
Merged

Clarify opportunity notification copy #18

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
147 changes: 125 additions & 22 deletions src/quant_strategy_plugins/strategy_plugin_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -946,6 +946,17 @@ def _review_attack_symbol(*sources: Mapping[str, Any]) -> str:
return ""


def _review_benchmark_symbol(*sources: Mapping[str, Any]) -> str:
for source in sources:
for container_key in ("metrics", "rebound_confirmation"):
container = source.get(container_key)
if isinstance(container, Mapping):
symbol = str(container.get("benchmark_symbol") or "").strip().upper()
if symbol:
return symbol
return ""


def _manual_review_source_title(
*,
panic_payload: Mapping[str, Any],
Expand Down Expand Up @@ -1101,23 +1112,70 @@ def _format_manual_review_notification_message(

vetoes = _message_reason_codes(_nested_mapping(payload, "arbiter").get("vetoes"))
attack_symbol = _review_attack_symbol(panic_payload, taco_payload) or target_label
benchmark_symbol = _review_benchmark_symbol(panic_payload, taco_payload)
source_title = _manual_review_source_title(panic_payload=panic_payload, taco_payload=taco_payload, locale=locale)
if locale == "zh-CN":
benchmark_label = benchmark_symbol or "基准指数"
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 panic_payload and taco_payload:
opportunity_summary = (
f"- 事件缓和与 VIX 恐慌回落同时出现，{benchmark_label}/{attack_symbol} 已从短期低点反弹，"
"属于恐慌后反转复核信号。"
)
elif panic_payload:
opportunity_summary = (
f"- VIX 已从极端恐慌高位回落，{benchmark_label}/{attack_symbol} 已从短期低点反弹，"
"出现恐慌后反转观察信号。"
)
else:
opportunity_summary = (
f"- 事件缓和后，{benchmark_label}/{attack_symbol} 出现反弹确认，出现事件反弹复核信号。"
)
situation_lines = [opportunity_summary]
if is_vetoed_opportunity_notice:
situation_lines.append(f"- 当前仍处于{route_status}。")
situation_lines.append(f"- 但当前仍处于{route_status}，说明策略环境还没有完全解除防守。")
if vetoes:
situation_lines.append(f"- {veto_text}。")
situation_lines.append(f"- {veto_text}，所以这条通知先作为人工检查线索。")
else:
situation_lines.append(f"- 当前状态：{route_status}。")
guidance_first = (
"- 人工复核恐慌是否已缓和，以及策略侧是否已按自身风控处理。"
if is_vetoed_opportunity_notice
else "- 结合策略自身风控、持仓状态和最新基本面判断是否需要干预。"
)
situation_lines.append(f"- 当前处于{route_status}，可以进入人工复核。")
if is_vetoed_opportunity_notice:
if panic_payload and taco_payload:
confirmation_line = (
f"- 先看 VIX 是否继续回落、事件是否继续缓和、{benchmark_label}/{attack_symbol} "
"反弹是否维持，避免把一次反抽误判为趋势恢复。"
)
elif panic_payload:
confirmation_line = (
f"- 先看 VIX 是否继续回落、{benchmark_label}/{attack_symbol} 反弹是否维持，"
"避免把一次反抽误判为趋势恢复。"
)
else:
confirmation_line = (
f"- 先看事件是否继续缓和、{benchmark_label}/{attack_symbol} 反弹是否维持，"
"避免把一次反抽误判为趋势恢复。"
)
guidance_lines = [
confirmation_line,
"- 对照策略侧当前仓位和风控状态，判断是否停止继续降风险、恢复观察，或继续保持防守。",
"- 如果 VIX 回升、价格跌回短期低点附近，忽略本次机会信号。",
]
else:
if panic_payload and taco_payload:
confirmation_target = f"VIX 和事件是否继续改善，{benchmark_label}/{attack_symbol}"
reversal_target = "VIX、事件或价格确认"
elif panic_payload:
confirmation_target = f"VIX 是否继续回落，{benchmark_label}/{attack_symbol}"
reversal_target = "VIX 或价格确认"
else:
confirmation_target = f"事件是否继续缓和，{benchmark_label}/{attack_symbol}"
reversal_target = "事件或价格确认"
guidance_lines = [
f"- 检查 {confirmation_target} 的价格确认是否延续。",
"- 对照策略侧仓位和风险预算，判断是否需要人工干预。",
f"- 如果 {reversal_target}反转，忽略本次信号。",
]
lines = [
f"【{card_prefix}｜{attack_symbol}｜{source_title}】",
f"日期：{as_of or '未知日期'}",
Expand All @@ -1131,27 +1189,73 @@ def _format_manual_review_notification_message(
lines.extend(
[
"建议操作：",
guidance_first,
"- 若后续 VIX 或价格确认反转，本信号需要重新评估。",
*guidance_lines,
]
)
return "\n".join(lines)

benchmark_label = benchmark_symbol or "benchmark"
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 panic_payload and taco_payload:
opportunity_summary = (
f"- Event de-escalation and a VIX panic pullback are both present; {benchmark_label}/{attack_symbol} "
"has rebounded from short-term lows, so this is a panic-reversal review signal."
)
elif panic_payload:
opportunity_summary = (
f"- VIX has pulled back from extreme panic levels, and {benchmark_label}/{attack_symbol} "
"has rebounded from short-term lows, so this is a panic-reversal watch signal."
)
else:
opportunity_summary = (
f"- After event de-escalation, {benchmark_label}/{attack_symbol} shows rebound confirmation, "
"so this is an event-rebound review signal."
)
situation_lines = [opportunity_summary]
if is_vetoed_opportunity_notice:
situation_lines.append(f"- Current state is still {route_status}.")
situation_lines.append(f"- Current state is still {route_status}, so the strategy environment is not fully out of defense.")
if vetoes:
situation_lines.append(f"- {veto_text}.")
situation_lines.append(f"- {veto_text}; treat this notification as a manual check cue for now.")
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."
)
situation_lines.append(f"- Current state is {route_status}, so it can move into manual review.")
if is_vetoed_opportunity_notice:
if panic_payload and taco_payload:
confirmation_line = (
f"- Check whether VIX keeps falling, the event keeps de-escalating, and the "
f"{benchmark_label}/{attack_symbol} rebound holds; avoid treating a one-day bounce as recovery."
)
elif panic_payload:
confirmation_line = (
f"- Check whether VIX keeps falling and whether the {benchmark_label}/{attack_symbol} rebound holds; "
"avoid treating a one-day bounce as recovery."
)
else:
confirmation_line = (
f"- Check whether the event keeps de-escalating and whether the "
f"{benchmark_label}/{attack_symbol} rebound holds; avoid treating a one-day bounce as recovery."
)
guidance_lines = [
confirmation_line,
"- Compare against the strategy-side exposure and risk-control state before deciding whether to stop further de-risking, return to watch, or stay defensive.",
"- Ignore this opportunity signal if VIX rises again or price falls back near short-term lows.",
]
else:
if panic_payload and taco_payload:
confirmation_target = f"VIX and event conditions keep improving and whether {benchmark_label}/{attack_symbol}"
reversal_target = "VIX, event context, or price confirmation"
elif panic_payload:
confirmation_target = f"VIX keeps falling and whether {benchmark_label}/{attack_symbol}"
reversal_target = "VIX or price confirmation"
else:
confirmation_target = f"the event keeps de-escalating and whether {benchmark_label}/{attack_symbol}"
reversal_target = "event context or price confirmation"
guidance_lines = [
f"- Check whether {confirmation_target} price confirmation continues.",
"- Compare against strategy-side exposure and risk budget before any manual intervention.",
f"- Ignore this signal if {reversal_target} reverses.",
]
lines = [
f"[{card_prefix} | {attack_symbol} | {source_title}]",
f"Date: {as_of or 'unknown date'}",
Expand All @@ -1165,8 +1269,7 @@ def _format_manual_review_notification_message(
lines.extend(
[
"Suggested action:",
guidance_first,
"- Reassess this signal if VIX or price confirmation later reverses.",
*guidance_lines,
]
)
return "\n".join(lines)
Expand Down
12 changes: 8 additions & 4 deletions tests/test_strategy_plugin_runner.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -405,8 +405,10 @@ def test_strategy_plugin_runner_can_enable_panic_reversal_inside_market_regime_c
zh_notification = payload["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 "QQQ/TQQQ 已从短期低点反弹" in zh_notification
assert "建议操作：" in zh_notification
assert "检查 VIX 是否继续回落" in zh_notification
assert "VIX 曾达到恐慌区间" in zh_notification
assert "VIX 已从高点回落" in zh_notification
assert "QQQ 3 日收益" in zh_notification
Expand Down Expand Up @@ -512,9 +514,11 @@ def test_market_regime_control_notification_surfaces_vetoed_panic_reversal() ->
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 "VIX 已从极端恐慌高位回落" in zh_notification
assert "但当前仍处于降风险状态，说明策略环境还没有完全解除防守" in zh_notification
assert "宏观降风险信号优先于 VIX 恐慌反转，所以这条通知先作为人工检查线索" in zh_notification
assert "建议操作：" 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
Expand Down Expand Up @@ -829,7 +833,7 @@ def test_strategy_plugin_runner_runs_taco_rebound_notification_mount_for_tqqq(tm
zh_notification = latest["localized_messages"]["notification"]["zh-CN"]
assert "【机会复核｜TQQQ｜TACO 事件反弹】" in zh_notification
assert "情况说明：" in zh_notification
assert "- 机会信号已触发。" in zh_notification
assert "事件缓和后，QQQ/TQQQ 出现反弹确认" in zh_notification
assert "事件：" in zh_notification
assert "价格确认：" in zh_notification
assert "建议操作：" in zh_notification
Expand Down