winning-wallet-finder_github/research/replay.py

#!/usr/bin/env python3
"""Set-replay harness: sweep candidate wallet SETS over the recorder tape
(2026-07-21, per the set-design discussion — "how many wallets can a
bankroll carry, and which composition?").

SEARCH TOOL ONLY. Verdicts still come exclusively from
research/forward_ledger.jsonl (README silo rules). What this buys over the
per-wallet bench: SET-level interactions — shared-equity compounding (a hot
wallet inflates everyone's 4% stakes), capital contention (all-or-nothing
cash gate), and paired comparison on the SAME tape (two live paper books
watch different weeks; replays of two sets watch identical ones).

Mechanics mirrored from the engine (copytrade.py, cited, NOT imported —
silo rule; parameters are read from live/copybot.paper.json read-only so
parity survives config edits):
  stake_usd   L322: class_pct × (cash + open cost), halved under 80% HWM,
              capped at THEIR cumulative stake, floored at min_order_usd.
  gate_buy    L384: all-or-nothing — cash < stake is a MISS, never partial.
  buy mirror  L403/_handle_their_buy: opens AND adds; per-tx clip merge;
              conviction floor on their trade USD; entry band.
  sell mirror _handle_their_sell: proportional (their_size/their_prev of
              OUR shares).
Execution = sim.Sim (calibrated FAK-print model: lag, +5c protected band =
price_guard_abs, crater no-match). Resolution = tape.build_resolved (the
742/742 chain-validated proxy); unresolved positions mark at last print.

Known v1 biases (identical across sets — rankings robust, absolutes soft):
  - no-backfill unknowable pre-tape: every first tape BUY counts as an OPEN
    (the real bot skips positions a wallet held before watching began);
  - exits fill at their sell print VWAP (no crater model on the way out);
  - sim optimism ≈ -2c/fill documented in FINDINGS (thresholds sit 2x out);
  - FAK re-quote retry (2026-07-20) not modelled — craters count as misses.
"""
import argparse
import json
import os
import sys
import time
from collections import defaultdict

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
sys.path.insert(0, HERE)
import tape                                    # noqa: E402
import sim as simmod                           # noqa: E402

DD_THRESHOLD, DD_FACTOR = 0.80, 0.5            # copytrade.py L306
OUT_DIR = os.path.join(HERE, "replay_out")


def paper_params():
    """Parity params read (read-only) from the paper bot's config."""
    c = json.load(open(os.path.join(ROOT, "live", "copybot.paper.json")))
    f = c["follow"]
    return {
        "class_pct": f.get("class_pct", {"volume": 0.04}),
        "min_their_usd": f.get("min_their_usd", 25.0),
        "min_entry": f.get("min_entry", 0.0),
        "max_entry": f.get("max_entry", 0.95),
        "buy_only": f.get("buy_only", True),
        "min_order_usd": c.get("risk", {}).get("min_order_usd", 5.0),
        "slip_cap": c.get("price_guard_abs", 0.05),
        "current_set": [{"wallet": w["wallet"].lower(),
                         "name": w.get("name", w["wallet"][:10]),
                         "class": w.get("class", "volume"),
                         "floor": w.get("floor")}
                        for w in c["wallets"]],
    }


def tape_p80_floor(db, wallet):
    """Conviction floor for a wallet with no pinned floor: p80 of its own
    tape BUY stakes — the same top-20% rule sync_floors pins from the
    trusted cache, derived from the only history the tape has."""
    r = db.execute("""
        SELECT quantile_cont(usd, 0.8) FROM (
          SELECT sum(price*size) usd FROM trades
          WHERE lower(wallet) = ? AND side = 'BUY'
          GROUP BY tx, asset)""", [wallet.lower()]).fetchone()
    return float(r[0]) if r and r[0] is not None else None


