ci: bump outcomes-joiner CronJob image tag alongside deployment-bot

The outcomes-joiner CronJob (k8s-manifests, Replay R2) runs the same bot
image; without this its tag would freeze at the sha it was created with
while the deployment moves on. Same sed, one more file.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

.gitea/workflows/ci.yml

@@ -178,7 +178,8 @@ jobs:
           # keep their current (still existing) tag in the registry.
           if [ "${{ steps.changes.outputs.build_bot }}" = "true" ]; then
             sed -i "s|image: .*polymarket-bot[^-].*|image: git.chemavx.xyz/chemavx/polymarket-bot:${TAG}|g" \
-              polymarket-bot/deployment-bot.yaml
+              polymarket-bot/deployment-bot.yaml \
+              polymarket-bot/cronjob-outcomes.yaml
           fi
           if [ "${{ steps.changes.outputs.build_api }}" = "true" ]; then
             sed -i "s|image: .*polymarket-bot-api.*|image: git.chemavx.xyz/chemavx/polymarket-bot-api:${TAG}|g" \

bot/data/db.py

@@ -650,6 +650,251 @@ class Database:
                     cooldown_reason   = EXCLUDED.cooldown_reason
             """, poly_market_id, last_status, retry_after, cooldown_reason)
 
+    # ── Replay R0: snapshot recorder ─────────────────────────────────────────
+
+    async def save_ext_snapshot(self, cycle_ts, ext) -> None:
+        """Persist the ExternalSignals snapshot for one cycle (Replay R0)."""
+        async with self._pool.acquire() as conn:
+            await conn.execute("""
+                INSERT INTO ext_snapshots (
+                    cycle_ts, btc_price, btc_change_24h, eth_price, eth_change_24h,
+                    btc_dominance, fear_greed_index, fear_greed_label,
+                    total_market_cap_change, valid
+                ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
+                ON CONFLICT (cycle_ts) DO NOTHING
+            """,
+                cycle_ts, ext.btc_price, ext.btc_change_24h,
+                ext.eth_price, ext.eth_change_24h, ext.btc_dominance,
+                ext.fear_greed_index, ext.fear_greed_label,
+                ext.total_market_cap_change, ext.valid,
+            )
+
+    async def upsert_markets(self, markets: list) -> None:
+        """Refresh market metadata (Replay R0) — replay rebuilds Market from here."""
+        rows = [
+            (m.id, m.condition_id, m.question, m.category, m.end_date, m.active)
+            for m in markets
+        ]
+        async with self._pool.acquire() as conn:
+            await conn.executemany("""
+                INSERT INTO markets (id, condition_id, question, category, end_date, active, last_seen)
+                VALUES ($1,$2,$3,$4,$5,$6, now())
+                ON CONFLICT (id) DO UPDATE SET
+                    condition_id = EXCLUDED.condition_id,
+                    question     = EXCLUDED.question,
+                    category     = EXCLUDED.category,
+                    end_date     = EXCLUDED.end_date,
+                    active       = EXCLUDED.active,
+                    last_seen    = now()
+            """, rows)
+
+    async def save_signal_records(self, cycle_ts, records: list[dict]) -> None:
+        """Batch-insert one cycle's decision records into signals (Replay R0)."""
+        if not records:
+            return
+        rows = [
+            (
+                r["market_id"], cycle_ts, cycle_ts,
+                r["polymarket_price"], r["category"], r["volume_24h"],
+                r["skip_reason"], r["family_key"],
+                r["prior_prob"], r["estimated_prob"], r["raw_final_prob"],
+                r["edge_gross"], r["edge_net"], r["regime_min_edge"],
+                r["days_to_resolution"], r["confidence"], r["direction"],
+                r["passed_gross"], r["passed_net"],
+                r["news_sentiment"], r["news_budget_skipped"],
+                r["guardrail_applied"], r["guardrail_changed_decision"],
+                r["feat_fg_lo"], r["feat_mom_lo"], r["feat_news_lo"],
+                r["feat_mfld_lo"], r["feat_btc_dom_lo"],
+                r["edge_gross"],  # legacy `edge` column mirrors edge_gross
+                r["acted_on"],
+            )
+            for r in records
+        ]
+        async with self._pool.acquire() as conn:
+            await conn.executemany("""
+                INSERT INTO signals (
+                    market_id, timestamp, cycle_ts,
+                    polymarket_price, category, volume_24h,
+                    skip_reason, family_key,
+                    prior_prob, estimated_prob, raw_final_prob,
+                    edge_gross, edge_net, regime_min_edge,
+                    days_to_resolution, confidence, direction,
+                    passed_gross, passed_net,
+                    news_sentiment, news_budget_skipped,
+                    guardrail_applied, guardrail_changed_decision,
+                    feat_fg_lo, feat_mom_lo, feat_news_lo,
+                    feat_mfld_lo, feat_btc_dom_lo,
+                    edge, acted_on
+                ) VALUES (
+                    $1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,
+                    $16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30
+                )
+            """, rows)
+
+    async def prune_signal_records(self, retention_days: int) -> int:
+        """Delete archive rows older than retention_days; returns rows deleted."""
+        async with self._pool.acquire() as conn:
+            result = await conn.execute(
+                "DELETE FROM signals WHERE timestamp < now() - ($1 || ' days')::interval",
+                str(retention_days),
+            )
+            await conn.execute(
+                "DELETE FROM ext_snapshots WHERE cycle_ts < now() - ($1 || ' days')::interval",
+                str(retention_days),
+            )
+        try:
+            return int(result.split()[-1])
+        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)
+
+    # ── 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(

bot/data/schema.sql

