From cf9465f38ae45ec870c8807fd7bf771d8b002b3b Mon Sep 17 00:00:00 2001 From: jaxperro Date: Mon, 27 Jul 2026 15:32:39 -0400 Subject: [PATCH] #26 wwf-leanbot (Study C stage-2) + #27 v2 band arm + shared concentration guards MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit leanbot: orders_matched inventory tracking on the nightly-published screened set (box never screens itself), frozen $150-500 one-sided crossing, paper FAK $100 capped at print+3c, complement routing for net-short leans, MAX_PER_EVENT=2, premium logged on every attempt. guards.py: ex-best-day / ex-top5 / top-event share — the cuts that caught both studies today, now machinery every grader reports nightly. grade_lag: v2 band arm (15-40c, forward-only from V2_FREEZE_TS, bar +$8) + three report-only control bands. Co-Authored-By: Claude Fable 5 --- fly.leanbot.toml | 28 ++ research/grade_lag.py | 29 +- research/grade_lean.py | 129 ++++++++ research/guards.py | 84 +++++ research/leanbot.Dockerfile | 5 + research/leanbot.py | 631 ++++++++++++++++++++++++++++++++++++ research/maker_set.py | 53 +++ 7 files changed, 958 insertions(+), 1 deletion(-) create mode 100644 fly.leanbot.toml create mode 100644 research/grade_lean.py create mode 100644 research/guards.py create mode 100644 research/leanbot.Dockerfile create mode 100644 research/leanbot.py create mode 100644 research/maker_set.py diff --git a/fly.leanbot.toml b/fly.leanbot.toml new file mode 100644 index 00000000..5c5edc31 --- /dev/null +++ b/fly.leanbot.toml @@ -0,0 +1,28 @@ +# wwf-leanbot — Study C Stage-2 maker-lean PAPER harness (research silo). +# No keys, no orders, code baked in image. The screened wallet set is +# published nightly by maker_set.py — the box never screens itself. +# Verdict binds to #26; graded nightly on the Mac (grade_lean.py). +app = "wwf-leanbot" +primary_region = "arn" + +[build] + dockerfile = "research/leanbot.Dockerfile" + +[[mounts]] + source = "lean" + destination = "/data" + +# read-only paper feed (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" diff --git a/research/grade_lag.py b/research/grade_lag.py index 812724b4..b2f4974e 100644 --- a/research/grade_lag.py +++ b/research/grade_lag.py @@ -20,6 +20,9 @@ STATE_PULL = os.path.join(HERE, ".lag_state.pull.json") ATT_PULL = os.path.join(HERE, ".lag_attempts.pull.jsonl") SET_PULL = os.path.join(HERE, ".lag_settles.pull.jsonl") FLYCTL = shutil.which("flyctl") or "/opt/homebrew/bin/flyctl" +# #27 freeze: v2 scores ONLY fills at/after this instant. v1 fills are +# not reusable for the band verdict (the band was found by slicing them). +V2_FREEZE_TS = 1785200000 # 2026-07-27 ~19:33Z, the commit that ships it def sftp(remote, local): @@ -120,7 +123,31 @@ def main(): print(f"[grade_lag] OBSERVATIONAL: median ask premium " f"{st.median(prem)*100:+.1f}c over {len(prem)} attempts " f"(kill-switch: >= +8c across 3 days)") - print("[grade_lag] verdict binds to the Study D pre-registration") + # ── #27 v2: the 15-40c complement band, forward fills only ─────────── + import guards + led = [] + try: + for ln in open(LEDGER): + try: + led.append(json.loads(ln)) + except Exception: + pass + except FileNotFoundError: + pass + print("[grade_lag] " + guards.line("v1 ALL-BANDS", guards.cuts(led), + pass_ev=4.0)) + v2 = [r for r in led if r.get("ts", 0) >= V2_FREEZE_TS + and 0.15 <= (r.get("price") or 0) < 0.40] + print("[grade_lag] " + guards.line("v2 BAND 15-40c", guards.cuts(v2), + pass_ev=8.0)) + for lo, hi, tag in ((0.0, 0.15, "<15c"), (0.40, 0.70, "40-70c"), + (0.70, 1.01, ">=70c")): + ctl = [r for r in led if r.get("ts", 0) >= V2_FREEZE_TS + and lo <= (r.get("price") or 0) < hi] + if ctl: + print("[grade_lag] control " + guards.line(tag, guards.cuts(ctl))) + print("[grade_lag] verdict binds to the Study D pre-registration (v1) " + "and #27 (v2 band; PASS needs EV>=+$8 AND all guards clear)") return 0 diff --git a/research/grade_lean.py b/research/grade_lean.py new file mode 100644 index 00000000..49a39eb1 --- /dev/null +++ b/research/grade_lean.py @@ -0,0 +1,129 @@ +#!/usr/bin/env python3 +"""Nightly chain-truth grading of the maker-lean harness (wwf-leanbot). + +Pulls /data/lean_state.json + lean_attempts.jsonl + lean_settles.jsonl off +the box (read-only), re-grades every settle with CTF payout vectors, +appends to research/lean_paper_ledger.jsonl (keyed asset+ts, idempotent), +and prints the pre-registered readouts for #26: paper-leg EV/fill WITH +the shared concentration guards (guards.py — ex-best-day, ex-top5, +top-event share; a headline PASS that fails them is CONCENTRATED and does +not graduate), plus the OBSERVATIONAL kill-switch — the median ask +premium over the lean-side print (>= 8c over 3 days = the tape's entry +was a mirage, no paper sample needed).""" +import json +import os +import shutil +import statistics as st +import subprocess +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +LEDGER = os.path.join(HERE, "lean_paper_ledger.jsonl") +STATE_PULL = os.path.join(HERE, ".lean_state.pull.json") +ATT_PULL = os.path.join(HERE, ".lean_attempts.pull.jsonl") +SET_PULL = os.path.join(HERE, ".lean_settles.pull.jsonl") +FLYCTL = shutil.which("flyctl") or "/opt/homebrew/bin/flyctl" + + +def sftp(remote, local): + subprocess.run([FLYCTL, "ssh", "sftp", "get", remote, local + ".new", + "-a", "wwf-leanbot"], 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) + return True + try: + os.remove(local + ".new") + except FileNotFoundError: + pass + return False + + +def main(): + ok = sftp("/data/lean_state.json", STATE_PULL) + sftp("/data/lean_attempts.jsonl", ATT_PULL) + sftp("/data/lean_settles.jsonl", SET_PULL) + if not ok: + print("[grade_lean] box unreachable or no state yet — skip") + return 0 + stt = json.load(open(STATE_PULL)) + 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 = stt.get("settled", []) + if os.path.exists(SET_PULL): # durable log beats trimmed state + seen = {(s["asset"], s["ts"]) for s in settled} + for ln in open(SET_PULL): + try: + s = json.loads(ln) + if (s["asset"], s["ts"]) not in seen: + settled.append(s) + except Exception: + pass + 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 + import guards + led = [] + try: + for ln in open(LEDGER): + try: + led.append(json.loads(ln)) + except Exception: + pass + except FileNotFoundError: + pass + # observational kill-switch: median premium across ALL attempts with a + # visible ask (fills, premium-skips, craters alike) + prem = [] + try: + for ln in open(ATT_PULL): + try: + a = json.loads(ln) + if a.get("premium") is not None: + prem.append(a["premium"]) + except Exception: + pass + except FileNotFoundError: + pass + c = stt.get("counters", {}) + cuts = guards.cuts(led) + print(f"[grade_lean] +{graded} settles ({flips} flips) · " + f"lifetime cross {c.get('crossings')} att {c.get('attempts')} " + f"fill {c.get('fills')} prem-skip {c.get('premium_skips')}") + print("[grade_lean] " + guards.line("PAPER LEG", cuts, pass_ev=2.0)) + if prem: + prem.sort() + print(f"[grade_lean] OBSERVATIONAL: median ask premium " + f"{st.median(prem)*100:+.1f}c over {len(prem)} attempts " + f"(kill-switch: >= +8c across 3 days)") + print("[grade_lean] verdict binds to the Study C Stage-2 (#26) pre-registration") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/research/guards.py b/research/guards.py new file mode 100644 index 00000000..7416f191 --- /dev/null +++ b/research/guards.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Concentration guards, shared by every harness grader (#26/#27). + +Two studies passed their headline bars this week on results that one +session or five episodes were carrying (Study C ex-best-day +$1.47 vs a ++$2 bar; Study D ex-best-day −$5.86 on a +$4.49 headline). These cuts +turn that lesson into machinery: every grader reports them nightly, and +a PASS whose ex-best-day EV is <= 0 is recorded CONCENTRATED and does +not graduate. + +rows: dicts with chain_pnl, cost, ts, and (optionally) event. +""" +import collections +import datetime as dt + + +def cuts(rows, pnl_key="chain_pnl"): + """-> dict of the guard readings (empty dict when there is nothing).""" + rows = [r for r in rows if r.get(pnl_key) is not None] + n = len(rows) + if not n: + return {} + tot = sum(r[pnl_key] for r in rows) + staked = sum(r.get("cost") or 0 for r in rows) + out = {"n": n, "pnl": round(tot, 2), "ev": round(tot / n, 2), + "pct_staked": round(100 * tot / staked, 1) if staked else None, + "wins": sum(1 for r in rows if r.get("chain_payout") == 1.0)} + out["hit"] = round(out["wins"] / n, 3) + # episode concentration + srt = sorted(rows, key=lambda r: -abs(r[pnl_key])) + for k in (1, 5): + if n > k: + ex = sum(r[pnl_key] for r in srt[k:]) + out[f"ex_top{k}_ev"] = round(ex / (n - k), 2) + out["top5_share"] = (round(100 * sum(r[pnl_key] for r in srt[:5]) / tot, 0) + if tot else None) + # TIME concentration — the guard both studies needed + byday = collections.defaultdict(lambda: [0, 0.0]) + for r in rows: + d = dt.datetime.utcfromtimestamp(r["ts"]).strftime("%m-%d") + byday[d][0] += 1 + byday[d][1] += r[pnl_key] + out["days"] = len(byday) + out["by_day"] = {d: [v[0], round(v[1], 0)] for d, v in sorted(byday.items())} + if len(byday) > 1: + best = max(byday.values(), key=lambda v: v[1]) + exn = n - best[0] + if exn > 0: + out["ex_best_day_ev"] = round((tot - best[1]) / exn, 2) + out["ex_best_day_n"] = exn + # EVENT concentration (the gate Study C's fade arm failed at 32%) + ev = collections.defaultdict(float) + for r in rows: + if r.get("event"): + ev[r["event"]] += r[pnl_key] + if ev and tot: + out["events"] = len(ev) + out["top_event_share"] = round( + 100 * max(ev.values(), key=abs) / tot, 0) + return out + + +def line(tag, c, pass_ev=None): + """One printable verdict line + the guard verdict.""" + if not c: + return f"{tag}: no graded rows" + s = (f"{tag}: n={c['n']} {c['wins']}W · EV/fill {c['ev']:+.2f} · " + f"hit {c['hit']:.3f} · {c['pct_staked']:+.0f}% staked · " + f"{c['days']}d") + if "ex_best_day_ev" in c: + s += f" · EX-BEST-DAY {c['ex_best_day_ev']:+.2f}" + if "ex_top5_ev" in c: + s += f" · ex-top5 {c['ex_top5_ev']:+.2f}" + if "top_event_share" in c: + s += f" · top-event {c['top_event_share']:+.0f}%" + if pass_ev is not None: + bars = [c["ev"] >= pass_ev, + c.get("ex_best_day_ev", -1) > 0, + c.get("ex_top5_ev", -1) > 0, + abs(c.get("top_event_share") or 0) < 30] + s += (" -> " + ("ALL GUARDS CLEAR" if all(bars) else + "CONCENTRATED (headline only)" if bars[0] else + "below bar")) + return s diff --git a/research/leanbot.Dockerfile b/research/leanbot.Dockerfile new file mode 100644 index 00000000..fc84b488 --- /dev/null +++ b/research/leanbot.Dockerfile @@ -0,0 +1,5 @@ +# wwf-leanbot — one dep, no git, no clone (same discipline as the others). +FROM python:3.12-slim +RUN pip install --no-cache-dir websocket-client +COPY research/leanbot.py /leanbot.py +CMD ["python3", "-u", "/leanbot.py"] diff --git a/research/leanbot.py b/research/leanbot.py new file mode 100644 index 00000000..63f44564 --- /dev/null +++ b/research/leanbot.py @@ -0,0 +1,631 @@ +#!/usr/bin/env python3 +"""leanbot — Study C Stage-2 (#26): maker inventory-leans at REAL execution. + +PAPER ONLY. No keys, no orders, no bot imports — self-contained (baked +into the wwf-leanbot image). Study C (#22) PASSED its tape-scored window +(+$2.17/lean, hit .576, n=20,461, 6 days) but every entry was booked at +the lean-side LAST PRINT — an entry no real order can have. Study B died +of exactly that assumption (tape-positive, −$8.51/fill at a real book). +This box answers the only question that matters: when a screened +maker-sharp's inventory crosses the conviction line, is the lean side +still buyable near that print? + +Two legs per lean, both recorded: + observational ask premium over the lean-side last print at trigger time + (median premium >= 8c over 3 days = the tape's entry was a + mirage — pre-registered kill-switch, no paper sample needed) + paper $100 FAK walk at the real ask, gated by the pre-declared + premium cap (+3c over the print; tighter than lagbot's 4c + because a lean carries no urgency — nothing is crashing) + +THE VERDICT BINDS TO #26. Entries graded nightly against chain truth +(grade_lean.py). Any semantics-altering fix bumps SEM_VER and restarts +the shakedown clock. + +FROZEN spec (from #22 + pre-declared; do not tune here): + set maker-sharps screened on strictly-prior tape, published + nightly by maker_set.py (the box never screens itself) + inventory rolling net/gross USD per (wallet, token) from the + orders_matched stream — inventory absorbed while quoting + trigger first crossing per (wallet, token, day) of + |net|*px >= $150 AND |net|/gross >= 0.6, lean px in [0.05,0.95] + band follow arm only: lean size $150-500 (the #22 verdict arm) + entry $100 FAK walk at asks <= min(print + 0.03, 0.95) + fee 3%*sh*min(p,1-p); MAX_PER_EVENT=2 concurrent +Live-only plumbing (never signal): 30-min warmup, worker pool + bounded +queue (throttle counted), map trims, registry-style settle passes. +""" +import json +import os +import queue +import ssl +import threading +import time +import urllib.request + +import websocket + +WS_URL = "wss://ws-live-data.polymarket.com" +CLOB = "https://clob.polymarket.com" +SET_URL = ("https://raw.githubusercontent.com/jaxperro/winning-wallet-finder/" + "main/research/params/maker_set.json") +STATE = os.environ.get("LEAN_STATE", "/data/lean_state.json") +ATTEMPTS = os.environ.get("LEAN_ATTEMPTS", "/data/lean_attempts.jsonl") +SETTLES = os.environ.get("LEAN_SETTLES", "/data/lean_settles.jsonl") +SEM_VER = "n1" # bump on ANY semantics-altering change + +# ── FROZEN — #22 params + pre-declared execution rule ────────────────────── +LEAN_USD_MIN = 150.0 # == maker_lean.LEAN_USD +LEAN_USD_MAX = 500.0 # the #22 follow arm's upper edge +NET_GROSS = 0.6 +BAND = (0.05, 0.95) +PREMIUM_CAP = 0.03 +STAKE = 100.0 +FEE_RATE = 0.03 +MAX_PER_EVENT = 2 +# ── LIVE-ONLY plumbing ───────────────────────────────────────────────────── +WARMUP_S = 1800 +SET_REFRESH_S = 3600 +BOOK_WORKERS = 4 +BOOK_QUEUE = 16 +CLOB_PASS_CAP = 25 +SETTLED_TRIM = 10000 +SKIPS_TRIM = 500 +INV_TRIM = 20000 # bounded inventory map +MAP_IDLE_TRIM_S = 86400 +TOP_LEVELS = 5 +SSL_CTX = ssl._create_unverified_context() + + +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()) + + +def walk_asks(asks, cap, stake_usd=STAKE): + """FAK against a CLOB ask list within cap; partial fills kept; top raw + levels recorded for offline replays. asks arrive UNSORTED.""" + 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, "top": top} + best = lv[0][0] + spent = shares = 0.0 + levels = 0 + for pxx, sz in lv: + if pxx > cap or spent >= stake_usd - 1e-9: + break + take = min(sz, (stake_usd - spent) / pxx) + if take <= 0: + break + shares += take + spent += take * pxx + levels += 1 + if not shares: + return {"filled": False, "best_ask": best, "top": top} + return {"filled": True, "vwap": spent / shares, "shares": shares, + "usd": spent, "levels": levels, "best_ask": best, "top": top, + "partial": spent < stake_usd - 1e-9} + + +def fresh_state(): + return { + "sem_ver": SEM_VER, "boot_ts": int(time.time()), + "open": {}, "settled": [], "skips": [], + "pnl_realized": 0.0, + "counters": {"crossings": 0, "attempts": 0, "fills": 0, + "craters": 0, "premium_skips": 0, "throttle_skip": 0, + "book_fail": 0, "settled_total": 0, "settle_clob": 0, + "skip": {"warmup": 0, "not_in_set": 0, "band_usd": 0, + "band_px": 0, "no_print": 0, "event_cap": 0, + "no_complement": 0, "already_fired": 0}}} + + +class LeanBot: + def __init__(self): + self.lock = threading.Lock() + self.boot = time.time() + self.state = fresh_state() + if os.path.exists(STATE): + try: + old = json.load(open(STATE)) + if old.get("sem_ver") == SEM_VER: + self.state = old + self.state["boot_ts"] = int(time.time()) + log(f"state resumed: {len(old.get('open', {}))} open · " + f"{old['counters']['settled_total']} settled") + else: + log(f"SEM_VER changed {old.get('sem_ver')}->{SEM_VER} — " + f"fresh state (shakedown clock restarts)") + except Exception as e: + log(f"state load failed ({e}) — fresh") + self.sharps = set() + self.set_meta = {} + self.set_ts = 0 + self.inv = {} # (wallet, asset) -> [net_usd, gross_usd, ts] + self.fired = set() # (wallet, asset, day) already triggered + self.last_px = {} # asset -> (ts, price) + self.cond_assets = {} # cond -> {asset, ...} + self.tok_meta = {} # asset -> (cond, event, title, outcome) + self.book_q = queue.Queue(BOOK_QUEUE) + + # ── the screened set (never computed here — published by maker_set.py) ── + def refresh_set(self): + try: + d = get_json(SET_URL, timeout=15) + ws = {w.lower() for w in d.get("wallets") or []} + if ws: + with self.lock: + self.sharps = ws + self.set_meta = {"n": len(ws), "as_of": d.get("as_of"), + "computed_at": d.get("computed_at")} + self.set_ts = time.time() + log(f"maker set: {len(ws)} wallets (as-of " + f"{time.strftime('%F', time.gmtime(d.get('as_of') or 0))})") + except Exception as e: + log(f"⚠ maker set fetch failed: {e}") + + # ── stream ───────────────────────────────────────────────────────────── + def on_trade(self, p): + """activity/trades — price marks + token metadata (the orders_matched + payload often carries empty conditionId/eventSlug; trades do not).""" + a = p.get("asset") + if not a: + return + try: + px = float(p.get("price") or 0) + except (TypeError, ValueError): + return + now = time.time() + with self.lock: + if 0 < px < 1: + self.last_px[a] = (now, px) + cond = p.get("conditionId") or "" + if cond: + self.cond_assets.setdefault(cond, set()).add(a) + self.tok_meta[a] = (cond, p.get("eventSlug") or "", + (p.get("title") or "")[:80], + p.get("outcome") or "") + + def on_match(self, p): + """activity/orders_matched — the MAKER side of every match. Inventory + absorbed while quoting, which is the #22 signal substrate.""" + w = (p.get("proxyWallet") or "").lower() + a = p.get("asset") + if not w or not a: + return + try: + px = float(p.get("price") or 0) + sz = float(p.get("size") or 0) + except (TypeError, ValueError): + return + if px <= 0 or sz <= 0: + return + now = time.time() + usd = px * sz + with self.lock: + cond = p.get("conditionId") or "" + if cond: + self.cond_assets.setdefault(cond, set()).add(a) + self.tok_meta.setdefault(a, (cond, p.get("eventSlug") or "", + (p.get("title") or "")[:80], + p.get("outcome") or "")) + if w not in self.sharps: + self.state["counters"]["skip"]["not_in_set"] += 1 + return + key = (w, a) + st = self.inv.get(key) or [0.0, 0.0, now] + st[0] += usd if p.get("side") == "BUY" else -usd + st[1] += usd + st[2] = now + self.inv[key] = st + if len(self.inv) > INV_TRIM: + for k in sorted(self.inv, key=lambda k: self.inv[k][2])[:2000]: + del self.inv[k] + if time.time() - self.boot < WARMUP_S: + self.state["counters"]["skip"]["warmup"] += 1 + return + self._maybe_fire(w, a, st, px, now) + + def _maybe_fire(self, w, a, st, px, now): # under lock + net, gross = st[0], st[1] + if gross < 1e-9: + return + lean_usd = abs(net) + if lean_usd < LEAN_USD_MIN or abs(net) / gross < NET_GROSS: + return + day = int(now // 86400) + if (w, a, day) in self.fired: + self.state["counters"]["skip"]["already_fired"] += 1 + return + self.state["counters"]["crossings"] += 1 + self.fired.add((w, a, day)) + if lean_usd >= LEAN_USD_MAX: # follow arm is $150-500 + self.state["counters"]["skip"]["band_usd"] += 1 + return + # lean side: net>0 = they are accumulating THIS token; net<0 = they + # are net short it, so the lean is the complement token + if net > 0: + target, lean_px = a, px + else: + cond = (self.tok_meta.get(a) or ("",))[0] + sibs = [x for x in self.cond_assets.get(cond, set()) if x != a] + if len(sibs) != 1: + self.state["counters"]["skip"]["no_complement"] += 1 + return + target = sibs[0] + lean_px = 1 - px + lp = self.last_px.get(target) + if not lp or now - lp[0] > 3600: + self.state["counters"]["skip"]["no_print"] += 1 + return + print_px = lp[1] + if not (BAND[0] <= print_px <= BAND[1]): + self.state["counters"]["skip"]["band_px"] += 1 + return + cond, event, title, outcome = self.tok_meta.get( + target, ("", "", "", "")) + if event and sum(1 for l in self.state["open"].values() + if l.get("event") == event) >= MAX_PER_EVENT: + self.state["counters"]["skip"]["event_cap"] += 1 + return + job = {"wallet": w, "src_asset": a, "asset": target, "cond": cond, + "event": event, "title": title, "outcome": outcome, + "lean_usd": round(lean_usd, 2), "net": round(net, 2), + "gross": round(gross, 2), "side": 1 if net > 0 else -1, + "print": round(print_px, 4), "lean_px": round(lean_px, 4), + "ts": round(now, 3)} + try: + self.book_q.put_nowait(job) + self.state["counters"]["attempts"] += 1 + except queue.Full: + self.state["counters"]["throttle_skip"] += 1 + self._skip(job, "throttle", None) + self.log_attempt({**job, "filled": False, "why": "throttle"}) + + # ── 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}") + with self.lock: + self.state["counters"]["book_fail"] += 1 + + def attempt(self, job): + t0 = time.time() + try: + bk = get_json(f"{CLOB}/book?token_id={job['asset']}") + except Exception: + with self.lock: + self.state["counters"]["book_fail"] += 1 + return + lat_ms = int((time.time() - t0) * 1000) + cap = min(job["print"] + PREMIUM_CAP, BAND[1]) + r = walk_asks(bk.get("asks"), cap) + premium = (r["best_ask"] - job["print"]) if r["best_ask"] else None + rec = {**job, "latency_ms": lat_ms, "best_ask": r["best_ask"], + "premium": round(premium, 4) if premium is not None else None, + "top": r["top"], "cap": round(cap, 4), "filled": r["filled"]} + with self.lock: + c = self.state["counters"] + if not r["filled"]: + why = "crater" if r["best_ask"] is None else "premium" + c["craters" if why == "crater" else "premium_skips"] += 1 + rec["why"] = why + self._skip(job, why, r["best_ask"]) + log(f"{why.upper()} {job['title'][:34]} print " + f"{job['print']:.2f} ask " + f"{r['best_ask'] if r['best_ask'] is not None else '—'} " + f"(lean ${job['lean_usd']:.0f}, {lat_ms}ms)") + self.persist() + else: + fee = FEE_RATE * r["shares"] * min(r["vwap"], 1 - r["vwap"]) + lot = {**job, "price": round(r["vwap"], 5), + "best_ask": r["best_ask"], "premium": rec["premium"], + "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 + rec.update({"price": lot["price"], "shares": lot["shares"], + "usd": lot["cost"], "fee": lot["fee"], + "partial": lot["partial"]}) + log(f"FILL {job['title'][:34]} @ {lot['price']:.3f} " + f"(print {job['print']:.2f} prem " + f"{rec['premium'] if rec['premium'] is not None else 0:+.3f}" + f", lean ${job['lean_usd']:.0f}, ${lot['cost']:.2f}, " + f"{lat_ms}ms)") + self.persist() + self.log_attempt(rec) + + def log_attempt(self, rec): + 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 _skip(self, job, why, ba): # under lock + self.state["skips"].append( + {"ts": int(job["ts"]), "asset": job["asset"], "why": why, + "print": job["print"], "ba": ba, "title": job["title"], + "lean_usd": job["lean_usd"]}) + del self.state["skips"][:-SKIPS_TRIM] + + # ── settles: CLOB backstop -> nightly chain truth ────────────────────── + def settle_clob_pass(self): + with self.lock: + now = time.time() + due = list(self.state["open"].keys()) + 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 + snap[lid] = (lot.get("cond"), lot["asset"]) + verdicts = {} + for lid, (cond, asset) in snap.items(): + if not cond: + continue + 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) + n += 1 + if n: + self.persist() + + def _settle(self, lid, lot, pay): # under lock + pnl = round(lot["shares"] * pay - lot["cost"] - lot["fee"], 2) + self.state["pnl_realized"] = round(self.state["pnl_realized"] + pnl, 2) + c = self.state["counters"] + c["settled_total"] += 1 + c["settle_clob"] += 1 + rec = {**lot, "payout": pay, "provisional": True, + "settled_ts": int(time.time()), "pnl": pnl} + self.state["settled"].append(rec) + del self.state["settled"][:-SETTLED_TRIM] + del self.state["open"][lid] + try: + with open(SETTLES, "a") as fh: + fh.write(json.dumps(rec) + "\n") + except Exception as e: + log(f"⚠ settles log write failed: {e}") + word = "WON" if pay == 1.0 else "refund" if pay == 0.5 else "lost" + log(f"SETTLE {word} {lot['title'][:34]} {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 trim_maps(self): # under lock (minute loop) + now = time.time() + if len(self.last_px) > 400000: + self.last_px = {a: v for a, v in self.last_px.items() + if now - v[0] < MAP_IDLE_TRIM_S} + day = int(now // 86400) + self.fired = {k for k in self.fired if k[2] >= day - 1} + + def feed_dict(self): # under lock + stt = self.state + return { + "mode": "paper", "study": "lean (Study C stage-2)", + "sem_ver": SEM_VER, "updated": int(time.time()), + "boot_ts": stt["boot_ts"], + "warmup_left_s": max(0, int(self.boot + WARMUP_S - time.time())), + "frozen": {"lean_usd": [LEAN_USD_MIN, LEAN_USD_MAX], + "net_gross": NET_GROSS, "premium_cap": PREMIUM_CAP, + "stake": STAKE, "band": BAND, "fee_rate": FEE_RATE, + "max_per_event": MAX_PER_EVENT}, + "set": self.set_meta, "set_age_s": int(time.time() - self.set_ts), + "tracked_inventories": len(self.inv), + "counters": stt["counters"], "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:], + "note": "verdict binds to #26; median ask-premium >= 8c over 3 " + "days is the observational kill-switch (graded nightly " + "from the attempts stream)"} + + +SUB = json.dumps({"action": "subscribe", "subscriptions": [ + {"topic": "activity", "type": "trades", "filters": ""}, + {"topic": "activity", "type": "orders_matched", "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 + if m.get("topic") != "activity": + return + try: + if m.get("type") == "trades": + bot.on_trade(m.get("payload") or {}) + elif m.get("type") == "orders_matched": + bot.on_match(m.get("payload") or {}) + except Exception as e: + log(f"⚠ on_message 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): + 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.""" + bot = LeanBot.__new__(LeanBot) + bot.state = fresh_state() + bot.lock = threading.Lock() + bot.boot = time.time() - WARMUP_S - 1 + bot.sharps = {"0xsharp"} + bot.set_meta, bot.set_ts = {}, time.time() + bot.inv, bot.fired = {}, set() + bot.last_px, bot.cond_assets, bot.tok_meta = {}, {}, {} + bot.book_q = queue.Queue(4) + jobs = [] + bot.book_q.put_nowait = lambda j: jobs.append(j) # capture, no network + + def mk(w, a, side, px, sz, cond="c1"): + return {"proxyWallet": w, "asset": a, "side": side, "price": px, + "size": sz, "conditionId": cond, "eventSlug": "ev1", + "title": "T", "outcome": "X"} + bot.on_trade({"asset": "tokA", "price": 0.30, "conditionId": "c1", + "eventSlug": "ev1", "title": "T", "outcome": "X"}) + bot.on_trade({"asset": "tokB", "price": 0.70, "conditionId": "c1", + "eventSlug": "ev1", "title": "T", "outcome": "Y"}) + # not in set -> ignored + bot.on_match(mk("0xrando", "tokA", "BUY", 0.30, 2000)) + assert not jobs, "non-set wallet fired" + # in set, sub-threshold ($60) -> no fire + bot.on_match(mk("0xsharp", "tokA", "BUY", 0.30, 200)) + assert not jobs, "sub-threshold fired" + # crosses $150 one-sided -> fires on tokA + bot.on_match(mk("0xsharp", "tokA", "BUY", 0.30, 400)) + assert len(jobs) == 1, f"expected 1 job, got {len(jobs)}" + assert jobs[0]["asset"] == "tokA" and jobs[0]["side"] == 1 + assert jobs[0]["print"] == 0.30, jobs[0]["print"] + # same (wallet, token, day) cannot fire twice + bot.on_match(mk("0xsharp", "tokA", "BUY", 0.30, 400)) + assert len(jobs) == 1, "re-fired same day" + # net-short lean routes to the complement token + bot.on_match(mk("0xsharp", "tokB", "SELL", 0.70, 400)) + assert len(jobs) == 2 and jobs[1]["asset"] == "tokA", jobs[1] + assert jobs[1]["side"] == -1 + # over the $500 band edge -> counted, no job + bot.on_match(mk("0xsharp", "tokZ", "BUY", 0.50, 4000)) + assert len(jobs) == 2, "over-band fired" + assert bot.state["counters"]["skip"]["band_usd"] == 1 + # walk_asks: cap respected, partials kept + r = walk_asks([{"price": 0.33, "size": 100}, {"price": 0.40, "size": 900}], + 0.33) + assert r["filled"] and abs(r["usd"] - 33.0) < 1e-6 and r["partial"], r + r2 = walk_asks([{"price": 0.99, "size": 10}], 0.33) + assert not r2["filled"] and r2["best_ask"] == 0.99, r2 + assert WARMUP_S >= 60, "warmup must exceed a burst window" + log("selftest OK") + + +def main(): + selftest() + bot = LeanBot() + bot.refresh_set() + if not bot.sharps: + log("⚠ no maker set yet — idling until the nightly publishes one") + threading.Thread(target=serve_feed, args=(bot,), daemon=True).start() + for i 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() + tick = 0 + while True: + time.sleep(60) + tick += 1 + try: + if time.time() - bot.set_ts > SET_REFRESH_S: + bot.refresh_set() + if tick % 2 == 0: + bot.settle_clob_pass() + with bot.lock: + bot.trim_maps() + if tick % 5 == 0: + bot.persist() + c = bot.state["counters"] + log(f"paper pnl {bot.state['pnl_realized']:+.2f} · open " + f"{len(bot.state['open'])} · cross {c['crossings']} " + f"att {c['attempts']} fill {c['fills']} " + f"prem-skip {c['premium_skips']} crat {c['craters']} · " + f"inv {len(bot.inv)} · set {len(bot.sharps)} · " + f"settled {c['settled_total']}") + except Exception as e: + log(f"⚠ loop error: {e}") + + +if __name__ == "__main__": + main() diff --git a/research/maker_set.py b/research/maker_set.py new file mode 100644 index 00000000..92b47f21 --- /dev/null +++ b/research/maker_set.py @@ -0,0 +1,53 @@ +#!/usr/bin/env python3 +"""Publish the walk-forward screened maker-sharp set for wwf-leanbot (#26). + +The bot must NEVER screen on its own live data — that is how a study +starts selecting the wallets that happened to win during its own window. +So the set is computed here, on tape strictly BEFORE today, exactly as +maker_lean.py's screen_asof does (the frozen #22 method), and published +to research/params/maker_set.json for the box to fetch. + +Wired into nightly.sh before the graders. Same contract as +informed_set.py (surgebot's input): committed file = frozen input. +""" +import json +import os +import sys +import time + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +import tape # noqa: E402 +import maker_lean as ml # noqa: E402 + +OUT = os.path.join(HERE, "params", "maker_set.json") + + +def main(): + db = tape.connect() + # midnight UTC today — the set may only see strictly-prior tape + t_cut = int(time.time() // 86400) * 86400 + sharps = ml.screen_asof(db, t_cut) + db.close() + out = { + "computed_at": int(time.time()), + "as_of": t_cut, + "method": "maker_lean.screen_asof (z>=2.5, >=6 resolved maker bets, " + "pnl>0) on orders_matched strictly before as_of", + "frozen_params": {"lean_usd": [ml.LEAN_USD, 500.0], + "net_gross": ml.NET_GROSS, "band": list(ml.BAND)}, + "n": len(sharps), + "wallets": sorted(sharps), + } + os.makedirs(os.path.dirname(OUT), exist_ok=True) + tmp = OUT + ".tmp" + json.dump(out, open(tmp, "w"), indent=1) + os.replace(tmp, OUT) + print(f"[maker_set] {len(sharps)} wallets as-of " + f"{time.strftime('%F', time.gmtime(t_cut))} -> params/maker_set.json", + flush=True) + return 0 + + +if __name__ == "__main__": + sys.exit(main())