def signals(db, wallets, t_lo=None, t_hi=None):
    """Per-tx clip-merged trades of the watched wallets, time-ordered.
    -> [{ts, wallet, asset, cond, side, vwap, size, usd, title}]"""
    ws = sorted({w.lower() for w in wallets})
    q = """SELECT min(ts) ts, lower(wallet) wallet, asset, any_value(cond) cond,
                  side, sum(price*size)/nullif(sum(size),0) vwap,
                  sum(size) size, sum(price*size) usd, any_value(title) title
           FROM trades WHERE lower(wallet) IN ({}) {} {}
           GROUP BY tx, lower(wallet), asset, side
           ORDER BY ts""".format(
        ",".join("?" * len(ws)),
        "AND ts >= ?" if t_lo else "", "AND ts <= ?" if t_hi else "")
    args = ws + ([t_lo] if t_lo else []) + ([t_hi] if t_hi else [])
    cols = ("ts", "wallet", "asset", "cond", "side", "vwap", "size", "usd", "title")
    return [dict(zip(cols, r)) for r in db.execute(q, args).fetchall()]


class Book:
    """The engine's book mechanics, replayed. One instance per (set, bankroll)."""

    def __init__(self, bankroll, prm, sim):
        self.cash = bankroll
        self.bankroll = bankroll
        self.prm = prm
        self.sim = sim
        self.hwm = bankroll
        self.pos = {}                   # asset -> {shares, cost, wallet}
        self.their = defaultdict(float)  # (wallet, asset) -> shares
        self.bets = []                  # closed + open records
        self.miss = defaultdict(list)   # family -> [records]
        self.dep_curve = []             # (ts, deployed, equity)

    def open_cost(self):
        return sum(p["cost"] for p in self.pos.values())

    def stake_usd(self, klass, their_total):
        eq = self.cash + self.open_cost()
        self.hwm = max(self.hwm, eq)
        frac = self.prm["class_pct"].get(klass, 0.04)
        if eq < DD_THRESHOLD * self.hwm:
            frac *= DD_FACTOR
        stake = frac * eq
        if their_total and stake > their_total:
            stake = their_total
        return max(stake, self.prm["min_order_usd"])

    def on_buy(self, s, klass, floor=None):
        their_prev = self.their[(s["wallet"], s["asset"])]
        self.their[(s["wallet"], s["asset"])] = their_prev + s["size"]
        if s["usd"] < (floor if floor else self.prm["min_their_usd"]):
            return                     # below the wallet's conviction floor
        if not (self.prm["min_entry"] <= s["vwap"] <= self.prm["max_entry"]):
            self.miss["entry_band"].append(s)
            return
        mine = self.pos.get(s["asset"])
        # ceiling arg is SHARES (their_prev + their_size), mirroring the
        # engine call site verbatim (copytrade L512/L524)
        stake_rule = self.stake_usd(klass, their_prev + s["size"])
        if mine:
            # ADD: one-market-one-stake — grow proportionally but never past
            # the stake rule for the whole position (copytrade L507-521)
            frac = s["size"] / their_prev if their_prev > 0 else 0
            room = stake_rule - mine["cost"]
            if room < self.prm["min_order_usd"]:
                return                          # silent skip, like the bot
            want = min(mine["shares"] * frac * s["vwap"], room)
            if want < self.prm["min_order_usd"]:
                return
        else:
            want = stake_rule
        if self.cash < want:
            self.miss["capital"].append({**s, "stake": want})
            return
        r = self.sim.try_buy(s["asset"], s["ts"], s["vwap"], stake_usd=want)
        if not r["filled"]:
            self.miss["crater"].append({**s, "stake": want})
            return
        self.cash -= r["cost"] + r["fee"]
        p = self.pos.setdefault(s["asset"], {"shares": 0.0, "cost": 0.0,
                                             "wallet": s["wallet"],
                                             "cond": s["cond"],
                                             "title": s["title"] or ""})
        p["shares"] += r["shares"]
        p["cost"] += r["cost"] + r["fee"]
        self.bets.append({"asset": s["asset"], "wallet": s["wallet"],
                          "ts": s["ts"], "price": r["price"],
                          "shares": r["shares"], "cost": r["cost"] + r["fee"],
                          "pnl": None})
        self.dep_curve.append((s["ts"], self.open_cost(),
                               self.cash + self.open_cost()))

    def on_sell(self, s):
        their_prev = self.their[(s["wallet"], s["asset"])]
        self.their[(s["wallet"], s["asset"])] = max(0.0, their_prev - s["size"])
        p = self.pos.get(s["asset"])
        if not p:
            return
        frac = 1.0 if their_prev <= 0 else min(1.0, s["size"] / their_prev)
        sh = p["shares"] * frac
        proceeds = sh * s["vwap"]
        f = simmod.fee(sh, s["vwap"])
        avg_cost = p["cost"] / p["shares"]
        self.cash += proceeds - f
        self._book_pnl(s["asset"], sh, proceeds - f - avg_cost * sh,
                       p["wallet"])
        p["shares"] -= sh
        p["cost"] -= avg_cost * sh
        if p["shares"] < 1e-9:
            del self.pos[s["asset"]]

    def _book_pnl(self, asset, shares, pnl, wallet):
        for b in self.bets:
            if b["asset"] == asset and b["pnl"] is None:
                b["pnl"] = pnl                  # first open lot takes it
                return
        self.bets.append({"asset": asset, "wallet": wallet, "ts": 0,
                          "price": 0, "shares": shares, "cost": 0, "pnl": pnl})

    def settle(self, payouts, marks):
        """Tape-end: proxy-resolved positions pay 1/0; the rest mark."""
        realized = sum(b["pnl"] for b in self.bets if b["pnl"] is not None)
        unresolved_mark = 0.0
        for a, p in list(self.pos.items()):
            pay = payouts.get(a)
            if pay is not None:
                self.cash += p["shares"] * pay        # redeem free
                self._book_pnl(a, p["shares"], p["shares"] * pay - p["cost"],
                               p["wallet"])
                realized += p["shares"] * pay - p["cost"]
                del self.pos[a]
            else:
                unresolved_mark += p["shares"] * marks.get(a, 0.0) - p["cost"]
        return realized, unresolved_mark


