From 347031e2bb3bd74678023adc9bbb822c96b9a3b1 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Tue, 29 Sep 2026 21:21:42 -0500 Subject: [PATCH] feat: judge paper trader survives a crash mid-session On 2026-09-29 the trader's console was killed at 10:21 ET (exit 0xC000013A) and it sat dead for the rest of the day. The scheduled task now re-fires every 5 min from 08:10 to 15:00 local, and IgnoreNew makes that a no-op while the trader is running. It also catches up if the machine was off at 08:10. A relaunched trader now picks up where the dead one left off: - positions still held at the broker are adopted, with the entry, original stop, fill time and high-water taken from today's journal and the backfilled tape, so the trail, time stop and 15:55 flat still apply. The working stop leg is reused; if none is working a new one is armed. A position with no known stop is closed rather than guessed at. - the per-symbol order count is rebuilt from the journal, so the cap holds across restarts. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01B3PXXn2tHBzdv6rYtGPz86 --- judge/live.py | 72 +++++++++++++++++++++++++++++++++++++++++++++- judge/run_live.cmd | 2 ++ 2 files changed, 73 insertions(+), 1 deletion(-) diff --git a/judge/live.py b/judge/live.py index d1b29b8..3c6d3e5 100644 --- a/judge/live.py +++ b/judge/live.py @@ -123,6 +123,7 @@ class Trader: "APCA_API_BASE_URL": "https://paper-api.alpaca.markets"})) self.art = load_model() self.day = now_et().strftime("%Y-%m-%d") + self.earlier = self._earlier_today() self.j = Journal(self.day) cfg = SnapshotScanConfig.from_env() cfg.price_min, cfg.price_max, cfg.min_pct, cfg.max_per_sweep = 1.0, 20.0, 20.0, 10 @@ -133,6 +134,70 @@ class Trader: self.subscribed: set = set() self.j.log("start", cutoff=self.art["cutoff"], risk_usd=RISK_USD, max_notional=MAX_NOTIONAL, dry=DRY) + def _earlier_today(self) -> dict: + """What earlier runs of today's session did (the task relaunches the trader if it dies): + orders per symbol, so the per-symbol cap holds across restarts, and each symbol's last fill.""" + out = {"orders": {}, "fills": {}} + f = OUT / "journal" / f"{self.day}.jsonl" + if not f.exists(): + return out + for line in f.read_text(encoding="utf-8").splitlines(): + try: + r = json.loads(line) + except ValueError: + continue + if r.get("kind") == "order": + out["orders"][r["symbol"]] = out["orders"].get(r["symbol"], 0) + 1 + elif r.get("kind") == "fill": + out["fills"][r["symbol"]] = r + return out + + async def adopt(self): + """Take over positions left by an earlier run today, so they still get the trail, time stop and EOD flat. + A position with no known stop (no journalled fill and no working stop order) is closed rather than guessed at.""" + try: + held = await asyncio.to_thread(self.client.get_positions, True) + except Exception as e: # noqa: BLE001 + self.j.log("error", where="adopt", err=str(e)[:300]); return + for x in held: + sym, qty, avg = x["symbol"], int(float(x["qty"])), float(x["avg_entry_price"]) + if qty <= 0: + continue + fill = self.earlier["fills"].get(sym) + try: + orders = await asyncio.to_thread(self.client.get_orders, "open", 50, True, [sym]) + except Exception: # noqa: BLE001 + orders = [] + stops = [o for o in orders if o.get("side") == "sell" and o.get("type") == "stop"] + for o in orders: + stops += [lg for lg in (o.get("legs") or []) if lg.get("side") == "sell" and lg.get("type") == "stop" + and lg.get("status") in ("new", "accepted", "held", "pending_new")] + stop = float(stops[0]["stop_price"]) if stops else (float(fill["stop"]) if fill else None) + if stop is None: + self.j.log("adopt_close", symbol=sym, qty=qty, why="no known stop") + try: + await asyncio.to_thread(self.client.close_position, sym) + except Exception as e: # noqa: BLE001 + self.j.log("error", where="adopt_close", symbol=sym, err=str(e)[:300]) + continue + base = float(fill["stop"]) if fill else stop # R is measured from the original stop, not a trailed one + risk = avg - base if avg > base else 0.03 * avg + fill_time = pd.Timestamp(fill["at"]) if fill else now_et() + try: + await self.add_symbol(sym, float(x.get("lastday_price") or avg)) + except Exception as e: # noqa: BLE001 + self.j.log("error", where="adopt_add", symbol=sym, err=str(e)[:300]) + pz = Pos(sym, qty, avg, stop, risk, fill_time, "", stops[0]["id"] if stops else None, avg, + fill.get("setup", "?") if fill else "?", float(fill.get("p") or 0) if fill else 0.0) + bk = self.books.get(sym) + if bk and bk.tt: # high-water since the fill, from the backfilled tape + k = np.searchsorted(np.array(bk.tt, dtype="datetime64[ns]"), np.datetime64(fill_time), side="left") + if bk.tp[k:]: + pz.hw = max(avg, max(bk.tp[k:])) + self.pos[sym] = pz + self.j.log("adopt", symbol=sym, qty=qty, entry=avg, stop=stop, leg_id=pz.leg_id, hw=pz.hw, fill_time=fill_time) + await self._reconcile(pz, bk.tp[-1] if bk and bk.tp else avg) + # --------------------------------------------------------------------- discovery + backfill def _paged(self, path, key, sym, start_utc, end_utc): rows, tok = [], None @@ -168,6 +233,7 @@ class Trader: return ctx = await asyncio.to_thread(self._context, sym, prev_close) bk = Book(ctx); self.books[sym] = bk + bk.n_trades = self.earlier["orders"].get(sym, 0) await self.subscribe([sym]) start = min(s for s, _ in ctx.windows) s_utc = start.tz_localize(ET).tz_convert("UTC").strftime("%Y-%m-%dT%H:%M:%SZ") @@ -409,7 +475,10 @@ class Trader: while True: n = now_et(); tick += 1 for sym, pz in list(self.pos.items()): - bk = self.books[sym] + bk = self.books.get(sym) + if bk is None: # adopted position whose backfill failed: manage it off the entry price + bk = self.books[sym] = Book(DayContext(sym, self.day, pz.entry, None, None, [])) + bk.ready = True if tick % 5 == 0 and await self._reconcile(pz, bk.tp[-1] if bk.tp else pz.entry): continue k = np.searchsorted(np.array(bk.tt[-5000:], dtype="datetime64[ns]"), np.datetime64(pz.fill_time), side="left") @@ -458,6 +527,7 @@ class Trader: self.j.log("stop", why="not a trading day"); return tasks = [asyncio.create_task(self.stream_loop()), asyncio.create_task(self.scan_loop()), asyncio.create_task(self.decide_loop())] + await self.adopt() await self.manage_loop() for t in tasks: t.cancel() diff --git a/judge/run_live.cmd b/judge/run_live.cmd index e0c3df2..6d5a1fe 100644 --- a/judge/run_live.cmd +++ b/judge/run_live.cmd @@ -1,6 +1,8 @@ @echo off rem Paper trial of the v2 judgement model (Alpaca paper "Momentum V2"). Started by Task Scheduler at 08:10 local rem (09:10 ET); the process exits by itself after 16:00 ET. Log: data\judge\logs\live_YYYY-MM-DD.log +rem The task re-fires every 5 min until 15:00 local (IgnoreNew): if the trader died it is relaunched, and it adopts +rem any open position and today's per-symbol order counts from the journal. setlocal cd /d "%~dp0.." set PYTHONIOENCODING=utf-8