cache schema v2: token-keyed upsert archive, provenance, raw prices, honest failures

- bets table gains asset (token id = position identity), src/ts (endpoint
  provenance + close time), resolved (False = early-sold in an unended market;
  won is a curPrice mark). Auto-migrates v1 in place (~8s, 3,697 exact dupes
  merged); legacy rows carry NULLs until their wallet refreshes.
- refresh is now an upsert by token instead of a wallet wipe: rows sliding out
  of the rolling pull window survive, so per-wallet history accumulates into a
  permanent archive. Same-asset rows from both endpoints (partially-closed
  positions) dedupe to the larger-stake row - kills the two-endpoint
  double-count class (~35k suspect pairs found in the audit).
- p stored raw (0 = avgPrice missing), clamped on read by get_bets, so missing
  prices stay distinguishable from real 0.1c longshots; insider CLI + oos clamp
  their own direct use.
- resolved_bets(strict=True): a failed page raises instead of returning a
  silently truncated history; get_bets no longer caches or marks failed pulls
  (pre-v2 an API error cached the wallet as empty-and-fresh for 14 days -
  Kruto2027 was a live victim of this last night).

Verified: migration 18,289,320 -> 18,285,623 rows; forced refreshes of two
sharps show 0 same-asset dups, 0 legacy/new mixing, clamped reads, conviction
stats intact (Kruto conv win 73%).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

insider.py

@@ -47,8 +47,22 @@ def _parse_end(end):
     return 0
 
 
-def resolved_bets(wallet, cutoff, max_pages=40):
+def resolved_bets(wallet, cutoff, max_pages=40, strict=False):
-    """Resolved bets with entry price, conditionId, resolution time, size."""
+    """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)
 

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
+**Schema v2 (2026-07-02) — token-keyed, provenance-tagged, archival.** `bets`
-`DELETE FROM bets WHERE wallet=?` then re-inserts a fresh pull, and that pull is a
+now carries `asset` (token id — the position identity), `src`/`ts` (endpoint
-**rolling 180-day window** (`WINDOW_DAYS`) capped at ~2k bets/endpoint
+provenance + close time), and `resolved` (False = early-sold position in a
-(`max_pages`). So wallet *coverage* grows (new wallets are kept), but any single
+market that hadn't ended at pull time; its `won` is a curPrice *mark*, not an
-wallet's history is a capped, rolling, overwrite-on-refresh snapshot — bets older
+outcome — scorers filter these). `p` is stored **raw** (0 = avgPrice missing)
-than ~180d are dropped on the next re-pull. For a permanent long-horizon archive,
+and clamped to [0.001, 0.999] by `get_bets` on read, so "missing price" stays
-use the append-style `../wide/pmkt.duckdb` subgraph dataset instead, or change the
+distinguishable from a real 0.1¢ longshot. Refresh is an **upsert by token**
-pull to upsert + drop the cutoff.
+(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)
 

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,
+Each wallet's resolved bets are stored once in cache.duckdb. Because we keep
-size) are stored once in cache.duckdb. Because we keep res_t per bet, ANY date
+res_t per bet, ANY date cutoff — pre-June-1, full window, future experiments —
-cutoff — pre-June-1, full window, future experiments — reads the same cached
+reads the same cached rows and filters locally. A pull only happens for wallets
-rows and filters locally. A pull only happens for wallets not seen, or older
+not seen, or older than MAX_AGE_DAYS.
-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()
+                "SELECT won,p,cond,res_t,size,asset,src,ts,resolved "
-            return [{"won": w, "p": p, "cond": c, "res_t": rt, "size": s}
+                "FROM bets WHERE wallet=?", [wallet]).fetchall()
-                    for w, p, c, rt, s in rows]
+            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 (?,?,?,?,?,?)",
+                "INSERT INTO bets(wallet,cond,asset,won,p,res_t,size,src,ts,resolved) "
-                [(wallet, b["cond"], b["won"], b["p"], b.get("res_t"), b.get("size"))
+                "VALUES (?,?,?,?,?,?,?,?,?,?)",
-                 for b in bets])
+                [(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):

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