From 16fd737262ab8889e442a4bd491cb8bacfb93a77 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 6 Jun 2026 17:39:45 +0800 Subject: [PATCH] Add panic reversal shadow plugin --- docs/examples/strategy_plugins.example.toml | 38 + docs/market-regime-control-plan.md | 30 +- docs/market-regime-control-plan.zh-CN.md | 10 +- pyproject.toml | 1 + src/quant_strategy_plugins/__init__.py | 10 + .../market_regime_control_plugin.py | 61 +- .../panic_reversal_shadow_plugin.py | 740 ++++++++++++++++++ .../strategy_plugin_runner.py | 106 +++ tests/test_market_regime_control_plugin.py | 56 ++ tests/test_panic_reversal_shadow_plugin.py | 102 +++ tests/test_strategy_plugin_runner.py | 153 +++- 11 files changed, 1281 insertions(+), 26 deletions(-) create mode 100644 src/quant_strategy_plugins/panic_reversal_shadow_plugin.py create mode 100644 tests/test_panic_reversal_shadow_plugin.py diff --git a/docs/examples/strategy_plugins.example.toml b/docs/examples/strategy_plugins.example.toml index 804ce41..8132c26 100644 --- a/docs/examples/strategy_plugins.example.toml +++ b/docs/examples/strategy_plugins.example.toml @@ -37,6 +37,9 @@ taco_opportunity_size_scalar = 0.0 crisis_enabled = true macro_enabled = true taco_enabled = true +# Research-only VIX panic reversal evidence. Keep disabled by default until +# event-window and no-regression reports justify promotion beyond notification. +panic_reversal_enabled = false [strategy_plugins.outputs] output_dir = "data/output/tqqq_growth_income/plugins/market_regime_control" @@ -66,6 +69,7 @@ realized_vol_requires_confirmation = true external_stress_actionable = false delever_risk_asset_scalar = 0.0 taco_enabled = false +panic_reversal_enabled = false crisis_enabled = true macro_enabled = true @@ -123,3 +127,37 @@ start_date = "2026-01-01" [strategy_plugins.outputs] output_dir = "data/output/tqqq_growth_income/plugins/taco_rebound_shadow" + +[[strategy_plugins]] +strategy = "tqqq_growth_income" +plugin = "panic_reversal_shadow" +enabled = false + +[strategy_plugins.inputs] +prices = "data/output/panic_reversal_shadow/input/price_history.csv" +external_context = "data/output/market_regime_control/input/external_context.csv" +benchmark_symbol = "QQQ" +attack_symbol = "TQQQ" +vix_symbols = ["VIX", "^VIX", "VIXCLS"] +vix3m_symbols = ["VIX3M", "^VIX3M", "VXV", "^VXV"] +start_date = "2010-01-01" + +[strategy_plugins.outputs] +output_dir = "data/output/tqqq_growth_income/plugins/panic_reversal_shadow" + +[[notification_targets]] +notification_target = "market_regime_notification" +plugin = "panic_reversal_shadow" +enabled = false + +[notification_targets.inputs] +prices = "data/output/panic_reversal_shadow/input/soxl_price_history.csv" +external_context = "data/output/market_regime_control/input/external_context.csv" +benchmark_symbol = "SOXX" +attack_symbol = "SOXL" +vix_symbols = ["VIX", "^VIX", "VIXCLS"] +vix3m_symbols = ["VIX3M", "^VIX3M", "VXV", "^VXV"] +start_date = "2016-06-06" + +[notification_targets.outputs] +output_dir = "data/output/market_regime_notification/plugins/panic_reversal_shadow" diff --git a/docs/market-regime-control-plan.md b/docs/market-regime-control-plan.md index 9151be1..6a5e241 100644 --- a/docs/market-regime-control-plan.md +++ b/docs/market-regime-control-plan.md @@ -50,6 +50,11 @@ Non-goals: - `taco_rebound_shadow` Handles TQQQ event-rebound notification. It emits manual-review notification context and local veto evidence, but it does not raise position size. +- `panic_reversal_shadow` + Handles research-only VIX panic-reversal notification after volatility has + fallen from a panic high and price confirmation is present. It emits + manual-review context only; the sample is still too small, so it is disabled + by default and cannot raise position size. The unified artifact exposes five main sections: @@ -74,15 +79,15 @@ The unified artifact exposes five main sections: The current order is risk-first: 1. `crisis_response_shadow` `true_crisis` or bubble fragility has top priority, - emits `risk_off`, and vetoes TACO. + emits `risk_off`, and vetoes TACO and panic reversal. 2. `macro_risk_governor` crisis state is next, emits `risk_off`, and vetoes - TACO. + TACO and panic reversal. 3. `macro_risk_governor` de-leveraging emits `risk_reduced`, scales down - leverage or risk-asset budget, and vetoes TACO. + leverage or risk-asset budget, and vetoes TACO and panic reversal. 4. Data-quality kill switches or blocked component states block opportunity-side actions. -5. TACO may emit `opportunity_watch` and manual-review notification only when - there is no crisis or macro de-risking route. +5. TACO or panic reversal may emit `opportunity_watch` and manual-review + notification only when there is no crisis or macro de-risking route. 6. Watch-only signals notify but never grant position-control authority. This keeps crisis, macro, and TACO behavior separate: defense first, @@ -99,13 +104,13 @@ Recommended policy: - TQQQ growth/income strategy Consumes `position_control` by default. `risk_off` moves toward cash-like or non-risk assets; `risk_reduced` lowers leverage or risk budget based on local - strategy configuration. TACO remains manual-review notification and local - veto context only. + strategy configuration. TACO and panic reversal remain manual-review + notification and local veto context only. - SOXL/SOXX trend/income strategy Does not mount the unified plugin by default and does not consume `position_control`. SOXL keeps its own reviewed SOXX trend and volatility - de-leveraging gates. Macro, crisis, and OSINT signals are delivered as general - notifications for manual review. + de-leveraging gates. Macro, crisis, panic reversal, and OSINT signals are + delivered as general notifications for manual review. - Global ETF, Russell 1000, and Mega Cap rotation strategies Support the unified plugin by default. `risk_reduced` should apply a 50% risk @@ -267,9 +272,10 @@ Design implications: - Rotation strategies: enable 50% risk scaling for `risk_reduced` and zero risk-asset budget for `risk_off`. -- TACO: - notification-only by default; it can surface opportunity context only when no - crisis or macro de-risking route is active. +- TACO / panic reversal: + notification-only by default; they can surface opportunity context only when + no crisis or macro de-risking route is active. Panic reversal remains disabled + by default until event-window and no-regression reports justify promotion. - AI audit: no trading authority by default. It may write audit conclusions and notification evidence only. diff --git a/docs/market-regime-control-plan.zh-CN.md b/docs/market-regime-control-plan.zh-CN.md index a1da87b..7959b03 100644 --- a/docs/market-regime-control-plan.zh-CN.md +++ b/docs/market-regime-control-plan.zh-CN.md @@ -33,6 +33,8 @@ 只有显式研究开关 `external_stress_actionable` 开启后，外部压力字段才允许进入可执行分数。 - `taco_rebound_shadow` 负责 TQQQ 事件反弹通知。它输出人工复核通知和本地 veto 线索，不直接提高仓位。 +- `panic_reversal_shadow` + 负责 VIX 恐慌高位回落后的研究通知。它要求 VIX 高位回落、VIX/VIX3M 期限结构和价格反弹确认同时满足，只输出人工复核通知；样本仍不足，默认不自动加仓，也不默认开启。 统一插件输出四组主要字段： @@ -55,7 +57,7 @@ 2. `macro_risk_governor` 的 `crisis` 其次，输出 `risk_off`，并 veto TACO。 3. `macro_risk_governor` 的 `delever` 输出 `risk_reduced`，降低杠杆或风险资产预算，并 veto TACO。 4. 数据质量 kill switch 或组件 blocked 状态会阻断机会侧动作。 -5. 只有没有危机和宏观降风险时，TACO 才能输出 `opportunity_watch` 和人工复核通知。 +5. 只有没有危机和宏观降风险时，TACO 或 panic reversal 才能输出 `opportunity_watch` 和人工复核通知。 6. watch-only 信号只通知，不给仓位权限。 这个顺序保证危机插件、宏观插件和 TACO 不冲突：防守优先，机会次之，通知和执行权限分离。 @@ -67,9 +69,9 @@ 建议消费规则： - TQQQ 杠杆增长收益策略 - 默认消费 `position_control`。`risk_off` 降到现金类或非风险资产；`risk_reduced` 按策略配置降低杠杆或风险预算；TACO 只触发人工复核和本地 veto。 + 默认消费 `position_control`。`risk_off` 降到现金类或非风险资产；`risk_reduced` 按策略配置降低杠杆或风险预算；TACO 和 panic reversal 只触发人工复核和本地 veto，不触发自动加仓。 - SOXL/SOXX 趋势收益策略 - 不默认挂载统一插件，也不消费 `position_control`。SOXL 继续只使用已经通过复核的 SOXX 自身趋势和波动率降杠杆门；宏观、危机和 OSINT 信号只进入通用通知，由人工决定是否干预。 + 不默认挂载统一插件，也不消费 `position_control`。SOXL 继续只使用已经通过复核的 SOXX 自身趋势和波动率降杠杆门；宏观、危机、panic reversal 和 OSINT 信号只进入通用通知，由人工决定是否干预。 - Global ETF、Russell 1000、Mega Cap 类轮动策略 默认支持统一插件。`risk_reduced` 建议做 50% 风险预算缩放，`risk_off` 建议归零风险资产预算。 - DCA 或收入型低频策略 @@ -157,5 +159,5 @@ TQQQ 2010-2026 真实产品窗口： - 杠杆策略：默认挂载统一插件，允许 `risk_off` 生效。 - 高波动行业杠杆策略：除非回测证明自动消费能提升收益/回撤组合，否则不默认挂载统一插件；SOXL 当前只接收通用通知。 - 轮动策略：默认开启 50% risk scaling 和 `risk_off` 归零。 -- TACO：默认通知-only；只有没有危机和宏观降风险时才允许提示机会。 +- TACO / panic reversal：默认通知-only；只有没有危机和宏观降风险时才允许提示机会。panic reversal 默认保持研究开关关闭，直到事件窗口和 no-regression 报告证明可提升权限。 - AI audit：默认不参与交易权限，只能写审计结论和通知证据。 diff --git a/pyproject.toml b/pyproject.toml index 633c1e2..93e49d4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ qsp-build-crisis-response-shadow-signal = "quant_strategy_plugins.crisis_respons qsp-build-macro-external-context = "quant_strategy_plugins.macro_external_context:main" qsp-build-macro-risk-governor-signal = "quant_strategy_plugins.macro_risk_governor_plugin:main" qsp-build-taco-rebound-shadow-signal = "quant_strategy_plugins.taco_rebound_shadow_plugin:main" +qsp-build-panic-reversal-shadow-signal = "quant_strategy_plugins.panic_reversal_shadow_plugin:main" qsp-run-strategy-plugins = "quant_strategy_plugins.strategy_plugin_runner:main" [tool.setuptools] diff --git a/src/quant_strategy_plugins/__init__.py b/src/quant_strategy_plugins/__init__.py index b7c69a9..ee4762e 100644 --- a/src/quant_strategy_plugins/__init__.py +++ b/src/quant_strategy_plugins/__init__.py @@ -18,6 +18,12 @@ build_market_regime_control_signal, write_market_regime_control_outputs, ) +from .panic_reversal_shadow_plugin import ( + PANIC_REVERSAL_PROFILE, + SCHEMA_VERSION as PANIC_REVERSAL_SHADOW_SCHEMA_VERSION, + build_panic_reversal_shadow_signal, + write_panic_reversal_shadow_outputs, +) from .strategy_plugin_runner import run_configured_plugins from .taco_rebound_shadow_plugin import ( SCHEMA_VERSION as TACO_REBOUND_SHADOW_SCHEMA_VERSION, @@ -33,15 +39,19 @@ "MACRO_RISK_GOVERNOR_SCHEMA_VERSION", "MARKET_REGIME_CONTROL_PROFILE", "MARKET_REGIME_CONTROL_SCHEMA_VERSION", + "PANIC_REVERSAL_PROFILE", + "PANIC_REVERSAL_SHADOW_SCHEMA_VERSION", "TACO_REBOUND_PROFILE", "TACO_REBOUND_SHADOW_SCHEMA_VERSION", "build_crisis_response_shadow_signal", "build_macro_risk_governor_signal", "build_market_regime_control_signal", + "build_panic_reversal_shadow_signal", "build_taco_rebound_shadow_signal", "run_configured_plugins", "write_crisis_response_shadow_outputs", "write_macro_risk_governor_outputs", "write_market_regime_control_outputs", + "write_panic_reversal_shadow_outputs", "write_taco_rebound_shadow_outputs", ] diff --git a/src/quant_strategy_plugins/market_regime_control_plugin.py b/src/quant_strategy_plugins/market_regime_control_plugin.py index cfa8650..5444b5a 100644 --- a/src/quant_strategy_plugins/market_regime_control_plugin.py +++ b/src/quant_strategy_plugins/market_regime_control_plugin.py @@ -17,6 +17,7 @@ COMPONENT_CRISIS = "crisis" COMPONENT_MACRO = "macro" COMPONENT_TACO = "taco" +COMPONENT_PANIC_REVERSAL = "panic_reversal" ROUTE_NO_ACTION = "no_action" ROUTE_WATCH = "watch" @@ -37,6 +38,7 @@ MACRO_ACTIVE_ROUTES = frozenset({"delever", "crisis"}) MACRO_WATCH_ROUTES = frozenset({"watch"}) TACO_ACTIVE_ROUTES = frozenset({"taco_rebound", "taco_fake_crisis"}) +PANIC_REVERSAL_ACTIVE_ROUTES = frozenset({"panic_reversal"}) def _as_bool(value: Any, *, default: bool = False) -> bool: @@ -92,6 +94,8 @@ def _component_key(payload: Mapping[str, Any]) -> str | None: return COMPONENT_MACRO if "taco" in plugin: return COMPONENT_TACO + if "panic_reversal" in plugin: + return COMPONENT_PANIC_REVERSAL return None @@ -104,7 +108,7 @@ def _normalize_component_signals( if not isinstance(payload, Mapping): continue component = str(key or "").strip().lower() - if component in {COMPONENT_CRISIS, COMPONENT_MACRO, COMPONENT_TACO}: + if component in {COMPONENT_CRISIS, COMPONENT_MACRO, COMPONENT_TACO, COMPONENT_PANIC_REVERSAL}: normalized[component] = payload continue inferred = _component_key(payload) @@ -165,6 +169,7 @@ def _compact_signal(payload: Mapping[str, Any] | None) -> dict[str, Any]: "manual_review_required", "rebound_context_active", "event_context_active", + "panic_reversal_context_active", "price_crisis_guard_active", "watch_label", "notification_reason", @@ -198,10 +203,12 @@ def build_market_regime_control_signal( crisis = components.get(COMPONENT_CRISIS) macro = components.get(COMPONENT_MACRO) taco = components.get(COMPONENT_TACO) + panic_reversal = components.get(COMPONENT_PANIC_REVERSAL) crisis_route = _normalized_route(crisis) macro_route = _normalized_route(macro) taco_route = _normalized_route(taco) + panic_reversal_route = _normalized_route(panic_reversal) crisis_active = bool(crisis_route in CRISIS_ACTIVE_ROUTES and not _blocked(crisis)) crisis_watch = bool( crisis_route in CRISIS_WATCH_ROUTES @@ -223,6 +230,25 @@ def build_market_regime_control_signal( ) ) taco_watch = bool(isinstance(taco, Mapping) and _normalized_action(taco) == ACTION_WATCH_ONLY and not _blocked(taco)) + panic_reversal_active = bool( + panic_reversal_route in PANIC_REVERSAL_ACTIVE_ROUTES + and not _blocked(panic_reversal) + and ( + _as_bool( + panic_reversal.get("manual_review_required") if isinstance(panic_reversal, Mapping) else None, + default=False, + ) + or _as_bool( + panic_reversal.get("panic_reversal_context_active") if isinstance(panic_reversal, Mapping) else None, + default=False, + ) + ) + ) + panic_reversal_watch = bool( + isinstance(panic_reversal, Mapping) + and _normalized_action(panic_reversal) == ACTION_WATCH_ONLY + and not _blocked(panic_reversal) + ) blocked = any(_blocked(payload) for payload in components.values()) final_route = ROUTE_NO_ACTION @@ -233,6 +259,7 @@ def build_market_regime_control_signal( leverage_scalar = 1.0 risk_asset_scalar = 1.0 taco_allowed = False + panic_reversal_allowed = False local_delever_veto_allowed = False crisis_defense_required = False blocked_actions: tuple[str, ...] = () @@ -248,10 +275,12 @@ def build_market_regime_control_signal( leverage_scalar = 0.0 risk_asset_scalar = 0.0 crisis_defense_required = True - blocked_actions = ("increase_leverage", "increase_risk", "taco_rebound_veto") + 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") + if panic_reversal_active: + vetoes.append("crisis_blocks_panic_reversal") elif macro_active and macro_route == "crisis": final_route = ROUTE_RISK_OFF suggested_action = ACTION_DEFEND @@ -260,10 +289,12 @@ def build_market_regime_control_signal( leverage_scalar = _clamp_ratio(macro.get("leverage_scalar") if isinstance(macro, Mapping) else None, default=0.0) risk_asset_scalar = _clamp_ratio(macro.get("risk_asset_scalar") if isinstance(macro, Mapping) else None, default=0.0) risk_budget_scalar = risk_asset_scalar - blocked_actions = ("increase_leverage", "increase_risk", "taco_rebound_veto") + 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") + if panic_reversal_active: + vetoes.append("macro_crisis_blocks_panic_reversal") elif macro_active: final_route = ROUTE_RISK_REDUCED suggested_action = ACTION_DELEVER @@ -272,29 +303,38 @@ def build_market_regime_control_signal( leverage_scalar = _clamp_ratio(macro.get("leverage_scalar") if isinstance(macro, Mapping) else None, default=0.0) risk_asset_scalar = _clamp_ratio(macro.get("risk_asset_scalar") if isinstance(macro, Mapping) else None, default=1.0) risk_budget_scalar = risk_asset_scalar - blocked_actions = ("increase_leverage", "taco_rebound_veto") + 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") + if panic_reversal_active: + vetoes.append("macro_delever_blocks_panic_reversal") elif blocked: final_route = ROUTE_BLOCKED suggested_action = ACTION_BLOCKED route_source = "data_quality" reason_codes.extend(f"{key}:blocked" for key, payload in components.items() if _blocked(payload)) - elif taco_active: + elif taco_active or panic_reversal_active: final_route = ROUTE_OPPORTUNITY_WATCH suggested_action = ACTION_NOTIFY_MANUAL_REVIEW - route_source = COMPONENT_TACO - taco_allowed = True + route_source = COMPONENT_TACO if taco_active else COMPONENT_PANIC_REVERSAL + taco_allowed = bool(taco_active) + panic_reversal_allowed = bool(panic_reversal_active) local_delever_veto_allowed = True - reason_codes.extend(f"taco:{code}" for code in _reason_codes(taco) or ("taco_rebound",)) - elif macro_watch or crisis_watch or taco_watch: + if taco_active: + reason_codes.extend(f"taco:{code}" for code in _reason_codes(taco) or ("taco_rebound",)) + if panic_reversal_active: + reason_codes.extend( + f"panic_reversal:{code}" for code in _reason_codes(panic_reversal) or ("panic_reversal",) + ) + elif macro_watch or crisis_watch or taco_watch or panic_reversal_watch: final_route = ROUTE_WATCH suggested_action = ACTION_WATCH_ONLY route_source = "watch" reason_codes.extend(f"macro:{code}" for code in _reason_codes(macro)) reason_codes.extend(f"crisis:{code}" for code in _reason_codes(crisis)) reason_codes.extend(f"taco:{code}" for code in _reason_codes(taco)) + reason_codes.extend(f"panic_reversal:{code}" for code in _reason_codes(panic_reversal)) notification = { "allowed": True, @@ -317,6 +357,8 @@ def build_market_regime_control_signal( "risk_asset_scalar": _clamp_ratio(risk_asset_scalar, default=1.0), "taco_allowed": taco_allowed, "taco_size_scalar": _clamp_ratio(taco_opportunity_size_scalar, default=0.0) if taco_allowed else 0.0, + "panic_reversal_allowed": panic_reversal_allowed, + "panic_reversal_size_scalar": 0.0, "local_delever_veto_allowed": local_delever_veto_allowed, "crisis_defense_required": crisis_defense_required, "blocked_actions": blocked_actions, @@ -348,6 +390,7 @@ def build_market_regime_control_signal( COMPONENT_CRISIS: _compact_signal(crisis), COMPONENT_MACRO: _compact_signal(macro), COMPONENT_TACO: _compact_signal(taco), + COMPONENT_PANIC_REVERSAL: _compact_signal(panic_reversal), }, "execution_controls": { "capital_impact": "strategy_opt_in", diff --git a/src/quant_strategy_plugins/panic_reversal_shadow_plugin.py b/src/quant_strategy_plugins/panic_reversal_shadow_plugin.py new file mode 100644 index 0000000..41b8619 --- /dev/null +++ b/src/quant_strategy_plugins/panic_reversal_shadow_plugin.py @@ -0,0 +1,740 @@ +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Mapping, Sequence + +import pandas as pd + +from .artifacts import write_json +from .plugin_signal_utils import flatten_for_csv, json_scalar, normalize_close, resolve_signal_date +from .russell_1000_multi_factor_defensive_snapshot import read_table +from .taco_panic_rebound_overlay_compare import ( + DEFAULT_ATTACK_SYMBOL, + DEFAULT_BENCHMARK_SYMBOL, + DEFAULT_PRICE_CRISIS_GUARD_DRAWDOWN, + DEFAULT_PRICE_CRISIS_GUARD_MA_DAYS, + DEFAULT_PRICE_CRISIS_GUARD_MA_SLOPE_DAYS, + build_price_crisis_guard_signal, +) +from .yfinance_prices import download_price_history + +SCHEMA_VERSION = "panic_reversal_shadow.v1" +SHADOW_MODE = "shadow" +PANIC_REVERSAL_PROFILE = "panic_reversal_shadow" +ROUTE_PANIC_REVERSAL = "panic_reversal" +ROUTE_WATCH = "watch" +ROUTE_NO_ACTION = "no_action" +ACTION_NOTIFY_MANUAL_REVIEW = "notify_manual_review" +ACTION_WATCH_ONLY = "watch_only" +ACTION_NO_ACTION = "no_action" + +DEFAULT_OUTPUT_DIR = "data/output/panic_reversal_shadow" +DEFAULT_START_DATE = "2010-01-01" +DEFAULT_MAX_PRICE_AGE_DAYS = 4 +DEFAULT_MAX_VOL_AGE_DAYS = 4 +DEFAULT_VIX_SYMBOLS = ("VIX", "^VIX", "VIXCLS") +DEFAULT_VIX3M_SYMBOLS = ("VIX3M", "^VIX3M", "VXV", "^VXV", "VXVCLS") +DEFAULT_VIX_HIGH_LOOKBACK_DAYS = 5 +DEFAULT_MIN_VIX_HIGH = 50.0 +DEFAULT_MIN_VIX_PULLBACK_FROM_HIGH = 0.10 +DEFAULT_REQUIRE_VIX_TERM_STRUCTURE = True +DEFAULT_MIN_VIX_VIX3M_RATIO = 1.0 +DEFAULT_PRICE_CONFIRMATION_LOOKBACK_DAYS = 5 +DEFAULT_MIN_BENCHMARK_REBOUND_FROM_LOW = 0.015 +DEFAULT_MIN_ATTACK_REBOUND_FROM_LOW = 0.04 +DEFAULT_MIN_BENCHMARK_3D_RETURN = 0.0 +DEFAULT_EVENT_STUDY_HORIZONS = (21, 42, 63) + + +def _as_str_tuple(raw: str | Sequence[str] | None) -> tuple[str, ...]: + if raw is None: + return () + values = raw.split(",") if isinstance(raw, str) else list(raw) + return tuple(str(value).strip().upper() for value in values if str(value).strip()) + + +def _normalize_external_context(external_context: pd.DataFrame | None) -> pd.DataFrame: + if external_context is None: + return pd.DataFrame() + frame = pd.DataFrame(external_context).copy() + if frame.empty: + return pd.DataFrame() + if "as_of" in frame.columns: + frame["as_of"] = pd.to_datetime(frame["as_of"], errors="coerce").dt.tz_localize(None).dt.normalize() + frame = frame.dropna(subset=["as_of"]).drop_duplicates("as_of", keep="last").set_index("as_of") + else: + frame.index = pd.to_datetime(frame.index, errors="coerce").tz_localize(None).normalize() + frame = frame.loc[frame.index.notna()] + frame.columns = frame.columns.astype(str).str.lower().str.strip() + return frame.sort_index() + + +def _metric_series( + close: pd.DataFrame, + external_context: pd.DataFrame, + *, + price_symbols: Sequence[str], + external_columns: Sequence[str], +) -> tuple[pd.Series, str]: + for symbol in _as_str_tuple(price_symbols): + if symbol in close.columns: + values = pd.to_numeric(close[symbol], errors="coerce").dropna() + values.name = symbol + return values, f"price:{symbol}" + for column in tuple(str(item).strip().lower() for item in external_columns if str(item).strip()): + if column in external_context.columns: + values = pd.to_numeric(external_context[column], errors="coerce").dropna() + values.name = column + return values, f"external_context:{column}" + return pd.Series(dtype=float), "" + + +def _value_on_or_before(series: pd.Series, date: pd.Timestamp) -> tuple[float | None, pd.Timestamp | None, int | None]: + clean = pd.to_numeric(series, errors="coerce").dropna().sort_index() + clean.index = pd.to_datetime(clean.index, errors="coerce").tz_localize(None).normalize() + clean = clean.loc[clean.index.notna()] + candidates = clean.loc[clean.index <= date] + if candidates.empty: + return None, None, None + value_date = pd.Timestamp(candidates.index[-1]).normalize() + return float(candidates.iloc[-1]), value_date, int((date - value_date).days) + + +def _last_observations(series: pd.Series, date: pd.Timestamp, count: int) -> pd.Series: + clean = pd.to_numeric(series, errors="coerce").dropna().sort_index() + clean.index = pd.to_datetime(clean.index, errors="coerce").tz_localize(None).normalize() + clean = clean.loc[clean.index.notna()] + return clean.loc[clean.index <= date].tail(max(1, int(count))) + + +def _recent_price_confirmation( + close: pd.DataFrame, + *, + signal_date: pd.Timestamp, + benchmark_symbol: str, + attack_symbol: str, + lookback_days: int, +) -> dict[str, float | int | str | None]: + index = pd.DatetimeIndex(close.index).sort_values() + if signal_date not in index: + return {"reason": "signal date missing from price index"} + signal_pos = int(index.get_loc(signal_date)) + lookback_start = max(0, signal_pos - max(1, int(lookback_days)) + 1) + window_index = index[lookback_start : signal_pos + 1] + benchmark = pd.to_numeric(close[benchmark_symbol].reindex(window_index), errors="coerce") + attack = pd.to_numeric(close[attack_symbol].reindex(window_index), errors="coerce") + + benchmark_close = float(benchmark.iloc[-1]) if benchmark.notna().any() else float("nan") + attack_close = float(attack.iloc[-1]) if attack.notna().any() else float("nan") + benchmark_low = float(benchmark.min()) if benchmark.notna().any() else float("nan") + attack_low = float(attack.min()) if attack.notna().any() else float("nan") + benchmark_rebound = benchmark_close / benchmark_low - 1.0 if benchmark_low > 0 else float("nan") + attack_rebound = attack_close / attack_low - 1.0 if attack_low > 0 else float("nan") + if signal_pos >= 3: + benchmark_3d_base = float(close[benchmark_symbol].iloc[signal_pos - 3]) + benchmark_3d_return = benchmark_close / benchmark_3d_base - 1.0 if benchmark_3d_base > 0 else float("nan") + else: + benchmark_3d_return = float("nan") + return { + "lookback_days": int(lookback_days), + "benchmark_symbol": benchmark_symbol, + "attack_symbol": attack_symbol, + "benchmark_rebound_from_recent_low": benchmark_rebound, + "attack_rebound_from_recent_low": attack_rebound, + "benchmark_3d_return": benchmark_3d_return, + "benchmark_recent_low": benchmark_low, + "attack_recent_low": attack_low, + } + + +def _score_checks(checks: Mapping[str, bool]) -> float: + if not checks: + return 0.0 + return round(sum(1.0 for value in checks.values() if bool(value)) / float(len(checks)), 4) + + +def _reason_text_for_failed_checks(checks: Mapping[str, bool], labels: Mapping[str, str]) -> str: + failed = [labels.get(key, key) for key, value in checks.items() if not bool(value)] + return "; ".join(failed) + + +def build_panic_reversal_shadow_signal( + price_history, + *, + external_context: pd.DataFrame | None = None, + as_of: str | None = None, + start_date: str = DEFAULT_START_DATE, + end_date: str | None = None, + benchmark_symbol: str = DEFAULT_BENCHMARK_SYMBOL, + attack_symbol: str = DEFAULT_ATTACK_SYMBOL, + vix_symbols: Sequence[str] = DEFAULT_VIX_SYMBOLS, + vix3m_symbols: Sequence[str] = DEFAULT_VIX3M_SYMBOLS, + max_price_age_days: int = DEFAULT_MAX_PRICE_AGE_DAYS, + max_vol_age_days: int = DEFAULT_MAX_VOL_AGE_DAYS, + vix_high_lookback_days: int = DEFAULT_VIX_HIGH_LOOKBACK_DAYS, + min_vix_high: float = DEFAULT_MIN_VIX_HIGH, + min_vix_pullback_from_high: float = DEFAULT_MIN_VIX_PULLBACK_FROM_HIGH, + require_vix_term_structure: bool = DEFAULT_REQUIRE_VIX_TERM_STRUCTURE, + min_vix_vix3m_ratio: float = DEFAULT_MIN_VIX_VIX3M_RATIO, + confirmation_lookback_days: int = DEFAULT_PRICE_CONFIRMATION_LOOKBACK_DAYS, + min_benchmark_rebound_from_low: float = DEFAULT_MIN_BENCHMARK_REBOUND_FROM_LOW, + min_attack_rebound_from_low: float = DEFAULT_MIN_ATTACK_REBOUND_FROM_LOW, + min_benchmark_3d_return: float = DEFAULT_MIN_BENCHMARK_3D_RETURN, + suppress_when_price_crisis_guard_active: bool = True, + crisis_guard_drawdown: float = DEFAULT_PRICE_CRISIS_GUARD_DRAWDOWN, + crisis_guard_ma_days: int = DEFAULT_PRICE_CRISIS_GUARD_MA_DAYS, + crisis_guard_ma_slope_days: int = DEFAULT_PRICE_CRISIS_GUARD_MA_SLOPE_DAYS, +) -> dict[str, Any]: + close = normalize_close(price_history) + if end_date is not None: + close = close.loc[close.index <= pd.Timestamp(end_date).tz_localize(None).normalize()].copy() + if start_date is not None: + close = close.loc[close.index >= pd.Timestamp(start_date).tz_localize(None).normalize()].copy() + requested_date, signal_date = resolve_signal_date(close, as_of) + signal_iso = signal_date.date().isoformat() + latest_price_date = pd.Timestamp(close.index.max()).normalize() + price_age_days = int((requested_date - signal_date).days) + + benchmark_symbol = str(benchmark_symbol).strip().upper() + attack_symbol = str(attack_symbol).strip().upper() + ext = _normalize_external_context(external_context) + vix, vix_source = _metric_series(close, ext, price_symbols=vix_symbols, external_columns=("vix", "vixcls")) + vix3m, vix3m_source = _metric_series( + close, + ext, + price_symbols=vix3m_symbols, + external_columns=("vix3m", "vxv", "vxvcls"), + ) + + kill_reasons: list[str] = [] + if benchmark_symbol not in close.columns: + kill_reasons.append(f"missing benchmark price data: {benchmark_symbol}") + if attack_symbol not in close.columns: + kill_reasons.append(f"missing attack price data: {attack_symbol}") + if price_age_days > int(max_price_age_days): + kill_reasons.append( + f"price data stale: signal_as_of={signal_iso}, requested_as_of={requested_date.date().isoformat()}" + ) + + vix_value, vix_date, vix_age_days = _value_on_or_before(vix, signal_date) + vix3m_value, vix3m_date, vix3m_age_days = _value_on_or_before(vix3m, signal_date) + if vix_value is None: + kill_reasons.append("missing VIX data") + elif vix_age_days is not None and vix_age_days > int(max_vol_age_days): + kill_reasons.append(f"VIX data stale: vix_as_of={vix_date.date().isoformat() if vix_date is not None else ''}") + if bool(require_vix_term_structure): + if vix3m_value is None: + kill_reasons.append("missing VIX3M data") + elif vix3m_age_days is not None and vix3m_age_days > int(max_vol_age_days): + kill_reasons.append( + f"VIX3M data stale: vix3m_as_of={vix3m_date.date().isoformat() if vix3m_date is not None else ''}" + ) + + vix_window = _last_observations(vix, signal_date, int(vix_high_lookback_days)) + vix_previous = float(vix_window.iloc[-2]) if len(vix_window) >= 2 else float("nan") + vix_high = float(vix_window.max()) if not vix_window.empty else float("nan") + vix_pullback_from_high = 1.0 - float(vix_value) / vix_high if vix_value is not None and vix_high > 0 else float("nan") + vix_vix3m_ratio = ( + float(vix_value) / float(vix3m_value) + if vix_value is not None and vix3m_value is not None and float(vix3m_value) > 0 + else float("nan") + ) + price_confirmation = ( + _recent_price_confirmation( + close, + signal_date=signal_date, + benchmark_symbol=benchmark_symbol, + attack_symbol=attack_symbol, + lookback_days=int(confirmation_lookback_days), + ) + if benchmark_symbol in close.columns and attack_symbol in close.columns + else {} + ) + crisis_guard_active = False + if not kill_reasons and bool(suppress_when_price_crisis_guard_active): + crisis_guard = build_price_crisis_guard_signal( + close, + start_date=start_date, + end_date=signal_iso, + benchmark_symbol=benchmark_symbol, + drawdown_threshold=float(crisis_guard_drawdown), + ma_days=int(crisis_guard_ma_days), + ma_slope_days=int(crisis_guard_ma_slope_days), + ) + crisis_guard_active = bool( + crisis_guard.reindex(crisis_guard.index.union(pd.DatetimeIndex([signal_date]))).sort_index().ffill().loc[signal_date] + ) + + checks = { + "price_data_usable": not bool(kill_reasons), + "vix_data_usable": vix_value is not None and (vix_age_days is None or vix_age_days <= int(max_vol_age_days)), + "vix_high_panic_level": pd.notna(vix_high) and vix_high >= float(min_vix_high), + "vix_reversed_from_high": pd.notna(vix_pullback_from_high) + and vix_pullback_from_high >= float(min_vix_pullback_from_high), + "vix_falling": vix_value is not None and pd.notna(vix_previous) and float(vix_value) < float(vix_previous), + "vix_term_structure_confirmed": (not bool(require_vix_term_structure)) + or (pd.notna(vix_vix3m_ratio) and vix_vix3m_ratio >= float(min_vix_vix3m_ratio)), + "benchmark_3d_return_positive": pd.notna(price_confirmation.get("benchmark_3d_return")) + and float(price_confirmation["benchmark_3d_return"]) > float(min_benchmark_3d_return), + "benchmark_rebound_from_low": pd.notna(price_confirmation.get("benchmark_rebound_from_recent_low")) + and float(price_confirmation["benchmark_rebound_from_recent_low"]) >= float(min_benchmark_rebound_from_low), + "attack_rebound_from_low": pd.notna(price_confirmation.get("attack_rebound_from_recent_low")) + and float(price_confirmation["attack_rebound_from_recent_low"]) >= float(min_attack_rebound_from_low), + "price_crisis_guard_clear": not bool(crisis_guard_active), + } + hard_check_labels = { + "price_data_usable": "required price/volatility data unavailable or stale", + "vix_data_usable": "VIX data unavailable or stale", + "vix_high_panic_level": "VIX has not reached panic threshold", + "vix_reversed_from_high": "VIX pullback from panic high below threshold", + "vix_falling": "VIX has not fallen versus previous observation", + "vix_term_structure_confirmed": "VIX/VIX3M term structure confirmation missing", + "benchmark_3d_return_positive": "benchmark 3d return below threshold", + "benchmark_rebound_from_low": "benchmark rebound from recent low below threshold", + "attack_rebound_from_low": "attack rebound from recent low below threshold", + "price_crisis_guard_clear": "price crisis guard active", + } + confirmed = all(bool(value) for value in checks.values()) + vix_reversal_watch = bool(checks["vix_high_panic_level"] and checks["vix_reversed_from_high"]) + canonical_route = ROUTE_PANIC_REVERSAL if confirmed else ROUTE_NO_ACTION + suggested_action = ACTION_NOTIFY_MANUAL_REVIEW if confirmed else ACTION_NO_ACTION + manual_review_required = bool(confirmed) + notification_reason = "" + suppression_reason = "" + if confirmed: + notification_reason = "panic volatility reversal with price confirmation" + elif kill_reasons or crisis_guard_active or vix_reversal_watch: + canonical_route = ROUTE_WATCH + suggested_action = ACTION_WATCH_ONLY + suppression_reason = "; ".join(kill_reasons) or _reason_text_for_failed_checks(checks, hard_check_labels) + + reason_codes: list[str] = [] + if confirmed: + reason_codes.extend(("panic_reversal", "vix_panic_reversal", "price_rebound_confirmation")) + elif canonical_route == ROUTE_WATCH: + reason_codes.append("panic_reversal_watch") + if crisis_guard_active: + reason_codes.append("price_crisis_guard_active") + + payload = { + "as_of": signal_iso, + "mode": SHADOW_MODE, + "schema_version": SCHEMA_VERSION, + "profile": PANIC_REVERSAL_PROFILE, + "canonical_route": canonical_route, + "suggested_action": suggested_action, + "manual_review_required": manual_review_required, + "notification_reason": notification_reason, + "suppression_reason": suppression_reason, + "panic_reversal_context_active": bool(confirmed), + "would_trade_if_enabled": False, + "reason_codes": tuple(dict.fromkeys(reason_codes)), + "reversal_confirmation": { + "confirmed": bool(confirmed), + "reason": "" if confirmed else suppression_reason, + "checks": checks, + "thresholds": { + "vix_high_lookback_days": int(vix_high_lookback_days), + "min_vix_high": float(min_vix_high), + "min_vix_pullback_from_high": float(min_vix_pullback_from_high), + "require_vix_term_structure": bool(require_vix_term_structure), + "min_vix_vix3m_ratio": float(min_vix_vix3m_ratio), + "confirmation_lookback_days": int(confirmation_lookback_days), + "min_benchmark_rebound_from_low": float(min_benchmark_rebound_from_low), + "min_attack_rebound_from_low": float(min_attack_rebound_from_low), + "min_benchmark_3d_return": float(min_benchmark_3d_return), + }, + }, + "panic_reversal_quality": { + "schema_version": "panic_reversal_quality.v1", + "quality_score": _score_checks(checks), + "checks": checks, + "warnings": list(kill_reasons) + + ([] if confirmed else [value for value in _reason_text_for_failed_checks(checks, hard_check_labels).split("; ") if value]), + "promotion_status": "shadow_only_insufficient_sample", + }, + "metrics": { + "vix": vix_value, + "vix_source": vix_source, + "vix_previous": vix_previous, + "vix_lookback_high": vix_high, + "vix_pullback_from_high": vix_pullback_from_high, + "vix3m": vix3m_value, + "vix3m_source": vix3m_source, + "vix_vix3m_ratio": vix_vix3m_ratio, + **price_confirmation, + }, + "price_crisis_guard_active": crisis_guard_active, + "data_freshness": { + "requested_as_of": requested_date.date().isoformat(), + "signal_as_of": signal_iso, + "prices_as_of": latest_price_date.date().isoformat(), + "price_age_days": price_age_days, + "max_price_age_days": int(max_price_age_days), + "vix_as_of": vix_date.date().isoformat() if vix_date is not None else None, + "vix_age_days": vix_age_days, + "vix3m_as_of": vix3m_date.date().isoformat() if vix3m_date is not None else None, + "vix3m_age_days": vix3m_age_days, + "max_vol_age_days": int(max_vol_age_days), + }, + "notification": { + "allowed": True, + "profile": "manual_review_only" if confirmed else "shadow_only", + "should_notify": bool(confirmed or canonical_route == ROUTE_WATCH), + "route": canonical_route, + "suggested_action": suggested_action, + "reason_codes": tuple(dict.fromkeys(reason_codes)), + }, + "execution_controls": { + "capital_impact": "none", + "broker_order_allowed": False, + "live_allocation_mutation_allowed": False, + "log_namespace": PANIC_REVERSAL_PROFILE, + "notification_profile": "manual_review_only" if confirmed else "shadow_only", + "intended_strategy_role": "panic_reversal_notification", + "selection_allowed": False, + "position_sizing_allowed": False, + "allocation_recommendation_allowed": False, + "strategy_runtime_metadata_allowed": True, + "position_control_shadow_only": True, + }, + "audit_summary": { + "route_source": PANIC_REVERSAL_PROFILE, + "final_route": canonical_route, + "suggested_action": suggested_action, + "reason_codes": tuple(dict.fromkeys(reason_codes)), + "note": "Research-only volatility panic reversal evidence; it cannot increase live allocation.", + }, + "generated_at": datetime.now(timezone.utc).isoformat(), + } + return json_scalar(payload) + + +def _bounded_ffill_to_index(series: pd.Series, index: pd.DatetimeIndex, *, max_age_days: int) -> pd.Series: + clean = pd.to_numeric(series, errors="coerce").dropna().sort_index() + clean.index = pd.to_datetime(clean.index, errors="coerce").tz_localize(None).normalize() + clean = clean.loc[clean.index.notna()] + aligned = clean.reindex(index).ffill() + valid_dates = pd.Series(clean.index, index=clean.index).reindex(index).ffill() + ages = (pd.Series(index, index=index) - valid_dates).dt.days + return aligned.where(ages <= int(max_age_days)) + + +def scan_panic_reversal_signals( + price_history, + *, + external_context: pd.DataFrame | None = None, + start_date: str = DEFAULT_START_DATE, + end_date: str | None = None, + benchmark_symbol: str = DEFAULT_BENCHMARK_SYMBOL, + attack_symbol: str = DEFAULT_ATTACK_SYMBOL, + vix_symbols: Sequence[str] = DEFAULT_VIX_SYMBOLS, + vix3m_symbols: Sequence[str] = DEFAULT_VIX3M_SYMBOLS, + max_vol_age_days: int = DEFAULT_MAX_VOL_AGE_DAYS, + vix_high_lookback_days: int = DEFAULT_VIX_HIGH_LOOKBACK_DAYS, + min_vix_high: float = DEFAULT_MIN_VIX_HIGH, + min_vix_pullback_from_high: float = DEFAULT_MIN_VIX_PULLBACK_FROM_HIGH, + require_vix_term_structure: bool = DEFAULT_REQUIRE_VIX_TERM_STRUCTURE, + min_vix_vix3m_ratio: float = DEFAULT_MIN_VIX_VIX3M_RATIO, + confirmation_lookback_days: int = DEFAULT_PRICE_CONFIRMATION_LOOKBACK_DAYS, + min_benchmark_rebound_from_low: float = DEFAULT_MIN_BENCHMARK_REBOUND_FROM_LOW, + min_attack_rebound_from_low: float = DEFAULT_MIN_ATTACK_REBOUND_FROM_LOW, + min_benchmark_3d_return: float = DEFAULT_MIN_BENCHMARK_3D_RETURN, + suppress_when_price_crisis_guard_active: bool = True, + crisis_guard_drawdown: float = DEFAULT_PRICE_CRISIS_GUARD_DRAWDOWN, + min_gap_trading_days: int = 21, +) -> pd.DataFrame: + close = normalize_close(price_history) + if start_date is not None: + close = close.loc[close.index >= pd.Timestamp(start_date).normalize()].copy() + if end_date is not None: + close = close.loc[close.index <= pd.Timestamp(end_date).normalize()].copy() + benchmark_symbol = str(benchmark_symbol).strip().upper() + attack_symbol = str(attack_symbol).strip().upper() + if benchmark_symbol not in close.columns: + raise ValueError(f"benchmark symbol {benchmark_symbol!r} missing from price history") + if attack_symbol not in close.columns: + raise ValueError(f"attack symbol {attack_symbol!r} missing from price history") + + ext = _normalize_external_context(external_context) + raw_vix, vix_source = _metric_series(close, ext, price_symbols=vix_symbols, external_columns=("vix", "vixcls")) + raw_vix3m, vix3m_source = _metric_series( + close, + ext, + price_symbols=vix3m_symbols, + external_columns=("vix3m", "vxv", "vxvcls"), + ) + if raw_vix.empty: + raise ValueError("VIX data is required for panic reversal scan") + index = pd.DatetimeIndex(close.index).sort_values() + vix = _bounded_ffill_to_index(raw_vix, index, max_age_days=int(max_vol_age_days)) + vix3m = ( + _bounded_ffill_to_index(raw_vix3m, index, max_age_days=int(max_vol_age_days)) + if not raw_vix3m.empty + else pd.Series(float("nan"), index=index) + ) + benchmark = pd.to_numeric(close[benchmark_symbol], errors="coerce") + attack = pd.to_numeric(close[attack_symbol], errors="coerce") + vix_high = vix.rolling(int(vix_high_lookback_days), min_periods=int(vix_high_lookback_days)).max() + vix_pullback = 1.0 - vix / vix_high + vix_ratio = vix / vix3m.where(vix3m > 0) + benchmark_rebound = benchmark / benchmark.rolling(int(confirmation_lookback_days), min_periods=1).min() - 1.0 + attack_rebound = attack / attack.rolling(int(confirmation_lookback_days), min_periods=1).min() - 1.0 + crisis_guard = ( + build_price_crisis_guard_signal( + close, + start_date=start_date, + end_date=end_date, + benchmark_symbol=benchmark_symbol, + drawdown_threshold=float(crisis_guard_drawdown), + ).reindex(index).ffill().fillna(False) + if bool(suppress_when_price_crisis_guard_active) + else pd.Series(False, index=index) + ) + vix_term_structure_ok = ( + vix_ratio.ge(float(min_vix_vix3m_ratio)) + if bool(require_vix_term_structure) + else pd.Series(True, index=index) + ) + signal = ( + vix_high.ge(float(min_vix_high)) + & vix_pullback.ge(float(min_vix_pullback_from_high)) + & vix.lt(vix.shift(1)) + & vix_term_structure_ok + & benchmark.pct_change(3).gt(float(min_benchmark_3d_return)) + & benchmark_rebound.ge(float(min_benchmark_rebound_from_low)) + & attack_rebound.ge(float(min_attack_rebound_from_low)) + & ~crisis_guard.astype(bool) + ).fillna(False) + + rows: list[dict[str, object]] = [] + last_signal_pos = -10_000 + for pos, date in enumerate(index): + if not bool(signal.loc[date]): + continue + if pos - last_signal_pos < int(min_gap_trading_days): + continue + rows.append( + { + "signal_date": date.date().isoformat(), + "benchmark_symbol": benchmark_symbol, + "attack_symbol": attack_symbol, + "vix": float(vix.loc[date]), + "vix_source": vix_source, + "vix3m": float(vix3m.loc[date]) if pd.notna(vix3m.loc[date]) else float("nan"), + "vix3m_source": vix3m_source, + "vix_lookback_high": float(vix_high.loc[date]), + "vix_pullback_from_high": float(vix_pullback.loc[date]), + "vix_vix3m_ratio": float(vix_ratio.loc[date]) if pd.notna(vix_ratio.loc[date]) else float("nan"), + "benchmark_3d_return": float(benchmark.pct_change(3).loc[date]), + "benchmark_rebound_from_recent_low": float(benchmark_rebound.loc[date]), + "attack_rebound_from_recent_low": float(attack_rebound.loc[date]), + "price_crisis_guard_active": bool(crisis_guard.loc[date]), + } + ) + last_signal_pos = pos + return pd.DataFrame(rows) + + +def run_panic_reversal_event_study( + price_history, + *, + external_context: pd.DataFrame | None = None, + start_date: str = DEFAULT_START_DATE, + end_date: str | None = None, + benchmark_symbol: str = DEFAULT_BENCHMARK_SYMBOL, + attack_symbol: str = DEFAULT_ATTACK_SYMBOL, + horizons: Sequence[int] = DEFAULT_EVENT_STUDY_HORIZONS, + entry_lag_trading_days: int = 1, + min_gap_trading_days: int = 21, + **scan_kwargs: Any, +) -> dict[str, pd.DataFrame]: + close = normalize_close(price_history) + if start_date is not None: + close = close.loc[close.index >= pd.Timestamp(start_date).normalize()].copy() + if end_date is not None: + close = close.loc[close.index <= pd.Timestamp(end_date).normalize()].copy() + index = pd.DatetimeIndex(close.index).sort_values() + signals = scan_panic_reversal_signals( + close, + external_context=external_context, + start_date=start_date, + end_date=end_date, + benchmark_symbol=benchmark_symbol, + attack_symbol=attack_symbol, + min_gap_trading_days=min_gap_trading_days, + **scan_kwargs, + ) + rows: list[dict[str, object]] = [] + benchmark_symbol = str(benchmark_symbol).strip().upper() + attack_symbol = str(attack_symbol).strip().upper() + for signal in signals.itertuples(index=False): + signal_date = pd.Timestamp(signal.signal_date).normalize() + if signal_date not in index: + continue + signal_pos = int(index.get_loc(signal_date)) + entry_pos = signal_pos + max(0, int(entry_lag_trading_days)) + if entry_pos >= len(index): + continue + entry_date = pd.Timestamp(index[entry_pos]).normalize() + for horizon in tuple(int(value) for value in horizons): + exit_pos = min(len(index) - 1, entry_pos + max(0, horizon)) + exit_date = pd.Timestamp(index[exit_pos]).normalize() + for symbol in (benchmark_symbol, attack_symbol): + entry_close = float(close.at[entry_date, symbol]) + exit_close = float(close.at[exit_date, symbol]) + if not pd.notna(entry_close) or not pd.notna(exit_close) or entry_close <= 0: + continue + path = pd.to_numeric(close[symbol].loc[(close.index >= entry_date) & (close.index <= exit_date)], errors="coerce") + drawdown = path / path.cummax() - 1.0 + rows.append( + { + "signal_date": signal.signal_date, + "entry_date": entry_date.date().isoformat(), + "exit_date": exit_date.date().isoformat(), + "symbol": symbol, + "horizon_days": horizon, + "entry_close": entry_close, + "exit_close": exit_close, + "return": exit_close / entry_close - 1.0 if entry_close > 0 else float("nan"), + "max_drawdown_after_entry": float(drawdown.min()) if not drawdown.empty else float("nan"), + } + ) + event_windows = pd.DataFrame(rows) + if event_windows.empty: + summary = pd.DataFrame( + columns=["symbol", "horizon_days", "trades", "avg_return", "median_return", "win_rate", "avg_max_drawdown"] + ) + else: + summary = ( + event_windows.groupby(["symbol", "horizon_days"], as_index=False) + .agg( + trades=("return", "count"), + avg_return=("return", "mean"), + median_return=("return", "median"), + win_rate=("return", lambda values: float((pd.Series(values) > 0.0).mean())), + avg_max_drawdown=("max_drawdown_after_entry", "mean"), + ) + .sort_values(["symbol", "horizon_days"]) + .reset_index(drop=True) + ) + return {"signals": signals, "event_windows": event_windows, "summary": summary} + + +def write_panic_reversal_shadow_outputs(payload: Mapping[str, Any], output_dir: str | Path) -> dict[str, Path]: + output_root = Path(output_dir) + signal_date = str(payload["as_of"]) + signal_dir = output_root / "signals" + audit_dir = output_root / "audit" + latest_path = output_root / "latest_signal.json" + dated_json_path = signal_dir / f"{signal_date}.json" + dated_csv_path = signal_dir / f"{signal_date}.csv" + evidence_csv_path = audit_dir / f"{signal_date}_evidence.csv" + + write_json(latest_path, payload) + write_json(dated_json_path, payload) + signal_dir.mkdir(parents=True, exist_ok=True) + audit_dir.mkdir(parents=True, exist_ok=True) + pd.DataFrame([flatten_for_csv(payload)]).to_csv(dated_csv_path, index=False) + evidence_payload = { + "as_of": payload.get("as_of"), + "canonical_route": payload.get("canonical_route"), + "suggested_action": payload.get("suggested_action"), + "manual_review_required": payload.get("manual_review_required"), + "notification_reason": payload.get("notification_reason"), + "suppression_reason": payload.get("suppression_reason"), + **flatten_for_csv(payload.get("reversal_confirmation", {})), + **flatten_for_csv(payload.get("panic_reversal_quality", {})), + **flatten_for_csv(payload.get("metrics", {})), + **flatten_for_csv(payload.get("data_freshness", {})), + } + pd.DataFrame([evidence_payload]).to_csv(evidence_csv_path, index=False) + return { + "latest_signal": latest_path, + "signal_json": dated_json_path, + "signal_csv": dated_csv_path, + "evidence_csv": evidence_csv_path, + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Build the research-only panic reversal shadow signal.") + input_group = parser.add_mutually_exclusive_group(required=True) + input_group.add_argument("--prices", help="Existing long price-history CSV with symbol/as_of/close columns") + input_group.add_argument("--download", action="store_true", help="Download adjusted price history through yfinance") + parser.add_argument("--external-context", default=None, help="Optional external_context CSV with vix/vix3m columns") + parser.add_argument("--as-of", default=None) + parser.add_argument("--price-start", default=DEFAULT_START_DATE) + parser.add_argument("--price-end", default=None) + parser.add_argument("--download-proxy", default=None, help="Optional yfinance proxy URL; YFINANCE_PROXY also works") + parser.add_argument("--start", dest="start_date", default=DEFAULT_START_DATE) + parser.add_argument("--end", dest="end_date", default=None) + parser.add_argument("--benchmark-symbol", default=DEFAULT_BENCHMARK_SYMBOL) + parser.add_argument("--attack-symbol", default=DEFAULT_ATTACK_SYMBOL) + parser.add_argument("--vix-symbols", default=",".join(DEFAULT_VIX_SYMBOLS)) + parser.add_argument("--vix3m-symbols", default=",".join(DEFAULT_VIX3M_SYMBOLS)) + parser.add_argument("--min-vix-high", type=float, default=DEFAULT_MIN_VIX_HIGH) + parser.add_argument("--min-vix-pullback-from-high", type=float, default=DEFAULT_MIN_VIX_PULLBACK_FROM_HIGH) + parser.add_argument("--min-vix-vix3m-ratio", type=float, default=DEFAULT_MIN_VIX_VIX3M_RATIO) + parser.add_argument("--disable-vix-term-structure", action="store_true") + parser.add_argument("--min-benchmark-rebound-from-low", type=float, default=DEFAULT_MIN_BENCHMARK_REBOUND_FROM_LOW) + parser.add_argument("--min-attack-rebound-from-low", type=float, default=DEFAULT_MIN_ATTACK_REBOUND_FROM_LOW) + parser.add_argument("--min-benchmark-3d-return", type=float, default=DEFAULT_MIN_BENCHMARK_3D_RETURN) + parser.add_argument("--output-dir", default=DEFAULT_OUTPUT_DIR) + return parser + + +def main(argv: Sequence[str] | None = None) -> int: + args = build_parser().parse_args(argv) + if args.download: + symbols = [args.benchmark_symbol, args.attack_symbol, *_as_str_tuple(args.vix_symbols), *_as_str_tuple(args.vix3m_symbols)] + symbols = list(dict.fromkeys(symbol for symbol in symbols if symbol and not symbol.startswith("^"))) + input_dir = Path(args.output_dir) / "input" + input_dir.mkdir(parents=True, exist_ok=True) + prices_path = input_dir / "panic_reversal_shadow_price_history.csv" + prices = download_price_history(symbols, start=args.price_start, end=args.price_end, proxy=args.download_proxy) + prices.to_csv(prices_path, index=False) + price_history = prices + else: + price_history = read_table(args.prices) + external_context = read_table(args.external_context) if args.external_context else None + payload = build_panic_reversal_shadow_signal( + price_history, + external_context=external_context, + as_of=args.as_of, + start_date=args.start_date, + end_date=args.end_date, + benchmark_symbol=args.benchmark_symbol, + attack_symbol=args.attack_symbol, + vix_symbols=_as_str_tuple(args.vix_symbols), + vix3m_symbols=_as_str_tuple(args.vix3m_symbols), + min_vix_high=float(args.min_vix_high), + min_vix_pullback_from_high=float(args.min_vix_pullback_from_high), + require_vix_term_structure=not bool(args.disable_vix_term_structure), + min_vix_vix3m_ratio=float(args.min_vix_vix3m_ratio), + min_benchmark_rebound_from_low=float(args.min_benchmark_rebound_from_low), + min_attack_rebound_from_low=float(args.min_attack_rebound_from_low), + min_benchmark_3d_return=float(args.min_benchmark_3d_return), + ) + paths = write_panic_reversal_shadow_outputs(payload, args.output_dir) + print( + "wrote panic reversal shadow signal " + f"as_of={payload['as_of']} route={payload['canonical_route']} " + f"action={payload['suggested_action']} latest={paths['latest_signal']}" + ) + return 0 + + +__all__ = [ + "ACTION_NOTIFY_MANUAL_REVIEW", + "PANIC_REVERSAL_PROFILE", + "ROUTE_PANIC_REVERSAL", + "SCHEMA_VERSION", + "build_panic_reversal_shadow_signal", + "main", + "run_panic_reversal_event_study", + "scan_panic_reversal_signals", + "write_panic_reversal_shadow_outputs", +] diff --git a/src/quant_strategy_plugins/strategy_plugin_runner.py b/src/quant_strategy_plugins/strategy_plugin_runner.py index 0ec9fce..136bf52 100644 --- a/src/quant_strategy_plugins/strategy_plugin_runner.py +++ b/src/quant_strategy_plugins/strategy_plugin_runner.py @@ -24,6 +24,11 @@ build_market_regime_control_signal, write_market_regime_control_outputs, ) +from .panic_reversal_shadow_plugin import ( + PANIC_REVERSAL_PROFILE, + build_panic_reversal_shadow_signal, + write_panic_reversal_shadow_outputs, +) from .russell_1000_multi_factor_defensive_snapshot import read_table from .taco_panic_rebound_research import DEFAULT_EVENT_SET, resolve_trade_war_event_set from .taco_rebound_shadow_plugin import ( @@ -37,6 +42,7 @@ PLUGIN_CRISIS_RESPONSE_SHADOW = "crisis_response_shadow" PLUGIN_MARKET_REGIME_CONTROL = MARKET_REGIME_CONTROL_PROFILE PLUGIN_MACRO_RISK_GOVERNOR = MACRO_RISK_GOVERNOR_PROFILE +PLUGIN_PANIC_REVERSAL_SHADOW = PANIC_REVERSAL_PROFILE PLUGIN_TACO_REBOUND_SHADOW = TACO_REBOUND_PROFILE SUPPORTED_PLUGIN_MODES = (SHADOW_MODE,) STRATEGY_PLUGIN_MESSAGE_SCHEMA_VERSION = "strategy_plugin_messages.v1" @@ -50,6 +56,7 @@ PLUGIN_CRISIS_RESPONSE_SHADOW: ("crisis_response_shadow.v1",), PLUGIN_MARKET_REGIME_CONTROL: ("market_regime_control.v1",), PLUGIN_MACRO_RISK_GOVERNOR: ("macro_risk_governor.v1",), + PLUGIN_PANIC_REVERSAL_SHADOW: ("panic_reversal_shadow.v1",), PLUGIN_TACO_REBOUND_SHADOW: ("taco_rebound_shadow.v2",), } PLUGIN_DEPRECATED_SUCCESSORS: dict[str, str] = { @@ -163,6 +170,15 @@ class PluginNotificationTargetPolicy: since_version="strategy_plugins.v1", description="Manual-review event rebound notifier for TQQQ only.", ), + PluginConsumptionPolicy( + plugin=PLUGIN_PANIC_REVERSAL_SHADOW, + strategy="tqqq_growth_income", + notification_allowed=True, + position_control_allowed=False, + evidence_status=EVIDENCE_NOTIFICATION_ONLY, + since_version="strategy_plugins.v1", + description="Research-only VIX panic reversal notifier for TQQQ manual review.", + ), ) PLUGIN_CONSUMPTION_POLICY_REGISTRY: dict[tuple[str, str], PluginConsumptionPolicy] = { (policy.plugin, policy.strategy): policy for policy in PLUGIN_CONSUMPTION_POLICIES @@ -178,6 +194,16 @@ class PluginNotificationTargetPolicy: description="General market-regime notice. Not mounted into an automated strategy runtime.", notification_role="general_market_regime_notification", ), + PluginNotificationTargetPolicy( + plugin=PLUGIN_PANIC_REVERSAL_SHADOW, + notification_target=GENERAL_MARKET_REGIME_NOTIFICATION_TARGET, + notification_allowed=True, + position_control_allowed=False, + evidence_status=EVIDENCE_NOTIFICATION_ONLY, + since_version="strategy_plugins.v1", + description="General research-only panic reversal notice. Not mounted into an automated strategy runtime.", + notification_role="panic_reversal_notification", + ), ) PLUGIN_NOTIFICATION_TARGET_POLICY_REGISTRY: dict[tuple[str, str], PluginNotificationTargetPolicy] = { (policy.plugin, policy.notification_target): policy for policy in PLUGIN_NOTIFICATION_TARGET_POLICIES @@ -205,6 +231,7 @@ class PluginNotificationTargetPolicy: "delever": {"en-US": "De-lever", "zh-CN": "降杠杆"}, "no_action": {"en-US": "No action", "zh-CN": "无动作"}, "opportunity_watch": {"en-US": "Opportunity watch", "zh-CN": "机会观察"}, + "panic_reversal": {"en-US": "Panic reversal", "zh-CN": "恐慌反转"}, "risk_off": {"en-US": "Risk off", "zh-CN": "风险关闭"}, "risk_reduced": {"en-US": "Risk reduced", "zh-CN": "风险降低"}, "taco_rebound": {"en-US": "TACO rebound", "zh-CN": "TACO 反弹"}, @@ -223,6 +250,7 @@ class PluginNotificationTargetPolicy: "crisis": {"en-US": "Crisis", "zh-CN": "危机"}, "data_quality": {"en-US": "Data quality", "zh-CN": "数据质量"}, "macro": {"en-US": "Macro", "zh-CN": "宏观"}, + "panic_reversal": {"en-US": "Panic reversal", "zh-CN": "恐慌反转"}, "taco": {"en-US": "TACO", "zh-CN": "TACO"}, } LOCALIZED_REASON_LABELS: dict[str, dict[str, str]] = { @@ -274,12 +302,17 @@ class PluginNotificationTargetPolicy: "zh-CN": "新高新低差观察", }, "pentagon_pizza_watch": {"en-US": "Pentagon pizza index watch", "zh-CN": "五角大楼比萨指数观察"}, + "panic_reversal": {"en-US": "Panic reversal context", "zh-CN": "恐慌反转上下文"}, + "panic_reversal_watch": {"en-US": "Panic reversal watch", "zh-CN": "恐慌反转观察"}, "put_call_stress_watch": {"en-US": "Put/call stress watch", "zh-CN": "Put/call 压力观察"}, + "price_crisis_guard_active": {"en-US": "Price crisis guard active", "zh-CN": "价格危机保护激活"}, + "price_rebound_confirmation": {"en-US": "Price rebound confirmation", "zh-CN": "价格反弹确认"}, "safe_haven_demand_watch": {"en-US": "Safe-haven demand watch", "zh-CN": "避险需求观察"}, "skew_high_watch": {"en-US": "SKEW high watch", "zh-CN": "SKEW 偏高观察"}, "taco_rebound": {"en-US": "TACO rebound context", "zh-CN": "TACO 反弹上下文"}, "true_crisis": {"en-US": "True crisis", "zh-CN": "真实危机"}, "vix_crisis_level": {"en-US": "VIX crisis level", "zh-CN": "VIX 危机水平"}, + "vix_panic_reversal": {"en-US": "VIX panic reversal", "zh-CN": "VIX 恐慌回落"}, "vix_spike": {"en-US": "VIX spike", "zh-CN": "VIX 尖峰"}, "vix_term_structure_inverted_watch": { "en-US": "VIX term-structure inversion watch", @@ -552,6 +585,55 @@ def _build_taco_rebound_kwargs(plugin_config: Mapping[str, Any]) -> dict[str, An return kwargs +def _build_panic_reversal_kwargs(plugin_config: Mapping[str, Any]) -> dict[str, Any]: + kwargs: dict[str, Any] = {} + string_keys = { + "as_of", + "start_date", + "end_date", + "benchmark_symbol", + "attack_symbol", + } + numeric_keys = { + "min_vix_high", + "min_vix_pullback_from_high", + "min_vix_vix3m_ratio", + "min_benchmark_rebound_from_low", + "min_attack_rebound_from_low", + "min_benchmark_3d_return", + "crisis_guard_drawdown", + } + integer_keys = { + "max_price_age_days", + "max_vol_age_days", + "vix_high_lookback_days", + "confirmation_lookback_days", + "crisis_guard_ma_days", + "crisis_guard_ma_slope_days", + } + bool_keys = { + "require_vix_term_structure", + "suppress_when_price_crisis_guard_active", + } + for key in string_keys: + if key in plugin_config and plugin_config[key] is not None: + kwargs[key] = str(plugin_config[key]).strip() + for key in numeric_keys: + if key in plugin_config and plugin_config[key] is not None: + kwargs[key] = float(plugin_config[key]) + for key in integer_keys: + if key in plugin_config and plugin_config[key] is not None: + kwargs[key] = int(plugin_config[key]) + for key in bool_keys: + if key in plugin_config and plugin_config[key] is not None: + kwargs[key] = _as_bool(plugin_config[key]) + if "vix_symbols" in plugin_config: + kwargs["vix_symbols"] = _as_str_tuple(plugin_config["vix_symbols"]) + if "vix3m_symbols" in plugin_config: + kwargs["vix3m_symbols"] = _as_str_tuple(plugin_config["vix3m_symbols"]) + return kwargs + + def _build_macro_risk_governor_kwargs(plugin_config: Mapping[str, Any]) -> dict[str, Any]: kwargs: dict[str, Any] = {} string_keys = { @@ -993,6 +1075,15 @@ def _build_taco_rebound_payload(price_history: pd.DataFrame, plugin_config: Mapp ) +def _build_panic_reversal_payload(price_history: pd.DataFrame, plugin_config: Mapping[str, Any]) -> dict[str, Any]: + external_context = _optional_table(plugin_config.get("external_context")) + return build_panic_reversal_shadow_signal( + price_history, + external_context=external_context, + **_build_panic_reversal_kwargs(plugin_config), + ) + + def _build_macro_risk_governor_payload(price_history: pd.DataFrame, plugin_config: Mapping[str, Any]) -> dict[str, Any]: external_context = _optional_table(plugin_config.get("external_context")) return build_macro_risk_governor_signal( @@ -1010,6 +1101,8 @@ def _build_market_regime_control_payload(price_history: pd.DataFrame, plugin_con components["macro"] = _build_macro_risk_governor_payload(price_history, plugin_config) 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) return build_market_regime_control_signal( components, strategy_policy=str(plugin_config.get("strategy_policy", "levered_growth_income_v1")).strip(), @@ -1132,6 +1225,11 @@ def _run_table_notification_target_plugin( build_payload=_build_taco_rebound_payload, write_outputs=write_taco_rebound_shadow_outputs, ) +PANIC_REVERSAL_SHADOW_SPEC = PluginExecutionSpec( + default_plugin=PLUGIN_PANIC_REVERSAL_SHADOW, + build_payload=_build_panic_reversal_payload, + write_outputs=write_panic_reversal_shadow_outputs, +) MACRO_RISK_GOVERNOR_SPEC = PluginExecutionSpec( default_plugin=PLUGIN_MACRO_RISK_GOVERNOR, build_payload=_build_macro_risk_governor_payload, @@ -1152,6 +1250,10 @@ def run_taco_rebound_shadow_plugin(plugin_config: Mapping[str, Any], default_mod return _run_table_strategy_plugin(plugin_config, default_mode, TACO_REBOUND_SHADOW_SPEC) +def run_panic_reversal_shadow_plugin(plugin_config: Mapping[str, Any], default_mode: str) -> PluginRunResult: + return _run_table_strategy_plugin(plugin_config, default_mode, PANIC_REVERSAL_SHADOW_SPEC) + + def run_macro_risk_governor_plugin(plugin_config: Mapping[str, Any], default_mode: str) -> PluginRunResult: return _run_table_strategy_plugin(plugin_config, default_mode, MACRO_RISK_GOVERNOR_SPEC) @@ -1164,12 +1266,14 @@ def run_market_regime_control_plugin(plugin_config: Mapping[str, Any], default_m PLUGIN_CRISIS_RESPONSE_SHADOW: run_crisis_response_shadow_plugin, PLUGIN_MARKET_REGIME_CONTROL: run_market_regime_control_plugin, PLUGIN_MACRO_RISK_GOVERNOR: run_macro_risk_governor_plugin, + PLUGIN_PANIC_REVERSAL_SHADOW: run_panic_reversal_shadow_plugin, PLUGIN_TACO_REBOUND_SHADOW: run_taco_rebound_shadow_plugin, } PLUGIN_SPECS: dict[str, PluginExecutionSpec] = { PLUGIN_CRISIS_RESPONSE_SHADOW: CRISIS_RESPONSE_SHADOW_SPEC, PLUGIN_MARKET_REGIME_CONTROL: MARKET_REGIME_CONTROL_SPEC, PLUGIN_MACRO_RISK_GOVERNOR: MACRO_RISK_GOVERNOR_SPEC, + PLUGIN_PANIC_REVERSAL_SHADOW: PANIC_REVERSAL_SHADOW_SPEC, PLUGIN_TACO_REBOUND_SHADOW: TACO_REBOUND_SHADOW_SPEC, } @@ -1321,6 +1425,7 @@ def main(argv: list[str] | None = None) -> int: "PLUGIN_CRISIS_RESPONSE_SHADOW", "PLUGIN_MARKET_REGIME_CONTROL", "PLUGIN_MACRO_RISK_GOVERNOR", + "PLUGIN_PANIC_REVERSAL_SHADOW", "PLUGIN_TACO_REBOUND_SHADOW", "PLUGIN_COMPATIBLE_STRATEGIES", "PLUGIN_COMPATIBLE_NOTIFICATION_TARGETS", @@ -1342,5 +1447,6 @@ def main(argv: list[str] | None = None) -> int: "run_crisis_response_shadow_plugin", "run_market_regime_control_plugin", "run_macro_risk_governor_plugin", + "run_panic_reversal_shadow_plugin", "run_taco_rebound_shadow_plugin", ] diff --git a/tests/test_market_regime_control_plugin.py b/tests/test_market_regime_control_plugin.py index 332a8c0..ca9d03b 100644 --- a/tests/test_market_regime_control_plugin.py +++ b/tests/test_market_regime_control_plugin.py @@ -89,10 +89,66 @@ def test_market_regime_control_taco_is_notification_with_local_veto_only() -> No assert payload["position_control"]["taco_allowed"] is True assert payload["position_control"]["local_delever_veto_allowed"] is True assert payload["position_control"]["taco_size_scalar"] == 0.25 + assert payload["position_control"]["panic_reversal_allowed"] is False assert payload["execution_controls"]["broker_order_allowed"] is False assert payload["execution_controls"]["live_allocation_mutation_allowed"] is False +def test_market_regime_control_panic_reversal_is_opportunity_watch_only() -> None: + payload = build_market_regime_control_signal( + { + "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, + "would_trade_if_enabled": False, + }, + } + ) + + assert payload["canonical_route"] == "opportunity_watch" + assert payload["suggested_action"] == "notify_manual_review" + assert payload["would_trade_if_enabled"] is False + assert payload["position_control"]["taco_allowed"] is False + assert payload["position_control"]["panic_reversal_allowed"] is True + assert payload["position_control"]["panic_reversal_size_scalar"] == 0.0 + assert payload["position_control"]["local_delever_veto_allowed"] is True + assert payload["execution_controls"]["broker_order_allowed"] is False + assert payload["execution_controls"]["live_allocation_mutation_allowed"] is False + + +def test_market_regime_control_macro_delever_blocks_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, + }, + } + ) + + assert payload["canonical_route"] == "risk_reduced" + assert payload["suggested_action"] == "delever" + assert payload["position_control"]["panic_reversal_allowed"] is False + assert "macro_delever_blocks_panic_reversal" in payload["arbiter"]["vetoes"] + + def test_market_regime_control_blocked_component_blocks_taco_opportunity() -> None: payload = build_market_regime_control_signal( { diff --git a/tests/test_panic_reversal_shadow_plugin.py b/tests/test_panic_reversal_shadow_plugin.py new file mode 100644 index 0000000..0eca601 --- /dev/null +++ b/tests/test_panic_reversal_shadow_plugin.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import json + +import pandas as pd + +from quant_strategy_plugins.panic_reversal_shadow_plugin import ( + ACTION_NOTIFY_MANUAL_REVIEW, + ROUTE_PANIC_REVERSAL, + build_panic_reversal_shadow_signal, + run_panic_reversal_event_study, + write_panic_reversal_shadow_outputs, +) + + +def _panic_reversal_prices(*, confirmed: bool = True) -> pd.DataFrame: + dates = pd.bdate_range("2025-04-01", periods=12) + qqq_path = [100.0, 96.0, 92.0, 88.0, 84.0, 82.0, 84.0, 87.0, 90.0, 92.0, 94.0, 96.0] + if confirmed: + tqqq_path = [100.0, 88.0, 78.0, 70.0, 64.0, 60.0, 66.0, 74.0, 82.0, 88.0, 94.0, 100.0] + else: + tqqq_path = [100.0, 88.0, 78.0, 70.0, 64.0, 60.0, 61.0, 62.0, 62.0, 63.0, 64.0, 65.0] + vix_path = [20.0, 32.0, 45.0, 54.0, 60.0, 58.0, 55.0, 48.0, 45.0, 42.0, 39.0, 35.0] + vix3m_path = [22.0, 28.0, 38.0, 45.0, 48.0, 47.0, 45.0, 41.0, 39.0, 37.0, 35.0, 33.0] + rows: list[dict[str, object]] = [] + for idx, as_of in enumerate(dates): + rows.append({"symbol": "QQQ", "as_of": as_of, "close": qqq_path[idx], "volume": 1_000_000}) + rows.append({"symbol": "TQQQ", "as_of": as_of, "close": tqqq_path[idx], "volume": 1_000_000}) + rows.append({"symbol": "VIX", "as_of": as_of, "close": vix_path[idx], "volume": 0}) + rows.append({"symbol": "VIX3M", "as_of": as_of, "close": vix3m_path[idx], "volume": 0}) + return pd.DataFrame(rows) + + +def test_panic_reversal_shadow_routes_confirmed_vix_reversal_to_manual_review() -> None: + payload = build_panic_reversal_shadow_signal( + _panic_reversal_prices(), + as_of="2025-04-10", + start_date="2025-04-01", + ) + + assert payload["canonical_route"] == ROUTE_PANIC_REVERSAL + assert payload["suggested_action"] == ACTION_NOTIFY_MANUAL_REVIEW + assert payload["manual_review_required"] is True + assert payload["panic_reversal_context_active"] is True + assert payload["would_trade_if_enabled"] is False + assert "panic_reversal" in payload["reason_codes"] + assert payload["reversal_confirmation"]["confirmed"] is True + assert payload["panic_reversal_quality"]["checks"]["vix_reversed_from_high"] is True + assert payload["panic_reversal_quality"]["checks"]["benchmark_rebound_from_low"] is True + assert payload["panic_reversal_quality"]["checks"]["attack_rebound_from_low"] is True + assert payload["execution_controls"]["broker_order_allowed"] is False + assert payload["execution_controls"]["live_allocation_mutation_allowed"] is False + assert payload["execution_controls"]["allocation_recommendation_allowed"] is False + + +def test_panic_reversal_shadow_waits_for_attack_price_confirmation() -> None: + payload = build_panic_reversal_shadow_signal( + _panic_reversal_prices(confirmed=False), + as_of="2025-04-10", + start_date="2025-04-01", + ) + + assert payload["canonical_route"] == "watch" + assert payload["suggested_action"] == "watch_only" + assert payload["manual_review_required"] is False + assert payload["panic_reversal_context_active"] is False + assert payload["would_trade_if_enabled"] is False + assert payload["reversal_confirmation"]["confirmed"] is False + assert payload["panic_reversal_quality"]["checks"]["attack_rebound_from_low"] is False + assert "attack rebound from recent low below threshold" in payload["suppression_reason"] + + +def test_panic_reversal_shadow_writes_artifacts(tmp_path) -> None: + payload = build_panic_reversal_shadow_signal( + _panic_reversal_prices(), + as_of="2025-04-10", + start_date="2025-04-01", + ) + + paths = write_panic_reversal_shadow_outputs(payload, tmp_path) + + assert paths["latest_signal"].exists() + assert paths["signal_json"].exists() + assert paths["signal_csv"].exists() + assert paths["evidence_csv"].exists() + latest = json.loads(paths["latest_signal"].read_text(encoding="utf-8")) + assert latest["canonical_route"] == ROUTE_PANIC_REVERSAL + assert latest["execution_controls"]["broker_order_allowed"] is False + + +def test_panic_reversal_event_study_reports_signal_windows() -> None: + result = run_panic_reversal_event_study( + _panic_reversal_prices(), + start_date="2025-04-01", + horizons=(1, 3), + entry_lag_trading_days=0, + min_gap_trading_days=5, + ) + + assert not result["signals"].empty + assert set(result["event_windows"]["symbol"]) == {"QQQ", "TQQQ"} + assert set(result["summary"]["horizon_days"]) == {1, 3} diff --git a/tests/test_strategy_plugin_runner.py b/tests/test_strategy_plugin_runner.py index c37a0f7..6fe47d1 100644 --- a/tests/test_strategy_plugin_runner.py +++ b/tests/test_strategy_plugin_runner.py @@ -19,6 +19,7 @@ PLUGIN_MARKET_REGIME_CONTROL, PLUGIN_NOTIFICATION_TARGET_POLICY_REGISTRY, PLUGIN_MACRO_RISK_GOVERNOR, + PLUGIN_PANIC_REVERSAL_SHADOW, PLUGIN_SCHEMA_VERSIONS, PLUGIN_TACO_REBOUND_SHADOW, STRATEGY_PLUGIN_LOG_SCHEMA_VERSION, @@ -142,6 +143,21 @@ def _taco_rebound_prices() -> pd.DataFrame: return pd.DataFrame(rows) +def _panic_reversal_prices() -> pd.DataFrame: + dates = pd.bdate_range("2025-04-01", periods=12) + qqq_path = [100.0, 96.0, 92.0, 88.0, 84.0, 82.0, 84.0, 87.0, 90.0, 92.0, 94.0, 96.0] + tqqq_path = [100.0, 88.0, 78.0, 70.0, 64.0, 60.0, 66.0, 74.0, 82.0, 88.0, 94.0, 100.0] + vix_path = [20.0, 32.0, 45.0, 54.0, 60.0, 58.0, 55.0, 48.0, 45.0, 42.0, 39.0, 35.0] + vix3m_path = [22.0, 28.0, 38.0, 45.0, 48.0, 47.0, 45.0, 41.0, 39.0, 37.0, 35.0, 33.0] + rows: list[dict[str, object]] = [] + for idx, as_of in enumerate(dates): + rows.append({"symbol": "QQQ", "as_of": as_of, "close": qqq_path[idx], "volume": 1_000_000}) + rows.append({"symbol": "TQQQ", "as_of": as_of, "close": tqqq_path[idx], "volume": 1_000_000}) + rows.append({"symbol": "VIX", "as_of": as_of, "close": vix_path[idx], "volume": 0}) + rows.append({"symbol": "VIX3M", "as_of": as_of, "close": vix3m_path[idx], "volume": 0}) + return pd.DataFrame(rows) + + def _macro_stress_prices() -> pd.DataFrame: dates = pd.bdate_range("2025-01-02", periods=260) rows: list[dict[str, object]] = [] @@ -348,6 +364,43 @@ def test_strategy_plugin_runner_runs_unified_market_regime_control_for_tqqq(tmp_ assert payload["log_record"]["canonical_route"] == "risk_reduced" +def test_strategy_plugin_runner_can_enable_panic_reversal_inside_market_regime_control(tmp_path) -> None: + prices_path = tmp_path / "market_regime_panic_prices.csv" + output_dir = tmp_path / STRATEGY_NAME / "plugins" / PLUGIN_MARKET_REGIME_CONTROL + _panic_reversal_prices().to_csv(prices_path, index=False) + config = { + "output_dir": str(tmp_path / "runner"), + "default_mode": "shadow", + "strategy_plugins": [ + { + "strategy": STRATEGY_NAME, + "plugin": PLUGIN_MARKET_REGIME_CONTROL, + "enabled": True, + "inputs": { + "prices": str(prices_path), + "as_of": "2025-04-10", + "start_date": "2025-04-01", + "crisis_enabled": False, + "macro_enabled": False, + "taco_enabled": False, + "panic_reversal_enabled": True, + }, + "outputs": {"output_dir": str(output_dir)}, + } + ], + } + + summary = run_configured_plugins(config) + + assert summary["strategy_plugins"][0]["status"] == "ok" + payload = json.loads((output_dir / "latest_signal.json").read_text(encoding="utf-8")) + assert payload["canonical_route"] == "opportunity_watch" + assert payload["position_control"]["panic_reversal_allowed"] is True + assert payload["position_control"]["panic_reversal_size_scalar"] == 0.0 + assert payload["position_control"]["taco_allowed"] is False + assert "panic_reversal:panic_reversal" in payload["position_control"]["reason_codes"] + + def test_strategy_plugin_runner_runs_general_market_regime_notification(tmp_path) -> None: prices_path = tmp_path / "market_regime_prices.csv" output_dir = tmp_path / GENERAL_MARKET_REGIME_NOTIFICATION_TARGET / "plugins" / PLUGIN_MARKET_REGIME_CONTROL @@ -430,7 +483,12 @@ def test_strategy_plugin_runner_contract_registry_prefers_unified_plugin() -> No assert set(PLUGIN_COMPATIBLE_NOTIFICATION_TARGETS[PLUGIN_MARKET_REGIME_CONTROL]) == { GENERAL_MARKET_REGIME_NOTIFICATION_TARGET, } + assert PLUGIN_COMPATIBLE_STRATEGIES[PLUGIN_PANIC_REVERSAL_SHADOW] == (STRATEGY_NAME,) + assert PLUGIN_COMPATIBLE_NOTIFICATION_TARGETS[PLUGIN_PANIC_REVERSAL_SHADOW] == ( + GENERAL_MARKET_REGIME_NOTIFICATION_TARGET, + ) assert PLUGIN_SCHEMA_VERSIONS[PLUGIN_MARKET_REGIME_CONTROL] == ("market_regime_control.v1",) + assert PLUGIN_SCHEMA_VERSIONS[PLUGIN_PANIC_REVERSAL_SHADOW] == ("panic_reversal_shadow.v1",) assert PLUGIN_DEPRECATED_SUCCESSORS[PLUGIN_CRISIS_RESPONSE_SHADOW] == PLUGIN_MARKET_REGIME_CONTROL assert PLUGIN_DEPRECATED_SUCCESSORS[PLUGIN_MACRO_RISK_GOVERNOR] == PLUGIN_MARKET_REGIME_CONTROL assert PLUGIN_DEPRECATED_SUCCESSORS[PLUGIN_TACO_REBOUND_SHADOW] == PLUGIN_MARKET_REGIME_CONTROL @@ -438,6 +496,10 @@ def test_strategy_plugin_runner_contract_registry_prefers_unified_plugin() -> No PLUGIN_MARKET_REGIME_CONTROL, SOXL_STRATEGY_NAME, ) not in PLUGIN_CONSUMPTION_POLICY_REGISTRY + assert ( + PLUGIN_PANIC_REVERSAL_SHADOW, + SOXL_STRATEGY_NAME, + ) not in PLUGIN_CONSUMPTION_POLICY_REGISTRY assert PLUGIN_NOTIFICATION_TARGET_POLICY_REGISTRY[ (PLUGIN_MARKET_REGIME_CONTROL, GENERAL_MARKET_REGIME_NOTIFICATION_TARGET) ].position_control_allowed is False @@ -719,7 +781,96 @@ def test_strategy_plugin_runner_rejects_taco_rebound_for_non_tqqq_strategy(tmp_p with pytest.raises(ValueError, match="strategy-limited"): run_configured_plugins(config) - assert not (output_dir / "latest_signal.json").exists() + +def test_strategy_plugin_runner_runs_panic_reversal_notification_mount_for_tqqq(tmp_path) -> None: + prices_path = tmp_path / "panic_prices.csv" + output_dir = tmp_path / STRATEGY_NAME / "plugins" / PLUGIN_PANIC_REVERSAL_SHADOW + _panic_reversal_prices().to_csv(prices_path, index=False) + config = { + "output_dir": str(tmp_path / "runner"), + "default_mode": "shadow", + "strategy_plugins": [ + { + "strategy": STRATEGY_NAME, + "plugin": PLUGIN_PANIC_REVERSAL_SHADOW, + "enabled": True, + "inputs": { + "prices": str(prices_path), + "as_of": "2025-04-10", + "start_date": "2025-04-01", + }, + "outputs": {"output_dir": str(output_dir)}, + } + ], + } + + summary = run_configured_plugins(config) + + result = summary["strategy_plugins"][0] + assert result["strategy"] == STRATEGY_NAME + assert result["plugin"] == PLUGIN_PANIC_REVERSAL_SHADOW + assert result["status"] == "ok" + assert "route=panic_reversal action=notify_manual_review" in result["message"] + latest = json.loads((output_dir / "latest_signal.json").read_text(encoding="utf-8")) + assert latest["manual_review_required"] is True + assert latest["would_trade_if_enabled"] is False + assert latest["execution_controls"]["position_control_allowed"] is False + assert latest["execution_controls"]["consumption_evidence_status"] == EVIDENCE_NOTIFICATION_ONLY + assert latest["localized_messages"]["labels"]["canonical_route"]["zh-CN"] == "恐慌反转" + + +def test_strategy_plugin_runner_rejects_panic_reversal_for_soxl_strategy_mount(tmp_path) -> None: + prices_path = tmp_path / "panic_prices.csv" + _panic_reversal_prices().to_csv(prices_path, index=False) + config = { + "output_dir": str(tmp_path / "runner"), + "default_mode": "shadow", + "strategy_plugins": [ + { + "strategy": SOXL_STRATEGY_NAME, + "plugin": PLUGIN_PANIC_REVERSAL_SHADOW, + "enabled": True, + "inputs": {"prices": str(prices_path), "as_of": "2025-04-10"}, + } + ], + } + + with pytest.raises(ValueError, match="strategy-limited"): + run_configured_plugins(config) + + +def test_strategy_plugin_runner_runs_panic_reversal_general_notification_target(tmp_path) -> None: + prices_path = tmp_path / "panic_prices.csv" + output_dir = tmp_path / GENERAL_MARKET_REGIME_NOTIFICATION_TARGET / "plugins" / PLUGIN_PANIC_REVERSAL_SHADOW + _panic_reversal_prices().to_csv(prices_path, index=False) + config = { + "output_dir": str(tmp_path / "runner"), + "default_mode": "shadow", + "notification_targets": [ + { + "notification_target": GENERAL_MARKET_REGIME_NOTIFICATION_TARGET, + "plugin": PLUGIN_PANIC_REVERSAL_SHADOW, + "enabled": True, + "inputs": { + "prices": str(prices_path), + "as_of": "2025-04-10", + "start_date": "2025-04-01", + }, + "outputs": {"output_dir": str(output_dir)}, + } + ], + } + + summary = run_configured_plugins(config) + + assert summary["strategy_plugins"] == [] + result = summary["notification_targets"][0] + assert result["plugin"] == PLUGIN_PANIC_REVERSAL_SHADOW + assert result["status"] == "ok" + latest = json.loads((output_dir / "latest_signal.json").read_text(encoding="utf-8")) + assert latest["target_type"] == "notification_target" + assert latest["execution_controls"]["position_control_allowed"] is False + assert latest["execution_controls"]["strategy_runtime_metadata_allowed"] is False def test_strategy_plugin_runner_can_skip_disabled_taco_notification_mount(tmp_path) -> None: