diff --git a/insider.py b/insider.py index b7f96c4f..e62b0fc2 100644 --- a/insider.py +++ b/insider.py @@ -47,8 +47,22 @@ def _parse_end(end): return 0 -def resolved_bets(wallet, cutoff, max_pages=40): - """Resolved bets with entry price, conditionId, resolution time, size.""" +def resolved_bets(wallet, cutoff, max_pages=40, strict=False): + """Resolved bets with entry price, conditionId, token (asset), resolution + time, size, and provenance (cache schema v2). + + * ``p`` is the RAW avgPrice (0 when the API omits it) — callers clamp for + the z math; storing raw keeps "missing price" distinguishable from a real + 0.1¢ longshot in the cache. + * ``asset`` (token id) is the position identity — it disambiguates the + two-endpoint union (same asset in /closed-positions and /positions is ONE + position seen twice, not two bets) and YES/NO both-sides holdings. + * ``resolved`` is False for early-sold positions in markets that had not + ended at pull time — their ``won`` is a curPrice mark, not an outcome. + * ``strict``: raise on a failed page pull instead of returning a silently + truncated history — a partial pull must never be cached as a wallet's + complete record. + """ now = time.time() out = [] for endpoint in ("/closed-positions", "/positions"): @@ -60,6 +74,8 @@ def resolved_bets(wallet, cutoff, max_pages=40): else: params["sizeThreshold"] = 0.0 page = sm.get_json(endpoint, params) + if page is None and strict: + raise RuntimeError(f"{endpoint} pull failed for {wallet} at offset {off}") if not page: break for p in page: @@ -69,17 +85,24 @@ def resolved_bets(wallet, cutoff, max_pages=40): if ts < cutoff: continue res_t = end or ts + resolved = bool(end) and end <= now else: + ts = None if not (cutoff <= end < now): continue res_t = end + resolved = True out.append({ "won": p.get("curPrice", 0) >= 0.5, - "p": max(0.001, min(0.999, p.get("avgPrice", 0) or 0)), + "p": p.get("avgPrice", 0) or 0, # raw — callers clamp "cond": p.get("conditionId"), + "asset": p.get("asset"), "res_t": res_t, "size": p.get("initialValue") or (p.get("avgPrice", 0) * p.get("totalBought", 0)), + "src": "closed" if endpoint == "/closed-positions" else "open", + "ts": ts, + "resolved": resolved, }) off += 50 if len(page) < 50: @@ -123,6 +146,8 @@ def analyze(cand): bets = resolved_bets(wallet, cutoff) if len(bets) < 15: return None + for b in bets: # v2 returns raw p — clamp for the z math + b["p"] = max(0.001, min(0.999, b["p"] or 0)) first_buy, _ = entry_times(wallet) total_trades = (sm.get_json("/traded", {"user": wallet}) or {}).get("traded", 0) diff --git a/live/README.md b/live/README.md index f9972dbf..cb2701bc 100644 --- a/live/README.md +++ b/live/README.md @@ -50,14 +50,25 @@ any archetype, any cutoff, the clean OOS test — now runs in **seconds** instea of hours of API pulls. `MAX_AGE_DAYS=14`: the broad pool refreshes biweekly; the watchlist is force-refreshed daily (`cache.invalidate`) for forward tracking. -**Retention gotcha — the cache is NOT append-only.** Each wallet's refresh does -`DELETE FROM bets WHERE wallet=?` then re-inserts a fresh pull, and that pull is a -**rolling 180-day window** (`WINDOW_DAYS`) capped at ~2k bets/endpoint -(`max_pages`). So wallet *coverage* grows (new wallets are kept), but any single -wallet's history is a capped, rolling, overwrite-on-refresh snapshot — bets older -than ~180d are dropped on the next re-pull. For a permanent long-horizon archive, -use the append-style `../wide/pmkt.duckdb` subgraph dataset instead, or change the -pull to upsert + drop the cutoff. +**Schema v2 (2026-07-02) — token-keyed, provenance-tagged, archival.** `bets` +now carries `asset` (token id — the position identity), `src`/`ts` (endpoint +provenance + close time), and `resolved` (False = early-sold position in a +market that hadn't ended at pull time; its `won` is a curPrice *mark*, not an +outcome — scorers filter these). `p` is stored **raw** (0 = avgPrice missing) +and clamped to [0.001, 0.999] by `get_bets` on read, so "missing price" stays +distinguishable from a real 0.1¢ longshot. Refresh is an **upsert by token** +(plus superseded legacy rows), not a wallet wipe: each pull still covers the +rolling `WINDOW_DAYS`, but rows that slide out of the window now *survive*, so +per-wallet history accumulates into a permanent archive. The same-asset row +from both endpoints (a partially-closed position) is deduped to the larger- +stake row instead of double-counting. Failed pulls are returned empty but NOT +cached and NOT marked pulled — they retry on the next call instead of +masquerading as "no bets" for `MAX_AGE_DAYS` (pre-v2, an API error could cache +a wallet as empty-and-fresh; that bug bit the watchlist in practice). Legacy v1 +rows keep NULLs in the new columns until their wallet's next refresh. The +migration runs automatically on first open (v1 → v2, exact-duplicate rows +merged). Per-endpoint pagination is still capped at ~2k bets (`max_pages`); +`../wide/pmkt.duckdb` remains the deep-history subgraph dataset. ## The clean test (why the favorites are a mirage) diff --git a/live/cache.py b/live/cache.py index 00b53221..6569ae83 100644 --- a/live/cache.py +++ b/live/cache.py @@ -1,11 +1,28 @@ #!/usr/bin/env python3 """Local cache of per-wallet resolved bets, so we stop re-pulling the data-api. -Each wallet's resolved bets (won, entry price p, conditionId, resolution time, -size) are stored once in cache.duckdb. Because we keep res_t per bet, ANY date -cutoff — pre-June-1, full window, future experiments — reads the same cached -rows and filters locally. A pull only happens for wallets not seen, or older -than MAX_AGE_DAYS. +Each wallet's resolved bets are stored once in cache.duckdb. Because we keep +res_t per bet, ANY date cutoff — pre-June-1, full window, future experiments — +reads the same cached rows and filters locally. A pull only happens for wallets +not seen, or older than MAX_AGE_DAYS. + +Schema v2 (migrated automatically on first open; legacy rows keep NULLs in the +new columns until their wallet refreshes): + * asset — token id, the position identity. Dedupes the two-endpoint union + (the same asset from /closed-positions AND /positions is one + position seen twice) and disambiguates YES/NO both-sides rows. + * src/ts — endpoint provenance ('closed'/'open') + close timestamp. + * resolved — False for early-sold positions in markets that hadn't ended at + pull time (their `won` is a curPrice mark, not an outcome). + * p — stored RAW (0 = avgPrice missing); get_bets clamps to + [0.001, 0.999] on read, so consumers see the same values as + before while the DB keeps missing-vs-real-longshot separable. + * upsert — refresh replaces only the re-pulled tokens instead of wiping + the wallet, so history beyond the rolling WINDOW_DAYS pull + accumulates (permanent archive instead of overwrite-on-refresh). + * failures — a failed pull is returned empty but NOT cached and NOT marked + pulled, so it retries next call instead of masquerading as + "wallet has no bets" for MAX_AGE_DAYS. Thread-safe: API pulls (the slow part) run outside the lock; only the small DuckDB reads/writes are serialized, so skill.py's worker pool still parallelizes @@ -49,7 +66,35 @@ def conv_cutoff(sizes, q=CONV_PCTILE): _lock = threading.Lock() _con = duckdb.connect(DB) _con.execute("""CREATE TABLE IF NOT EXISTS bets( - wallet TEXT, cond TEXT, won BOOLEAN, p DOUBLE, res_t BIGINT, size DOUBLE)""") + wallet TEXT, cond TEXT, asset TEXT, won BOOLEAN, p DOUBLE, res_t BIGINT, + size DOUBLE, src TEXT, ts BIGINT, resolved BOOLEAN)""") + + +def _migrate_v2(): + """One-shot in-place migration of a v1 `bets` table (no asset/src/ts/resolved + columns). Rebuilds via SELECT DISTINCT — v1 had no position identity, so its + few thousand byte-identical duplicate rows are unrecoverable noise and are + merged. Legacy rows keep NULLs in the new columns until their wallet is + re-pulled; `p` stays clamped for them (raw-p is forward-only).""" + cols = {r[0] for r in _con.execute("DESCRIBE bets").fetchall()} + if "asset" in cols: + return + n0 = _con.execute("SELECT count(*) FROM bets").fetchone()[0] + _con.execute("BEGIN") + _con.execute("""CREATE TABLE bets_v2( + wallet TEXT, cond TEXT, asset TEXT, won BOOLEAN, p DOUBLE, res_t BIGINT, + size DOUBLE, src TEXT, ts BIGINT, resolved BOOLEAN)""") + _con.execute("""INSERT INTO bets_v2(wallet, cond, won, p, res_t, size) + SELECT DISTINCT wallet, cond, won, p, res_t, size FROM bets""") + _con.execute("DROP TABLE bets") + _con.execute("ALTER TABLE bets_v2 RENAME TO bets") + _con.execute("COMMIT") + n1 = _con.execute("SELECT count(*) FROM bets").fetchone()[0] + print(f"[cache] migrated bets to schema v2: {n0:,} -> {n1:,} rows " + f"({n0 - n1:,} exact duplicates merged)", flush=True) + + +_migrate_v2() _con.execute("CREATE INDEX IF NOT EXISTS bets_w ON bets(wallet)") _con.execute("CREATE TABLE IF NOT EXISTS pulled(wallet TEXT PRIMARY KEY, pulled_at BIGINT)") _con.execute("CREATE TABLE IF NOT EXISTS entries(wallet TEXT, cond TEXT, first_buy BIGINT)") @@ -79,30 +124,63 @@ def get_entries(wallet): return first_buy +def _bet_row(won, p, cond, res_t, size, asset, src, ts, resolved): + """The dict shape get_bets returns — p clamped on read so consumer math is + unchanged while the DB stores it raw.""" + return {"won": won, "p": max(0.001, min(0.999, p or 0)), "cond": cond, + "res_t": res_t, "size": size, "asset": asset, "src": src, + "ts": ts, "resolved": resolved} + + def get_bets(wallet): - """Resolved bets for a wallet — from cache if fresh, else pull and store.""" + """Resolved bets for a wallet — from cache if fresh, else pull and upsert.""" now = time.time() with _lock: r = _con.execute("SELECT pulled_at FROM pulled WHERE wallet=?", [wallet]).fetchone() if r and now - r[0] < MAX_AGE_DAYS * 86400: rows = _con.execute( - "SELECT won,p,cond,res_t,size FROM bets WHERE wallet=?", [wallet]).fetchall() - return [{"won": w, "p": p, "cond": c, "res_t": rt, "size": s} - for w, p, c, rt, s in rows] + "SELECT won,p,cond,res_t,size,asset,src,ts,resolved " + "FROM bets WHERE wallet=?", [wallet]).fetchall() + return [_bet_row(*row) for row in rows] # cache miss / stale -> pull (slow, outside the lock so workers stay parallel) try: - bets = insider.resolved_bets(wallet, now - WINDOW_DAYS * 86400) + bets = insider.resolved_bets(wallet, now - WINDOW_DAYS * 86400, strict=True) except Exception: - bets = [] + return [] # transient API failure — do NOT cache or mark pulled; + # the next call retries instead of trusting a bad pull + # one row per token: the endpoint union returns the same asset twice for a + # partially-closed position (closed portion + open remainder) — keep the + # larger-stake row rather than double-counting one position as two bets. + best = {} + for b in bets: + k = (b["cond"], b.get("asset")) + if k not in best or (b.get("size") or 0) > (best[k].get("size") or 0): + best[k] = b + bets = list(best.values()) with _lock: - _con.execute("DELETE FROM bets WHERE wallet=?", [wallet]) + # upsert: replace only what this pull re-observed — re-pulled tokens, plus + # any legacy (pre-v2, NULL-asset) rows of the re-pulled markets they + # supersede. Rows older than the rolling pull window survive, so per-wallet + # history now accumulates instead of being overwritten each refresh. + assets = [b["asset"] for b in bets if b.get("asset")] + conds = list({b["cond"] for b in bets if b.get("cond")}) + _con.execute( + """DELETE FROM bets WHERE wallet = ? + AND (asset IN (SELECT UNNEST(?::VARCHAR[])) + OR (asset IS NULL AND cond IN (SELECT UNNEST(?::VARCHAR[]))))""", + [wallet, assets, conds]) if bets: _con.executemany( - "INSERT INTO bets(wallet,cond,won,p,res_t,size) VALUES (?,?,?,?,?,?)", - [(wallet, b["cond"], b["won"], b["p"], b.get("res_t"), b.get("size")) - for b in bets]) + "INSERT INTO bets(wallet,cond,asset,won,p,res_t,size,src,ts,resolved) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + [(wallet, b["cond"], b.get("asset"), b["won"], b.get("p"), + b.get("res_t"), b.get("size"), b.get("src"), b.get("ts"), + b.get("resolved")) for b in bets]) _con.execute("INSERT OR REPLACE INTO pulled VALUES (?,?)", [wallet, int(now)]) - return bets + rows = _con.execute( + "SELECT won,p,cond,res_t,size,asset,src,ts,resolved " + "FROM bets WHERE wallet=?", [wallet]).fetchall() + return [_bet_row(*row) for row in rows] def invalidate(wallets): diff --git a/oos.py b/oos.py index 7a919406..d4bf3904 100644 --- a/oos.py +++ b/oos.py @@ -28,6 +28,8 @@ def score_pre(wallet): if SEL_T0 <= b["res_t"] <= SEL_T1] if len(bets) < MIN_BETS: return None + for b in bets: # v2 returns raw p — clamp for the z math + b["p"] = max(0.001, min(0.999, b["p"] or 0)) wins = sum(1 for b in bets if b["won"]) exp = sum(b["p"] for b in bets) var = sum(b["p"] * (1 - b["p"]) for b in bets) or 1e-9