feat(018): sub-second exits on a dedicated thread + fast decision pipeline
Exits move off the single-threaded scheduler onto their own daemon thread at 1s (mirrors the fast trigger thread). Dedicated research connection (shared facade isn't thread-safe); scheduler exits step auto-disabled when the thread runs (single-caller invariant — no racing instance state, no double-submit); manage() already idempotent (stop ratchets >1c up only). 3 driver tests + 539 suite green. On fast small-caps this is the 'get out fast' the owner asked for. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
1 parent
375973fe21
commit
5896d291d8
5 files changed
+139
-2
No files matched your search
@@ -78,6 +78,8 @@ x-app: &app
|
|||||||
TRADING_EXIT_TRAIL_AFTER_R: '1.0'
|
TRADING_EXIT_TRAIL_AFTER_R: '1.0'
|
||||||
TRADING_EXIT_TRAIL_MODE: 'prior_low'
|
TRADING_EXIT_TRAIL_MODE: 'prior_low'
|
||||||
TRADING_EXIT_MANAGE_INTERVAL_SECONDS: '4' # 12->4s: react to fast small-cap round-trips (was giving back gains)
|
TRADING_EXIT_MANAGE_INTERVAL_SECONDS: '4' # 12->4s: react to fast small-cap round-trips (was giving back gains)
|
||||||
|
TRADING_EXIT_THREAD_ENABLED: '1'
|
||||||
|
TRADING_EXIT_THREAD_INTERVAL_SECONDS: '1.0'
|
||||||
TRADING_EXIT_TRAIL_R_STEP: '0.25'
|
TRADING_EXIT_TRAIL_R_STEP: '0.25'
|
||||||
TRADING_LIQUIDITY_MAX_VOLUME_PCT: '0.025'
|
TRADING_LIQUIDITY_MAX_VOLUME_PCT: '0.025'
|
||||||
TRADING_MAX_BACKOUTS_PER_SYMBOL: '3'
|
TRADING_MAX_BACKOUTS_PER_SYMBOL: '3'
|
||||||
|
|||||||
@@ -50,6 +50,8 @@ x-app-env: &app-env
|
|||||||
TRADING_EXIT_TRAIL_MODE: "prior_low"
|
TRADING_EXIT_TRAIL_MODE: "prior_low"
|
||||||
TRADING_EXIT_TRAIL_AFTER_R: "1.0"
|
TRADING_EXIT_TRAIL_AFTER_R: "1.0"
|
||||||
TRADING_EXIT_MANAGE_INTERVAL_SECONDS: '4' # 12->4s: react to fast small-cap round-trips (was giving back gains)
|
TRADING_EXIT_MANAGE_INTERVAL_SECONDS: '4' # 12->4s: react to fast small-cap round-trips (was giving back gains)
|
||||||
|
TRADING_EXIT_THREAD_ENABLED: '1'
|
||||||
|
TRADING_EXIT_THREAD_INTERVAL_SECONDS: '1.0'
|
||||||
TRADING_EXIT_TRAIL_R_STEP: "0.25"
|
TRADING_EXIT_TRAIL_R_STEP: "0.25"
|
||||||
TRADING_EXIT_BREAKEVEN_R: "1.0"
|
TRADING_EXIT_BREAKEVEN_R: "1.0"
|
||||||
TRADING_EXIT_PROFIT_TIERS: "8:3,15:9,25:18,40:30"
|
TRADING_EXIT_PROFIT_TIERS: "8:3,15:9,25:18,40:30"
|
||||||
|
|||||||
+32
-2
@@ -1085,9 +1085,17 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
# --- live exit management (breakeven / trail / scale / first-red) -------
|
# --- live exit management (breakeven / trail / scale / first-red) -------
|
||||||
exit_cfg = ExitConfig.from_env()
|
exit_cfg = ExitConfig.from_env()
|
||||||
|
|
||||||
|
# Exits run on their OWN thread (plan 018) with a DEDICATED research
|
||||||
|
# connection — the shared rt["research_con"] is not thread-safe, and the
|
||||||
|
# exit loop reads bars concurrently with the scheduler. Open it lazily so a
|
||||||
|
# DB hiccup at boot doesn't stop the loop; None falls back to the shared
|
||||||
|
# connection (still correct, just not concurrency-isolated).
|
||||||
|
_exit_con: dict = {"c": None}
|
||||||
|
|
||||||
def _rth_bars(sym):
|
def _rth_bars(sym):
|
||||||
from research import query as rq
|
from research import query as rq
|
||||||
b = rq.query_minute_bars(rt["research_con"], sym, _now_session_date())
|
con = _exit_con["c"] or rt["research_con"]
|
||||||
|
b = rq.query_minute_bars(con, sym, _now_session_date())
|
||||||
if b is not None and len(b) and "is_regular_hours" in b.columns:
|
if b is not None and len(b) and "is_regular_hours" in b.columns:
|
||||||
b = b[b["is_regular_hours"] == True].reset_index(drop=True) # noqa: E712
|
b = b[b["is_regular_hours"] == True].reset_index(drop=True) # noqa: E712
|
||||||
return b
|
return b
|
||||||
@@ -1291,8 +1299,12 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
scheduler.add("command", step_commands,
|
scheduler.add("command", step_commands,
|
||||||
float(os.environ.get("OPERATOR_COMMAND_INTERVAL_SECONDS", "5")),
|
float(os.environ.get("OPERATOR_COMMAND_INTERVAL_SECONDS", "5")),
|
||||||
enabled=(rt["execution"] is not None))
|
enabled=(rt["execution"] is not None))
|
||||||
|
# SINGLE-CALLER INVARIANT (plan 018): the scheduler exits step runs ONLY when
|
||||||
|
# the dedicated exit thread is OFF — otherwise manage() would be called from
|
||||||
|
# two threads (racing instance state + double-submit). Thread default ON.
|
||||||
scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")),
|
scheduler.add("exits", step_manage_exits, float(os.environ.get("TRADING_EXIT_MANAGE_INTERVAL_SECONDS", "12")),
|
||||||
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
enabled=(_flag("TRADING_EXECUTION_ENABLED", "1")
|
||||||
|
and not _flag("TRADING_EXIT_THREAD_ENABLED", "1")))
|
||||||
scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")),
|
scheduler.add("eod", step_eod_flatten, float(os.environ.get("TRADING_EOD_INTERVAL_SECONDS", "30")),
|
||||||
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
enabled=_flag("TRADING_EXECUTION_ENABLED", "1"))
|
||||||
scheduler.add("notify", step_notify, float(os.environ.get("TELEGRAM_NOTIFY_INTERVAL_SECONDS", "45")),
|
scheduler.add("notify", step_notify, float(os.environ.get("TELEGRAM_NOTIFY_INTERVAL_SECONDS", "45")),
|
||||||
@@ -1339,6 +1351,24 @@ def main(argv: list[str] | None = None) -> int:
|
|||||||
trigger_thread = threading.Thread(target=trigger_loop, name="trigger", daemon=True)
|
trigger_thread = threading.Thread(target=trigger_loop, name="trigger", daemon=True)
|
||||||
trigger_thread.start()
|
trigger_thread.start()
|
||||||
print(f"[boot] fast trigger thread started (every {trigger_interval:.0f}s)")
|
print(f"[boot] fast trigger thread started (every {trigger_interval:.0f}s)")
|
||||||
|
# exits on their OWN thread (plan 018): sub-second reaction, independent of
|
||||||
|
# the scheduler queue. Gated so the slow scheduler "exits" step is disabled
|
||||||
|
# whenever this runs (single-caller invariant — see below).
|
||||||
|
_exit_thread_on = (_flag("TRADING_EXIT_THREAD_ENABLED", "1")
|
||||||
|
and exit_mgr is not None and client is not None)
|
||||||
|
if _exit_thread_on:
|
||||||
|
from runtime.exit_thread import run_exit_loop
|
||||||
|
_exit_interval = float(os.environ.get("TRADING_EXIT_THREAD_INTERVAL_SECONDS", "1.0"))
|
||||||
|
|
||||||
|
def exit_loop():
|
||||||
|
try: # dedicated connection so bars reads don't race the scheduler
|
||||||
|
_exit_con["c"] = open_research_db("market")
|
||||||
|
except Exception as exc: # noqa: BLE001 — fall back to shared con
|
||||||
|
print(f"[exit-thread] dedicated DB unavailable, using shared: {exc}", flush=True)
|
||||||
|
run_exit_loop(exit_mgr, stop_event, _exit_interval)
|
||||||
|
|
||||||
|
threading.Thread(target=exit_loop, name="exits", daemon=True).start()
|
||||||
|
print(f"[boot] exit thread started (every {_exit_interval:.1f}s) — scheduler exits step OFF")
|
||||||
# news enrichment on its OWN thread (never blocks the scheduler's trade steps)
|
# news enrichment on its OWN thread (never blocks the scheduler's trade steps)
|
||||||
if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
|
if ollama_cfg.enabled and ollama_cfg.enrichment_enabled:
|
||||||
threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()
|
threading.Thread(target=enrich_loop, name="enrich", daemon=True).start()
|
||||||
|
|||||||
@@ -0,0 +1,43 @@
|
|||||||
|
"""Sub-second exit management on a dedicated daemon thread (plan 018).
|
||||||
|
|
||||||
|
Exits used to run as a step on the single-threaded main scheduler, so real
|
||||||
|
reaction drifted past the interval whenever a slower step (ingest, discover,
|
||||||
|
news) held the thread. On a $2-20 small-cap that round-trips in seconds, that
|
||||||
|
gave back gains. This runs LiveExitManager.manage() on its OWN thread, at a
|
||||||
|
~1s cadence, independent of the scheduler queue — the symmetric twin of the
|
||||||
|
fast trigger thread that already handles entries.
|
||||||
|
|
||||||
|
Two invariants make this safe (audited 2026-07-19):
|
||||||
|
* SINGLE CALLER — the scheduler's "exits" step is removed when the thread
|
||||||
|
runs, so manage() is invoked from exactly one thread; its mutable instance
|
||||||
|
state (_tracked / _naked_passes / _empty_snapshot_streak) never races.
|
||||||
|
* IDEMPOTENT manage() — the stop ratchet only replaces when the desired stop
|
||||||
|
is >1c ABOVE the resting one (exit_manager.py), so a 1s cadence re-submits
|
||||||
|
nothing once a level is set; catastrophe/scale close the position, so the
|
||||||
|
next pass sees it gone. The manager must also get its OWN research DB
|
||||||
|
connection for bars (the shared facade connection is not thread-safe).
|
||||||
|
"""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import Callable
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def run_exit_loop(exit_mgr, stop_event, interval: float = 1.0,
|
||||||
|
on_error: Callable[[Exception], None] | None = None) -> None:
|
||||||
|
"""Drive ``exit_mgr.manage()`` every ``interval`` seconds until
|
||||||
|
``stop_event`` is set. Mirrors ``trigger_loop``: waits FIRST (so a
|
||||||
|
pre-set stop returns immediately without a manage call), and a manage()
|
||||||
|
exception is caught — a live exit loop must never die on a broker blip and
|
||||||
|
leave positions unmanaged."""
|
||||||
|
while not stop_event.wait(interval):
|
||||||
|
try:
|
||||||
|
exit_mgr.manage()
|
||||||
|
except Exception as exc: # noqa: BLE001 — one bad pass must not kill exits
|
||||||
|
if on_error is not None:
|
||||||
|
on_error(exc)
|
||||||
|
else:
|
||||||
|
logger.warning("exit pass failed (continuing): %s", exc)
|
||||||
@@ -0,0 +1,60 @@
|
|||||||
|
"""Plan 018 — sub-second exit management on its own thread.
|
||||||
|
|
||||||
|
The driver (run_exit_loop) must: call manage() every interval, survive a
|
||||||
|
manage() exception without dying (a live exit loop that crashes leaves
|
||||||
|
positions unmanaged), and stop promptly when the stop_event is set."""
|
||||||
|
|
||||||
|
import threading
|
||||||
|
import time
|
||||||
|
|
||||||
|
from runtime.exit_thread import run_exit_loop
|
||||||
|
|
||||||
|
|
||||||
|
def test_loop_calls_manage_repeatedly_then_stops():
|
||||||
|
calls = {"n": 0}
|
||||||
|
stop = threading.Event()
|
||||||
|
|
||||||
|
class Mgr:
|
||||||
|
def manage(self):
|
||||||
|
calls["n"] += 1
|
||||||
|
if calls["n"] >= 3:
|
||||||
|
stop.set() # end after 3 passes
|
||||||
|
return []
|
||||||
|
|
||||||
|
run_exit_loop(Mgr(), stop, interval=0.01)
|
||||||
|
assert calls["n"] == 3
|
||||||
|
|
||||||
|
|
||||||
|
def test_loop_survives_a_manage_exception():
|
||||||
|
calls = {"n": 0}
|
||||||
|
stop = threading.Event()
|
||||||
|
errs = []
|
||||||
|
|
||||||
|
class Mgr:
|
||||||
|
def manage(self):
|
||||||
|
calls["n"] += 1
|
||||||
|
if calls["n"] == 1:
|
||||||
|
raise RuntimeError("broker blip") # must NOT kill the loop
|
||||||
|
if calls["n"] >= 3:
|
||||||
|
stop.set()
|
||||||
|
return []
|
||||||
|
|
||||||
|
run_exit_loop(Mgr(), stop, interval=0.01, on_error=errs.append)
|
||||||
|
assert calls["n"] == 3 # kept going after the exception
|
||||||
|
assert len(errs) == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_stop_event_ends_loop_without_calling_manage():
|
||||||
|
calls = {"n": 0}
|
||||||
|
stop = threading.Event()
|
||||||
|
stop.set() # already stopped
|
||||||
|
|
||||||
|
class Mgr:
|
||||||
|
def manage(self):
|
||||||
|
calls["n"] += 1
|
||||||
|
return []
|
||||||
|
|
||||||
|
t0 = time.monotonic()
|
||||||
|
run_exit_loop(Mgr(), stop, interval=5.0) # would hang 5s if it waited first
|
||||||
|
assert calls["n"] == 0
|
||||||
|
assert time.monotonic() - t0 < 1.0
|
||||||
Reference in new issue
Block a user