"""Orchestrator: market data → quote → paper/live → risk → status.""" from __future__ import annotations import asyncio import logging import os import time from pathlib import Path from typing import Optional, Union from .config import AppConfig from .market_data import BookTop, MarketDataFeed, discover_market from .paper_engine import PaperEngine from .quoter import Quote, QuoteLadder, build_quote, build_ladder from .momentum_1m import Momentum1m from .regime import analyze_regime, RegimeLeanHold from .risk import RiskManager from .strategy_policy import ( MarkoutTracker, day_pnl_policy, effective_half_bps, inventory_policy, merge_allows, ofi_policy, side_allow_hysteresis, ) from .status import StatusWriter log = logging.getLogger("lighter_mm.bot") def _ensure_momentum(bot) -> "Momentum1m": """Guard: live/paper must never crash on missing _momentum.""" m = getattr(bot, "_momentum", None) if m is None: bot._momentum = Momentum1m() log.warning("_momentum was missing — recreated (bug guard)") return bot._momentum class MicroMMBot: """Paper-mode bot (unchanged behaviour).""" def __init__(self, cfg: AppConfig, root: Path): if cfg.is_live: raise RuntimeError( "Use LiveMicroMMBot for live. " "Set mode=paper for this class." ) self.cfg = cfg self.root = root self.engine = PaperEngine(cfg.capital.paper_equity_usd, cfg.paper) self.risk = RiskManager(cfg.risk, cfg.capital.paper_equity_usd) status_path = root / cfg.status.path self.status = StatusWriter(status_path, cfg.status.console_every_s) self.meta = None self.feed: Optional[MarketDataFeed] = None self._queue: asyncio.Queue[BookTop] = asyncio.Queue(maxsize=256) self._last_quote_ts = 0.0 self._start = time.time() self._momentum = Momentum1m() self._stop = asyncio.Event() async def setup(self) -> None: self.meta = await discover_market( self.cfg.venue.rest_url, self.cfg.venue.symbol, self.cfg.venue.market_id, ) log.info( "Market %s id=%s mark=%.1f size_dec=%s price_dec=%s min_base=%s", self.meta.symbol, self.meta.market_id, self.meta.mark_price, self.meta.size_decimals, self.meta.price_decimals, self.meta.min_base_amount, ) self.feed = MarketDataFeed( rest_url=self.cfg.venue.rest_url, ws_url=self.cfg.venue.ws_url, market_id=self.meta.market_id, on_top=self._enqueue_top, ) def _enqueue_top(self, top: BookTop) -> None: try: self._queue.put_nowait(top) except asyncio.QueueFull: try: _ = self._queue.get_nowait() except asyncio.QueueEmpty: pass try: self._queue.put_nowait(top) except asyncio.QueueFull: pass async def _process_top(self, top: BookTop) -> None: fills = self.engine.on_book(top) for f in fills: log.info( "FILL %s %.6f BTC @ %.1f (mid %.1f)", f.side, f.size_btc, f.price, f.mid_at_fill, ) decision = self.risk.evaluate(self.engine, top.mid) now = time.time() interval = self.cfg.quoting.requote_interval_ms / 1000.0 q: Optional[Quote] = self.engine.state.quote if decision.killed: self.engine.set_quote(None) q = None elif (now - self._last_quote_ts) >= interval or q is None: q = build_quote( top, self.engine.state.inventory_btc, self.cfg.quoting, self.meta.price_decimals, self.meta.size_decimals, self.meta.min_base_amount, self.meta.min_quote_amount, ) if q: if not decision.allow_bid: q.bid_px = round(q.bid_px * 0.5, self.meta.price_decimals) if not decision.allow_ask: q.ask_px = round(q.ask_px * 1.5, self.meta.price_decimals) if q.bid_px >= q.ask_px: q = None self.engine.set_quote(q) self._last_quote_ts = now self._write_status(top, q, decision) dur = self.cfg.run.duration_s if dur and (now - self._start) >= dur: log.info("duration_s=%.0f reached — stopping", dur) self.stop() def _write_status(self, top: BookTop, q: Optional[Quote], decision) -> None: mid = top.mid payload = { "mode": "paper", "symbol": self.meta.symbol if self.meta else self.cfg.venue.symbol, "market_id": self.meta.market_id if self.meta else None, "mid": mid, "bid": top.bid, "ask": top.ask, "quote": None if q is None else { "bid_px": q.bid_px, "ask_px": q.ask_px, "size_btc": q.size_btc, "skew_bps": q.skew_bps, }, "inventory_btc": self.engine.state.inventory_btc, "inventory_usd": self.engine.state.inventory_btc * mid, "cash_usd": self.engine.state.cash_usd, "equity": self.engine.equity(mid), "equity0": self.engine.state.equity0, "capital_usd": self.engine.equity(mid), "capital0_usd": self.cfg.capital.paper_equity_usd, "realized_pnl": self.engine.state.realized_pnl, "day_pnl": self.engine.day_pnl(mid), "day_pnl_pct": self.engine.day_pnl_pct(mid), "fills": self.engine.fills_today(), "killed": decision.killed, "risk_reason": decision.reason, "lot_usd": [self.cfg.quoting.lot_usd_min, self.cfg.quoting.lot_usd_max], "half_spread_bps": float(getattr(self.cfg.quoting, "half_spread_bps", 0.0)), "rest_threshold_bps": float(getattr(self.cfg.quoting, "rest_threshold_bps", 0.0)), "min_rest_s": float(getattr(self.cfg.quoting, "min_rest_s", 0.0)), "max_inventory_btc": float(getattr(self.cfg.quoting, "max_inventory_btc", 0.0)), "kill_day_pnl_pct": float(getattr(self.cfg.risk, "kill_day_pnl_pct", -1.0)), "soft_leverage_min": float(getattr(self.cfg.risk, "soft_leverage_min", 0.0)), "soft_leverage_max": float(getattr(self.cfg.risk, "soft_leverage_max", 0.0)), "max_notional_usd": float(getattr(self.cfg.risk, "max_notional_usd", 0.0)), "soft_leverage": ( (abs(float(self.engine.state.inventory_btc) * float(mid or 0.0)) / max(float(self.engine.equity(mid) or 0.0), 1e-9)) if mid else 0.0 ), "uptime_s": time.time() - self._start, "ws_ok": bool(self.feed and self.feed.ws_ok), "open_positions": self.engine.open_positions(), # Net inventory sides (MM has at most one side open) — source of truth for UI L/S "n_long": ( 1 if float(self.engine.state.inventory_btc) > 1e-8 else 0 ), "n_short": ( 1 if float(self.engine.state.inventory_btc) < -1e-8 else 0 ), "position_side": ( "long" if float(self.engine.state.inventory_btc) > 1e-8 else ( "short" if float(self.engine.state.inventory_btc) < -1e-8 else "flat" ) ), "close_events": self.engine.close_events(30), "avg_buy_px": self.engine.avg_buy_px(), "avg_sell_px": self.engine.avg_sell_px(), "price_history": self.engine.price_history_list(), "candles": self.engine.candles_list() if hasattr(self.engine, "candles_list") else [], "candle_sec": 300, "momentum_1m": self._momentum.snapshot() if getattr(self, "_momentum", None) else None, "last_fills": [ { "id": f.id, "side": f.side, "price": f.price, "size_btc": f.size_btc, "notional_usd": round(f.price * f.size_btc, 4), "fee_usd": f.fee_usd, "mid_at_fill": f.mid_at_fill, "ts": f.ts, } for f in self.engine.state.fills[-100:] ], } self.status.write(payload) def stop(self) -> None: self._stop.set() if self.feed: self.feed.stop() async def run(self) -> None: await self.setup() assert self.feed is not None feed_task = asyncio.create_task(self.feed.run()) try: while not self._stop.is_set(): try: top = await asyncio.wait_for(self._queue.get(), timeout=0.5) except asyncio.TimeoutError: dur = self.cfg.run.duration_s if dur and (time.time() - self._start) >= dur: log.info("duration_s=%.0f reached — stopping", dur) self.stop() continue await self._process_top(top) finally: top = self.feed.last_top was_ws = bool(self.feed.ws_ok) self.feed.stop() feed_task.cancel() try: await feed_task except asyncio.CancelledError: pass if top.ok(): d = self.risk.evaluate(self.engine, top.mid) self.feed.ws_ok = was_ws self._write_status(top, self.engine.state.quote, d) log.info( "Stopped. fills=%s equity=%.2f dayPnL=%.2f opens=%s closes=%s", self.engine.fills_today(), self.engine.equity(top.mid if top.ok() else 0), self.engine.day_pnl(top.mid if top.ok() else 0), len(self.engine.state.open_lots), len(self.engine.state.closed_trades), ) class LiveMicroMMBot: """Live micro-MM: real post-only quotes on Lighter BTC.""" def __init__(self, cfg: AppConfig, root: Path): if not cfg.is_live: raise RuntimeError( "LiveMicroMMBot requires mode=live + live_enabled + LIVE=1" ) for key in ( "LIGHTER_API_PRIVATE_KEY", "LIGHTER_ACCOUNT_INDEX", "LIGHTER_API_KEY_INDEX", ): if not os.environ.get(key): raise RuntimeError(f"Missing env {key}") self.cfg = cfg self.root = root self.meta = None self.feed: Optional[MarketDataFeed] = None self.engine = None # set in setup self.client = None self.risk: Optional[RiskManager] = None status_path = root / cfg.status.path self.status = StatusWriter(status_path, cfg.status.console_every_s) self._queue: asyncio.Queue[BookTop] = asyncio.Queue(maxsize=256) self._last_quote_ts = 0.0 self._last_refresh_ts = 0.0 self._start = time.time() self._stop = asyncio.Event() self._quote_cycles = 0 self._ladder_task: asyncio.Task | None = None self._refresh_task: asyncio.Task | None = None self._momentum = Momentum1m() self._markout = MarkoutTracker( path=root / "data" / "stats_book" / "markout.jsonl", half_bps_ref=float(getattr(cfg.quoting, "half_spread_bps", 1.2) or 1.2), ) self._fills_seen = 0 self._inv_band = "green" self._prev_allow_bid = True self._prev_allow_ask = True self._account_index = int(os.environ["LIGHTER_ACCOUNT_INDEX"]) self._api_key_index = int(os.environ["LIGHTER_API_KEY_INDEX"]) # private key read once — never logged self._api_private_key = os.environ["LIGHTER_API_PRIVATE_KEY"].strip() self._regime_hold = RegimeLeanHold(hold_s=0.0) # Denis: JAMAIS hold — pass-through self._regime = None async def setup(self) -> None: from lighter import SignerClient from .live_engine import LiveEngine from .rate_limit import get_limiter, install_rest_rate_limit from .socks_proxy import install_socks_proxy install_socks_proxy() rl = self.cfg.rate_limit install_rest_rate_limit( max_per_window=rl.max_per_window, window_s=rl.window_s, on_429_backoff_s=rl.on_429_backoff_s, ) lim = get_limiter() log.info( "REST pace target ~%s/%ss (Lighter hard 40/60); 429 backoff %.1fs", rl.max_per_window, int(rl.window_s), rl.on_429_backoff_s, ) self.meta = await discover_market( self.cfg.venue.rest_url, self.cfg.venue.symbol, self.cfg.venue.market_id, ) log.info( "LIVE Market %s id=%s mark=%.1f size_dec=%s price_dec=%s", self.meta.symbol, self.meta.market_id, self.meta.mark_price, self.meta.size_decimals, self.meta.price_decimals, ) self.client = SignerClient( self.cfg.venue.rest_url, self.account_index, {self._api_key_index: self._api_private_key}, ) # scrub local copy reference after client owns it self._api_private_key = "" self.engine = LiveEngine( client=self.client, account_index=self._account_index, api_key_index=self._api_key_index, market_id=self.meta.market_id, price_decimals=self.meta.price_decimals, size_decimals=self.meta.size_decimals, equity_usd=self.cfg.capital.target_working_usd, ) self.engine.markout = self._markout # type: ignore[attr-defined] await self.engine.refresh_account() eq0 = self.engine.equity(self.meta.mark_price or 0) # Keep same-UTC-day baseline if session_state restored it; else set once. self.engine.persist_day_equity(eq0) eq_day = self.engine.state.day_start_equity or eq0 self.risk = RiskManager( self.cfg.risk, eq_day or self.cfg.capital.target_working_usd ) log.info( "LIVE account=%s equity0=%.2f day_equity0=%.2f avail=%.2f inv=%.6f fills=%s", self._account_index, eq0, eq_day, self.engine.state.available_balance, self.engine.state.inventory_btc, self.engine.fills_today(), ) self.feed = MarketDataFeed( rest_url=self.cfg.venue.rest_url, ws_url=self.cfg.venue.ws_url, market_id=self.meta.market_id, on_top=self._enqueue_top, ) @property def account_index(self) -> int: return self._account_index def _enqueue_top(self, top: BookTop) -> None: try: self._queue.put_nowait(top) except asyncio.QueueFull: try: _ = self._queue.get_nowait() except asyncio.QueueEmpty: pass try: self._queue.put_nowait(top) except asyncio.QueueFull: pass async def _process_top(self, top: BookTop) -> None: assert self.engine and self.risk and self.meta self.engine.on_book(top) now = time.time() # Account refresh paced (~12s). NEVER await on the book loop — a full # rate window wait (~60s) used to freeze status.json and trip the # watch into a restart storm (empty book → no fills). refresh_every = float(getattr(self.cfg.quoting, "account_refresh_s", 12.0) or 12.0) # Empty book + tight rate window: skip account polls so place can acquire. empty_book = int(getattr(self.engine, "_working_n_levels", 0) or 0) <= 0 rate_tight = False if empty_book: try: from .rate_limit import get_limiter lim = get_limiter() snap = lim.snapshot() if lim else None if snap: room = int(snap.get("max_per_window") or 37) - int(snap.get("in_window") or 0) rate_tight = room < 6 except Exception: rate_tight = False if (now - self._last_refresh_ts) >= refresh_every: rt = self._refresh_task if rt is None or rt.done(): self._last_refresh_ts = now if empty_book and rate_tight: log.info("skip account refresh — empty book, rate tight (prioritize place)") else: async def _do_refresh(): try: await self.engine.refresh_account() except Exception as e: log.warning("refresh_account: %s", e) self._refresh_task = asyncio.create_task(_do_refresh()) decision = self.risk.evaluate(self.engine, top.mid) interval = self.cfg.quoting.requote_interval_ms / 1000.0 q: Optional[Quote] = self.engine.state.quote if decision.killed: log.error("KILL SWITCH: %s — cancel all + stop", decision.reason) await self.engine.cancel_all() self.engine.set_quote(None) q = None self._write_status(top, q, decision) self.stop() return # Always feed 1m momentum + keep dashboard fresh before any REST wait. mom = _ensure_momentum(self) mom.on_mid(top.mid, now) try: self._markout.on_mid(top.mid, now) except Exception: pass # OFI snapshot each tick (log path + gate); actions only when gate open try: ofi_now = None if self.feed is not None and getattr(self.feed, "book", None) is not None: ofi_now = self.feed.book.ofi(3) self._last_ofi = ofi_now self._markout.last_ofi = ofi_now except Exception: self._last_ofi = getattr(self, "_last_ofi", None) # Warm 1m regime from 5m candle closes so quiet/active works immediately. if len(mom.candles) < 3 and hasattr(self.engine, "candles_list"): try: pts = [] for c in (self.engine.candles_list() or [])[-40:]: ts = float(c.get("t") or c.get("time") or 0) mid = float(c.get("c") or c.get("close") or 0) if ts > 0 and mid > 0: # spread 5m bar into 5 synthetic 1m closes at same mid for k in range(5): pts.append((ts + k * 60.0, mid)) if pts: mom.seed_from_mids(pts) except Exception as e: log.debug("momentum seed: %s", e) self._write_status(top, q, decision) # Health-check cadence: evaluate rest vs refresh. apply_ladder no-ops # (no cancel_all) when mid hasn't moved past rest_threshold_bps. if (now - self._last_quote_ts) >= interval or q is None: allow_bid = decision.allow_bid allow_ask = decision.allow_ask qc = self.cfg.quoting rest_thr = float(getattr(qc, "rest_threshold_bps", None) or 3.0) batch_place = bool(getattr(qc, "batch_place", True)) min_rest_s = float(getattr(qc, "min_rest_s", None) or 40.0) inv = float(self.engine.state.inventory_btc) inv_pol = inventory_policy(inv) self._inv_band = inv_pol.band day_pct = float(self.engine.day_pnl_pct(top.mid)) day_pol = day_pnl_policy(day_pct, inv) allow_bid, allow_ask, _pol_r = merge_allows( allow_bid, allow_ask, inv_pol, day_pol, inv ) allow_bid, allow_ask = side_allow_hysteresis( getattr(self, "_prev_allow_bid", None), getattr(self, "_prev_allow_ask", None), allow_bid, allow_ask, day_pnl_pct=day_pct, inventory_btc=inv, ) self._prev_allow_bid = allow_bid self._prev_allow_ask = allow_ask # OFI overlay (complement): actions only if adverse-OFI markout gate open ofi_now = getattr(self, "_last_ofi", None) ofi_pol = ofi_policy( ofi_now, enabled=bool(self._markout.ofi_actions_enabled()) ) self._ofi_pol_reason = ofi_pol.reason if not ofi_pol.allow_bid: allow_bid = False if not ofi_pol.allow_ask: allow_ask = False # Effective half (≥1.2 floor) + markout widen (already ticked) widen = self._markout.current_widen_bps(now) self._markout.half_bps_ref = float(getattr(qc, "half_spread_bps", 1.2) or 1.2) sigma = 0.0 try: sigma = float(mom.sigma_5m_bps()) except Exception: sigma = 0.0 # mom half_mult already ≥1; fold into effective half override eff_half = effective_half_bps( float(getattr(qc, "half_spread_bps", 1.2) or 1.2), sigma_5m_bps=sigma, markout_widen_bps=widen, day_half_mult=day_pol.half_mult, mom_half_mult=mom.half_spread_mult(), ) self._eff_half = eff_half size_day = float(day_pol.size_mult or 1.0) # Bear/bull size asymmetry BEFORE orders (Denis): crash risk → big bids; # melt-up risk → big asks. Neutral keeps symmetric random band. try: candles = ( self.engine.candles_list() if hasattr(self.engine, "candles_list") else [] ) mids_1m = list(getattr(mom, "candles", []) or []) reg = analyze_regime( candles, mids_1m=mids_1m, lot_usd_min=float(qc.lot_usd_min), lot_usd_max=float(qc.lot_usd_max), ) except Exception as e: log.debug("regime analyze: %s", e) from .regime import RegimeSignal reg = RegimeSignal(reasons=["regime_error"]) reg = self._regime_hold.apply( reg, now=now, lot_usd_min=float(qc.lot_usd_min), lot_usd_max=float(qc.lot_usd_max), ) self._regime = reg lot_bid_usd = reg.lot_bid_usd lot_ask_usd = reg.lot_ask_usd # Only override random when lean is active (keep neutral = random band) # Denis 26/10: regime lean CUT (bilan négatif) — observability only, # lots always symmetric random band. if True or reg.lean == "neutral": lot_bid_usd = None lot_ask_usd = None reg.lot_bid_usd = None reg.lot_ask_usd = None ladder = None if allow_bid or allow_ask: mom_skew = mom.bias_bps() ladder = build_ladder( top, inv, qc, self.meta.price_decimals, self.meta.size_decimals, self.meta.min_base_amount, self.meta.min_quote_amount, allow_bid=allow_bid, allow_ask=allow_ask, extra_skew_bps=mom_skew, half_spread_mult=1.0, # already in eff_half join_extra_ticks=mom.join_extra_ticks(), size_bid_mult=inv_pol.size_bid_mult * size_day * ofi_pol.size_bid_mult, size_ask_mult=inv_pol.size_ask_mult * size_day * ofi_pol.size_ask_mult, half_bid_mult=inv_pol.half_bid_mult, half_ask_mult=inv_pol.half_ask_mult, half_bid_add_bps=ofi_pol.half_bid_add_bps, half_ask_add_bps=ofi_pol.half_ask_add_bps, half_spread_bps_override=eff_half, depth_sizing=True, lot_bid_usd=lot_bid_usd, lot_ask_usd=lot_ask_usd, ) # Non-blocking: REST place/cancel can take many seconds under the # ~37/min pace. Keep the book loop + status heartbeats alive. if self._ladder_task is not None and not self._ladder_task.done(): pass # previous ladder still placing — skip pile-up else: self._last_quote_ts = now self._quote_cycles += 1 if self._quote_cycles == 1: n_lv = 0 if ladder is None else len(ladder.levels) log.info( "First ladder cycle mid=%.1f levels=%s bid=%s ask=%s sz/lvl=%s " "rest_thr=%.2f bps", top.mid, n_lv, None if not ladder else ladder.bid_px, None if not ladder else ladder.ask_px, None if not ladder else ladder.size_btc, rest_thr, ) async def _do_ladder(lad=ladder, ab=allow_bid, aa=allow_ask, rt=rest_thr, bp_=batch_place, mr=min_rest_s, t=top, d=decision): try: await self.engine.apply_ladder( lad, allow_bid=ab and lad is not None, allow_ask=aa and lad is not None, rest_threshold_bps=rt, batch_place=bp_, min_rest_s=mr, ) except Exception as e: log.warning("apply_ladder task: %s", e) finally: try: self._write_status(t, self.engine.state.quote, d) except Exception: pass self._ladder_task = asyncio.create_task(_do_ladder()) dur = self.cfg.run.duration_s if dur and (now - self._start) >= dur: log.info("duration_s=%.0f reached — cancel all + stop", dur) await self.engine.cancel_all() self.stop() def _write_status(self, top: BookTop, q: Optional[Quote], decision) -> None: assert self.engine mid = top.mid try: from .rate_limit import get_limiter _lim = get_limiter() rl_snap = _lim.snapshot() if _lim else None except Exception: rl_snap = None payload = { "mode": "live", "symbol": self.meta.symbol if self.meta else self.cfg.venue.symbol, "market_id": self.meta.market_id if self.meta else None, "account_index": self._account_index, "mid": mid, "bid": top.bid, "ask": top.ask, "ladder_orders": getattr(self.engine.state, "ladder_orders", 0), "ladder_bids": getattr(self.engine.state, "ladder_bids", 0), "ladder_asks": getattr(self.engine.state, "ladder_asks", 0), "ladder_refresh": getattr(self.engine.state, "last_refresh_reason", ""), "ladder_levels": ( self.engine.ladder_levels_for_status() if hasattr(self.engine, "ladder_levels_for_status") else [] ), "quote_target": ( self.engine.quote_target_for_status() if hasattr(self.engine, "quote_target_for_status") else None ), "quotes_live": int(getattr(self.engine.state, "ladder_orders", 0) or 0) > 0, "quote": ( None if ( q is None or int(getattr(self.engine.state, "ladder_orders", 0) or 0) <= 0 ) else { "bid_px": float(getattr(self.engine, "_working_bid0", 0) or q.bid_px), "ask_px": float(getattr(self.engine, "_working_ask0", 0) or q.ask_px), "size_btc": q.size_btc, "skew_bps": q.skew_bps, } ), "working_mid": getattr(self.engine, "_working_mid", 0.0), "working_bid": getattr(self.engine, "_working_bid0", 0.0), "working_ask": getattr(self.engine, "_working_ask0", 0.0), "book_bid": top.bid, "book_ask": top.ask, "n_bid_levels": int(getattr(self.cfg.quoting, "n_bid_levels", 1)), "n_ask_levels": int(getattr(self.cfg.quoting, "n_ask_levels", 1)), "level_spacing_bps": float(getattr(self.cfg.quoting, "level_spacing_bps", 0.5)), "cancel_all_count": int(getattr(self.engine, "_cancel_all_count", 0)), "rest_skip_count": int(getattr(self.engine, "_rest_skip_count", 0)), "inventory_btc": self.engine.state.inventory_btc, "inventory_usd": self.engine.state.inventory_btc * mid, "cash_usd": self.engine.state.cash_usd, "equity": self.engine.equity(mid), "equity0": self.engine.state.equity0, "capital_usd": self.engine.equity(mid), "capital0_usd": self.engine.state.equity0, "available_balance": self.engine.state.available_balance, "collateral": self.engine.state.collateral, "unrealized_pnl": self.engine.state.unrealized_pnl, "pending_orders": self.engine.state.pending_orders, "realized_pnl": self.engine.state.realized_pnl, "day_pnl": self.engine.day_pnl(mid), "day_pnl_pct": self.engine.day_pnl_pct(mid), "fills": self.engine.fills_today(), "fills_buy": self.engine.fills_buy_count(), "fills_sell": self.engine.fills_sell_count(), "killed": decision.killed, "risk_reason": decision.reason, "last_error": self.engine.state.last_error, "rate_limit": rl_snap, "quote_cycles": self._quote_cycles, "lot_usd": [self.cfg.quoting.lot_usd_min, self.cfg.quoting.lot_usd_max], "half_spread_bps": float(getattr(self.cfg.quoting, "half_spread_bps", 0.0)), "rest_threshold_bps": float(getattr(self.cfg.quoting, "rest_threshold_bps", 0.0)), "min_rest_s": float(getattr(self.cfg.quoting, "min_rest_s", 0.0)), "max_inventory_btc": float(getattr(self.cfg.quoting, "max_inventory_btc", 0.0)), "kill_day_pnl_pct": float(getattr(self.cfg.risk, "kill_day_pnl_pct", -1.0)), "soft_leverage_min": float(getattr(self.cfg.risk, "soft_leverage_min", 0.0)), "soft_leverage_max": float(getattr(self.cfg.risk, "soft_leverage_max", 0.0)), "max_notional_usd": float(getattr(self.cfg.risk, "max_notional_usd", 0.0)), "soft_leverage": ( (abs(float(self.engine.state.inventory_btc) * float(mid or 0.0)) / max(float(self.engine.equity(mid) or 0.0), 1e-9)) if mid else 0.0 ), "uptime_s": time.time() - self._start, "ws_ok": bool(self.feed and self.feed.ws_ok), "open_positions": self.engine.open_positions(), # Net inventory sides — source of truth for dashboard L / S (not ladder bids) "n_long": 1 if float(self.engine.state.inventory_btc) > 1e-8 else 0, "n_short": 1 if float(self.engine.state.inventory_btc) < -1e-8 else 0, "position_side": ( "long" if float(self.engine.state.inventory_btc) > 1e-8 else ( "short" if float(self.engine.state.inventory_btc) < -1e-8 else "flat" ) ), "close_events": self.engine.close_events(30), "avg_buy_px": self.engine.avg_buy_px(), "avg_sell_px": self.engine.avg_sell_px(), "price_history": self.engine.price_history_list(), "candles": self.engine.candles_list() if hasattr(self.engine, "candles_list") else [], "candle_sec": 300, "momentum_1m": self._momentum.snapshot() if getattr(self, "_momentum", None) else None, "regime": ( self._regime.snapshot() if getattr(self, "_regime", None) is not None and hasattr(self._regime, "snapshot") else None ), "regime_lean": getattr(getattr(self, "_regime", None), "lean", None), "regime_score": ( float(getattr(self._regime, "score", 0.0)) if getattr(self, "_regime", None) is not None else None ), "lot_bid_usd": getattr(getattr(self, "_regime", None), "lot_bid_usd", None), "lot_ask_usd": getattr(getattr(self, "_regime", None), "lot_ask_usd", None), "inv_band": getattr(self, "_inv_band", "green"), "markout": self._markout.snapshot() if getattr(self, "_markout", None) else None, "ofi": getattr(self, "_last_ofi", None), "ofi_policy": getattr(self, "_ofi_pol_reason", None), "half_spread_bps_effective": float(getattr(self, "_eff_half", 0) or 0), # Persisted across reboot: last ~5h / up to 500 fills (UTC day store) "last_fills": self.engine.last_fills_for_status(), "session_day": ( self.engine._session.day if getattr(self.engine, "_session", None) is not None else None ), "session_day_tz": "UTC", } self.status.write(payload) def stop(self) -> None: self._stop.set() if self.feed: self.feed.stop() async def _status_heartbeat_loop(self) -> None: """Independent of book/REST waits — status must stay fresh for watchers.""" while not self._stop.is_set(): try: await asyncio.sleep(2.0) if self._stop.is_set(): break last = self.feed.last_top if self.feed else None if last is None or not last.ok() or not self.risk or not self.engine: continue mom = _ensure_momentum(self) mom.on_mid(last.mid, time.time()) d = self.risk.evaluate(self.engine, last.mid) self._write_status(last, self.engine.state.quote, d) except asyncio.CancelledError: raise except Exception as e: log.debug("status heartbeat loop: %s", e) async def run(self) -> None: await self.setup() assert self.feed is not None and self.engine is not None feed_task = asyncio.create_task(self.feed.run()) hb_task = asyncio.create_task(self._status_heartbeat_loop()) try: while not self._stop.is_set(): try: top = await asyncio.wait_for(self._queue.get(), timeout=0.5) except asyncio.TimeoutError: dur = self.cfg.run.duration_s if dur and (time.time() - self._start) >= dur: log.info("duration_s=%.0f reached — stopping", dur) await self.engine.cancel_all() self.stop() # Heartbeat: keep status.json fresh even if book queue is quiet # or previous apply_ladder is between cycles. last = self.feed.last_top if self.feed else None if last is not None and last.ok() and self.risk: try: mom = _ensure_momentum(self) mom.on_mid(last.mid, time.time()) d = self.risk.evaluate(self.engine, last.mid) self._write_status(last, self.engine.state.quote, d) except Exception as e: log.debug("status heartbeat: %s", e) continue await self._process_top(top) finally: hb_task.cancel() try: await hb_task except asyncio.CancelledError: pass try: await self.engine.cancel_all() except Exception as e: log.warning("final cancel_all: %s", e) top = self.feed.last_top was_ws = bool(self.feed.ws_ok) self.feed.stop() feed_task.cancel() try: await feed_task except asyncio.CancelledError: pass if top.ok() and self.risk: d = self.risk.evaluate(self.engine, top.mid) self.feed.ws_ok = was_ws self._write_status(top, self.engine.state.quote, d) if self.client: try: await self.client.close() except Exception: pass log.info( "LIVE stopped. cycles=%s fills=%s equity=%.2f dayPnL=%.2f", self._quote_cycles, self.engine.fills_today(), self.engine.equity(top.mid if top.ok() else 0), self.engine.day_pnl(top.mid if top.ok() else 0), )