feat: pre-trade hard gate — block no-stop BUYs, show blocked as non-actionable

Dockerfile

@@ -3,10 +3,12 @@ FROM python:3.12-slim
 
 WORKDIR /app
 
-# Install system dependencies (libpq for psycopg2)
+# Install system dependencies (libpq for psycopg2; git for risk-engine's
+# trading-py-commons git dependency during pip install)
 RUN apt-get update && apt-get install -y --no-install-recommends \
     build-essential \
     libpq-dev \
+    git \
     && rm -rf /var/lib/apt/lists/*
 
 # Create non-root user

decision_engine/checklist.py

@@ -65,6 +65,7 @@ class ChecklistResult:
     # Aggregated status
     all_checks_passed: bool = False
     status: str = "REVIEW"                    # "GO" | "REVIEW" | "BLOCKED"
+    block_reasons: list = field(default_factory=list)  # why BLOCKED (for the alert)
 
     # Optional detail fields for display
     earnings_date: Optional[str] = None       # e.g. "2026-03-12"
@@ -83,6 +84,7 @@ def to_dict(self) -> dict:
             "regime_compatible": self.regime_compatible,
             "all_checks_passed": self.all_checks_passed,
             "status": self.status,
+            "block_reasons": list(self.block_reasons),
             "earnings_date": self.earnings_date,
             "earnings_days_away": self.earnings_days_away,
             "earnings_verified": self.earnings_verified,
@@ -238,16 +240,33 @@ def evaluate(
             and result.regime_compatible
         )
 
-        # Status — hard gates first
+        # Status — hard gates first. Each triggers BLOCKED and records a
+        # human-readable reason so the alert can say exactly why.
         earnings_days = result.earnings_days_away
+        block_reasons: list = []
+
+        # A BUY whose plan has no valid stop must never be actionable (the MOH
+        # lesson). Scoped to when a plan exists — a missing plan entirely is the
+        # plan-engine-disabled path, which stays REVIEW (only earnings hard-gates).
+        if trade_plan is not None and not result.stop_loss_defined:
+            block_reasons.append("no stop plan")
+
         is_earnings_blocked = (
             earnings_days is not None
             and earnings_days <= EARNINGS_HARD_GATE_DAYS
         )
+        if is_earnings_blocked:
+            block_reasons.append(f"earnings in {earnings_days}d")
+
         # Size block only applies when we have plan data
         is_size_blocked = (
             trade_plan is not None and trade_plan.risk_pct > MAX_RISK_PCT_BLOCKED
         )
+        if is_size_blocked:
+            block_reasons.append(
+                f"risk {trade_plan.risk_pct:.1f}% > {MAX_RISK_PCT_BLOCKED:.0f}%"
+            )
+
         # Regime-conditional stocks in wrong regime → hard block.
         # If allowed_regimes is set (from tier data or rules.yaml), the stock
         # was proven via backtesting to only work in specific regimes.
@@ -256,8 +275,11 @@ def evaluate(
             allowed_regimes is not None
             and not result.regime_compatible
         )
+        if is_regime_blocked:
+            block_reasons.append(f"regime {regime_id} not in {sorted(allowed_regimes)}")
 
-        if is_earnings_blocked or is_size_blocked or is_regime_blocked:
+        result.block_reasons = block_reasons
+        if block_reasons:
             result.status = "BLOCKED"
         elif result.all_checks_passed:
             result.status = "GO"

decision_engine/config.py

@@ -110,6 +110,13 @@ class Settings(BaseServiceSettings):
         description="Enable publishing rules to Redis for cross-service access"
     )
 
+    # Pre-trade hard gate: publish BLOCKED BUYs as NON-ACTIONABLE marked alerts
+    # (trader sees the setup + why it was filtered). False = silently suppress them.
+    hard_gate_enabled: bool = Field(
+        True,
+        description="Publish checklist-BLOCKED BUY signals as non-actionable alerts (vs suppress)"
+    )
+
     # Risk engine settings
     risk_engine_enabled: bool = Field(
         True,

decision_engine/service.py

@@ -441,20 +441,38 @@ def handle_indicator_event(self, event: dict):
                             f"Checklist evaluation failed for {symbol}: {exc}"
                         )
 
-                # Enforce BLOCKED checklist: suppress publication entirely.
-                # A BLOCKED status means a hard gate failed (e.g. earnings
-                # imminent, no stop loss defined) — the signal MUST NOT reach
-                # the alert-service or the trader may act on it.
+                # Hard gate: a BLOCKED checklist means a hard rule failed
+                # (no stop plan, earnings imminent, risk too high, wrong regime).
+                # The signal is NON-ACTIONABLE. We publish it MARKED BLOCKED so
+                # the trader sees the setup and exactly why it was filtered — but
+                # skip the actionable gates below (risk-engine etc.), since it is
+                # not tradeable. HARD_GATE_ENABLED=false reverts to silent suppression.
                 if (
                     checklist_result is not None
                     and checklist_result.status == "BLOCKED"
                 ):
-                    m.SIGNALS_REJECTED.labels(reason="checklist_blocked").inc()
-                    logger.warning(
-                        f"Checklist BLOCKED for {symbol} — suppressing "
-                        f"{aggregated_signal.signal_type.value} signal "
-                        f"(confidence={aggregated_signal.aggregate_confidence:.2f})"
-                    )
+                    reasons = ", ".join(checklist_result.block_reasons) or "hard gate"
+                    if not getattr(self.settings, "hard_gate_enabled", True):
+                        m.SIGNALS_REJECTED.labels(reason="checklist_blocked").inc()
+                        logger.warning(
+                            f"Checklist BLOCKED for {symbol} ({reasons}) — "
+                            f"suppressing (hard_gate disabled)"
+                        )
+                        return
+                    # Debounce so a persistently-blocked symbol doesn't repeat every cycle.
+                    if self._should_publish(symbol, aggregated_signal):
+                        self.producer.publish_decision(
+                            aggregated_signal,
+                            indicators,
+                            risk_result=None,
+                            trade_plan=trade_plan,
+                            checklist_result=checklist_result,
+                        )
+                        self._last_publish[symbol] = datetime.utcnow()
+                        logger.warning(
+                            f"Checklist BLOCKED for {symbol} ({reasons}) — "
+                            f"published NON-ACTIONABLE alert"
+                        )
                     return
 
                 # Suppress BUY signals with R:R below minimum threshold.

tests/test_checklist.py

@@ -108,10 +108,14 @@ def test_to_dict_contains_all_keys(self):
             "stop_loss_defined", "position_sized_correctly", "rr_ratio_acceptable",
             "no_earnings_imminent", "regime_compatible", "all_checks_passed",
             "status", "earnings_date", "earnings_days_away", "earnings_verified",
-            "regime_id", "risk_pct", "rr_ratio",
+            "regime_id", "risk_pct", "rr_ratio", "block_reasons",
         ):
             self.assertIn(key, d)
 
+    def test_block_reasons_serialize(self):
+        result = ChecklistResult(status="BLOCKED", block_reasons=["no stop plan"])
+        self.assertEqual(result.to_dict()["block_reasons"], ["no stop plan"])
+
     def test_go_status_serializes(self):
         result = ChecklistResult(status="GO", all_checks_passed=True)
         self.assertEqual(result.to_dict()["status"], "GO")
@@ -133,6 +137,18 @@ def test_zero_stop_price_fails(self):
         result = ev.evaluate(_make_plan(stop_price=0.0), "BULL", "WPM")
         self.assertFalse(result.stop_loss_defined)
 
+    def test_no_stop_plan_hard_blocks(self):
+        """A plan with no valid stop is a hard block (the MOH lesson)."""
+        ev = _make_evaluator()
+        result = ev.evaluate(_make_plan(stop_price=0.0), "BULL", "WPM")
+        self.assertEqual(result.status, "BLOCKED")
+        self.assertIn("no stop plan", result.block_reasons)
+
+    def test_valid_stop_not_blocked_for_stop_reason(self):
+        ev = _make_evaluator()
+        result = ev.evaluate(_make_plan(stop_price=9.5), "BULL", "WPM")
+        self.assertNotIn("no stop plan", result.block_reasons)
+
 
 # ---------------------------------------------------------------------------
 # Position sizing check

tests/test_hard_gate.py

@@ -0,0 +1,97 @@
+"""
+Tests for the pre-trade hard gate at the emit path.
+
+A BLOCKED checklist means a hard rule failed (no stop plan, earnings imminent,
+risk too high, wrong regime). With hard_gate_enabled=True the signal is PUBLISHED
+marked BLOCKED (non-actionable, so the trader sees why it was filtered); with
+hard_gate_enabled=False it is silently suppressed.
+"""
+
+import unittest
+from datetime import datetime, timezone
+from unittest.mock import MagicMock
+
+from decision_engine.service import DecisionEngineService
+from decision_engine.config import Settings
+from decision_engine.checklist import ChecklistResult
+from decision_engine.rules.base import SignalType
+from decision_engine.models.signals import AggregatedSignal
+
+
+def _make_event(symbol: str = "CCJ") -> dict:
+    return {
+        "event_type": "INDICATOR_UPDATE",
+        "data": {
+            "symbol": symbol,
+            "indicators": {"RSI_14": 45.0, "close": 50.0},
+            "time": "2026-02-25T15:00:00Z",
+        },
+    }
+
+
+def _buy_signal(symbol: str = "CCJ") -> AggregatedSignal:
+    return AggregatedSignal(
+        symbol=symbol,
+        signal_type=SignalType.BUY,
+        aggregate_confidence=0.9,
+        primary_reasoning="test buy",
+        contributing_signals=[],
+        timestamp=datetime.now(timezone.utc),
+        regime_id="BULL",
+    )
+
+
+def _blocked_checklist() -> ChecklistResult:
+    return ChecklistResult(
+        stop_loss_defined=False,
+        status="BLOCKED",
+        block_reasons=["no stop plan"],
+    )
+
+
+class TestHardGateEmit(unittest.TestCase):
+    def _build_service(self, hard_gate_enabled: bool) -> DecisionEngineService:
+        settings = MagicMock(spec=Settings)
+        settings.hard_gate_enabled = hard_gate_enabled
+        svc = DecisionEngineService(settings)
+        svc._config = {"active_tickers_only": False, "active_tickers": {}}
+        svc.state_manager = MagicMock()
+        svc.tier_reader = None
+        # A BUY signal reaches the trade-plan + checklist path.
+        svc._evaluate_rules = MagicMock(return_value=_buy_signal())
+        svc.trade_plan_engine = MagicMock()
+        svc.trade_plan_engine.generate.return_value = MagicMock(
+            rr_warning=None, warnings=[], plan_valid=True
+        )
+        svc.checklist_evaluator = MagicMock()
+        svc.checklist_evaluator.evaluate.return_value = _blocked_checklist()
+        svc._should_publish = MagicMock(return_value=True)
+        svc.producer = MagicMock()
+        return svc
+
+    def test_blocked_buy_is_published_when_enabled(self):
+        svc = self._build_service(hard_gate_enabled=True)
+        svc.handle_indicator_event(_make_event("CCJ"))
+
+        svc.producer.publish_decision.assert_called_once()
+        # It is published with the BLOCKED checklist so alert-service can mark it.
+        kwargs = svc.producer.publish_decision.call_args.kwargs
+        self.assertEqual(kwargs["checklist_result"].status, "BLOCKED")
+        self.assertIn("no stop plan", kwargs["checklist_result"].block_reasons)
+
+    def test_blocked_buy_is_suppressed_when_disabled(self):
+        svc = self._build_service(hard_gate_enabled=False)
+        svc.handle_indicator_event(_make_event("CCJ"))
+
+        svc.producer.publish_decision.assert_not_called()
+
+    def test_blocked_buy_respects_debounce(self):
+        """Even enabled, a debounced (already-published) symbol is not re-sent."""
+        svc = self._build_service(hard_gate_enabled=True)
+        svc._should_publish = MagicMock(return_value=False)
+        svc.handle_indicator_event(_make_event("CCJ"))
+        svc.producer.publish_decision.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()