#!/usr/bin/env python3 """copybot.py — push-driven, live-capable Polymarket copy-trader. Marries the two halves you already built: * a layered **push** trigger stack — T0 is the RTDS trade stream (`RtdsListener`: every platform trade, wallet-attributed, ~1s — README gotcha 18), backstopped by Alchemy's Address-Activity webhook (~3s, POST /alchemy), a 5-min backstop poll over per-wallet trade cursors, and the reconcile janitor. The live bot also streams its OWN fills (`UserFillsListener`) to adopt in-play holds the moment they match. * copytrade.py's hardened **execution engine** — paper executor + the LIVE executor on the unified SDK (`polymarket-client`; py-clob-client is archived and the CLOB rejects its orders — README gotcha 16), risk gates, absolute price guard, the depth gate (stake ≤10% of visible 5c ask depth), no-backfill seeding, proportional entry/exit mirroring, and the pending-order registry for in-play `delayed` holds (gotcha 17: an accepted-but-held order is NEVER a rejection). Flow: Alchemy POST /alchemy → enrich the tx via the Polymarket data-API (market, side, price, size) → FollowFilter — the "only the trades I actually want" gate → CopyTrader.handle_trade — sizes + places under the risk gates The execution engine is unchanged; this file only swaps the *trigger* from a poll loop to a push, and inserts the follow-filter in front of it. SAFETY — paper by default. Live trading needs ALL of: 1. "mode": "live" in config.json, 2. the --live flag, 3. typing the confirmation phrase (env LIVE_CONFIRM on the worker), 4. polymarket-client installed + live.private_key (env LIVE_PRIVATE_KEY). Sizing is paper-parity 4%-of-equity floored at the venue $1 minimum (hard caps retired 2026-07-10, user decision); the geo-gate is fatal in live mode. This is real money in live mode. Endpoints (stdlib http server, binds $PORT or 8080): POST /alchemy ← point the Alchemy webhook here GET /health ← uptime check Usage: python3 copybot.py # paper, listen for webhooks python3 copybot.py --conviction-from-sharps live/watch_sharps.json python3 copybot.py --test-wallet 0x.. # dry-run the pipeline on a wallet's # latest trade, then exit (no server) python3 copybot.py --live # live (needs mode:live + confirm) """ import argparse import uuid import hashlib import hmac import json import os import ssl import sys import threading import time import urllib.error import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from copytrade import ( # the execution engine (sizing, gates, executors) CopyTrader, PaperExecutor, DEFAULT_CONFIG, load_json, save_json, new_state, recent_trades, confirm_live, clob_price, book_depth, ) import smart_money as sm # noqa: E402 from smart_money import SSL_CTX # noqa: E402 CLOB_API = "https://clob.polymarket.com" # ── on-chain resolution (ConditionalTokens payout vectors) ─────────────────── # The CLOB's `winner` flags NEVER populate for operator-resolved markets # (in-play set winners, game O/Us — the whale class's staple), and data-api # curPrice on a dead book reads 0.5 whether the market refunded or one side # won (2026-07-06 audit: four resolved positions sat unsettleable for hours). # The chain is the source every redeem actually pays from: payoutNumerators/ # payoutDenominator on the CTF contract — 1/0 winners, [0.5,0.5] refunds, # denominator 0 = not resolved. Selectors are keccak4 of the signatures. CTF_ADDR = "0x4D97DCd97eC945f40cF65F87097ACe5EA0476045" _SEL_DEN = "0xdd34de67" # payoutDenominator(bytes32) _SEL_NUM = "0x0504c814" # payoutNumerators(bytes32,uint256) _RPC_URL = None # resolved once in main() (env/config), stays None without a key _RPC_FAILS = 0 # consecutive onchain_payouts failures (closes #8: a dead # key silently stalled tier-2 settlement forever) _PAYOUTS = {} # cond -> [p0, p1], cached once resolved (immutable) def _rpc(method, params): body = json.dumps({"jsonrpc": "2.0", "id": 1, "method": method, "params": params}).encode() req = urllib.request.Request(_RPC_URL, data=body, headers={"Content-Type": "application/json"}) with urllib.request.urlopen(req, timeout=15, context=SSL_CTX) as r: return json.loads(r.read())["result"] def _eth_call(data): return _rpc("eth_call", [{"to": CTF_ADDR, "data": data}, "latest"]) def onchain_payouts(cond): """[payout_outcome0, payout_outcome1] for a resolved condition (order matches the CLOB market's tokens[] order), or None if unresolved / no RPC configured.""" if not _RPC_URL or not cond: return None if cond in _PAYOUTS: return _PAYOUTS[cond] global _RPC_FAILS try: c = cond[2:].rjust(64, "0") den = int(_eth_call(_SEL_DEN + c), 16) _RPC_FAILS = 0 if not den: return None nums = [int(_eth_call(_SEL_NUM + c + hex(i)[2:].rjust(64, "0")), 16) for i in (0, 1)] _PAYOUTS[cond] = [n / den for n in nums] return _PAYOUTS[cond] except Exception: _RPC_FAILS += 1 return None # ── T0b: fills straight from the tx receipt (Alchemy push) ────────────────── # RTDS doesn't emit every market (2026-07-13: badaf's "Credible public sale" # entry was missed on BOTH books — RTDS silent, and the data-api indexer # lagged past the 600s window, so the ~3s Alchemy push detected a trade it # couldn't fetch). The push carries the tx hash, and the fill is fully # decodable from the receipt — so no market's copy speed depends on RTDS # coverage or indexer freshness. _ORDER_FILLED = "0xd543adfd945773f1a62f74f0ee55a5e3b9b1a28262980ba90b1a89f2ea84d8ee" _EXCHANGES = { # new-stack (pUSD) exchange contracts, the only trusted # OrderFilled emitters — an attacker CAN trigger the webhook with a dust # transfer, so never decode fills from arbitrary contracts' events. # Both observed across 200/200 shadow-ledger fills (2026-07-14). "0xe111180000d2663c0091e4f400237545b87b996b", "0xe2222d279d744050d28e00520010520000310f59", } _TOK_MKT = {} # token id -> (title, outcome, cond), immutable _TOK_LOCK = threading.Lock() def _token_market(token): """token id -> (title, outcome, conditionId) via the gamma API, or None. A failed lookup means the caller drops the seed — the backstop poll still copies the trade once the indexer catches up (honest, never guessy).""" with _TOK_LOCK: if token in _TOK_MKT: return _TOK_MKT[token] try: req = urllib.request.Request( f"https://gamma-api.polymarket.com/markets?clob_token_ids={token}", headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=12, context=SSL_CTX) as r: m = (json.loads(r.read().decode()) or [{}])[0] toks = json.loads(m["clobTokenIds"]) got = (m.get("question") or "", json.loads(m["outcomes"])[toks.index(token)], m.get("conditionId")) except Exception: return None with _TOK_LOCK: _TOK_MKT[token] = got return got def fills_from_tx(tx, wallet): """The watched wallet's fills decoded from the receipt, as data-api-shaped trade dicts ([] without an RPC / on any failure — backstops still cover). OrderFilled topics: (sig, orderHash, maker, taker); the wallet's own order is the maker==wallet row (aggregate when it took, per-order when it made). Data words: (assetClassGiven, tokenId, amountGiven, amountTaken, fee…) in 6dp base units; class 0 = gave cash = BUY, else gave tokens = SELL. Validated 200/200 against shadow-ledger fills — side, size, price exact.""" if not _RPC_URL: return [] try: rcpt = _rpc("eth_getTransactionReceipt", [tx]) except Exception: return [] if not rcpt: return [] try: # block ts keeps the staleness gate + lag stat honest ts = int(_rpc("eth_getBlockByNumber", [rcpt["blockNumber"], False])["timestamp"], 16) except Exception: ts = int(time.time()) w40, out = wallet[2:].lower(), [] for lg in rcpt.get("logs", []): t = lg.get("topics") or [] if (len(t) < 4 or t[0].lower() != _ORDER_FILLED or (lg.get("address") or "").lower() not in _EXCHANGES or t[2][-40:].lower() != w40): continue d = lg.get("data", "0x")[2:] wd = [int(d[i:i + 64], 16) for i in range(0, len(d), 64)] if len(wd) < 4: continue side = "BUY" if wd[0] == 0 else "SELL" token, given, taken = str(wd[1]), wd[2] / 1e6, wd[3] / 1e6 usd, size = (given, taken) if side == "BUY" else (taken, given) if not size: continue meta = _token_market(token) if not meta: log(f"chainseed: no metadata for token …{token[-8:]} " f"({tx[:10]}…) — left to the backstop poll") continue out.append({"transactionHash": tx, "asset": token, "side": side, "size": size, "price": round(usd / size, 6), "usdcSize": round(usd, 2), "title": meta[0], "outcome": meta[1], "conditionId": meta[2], "timestamp": ts}) return out # follow-filter defaults — merged under cfg["follow"]; permissive so nothing is # silently dropped until you opt in. The engine's risk caps bound everything # regardless of these. FOLLOW_DEFAULT = { "buy_only": True, # SELLs only ever close a position we already hold "min_their_usd": 0.0, # global conviction floor: ignore their bets below this "per_wallet_min_usd": {}, # {wallet: usd} — overrides the global floor per wallet "min_entry": 0.0, # only copy entries with their fill price in this band "max_entry": 1.0, # (the archetype/copyability zone; 0.35–0.70 = value) } RECENT_TRADE_WINDOW_S = 600 # webhook just told us a trade happened; ignore stale # defaults — config keys feed_path / fill_log override them (LIVE_ROLLOUT 1.1: # a live-money run must never clobber the paper book's feed or fills ledger) FILL_LOG = "copybot_fills.jsonl" # append-only ledger of every copy fill + lag/slippage FEED = os.path.join("live", "copybot_live.json") # published feed the trading dashboard reads FEED_PUSH_MIN_S = 120 # min seconds between feed git-pushes (commit-on-change) # Polymarket taker fee (Fee Structure V2, live since 2026-03-30): # fee = shares × rate × p × (1−p) # charged on marketable BUYs and SELLs (we always take — FOK/market copies); # redeeming a resolved position on-chain is fee-free. Rate is per market # category — sports 0.03, finance/politics/tech 0.04, econ/culture/weather 0.05, # crypto 0.07, geopolitics 0. The follow set is currently all-sports; override # with "taker_fee_rate" in config.json if that changes. TAKER_FEE_RATE = 0.03 def taker_fee(shares, price, rate): return shares * rate * price * (1.0 - price) def log(m): print(f"{time.strftime('%H:%M:%S')} {m}", flush=True) # ── market resolution lookup (for settling held positions at resolution) ───── _MKT_CACHE = {} # cond -> market dict, cached only once resolved _MKT_LOCK = threading.Lock() def _market(cond): with _MKT_LOCK: if cond in _MKT_CACHE: return _MKT_CACHE[cond] try: req = urllib.request.Request(f"{CLOB_API}/markets/{cond}", headers={"User-Agent": "Mozilla/5.0"}) with urllib.request.urlopen(req, timeout=12, context=SSL_CTX) as r: m = json.loads(r.read().decode()) or {} except Exception: return None # only cache a RESOLVED market (it won't change); re-check live ones each pass. # CLOB semantics: unresolved markets report winner=False on EVERY token — # resolution flips exactly one to True. Only a True winner means resolved. if any(t.get("winner") is True for t in (m.get("tokens") or [])): with _MKT_LOCK: _MKT_CACHE[cond] = m return m def market_tokens(cond): m = _market(cond) return (m.get("tokens") or []) if m else None def market_neg_risk(cond): """True if the market settles through the Neg-Risk adapter (different redeem path). Best-effort from the CLOB market metadata.""" m = _market(cond) return bool(m.get("neg_risk")) if m else False def resolution_price(token_id, cond, outcome=None): """Settled price of our held token — 1.0 won, 0.0 lost, 0.5 refunded (50/50 resolution: walkovers/abandonments), None if not resolved yet. Two tiers: 1. CLOB `winner` flags — authoritative when present. CRITICAL semantics: winner=False on EVERY token of an UNRESOLVED market means "not yet", not "lost" (treating False as lost booked four winning live bets as -$180 of instant losses on 2026-07-02). Token-id match first — outcome labels are venue strings, the token id is the position. 2. On-chain CTF payout vector — the flags NEVER populate for operator-resolved markets (in-play set winners / game O/Us) and never reflect 50/50 refunds; the chain records both (2026-07-06: four refunded Gojo/Heide positions sat locked for hours).""" toks = market_tokens(cond) if not toks: return None if any(t.get("winner") is True for t in toks): def winp(t): return 1.0 if t.get("winner") is True else 0.0 for t in toks: if str(t.get("token_id")) == str(token_id): return winp(t) if outcome is not None: for t in toks: if t.get("outcome") == outcome: return winp(t) return None # tier 2: no winner flag — ask the chain (only meaningful once trading closed) m = _market(cond) if not (m and m.get("closed")): return None po = onchain_payouts(cond) if po is None: return None for i, t in enumerate(toks[:2]): if str(t.get("token_id")) == str(token_id): return po[i] return None class LedgerPaperExecutor(PaperExecutor): """Paper executor that records each fill (side/token/shares/price), so the run can track free cash, realized P&L, and lag/slippage. Paper fills at the live CLOB price the engine fetched — already capturing detection + price-drift lag, unlike the dashboard's zero-lag assumption.""" def __init__(self): self.fills = [] # [{side, token, shares, price}] since last drain def buy(self, token_id, shares, price, meta): r = super().buy(token_id, shares, price, meta) if r["ok"]: # a paper FAK no-match is not a fill self.fills.append({"side": "BUY", "token": token_id, "shares": r["filled_shares"], "price": r["price"]}) return r def sell(self, token_id, shares, price, meta): r = super().sell(token_id, shares, price, meta) self.fills.append({"side": "SELL", "token": token_id, "shares": r["filled_shares"], "price": r["price"]}) return r class LedgerLiveExecutor: """Live executor on the unified SDK (polymarket-client). py-clob-client was ARCHIVED May 2026 — the CLOB rejects its order format globally ('invalid order version'), which is why this no longer extends LiveExecutor. * **Marketable FAK/FOK** via place_market_order — the SDK owns tick conformity, neg-risk exchange routing, and fee handling internally (the hand-rolled tick rounding here crashed the bot twice on 2026-07-09; all of that machinery is gone). * **Protected prices** — max_price (BUY) / min_price (SELL) bound the fill at the engine's quoted price ± live.max_slippage_pct (default 5%), replacing the old round-to-tick limit. Clamped to [0.01, 0.99], valid on both 1c and 0.1c books, so the SDK's band check can't raise. * **Fill recording** — same ledger as paper. AcceptedOrder reports the matched amounts: BUY gives collateral (making) for shares (taking), SELL the reverse; avg fill price falls out of the ratio. * **Never raises into the trade loop** — any exception is an honest ok:False (the engine records a missed row). """ live = True def __init__(self, cfg): try: from polymarket import SecureClient except ImportError: sys.exit("Live mode needs the unified SDK: " "pip install --pre polymarket-client") live = cfg.get("live", {}) if not live.get("private_key"): sys.exit("Live mode needs live.private_key in the config.") # wallet auto-resolves to the signer's Deposit Wallet; no api_key at # runtime — trading approvals already exist (host/order_probe_v2.py). # create() is ready to use as-is (__enter__ is a no-op returning self). self.client = SecureClient.create(private_key=live["private_key"]) self.fills = [] ot = str(live.get("order_type", "FAK")).upper() self._otype = ot if ot in ("FAK", "FOK") else "FAK" self._slip = float(live.get("max_slippage_pct", 0.05)) def _shares_held(self, token_id): """Exchange-view share balance for a token — chain truth, the arbiter when an order's fate is ambiguous.""" b = self.client.get_balance_allowance(asset_type="CONDITIONAL", token_id=str(token_id)) return b.balance / 1e6 def _settle_uncertain(self, token_id, side, bal0, price, order_id=None, deadline_s=20, cancel=True): """An order may be resting/held at the exchange (in-play 'delayed' acceptance, or an exception after posting). Poll it to a terminal state, cancel whatever remains (unless the caller keeps it alive for the pending registry), and return (filled, avg_price) from the exchange's own balance diff. INVARIANT (2026-07-10 incident: six in-play acceptances were logged as misses and filled untracked minutes later): no order outlives this call untracked — it either reports its fill here or is handed to state["pending_orders"].""" import time px = price deadline = time.time() + deadline_s while order_id and time.time() < deadline: time.sleep(2) try: o = self.client.get_order(order_id=order_id) if float(o.size_matched or 0) > 0: px = float(o.price or price) if o.status not in ("live", "delayed"): break except Exception: # gone from the open view — terminal break if cancel: try: if order_id: self.client.cancel_order(order_id=order_id) else: # exception path: sweep the whole token ids = [o.id for o in self.client.list_open_orders( token_id=str(token_id))] if ids: self.client.cancel_orders(order_ids=ids) except Exception: pass # cancel of a just-matched order — fine try: bal1 = self._shares_held(token_id) except Exception: return 0.0, price # can't read truth: claim nothing; the # chain_cash_gap alarm catches the rest filled = (bal1 - bal0) if side == "BUY" else (bal0 - bal1) return max(filled, 0.0), px def _order(self, token_id, shares, price, side): import math sz = math.floor(shares * 100) / 100.0 # cost never exceeds the gated stake try: bal0 = self._shares_held(token_id) except Exception as e: # no truth anchor → refuse to place at all return {"ok": False, "filled_shares": 0.0, "price": price, "resp": f"pre-check failed: {e}", "paper": False} # protected prices must (a) scale WITH the price — flat 2dp rounding # zeroed the bound on sub-penny books — and (b) CONFORM to the tick: # the SDK rejects more decimals than the book's tick allows # (2026-07-13: a 4dp bound on a 1c SPX book failed a winning copy). # The quoted price is always a tick multiple, so its own decimal # count is the finest precision that is safe on this book; a coarser # bound is valid on finer ticks, never the reverse. d = 2 for cand in (2, 3, 4): if abs(round(price, cand) - price) < 1e-9: d = cand break else: d = 4 try: if side == "BUY": # share-flooring can shave a gated $1.00 stake to $0.99, which # the venue rejects ('invalid amount for a marketable BUY') — # 7 of 13 live rejections in the first 3 days. Bump to the $1 # minimum; worst case pays a cent over the gated stake. amt = max(round(sz * price, 2), 1.0) r = self.client.place_market_order( token_id=token_id, side="BUY", amount=amt, max_price=min(max(round(price * (1 + self._slip), d), price), 0.99), order_type=self._otype) else: r = self.client.place_market_order( token_id=token_id, side="SELL", shares=sz, min_price=min(max(round(price * (1 - self._slip), d), 10 ** -d), price), order_type=self._otype) except Exception as e: # NEVER raise into the trade loop — # but a timed-out post may still be resting: sweep + measure first filled, px = self._settle_uncertain(token_id, side, bal0, price) if filled > 0: return {"ok": True, "filled_shares": filled, "price": px, "resp": f"filled despite client error: {e}", "paper": False} return {"ok": False, "filled_shares": 0.0, "price": price, "resp": f"exception: {e}", "paper": False} if not getattr(r, "ok", False): # RejectedOrder: typed code + message return {"ok": False, "filled_shares": 0.0, "price": price, "resp": f"{getattr(r, 'code', '?')}: {getattr(r, 'message', r)}", "paper": False} making = float(r.making_amount or 0) # what we gave (matched) taking = float(r.taking_amount or 0) # what we got (matched) filled, usd = (taking, making) if side == "BUY" else (making, taking) px = usd / filled if filled else price if filled <= 0: # ACCEPTED with zero matched = in-play 'delayed'/'live' hold, NOT # a rejection (the 2026-07-10 lesson). Wait briefly in-call; if # still held, hand the order to the PENDING registry — the # heartbeat resolver adopts the fill when it lands or cancels at # TTL (the 20s cancel-everything version forfeited a Rune-Eaters # hold that filled at +4.5min and paid +$7.50). filled, px = self._settle_uncertain(token_id, side, bal0, price, order_id=r.order_id, deadline_s=8, cancel=False) if filled <= 0: return {"ok": False, "filled_shares": 0.0, "price": price, "pending": {"order_id": r.order_id, "bal0": bal0, "sz": sz}, "resp": {"order_id": r.order_id, "status": r.status, "note": "in-play hold — pending resolver"}, "paper": False} return {"ok": filled > 0, "filled_shares": filled, "price": px, "resp": {"order_id": r.order_id, "status": r.status, "making": making, "taking": taking, "trades": len(r.trade_ids)}, "paper": False} def buy(self, token_id, shares, price, meta): r = self._order(token_id, shares, price, "BUY") if r["ok"]: self.fills.append({"side": "BUY", "token": token_id, "shares": r["filled_shares"], "price": r["price"]}) return r def sell(self, token_id, shares, price, meta): r = self._order(token_id, shares, price, "SELL") if r["ok"]: self.fills.append({"side": "SELL", "token": token_id, "shares": r["filled_shares"], "price": r["price"]}) return r # ── follow-filter — "just the ones I want to follow" ──────────────────────── class FollowFilter: """Decides whether one of their trades is worth handing to the engine. This is the selection gate that sits in front of execution. A BUY must be in your follow set, clear the conviction (stake-size) floor, and land in the entry-price band. A SELL always passes — the engine then mirrors it only if we actually hold the token. """ def __init__(self, cfg): f = {**FOLLOW_DEFAULT, **cfg.get("follow", {})} self.buy_only = f["buy_only"] self.min_their_usd = float(f["min_their_usd"]) self.per_wallet = {k.lower(): float(v) for k, v in f["per_wallet_min_usd"].items()} self.min_entry = float(f["min_entry"]) self.max_entry = float(f["max_entry"]) # whale-class wallets (follow.wallet_class) are followed on EVERY trade: # no conviction floor at all — their whole book is the signal (both # current whales win ~92% across all bets, so size-filtering them only # costs flow). The entry-price band still applies (execution guard). self.whales = {w.lower() for w, c in (f.get("wallet_class") or {}).items() if c == "whale"} wl = cfg.get("watchlist") or [w["wallet"] for w in cfg.get("watch", [])] self.wallets = {w.lower() for w in wl} def floor(self, wallet): if wallet.lower() in self.whales: return 0.0 return self.per_wallet.get(wallet.lower(), self.min_their_usd) def check(self, wallet, t): """-> (follow: bool, reason_if_skipped: str|None).""" if wallet.lower() not in self.wallets: return False, "wallet not in follow set" side = t.get("side") if side == "SELL": return True, None # engine exits only if we hold if side != "BUY": return False, f"side {side}" usd = t.get("usdcSize") or t.get("size", 0) * t.get("price", 0) fl = self.floor(wallet) if usd < fl: return False, f"${usd:,.0f} < conviction floor ${fl:,.0f}" p = t.get("price", 0) if not (self.min_entry <= p <= self.max_entry): return False, f"entry {p:.2f} outside [{self.min_entry:.2f},{self.max_entry:.2f}]" return True, None def describe(self): pw = f" · {len(self.per_wallet)} per-wallet floors" if self.per_wallet else "" wh = f" · {len(self.whales)} whales follow-all" if self.whales else "" return (f"follow filter · {'BUY-only' if self.buy_only else 'BUY+SELL'} · " f"conviction ≥ ${self.min_their_usd:,.0f}{pw}{wh} · " f"entry [{self.min_entry:.2f},{self.max_entry:.2f}]") # ── T0: the real-time trade stream (RTDS) ─────────────────────────────────── class RtdsListener: """PRIMARY detection: Polymarket's real-time data socket streams EVERY platform trade, wallet-attributed, ~1s after it happens (measured 2026-07-10: median 0.8s over 22k msgs, ~45 msg/s at US-evening peak, zero drops in 45 min). topic=activity/type=trades is undocumented but official (it powers polymarket.com's live feed and is spec'd in Polymarket/real-time-data-client). Server-side filters are broken (real-time-data-client#34) — subscribe unfiltered, filter client-side. Resilience: if the lib is missing or the stream dies, detection degrades to the existing backstops (Alchemy push ~3s, 300s poll, reconcile janitor) — never to zero. Reconnects with capped backoff forever.""" URL = "wss://ws-live-data.polymarket.com" def __init__(self, bot): self.bot = bot self.watched = {w.lower() for w in bot.cfg.get("watchlist", [])} self.last_msg = 0.0 # last firehose message of any kind self.hits = 0 # Set E messages dispatched self.state = "off" def start(self): try: import websocket # websocket-client (fly.Dockerfile pin) except ImportError: log("rtds: websocket-client not installed — listener OFF " "(backstops cover detection)") return False threading.Thread(target=self._run, args=(websocket,), daemon=True, name="rtds").start() return True def status(self): if self.state != "up": return self.state age = time.time() - self.last_msg if self.last_msg else -1 return f"up {age:.0f}s" if age >= 0 else "up" def _run(self, websocket): backoff = 1 sub = json.dumps({"action": "subscribe", "subscriptions": [ {"topic": "activity", "type": "trades", "filters": ""}]}) def on_open(ws): ws.send(sub) self.state = "up" self.last_msg = time.time() # fresh baseline for the stale check log("rtds: connected — activity/trades stream up (T0 detection)") def ping(): # app-level ping keeps RTDS alive while ws.keep_running: time.sleep(5) try: ws.send('{"action":"ping"}') except Exception: break # SILENT-STALE guard (2026-07-11 04:00Z: the socket sat # "up" delivering nothing for 35 min — no close event, so # the reconnect loop never fired; backstops covered). # The platform firehose never goes quiet for 2 min even # at 4am UTC — force-close to trigger a clean reconnect. # A false trip is just a harmless retry. if time.time() - self.last_msg > 120: log("rtds: stream silent 120s — forcing reconnect") try: ws.close() except Exception: pass break threading.Thread(target=ping, daemon=True).start() def on_message(ws, raw): self._handle(raw) while True: try: app = websocket.WebSocketApp(self.URL, on_open=on_open, on_message=on_message) app.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) except Exception as e: log(f"rtds: listener error {str(e)[:80]}") self.state = "down" log(f"rtds: stream down — reconnect in {backoff}s " "(backstops still detecting)") time.sleep(backoff) backoff = min(backoff * 2, 60) def _handle(self, raw): self.last_msg = time.time() try: m = json.loads(raw) except Exception: return if m.get("topic") != "activity" or m.get("type") != "trades": return p = m.get("payload") or {} w = (p.get("proxyWallet") or "").lower() if w not in self.watched: return ts = p.get("timestamp") or 0 if ts > 1e12: ts /= 1000.0 lat = (time.time() - ts) if ts else None self.hits += 1 name = self.bot.names.get(w, w[:10]) log(f"rtds: {name} {p.get('side')} {p.get('size')} @ {p.get('price')}" f" · {str(p.get('title'))[:40]}" f"{f' · lat {lat:.1f}s' if lat is not None else ''}") # shadow ledger — the durable record the RTDS-vs-Alchemy comparison # reads (Fly logs are ephemeral; this file rides the publish commit). # `seen` = another trigger already processed this tx before RTDS won. try: tx = p.get("transactionHash") with open(os.path.join(self.bot.here, self.bot.shadow_log), "a") as fh: fh.write(json.dumps({ "ts": round(time.time(), 2), "trade_ts": round(ts, 2), "lat_s": round(lat, 2) if lat is not None else None, "wallet": w, "name": name, "side": p.get("side"), "size": p.get("size"), "price": p.get("price"), "title": str(p.get("title") or "")[:60], "tx": tx, "seen": bool(tx and tx in self.bot.engine.seen)}) + "\n") if self.hits % 500 == 0: # keep the committed file bounded path = os.path.join(self.bot.here, self.bot.shadow_log) lines = open(path).readlines()[-2000:] open(path, "w").writelines(lines) except Exception: pass # SEED the funnel with the RTDS payload itself (2026-07-13): the # message carries every field handle_trade needs, so the copy no # longer waits on the data-api indexer — which lagged 130-305s on # badaf/1kto1m crypto+index markets even though RTDS delivered at # ~0.3s. on_wallet_activity still re-fetches and merges (fill-split # clips), deduped by tx, so nothing is lost; the seed just guarantees # the trade is in the candidate set at ~1s regardless of the indexer. try: usd = float(p.get("size") or 0) * float(p.get("price") or 0) except (TypeError, ValueError): usd = 0 seed = {"transactionHash": tx, "asset": p.get("asset"), "side": p.get("side"), "size": p.get("size"), "price": p.get("price"), "usdcSize": round(usd, 2), "title": p.get("title"), "outcome": p.get("outcome"), "conditionId": p.get("conditionId"), "eventSlug": p.get("eventSlug") or p.get("slug"), "timestamp": int(ts) if ts else None} try: # same funnel as the Alchemy push — self.bot.on_wallet_activity(w, seed_trade=seed) # locks internally except Exception as e: log(f"rtds handler error: {e}") # ── own-fill push: the CLOB user channel ───────────────────────────────────── class UserFillsListener: """LIVE only. Streams OUR order/trade lifecycle from the canonical CLOB user channel and turns a matching event into an IMMEDIATE resolve_pendings() pass — in-play holds adopt the second they match instead of on the next 60s tick. Events only TRIGGER the resolver; get_order stays the single arbiter (the 2026-07-12 anti-phantom-cash invariant books nothing from ws payloads). Auth = the CLOB L2 creds the SDK already derived (client.credentials). On any failure the listener logs and stays down while the 60s poll carries — never below today.""" URL = "wss://ws-subscriptions-clob.polymarket.com/ws/user" def __init__(self, bot): self.bot = bot self.state = "off" self.last_msg = 0.0 def start(self): try: import websocket # noqa: F401 creds = self.bot.engine.ex.client.credentials self._auth = {"apiKey": creds.key, "secret": creds.secret, "passphrase": creds.passphrase} except Exception as e: log(f"user-ws: unavailable ({type(e).__name__}: {str(e)[:60]}) — " "pending adoption stays on the 60s poll") return False import websocket threading.Thread(target=self._run, args=(websocket,), daemon=True, name="user-ws").start() return True def _pending_ids(self): return {p.get("order_id") for p in self.bot.engine.state.get("pending_orders", [])} def _run(self, websocket): backoff = 1 def on_open(ws): # all-markets subscription: omit "markets" entirely (the SDK's # own UserSpec does the same) — no per-market management needed ws.send(json.dumps({"type": "user", "auth": self._auth})) self.state = "up" self.last_msg = time.time() log("user-ws: connected — own-fill events live") def ping(): while ws.keep_running: time.sleep(10) try: ws.send("PING") except Exception: break if time.time() - self.last_msg > 30: log("user-ws: silent 30s — forcing reconnect") try: ws.close() except Exception: pass break threading.Thread(target=ping, daemon=True).start() def on_message(ws, raw): self.last_msg = time.time() if raw == "PONG": return try: m = json.loads(raw) except Exception: return events = m if isinstance(m, list) else [m] hit = False for ev in events: if not isinstance(ev, dict): continue et = ev.get("event_type") ids = set() if et == "order" and ev.get("type") == "UPDATE" \ and float(ev.get("size_matched") or 0) > 0: ids.add(ev.get("id")) elif et == "trade" and ev.get("status") in ( "MATCHED", "MINED", "CONFIRMED"): ids.add(ev.get("taker_order_id")) for mo in ev.get("maker_orders") or []: if isinstance(mo, dict): ids.add(mo.get("order_id")) if ids & self._pending_ids(): hit = True if hit: log("user-ws: pending order matched — resolving now") try: self.bot.resolve_pendings() except Exception as e: log(f"user-ws resolver error: {e}") while True: try: app = websocket.WebSocketApp(self.URL, on_open=on_open, on_message=on_message) app.run_forever() # audit 3.6: this socket carries live L2 # API creds — VERIFY certs (CERT_NONE was cargo-culted from the # read-only RTDS pattern, where it stays) except Exception as e: log(f"user-ws: listener error {str(e)[:80]}") self.state = "down" log(f"user-ws: down — reconnect in {backoff}s (60s poll carries)") time.sleep(backoff) backoff = min(backoff * 2, 120) # ── the push → filter → execute bridge ────────────────────────────────────── class Copybot: def __init__(self, cfg, engine, filt, redeemer=None): self.cfg = cfg self.engine = engine self.filt = filt self.redeemer = redeemer # gap 2: on-chain redeem of resolved positions (live) # trade-by-trade Discord pings retired 2026-07-04: the only Discord # output now is the daily sharp digest (live/discord_daily.py) self.names = {} for w in cfg.get("watch", []): self.names[w["wallet"].lower()] = w.get("name", w["wallet"][:10]) self.skipped = set() # tx we've already evaluated-and-skipped (no re-log) self.negrisk_warned = set() # conds we've already warned need manual redeem self.lock = threading.Lock() # serialize engine/settle access (webhook is threaded) self.here = os.path.dirname(os.path.abspath(__file__)) self.feed_path = cfg.get("feed_path", FEED) self.fill_log = cfg.get("fill_log", FILL_LOG) self.shadow_log = cfg.get("shadow_log", "rtds_shadow.jsonl") # persisted across restarts via the engine's state file self.conds = engine.state.setdefault("conds", {}) # token_id -> conditionId (open positions) engine.state.setdefault("cash", cfg["bankroll_usd"]) # free cash (recycles on sell/resolution) engine.state.setdefault("lag", {"n": 0, "sum_s": 0.0, "sum_slip_pct": 0.0}) engine.state.setdefault("fees_paid", 0.0) # 24h rolling lag window — backfill from the fills ledger on first boot # after the rolling-window change (older state has 'lag' but no 'lag_recent') if "lag_recent" not in engine.state: engine.state["lag_recent"] = self._backfill_lag_recent() self.fee_rate = float(cfg.get("taker_fee_rate", TAKER_FEE_RATE)) def _backfill_lag_recent(self, window_s=86400): """[[ts, lag_s, slip], …] for fills in the last window_s, read from the fills ledger — seeds the rolling avg so it's populated immediately.""" out = [] now = time.time() try: with open(os.path.join(self.here, self.fill_log)) as fh: for line in fh: try: r = json.loads(line) except Exception: continue ts, lag = r.get("ts"), r.get("detect_lag_s") if ts and lag is not None and ts >= now - window_s: out.append([round(ts, 1), round(lag, 1), round(r.get("slippage_pct") or 0, 5)]) except FileNotFoundError: pass return out def _drain_fills(self): """Apply cash flows from any fills the engine just made; return the BUY fills so the caller can log lag/slippage against the source trade. Every marketable fill (buy or sell) pays the taker fee — in live mode the protocol charges it at match time, so the paper book must charge it too or it overstates the edge.""" ex = self.engine.ex buys = [] # atomic snapshot-and-swap (audit 3.1): two threads draining the same # shared list each applied every fill's cash delta before one cleared fills, ex.fills = (ex.fills if hasattr(ex, "fills") else []), [] if fills: for f in fills: sign = -1 if f["side"] == "BUY" else 1 fee = taker_fee(f["shares"], f["price"], self.fee_rate) f["fee"] = round(fee, 4) self.engine.state["cash"] += sign * f["shares"] * f["price"] - fee self.engine.state["fees_paid"] = self.engine.state.get("fees_paid", 0.0) + fee if f["side"] == "BUY": buys.append(f) else: # link the mirror-exit to its bet record so the sold leg shows # up in per-bet P&L (feed) — cash above is already correct. # Accumulate UNROUNDED (rounding per-fill drifted the ledger a # few cents per trim); the feed rounds at render time. b = self.engine.state.get("bets", {}).get(f["token"]) if b: b["sold_shares"] = b.get("sold_shares", 0) + f["shares"] b["sold_proceeds"] = (b.get("sold_proceeds", 0) + f["shares"] * f["price"] - fee) # sells go to the fills ledger too — the audit had no sell # trail to reconcile the ledger against try: with open(os.path.join(self.here, self.fill_log), "a") as fh: fh.write(json.dumps({ "ts": round(time.time(), 1), "side": "SELL", "token": f["token"], "shares": round(f["shares"], 4), "price": round(f["price"], 4), "fee": f["fee"], "mode": "live" if self.engine.ex.live else "paper", }) + "\n") except Exception: pass return buys def ledger_drift(self): """cash minus what the ledger implies it should be. The invariant: cash = bank + Σadjustments + Σsettled-bet P&L + Σopen(-cost-fee+sold). Non-zero means a booking bug — the audit found +$15.45 of Jul-5 accounting-migration residue this check would have caught same-day.""" st = self.engine.state bets = st.get("bets", {}) adj = sum(a["amount"] for a in st.get("adjustments", [])) realized = sum(b["pnl"] for b in bets.values() if b.get("pnl") is not None) # in-flight flows keyed on "no P&L booked yet", NOT on my_pos membership: # a fully-exited bet leaves my_pos immediately but only gets its pnl at # the next write_feed reconcile — keying on my_pos made the invariant # jump during exactly that window flows = sum(-(b["cost"] + (b.get("fee") or 0)) + (b.get("sold_proceeds") or 0) for b in bets.values() if b.get("pnl") is None) return st.get("cash", 0) - (self.cfg["bankroll_usd"] + adj + realized + flows) def _book_snapshot(self, token): """Top-of-book spread + $-depth within 5c of touch, captured per copy into the fills ledger. The calibration experiment's weakest model assumption is the flat SLIP haircut — 2026-07-08 probe: an open position showed a 9c spread with $77 of bids within 5c vs our $40 stakes. Weeks of these snapshots = an EMPIRICAL fill model for the backtest and a depth gate before sizing up. NB captured just AFTER our own fill, so ask-side depth is net of what we took — fine for a first-order model. Best-effort: never blocks or fails a copy.""" return book_depth(token) # dedupe 2026-07-19 (closes #6): was a line-for-line copy def _record_lag(self, wallet, t, fill): """Gap 1 — log the detection lag and price slippage of a copy: their fill time/price vs ours. Appends to copybot_fills.jsonl and tracks running averages for the summary, so the live cost of lag is measurable.""" now = time.time() their_ts = t.get("timestamp", 0) or 0 detect_s = (now - their_ts) if their_ts else None their_p = t.get("price", 0) or 0 my_p = fill["price"] slip_pct = (my_p - their_p) / their_p if their_p else 0.0 rec = { "ts": round(now, 1), "wallet": wallet, "token": str(fill["token"]), "name": self.names.get(wallet.lower(), wallet[:10]), "outcome": t.get("outcome"), "title": (t.get("title") or "")[:80], "detect_lag_s": round(detect_s, 1) if detect_s is not None else None, "their_price": round(their_p, 4), "my_price": round(my_p, 4), "slippage_pct": round(slip_pct, 4), "shares": round(fill["shares"], 2), "cost": round(fill["shares"] * my_p, 2), "fee": fill.get("fee", 0), "mode": "live" if self.engine.ex.live else "paper", "book": self._book_snapshot(fill["token"]), } try: with open(os.path.join(self.here, self.fill_log), "a") as fh: fh.write(json.dumps(rec) + "\n") except Exception: pass if detect_s is not None: lag = self.engine.state["lag"] lag["n"] += 1 # lifetime copy count (kept for the total) lag["sum_s"] += detect_s lag["sum_slip_pct"] += slip_pct # rolling 24h window so the reported avg reflects CURRENT execution # (push mode ~3s), not a lifetime average dragged up by the retired # 60s poll era (which made a 3s bot read as 48s) rec24 = self.engine.state.setdefault("lag_recent", []) rec24.append([round(now, 1), round(detect_s, 1), round(slip_pct, 5)]) cutoff = now - 86400 self.engine.state["lag_recent"] = [r for r in rec24 if r[0] >= cutoff] # record the placed bet for the live dashboard feed. AGGREGATE across # fills: an ADD to an existing open position must accumulate shares/ # cost/fees, not overwrite the record with just the last fill — that # made Cost show one fill while P&L settled the whole position. bets = self.engine.state.setdefault("bets", {}) prev = bets.get(fill["token"]) if prev and prev.get("status") == "open": # accumulate UNROUNDED (per-fill rounding drifted the ledger); the # feed rounds at render time sh = prev["shares"] + fill["shares"] cost = prev["cost"] + fill["shares"] * my_p prev.update(shares=sh, cost=cost, my_price=(cost / sh) if sh else prev["my_price"], fee=(prev.get("fee") or 0) + fill.get("fee", 0)) else: self._archive_settled(bets, fill["token"]) bets[fill["token"]] = { "token": fill["token"], "wallet": wallet, "name": self.names.get(wallet.lower(), wallet[:10]), "outcome": t.get("outcome"), "title": (t.get("title") or "")[:90], "their_price": round(their_p, 4), "my_price": round(my_p, 4), "slippage_pct": round(slip_pct, 4), "lag_s": round(detect_s, 1) if detect_s is not None else None, "shares": round(fill["shares"], 2), "cost": round(fill["shares"] * my_p, 2), "fee": fill.get("fee", 0), "opened": int(their_ts or now), "status": "open", "exit_price": None, "pnl": None, "settled": None, } log(f" ↳ lag {('%.0fs' % detect_s) if detect_s is not None else '?'} · " f"their {their_p:.3f} → mine {my_p:.3f} ({slip_pct:+.1%} slippage)") @staticmethod def _archive_settled(bets, tok): """Re-entering a token whose previous bet already settled must not OVERWRITE it — the dict is keyed by token and the old record's pnl vanished from realized (2026-07-09: WTI −9.97 and a Dota +19.30 clobbered; drift alarm caught it). Move the settled record to a history key; every aggregator iterates values(), so archived rows keep counting in realized/feed/win-rates.""" prev = bets.get(tok) if prev is not None and prev.get("status") != "open": bets[f"{tok}#{prev.get('settled') or int(time.time())}"] = prev def _synth_bet(self, tok, pos): """Bet record for an open my_pos position that has none (their_price None marks it synthesized — lag/slippage unknowable, the source trade is gone). Fee is ESTIMATED with the same taker_fee formula _drain_fills charges, so for a position whose cash WAS debited the ledger closes to 0 (same inputs, same result) — while a never-debited orphan shows drift = cost+fee exactly, which is what check_book's heal keys on.""" bets = self.engine.state.setdefault("bets", {}) self._archive_settled(bets, tok) b = bets.get(tok) or {} sh = pos.get("shares") or 0 cost = pos.get("cost") or 0 fee = b.get("fee") or (taker_fee(sh, cost / sh, self.fee_rate) if sh else 0) bets[tok] = { "token": tok, "wallet": pos.get("wallet", ""), "name": self.names.get((pos.get("wallet") or "").lower()) or b.get("name") or "?", "outcome": pos.get("outcome"), "title": (pos.get("title") or "")[:90], "their_price": None, "my_price": round(cost / sh, 4) if sh else None, "slippage_pct": None, "shares": round(sh, 2), "cost": round(cost, 2), "fee": round(fee, 4), "opened": b.get("opened") or int(time.time()), "status": "open", "exit_price": None, "pnl": None, "settled": None, } return bets[tok] def _record_untracked_buy(self, f): """A drained BUY fill no _record_lag call claimed — the handler that placed it died between handle_trade and _drain_fills (webhook catches the exception, the fill sits in ex.fills), so a LATER drain booked its cash under some other trade's iteration. Before 2026-07-08 that fill vanished: position in my_pos, cash debited, but no bet record and no conds entry — the invisible-orphan seam behind the $+36.35 drift. Book it now: audit line, bet record (from the fill + my_pos attribution), conds from the position.""" tok = f["token"] pos = self.engine.state["my_pos"].get(tok, {}) if pos.get("cond") and tok not in self.conds: self.conds[tok] = pos["cond"] try: with open(os.path.join(self.here, self.fill_log), "a") as fh: fh.write(json.dumps({ "ts": round(time.time(), 1), "side": "BUY", "untracked": True, "token": str(tok), "shares": round(f["shares"], 4), "price": round(f["price"], 4), "fee": f.get("fee", 0), "mode": "live" if self.engine.ex.live else "paper", }) + "\n") except Exception: pass bets = self.engine.state.setdefault("bets", {}) prev = bets.get(tok) if prev and prev.get("status") == "open": # ADD to an open record: accumulate unrounded, like _record_lag sh = prev["shares"] + f["shares"] cost = prev["cost"] + f["shares"] * f["price"] prev.update(shares=sh, cost=cost, my_price=(cost / sh) if sh else prev["my_price"], fee=(prev.get("fee") or 0) + f.get("fee", 0)) else: self._archive_settled(bets, tok) bets[tok] = { "token": tok, "wallet": pos.get("wallet", ""), "name": self.names.get((pos.get("wallet") or "").lower(), "?"), "outcome": pos.get("outcome"), "title": (pos.get("title") or "")[:90], "their_price": None, "my_price": round(f["price"], 4), "slippage_pct": None, "shares": round(f["shares"], 2), "cost": round(f["shares"] * f["price"], 2), "fee": f.get("fee", 0), "opened": int(time.time()), "status": "open", "exit_price": None, "pnl": None, "settled": None, } log(f" ↳ untracked buy booked: {(pos.get('title') or '?')[:42]} — " f"{f['shares']:.1f}sh @ {f['price']:.3f}") def _ledger_buy_tokens(self): """Tokens with a BUY line in the fills ledger — i.e. positions whose cash demonstrably went through _drain_fills. Lines from before 2026-07-08 carry no 'token' (they can't vouch for anyone) and no 'side' (they are all buys — sells only started logging 2026-07-06).""" toks = set() try: with open(os.path.join(self.here, self.fill_log)) as fh: for line in fh: try: r = json.loads(line) except Exception: continue if r.get("side", "BUY") == "BUY" and r.get("token"): toks.add(str(r["token"])) except FileNotFoundError: pass return toks def check_book(self, heal_cash=False): """The book invariant (HANDOFF proper fix, Option A), asserted after every trade and at boot: every my_pos token has a bet record, a conds entry, and its cost debited from cash. Records and conds self-correct here. The cash leg only heals at boot (heal_cash=True) and only when ledger_drift matches ONE candidate's cost+fee — candidates are positions with no drained-fill evidence (no ledger BUY line, record synthesized) — so a drift that matches nothing stays loudly visible instead of being papered over. Callers hold self.lock (webhook path) or run before threads start (boot).""" st = self.engine.state mp = st["my_pos"] bets = st.setdefault("bets", {}) fixed = False for tok, pos in list(mp.items()): b = bets.get(tok) if not b or b.get("status") != "open": self._synth_bet(tok, pos) log(f"⚠ BOOK: synthesized missing bet record — " f"{(pos.get('title') or '?')[:42]}") fixed = True if tok not in self.conds: if pos.get("cond"): self.conds[tok] = pos["cond"] log(f"⚠ BOOK: backfilled conds from my_pos — " f"{(pos.get('title') or '?')[:42]}") fixed = True else: log(f"⚠ BOOK: no conditionId for " f"{(pos.get('title') or '?')[:42]} — can't settle it " f"until a reconcile pass learns the market") drift = self.ledger_drift() if heal_cash and abs(drift) > 0.01: vouched = self._ledger_buy_tokens() for tok in list(mp): b = bets.get(tok) or {} # a real _record_lag record (their_price set) or a ledger BUY # line proves the fill was drained — cash side is fine if str(tok) in vouched or b.get("their_price") is not None: continue gap = (b.get("cost") or mp[tok].get("cost") or 0) + (b.get("fee") or 0) if gap > 0 and abs(drift - gap) <= max(0.10, 0.01 * gap): st["cash"] -= gap st["fees_paid"] = st.get("fees_paid", 0.0) + (b.get("fee") or 0) try: # the late debit IS the missed drain with open(os.path.join(self.here, self.fill_log), "a") as fh: fh.write(json.dumps({ "ts": round(time.time(), 1), "side": "BUY", "healed": True, "token": str(tok), "shares": round(mp[tok].get("shares", 0), 4), "price": b.get("my_price"), "fee": b.get("fee", 0), "mode": "live" if self.engine.ex.live else "paper", }) + "\n") except Exception: pass log(f"⚠ BOOK HEALED: debited ${gap:.2f} never-drained cost — " f"{(b.get('title') or '?')[:42]} (drift was ${drift:+.2f})") drift = self.ledger_drift() fixed = True if fixed: self.engine.persist() if abs(drift) > 0.05: # pennies = 4dp fill rounding, not a bug log(f"⚠ LEDGER DRIFT ${drift:+.2f} — check_book could not attribute it") return drift def write_feed(self): with self.lock: # audit 3.1: reconcile mutates bet records return self._write_feed_locked() def _write_feed_locked(self): """Publish the bot's live book to live/copybot_live.json — the feed the top of jaxperro.com/trading reads. Reconciles any open bet no longer held (mirror-sold) to 'closed'.""" st = self.engine.state bets = st.setdefault("bets", {}) mp = st["my_pos"] # my_pos -> bets: a position can land in my_pos without a bet record when # its fill was drained in a context where _record_lag didn't fire (e.g. a # second market on the same event processed in one batch). It's cash- # correct (open_exposure/open_count read my_pos) but INVISIBLE in the feed # table, so the header showed "2 open" while the table showed 1. Synthesize # the missing record from my_pos so the counts always match what's shown. for tok, p in mp.items(): b = bets.get(tok) if not b or b.get("status") != "open": self._synth_bet(tok, p) for tok, b in bets.items(): if b["status"] == "open" and tok not in mp: b["status"] = "closed" b["settled"] = b["settled"] or int(time.time()) # fully mirror-sold: realize the sold leg on the bet record # (cash was already credited fill-by-fill in _drain_fills) if b.get("sold_proceeds") is not None and b.get("pnl") is None: b["pnl"] = round(b["sold_proceeds"] - b["cost"] - (b.get("fee") or 0), 2) if b.get("sold_shares"): b["exit_price"] = round(b["sold_proceeds"] / b["sold_shares"], 4) bank = self.cfg["bankroll_usd"] exp = self.engine.open_exposure() cash = st.get("cash", bank) lag = st.get("lag", {}) missed = st.get("missed", []) for m in missed: # display names for the feed m["name"] = self.names.get((m.get("wallet") or "").lower(), (m.get("wallet") or "")[:10]) feed = { "mode": "live" if self.engine.ex.live else "paper", "bankroll": bank, "stake": round(self.engine.stake_usd(), 2), "stake_pct": self.cfg["bankroll_pct"], "event_cap": self.engine.risk.get("max_per_event"), "hwm": round(st.get("hwm", 0.0), 2), "cash": round(cash, 2), "deployed": round(exp, 2), "reserve": round(st.get("reserve", 0.0), 2), # banked profit, never bet # realized excludes audited ledger adjustments (they're bookkeeping # corrections, not P&L) and the feed carries the drift so the # dashboard shows a broken ledger instead of hiding one "realized": round(cash + exp + st.get("reserve", 0.0) - bank - sum(a["amount"] for a in st.get("adjustments", [])), 2), "adjustments": round(sum(a["amount"] for a in st.get("adjustments", [])), 2), "ledger_drift": round(self.ledger_drift(), 2), "open_count": len(mp), "fees_paid": round(st.get("fees_paid", 0.0), 2), "fee_rate": self.fee_rate, # avg_s / avg_slip_pct are the 24h ROLLING view (current execution) # falling back to the LIFETIME average when the window is empty — # the dashboard's Lag tile showed "—" whenever a book went a day # without copies (2026-07-13 user report). n = lifetime copies, # n24 = fills in the window. "lag": (lambda t: {"n": lag.get("n", 0), "n24": t[0], "avg_s": (round(t[1], 1) if t[1] is not None else (round(lag["sum_s"] / lag["n"], 1) if lag.get("n") else None)), "avg_slip_pct": (round(t[2], 4) if t[2] is not None else (round(lag["sum_slip_pct"] / lag["n"], 4) if lag.get("n") else None)), "window_h": 24})(self.lag_24h()), "wallets": [w.get("name", w["wallet"][:10]) for w in self.cfg.get("watch", [])], "classes": {w.get("name", w["wallet"][:10]): self.engine.wallet_class(w["wallet"]) for w in self.cfg.get("watch", [])}, "floors": {self.names.get(a, a[:10]): v for a, v in self.filt.per_wallet.items()}, # state accumulators are unrounded — round display fields at render "bets": [{**b, **{k: round(b[k], 2) for k in ("shares", "cost", "sold_shares", "sold_proceeds", "fee") if b.get(k) is not None}, **({"my_price": round(b["my_price"], 4)} if b.get("my_price") else {})} for b in sorted(bets.values(), key=lambda b: b.get("settled") or b.get("opened") or 0, reverse=True)[:100]], "missed": sorted(missed, key=lambda m: m.get("settled") or m.get("ts") or 0, reverse=True)[:60], "missed_pnl": round(sum(m["pnl"] for m in missed if m.get("pnl") is not None), 2), } # only (re)write — and so only commit — when the meaningful content changed, # not on every poll. The "updated" stamp advances only on real change, so the # scheduled runner doesn't spam a commit every 5 minutes. sig = hashlib.md5(json.dumps(feed, sort_keys=True).encode()).hexdigest() if sig == self.engine.state.get("feed_sig"): return self.engine.state["feed_sig"] = sig feed["updated"] = int(time.time()) path = os.path.join(self.here, self.feed_path) os.makedirs(os.path.dirname(path), exist_ok=True) tmp = path + ".tmp" json.dump(feed, open(tmp, "w"), indent=1) os.replace(tmp, path) def publish_feed(self): """Commit + push the feed, the state file, and the fills ledger so (a) the public dashboard reads the current book, (b) the book survives machine loss, and (c) the per-fill lag/slippage evidence is preserved — the Actions runner used to discard copybot_fills.jsonl on every run. Committing state here (not just the feed) also keeps `git pull --rebase` from wedging on a dirty tracked file now that this local poller is the sole runner. Throttled and commit-on-change. Best-effort — never crashes the run.""" st = self.engine.state now = time.time() if now - st.get("feed_pushed_at", 0) < FEED_PUSH_MIN_S: return st["feed_pushed_at"] = now try: import subprocess repo = self.here # publish only when the FEED itself changed (a bet placed/settled) — # the state file churns bookkeeping every cycle and would otherwise # commit every FEED_PUSH_MIN_S forever. unchanged = subprocess.run( ["git", "-C", repo, "diff", "--quiet", "--", self.feed_path], capture_output=True).returncode == 0 # a brand-new feed path is UNTRACKED — git diff reports no change # for it, which silently suppressed the real-money feed's first # publish forever (found 2026-07-09, /live stuck at 404) untracked = subprocess.run( ["git", "-C", repo, "ls-files", "--error-unmatch", self.feed_path], capture_output=True).returncode != 0 if unchanged and not untracked: return paths = [p for p in (self.feed_path, self.engine.state_path, self.fill_log, self.shadow_log) if os.path.exists(os.path.join(repo, p))] subprocess.run(["git", "-C", repo, "add", "-f"] + paths, capture_output=True) if subprocess.run(["git", "-C", repo, "diff", "--cached", "--quiet"], capture_output=True).returncode == 0: return # nothing changed c = subprocess.run(["git", "-C", repo, "commit", "-q", "-m", "copybot: live paper feed [skip ci]"], capture_output=True, text=True) if c.returncode != 0: return subprocess.run(["git", "-C", repo, "pull", "--rebase", "--autostash", "-q", "origin", "main"], capture_output=True) # a conflicted rebase wedges the repo (UU) and silently kills every # future publish (2026-07-08: a boot raced the BOOK RESET push — the # stale clone rebased old-book state onto the reset and sat wedged # until reboot). Abort, resync to origin, re-commit the RUNNING book # on top: the process is the book's single writer, so its memory # wins the file; remote state surgery must stop the bot first. if subprocess.run(["git", "-C", repo, "rev-parse", "-q", "--verify", "REBASE_HEAD"], capture_output=True).returncode == 0: log("⚠ publish rebase conflicted — resyncing to origin and " "re-committing the live book") subprocess.run(["git", "-C", repo, "rebase", "--abort"], capture_output=True) subprocess.run(["git", "-C", repo, "reset", "--hard", "origin/main"], capture_output=True) self.engine.persist() self.write_feed() subprocess.run(["git", "-C", repo, "add", "-f"] + paths, capture_output=True) subprocess.run(["git", "-C", repo, "commit", "-q", "-m", "copybot: live paper feed (resynced after " "conflicted rebase) [skip ci]"], capture_output=True) try: # audit 3.4: stale-writer suicide guard up = subprocess.run(["git", "-C", repo, "show", "@{u}:" + self.engine.state_path], capture_output=True, text=True) if up.returncode == 0: ub = (json.loads(up.stdout).get("boot") or {}) ours = st.get("boot") or {} if (ub.get("id") and ours.get("id") and ub["id"] != ours["id"] and ub.get("ts", 0) > ours.get("ts", 0)): log("⚠ a NEWER boot owns the book on origin — this " "stale writer is exiting instead of overwriting it") os._exit(0) except Exception: pass p = subprocess.run(["git", "-C", repo, "push", "-q", "origin", "main"], capture_output=True, text=True) if p.returncode == 0: self._pub_fails = 0 log("published live feed → dashboard") else: # closes #10: a dead GITHUB_TOKEN ends durability SILENTLY — # after 10 straight failures, say so where someone will look self._pub_fails = getattr(self, "_pub_fails", 0) + 1 log("feed push failed (will retry next change)") if self._pub_fails == 10: self.engine.alert( "⚠ feed publish failing repeatedly — book durability " "is OFF (check GITHUB_TOKEN)", discord_text="🚨 **PUBLISH FAILING** — 10 straight " "push failures; the book is not being backed up " "(GITHUB_TOKEN dead?)") except Exception as e: log(f"feed publish error: {e}") def seed(self): """Load each watched wallet's current positions so exits mirror correctly and we never backfill a position they held before we started.""" for wallet in self.cfg.get("watchlist", []): self.engine.seed_wallet(wallet) def baseline(self): """Mark every currently-visible trade as already seen, so a poll run only copies trades that happen AFTER startup (the forward equivalent of the dashboard's June-1 START — no retro-copying of history).""" n = fresh = 0 cutoff = time.time() - RECENT_TRADE_WINDOW_S for wallet in self.cfg.get("watchlist", []): for t in recent_trades(wallet): tx = t.get("transactionHash") if not tx or tx in self.engine.seen: continue # trades YOUNGER than the stale window stay unseen: the first # post-boot poll copies them like a late webhook would. Before # 2026-07-09 a crash-restart baselined away fresh qualifying # trades — two live-executor crashes ate Kruto's 21:21 handicap # and badaf's 21:53 Epic entries exactly this way. if (t.get("timestamp") or 0) >= cutoff: fresh += 1 continue self.engine.seen.add(tx) n += 1 self.engine.persist() log(f"baseline: {n} historical trades marked seen · {fresh} fresh trades " f"left copyable — only NEW trades from now") def resolve_pendings(self): """Settle state["pending_orders"] — in-play holds the executor handed off instead of cancelling (2026-07-10 registry). Each pending either ADOPTS its fill (full bookkeeping: spend, position, cash drain, ledger row, bet record) or expires at TTL into a cancel + honest miss. The exchange's balance diff is the fill arbiter, same as the executor's own uncertain path.""" st = self.engine.state pend = st.get("pending_orders") or [] if not pend or not self.engine.ex.live: return # user-ws events and the heartbeat both call this — one pass at a # time; a concurrent caller just skips (the running pass adopts) rl = getattr(self, "_resolve_lock", None) if rl is None: rl = self._resolve_lock = threading.Lock() if not rl.acquire(blocking=False): return try: with self.lock: # audit 3.1: adoption mutates cash/my_pos/spend self._resolve_pendings_locked(st) finally: rl.release() def _resolve_pendings_locked(self, st): pend = st.get("pending_orders") or [] ex = self.engine.ex now = time.time() keep = [] for p in pend: tok = p["token"] px, status, matched = p["price"], "gone", 0.0 order_view = False # did the exchange answer for THIS order? try: o = ex.client.get_order(order_id=p["order_id"]) matched = float(o.size_matched or 0) status = o.status order_view = True if matched > 0: px = float(o.price or px) except Exception: pass # gone from open view — terminal expired = now - p["ts"] > p.get("ttl_s", 600) if status in ("live", "delayed") and not expired and matched <= 0: keep.append(p) # still held — check again next tick continue if status in ("live", "delayed"): # expired: kill the remainder try: ex.client.cancel_order(order_id=p["order_id"]) except Exception: pass # THE 2026-07-12 LESSON (+$7.86 phantom cash): with several # pendings on ONE token, each balance-diff window sees the SAME # balance move, so max(matched, diff) booked one real fill many # times (a 1.56-share position "sold" 4.52 shares). The order's # own matched size is authoritative; the diff is a FALLBACK for # orders gone from the open view, and everything caps at the # order's own size. sz_cap = p.get("sz") or p.get("shares_req") or float("inf") filled = min(matched, sz_cap) if filled <= 0 and not order_view: # the exchange no longer knows the order — ONLY then is the # balance diff worth consulting (a definitive killed/0 answer # above means unfilled, full stop) try: bal1 = ex._shares_held(tok) diff = (bal1 - p["bal0"]) if p["side"] == "BUY" \ else (p["bal0"] - bal1) filled = min(max(diff, 0.0), sz_cap) except Exception: pass if filled <= 0.01: log(f"pending expired unfilled: {p['outcome']} · {p['title'][:40]}") if p["side"] == "BUY" and not p.get("is_add"): self.engine.record_miss( p["wallet"], tok, p.get("cond"), p["title"], p["outcome"], p["price"], p.get("stake", 0), f"in-play hold expired unfilled ({int(now - p['ts'])}s)") continue spent = filled * px if p["side"] == "BUY": st["spend"]["usd"] += spent mine = st["my_pos"].get(tok) if p.get("is_add") and mine: mine["shares"] += filled mine["cost"] += spent if p.get("cond"): mine.setdefault("cond", p["cond"]) else: st["my_pos"][tok] = { "shares": filled, "cost": spent, "title": p["title"], "outcome": p["outcome"], "event": p.get("event"), "wallet": p["wallet"], "cond": p.get("cond")} if p.get("cond"): self.conds[tok] = p["cond"] ex.fills.append({"side": "BUY", "token": tok, "shares": filled, "price": px}) synth = {"timestamp": p.get("their_ts"), "price": p["their_price"], "outcome": p["outcome"], "title": p["title"]} for f in self._drain_fills(): self._record_lag(p["wallet"], synth, f) self.engine.alert( f"PENDING FILLED · {p['outcome']} · {p['title'][:40]} — " f"buy {filled:.2f} @ {px:.3f} (${spent:.2f}, held " f"{int(now - p['ts'])}s)", discord_text=(f"🟢 **OPEN (in-play hold filled)** [LIVE]\n" f"{p['outcome']} · {p['title'][:60]}\n" f"buy {filled:.2f} @ {px:.3f} = **${spent:.2f}** " f"(held {int(now - p['ts'])}s)")) else: # SELL adoption: reduce the position mine = st["my_pos"].get(tok) if not mine or not mine.get("shares"): log(f"pending sell adoption: no position left for " f"{p['title'][:36]} — dropping (nothing to book)") continue filled = min(filled, mine["shares"]) # book can't sell more frac = min(1.0, filled / mine["shares"]) mine["cost"] *= (1 - frac) mine["shares"] -= filled if mine["shares"] <= 0.01: st["my_pos"].pop(tok, None) ex.fills.append({"side": "SELL", "token": tok, "shares": filled, "price": px}) self._drain_fills() self.engine.alert( f"PENDING EXIT FILLED · {p['outcome']} · {p['title'][:40]} — " f"sold {filled:.2f} @ {px:.3f}", discord_text=(f"🔴 **EXIT (in-play hold filled)** [LIVE]\n" f"{p['outcome']} · {p['title'][:60]}\n" f"sold {filled:.2f} @ {px:.3f} = " f"**${filled * px:.2f}**")) st["pending_orders"] = keep self.engine.persist() MAX_EXIT_RETRIES = 10 def retry_stuck_exits(self): """LIVE_ROLLOUT 1.6: re-attempt failed mirror-exits each heartbeat. A thin/no-bid book at copy time must not silently turn a scalp into a hold-to-resolution — retry up to MAX_EXIT_RETRIES ticks, then page Discord (⚠ EXIT STUCK) and let the position ride knowingly.""" st = self.engine.state retries = st.get("exit_retries") or [] if not retries: return with self.lock: # audit 3.1: sells + fill drains mutate the book self._retry_stuck_exits_locked(st, retries) def _retry_stuck_exits_locked(self, st, retries): keep = [] for r in retries: tok = r["token"] mine = st["my_pos"].get(tok) if not mine or mine.get("shares", 0) <= 0.01: continue # settled/sold meanwhile — done if any(po.get("token") == tok for po in st.get("pending_orders", [])): keep.append(r) # resolver owns this token right now continue shares = min(r["shares"], mine["shares"]) if getattr(self.engine.ex, "live", False): try: # closes #7: never retry a ghost chain_sh = self.engine.ex._shares_held(tok) except Exception: keep.append(r) continue if chain_sh < 0.01: log(f"retry: ghost holding (chain 0) — dropping " f"{r.get('label','?')[:40]}") st["my_pos"].pop(tok, None) continue shares = min(shares, chain_sh) price = self.engine._live_price(tok, "sell") if price is None: res = {"ok": False, "resp": "no bid side"} else: res = self.engine.ex.sell(tok, shares, price, {}) if res["ok"]: frac = min(1.0, res["filled_shares"] / mine["shares"]) \ if mine["shares"] else 1.0 mine["cost"] *= (1 - frac) mine["shares"] -= res["filled_shares"] if mine["shares"] <= 0.01: st["my_pos"].pop(tok, None) self._drain_fills() self.engine.alert( f"EXIT RECOVERED · {r['label'][:46]} — sold " f"{res['filled_shares']:.2f} @ {res['price']:.3f} " f"(attempt {r['attempts'] + 1})", discord_text=(f"🔴 **EXIT recovered** [LIVE]\n{r['label'][:60]}\n" f"sold {res['filled_shares']:.2f} @ " f"{res['price']:.3f} on retry " f"{r['attempts'] + 1}")) continue if res.get("pending"): # in-play hold — registry owns it now st.setdefault("pending_orders", []).append({ **res["pending"], "token": tok, "side": "SELL", "wallet": mine.get("wallet", ""), "title": mine.get("title", ""), "outcome": mine.get("outcome", ""), "event": mine.get("event"), "cond": mine.get("cond"), "their_price": price, "their_ts": None, "price": price, "is_add": False, "stake": shares * price, "ts": time.time(), "ttl_s": 600}) continue r["attempts"] += 1 if r["attempts"] >= self.MAX_EXIT_RETRIES: self.engine.alert( f"⚠ EXIT STUCK · {r['label'][:46]} — {r['attempts']} retries " "failed; position rides to resolution", discord_text=(f"🚨 **EXIT STUCK** [LIVE]\n{r['label'][:60]}\n" f"{r['attempts']} sell retries failed " f"(last: {str(res.get('resp'))[:60]}) — " "position will ride to resolution")) continue keep.append(r) st["exit_retries"] = keep self.engine.persist() DUST_MIN_USD = 0.10 # below this a sweep sell isn't worth the fee DUST_WALLET = "0x455e252e45Ee46d6C4cc1c8fAdD3899d68f245a1" # deposit wallet def sweep_dust(self, cycle): """Exit-dust reclaimer (2026-07-17): proportional mirror-exits leave sub-share remainders on chain after the book closes the bet (found by the platform-vs-book reconcile: 4 residues worth ~$1.90). Shortly after boot and ~daily, market-sell any LIVE-market wallet holding the book does NOT track and credit the proceeds as a documented ADJUSTMENT — recovery of already-settled bets, never new realized P&L (the ledger invariant needs the cash move in exactly one term). Never touches my_pos / pending / retry tokens; a failed sell (dust books FAK no-match, sub-$1 rejects) just waits for the next pass.""" if not getattr(self.engine.ex, "live", False): return if cycle != 5 and cycle % 1440 != 0: return pos = sm.get_json("/positions", {"user": self.DUST_WALLET, "sizeThreshold": 0}) or [] st = self.engine.state skip = (set(st["my_pos"]) | {p.get("token") for p in st.get("pending_orders", [])} | {r.get("token") for r in st.get("exit_retries", [])}) recovered, sold, tried = 0.0, [], 0 for p in pos: tok, val = p.get("asset"), p.get("currentValue") or 0 cur, shares = p.get("curPrice") or 0, p.get("size") or 0 if not tok or tok in skip or val < self.DUST_MIN_USD or shares <= 0: continue if tried >= 10: break # audit 3.2: the data-api LAGS the chain — a residual the chain # already zeroed (platform auto-redeem) keeps reappearing here. # Gate every action on exchange-view truth; act only on confirmed # shares; credit proportionally so a stale row books ~$0. try: chain_sh = self.engine.ex._shares_held(tok) except Exception: continue # no chain truth -> no action if chain_sh < max(0.01, shares * 0.5): continue # data-api ghost — skip forever if chain_sh < shares: val *= chain_sh / shares shares = chain_sh if p.get("redeemable") or cur >= 0.99: # resolved WINNER residual (the 2026-07-17 $1.90 case): the # tracked-position redeemer never sees untracked tokens, so # collect here. Refunds/winners pay at the vector price. cond = p.get("conditionId") if not (self.redeemer and cond) or market_neg_risk(cond): continue tried += 1 ok, info = self.redeemer.try_redeem(cond) if ok: with self.lock: st["cash"] += val recovered += val sold.append(f"redeemed {shares:.2f}sh ≈ ${val:.2f} " f"{str(p.get('title'))[:36]}") log(f"DUST REDEEM +${val:.2f} · {sold[-1]}") else: log(f"dust redeem failed ({str(info)[:40]}) — retry next pass") continue if not 0.01 <= cur <= 0.99: continue tried += 1 quote = clob_price(tok, "sell") if not quote: continue with self.lock: before = st["cash"] r = self.engine.ex.sell(tok, shares, quote, {"title": p.get("title") or "dust"}) self._drain_fills() delta = st["cash"] - before if r.get("ok") and delta > 0: recovered += delta sold.append(f"{shares:.2f}sh @ {r['price']:.3f} " f"{str(p.get('title'))[:36]}") log(f"DUST SWEEP +${delta:.2f} · {sold[-1]}") else: log(f"dust sweep skip ({str(r.get('resp'))[:50]}) · " f"{str(p.get('title'))[:40]}") if recovered > 0: st.setdefault("adjustments", []).append({ "ts": int(time.time()), "amount": round(recovered, 6), "note": f"dust sweep: {len(sold)} residual holdings sold " "(recovery of closed bets, not P&L)"}) self.engine.persist() self.engine.alert( f"dust sweep recovered ${recovered:.2f} ({len(sold)} holdings)", discord_text=(f"🧹 **DUST SWEEP** [LIVE] +${recovered:.2f}\n" + "\n".join(sold[:8]))) _chain_bal = (0.0, None) # (checked_at, usdc) — cached; poll ≤1/min def chain_cash_gap(self): """LIVE only (LIVE_ROLLOUT 1.4): book cash minus the exchange's actual USDC balance — the real-money version of ledger_drift. None in paper mode or on a failed fetch (never treat silence as a number).""" if not self.engine.ex.live: return None now = time.time() ts, bal = self._chain_bal if now - ts > 60: try: # unified SDK: the deposit wallet's pUSD as the exchange counts # it (LIVE_ROLLOUT 1.4 anchor — was the emptied legacy proxy # via py-clob-client, which alarmed CASH≠CHAIN +24.73) r = self.engine.ex.client.get_balance_allowance( asset_type="COLLATERAL") bal = r.balance / 1e6 self._chain_bal = (now, bal) except Exception: return None return None if bal is None else self.engine.state.get("cash", 0) - bal def lag_24h(self): """(count, avg_lag_s, avg_slip_pct) over fills in the last 24h — the CURRENT-execution view. The lifetime average buried push mode's ~3s under the retired 60s poll era (a 3s bot read as 48s).""" now = time.time() rec = [r for r in self.engine.state.get("lag_recent", []) if r[0] >= now - 86400] if not rec: return 0, None, None n = len(rec) return n, sum(r[1] for r in rec) / n, sum(r[2] for r in rec) / n def summary(self, cycle): bank = self.cfg["bankroll_usd"] stake = self.engine.stake_usd() # dynamic: pct of current equity exp = self.engine.open_exposure() cash = self.engine.state.get("cash", bank) reserve = self.engine.state.get("reserve", 0.0) adj = sum(a["amount"] for a in self.engine.state.get("adjustments", [])) realized = cash + exp + reserve - bank - adj # see _drain_fills / settle_resolved n = len(self.engine.state["my_pos"]) lag = self.engine.state.get("lag", {}) lagstr = "" n24, avg_s, avg_slip = self.lag_24h() if n24: lagstr = (f" · {lag.get('n', 0)} copies · 24h lag {avg_s:.0f}s " f"slip {avg_slip:+.1%} ({n24})") elif lag.get("n"): lagstr = (f" · {lag['n']} copies avg lag {lag['sum_s']/lag['n']:.0f}s " f"slip {lag['sum_slip_pct']/lag['n']:+.1%}") bankstr = f" · banked ${reserve:,.0f}" if reserve else "" drift = self.ledger_drift() driftstr = f" · ⚠ LEDGER DRIFT ${drift:+.2f}" if abs(drift) > 0.05 else "" gap = self.chain_cash_gap() if gap is not None and abs(gap) > 1.0: # ≥$1: beyond fee/rounding float driftstr += f" · ⚠ CASH≠CHAIN ${gap:+.2f}" rtds = getattr(self, "rtds", None) if rtds is not None: st = rtds.status() driftstr += (f" · rtds {st}" if st.startswith("up") else f" · ⚠ rtds {st}") uw = getattr(self, "userws", None) if uw is not None: driftstr += (" · userws up" if uw.state == "up" else f" · ⚠ userws {uw.state}") if _RPC_URL and _RPC_FAILS >= 5: driftstr += f" · ⚠ rpc down ({_RPC_FAILS} fails)" log(f"[{cycle}] open {n} · deployed ${exp:,.0f} · free ${cash:,.0f}/${bank:,.0f}" f"{bankstr} · realized ${realized:+,.2f}{lagstr}{driftstr}" + (f" · CAN'T OPEN (free < ${stake:,.0f} stake — bets missed)" if cash < stake else "")) def _fetch_since_cursor(self, wallet): """H3 fix (2026-07-13): a burst trader can push unprocessed rows past a single newest-100 fetch (gkmg: 30+ clips in 15 min). Paginate the activity feed back to the last processed timestamp per wallet, capped at 5 pages — the cursor makes every fetch lossless regardless of burst size; the cap bounds a cold start. Cursor moves only forward and only after a fetch, so a failed page just re-reads next time.""" cur = self.engine.state.setdefault("trade_cursor", {}) since = cur.get(wallet.lower(), 0) if not since: # COLD START: no cursor yet — behave like the pre-cursor fetch # (fresh trades only). Walking history here poured 500 rows/wallet # of weeks-old trades through the funnel at the 2026-07-13 02:10 # boot and spammed ~230 phantom "too slow" missed rows per book. since = time.time() - RECENT_TRADE_WINDOW_S rows, offset = [], 0 for _ in range(5): page = recent_trades(wallet, limit=100, offset=offset) if not page: break rows.extend(page) oldest = min((t.get("timestamp") or 0) for t in page) if oldest <= since or len(page) < 100: break offset += 100 if rows: newest = max((t.get("timestamp") or 0) for t in rows) cur[wallet.lower()] = max(since, newest) return [t for t in rows if (t.get("timestamp") or 0) > since - 600] def on_wallet_activity(self, wallet, ignore_stale=False, seed_trade=None, hint_txs=None): """A watched wallet just transacted — pull its latest trades and route any new, recent one through the filter and (if it passes) the engine. `seed_trade` (from the RTDS push) is merged into the fetched set so a lagging data-api indexer can't delay the copy — deduped by tx. `hint_txs` (from the Alchemy push) are receipt hashes: any hash the fetch didn't return is decoded straight from the chain (fills_from_tx), so an RTDS-silent market still copies at push speed.""" name = self.names.get(wallet.lower(), wallet[:10] + "…") trades = self._fetch_since_cursor(wallet) if seed_trade and seed_trade.get("transactionHash") and seed_trade.get("asset"): if seed_trade["transactionHash"] not in {t.get("transactionHash") for t in trades}: trades = trades + [seed_trade] have = {t.get("transactionHash") for t in trades} for tx in (hint_txs or []): if tx in have or tx in self.engine.seen or tx in self.skipped: continue have.add(tx) for f in fills_from_tx(tx, wallet): log(f"chainseed: {name} {f['side']} {f['size']:g} @ " f"{f['price']:.3f} · {str(f['title'])[:40]}") trades = trades + [f] # ONE conviction bet often arrives as several fills — a sweep through # the book or rapid clip entries (gkmg 2026-07-09: a $612 MOUZ entry = # 3×$204 same-second rows, every clip sub-floor while the backtest's # position-level view takes the bet; he sold +60% five minutes later). # Merge same-token BUY rows within 120s into the bet the sharp made; # component txs are all marked processed. SELLs stay per-fill — the # proportional mirror handles those correctly clip by clip. merged, by_tok = [], {} for t in sorted(trades, key=lambda x: x.get("timestamp", 0)): if (t.get("side") != "BUY" or not t.get("asset") or not t.get("transactionHash") or t.get("transactionHash") in self.engine.seen or t.get("transactionHash") in self.skipped): merged.append(t) continue prev = by_tok.get(t["asset"]) if prev is not None and (t.get("timestamp", 0) or 0) - (prev.get("timestamp", 0) or 0) <= 120: sz_p, sz_t = prev.get("size") or 0, t.get("size") or 0 if sz_p + sz_t: prev["price"] = (((prev.get("price") or 0) * sz_p + (t.get("price") or 0) * sz_t) / (sz_p + sz_t)) prev["size"] = sz_p + sz_t prev["usdcSize"] = (prev.get("usdcSize") or 0) + (t.get("usdcSize") or 0) prev["timestamp"] = max(prev.get("timestamp", 0) or 0, t.get("timestamp", 0) or 0) prev.setdefault("_extra_txs", []).append(t["transactionHash"]) else: by_tok[t["asset"]] = t merged.append(t) # oldest-first so the engine's position math stays causal for t in sorted(merged, key=lambda x: x.get("timestamp", 0)): tx = t.get("transactionHash") extras = t.pop("_extra_txs", []) if not tx or tx in self.engine.seen or tx in self.skipped: self.skipped.update(extras) continue # filter FIRST (before the stale gate) so we know whether a trade we # were too slow on WOULD have qualified — a qualifying miss is worth # recording; a below-floor/out-of-band one is a deliberate skip. follow, reason = self.filt.check(wallet, t) stale = not ignore_stale and time.time() - t.get("timestamp", 0) > RECENT_TRADE_WINDOW_S if follow and stale: # a bet we'd have copied but didn't catch in time — webhook missed # it, the bot was down, or the market resolved faster than we poll # (5-min crypto, in-play). Log it as MISSED so it's visible on the # dashboard instead of silently dropped; settle_resolved values it # hypothetically like every other miss. record_miss dedups by token # so reconcile_entries can't double-count the same position. self.skipped.update([tx, *extras]) if t.get("side") == "BUY" and t.get("asset"): late_m = (time.time() - (t.get("timestamp") or 0)) / 60.0 with self.lock: self.engine.record_miss( wallet, t["asset"], t.get("conditionId"), t.get("title") or "", t.get("outcome") or "", t.get("price") or 0, self.engine.stake_usd(wallet), f"too slow to follow ({late_m:.0f}m late)") self.engine.persist() log(f"MISS {name}: {t.get('side')} {t.get('outcome','?')} " f"@ {t.get('price',0):.3f} — too slow ({late_m:.0f}m late)") continue if not follow: self.skipped.update([tx, *extras]) log(f"skip {name}: {t.get('side')} {t.get('outcome','?')} " f"@ {t.get('price',0):.3f} — {reason}") continue log(f"FOLLOW {name}: {t.get('side')} {t.get('outcome','?')} " f"@ {t.get('price',0):.3f} (${t.get('usdcSize',0):,.0f})") with self.lock: self.engine.handle_trade(wallet, t) # sizes, gates, places (paper/live) self.engine.seen.update(extras) # component fills of the merged bet tok = t.get("asset") if tok in self.engine.state["my_pos"] and tok not in self.conds: self.conds[tok] = t.get("conditionId") # remember for settling for f in self._drain_fills(): if f["token"] == tok: # the fill from this copy self._record_lag(wallet, t, f) else: # a leftover fill from a handler that died mid-trade — # its cash was just debited above; record it or it # becomes an invisible orphan (the 2026-07-08 drift) self._record_untracked_buy(f) self.check_book() def reconcile_exits(self): """Exits the signal made while we weren't listening. RECENT_TRADE_WINDOW_S (10 min) skips stale trades, so a SELL during downtime/restart never mirrors — the 2026-07-06 audit found the bot holding McCormick/Sakamoto a day after the whale had sold it for +$4.2k. For every open copy, ask the data-api (market-filtered, so no pagination cap) whether the signal still holds the token: gone + market still trading -> mirror-exit ALL of ours now at the live price; gone + market closed -> leave it for settle_resolved (selling into a dead book would book winners as scratches). Runs at boot and every backstop poll.""" with self.lock: mp = self.engine.state["my_pos"] checks = [] for token in list(mp): b = self.engine.state.get("bets", {}).get(token) cond = self.conds.get(token) if b and b.get("wallet") and cond: checks.append((token, cond, b["wallet"])) for token, cond, wallet in checks: if token not in mp: continue # settled/sold earlier this pass # FAIL-SAFE: get_json returns None on failure and [] on a real # empty — silence must NEVER read as "they exited" (the dry-run # of this very fix tried to liquidate the whole book when the # API blipped). An exit needs three affirmative facts: # 1. their open positions on this market: fetched AND empty, # 2. their closed positions: fetched AND contain our token # (they demonstrably had it and closed it), # 3. the market itself: fetched AND still trading. ps = sm.get_json("/positions", {"user": wallet, "market": cond, "limit": 10, "sizeThreshold": 0}) if ps is None: continue # API failure — retry next pass held = sum(p.get("size", 0) or 0 for p in ps if str(p.get("asset")) == str(token)) book = self.engine.state["their_pos"].setdefault(wallet, {}) if held > 0: book[token] = held # refresh sell-fraction basis continue cps = sm.get_json("/closed-positions", {"user": wallet, "market": cond, "limit": 10}) if cps is None or not any(str(p.get("asset")) == str(token) for p in cps): continue # can't corroborate the exit m = _market(cond) if not m: continue # market state unknown — don't act book[token] = 0.0 if m.get("closed"): continue # resolved -> settle path pays truth pos = mp.get(token) if not pos: continue # closes #7: the BOOK's share count can drift above the CHAIN's # (venue rounding on partial exits — the BetBoom 0.12sh ghost # looped "not enough balance" forever). Chain is the arbiter: # clamp the exit to what actually exists; a ghost (<0.01sh on # chain) is dropped outright, loudly, instead of retried. if getattr(self.engine.ex, "live", False): try: chain_sh = self.engine.ex._shares_held(token) except Exception: continue # no chain truth -> no action this pass if chain_sh < 0.01: log(f"reconcile: ghost position (book " f"{pos['shares']:.2f}sh, chain 0) — dropping " f"{pos.get('title','?')[:40]}") del mp[token] continue pos["shares"] = min(pos["shares"], chain_sh) name = self.names.get(wallet.lower(), wallet[:10]) log(f"reconcile: {name} exited {pos.get('title','?')[:42]} while we " f"weren't listening — mirror-exiting {pos['shares']:.1f}sh now") # their_prev<=0 -> frac 1.0: sell everything we hold self.engine._handle_their_sell( token, 0, 0, f"{pos.get('outcome','?')} · {pos.get('title','?')[:42]}") for f in self._drain_fills(): # book the sell's cash + sold-leg self._record_untracked_buy(f) # (returned buys = leftovers) # LIVE_ROLLOUT 1.6 — a FAK sell on a thin book can fill 0 and # the position silently rides to resolution. This pass re-fires # every backstop poll; count the attempts and raise the alarm # once it looks stuck (still held after N tries). if token in mp: att = self.engine.state.setdefault("exit_attempts", {}) att[token] = att.get(token, 0) + 1 if att[token] == 10: self.engine.alert( f"⚠ EXIT STUCK: {pos.get('title','?')[:42]} — " f"{att[token]} mirror-exit attempts unfilled; " f"check the book / exit manually in the UI.") else: self.engine.state.get("exit_attempts", {}).pop(token, None) self.engine.persist() def reconcile_entries(self): """Entries the signal made while we weren't listening — the mirror of reconcile_exits. Every boot baselines history (no retro-copying), so a trade during downtime vanishes: not copied, not even MISSED. The backtest reads positions (state) while the bot reads trades (events), so downtime opens showed up only in the backtest — that asymmetry is how this hole was found (2026-07-07: 9 ArbTrader positions opened into a dead listener during the Fly trial-expiry crash-loop). For each followed wallet: any CURRENT position that (a) isn't seeded, held, in bet history, or already recorded, and (b) clears the follow filter, becomes a missed bet — reason 'bot offline'. Never copied: the entry is stale by definition; record the truth and settle it hypothetically.""" with self.lock: st = self.engine.state bets = st.get("bets", {}) missed_toks = {m["token"] for m in st.get("missed", [])} # tokens adjudicated 'accumulated via sub-floor clips, not a miss' # — persisted so each backstop pass doesn't re-fetch and re-log no_miss = set(st.setdefault("no_miss_toks", [])) missed_toks |= no_miss for w in self.cfg.get("watchlist", []): ps = sm.get_json("/positions", {"user": w, "limit": 500, "sizeThreshold": 0}) if ps is None: continue # API failure — retry next pass seeded = set(st.get("seed_tokens", {}).get(w, [])) name = self.names.get(w.lower(), w[:10]) for p in ps: tok = p.get("asset") if (not tok or tok in seeded or tok in st["my_pos"] or tok in bets or tok in missed_toks): continue iv = p.get("initialValue") or 0 if iv < self.engine.risk.get("min_order_usd", 5.0): continue # literal dust — not a signal trade = {"side": "BUY", "usdcSize": iv, "price": p.get("avgPrice", 0) or 0, "outcome": p.get("outcome"), "title": p.get("title")} ok, _ = self.filt.check(w, trade) if not ok: continue # the live path gates PER TRADE; a position can accumulate # past the floor via sub-floor clips the bot rightly skipped # one by one (2026-07-08: 0xbadaf319's paired YES+NO arb # clips of $6–37 built $40+ positions that read as # 'missed · bot offline' while the bot was UP and skipping # every fill). Only a genuine miss if some SINGLE buy on # this market would have passed the same filter. Market- # filtered /trades = full history for the market (no # pagination window, unlike /activity), usdcSize # synthesized where that endpoint omits it. API failure # falls through to the old behavior (record the miss). cond = p.get("conditionId") trs = sm.get_json("/trades", {"user": w, "market": cond, "limit": 100}) if cond else None if trs is not None: for t_ in trs: if not t_.get("usdcSize"): t_["usdcSize"] = (t_.get("size") or 0) * (t_.get("price") or 0) # cluster fills ≤120s apart — the same merge rule the # live path applies: a fill-split conviction entry is # ONE bet, and must count as a real miss buys = sorted((t_ for t_ in trs if t_.get("side") == "BUY" and str(t_.get("asset")) == str(tok)), key=lambda x: x.get("timestamp", 0) or 0) best, cur, last_ts = None, None, None for t_ in buys: ts_ = t_.get("timestamp", 0) or 0 if cur is not None and ts_ - last_ts <= 120: cur = dict(t_, usdcSize=(cur.get("usdcSize") or 0) + (t_.get("usdcSize") or 0)) else: cur = dict(t_) last_ts = ts_ if best is None or (cur.get("usdcSize") or 0) > (best.get("usdcSize") or 0): best = cur if best is None or not self.filt.check(w, best)[0]: log(f"reconcile: {name} accumulated " f"{(p.get('title') or '?')[:38]} via sub-floor " f"clips (${iv:,.0f} total) — not copyable under " f"per-trade rules, not a miss") no_miss.add(tok) st["no_miss_toks"] = sorted(no_miss) missed_toks.add(tok) continue want = self.engine.stake_usd(w, iv) self.engine.record_miss( w, tok, p.get("conditionId"), p.get("title") or "", p.get("outcome") or "", trade["price"], want, "not copied in the detection window (reconciled)") missed_toks.add(tok) log(f"reconcile: {name} entered {(p.get('title') or '?')[:42]} " f"while we weren't listening — recorded as missed") # exit-aware missed settling: a missed bet whose sharp FULLY # EXITED pre-resolution settles at THEIR exit — the mirror # counterfactual — instead of riding to resolution (Kruto's 3c # Hive entry 2026-07-09: he banked 16x selling at 48c; the # hold-to-resolution valuation would have shown the map result # instead). Same three affirmative facts as reconcile_exits — # silence never reads as an exit. Entry+exit taker fees charged, # like every mirrored sell. for m in st.get("missed", []): # (B) rows already SETTLED at resolution (won/lost/refund): # fast in-play markets can resolve before this 5-min pass sees # the sharp's sell, and rows from before 2026-07-09 predate # exit-aware settling. One idempotent re-check per row: a # redeem prints EXACTLY the payout, so an exit print >2c away # means the sharp sold pre-resolution — revalue the # counterfactual at their exit (mirror semantics, both fees). if (m.get("status") in ("won", "lost", "refund") and not m.get("exit_checked") and m.get("cond") and m.get("wallet") and m.get("token")): cps = sm.get_json("/closed-positions", {"user": m["wallet"], "market": m["cond"], "limit": 10}) if cps is None: continue # API blip — retry next pass m["exit_checked"] = True # examined once, forever row = next((p for p in cps if str(p.get("asset")) == str(m["token"])), None) if row is not None: tb = row.get("totalBought") or 0 xp = ((row.get("avgPrice") or 0) + ((row.get("realizedPnl") or 0) / tb if tb else 0)) payout = {"won": 1.0, "lost": 0.0, "refund": 0.5}[m["status"]] if abs(xp - payout) > 0.02: p = m.get("price") or 0.5 sh = m["stake"] / max(p, 0.001) fee_in = taker_fee(sh, p, self.fee_rate) fee_out = taker_fee(sh, xp, self.fee_rate) old_pnl = m.get("pnl") m.update(status="sold", exit_price=round(xp, 4), pnl=round(sh * xp - m["stake"] - fee_in - fee_out, 2)) log(f"missed-reval: {(m.get('title') or '?')[:38]} — sharp " f"sold @ {xp:.3f} pre-resolution; counterfactual " f"${old_pnl} -> ${m['pnl']:+.2f} (sold)") continue if (m.get("status") != "open" or not m.get("cond") or not m.get("wallet") or not m.get("token")): continue ps = sm.get_json("/positions", {"user": m["wallet"], "market": m["cond"], "limit": 10, "sizeThreshold": 0}) if ps is None: continue if any(str(p.get("asset")) == str(m["token"]) and (p.get("size") or 0) > 0 for p in ps): continue # still held — resolution path waits cps = sm.get_json("/closed-positions", {"user": m["wallet"], "market": m["cond"], "limit": 10}) if cps is None: continue row = next((p for p in cps if str(p.get("asset")) == str(m["token"])), None) if row is None: continue mk = _market(m["cond"]) if not mk or mk.get("closed"): continue # resolved -> chain-truth settle path tb = row.get("totalBought") or 0 xp = (row.get("avgPrice") or 0) + ((row.get("realizedPnl") or 0) / tb if tb else 0) p = m.get("price") or 0.5 sh = m["stake"] / max(p, 0.001) fee_in = taker_fee(sh, p, self.fee_rate) fee_out = taker_fee(sh, xp, self.fee_rate) m.update(status="sold", exit_price=round(xp, 4), pnl=round(sh * xp - m["stake"] - fee_in - fee_out, 2), settled=int(time.time())) log(f"missed-settle: {m.get('name') or m['wallet'][:8]} exited " f"{(m.get('title') or '?')[:38]} @ {xp:.3f} — " f"hypothetical ${m['pnl']:+.2f} (sold)") self.engine.persist() def settle_resolved(self): # audit 4: this ran on EVERY webhook POST ahead of the copy itself — # dozens of redundant CLOB fetches during clip bursts. 20s throttle; # the 60s heartbeat still guarantees timely settles. if time.time() - getattr(self, "_settle_last", 0) < 20: return self._settle_last = time.time() """Free capital like the dashboard: when an open position's market has resolved, settle it at the winner price (1/0), recycle the cash, and tally realized P&L. This is the resolution path the engine's sell-only mirror lacks — without it the $1k never recycles for held-to-resolution bets.""" with self.lock: mp = self.engine.state["my_pos"] for token in list(mp): cond = self.conds.get(token) if not cond: continue wp = resolution_price(token, cond, mp[token].get("outcome")) if wp is None: continue # not resolved yet pos = mp[token] # gap 2 — LIVE: redeem winning shares on-chain so the freed USDC is # actually back in the wallet (paper just recycles a number). Losers # are worth $0, no redeem; 50/50 refunds redeem at $0.50/share. If the # redeem fails, keep the position and retry next pass. if self.redeemer and wp > 0: neg = market_neg_risk(cond) if neg and cond not in self.negrisk_warned: self.negrisk_warned.add(cond) self.engine.alert(f"⚠ {pos.get('title','?')[:42]} is a NEG-RISK " f"market — auto-redeem unsupported; redeem " f"manually in the Polymarket UI.") if not neg: ok, info = self.redeemer.try_redeem(cond) if not ok: log(f" ⚠ redeem failed ({info}) — keeping position, will retry") continue log(f" ↳ redeemed on-chain: {info}") proceeds = pos["shares"] * wp # redeem is fee-free on-chain b = self.engine.state.get("bets", {}).get(token) fee_in = (b or {}).get("fee") or 0 # entry taker fee, already off cash # a TRIMmed bet realized part of itself pre-resolution: count the # sold leg and charge the ORIGINAL cost (pos["cost"] was reduced # proportionally at each trim), so per-bet pnl = whole-bet truth sold = (b or {}).get("sold_proceeds") or 0 base_cost = b["cost"] if b else pos["cost"] pnl = proceeds + sold - base_cost - fee_in self.engine.state["cash"] += proceeds # recycle freed capital status = "won" if wp > 0.5 else "lost" if wp < 0.5 else "refund" if b: b.update(status=status, exit_price=wp, pnl=round(pnl, 2), settled=int(time.time())) del mp[token] self.conds.pop(token, None) tag = {"won": "WON ✅", "lost": "LOST ❌", "refund": "REFUND ↩ (50/50)"}[status] label = f"{pos.get('outcome','?')} · {pos.get('title','?')[:42]}" self.engine.alert( f"SETTLE {label} — {tag} {pos['shares']:.0f}sh -> " f"${proceeds:.2f} (P&L ${pnl:+.2f})", discord_text=(f"🏁 **SETTLE** {tag}\n{label}\n" f"${pos['cost']:.2f} cost -> ${proceeds:.2f} " f"= **${pnl:+.2f}**")) # settle MISSED bets hypothetically: what the skipped stake would have # returned (entry fee included; redeem free) — the live counterpart of # the backtest's Missed P&L, "the cost of a small bankroll". for m in self.engine.state.get("missed", []): if m.get("status") != "open" or not m.get("cond"): continue wp = resolution_price(m["token"], m["cond"], m.get("outcome")) if wp is None: continue p = m.get("price") or 0.5 fee = taker_fee(m["stake"] / p, p, self.fee_rate) pnl = (m["stake"] / p) * wp - m["stake"] - fee m.update(status=("won" if wp > 0.5 else "lost" if wp < 0.5 else "refund"), pnl=round(pnl, 2), settled=int(time.time())) self.engine.persist() # ── Alchemy webhook plumbing ──────────────────────────────────────────────── def verify(raw, sig, signing_key): if not signing_key: return True # verification off if unconfigured digest = hmac.new(signing_key.encode(), raw, hashlib.sha256).hexdigest() return hmac.compare_digest(digest, sig or "") def addresses_in_payload(payload, watched): """-> {watched wallet: sorted tx hashes}. The hashes ride along so a trade the data-api can't return yet is decoded from its receipt instead of waiting on the indexer (fills_from_tx).""" out = {} for a in payload.get("event", {}).get("activity", []): for k in ("fromAddress", "toAddress"): v = (a.get(k) or "").lower() if v in watched: txs = out.setdefault(v, set()) if a.get("hash"): txs.add(a["hash"]) return {w: sorted(txs) for w, txs in out.items()} def make_handler(bot, signing_key): watched = {w.lower() for w in bot.cfg.get("watchlist", [])} class Handler(BaseHTTPRequestHandler): def _send(self, code, body="ok"): self.send_response(code) self.send_header("Content-Type", "text/plain") self.end_headers() self.wfile.write(body.encode()) def do_GET(self): self._send(200, "alive" if self.path == "/health" else "copybot") def do_POST(self): raw = self.rfile.read(int(self.headers.get("Content-Length", 0))) if not verify(raw, self.headers.get("x-alchemy-signature"), signing_key): log("⚠ bad signature — rejected") return self._send(401, "bad signature") self._send(200) # ack fast; Alchemy retries non-2xx try: bot.settle_resolved() # recycle any newly-resolved positions payload = json.loads(raw or b"{}") for w, txs in addresses_in_payload(payload, watched).items(): bot.on_wallet_activity(w, hint_txs=txs) bot.write_feed() # refresh + publish the dashboard feed bot.publish_feed() except Exception as e: log(f"handler error: {e}") def log_message(self, *a): pass return Handler # ── config / cli ──────────────────────────────────────────────────────────── def _pctl80(vals): s = sorted(v for v in vals if v and v > 0) if not s: return None k = (len(s) - 1) * 0.8 f = int(k) return s[f] if f + 1 >= len(s) else s[f] + (s[f + 1] - s[f]) * (k - f) def derive_floor(wallet): """Cacheless p80 conviction floor: the 80th percentile of the wallet's recent position stakes from the data-api (last 500 closed + current open). The Railway worker has no cache.duckdb, so this is how an auto floor gets computed at boot; it refreshes on every restart.""" sizes = [] for p in (sm.get_json("/closed-positions", {"user": wallet, "limit": 500, "sortBy": "TIMESTAMP", "sortDirection": "DESC"}) or []) + (sm.get_json("/positions", {"user": wallet, "limit": 500, "sizeThreshold": 0}) or []): v = p.get("initialValue") or 0 if v > 0: sizes.append(v) return _pctl80(sizes) def normalize_follow_config(cfg): """Expand the compact follow format into the legacy structures the filter and engine read. One entry per wallet: "wallets": [{"wallet": "0x…", "name": "…", "class": "volume"|"whale", "floor": 123.0 (optional)}] -> watchlist / watch / follow.wallet_class / follow.per_wallet_min_usd. Old-format configs (no "wallets" key, e.g. config.live.json) pass through untouched. Floors: explicit "floor" wins; whales need none (follow-all); otherwise derived from the data-api at boot (see derive_floor).""" ws = cfg.get("wallets") if not ws: return cfg cfg["watchlist"] = [w["wallet"] for w in ws] cfg["watch"] = [{"wallet": w["wallet"], "name": w.get("name", w["wallet"][:10])} for w in ws] f = cfg.setdefault("follow", {}) f["wallet_class"] = {w["wallet"].lower(): w.get("class", "volume") for w in ws} floors = {} for w in ws: addr = w["wallet"].lower() name = w.get("name", addr[:10]) if w.get("class", "volume") == "whale": continue # follow-all: floors ignored if w.get("floor") is not None: floors[addr] = float(w["floor"]) else: fl = derive_floor(w["wallet"]) if fl is not None: floors[addr] = round(fl, 2) log(f"floor[{name}] auto p80 = ${fl:,.0f}") else: log(f"⚠ floor[{name}] underivable — falls back to the global " f"min_their_usd ${f.get('min_their_usd', 0):,.0f}") f["per_wallet_min_usd"] = floors return cfg def load_cfg(path): if not os.path.exists(path): sys.exit(f"No config at {path}.") cfg = {**DEFAULT_CONFIG, **load_json(path, {})} cfg["risk"] = {**DEFAULT_CONFIG["risk"], **cfg.get("risk", {})} cfg["live"] = {**DEFAULT_CONFIG["live"], **cfg.get("live", {})} # env overrides for the credentials (LIVE_ROLLOUT 1.2): the Fly worker # clones the repo, and config.live.json is gitignored — secrets reach a # cloud box as env vars only. Env wins over file so a deployed worker # can't accidentally trade an old key from a stray config. for env, key in (("LIVE_PRIVATE_KEY", "private_key"), ("LIVE_FUNDER_ADDRESS", "funder_address"), ("LIVE_SIGNATURE_TYPE", "signature_type")): v = (os.environ.get(env) or "").strip() # secrets often arrive with a if v: # trailing newline from paste cfg["live"][key] = int(v) if key == "signature_type" else v cfg["follow"] = {**FOLLOW_DEFAULT, **cfg.get("follow", {})} normalize_follow_config(cfg) # accept either "watchlist" (addresses) or "watch" ([{wallet,name}]); fill the gap if not cfg.get("watchlist") and cfg.get("watch"): cfg["watchlist"] = [w["wallet"] for w in cfg["watch"]] if not cfg.get("watch") and cfg.get("watchlist"): cfg["watch"] = [{"wallet": w, "name": w[:10]} for w in cfg["watchlist"]] return cfg def conviction_floors_from_sharps(path): """Per-wallet conviction floor = that wallet's avg bet size, so only their above-average ('conviction') bets get copied — the edge per the research.""" rows = load_json(path, []) return {r["wallet"].lower(): float(r["avg_bet"]) for r in rows if r.get("avg_bet")} def main(): ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--config", default="config.json") ap.add_argument("--state", default="copybot_state.json") ap.add_argument("--live", action="store_true", help="enable live trading (also needs mode:live in config)") ap.add_argument("--conviction-from-sharps", metavar="PATH", help="set each wallet's conviction floor to its avg_bet from " "this sharps json (e.g. live/watch_sharps.json)") ap.add_argument("--test-wallet", metavar="0x...", help="dry-run: route this wallet's latest trade through the " "pipeline once (paper), print the decision, then exit") ap.add_argument("--poll", type=int, metavar="SECONDS", help="run forward by polling every SECONDS (instead of waiting " "for Alchemy webhooks) — lets a paper run go without " "deploying the webhook. Use < 600 so no trade is missed.") ap.add_argument("--poll-once", action="store_true", help="run ONE poll pass (settle, copy new trades, write feed) " "then exit — for a scheduled runner (GitHub Actions cron). " "State persists across runs via --state.") args = ap.parse_args() cfg = load_cfg(args.config) if args.conviction_from_sharps: floors = conviction_floors_from_sharps(args.conviction_from_sharps) cfg["follow"]["per_wallet_min_usd"] = { **cfg["follow"].get("per_wallet_min_usd", {}), **floors} log(f"loaded {len(floors)} per-wallet conviction floors from " f"{args.conviction_from_sharps}") want_live = args.live and cfg.get("mode") == "live" if args.live and cfg.get("mode") != "live": sys.exit('--live given but config "mode" is not "live". Refusing to trade.') state = load_json(args.state, new_state()) redeemer = None if want_live: # LIVE_ROLLOUT 1.5 — the geo-gate is FATAL for real money: an # unauthenticated dummy order separates geo-blocked (403 "restricted # in your region") from allowed (auth/validation error). Paper mode # only reads, so it merely warns (host/start.sh); live must refuse. try: req = urllib.request.Request(CLOB_API + "/order", method="POST", data=b"{}", headers={"User-Agent": "Mozilla/5.0", "Content-Type": "application/json"}) urllib.request.urlopen(req, timeout=15, context=SSL_CTX) geo_body = "" except urllib.error.HTTPError as e: geo_body = e.read().decode(errors="replace")[:200] if e.code == 403 and "restricted" in geo_body.lower(): sys.exit("GEO-BLOCKED: this box cannot place Polymarket orders " f"({geo_body[:100]}). Run from the Fly worker (Stockholm) " "or an unrestricted country — see LIVE_ROLLOUT.md 0.1.") except Exception as e: sys.exit(f"geo-gate probe failed ({e}) — refusing to arm live " "without a TRADABLE verdict (LIVE_ROLLOUT.md 0.1)") log("geo-gate: TRADABLE — order endpoint reachable") confirm_live(cfg) executor = LedgerLiveExecutor(cfg) # FOK marketable orders + fill recording if cfg.get("live", {}).get("auto_redeem", True): try: from redeem import Redeemer # 2026-07-19 rewrite (closes #4): shares the executor's # SecureClient — gasless redeemPositions FROM the deposit # wallet with pUSD collateral (the web3/EOA/USDC.e version # was a silent no-op on this stack) redeemer = Redeemer(executor.client) log("on-chain auto-redeem ENABLED (pUSD, deposit-wallet gasless)") except Exception as e: log(f"⚠ auto-redeem unavailable ({e}) — resolved winners must be " f"redeemed manually in the UI.") else: executor = LedgerPaperExecutor() # tracks cash flows for realized-P&L reporting engine = CopyTrader(cfg, state, executor, args.state) # per-trade Discord pings retired for PAPER (daily digest only) — but the # LIVE book pings every real placement/exit/settle: eyes on every dollar # during the supervised test. Webhook via env secret only (a webhook URL # committed to this public repo leaked once already — 2026-07-09). engine.webhook = ((os.environ.get("DISCORD_WEBHOOK") or "").strip() if want_live else "") filt = FollowFilter(cfg) bot = Copybot(cfg, engine, filt, redeemer=redeemer) # single-writer invariant (audit 3.4): every boot stamps the state; a # stale process that survived `machine stop` (gotcha 15c) sees a newer # boot on origin at publish time and yields instead of overwriting it. engine.state["boot"] = {"id": uuid.uuid4().hex[:12], "ts": int(time.time())} # on-chain resolution RPC (payout vectors for operator-resolved markets): # env ALCHEMY_RPC_URL wins (the Fly worker has no config.json), else the # local config's alchemy_key. Without either, tier-2 settlement is off and # operator-resolved/refunded positions stay open (pre-2026-07-06 behavior). global _RPC_URL _RPC_URL = (os.environ.get("ALCHEMY_RPC_URL") or (f"https://polygon-mainnet.g.alchemy.com/v2/{cfg['alchemy_key']}" if cfg.get("alchemy_key") else None)) mode = "LIVE — REAL MONEY" if executor.live else "PAPER (no orders placed)" log(f"copybot · mode: {mode}") log(f"on-chain settle fallback: {'ON' if _RPC_URL else 'OFF — set ALCHEMY_RPC_URL'}") log(f"watching {len(cfg.get('watchlist', []))} wallets · {filt.describe()}") guard = cfg.get("price_guard_abs", cfg.get("price_guard_pct", 0.05)) def _cap(v): return "off" if v >= 1e5 else f"${v:,.0f}" log(f"bankroll ${cfg['bankroll_usd']:.2f} @ {cfg['bankroll_pct']:.1%}/entry · " f"guard +{guard:.2f} abs · " f"caps: {_cap(cfg['risk']['max_trade_usd'])}/trade, " f"{_cap(cfg['risk']['daily_spend_cap_usd'])}/day, " f"{_cap(cfg['risk']['max_total_exposure_usd'])} exposure") bot.seed() # boot invariant pass: rebuild any missing bet/conds records and — only # here, where no trade is in flight — heal a never-debited orphan's cash # if the drift matches it exactly (HANDOFF proper fix, Option A) bot.check_book(heal_cash=True) # T0 detection — started AFTER seed()/check_book (closes #5: a trade # arriving before their_pos is seeded made an ADD look like a fresh # OPEN and broke sell-fraction basis). Backstops cover the boot gap. # The RTDS trade stream (~1s, wallet-attributed). Paper # runs it by default (the 24h shadow run, 2026-07-10); the REAL-MONEY # role stays on the proven Alchemy+poll stack until the shadow validates, # then: flyctl secrets set RTDS_DETECT=1 -a wwf-copybot-live. rtds_default = "0" if os.environ.get("COPYBOT_ROLE", "paper") == "live" else "1" if (os.environ.get("RTDS_DETECT") or rtds_default) == "1": bot.rtds = RtdsListener(bot) bot.rtds.start() else: log("rtds: T0 listener disabled for this role (RTDS_DETECT=1 enables)") if want_live: # own-fill push (2026-07-13): in-play holds adopt the moment they # match instead of on the next 60s resolver tick bot.userws = UserFillsListener(bot) bot.userws.start() # one-shot pipeline test: no server, just push a wallet's latest trade through if args.test_wallet: log(f"--test-wallet: routing {args.test_wallet[:10]}…'s latest activity " f"through filter + engine (paper)") bot.on_wallet_activity(args.test_wallet, ignore_stale=True) log("test done.") return # single-pass mode for a scheduled runner (GitHub Actions cron). State persists # across runs via --state, so the $1k book carries forward run to run. On the # very first run there's no baseline → mark history seen and copy nothing, so we # only copy trades that happen after the test starts. if args.poll_once: if not bot.engine.state.get("baselined"): bot.baseline() bot.engine.state["baselined"] = True bot.write_feed() bot.engine.persist() log("first run — baselined history; published online feed, copied nothing") return bot.settle_resolved() bot.reconcile_exits() bot.reconcile_entries() for w in cfg.get("watchlist", []): bot.on_wallet_activity(w) bot.summary(0) bot.write_feed() bot.engine.persist() log("poll-once complete") return # forward poll mode: run the same filter+engine pipeline by polling, so a paper # run works today without the deployed Alchemy webhook. (Production push uses the # webhook below; behaviour through the filter+engine is identical either way.) if args.poll: bot.baseline() bot.reconcile_exits() # catch exits made while we were down bot.reconcile_entries() # ...and entries: record them as missed log(f"poll mode · every {args.poll}s · Ctrl-C to stop") bot.write_feed() # publish an initial "online" snapshot bot.publish_feed() cycle = 0 try: while True: bot.resolve_pendings() # adopt/expire in-play held orders bot.retry_stuck_exits() # LIVE_ROLLOUT 1.6 bot.settle_resolved() # recycle capital at resolution if cycle % 5 == 0: bot.reconcile_exits() bot.reconcile_entries() for w in cfg.get("watchlist", []): try: bot.on_wallet_activity(w) except Exception as e: # parity with the push handler's log(f"poll error {w[:10]}…: {e}") # guard — never die cycle += 1 bot.summary(cycle) bot.write_feed() # refresh the dashboard feed each cycle bot.publish_feed() # push to GitHub (throttled, on change) time.sleep(args.poll) except KeyboardInterrupt: log("stopped.") return signing_key = (os.environ.get("ALCHEMY_SIGNING_KEY") or cfg.get("alchemy_signing_key", "")) port = int(os.environ.get("PORT", 8080)) bot.baseline() bot.reconcile_exits() # catch exits made while we were down bot.reconcile_entries() # ...and entries: record them as missed # webhook mode is event-driven, but the book must not depend on the next # push arriving: a heartbeat thread settles resolved positions, refreshes # the feed, and logs the summary every 60s — plus a FULL backstop poll of # every wallet each 5th tick, so a dropped/misconfigured push costs at most # ~5 minutes of lag instead of a silent miss. def _heartbeat(): cycle = 0 while True: time.sleep(60) cycle += 1 try: bot.resolve_pendings() # adopt/expire in-play held orders bot.retry_stuck_exits() # LIVE_ROLLOUT 1.6 bot.settle_resolved() bot.sweep_dust(cycle) # reclaim untracked exit residue (live) if cycle % 5 == 0: bot.reconcile_exits() bot.reconcile_entries() for w in cfg.get("watchlist", []): bot.on_wallet_activity(w) bot.summary(cycle) bot.write_feed() bot.publish_feed() except Exception as e: log(f"heartbeat error: {e}") threading.Thread(target=_heartbeat, daemon=True).start() log(f"push mode · listening on :{port} · POST /alchemy · " f"signature-verify {'ON' if signing_key else 'OFF'} · " f"heartbeat 60s · backstop poll 300s") bot.write_feed() bot.publish_feed() ThreadingHTTPServer(("0.0.0.0", port), make_handler(bot, signing_key)).serve_forever() if __name__ == "__main__": main()