diff --git a/bot/data/db.py b/bot/data/db.py index 135f620..3a372a5 100644 --- a/bot/data/db.py +++ b/bot/data/db.py @@ -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( diff --git a/bot/data/schema.sql b/bot/data/schema.sql index 3506db4..34eed30 100644 --- a/bot/data/schema.sql +++ b/bot/data/schema.sql @@ -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() +); diff --git a/bot/outcomes.py b/bot/outcomes.py new file mode 100644 index 0000000..ed60517 --- /dev/null +++ b/bot/outcomes.py @@ -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() diff --git a/tests/test_outcomes.py b/tests/test_outcomes.py new file mode 100644 index 0000000..ea4caf3 --- /dev/null +++ b/tests/test_outcomes.py @@ -0,0 +1,174 @@ +"""Replay R2 tests — outcome fetching and calibration scoring.""" +import asyncio +import math + +import pytest + +from bot.data.polymarket import MarketResolution +from bot.outcomes import ( + LOGLOSS_EPS, + compute_calibration, + fetch_outcomes, + print_report, +) +from datetime import datetime, timezone + + +class FakePoly: + """get_market_resolution stand-in driven by a dict of canned responses.""" + + def __init__(self, responses: dict): + self.responses = responses + self.calls: list[str] = [] + + async def get_market_resolution(self, market_id: str): + self.calls.append(market_id) + return self.responses.get(market_id) + + +RESOLVED_AT = datetime(2026, 7, 1, 12, 0, tzinfo=timezone.utc) + + +def _row(market_id="m1", category="politics", est=0.6, prior=0.5, outcome=1.0): + return { + "market_id": market_id, + "category": category, + "estimated_prob": est, + "prior_prob": prior, + "outcome": outcome, + } + + +# ── fetch_outcomes ─────────────────────────────────────────────────────────── + +def test_fetch_keeps_only_definitive_resolutions(): + poly = FakePoly({ + "yes": MarketResolution(resolved=True, resolution=1.0, + resolved_at=RESOLVED_AT), + "no": MarketResolution(resolved=True, resolution=0.0, + resolved_at=None), + "open": MarketResolution(resolved=False), + "disputed": MarketResolution(resolved=False), + "apierror": None, # get_market_resolution returns None on HTTP errors + }) + out = asyncio.run( + fetch_outcomes(poly, ["yes", "no", "open", "disputed", "apierror"]) + ) + assert poly.calls == ["yes", "no", "open", "disputed", "apierror"] + assert out == [ + {"market_id": "yes", "outcome": 1.0, "resolved_at": RESOLVED_AT}, + {"market_id": "no", "outcome": 0.0, "resolved_at": None}, + ] + + +def test_fetch_empty_list_is_noop(): + poly = FakePoly({}) + assert asyncio.run(fetch_outcomes(poly, [])) == [] + assert poly.calls == [] + + +# ── compute_calibration ────────────────────────────────────────────────────── + +def test_no_rows_returns_none(): + assert compute_calibration([]) is None + + +def test_single_row_known_values(): + m = compute_calibration([_row(est=0.8, prior=0.6, outcome=1.0)]) + assert m["n_evaluations"] == 1 + assert m["n_markets"] == 1 + assert m["brier_model"] == pytest.approx((0.8 - 1.0) ** 2) + assert m["brier_prior"] == pytest.approx((0.6 - 1.0) ** 2) + assert m["logloss_model"] == pytest.approx(-math.log(0.8)) + assert m["logloss_prior"] == pytest.approx(-math.log(0.6)) + # one market: macro == micro + assert m["brier_model_macro"] == pytest.approx(m["brier_model"]) + assert m["brier_prior_macro"] == pytest.approx(m["brier_prior"]) + + +def test_logloss_no_outcome_branch(): + m = compute_calibration([_row(est=0.2, prior=0.7, outcome=0.0)]) + assert m["logloss_model"] == pytest.approx(-math.log(0.8)) + assert m["logloss_prior"] == pytest.approx(-math.log(0.3)) + + +def test_logloss_clipping_keeps_hard_miss_finite(): + # A hard 1.0 estimate on a NO outcome must not produce inf. + m = compute_calibration([_row(est=1.0, prior=0.5, outcome=0.0)]) + assert math.isfinite(m["logloss_model"]) + assert m["logloss_model"] == pytest.approx(-math.log(LOGLOSS_EPS)) + + +def test_micro_weights_evaluations_macro_weights_markets(): + # Market a: 3 evaluations, model error 0.1; market b: 1 evaluation, error 0.5. + rows = [ + _row(market_id="a", est=0.9, prior=0.8, outcome=1.0), + _row(market_id="a", est=0.9, prior=0.8, outcome=1.0), + _row(market_id="a", est=0.9, prior=0.8, outcome=1.0), + _row(market_id="b", est=0.5, prior=0.6, outcome=1.0), + ] + m = compute_calibration(rows) + assert m["n_evaluations"] == 4 + assert m["n_markets"] == 2 + # micro: (3*0.01 + 0.25) / 4 ; macro: (0.01 + 0.25) / 2 + assert m["brier_model"] == pytest.approx((3 * 0.01 + 0.25) / 4) + assert m["brier_model_macro"] == pytest.approx((0.01 + 0.25) / 2) + assert m["brier_prior"] == pytest.approx((3 * 0.04 + 0.16) / 4) + assert m["brier_prior_macro"] == pytest.approx((0.04 + 0.16) / 2) + + +def test_model_beating_market_gives_negative_delta(): + # est closer to the outcome than the price on every row + rows = [ + _row(market_id="a", est=0.8, prior=0.6, outcome=1.0), + _row(market_id="b", est=0.3, prior=0.45, outcome=0.0), + ] + m = compute_calibration(rows) + assert m["brier_model"] < m["brier_prior"] + assert m["logloss_model"] < m["logloss_prior"] + + +def test_per_category_grouping_and_unknown(): + rows = [ + _row(market_id="a", category="politics", est=0.8, prior=0.6, outcome=1.0), + _row(market_id="b", category="politics", est=0.7, prior=0.6, outcome=1.0), + _row(market_id="c", category=None, est=0.4, prior=0.5, outcome=0.0), + ] + m = compute_calibration(rows) + assert set(m["per_category"]) == {"politics", "unknown"} + pol = m["per_category"]["politics"] + assert pol["n"] == 2 and pol["markets"] == 2 + assert pol["brier_model"] == pytest.approx((0.04 + 0.09) / 2) + unk = m["per_category"]["unknown"] + assert unk["n"] == 1 and unk["markets"] == 1 + assert unk["brier_model"] == pytest.approx(0.16) + + +def test_repeated_market_counts_once_in_markets(): + rows = [ + _row(market_id="a", est=0.8, prior=0.6, outcome=1.0), + _row(market_id="a", est=0.7, prior=0.55, outcome=1.0), + ] + m = compute_calibration(rows) + assert m["n_markets"] == 1 + assert m["per_category"]["politics"]["markets"] == 1 + + +# ── print_report ───────────────────────────────────────────────────────────── + +def test_report_handles_no_metrics(capsys): + print_report(None, "R0 archive") + assert "no scorable rows yet" in capsys.readouterr().out + + +def test_report_prints_all_metric_lines(capsys): + m = compute_calibration([ + _row(market_id="a", est=0.8, prior=0.6, outcome=1.0), + _row(market_id="b", category=None, est=0.4, prior=0.5, outcome=0.0), + ]) + print_report(m, "R0 archive") + out = capsys.readouterr().out + assert "2 evaluations, 2 markets" in out + for label in ("Brier micro", "Brier macro", "logloss micro", + "politics", "unknown"): + assert label in out