winning-wallet-finder_github/research/study_oracle.py

#!/usr/bin/env python3
"""Study B — crypto oracle fair value vs the book.

The tape's `crypto_prices` aux stream IS the venue's settlement feed
(Binance-sourced, ms-stamped, ~1/s per symbol since 2026-07-19 19:55). Every
strike/sprint crypto market is a digital option on that feed, so fair value
is computable tick-by-tick with no basis risk:

    above K, expiry T:  fair(Yes) = Phi( ln(S_t/K) / (sigma*sqrt(tau)) )
    between K1..K2:     Phi(ln(K2/S)/sv) - Phi(ln(K1/S)/sv)
    sprint (window t0..t1): strike = S_{t0} read from the same feed
    Down/No tokens: 1 - fair(up-side).   sigma = trailing 30min realized
    vol of 1s log-returns (drift negligible at these horizons).

Signal: at a market print, edge = fair - print >= E for that token ->
simulated FAK entry (calibrated sim), hold to resolution (tape truth).

IMPORTANT scope honesty: tick coverage is ~21h, so there is no holdout —
this run only CHOOSES E (grid below) and freezes it; ALL belief is deferred
to the forward ledger. Also grades the live bot's own crypto fills against
fair value at their fill times (objective score of the 0xbadaf319-class
copies)."""
import bisect
import json
import math
import os
import statistics as st
import time

import tape
import sim as simmod

HERE = os.path.dirname(os.path.abspath(__file__))
PARAMS_F = os.path.join(HERE, "params", "study_oracle.json")

EDGE_GRID = [0.04, 0.07, 0.10]
VOL_WIN_S = 1800
TAU_MIN, TAU_MAX = 60, 12 * 3600
COOLDOWN_S = 300
STAKE = 100.0
MIN_FILLS = 30
UP_WORDS = {"up", "yes"}
DOWN_WORDS = {"down", "no"}


def phi(x):
    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))


class TickSeries:
    def __init__(self, ticks):
        self.ts = [t for t, _ in ticks]
        self.px = [p for _, p in ticks]

    def at(self, t):
        i = bisect.bisect_right(self.ts, t) - 1
        return self.px[i] if i >= 0 else None

    def vol_1s(self, t, win=VOL_WIN_S):
        """stdev of 1s log returns over the trailing window (per-sqrt-second)."""
        lo = bisect.bisect_left(self.ts, t - win)
        hi = bisect.bisect_right(self.ts, t)
        if hi - lo < 60:
            return None
        rets = []
        for i in range(lo + 1, hi):
            dt = self.ts[i] - self.ts[i - 1]
            if dt <= 0:
                continue
            r = math.log(self.px[i] / self.px[i - 1]) / math.sqrt(dt)
            rets.append(r)
        return st.pstdev(rets) if len(rets) >= 30 else None


def fair_value(mkt, up_side, S, sigma, t):
    tau = mkt["t1"] - t
    if not (TAU_MIN <= tau <= TAU_MAX) or not S or not sigma:
        return None
    sv = sigma * math.sqrt(tau)
    if sv <= 0:
        return None
    k = mkt["kind"]
    if k == "sprint":
        if mkt.get("s0") is None:
            return None
        f_up = phi(math.log(S / mkt["s0"]) / sv)
    elif k == "above":
        f_up = phi(math.log(S / mkt["k1"]) / sv)
    elif k == "below":
        f_up = 1.0 - phi(math.log(S / mkt["k1"]) / sv)
    elif k == "between":
        f_up = phi(math.log(mkt["k2"] / S) / sv) - phi(math.log(mkt["k1"] / S) / sv)
    else:
        return None
    return f_up if up_side else 1.0 - f_up


def outcome_map(db):
    """asset -> lowercase outcome name, from the orders_matched aux stream."""
    rows = db.execute("""
      SELECT json_extract_string(payload,'$.asset'),
             lower(any_value(json_extract_string(payload,'$.outcome')))
      FROM aux WHERE type = 'orders_matched'
        AND json_extract_string(payload,'$.outcome') != ''
      GROUP BY 1""").fetchall()
    return {a: o for a, o in rows if a and o}


def crypto_universe(db, outcomes, series):
    """Parseable crypto tokens with a knowable side + tick coverage."""
    rows = db.execute("""
      SELECT asset, any_value(title), min(ts), max(ts)
      FROM trades GROUP BY asset""").fetchall()
    out = []
    for asset, title, lo, hi in rows:
        mkt = tape.crypto_parse(title or "")
        if not mkt or mkt["sym"] not in series:
            continue
        o = outcomes.get(asset, "")
        up = o in UP_WORDS or (o == "" and mkt["kind"] != "sprint")
        if o and o not in UP_WORDS | DOWN_WORDS:
            continue                      # unknown side label — skip honestly
        if mkt["kind"] == "sprint":
            if not o:
                continue                  # sprints NEED the Up/Down label
            mkt["s0"] = series[mkt["sym"]].at(mkt["t0"])
            if mkt["s0"] is None:
                continue
        out.append({"asset": asset, "mkt": mkt, "up": up, "title": title})
    return out


