bot/data/db.py

@@ -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

@@ -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

@@ -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

@@ -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:
+def _days_to_resolution(end_date: str, as_of: Optional[datetime] = None) -> int:
-    """Return calendar days until market resolution, or 30 if unknown."""
+    """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 ──────────────────────────────────

tests/test_replay_engine.py

@@ -0,0 +1,367 @@
+"""
+Tests for the Replay R1 replay core (bot/replay.py) and the as_of clock
+injection in BayesianStrategy.evaluate().
+
+The central contract is round-trip fidelity: a decision recorded by R0 and
+replayed through replay_cycle() with the same strategy constants must match
+field-for-field (matched=True, mismatch_field=None).  Each round-trip test
+produces the "archive" by running the real evaluate() with FakeNews, then
+replays the drained record as if it had been read back from the signals table.
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+import bot.strategy.bayesian as bayesian
+from bot.data.polymarket import Market, market_family_key
+from bot.strategy.bayesian import BayesianStrategy, _days_to_resolution
+from bot.replay import (
+    ReplayNews,
+    build_ext,
+    build_market,
+    replay_cycle,
+    strategy_config_hash,
+)
+
+from tests.test_news_guardrail import FakeNews, _sentiment_for
+
+
+def _end_date(days_ahead: int = 20) -> str:
+    dt = datetime.now(timezone.utc) + timedelta(days=days_ahead)
+    return dt.strftime("%Y-%m-%dT00:00:00Z")
+
+
+def _make_market(
+    yes_price: float,
+    question: str = "Will John Smith win the election?",
+    category: str = "politics",
+    market_id: str = "mkt-replay-1",
+    end_date: str = None,
+) -> Market:
+    return Market(
+        id=market_id,
+        condition_id="cond-replay-1",
+        question=question,
+        yes_token_id="yes-tok",
+        no_token_id="no-tok",
+        yes_price=yes_price,
+        no_price=1.0 - yes_price,
+        volume_24h=50_000.0,
+        end_date=end_date if end_date is not None else _end_date(),
+        active=True,
+        category=category,
+    )
+
+
+def _snapshot(valid: bool = True) -> dict:
+    """An ext_snapshots row as read back from the DB."""
+    return {
+        "btc_price": 100_000.0,
+        "btc_change_24h": 0.0,
+        "eth_price": 4_000.0,
+        "eth_change_24h": 0.0,
+        "btc_dominance": 50.0,
+        "fear_greed_index": 50,
+        "fear_greed_label": "neutral",
+        "total_market_cap_change": 0.0,
+        "valid": valid,
+    }
+
+
+def _market_row(market: Market) -> dict:
+    """A markets-table row for the given Market."""
+    return {
+        "id": market.id,
+        "condition_id": market.condition_id,
+        "question": market.question,
+        "category": market.category,
+        "end_date": market.end_date,
+    }
+
+
+def _record_with_live_evaluate(
+    market: Market,
+    news=None,
+    families: set = frozenset(),
+) -> dict:
+    """Run the real evaluate() and return the R0 record it produced —
+    the same dict save_signal_records() would have archived."""
+    strategy = BayesianStrategy(news=news, manifold=None, db=None)
+    asyncio.run(strategy.evaluate(market, build_ext(_snapshot()), set(families)))
+    return strategy.drain_cycle_records()[0]
+
+
+def _replay_one(record: dict, market: Market, snapshot: dict = None) -> dict:
+    cycle_ts = datetime.now(timezone.utc)
+    decisions = asyncio.run(replay_cycle(
+        cycle_ts,
+        snapshot or _snapshot(),
+        [record],
+        {market.id: _market_row(market)},
+    ))
+    assert len(decisions) == 1
+    return decisions[0]
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Clock injection
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_days_to_resolution_uses_injected_clock():
+    end = "2026-08-01T00:00:00Z"
+    as_of = datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc)
+    assert _days_to_resolution(end, as_of) == 29
+    assert _days_to_resolution(end, as_of - timedelta(days=60)) == 89
+
+
+def test_default_clock_is_wall_clock():
+    end = _end_date(days_ahead=40)
+    assert _days_to_resolution(end) == _days_to_resolution(
+        end, datetime.now(timezone.utc)
+    )
+
+
+def test_as_of_changes_regime_threshold():
+    """Same politics market: <30 d out → regime 0.08; replayed from 60 d
+    earlier → regime 0.12.  The clock, not the wall time, must decide."""
+    market = _make_market(0.470)
+    sentiment = _sentiment_for(0.470, 0.601)
+
+    def _regime(as_of):
+        strategy = BayesianStrategy(news=FakeNews(sentiment), manifold=None, db=None)
+        asyncio.run(strategy.evaluate(
+            market, build_ext(_snapshot()), set(), as_of=as_of,
+        ))
+        return strategy.drain_cycle_records()[0]["regime_min_edge"]
+
+    now = datetime.now(timezone.utc)
+    assert _regime(now) == pytest.approx(0.08)
+    assert _regime(now - timedelta(days=60)) == pytest.approx(0.12)
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Round-trip fidelity: record with live evaluate(), replay, expect match
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_roundtrip_confidence_skip():
+    """Georgia signature: edge passes, confidence blocks — full-field match."""
+    sentiment = _sentiment_for(0.470, 0.601)
+    market = _make_market(0.470)
+    record = _record_with_live_evaluate(market, news=FakeNews(sentiment))
+    assert record["skip_reason"] == "confidence"
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["mismatch_field"] is None
+    assert decision["skip_reason"] == "confidence"
+    assert decision["estimated_prob"] == pytest.approx(record["estimated_prob"])
+    assert decision["edge_net"] == pytest.approx(record["edge_net"])
+    assert decision["confidence"] == pytest.approx(record["confidence"])
+    assert decision["direction"] == record["direction"]
+    assert decision["would_trade"] is False
+
+
+def test_roundtrip_edge_net_skip():
+    market = _make_market(0.50)
+    record = _record_with_live_evaluate(market)
+    assert record["skip_reason"] == "edge_net"
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["would_trade"] is False
+
+
+def test_roundtrip_guardrail_clamp():
+    """Clamped posterior must reproduce exactly (raw != final in archive)."""
+    market = _make_market(0.845)
+    record = _record_with_live_evaluate(
+        market, news=FakeNews(_sentiment_for(0.845, 0.431))
+    )
+    assert record["guardrail_applied"] is True
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["raw_final_prob"] == pytest.approx(record["raw_final_prob"])
+    assert decision["estimated_prob"] == pytest.approx(record["estimated_prob"])
+
+
+def test_roundtrip_prior_extreme():
+    market = _make_market(0.03)
+    record = _record_with_live_evaluate(market)
+    assert record["skip_reason"] == "prior_extreme"
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["skip_reason"] == "prior_extreme"
+
+
+def test_roundtrip_family_skip():
+    """Family-skipped rows replay with their own family injected as occupied."""
+    market = _make_market(0.50)
+    record = _record_with_live_evaluate(
+        market, families={market_family_key(market)}
+    )
+    assert record["skip_reason"] == "family"
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["skip_reason"] == "family"
+
+
+def test_roundtrip_unsupported():
+    market = _make_market(0.50, question="Will it rain tomorrow?", category="")
+    record = _record_with_live_evaluate(market)
+    assert record["skip_reason"] == "unsupported"
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+
+
+def test_roundtrip_no_signals():
+    """ext.valid=False archived → replay rebuilds the invalid snapshot."""
+    market = _make_market(0.50)
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    asyncio.run(strategy.evaluate(market, build_ext(_snapshot(valid=False)), set()))
+    record = strategy.drain_cycle_records()[0]
+    assert record["skip_reason"] == "no_signals"
+
+    decision = _replay_one(record, market, snapshot=_snapshot(valid=False))
+    assert decision["matched"] is True
+
+
+def test_roundtrip_trade_path(monkeypatch):
+    """skip_reason=None (tradeable) round-trips with would_trade=True.
+    Politics can't clear MIN_CONFIDENCE=0.55 (the known ceiling), so the
+    gate is lowered for this test only — both record and replay see the
+    same constant, which is exactly the config_hash contract."""
+    monkeypatch.setattr(bayesian, "MIN_CONFIDENCE", 0.45)
+    sentiment = _sentiment_for(0.470, 0.601)
+    market = _make_market(0.470)
+    record = _record_with_live_evaluate(market, news=FakeNews(sentiment))
+    assert record["skip_reason"] is None
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["skip_reason"] is None
+    assert decision["would_trade"] is True
+    assert decision["direction"] == "BUY_YES"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Replay-specific semantics
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_budget_skipped_row_replays_without_news():
+    """A budget-skipped archive row (sentiment 0.0) must replay to the same
+    no-news decision — and never consume a replay-side budget."""
+    market = _make_market(0.50)
+    strategy = BayesianStrategy(news=FakeNews(0.9), manifold=None, db=None)
+    strategy._news_queries_this_cycle = bayesian.MAX_NEWS_QUERIES_PER_CYCLE
+    asyncio.run(strategy.evaluate(market, build_ext(_snapshot()), set()))
+    record = strategy.drain_cycle_records()[0]
+    assert record["news_budget_skipped"] is True
+    assert record["news_sentiment"] == 0.0
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is True
+    assert decision["estimated_prob"] == pytest.approx(record["estimated_prob"])
+
+
+def test_reentry_guard_row_is_recalibrated_not_compared():
+    """record_skip() rows carry no decision fields; the replay re-evaluates
+    them (calibration data) but marks them non-comparable."""
+    market = _make_market(0.50)
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    strategy.record_skip(market, "reentry_guard")
+    record = strategy.drain_cycle_records()[0]
+
+    decision = _replay_one(record, market)
+    assert decision["matched"] is None
+    assert decision["recorded_skip_reason"] == "reentry_guard"
+    # Re-evaluated on its merits: a full decision despite the recorded skip
+    assert decision["estimated_prob"] is not None
+    assert decision["skip_reason"] == "edge_net"
+
+
+def test_missing_market_row_flagged_not_crashed():
+    market = _make_market(0.50)
+    record = _record_with_live_evaluate(market)
+
+    decisions = asyncio.run(replay_cycle(
+        datetime.now(timezone.utc), _snapshot(), [record], {},
+    ))
+    assert decisions[0]["matched"] is False
+    assert decisions[0]["mismatch_field"] == "market_missing"
+
+
+def test_mismatch_detected_when_config_differs(monkeypatch):
+    """Counterfactual sanity: replaying under a different guardrail band
+    must produce matched=False with the diverging field named."""
+    market = _make_market(0.845)
+    record = _record_with_live_evaluate(
+        market, news=FakeNews(_sentiment_for(0.845, 0.431))
+    )
+    assert record["guardrail_applied"] is True
+
+    monkeypatch.setattr(bayesian, "MAX_NEWS_ONLY_PROB_SHIFT", 0.10)
+    decision = _replay_one(record, market)
+    assert decision["matched"] is False
+    # Tighter clamp (prior 0.845 ± 0.10 → est 0.745): edge_net drops from
+    # 0.21 to 0.06 < regime 0.08, so the skip flips confidence → edge_net
+    # and skip_reason is the first field _compare() sees diverge.
+    assert decision["mismatch_field"] == "skip_reason"
+    assert decision["skip_reason"] == "edge_net"
+
+
+def test_multi_row_cycle_preserves_order_and_isolation():
+    """Rows replay independently within a cycle: a family skip and a full
+    evaluation with different sentiments don't bleed into each other."""
+    m1 = _make_market(0.470, market_id="m1")
+    m2 = _make_market(
+        0.50, market_id="m2",
+        question="Will Jane Doe win the Georgia Senate race?",
+    )
+    r1 = _record_with_live_evaluate(m1, news=FakeNews(_sentiment_for(0.470, 0.601)))
+    r2 = _record_with_live_evaluate(m2)  # no news → edge_net skip
+
+    decisions = asyncio.run(replay_cycle(
+        datetime.now(timezone.utc),
+        _snapshot(),
+        [r1, r2],
+        {"m1": _market_row(m1), "m2": _market_row(m2)},
+    ))
+    assert [d["market_id"] for d in decisions] == ["m1", "m2"]
+    assert all(d["matched"] is True for d in decisions)
+    assert decisions[0]["skip_reason"] == "confidence"
+    assert decisions[1]["skip_reason"] == "edge_net"
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Run tagging
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_config_hash_stable_and_sensitive(monkeypatch):
+    h1 = strategy_config_hash()
+    assert strategy_config_hash() == h1
+    monkeypatch.setattr(bayesian, "MAX_NEWS_ONLY_PROB_SHIFT", 0.10)
+    assert strategy_config_hash() != h1
+
+
+def test_replay_news_returns_current_sentiment():
+    news = ReplayNews()
+    assert asyncio.run(news.get_sentiment("q")) == 0.0
+    news.sentiment = -0.42
+    assert asyncio.run(news.get_sentiment("q")) == -0.42
+
+
+def test_build_market_reconstruction():
+    market = _make_market(0.37)
+    record = _record_with_live_evaluate(market)
+    rebuilt = build_market(_market_row(market), record)
+    assert rebuilt.id == market.id
+    assert rebuilt.yes_price == pytest.approx(0.37)
+    assert rebuilt.volume_24h == pytest.approx(market.volume_24h)
+    assert rebuilt.end_date == market.end_date
+    assert rebuilt.category == "politics"
+    assert market_family_key(rebuilt) == market_family_key(market)