Triggers: price-staleness guard — don't fire on a frozen quote

ArmedTrigger stamps price_ts on each update; update_price(None) (a failed batch
price fetch) now CLEARS the price instead of leaving the last value live; and
fires() requires the price to be fresher than max_price_age_s (env
TRIGGER_PRICE_MAX_AGE=12). A stalled feed can no longer fire a breakout into a
name that has already reverted.

Tests: stale price blocks fire, fresh re-enables, None clears; 18 pass.

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

run_live_paper.py

@@ -267,6 +267,7 @@ def build_runtime(args: argparse.Namespace) -> dict:
         rvol_min=float(os.environ.get("TRIGGER_RVOL_MIN", "2.0")),
         min_range_pct=float(os.environ.get("TRIGGER_MIN_RANGE_PCT", "0.004")),
         min_dollar_vol=float(os.environ.get("TRIGGER_MIN_DOLLAR_VOL", "0")),
+        max_price_age_s=float(os.environ.get("TRIGGER_PRICE_MAX_AGE", "12")),
     )
 
     return {

runtime/triggers.py

@@ -25,6 +25,7 @@ from __future__ import annotations
 
 import functools
 import threading
+import time
 from dataclasses import dataclass, field
 
 WAITING = "waiting"
@@ -58,6 +59,7 @@ class ArmedTrigger:
     stop: float | None = None     # opening-range low, cushioned (protective stop)
     range_pct: float = 0.0        # (high-low)/high — how wide the opening range is
     price: float | None = None    # latest live trade price
+    price_ts: float = 0.0         # monotonic time of that price (staleness guard)
     cum_volume: float = 0.0       # session cumulative volume (for liquidity sizing)
     dollar_vol: float = 0.0       # recent $-volume (liquidity gate: skip thin spikes)
     catalyst: str = ""            # fresh news headline (if any) — why it's hot
@@ -104,6 +106,10 @@ class ArmedTriggerBook:
     # qualifies — thin names then fill at the top of a one-bar spike and reverse
     # (observed live: WKSP/LNKS/UWMC entered on 100-400 share bars). 0 = off.
     min_dollar_vol: float = 0.0
+    # max age (s) of the last price before a trigger may fire. A stalled feed
+    # would otherwise let a name fire on a frozen quote into a reverted market.
+    # 0 = off.
+    max_price_age_s: float = 0.0
     triggers: dict[str, ArmedTrigger] = field(default_factory=dict)
     _lock: object = field(default_factory=threading.RLock, repr=False, compare=False)
 
@@ -162,12 +168,20 @@ class ArmedTriggerBook:
     @_synced
     def update_price(self, symbol: str, price: float | None) -> None:
         t = self.triggers.get(symbol)
-        if t is not None and price is not None:
+        if t is None:
+            return
+        if price is None:
+            t.price = None       # failed fetch -> clear, so we can't fire on stale
+        else:
             t.price = float(price)
+            t.price_ts = time.monotonic()
 
     @_synced
     def fires(self) -> list[ArmedTrigger]:
-        """Armed triggers whose live price has reached or crossed the trigger."""
+        """Armed triggers whose live price has reached or crossed the trigger —
+        and whose price is fresh (a stalled feed must not fire on a frozen quote)."""
+        now = time.monotonic()
+        max_age = self.max_price_age_s
         return [
             t
             for t in self.triggers.values()
@@ -175,6 +189,7 @@ class ArmedTriggerBook:
             and t.price is not None
             and t.trigger is not None
             and t.price >= t.trigger
+            and (max_age <= 0 or (now - t.price_ts) <= max_age)
         ]
 
     @_synced

tests/unit/test_triggers.py

@@ -102,6 +102,21 @@ def test_fires_only_when_price_crosses():
     assert [t.symbol for t in b.fires()] == ["AAA"]
 
 
+def test_fire_blocked_when_price_is_stale():
+    """A stalled feed must not fire on a frozen quote — fires() requires the
+    price to be fresher than max_price_age_s."""
+    b = ArmedTriggerBook(gap_min=3, rvol_min=2, max_price_age_s=5)
+    b.arm([_cand("AAA", trig=10.0)])
+    b.update_price("AAA", 10.5)                 # fresh cross -> fires
+    assert [t.symbol for t in b.fires()] == ["AAA"]
+    b.triggers["AAA"].price_ts -= 10            # age the price past the window
+    assert b.fires() == []                      # stale -> no fire
+    b.update_price("AAA", 10.5)                 # fresh again -> fires
+    assert [t.symbol for t in b.fires()] == ["AAA"]
+    b.update_price("AAA", None)                 # failed fetch clears the price
+    assert b.fires() == [] and b.triggers["AAA"].price is None
+
+
 def test_fired_state_pinned_across_rearm():
     b = ArmedTriggerBook(max_armed=2, gap_min=3, rvol_min=2)
     b.arm([_cand("AAA"), _cand("BBB")])