@@ -318,3 +318,137 @@ CREATE TABLE IF NOT EXISTS manifold_eval_cooldown (
 );
 
 CREATE INDEX IF NOT EXISTS idx_mfld_cooldown_retry ON manifold_eval_cooldown(retry_after);
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Replay R0: snapshot recorder — the archive the replay engine reads from
+--
+-- The signals table (Phase 2/5 schema) never had a writer; R0 makes it the
+-- per-(market, cycle) decision archive.  One row per evaluated market per
+-- cycle, carrying both the INPUTS the strategy saw (external signals, news
+-- sentiment, per-feature log-odds) and the OUTPUTS it produced (probs, edges,
+-- gates, skip_reason).  A replay run rebuilds Market/ExternalSignals from
+-- these rows plus ext_snapshots and re-executes evaluate() deterministically.
+--
+-- cycle_ts groups all rows of one trading cycle and joins them to their
+-- ext_snapshots row (same timestamp; no FK to keep writes independent).
+-- days_to_resolution is persisted so replay does not depend on wall-clock.
+-- news_budget_skipped distinguishes "GNews had nothing" from "GNews was not
+-- asked this cycle" (5-query budget) — without it politics replay would treat
+-- budget starvation as absence of news.
+-- Retention: rows older than SIGNALS_RETENTION_DAYS (default 90) are pruned.
+-- ─────────────────────────────────────────────────────────────────────────────
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS cycle_ts            TIMESTAMPTZ;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS category            TEXT;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS prior_prob          DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS raw_final_prob      DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS days_to_resolution  INTEGER;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS volume_24h          DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS news_sentiment      DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS news_budget_skipped BOOLEAN;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS guardrail_applied   BOOLEAN;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS guardrail_changed_decision BOOLEAN;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_fg_lo          DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_mom_lo         DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_news_lo        DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_mfld_lo        DOUBLE PRECISION;
+ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_btc_dom_lo     DOUBLE PRECISION;
+
+CREATE INDEX IF NOT EXISTS idx_signals_cycle ON signals(cycle_ts);
+
+-- One row per trading cycle: the ExternalSignals snapshot every market in
+-- that cycle was evaluated against.  Written once per cycle before the
+-- evaluation loop; signals rows join on cycle_ts.
+CREATE TABLE IF NOT EXISTS ext_snapshots (
+    cycle_ts                TIMESTAMPTZ PRIMARY KEY,
+    btc_price               DOUBLE PRECISION,
+    btc_change_24h          DOUBLE PRECISION,
+    eth_price               DOUBLE PRECISION,
+    eth_change_24h          DOUBLE PRECISION,
+    btc_dominance           DOUBLE PRECISION,
+    fear_greed_index        INTEGER,
+    fear_greed_label        TEXT,
+    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);
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- 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()
+);

bot/main.py

@@ -43,6 +43,14 @@ PAPER_BANKROLL = float(os.getenv("PAPER_BANKROLL", "10000"))
 # position per 10 minutes.
 RESOLUTION_CHECK_INTERVAL = 10
 