def replay(db, wallets_cfg, bankroll, prm, sim, t_lo=None, t_hi=None):
    klass = {w["wallet"]: w.get("class", "volume") for w in wallets_cfg}
    floors = {w["wallet"]: (w.get("floor") or tape_p80_floor(db, w["wallet"]))
              for w in wallets_cfg}
    book = Book(bankroll, prm, sim)
    for s in signals(db, list(klass), t_lo, t_hi):
        if s["side"] == "BUY":
            book.on_buy(s, klass[s["wallet"]], floors.get(s["wallet"]))
        else:
            book.on_sell(s)     # buy_only: their SELLs only ever CLOSE ours
    # resolution + marks
    tape.build_resolved(db)
    payouts = {a: float(p) for a, p in db.execute(
        "SELECT asset, payout FROM res_tok WHERE payout IS NOT NULL").fetchall()}
    marks = {}
    if book.pos:
        marks = {a: float(m) for a, m in db.execute(
            "SELECT asset, arg_max(price, ts) FROM trades WHERE asset IN ({}) "
            "GROUP BY asset".format(",".join("?" * len(book.pos))),
            list(book.pos)).fetchall()}
    realized, mark = book.settle(payouts, marks)
    dep = [d for _, d, _ in book.dep_curve]
    eqs = [e for _, _, e in book.dep_curve]
    per_wallet = defaultdict(float)
    for b in book.bets:
        if b["pnl"] is not None:
            per_wallet[b["wallet"]] += b["pnl"]
    return {
        "bankroll": bankroll, "copies": len(book.bets),
        "realized": round(realized, 2), "open_mark": round(mark, 2),
        "end_equity": round(book.cash + book.open_cost() + mark, 2),
        "misses": {k: len(v) for k, v in book.miss.items()},
        "capital_miss_hypo": round(_hypo(book.miss.get("capital", []),
                                         payouts), 2),
        "peak_deploy_pct": round(100 * max((d / e for d, e in zip(dep, eqs)),
                                           default=0.0), 1),
        "mean_deploy": round(sum(dep) / len(dep), 2) if dep else 0.0,
        "per_wallet": {w: round(p, 2) for w, p in sorted(per_wallet.items())},
    }


