winning-wallet-finder_github/research/sim.py

#!/usr/bin/env python3
"""Execution replayer calibrated on the live bot's OWN ledger.

Model: a signal at t with reference price p_ref becomes a marketable FAK
arriving at t+lag with protected cap p_ref*(1+slip_cap). The tape has no
book stream, so standing liquidity at arrival is proxied by PRINTS: the
order fills at the first trade print on the token inside
(arrive, arrive+hold_s] whose price is inside the cap — else it dies
no-match (the crater). hold_s is NOT a free choice: `calibrate()` fits it
so the model best separates the bot's real live fills (should fill) from
its real FAK-rejected misses (should miss), and reports fill-price error
with the bot's own prints EXCLUDED (else the validation is circular — our
fill is itself a tape print).

Fees mirror the venue: fee = rate * shares * min(p, 1-p) (verified against
the live ledger: 7.81sh @ .64 -> $0.0844, 5.26sh @ .95 -> $0.0075).

Everything is deterministic — scenarios (lag percentiles) not RNG.
"""
import json
import os

HERE = os.path.dirname(os.path.abspath(__file__))
ROOT = os.path.dirname(HERE)
BOT_WALLET = "0x455e252e45ee46d6c4cc1c8fadd3899d68f245a1"

FEE_RATE = 0.03
LAG_P50, LAG_P90 = 6.7, 66.4       # live ledger 2026-07-20 (102 BUY fills)


def fee(shares, price, rate=FEE_RATE):
    return rate * shares * min(price, 1.0 - price)


class Sim:
    def __init__(self, db, lag_s=LAG_P50, slip_cap=0.05, hold_s=10,
                 fee_rate=FEE_RATE, exclude_wallet=None, fill="first"):
        self.db = db
        self.lag_s = lag_s
        self.slip_cap = slip_cap
        self.hold_s = hold_s
        self.fee_rate = fee_rate
        self.excl = (exclude_wallet or "").lower()
        self.fill = fill               # "first" print, or "worst" (pessimistic)

    def first_print(self, asset, t0, t1, cap=None):
        """First trade print on asset in (t0, t1], optionally inside cap."""
        q = """SELECT ts, price FROM trades
               WHERE asset = ? AND ts > ? AND ts <= ?"""
        args = [asset, t0, t1]
        if self.excl:
            q += " AND lower(wallet) != ?"
            args.append(self.excl)
        if cap is not None:
            q += " AND price <= ?"
            args.append(cap)
        q += " ORDER BY ts LIMIT 1"
        r = self.db.execute(q, args).fetchone()
        return r  # (ts, price) or None

    def try_buy(self, asset, t_sig, p_ref, stake_usd=100.0, lag_s=None):
        """-> dict(filled, price, shares, cost, fee) — FAK with protected cap."""
        lag = self.lag_s if lag_s is None else lag_s
        arrive = t_sig + lag
        cap = min(p_ref * (1 + self.slip_cap), 0.99)
        if self.fill == "worst":       # pay the top of the burst
            # (ORDER BY form: max(ts),max(price) trips a duckdb-internal
            # statistics-propagation assertion on this temp-table layout)
            pr = self.db.execute("""SELECT ts, price FROM trades
                 WHERE asset = ? AND ts > ? AND ts <= ? AND price <= ?
                 ORDER BY price DESC LIMIT 1""",
                 [asset, arrive, arrive + self.hold_s, cap]).fetchone()
        else:
            pr = self.first_print(asset, arrive, arrive + self.hold_s, cap)
        if not pr:
            return {"filled": False, "reason": "no print inside band (crater)"}
        px = float(pr[1])
        shares = stake_usd / px
        return {"filled": True, "price": px, "shares": shares,
                "cost": shares * px, "fee": fee(shares, px, self.fee_rate),
                "fill_ts": pr[0]}

    def markout(self, asset, t_fill, horizon_s):
        """Last print at/before t_fill+horizon (None if nothing printed)."""
        r = self.db.execute("""SELECT price FROM trades WHERE asset = ?
              AND ts > ? AND ts <= ? ORDER BY ts DESC LIMIT 1""",
              [asset, t_fill, t_fill + horizon_s]).fetchone()
        return r[0] if r else None


