Tune macro volatility confirmation defaults

Author: Pigbibi

src/quant_strategy_plugins/macro_risk_governor_plugin.py

@@ -245,7 +245,8 @@ def build_macro_risk_governor_signal(
     benchmark_drawdown_watch: float = -0.10,
     benchmark_drawdown_crisis: float = -0.18,
     realized_vol_window: int = 10,
-    realized_vol_threshold: float = 0.35,
+    realized_vol_threshold: float = 0.30,
+    realized_vol_requires_confirmation: bool = True,
     vix_watch_level: float = 28.0,
     vix_crisis_level: float = 35.0,
     vix_spike_lookback_days: int = 5,
@@ -261,7 +262,7 @@ def build_macro_risk_governor_signal(
     delever_score_threshold: float = 5.0,
     crisis_score_threshold: float = 7.0,
     delever_leverage_scalar: float = 0.0,
-    delever_risk_asset_scalar: float = 1.0,
+    delever_risk_asset_scalar: float = 0.0,
     crisis_leverage_scalar: float = 0.0,
     crisis_risk_asset_scalar: float = 0.0,
 ) -> dict[str, Any]:
@@ -447,12 +448,36 @@ def build_macro_risk_governor_signal(
         threshold=float(pizza_index_watch_level),
         actionable=False,
     )
+    realized_vol_confirmed_for_action = None
+    realized_vol_check = checks.get("benchmark_realized_volatility_high")
+    if realized_vol_check is not None:
+        volatility_active = bool(realized_vol_check.get("active", False))
+        if volatility_active:
+            confirmation_checks = (
+                "vix_watch_level",
+                "vix_crisis_level",
+                "vix_spike",
+                "credit_pair_stress",
+                "hy_oas_watch_level",
+                "hy_oas_widening",
+                "financial_stress_index_high",
+            )
+            realized_vol_confirmed_for_action = any(
+                bool(checks.get(name, {}).get("active", False)) for name in confirmation_checks
+            )
+            if bool(realized_vol_requires_confirmation) and not realized_vol_confirmed_for_action:
+                realized_vol_check["actionable"] = False
+                realized_vol_check["suppression_reason"] = "missing_volatility_stress_confirmation"
+        realized_vol_check["confirmation_required"] = bool(realized_vol_requires_confirmation)
+        realized_vol_check["confirmed_for_action"] = realized_vol_confirmed_for_action
     evidence["metrics"].update(
         {
             "hy_oas": hy_oas,
             "hy_oas_delta_63d": hy_oas_delta,
             "financial_stress": financial_stress,
             "pentagon_pizza_index": pizza_index,
+            "benchmark_realized_volatility_requires_confirmation": bool(realized_vol_requires_confirmation),
+            "benchmark_realized_volatility_confirmed_for_action": realized_vol_confirmed_for_action,
         }
     )
 
@@ -626,11 +651,18 @@ def build_parser() -> argparse.ArgumentParser:
     )
     parser.add_argument("--max-price-age-days", type=int, default=DEFAULT_MAX_PRICE_AGE_DAYS)
     parser.add_argument("--max-external-context-age-days", type=int, default=DEFAULT_MAX_EXTERNAL_CONTEXT_AGE_DAYS)
+    parser.add_argument("--realized-vol-threshold", type=float, default=0.30)
+    parser.add_argument(
+        "--realized-vol-requires-confirmation",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+        help="Require VIX, credit, or financial-stress confirmation before realized volatility contributes to actionable score.",
+    )
     parser.add_argument("--watch-score-threshold", type=float, default=3.0)
     parser.add_argument("--delever-score-threshold", type=float, default=5.0)
     parser.add_argument("--crisis-score-threshold", type=float, default=7.0)
     parser.add_argument("--delever-leverage-scalar", type=float, default=0.0)
-    parser.add_argument("--delever-risk-asset-scalar", type=float, default=1.0)
+    parser.add_argument("--delever-risk-asset-scalar", type=float, default=0.0)
     parser.add_argument("--output-dir", default=DEFAULT_OUTPUT_DIR)
     return parser
 
@@ -663,6 +695,8 @@ def main(argv: list[str] | None = None) -> int:
         credit_pairs=credit_pairs,
         max_price_age_days=args.max_price_age_days,
         max_external_context_age_days=args.max_external_context_age_days,
+        realized_vol_threshold=args.realized_vol_threshold,
+        realized_vol_requires_confirmation=args.realized_vol_requires_confirmation,
         watch_score_threshold=args.watch_score_threshold,
         delever_score_threshold=args.delever_score_threshold,
         crisis_score_threshold=args.crisis_score_threshold,

src/quant_strategy_plugins/strategy_plugin_runner.py

@@ -331,6 +331,9 @@ def _build_macro_risk_governor_kwargs(plugin_config: Mapping[str, Any]) -> dict[
         "credit_relative_lookback_days",
         "hy_oas_delta_lookback_days",
     }
+    bool_keys = {
+        "realized_vol_requires_confirmation",
+    }
     for key in string_keys:
         if key in plugin_config and plugin_config[key] is not None:
             kwargs[key] = str(plugin_config[key]).strip()
