diff --git a/bot/data/db.py b/bot/data/db.py index fa04fd1..135f620 100644 --- a/bot/data/db.py +++ b/bot/data/db.py @@ -747,6 +747,85 @@ class Database: except (ValueError, IndexError): return 0 + # ── Replay R1: replay core ─────────────────────────────────────────────── + + async def get_replay_cycles(self, from_ts, to_ts) -> list: + """Return the cycle_ts values with archived decisions in [from_ts, to_ts).""" + async with self._pool.acquire() as conn: + rows = await conn.fetch(""" + SELECT DISTINCT cycle_ts FROM signals + WHERE cycle_ts >= $1 AND cycle_ts < $2 + ORDER BY cycle_ts + """, from_ts, to_ts) + return [r["cycle_ts"] for r in rows] + + async def get_ext_snapshot(self, cycle_ts) -> Optional[dict]: + """Return one cycle's ExternalSignals snapshot, or None if missing.""" + async with self._pool.acquire() as conn: + row = await conn.fetchrow( + "SELECT * FROM ext_snapshots WHERE cycle_ts = $1", cycle_ts + ) + return dict(row) if row else None + + async def get_cycle_signal_rows(self, cycle_ts) -> list[dict]: + """Return one cycle's archived decision rows in original evaluation + order (id = insertion order = the order main.py evaluated them).""" + async with self._pool.acquire() as conn: + rows = await conn.fetch( + "SELECT * FROM signals WHERE cycle_ts = $1 ORDER BY id", cycle_ts + ) + return [dict(r) for r in rows] + + async def get_markets_by_ids(self, market_ids: list[str]) -> dict[str, dict]: + """Return market metadata rows keyed by id (for Market reconstruction).""" + if not market_ids: + return {} + async with self._pool.acquire() as conn: + rows = await conn.fetch( + "SELECT * FROM markets WHERE id = ANY($1::text[])", market_ids + ) + return {r["id"]: dict(r) for r in rows} + + async def save_replay_run(self, run: dict) -> None: + async with self._pool.acquire() as conn: + await conn.execute(""" + INSERT INTO replay_runs ( + run_id, git_sha, config_hash, config_json, + from_ts, to_ts, cycles, decisions, matched, mismatched, note + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11) + """, + run["run_id"], run["git_sha"], run["config_hash"], + run["config_json"], run["from_ts"], run["to_ts"], + run["cycles"], run["decisions"], run["matched"], + run["mismatched"], run["note"], + ) + + async def save_replay_decisions(self, run_id: str, decisions: list[dict]) -> None: + if not decisions: + return + rows = [ + ( + run_id, d["cycle_ts"], d["market_id"], + d["skip_reason"], d["prior_prob"], d["estimated_prob"], + d["raw_final_prob"], d["edge_gross"], d["edge_net"], + d["regime_min_edge"], d["days_to_resolution"], + d["confidence"], d["direction"], d["would_trade"], + d["recorded_skip_reason"], d["matched"], d["mismatch_field"], + ) + for d in decisions + ] + async with self._pool.acquire() as conn: + await conn.executemany(""" + INSERT INTO replay_decisions ( + run_id, cycle_ts, market_id, + skip_reason, prior_prob, estimated_prob, + raw_final_prob, edge_gross, edge_net, + regime_min_edge, days_to_resolution, + confidence, direction, would_trade, + recorded_skip_reason, matched, mismatch_field + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) + """, rows) + async def mark_manifold_audit_used(self, audit_id: str) -> None: async with self._pool.acquire() as conn: await conn.execute( diff --git a/bot/data/schema.sql b/bot/data/schema.sql index f6d55d6..3506db4 100644 --- a/bot/data/schema.sql +++ b/bot/data/schema.sql @@ -370,3 +370,64 @@ CREATE TABLE IF NOT EXISTS ext_snapshots ( total_market_cap_change DOUBLE PRECISION, valid BOOLEAN ); + +-- ───────────────────────────────────────────────────────────────────────────── +-- Replay R1: replay core — re-execute evaluate() over the R0 archive +-- +-- A replay run reads cycles from signals + ext_snapshots + markets, rebuilds +-- the exact inputs (including archived news_sentiment — GNews is never called), +-- re-runs BayesianStrategy.evaluate() with the archived cycle_ts as clock, and +-- writes one replay_decisions row per (cycle, market). +-- +-- replay_runs tags every run with the code (git_sha) and strategy constants +-- (config_hash) that produced it: two runs over the same window with different +-- config_hash values are a counterfactual comparison; same config_hash against +-- the recorded rows is a determinism check (mismatches should be 0, modulo +-- day-boundary crossings between cycle_ts and the original wall-clock). +-- +-- matched: replayed decision equals the recorded one (skip_reason, probs, +-- confidence, direction). NULL when not comparable — e.g. reentry_guard +-- rows, recorded outside evaluate() with no decision fields to compare; +-- the replay still re-evaluates them, which is extra calibration data. +-- mismatch_field: first field that differed, for triage. +-- ───────────────────────────────────────────────────────────────────────────── +CREATE TABLE IF NOT EXISTS replay_runs ( + run_id TEXT PRIMARY KEY, + created_at TIMESTAMPTZ DEFAULT NOW(), + git_sha TEXT, + config_hash TEXT, + config_json TEXT, + from_ts TIMESTAMPTZ, + to_ts TIMESTAMPTZ, + cycles INTEGER, + decisions INTEGER, + matched INTEGER, + mismatched INTEGER, + note TEXT +); + +CREATE TABLE IF NOT EXISTS replay_decisions ( + id SERIAL PRIMARY KEY, + run_id TEXT NOT NULL, + cycle_ts TIMESTAMPTZ NOT NULL, + market_id TEXT NOT NULL, + -- replayed outputs (same semantics as the signals columns) + skip_reason TEXT, + prior_prob DOUBLE PRECISION, + estimated_prob DOUBLE PRECISION, + raw_final_prob DOUBLE PRECISION, + edge_gross DOUBLE PRECISION, + edge_net DOUBLE PRECISION, + regime_min_edge DOUBLE PRECISION, + days_to_resolution INTEGER, + confidence DOUBLE PRECISION, + direction TEXT, + would_trade BOOLEAN, + -- fidelity vs the recorded signals row + recorded_skip_reason TEXT, + matched BOOLEAN, + mismatch_field TEXT +); + +CREATE INDEX IF NOT EXISTS idx_replay_decisions_run ON replay_decisions(run_id); +CREATE INDEX IF NOT EXISTS idx_replay_decisions_mkt ON replay_decisions(market_id); diff --git a/bot/replay.py b/bot/replay.py new file mode 100644 index 0000000..6f8ece9 --- /dev/null +++ b/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() diff --git a/bot/strategy/bayesian.py b/bot/strategy/bayesian.py index ebc03e1..8f12443 100644 --- a/bot/strategy/bayesian.py +++ b/bot/strategy/bayesian.py @@ -167,15 +167,22 @@ def _regime_min_edge(category: str, days_to_resolution: int) -> float: return 0.10 # tech, crypto/finance, events, default -def _days_to_resolution(end_date: str) -> int: - """Return calendar days until market resolution, or 30 if unknown.""" +def _days_to_resolution(end_date: str, as_of: Optional[datetime] = None) -> int: + """Return calendar days until market resolution, or 30 if unknown. + + as_of (Replay R1): reference clock for the computation. None (production) + means wall-clock now; a replay run passes the archived cycle_ts so + days-to-resolution — and therefore the regime edge threshold — is computed + against the moment the decision was originally made. + """ if not end_date: return 30 # conservative: treat as medium-term try: dt = datetime.fromisoformat(end_date.replace("Z", "+00:00")) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) - days = (dt - datetime.now(timezone.utc)).days + now = as_of if as_of is not None else datetime.now(timezone.utc) + days = (dt - now).days return max(0, days) except (ValueError, TypeError): return 30 @@ -457,10 +464,17 @@ class BayesianStrategy: market: Market, ext: ExternalSignals, occupied_families: set[str], + as_of: Optional[datetime] = None, ) -> Optional[TradingSignal]: """ Evaluate a market and return a TradingSignal if actionable. + as_of (Replay R1): clock injection — None in production (wall-clock + now); a replay passes the archived cycle_ts so the regime threshold + matches the original decision moment. Only days-to-resolution + depends on the clock; everything else is a pure function of + (market, ext, occupied_families) and the news/manifold clients. + Returns None with a structured log line in all skip cases. Skip reasons (Phase 5 observability): SKIP_UNSUPPORTED — category not supported @@ -558,7 +572,7 @@ class BayesianStrategy: return None # ── Phase 4: regime min-edge ───────────────────────────────────────── - days = _days_to_resolution(market.end_date) + days = _days_to_resolution(market.end_date, as_of) regime_min = _regime_min_edge(category, days) # ── Bayesian probability estimation ────────────────────────────────── diff --git a/tests/test_replay_engine.py b/tests/test_replay_engine.py new file mode 100644 index 0000000..1f79683 --- /dev/null +++ b/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)