From b3a47d775dc0c1df8c2cda0e45c9891fdf597118 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Sun, 19 Jul 2026 08:44:10 -0500 Subject: [PATCH] feat(019): discovery on its own thread + tuner heartbeat (honest stale alert) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Telemetry: discovery took 6-30s ON the single scheduler, freezing ingest/arm so the trigger read stale bars — the real 'late edge detection'. Now discovery runs on its own daemon thread with a dedicated DB connection; the watchlist it mutates is a lock-guarded Watchlist (list-like drop-in, no torn reads); the scheduler discover step auto-disables (single-writer). Loop is now 3 fast threads (entries/exits/discovery) + slow scheduler for tolerant work. Also: nightly_tune writes a heartbeat every run, so the boot alert distinguishes 'tuner not running' (real) from 'ran, no profitable ORB config, kept params' (expected — the strategy has no edge). Kills the misleading Telegram alarm. 543 tests green. Co-Authored-By: Claude Fable 5 --- deploy/docker/truenas-compose.yml | 1 + docker-compose.yml | 1 + nightly_tune.py | 19 ++++++ plans/README.md | 2 +- run_live_paper.py | 86 +++++++++++++++++++++------- runtime/watchlist.py | 68 ++++++++++++++++++++++ tests/unit/runtime/test_watchlist.py | 58 +++++++++++++++++++ 7 files changed, 214 insertions(+), 21 deletions(-) create mode 100644 runtime/watchlist.py create mode 100644 tests/unit/runtime/test_watchlist.py diff --git a/deploy/docker/truenas-compose.yml b/deploy/docker/truenas-compose.yml index 1137d4a..8394446 100644 --- a/deploy/docker/truenas-compose.yml +++ b/deploy/docker/truenas-compose.yml @@ -79,6 +79,7 @@ x-app: &app 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_THREAD_ENABLED: '1' + DISCOVER_THREAD_ENABLED: '1' TRADING_EXIT_THREAD_INTERVAL_SECONDS: '1.0' TRADING_EXIT_TRAIL_R_STEP: '0.25' TRADING_LIQUIDITY_MAX_VOLUME_PCT: '0.025' diff --git a/docker-compose.yml b/docker-compose.yml index bf3a06f..51a14d1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -51,6 +51,7 @@ x-app-env: &app-env 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_THREAD_ENABLED: '1' + DISCOVER_THREAD_ENABLED: '1' TRADING_EXIT_THREAD_INTERVAL_SECONDS: '1.0' TRADING_EXIT_TRAIL_R_STEP: "0.25" TRADING_EXIT_BREAKEVEN_R: "1.0" diff --git a/nightly_tune.py b/nightly_tune.py index f474366..b6286f0 100644 --- a/nightly_tune.py +++ b/nightly_tune.py @@ -27,9 +27,25 @@ from research.multi_schema import open_research_db load_dotenv() +def _write_heartbeat(outcome: str) -> None: + """Record that the tuner RAN (regardless of whether it found a profitable + config), so the boot staleness check can tell 'tuner not running' (real + problem) from 'tuner ran, no profitable ORB config, kept params' (expected + when the strategy has no edge — the common case). Best-effort.""" + try: + path = os.path.join(os.environ.get("DATA_DIR", "./data"), "tuner_heartbeat.json") + os.makedirs(os.path.dirname(path) or ".", exist_ok=True) + with open(path, "w") as f: + json.dump({"ran_at": datetime.now(timezone.utc).isoformat(), + "outcome": outcome}, f) + except Exception: # noqa: BLE001 — a heartbeat must never break tuning + pass + + def main(): # Tune the profitable setup only. os.environ["STRATEGY_SETUPS"] = "opening_range_break" + _write_heartbeat("started") # 1) accumulate the latest sessions (best-effort; never abort tuning on this) try: @@ -45,6 +61,7 @@ def main(): sessions = load_sessions(con) print(f"tuning over {len(sessions)} gapper sessions", flush=True) if len(sessions) < 10: + _write_heartbeat("kept:too_few_sessions") print("too few sessions — keeping current learned_params.json") return @@ -71,6 +88,7 @@ def main(): best = (p, m, score) if best is None: + _write_heartbeat("kept:no_profitable_config") print("no profitable ORB config on this sample — keeping current params") return @@ -97,6 +115,7 @@ def main(): f.flush() os.fsync(f.fileno()) os.replace(tmp, path) + _write_heartbeat("wrote:profitable") print(f"wrote {path}:\n{json.dumps(out, indent=2)}", flush=True) diff --git a/plans/README.md b/plans/README.md index e644878..738e1f4 100644 --- a/plans/README.md +++ b/plans/README.md @@ -35,7 +35,7 @@ your row when done. | 016 | EDGAR dilution enrichment — unbreak silent nightly refresh (UTC current_date bug + missing float_cache DDL) + pre-register dilution_veto_v1 keep-or-zero experiment | P2 | M | 006 | **DONE (2026-07-17, review-APPROVED first pass; f6999c7 merged). iv_dilution_gate keep-or-zero (fail-open on unfetched symbols, early results biased toward zero never inflated), dilution_veto_v1 self-registers via the experiments/*.json glob, nightly adjudication live. Prod backfill seeded same day (304 symbols swept, 245 filings / 91 symbols). FIRST ADJUDICATION 2026-07-17: **FAIL** (n=2044, expR_delta +0.0012 vs +0.05 bar) — but only 64/2069 fires (3%) had filing coverage (backfill covers recent actives only; historical fires read unflagged by fail-open design), AND the flagged fires ran BETTER than unflagged (baseline −0.038 vs −0.377, n=64): early evidence points AGAINST the dilution-trap thesis, not merely underpowered. Bar-design lesson: min_n counted ALL fires not flagged ones. Verdict terminal (no goalpost-moving); filings keep accruing nightly. IF revisited: dilution_veto_v2 with a flagged-fire min_n metric, after ~3-4 weeks coverage.** | | 017 | L2 order-book depth recorder **[VERDICT 2026-07-18: order-book overhang is REAL but too weak to trade. n=2191 under the real deployed exit: high-overhang fires -0.676R vs low -0.759R = +0.083R separation, SAME direction as the sniff (held from n=23 to n=2191 — genuinely real, not noise). BUT both buckets deeply negative; gating fails the random-gate null (delta +0.28 was mostly exposure-shrink). It reduces losses, does NOT create profit or beat a random gate. Live gap>=10% band: n=13/74, no signal. FAIL under the pre-registered bar. Same shape as the regime gate: real modest signal on a negative-EV base = less-bad, not positive. depth_features(2202) + code retained for a possible broader-population redesign; NOT wired live. Databento spend ~$16 free credits. REDESIGN LEAD REFUTED 2026-07-18: gap x overhang grid under real exit shows lower-gap is the WORST cell (0-3% -0.831R), not a hidden edge; NO positive cell exists anywhere. Exhaustive search complete — no positive subpopulation in gap-ORB.]** — the one untested selection axis; FREE-first (Phase 0-1 = $0 sniff test + recorder), paid feed only after the data earns it, pre-registered verdict before any live use | P1 | L | 006 | TODO (owner-approved to plan, no spend yet) | | 018 | Sub-second exit management on a dedicated thread | P1 | M | — | **DONE (5896d29): exits on own 1s thread + dedicated conn, scheduler step auto-off, idempotent. Also shipped: fast pipeline (ingest 5s/arm 8s/trigger 2s) + emit self-heal.** | -| 019 | Move discovery off the scheduler thread — telemetry shows it blocks 6-30s and staleness the whole loop (the real 'late edge detection' cause) | P1 | M | 018 | TODO | +| 019 | Move discovery off the scheduler thread | P1 | M | 018 | **DONE: discovery on own thread + dedicated conn + thread-safe Watchlist; scheduler discover step auto-off; single-writer. Also: tuner heartbeat fixes the false 'tuner not running' alert (it runs, finds no profitable config = expected).** | Status values: TODO | IN PROGRESS | DONE | BLOCKED (with one-line reason) | REJECTED (with one-line rationale) diff --git a/run_live_paper.py b/run_live_paper.py index 52f6090..b7f83de 100644 --- a/run_live_paper.py +++ b/run_live_paper.py @@ -56,6 +56,7 @@ from research.ingestion.news_enrichment import enrich_recent_news from config import OllamaConfig from research.multi_schema import open_research_db from runtime.exit_manager import LiveExitManager +from runtime.watchlist import Watchlist from runtime.halt_guard import is_halt_gap from runtime.market_calendar import ( days_to_next_session, @@ -150,14 +151,33 @@ def _load_learned(now: datetime | None = None) -> dict: age = _learned_params_age_days(learned.get("as_of"), now) max_age = float(os.environ.get("LEARNED_PARAMS_MAX_AGE_DAYS", "3")) if age is not None and age > max_age: - msg = (f"learned_params is {age:.0f} days stale (as_of " - f"{str(learned.get('as_of'))[:10]}) — tuner not running?") - print(f"[boot] WARNING: {msg}", flush=True) + # Distinguish a REAL failure (tuner not running) from the EXPECTED case + # (tuner runs nightly but finds no profitable ORB config, so it keeps + # the old params — the common state when the strategy has no edge). The + # tuner writes a heartbeat every run regardless of outcome. + hb_age = None try: - from runtime.notifier import send_telegram - send_telegram(f"⚠️ {msg}") - except Exception: # noqa: BLE001 — never let a note block boot - pass + hb_path = os.path.join(os.environ.get("DATA_DIR", "./data"), "tuner_heartbeat.json") + if os.path.exists(hb_path): + with open(hb_path) as fh: + hb_age = _learned_params_age_days(json.load(fh).get("ran_at"), now) + except Exception: # noqa: BLE001 + hb_age = None + if hb_age is not None and hb_age <= max_age: + # tuner IS running — params are old only because nothing profitable + # was found. Informational, not an alarm. + msg = (f"tuner running (last {hb_age:.0f}d ago) but no profitable ORB " + f"config found — params kept (as_of {str(learned.get('as_of'))[:10]})") + print(f"[boot] note: {msg}", flush=True) + else: + msg = (f"learned_params {age:.0f}d stale AND tuner heartbeat " + f"{'missing' if hb_age is None else f'{hb_age:.0f}d old'} — tuner NOT running") + print(f"[boot] WARNING: {msg}", flush=True) + try: + from runtime.notifier import send_telegram + send_telegram(f"⚠️ {msg}") + except Exception: # noqa: BLE001 — never let a note block boot + pass return learned @@ -433,7 +453,7 @@ def build_runtime(args: argparse.Namespace) -> dict: ) return { - "symbols": symbols, + "symbols": Watchlist(symbols), # plan 019: thread-safe (discovery mutates on its own thread) "symbols_source": symbols_source, # per-symbol IEX->consolidated volume correction (research.ingestion.volume_ratio): # measured each morning from the delayed free SIP feed, applied at ingest so live @@ -501,7 +521,7 @@ def main(argv: list[str] | None = None) -> int: args = parse_args(argv) rt = build_runtime(args) ollama_cfg = rt["ollama_cfg"] - symbols: list[str] = rt["symbols"] + symbols: Watchlist = rt["symbols"] client = rt["client"] print(f"[boot] session={rt['session_id']} db={_db_target_desc()}") @@ -522,8 +542,7 @@ def main(argv: list[str] | None = None) -> int: provider = rt["watcher"].provider if hasattr(provider, "add_symbol"): provider.add_symbol(symbol) - if symbol not in symbols: - symbols.append(symbol) + symbols.append(symbol) # Watchlist dedupes state = DashboardState( rt["event_db"], @@ -549,10 +568,10 @@ def main(argv: list[str] | None = None) -> int: ) if discovered: print(f"[boot] screener added: {', '.join(discovered)}") - symbols = sorted(set(symbols) | set(discovered)) + symbols.replace(sorted(set(symbols.snapshot()) | set(discovered))) else: print("[boot] screener unavailable — using static symbol list") - daily = ingest_daily_history(rt["research_con"], client, symbols, days=30) + daily = ingest_daily_history(rt["research_con"], client, list(symbols), days=30) print(f"[boot] daily backfill: {daily.daily_rows} rows" + (f" (errors: {daily.errors})" if daily.errors else "")) @@ -794,11 +813,11 @@ def main(argv: list[str] | None = None) -> int: done = command_consumer.poll() return f"operator commands executed: {len(done)}" if done else None - def step_discover(): + def step_discover(con=None): if not client: return "skipped (no keys)" res = run_discovery( - rt["store"], rt["research_con"], client, _now_session_date(), + rt["store"], con or rt["research_con"], client, _now_session_date(), mode=EventMode.PAPER, correlation_id=rt["session_id"], price_min=float(os.environ.get("WATCHER_PRICE_MIN", "1")), price_max=float(os.environ.get("WATCHER_PRICE_MAX", "20")), @@ -806,14 +825,14 @@ def main(argv: list[str] | None = None) -> int: ) # fold newly-screened names into the live watchlist so the next ingest # pulls their bars and the watcher evaluates them - added = [s for s in res.universe if s not in symbols] - symbols.extend(added) + before = set(symbols.snapshot()) + symbols.extend(res.universe) + added = [s for s in res.universe if s not in before] # Keep the watchlist BOUNDED so it rotates toward current movers instead # of growing all session. (Ingestion is now batched so an over-cap list # no longer 400s, but an unbounded list is still wasteful and stale.) max_uni = int(os.environ.get("WATCHER_MAX_UNIVERSE", "80")) - if len(symbols) > max_uni: - del symbols[: len(symbols) - max_uni] + symbols.trim(max_uni) return (f"universe={len(res.universe)} gappers={len(res.gappers)}" + (f" +{len(added)} new" if added else "")) @@ -1277,9 +1296,14 @@ def main(argv: list[str] | None = None) -> int: scheduler = Scheduler() scheduler.add("news", step_news, float(os.environ.get("ALPACA_NEWS_INTERVAL_SECONDS", "120")), enabled=_flag("ALPACA_NEWS_ENABLED", "1")) + # SINGLE-WRITER (plan 019): the scheduler discover step runs ONLY when the + # dedicated discovery thread is OFF. Discovery takes 6-30s (telemetry) and + # on the scheduler it froze ingest/arm -> the trigger read stale bars. The + # thread moves that block off the time-critical path. Thread default ON. scheduler.add("discover", step_discover, float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")), - enabled=_flag("DISCOVER_ENABLED", "1")) + enabled=(_flag("DISCOVER_ENABLED", "1") + and not _flag("DISCOVER_THREAD_ENABLED", "1"))) scheduler.add("vratio", step_vratio, float(os.environ.get("VOLUME_RATIO_INTERVAL_SECONDS", "300")), enabled=_flag("VOLUME_RATIO_ENABLED", "1")) scheduler.add("ingest", step_ingest, float(os.environ.get("LIVE_BARS_INTERVAL_SECONDS", "60")), @@ -1369,6 +1393,28 @@ def main(argv: list[str] | None = None) -> int: 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") + # discovery on its OWN thread (plan 019): its 6-30s pass no longer blocks + # ingest/arm, so the trigger thread never reads stale bars. Own DB + # connection (shared facade isn't thread-safe); the watchlist it mutates is + # a thread-safe Watchlist; scheduler discover step disabled when this runs. + if _flag("DISCOVER_THREAD_ENABLED", "1") and _flag("DISCOVER_ENABLED", "1") and client is not None: + from runtime.exit_thread import run_exit_loop as _run_interval_loop # generic driver + _disc_interval = float(os.environ.get("DISCOVER_INTERVAL_SECONDS", "300")) + _disc_con: dict = {"c": None} + + def discover_loop(): + try: + _disc_con["c"] = open_research_db("market") + except Exception as exc: # noqa: BLE001 + print(f"[discover-thread] dedicated DB unavailable, using shared: {exc}", flush=True) + + class _DiscStep: + def manage(_self): # adapt to the generic loop driver + step_discover(con=_disc_con["c"]) + _run_interval_loop(_DiscStep(), stop_event, _disc_interval) + + threading.Thread(target=discover_loop, name="discover", daemon=True).start() + print(f"[boot] discovery thread started (every {_disc_interval:.0f}s) — scheduler discover step OFF") # news enrichment on its OWN thread (never blocks the scheduler's trade steps) if ollama_cfg.enabled and ollama_cfg.enrichment_enabled: threading.Thread(target=enrich_loop, name="enrich", daemon=True).start() diff --git a/runtime/watchlist.py b/runtime/watchlist.py new file mode 100644 index 0000000..a87de04 --- /dev/null +++ b/runtime/watchlist.py @@ -0,0 +1,68 @@ +"""Thread-safe watchlist (plan 019). + +Discovery runs on its own thread (so its 6-30s pass never blocks ingest/arm) +and mutates the live watchlist while the scheduler's ingest/arm/watch steps +iterate it. A bare list mutated across threads yields torn reads / "list +changed during iteration". This is a small lock-guarded, list-like drop-in: +every read (len, ``in``, iteration, indexing/slicing) takes a consistent +snapshot under the lock; mutations (append/extend/trim) hold the lock. Order +is preserved and adds dedupe (a name already watched isn't re-added).""" + +from __future__ import annotations + +import threading +from collections.abc import Iterable + + +class Watchlist: + def __init__(self, initial: Iterable[str] | None = None): + self._lock = threading.Lock() + self._items: list[str] = [] + if initial: + self.extend(initial) + + # -- reads (each snapshots under the lock) ---------------------------- + def snapshot(self) -> tuple[str, ...]: + with self._lock: + return tuple(self._items) + + def __len__(self) -> int: + with self._lock: + return len(self._items) + + def __contains__(self, item) -> bool: + with self._lock: + return item in self._items + + def __iter__(self): + # iterate a snapshot copy so a concurrent mutation can't tear the loop + return iter(self.snapshot()) + + def __getitem__(self, idx): + with self._lock: + return self._items[idx] # supports symbols[:N] slicing (arm) + + # -- mutations (locked, dedupe-preserving-order) ---------------------- + def append(self, name: str) -> None: + with self._lock: + if name not in self._items: + self._items.append(name) + + def extend(self, names: Iterable[str]) -> None: + with self._lock: + seen = set(self._items) + for n in names: + if n not in seen: + self._items.append(n) + seen.add(n) + + def trim(self, max_len: int) -> None: + """Bound to the most-recent ``max_len`` names (drops oldest).""" + with self._lock: + if len(self._items) > max_len: + del self._items[: len(self._items) - max_len] + + def replace(self, names: Iterable[str]) -> None: + """Atomically swap the whole list (boot-time universe resolution).""" + with self._lock: + self._items = list(dict.fromkeys(names)) diff --git a/tests/unit/runtime/test_watchlist.py b/tests/unit/runtime/test_watchlist.py new file mode 100644 index 0000000..e54385e --- /dev/null +++ b/tests/unit/runtime/test_watchlist.py @@ -0,0 +1,58 @@ +"""Thread-safe watchlist (plan 019): discovery mutates it on its own thread +while ingest/arm iterate it on the scheduler thread. A list-like object that +locks internally, so existing readers (len, `in`, `for`) work unchanged and +every read sees a consistent snapshot — never a torn/half-updated list.""" + +import threading + +from runtime.watchlist import Watchlist + + +def test_listlike_reads(): + w = Watchlist(["AAA", "BBB"]) + assert len(w) == 2 + assert "AAA" in w and "ZZZ" not in w + assert list(w) == ["AAA", "BBB"] + assert w.snapshot() == ("AAA", "BBB") + + +def test_add_dedupes_and_preserves_order(): + w = Watchlist(["AAA"]) + w.extend(["BBB", "AAA", "CCC"]) # AAA already present -> not duplicated + assert list(w) == ["AAA", "BBB", "CCC"] + w.append("DDD") + assert "DDD" in w + + +def test_trim_keeps_most_recent(): + w = Watchlist(["A", "B", "C", "D", "E"]) + w.trim(3) # bound to newest 3 + assert list(w) == ["C", "D", "E"] + w.trim(10) # no-op when under cap + assert len(w) == 3 + + +def test_concurrent_mutation_never_tears(): + w = Watchlist([f"S{i}" for i in range(20)]) + stop = threading.Event() + errs = [] + + def churn(): + i = 0 + while not stop.is_set(): + w.extend([f"N{i}"]); w.trim(30); i += 1 + + def read(): + try: + for _ in range(20000): + n = len(w) + assert n == len(list(w)) # snapshot self-consistent + _ = "S0" in w + except Exception as e: # noqa: BLE001 + errs.append(e) + finally: + stop.set() + + t = threading.Thread(target=churn); t.start() + read(); t.join() + assert not errs