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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01B3PXXn2tHBzdv6rYtGPz86
This commit is contained in:
1 parent
c518747387
commit
347031e2bb
2 files changed
+73
-1
No files matched your search
+71
-1
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user