# ── calibration against the live ledger ─────────────────────────────────────

def _live_attempts(tape_lo, tape_hi):
    """Real BUY attempts inside the tape window:
    fills from copybot_fills.live.jsonl (label filled=True) and FAK
    no-match misses from copybot_state.live.json (label filled=False)."""
    fills = []
    for ln in open(os.path.join(ROOT, "copybot_fills.live.jsonl")):
        r = json.loads(ln)
        if r.get("untracked") or r.get("side") == "SELL":
            continue
        if r.get("detect_lag_s") is None or not r.get("their_price"):
            continue
        t_sig = r["ts"] - r["detect_lag_s"]
        if not (tape_lo <= t_sig <= tape_hi - 120):
            continue
        fills.append({"filled": True, "asset": str(r["token"]),
                      "t_sig": t_sig, "p_ref": r["their_price"],
                      "lag": r["detect_lag_s"], "actual_px": r["my_price"]})
    st = json.load(open(os.path.join(ROOT, "copybot_state.live.json")))
    misses = []
    for m in st.get("missed", []):
        if "no orders found to match" not in str(m.get("reason", "")):
            continue
        if not (tape_lo <= m["ts"] <= tape_hi - 120):
            continue
        misses.append({"filled": False, "asset": str(m["token"]),
                       "t_sig": m["ts"], "p_ref": m["price"], "lag": LAG_P50})
    return fills, misses


def calibrate(db, tape_lo, tape_hi, holds=(3, 5, 10, 20, 45, 90)):
    """Fit hold_s on real outcomes; report the confusion + price error."""
    fills, misses = _live_attempts(tape_lo, tape_hi)
    out = {"n_fills": len(fills), "n_misses": len(misses), "grid": {}}
    best = None
    for h in holds:
        sim = Sim(db, hold_s=h, exclude_wallet=BOT_WALLET)
        tp = sum(1 for a in fills
                 if sim.try_buy(a["asset"], a["t_sig"], a["p_ref"],
                                lag_s=a["lag"])["filled"])
        tn = sum(1 for a in misses
                 if not sim.try_buy(a["asset"], a["t_sig"], a["p_ref"],
                                    lag_s=a["lag"])["filled"])
        acc = (tp + tn) / max(len(fills) + len(misses), 1)
        out["grid"][h] = {"fill_recall": tp / max(len(fills), 1),
                          "miss_recall": tn / max(len(misses), 1),
                          "acc": round(acc, 3)}
        if best is None or acc > best[1]:
            best = (h, acc)
    out["hold_s"] = best[0]
    sim = Sim(db, hold_s=best[0], exclude_wallet=BOT_WALLET)
    errs, signed = [], []
    for a in fills:
        r = sim.try_buy(a["asset"], a["t_sig"], a["p_ref"], lag_s=a["lag"])
        if r["filled"]:
            errs.append(abs(r["price"] - a["actual_px"]))
            signed.append(r["price"] - a["actual_px"])
    errs.sort()
    if errs:
        out["px_err_p50"] = round(errs[len(errs) // 2], 4)
        out["px_err_p90"] = round(errs[int(len(errs) * 0.9)], 4)
        out["px_within_1c"] = round(sum(e <= 0.01 for e in errs) / len(errs), 3)
        # signed bias: negative = sim fills cheaper than reality = OPTIMISTIC
        # (study EVs must clear |bias| + noise before they mean anything)
        out["px_bias_mean"] = round(sum(signed) / len(signed), 4)
    return out


if __name__ == "__main__":
    import tape
    db = tape.connect()
    lo, hi = db.execute("SELECT min(ts), max(ts) FROM trades").fetchone()
    cal = calibrate(db, lo, hi)
    print(json.dumps(cal, indent=2))
    json.dump(cal, open(os.path.join(HERE, "params", "sim_calibration.json"),
                        "w"), indent=1)