+# Replay R0: persist per-(market, cycle) decision records + the ExternalSignals
+# snapshot each cycle, so the replay engine can re-run past decisions.  The
+# recorder must never break trading — every write is wrapped in try/except.
+SIGNAL_RECORDER_ENABLED = os.getenv("SIGNAL_RECORDER_ENABLED", "true").lower() == "true"
+SIGNALS_RETENTION_DAYS = int(os.getenv("SIGNALS_RETENTION_DAYS", "90"))
+# Prune the archive roughly once a day at the 60s cycle cadence.
+SIGNALS_PRUNE_INTERVAL_CYCLES = 1440
+
 
 async def check_resolutions(
     poly: PolymarketClient,
@@ -122,6 +130,16 @@ async def run_trading_loop(
             # 2. Get external signals
             ext_data = await external.get_all_signals()
 
+            # 2b. Replay R0: archive this cycle's inputs (ext snapshot + market
+            # metadata).  cycle_ts groups all signals rows of this cycle.
+            cycle_ts = datetime.now(timezone.utc)
+            if SIGNAL_RECORDER_ENABLED:
+                try:
+                    await db.save_ext_snapshot(cycle_ts, ext_data)
+                    await db.upsert_markets(markets)
+                except Exception as exc:
+                    log.warning("Signal recorder (inputs) failed: %s", exc)
+
             # 3. Build occupied_families from the current open portfolio positions.
             #    This prevents re-entering a family where we already hold a position.
             #    We also pull from DB to survive pod restarts.
@@ -176,6 +194,7 @@ async def run_trading_loop(
 
             reentry_guard_count = 0
             cycle_trades = 0
+            traded_market_ids: set[str] = set()
             for market in markets:
                 if market.id in inverted_guard:
                     log.info(
@@ -183,6 +202,7 @@ async def run_trading_loop(
                         market.id, market.question[:60],
                     )
                     reentry_guard_count += 1
+                    strategy.record_skip(market, "reentry_guard")
                     continue
 
                 # evaluate() returns None for all skips — reasons are logged internally
@@ -214,6 +234,7 @@ async def run_trading_loop(
                     # Block this family for the rest of the cycle (Phase 2)
                     occupied_families.add(signal.family_key)
                     cycle_trades += 1
+                    traded_market_ids.add(market.id)
                     # Mark manifold audit record as used in this trade
                     if signal.mfld_audit_id:
                         try:
@@ -221,6 +242,28 @@ async def run_trading_loop(
                         except Exception as exc:
                             log.warning("Failed to mark manifold audit used: %s", exc)
 
+            # 7b. Replay R0: flush this cycle's decision records to the archive.
+            # acted_on marks records whose signal actually became a trade
+            # (evaluate() can emit a signal that risk sizing later rejects).
+            records = strategy.drain_cycle_records()
+            if SIGNAL_RECORDER_ENABLED and records:
+                for rec in records:
+                    if rec["market_id"] in traded_market_ids:
+                        rec["acted_on"] = True
+                try:
+                    await db.save_signal_records(cycle_ts, records)
+                except Exception as exc:
+                    log.warning("Signal recorder (records) failed: %s", exc)
+                if cycle_count % SIGNALS_PRUNE_INTERVAL_CYCLES == 1:
+                    try:
+                        pruned = await db.prune_signal_records(SIGNALS_RETENTION_DAYS)
+                        log.info(
+                            "Signal archive pruned: %d rows older than %d days removed",
+                            pruned, SIGNALS_RETENTION_DAYS,
+                        )
+                    except Exception as exc:
+                        log.warning("Signal archive prune failed: %s", exc)
+
             # 8. [CYCLE SUMMARY] — one block per cycle, stable format for grep/compare
             stats = strategy.get_cycle_stats()
             legacy_incomplete_count = await db.get_legacy_incomplete_count()
@@ -277,6 +320,20 @@ async def run_trading_loop(
                 manifold_summary,
             )
 
+            # NEWS SUMMARY — one compact line, only on cycles where at least
+            # one market had a material GNews contribution (never an empty
+            # section on news-less cycles).
+            if stats["news_with_material"] > 0:
+                log.info(
+                    "NEWS SUMMARY | with_news=%d | avg_shift=%+.2f | "
+                    "max_shift=%+.2f | guardrail_applied=%d | changed_decisions=%d",
+                    stats["news_with_material"],
+                    stats["news_avg_shift"],
+                    stats["news_max_shift"],
+                    stats["news_guardrail_applied"],
+                    stats["news_changed_decisions"],
+                )
+
             # 9. Update daily metrics
             await metrics.update_daily_summary()
 

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

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

@@ -84,6 +84,27 @@ def _env_bool(name: str, default: bool) -> bool:
 MANIFOLD_SIGNAL_ENABLED = _env_bool("MANIFOLD_SIGNAL_ENABLED", False)
 MANIFOLD_AUDIT_ENABLED  = _env_bool("MANIFOLD_AUDIT_ENABLED", True)
 
+# ── GNews guardrail (catastrophic fuse) ────────────────────────────────────────
+# Post-mortem NVIDIA 631181: a single strong signal (legacy Manifold 0.13 at
+# weight 0.6) flipped a 0.845 market to 0.431 and lost.  With Manifold now
+# observational-only and macro signals gated behind is_non_price, GNews
+# (weight 1.5) is the only live signal that can move politics markets 20-30 pp
+# against the order-book consensus.  This is NOT a fine calibration — it is a
+# fuse against the extreme case: one uncorroborated signal violently inverting
+# the market.
+#
+# NEWS_GUARDRAIL_ENABLED: master switch for the fuse.
+# MAX_NEWS_ONLY_PROB_SHIFT: when GNews is the ONLY material signal, the final
+#   probability is clamped to prior ± this value.  0.25 still allows a 25 pp
+#   move (edge_net 0.21 after costs) — trades still happen, sizing is bounded.
+# NEWS_MATERIAL_LOGODDS_THRESHOLD: a signal counts as *material* iff its
+#   |log-odds contribution| >= this value.  Below it, a signal is noise and
+#   does NOT count as corroboration.  If ANY other signal (fg, momentum,
+#   btc_dom, manifold) is material, the fuse does not apply.
+NEWS_GUARDRAIL_ENABLED = _env_bool("NEWS_GUARDRAIL_ENABLED", True)
+MAX_NEWS_ONLY_PROB_SHIFT = float(os.getenv("MAX_NEWS_ONLY_PROB_SHIFT", "0.25"))
+NEWS_MATERIAL_LOGODDS_THRESHOLD = float(os.getenv("NEWS_MATERIAL_LOGODDS_THRESHOLD", "0.10"))
+
 # GNews free tier: 100 req/day.  We limit to 5 queries per trading cycle
 # (politics markets only) and rely on 6 h cache to stay within budget.
 MAX_NEWS_QUERIES_PER_CYCLE = 5
@@ -146,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
@@ -179,6 +207,42 @@ def has_token(text: str, token: str) -> bool:
 # Phase 3 — GNews priority scoring
 # ─────────────────────────────────────────────────────────────────────────────
 
+def apply_news_guardrail(
+    prior: float,
+    raw_final_prob: float,
+    feat_news_lo: float,
+    other_feats_lo: tuple[float, ...],
+) -> tuple[float, bool]:
+    """
+    GNews guardrail (catastrophic fuse).
+
+    Clamp raw_final_prob to prior ± MAX_NEWS_ONLY_PROB_SHIFT when ALL hold:
+      1. NEWS_GUARDRAIL_ENABLED
+      2. |feat_news_lo| >= NEWS_MATERIAL_LOGODDS_THRESHOLD (news is material)
+      3. every other signal's |log-odds contribution| is below the threshold
+         (GNews is the ONLY material signal — no corroboration)
+
+    Returns (final_prob, guardrail_applied).  guardrail_applied is True only
+    when the clamp actually changed the value; a raw_final_prob already inside
+    the band passes through untouched with applied=False.
+
+    Module globals are read at call time so tests can monkeypatch them.
+    """
+    if not NEWS_GUARDRAIL_ENABLED:
+        return raw_final_prob, False
+    if abs(feat_news_lo) < NEWS_MATERIAL_LOGODDS_THRESHOLD:
+        return raw_final_prob, False
+    if any(abs(v) >= NEWS_MATERIAL_LOGODDS_THRESHOLD for v in other_feats_lo):
+        return raw_final_prob, False  # corroborated — fuse does not apply
+    clamped = min(
+        max(raw_final_prob, prior - MAX_NEWS_ONLY_PROB_SHIFT),
+        prior + MAX_NEWS_ONLY_PROB_SHIFT,
+    )
+    if clamped == raw_final_prob:
+        return raw_final_prob, False
+    return clamped, True
+
+
 def gnews_priority(market: Market, news: "NewsClient") -> float:
     """
     Score a market for GNews query priority (higher = more valuable to query).
@@ -300,6 +364,13 @@ class BayesianStrategy:
         # (edge_gross, edge_net, regime_min) for every market that reached the
         # edge computation stage (passed prior-extreme, family, unsupported filters)
         self._evaluated_edges: list[tuple[float, float, float]] = []
+        # GNews guardrail observability — only markets with material news
+        self._news_shifts: list[float] = []       # final_prob - prior, signed
+        self._news_guardrail_applied: int = 0
+        self._news_changed_decisions: int = 0
+        # Replay R0: per-(market, cycle) decision records, drained by main.py
+        # into the signals table after each evaluation loop.
+        self._cycle_records: list[dict] = []
 
     def reset_cycle(self) -> None:
         """Call once at the start of each trading cycle to reset per-cycle counters."""
@@ -311,6 +382,54 @@ class BayesianStrategy:
         self._manifold_fetched = 0
         self._manifold_on_trade = 0
         self._evaluated_edges = []
+        self._news_shifts = []
+        self._news_guardrail_applied = 0
+        self._news_changed_decisions = 0
+        self._cycle_records = []
+
+    def record_skip(self, market: Market, skip_reason: str) -> None:
+        """Record a skip decided OUTSIDE evaluate() (e.g. reentry_guard in main)."""
+        self._record(market, skip_reason=skip_reason)
+
+    def drain_cycle_records(self) -> list[dict]:
+        """Return and clear this cycle's decision records (Replay R0)."""
+        records, self._cycle_records = self._cycle_records, []
+        return records
+
+    def _record(self, market: Market, skip_reason: Optional[str], **fields) -> None:
+        """Append one decision record. Early skips leave most fields None —
+        the archive still shows the market existed and why it went no further."""
+        rec = {
+            "market_id": market.id,
+            "polymarket_price": market.yes_price,
+            "category": market.category,
+            "volume_24h": market.volume_24h,
+            "skip_reason": skip_reason,
+            "family_key": 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,
+            "passed_gross": None,
+            "passed_net": None,
+            "news_sentiment": None,
+            "news_budget_skipped": None,
+            "guardrail_applied": None,
+            "guardrail_changed_decision": None,
+            "feat_fg_lo": None,
+            "feat_mom_lo": None,
+            "feat_news_lo": None,
+            "feat_mfld_lo": None,
+            "feat_btc_dom_lo": None,
+            "acted_on": False,
+        }
+        rec.update(fields)
+        self._cycle_records.append(rec)
 
     def get_cycle_stats(self) -> dict:
         """Return per-cycle counters for the [CYCLE SUMMARY] log block."""
@@ -330,6 +449,14 @@ class BayesianStrategy:
             "gross_gt_004": sum(1 for g in all_gross if g > 0.04),
             "manifold_matches_accepted": self._manifold_on_trade,
             "manifold_matches_rejected": self._manifold_fetched - self._manifold_on_trade,
+            # GNews guardrail — markets with |news_lo| >= NEWS_MATERIAL_LOGODDS_THRESHOLD
+            "news_with_material":     len(self._news_shifts),
+            "news_avg_shift":         (sum(self._news_shifts) / len(self._news_shifts))
+                                      if self._news_shifts else 0.0,
+            "news_max_shift":         max(self._news_shifts, key=abs)
+                                      if self._news_shifts else 0.0,
+            "news_guardrail_applied": self._news_guardrail_applied,
+            "news_changed_decisions": self._news_changed_decisions,
         }
 
     async def evaluate(
@@ -337,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
@@ -395,6 +529,7 @@ class BayesianStrategy:
                 "SKIP_UNSUPPORTED  %-50s | cat=%r",
                 market.question[:50], category,
             )
+            self._record(market, skip_reason="unsupported")
             return None
 
         if not ext.valid:
@@ -402,6 +537,7 @@ class BayesianStrategy:
                 "SKIP_NO_SIGNALS   %-50s | reason=external data unavailable",
                 market.question[:50],
             )
+            self._record(market, skip_reason="no_signals")
             return None
 
         # ── Phase 1: prior + prior-extreme filter ────────────────────────────
@@ -413,6 +549,7 @@ class BayesianStrategy:
                 "SKIP_PRIOR_EXTREME %-50s | cat=%-12s | prior=%.3f | reason=prior<0.08",
                 market.question[:50], category, market.yes_price,
             )
+            self._record(market, skip_reason="prior_extreme", prior_prob=prior)
             return None
         if market.yes_price > 0.92:
             self._skip_prior_extreme += 1
@@ -420,6 +557,7 @@ class BayesianStrategy:
                 "SKIP_PRIOR_EXTREME %-50s | cat=%-12s | prior=%.3f | reason=prior>0.92",
                 market.question[:50], category, market.yes_price,
             )
+            self._record(market, skip_reason="prior_extreme", prior_prob=prior)
             return None
 
         # ── Phase 2: family deduplication ────────────────────────────────────
@@ -430,10 +568,11 @@ class BayesianStrategy:
                 "SKIP_FAMILY        %-50s | cat=%-12s | family=%s",
                 market.question[:50], category, family,
             )
+            self._record(market, skip_reason="family", prior_prob=prior, family_key=family)
             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 ──────────────────────────────────
@@ -503,6 +642,11 @@ class BayesianStrategy:
         # Phase 3: caller has pre-sorted markets by gnews_priority() so the
         # highest-value markets reach this block first.
         news_log_adj = 0.0
+        news_sentiment = 0.0
+        # Replay R0: True when GNews was never consulted for this market this
+        # cycle (budget exhausted) — a replay must not read feat_news_lo=0.0 as
+        # "there was no news".
+        news_budget_skipped = False
         # self._news.enabled gates the whole block: with no GNews API key the
         # client is a no-op, so we must not consume (or report) query budget for
         # it — see NewsClient.enabled.
@@ -511,9 +655,11 @@ class BayesianStrategy:
                 self._news_queries_this_cycle += 1
                 sentiment = await self._news.get_sentiment(market.question)
                 if abs(sentiment) > 0.05:
+                    news_sentiment = sentiment
                     news_log_adj = sentiment * NEWS_LOGODDS_WEIGHT
                     sources.append(f"GNews: {sentiment:+.2f}")
             else:
+                news_budget_skipped = True
                 log.info(
                     "SKIP_GNEWS_PRIORITY %-50s | reason=cycle budget %d reached",
                     market.question[:50], MAX_NEWS_QUERIES_PER_CYCLE,
@@ -652,8 +798,31 @@ class BayesianStrategy:
         # Posterior via log-odds updating
         log_odds_prior = math.log(prior / (1 - prior))
         total_adj = sum(adjustments)
-        estimated_prob = _sigmoid(log_odds_prior + total_adj * 2 + news_log_adj + manifold_log_adj)
-        estimated_prob = max(0.05, min(0.95, estimated_prob))
+        # raw_final_prob: posterior BEFORE the news guardrail.
+        raw_final_prob = _sigmoid(log_odds_prior + total_adj * 2 + news_log_adj + manifold_log_adj)
+        raw_final_prob = max(0.05, min(0.95, raw_final_prob))
+
+        # Per-feature log-odds contributions (Phase 6) — computed here (not
+        # after the edge gate) because the guardrail below needs them to decide
+        # signal materiality.
+        # fg / mom / btc_dom: probability-delta × 2 → log-odds.
+        # news / mfld: already log-odds (LOGODDS_WEIGHT already applied).
+        feat_fg_lo      = _fg_contribution * 2
+        feat_mom_lo     = _momentum_contribution * 2
+        feat_news_lo    = news_log_adj
+        feat_mfld_lo    = manifold_log_adj
+        feat_btc_dom_lo = _btc_dom_contribution * 2
+
+        # ── GNews guardrail (catastrophic fuse) ──────────────────────────────
+        # When GNews is the ONLY material signal, clamp the posterior to
+        # prior ± MAX_NEWS_ONLY_PROB_SHIFT.  estimated_prob (post-guardrail) is
+        # what edge/trading uses; raw_final_prob is kept for observability.
+        estimated_prob, news_guardrail_applied = apply_news_guardrail(
+            prior,
+            raw_final_prob,
+            feat_news_lo,
+            (feat_fg_lo, feat_mom_lo, feat_btc_dom_lo, feat_mfld_lo),
+        )
 
         # ── Phase 1: edge_gross and edge_net ─────────────────────────────────
         raw_edge = estimated_prob - market.yes_price
@@ -675,15 +844,6 @@ class BayesianStrategy:
         if manifold_log_adj != 0.0:
             confidence = min(confidence_cap, confidence + 0.08)
 
-        # Per-feature log-odds contributions (Phase 6).
-        # fg / mom / btc_dom: probability-delta × 2 → log-odds.
-        # news / mfld: already log-odds (LOGODDS_WEIGHT already applied).
-        feat_fg_lo      = _fg_contribution * 2
-        feat_mom_lo     = _momentum_contribution * 2
-        feat_news_lo    = news_log_adj
-        feat_mfld_lo    = manifold_log_adj
-        feat_btc_dom_lo = _btc_dom_contribution * 2
-
         feat_str = (
             f"fg_lo={feat_fg_lo:+.4f} mom_lo={feat_mom_lo:+.4f} "
             f"news_lo={feat_news_lo:+.4f} mfld_lo={feat_mfld_lo:+.4f} "
@@ -695,6 +855,80 @@ class BayesianStrategy:
         passed_net = edge_net >= regime_min
         can_trade = passed_net and confidence >= MIN_CONFIDENCE
 
+        # ── Guardrail decision impact ────────────────────────────────────────
+        # True when the un-clamped posterior's edge crossed the regime gate but
+        # the clamped one no longer does — i.e. the fuse PREVENTED a trade.
+        # Confidence is invariant under the clamp (it depends only on signal
+        # agreement), so the edge gate is the only component that can flip.
+        guardrail_changed_trade_decision = False
+        if news_guardrail_applied:
+            raw_edge_net = abs(raw_final_prob - market.yes_price) - TOTAL_COST_RATE
+            guardrail_changed_trade_decision = (
+                raw_edge_net >= regime_min and edge_net < regime_min
+            )
+
+        # ── Guardrail observability — ONLY markets with material news ───────
+        # Gated on materiality so the ~145 markets/cycle without news don't
+        # flood the logs.  posterior_before_news = everything except GNews.
+        news_is_material = abs(feat_news_lo) >= NEWS_MATERIAL_LOGODDS_THRESHOLD
+        if news_is_material:
+            posterior_before_news = max(0.05, min(0.95, _sigmoid(
+                log_odds_prior + total_adj * 2 + manifold_log_adj
+            )))
+            self._news_shifts.append(estimated_prob - prior)
+            if news_guardrail_applied:
+                self._news_guardrail_applied += 1
+            if guardrail_changed_trade_decision:
+                self._news_changed_decisions += 1
+            log.info(
+                "NEWS_MATERIAL      %-50s | cat=%-12s | family=%-28s | "
+                "prior=%.3f | before_news=%.3f | raw=%.3f | final=%.3f | "
+                "sent=%+.2f | news_lo=%+.4f | "
+                "edge_before_news=%.3f | edge_after_raw=%.3f | edge_after_guardrail=%.3f | "
+                "guardrail=%s | changed_decision=%s | max_shift=%.2f",
+                market.question[:50], category, family,
+                prior, posterior_before_news, raw_final_prob, estimated_prob,
+                news_sentiment, feat_news_lo,
+                abs(posterior_before_news - market.yes_price),
+                abs(raw_final_prob - market.yes_price),
+                edge_gross,
+                "applied" if news_guardrail_applied else "none",
+                str(guardrail_changed_trade_decision).lower(),
+                MAX_NEWS_ONLY_PROB_SHIFT,
+            )
+
+        # Replay R0: full decision record — same fields for skip and trade paths.
+        # skip_reason granularity: "edge_net" when the edge gate failed,
+        # "confidence" when only the confidence gate blocked the trade.
+        self._record(
+            market,
+            skip_reason=(
+                None if can_trade
+                else ("edge_net" if not passed_net else "confidence")
+            ),
+            family_key=family,
+            prior_prob=prior,
+            estimated_prob=estimated_prob,
+            raw_final_prob=raw_final_prob,
+            edge_gross=edge_gross,
+            edge_net=edge_net,
+            regime_min_edge=regime_min,
+            days_to_resolution=days,
+            confidence=confidence,
+            direction=direction,
+            passed_gross=passed_gross,
+            passed_net=passed_net,
+            news_sentiment=news_sentiment,
+            news_budget_skipped=news_budget_skipped,
+            guardrail_applied=news_guardrail_applied,
+            guardrail_changed_decision=guardrail_changed_trade_decision,
+            feat_fg_lo=feat_fg_lo,
+            feat_mom_lo=feat_mom_lo,
+            feat_news_lo=feat_news_lo,
+            feat_mfld_lo=feat_mfld_lo,
+            feat_btc_dom_lo=feat_btc_dom_lo,
+        )
+
         if not can_trade:
             # Increment the appropriate edge-net counter
             if edge_net <= 0:
@@ -723,8 +957,21 @@ class BayesianStrategy:
             )
             return None
 
+        # When GNews participated, expose raw vs final and the guardrail verdict
+        # (Task 4 of the guardrail spec); otherwise keep the legacy format.
+        if news_log_adj != 0.0:
+            prob_part = (
+                f"Prior=poly({prior:.3f}) → raw={raw_final_prob:.3f} "
+                f"→ final={estimated_prob:.3f} | "
+                f"GNews sent={news_sentiment:+.2f} | "
+                f"guardrail={'applied' if news_guardrail_applied else 'none'} | "
+                f"changed_decision={str(guardrail_changed_trade_decision).lower()} | "
+                f"max_shift={MAX_NEWS_ONLY_PROB_SHIFT:.2f} | "
+            )
+        else:
+            prob_part = f"Prior=poly({prior:.3f}) → estimate={estimated_prob:.3f} | "
         reasoning = (
-            f"Prior=poly({prior:.3f}) → estimate={estimated_prob:.3f} | "
+            prob_part +
             f"Poly price={market.yes_price:.3f} | "
             f"edge_gross={edge_gross:+.3f} | edge_net={edge_net:+.3f} | "
             f"regime_min={regime_min:.2f} | days={days} | "

requirements.txt

@@ -1,7 +1,7 @@
 # Core
 asyncpg==0.29.0
 httpx==0.27.0
-fastapi==0.138.2
+fastapi==0.111.0
 uvicorn[standard]==0.29.0
 pydantic==2.7.0
 

tests/test_news_guardrail.py

@@ -0,0 +1,247 @@
+"""
+Tests for the GNews guardrail (catastrophic fuse).
+
+Post-mortem NVIDIA 631181: one uncorroborated signal at high weight flipped a
+0.845 market to 0.431.  With Manifold observational-only and macro signals
+gated behind is_non_price, GNews is the only live signal able to move politics
+markets 20-30 pp against the order-book consensus.  The fuse clamps the
+posterior to prior ± MAX_NEWS_ONLY_PROB_SHIFT when GNews is the ONLY material
+signal (|log-odds| >= NEWS_MATERIAL_LOGODDS_THRESHOLD); any other material
+signal counts as corroboration and disables the clamp.
+
+Politics markets have no macro adjustments, so full-path tests exercise the
+"GNews only" branch naturally; the corroboration branch is tested through the
+pure helper apply_news_guardrail().
+
+evaluate() emits a NEWS_MATERIAL log line for every market whose news
+contribution is material (trade or skip); tests parse it via caplog.
+"""
+import asyncio
+import logging
+import math
+import re
+
+import pytest
+
+import bot.strategy.bayesian as bayesian
+from bot.data.external import ExternalSignals
+from bot.data.polymarket import Market
+from bot.strategy.bayesian import (
+    NEWS_LOGODDS_WEIGHT,
+    BayesianStrategy,
+    apply_news_guardrail,
+)
+
+NEWS_MATERIAL_RE = re.compile(
+    r"NEWS_MATERIAL.*raw=(\d+\.\d+) \| final=(\d+\.\d+).*"
+    r"guardrail=(applied|none) \| changed_decision=(true|false)"
+)
+
+
+def _logodds(p: float) -> float:
+    return math.log(p / (1 - p))
+
+
+def _sentiment_for(prior: float, target_raw: float) -> float:
+    """Sentiment that moves `prior` to exactly `target_raw` via GNews alone."""
+    return (_logodds(target_raw) - _logodds(prior)) / NEWS_LOGODDS_WEIGHT
+
+
+class FakeNews:
+    """Deterministic NewsClient stub returning a fixed sentiment."""
+
+    enabled = True
+
+    def __init__(self, sentiment: float) -> None:
+        self._sentiment = sentiment
+
+    async def get_sentiment(self, question: str) -> float:
+        return self._sentiment
+
+    def get_freshness(self, question: str) -> float:
+        return 1.0
+
+
+def _make_market(yes_price: float) -> Market:
+    return Market(
+        id="mkt-guardrail-1",
+        condition_id="cond-guardrail-1",
+        question="Will John Smith win the election?",
+        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="2026-07-15T00:00:00Z",  # politics <30 d → regime_min 0.08
+        active=True,
+        category="politics",
+    )
+
+
+def _make_signals() -> ExternalSignals:
+    # Neutral macro environment; irrelevant for politics (gated) but explicit.
+    return ExternalSignals(
+        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=True,
+    )
+
+
+def _evaluate(yes_price: float, sentiment: float, caplog) -> tuple[
+    BayesianStrategy, tuple[float, float, str, str]
+]:
+    """Run evaluate() on a politics market and parse the NEWS_MATERIAL line."""
+    strategy = BayesianStrategy(news=FakeNews(sentiment), manifold=None, db=None)
+    market = _make_market(yes_price)
+    with caplog.at_level(logging.INFO, logger="bot.strategy.bayesian"):
+        asyncio.run(strategy.evaluate(market, _make_signals(), occupied_families=set()))
+    for record in caplog.records:
+        m = NEWS_MATERIAL_RE.search(record.getMessage())
+        if m:
+            return strategy, (
+                float(m.group(1)), float(m.group(2)), m.group(3), m.group(4)
+            )
+    pytest.fail(
+        "No NEWS_MATERIAL log line found; got: "
+        f"{[r.getMessage() for r in caplog.records]}"
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 1 — extreme uncorroborated shift: clamp to prior - MAX_NEWS_ONLY_PROB_SHIFT
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_extreme_news_only_shift_is_clamped(caplog):
+    """prior=0.845, raw 0.431 (NVIDIA signature) → final clamped to 0.595."""
+    strategy, (raw, final, guardrail, _) = _evaluate(
+        yes_price=0.845, sentiment=_sentiment_for(0.845, 0.431), caplog=caplog
+    )
+    assert raw == pytest.approx(0.431, abs=1e-3)
+    assert guardrail == "applied"
+    assert final >= 0.595
+    assert final == pytest.approx(0.845 - bayesian.MAX_NEWS_ONLY_PROB_SHIFT, abs=1e-3)
+    assert strategy.get_cycle_stats()["news_guardrail_applied"] == 1
+    assert strategy.get_cycle_stats()["news_with_material"] == 1
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 2 — moderate shift inside the band: passes through untouched
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_moderate_news_shift_inside_band_not_clamped(caplog):
+    """prior=0.50, raw 0.62 → within ±0.25 band → final=0.62, no clamp."""
+    strategy, (raw, final, guardrail, _) = _evaluate(
+        yes_price=0.50, sentiment=_sentiment_for(0.50, 0.62), caplog=caplog
+    )
+    assert raw == pytest.approx(0.62, abs=1e-3)
+    assert final == pytest.approx(0.62, abs=1e-3)
+    assert guardrail == "none"
+    assert strategy.get_cycle_stats()["news_guardrail_applied"] == 0
+    # Still counted as a material-news market for the NEWS SUMMARY.
+    assert strategy.get_cycle_stats()["news_with_material"] == 1
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 3 — corroboration: any other material signal disables the fuse
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_corroborated_news_not_clamped():
+    """GNews material + another signal >= threshold → raw passes without clamp."""
+    news_lo = _logodds(0.20) - _logodds(0.50)  # ≈ -1.386, clearly material
+    final, applied = apply_news_guardrail(
+        prior=0.50,
+        raw_final_prob=0.20,
+        feat_news_lo=news_lo,
+        other_feats_lo=(0.0, 0.15, 0.0, 0.0),  # one corroborating signal
+    )
+    assert final == 0.20
+    assert applied is False
+
+
+def test_corroboration_threshold_is_inclusive():
+    """|other| == threshold exactly counts as corroboration (>=, not >)."""
+    final, applied = apply_news_guardrail(
+        prior=0.50,
+        raw_final_prob=0.20,
+        feat_news_lo=-1.386,
+        other_feats_lo=(bayesian.NEWS_MATERIAL_LOGODDS_THRESHOLD, 0.0, 0.0, 0.0),
+    )
+    assert final == 0.20
+    assert applied is False
+
+
+def test_uncorroborated_helper_clamps():
+    """Same shift with only noise elsewhere → clamped to prior - 0.25."""
+    final, applied = apply_news_guardrail(
+        prior=0.50,
+        raw_final_prob=0.20,
+        feat_news_lo=-1.386,
+        other_feats_lo=(0.05, -0.09, 0.0, 0.0),  # all below threshold → noise
+    )
+    assert final == pytest.approx(0.25)
+    assert applied is True
+
+
+def test_sub_material_news_never_clamped():
+    """|news_lo| below threshold → fuse not armed, whatever the shift."""
+    final, applied = apply_news_guardrail(
+        prior=0.50,
+        raw_final_prob=0.10,
+        feat_news_lo=0.09,
+        other_feats_lo=(0.0, 0.0, 0.0, 0.0),
+    )
+    assert final == 0.10
+    assert applied is False
+
+
+def test_guardrail_disabled_passthrough(monkeypatch):
+    monkeypatch.setattr(bayesian, "NEWS_GUARDRAIL_ENABLED", False)
+    final, applied = apply_news_guardrail(
+        prior=0.845,
+        raw_final_prob=0.431,
+        feat_news_lo=-1.974,
+        other_feats_lo=(0.0, 0.0, 0.0, 0.0),
+    )
+    assert final == 0.431
+    assert applied is False
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Test 4 — changed_decision: the clamp moves the edge from tradeable to not
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_guardrail_changed_trade_decision(monkeypatch, caplog):
+    """
+    With max_shift=0.10 the clamped edge (0.10 gross, 0.06 net) falls below the
+    politics <30 d regime gate (0.08) while the raw edge (0.414 gross, 0.374
+    net) crossed it → the fuse prevented the trade → changed_decision=true.
+
+    (With the default 0.25 the clamped edge_net is 0.21, above every regime
+    minimum, so the flag can only fire with a tighter configured band.)
+    """
+    monkeypatch.setattr(bayesian, "MAX_NEWS_ONLY_PROB_SHIFT", 0.10)
+    strategy, (raw, final, guardrail, changed) = _evaluate(
+        yes_price=0.845, sentiment=_sentiment_for(0.845, 0.431), caplog=caplog
+    )
+    assert raw == pytest.approx(0.431, abs=1e-3)
+    assert final == pytest.approx(0.745, abs=1e-3)
+    assert guardrail == "applied"
+    assert changed == "true"
+    stats = strategy.get_cycle_stats()
+    assert stats["news_changed_decisions"] == 1
+    assert stats["news_guardrail_applied"] == 1
+
+
+def test_default_band_does_not_change_decision(caplog):
+    """Default 0.25 band: clamp binds but edge_net 0.21 still crosses the gate."""
+    _, (_, _, guardrail, changed) = _evaluate(
+        yes_price=0.845, sentiment=_sentiment_for(0.845, 0.431), caplog=caplog
+    )
+    assert guardrail == "applied"
+    assert changed == "false"

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

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)

tests/test_signal_recorder.py

@@ -0,0 +1,224 @@
+"""
+Tests for the Replay R0 snapshot recorder (strategy-side record accumulation).
+
+Every evaluate() call must leave exactly one record in _cycle_records, whatever
+exit path it takes, so the signals archive is a complete account of each cycle.
+DB persistence itself (save_signal_records) is exercised in prod; these tests
+cover the record-building contract the replay engine will rely on:
+
+  - one record per market per evaluate() call, drained per cycle
+  - skip_reason granularity (prior_extreme / family / edge_net / confidence /
+    unsupported / reentry_guard via record_skip)
+  - full input/output fields on records that reached edge computation
+  - news_budget_skipped distinguishes "not asked" from "no news"
+"""
+import asyncio
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+import bot.strategy.bayesian as bayesian
+from bot.data.external import ExternalSignals
+from bot.data.polymarket import Market
+from bot.strategy.bayesian import (
+    MAX_NEWS_QUERIES_PER_CYCLE,
+    BayesianStrategy,
+)
+
+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-recorder-1",
+) -> Market:
+    return Market(
+        id=market_id,
+        condition_id="cond-recorder-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(),  # ~20 d → politics regime_min 0.08
+        active=True,
+        category=category,
+    )
+
+
+def _make_signals() -> ExternalSignals:
+    return ExternalSignals(
+        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=True,
+    )
+
+
+def _evaluate(strategy: BayesianStrategy, market: Market, families=None) -> None:
+    asyncio.run(strategy.evaluate(market, _make_signals(), families or set()))
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Full-evaluation records: every input/output field the replay needs
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_confidence_skip_record_has_full_fields():
+    """Politics market whose edge passes but confidence blocks (the known
+    politics ceiling): record must carry the complete decision context."""
+    sentiment = _sentiment_for(0.470, 0.601)  # Georgia signature: edge_net 0.091
+    strategy = BayesianStrategy(news=FakeNews(sentiment), manifold=None, db=None)
+    market = _make_market(0.470)
+    _evaluate(strategy, market)
+
+    records = strategy.drain_cycle_records()
+    assert len(records) == 1
+    rec = records[0]
+    assert rec["market_id"] == "mkt-recorder-1"
+    assert rec["skip_reason"] == "confidence"
+    assert rec["category"] == "politics"
+    assert rec["polymarket_price"] == pytest.approx(0.470)
+    assert rec["prior_prob"] == pytest.approx(0.470)
+    assert rec["estimated_prob"] == pytest.approx(0.601, abs=1e-3)
+    assert rec["raw_final_prob"] == pytest.approx(0.601, abs=1e-3)
+    assert rec["edge_net"] == pytest.approx(0.091, abs=1e-3)
+    assert rec["regime_min_edge"] == pytest.approx(0.08)
+    assert rec["passed_net"] is True
+    assert rec["confidence"] == pytest.approx(0.50)
+    assert rec["direction"] == "BUY_YES"
+    assert rec["news_sentiment"] == pytest.approx(sentiment, abs=1e-6)
+    assert rec["feat_news_lo"] != 0.0
+    assert rec["news_budget_skipped"] is False
+    assert rec["guardrail_applied"] is False
+    assert rec["guardrail_changed_decision"] is False
+    assert rec["days_to_resolution"] is not None
+    assert rec["acted_on"] is False
+
+
+def test_edge_net_skip_record():
+    """No news, no edge → skip_reason=edge_net with passed_net False."""
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    market = _make_market(0.50)
+    _evaluate(strategy, market)
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["skip_reason"] == "edge_net"
+    assert rec["passed_net"] is False
+    assert rec["estimated_prob"] == pytest.approx(0.50, abs=1e-3)
+    assert rec["feat_news_lo"] == 0.0
+
+
+def test_guardrail_fields_recorded_when_clamped():
+    """Guardrail clamp shows up in the record (applied=True, raw != final)."""
+    strategy = BayesianStrategy(
+        news=FakeNews(_sentiment_for(0.845, 0.431)), manifold=None, db=None
+    )
+    market = _make_market(0.845)
+    _evaluate(strategy, market)
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["guardrail_applied"] is True
+    assert rec["raw_final_prob"] == pytest.approx(0.431, abs=1e-3)
+    assert rec["estimated_prob"] == pytest.approx(
+        0.845 - bayesian.MAX_NEWS_ONLY_PROB_SHIFT, abs=1e-3
+    )
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Early-skip records: minimal but present
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_prior_extreme_record():
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    _evaluate(strategy, _make_market(0.03))
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["skip_reason"] == "prior_extreme"
+    assert rec["polymarket_price"] == pytest.approx(0.03)
+    assert rec["prior_prob"] == pytest.approx(0.05)  # clamped prior
+    assert rec["estimated_prob"] is None
+    assert rec["edge_net"] is None
+
+
+def test_family_skip_record():
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    market = _make_market(0.50)
+    from bot.data.polymarket import market_family_key
+    _evaluate(strategy, market, families={market_family_key(market)})
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["skip_reason"] == "family"
+    assert rec["family_key"] is not None
+
+
+def test_unsupported_record():
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    market = _make_market(0.50, question="Will it rain tomorrow?", category="")
+    _evaluate(strategy, market)
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["skip_reason"] == "unsupported"
+
+
+def test_record_skip_external_reason():
+    """main.py records reentry-guard skips through record_skip()."""
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    strategy.record_skip(_make_market(0.50), "reentry_guard")
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["skip_reason"] == "reentry_guard"
+    assert rec["estimated_prob"] is None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# Budget flag + cycle lifecycle
+# ─────────────────────────────────────────────────────────────────────────────
+
+def test_news_budget_skipped_flag():
+    """With the cycle budget exhausted, the record must say GNews was never
+    asked — feat_news_lo=0.0 alone would be indistinguishable from no-news."""
+    strategy = BayesianStrategy(news=FakeNews(0.9), manifold=None, db=None)
+    strategy._news_queries_this_cycle = MAX_NEWS_QUERIES_PER_CYCLE
+    _evaluate(strategy, _make_market(0.50))
+
+    rec = strategy.drain_cycle_records()[0]
+    assert rec["news_budget_skipped"] is True
+    assert rec["news_sentiment"] == 0.0
+    assert rec["feat_news_lo"] == 0.0
+
+
+def test_drain_empties_and_reset_clears():
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    _evaluate(strategy, _make_market(0.50))
+    assert len(strategy.drain_cycle_records()) == 1
+    assert strategy.drain_cycle_records() == []
+
+    _evaluate(strategy, _make_market(0.50))
+    strategy.reset_cycle()
+    assert strategy.drain_cycle_records() == []
+
+
+def test_one_record_per_market_accumulates_in_order():
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    _evaluate(strategy, _make_market(0.03, market_id="m1"))   # prior_extreme
+    _evaluate(strategy, _make_market(0.50, market_id="m2"))   # edge_net
+    _evaluate(strategy, _make_market(0.97, market_id="m3"))   # prior_extreme
+
+    records = strategy.drain_cycle_records()
+    assert [r["market_id"] for r in records] == ["m1", "m2", "m3"]
+    assert [r["skip_reason"] for r in records] == [
+        "prior_extreme", "edge_net", "prior_extreme",
+    ]