#!/usr/bin/env python3 """Executor 1h supervisor: every ~25s check status/watch; fix structural issues. Does not talk to Denis. Restarts ONLY via restart_live_safe. Coordinates peers. """ from __future__ import annotations import json import subprocess import time from pathlib import Path ROOT = Path("/workspace/lighter-mm") DATA = ROOT / "data" LOG = DATA / "hour_supervisor.log" REPORT = DATA / "hour_supervisor_report.json" STATUS = DATA / "status.json" WATCH_STATE = DATA / "watch_5s_1h_state.json" VENV = ROOT / ".venv" / "bin" / "python" SAFE = ROOT / "scripts" / "restart_live_safe.py" END_S = float((DATA / "hour_supervisor_end.txt").read_text().strip()) if (DATA / "hour_supervisor_end.txt").exists() else time.time() + 3300 def log(msg: str) -> None: line = f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}" with LOG.open("a") as f: f.write(line + "\n") print(line, flush=True) def pgrep(pat: str) -> list[str]: try: out = subprocess.check_output(["pgrep", "-af", pat], text=True) return [l for l in out.splitlines() if l.strip() and "hour_supervisor" not in l and "pgrep" not in l] except subprocess.CalledProcessError: return [] def assert_watch_constants() -> bool: """Peer agents sometimes overwrite watch_5s_1h.py — put back non-hair-trigger thresholds.""" wp = ROOT / "scripts" / "watch_5s_1h.py" try: txt = wp.read_text() except Exception: return False changed = False import re def set_const(name, val): nonlocal txt, changed m = re.search(rf'^{name}\s*=\s*(\S+)', txt, flags=re.M) if not m: return if m.group(1).split()[0] != str(val): txt = re.sub(rf'^{name}\s*=\s*\S+', f'{name} = {val}', txt, count=1, flags=re.M) changed = True set_const('TICK_S', 1) set_const('STALE_S', 90) set_const('BOOT_GRACE_S', 90) set_const('RESTART_COOLDOWN_S', 90) if changed: wp.write_text(txt) log(f"FIX watch_constants restored TICK=1 STALE=90 — reload watch") # reload watch preserving ends try: ws = json.loads(WATCH_STATE.read_text()) rem = max(60, int(ws.get("ends_at", time.time() + 600) - time.time())) except Exception: rem = max(60, int(END_S - time.time())) for l in pgrep("scripts/watch_5s_1h.py"): try: pid = int(l.split()[0]) import os as _os _os.kill(pid, 15) except Exception: pass time.sleep(1) subprocess.Popen( [str(VENV), "-B", str(ROOT / "scripts" / "watch_5s_1h.py")], cwd=str(ROOT), env={**dict(__import__("os").environ), "WATCH_DURATION_S": str(rem)}, stdout=open(DATA / "watch_5s_1h.nohup.log", "a"), stderr=subprocess.STDOUT, ) return True return False def main() -> int: DATA.mkdir(parents=True, exist_ok=True) t0 = time.time() actions = [] snaps = [] s0 = None try: s0 = json.loads(STATUS.read_text()) except Exception: pass log(f"START end_in={END_S - time.time():.0f}s fills0={(s0 or {}).get('fills')} day0={(s0 or {}).get('day_pnl')}") while time.time() < END_S: try: s = json.loads(STATUS.read_text()) age = time.time() - float(s.get("updated_at") or 0) except Exception as e: s, age = None, 1e9 log(f"status_read_err {e}") lives = [l for l in pgrep("scripts/run_live.py") if "python" in l] watches = [l for l in pgrep("scripts/watch_5s_1h.py") if "python" in l] peers = pgrep("restart_live_safe.py") ws = {} try: ws = json.loads(WATCH_STATE.read_text()) except Exception: pass mom = (s or {}).get("momentum_1m") or {} rl = (s or {}).get("rate_limit") or {} bids = int((s or {}).get("ladder_bids") or 0) asks = int((s or {}).get("ladder_asks") or 0) fills = (s or {}).get("fills") # fills_2m now = time.time() fills_2m = 0 for f in (s or {}).get("last_fills") or []: ts = f.get("ts") or f.get("time") if ts is None: continue ts = float(ts) if ts > 1e11: ts /= 1000 if now - ts < 120: fills_2m += 1 mid = float((s or {}).get("mid") or 0) wb = float((s or {}).get("working_bid") or 0) wa = float((s or {}).get("working_ask") or 0) gap = None if mid > 0 and (wb > 0 or wa > 0): gaps = [] if wb > 0: gaps.append(abs(mid - wb) / mid * 1e4) if wa > 0: gaps.append(abs(wa - mid) / mid * 1e4) gap = max(gaps) if gaps else None snap = { "ts": now, "age": round(age, 1), "ladder": [bids, asks], "fills": fills, "fills_2m": fills_2m, "day": (s or {}).get("day_pnl"), "inv": (s or {}).get("inventory_btc"), "gap": gap, "half_mult": mom.get("half_mult"), "regime": mom.get("regime"), "join_extra": mom.get("join_extra_ticks"), "rate": f"{rl.get('in_window')}/{rl.get('max_per_window')}", "n_live": len(lives), "n_watch": len(watches), "refresh": (s or {}).get("ladder_refresh"), "killed": (s or {}).get("killed"), } snaps.append(snap) snaps = snaps[-200:] if int(time.time()) % 120 < 30: assert_watch_constants() # Fix paths if len(lives) == 0: if peers: log("FIX skip process_dead — peer restart running") else: log("FIX process_dead → restart_live_safe") actions.append({"ts": now, "act": "restart", "reason": "process_dead"}) subprocess.run( [str(VENV), "-B", str(SAFE), "--reason", "hour_sup_process_dead"], cwd=str(ROOT), timeout=120, ) time.sleep(20) elif len(lives) > 1: if not peers: log(f"FIX multi_live n={len(lives)} → restart_live_safe dedupe") actions.append({"ts": now, "act": "restart", "reason": "multi_live"}) subprocess.run( [str(VENV), "-B", str(SAFE), "--reason", "hour_sup_dedupe"], cwd=str(ROOT), timeout=120, ) time.sleep(20) elif age > 150 and not peers: log(f"FIX status_age={age:.0f}s → restart_live_safe") actions.append({"ts": now, "act": "restart", "reason": f"stale_{age:.0f}"}) subprocess.run( [str(VENV), "-B", str(SAFE), "--reason", "hour_sup_status_stale"], cwd=str(ROOT), timeout=120, ) time.sleep(20) elif not watches: rem = max(60, int(END_S - time.time())) log(f"FIX watch_dead → relaunch remain={rem}s") actions.append({"ts": now, "act": "relaunch_watch", "remain": rem}) subprocess.Popen( [str(VENV), "-B", str(ROOT / "scripts" / "watch_5s_1h.py")], cwd=str(ROOT), env={**dict(__import__("os").environ), "WATCH_DURATION_S": str(rem)}, stdout=open(DATA / "watch_5s_1h.nohup.log", "a"), stderr=subprocess.STDOUT, ) # Structural: quiet half_mult must be 1.0 if mom.get("regime") == "quiet" and mom.get("half_mult") not in (None, 1.0): log(f"STRUCT quiet half_mult={mom.get('half_mult')} — code regression!") actions.append({"ts": now, "act": "alert_quiet_mult", "half": mom.get("half_mult")}) if bids >= 4 and asks >= 4 and fills_2m == 0 and float((s or {}).get("uptime_s") or 0) > 120: log( f"WARN fills_starve_2m gap={gap} half={mom.get('half_mult')} " f"reg={mom.get('regime')} ref={(s or {}).get('ladder_refresh')} " f"rate={rl.get('in_window')}/{rl.get('max_per_window')}" ) if int(time.time()) % 60 < 28: # ~once/min in 25s loop log( f"HB L={bids}+{asks} fills={fills} f2m={fills_2m} day={(s or {}).get('day_pnl')} " f"age={age:.1f} gap={None if gap is None else round(gap,2)} " f"reg={mom.get('regime')} half={mom.get('half_mult')} " f"rate={rl.get('in_window')}/{rl.get('max_per_window')} " f"live={len(lives)} watch={len(watches)}" ) REPORT.write_text( json.dumps( { "started": t0, "ends": END_S, "actions": actions, "last": snap, "fills_start": (s0 or {}).get("fills"), "day_start": (s0 or {}).get("day_pnl"), "snaps_n": len(snaps), }, indent=2, ) ) time.sleep(25) try: s = json.loads(STATUS.read_text()) except Exception: s = {} summary = { "started": t0, "ended": time.time(), "fills_start": (s0 or {}).get("fills"), "fills_end": s.get("fills"), "day_start": (s0 or {}).get("day_pnl"), "day_end": s.get("day_pnl"), "equity_end": s.get("equity"), "inv_end": s.get("inventory_btc"), "actions": actions, "last": snaps[-1] if snaps else None, "n_snaps": len(snaps), } REPORT.write_text(json.dumps(summary, indent=2)) log( f"END fills={(s0 or {}).get('fills')}->{s.get('fills')} " f"day={(s0 or {}).get('day_pnl')}->{s.get('day_pnl')} actions={len(actions)}" ) return 0 if __name__ == "__main__": raise SystemExit(main())