"""Live trading engine: post-only quotes via Lighter SignerClient.""" from __future__ import annotations import asyncio import json import logging import time import uuid from collections import deque from dataclasses import dataclass, field from pathlib import Path from typing import List, Optional, Tuple from lighter import AccountApi, SignerClient from .market_data import BookTop from .quoter import Quote, QuoteLadder, QuoteLevel from .rate_limit import get_limiter, is_rate_limit_error SEQUENTIAL_PACED_PLACE = True # one-by-one placement (no burst) PLACE_GAP_S = 2.0 # Denis: 2s/trade → ~30/min < API 37 from .session_persist import SessionStore, fill_to_dict, trim_for_ui log = logging.getLogger("lighter_mm.live") @dataclass class Fill: id: str ts: float side: str # buy | sell price: float size_btc: float fee_usd: float mid_at_fill: float @dataclass class OpenLot: id: str side: str price: float size_btc: float notional_usd: float ts: float @dataclass class ClosedTrade: id: str ts: float side: str entry_price: float exit_price: float size_btc: float pnl_usd: float fee_usd: float profitable: bool @dataclass class LiveState: equity0: float cash_usd: float inventory_btc: float = 0.0 realized_pnl: float = 0.0 fills: List[Fill] = field(default_factory=list) open_lots: List[OpenLot] = field(default_factory=list) closed_trades: List[ClosedTrade] = field(default_factory=list) day_start_ts: float = field(default_factory=time.time) day_start_equity: float = 0.0 quote: Optional[Quote] = None last_mid: float = 0.0 prev_mid: float = 0.0 buy_notional: float = 0.0 buy_qty: float = 0.0 sell_notional: float = 0.0 sell_qty: float = 0.0 price_history: deque = field(default_factory=lambda: deque(maxlen=600)) # 5m OHLC; forming candle updates every book tick (realtime tip) candles: deque = field(default_factory=lambda: deque(maxlen=288)) # ~24h available_balance: float = 0.0 collateral: float = 0.0 unrealized_pnl: float = 0.0 pending_orders: int = 0 last_error: str = "" ladder_orders: int = 0 ladder_bids: int = 0 ladder_asks: int = 0 last_refresh_reason: str = "" def __post_init__(self) -> None: if not self.day_start_equity: self.day_start_equity = self.equity0 def _to_int_amount(value: float, decimals: int) -> int: return int(round(value * (10**decimals))) def _from_str_amount(s: str | float | int | None) -> float: if s is None: return 0.0 try: return float(s) except (TypeError, ValueError): return 0.0 def infer_fill_price( *, delta: float, inv: float, avg_entry: float, prev_inv: float, prev_avg: float, mid: float, max_mid_dev: float = 0.02, ) -> float: """Infer execution price from inventory change (not exchange avg_entry alone). Sell fills must NOT use avg_entry — exchange keeps the entry average while inventory is still open, so sells would wrongly match the buy price. """ size = abs(delta) if size <= 1e-12: return 0.0 px = 0.0 if delta < 0: # Sell / reduce long / add short: prefer mid if mid > 0: px = mid elif avg_entry > 0: px = avg_entry else: # Buy: reverse-engineer when adding to / opening a long if prev_inv >= -1e-12 and inv > 1e-12 and avg_entry > 0: if abs(prev_inv) <= 1e-12: px = avg_entry elif prev_inv > 0 and prev_avg > 0: px = (avg_entry * inv - prev_avg * prev_inv) / delta if px <= 0: if mid > 0: px = mid elif avg_entry > 0: px = avg_entry if px > 0 and mid > 0 and abs(px - mid) / mid > max_mid_dev: # Insane vs mid → fall back px = mid return px # Rate-window policy for ladder refresh (Lighter soft-cap ~37/60s). RATE_REFRESH_HEADROOM = 8 # more slack — mid races burned 36/37 # Hard mid-drift wipe threshold. Was 8.0 — BTC often prints 8–12 bps and # toxic_* → cancel_all left 0+0 for ~30–40s under PLACE_GAP (Denis t192u). MAX_STALE_KEEP_BPS = 22.0 # Force cancel+replace when working_mid drifts this far from market mid. # Gap between rest_threshold (~2) and MAX_STALE (22) used to soft_keep forever # (incl. soft_keep_toxic), freezing quotes at 8–20 bps until restart. # Toxic micro-through alone (moved < this) still soft_keeps — strategy #1. FORCE_RECENTER_BPS = 14.0 # less touch churn on mid races (was 10) # Through-market threshold (bps). 0.5 thrashed: place→toxic~1.5→clear→empty. TOXIC_EPS_BPS = 2.5 POST_PLACE_GRACE_S = 30.0 # Adverse mid-cross: NEVER triggers cancel_all (soft_keep only). Threshold kept # high so we don't even attempt touch churn on join-flicker. ADVERSE_CROSS_MIN_BPS = 15.0 ADVERSE_CROSS_GRACE_S = 30.0 # Full cancel_all rebuild ONLY for mid_moved ≥ MAX_STALE (22), with this cooldown. FULL_REBUILD_COOLDOWN_S = 90.0 REBUILD_RATE_NEED = 5 # Min gap between incremental touch recenters (no cancel_all). TOUCH_RECENTER_COOLDOWN_S = 60.0 # Large mid-cross / FORCE_RECENTER drift: shorter CD so L0 is not toxic for 60s. TOUCH_RECENTER_LARGE_COOLDOWN_S = 12.0 # Thru/drift threshold for short touch CD. FORCE_RECENTER stays 14 (less churn); # 8 bps catches the live freeze band (12–18 bps) without 60s toxic L0. TOUCH_RECENTER_LARGE_THRU_BPS = 8.0 # Target ladder depth (config n_*_levels is 8). Incomplete → side_trim topup. LADDER_TARGET_LEVELS = 8 SIDE_TRIM_TOPUP_COOLDOWN_S = 15.0 def decide_rate_constrained_refresh( *, in_window: int, max_per_window: int, working_n_levels: int, reason: str, mid: float, working_mid: float, working_bid0: float, working_ask0: float, rest_threshold_bps: float, max_stale_keep_bps: float = MAX_STALE_KEEP_BPS, force_recenter_bps: float = FORCE_RECENTER_BPS, rate_headroom: int = RATE_REFRESH_HEADROOM, toxic_eps_bps: float = TOXIC_EPS_BPS, seconds_since_place: float = 1e9, post_place_grace_s: float = POST_PLACE_GRACE_S, ) -> str: """Anti-thrash rate gate. PIN: MAX_STALE=22 FORCE_RECENTER=14 GRACE=30 TOXIC=2.5 GAP=2.0 CD=90 - empty book → proceed / defer_place (never idle) - force (initial/post_fill/risk/adverse_cross/side_trim) → proceed - mid_moved with stale ≥ FORCE_RECENTER may wipe (recenter frozen quotes) - extreme mid-stale (≥ MAX_STALE) may wipe; rate full → clear_stale - toxic_* / through-market NEVER wipe by toxic reason alone (Denis t192u) - adverse_cross_* (bid>=mid or ask<=mid) ALWAYS force proceed (post-only) - soft_refresh / soft_keep: ALWAYS defer_keep - other mild mid reasons → defer_keep even with rate room """ max_pw = int(max_per_window or 37) in_w = int(in_window or 0) rate_ok = in_w < max_pw - int(rate_headroom) reason_s = str(reason or "") is_risk = reason_s.startswith("risk_side") is_adverse = reason_s.startswith("adverse_cross") # legacy; no longer emitted as wipe is_touch = reason_s == "touch_recenter" force_proceed = ( reason_s in ("initial_place", "post_fill_replenish", "ladder_cleared", "side_trim", "touch_recenter") or reason_s.startswith("side_trim") or reason_s.startswith("adverse_cross") or is_risk ) stale_bps = 1e9 if working_mid > 0 and mid > 0: stale_bps = abs(mid - working_mid) / working_mid * 10_000.0 # through_bps kept for API/compat metrics only — NEVER drives hard wipe. through_bps = 0.0 if mid > 0: eps = mid * float(toxic_eps_bps) / 10_000.0 if working_bid0 > 0 and working_bid0 >= mid + eps: through_bps = max(through_bps, (working_bid0 - mid) / mid * 10_000.0) if working_ask0 > 0 and working_ask0 <= mid - eps: through_bps = max(through_bps, (mid - working_ask0) / mid * 10_000.0) _ = through_bps # intentionally unused for wipe decisions (Denis t192u) # mid_moved_* with meaningful drift must proceed (else decide deferred # soft_keep while _should_refresh returned need=True — freeze bug). # FORCE_RECENTER band (≥10) also proceeds — wipe only happens at MAX_STALE # in _should_refresh; decide must not freeze a needed refresh. mid_force = ( reason_s.startswith("mid_moved_") and stale_bps >= float(force_recenter_bps) ) hard_stale = ( stale_bps >= float(max_stale_keep_bps) or mid_force or is_risk or reason_s == "ladder_cleared" ) _ = is_adverse # kept for API/compat; never drives wipe # Empty first: rate full → defer_place; never force-proceed into 37/37. if int(working_n_levels) <= 0: return "proceed" if rate_ok else "defer_place" if force_proceed: if rate_ok or reason_s == "post_fill_replenish": return "proceed" # Book resting + no rate room → KEEP (never defer_place duplicates / 0+0 cancel) if int(working_n_levels) > 0 and ( is_adverse or reason_s.startswith("mid_moved_") or reason_s.startswith("side_trim") ): return "defer_keep" return "defer_place" # Toxic through-market: NEVER wipe / clear_stale (empty book thrash). if reason_s.startswith("toxic_") or reason_s.startswith("soft_keep_toxic"): _ = rest_threshold_bps return "defer_keep" if float(seconds_since_place) < float(post_place_grace_s) and not is_risk: if hard_stale: return "proceed" if rate_ok else "clear_stale" return "defer_keep" # Only hard mid-stale / risk may wipe. Soft refresh never wipes. if hard_stale or is_risk: return "proceed" if rate_ok else "clear_stale" # soft_refresh/soft_keep: never wipe (empty book thrash under PLACE_GAP). if reason_s.startswith("soft_refresh") or reason_s.startswith("soft_keep"): _ = rest_threshold_bps return "defer_keep" _ = rest_threshold_bps # unused for non-soft mild; API compat return "defer_keep" class LiveEngine: """Places real post-only bid/ask; state mirrors PaperEngine for risk/status.""" MAX_CLOSE_EVENTS = 30 def __init__( self, client: SignerClient, account_index: int, api_key_index: int, market_id: int, price_decimals: int, size_decimals: int, equity_usd: float, ): self.client = client self.account_index = account_index self.api_key_index = api_key_index self.market_id = market_id self.price_decimals = price_decimals self.size_decimals = size_decimals self.state = LiveState(equity0=equity_usd, cash_usd=equity_usd) self._client_order_seq = int(time.time() * 1000) % 1_000_000_000 self._bid_order_index: Optional[int] = None self._ask_order_index: Optional[int] = None self._working_mid: float = 0.0 self._working_bid0: float = 0.0 self._working_ask0: float = 0.0 self._working_allow_bid: Optional[bool] = None self._working_allow_ask: Optional[bool] = None self._working_n_levels: int = 0 self._working_levels: list[dict] = [] # resting ladder for status/UI self._last_side_trim_ts: float = 0.0 self._quote_target: Optional[Quote] = None # desired ladder (may differ while placing) self._last_place_ts: float = 0.0 self._last_full_rebuild_ts: float = 0.0 # cooldown for cancel_all rebuilds self._cancel_all_count: int = 0 self._place_batch_count: int = 0 self._rest_skip_count: int = 0 self._replenish_after_fill: bool = False # 1-for-1 replace: fills per ladder side awaiting ONE replacement each. self._fills_to_replace: dict = {"bid": 0, "ask": 0} self._prev_inventory: float = 0.0 self._prev_avg_entry: float = 0.0 self._boot_synced: bool = False # first refresh = sync inv, no fake fill self._day_equity_restored: bool = False self._persisted_close_events: List[dict] = [] data_dir = Path(__file__).resolve().parents[1] / "data" data_dir.mkdir(parents=True, exist_ok=True) self._fills_log = data_dir / "fills_live.jsonl" self._candles_path = data_dir / "candles_5m.json" self._session_store = SessionStore(data_dir / "session_state.json") self._session = self._session_store.load_or_rotate() # Recover same-day fills from jsonl if session file was empty seeded = self._session_store.seed_from_jsonl( self._session, self._fills_log ) if seeded: self._session_store.save(self._session) self._apply_session_to_state() self._load_candles() def set_quote(self, q: Optional[Quote]) -> None: self.state.quote = q def _levels_to_status(self, levels: List[QuoteLevel]) -> list[dict]: out: list[dict] = [] for lv in levels: out.append( { "side": lv.side, "level": int(lv.level), "price": float(lv.price), "size_btc": float(lv.size_btc), "notional_usd": float(lv.price) * float(lv.size_btc), } ) return out def _mark_working_placed( self, levels: List[QuoteLevel], ladder: QuoteLadder, n_placed: int, allow_bid: bool, allow_ask: bool, ) -> None: """Record resting ladder after a successful place (source of truth for UI).""" n_bid = sum(1 for lv in levels if lv.side == "bid") n_ask = sum(1 for lv in levels if lv.side == "ask") self.state.ladder_orders = int(n_placed) self.state.pending_orders = int(n_placed) self.state.ladder_bids = n_bid self.state.ladder_asks = n_ask self.state.quote = ladder.as_quote() self._quote_target = ladder.as_quote() self._working_levels = self._levels_to_status(levels) self._working_mid = float(ladder.mid or 0.0) # Best resting from actual placed levels (not desired-only) bids = [lv.price for lv in levels if lv.side == "bid"] asks = [lv.price for lv in levels if lv.side == "ask"] self._working_bid0 = float(max(bids) if bids else (ladder.bid_px or 0.0)) self._working_ask0 = float(min(asks) if asks else (ladder.ask_px or 0.0)) if self.state.quote is not None: if self._working_bid0 > 0: self.state.quote.bid_px = self._working_bid0 if self._working_ask0 > 0: self.state.quote.ask_px = self._working_ask0 self._working_n_levels = len(levels) self._working_allow_bid = allow_bid self._working_allow_ask = allow_ask def ladder_levels_for_status(self) -> list[dict]: return list(self._working_levels) def quote_target_for_status(self) -> Optional[dict]: q = self._quote_target if q is None: return None return { "bid_px": q.bid_px, "ask_px": q.ask_px, "size_btc": q.size_btc, "skew_bps": q.skew_bps, } def _apply_session_to_state(self) -> None: """Hydrate in-memory fills / day baseline from persisted session.""" st = self.state sess = self._session st.fills = [] for d in sess.fills: try: st.fills.append( Fill( id=str(d.get("id") or uuid.uuid4().hex[:12]), ts=float(d["ts"]), side=str(d["side"]), price=float(d["price"]), size_btc=float(d["size_btc"]), fee_usd=float(d.get("fee_usd") or 0.0), mid_at_fill=float(d.get("mid_at_fill") or d["price"]), ) ) except Exception: continue st.buy_notional = float(sess.buy_notional or 0.0) st.buy_qty = float(sess.buy_qty or 0.0) st.sell_notional = float(sess.sell_notional or 0.0) st.sell_qty = float(sess.sell_qty or 0.0) if sess.day_start_ts: st.day_start_ts = float(sess.day_start_ts) if sess.day_start_equity and sess.day_start_equity > 0: st.day_start_equity = float(sess.day_start_equity) st.equity0 = float(sess.day_start_equity) self._day_equity_restored = True self._persisted_close_events = list(sess.close_events or []) log.info( "session hydrated: fills=%s buy=%s sell=%s day_equity0=%.2f restored=%s", len(st.fills), sess.fills_buy, sess.fills_sell, st.day_start_equity, self._day_equity_restored, ) def persist_day_equity(self, equity: float) -> None: """Set day baseline once per UTC day (skip if already restored).""" if self._day_equity_restored and self.state.day_start_equity > 0: return if equity <= 0: return self.state.day_start_equity = equity self.state.equity0 = equity if not self._session.day_start_ts: self._session.day_start_ts = time.time() self._session.day_start_equity = equity self._day_equity_restored = True self._session_store.save(self._session) def _persist_session(self) -> None: sess = self._session sess.fills = [fill_to_dict(f) for f in self.state.fills] sess.buy_notional = self.state.buy_notional sess.buy_qty = self.state.buy_qty sess.sell_notional = self.state.sell_notional sess.sell_qty = self.state.sell_qty sess.day_start_equity = self.state.day_start_equity sess.day_start_ts = self.state.day_start_ts typed = [] if self.state.closed_trades: typed = [ { "id": c.id, "pnl_usd": round(c.pnl_usd, 6), "size_btc": c.size_btc, "ts": c.ts, "profitable": c.profitable, "side": c.side, "entry_price": c.entry_price, "exit_price": c.exit_price, } for c in self.state.closed_trades[-self.MAX_CLOSE_EVENTS :] ] if typed: sess.close_events = typed elif self._persisted_close_events: sess.close_events = list(self._persisted_close_events) self._session_store.save(sess) def fills_buy_count(self) -> int: return sum(1 for f in self.state.fills if f.side == "buy") def fills_sell_count(self) -> int: return sum(1 for f in self.state.fills if f.side == "sell") def last_fills_for_status(self, n: int | None = None) -> List[dict]: """Return fills for dashboard Trades: last ~5h, capped ~500.""" rows = [fill_to_dict(f) for f in self.state.fills] ui = trim_for_ui(rows) if n is not None and n > 0: return ui[-n:] return ui def _next_client_order_index(self) -> int: self._client_order_seq = (self._client_order_seq + 1) % 9_000_000_000_000_000_000 return self._client_order_seq async def _auth(self) -> str: token, err = self.client.create_auth_token_with_expiry() if err or not token: raise RuntimeError(f"auth token failed: {err}") return token async def _maybe_429_backoff(self, err: object) -> None: """On rare 429: short backoff via shared limiter (not a long brake).""" if not is_rate_limit_error(err): return lim = get_limiter() if lim is not None: await lim.on_429() else: log.warning("429 without limiter installed — sleep 1.5s") import asyncio await asyncio.sleep(1.5) async def refresh_account(self) -> None: api = AccountApi(self.client.api_client) try: accs = await api.account(by="index", value=str(self.account_index)) except Exception as e: await self._maybe_429_backoff(e) raise if not accs.accounts: raise RuntimeError("account not found") a = accs.accounts[0] avail = _from_str_amount(a.available_balance) coll = _from_str_amount(a.collateral) tav = _from_str_amount(a.total_asset_value) self.state.available_balance = avail self.state.collateral = coll acct_pending = int(a.pending_order_count or 0) # Lighter often reports pending_order_count=0 while limits are open. if acct_pending > 0: self.state.pending_orders = acct_pending elif self.state.ladder_orders > 0: self.state.pending_orders = self.state.ladder_orders else: self.state.pending_orders = acct_pending inv = 0.0 avg_entry = 0.0 upnl = 0.0 rpnl = 0.0 for p in a.positions or []: if int(getattr(p, "market_id", -1)) != self.market_id: continue size = _from_str_amount(p.position) sign = int(p.sign or 1) inv = size * sign # long positive, short negative avg_entry = _from_str_amount(p.avg_entry_price) upnl = _from_str_amount(p.unrealized_pnl) rpnl = _from_str_amount(p.realized_pnl) break # Boot: adopt exchange inventory without inventing a fill. if not self._boot_synced: self._prev_inventory = inv self._prev_avg_entry = avg_entry self._boot_synced = True log.info( "boot inventory sync inv=%+.6f avg_entry=%.1f (no fill recorded)", inv, avg_entry, ) elif abs(inv - self._prev_inventory) > 1e-9: # Real inventory change since last refresh → record fill delta = inv - self._prev_inventory side = "buy" if delta > 0 else "sell" size = abs(delta) mid = self.state.last_mid or 0.0 px = infer_fill_price( delta=delta, inv=inv, avg_entry=avg_entry, prev_inv=self._prev_inventory, prev_avg=self._prev_avg_entry, mid=mid, ) log.info( "fill_price side=%s delta=%+.6f mid=%.1f avg_entry=%.1f " "prev_avg=%.1f recorded_px=%.1f", side, delta, mid, avg_entry, self._prev_avg_entry, px, ) if px > 0 and size > 0: self._record_fill(side, px, size, mid if mid > 0 else px) self._prev_inventory = inv self._prev_avg_entry = avg_entry self.state.inventory_btc = inv self.state.unrealized_pnl = upnl # Prefer total_asset_value; fallback collateral + upnl equity = tav if tav > 0 else (coll + upnl) if equity <= 0: equity = avail if avail > 0 else self.state.equity0 self.state.cash_usd = equity - inv * (self.state.last_mid or 0.0) self.state.realized_pnl = rpnl if self.state.day_start_equity <= 0 and not self._day_equity_restored: self.persist_day_equity(equity) def _record_fill(self, side: str, price: float, size: float, mid: float) -> None: st = self.state notional = price * size if side == "buy": st.buy_notional += notional st.buy_qty += size else: st.sell_notional += notional st.sell_qty += size f = Fill( id=uuid.uuid4().hex[:12], ts=time.time(), side=side, price=price, size_btc=size, fee_usd=0.0, mid_at_fill=mid, ) st.fills.append(f) # Markout tracker (widen half on toxic flow — no book clear) # Stamp ofi_at_fill from latest book OFI (set each quote cycle on tracker). ofi_at_fill = None try: mk = getattr(self, "markout", None) if mk is not None: ofi_at_fill = getattr(mk, "last_ofi", None) mk.on_fill(side, mid if mid > 0 else price, f.ts, ofi=ofi_at_fill) except Exception: pass self._replenish_after_fill = True # restore max ladder after hit # Denis: 1 fill -> exactly 1 replacement order on that side (no rebuild burst). lside = "bid" if side == "buy" else "ask" self._fills_to_replace[lside] = min(8, self._fills_to_replace.get(lside, 0) + 1) # open lot bookkeeping (simplified) if side == "buy": st.open_lots.append( OpenLot( id=uuid.uuid4().hex[:10], side="long", price=price, size_btc=size, notional_usd=notional, ts=f.ts, ) ) else: st.open_lots.append( OpenLot( id=uuid.uuid4().hex[:10], side="short", price=price, size_btc=size, notional_usd=notional, ts=f.ts, ) ) try: with self._fills_log.open("a") as fh: fh.write( json.dumps( { "id": f.id, "ts": f.ts, "side": f.side, "price": f.price, "size_btc": f.size_btc, "notional_usd": round(notional, 4), "mid_at_fill": f.mid_at_fill, "ofi_at_fill": ofi_at_fill, } ) + "\n" ) except Exception: pass try: self._persist_session() except Exception as e: log.warning("session persist after fill: %s", e) log.info("LIVE FILL %s %.6f BTC @ %.1f", side, size, price) def on_book(self, top: BookTop) -> List[Fill]: """Update mid / history; fills come from account refresh.""" st = self.state st.prev_mid = st.last_mid or top.mid st.last_mid = top.mid if top.ok() and top.mid > 0: now = time.time() st.price_history.append({"ts": now, "mid": top.mid}) self._update_candle(top.mid, now) return [] async def cancel_all(self) -> Tuple[bool, str]: try: _, resp, err = await self.client.cancel_all_orders( time_in_force=self.client.CANCEL_ALL_TIF_IMMEDIATE, timestamp_ms=0, cancel_all_market_index=self.market_id, api_key_index=self.api_key_index, ) self._bid_order_index = None self._ask_order_index = None if err: self.state.last_error = str(err) log.warning("cancel_all error: %s", err) await self._maybe_429_backoff(err) return False, str(err) self._cancel_all_count += 1 log.info( "cancel_all ok (#%s): %s", self._cancel_all_count, getattr(resp, "code", resp), ) return True, "" except Exception as e: self.state.last_error = str(e) log.warning("cancel_all exception: %s", e) await self._maybe_429_backoff(e) return False, str(e) async def _cancel_one(self, order_index: Optional[int]) -> None: if order_index is None: return try: _, _, err = await self.client.cancel_order( market_index=self.market_id, order_index=order_index, api_key_index=self.api_key_index, ) if err: log.debug("cancel_order %s: %s", order_index, err) await self._maybe_429_backoff(err) except Exception as e: log.debug("cancel_order %s exception: %s", order_index, e) await self._maybe_429_backoff(e) async def _place_limit( self, *, is_ask: bool, price: float, size_btc: float ) -> Tuple[Optional[int], str]: base_amount = _to_int_amount(size_btc, self.size_decimals) price_i = _to_int_amount(price, self.price_decimals) if base_amount <= 0 or price_i <= 0: return None, "invalid size/price" coi = self._next_client_order_index() try: tx, resp, err = await self.client.create_order( market_index=self.market_id, client_order_index=coi, base_amount=base_amount, price=price_i, is_ask=is_ask, order_type=self.client.ORDER_TYPE_LIMIT, time_in_force=self.client.ORDER_TIME_IN_FORCE_POST_ONLY, reduce_only=False, order_expiry=self.client.DEFAULT_28_DAY_ORDER_EXPIRY, api_key_index=self.api_key_index, ) if err: self.state.last_error = str(err) log.warning( "create_order %s err=%s px=%.1f sz=%.6f", "ask" if is_ask else "bid", err, price, size_btc, ) await self._maybe_429_backoff(err) return None, str(err) code = getattr(resp, "code", None) log.info( "LIVE ORDER %s px=%.1f sz=%.6f coi=%s code=%s", "ASK" if is_ask else "BID", price, size_btc, coi, code, ) # order_index resolved later via active orders; return client index as marker return coi, "" except Exception as e: self.state.last_error = str(e) log.warning("create_order exception: %s", e) await self._maybe_429_backoff(e) return None, str(e) async def sync_open_order_indices(self) -> None: """Map active orders to bid/ask indices for targeted cancel.""" try: auth = await self._auth() orders = await self.client.order_api.account_active_orders( authorization=auth, account_index=self.account_index, market_id=self.market_id, market_type="perp", ) bid_idx = None ask_idx = None n = 0 n_bid = 0 n_ask = 0 for o in getattr(orders, "orders", None) or []: n += 1 try: oi = int(o.order_index) except Exception: continue if bool(o.is_ask): ask_idx = oi n_ask += 1 else: bid_idx = oi n_bid += 1 self._bid_order_index = bid_idx self._ask_order_index = ask_idx self.state.ladder_orders = n self.state.ladder_bids = n_bid self.state.ladder_asks = n_ask # Exchange is truth for counts; if empty, drop local resting levels/quote if n <= 0: self._working_levels = [] self._working_n_levels = 0 self._working_bid0 = 0.0 self._working_ask0 = 0.0 self._working_mid = 0.0 self.state.quote = None elif self._working_levels and (n_bid + n_ask) != len(self._working_levels): # Counts drifted (partial fills) — keep prices, trim counts already set pass except Exception as e: log.debug("sync_open_order_indices: %s", e) async def _sign_create( self, *, is_ask: bool, price: float, size_btc: float, nonce: int ): """Sign one create_order; returns (tx_type, tx_info, coi, err).""" base_amount = _to_int_amount(size_btc, self.size_decimals) price_i = _to_int_amount(price, self.price_decimals) if base_amount <= 0 or price_i <= 0: return None, None, None, "invalid size/price" coi = self._next_client_order_index() tx_type, tx_info, _tx_hash, err = self.client.sign_create_order( market_index=self.market_id, client_order_index=coi, base_amount=base_amount, price=price_i, is_ask=is_ask, order_type=self.client.ORDER_TYPE_LIMIT, time_in_force=self.client.ORDER_TIME_IN_FORCE_POST_ONLY, reduce_only=False, order_expiry=self.client.DEFAULT_28_DAY_ORDER_EXPIRY, nonce=nonce, api_key_index=self.api_key_index, ) if err: return None, None, None, str(err) return tx_type, tx_info, coi, None async def _place_batch(self, levels: List[QuoteLevel]) -> Tuple[int, str]: """Place N post-only orders via send_tx_batch (1 REST) when possible. Falls back to sequential create_order on batch failure. Returns (n_placed, err). """ if not levels: return 0, "" # Denis 2026-10-25: place replacement / ladder orders ONE BY ONE, paced # (no burst). Best levels first (inner -> outer, alternating sides). # Stop immediately on 429; remaining levels are replenished next cycle. if SEQUENTIAL_PACED_PLACE: bids = [lv for lv in levels if lv.side == "bid"] asks = [lv for lv in levels if lv.side == "ask"] order: List[QuoteLevel] = [] for i in range(max(len(bids), len(asks))): if i < len(bids): order.append(bids[i]) if i < len(asks): order.append(asks[i]) n_ok, last_err = 0, "" for i, lv in enumerate(order): if i: await asyncio.sleep(PLACE_GAP_S) coi, err = await self._place_limit( is_ask=(lv.side == "ask"), price=lv.price, size_btc=lv.size_btc, ) if coi is not None: n_ok += 1 self._last_place_ts = time.time() elif err: last_err = err if is_rate_limit_error(err): self._replenish_after_fill = True log.warning("paced place: 429 after %d/%d — rest next cycle", n_ok, len(order)) break if n_ok < len(order): self._replenish_after_fill = True return n_ok, last_err # Prefer batch with explicit nonces under the nonce-manager lock. try: placed = await self._place_batch_locked(levels) if placed > 0: self._place_batch_count += 1 self._last_place_ts = time.time() return placed, "" except Exception as e: if is_rate_limit_error(e): # 429 on batch: do NOT fan out into N sequential creates (that # turned 1 rejected call into 16 more and caused 429 storms). log.warning("send_tx_batch 429 — backoff, skip sequential fallback (retry next cycle)") await self._maybe_429_backoff(e) return 0, "429" log.warning("send_tx_batch failed (%s) — sequential fallback", e) # Sequential fallback (expensive on rate limit — use sparingly). n_ok = 0 last_err = "" for lv in levels: coi, err = await self._place_limit( is_ask=(lv.side == "ask"), price=lv.price, size_btc=lv.size_btc, ) if coi is not None: n_ok += 1 elif err: last_err = err if n_ok: self._last_place_ts = time.time() return n_ok, last_err async def _place_batch_locked(self, levels: List[QuoteLevel]) -> int: """Sign all levels with sequential nonces, send_tx_batch (chunked).""" nm = self.client.nonce_manager api_key = self.api_key_index # Chunk to stay safe on batch size limits. chunk_size = 24 total_placed = 0 for offset in range(0, len(levels), chunk_size): chunk = levels[offset : offset + chunk_size] async with nm.lock(api_key): tx_types: List[int] = [] tx_infos: List[str] = [] coins: List[int] = [] sides: List[str] = [] for lv in chunk: _k, nonce = await nm.async_next_nonce(api_key) tx_type, tx_info, coi, err = await self._sign_create( is_ask=(lv.side == "ask"), price=lv.price, size_btc=lv.size_btc, nonce=nonce, ) if err or tx_type is None or tx_info is None: log.warning( "sign_create %s L%s failed: %s", lv.side, lv.level, err, ) nm.acknowledge_failure(api_key) continue # tx_type from signer may be str or int try: tx_types.append(int(tx_type)) except (TypeError, ValueError): tx_types.append(tx_type) # type: ignore[arg-type] tx_infos.append(tx_info) coins.append(coi) sides.append(lv.side) if not tx_types: continue try: resp = await self.client.send_tx_batch( tx_types=tx_types, tx_infos=tx_infos ) except Exception as e: # Batch consumed N nonces optimistically — hard refresh. try: await nm.async_hard_refresh_nonce(api_key) except Exception: nm.acknowledge_failure(api_key) raise code = getattr(resp, "code", None) hashes = getattr(resp, "tx_hash", None) or [] n = len(hashes) if hashes else len(tx_types) total_placed += n for i, (coi, side) in enumerate(zip(coins, sides)): lv = chunk[i] if i < len(chunk) else None px = lv.price if lv else 0 sz = lv.size_btc if lv else 0 log.info( "LIVE ORDER %s px=%.1f sz=%.6f coi=%s batch_code=%s", "ASK" if side == "ask" else "BID", px, sz, coi, code, ) return total_placed async def _replace_one_for_one(self, ladder, allow_bid: bool, allow_ask: bool) -> None: """Place exactly one replacement order per fill, one after the other.""" self.state.last_refresh_reason = "replace_1for1" # Bail if rate nearly full — keep counters, retry next cycle. try: snap = get_limiter().snapshot() room = int(snap.get("max_per_window") or 37) - int(snap.get("in_window") or 0) if room < 3: self.state.last_refresh_reason = "defer_rate_1for1" self._replenish_after_fill = True return except Exception: pass placed_any = False for side, allowed in (("bid", allow_bid), ("ask", allow_ask)): n = self._fills_to_replace.get(side, 0) if n <= 0: continue if not allowed: self._fills_to_replace[side] = 0 continue rungs = sorted((lv for lv in ladder.levels if lv.side == side), key=lambda lv: lv.level) if not rungs: self._fills_to_replace[side] = 0 continue # Outermost rungs (the filled inner one is gone; remaining shift in). targets = rungs[-n:] if n <= len(rungs) else rungs for lv in targets: if placed_any: await asyncio.sleep(PLACE_GAP_S) coi, err = await self._place_limit( is_ask=(side == "ask"), price=lv.price, size_btc=lv.size_btc, ) if coi is None: log.warning("replace_1for1 %s failed (%s) — retry next cycle", side, err) if err and is_rate_limit_error(err): return # keep counters; _replenish_after_fill stays True continue placed_any = True self._fills_to_replace[side] = max(0, self._fills_to_replace[side] - 1) self._last_place_ts = time.time() try: self._working_levels.extend(self._levels_to_status([lv])) except Exception: pass log.info("replace_1for1 %s px=%.1f sz=%.6f (fill -> 1 order)", side, lv.price, lv.size_btc) if sum(self._fills_to_replace.values()) == 0: self._replenish_after_fill = False def _clear_working_for_replenish(self, allow_bid: bool, allow_ask: bool) -> None: """After cancel under rate pressure: empty working markers, force replenish.""" self._bid_order_index = None self._ask_order_index = None self._working_mid = 0.0 self._working_bid0 = 0.0 self._working_ask0 = 0.0 self._working_n_levels = 0 self._working_levels = [] self._working_allow_bid = allow_bid self._working_allow_ask = allow_ask self._last_place_ts = 0.0 self._replenish_after_fill = True self.state.ladder_orders = 0 self.state.ladder_bids = 0 self.state.ladder_asks = 0 self.state.pending_orders = 0 # No phantom quote while book empty — UI must not show fake resting quotes self.state.quote = None def _select_levels_for_budget( self, levels: List[QuoteLevel], snap: Optional[dict] ) -> Tuple[List[QuoteLevel], bool]: """If rate room is tight, keep only top-of-book (≤2 per side). Returns (levels_to_place_now, partial). Batch place is 1 REST call, but sequential fallback burns the window; shrinking the ladder under pressure avoids wipe→sparse holes. """ if not levels or not snap: return levels, False try: room = int(snap.get("max_per_window") or 37) - int(snap.get("in_window") or 0) except (TypeError, ValueError): return levels, False # After cancel_all we typically have room-1. Need ≥2 free-ish for place+slack. if room >= 3: return levels, False # Top 2 levels each side (level 0, 1). bids = sorted([lv for lv in levels if lv.side == "bid"], key=lambda x: x.level) asks = sorted([lv for lv in levels if lv.side == "ask"], key=lambda x: x.level) trimmed = bids[:2] + asks[:2] if len(trimmed) >= len(levels): return levels, False return trimmed, True async def _wait_rate_room(self, need: int = 2, timeout_s: float = 20.0) -> bool: """Block until limiter has `need` free slots (or timeout). Empty-book priority.""" import asyncio deadline = time.monotonic() + float(timeout_s) while time.monotonic() < deadline: try: lim = get_limiter() snap = lim.snapshot() if lim else None except Exception: snap = None if snap is None: return True room = int(snap.get("max_per_window") or 37) - int(snap.get("in_window") or 0) if room >= int(need): return True await asyncio.sleep(0.25) return False def _mid_moved_bps(self, mid: float) -> float: w = self._working_mid if w <= 0 or mid <= 0: return 1e9 return abs(mid - w) / w * 10_000.0 async def _incremental_touch_recenter( self, ladder: "QuoteLadder", allow_bid: bool, allow_ask: bool ) -> None: """Replace best bid/ask only — NEVER cancel_all. Cancels tracked top-of-book indices (if any) then places new L0 from the desired ladder. Deeper levels keep resting. Used for FORCE_RECENTER band so the book never goes 0+0 for ~40s under rate limit. """ self.state.last_refresh_reason = "placing(touch_recenter)" self._quote_target = ladder.as_quote() # Desired L0 bid_lv = next((lv for lv in (ladder.levels or []) if lv.side == "bid" and int(lv.level) == 0), None) ask_lv = next((lv for lv in (ladder.levels or []) if lv.side == "ask" and int(lv.level) == 0), None) if bid_lv is None: bid_lv = next((lv for lv in (ladder.levels or []) if lv.side == "bid"), None) if ask_lv is None: ask_lv = next((lv for lv in (ladder.levels or []) if lv.side == "ask"), None) # Cancel only known top indices (not market-wide) for idx_attr in ("_bid_order_index", "_ask_order_index"): oi = getattr(self, idx_attr, None) if oi is not None: try: await self._cancel_one(int(oi)) except Exception as e: log.debug("touch_recenter cancel %s: %s", idx_attr, e) setattr(self, idx_attr, None) ok = await self._wait_rate_room(need=2, timeout_s=8.0) if not ok: self.state.last_refresh_reason = "defer_rate_keep(touch_recenter)" log.warning("touch_recenter deferred — keep deeper resting book") return placed = 0 if allow_bid and bid_lv is not None: coi, err = await self._place_limit( is_ask=False, price=float(bid_lv.price), size_btc=float(bid_lv.size_btc) ) if coi is not None: placed += 1 self._working_bid0 = float(bid_lv.price) else: log.warning("touch_recenter bid place failed: %s", err) if allow_ask and ask_lv is not None: if placed: await asyncio.sleep(PLACE_GAP_S) coi, err = await self._place_limit( is_ask=True, price=float(ask_lv.price), size_btc=float(ask_lv.size_btc) ) if coi is not None: placed += 1 self._working_ask0 = float(ask_lv.price) else: log.warning("touch_recenter ask place failed: %s", err) if ladder.mid: self._working_mid = float(ladder.mid) if placed > 0: self._last_place_ts = time.time() self._last_touch_recenter_ts = time.time() # Refresh counts from exchange when possible; else bump locally try: await self.sync_open_order_indices() except Exception: pass if int(self.state.ladder_bids or 0) + int(self.state.ladder_asks or 0) <= 0: # sync empty/failed — keep prior working level list, update best px if self.state.quote is not None: if self._working_bid0 > 0: self.state.quote.bid_px = self._working_bid0 if self._working_ask0 > 0: self.state.quote.ask_px = self._working_ask0 self.state.last_refresh_reason = f"placed_touch_recenter({placed})" log.info( "touch_recenter placed=%s bid0=%.1f ask0=%.1f mid=%.1f (no cancel_all)", placed, self._working_bid0 or 0, self._working_ask0 or 0, ladder.mid or 0, ) else: self.state.last_refresh_reason = "defer_rate_keep(touch_recenter)" async def _apply_side_trim( self, ladder: "QuoteLadder", allow_bid: bool, allow_ask: bool ) -> None: """Side change helper — never stick 0+x / x+0 >10s; target 8+8. Unless |inv| > 0.5 BTC (true danger), force both sides and TOP-UP the short side without cancel_all. Full wipe only when inv forces a cut. """ keep_bid = bool(allow_bid) keep_ask = bool(allow_ask) inv_abs = abs(float(getattr(self.state, "inventory_btc", 0.0) or 0.0)) # Soft inv / day cuts left 0+8 for minutes — force 8+8 unless dangerous. if inv_abs <= 0.5: keep_bid = True keep_ask = True wb = int(self.state.ladder_bids or 0) wa = int(self.state.ladder_asks or 0) self._working_allow_bid = keep_bid self._working_allow_ask = keep_ask self._quote_target = ladder.as_quote() # Both allowed → target 8+8. Top-up ONLY missing rungs (no duplicate 8). if keep_bid and keep_ask: target = int(LADDER_TARGET_LEVELS) if wb >= target and wa >= target: self.state.last_refresh_reason = "rest_side_trim" self._last_side_trim_ts = time.time() return need_bid = max(0, target - wb) if keep_bid else 0 need_ask = max(0, target - wa) if keep_ask else 0 bid_lvs = sorted( (lv for lv in (ladder.levels or []) if lv.side == "bid"), key=lambda lv: int(lv.level), ) ask_lvs = sorted( (lv for lv in (ladder.levels or []) if lv.side == "ask"), key=lambda lv: int(lv.level), ) levels = [] # Outer rungs first for holes (already have inner N → place L[N..]). if need_bid > 0 and bid_lvs: levels.extend(bid_lvs[-need_bid:] if len(bid_lvs) >= need_bid else bid_lvs) if need_ask > 0 and ask_lvs: levels.extend(ask_lvs[-need_ask:] if len(ask_lvs) >= need_ask else ask_lvs) if not levels: # Do NOT arm post_fill_replenish — empty fills_to_replace + that # flag falls through to cancel_all wipe (live 10:01). self.state.last_refresh_reason = "defer_rate_place(side_trim)" self._last_side_trim_ts = time.time() return self.state.last_refresh_reason = "placing(side_trim_topup)" self._last_side_trim_ts = time.time() log.info( "side_trim topup need_bid=%s need_ask=%s (have b=%s a=%s inv=%.4f) levels=%s", need_bid, need_ask, wb, wa, inv_abs, len(levels), ) ok = await self._wait_rate_room(need=2, timeout_s=20.0) if not ok: self.state.last_refresh_reason = "defer_rate_place(side_trim)" return n_placed, _err = await self._place_batch(levels) if n_placed > 0: if ladder.mid: self._working_mid = float(ladder.mid) self._last_place_ts = time.time() try: await self.sync_open_order_indices() except Exception: pass # Never regress below known resting counts if sync glitches. if int(self.state.ladder_bids or 0) < wb: self.state.ladder_bids = wb if int(self.state.ladder_asks or 0) < wa: self.state.ladder_asks = wa self.state.ladder_orders = int(self.state.ladder_bids or 0) + int( self.state.ladder_asks or 0 ) self._working_n_levels = int(self.state.ladder_orders or 0) self.state.pending_orders = int(self.state.ladder_orders or 0) self.state.last_refresh_reason = f"placed_side_trim_topup({n_placed})" log.info( "side_trim topup placed=%s now b=%s a=%s", n_placed, self.state.ladder_bids, self.state.ladder_asks, ) # Incomplete holes → next side_trim via incomplete check, not wipe. self._replenish_after_fill = False else: self.state.last_refresh_reason = "defer_rate_place(side_trim)" return # Risk one-sided already correct → rest (only when |inv|>0.5 cut). if (keep_bid and not keep_ask and wb >= 4 and wa == 0) or ( keep_ask and not keep_bid and wa >= 4 and wb == 0 ): self.state.last_refresh_reason = "rest_side_trim" log.info( "side_trim already one-sided b=%s a=%s allow=%s/%s inv=%.4f — rest", wb, wa, keep_bid, keep_ask, inv_abs, ) return levels = [ lv for lv in (ladder.levels or []) if (lv.side == "bid" and keep_bid) or (lv.side == "ask" and keep_ask) ] self.state.last_refresh_reason = "placing(side_trim)" self._last_side_trim_ts = time.time() log.info( "side_trim cancel+place kept bid=%s ask=%s levels=%s", keep_bid, keep_ask, len(levels), ) if self._working_n_levels > 0 or int(getattr(self.state, "ladder_orders", 0) or 0) > 0: await self.cancel_all() self._last_full_rebuild_ts = time.time() self._working_levels = [] self._working_n_levels = 0 self._working_bid0 = 0.0 self._working_ask0 = 0.0 self._working_mid = 0.0 self.state.ladder_orders = 0 self.state.ladder_bids = 0 self.state.ladder_asks = 0 self.state.pending_orders = 0 self.state.quote = None self._working_allow_bid = keep_bid self._working_allow_ask = keep_ask if not levels: self._replenish_after_fill = True self.state.last_refresh_reason = "defer_rate_place(side_trim)" return ok = await self._wait_rate_room(need=2, timeout_s=20.0) if not ok: self._replenish_after_fill = True self.state.last_refresh_reason = "defer_rate_place(side_trim)" return n_placed, _err = await self._place_batch(levels) if n_placed > 0: self._mark_working_placed(levels, ladder, n_placed, keep_bid, keep_ask) self._last_place_ts = time.time() self.state.last_refresh_reason = f"placed_side_trim({n_placed})" self._replenish_after_fill = ( n_placed < len(levels) or ( keep_bid and keep_ask and ( int(self.state.ladder_bids or 0) < 8 or int(self.state.ladder_asks or 0) < 8 ) ) ) else: self._replenish_after_fill = True self.state.last_refresh_reason = "defer_rate_place(side_trim)" def _should_refresh_ladder( self, ladder: Optional[QuoteLadder], allow_bid: bool, allow_ask: bool, rest_threshold_bps: float, min_rest_s: float = 40.0, ) -> Tuple[bool, str]: """Decide whether to wipe+replace. Default = rest (no cancel_all).""" if ladder is None: if self._working_n_levels > 0 or int(getattr(self.state, "ladder_orders", 0) or 0) > 0: return True, "ladder_cleared" return False, "idle" # Empty book (exchange count OR local): ALWAYS replenish. # Ghost working prices after cancel must NOT drive toxic_* loops. n_orders = int(getattr(self.state, "ladder_orders", 0) or 0) if n_orders <= 0 or self._working_n_levels <= 0 or self._working_mid <= 0: if ( self._working_n_levels > 0 or self._working_bid0 > 0 or self._working_ask0 > 0 or self._working_mid > 0 ): self._working_levels = [] self._working_n_levels = 0 self._working_bid0 = 0.0 self._working_ask0 = 0.0 self._working_mid = 0.0 self.state.quote = None return True, "initial_place" # Anti-thrash: after a place, force rest unless risk side flipped. aged = (time.time() - self._last_place_ts) if self._last_place_ts > 0 else 1e9 side_changed = ( self._working_allow_bid != allow_bid or self._working_allow_ask != allow_ask ) # Mid-cross beats post_fill: otherwise 1-for-1 loops while L0 stays toxic. _mid_early = float(ladder.mid or 0.0) if _mid_early > 0 and ( (self._working_bid0 > 0 and self._working_bid0 >= _mid_early) or (self._working_ask0 > 0 and self._working_ask0 <= _mid_early) ): last_touch = float(getattr(self, "_last_touch_recenter_ts", 0.0) or 0.0) bid_thru_e = ( (self._working_bid0 - _mid_early) / _mid_early * 10_000.0 if self._working_bid0 > 0 else 0.0 ) ask_thru_e = ( (_mid_early - self._working_ask0) / _mid_early * 10_000.0 if self._working_ask0 > 0 else 0.0 ) thru_a = max(bid_thru_e, ask_thru_e, 0.0) # Micro-cross: 60s CD (rate thrash). Large thru ≥ 8 bps: 12s. cross_cd = float(TOUCH_RECENTER_COOLDOWN_S) if thru_a >= float(TOUCH_RECENTER_LARGE_THRU_BPS): cross_cd = float(TOUCH_RECENTER_LARGE_COOLDOWN_S) if (last_touch <= 0.0 or (time.time() - last_touch) >= cross_cd) and ( allow_bid or allow_ask ): which = ( "bid" if self._working_bid0 >= _mid_early and ( self._working_ask0 <= 0 or self._working_ask0 > _mid_early or bid_thru_e >= ask_thru_e ) else "ask" ) return True, f"adverse_cross_{which}_{thru_a:.2f}bps" # Large toxic L0 must not fall into 1-for-1 while CD ticks. if thru_a >= float(TOUCH_RECENTER_LARGE_THRU_BPS): return False, f"soft_keep_cross_cd_{thru_a:.2f}bps" # After a fill: replace immediately (Denis: always keep ladder count, # no delay when an order leaves). 1-for-1 path does not cancel_all. if self._replenish_after_fill: return True, "post_fill_replenish" if side_changed and aged >= 5.0: # Prefer side_trim (cancel disallowed only / no full thrash). # Full risk_side wipe left 0+0 for ~30s under PLACE_GAP=2. keeping = (allow_bid or allow_ask) if keeping: return True, "side_trim" return True, "risk_side_change" # Incomplete ladder (sticky 3+8 / 7+2): top-up to 8+8 without side flip. # |inv|≤0.5 always wants both sides (strategy #1 / side_trim force). if allow_bid and allow_ask: inv_abs = abs(float(getattr(self.state, "inventory_btc", 0.0) or 0.0)) if inv_abs <= 0.5: wb = int(getattr(self.state, "ladder_bids", 0) or 0) wa = int(getattr(self.state, "ladder_asks", 0) or 0) # Tests / brief desync: derive from local resting levels if counts unset. if (wb + wa) <= 0 and self._working_levels: wb = sum(1 for x in self._working_levels if x.get("side") == "bid") wa = sum(1 for x in self._working_levels if x.get("side") == "ask") target = int(LADDER_TARGET_LEVELS) known = (wb + wa) > 0 if known and (wb < target or wa < target): last_top = float(getattr(self, "_last_side_trim_ts", 0.0) or 0.0) if last_top <= 0.0 or ( (time.time() - last_top) >= float(SIDE_TRIM_TOPUP_COOLDOWN_S) ): return True, "side_trim" # Through-market / mid drift (Denis t165u/t192u): # - toxic through NEVER cancel_all (0+0 thrash under PLACE_GAP). # - only extreme mid_moved (≥ MAX_STALE) may full-rebuild, and only # after FULL_REBUILD_COOLDOWN_S since the last cancel_all rebuild. mid = float(ladder.mid or 0.0) ask_thru = bid_thru = 0.0 if mid > 0: if self._working_ask0 > 0: ask_thru = (mid - self._working_ask0) / mid * 10_000.0 if self._working_bid0 > 0: bid_thru = (self._working_bid0 - mid) / mid * 10_000.0 moved = self._mid_moved_bps(ladder.mid) thru = max(float(ask_thru), float(bid_thru), 0.0) last_rb = float(getattr(self, "_last_full_rebuild_ts", 0.0) or 0.0) cooldown_ok = (last_rb <= 0.0) or ( (time.time() - last_rb) >= float(FULL_REBUILD_COOLDOWN_S) ) # Full cancel_all rebuild ONLY at extreme mid drift (≥ MAX_STALE=22) + cooldown. # Checked BEFORE adverse soft_keep so a 22bps+ move still recenters. if moved >= MAX_STALE_KEEP_BPS: if cooldown_ok: return True, f"mid_moved_{moved:.2f}bps" return False, f"soft_keep_mid_cd_{moved:.2f}bps" # MID-CROSS (bid>=mid or ask<=mid): incremental touch_recenter ONLY. # Never soft_keep crossed L0 (got picked / day bleed). Never cancel_all # (prior adverse_cross wipe → 0+0 ~40s). Any through > 0 bps counts. cross_bid = bool(mid > 0 and self._working_bid0 > 0 and self._working_bid0 >= mid) cross_ask = bool(mid > 0 and self._working_ask0 > 0 and self._working_ask0 <= mid) if cross_bid or cross_ask: # NO post-place grace on mid-cross: 1-for-1 fills reset _last_place_ts # and used to soft_keep toxic L0 forever. Cooldown alone paces churn. last_touch = float(getattr(self, "_last_touch_recenter_ts", 0.0) or 0.0) thru_a = max(float(bid_thru), float(ask_thru), 0.0) cross_cd = float(TOUCH_RECENTER_COOLDOWN_S) if thru_a >= float(TOUCH_RECENTER_LARGE_THRU_BPS): cross_cd = float(TOUCH_RECENTER_LARGE_COOLDOWN_S) touch_ok = (last_touch <= 0.0) or ( (time.time() - last_touch) >= float(cross_cd) ) if touch_ok and (allow_bid or allow_ask): which = ( "bid" if cross_bid and (not cross_ask or float(bid_thru) >= float(ask_thru)) else "ask" ) return True, f"adverse_cross_{which}_{thru_a:.2f}bps" return False, f"soft_keep_cross_cd_{thru_a:.2f}bps" # FORCE_RECENTER band: incremental touch recenter (no cancel_all), else soft_keep. if moved >= FORCE_RECENTER_BPS: # Skip post-place grace in FORCE band (fills reset _last_place_ts → freeze). # Use short CD so quotes don't sit 14–20 bps off mid for a full minute. last_touch = float(getattr(self, "_last_touch_recenter_ts", 0.0) or 0.0) force_cd = float(TOUCH_RECENTER_LARGE_COOLDOWN_S) touch_ok = (last_touch <= 0.0) or ( (time.time() - last_touch) >= float(force_cd) ) if touch_ok and (allow_bid or allow_ask): return True, "touch_recenter" return False, f"soft_keep_force_{moved:.2f}bps" # Through-market without mid-cross (join flicker < 0): KEEP (strategy #1). if ask_thru >= TOXIC_EPS_BPS or bid_thru >= TOXIC_EPS_BPS: which = "ask" if float(ask_thru) >= float(bid_thru) else "bid" return False, f"soft_keep_toxic_{which}_{thru:.2f}bps" if aged < POST_PLACE_GRACE_S: return False, f"post_place_grace_{aged:.1f}s" if side_changed: if allow_bid or allow_ask: return True, "side_trim" return True, "risk_side_change" if aged < min_rest_s and not self._replenish_after_fill: return False, f"min_rest_{aged:.1f}s" # Soft mid gap: KEEP the book (Denis t165u — 0+0 / half ladder # kills fills). Mild drift must NOT cancel_all; only hard mid_moved # (above) / risk / post_fill may rebuild. rest_threshold for compat. thr = float(rest_threshold_bps or 0.0) if thr > 0.0: mild_thru = max(float(ask_thru), float(bid_thru), 0.0) gap = max(float(moved), mild_thru) if gap >= thr: return False, f"soft_keep_{gap:.2f}bps" return False, "rest" async def apply_ladder( self, ladder: Optional[QuoteLadder], allow_bid: bool, allow_ask: bool, *, rest_threshold_bps: float = 3.0, batch_place: bool = True, min_rest_s: float = 25.0, ) -> None: """Rest-first ladder: place once, leave resting for fills. cancel_all ONLY when refresh is required (mid moved / risk side / initial / drained). Health checks that call this with an unchanged mid are no-ops — no cancel_all loop. """ need, reason = self._should_refresh_ladder( ladder, allow_bid, allow_ask, rest_threshold_bps, min_rest_s=min_rest_s ) if not need: self._rest_skip_count += 1 if ladder is not None: self._quote_target = ladder.as_quote() # Keep state.quote / ladder_levels = resting book (do not overwrite with desired) if self._working_levels and self.state.quote is None: # Rebuild quote from resting best if lost self.state.quote = Quote( bid_px=float(self._working_bid0 or 0.0), ask_px=float(self._working_ask0 or 0.0), size_btc=float( next((x.get("size_btc") or 0.0 for x in self._working_levels), 0.0) ), mid=float(self._working_mid or 0.0), skew_bps=float(getattr(self._quote_target, "skew_bps", 0.0) or 0.0), ) # Keep soft_keep_* / grace reason visible (was rest(N) — hid freeze bugs). if str(reason).startswith(("soft_keep", "post_place_grace", "min_rest")): self.state.last_refresh_reason = str(reason) else: self.state.last_refresh_reason = f"rest({self._rest_skip_count})" return # Refresh path: one cancel_all + batch place (not every tick). # Under a nearly-full L1 window: NEVER keep toxic / mid-moved quotes # resting (that starved fills — "manque de trade"). Clear them without # re-placing; replenish as soon as budget allows. # Belt: toxic_* must never reach cancel_all (Denis t192u). if str(reason).startswith("toxic_"): self._rest_skip_count += 1 self.state.last_refresh_reason = f"defer_toxic({reason})" if ladder is not None: self._quote_target = ladder.as_quote() log.warning( "defer toxic wipe (%s): keep resting book (never cancel_all→0+0)", reason, ) return # EARLY mid-cross: never let 1-for-1 skip L0 fix. if ladder is not None and ( reason == "touch_recenter" or str(reason).startswith("adverse_cross_") ): await self._incremental_touch_recenter(ladder, allow_bid, allow_ask) return # EARLY post_fill: 1-for-1 must not be deferred by rate/grace gate. if reason == "post_fill_replenish" and ladder is not None and ladder.levels: n_rep = sum(getattr(self, "_fills_to_replace", {}).values()) if self._working_n_levels > 0 and n_rep > 0: await self._replace_one_for_one(ladder, allow_bid, allow_ask) return # Flag set without fill counters (topup/incomplete) — NEVER cancel_all. self._replenish_after_fill = False await self._apply_side_trim(ladder, allow_bid, allow_ask) return # EARLY side_trim: has its own rate wait; must not defer_keep incomplete 3+8. if reason == "side_trim" and ladder is not None: await self._apply_side_trim(ladder, allow_bid, allow_ask) return try: from .rate_limit import get_limiter lim = get_limiter() snap = lim.snapshot() if lim else None except Exception: snap = None mid_now = float(ladder.mid) if ladder is not None else float(self.state.last_mid or 0.0) aged_place = ( (time.time() - self._last_place_ts) if self._last_place_ts > 0 else 1e9 ) action = decide_rate_constrained_refresh( in_window=int((snap or {}).get("in_window") or 0), max_per_window=int((snap or {}).get("max_per_window") or 37), working_n_levels=int(self._working_n_levels), reason=reason, mid=mid_now, working_mid=float(self._working_mid or 0.0), working_bid0=float(self._working_bid0 or 0.0), working_ask0=float(self._working_ask0 or 0.0), rest_threshold_bps=float(rest_threshold_bps), seconds_since_place=float(aged_place), ) # No snap → proceed (limiter not installed). if snap is None: action = "proceed" if action == "clear_stale": self.state.last_refresh_reason = f"stale_clear_rate({reason})" log.warning( "stale_clear under rate pressure (%s): window %s/%s — " "cancel then replenish (never leave empty)", reason, snap.get("in_window") if snap else "?", snap.get("max_per_window") if snap else "?", ) if self._working_n_levels > 0: await self.cancel_all() self._last_full_rebuild_ts = time.time() self._clear_working_for_replenish(allow_bid, allow_ask) if ladder is not None: self._quote_target = ladder.as_quote() # Same path as empty-book defer_place: wait for rate then place. action = "defer_place" if action == "defer_place": # Empty book is worse than a brief wait: block for rate room, then # fall through to place (skip cancel_all — nothing to wipe). self.state.last_refresh_reason = f"wait_rate_place({reason})" self._replenish_after_fill = True log.warning( "empty book under rate pressure (%s): window %s/%s — wait then place", reason, (snap or {}).get("in_window"), (snap or {}).get("max_per_window"), ) ok = await self._wait_rate_room(need=2, timeout_s=55.0) if not ok: # One more window: never idle-return empty if ladder desired. ok = await self._wait_rate_room(need=2, timeout_s=55.0) if not ok: self.state.last_refresh_reason = f"defer_rate_place({reason})" self._replenish_after_fill = True if ladder is not None: self._quote_target = ladder.as_quote() log.warning( "empty-book still rate-blocked after waits (%s) — will retry", reason, ) return # Refresh snap after wait; proceed to place without cancel. try: lim = get_limiter() snap = lim.snapshot() if lim else None except Exception: snap = None self.state.last_refresh_reason = reason # jump to place path below (skip cancel_all) if ladder is None or not ladder.levels: self.state.quote = None self._clear_working_for_replenish(allow_bid, allow_ask) return levels = [ lv for lv in ladder.levels if (lv.side == "bid" and allow_bid) or (lv.side == "ask" and allow_ask) ] if not levels: self._clear_working_for_replenish(allow_bid, allow_ask) return levels, partial = self._select_levels_for_budget(levels, snap) if partial: self._replenish_after_fill = True if batch_place: n_placed, err = await self._place_batch(levels) else: n_placed = 0 err = "" for lv in levels: coi, e = await self._place_limit( is_ask=(lv.side == "ask"), price=lv.price, size_btc=lv.size_btc, ) if coi is not None: n_placed += 1 elif e: err = e if n_placed: self._last_place_ts = time.time() if n_placed < max(1, len(levels) // 2): if ladder is not None: self._quote_target = ladder.as_quote() self._clear_working_for_replenish(allow_bid, allow_ask) self.state.last_error = f"place_failed:{err or 'n='+str(n_placed)}" log.warning( "empty-book place sparse n=%s/%s (reason=%s err=%s)", n_placed, len(levels), reason, err or "", ) return self._mark_working_placed(levels, ladder, n_placed, allow_bid, allow_ask) if not partial: self._replenish_after_fill = False self.state.last_error = "" log.info( "ladder placed n=%s (bid=%s ask=%s) mid=%.1f reason=%s (after rate wait) err=%s", n_placed, self.state.ladder_bids, self.state.ladder_asks, ladder.mid, reason, err or "", ) return if action == "defer_keep": self.state.last_refresh_reason = f"defer_rate({reason})" log.warning( "defer ladder refresh (%s): rate window %s/%s — keep near-market resting", reason, (snap or {}).get("in_window"), (snap or {}).get("max_per_window"), ) if ladder is not None: self._quote_target = ladder.as_quote() return # --- 1-for-1 replacement after a fill: NO cancel_all, NO rebuild. --- # Each fill removed one rung; add one new outer rung on that side so # the ladder keeps the same count (e.g. 8+8). Sequential + paced. if ( reason == "post_fill_replenish" and self._working_n_levels > 0 and ladder is not None and ladder.levels and sum(self._fills_to_replace.values()) > 0 ): await self._replace_one_for_one(ladder, allow_bid, allow_ask) return # --- touch_recenter / adverse mid-cross: replace L0 only, never cancel_all --- if ( ladder is not None and (reason == "touch_recenter" or str(reason).startswith("adverse_cross_")) ): await self._incremental_touch_recenter(ladder, allow_bid, allow_ask) return # --- side_trim: day-flatten / inv-red drops ONE side without nuking both --- # Keep reducing-side quotes resting; cancel_all only if disallowed side # still has working levels (Lighter has no cancel-by-side API here). # Then place only missing reducing levels — never leave 0+0 for minutes. if reason == "side_trim" and ladder is not None: await self._apply_side_trim(ladder, allow_bid, allow_ask) return self.state.last_refresh_reason = f"placing({reason})" self._replenish_after_fill = False self._fills_to_replace = {"bid": 0, "ask": 0} # PREFLIGHT: never cancel_all into 0+0 without rate room for immediate re-place. # Prior bug: cancel first → wait 20–55s under PLACE_GAP → 38s empty book. will_wipe = ( self._working_n_levels > 0 or int(getattr(self.state, "ladder_orders", 0) or 0) > 0 ) if will_wipe and ( str(reason).startswith("mid_moved_") or str(reason).startswith("adverse_cross_") ): ok_pre = await self._wait_rate_room( need=int(REBUILD_RATE_NEED), timeout_s=3.0 ) if not ok_pre: self.state.last_refresh_reason = f"defer_rate_keep({reason})" log.warning( "defer rebuild (%s): keep resting book (need ≥%s rate slots)", reason, REBUILD_RATE_NEED, ) if ladder is not None: self._quote_target = ladder.as_quote() return log.info( "ladder refresh reason=%s cancel_all_total_before=%s working_n=%s", reason, self._cancel_all_count, self._working_n_levels, ) # Skip wipe when nothing is working — saves 1 rate slot after stale_clear. if will_wipe: await self.cancel_all() self._last_full_rebuild_ts = time.time() self._bid_order_index = None self._ask_order_index = None # Book wiped — clear ALL working markers NOW (no toxic ghosts while # place is in flight / if place fails under rate). Next tick sees # ladder_orders==0 → initial_place, never sticky toxic_*. self._working_levels = [] self._working_n_levels = 0 self._working_bid0 = 0.0 self._working_ask0 = 0.0 self._working_mid = 0.0 self.state.ladder_orders = 0 self.state.ladder_bids = 0 self.state.ladder_asks = 0 self.state.pending_orders = 0 self.state.quote = None if ladder is not None: self._quote_target = ladder.as_quote() if ladder is None or not ladder.levels: self._quote_target = None self._clear_working_for_replenish(allow_bid, allow_ask) self._replenish_after_fill = False return # Filter by allow flags (builder usually already did). levels = [ lv for lv in ladder.levels if (lv.side == "bid" and allow_bid) or (lv.side == "ask" and allow_ask) ] if not levels: self._quote_target = None self._clear_working_for_replenish(allow_bid, allow_ask) self._replenish_after_fill = False return # After cancel_all (or empty initial_place): ALWAYS wait for rate room # then place in THIS call — never return with empty book. self.state.last_refresh_reason = f"wait_rate_place({reason})" ok_room = await self._wait_rate_room(need=2, timeout_s=25.0) if not ok_room: self._replenish_after_fill = True self.state.last_refresh_reason = f"defer_rate_place({reason})" log.warning( "post-cancel/empty wait_rate timeout (%s) — retry next tick (book empty)", reason, ) return try: lim = get_limiter() snap = lim.snapshot() if lim else None except Exception: snap = None self.state.last_refresh_reason = f"placing({reason})" # Budget-aware: if window still tight after cancel, place top-of-book # first (2+2) and mark replenish for the rest — avoids wipe→sparse hole. levels, partial = self._select_levels_for_budget(levels, snap) if partial: self._replenish_after_fill = True log.info( "budget-aware place: top %s levels now (rate %s/%s), rest later", len(levels), (snap or {}).get("in_window"), (snap or {}).get("max_per_window"), ) if batch_place: n_placed, err = await self._place_batch(levels) else: n_placed = 0 err = "" for lv in levels: coi, e = await self._place_limit( is_ask=(lv.side == "ask"), price=lv.price, size_btc=lv.size_btc, ) if coi is not None: n_placed += 1 elif e: err = e if n_placed: self._last_place_ts = time.time() if n_placed < max(1, len(levels) // 2): # Failed / sparse place after cancel_all → do NOT mark working. if ladder is not None: self._quote_target = ladder.as_quote() self._clear_working_for_replenish(allow_bid, allow_ask) self.state.last_error = f"place_failed:{err or 'n='+str(n_placed)}" log.warning( "ladder place sparse n=%s/%s after cancel (reason=%s err=%s) — will retry", n_placed, len(levels), reason, err or "", ) return self._mark_working_placed(levels, ladder, n_placed, allow_bid, allow_ask) if not partial: self._replenish_after_fill = False self.state.last_error = "" log.info( "ladder placed n=%s (bid=%s ask=%s) mid=%.1f reason=%s err=%s " "cancel_all_total=%s rest_skips=%s", n_placed, self.state.ladder_bids, self.state.ladder_asks, ladder.mid, reason, err or "", self._cancel_all_count, self._rest_skip_count, ) async def apply_quote( self, q: Optional[Quote], allow_bid: bool, allow_ask: bool ) -> None: """Backward-compat: wrap single quote as 1+1 ladder with rest logic.""" if q is None: await self.apply_ladder(None, allow_bid, allow_ask) return levels: List[QuoteLevel] = [] if allow_bid and q.bid_px > 0 and q.size_btc > 0: levels.append( QuoteLevel(side="bid", level=0, price=q.bid_px, size_btc=q.size_btc) ) if allow_ask and q.ask_px > 0 and q.size_btc > 0: levels.append( QuoteLevel(side="ask", level=0, price=q.ask_px, size_btc=q.size_btc) ) ladder = QuoteLadder( levels=levels, mid=q.mid, skew_bps=q.skew_bps, size_btc=q.size_btc, bid_px=q.bid_px if allow_bid else 0.0, ask_px=q.ask_px if allow_ask else 0.0, ) await self.apply_ladder(ladder, allow_bid, allow_ask) # --- PaperEngine-compatible helpers for risk/status --- def equity(self, mid: Optional[float] = None) -> float: m = mid if mid is not None else self.state.last_mid # Prefer exchange-reported equity approximation if self.state.collateral > 0: return self.state.collateral + self.state.unrealized_pnl return self.state.cash_usd + self.state.inventory_btc * (m or 0.0) def day_pnl(self, mid: Optional[float] = None) -> float: return self.equity(mid) - self.state.day_start_equity def day_pnl_pct(self, mid: Optional[float] = None) -> float: base = self.state.day_start_equity or self.state.equity0 if base <= 0: return 0.0 return 100.0 * self.day_pnl(mid) / base def fills_today(self) -> int: return len(self.state.fills) def avg_buy_px(self) -> Optional[float]: st = self.state if st.buy_qty <= 1e-12: return None return st.buy_notional / st.buy_qty def avg_sell_px(self) -> Optional[float]: st = self.state if st.sell_qty <= 1e-12: return None return st.sell_notional / st.sell_qty def price_history_list(self) -> List[dict]: return list(self.state.price_history) CANDLE_SEC = 300 # 5 minutes def _update_candle(self, mid: float, ts: float) -> None: """5m OHLC; forming candle updates every book tick (no lag).""" bucket = int(ts // self.CANDLE_SEC) * self.CANDLE_SEC candles = self.state.candles if not candles or int(candles[-1]["t"]) != bucket: candles.append({"t": float(bucket), "o": mid, "h": mid, "l": mid, "c": mid}) else: c = candles[-1] if mid > c["h"]: c["h"] = mid if mid < c["l"]: c["l"] = mid c["c"] = mid # throttle disk write ~every 15s now_m = int(ts) // 15 if getattr(self, "_candle_save_slot", None) != now_m: self._candle_save_slot = now_m self._save_candles() def candles_list(self) -> List[dict]: return list(self.state.candles) def _load_candles(self) -> None: path = getattr(self, "_candles_path", None) if path is None or not path.exists(): return try: raw = json.loads(path.read_text()) rows = raw.get("candles") or [] for c in rows[-288:]: if not c or c.get("o") is None: continue self.state.candles.append( { "t": float(c["t"]), "o": float(c["o"]), "h": float(c["h"]), "l": float(c["l"]), "c": float(c["c"]), } ) log.info("loaded %d candles from %s", len(self.state.candles), path.name) except Exception as e: log.warning("candle load: %s", e) def _save_candles(self) -> None: path = getattr(self, "_candles_path", None) if path is None: return try: path.write_text( json.dumps( { "candle_sec": self.CANDLE_SEC, "updated_at": time.time(), "candles": list(self.state.candles), } ) ) except Exception as e: log.warning("candle save: %s", e) def open_positions(self) -> List[dict]: if abs(self.state.inventory_btc) < 1e-12: return [] side = "long" if self.state.inventory_btc > 0 else "short" px = self._prev_avg_entry or self.state.last_mid sz = abs(self.state.inventory_btc) return [ { "id": "live", "side": side, "price": px, "size_btc": sz, "notional_usd": round(px * sz, 4), "ts": self.state.day_start_ts, } ] def close_events(self, n: int | None = None) -> List[dict]: n = n if n is not None else self.MAX_CLOSE_EVENTS if self.state.closed_trades: recent = self.state.closed_trades[-n:] return [ { "id": c.id, "pnl_usd": round(c.pnl_usd, 6), "size_btc": c.size_btc, "ts": c.ts, "profitable": c.profitable, "side": c.side, "entry_price": c.entry_price, "exit_price": c.exit_price, } for c in recent ] persisted = getattr(self, "_persisted_close_events", None) or [] return list(persisted[-n:])