diff --git a/run_live_paper.py b/run_live_paper.py index 8f2cf48..08e2100 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -450,6 +450,7 @@ def build_runtime(args: argparse.Namespace) -> dict: 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")), + fresh_cross_tol=float(os.environ.get("TRIGGER_FRESH_CROSS_TOL", "0.015")), ) return { diff --git a/runtime/triggers.py b/runtime/triggers.py index 063d350..fb3e854 100644 --- a/runtime/triggers.py +++ b/runtime/triggers.py @@ -73,6 +73,12 @@ class ArmedTrigger: day_open: float | None = None # session-open price (for the day-extension gate) state: str = WAITING rank: int = 0 + # live price has been observed BELOW the trigger since arming — a fire is a + # fresh CROSS, not "price happens to stand above the level". Without this, + # any blind window (boot after a blocked spell, halt, feed outage) fires + # instantly on stale, already-extended breaks (SNXX 2026-07-21: bought a + # 15-min-old break +1.6% above trigger 18s after restart). + seen_below: bool = False def distance_pct(self) -> float | None: """How far live price is from the trigger, as a fraction. @@ -126,6 +132,11 @@ class ArmedTriggerBook: # would otherwise let a name fire on a frozen quote into a reverted market. # 0 = off. max_price_age_s: float = 0.0 + # fire without a seen-below cross only within this fraction ABOVE the trigger + # ("at the break" tolerance). Beyond it, require a pullback below the trigger + # and a fresh re-cross — stale breaks discovered after a blind window are not + # entries, they're chases. 0 = always require the cross. + fresh_cross_tol: float = 0.015 triggers: dict[str, ArmedTrigger] = field(default_factory=dict) _lock: object = field(default_factory=threading.RLock, repr=False, compare=False) @@ -209,6 +220,8 @@ class ArmedTriggerBook: else: t.price = float(price) t.price_ts = time.monotonic() + if t.trigger is not None and t.price < t.trigger: + t.seen_below = True @_synced def fires(self) -> list[ArmedTrigger]: @@ -223,6 +236,8 @@ class ArmedTriggerBook: and t.price is not None and t.trigger is not None and t.price >= t.trigger + and (t.seen_below + or t.price <= t.trigger * (1.0 + self.fresh_cross_tol)) and (max_age <= 0 or (now - t.price_ts) <= max_age) ] diff --git a/tests/unit/test_triggers.py b/tests/unit/test_triggers.py index 3b13ed4..7258ba1 100644 --- a/tests/unit/test_triggers.py +++ b/tests/unit/test_triggers.py @@ -107,6 +107,7 @@ def test_fire_blocked_when_price_is_stale(): 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", 9.9) # below the level (fresh-cross arm) 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 @@ -424,3 +425,37 @@ def test_book_falls_back_to_legacy_rvol_without_tod(): b = ArmedTriggerBook(gap_min=3, rvol_min=2) b.arm([_cand("LEGACY", rvol=3.0)]) # no rvol_tod key at all assert b.triggers["LEGACY"].state == ARMED + + +def _prime(b, sym): + b.arm([_cand(sym)]) + return b.triggers[sym] + + +def test_fire_requires_fresh_cross_not_stale_break(): + """SNXX 2026-07-21: after a blind window the price already stood +1.6% above + the trigger — that is a stale break, not an entry. No fire until price pulls + back below the trigger and re-crosses.""" + b = ArmedTriggerBook(gap_min=3, rvol_min=2, fresh_cross_tol=0.015) + _prime(b, "SNXX") + b.update_price("SNXX", 10.17) # first tick: +1.7% above trigger 10.0 + assert b.fires() == [] + b.update_price("SNXX", 9.95) # pullback below the level + b.update_price("SNXX", 10.02) # fresh re-cross + assert [t.symbol for t in b.fires()] == ["SNXX"] + + +def test_fire_allowed_at_the_break_within_tolerance(): + b = ArmedTriggerBook(gap_min=3, rvol_min=2, fresh_cross_tol=0.015) + _prime(b, "GAP") + b.update_price("GAP", 10.01) # +0.1%: AT the break, no cross seen + assert [t.symbol for t in b.fires()] == ["GAP"] + + +def test_fire_normal_cross_still_works(): + b = ArmedTriggerBook(gap_min=3, rvol_min=2) + _prime(b, "NORM") + b.update_price("NORM", 9.90) + assert b.fires() == [] + b.update_price("NORM", 10.05) + assert [t.symbol for t in b.fires()] == ["NORM"]