Entry-path unification (step 1): shared anti-chase gates on the live auto path

Closes the safety gap the consolidation pass found: the anti-chase / halt
guards (the source-trader blow-up protections) ran ONLY on the fast trigger
path, leaving live auto entries — which is where 100% of fills actually
originate — unprotected.

- Extract _anti_chase_skip(entry_ref, trigger, day_open, halted) from
  submit_breakout_now (pure refactor; the fast path is behaviour-identical,
  60 trigger/exec tests still green).
- Thread day_open through the setup dict (setup_evaluator) +
  query_ready_signals_snapshot so the auto path has the gate input.
- request_approvals_for_ready_signals calls the shared gate behind
  TRADING_UNIFIED_ENTRY (env default on). The auto path rests a limit AT the
  trigger so it can't chase above it (over_extended is a no-op); the gate that
  bites is day-extension — don't enter a name whose breakout level is already
  >30% above the session open. Fail-open on missing day_open; halted=False (a
  resting limit won't fill mid-halt). Emits rule_type=over_extended_day on skip.
- Boot log shows "anti-chase ON (auto path)".

Tests: _anti_chase_skip logic, parabolic-day blocked when on, allowed when off,
fail-open on missing day_open, normal day allowed. Full path suites green (73).

Note: this is step 1 (the safety win). Full single-chokepoint + fill-model
unification + halt-status-on-auto-path remain staged per docs/CONSOLIDATION_PLAN.md.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

run_live_paper.py

@@ -806,6 +806,7 @@ def main(argv: list[str] | None = None) -> int:
     if _es is not None:
         print(
             f"[boot] entry gates: VWAP {'ENFORCED' if _es.require_above_vwap else 'shadow-only'}"
+            f", anti-chase {'ON (auto path)' if _es.unified_entry else 'fast-path only'}"
             f", daily-cap {_es.max_fresh_entries_per_day or 'off'}"
             f", risk-cap ${_es.max_risk_dollars:.0f}"
         )

storage/projections.py

@@ -229,6 +229,7 @@ def query_ready_signals_snapshot(store, session_id: str | None = None) -> list[d
             "quality_score": signal_data.get("quality_score"),
             "vwap": signal_data.get("vwap"),
             "above_vwap": signal_data.get("above_vwap"),
+            "day_open": signal_data.get("day_open"),
             "timestamp": event.get("timestamp"),
         }
     return sorted(

strategy/evaluation/setup_evaluator.py

@@ -157,10 +157,11 @@ def evaluate_setup(
                 "quality_score": quality.score,
                 "quality_grade": quality.grade,
                 "confidence": round(score_pct / 100.0, 4),
-                # surfaced for the live VWAP entry gate + observability
+                # surfaced for the live VWAP + anti-chase entry gates + observability
                 "vwap": (round(float(levels.vwap), 4)
                          if levels.vwap is not None else None),
                 "above_vwap": bool(above_vwap),
+                "day_open": round(float(open_price), 4) if open_price > 0 else None,
             }
         )
 

tests/unit/test_trading_execution.py

@@ -67,7 +67,7 @@ class Clock:
 
 
 def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45,
-                above_vwap=None, vwap=None):
+                above_vwap=None, vwap=None, day_open=None):
     signal_data = {
         "entry_price": entry,
         "stop_loss_price": stop,
@@ -77,6 +77,8 @@ def emit_signal(store, symbol="GOOD", entry=14.0, stop=13.45,
         signal_data["above_vwap"] = above_vwap
         signal_data["vwap"] = vwap if vwap is not None else (
             round(entry - 1.0, 2) if above_vwap else round(entry + 1.0, 2))
+    if day_open is not None:
+        signal_data["day_open"] = day_open
     store.emit(
         SignalReadyEvent(
             timestamp=T0,
@@ -252,6 +254,41 @@ def test_vwap_gate_fail_open_on_missing_field(store):
     assert len(service.request_approvals_for_ready_signals()) == 1
 
 
+def test_anti_chase_skip_logic(store):
+    svc = make_service(store)   # defaults: ext 0.15, day-ext 0.30, halt-guard on
+    assert svc._anti_chase_skip(12.0, 10.0, None, False) == "over_extended"        # +20% > 15%
+    assert svc._anti_chase_skip(14.0, 14.0, 10.0, False) == "over_extended_day"    # +40% > 30%
+    assert svc._anti_chase_skip(10.0, 10.0, 10.0, True) == "halted_symbol"
+    assert svc._anti_chase_skip(10.0, 10.0, 9.0, False) is None                    # clean
+    assert svc._anti_chase_skip(14.0, 14.0, None, False) is None                   # fail-open day_open
+
+
+def test_unified_entry_blocks_parabolic_day_on_auto_path(store):
+    # entry 14 is +40% above the day open 10 -> day-extension gate blocks it
+    emit_signal(store, symbol="PARA", entry=14.0, stop=13.45, above_vwap=True, day_open=10.0)
+    created = make_service(store, unified_entry=True).request_approvals_for_ready_signals()
+    assert created == []
+    rr = [json.loads(e["payload_json"]) for e in store.query_events(event_type="risk_rule_triggered")
+          if json.loads(e["payload_json"])["rule_type"] == "over_extended_day"]
+    assert len(rr) == 1 and rr[0]["action_taken"] == "skipped_entry"
+
+
+def test_unified_entry_off_allows_parabolic_day(store):
+    emit_signal(store, symbol="PARA", entry=14.0, stop=13.45, above_vwap=True, day_open=10.0)
+    assert len(make_service(store, unified_entry=False).request_approvals_for_ready_signals()) == 1
+
+
+def test_unified_entry_fail_open_missing_day_open(store):
+    emit_signal(store, symbol="NODO", entry=14.0, stop=13.45, above_vwap=True)  # no day_open
+    assert len(make_service(store, unified_entry=True).request_approvals_for_ready_signals()) == 1
+
+
+def test_unified_entry_allows_normal_day(store):
+    # entry 14 is +7.7% above day open 13 -> under the 30% ceiling -> allowed
+    emit_signal(store, symbol="OKDAY", entry=14.0, stop=13.45, above_vwap=True, day_open=13.0)
+    assert len(make_service(store, unified_entry=True).request_approvals_for_ready_signals()) == 1
+
+
 def test_no_duplicate_requests_for_same_symbol(store):
     emit_signal(store)
     service = make_service(store)

trading_execution.py

@@ -86,6 +86,11 @@ class ExecutionSettings:
     # measurement; entries are only SKIPPED when this is True. Fail-open: a signal
     # with above_vwap=None (missing the field) is never blocked.
     require_above_vwap: bool = False
+    # UNIFIED ENTRY: run the fast path's shared anti-chase gates (over_extended,
+    # day-extension, halt) on the LIVE auto path too. These previously ran ONLY on
+    # the fast trigger path, leaving live auto entries unprotected (see the
+    # consolidation plan). Fail-open on missing data. False = pre-unification.
+    unified_entry: bool = False
     max_daily_loss_pct: float = 0.03
     # on a daily-loss breach, also flatten open positions + cancel unfilled entries
     flatten_on_breach: bool = True
@@ -225,6 +230,7 @@ class ExecutionSettings:
             concentrate_top_n=int(values.get("TRADING_CONCENTRATE_TOP_N", "0")),
             max_fresh_entries_per_day=int(values.get("TRADING_MAX_FRESH_ENTRIES_PER_DAY", "0")),
             require_above_vwap=flag("TRADING_REQUIRE_ABOVE_VWAP", "1"),
+            unified_entry=flag("TRADING_UNIFIED_ENTRY", "1"),
             liquidity_max_volume_pct=float(
                 values.get("TRADING_LIQUIDITY_MAX_VOLUME_PCT", "0.0")
             ),
@@ -679,6 +685,35 @@ class TradingExecutionService:
                     self._requested_symbols.add(symbol)  # don't re-evaluate every tick
                     continue
 
+            # Anti-chase / halt gates on the LIVE auto path (the entry-path
+            # unification — these ran ONLY on the fast trigger path before). The
+            # auto path rests a limit AT the trigger, so it can't chase above it
+            # (over_extended is a no-op here); the gate that bites is day-extension:
+            # don't enter a name whose breakout level is already parabolic vs the
+            # session open. halted=False (a resting limit won't fill mid-halt).
+            # Fail-open: a missing day_open skips only the day-extension check.
+            if self.settings.unified_entry:
+                entry_f0 = float(entry)
+                skip = self._anti_chase_skip(entry_f0, entry_f0,
+                                             signal.get("day_open"), False)
+                if skip is not None:
+                    self.store.emit(
+                        RiskRuleTriggeredEvent(
+                            timestamp=datetime.now(),
+                            mode=self.mode,
+                            correlation_id=self.session_id,
+                            message=(f"{symbol} entry blocked: {skip} "
+                                     f"(entry={entry}, day_open={signal.get('day_open')})"),
+                            rule_type=skip,
+                            rule_value=float(signal.get("day_open") or 0.0),
+                            current_state={"symbol": symbol, "entry": entry_f0,
+                                           "day_open": signal.get("day_open")},
+                            action_taken="skipped_entry",
+                        )
+                    )
+                    self._requested_symbols.add(symbol)  # don't re-evaluate every tick
+                    continue
+
             eq = self._current_equity()
             # cap this entry to BOTH the per-position limit AND the portfolio's
             # remaining gross-notional budget (so the book can't pile into ~100%).
@@ -805,6 +840,30 @@ class TradingExecutionService:
             "error": result.error,
         }
 
+    def _anti_chase_skip(self, entry_ref: float, trigger: float,
+                         day_open: float | None, halted: bool) -> str | None:
+        """Shared anti-chase / halt entry gates, used by BOTH the fast trigger path
+        and (behind TRADING_UNIFIED_ENTRY) the slow auto path. Returns a skip reason
+        or None. Fail-open: a missing/zero day_open skips only the day-extension
+        check. For a resting-limit fill (slow path) entry_ref==trigger so
+        over_extended is a no-op — day_extension + halt are the gates that bite."""
+        ext_max = self.settings.extension_max_pct
+        if (ext_max > 0 and trigger and float(trigger) > 0
+                and entry_ref > float(trigger) * (1.0 + ext_max)):
+            return "over_extended"
+        # Day-extension ceiling: the stock is already parabolic on the DAY (far
+        # above the session open) — chasing that is the "+100% then halts down"
+        # trap even when the trigger break itself looks clean.
+        day_max = self.settings.day_extension_max_pct
+        if (day_max > 0 and day_open is not None and float(day_open) > 0
+                and entry_ref > float(day_open) * (1.0 + day_max)):
+            return "over_extended_day"
+        # Halt guard: you can't exit during a trading halt and it gaps through the
+        # stop on resume — don't enter a name flagged as halted.
+        if halted and self.settings.halt_guard_enabled:
+            return "halted_symbol"
+        return None
+
     @_locked
     def submit_breakout_now(
         self,
@@ -884,20 +943,9 @@ class TradingExecutionService:
         # the top tick that reverses/halts down. (Overnight +100% gappers are
         # already excluded upstream by the trigger book's gap_max; this catches
         # the intraday chase / gap-through that gap_max can't see.)
-        ext_max = self.settings.extension_max_pct
-        if ext_max > 0 and entry_ref > trigger_f * (1.0 + ext_max):
-            return {"ok": False, "skipped": "over_extended"}
-        # Day-extension ceiling: the stock is already parabolic on the DAY (far
-        # above the session open) — chasing that is the "+100% then halts down"
-        # trap even when the trigger break itself looks clean.
-        day_max = self.settings.day_extension_max_pct
-        if (day_max > 0 and day_open is not None and float(day_open) > 0
-                and entry_ref > float(day_open) * (1.0 + day_max)):
-            return {"ok": False, "skipped": "over_extended_day"}
-        # Halt guard: you can't exit during a trading halt and it gaps through the
-        # stop on resume — don't enter a name the caller flagged as halted.
-        if halted and self.settings.halt_guard_enabled:
-            return {"ok": False, "skipped": "halted_symbol"}
+        skip = self._anti_chase_skip(entry_ref, trigger_f, day_open, halted)
+        if skip is not None:
+            return {"ok": False, "skipped": skip}
 
         eq = self._current_equity()
         liq = self.settings.liquidity_max_volume_pct