@@ -340,6 +343,9 @@ def _build_macro_risk_governor_kwargs(plugin_config: Mapping[str, Any]) -> dict[
     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 "credit_pairs" in plugin_config:

tests/test_macro_risk_governor_plugin.py

@@ -14,11 +14,11 @@
 )
 
 
-def _macro_prices(*, stress: bool = False) -> pd.DataFrame:
+def _macro_prices(*, stress: bool = False, volatility_spike: bool = False, vix_level: float = 15.0) -> pd.DataFrame:
     dates = pd.bdate_range("2025-01-02", periods=260)
     rows: list[dict[str, object]] = []
     qqq = pd.Series([100.0 + idx * 0.10 for idx in range(len(dates))], index=dates)
-    vix = pd.Series(15.0, index=dates)
+    vix = pd.Series(float(vix_level), index=dates)
     hyg = pd.Series(100.0, index=dates)
     ief = pd.Series(100.0, index=dates)
     if stress:
@@ -35,6 +35,11 @@ def _macro_prices(*, stress: bool = False) -> pd.DataFrame:
             [100.0 + idx * (3.0 / 21.0) for idx in range(22)],
             index=dates[-22:],
         )
+    if volatility_spike:
+        qqq.iloc[-10:] = pd.Series(
+            [130.0, 122.0, 131.0, 121.0, 132.0, 122.0, 133.0, 123.0, 134.0, 124.0],
+            index=dates[-10:],
+        )
     prices = {
         "QQQ": qqq,
         "TQQQ": qqq * 3.0,
@@ -69,7 +74,7 @@ def test_macro_risk_governor_routes_stress_to_delever_when_below_crisis_threshol
     assert payload["suggested_action"] == "delever"
     assert payload["would_trade_if_enabled"] is True
     assert payload["leverage_scalar"] == 0.0
-    assert payload["risk_asset_scalar"] == 1.0
+    assert payload["risk_asset_scalar"] == 0.0
     assert "vix_crisis_level" in payload["reason_codes"]
     assert payload["checks"]["pentagon_pizza_watch"]["actionable"] is False
 
@@ -109,6 +114,44 @@ def test_macro_risk_governor_keeps_pizza_index_watch_only() -> None:
     assert payload["checks"]["pentagon_pizza_watch"]["actionable"] is False
 
 
+def test_macro_risk_governor_requires_confirmation_for_realized_volatility_action() -> None:
+    payload = build_macro_risk_governor_signal(
+        _macro_prices(volatility_spike=True),
+        as_of="2025-12-31",
+        watch_score_threshold=1.0,
+        delever_score_threshold=1.0,
+        crisis_score_threshold=99.0,
+    )
+
+    volatility_check = payload["checks"]["benchmark_realized_volatility_high"]
+    assert volatility_check["active"] is True
+    assert volatility_check["actionable"] is False
+    assert volatility_check["confirmed_for_action"] is False
+    assert volatility_check["suppression_reason"] == "missing_volatility_stress_confirmation"
+    assert payload["actionable_score"] == 0.0
+    assert payload["total_score"] == 1.0
+    assert payload["canonical_route"] == ROUTE_WATCH
+    assert payload["suggested_action"] == "watch_only"
+
+
+def test_macro_risk_governor_allows_realized_volatility_action_when_vix_confirms() -> None:
+    payload = build_macro_risk_governor_signal(
+        _macro_prices(volatility_spike=True, vix_level=30.0),
+        as_of="2025-12-31",
+        watch_score_threshold=1.0,
+        delever_score_threshold=1.0,
+        crisis_score_threshold=99.0,
+    )
+
+    volatility_check = payload["checks"]["benchmark_realized_volatility_high"]
+    assert volatility_check["active"] is True
+    assert volatility_check["actionable"] is True
+    assert volatility_check["confirmed_for_action"] is True
+    assert payload["actionable_score"] >= 1.0
+    assert payload["canonical_route"] == ROUTE_DELEVER
+    assert payload["suggested_action"] == "delever"
+
+
 def test_macro_risk_governor_writes_json_csv_and_evidence(tmp_path) -> None:
     payload = build_macro_risk_governor_signal(_macro_prices(stress=True), as_of="2025-12-31")
 

tests/test_strategy_plugin_runner.py

@@ -256,7 +256,7 @@ def test_strategy_plugin_runner_runs_unified_market_regime_control_for_tqqq(tmp_
     assert payload["plugin"] == PLUGIN_MARKET_REGIME_CONTROL
     assert payload["canonical_route"] == "risk_reduced"
     assert payload["position_control"]["leverage_scalar"] == 0.0
-    assert payload["position_control"]["risk_asset_scalar"] == 1.0
+    assert payload["position_control"]["risk_asset_scalar"] == 0.0
     assert payload["position_control"]["taco_allowed"] is False
     assert payload["execution_controls"]["strategy_runtime_metadata_allowed"] is True
     assert payload["execution_controls"]["broker_order_allowed"] is False