#!/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 re
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, _SeenTx,
)
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 _gamma_markets(qs):
    """One gamma /markets fetch (separate so tests can stub the HTTP edge)."""
    req = urllib.request.Request(
        f"https://gamma-api.polymarket.com/markets?{qs}",
        headers={"User-Agent": "Mozilla/5.0"})
    with urllib.request.urlopen(req, timeout=12, context=SSL_CTX) as r:
        return json.loads(r.read().decode()) or []


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).
    Gamma's DEFAULT listing hides closed markets (#18 follow-up: the resolved
    Odyssey market returned [] until &closed=true) — and closed markets are
    exactly what repair_market_meta needs, so fall back to the closed view."""
    with _TOK_LOCK:
        if token in _TOK_MKT:
            return _TOK_MKT[token]
    for extra in ("", "&closed=true"):
        try:
            ms = _gamma_markets(f"clob_token_ids={token}{extra}")
            if not ms:
                continue
            m = ms[0]
            toks = json.loads(m["clobTokenIds"])
            got = (m.get("question") or "",
                   json.loads(m["outcomes"])[toks.index(token)],
                   m.get("conditionId"))
        except Exception:
            continue
        with _TOK_LOCK:
            _TOK_MKT[token] = got
        return got
    return None


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)
    # #20/#21 dark flags (pre-registered; flip only at the Friday read):
    "entry_mode": "taker",     # "maker" = rest a GTC bid at their price
    "exit_mode": "mirror",     # "hold"  = never mirror their sells
    "maker_ttl_s": 60,         # registry-side cancel for resting maker bids
    # #25 dark flag (pre-registered): what level the conviction floor checks.
    # "trade"    = each individual fill must clear the floor (current)
    # "position" = a rolling net-USD aggregate per (wallet, token) counts —
    #   clip-built conviction passes once the aggregate crosses the floor
    #   (the 1kto1m class: 36 sub-floor-clip positions in one parity week).
    #   In-memory aggregate: rebuilds from flow after a restart (declared).
    "conviction_scope": "trade",
}

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

# FAK re-quote retry, per market niche. Waits are the measured crater-refill
# times (research/params/requote_timing.json, 775k crater prints, 2026-07-20):
# crypto books refill <4s 94% of the time, esports 83% by 10s, sports needs
# ~25s (70% at 10s -> 76% at 25s); geo/politics/other tails run minutes, so
# 25s is the cap — beyond that the price guard is the protection, not the
# retry. Patterns are the validated research/tape.py classifier, first match
# wins (esports before sports: "LoL: A vs B" must not fall through to
# sports on " vs "). Config: fak_retry_niche_s overrides per key;
# fak_retry_s stays the fallback for unmatched niches and 0 disables the
# whole feature.
FAK_RETRY_NICHE_DEFAULT = {"crypto": 4.0, "esports": 10.0, "sports": 25.0,
                           "geo": 25.0, "politics": 25.0, "other": 25.0}
NICHE_PATTERNS = [
    ("esports", ["lol:", "dota", "cs2", "csgo", "valorant", "esports",
                 "bilibili", "map ", "game 1", "game 2", "game 3"]),
    ("tennis", ["tennis", "atp", "wta", "wimbledon", "set winner"]),
    ("sports", [" vs. ", " vs ", " @ ", "mlb", "nba", "nhl", "ufc",
                "world cup", "f1", "grand prix", "fifa"]),
    ("crypto", ["bitcoin", "btc", "ethereum", "solana", "xrp", "doge",
                "price of", "up or down"]),
    ("politics", ["election", "president", "senate", "governor", "mayor",
                  "nominee", "impeach", "tariff", "fed ", "rate cut"]),
    ("geo", ["iran", "israel", "russia", "ukraine", "china", "taiwan",
             "ceasefire", "strike", "nato"]),
]


def market_niche(title):
    t = (title or "").lower()
    for label, pats in NICHE_PATTERNS:
        if any(p in t for p in pats):
            return label
    return "other"


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 _iso_end(s):
    """CLOB end_date_iso -> epoch (None on parse trouble). Display metadata
    only — in-play markets usually resolve well before the scheduled end."""
    try:
        import calendar
        return calendar.timegm(time.strptime(
            s.replace("+00:00", "Z")[:20], "%Y-%m-%dT%H:%M:%SZ"))
    except Exception:
        return None


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
            msg = str(getattr(r, "message", r))
            # tick-size conformance retry (2026-07-24, #24 audit): `d` infers
            # precision from the INPUT price's decimals, but a sharp's VWAP
            # (e.g. 0.485) on a 0.01-tick market yields an off-tick bound the
            # venue rejects. The rejection message names the real tick —
            # requantize the bound to it (BUY floors, SELL ceils: both only
            # tighten) and retry ONCE.
            m_t = re.search(r"tick size (0\.0*1)", msg)
            if m_t:
                tick = float(m_t.group(1))
                dp = max(0, -int(math.floor(math.log10(tick) + 1e-9)))
                try:
                    if side == "BUY":
                        mp = max(tick, math.floor(
                            min(max(price * (1 + self._slip), price), 0.99)
                            / tick) * tick)
                        r = self.client.place_market_order(
                            token_id=token_id, side="BUY",
                            amount=max(round(sz * price, 2), 1.0),
                            max_price=round(mp, dp), order_type=self._otype)
                    else:
                        mp = max(tick, math.ceil(
                            min(max(price * (1 - self._slip), tick), price)
                            / tick) * tick)
                        r = self.client.place_market_order(
                            token_id=token_id, side="SELL", shares=sz,
                            min_price=round(mp, dp), order_type=self._otype)
                except Exception as e:
                    return {"ok": False, "filled_shares": 0.0, "price": price,
                            "resp": f"exception (tick retry): {e}",
                            "paper": False}
            if not getattr(r, "ok", False):
                return {"ok": False, "filled_shares": 0.0, "price": price,
                        "resp": f"{getattr(r, 'code', '?')}: "
                                f"{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 maker_buy(self, token_id, shares, price, meta):
        """#20 dark path (live): rest a GTC limit BUY at the sharp's own
        price; the pending registry adopts the fill or cancels at its TTL.
        An instant cross books like a fill (taker at their price or better —
        the same semantics the backtest counted); any instant PARTIAL cancels
        its remainder so no order outlives this call untracked (the 07-10
        invariant). Venue-side GTD needs >=3min, so TTL is registry-side."""
        import math
        sz = math.floor(shares * 100) / 100.0
        try:
            bal0 = self._shares_held(token_id)
        except Exception as e:
            return {"ok": False, "filled_shares": 0.0, "price": price,
                    "resp": f"pre-check failed: {e}", "paper": False}
        d = 2
        for cand in (2, 3, 4):
            if abs(round(price, cand) - price) < 1e-9:
                d = cand
                break
        else:
            d = 4
        try:
            r = self.client.place_limit_order(
                token_id=token_id, side="BUY", size=sz, price=round(price, d))
        except Exception as e:      # a timed-out post may still be resting
            filled, px = self._settle_uncertain(token_id, "BUY", bal0, price)
            if filled > 0:
                self.fills.append({"side": "BUY", "token": token_id,
                                   "shares": filled, "price": px,
                                   "maker": True})
                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):
            return {"ok": False, "filled_shares": 0.0, "price": price,
                    "resp": f"{getattr(r, 'code', '?')}: "
                            f"{getattr(r, 'message', r)}", "paper": False}
        making = float(r.making_amount or 0)
        taking = float(r.taking_amount or 0)
        filled = taking
        if filled > 0:
            px = making / filled if filled else price
            if filled < sz - 0.01:          # partial cross: kill the rest
                try:
                    self.client.cancel_order(order_id=r.order_id)
                except Exception:
                    pass
            self.fills.append({"side": "BUY", "token": token_id,
                               "shares": filled, "price": px, "maker": True})
            return {"ok": True, "filled_shares": filled, "price": px,
                    "resp": {"order_id": r.order_id, "status": r.status},
                    "paper": False}
        return {"ok": False, "filled_shares": 0.0, "price": price,
                "pending": {"order_id": r.order_id, "bal0": bal0, "sz": sz,
                            "mode": "maker"},
                "resp": {"order_id": r.order_id, "status": r.status,
                         "note": "maker bid resting"}, "paper": False}

    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.entry_mode = f.get("entry_mode", "taker")
        self.exit_mode = f.get("exit_mode", "mirror")
        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}
        # #25: conviction_scope "position" — rolling net-USD per (wallet,
        # token), fed by EVERY in-set trade (including sub-floor clips the
        # engine never sees). In-memory; rebuilds from flow after restart.
        self.scope = f.get("conviction_scope", "trade")
        self.agg = {}                  # (wallet, token) -> [net_usd, last_ts]

    def _agg_update(self, wallet, t):
        a = t.get("asset")
        if not a:
            return 0.0
        usd = t.get("usdcSize") or t.get("size", 0) * t.get("price", 0) or 0
        key = (wallet.lower(), str(a))
        st = self.agg.get(key) or [0.0, 0.0]
        st[0] = max(0.0, st[0] + (usd if t.get("side") == "BUY" else -usd))
        st[1] = time.time()
        self.agg[key] = st
        if len(self.agg) > 8000:       # prune stalest tenth, bound memory
            for k in sorted(self.agg, key=lambda k: self.agg[k][1])[:800]:
                del self.agg[k]
        return st[0]

    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":
            if self.scope == "position":
                self._agg_update(wallet, t)         # sells shrink the aggregate
            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 self.scope == "position":
            net = self._agg_update(wallet, t)
            if usd < fl and net < fl:
                return False, (f"clip ${usd:,.0f} · aggregate ${net:,.0f} "
                               f"< conviction floor ${fl:,.0f}")
        elif 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 ""
        md = (f" · entry_mode {self.entry_mode} · exit_mode {self.exit_mode}"
              if (self.entry_mode, self.exit_mode) != ("taker", "mirror")
              else "")
        md += (f" · conviction_scope {self.scope}"
               if self.scope != "trade" 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}]{md}")


# ── 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 {}
        # #20 paper leg: every platform print passes here — fill any resting
        # paper maker bid the print crosses. Cheap no-op unless bids exist.
        if self.bot.engine.state.get("paper_maker"):
            try:
                self.bot.paper_maker_probe(p)
            except Exception as e:
                log(f"⚠ paper_maker_probe error: {e}")
        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}
        # #18: some markets stream with empty title/outcome/conditionId (the
        # 07-20 Odyssey copy) — a cond-less book entry is UNSETTLEABLE. Enrich
        # from the token id (gamma, cached) before seeding; on a miss the copy
        # still goes through and repair_market_meta backfills later.
        if seed["asset"] and not (seed["conditionId"] and seed["title"]):
            meta = _token_market(seed["asset"])
            if meta:
                seed["title"] = seed["title"] or meta[0]
                seed["outcome"] = seed["outcome"] or meta[1]
                seed["conditionId"] = seed["conditionId"] or meta[2]
        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 = _SeenTx()   # evaluated-and-skipped txs, recency-capped (closes #9)
        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")
        # #21: append-only ignored-exit ledger (derived per book)
        self.ignored_log = self.fill_log.replace("fills", "ignored_exits")
        self.reject_log = self.fill_log.replace("fills", "rejects")
        # persisted across restarts via the engine's state file
        self.conds = engine.state.setdefault("conds", {})   # token_id -> conditionId (open positions)
        # #18 empty-cond repair: in-memory backoff/alarm bookkeeping (a reboot
        # restarts the 1h clock — fine, the repair itself is what matters)
        self._meta_fail = {}           # token -> (first_fail_ts, last_attempt_ts)
        self._meta_warned = set()      # tokens already alarmed as unsettleable
        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))
        # FAK no-match re-quote retry wait (s); 0 disables (miss recorded at
        # the first rejection, pre-2026-07-20 behavior). Per-niche waits from
        # the measured crater-refill times (see FAK_RETRY_NICHE_DEFAULT);
        # fak_retry_s is the fallback for niches without a measurement.
        self.fak_retry_s = float(cfg.get("fak_retry_s", 10))
        self.fak_retry_niche = {
            k: float(v) for k, v in {**FAK_RETRY_NICHE_DEFAULT,
                                     **(cfg.get("fak_retry_niche_s") or {})}.items()}
        # append-only history spools (2026-07-21 full-history feed): rows
        # trimmed out of state (bets >30d via _spool_old_bets, missed past
        # the 200 cap via engine.on_miss_spool) land here and keep feeding
        # the *_full.json companion feed + lifetime wallet_pnl rollup.
        self.bets_archive = self.fill_log.replace("fills", "bets_archive")
        self.missed_archive = self.fill_log.replace("fills", "missed_archive")

    def _fak_wait(self, title):
        return self.fak_retry_niche.get(market_niche(title), self.fak_retry_s)

    def spool_missed(self, rows):
        """engine.on_miss_spool hook: archive rows trimmed off state.missed
        (display name stamped now — the archive never sees write_feed)."""
        with open(os.path.join(self.here, self.missed_archive), "a") as fh:
            for m in rows:
                m["name"] = self.names.get((m.get("wallet") or "").lower(),
                                           (m.get("wallet") or "")[:10])
                fh.write(json.dumps(m) + "\n")
        log(f"spooled {len(rows)} missed rows → {self.missed_archive}")

    def _load_archive(self, path):
        rows = []
        try:
            with open(os.path.join(self.here, path)) as fh:
                for ln in fh:
                    try:
                        rows.append(json.loads(ln))
                    except Exception:
                        pass
        except FileNotFoundError:
            pass
        return rows

    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
                # bot-health "last trade placed" (2026-07-21 dashboard ask)
                src = (self.engine.state["my_pos"].get(f["token"])
                       or self.engine.state.get("bets", {}).get(f["token"]) or {})
                self.engine.state["last_fill"] = {
                    "ts": int(time.time()), "side": f["side"],
                    "price": round(f["price"], 4),
                    "shares": round(f["shares"], 2),
                    "title": (src.get("title") or "")[:60],
                    "outcome": src.get("outcome")}
                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)
                    + st.get("spooled_pnl", 0.0))   # closes #9: spooled history
        # 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 write_reject(self, rec):
        """#24 v1 instrumentation: one JSONL row per follow-filter rejection
        of an IN-SET wallet's BUY (floor / entry-band). Append-only; splits
        the reconciliation's UNSEEN bucket into screen-vs-universe honestly
        (the 1kto1m clip class was invisible without this). Not a behavior
        change — the signal was already being skipped, now it leaves a row."""
        rec["mode"] = "live" if self.engine.ex.live else "paper"
        try:
            with open(os.path.join(self.here, self.reject_log), "a") as fh:
                fh.write(json.dumps(rec) + "\n")
        except Exception:
            pass

    def write_ignored_exit(self, rec):
        """#21 hold-through instrumentation: one JSONL row per ignored
        mirror-exit (the forward counterfactual's raw data) + a trimmed tail
        in state for the feed."""
        rec["mode"] = "live" if self.engine.ex.live else "paper"
        try:
            with open(os.path.join(self.here, self.ignored_log), "a") as fh:
                fh.write(json.dumps(rec) + "\n")
        except Exception:
            pass
        tail = self.engine.state.setdefault("ignored_exits", [])
        tail.append(rec)
        del tail[:-200]

    # ── #20 paper-maker registry (paper books only; live uses the pending
    #    registry with the same TTL semantics) ─────────────────────────────
    def paper_maker_probe(self, p):
        """RTDS shows every platform print — fill any resting paper maker
        bid the print crosses (price <= bid, fill AT our bid)."""
        tok = str(p.get("asset") or "")
        try:
            px = float(p.get("price") or 0)
        except (TypeError, ValueError):
            return
        if not tok or px <= 0:
            return
        with self.lock:
            pend = self.engine.state.get("paper_maker") or []
            if not pend:
                return
            keep = []
            for e in pend:
                if str(e["token"]) != tok or px > e["price"] + 1e-9:
                    keep.append(e)
                    continue
                self._book_paper_maker_fill(e)
            self.engine.state["paper_maker"] = keep

    def _book_paper_maker_fill(self, e):     # under self.lock
        st = self.engine.state
        filled, px = e["sz"], e["price"]
        spent = filled * px
        st["spend"]["usd"] += spent
        mine = st["my_pos"].get(e["token"])
        if e.get("is_add") and mine:
            mine["shares"] += filled
            mine["cost"] += spent
        else:
            st["my_pos"][e["token"]] = {
                "shares": filled, "cost": spent, "title": e.get("title"),
                "outcome": e.get("outcome"), "event": e.get("event"),
                "wallet": e.get("wallet"), "cond": e.get("cond")}
        self.engine.ex.fills.append({"side": "BUY", "token": e["token"],
                                     "shares": filled, "price": px,
                                     "maker": True})
        for f in self._drain_fills():
            synth = {"timestamp": e.get("their_ts"),
                     "price": e.get("their_price"),
                     "outcome": e.get("outcome"), "title": e.get("title")}
            self._record_lag(e.get("wallet", ""), synth, f)
        ms = st.setdefault("maker_stats", {"fills": 0, "misses": 0})
        ms["fills"] += 1
        held = time.time() - e["ts"]
        self.engine.alert(
            f"MAKER FILLED · {e.get('outcome')} · {str(e.get('title'))[:40]} — "
            f"buy {filled:.2f} @ {px:.3f} (${spent:.2f}, rested {held:.0f}s)",
            discord_text=(f"🟢 **OPEN (maker bid filled)** [PAPER]\n"
                          f"{e.get('outcome')} · {str(e.get('title'))[:60]}\n"
                          f"buy {filled:.2f} @ {px:.3f} = **${spent:.2f}** "
                          f"(rested {held:.0f}s)"))
        self.engine.persist()

    def paper_maker_sweep(self):
        """Expire resting paper bids at TTL into honest misses."""
        st = self.engine.state
        if not st.get("paper_maker"):
            return
        now = time.time()
        with self.lock:
            keep = []
            for e in st.get("paper_maker") or []:
                if now - e["ts"] <= e.get("ttl_s", 60):
                    keep.append(e)
                    continue
                ms = st.setdefault("maker_stats", {"fills": 0, "misses": 0})
                ms["misses"] += 1
                if not e.get("is_add"):
                    self.engine.record_miss(
                        e.get("wallet", ""), e["token"], e.get("cond"),
                        e.get("title"), e.get("outcome"), e["price"],
                        e.get("stake", 0),
                        f"maker bid unfilled ({int(now - e['ts'])}s)")
                log(f"MAKER expired unfilled: {e.get('outcome')} · "
                    f"{str(e.get('title'))[:40]}")
            st["paper_maker"] = keep

    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"]),
            **({"maker": True} if fill.get("maker") else {}),
        }
        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 fak_requote_retry(self, ctx):
        """One-shot re-quote retry for an OPEN whose FAK died unmatched
        (2026-07-20: 13 of 48h's misses were 'no orders found to match' — the
        copy arrives in the crater the sharp just swept, before makers
        requote; the resolved ones were net-positive would-be P&L). Sleep
        OUTSIDE the bot lock (a burst is exactly when these fire), then
        re-run the whole gated buy path — fresh quote, price guard, depth
        gate — exactly once (retry=True can't re-schedule). Mirrors the
        webhook call site's fill drain so a retry fill books its lag + bet
        rows instead of surfacing later as an untracked orphan; a second
        rejection records the miss inside _handle_their_buy, tagged
        'twice (re-quote retry)'. Installed on engine.on_fak_reject by
        main() when cfg fak_retry_s > 0. The wait is PER NICHE (_fak_wait):
        measured crater-refill times, crypto 4s / esports 10s / sports+slow
        books 25s (research requote_timing, 2026-07-20)."""
        wait = self._fak_wait(ctx["title"])

        def run():
            time.sleep(wait)
            tok = ctx["token"]
            try:
                with self.lock:
                    self.engine._handle_their_buy(
                        ctx["wallet"], tok, 0.0, ctx["their_price"],
                        ctx["label"], ctx["title"], ctx["outcome"],
                        event=ctx["event"], cond=ctx["cond"],
                        their_ts=ctx["their_ts"], retry=True)
                    self.engine.persist()
                    if tok in self.engine.state["my_pos"] \
                            and tok not in self.conds and ctx["cond"]:
                        self.conds[tok] = ctx["cond"]
                    synth = {"timestamp": ctx["their_ts"],
                             "price": ctx["their_price"],
                             "outcome": ctx["outcome"], "title": ctx["title"]}
                    for f in self._drain_fills():
                        if f["token"] == tok:
                            self._record_lag(ctx["wallet"], synth, f)
                        else:
                            self._record_untracked_buy(f)
                    self.check_book()
            except Exception as e:
                log(f"re-quote retry error ({ctx['label'][:36]}): {e}")
        log(f"FAK no-match — re-quote retry in {wait:.0f}s "
            f"({market_niche(ctx['title'])}): {ctx['label'][:48]}")
        threading.Thread(target=run, daemon=True).start()

    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", [])
        # display names: current follow set first, then names STORED on bet
        # records (a retired wallet like LSB1 keeps its name instead of
        # splitting into an address-prefix twin card — found 2026-07-21 when
        # per-wallet missed totals landed)
        arch_bets = self._load_archive(self.bets_archive)
        arch_missed = self._load_archive(self.missed_archive)
        all_bets = arch_bets + list(bets.values())
        addr_names = {(b.get("wallet") or "").lower(): b["name"]
                      for b in all_bets
                      if b.get("wallet") and b.get("name")
                      and not str(b["name"]).startswith("0x")}
        for m in missed:
            w = (m.get("wallet") or "").lower()
            m["name"] = self.names.get(w) or addr_names.get(w) or w[:10]
        # expected-resolution ETA (2026-07-23, display metadata only): one
        # cached market fetch per open bet lifetime, <=8 per publish pass;
        # "end_ts" in the record marks fetched-even-if-unknown (no refetch
        # loops on metadata-less markets)
        fetched = 0
        for tok, b in bets.items():
            if b.get("status") != "open" or "end_ts" in b or fetched >= 8:
                continue
            cond = (mp.get(tok) or {}).get("cond")
            if not cond:
                b["end_ts"] = None
                continue
            m = _market(cond) or {}
            b["end_ts"] = _iso_end(m.get("end_date_iso") or "")
            fetched += 1
        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],
                               # window empty -> we show the LIFETIME mean, so
                               # SAY so; the tile used to label it "24h" and
                               # quietly report a number dragged by the retired
                               # poll era (2026-07-27 user report)
                               "basis": ("24h" if t[1] is not None
                                         else "lifetime" if lag.get("n")
                                         else "none"),
                               "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),
            "last_fill": st.get("last_fill"),
        }
        # LIFETIME per-wallet rollup by display NAME (wallets change address —
        # AIcAIc did — and the archives carry rows the state has trimmed).
        # 2026-07-21: the dashboard cards were silently summing the rolling
        # window; the -5.62 it showed for AIcAIc was really -15.84 lifetime.
        wp = {}

        def _w(nm):
            return wp.setdefault(nm, {"realized": 0.0, "n": 0, "wins": 0,
                                      "open_cost": 0.0, "open_n": 0,
                                      "miss_n": 0, "miss_pnl": 0.0,
                                      "last_ts": 0})
        for b in all_bets:
            w = _w(b.get("name") or "?")
            if b.get("pnl") is not None:
                w["realized"] += b["pnl"]
                w["n"] += 1
                w["wins"] += b.get("status") == "won"
            elif b.get("status") == "open":
                w["open_cost"] += b.get("cost") or 0
                w["open_n"] += 1
            w["last_ts"] = max(w["last_ts"],
                               int(b.get("settled") or b.get("opened") or 0))
        for m in arch_missed + missed:      # would-be P&L of every skip ever
            w = _w(m.get("name") or "?")
            w["miss_n"] += 1
            if m.get("pnl") is not None:
                w["miss_pnl"] += m["pnl"]
        for w in wp.values():
            w["realized"] = round(w["realized"], 2)
            w["open_cost"] = round(w["open_cost"], 2)
            w["miss_pnl"] = round(w["miss_pnl"], 2)
        feed["wallet_pnl"] = wp
        # 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)
        # the FULL companion feed (dashboard "view all" + per-wallet tabs):
        # every bet and every miss ever — state + spooled archives. Written
        # only when the main feed changed (same sig gate).
        full = {"mode": feed["mode"], "updated": feed["updated"],
                "wallet_pnl": wp,
                "bets": sorted(all_bets,
                               key=lambda b: b.get("settled") or b.get("opened") or 0,
                               reverse=True),
                "missed": sorted(arch_missed + missed,
                                 key=lambda m: m.get("settled") or m.get("ts") or 0,
                                 reverse=True)}
        fpath = os.path.join(self.here,
                             self.feed_path.replace(".json", "_full.json"))
        json.dump(full, open(fpath + ".tmp", "w"), indent=1)
        os.replace(fpath + ".tmp", fpath)

    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.feed_path.replace(".json", "_full.json"),
                                 self.engine.state_path, self.fill_log,
                                 self.ignored_log, self.reject_log,
                                 self.shadow_log, self.bets_archive,
                                 self.missed_archive)
                     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:
                is_maker = p.get("mode") == "maker"
                why = "maker bid unfilled" if is_maker \
                    else "in-play hold expired unfilled"
                log(f"pending expired unfilled: {p['outcome']} · "
                    f"{p['title'][:40]}" + (" (maker)" if is_maker else ""))
                if is_maker:
                    ms = st.setdefault("maker_stats", {"fills": 0, "misses": 0})
                    ms["misses"] += 1
                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"{why} ({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()

    SPOOL_AFTER_S = 30 * 86400   # settled bets older than this leave the state

    def _spool_old_bets(self, cycle):
        """Closes #9: the git-committed state grew forever (every settled bet
        stays in bets{} so realized/drift keep counting it). Once a day, move
        bets settled >30d ago to an append-only side ledger and fold their
        pnl into state["spooled_pnl"] — ledger_drift() adds it back, so the
        invariant cash = bank + adj + Σpnl + flows is unchanged. The feed's
        resolved table and per-wallet stats become a 30-day window (the
        archive keeps the full history)."""
        if cycle % 1440 != 20:
            return
        st = self.engine.state
        cut = time.time() - self.SPOOL_AFTER_S
        arch = self.bets_archive
        with self.lock:
            old = {k: b for k, b in st.get("bets", {}).items()
                   if b.get("pnl") is not None and (b.get("settled") or 0)
                   and b["settled"] < cut}
            if not old:
                return
            with open(os.path.join(self.here, arch), "a") as fh:
                for k, b in old.items():
                    fh.write(json.dumps({"key": k, **b}) + "\n")
                    st["spooled_pnl"] = st.get("spooled_pnl", 0.0) + (b["pnl"] or 0)
                    del st["bets"][k]
            self.engine.persist()
            log(f"spooled {len(old)} settled bets (>30d) → {arch}")

    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)
        # MEDIAN, not mean (2026-07-27): one 269s early-rollout outlier made a
        # 3s bot read as 25s. The tile is a health signal, not an accounting
        # total, so the typical fill is the honest summary.
        lags = sorted(r[1] for r in rec)
        med = lags[n // 2] if n % 2 else (lags[n // 2 - 1] + lags[n // 2]) / 2
        return n, med, 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)"
        if len(self.skipped) > 20000:      # closes #9: was unbounded (MBs/month)
            for k in sorted(self.skipped, key=self.skipped.get)[:-15000]:
                del self.skipped[k]
        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)
            # #24 v1: an in-set wallet's BUY that the filter rejects leaves a
            # row (floor/band skips were invisible — screen gap wearing a
            # universe-gap costume in the reconciliation)
            if (not follow and reason and t.get("side") == "BUY"
                    and "not in follow set" not in reason):
                self.write_reject({
                    "ts": round(time.time(), 1), "wallet": wallet,
                    "token": str(t.get("asset") or ""),
                    "cond": t.get("conditionId"),
                    "usd": round(t.get("usdcSize") or
                                 (t.get("size", 0) * t.get("price", 0)), 2),
                    "price": t.get("price"), "reason": reason,
                    "title": (t.get("title") or "")[:80]})
            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")
                # falsy-overwrite, not key-presence (#18): a seed that stored
                # cond="" must not block the indexer's richer row from fixing it
                if tok in self.engine.state["my_pos"] and not self.conds.get(tok):
                    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])
                if (self.engine.cfg.get("follow") or {}).get(
                        "exit_mode", "mirror") == "hold":
                    # #21: they exited while we weren't listening — log the
                    # ignored exit ONCE (the flag stops backstop-poll spam;
                    # we keep holding, so the trigger condition never clears)
                    if not pos.get("hold_thru_logged"):
                        pos["hold_thru_logged"] = int(time.time())
                        log(f"reconcile: {name} exited "
                            f"{pos.get('title','?')[:42]} — HOLD-THROUGH "
                            f"(exit_mode=hold)")
                        self.engine._handle_their_sell(
                            token, 0, 0, f"{pos.get('outcome','?')} · "
                                         f"{pos.get('title','?')[:42]}")
                    continue
                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()

    META_RETRY_S = 300             # per-token backoff between failed lookups
    META_WARN_S = 3600             # empty-cond position older than this alarms

    def repair_market_meta(self):
        """#18: a copy seeded from a metadata-less RTDS row books with
        title/outcome/cond == "" and settle_resolved's `if not cond` skip makes
        it silently UNSETTLEABLE — the 07-20 Odyssey bet sat open while the
        venue auto-redeemed it on-chain, alarming CASH≠CHAIN -$1.75 until
        manual state surgery. Each settle pass: re-resolve empty conds from
        the token id (gamma, cached), backfill the position/bet/missed
        records, and alarm ONCE per token still unsettleable after an hour.
        Caller holds self.lock (settle_resolved)."""
        st = self.engine.state
        now = time.time()
        empties = [(t, p) for t, p in st["my_pos"].items() if not self.conds.get(t)]
        for m in st.get("missed", []):     # cond-less misses skip hypothetical
            if m.get("status") == "open" and not m.get("cond") and m.get("token"):
                empties.append((m["token"], m))
        for tok, rec in empties:
            first, last = self._meta_fail.get(tok, (now, 0.0))
            if now - last >= self.META_RETRY_S:
                meta = _token_market(tok)
                if meta and meta[2]:
                    self._meta_fail.pop(tok, None)
                    self._meta_warned.discard(tok)
                    b = st.get("bets", {}).get(tok)
                    for d in (rec, b) if b is not None else (rec,):
                        d["title"] = d.get("title") or meta[0]
                        d["outcome"] = d.get("outcome") or meta[1]
                        if "cond" in d or d is rec:
                            d["cond"] = d.get("cond") or meta[2]
                    if tok in st["my_pos"]:
                        self.conds[tok] = meta[2]
                    log(f"meta repair (#18): …{tok[-8:]} -> "
                        f"{str(meta[0])[:40]} · {meta[1]}")
                    continue
                self._meta_fail[tok] = (first, now)
            if (tok in st["my_pos"] and now - first > self.META_WARN_S
                    and tok not in self._meta_warned):
                self._meta_warned.add(tok)
                self.engine.alert(
                    f"⚠ …{tok[-8:]} ({str(rec.get('title') or '?')[:36]}) has had "
                    f"no condition id for {(now - first) / 3600:.1f}h — "
                    f"UNSETTLEABLE until metadata resolves (#18)")

    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:
            self.repair_market_meta()   # #18: un-stick cond-less books first
            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)
    if bot.fak_retry_s > 0:      # FAK crater misses get one re-quote retry
        engine.on_fak_reject = bot.fak_requote_retry
    engine.on_ignored_exit = bot.write_ignored_exit    # #21 instrumentation
    engine.on_miss_spool = bot.spool_missed   # missed history survives the cap
    # 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()
                if cycle % 30 == 0:
                    # closes #12: entries-reconcile is miss BOOKKEEPING, not a
                    # trading decision — the dominant steady-state API load
                    # (positions?limit=500/wallet) needs no 5-min cadence
                    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.paper_maker_sweep()    # #20: expire resting paper bids
                bot.retry_stuck_exits()    # LIVE_ROLLOUT 1.6
                bot.settle_resolved()
                bot.sweep_dust(cycle)      # reclaim untracked exit residue (live)
                bot._spool_old_bets(cycle) # cap state growth (closes #9)
                if cycle % 5 == 0:
                    bot.reconcile_exits()
                if cycle % 30 == 0:
                    # closes #12: entries-reconcile is miss BOOKKEEPING, not a
                    # trading decision — the dominant steady-state API load
                    # (positions?limit=500/wallet) needs no 5-min cadence
                    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()
