polymarket-bot/bot/replay.py

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