momentum devandClaude Opus 5.5
53725854f4
At the first exit-thread pass >= TRADING_EXIT_FAIL_FAST_SECONDS after the fill,
flatten if price has already traded TRADING_EXIT_FAIL_FAST_R x R below entry.
It is checked once per position. A position first seen after its window (a
restart or a carry) is never judged late. 0 seconds = off (the default).
Evidence, from tick data on 1,093 live-band v1 ORB breaks (Jun-Sep 2026) with
outcomes under the current exit rules:
- 60 s / 0.3R: +0.060R per trade Jun-Jul (90% CI +0.015..+0.102) and +0.062R
Aug-Sep (+0.029..+0.094). It fired on 29% of trades and cut 22 runners.
- Every threshold from -0.15R to -0.4R at 60 s was positive in both halves,
a plateau rather than one fitted cell.
- On our own 194 real fills (10-s bars): +0.015R. Same sign, not yet
significant.
It is deliberately narrow. The same study found that later "bail out below
+1R" rules kill runners: the LLM trade-manager pilot, the hybrid bailout, and
orderflow H1 (Jul 17). The first minute is the one window where failure tells
you something. Tape-flow features (aggressor buy share, print rate, big
prints, quote imbalance, spread) separated nothing (AUC 0.44-0.59, not stable
across halves).
Live-only: it needs sub-minute prices, so simulate_exit (minute bars) does not
model it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01E3S5EFKThFVrjmN4HHsHMN
100 lines
3.3 KiB
Python
100 lines
3.3 KiB
Python
"""Fail-fast: a breakout that dips >= fail_fast_r * R inside its first window is exited once."""
|
|
|
|
import pandas as pd
|
|
|
|
from runtime.exit_manager import LiveExitManager
|
|
from strategy.exits import ExitConfig
|
|
|
|
FILL = pd.Timestamp("2026-09-28 13:40:00") # UTC-naive, as _entry_time returns
|
|
|
|
|
|
class _Store:
|
|
def __init__(self):
|
|
self.events = []
|
|
|
|
def emit(self, e):
|
|
self.events.append(e)
|
|
|
|
|
|
class _Broker:
|
|
def __init__(self):
|
|
self.price = 10.0
|
|
self.closed = []
|
|
self.canceled = []
|
|
|
|
def get_positions(self):
|
|
return [{"symbol": "AAA", "avg_entry_price": "10.0", "qty": "100", "side": "long",
|
|
"current_price": str(self.price)}]
|
|
|
|
def get_orders(self, status="all", limit=500, nested=True, symbols=None):
|
|
return [{"id": "buy1", "symbol": "AAA", "side": "buy", "filled_qty": "100",
|
|
"filled_at": FILL.isoformat() + "Z", "status": "filled",
|
|
"legs": [{"id": "stop1", "symbol": "AAA", "type": "stop", "side": "sell",
|
|
"stop_price": "9.0", "status": "held"}]}]
|
|
|
|
def get_clock(self):
|
|
return {"is_open": True}
|
|
|
|
def cancel_order(self, oid):
|
|
self.canceled.append(oid)
|
|
|
|
def close_position(self, symbol, qty=None, percentage=None):
|
|
self.closed.append(symbol)
|
|
return {"id": "c"}
|
|
|
|
|
|
def _mgr(broker, clock, seconds=60.0, r=0.3):
|
|
cfg = ExitConfig(catastrophe_pct=0.0, enforce_stop_grace_passes=0, fail_fast_seconds=seconds,
|
|
fail_fast_r=r, default_trail_r_pct=0.0)
|
|
return LiveExitManager(broker, _Store(), lambda s: pd.DataFrame(), cfg=cfg,
|
|
session_id="t", now_fn=lambda: clock["now"])
|
|
|
|
|
|
def _run(prices_by_second, seconds=60.0, first_seen=5):
|
|
b = _Broker(); clock = {"now": FILL}
|
|
m = _mgr(b, clock, seconds)
|
|
for s, px in prices_by_second:
|
|
if s < first_seen:
|
|
continue
|
|
clock["now"] = FILL + pd.Timedelta(seconds=s); b.price = px
|
|
m.manage()
|
|
return b, m
|
|
|
|
|
|
def test_deep_early_dip_exits_once_at_the_window():
|
|
# R = 1.00 (entry 10, stop 9); dips to 9.65 (-0.35R) at 20 s, recovers, checked at 60 s
|
|
b, m = _run([(5, 10.0), (20, 9.65), (40, 9.9), (59, 10.02), (60, 10.05), (61, 10.1)])
|
|
assert b.closed == ["AAA"]
|
|
|
|
|
|
def test_shallow_dip_holds():
|
|
b, _ = _run([(5, 10.0), (20, 9.8), (60, 9.95), (75, 9.9)]) # -0.2R only
|
|
assert b.closed == []
|
|
|
|
|
|
def test_no_exit_before_the_window_ends():
|
|
b, _ = _run([(5, 10.0), (20, 9.5), (59, 9.5)])
|
|
assert b.closed == []
|
|
|
|
|
|
def test_position_first_seen_late_is_never_judged():
|
|
b, _ = _run([(200, 9.5), (201, 9.4)], first_seen=200)
|
|
assert b.closed == []
|
|
|
|
|
|
def test_off_by_default_does_nothing():
|
|
cfg = ExitConfig()
|
|
assert cfg.fail_fast_seconds == 0.0
|
|
b = _Broker(); b.price = 9.0
|
|
clock = {"now": FILL + pd.Timedelta(seconds=90)}
|
|
LiveExitManager(b, _Store(), lambda s: pd.DataFrame(), cfg=ExitConfig(catastrophe_pct=0.0,
|
|
enforce_stop_grace_passes=0, default_trail_r_pct=0.0),
|
|
now_fn=lambda: clock["now"]).manage()
|
|
assert b.closed == []
|
|
|
|
|
|
def test_env_parsing():
|
|
cfg = ExitConfig.from_env({"TRADING_EXIT_FAIL_FAST_SECONDS": "60", "TRADING_EXIT_FAIL_FAST_R": "0.3"})
|
|
assert cfg.fail_fast_seconds == 60.0 and cfg.fail_fast_r == 0.3
|
|
assert "fail-fast -0.3R@60s" in cfg.describe()
|