def _hypo(capital_misses, payouts):
    """What the capital misses would have paid at resolution (stake-sized)."""
    tot = 0.0
    for m in capital_misses:
        pay = payouts.get(m["asset"])
        if pay is not None and m["vwap"] > 0:
            tot += m["stake"] / m["vwap"] * pay - m["stake"]
    return tot


def main():
    ap = argparse.ArgumentParser(description="replay wallet sets over the tape")
    ap.add_argument("--sets", default=os.path.join(HERE, "params",
                                                   "replay_sets.json"))
    ap.add_argument("--bankrolls", default="500,1000,2000,5000")
    ap.add_argument("--loo", action="store_true",
                    help="leave-one-out marginals at each bankroll")
    ap.add_argument("--lag", type=float, default=simmod.LAG_P50)
    args = ap.parse_args()

    prm = paper_params()
    sets = {"current": prm["current_set"]}
    if os.path.exists(args.sets):
        for name, ws in json.load(open(args.sets)).items():
            sets[name] = [{"wallet": w["wallet"].lower(),
                           "name": w.get("name", w["wallet"][:10]),
                           "class": w.get("class", "volume")} for w in ws]

    db = tape.connect()
    lo, hi = db.execute("SELECT min(ts), max(ts) FROM trades").fetchone()
    print(f"tape window: {time.strftime('%m-%d %H:%M', time.gmtime(lo))} -> "
          f"{time.strftime('%m-%d %H:%M', time.gmtime(hi))} UTC "
          f"({(hi - lo) / 86400:.2f} days)")
    sim = simmod.Sim(db, lag_s=args.lag, slip_cap=prm["slip_cap"],
                     exclude_wallet=simmod.BOT_WALLET)

    out = {"ran_at": int(time.time()), "tape": [lo, hi], "lag_s": args.lag,
           "results": {}}
    for name, ws in sets.items():
        for bank in [float(b) for b in args.bankrolls.split(",")]:
            r = replay(db, ws, bank, prm, sim)
            out["results"][f"{name}@{bank:.0f}"] = r
            m = r["misses"]
            print(f"{name:24s} ${bank:>6.0f}  copies {r['copies']:3d}  "
                  f"realized {r['realized']:+9.2f}  open {r['open_mark']:+8.2f} "
                  f" deploy μ${r['mean_deploy']:.0f}/pk{r['peak_deploy_pct']}%"
                  f"  miss cap:{m.get('capital', 0)} crater:{m.get('crater', 0)}"
                  f" band:{m.get('entry_band', 0)}"
                  f"  capmiss_hypo {r['capital_miss_hypo']:+.2f}")
            if args.loo and len(ws) > 1 and bank == 1000.0:
                base = r["realized"]
                for drop in ws:
                    sub = [w for w in ws if w is not drop]
                    rr = replay(db, sub, bank, prm, sim)
                    print(f"    -{drop['name']:20s} marginal "
                          f"{base - rr['realized']:+9.2f}  "
                          f"(set realized {rr['realized']:+9.2f})")
    os.makedirs(OUT_DIR, exist_ok=True)
    path = os.path.join(OUT_DIR, f"replay_{int(time.time())}.json")
    json.dump(out, open(path, "w"), indent=1)
    print(f"\nwrote {path}")


if __name__ == "__main__":
    main()