chemavx/polymarket-bot
1
0
Fork 0
Code Issues Pull Requests 10 Actions Packages Projects Releases 1 Wiki Activity

feat(replay): R1 replay core — clock injection + replay of archived cycles

Browse Source 
Re-executes BayesianStrategy.evaluate() over the R0 archive and stores
results in replay_runs/replay_decisions, tagged with git sha + a hash of
the strategy constants (same hash vs archive = determinism check,
different hash = counterfactual run).

- bayesian.py: optional as_of param on evaluate()/_days_to_resolution()
  (clock injection; default None = wall clock, prod behavior unchanged —
  the only touch to frozen code, purely additive)
- bot/replay.py: replay engine + CLI (python -m bot.replay --from --to);
  ReplayNews feeds archived sentiment back (GNews never called, per-cycle
  budget bypassed — archived sentiment already encodes it); manifold/db
  not wired (observational-only in prod); recorded-vs-replayed compare
  at 1e-9 tolerance
- schema.sql: replay_runs + replay_decisions (+ indexes), idempotent
- db.py: 6 replay accessors/writers
- tests: 19 new round-trip fidelity tests (104 total green)

Validated against a real prod cycle (2026-07-02T14:03:15Z, 46 markets,
4 skip paths incl. the Georgia confidence record): 46/46 matched,
max float delta 0.0.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
chemavx
2026-07-02 14:05:25 +00:00
co-authored by Claude Fable 5
parent eb4f67414a
commit 0ac48ba7f8
5 changed files with 919 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot/data/db.py
+79
View File
@@ -747,6 +747,85 @@ class Database:
except (ValueError, IndexError):
return 0
# ── Replay R1: replay core ───────────────────────────────────────────────
async def get_replay_cycles(self, from_ts, to_ts) -> list:
"""Return the cycle_ts values with archived decisions in [from_ts, to_ts)."""
async with self._pool.acquire() as conn:
rows = await conn.fetch("""
SELECT DISTINCT cycle_ts FROM signals
WHERE cycle_ts >= $1 AND cycle_ts < $2
ORDER BY cycle_ts
""", from_ts, to_ts)
return [r["cycle_ts"] for r in rows]
async def get_ext_snapshot(self, cycle_ts) -> Optional[dict]:
"""Return one cycle's ExternalSignals snapshot, or None if missing."""
async with self._pool.acquire() as conn:
row = await conn.fetchrow(
"SELECT * FROM ext_snapshots WHERE cycle_ts = $1", cycle_ts
)
return dict(row) if row else None
async def get_cycle_signal_rows(self, cycle_ts) -> list[dict]:
"""Return one cycle's archived decision rows in original evaluation
order (id = insertion order = the order main.py evaluated them)."""
async with self._pool.acquire() as conn:
rows = await conn.fetch(
"SELECT * FROM signals WHERE cycle_ts = $1 ORDER BY id", cycle_ts
)
return [dict(r) for r in rows]
async def get_markets_by_ids(self, market_ids: list[str]) -> dict[str, dict]:
"""Return market metadata rows keyed by id (for Market reconstruction)."""
if not market_ids:
return {}
async with self._pool.acquire() as conn:
rows = await conn.fetch(
"SELECT * FROM markets WHERE id = ANY($1::text[])", market_ids
)
return {r["id"]: dict(r) for r in rows}
async def save_replay_run(self, run: dict) -> None:
async with self._pool.acquire() as conn:
await conn.execute("""
INSERT INTO replay_runs (
run_id, git_sha, config_hash, config_json,
from_ts, to_ts, cycles, decisions, matched, mismatched, note
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
""",
run["run_id"], run["git_sha"], run["config_hash"],
run["config_json"], run["from_ts"], run["to_ts"],
run["cycles"], run["decisions"], run["matched"],
run["mismatched"], run["note"],
)
async def save_replay_decisions(self, run_id: str, decisions: list[dict]) -> None:
if not decisions:
return
rows = [
(
run_id, d["cycle_ts"], d["market_id"],
d["skip_reason"], d["prior_prob"], d["estimated_prob"],
d["raw_final_prob"], d["edge_gross"], d["edge_net"],
d["regime_min_edge"], d["days_to_resolution"],
d["confidence"], d["direction"], d["would_trade"],
d["recorded_skip_reason"], d["matched"], d["mismatch_field"],
)
for d in decisions
]
async with self._pool.acquire() as conn:
await conn.executemany("""
INSERT INTO replay_decisions (
run_id, cycle_ts, market_id,
skip_reason, prior_prob, estimated_prob,
raw_final_prob, edge_gross, edge_net,
regime_min_edge, days_to_resolution,
confidence, direction, would_trade,
recorded_skip_reason, matched, mismatch_field
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
""", rows)
async def mark_manifold_audit_used(self, audit_id: str) -> None:
async with self._pool.acquire() as conn:
await conn.execute(
bot/data/schema.sql
+61
View File
@@ -370,3 +370,64 @@ CREATE TABLE IF NOT EXISTS ext_snapshots (
total_market_cap_change DOUBLE PRECISION,
valid BOOLEAN
);
-- ─────────────────────────────────────────────────────────────────────────────
-- Replay R1: replay core — re-execute evaluate() over the R0 archive
--
-- A replay run reads cycles from signals + ext_snapshots + markets, rebuilds
-- the exact inputs (including archived news_sentiment — GNews is never called),
-- re-runs BayesianStrategy.evaluate() with the archived cycle_ts as clock, and
-- writes one replay_decisions row per (cycle, market).
--
-- replay_runs tags every run with the code (git_sha) and strategy constants
-- (config_hash) that produced it: two runs over the same window with different
-- config_hash values are a counterfactual comparison; same config_hash against
-- the recorded rows is a determinism check (mismatches should be 0, modulo
-- day-boundary crossings between cycle_ts and the original wall-clock).
--
-- matched: replayed decision equals the recorded one (skip_reason, probs,
-- confidence, direction). NULL when not comparable — e.g. reentry_guard
-- rows, recorded outside evaluate() with no decision fields to compare;
-- the replay still re-evaluates them, which is extra calibration data.
-- mismatch_field: first field that differed, for triage.
-- ─────────────────────────────────────────────────────────────────────────────
CREATE TABLE IF NOT EXISTS replay_runs (
run_id TEXT PRIMARY KEY,
created_at TIMESTAMPTZ DEFAULT NOW(),
git_sha TEXT,
config_hash TEXT,
config_json TEXT,
from_ts TIMESTAMPTZ,
to_ts TIMESTAMPTZ,
cycles INTEGER,
decisions INTEGER,
matched INTEGER,
mismatched INTEGER,
note TEXT
);
CREATE TABLE IF NOT EXISTS replay_decisions (
id SERIAL PRIMARY KEY,
run_id TEXT NOT NULL,
cycle_ts TIMESTAMPTZ NOT NULL,
market_id TEXT NOT NULL,
-- replayed outputs (same semantics as the signals columns)
skip_reason TEXT,
prior_prob DOUBLE PRECISION,
estimated_prob DOUBLE PRECISION,
raw_final_prob DOUBLE PRECISION,
edge_gross DOUBLE PRECISION,
edge_net DOUBLE PRECISION,
regime_min_edge DOUBLE PRECISION,
days_to_resolution INTEGER,
confidence DOUBLE PRECISION,
direction TEXT,
would_trade BOOLEAN,
-- fidelity vs the recorded signals row
recorded_skip_reason TEXT,
matched BOOLEAN,
mismatch_field TEXT
);
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);
bot/replay.py
+394
View File
@@ -0,0 +1,394 @@
"""
Replay R1 — replay core.
Re-executes BayesianStrategy.evaluate() over the R0 archive (signals +
ext_snapshots + markets) and stores the outcome in replay_runs /
replay_decisions.
Determinism contract: evaluate() is a pure function of
(market, ext, occupied_families, as_of) plus the news client, so a replay
rebuilds exactly those four inputs from the archive:
market — metadata from `markets`, per-cycle price/volume from `signals`
ext — the cycle's `ext_snapshots` row
families — a family-skipped row replays with its own family_key occupied;
every other row replays with no occupancy (the recorded
skip_reason already reflects the original portfolio state)
as_of — the archived cycle_ts (clock injection, Replay R1)
GNews is never called: ReplayNews feeds back the archived news_sentiment.
The per-cycle query budget is bypassed (reset before every market) because
the archived sentiment already encodes the budget's effect — a
budget-skipped market was recorded with sentiment 0.0.
Manifold and the DB are not wired into the replayed strategy (manifold=None,
db=None): the signal is observational-only in production (feat_mfld_lo is
always 0.0 in the archive), so the replay reproduces decisions without
touching cooldowns or audit tables. If MANIFOLD_SIGNAL_ENABLED is ever
turned on, replayed decisions will diverge from recorded ones and the
matched/mismatch_field columns will say so.
Run tagging: every run stores the git sha and a hash of the strategy
constants. Same config_hash vs the archive = determinism check (expect 0
mismatches, modulo UTC-day-boundary crossings between cycle_ts and the
original wall-clock). Different config_hash = counterfactual run.
CLI:
python -m bot.replay --from 2026-07-02T00:00:00Z --to 2026-07-03 --note "..."
"""
import argparse
import asyncio
import hashlib
import json
import logging
import os
import subprocess
import uuid
from collections import Counter
from datetime import datetime, timedelta, timezone
from typing import Optional
import bot.strategy.bayesian as bayesian
from bot.data.db import Database
from bot.data.external import ExternalSignals
from bot.data.polymarket import Market
from bot.strategy.bayesian import BayesianStrategy
log = logging.getLogger(__name__)
# Absolute float tolerance for recorded-vs-replayed comparison. Archived
# values are float8 (exact IEEE-754 round-trip of Python floats), so any real
# divergence is far larger than this.
FLOAT_TOL = 1e-9
# Strategy constants that define a replay configuration. Hashed into
# replay_runs.config_hash; read from the module at call time so a
# counterfactual run can monkeypatch them and be tagged distinctly.
CONFIG_KEYS = (
"SPREAD_ESTIMATE",
"COMMISSION_RATE",
"MIN_CONFIDENCE",
"NEWS_LOGODDS_WEIGHT",
"MANIFOLD_LOGODDS_WEIGHT",
"MANIFOLD_SIGNAL_ENABLED",
"NEWS_GUARDRAIL_ENABLED",
"MAX_NEWS_ONLY_PROB_SHIFT",
"NEWS_MATERIAL_LOGODDS_THRESHOLD",
"MAX_NEWS_QUERIES_PER_CYCLE",
)
# Rows recorded outside evaluate() (via record_skip) carry no decision fields;
# the replay still re-evaluates them for calibration but cannot compare.
NON_COMPARABLE_SKIPS = {"reentry_guard"}
def strategy_config() -> dict:
return {k: getattr(bayesian, k) for k in CONFIG_KEYS}
def strategy_config_hash() -> str:
blob = json.dumps(strategy_config(), sort_keys=True)
return hashlib.sha256(blob.encode()).hexdigest()[:12]
def _git_sha() -> str:
sha = os.getenv("GIT_SHA", "")
if sha:
return sha
try:
return subprocess.run(
["git", "rev-parse", "--short", "HEAD"],
capture_output=True, text=True, timeout=5,
).stdout.strip() or "unknown"
except (OSError, subprocess.SubprocessError):
return "unknown"
class ReplayNews:
"""NewsClient stand-in that feeds archived sentiment back into evaluate().
No HTTP, no cache: the engine sets `sentiment` to the archived value
before each evaluate() call. Values below evaluate()'s 0.05 materiality
threshold were archived as 0.0, so the round-trip is exact.
"""
enabled = True
def __init__(self) -> None:
self.sentiment: float = 0.0
async def get_sentiment(self, question: str) -> float:
return self.sentiment
def get_freshness(self, question: str) -> float:
return 1.0 # only used by gnews_priority(), which replay never calls
def build_ext(snapshot: dict) -> ExternalSignals:
"""Rebuild the ExternalSignals a cycle was evaluated against."""
return ExternalSignals(
btc_price=snapshot["btc_price"],
btc_change_24h=snapshot["btc_change_24h"],
eth_price=snapshot["eth_price"],
eth_change_24h=snapshot["eth_change_24h"],
btc_dominance=snapshot["btc_dominance"],
fear_greed_index=snapshot["fear_greed_index"],
fear_greed_label=snapshot["fear_greed_label"],
total_market_cap_change=snapshot["total_market_cap_change"],
valid=snapshot["valid"],
)
def build_market(market_row: dict, signal_row: dict) -> Market:
"""Rebuild a Market: metadata from `markets`, per-cycle state from `signals`.
Token ids are irrelevant to evaluate() and left empty; no_price is the
YES complement (evaluate() never reads it either).
"""
yes_price = signal_row["polymarket_price"]
return Market(
id=market_row["id"],
condition_id=market_row["condition_id"] or "",
question=market_row["question"],
yes_token_id="",
no_token_id="",
yes_price=yes_price,
no_price=1.0 - yes_price,
volume_24h=signal_row["volume_24h"] or 0.0,
end_date=market_row["end_date"] or "",
active=True,
category=signal_row["category"] or (market_row["category"] or ""),
)
def _compare(recorded: dict, replayed: dict) -> Optional[str]:
"""Return the first field where replayed diverges from recorded, or None."""
if recorded["skip_reason"] != replayed["skip_reason"]:
return "skip_reason"
for field in ("prior_prob", "estimated_prob", "raw_final_prob",
"edge_net", "confidence"):
a, b = recorded[field], replayed[field]
if a is None and b is None:
continue
if a is None or b is None or abs(a - b) > FLOAT_TOL:
return field
if recorded["direction"] != replayed["direction"]:
return "direction"
return None
async def replay_cycle(
cycle_ts: datetime,
snapshot: dict,
signal_rows: list[dict],
market_rows: dict[str, dict],
) -> list[dict]:
"""Re-evaluate one archived cycle; returns one decision dict per row.
Pure with respect to the DB — everything it needs is passed in, so tests
can drive it with synthetic rows.
"""
news = ReplayNews()
strategy = BayesianStrategy(news=news, manifold=None, db=None)
ext = build_ext(snapshot)
decisions: list[dict] = []
for row in signal_rows:
recorded_skip = row["skip_reason"]
decision = {
"cycle_ts": cycle_ts,
"market_id": row["market_id"],
"skip_reason": None,
"prior_prob": None,
"estimated_prob": None,
"raw_final_prob": None,
"edge_gross": None,
"edge_net": None,
"regime_min_edge": None,
"days_to_resolution": None,
"confidence": None,
"direction": None,
"would_trade": None,
"recorded_skip_reason": recorded_skip,
"matched": None,
"mismatch_field": None,
}
market_row = market_rows.get(row["market_id"])
if market_row is None:
# Should not happen (R0 upserts markets every cycle) — record the
# gap instead of crashing the run.
decision["matched"] = False
decision["mismatch_field"] = "market_missing"
decisions.append(decision)
continue
market = build_market(market_row, row)
# A family-skipped row replays against its own occupied family; all
# other rows replay unoccupied — their recorded skip_reason already
# reflects whatever portfolio state existed, and evaluate() checks
# the family gate before anything portfolio-dependent.
families = (
{row["family_key"]}
if recorded_skip == "family" and row["family_key"]
else set()
)
news.sentiment = row["news_sentiment"] or 0.0
# Bypass the per-cycle GNews budget: archived sentiment already
# encodes it (budget-skipped markets were recorded with 0.0).
strategy._news_queries_this_cycle = 0
signal = await strategy.evaluate(market, ext, families, as_of=cycle_ts)
rec = strategy.drain_cycle_records()[-1]
decision.update(
skip_reason=rec["skip_reason"],
prior_prob=rec["prior_prob"],
estimated_prob=rec["estimated_prob"],
raw_final_prob=rec["raw_final_prob"],
edge_gross=rec["edge_gross"],
edge_net=rec["edge_net"],
regime_min_edge=rec["regime_min_edge"],
days_to_resolution=rec["days_to_resolution"],
confidence=rec["confidence"],
direction=rec["direction"],
would_trade=signal is not None,
)
if recorded_skip in NON_COMPARABLE_SKIPS:
decision["matched"] = None # re-evaluated for calibration only
else:
mismatch = _compare(row, rec)
decision["matched"] = mismatch is None
decision["mismatch_field"] = mismatch
decisions.append(decision)
return decisions
async def run_replay(
db: Database,
from_ts: datetime,
to_ts: datetime,
note: str = "",
limit_cycles: Optional[int] = None,
) -> dict:
"""Replay every archived cycle in [from_ts, to_ts) and persist the run.
Returns the replay_runs row (plus a mismatch_fields Counter) for reporting.
"""
run_id = str(uuid.uuid4())
cycles = await db.get_replay_cycles(from_ts, to_ts)
if limit_cycles:
cycles = cycles[:limit_cycles]
decisions_total = 0
matched = 0
mismatched = 0
mismatch_fields: Counter = Counter()
skipped_cycles = 0
for cycle_ts in cycles:
snapshot = await db.get_ext_snapshot(cycle_ts)
if snapshot is None:
skipped_cycles += 1
log.warning("Replay: no ext_snapshot for cycle %s — skipped", cycle_ts)
continue
signal_rows = await db.get_cycle_signal_rows(cycle_ts)
market_rows = await db.get_markets_by_ids(
[r["market_id"] for r in signal_rows]
)
decisions = await replay_cycle(cycle_ts, snapshot, signal_rows, market_rows)
await db.save_replay_decisions(run_id, decisions)
decisions_total += len(decisions)
for d in decisions:
if d["matched"] is True:
matched += 1
elif d["matched"] is False:
mismatched += 1
mismatch_fields[d["mismatch_field"]] += 1
run = {
"run_id": run_id,
"git_sha": _git_sha(),
"config_hash": strategy_config_hash(),
"config_json": json.dumps(strategy_config(), sort_keys=True),
"from_ts": from_ts,
"to_ts": to_ts,
"cycles": len(cycles) - skipped_cycles,
"decisions": decisions_total,
"matched": matched,
"mismatched": mismatched,
"note": note,
}
await db.save_replay_run(run)
run["mismatch_fields"] = dict(mismatch_fields)
run["skipped_cycles"] = skipped_cycles
return run
def _parse_ts(value: str) -> datetime:
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt
async def _amain(args: argparse.Namespace) -> None:
db = Database()
await db.connect()
try:
run = await run_replay(
db,
from_ts=args.from_ts,
to_ts=args.to_ts,
note=args.note,
limit_cycles=args.limit_cycles,
)
finally:
await db.disconnect()
comparable = run["matched"] + run["mismatched"]
print(f"run_id : {run['run_id']}")
print(f"git_sha : {run['git_sha']} config_hash: {run['config_hash']}")
print(f"window : {run['from_ts'].isoformat()} → {run['to_ts'].isoformat()}")
print(f"cycles : {run['cycles']} (skipped: {run['skipped_cycles']})")
print(f"decisions : {run['decisions']} ({comparable} comparable)")
print(f"matched : {run['matched']}")
print(f"mismatched : {run['mismatched']}")
if run["mismatch_fields"]:
for field, count in sorted(run["mismatch_fields"].items(), key=lambda x: -x[1]):
print(f" {field}: {count}")
def main() -> None:
parser = argparse.ArgumentParser(
prog="python -m bot.replay",
description="Replay archived trading cycles through the current strategy.",
)
now = datetime.now(timezone.utc)
parser.add_argument(
"--from", dest="from_ts", type=_parse_ts,
default=now - timedelta(hours=24),
help="window start, ISO-8601 (default: 24h ago)",
)
parser.add_argument(
"--to", dest="to_ts", type=_parse_ts, default=now,
help="window end, ISO-8601, exclusive (default: now)",
)
parser.add_argument("--note", default="", help="free-text tag for replay_runs")
parser.add_argument(
"--limit-cycles", type=int, default=None,
help="replay at most N cycles (smoke runs)",
)
args = parser.parse_args()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
)
# evaluate() logs one INFO line per market — thousands per replay window.
logging.getLogger("bot.strategy.bayesian").setLevel(logging.WARNING)
asyncio.run(_amain(args))
if __name__ == "__main__":
main()
bot/strategy/bayesian.py
+18 -4
View File
@@ -167,15 +167,22 @@ def _regime_min_edge(category: str, days_to_resolution: int) -> float:
return 0.10 # tech, crypto/finance, events, default
def _days_to_resolution(end_date: str) -> int:
"""Return calendar days until market resolution, or 30 if unknown."""
def _days_to_resolution(end_date: str, as_of: Optional[datetime] = None) -> int:
"""Return calendar days until market resolution, or 30 if unknown.
as_of (Replay R1): reference clock for the computation. None (production)
means wall-clock now; a replay run passes the archived cycle_ts so
days-to-resolution — and therefore the regime edge threshold — is computed
against the moment the decision was originally made.
"""
if not end_date:
return 30 # conservative: treat as medium-term
try:
dt = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
days = (dt - datetime.now(timezone.utc)).days
now = as_of if as_of is not None else datetime.now(timezone.utc)
days = (dt - now).days
return max(0, days)
except (ValueError, TypeError):
return 30
@@ -457,10 +464,17 @@ class BayesianStrategy:
market: Market,
ext: ExternalSignals,
occupied_families: set[str],
as_of: Optional[datetime] = None,
) -> Optional[TradingSignal]:
"""
Evaluate a market and return a TradingSignal if actionable.
as_of (Replay R1): clock injection — None in production (wall-clock
now); a replay passes the archived cycle_ts so the regime threshold
matches the original decision moment. Only days-to-resolution
depends on the clock; everything else is a pure function of
(market, ext, occupied_families) and the news/manifold clients.
Returns None with a structured log line in all skip cases.
Skip reasons (Phase 5 observability):
SKIP_UNSUPPORTED — category not supported
@@ -558,7 +572,7 @@ class BayesianStrategy:
return None
# ── Phase 4: regime min-edge ─────────────────────────────────────────
days = _days_to_resolution(market.end_date)
days = _days_to_resolution(market.end_date, as_of)
regime_min = _regime_min_edge(category, days)
# ── Bayesian probability estimation ──────────────────────────────────