diff --git a/judge/live.py b/judge/live.py index 7fa2c04..d1b29b8 100644 --- a/judge/live.py +++ b/judge/live.py @@ -111,6 +111,7 @@ class Pos: hw: float setup: str p: float + next_try: pd.Timestamp | None = None # throttles exit retries after a broker error class Trader: @@ -333,29 +334,84 @@ class Trader: fq = int(float(filled["filled_qty"])); fpx = float(filled["filled_avg_price"]) legs = filled.get("legs") or [] leg_id = legs[0]["id"] if legs else None - self.pos[sym] = Pos(sym, fq, fpx, stop, fpx - stop, now_et(), oid, leg_id, fpx, d.setup, d.p) + self.pos[sym] = pz = Pos(sym, fq, fpx, stop, fpx - stop, now_et(), oid, leg_id, fpx, d.setup, d.p) self.j.log("fill", symbol=sym, qty=fq, price=fpx, stop=stop, leg_id=leg_id, slippage_vs_ask=fpx - entry) + if filled.get("status") != "filled" or not leg_id: + # a partial fill whose parent was cancelled takes its OTO stop leg down with it: re-arm a plain stop + await self._arm_stop(pz) - # --------------------------------------------------------------------- exits - async def exit_pos(self, pz: Pos, reason: str, last: float): + async def _arm_stop(self, pz: Pos): try: - if pz.leg_id: - await asyncio.to_thread(self.client.cancel_order, pz.leg_id) - await asyncio.to_thread(self.client.close_position, pz.symbol) + o = await asyncio.to_thread(self.client.submit_order, pz.symbol, pz.qty, "sell", "stop", "day", None, + round(pz.stop, 2), None, None, f"jv2s-{pz.symbol}-{now_et():%H%M%S}") + pz.leg_id = o.get("id") + self.j.log("stop_armed", symbol=pz.symbol, qty=pz.qty, stop=round(pz.stop, 2), leg_id=pz.leg_id) except Exception as e: # noqa: BLE001 - self.j.log("error", where="exit", symbol=pz.symbol, err=str(e)[:300]); return - self.j.log("exit", symbol=pz.symbol, reason=reason, last=last, entry=pz.entry, r_est=(last - pz.entry) / pz.risk) + pz.leg_id = None + self.j.log("error", where="arm_stop", symbol=pz.symbol, err=str(e)[:300]) + + async def _broker_qty(self, sym) -> int: + ps = await asyncio.to_thread(self.client.get_positions, True) + return sum(int(float(x["qty"])) for x in ps if x.get("symbol") == sym) + + def _closed(self, pz: Pos, reason: str, price: float): + self.j.log("exit", symbol=pz.symbol, reason=reason, price=price, entry=pz.entry, r_est=(price - pz.entry) / pz.risk) self.pos.pop(pz.symbol, None) bk = self.books.get(pz.symbol) if bk: bk.cooldown_until = now_et() + pd.Timedelta(seconds=engine.EXEC["cooldown_s"]) + async def _reconcile(self, pz: Pos, last: float) -> bool: + """Match our record to the broker. True if the position is gone (and has been recorded as closed). + If the stop leg is no longer working but shares remain, a fresh stop is armed.""" + leg = None + if pz.leg_id: + try: + leg = await asyncio.to_thread(self._order, pz.leg_id) + except Exception: # noqa: BLE001 + leg = None + if leg and leg.get("status") == "filled": + self._closed(pz, "stop", float(leg.get("filled_avg_price") or pz.stop)); return True + try: + held = await self._broker_qty(pz.symbol) + except Exception as e: # noqa: BLE001 + self.j.log("error", where="reconcile", symbol=pz.symbol, err=str(e)[:200]); return False + if held <= 0: + self._closed(pz, "flat_at_broker", last); return True + if held != pz.qty: + self.j.log("qty_sync", symbol=pz.symbol, ours=pz.qty, broker=held); pz.qty = held + if leg is None or leg.get("status") in ("canceled", "expired", "rejected", "replaced", "done_for_day"): + await self._arm_stop(pz) + return False + + # --------------------------------------------------------------------- exits + async def exit_pos(self, pz: Pos, reason: str, last: float): + if pz.next_try is not None and now_et() < pz.next_try: + return + if await self._reconcile(pz, last): # the broker stop may already have closed it + return + try: + if pz.leg_id: + try: + await asyncio.to_thread(self.client.cancel_order, pz.leg_id) + except Exception: # noqa: BLE001 (leg filled/cancelled in the meantime: close_position decides) + pass + await asyncio.to_thread(self.client.close_position, pz.symbol) + except Exception as e: # noqa: BLE001 + self.j.log("error", where="exit", symbol=pz.symbol, err=str(e)[:300]) + if not await self._reconcile(pz, last): + pz.next_try = now_et() + pd.Timedelta(seconds=5) + return + self._closed(pz, reason, last) + async def manage_loop(self): tick = 0 while True: n = now_et(); tick += 1 for sym, pz in list(self.pos.items()): bk = self.books[sym] + 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") recent = bk.tp[-5000:][k:] last = recent[-1] if recent else pz.entry @@ -375,16 +431,7 @@ class Trader: self.j.log("stop_move", symbol=sym, frm=pz.stop, to=round(new, 2)); pz.stop = new except Exception as e: # noqa: BLE001 self.j.log("error", where="replace", symbol=sym, err=str(e)[:200]) - if tick % 5 == 0 and pz.leg_id: - try: - leg = await asyncio.to_thread(self._order, pz.leg_id) - if leg.get("status") == "filled": - self.j.log("exit", symbol=sym, reason="stop", price=float(leg.get("filled_avg_price") or 0), - entry=pz.entry, r_est=(float(leg.get("filled_avg_price") or 0) - pz.entry) / pz.risk) - self.pos.pop(sym, None) - bk.cooldown_until = n + pd.Timedelta(seconds=engine.EXEC["cooldown_s"]) - except Exception: # noqa: BLE001 - pass + await self._reconcile(pz, last) if n.time() >= pd.Timestamp("16:00").time() and not self.pos: return await asyncio.sleep(1.0)