diff --git a/research/grade_oracle.py b/research/grade_oracle.py index dfa35d09..a2289fb8 100644 --- a/research/grade_oracle.py +++ b/research/grade_oracle.py @@ -21,10 +21,28 @@ FLYCTL = shutil.which("flyctl") or "/opt/homebrew/bin/flyctl" EDGE_GRID = (0.04, 0.07, 0.10) # cells keyed str(E): "0.04"/"0.07"/"0.1" +def sftp(remote, local): + subprocess.run([FLYCTL, "ssh", "sftp", "get", remote, local + ".new", + "-a", "wwf-oraclebot"], capture_output=True, + timeout=600, stdin=subprocess.DEVNULL) + if os.path.exists(local + ".new") and os.path.getsize(local + ".new") > 0: + os.replace(local + ".new", local) # keep last good pull on failure + return True + try: + os.remove(local + ".new") + except FileNotFoundError: + pass + return False + + def main(): r = subprocess.run([FLYCTL, "ssh", "sftp", "get", "/data/oracle_state.json", TMP, "-a", "wwf-oraclebot"], capture_output=True, timeout=300, stdin=subprocess.DEVNULL) + sftp("/data/oracle_attempts.jsonl", + os.path.join(HERE, ".oracle_attempts.pull.jsonl")) + sftp("/data/oracle_markouts.jsonl", + os.path.join(HERE, ".oracle_markouts.pull.jsonl")) if not os.path.exists(TMP) or os.path.getsize(TMP) == 0: print("[grade_oracle] box unreachable or no state yet — skip") return 0 diff --git a/research/grade_surge.py b/research/grade_surge.py index cfecae92..d854b86f 100644 --- a/research/grade_surge.py +++ b/research/grade_surge.py @@ -40,6 +40,8 @@ def sftp(remote, local): def main(): ok = sftp("/data/surge2_state.json", STATE_PULL) sftp("/data/surge_attempts.jsonl", ATT_PULL) + sftp("/data/surge_markouts.jsonl", + os.path.join(HERE, ".surge_markouts.pull.jsonl")) if not ok: print("[grade_surge] box unreachable or no state yet — skip") return 0 diff --git a/research/oraclebot.py b/research/oraclebot.py index 161a41bb..3a13f74f 100644 --- a/research/oraclebot.py +++ b/research/oraclebot.py @@ -34,9 +34,19 @@ Clocks: prints are stamped at ARRIVAL and crypto ticks at the payload's venue ms timestamp — the same two clocks the tape records (recorder stamps trades on arrival; load_ticks prefers payload ms). Staleness guards use the arrival clock only, so venue clock skew can't fake freshness. + +OBSERVATIONAL instrumentation (2026-07-22, NOT semantics — SEM_VER +unchanged): every attempt (fill/crater/fail/throttle) appends a line with +top-5 asks + top-3 bids + fair/edge to /data/oracle_attempts.jsonl (real +book depth at mispricing moments — maker-study groundwork; state's skips +list trims at 500 and loses this otherwise), and every fill schedules book +re-reads at +60s/+300s/+1800s (skipping offsets past t1) whose best +bid/ask land in /data/oracle_markouts.jsonl. Pending re-reads die on +restart (counted, acceptable — observation only). """ import bisect import calendar +import heapq import json import math import os @@ -53,8 +63,13 @@ import websocket WS_URL = "wss://ws-live-data.polymarket.com" CLOB = "https://clob.polymarket.com" STATE = os.environ.get("ORACLE_STATE", "/data/oracle_state.json") +ATTEMPTS = os.environ.get("ORACLE_ATTEMPTS", "/data/oracle_attempts.jsonl") +MARKOUTS = os.environ.get("ORACLE_MARKOUTS", "/data/oracle_markouts.jsonl") DEBUG = bool(os.environ.get("ORACLE_DEBUG")) SEM_VER = "b1" # bump on ANY semantics-altering change +MARKOUT_OFFSETS = (60, 300, 1800) # observational re-reads per fill +BID_LEVELS = 3 +TOP_LEVELS = 5 # raw ask levels recorded per attempt # ── FROZEN — must equal study_oracle.py verbatim ──────────────────────────── EDGE_GRID = [0.04, 0.07, 0.10] @@ -218,8 +233,9 @@ def walk_asks(asks, cap, stake_usd=STAKE): gets). The first level's price is also the best-ask-only read (~$5- deployable). asks arrive UNSORTED from /book.""" lv = sorted(((float(a["price"]), float(a["size"])) for a in asks or [])) + top = [[p, s] for p, s in lv[:TOP_LEVELS]] if not lv: - return {"filled": False, "best_ask": None} + return {"filled": False, "best_ask": None, "top": top} best = lv[0][0] spent = shares = 0.0 levels = 0 @@ -233,9 +249,9 @@ def walk_asks(asks, cap, stake_usd=STAKE): spent += take * px levels += 1 if not shares: - return {"filled": False, "best_ask": best} + return {"filled": False, "best_ask": best, "top": top} return {"filled": True, "vwap": spent / shares, "shares": shares, - "usd": spent, "levels": levels, "best_ask": best, + "usd": spent, "levels": levels, "best_ask": best, "top": top, "partial": spent < stake_usd - 1e-9} @@ -339,6 +355,7 @@ class OracleBot: self.last_trig = {} # not persisted — see WARMUP_S comment self.lock = threading.Lock() self.book_q = queue.Queue(BOOK_QUEUE) + self.mo_heap = [] # (due_ts, asset, fill_id, offset) self.trades_seen = 0 # unlocked cosmetic counter (firehose # rate) — folded into state at heartbeat @@ -445,6 +462,7 @@ class OracleBot: if edge >= E: self.state["cells"][str(E)]["throttle"] += 1 self._skip(job, "throttle", None) + self.log_attempt({**job, "filled": False, "why": "throttle"}) # ── paper execution (workers; network OUT of the lock) ───────────────── def book_worker(self): @@ -463,10 +481,17 @@ class OracleBot: with self.lock: self.state["counters"]["skip"]["book_fail"] += 1 self._skip(job, "book_fail", None) + self.log_attempt({**job, "filled": False, "why": "book_fail"}) return lat_ms = int((time.time() - t_req) * 1000) cap = min(job["p_ref"] * (1 + SLIP_CAP), 0.99) r = walk_asks(book.get("asks"), cap) + bid_top = sorted(((float(b["price"]), float(b["size"])) + for b in book.get("bids") or []), + reverse=True)[:BID_LEVELS] + rec = {**job, "cap": round(cap, 4), "latency_ms": lat_ms, + "top": r["top"], "bid_top": [[p, s] for p, s in bid_top], + "best_ask": r["best_ask"], "filled": r["filled"]} with self.lock: c = self.state["counters"] if not r["filled"]: @@ -479,26 +504,37 @@ class OracleBot: log(f"CRATER {job['title'][:38]} ref {job['p_ref']:.3f} " f"ask {ba} edge {job['edge']:+.3f} ({lat_ms}ms)") self.persist() - return - fee = FEE_RATE * r["shares"] * min(r["vwap"], 1 - r["vwap"]) - lot = {**job, "cap": round(cap, 4), "price": round(r["vwap"], 5), - "best_ask": r["best_ask"], - "shares": round(r["shares"], 4), - "cost": round(r["usd"], 2), "fee": round(fee, 4), - "levels": r["levels"], "partial": r["partial"], - "latency_ms": lat_ms} - self.state["open"][f"{job['asset']}:{job['ts']}"] = lot - c["fills"] += 1 - for E in EDGE_GRID: - if job["edge"] >= E: - cell = self.state["cells"][str(E)] - cell["fills"] += 1 - cell["staked"] = round(cell["staked"] + lot["cost"], 2) - log(f"FILL {job['outcome']} · {job['title'][:38]} @ " - f"{lot['price']:.3f} (${lot['cost']:.2f}" - f"{' partial' if lot['partial'] else ''}, " - f"edge {job['edge']:+.3f}, {lat_ms}ms)") - self.persist() + else: + fee = FEE_RATE * r["shares"] * min(r["vwap"], 1 - r["vwap"]) + lot = {**job, "cap": round(cap, 4), + "price": round(r["vwap"], 5), + "best_ask": r["best_ask"], + "shares": round(r["shares"], 4), + "cost": round(r["usd"], 2), "fee": round(fee, 4), + "levels": r["levels"], "partial": r["partial"], + "latency_ms": lat_ms} + self.state["open"][f"{job['asset']}:{job['ts']}"] = lot + c["fills"] += 1 + for E in EDGE_GRID: + if job["edge"] >= E: + cell = self.state["cells"][str(E)] + cell["fills"] += 1 + cell["staked"] = round(cell["staked"] + lot["cost"], 2) + rec.update({"price": lot["price"], "shares": lot["shares"], + "usd": lot["cost"], "fee": lot["fee"], + "partial": lot["partial"]}) + for off in MARKOUT_OFFSETS: # observational exit marks + if job["t1"] and job["ts"] + off > job["t1"]: + continue + heapq.heappush(self.mo_heap, + (job["ts"] + off, job["asset"], + f"{job['asset']}:{job['ts']}", off)) + log(f"FILL {job['outcome']} · {job['title'][:38]} @ " + f"{lot['price']:.3f} (${lot['cost']:.2f}" + f"{' partial' if lot['partial'] else ''}, " + f"edge {job['edge']:+.3f}, {lat_ms}ms)") + self.persist() + self.log_attempt(rec) def _skip(self, job, why, ba): # under lock self.state["skips"].append( @@ -507,6 +543,51 @@ class OracleBot: "title": job["title"]}) del self.state["skips"][:-SKIPS_TRIM] + def log_attempt(self, rec): + """Append-only full attempt stream (observational — maker-study + groundwork; the state's skips list trims and loses book detail).""" + try: + with open(ATTEMPTS, "a") as fh: + fh.write(json.dumps(rec) + "\n") + except Exception as e: + log(f"⚠ attempts log write failed: {e}") + + def markout_worker(self): + """Observational: due book re-reads -> best bid/ask marks. Never + touches the signal path; pending marks die on restart (counted).""" + while True: + due = None + with self.lock: + if self.mo_heap and self.mo_heap[0][0] <= time.time(): + due = heapq.heappop(self.mo_heap) + if due is None: + time.sleep(2) + continue + due_ts, asset, fid, off = due + rec = {"fill_id": fid, "asset": asset, "offset_s": off, + "due_ts": round(due_ts, 1), + "read_ts": round(time.time(), 3)} + try: + book = get_json(f"{CLOB}/book?token_id={asset}") + bids = sorted(((float(b["price"]), float(b["size"])) + for b in book.get("bids") or []), reverse=True) + asks = sorted(((float(a["price"]), float(a["size"])) + for a in book.get("asks") or [])) + rec["bid"], rec["bid_sz"] = (bids[0] if bids else (None, None)) + rec["ask"], rec["ask_sz"] = (asks[0] if asks else (None, None)) + key = "mo_done" + except Exception as e: + rec["err"] = str(e)[:40] + key = "mo_err" + with self.lock: + c = self.state["counters"] + c[key] = c.get(key, 0) + 1 + try: + with open(MARKOUTS, "a") as fh: + fh.write(json.dumps(rec) + "\n") + except Exception as e: + log(f"⚠ markouts log write failed: {e}") + # ── settlement ───────────────────────────────────────────────────────── # (a) instant, from the same feed the venue settles on (under lock) def settle_provisional(self): @@ -774,6 +855,8 @@ def main(): threading.Thread(target=serve_feed, args=(bot,), daemon=True).start() for _ in range(BOOK_WORKERS): threading.Thread(target=bot.book_worker, daemon=True).start() + for _ in range(2): + threading.Thread(target=bot.markout_worker, daemon=True).start() for tag in ("a", "b"): threading.Thread(target=run_conn, args=(tag, bot), daemon=True).start() n = 0 diff --git a/research/surgebot.py b/research/surgebot.py index faa58f7c..bf40abeb 100644 --- a/research/surgebot.py +++ b/research/surgebot.py @@ -34,7 +34,15 @@ the nightly CTF payout re-grade (grade_surge.py -> surge_meas_ledger.jsonl; v1's surge_paper_ledger.jsonl is closed and untouched, as is its state at /data/surge_state.json). Any semantics-altering fix bumps SEM_VER and restarts the shakedown clock. + +OBSERVATIONAL instrumentation (2026-07-22, NOT semantics — SEM_VER +unchanged): attempts also record top-3 bids (spread context; exit-fill +realism), and every fill schedules book re-reads at +60s/+300s/+1800s whose +best bid/ask land in /data/surge_markouts.jsonl — real exit marks for the +markout-exit study instead of print-inferred guesses. Pending re-reads are +lost on restart (counted, acceptable — observation only). """ +import heapq import json import os import queue @@ -52,7 +60,10 @@ SET_URL = ("https://raw.githubusercontent.com/jaxperro/winning-wallet-finder/" "main/research/params/informed_set.json") STATE = os.environ.get("SURGE_STATE", "/data/surge2_state.json") ATTEMPTS = os.environ.get("SURGE_ATTEMPTS", "/data/surge_attempts.jsonl") +MARKOUTS = os.environ.get("SURGE_MARKOUTS", "/data/surge_markouts.jsonl") SEM_VER = "a2" # bump on ANY semantics-altering change +MARKOUT_OFFSETS = (60, 300, 1800) # observational re-reads per fill +BID_LEVELS = 3 # ── FROZEN — must equal params/study_flow.json / forward.py ──────────────── FLOW_USD = 300.0 @@ -174,6 +185,7 @@ class Surge: self.last_trig = {} self.lock = threading.Lock() self.book_q = queue.Queue(BOOK_QUEUE) + self.mo_heap = [] # (due_ts, asset, fill_id, offset) # ── config ───────────────────────────────────────────────────────────── def load_set(self): @@ -254,6 +266,9 @@ class Surge: lat_ms = int((time.time() - t_req) * 1000) cap = min(job["p_ref"] * (1 + SLIP_CAP), 0.99) r = walk_asks(book.get("asks"), cap) + bid_top = sorted(((float(b["price"]), float(b["size"])) + for b in book.get("bids") or []), + reverse=True)[:BID_LEVELS] end_ts = None # expected resolution (dashboard ETA) if r["filled"]: try: @@ -262,8 +277,8 @@ class Surge: except Exception: pass rec = {**job, "cap": round(cap, 4), "latency_ms": lat_ms, - "top": r["top"], "best_ask": r["best_ask"], - "filled": r["filled"]} + "top": r["top"], "bid_top": [[p, s] for p, s in bid_top], + "best_ask": r["best_ask"], "filled": r["filled"]} with self.lock: c = self.state["counters"] if not r["filled"]: @@ -285,6 +300,12 @@ class Surge: rec.update({"price": lot["price"], "shares": lot["shares"], "usd": lot["cost"], "fee": lot["fee"], "partial": lot["partial"], "end_ts": end_ts}) + for off in MARKOUT_OFFSETS: # observational exit marks + if end_ts and job["ts"] + off > end_ts: + continue + heapq.heappush(self.mo_heap, + (job["ts"] + off, job["asset"], + f"{job['asset']}:{job['ts']}", off)) log(f"FILL {job['outcome']} · {job['title'][:38]} @ " f"{lot['price']:.3f} (${lot['cost']:.2f}" f"{' partial' if lot['partial'] else ''}, " @@ -301,6 +322,42 @@ class Surge: except Exception as e: log(f"⚠ attempts log write failed: {e}") + def markout_worker(self): + """Observational: due book re-reads -> best bid/ask marks. Never + touches the signal path; pending marks die on restart (counted).""" + while True: + due = None + with self.lock: + if self.mo_heap and self.mo_heap[0][0] <= time.time(): + due = heapq.heappop(self.mo_heap) + if due is None: + time.sleep(2) + continue + due_ts, asset, fid, off = due + rec = {"fill_id": fid, "asset": asset, "offset_s": off, + "due_ts": round(due_ts, 1), + "read_ts": round(time.time(), 3)} + try: + book = get_json(f"{CLOB}/book?token_id={asset}") + bids = sorted(((float(b["price"]), float(b["size"])) + for b in book.get("bids") or []), reverse=True) + asks = sorted(((float(a["price"]), float(a["size"])) + for a in book.get("asks") or [])) + rec["bid"], rec["bid_sz"] = (bids[0] if bids else (None, None)) + rec["ask"], rec["ask_sz"] = (asks[0] if asks else (None, None)) + key = "mo_done" + except Exception as e: + rec["err"] = str(e)[:40] + key = "mo_err" + with self.lock: + c = self.state["counters"] + c[key] = c.get(key, 0) + 1 + try: + with open(MARKOUTS, "a") as fh: + fh.write(json.dumps(rec) + "\n") + except Exception as e: + log(f"⚠ markouts log write failed: {e}") + def _skip(self, job, why, ba): # under lock self.state["skips"].append( {"ts": int(job["ts"]), "asset": job["asset"], "why": why, @@ -494,6 +551,8 @@ def main(): threading.Thread(target=serve_feed, args=(bot,), daemon=True).start() for _ in range(BOOK_WORKERS): threading.Thread(target=bot.book_worker, daemon=True).start() + for _ in range(2): + threading.Thread(target=bot.markout_worker, daemon=True).start() for tag in ("a", "b"): threading.Thread(target=run_conn, args=(tag, bot), daemon=True).start() last_set = time.time()