From 415fe377de795f0855c63a9983c6936f754cfd0a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Tue, 26 May 2026 14:27:14 +0800 Subject: [PATCH] Add strategy plugin alert guidance --- pyproject.toml | 2 +- src/quant_platform_kit/common/__init__.py | 4 + .../common/strategy_plugins.py | 124 ++++++++++++++++++ tests/test_strategy_plugins.py | 9 ++ 4 files changed, 138 insertions(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 1f82521..f0c517d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "quant-platform-kit" -version = "0.7.31" +version = "0.7.32" description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies." readme = "README.md" requires-python = ">=3.9" diff --git a/src/quant_platform_kit/common/__init__.py b/src/quant_platform_kit/common/__init__.py index d1ed075..7aa1cba 100644 --- a/src/quant_platform_kit/common/__init__.py +++ b/src/quant_platform_kit/common/__init__.py @@ -56,8 +56,10 @@ StrategyPluginDefinition, StrategyPluginMountConfig, StrategyPluginSignal, + build_strategy_plugin_alert_guidance, build_strategy_plugin_alert_key, build_strategy_plugin_alert_messages, + build_strategy_plugin_alert_scope_note, build_strategy_plugin_notification_lines, build_strategy_plugin_report_payload, load_configured_strategy_plugin_signals, @@ -117,8 +119,10 @@ "StrategyPluginDefinition", "StrategyPluginMountConfig", "StrategyPluginSignal", + "build_strategy_plugin_alert_guidance", "build_strategy_plugin_alert_key", "build_strategy_plugin_alert_messages", + "build_strategy_plugin_alert_scope_note", "build_strategy_plugin_notification_lines", "build_strategy_plugin_report_payload", "build_runtime_target", diff --git a/src/quant_platform_kit/common/strategy_plugins.py b/src/quant_platform_kit/common/strategy_plugins.py index 13200ac..f3598a9 100644 --- a/src/quant_platform_kit/common/strategy_plugins.py +++ b/src/quant_platform_kit/common/strategy_plugins.py @@ -28,6 +28,35 @@ } ) TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES = frozenset({"tqqq_growth_income"}) +_DEFAULT_STRATEGY_PLUGIN_ALERT_GUIDANCE: Mapping[tuple[str, str, str], str] = { + ( + PLUGIN_CRISIS_RESPONSE_SHADOW, + "true_crisis", + "defend", + ): ( + "Consider reducing leveraged exposure, moving to defensive or cash-like positions, " + "and pausing new risk additions until the signal de-escalates." + ), + ( + PLUGIN_CRISIS_RESPONSE_SHADOW, + "no_action", + "blocked", + ): ( + "The crisis route was blocked by a guard; review data freshness and context before " + "acting on the signal." + ), + ( + PLUGIN_TACO_REBOUND_SHADOW, + "taco_rebound", + "notify_manual_review", + ): ( + "Manual review only: consider a small, pre-sized probe or staged entry with a " + "predefined invalidation level; avoid full-size deployment from this alert alone." + ), +} +_DEFAULT_STRATEGY_PLUGIN_ALERT_SCOPE_NOTE = ( + "Manual review notice only; the plugin does not place orders or change allocations." +) @dataclass(frozen=True) @@ -445,6 +474,64 @@ def should_alert_strategy_plugin_signal(signal: StrategyPluginSignal) -> bool: ) +def build_strategy_plugin_alert_guidance( + signal: StrategyPluginSignal, + *, + translator: Callable[..., str] | None = None, +) -> str | None: + plugin = _normalize_strategy_plugin_field(getattr(signal, "plugin", None)) + route = _normalize_strategy_plugin_field(getattr(signal, "canonical_route", None)) + action = _normalize_strategy_plugin_field(getattr(signal, "suggested_action", None)) + translated = _translate_first( + translator, + ( + f"strategy_plugin_guidance_{plugin}_{route}_{action}", + f"strategy_plugin_guidance_{plugin}_{route}", + f"strategy_plugin_guidance_{plugin}_{action}", + f"strategy_plugin_guidance_{plugin}", + f"strategy_plugin_guidance_{route}_{action}", + f"strategy_plugin_guidance_{action}", + ), + ) + if translated: + return translated + return _DEFAULT_STRATEGY_PLUGIN_ALERT_GUIDANCE.get((plugin, route, action)) + + +def build_strategy_plugin_alert_scope_note( + signal: StrategyPluginSignal, + *, + translator: Callable[..., str] | None = None, +) -> str | None: + controls = getattr(signal, "execution_controls", {}) or {} + if not isinstance(controls, Mapping): + controls = {} + notification_profile = str(controls.get("notification_profile") or "").strip().lower() + if notification_profile != "shadow_only" and any( + _as_bool(controls.get(field), default=False) + for field in ( + "broker_order_allowed", + "repository_broker_write_allowed", + "live_allocation_mutation_allowed", + "repository_allocation_mutation_allowed", + "allocation_recommendation_allowed", + "position_sizing_allowed", + "selection_allowed", + ) + ): + return None + return ( + _translate_first( + translator, + ( + f"strategy_plugin_alert_scope_{_normalize_strategy_plugin_field(getattr(signal, 'plugin', None))}", + "strategy_plugin_alert_scope", + ), + ) + or _DEFAULT_STRATEGY_PLUGIN_ALERT_SCOPE_NOTE + ) + + def build_strategy_plugin_alert_key( signal: StrategyPluginSignal, *, @@ -504,6 +591,8 @@ def build_strategy_plugin_alert_messages( translated_route = translate_strategy_plugin_value("route", route, translator=translator) translated_action = translate_strategy_plugin_value("action", action, translator=translator) strategy = str(strategy_label or getattr(signal, "strategy", None) or "").strip() or "unknown" + guidance = build_strategy_plugin_alert_guidance(signal, translator=translator) + scope_note = build_strategy_plugin_alert_scope_note(signal, translator=translator) subject = _translate( translator, "strategy_plugin_alert_subject", @@ -567,6 +656,24 @@ def build_strategy_plugin_alert_messages( ), ] ) + if guidance: + body_lines.append( + _translate( + translator, + "strategy_plugin_alert_guidance", + fallback="Manual guidance: {guidance}", + guidance=guidance, + ) + ) + if scope_note: + body_lines.append( + _translate( + translator, + "strategy_plugin_alert_scope_note", + fallback="Scope: {scope_note}", + scope_note=scope_note, + ) + ) metadata = { "strategy": getattr(signal, "strategy", None), "strategy_label": strategy, @@ -577,6 +684,8 @@ def build_strategy_plugin_alert_messages( "suggested_action": getattr(signal, "suggested_action", None), "would_trade_if_enabled": bool(getattr(signal, "would_trade_if_enabled", False)), "context_label": context or None, + "guidance": guidance, + "scope_note": scope_note, } messages.append( StrategyPluginAlertMessage( @@ -687,6 +796,21 @@ def _translate( return translated if translated != key else fallback.format(**kwargs) +def _translate_first( + translator: Callable[..., str] | None, + keys: Sequence[str], +) -> str | None: + if translator is None: + return None + for key in keys: + translated = translator(key) + if translated != key: + text = str(translated).strip() + if text: + return text + return None + + def _required_string(value: Any, *, field_name: str) -> str: text = _optional_string(value) if text is None: diff --git a/tests/test_strategy_plugins.py b/tests/test_strategy_plugins.py index 5369ae7..14387f8 100644 --- a/tests/test_strategy_plugins.py +++ b/tests/test_strategy_plugins.py @@ -334,10 +334,14 @@ def test_strategy_plugin_true_crisis_builds_generic_alert_message(self): "strategy_plugin_alert_action": "action={action}", "strategy_plugin_alert_mode": "mode={mode}", "strategy_plugin_alert_as_of": "as_of={as_of}", + "strategy_plugin_alert_guidance": "guidance={guidance}", + "strategy_plugin_alert_scope_note": "scope={scope_note}", "strategy_plugin_name_crisis_response_shadow": "Crisis", "strategy_plugin_mode_shadow": "shadow", "strategy_plugin_route_true_crisis": "true crisis", "strategy_plugin_action_defend": "defend", + "strategy_plugin_guidance_crisis_response_shadow_true_crisis_defend": "reduce leverage or move to cash", + "strategy_plugin_alert_scope": "manual review only", } alerts = build_strategy_plugin_alert_messages( @@ -355,9 +359,12 @@ def test_strategy_plugin_true_crisis_builds_generic_alert_message(self): self.assertIn("status=true crisis", alerts[0].body) self.assertIn("action=defend", alerts[0].body) self.assertIn("mode=shadow", alerts[0].body) + self.assertIn("guidance=reduce leverage or move to cash", alerts[0].body) + self.assertIn("scope=manual review only", alerts[0].body) self.assertNotIn("would_trade=", alerts[0].body) self.assertNotIn("source=", alerts[0].body) self.assertTrue(alerts[0].metadata["would_trade_if_enabled"]) + self.assertEqual(alerts[0].metadata["guidance"], "reduce leverage or move to cash") def test_taco_rebound_notification_alerts_without_trade_flag(self): signal = validate_strategy_plugin_signal_payload( @@ -377,6 +384,8 @@ def test_taco_rebound_notification_alerts_without_trade_flag(self): self.assertEqual(len(alerts), 1) self.assertIn("taco_rebound_shadow", alerts[0].subject) + self.assertIn("small, pre-sized probe", alerts[0].body) + self.assertIn("does not place orders", alerts[0].body) self.assertFalse(alerts[0].metadata["would_trade_if_enabled"])