feat(replay): R2 outcomes + calibration metrics
Scores every archived estimate against reality — the sample multiplier the phase plan calls for: Brier/log-loss of estimated_prob benchmarked against the market price (prior_prob) on the same rows, over ALL evaluations with a resolved outcome, not just executed trades. - schema.sql: market_outcomes (one row per resolved market; outcome = final YES price 1.0/0.0, UMA-final only) - bot/outcomes.py: CLI (python -m bot.outcomes) with two phases — fetch resolutions for archived markets via the existing get_market_resolution() (open/disputed/ambiguous markets simply retry next invocation; no data-loss urgency, Gamma reports past resolutions at any time), then compute calibration: Brier micro (per evaluation) / macro (per market — the honest sample size given ~1 eval/min autocorrelation), log-loss with 1e-9 clipping, per-category breakdown. --run-id scores a replay run's re-estimates instead of the archive (counterfactual calibration). - db.py: 4 accessors (pending markets, outcome upsert, coverage, calibration rows for archive or run) - tests: 12 new (116 total green); compute_calibration is a pure function driven by literals No prod behavior change: the bot never imports bot.outcomes; the only shared surface is the idempotent schema migration (dry-run BEGIN/ROLLBACK clean against prod). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
chemavx
co-authored by
Claude Fable 5
Claude Fable 5
parent
6c544e46e2
commit
124b6d8558
@@ -826,6 +826,75 @@ class Database:
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
|
||||
""", rows)
|
||||
|
||||
# ── Replay R2: outcomes + calibration metrics ────────────────────────────
|
||||
|
||||
async def get_unresolved_archived_market_ids(self) -> list[str]:
|
||||
"""Archived markets (present in signals) with no stored outcome yet."""
|
||||
async with self._pool.acquire() as conn:
|
||||
rows = await conn.fetch("""
|
||||
SELECT DISTINCT s.market_id FROM signals s
|
||||
LEFT JOIN market_outcomes o ON o.market_id = s.market_id
|
||||
WHERE o.market_id IS NULL
|
||||
ORDER BY s.market_id
|
||||
""")
|
||||
return [r["market_id"] for r in rows]
|
||||
|
||||
async def upsert_market_outcome(
|
||||
self, market_id: str, outcome: float, resolved_at
|
||||
) -> None:
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute("""
|
||||
INSERT INTO market_outcomes (market_id, outcome, resolved_at)
|
||||
VALUES ($1, $2, $3)
|
||||
ON CONFLICT (market_id) DO UPDATE
|
||||
SET outcome = EXCLUDED.outcome,
|
||||
resolved_at = EXCLUDED.resolved_at,
|
||||
fetched_at = NOW()
|
||||
""", market_id, outcome, resolved_at)
|
||||
|
||||
async def get_outcome_coverage(self) -> dict:
|
||||
"""How much of the archive is scorable: resolved vs archived markets."""
|
||||
async with self._pool.acquire() as conn:
|
||||
row = await conn.fetchrow("""
|
||||
SELECT
|
||||
(SELECT COUNT(DISTINCT market_id) FROM signals) AS archived,
|
||||
(SELECT COUNT(*) FROM market_outcomes
|
||||
WHERE market_id IN (SELECT DISTINCT market_id FROM signals)
|
||||
) AS resolved
|
||||
""")
|
||||
return dict(row)
|
||||
|
||||
async def get_calibration_rows(self, run_id: Optional[str] = None) -> list[dict]:
|
||||
"""Every archived evaluation with a full estimate AND a known outcome.
|
||||
|
||||
run_id None scores the R0 archive (signals); a run_id scores that
|
||||
replay run's re-estimates instead (counterfactual calibration).
|
||||
Rows without estimated_prob (skipped before estimation: prior_extreme,
|
||||
unsupported, family, no_signals) carry no model prediction to score.
|
||||
"""
|
||||
async with self._pool.acquire() as conn:
|
||||
if run_id is None:
|
||||
rows = await conn.fetch("""
|
||||
SELECT s.market_id, s.category,
|
||||
s.estimated_prob, s.prior_prob, o.outcome
|
||||
FROM signals s
|
||||
JOIN market_outcomes o ON o.market_id = s.market_id
|
||||
WHERE s.estimated_prob IS NOT NULL
|
||||
AND s.prior_prob IS NOT NULL
|
||||
""")
|
||||
else:
|
||||
rows = await conn.fetch("""
|
||||
SELECT d.market_id, m.category,
|
||||
d.estimated_prob, d.prior_prob, o.outcome
|
||||
FROM replay_decisions d
|
||||
JOIN market_outcomes o ON o.market_id = d.market_id
|
||||
LEFT JOIN markets m ON m.id = d.market_id
|
||||
WHERE d.run_id = $1
|
||||
AND d.estimated_prob IS NOT NULL
|
||||
AND d.prior_prob IS NOT NULL
|
||||
""", run_id)
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
async def mark_manifold_audit_used(self, audit_id: str) -> None:
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute(
|
||||
|
||||
@@ -431,3 +431,24 @@ CREATE TABLE IF NOT EXISTS replay_decisions (
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_replay_decisions_run ON replay_decisions(run_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_replay_decisions_mkt ON replay_decisions(market_id);
|
||||
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
-- Replay R2: outcomes + calibration metrics
|
||||
--
|
||||
-- One row per resolved market, fetched from the Gamma API via
|
||||
-- get_market_resolution() (UMA-final only: a market closed but still in
|
||||
-- proposal/dispute is not stored). outcome is the final YES price:
|
||||
-- 1.0 = YES won, 0.0 = NO won.
|
||||
--
|
||||
-- Joining signals (or replay_decisions) to market_outcomes scores every
|
||||
-- archived estimate against reality — Brier / log-loss of estimated_prob
|
||||
-- benchmarked against the market price (prior_prob) on the same rows,
|
||||
-- answering "does the model add value over the market?" across ALL
|
||||
-- evaluations, not just executed trades.
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
CREATE TABLE IF NOT EXISTS market_outcomes (
|
||||
market_id TEXT PRIMARY KEY,
|
||||
outcome DOUBLE PRECISION NOT NULL,
|
||||
resolved_at TIMESTAMPTZ,
|
||||
fetched_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
+208
@@ -0,0 +1,208 @@
|
||||
"""
|
||||
Replay R2 — outcomes + calibration metrics.
|
||||
|
||||
Two phases, one CLI:
|
||||
|
||||
1. Fetch: for every archived market (present in `signals`) without a stored
|
||||
outcome, ask the Gamma API via PolymarketClient.get_market_resolution()
|
||||
— the same UMA-finality gate the trading loop uses to settle positions.
|
||||
Definitive resolutions are upserted into `market_outcomes`; open, disputed
|
||||
or ambiguous markets are simply retried on the next invocation. There is
|
||||
no data-loss urgency here (unlike the R0 recorder): Gamma reports past
|
||||
resolutions at any time, so running this lazily loses nothing.
|
||||
|
||||
2. Score: join archived estimates to outcomes and compute Brier / log-loss of
|
||||
estimated_prob, benchmarked against the market price (prior_prob) on the
|
||||
same rows. This scores ALL evaluations with a full estimate — the sample
|
||||
multiplier the phase plan calls for — not just executed trades. With
|
||||
--run-id it scores a replay run's re-estimates instead (counterfactual
|
||||
calibration: did config X predict better than the market?).
|
||||
|
||||
Reading the numbers: lower is better for both metrics; model < prior means
|
||||
the model added information over the market price. Micro averages weight
|
||||
every evaluation equally, so long-lived markets (~1 evaluation/min while in
|
||||
the universe) dominate; macro averages score each market once (mean of its
|
||||
evaluations) and answer the same question per market. Evaluations of one
|
||||
market minutes apart are highly autocorrelated — n_evaluations overstates
|
||||
the effective sample size, n_markets is the honest one.
|
||||
|
||||
CLI:
|
||||
python -m bot.outcomes # fetch new outcomes, then score archive
|
||||
python -m bot.outcomes --fetch-only
|
||||
python -m bot.outcomes --metrics-only
|
||||
python -m bot.outcomes --run-id UUID # score a replay run (implies no fetch)
|
||||
"""
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
from collections import defaultdict
|
||||
from typing import Optional
|
||||
|
||||
from bot.data.db import Database
|
||||
from bot.data.polymarket import PolymarketClient
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Clip probabilities before log() so a (theoretical) hard 0/1 estimate on a
|
||||
# wrong outcome scores ~20.7 nats instead of infinity poisoning the mean.
|
||||
LOGLOSS_EPS = 1e-9
|
||||
|
||||
|
||||
async def fetch_outcomes(poly, market_ids: list[str]) -> list[dict]:
|
||||
"""Resolve archived markets against Gamma; returns only definitive ones.
|
||||
|
||||
Sequential on purpose: ~50 markets per invocation, and the Gamma API has
|
||||
no bulk endpoint. get_market_resolution() already returns None on API
|
||||
errors and resolved=False on open/disputed/ambiguous markets.
|
||||
"""
|
||||
resolved = []
|
||||
for market_id in market_ids:
|
||||
res = await poly.get_market_resolution(market_id)
|
||||
if res is None or not res.resolved or res.resolution is None:
|
||||
continue
|
||||
resolved.append({
|
||||
"market_id": market_id,
|
||||
"outcome": res.resolution,
|
||||
"resolved_at": res.resolved_at,
|
||||
})
|
||||
return resolved
|
||||
|
||||
|
||||
def _logloss(p: float, outcome: float) -> float:
|
||||
p = min(max(p, LOGLOSS_EPS), 1.0 - LOGLOSS_EPS)
|
||||
return -math.log(p) if outcome == 1.0 else -math.log(1.0 - p)
|
||||
|
||||
|
||||
def compute_calibration(rows: list[dict]) -> Optional[dict]:
|
||||
"""Score estimated_prob vs prior_prob against outcomes; None if no rows.
|
||||
|
||||
rows: dicts with market_id, category, estimated_prob, prior_prob, outcome.
|
||||
Pure function — the CLI feeds it DB rows, tests feed it literals.
|
||||
"""
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
n = len(rows)
|
||||
brier_model = sum((r["estimated_prob"] - r["outcome"]) ** 2 for r in rows) / n
|
||||
brier_prior = sum((r["prior_prob"] - r["outcome"]) ** 2 for r in rows) / n
|
||||
logloss_model = sum(_logloss(r["estimated_prob"], r["outcome"]) for r in rows) / n
|
||||
logloss_prior = sum(_logloss(r["prior_prob"], r["outcome"]) for r in rows) / n
|
||||
|
||||
by_market: dict[str, list[dict]] = defaultdict(list)
|
||||
for r in rows:
|
||||
by_market[r["market_id"]].append(r)
|
||||
market_briers = [
|
||||
(
|
||||
sum((r["estimated_prob"] - r["outcome"]) ** 2 for r in mrows) / len(mrows),
|
||||
sum((r["prior_prob"] - r["outcome"]) ** 2 for r in mrows) / len(mrows),
|
||||
)
|
||||
for mrows in by_market.values()
|
||||
]
|
||||
brier_model_macro = sum(b[0] for b in market_briers) / len(market_briers)
|
||||
brier_prior_macro = sum(b[1] for b in market_briers) / len(market_briers)
|
||||
|
||||
by_category: dict[str, list[dict]] = defaultdict(list)
|
||||
for r in rows:
|
||||
by_category[r["category"] or "unknown"].append(r)
|
||||
per_category = {
|
||||
cat: {
|
||||
"n": len(crows),
|
||||
"markets": len({r["market_id"] for r in crows}),
|
||||
"brier_model": sum((r["estimated_prob"] - r["outcome"]) ** 2
|
||||
for r in crows) / len(crows),
|
||||
"brier_prior": sum((r["prior_prob"] - r["outcome"]) ** 2
|
||||
for r in crows) / len(crows),
|
||||
}
|
||||
for cat, crows in sorted(by_category.items())
|
||||
}
|
||||
|
||||
return {
|
||||
"n_evaluations": n,
|
||||
"n_markets": len(by_market),
|
||||
"brier_model": brier_model,
|
||||
"brier_prior": brier_prior,
|
||||
"brier_model_macro": brier_model_macro,
|
||||
"brier_prior_macro": brier_prior_macro,
|
||||
"logloss_model": logloss_model,
|
||||
"logloss_prior": logloss_prior,
|
||||
"per_category": per_category,
|
||||
}
|
||||
|
||||
|
||||
def print_report(metrics: Optional[dict], source: str) -> None:
|
||||
if metrics is None:
|
||||
print(f"calibration : no scorable rows yet for {source} "
|
||||
"(no archived estimate has a resolved outcome)")
|
||||
return
|
||||
print(f"calibration : {source} — {metrics['n_evaluations']} evaluations, "
|
||||
f"{metrics['n_markets']} markets")
|
||||
print(f"{'':14s}{'model':>10s}{'market':>10s}{'delta':>10s}")
|
||||
for label, m_key, p_key in (
|
||||
("Brier micro", "brier_model", "brier_prior"),
|
||||
("Brier macro", "brier_model_macro", "brier_prior_macro"),
|
||||
("logloss micro", "logloss_model", "logloss_prior"),
|
||||
):
|
||||
m, p = metrics[m_key], metrics[p_key]
|
||||
print(f" {label:12s}{m:>10.4f}{p:>10.4f}{m - p:>+10.4f}")
|
||||
print(" (delta < 0 = model beats the market price)")
|
||||
for cat, c in metrics["per_category"].items():
|
||||
print(f" {cat:12s}n={c['n']:<6d} markets={c['markets']:<3d} "
|
||||
f"brier model {c['brier_model']:.4f} vs market {c['brier_prior']:.4f}")
|
||||
|
||||
|
||||
async def _amain(args: argparse.Namespace) -> None:
|
||||
db = Database()
|
||||
await db.connect()
|
||||
try:
|
||||
if not args.metrics_only and args.run_id is None:
|
||||
pending = await db.get_unresolved_archived_market_ids()
|
||||
poly = PolymarketClient()
|
||||
try:
|
||||
resolved = await fetch_outcomes(poly, pending)
|
||||
finally:
|
||||
await poly.close()
|
||||
for out in resolved:
|
||||
await db.upsert_market_outcome(
|
||||
out["market_id"], out["outcome"], out["resolved_at"]
|
||||
)
|
||||
print(f"outcomes : {len(resolved)} newly resolved "
|
||||
f"(of {len(pending)} pending markets checked)")
|
||||
|
||||
coverage = await db.get_outcome_coverage()
|
||||
print(f"coverage : {coverage['resolved']}/{coverage['archived']} "
|
||||
"archived markets resolved")
|
||||
|
||||
if args.fetch_only:
|
||||
return
|
||||
rows = await db.get_calibration_rows(run_id=args.run_id)
|
||||
source = f"replay run {args.run_id}" if args.run_id else "R0 archive"
|
||||
print_report(compute_calibration(rows), source)
|
||||
finally:
|
||||
await db.disconnect()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="python -m bot.outcomes",
|
||||
description="Fetch market resolutions and score archived estimates.",
|
||||
)
|
||||
parser.add_argument("--fetch-only", action="store_true",
|
||||
help="only fetch/store outcomes, skip metrics")
|
||||
parser.add_argument("--metrics-only", action="store_true",
|
||||
help="skip the Gamma fetch, score what is stored")
|
||||
parser.add_argument("--run-id", default=None,
|
||||
help="score a replay run's re-estimates instead of "
|
||||
"the R0 archive (implies --metrics-only)")
|
||||
args = parser.parse_args()
|
||||
if args.fetch_only and args.metrics_only:
|
||||
parser.error("--fetch-only and --metrics-only are mutually exclusive")
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
)
|
||||
asyncio.run(_amain(args))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user