research: oraclebot — Study B real-time paper harness (wwf-oraclebot) + nightly chain grading

Mirrors forward.py::score_oracle semantics verbatim (constants frozen from
study_oracle/tape); live-only suppress-only guards; $100 FAK walks the live
asks inside p_ref*1.05; three settle layers (own-feed tick, CLOB flags,
nightly CTF re-grade via grade_oracle.py -> oracle_paper_ledger.jsonl).
Shakedown 2026-07-22: 52 events/9 fills/43 craters at p50 541ms; 30/30
event agreement vs tape scorer on the comparable set. Verdict still binds
to forward_ledger only (#17).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

fly.oraclebot.toml

@@ -0,0 +1,29 @@
+# wwf-oraclebot — Study B crypto-oracle PAPER harness (research silo).
+# No keys, no orders, no repo clone: code baked in image (recorder pattern).
+# Believing its numbers is gated on the #17 forward verdict; the harness
+# measures real-time capturability only. State on the volume; the Mac
+# nightly pulls + grades (grade_oracle.py).
+app = "wwf-oraclebot"
+primary_region = "arn"
+
+[build]
+  dockerfile = "research/oraclebot.Dockerfile"
+
+[[mounts]]
+  source = "oracle"
+  destination = "/data"
+
+# read-only paper feed for jaxperro.com/oracle (GET /feed, CORS-open)
+[http_service]
+  internal_port = 8080
+  force_https = true
+  auto_stop_machines = false
+  auto_start_machines = true
+  min_machines_running = 1
+
+[[restart]]
+  policy = "always"
+
+[[vm]]
+  size = "shared-cpu-1x"
+  memory = "256mb"

research/grade_oracle.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Nightly chain-truth grading of the oracle paper harness (wwf-oraclebot).
+
+Pulls /data/oracle_state.json off the box (read-only — never writes back),
+re-grades every settled entry with CTF payout vectors (payouts.truth; the
+harness's provisional settles — own-feed ticks and CLOB winner flags — can
+both be wrong on operator-resolved/refunded markets), and appends one row
+per settle to research/oracle_paper_ledger.jsonl (keyed asset+ts,
+idempotent). The per-E-cell rollup printed here is the CAPTURABILITY read —
+it is NOT the #17 verdict (forward_ledger.jsonl is)."""
+import json
+import os
+import shutil
+import subprocess
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+LEDGER = os.path.join(HERE, "oracle_paper_ledger.jsonl")
+TMP = os.path.join(HERE, ".oracle_state.pull.json")
+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 main():
+    r = subprocess.run([FLYCTL, "ssh", "sftp", "get", "/data/oracle_state.json",
+                        TMP, "-a", "wwf-oraclebot"], capture_output=True,
+                       timeout=300, stdin=subprocess.DEVNULL)
+    if not os.path.exists(TMP) or os.path.getsize(TMP) == 0:
+        print("[grade_oracle] box unreachable or no state yet — skip")
+        return 0
+    st = json.load(open(TMP))
+    os.remove(TMP)
+    sys.path.insert(0, os.path.join(HERE, "..", "live"))
+    import payouts
+    have = set()
+    ledger_rows = 0
+    try:
+        for ln in open(LEDGER):
+            try:
+                d = json.loads(ln)
+                have.add((d["asset"], d["ts"]))
+                ledger_rows += 1
+            except Exception:
+                pass
+    except FileNotFoundError:
+        pass
+    settled = st.get("settled", [])
+    # rotation-loss guard: SETTLED_TRIM must never rotate a settle out of
+    # state before a nightly grades it
+    total_settles = st.get("counters", {}).get("settled_total", 0)
+    if total_settles > ledger_rows + len(settled):
+        print(f"[grade_oracle] ⚠ TAPE LOSS: {total_settles} lifetime settles "
+              f"but only {ledger_rows} graded + {len(settled)} in state — "
+              f"raise SETTLED_TRIM or grade more often")
+    new = [s for s in settled if (s["asset"], s["ts"]) not in have]
+    if new:
+        payouts.ensure(sorted({s["cond"] for s in new if s.get("cond")}))
+    graded = flips = 0
+    with open(LEDGER, "a") as fh:
+        for s in new:
+            t = payouts.truth(s.get("cond"), s.get("asset")) if s.get("cond") else None
+            pay = s["payout"] if t is None else t
+            pnl = round(s["shares"] * pay - s["cost"] - s["fee"], 2)
+            if t is not None and abs(t - s["payout"]) > 1e-9:
+                flips += 1
+            fh.write(json.dumps({**s, "chain_payout": pay, "chain_pnl": pnl,
+                                 "flip": t is not None
+                                 and abs(t - s["payout"]) > 1e-9}) + "\n")
+            graded += 1
+    # per-E-cell rollup from the LEDGER (chain truth): row in cell iff
+    # edge >= E — same bucketing as forward.py
+    cells = {str(E): dict(fills=0, wins=0, refunds=0, pnl=0.0)
+             for E in EDGE_GRID}
+    try:
+        for ln in open(LEDGER):
+            d = json.loads(ln)
+            for E in EDGE_GRID:
+                if d.get("edge", 0) >= E:
+                    g = cells[str(E)]
+                    g["fills"] += 1
+                    g["wins"] += d["chain_payout"] == 1.0
+                    g["refunds"] += d["chain_payout"] == 0.5
+                    g["pnl"] += d["chain_pnl"]
+    except FileNotFoundError:
+        pass
+    c = st.get("counters", {})
+    print(f"[grade_oracle] +{graded} settles ({flips} provisional flips) · "
+          f"realized ${st.get('pnl_realized', 0):+.2f} · "
+          f"open {len(st.get('open', {}))} · lifetime ev {c.get('events')} "
+          f"att {c.get('attempts')} fill {c.get('fills')} "
+          f"crater {c.get('craters')}")
+    for E in EDGE_GRID:
+        g = cells[str(E)]
+        ev = g["pnl"] / g["fills"] if g["fills"] else None
+        hit = g["wins"] / g["fills"] if g["fills"] else None
+        print(f"[grade_oracle]   E>={E:.2f}: {g['fills']} fills · "
+              f"EV/fill {'—' if ev is None else f'${ev:+.2f}'} · "
+              f"hit {'—' if hit is None else f'{hit:.3f}'} · "
+              f"pnl ${g['pnl']:+.2f}"
+              + (f" · {g['refunds']} refunds" if g["refunds"] else ""))
+    print("[grade_oracle] capturability read — NOT the #17 verdict "
+          "(forward_ledger is; bar: first cell to 30 fwd fills, "
+          "PASS EV>=+$6 & hit>=0.55, KILL EV<=0 at 50)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

research/nightly.sh

@@ -53,10 +53,11 @@ done
 "$PY" forward.py >> forward.log 2>&1
 "$PY" informed_set.py >> forward.log 2>&1     # surge harness reads this daily
 "$PY" grade_surge.py >> forward.log 2>&1 || true   # paper book -> chain truth
+"$PY" grade_oracle.py >> forward.log 2>&1 || true  # oracle paper -> chain truth
 
 cd ..
 git add research/forward_ledger.jsonl research/params/informed_set.json \
-        research/surge_paper_ledger.jsonl 2>/dev/null
+        research/surge_paper_ledger.jsonl research/oracle_paper_ledger.jsonl 2>/dev/null
 if ! git diff --cached --quiet; then
   git commit -q -m "research: forward ledger $(date -u +%F) [skip ci]"
   git pull --rebase --autostash -q && git push -q

research/oraclebot.Dockerfile

@@ -0,0 +1,6 @@
+# wwf-oraclebot — one dep, no git, no clone (a paper harness must not depend
+# on GitHub being up to boot; same discipline as surgebot/recorder).
+FROM python:3.12-slim
+RUN pip install --no-cache-dir websocket-client
+COPY research/oraclebot.py /oraclebot.py
+CMD ["python3", "-u", "/oraclebot.py"]

research/oraclebot.py

@@ -0,0 +1,814 @@
+#!/usr/bin/env python3
+"""oraclebot — Study B's crypto-oracle fair-value signal as a real-time
+PAPER harness.
+
+PAPER ONLY. No keys, no orders, no bot imports — self-contained
+(recorder-style: baked into the wwf-oraclebot image, must not depend on the
+repo to boot). It answers the two questions the tape sim cannot (#17): does
+fair value compute in real time on the venue's own settlement feed, and is
+the mispricing CAPTURABLE at real book latency (~200ms)? The sim's 86%
+crater rate and winner's-curse inversion come from a 6.7s-lag fill model
+fitted on sports FAKs; this box re-measures that race directly.
+
+THE #17 VERDICT BINDS TO forward_ledger.jsonl ONLY. A good harness cannot
+rescue a forward KILL — it can only seed a new pre-registered v2. Entries
+are graded nightly against chain truth on the Mac (grade_oracle.py); any
+semantics-altering fix bumps SEM_VER and restarts the shakedown clock.
+
+FROZEN semantics (mirror research/forward.py::score_oracle; constants must
+equal study_oracle.py / tape.py verbatim — do not tune here):
+  fair       Phi(ln(S/K)/(sigma*sqrt(tau))); sigma = trailing 30-min
+             realized vol of 1s log-returns; tau in [60s, 12h];
+             Down/No token = 1 - fair(up-side); sprint strike = S at t0
+  trigger    at a trade print, edge = fair - print >= 0.04; cooldown 300s
+             per token set at event time; each event counted in every grid
+             cell it clears (cell keys str(E): "0.04"/"0.07"/"0.1")
+  execution  $100 FAK vs live CLOB asks inside min(p_ref*1.05, 0.99);
+             fee 3% * shares * min(p, 1-p); partial fills kept
+Live-only guards (each can only SUPPRESS a trigger the tape sim would
+count, never add one): warmup, tick staleness, sprint pre-window gate (the
+tape scorer reads s0 from the FUTURE window-open tick for prints before t0
+— live can never know it), s0/t1 slack, book-fetch throttle.
+
+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.
+"""
+import bisect
+import calendar
+import json
+import math
+import os
+import queue
+import re
+import ssl
+import statistics as st
+import threading
+import time
+import urllib.request
+
+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")
+DEBUG = bool(os.environ.get("ORACLE_DEBUG"))
+SEM_VER = "b1"                    # bump on ANY semantics-altering change
+
+# ── FROZEN — must equal study_oracle.py verbatim ────────────────────────────
+EDGE_GRID = [0.04, 0.07, 0.10]
+VOL_WIN_S = 1800
+TAU_MIN, TAU_MAX = 60, 12 * 3600
+COOLDOWN_S = 300
+STAKE = 100.0
+UP_WORDS = {"up", "yes"}
+DOWN_WORDS = {"down", "no"}
+# ── FROZEN — must equal sim.py / surgebot.py verbatim ───────────────────────
+FEE_RATE = 0.03
+SLIP_CAP = 0.05
+# ── LIVE-ONLY — can only suppress, never add, triggers ──────────────────────
+WARMUP_S = 1800                   # per-symbol; must contain COOLDOWN_S (the
+COOLDOWN_CONTAINMENT = True       # restart argument: cooldowns are not
+                                  # persisted — safe because warmup blocks
+                                  # all triggers past any pre-restart
+                                  # cooldown horizon; selftest asserts it)
+TICK_STALE_S = 10                 # arrival-clock; stale feed => no fair
+TICK_RETAIN_S = 7200              # covers vol window + hourly-sprint t0
+PROV_SLACK_S = 5                  # deciding tick must be this close to t0/t1
+TIE_EPS = 1e-4                    # 1bp of boundary => defer to CLOB/chain
+SETTLE_GRACE_S = 120
+BOOK_WORKERS = 4
+BOOK_QUEUE = 16                   # overflow => throttle_skip (still counted)
+CLOB_PASS_CAP = 25                # GETs per backstop pass, round-robin
+SETTLED_TRIM = 10000              # nightly ledger is the durable record
+SKIPS_TRIM = 500
+DST_WARN_TS = calendar.timegm((2026, 10, 25, 0, 0, 0))   # EDT ends 11-01
+YEAR_WARN_TS = calendar.timegm((2026, 12, 15, 0, 0, 0))  # YEAR rolls
+SSL_CTX = ssl._create_unverified_context()
+
+SYMS = ("btcusdt", "ethusdt", "solusdt", "xrpusdt", "bnbusdt", "dogeusdt")
+
+
+def log(m):
+    print(f"{time.strftime('%H:%M:%S')}  {m}", flush=True)
+
+
+def get_json(url, timeout=8):
+    req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
+    with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as r:
+        return json.loads(r.read().decode())
+
+
+# ── title parsing — FROZEN, ported verbatim from research/tape.py ───────────
+
+MONTHS = {m: i + 1 for i, m in enumerate(
+    ["january", "february", "march", "april", "may", "june", "july",
+     "august", "september", "october", "november", "december"])}
+
+SYMBOLS = {"bitcoin": "btcusdt", "btc": "btcusdt",
+           "ethereum": "ethusdt", "eth": "ethusdt",
+           "solana": "solusdt", "sol": "solusdt",
+           "xrp": "xrpusdt", "bnb": "bnbusdt", "doge": "dogeusdt",
+           "dogecoin": "dogeusdt"}
+
+ET_OFF = 4 * 3600                 # EDT (July) = UTC-4
+YEAR = 2026                       # tape era; date guard warns at year roll
+
+RE_SPRINT = re.compile(
+    r"(?i)^(\w+)\s+up or down\s*-\s*(\w+)\s+(\d+),\s*"
+    r"(\d+)(?::(\d+))?(am|pm)-(\d+)(?::(\d+))?(pm|am)\s*et")
+RE_HOURLY = re.compile(
+    r"(?i)^(\w+)\s+(above|below)\s+([\d,\.]+)\s+on\s+(\w+)\s+(\d+),\s*"
+    r"(\d+)(?::(\d+))?\s*(am|pm)\s*et")
+RE_DAILY = re.compile(
+    r"(?i)price of (\w+) be (above|below|between)\s+\$?([\d,\.]+)"
+    r"(?:\s+and\s+\$?([\d,\.]+))?\s+on\s+(\w+)\s+(\d+)")
+# One-touch "dip to / reach $K" titles are NOT parsed on purpose (barrier,
+# not terminal digital) — crypto_parse returns None and they never trigger.
+
+
+def _num(s):
+    return float(s.replace(",", "")) if s else None
+
+
+def _clock(h, m, ap):
+    h = int(h) % 12 + (12 if ap.lower() == "pm" else 0)
+    return h, int(m or 0)
+
+
+def _et(mon, day, hh, mm):
+    # mktime(struct, isdst=0) - time.timezone == timegm on any libc that
+    # honors isdst=0 (trivially exact on the UTC Fly box); selftest proves
+    # the identity on the running box with canned-title epochs.
+    return time.mktime(time.struct_time(
+        (YEAR, mon, day, 0, 0, 0, 0, 0, 0))) - time.timezone + hh * 3600 \
+        + mm * 60 + ET_OFF
+
+
+def crypto_parse(title):
+    """-> dict(sym, kind, k1, k2, t0, t1) or None. kind: sprint/above/
+    below/between. t0 only for sprints (window open)."""
+    t = title or ""
+    m = RE_SPRINT.match(t)
+    if m:
+        sym = SYMBOLS.get(m.group(1).lower())
+        mon = MONTHS.get(m.group(2).lower())
+        if not sym or not mon:
+            return None
+        day = int(m.group(3))
+        h0, m0 = _clock(m.group(4), m.group(5), m.group(6))
+        h1, m1 = _clock(m.group(7), m.group(8), m.group(9))
+        return {"sym": sym, "kind": "sprint", "k1": None, "k2": None,
+                "t0": _et(mon, day, h0, m0), "t1": _et(mon, day, h1, m1)}
+    m = RE_HOURLY.match(t)
+    if m:
+        sym, mon = SYMBOLS.get(m.group(1).lower()), MONTHS.get(m.group(4).lower())
+        if not sym or not mon:
+            return None
+        h, mi = _clock(m.group(6), m.group(7), m.group(8))
+        return {"sym": sym, "kind": m.group(2).lower(), "k1": _num(m.group(3)),
+                "k2": None, "t0": None,
+                "t1": _et(mon, int(m.group(5)), h, mi)}
+    m = RE_DAILY.search(t)
+    if m:
+        sym, mon = SYMBOLS.get(m.group(1).lower()), MONTHS.get(m.group(5).lower())
+        if not sym or not mon:
+            return None
+        return {"sym": sym, "kind": m.group(2).lower(), "k1": _num(m.group(3)),
+                "k2": _num(m.group(4)), "t0": None,
+                "t1": _et(mon, int(m.group(6)), 12, 0)}   # dailies: 12PM ET
+    return None
+
+
+# ── fair value — FROZEN, ported verbatim from research/study_oracle.py ──────
+
+def phi(x):
+    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))
+
+
+def fair_value(mkt, up_side, S, sigma, t):
+    tau = mkt["t1"] - t
+    if not (TAU_MIN <= tau <= TAU_MAX) or not S or not sigma:
+        return None
+    sv = sigma * math.sqrt(tau)
+    if sv <= 0:
+        return None
+    k = mkt["kind"]
+    if k == "sprint":
+        if mkt.get("s0") is None:
+            return None
+        f_up = phi(math.log(S / mkt["s0"]) / sv)
+    elif k == "above":
+        f_up = phi(math.log(S / mkt["k1"]) / sv)
+    elif k == "below":
+        f_up = 1.0 - phi(math.log(S / mkt["k1"]) / sv)
+    elif k == "between":
+        f_up = phi(math.log(mkt["k2"] / S) / sv) - phi(math.log(mkt["k1"] / S) / sv)
+    else:
+        return None
+    return f_up if up_side else 1.0 - f_up
+
+
+# ── paper FAK ───────────────────────────────────────────────────────────────
+
+def walk_asks(asks, cap, stake_usd=STAKE):
+    """FAK against a CLOB ask list: consume ascending price levels <= cap
+    until stake_usd is spent; partial fills kept (that IS what a real FAK
+    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 []))
+    if not lv:
+        return {"filled": False, "best_ask": None}
+    best = lv[0][0]
+    spent = shares = 0.0
+    levels = 0
+    for px, sz in lv:
+        if px > cap or spent >= stake_usd - 1e-9:
+            break
+        take = min(sz, (stake_usd - spent) / px)
+        if take <= 0:
+            break
+        shares += take
+        spent += take * px
+        levels += 1
+    if not shares:
+        return {"filled": False, "best_ask": best}
+    return {"filled": True, "vwap": spent / shares, "shares": shares,
+            "usd": spent, "levels": levels, "best_ask": best,
+            "partial": spent < stake_usd - 1e-9}
+
+
+# ── live tick engine ────────────────────────────────────────────────────────
+
+class LiveTicks:
+    """Per-symbol venue-time series. at()/vol_1s() are FROZEN verbatim from
+    study_oracle.TickSeries; the rest is live-only maintenance (monotonic
+    dedupe absorbs the twin sockets, retention trim, warmup/freshness)."""
+
+    def __init__(self):
+        self.ts = []
+        self.px = []
+        self.last_arrival = 0.0
+        self.ooo = 0                   # duplicate/out-of-order drops
+
+    def add(self, t, p, now):
+        self.last_arrival = now
+        if not p:
+            return                     # mirrors load_ticks' `if v`
+        if self.ts and t <= self.ts[-1]:
+            self.ooo += 1
+            return
+        self.ts.append(t)
+        self.px.append(p)
+
+    def at(self, t):                   # FROZEN
+        i = bisect.bisect_right(self.ts, t) - 1
+        return self.px[i] if i >= 0 else None
+
+    def at_ts(self, t):
+        """venue timestamp of the tick at() would use — for slack guards."""
+        i = bisect.bisect_right(self.ts, t) - 1
+        return self.ts[i] if i >= 0 else None
+
+    def vol_1s(self, t, win=VOL_WIN_S):  # FROZEN
+        lo = bisect.bisect_left(self.ts, t - win)
+        hi = bisect.bisect_right(self.ts, t)
+        if hi - lo < 60:
+            return None
+        rets = []
+        for i in range(lo + 1, hi):
+            dt = self.ts[i] - self.ts[i - 1]
+            if dt <= 0:
+                continue
+            r = math.log(self.px[i] / self.px[i - 1]) / math.sqrt(dt)
+            rets.append(r)
+        return st.pstdev(rets) if len(rets) >= 30 else None
+
+    def trim(self, now):
+        cut = bisect.bisect_left(self.ts, now - TICK_RETAIN_S)
+        if cut:
+            del self.ts[:cut]
+            del self.px[:cut]
+
+    def warm(self, now):
+        return bool(self.ts) and self.ts[0] <= now - WARMUP_S
+
+    def fresh(self, now):
+        return now - self.last_arrival <= TICK_STALE_S
+
+
+# ── the bot ─────────────────────────────────────────────────────────────────
+
+def fresh_state():
+    return {
+        "sem_ver": SEM_VER, "boot_ts": int(time.time()),
+        "open": {}, "settled": [], "skips": [],
+        "cells": {str(E): {"events": 0, "attempts": 0, "fills": 0,
+                           "craters": 0, "throttle": 0, "wins": 0,
+                           "refunds": 0, "staked": 0.0, "pnl": 0.0}
+                  for E in EDGE_GRID},
+        "pnl_realized": 0.0,
+        "counters": {"trades_seen": 0, "crypto_prints": 0, "events": 0,
+                     "attempts": 0, "fills": 0, "craters": 0,
+                     "throttle_skip": 0, "settled_total": 0,
+                     "settle_tick": 0, "settle_clob": 0,
+                     "skip": {"unknown_side": 0, "sprint_no_label": 0,
+                              "pre_window": 0, "s0_gap": 0, "warmup": 0,
+                              "stale": 0, "no_vol": 0, "tau": 0,
+                              "book_fail": 0}}}
+
+
+class OracleBot:
+    def __init__(self):
+        self.state = fresh_state()
+        if os.path.exists(STATE):
+            try:
+                loaded = json.load(open(STATE))
+                if loaded.get("sem_ver") != SEM_VER:
+                    log(f"⚠ SEM_VER {loaded.get('sem_ver')} -> {SEM_VER} — "
+                        f"semantics changed; shakedown clock restarts here")
+                    loaded["sem_ver"] = SEM_VER
+                self.state = loaded
+                log(f"resumed: pnl {self.state['pnl_realized']:+.2f} · "
+                    f"{len(self.state['open'])} open · "
+                    f"{self.state['counters']['fills']} lifetime fills")
+            except Exception as e:
+                log(f"⚠ state load failed ({e}) — fresh book")
+        self.ticks = {s: LiveTicks() for s in SYMS}
+        self.last_trig = {}            # not persisted — see WARMUP_S comment
+        self.lock = threading.Lock()
+        self.book_q = queue.Queue(BOOK_QUEUE)
+        self.trades_seen = 0           # unlocked cosmetic counter (firehose
+                                       # rate) — folded into state at heartbeat
+
+    # ── streams ────────────────────────────────────────────────────────────
+    def on_tick(self, payload):
+        now = time.time()
+        items = payload if isinstance(payload, list) else [payload]
+        with self.lock:
+            for it in items:
+                try:
+                    s = self.ticks.get((it.get("symbol") or "").lower())
+                    if s is None:
+                        continue
+                    t = float(it.get("timestamp") or 0)
+                    t = t / 1000.0 if t > 1e12 else (t or now)
+                    s.add(t, float(it.get("value") or 0), now)
+                except Exception:
+                    continue
+
+    def on_trade(self, p):
+        title = p.get("title") or ""
+        mkt = crypto_parse(title)      # pure — outside the lock
+        if not mkt:
+            return
+        asset = p.get("asset")
+        try:
+            px = float(p.get("price") or 0)
+        except (TypeError, ValueError):
+            return
+        if not asset or not (0 < px < 1):
+            return
+        s = self.ticks.get(mkt["sym"])
+        if s is None:
+            return
+        now = time.time()              # print clock = arrival (tape mirror)
+        o = (p.get("outcome") or "").lower()
+        with self.lock:
+            c = self.state["counters"]
+            sk = c["skip"]
+            c["crypto_prints"] += 1
+            # side — FROZEN mirror of study_oracle.crypto_universe()
+            if o and o not in UP_WORDS | DOWN_WORDS:
+                sk["unknown_side"] += 1
+                return
+            up = o in UP_WORDS or (o == "" and mkt["kind"] != "sprint")
+            if mkt["kind"] == "sprint" and not o:
+                sk["sprint_no_label"] += 1
+                return
+            # LIVE-ONLY gates (suppress-only)
+            if not s.warm(now):
+                sk["warmup"] += 1
+                return
+            if not s.fresh(now):
+                sk["stale"] += 1
+                return
+            s0 = None
+            if mkt["kind"] == "sprint":
+                if now < mkt["t0"]:
+                    sk["pre_window"] += 1   # tape scorer's s0 lookahead
+                    return
+                s0 = s.at(mkt["t0"])
+                t0_tick = s.at_ts(mkt["t0"])
+                if s0 is None or mkt["t0"] - t0_tick > PROV_SLACK_S:
+                    sk["s0_gap"] += 1
+                    return
+                mkt = {**mkt, "s0": s0}
+            S = s.at(now)
+            sig = s.vol_1s(now)
+            f = fair_value(mkt, up, S, sig, now)
+            if f is None:
+                tau = mkt["t1"] - now
+                sk["tau" if not (TAU_MIN <= tau <= TAU_MAX) else "no_vol"] += 1
+                return
+            edge = f - px
+            if DEBUG:
+                log(f"DBG ts={now:.3f} asset={asset} kind={mkt['kind']} "
+                    f"sym={mkt['sym']} fair={f:.4f} print={px:.3f} "
+                    f"edge={edge:+.4f} vol={sig:.2e} tau={int(mkt['t1']-now)}")
+            if edge < min(EDGE_GRID):
+                return
+            if now - self.last_trig.get(asset, 0) < COOLDOWN_S:
+                return
+            self.last_trig[asset] = now  # set PRE-fetch: twin dups self-block
+            c["events"] += 1
+            for E in EDGE_GRID:
+                if edge >= E:
+                    self.state["cells"][str(E)]["events"] += 1
+            job = {"asset": asset, "cond": p.get("conditionId"),
+                   "title": title[:60], "kind": mkt["kind"], "sym": mkt["sym"],
+                   "outcome": p.get("outcome"), "up": up,
+                   "k1": mkt.get("k1"), "k2": mkt.get("k2"),
+                   "t0": mkt.get("t0"), "t1": mkt["t1"], "s0": s0,
+                   "ts": round(now, 3), "p_ref": px, "fair": round(f, 4),
+                   "edge": round(edge, 4), "sig": round(sig, 8)}
+            try:
+                self.book_q.put_nowait(job)
+                c["attempts"] += 1
+                for E in EDGE_GRID:
+                    if edge >= E:
+                        self.state["cells"][str(E)]["attempts"] += 1
+            except queue.Full:
+                c["throttle_skip"] += 1
+                for E in EDGE_GRID:
+                    if edge >= E:
+                        self.state["cells"][str(E)]["throttle"] += 1
+                self._skip(job, "throttle", None)
+
+    # ── paper execution (workers; network OUT of the lock) ─────────────────
+    def book_worker(self):
+        while True:
+            job = self.book_q.get()
+            try:
+                self.attempt(job)
+            except Exception as e:
+                log(f"⚠ attempt error: {e}")
+
+    def attempt(self, job):
+        t_req = time.time()
+        try:
+            book = get_json(f"{CLOB}/book?token_id={job['asset']}")
+        except Exception:
+            with self.lock:
+                self.state["counters"]["skip"]["book_fail"] += 1
+                self._skip(job, "book_fail", None)
+            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)
+        with self.lock:
+            c = self.state["counters"]
+            if not r["filled"]:
+                c["craters"] += 1
+                for E in EDGE_GRID:
+                    if job["edge"] >= E:
+                        self.state["cells"][str(E)]["craters"] += 1
+                self._skip(job, "crater", r["best_ask"])
+                ba = "—" if r["best_ask"] is None else f"{r['best_ask']:.3f}"
+                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()
+
+    def _skip(self, job, why, ba):     # under lock
+        self.state["skips"].append(
+            {"ts": int(job["ts"]), "asset": job["asset"], "why": why,
+             "p_ref": job["p_ref"], "edge": job["edge"], "ba": ba,
+             "title": job["title"]})
+        del self.state["skips"][:-SKIPS_TRIM]
+
+    # ── settlement ─────────────────────────────────────────────────────────
+    # (a) instant, from the same feed the venue settles on (under lock)
+    def settle_provisional(self):
+        now = time.time()
+        n = 0
+        for lid, lot in list(self.state["open"].items()):
+            if now < lot["t1"] + SETTLE_GRACE_S:
+                continue
+            s = self.ticks[lot["sym"]]
+            S1 = s.at(lot["t1"])
+            t1_tick = s.at_ts(lot["t1"])
+            if S1 is None or lot["t1"] - t1_tick > PROV_SLACK_S:
+                continue               # buffer gap at t1 -> CLOB/chain layers
+            k = lot["kind"]
+            if k == "sprint":
+                bounds, yes = [lot["s0"]], S1 > lot["s0"]
+            elif k == "above":
+                bounds, yes = [lot["k1"]], S1 > lot["k1"]
+            elif k == "below":
+                bounds, yes = [lot["k1"]], S1 < lot["k1"]
+            elif k == "between":
+                bounds = [lot["k1"], lot["k2"]]
+                yes = lot["k1"] <= S1 <= lot["k2"]
+            else:
+                continue
+            # boundary tie: the venue's exact settle print can differ — defer
+            if any(b and abs(S1 - b) / b < TIE_EPS for b in bounds):
+                continue
+            self._settle(lid, lot, 1.0 if yes == lot["up"] else 0.0, "tick")
+            n += 1
+        if n:
+            self.persist()
+
+    # (b) CLOB closed/winner backstop for what (a) couldn't decide
+    def settle_clob_pass(self):
+        with self.lock:
+            now = time.time()
+            due = [lid for lid, lot in self.state["open"].items()
+                   if now > lot["t1"] + SETTLE_GRACE_S]
+            due.sort(key=lambda lid: self.state["open"][lid].get("clob_ck", 0))
+            due = due[:CLOB_PASS_CAP]
+            snap = {}
+            for lid in due:
+                lot = self.state["open"][lid]
+                lot["clob_ck"] = now   # round-robin stamp
+                snap[lid] = (lot["cond"], lot["asset"])
+        verdicts = {}
+        for lid, (cond, asset) in snap.items():
+            try:
+                m = get_json(f"{CLOB}/markets/{cond}")
+            except Exception:
+                continue
+            if not m.get("closed"):
+                continue
+            for t in m.get("tokens") or []:
+                if str(t.get("token_id")) == str(asset):
+                    verdicts[lid] = 1.0 if t.get("winner") else 0.0
+        if not verdicts:
+            return
+        with self.lock:
+            n = 0
+            for lid, pay in verdicts.items():
+                lot = self.state["open"].get(lid)
+                if lot:
+                    self._settle(lid, lot, pay, "clob")
+                    n += 1
+            if n:
+                self.persist()
+
+    def _settle(self, lid, lot, pay, source):   # under lock
+        pnl = round(lot["shares"] * pay - lot["cost"] - lot["fee"], 2)
+        self.state["pnl_realized"] = round(self.state["pnl_realized"] + pnl, 2)
+        for E in EDGE_GRID:
+            if lot["edge"] >= E:
+                cell = self.state["cells"][str(E)]
+                cell["pnl"] = round(cell["pnl"] + pnl, 2)
+                cell["wins"] += pay == 1.0
+                cell["refunds"] += pay == 0.5
+        c = self.state["counters"]
+        c["settled_total"] += 1
+        c["settle_tick" if source == "tick" else "settle_clob"] += 1
+        self.state["settled"].append(
+            {**lot, "payout": pay, "source": source, "provisional": True,
+             "settled_ts": int(time.time()), "pnl": pnl})
+        del self.state["settled"][:-SETTLED_TRIM]
+        del self.state["open"][lid]
+        word = "WON" if pay == 1.0 else "refund" if pay == 0.5 else "lost"
+        log(f"SETTLE[{source}] {word} {lot['title'][:36]} {pnl:+.2f} · "
+            f"realized {self.state['pnl_realized']:+.2f}")
+
+    # ── plumbing ───────────────────────────────────────────────────────────
+    def persist(self):
+        tmp = STATE + ".tmp"
+        json.dump(self.state, open(tmp, "w"))
+        os.replace(tmp, STATE)
+
+    def feed_dict(self):               # under lock
+        now = time.time()
+        syms = {}
+        for sym, s in self.ticks.items():
+            syms[sym] = {
+                "px": s.px[-1] if s.px else None,
+                "tick_age_s": round(now - s.last_arrival, 1)
+                if s.last_arrival else None,
+                "n_buf": len(s.ts), "warm": s.warm(now),
+                "warm_eta_s": max(0, int(s.ts[0] + WARMUP_S - now))
+                if s.ts else None,
+                "fresh": s.fresh(now), "vol_1s": s.vol_1s(now)}
+        stt = self.state
+        return {
+            "mode": "paper", "study": "oracle-B (#17)", "sem_ver": SEM_VER,
+            "updated": int(now), "boot_ts": stt["boot_ts"],
+            "frozen": {"edge_grid": EDGE_GRID, "vol_win_s": VOL_WIN_S,
+                       "tau_min": TAU_MIN, "tau_max": TAU_MAX,
+                       "cooldown_s": COOLDOWN_S, "stake": STAKE,
+                       "fee_rate": FEE_RATE, "slip_cap": SLIP_CAP},
+            "symbols": syms, "cells": stt["cells"],
+            "pnl_realized": stt["pnl_realized"], "open_n": len(stt["open"]),
+            "open": [{**l, "id": lid}
+                     for lid, l in list(stt["open"].items())[-200:]],
+            "settled": stt["settled"][-200:], "skips": stt["skips"][-100:],
+            "counters": stt["counters"],
+            "note": "verdict binds to forward_ledger.jsonl (#17); this feed "
+                    "measures real-time capturability only"}
+
+
+# ── streams / feed / boot — surgebot pattern ────────────────────────────────
+
+SUB = json.dumps({"action": "subscribe", "subscriptions": [
+    {"topic": "activity", "type": "trades", "filters": ""},
+    {"topic": "crypto_prices", "type": "*", "filters": ""}]})
+
+
+def run_conn(tag, bot):
+    backoff = 2
+    while True:
+        state = {"fresh": time.time()}
+
+        def on_open(ws):
+            ws.send(SUB)
+            state["fresh"] = time.time()
+            log(f"ws[{tag}]: connected")
+
+            def ping():
+                while ws.keep_running:
+                    time.sleep(5)
+                    try:
+                        ws.send('{"action":"ping"}')
+                    except Exception:
+                        break
+                    if time.time() - state["fresh"] > 20:
+                        try:
+                            ws.close()
+                        except Exception:
+                            pass
+                        break
+            threading.Thread(target=ping, daemon=True).start()
+
+        def on_message(ws, raw):
+            state["fresh"] = time.time()
+            try:
+                m = json.loads(raw)
+            except Exception:
+                return
+            topic = m.get("topic")
+            if topic == "activity" and m.get("type") == "trades":
+                bot.trades_seen += 1
+                try:
+                    bot.on_trade(m.get("payload") or {})
+                except Exception as e:
+                    log(f"⚠ on_trade error: {e}")
+            elif topic == "crypto_prices":
+                try:
+                    bot.on_tick(m.get("payload") or {})
+                except Exception as e:
+                    log(f"⚠ on_tick error: {e}")
+
+        try:
+            app = websocket.WebSocketApp(WS_URL, on_open=on_open,
+                                         on_message=on_message)
+            app.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
+        except Exception as e:
+            log(f"ws[{tag}]: {str(e)[:60]}")
+        time.sleep(backoff + (1 if tag == "b" else 0))
+        backoff = min(backoff * 2, 30)
+        if time.time() - state["fresh"] < 60:
+            backoff = 2
+
+
+def serve_feed(bot, port=8080):
+    """Read-only public feed for the /oracle dashboard (paper data only —
+    nothing here can place an order or mutate state). CORS-open."""
+    from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+    class H(BaseHTTPRequestHandler):
+        def log_message(self, *a):
+            pass
+
+        def do_GET(self):
+            if self.path.split("?")[0] != "/feed":
+                self.send_response(404)
+                self.end_headers()
+                return
+            with bot.lock:
+                body = json.dumps(bot.feed_dict()).encode()
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Access-Control-Allow-Origin", "*")
+            self.send_header("Cache-Control", "no-store")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+    ThreadingHTTPServer(("0.0.0.0", port), H).serve_forever()
+
+
+def selftest():
+    """Boot gate — hard exit on failure. Proves the _et/mktime identity on
+    THIS box, the parsers, fair-value spots, the FAK walk, and the
+    restart-containment invariant."""
+    m = crypto_parse("Bitcoin Up or Down - July 22, 3PM-4PM ET")
+    assert m and m["kind"] == "sprint" and m["sym"] == "btcusdt", m
+    assert m["t0"] == calendar.timegm((2026, 7, 22, 19, 0, 0)), m["t0"]
+    assert m["t1"] == calendar.timegm((2026, 7, 22, 20, 0, 0)), m["t1"]
+    m = crypto_parse("Ethereum above 3,600 on July 22, 4PM ET")
+    assert m and m["kind"] == "above" and m["k1"] == 3600.0, m
+    assert m["t1"] == calendar.timegm((2026, 7, 22, 20, 0, 0)), m["t1"]
+    m = crypto_parse("Will the price of Solana be between $150 and $170 "
+                     "on July 25?")
+    assert m and m["kind"] == "between" and m["k2"] == 170.0, m
+    assert m["t1"] == calendar.timegm((2026, 7, 25, 16, 0, 0)), m["t1"]
+    assert crypto_parse("Will Bitcoin dip to $110,000 in July?") is None
+    assert phi(0) == 0.5
+    mkt = {"kind": "above", "k1": 100.0, "k2": None, "t0": None, "t1": 1000.0}
+    assert abs(fair_value(mkt, True, 100.0, 1e-4, 400.0) - 0.5) < 1e-9
+    assert fair_value(mkt, True, 120.0, 2e-4, 400.0) > 0.99
+    assert fair_value(mkt, False, 120.0, 2e-4, 400.0) < 0.01
+    assert fair_value(mkt, True, 100.0, 1e-4, 970.0) is None       # tau<60
+    assert fair_value(mkt, True, 100.0, 1e-4, 1000 - 13 * 3600) is None
+    r = walk_asks([{"price": "0.60", "size": "200"},
+                   {"price": "0.55", "size": "100"}], cap=0.62)
+    # sizes are SHARES: 100sh@0.55 = $55, then 75sh@0.60 = $45 -> full $100
+    assert r["filled"] and abs(r["usd"] - 100.0) < 1e-6, r
+    assert r["levels"] == 2 and r["best_ask"] == 0.55 and not r["partial"], r
+    assert 0.55 < r["vwap"] < 0.60 and abs(r["shares"] - 175.0) < 1e-6, r
+    r = walk_asks([{"price": "0.55", "size": "10"}], cap=0.60)
+    assert r["filled"] and r["partial"] and abs(r["usd"] - 5.5) < 1e-9, r
+    r = walk_asks([{"price": "0.70", "size": "10"}], cap=0.60)
+    assert not r["filled"] and r["best_ask"] == 0.70, r
+    assert not walk_asks([], 0.5)["filled"]
+    s = LiveTicks()
+    p = 100.0
+    for i in range(120):
+        p *= math.exp(1e-4 if i % 2 == 0 else -1e-4)
+        s.add(1000.0 + i, p, 0)
+    v = s.vol_1s(1000.0 + 119)
+    assert v and abs(v - 1e-4) < 2e-5, v
+    assert WARMUP_S >= COOLDOWN_S      # restart-containment invariant
+    log("selftest OK")
+
+
+def main():
+    selftest()
+    bot = OracleBot()
+    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 tag in ("a", "b"):
+        threading.Thread(target=run_conn, args=(tag, bot), daemon=True).start()
+    n = 0
+    while True:
+        time.sleep(60)
+        n += 1
+        now = time.time()
+        with bot.lock:
+            for s in bot.ticks.values():
+                s.trim(now)
+            bot.last_trig = {a: t for a, t in bot.last_trig.items()
+                             if now - t < COOLDOWN_S}
+            bot.settle_provisional()
+        if n % 2 == 0:
+            bot.settle_clob_pass()
+        if n % 5 == 0:
+            with bot.lock:
+                bot.persist()
+        if now > DST_WARN_TS:
+            log("⚠ ET_OFF=EDT expires at DST end 2026-11-01 — review parser")
+        if now > YEAR_WARN_TS:
+            log("⚠ YEAR=2026 rolls soon — parser dates will break")
+        with bot.lock:
+            c = bot.state["counters"]
+            c["trades_seen"] = bot.trades_seen
+            ticks = " ".join(f"{sym[:3]}:{len(s.ts)}"
+                             for sym, s in bot.ticks.items())
+            msg = (f"paper pnl {bot.state['pnl_realized']:+.2f} · "
+                   f"open {len(bot.state['open'])} · ev {c['events']} "
+                   f"att {c['attempts']} fill {c['fills']} "
+                   f"crat {c['craters']} thr {c['throttle_skip']} · "
+                   f"settle t{c['settle_tick']}/c{c['settle_clob']} · "
+                   f"ticks {ticks}")
+        log(msg)
+
+
+if __name__ == "__main__":
+    main()