""" 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()