Register TACO rebound notification plugin

Author: Pigbibi

docs/strategy_plugin_runtime_contract.md

@@ -68,7 +68,13 @@ The default registry currently defines:
 
 | Plugin | Supported strategies | Supported mode | Escalated alert channel |
 | --- | --- | --- | --- |
-| `crisis_response_shadow` | `tqqq_growth_income`, `soxl_soxx_trend_income` | `shadow` | `email` |
+| `crisis_response_shadow` | `tqqq_growth_income`, `soxl_soxx_trend_income` | `shadow` | `email`, `sms`, `push`, `telegram` |
+| `taco_rebound_shadow` | `tqqq_growth_income` | `shadow` | `email`, `sms`, `push`, `telegram` |
+
+`taco_rebound_shadow` is notification-only. Its artifact may escalate a
+manual-review alert when a TACO-style rebound context is active, but it must not
+recommend position size, mutate live allocation, or imply broker order
+permission.
 
 To expand a plugin later, update the shared definition or pass an explicit
 definition registry into the parser/loader. This keeps future plugin eligibility

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "quant-platform-kit"
-version = "0.7.30"
+version = "0.7.31"
 description = "Shared broker adapters, domain models, execution ports, and notification utilities for QuantStrategyLab strategies."
 readme = "README.md"
 requires-python = ">=3.9"

src/quant_platform_kit/common/__init__.py

@@ -42,6 +42,7 @@
     CRISIS_RESPONSE_SHADOW_SUPPORTED_STRATEGIES,
     DEFAULT_STRATEGY_PLUGIN_DEFINITIONS,
     PLUGIN_CRISIS_RESPONSE_SHADOW,
+    PLUGIN_TACO_REBOUND_SHADOW,
     PLUGIN_MODE_SHADOW,
     STRATEGY_PLUGIN_ALERT_CHANNEL_EMAIL,
     STRATEGY_PLUGIN_ALERT_CHANNEL_PUSH,
@@ -50,6 +51,7 @@
     STRATEGY_PLUGIN_ALERT_ACTIONS,
     STRATEGY_PLUGIN_NON_ALERT_ROUTES,
     SUPPORTED_STRATEGY_PLUGIN_MODES,
+    TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES,
     StrategyPluginAlertMessage,
     StrategyPluginDefinition,
     StrategyPluginMountConfig,
@@ -77,6 +79,7 @@
     "DEFAULT_TERMINAL_FUNDING_BLOCK_SKIP_REASONS",
     "DEFAULT_TERMINAL_STRATEGY_RUN_STAGES",
     "PLUGIN_CRISIS_RESPONSE_SHADOW",
+    "PLUGIN_TACO_REBOUND_SHADOW",
     "PLUGIN_MODE_SHADOW",
     "STAGE_COMPLETED",
     "STAGE_DRY_RUN_COMPLETED",
@@ -94,6 +97,7 @@
     "STRATEGY_PLUGIN_ALERT_ACTIONS",
     "STRATEGY_PLUGIN_NON_ALERT_ROUTES",
     "SUPPORTED_STRATEGY_PLUGIN_MODES",
+    "TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES",
     "filter_execution_blocking_skips",
     "is_terminal_funding_block",
     "is_terminal_strategy_run_stage",

src/quant_platform_kit/common/strategy_plugins.py

@@ -11,6 +11,7 @@
 from typing import Any, Callable
 
 PLUGIN_CRISIS_RESPONSE_SHADOW = "crisis_response_shadow"
+PLUGIN_TACO_REBOUND_SHADOW = "taco_rebound_shadow"
 PLUGIN_MODE_SHADOW = "shadow"
 STRATEGY_PLUGIN_ALERT_CHANNEL_EMAIL = "email"
 STRATEGY_PLUGIN_ALERT_CHANNEL_SMS = "sms"
@@ -26,6 +27,7 @@
         "soxl_soxx_trend_income",
     }
 )
+TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES = frozenset({"tqqq_growth_income"})
 
 
 @dataclass(frozen=True)
@@ -76,7 +78,18 @@ def supports_strategy(self, strategy: str) -> bool:
             STRATEGY_PLUGIN_ALERT_CHANNEL_PUSH,
             STRATEGY_PLUGIN_ALERT_CHANNEL_TELEGRAM,
         ),
-    )
+    ),
+    PLUGIN_TACO_REBOUND_SHADOW: StrategyPluginDefinition(
+        plugin=PLUGIN_TACO_REBOUND_SHADOW,
+        supported_strategies=TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES,
+        supported_modes=SUPPORTED_STRATEGY_PLUGIN_MODES,
+        alert_channels=(
+            STRATEGY_PLUGIN_ALERT_CHANNEL_EMAIL,
+            STRATEGY_PLUGIN_ALERT_CHANNEL_SMS,
+            STRATEGY_PLUGIN_ALERT_CHANNEL_PUSH,
+            STRATEGY_PLUGIN_ALERT_CHANNEL_TELEGRAM,
+        ),
+    ),
 }
 
 

