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