def run_study(db, hold_s):
    series = {s: TickSeries(tape.load_ticks(db, s))
              for s in ("btcusdt", "ethusdt", "solusdt", "xrpusdt",
                        "bnbusdt", "dogeusdt")}
    tick_lo = min(s.ts[0] for s in series.values() if s.ts)
    outcomes = outcome_map(db)
    tape.build_resolved(db)
    uni = crypto_universe(db, outcomes, series)
    payout = {a: p for a, p in db.execute(
        "SELECT asset, payout::DOUBLE FROM res_tok").fetchall()}
    print(f"crypto universe: {len(uni)} tokens with side + ticks "
          f"({sum(1 for u in uni if u['asset'] in payout)} resolved in-tape)")

    events = []                          # candidate mispricings at prints
    for u in uni:
        prints = db.execute("""SELECT ts, price FROM trades
            WHERE asset = ? AND ts >= ? ORDER BY ts""",
            [u["asset"], tick_lo]).fetchall()
        s = series[u["mkt"]["sym"]]
        last_ev = 0.0
        for ts, px in prints:
            if ts - last_ev < COOLDOWN_S:
                continue
            S = s.at(ts)
            sig = s.vol_1s(ts)
            f = fair_value(u["mkt"], u["up"], S, sig, ts)
            if f is None:
                continue
            edge = f - float(px)
            if edge > 0.02:              # collect loosely; grid filters below
                last_ev = ts
                events.append({"asset": u["asset"], "ts": ts, "p_ref": float(px),
                               "fair": round(f, 4), "edge": round(edge, 4),
                               "kind": u["mkt"]["kind"], "title": u["title"]})
    print(f"candidate mispricing events (edge > 2c): {len(events)}")

    sim = simmod.Sim(db, hold_s=hold_s)
    grid = {}
    for E in EDGE_GRID:
        sel = [e for e in events if e["edge"] >= E]
        fills = wins = 0
        pnl = staked = 0.0
        misses = pending = 0
        for e in sel:
            r = sim.try_buy(e["asset"], e["ts"], e["p_ref"], stake_usd=STAKE)
            if not r["filled"]:
                misses += 1
                continue
            pay = payout.get(e["asset"])
            if pay is None:
                pending += 1
                continue
            fills += 1
            staked += r["cost"]
            pnl += r["shares"] * (pay - r["price"]) - r["fee"]
            wins += pay == 1.0
        grid[E] = {"events": len(sel), "fills": fills, "misses": misses,
                   "pending": pending,
                   "ev_per_fill": round(pnl / fills, 2) if fills else None,
                   "hit": round(wins / fills, 3) if fills else None,
                   "pnl": round(pnl, 2)}
        print(f"E >= {E:.2f}: {grid[E]}")
    eligible = [(E, g) for E, g in grid.items()
                if g["fills"] >= MIN_FILLS and g["ev_per_fill"] is not None]
    frozen_E = max(eligible, key=lambda eg: eg[1]["ev_per_fill"])[0] \
        if eligible else None
    return {"grid": grid, "frozen_edge": frozen_E, "n_universe": len(uni),
            "tick_lo": tick_lo}, events


def grade_our_fills(db):
    """Fair-value edge of the live bot's own crypto fills at fill time."""
    series = {}
    graded = []
    for ln in open(os.path.join(tape.ROOT, "copybot_fills.live.jsonl")):
        r = json.loads(ln)
        if r.get("side") == "SELL" or r.get("untracked"):
            continue
        mkt = tape.crypto_parse(r.get("title") or "")
        if not mkt:
            continue
        sym = mkt["sym"]
        if sym not in series:
            series[sym] = TickSeries(tape.load_ticks(db, sym))
        s = series[sym]
        if not s.ts or r["ts"] < s.ts[0] or mkt["kind"] == "sprint":
            continue
        S, sig = s.at(r["ts"]), s.vol_1s(r["ts"])
        up = (r.get("outcome") or "").lower() in UP_WORDS
        f = fair_value(mkt, up, S, sig, r["ts"])
        if f is None:
            continue
        graded.append({"title": r["title"][:60], "outcome": r.get("outcome"),
                       "px": r["my_price"], "fair": round(f, 3),
                       "edge": round(f - r["my_price"], 3),
                       "wallet": r.get("name")})
    return graded


def main():
    db = tape.connect()
    cal = json.load(open(os.path.join(HERE, "params", "sim_calibration.json")))
    res, events = run_study(db, cal["hold_s"])
    graded = grade_our_fills(db)
    print(f"\nour crypto fills graded vs fair value: {len(graded)}")
    for g in graded:
        print(f"  {g['edge']:+.3f}  {g['wallet']:<12} {g['outcome']:<4} "
              f"@{g['px']:.3f} fair {g['fair']:.3f}  {g['title']}")
    json.dump({**res, "our_fills_graded": graded,
               "frozen_at": time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime()),
               "note": "NO holdout exists (21h ticks) — belief deferred "
                       "entirely to forward_ledger"},
              open(PARAMS_F, "w"), indent=1, default=float)
    print(f"\nfroze {PARAMS_F}")


if __name__ == "__main__":
    main()