tests/test_strategy_plugins.py

@@ -7,11 +7,13 @@
     CRISIS_RESPONSE_SHADOW_SUPPORTED_STRATEGIES,
     DEFAULT_STRATEGY_PLUGIN_DEFINITIONS,
     PLUGIN_CRISIS_RESPONSE_SHADOW,
+    PLUGIN_TACO_REBOUND_SHADOW,
     PLUGIN_MODE_SHADOW,
     STRATEGY_PLUGIN_ALERT_CHANNEL_EMAIL,
     STRATEGY_PLUGIN_ALERT_CHANNEL_PUSH,
     STRATEGY_PLUGIN_ALERT_CHANNEL_SMS,
     STRATEGY_PLUGIN_ALERT_CHANNEL_TELEGRAM,
+    TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES,
     StrategyPluginDefinition,
     build_strategy_plugin_alert_messages,
     build_strategy_plugin_notification_lines,
@@ -111,6 +113,34 @@ def test_default_plugin_definition_limits_crisis_response_to_supported_strategie
             mode=PLUGIN_MODE_SHADOW,
         )
 
+    def test_default_plugin_definition_limits_taco_rebound_to_tqqq_notifications(self):
+        definition = DEFAULT_STRATEGY_PLUGIN_DEFINITIONS[PLUGIN_TACO_REBOUND_SHADOW]
+
+        self.assertEqual(definition.supported_strategies, TACO_REBOUND_SHADOW_SUPPORTED_STRATEGIES)
+        self.assertEqual(
+            definition.alert_channels,
+            (
+                STRATEGY_PLUGIN_ALERT_CHANNEL_EMAIL,
+                STRATEGY_PLUGIN_ALERT_CHANNEL_SMS,
+                STRATEGY_PLUGIN_ALERT_CHANNEL_PUSH,
+                STRATEGY_PLUGIN_ALERT_CHANNEL_TELEGRAM,
+            ),
+        )
+        validate_strategy_plugin_compatibility(
+            strategy="tqqq_growth_income",
+            plugin=PLUGIN_TACO_REBOUND_SHADOW,
+            mode=PLUGIN_MODE_SHADOW,
+        )
+        with self.assertRaisesRegex(
+            ValueError,
+            "taco_rebound_shadow does not support strategy soxl_soxx_trend_income",
+        ):
+            validate_strategy_plugin_compatibility(
+                strategy="soxl_soxx_trend_income",
+                plugin=PLUGIN_TACO_REBOUND_SHADOW,
+                mode=PLUGIN_MODE_SHADOW,
+            )
+
     def test_parse_strategy_plugin_mounts_rejects_unsupported_crisis_response_strategy(self):
         raw = [
             {
@@ -126,6 +156,20 @@ def test_parse_strategy_plugin_mounts_rejects_unsupported_crisis_response_strate
         ):
             parse_strategy_plugin_mounts(raw)
 
+    def test_parse_strategy_plugin_mounts_accepts_taco_rebound_tqqq_notification(self):
+        mounts = parse_strategy_plugin_mounts(
+            [
+                {
+                    "strategy": "tqqq_growth_income",
+                    "plugin": PLUGIN_TACO_REBOUND_SHADOW,
+                    "signal_path": "gs://bucket/taco/latest_signal.json",
+                }
+            ]
+        )
+
+        self.assertEqual(mounts[0].strategy, "tqqq_growth_income")
+        self.assertEqual(mounts[0].plugin, PLUGIN_TACO_REBOUND_SHADOW)
+
     def test_plugin_definition_can_extend_future_strategy_support(self):
         raw = [
             {
@@ -315,6 +359,26 @@ def test_strategy_plugin_true_crisis_builds_generic_alert_message(self):
         self.assertNotIn("source=", alerts[0].body)
         self.assertTrue(alerts[0].metadata["would_trade_if_enabled"])
 
+    def test_taco_rebound_notification_alerts_without_trade_flag(self):
+        signal = validate_strategy_plugin_signal_payload(
+            {
+                **_signal_payload(plugin=PLUGIN_TACO_REBOUND_SHADOW),
+                "schema_version": "taco_rebound_shadow.v2",
+                "canonical_route": "taco_rebound",
+                "suggested_action": "notify_manual_review",
+                "would_trade_if_enabled": False,
+                "manual_review_required": True,
+            },
+            source_uri="gs://bucket/taco/latest_signal.json",
+        )
+
+        self.assertTrue(should_alert_strategy_plugin_signal(signal))
+        alerts = build_strategy_plugin_alert_messages([signal], strategy_label="TQQQ Growth Income")
+
+        self.assertEqual(len(alerts), 1)
+        self.assertIn("taco_rebound_shadow", alerts[0].subject)
+        self.assertFalse(alerts[0].metadata["would_trade_if_enabled"])
+
 
 if __name__ == "__main__":
     unittest.main()