docs: README de portfolio — arquitectura, decisiones, resultados y archivado

Reescritura completa como pieza de portfolio: problema/solución, diagrama
Mermaid de arquitectura, decisiones de diseño, stack y sección de estado
del proyecto (archivado por el bloqueo regulatorio de la DGOJ a Polymarket
en España, mayo 2026, con enlaces a las fuentes).

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

.gitea/workflows/ci.yml

@@ -1,9 +1,10 @@
 name: CI/CD
 
+# Decommission (2026-07): trigger switched from push:main to manual only, so
+# documentation commits don't rebuild/redeploy images. Run from the Gitea UI
+# (Actions → Run workflow) if a rebuild is ever needed again.
 on:
-  push:
-    branches:
-      - main
+  workflow_dispatch:
 
 env:
   REGISTRY: gitea.gitea.svc.cluster.local:3000
@@ -19,15 +20,64 @@ jobs:
         uses: actions/checkout@v4
         with:
           ssl-verify: false
+          # Full history: needed to diff against github.event.before
+          fetch-depth: 0
 
       - name: Set image tag
         id: tag
         run: echo "TAG=${GITHUB_SHA::8}" >> $GITHUB_OUTPUT
 
+      - name: Detect changed components
+        id: changes
+        run: |
+          BEFORE="${{ github.event.before }}"
+          CHANGED=""
+          case "$BEFORE" in
+            ""|0000000000000000000000000000000000000000)
+              echo "First push or unknown base — building all images"
+              CHANGED="__all__"
+              ;;
+            *)
+              if git cat-file -e "$BEFORE" 2>/dev/null; then
+                CHANGED=$(git diff --name-only "$BEFORE" "$GITHUB_SHA")
+              else
+                echo "Base commit $BEFORE not in history (force push?) — building all images"
+                CHANGED="__all__"
+              fi
+              ;;
+          esac
+          echo "Changed files:"
+          echo "$CHANGED"
+
+          if [ "$CHANGED" = "__all__" ]; then
+            BOT=true; API=true; DASH=true
+          else
+            BOT=false; API=false; DASH=false
+            matches() { echo "$CHANGED" | grep -qE "$1"; }
+            # The workflow itself affects every image build
+            if matches '^\.gitea/workflows/ci\.yml$'; then BOT=true; API=true; DASH=true; fi
+            # bot and api images both COPY bot/, api/ and requirements.txt
+            if matches '^(bot/|api/|requirements\.txt$)'; then BOT=true; API=true; fi
+            if matches '^Dockerfile$'; then BOT=true; fi
+            if matches '^Dockerfile\.api$'; then API=true; fi
+            # dashboard image builds from the dashboard/ context only
+            if matches '^dashboard/'; then DASH=true; fi
+          fi
+
+          ANY=false
+          if [ "$BOT" = "true" ] || [ "$API" = "true" ] || [ "$DASH" = "true" ]; then ANY=true; fi
+          echo "build_bot=$BOT"        >> $GITHUB_OUTPUT
+          echo "build_api=$API"        >> $GITHUB_OUTPUT
+          echo "build_dashboard=$DASH" >> $GITHUB_OUTPUT
+          echo "build_any=$ANY"        >> $GITHUB_OUTPUT
+          echo "Will build: bot=$BOT api=$API dashboard=$DASH"
+
       - name: Log in to registry
+        if: steps.changes.outputs.build_any == 'true'
         run: echo "${{ secrets.CI_TOKEN }}" | docker login gitea.gitea.svc.cluster.local:3000 -u chemavx --password-stdin
 
       - name: Create buildx builder
+        if: steps.changes.outputs.build_any == 'true'
         run: |
           cat > /tmp/buildkitd.toml << 'EOF'
           [registry."registry-cache.registry-cache.svc.cluster.local:5000"]
@@ -50,6 +100,7 @@ jobs:
           docker buildx inspect --bootstrap
 
       - name: Build and push bot image
+        if: steps.changes.outputs.build_bot == 'true'
         run: |
           TAG=${{ steps.tag.outputs.TAG }}
           docker buildx build \
@@ -61,6 +112,7 @@ jobs:
             -f Dockerfile .
 
       - name: Build and push API image
+        if: steps.changes.outputs.build_api == 'true'
         run: |
           TAG=${{ steps.tag.outputs.TAG }}
           docker buildx build \
@@ -72,6 +124,7 @@ jobs:
             -f Dockerfile.api .
 
       - name: Build and push dashboard image
+        if: steps.changes.outputs.build_dashboard == 'true'
         run: |
           TAG=${{ steps.tag.outputs.TAG }}
           docker buildx build \
@@ -84,6 +137,7 @@ jobs:
             dashboard
 
       - name: Verify images in registry
+        if: steps.changes.outputs.build_any == 'true'
         run: |
           TAG=${{ steps.tag.outputs.TAG }}
           check_image() {
@@ -98,11 +152,18 @@ jobs:
             fi
             echo "OK: chemavx/${image}:${TAG} verified in registry"
           }
+          if [ "${{ steps.changes.outputs.build_bot }}" = "true" ]; then
             check_image polymarket-bot
+          fi
+          if [ "${{ steps.changes.outputs.build_api }}" = "true" ]; then
             check_image polymarket-bot-api
+          fi
+          if [ "${{ steps.changes.outputs.build_dashboard }}" = "true" ]; then
             check_image polymarket-bot-dashboard
+          fi
 
       - name: Update k8s manifests
+        if: steps.changes.outputs.build_any == 'true'
         run: |
           pip3 install pyyaml -q
 
@@ -114,12 +175,21 @@ jobs:
           git clone ${{ env.K8S_MANIFESTS_REPO }} /tmp/k8s-manifests
           cd /tmp/k8s-manifests
 
+          # Only bump the tag of images that were actually rebuilt: the others
+          # 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" \
               polymarket-bot/deployment-api.yaml
+          fi
+          if [ "${{ steps.changes.outputs.build_dashboard }}" = "true" ]; then
             sed -i "s|image: .*polymarket-bot-dashboard.*|image: git.chemavx.xyz/chemavx/polymarket-bot-dashboard:${TAG}|g" \
               polymarket-bot/deployment-dashboard.yaml
+          fi
           sed -i "s|imagePullPolicy: Never|imagePullPolicy: Always|g" \
             polymarket-bot/deployment-bot.yaml \
             polymarket-bot/deployment-api.yaml \
@@ -154,10 +224,21 @@ jobs:
           TAG: ${{ steps.tag.outputs.TAG }}
           JOB_STATUS: ${{ job.status }}
           TELEGRAM_TOKEN: ${{ secrets.TELEGRAM_BOT_TOKEN }}
+          BUILD_BOT: ${{ steps.changes.outputs.build_bot }}
+          BUILD_API: ${{ steps.changes.outputs.build_api }}
+          BUILD_DASH: ${{ steps.changes.outputs.build_dashboard }}
         run: |
           TAG="${TAG:-${GITHUB_SHA:0:8}}"
+          BUILT=""
+          [ "$BUILD_BOT" = "true" ]  && BUILT="${BUILT}bot "
+          [ "$BUILD_API" = "true" ]  && BUILT="${BUILT}api "
+          [ "$BUILD_DASH" = "true" ] && BUILT="${BUILT}dashboard "
           if [ "$JOB_STATUS" = "success" ]; then
-            MSG="✅ Deploy polymarket-bot:${TAG} completado"
+            if [ -n "$BUILT" ]; then
+              MSG="✅ Deploy polymarket-bot:${TAG} completado (imágenes: ${BUILT% })"
+            else
+              MSG="✅ CI polymarket-bot:${TAG} OK — sin cambios de imagen, nada que desplegar"
+            fi
           else
             MSG="❌ Deploy polymarket-bot:${TAG} fallido (status: ${JOB_STATUS})"
           fi

README.md

@@ -0,0 +1,169 @@
+# polymarket-bot
+
+Bot de **paper trading** para mercados de predicción de Polymarket: estrategia
+bayesiana sobre el precio de mercado como prior, señales externas en log-odds,
+gates de riesgo en cascada y un motor de replay determinista para calibración
+offline. API FastAPI + dashboard React. Desplegado en k3s vía GitOps
+(Gitea Actions → registry → ArgoCD), con PostgreSQL, secrets por Infisical y
+observabilidad completa (Grafana, uptime-kuma, Telegram).
+
+> **📦 Estado del proyecto: ARCHIVADO (julio 2026).**
+> El 26 de mayo de 2026 la DGOJ [ordenó el bloqueo cautelar de Polymarket y
+> Kalshi en España](https://www.ordenacionjuego.es/novedades/dgoj-abre-expediente-sancionador-plataformas-polymarket-kalshi-ordena-bloqueo-sus-webs)
+> por operar sin licencia de juego ([CoinDesk](https://www.coindesk.com/policy/2026/05/26/spain-joins-growing-list-of-countries-shutting-out-polymarket-and-kalshi)).
+> Con el bloqueo a nivel de ISP la fuente de datos primaria desaparece, así que
+> el proyecto se jubila de forma ordenada: datos respaldados y verificados,
+> resultados documentados en el [informe final](docs/informe-final.md), e
+> infraestructura apagada vía GitOps. El cierre es parte de la historia del
+> proyecto, no un abandono — los módulos centrales son agnósticos de la fuente
+> y el pivote natural es Manifold Markets (API abierta, play-money, sin
+> fricción regulatoria).
+
+## El problema y la solución
+
+Los mercados de predicción publican probabilidades implícitas en el precio.
+La hipótesis: en mercados de bajo volumen hay ineficiencias detectables
+combinando el precio con señales externas (noticias, sentimiento cripto,
+mercados espejo en otras plataformas). El bot la pone a prueba **sin arriesgar
+dinero**, con la disciplina de un sistema real:
+
+1. **Prior desde el precio** de Polymarket (mid del order book CLOB).
+2. **Ajuste bayesiano en log-odds** con señales independientes: GNews (con
+   presupuesto de 5 consultas/día y *guardrail* que limita el ajuste a
+   prior±0,25), Fear&Greed, momentum BTC, dominancia BTC, y Manifold Markets
+   como señal cruzada (en modo observacional tras detectarse un bug de
+   inversión — auditado en 165k filas).
+3. **Gates en cascada** antes de operar: prior extremo (<0,08 / >0,92), edge
+   bruto mínimo por régimen de volatilidad, edge neto positivo tras comisión
+   (2%) y spread, confianza ≥0,55, conflicto de familias, guard de reentrada.
+4. **Agrupación por familias de mercados**: mercados hermanos del mismo evento
+   (candidatos de una misma elección, umbrales de un mismo precio) comparten
+   `family_key`; solo se mantiene la posición de mayor edge para no apilar la
+   misma apuesta con dos nombres.
+5. **Sizing por Kelly fraccionado** con techo por posición sobre un bankroll
+   simulado de $10.000.
+
+Resultado honesto en 81 días: 12 trades (n estadísticamente irrelevante, y el
+informe lo subraya), realized **+$247,78**, y — más interesante — **~223.000
+evaluaciones archivadas con su decisión reproducible**, porque el sistema
+prefirió no operar antes que operar sin ventaja medible.
+Detalle completo: [docs/informe-final.md](docs/informe-final.md).
+
+## Arquitectura
+
+```mermaid
+flowchart LR
+    subgraph ext["Fuentes externas"]
+        PM["Polymarket<br/>CLOB + Gamma API"]
+        GN["GNews"]
+        MF["Manifold Markets"]
+        CG["CoinGecko /<br/>alternative.me"]
+    end
+
+    subgraph k3s["k3s · namespace polymarket-bot"]
+        BOT["bot (ciclo ~64s)<br/>estrategia bayesiana + gates"]
+        PG[("PostgreSQL 16<br/>trades · signals · replay")]
+        API["api (FastAPI :8000)"]
+        DASH["dashboard (React/nginx)"]
+        CJ1["CronJob metrics-retention 00:10"]
+        CJ2["CronJob outcomes-joiner 00:30"]
+    end
+
+    subgraph obs["Observabilidad"]
+        GRAF["Grafana"]
+        UK["uptime-kuma"]
+        TG["Telegram"]
+    end
+
+    PM --> BOT
+    GN --> BOT
+    MF --> BOT
+    CG --> BOT
+    BOT --> PG
+    PG --> API
+    API --> DASH
+    API --> GRAF
+    UK --> DASH
+    BOT --> TG
+    PM --> CJ2
+    CJ2 --> PG
+    CJ1 --> PG
+
+    subgraph ci["GitOps"]
+        GA["Gitea Actions"] --> REG["registry"] --> ARGO["ArgoCD<br/>prune + selfHeal"]
+    end
+    ARGO -.-> k3s
+```
+
+## Decisiones de diseño
+
+- **Paper trading por diseño, no como demo**: el bot exigía ≥10 mercados
+  resueltos y ≥30 días antes de considerarse promocionable a dinero real, y
+  reportaba `n/a (insufficient_sample)` en vez de métricas infladas.
+- **Todo skip es un dato**: cada evaluación archiva prior, estimación, edge
+  bruto/neto, descomposición por feature en log-odds y el gate exacto que la
+  bloqueó (~55k filas/día en `signals`).
+- **Replay determinista** (julio 2026): inyección de reloj + re-ejecución de
+  ciclos archivados con 22.697/22.697 decisiones idénticas; joiner diario de
+  resoluciones UMA para calibrar sobre *todas* las evaluaciones, no solo los
+  trades.
+- **Los errores se cierran y se documentan**: el bug de inversión de Manifold
+  y los conflictos de familia cerraron 5 posiciones en abril; la señal pasó a
+  observacional-only con auditoría completa en vez de eliminarse.
+- **GitOps estricto**: nunca `kubectl` para cambios persistentes (ArgoCD con
+  `selfHeal` revierte cualquier parche manual); secrets fuera de git vía
+  Infisical operator; smoke test PostSync con notificación a Telegram.
+
+## Componentes
+
+| Componente | Código | Imagen | CMD |
+|---|---|---|---|
+| bot | `bot/` | `polymarket-bot` | `python3 -m bot.main` |
+| api | `api/` (+ `bot/` como librería) | `polymarket-bot-api` | `uvicorn api.main:app` |
+| dashboard | `dashboard/` | `polymarket-bot-dashboard` | nginx estático |
+
+Dashboard (hasta el apagado): https://polymarket.chemavx.xyz
+
+## Stack
+
+Python 3 (asyncio) · FastAPI · React + Vite · PostgreSQL 16 · k3s · ArgoCD ·
+Gitea Actions + BuildKit · Infisical (secrets) · Grafana + Prometheus ·
+uptime-kuma · Telegram Bot API.
+
+## CI/CD
+
+> ⚠️ **Decomisión**: el trigger automático `push:main` se desactivó en la Fase 1
+> del archivado; el workflow queda solo bajo `workflow_dispatch` manual.
+
+`.gitea/workflows/ci.yml` construye **solo las imágenes cuyos ficheros
+cambiaron** en el push (diff contra `github.event.before`):
+
+- `bot/`, `api/`, `requirements.txt` → bot + api (ambas imágenes copian las
+  mismas fuentes Python; solo cambia el CMD)
+- `Dockerfile` → bot · `Dockerfile.api` → api · `dashboard/` → dashboard
+- `.gitea/workflows/ci.yml`, primer push o force-push → todas (fallback seguro)
+- `tests/`, docs → ninguna (la CI no construye ni despliega nada)
+
+Las imágenes se tagean con `${GITHUB_SHA::8}`; el CI actualiza solo los
+deployments reconstruidos en `k8s-manifests/polymarket-bot/` y ArgoCD
+sincroniza vía webhook en segundos.
+
+## Tests
+
+```bash
+python3 -m pytest tests/ -q
+```
+
+## Datos y backup
+
+Base de datos respaldada y verificada (2026-07-06): `pg_dump -Fc` + exports
+CSV.gz de las tablas de señales y auditoría, con checksums y restore de prueba
+(recuentos idénticos 11/11 tablas). Ubicación: almacenamiento de backups del
+clúster con sync offsite. El dump final se toma en la fase de apagado.
+
+## Documentación
+
+- [Informe final de paper trading](docs/informe-final.md) — resultados,
+  pipeline de datos y estado de la BD al cierre.
+- `docs/pivot-manifold.md` — notas de diseño del posible pivote (Fase 5,
+  pendiente).

api/main.py

@@ -11,6 +11,12 @@ from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
 from bot.data.db import Database
+from bot.executor.paper import cash_available
+from bot.metrics.sharpe import (
+    MIN_DAYS_OBSERVED,
+    MIN_RESOLVED_TRADES,
+    sharpe_with_gate,
+)
 
 # Phase 6 format (Phase 6+): values already in log-odds space.
 # "fg_lo=+0.1200 mom_lo=+0.0000 news_lo=+0.0000 mfld_lo=-0.7483 btc_dom_lo=+0.0000"
@@ -209,6 +215,66 @@ async def get_attribution():
     return {"attribution": attribution, "total_attributed_trades": total}
 
 
+@app.get("/api/metrics/manifold-matches")
+async def get_manifold_matches():
+    """Manifold match audit — version-split summary and recent match attempts.
+
+    summary.current_version — stats for the active matcher (MANIFOLD_MATCHER_VERSION):
+      version          — the matcher version string
+      total_accepted   — matches accepted (score >= 0.40, inversion unambiguous)
+      total_rejected   — matches rejected (low score or ambiguous inversion)
+      total_no_results — no Manifold market found or API error
+      avg_match_score  — mean Jaccard score for accepted matches
+      used_in_trade    — accepted matches that were actually executed
+    summary.all_time — accepted/rejected/no_results across every matcher version.
+    summary.legacy.accepted_without_outcome_type — pre-outcome-guard accepted
+      records that the current matcher would reject (not counted in current_version).
+    summary.trades_dominated_by_mfld — non-excluded accepted-match trades where
+      feat_mfld_lo is the largest signal (consistent with attribution/features,
+      which also exclude excluded_from_metrics trades).
+    summary.unique_markets — distinct-market coverage (per-market, not per-attempt):
+      evaluated      — DISTINCT poly_market_id in manifold_match_audit (all versions)
+      accepted       — DISTINCT poly_market_id accepted by the current matcher
+      coverage_rate  — accepted / evaluated (null when evaluated=0)
+
+    recent_matches: last 50 rows from manifold_match_audit, newest first, each
+      tagged with matcher_version.
+      used_in_trade=True only when status='accepted' AND a trade was actually executed.
+    """
+    data = await db.get_manifold_matches(limit=50)
+    for match in data["recent_matches"]:
+        ts = match.get("timestamp")
+        if ts is not None and hasattr(ts, "isoformat"):
+            match["timestamp"] = ts.isoformat()
+    return data
+
+
+@app.get("/api/metrics/manifold-coverage")
+async def get_manifold_coverage():
+    """Manifold coverage by semantic market category, counted by UNIQUE market.
+
+    Unlike the raw audit counters (which count per-attempt and are inflated by the
+    trading loop re-evaluating the same markets), this measures real coverage:
+    how many DISTINCT markets in each category the matcher found an accepted
+    Manifold counterpart for. Base table is manifold_match_audit filtered to the
+    current matcher (v3_outcome_guard); category is inferred from trade family_key
+    when available, otherwise from the question text.
+
+    coverage_by_category — one entry per category, ordered by unique_evaluated DESC:
+      category          — gubernatorial | mayoral | senate | primary-republican |
+                          primary-democrat | big-tech | geopolitics | other
+      unique_evaluated  — distinct markets audited in this category
+      unique_accepted   — distinct markets with at least one accepted match
+      unique_rejected   — distinct markets with at least one rejected match
+      unique_no_results — distinct markets with at least one no_results outcome
+      coverage_rate     — unique_accepted / unique_evaluated (null if evaluated=0)
+    summary — total_unique_evaluated, total_unique_accepted, overall_coverage_rate
+      (null if nothing evaluated), categories_with_coverage (categories with
+      unique_accepted > 0).
+    """
+    return await db.get_manifold_coverage_by_category()
+
+
 @app.get("/api/summary")
 async def get_summary():
     """Dashboard summary card data.
@@ -219,28 +285,49 @@ async def get_summary():
     PnL and performance metrics come from the latest metrics_daily snapshot,
     which is written by the bot every cycle via MetricsTracker.update_daily_summary().
     After Fix 3, that snapshot is also DB-computed — not dependent on pod restarts.
+    sharpe_ratio is the exception: it is recomputed live here from the daily
+    PnL-close series (same sharpe_with_gate the tracker uses), so the
+    explanation fields (sharpe_status, days_observed) always match the value.
     """
-    latest_metrics, open_trades, all_trades, inverted, legacy_count = await asyncio.gather(
+    latest_metrics, counts, position_data, inverted, legacy_count, daily_closes = (
+        await asyncio.gather(
             db.get_metrics_history(days=1),
-        db.get_recent_trades(limit=500, status="open"),
-        db.get_recent_trades(limit=500),
+            db.compute_metrics_from_db(),
+            db.get_open_position_data(),
             db.get_recently_closed_inverted(hours=24),
             db.get_legacy_incomplete_count(),
+            db.get_daily_pnl_closes(),
+        )
     )
 
     latest = latest_metrics[0] if latest_metrics else {}
     paper_bankroll = float(os.getenv("PAPER_BANKROLL", "10000"))
-    total_deployed = sum(t.get("net_cost", 0) for t in open_trades)
+    total_trades = int(counts["total_trades"] or 0)
+    resolved_count = int(counts.get("resolved_count") or 0)
+    # Same source PaperExecutor.initialize() uses to reconstruct cash:
+    # total_net_cost_open = SUM(net_cost) over open trades, uncapped.
+    _, total_net_cost_open = position_data
+    total_deployed = total_net_cost_open
+
+    # Sharpe: computed live from the daily PnL curve (same function the
+    # tracker uses for the snapshot). None + status while the minimum-sample
+    # gate (>=30 days observed, >=10 resolved trades) is not met — a Sharpe
+    # over 1 resolved trade is statistically meaningless.
+    days_observed = len(daily_closes)
+    sharpe, sharpe_status = sharpe_with_gate(daily_closes, paper_bankroll, resolved_count)
+
+    win_rate = latest.get("win_rate")
+    calibration = latest.get("calibration_score")
 
     return {
         # ── Portfolio state (live from DB) ──────────────────────────────────
         "paper_mode":              os.getenv("PAPER_MODE", "true") == "true",
         "paper_bankroll":          paper_bankroll,
-        "total_trades":            len(all_trades),           # exact, from DB
-        "open_trades_count":       len(open_trades),          # exact, from DB
-        "closed_trades_count":     len(all_trades) - len(open_trades),  # exact
+        "total_trades":            total_trades,              # COUNT(*), uncapped
+        "open_trades_count":       int(counts["open_count"] or 0),    # COUNT(*), uncapped
+        "closed_trades_count":     int(counts["closed_count"] or 0),  # COUNT(*), uncapped
         "total_deployed":          total_deployed,            # exact, from DB
-        "cash_available":          max(0.0, paper_bankroll - total_deployed),  # exact
+        "cash_available":          cash_available(paper_bankroll, total_net_cost_open),
         "legacy_incomplete_count": legacy_count,              # exact, from DB
         "reentry_guard_blocks_24h": len(inverted),            # exact, from DB
 
@@ -248,31 +335,41 @@ async def get_summary():
         # unrealized_pnl_est: open positions, edge_net × net_cost − fee.
         #   Estimated — uses model signal, not live price. Source: open trades.
         # realized_pnl: closed positions with known resolution.
-        #   Exact — computed from (resolution − entry_price) × shares.
+        #   Exact — payout − net_cost per trade (net of fee), matches logs/Telegram.
         # total_pnl: sum of both.
         "unrealized_pnl_est": latest.get("unrealized_pnl_est") or 0,
         "realized_pnl":       latest.get("realized_pnl") or 0,
         "total_pnl":          latest.get("total_pnl") or 0,
 
-        # ── Performance metrics (from latest metrics_daily snapshot) ─────────
+        # ── Performance metrics ──────────────────────────────────────────────
         # win_rate: fraction of resolved closed trades where close_pnl > 0.
         #   null if fewer than 5 resolved trades. Source: closed+resolved trades.
-        # sharpe_ratio: 0.0 — requires daily-return time series (not yet tracked).
+        # sharpe_ratio: annualized Sharpe of the daily total_pnl curve, computed
+        #   live from metrics_daily. null while the minimum-sample gate fails
+        #   (sharpe_status explains why). Source: bot/metrics/sharpe.py.
         # calibration_score: 1 − Brier score on resolved trades (higher = better).
         #   null if fewer than 10 resolved trades. Source: closed+resolved trades.
-        "win_rate":           latest.get("win_rate"),         # null if < 5 resolved
-        "sharpe_ratio":       latest.get("sharpe_ratio") or 0,  # 0.0 until tracked
-        "calibration_score":  latest.get("calibration_score"),  # null if < 10 resolved
+        "win_rate":           win_rate,                       # null if < 5 resolved
+        "sharpe_ratio":       sharpe,                         # null if gate fails
+        "sharpe_status":      sharpe_status,  # ok | insufficient_sample | zero_variance
+        "days_observed":      days_observed,
+        "min_days_required":  MIN_DAYS_OBSERVED,
+        "min_resolved_required": MIN_RESOLVED_TRADES,
+        "calibration_score":  calibration,                    # null if < 10 resolved
 
-        # ── Counters from snapshot ───────────────────────────────────────────
-        "resolved_count":     latest.get("resolved_count") or 0,
+        # ── Counters (live from DB) ──────────────────────────────────────────
+        "resolved_count":     resolved_count,
 
         # ── Promotion gate ───────────────────────────────────────────────────
-        # All thresholds must pass; null metrics count as not-ready.
+        # Never promote on a tiny sample: requires the resolved/days minimums
+        # AND non-null metrics AND all thresholds. A single lucky resolved
+        # trade must not flip this to true.
         "promotion_ready": (
-            (latest.get("sharpe_ratio") or 0) >= 0.5
-            and (latest.get("win_rate") or 0) >= 0.52
-            and (latest.get("calibration_score") or 0) >= 0.7
-            and len(all_trades) >= 50
+            resolved_count >= MIN_RESOLVED_TRADES
+            and days_observed >= MIN_DAYS_OBSERVED
+            and win_rate is not None and win_rate >= 0.52
+            and calibration is not None and calibration >= 0.7
+            and sharpe is not None and sharpe >= 0.5
+            and total_trades >= 50
         ),
     }

bot/data/db.py

@@ -4,6 +4,8 @@ import os
 from typing import Optional
 import asyncpg
 
+from bot.data.manifold import MANIFOLD_MATCHER_VERSION
+
 log = logging.getLogger(__name__)
 
 
@@ -36,11 +38,15 @@ class Database:
                     entry_price, shares, fee_usdc, net_cost, timestamp, reasoning, paper,
                     edge_gross, edge_net, prior_prob, final_prob,
                     mid_price, spread_estimate, commission, family_key,
-                    feat_fg_lo, feat_mom_lo, feat_news_lo, feat_mfld_lo, feat_btc_dom_lo
+                    feat_fg_lo, feat_mom_lo, feat_news_lo, feat_mfld_lo, feat_btc_dom_lo,
+                    mfld_market_id, mfld_market_title, mfld_market_url,
+                    mfld_prob_raw, mfld_prob_final, mfld_inverted,
+                    mfld_match_score, mfld_match_reason, mfld_match_status
                 ) 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
+                    $21,$22,$23,$24,$25,
+                    $26,$27,$28,$29,$30,$31,$32,$33,$34
                 )
                 ON CONFLICT (id) DO NOTHING
             """,
@@ -53,6 +59,10 @@ class Database:
                 # Phase 6 feature log-odds
                 trade.feat_fg_lo, trade.feat_mom_lo, trade.feat_news_lo,
                 trade.feat_mfld_lo, trade.feat_btc_dom_lo,
+                # Manifold audit fields
+                trade.mfld_market_id, trade.mfld_market_title, trade.mfld_market_url,
+                trade.mfld_prob_raw, trade.mfld_prob_final, trade.mfld_inverted,
+                trade.mfld_match_score, trade.mfld_match_reason, trade.mfld_match_status,
             )
 
     async def save_daily_metrics(self, metrics: dict) -> None:
@@ -142,6 +152,20 @@ class Database:
             """)
             return [dict(r) for r in rows]
 
+    async def get_open_trades_for_market(self, market_id: str) -> list[dict]:
+        """Return direction, shares and net_cost for each open trade in a market.
+
+        Used by PaperExecutor.close_position() to compute the settlement
+        payout per direction (BUY_NO pays out when resolution = 0.0).
+        """
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch(
+                "SELECT direction, shares, net_cost FROM trades "
+                "WHERE market_id = $1 AND closed_at IS NULL",
+                market_id,
+            )
+            return [dict(r) for r in rows]
+
     async def close_paper_position(
         self, market_id: str, reason: str = "", resolution: Optional[float] = None
     ) -> None:
@@ -149,19 +173,29 @@ class Database:
 
         resolution: 1.0 if YES resolved, 0.0 if NO resolved, None if unknown
         (legacy closes, inversion fixes). When resolution is provided, close_pnl
-        is computed in SQL so it matches the stored entry_price and shares exactly.
+        is computed in SQL per row as payout − net_cost — NET of fee, the single
+        PnL definition shared with PaperExecutor.close_position() (logs/Telegram):
+          BUY_YES: resolution * shares − net_cost
+          BUY_NO:  (1 − resolution) * shares − net_cost
+        paper.py aggregates payout − net_cost over these same open rows, so
+        SUM(close_pnl) per market equals the pnl it reports exactly. The
+        aggregate is intentionally NOT passed in as a parameter: writing it to
+        every row would double-count markets with more than one open trade.
         """
         async with self._pool.acquire() as conn:
+            # $3 is cast on every use: Postgres cannot infer the parameter type
+            # from a bare "$3 IS NOT NULL" and fails the prepare with
+            # AmbiguousParameterError otherwise.
             await conn.execute("""
                 UPDATE trades
                 SET closed_at    = NOW(),
                     close_reason = $2,
-                    resolution   = $3,
+                    resolution   = $3::double precision,
                     close_pnl    = CASE
-                        WHEN $3 IS NOT NULL AND direction = 'BUY_YES'
-                            THEN ($3::double precision - entry_price) * shares
-                        WHEN $3 IS NOT NULL AND direction = 'BUY_NO'
-                            THEN ((1.0 - $3::double precision) - entry_price) * shares
+                        WHEN $3::double precision IS NOT NULL AND direction = 'BUY_YES'
+                            THEN ($3::double precision * shares) - net_cost
+                        WHEN $3::double precision IS NOT NULL AND direction = 'BUY_NO'
+                            THEN ((1.0 - $3::double precision) * shares) - net_cost
                         ELSE NULL
                     END
                 WHERE market_id = $1 AND closed_at IS NULL
@@ -218,8 +252,14 @@ class Database:
                   COUNT(*)                                                AS total_trades,
                   COUNT(*) FILTER (WHERE closed_at IS NULL)              AS open_count,
                   COUNT(*) FILTER (WHERE closed_at IS NOT NULL)          AS closed_count,
+                  -- excluded_from_metrics trades are omitted from resolved_count,
+                  -- realized_pnl, wins_realized, and calibration_score.
+                  -- resolved_count does NOT require final_prob: legacy trades
+                  -- without signal data still count as resolved. Calibration
+                  -- below keeps the final_prob requirement (it needs the
+                  -- estimated probability to score).
                   COUNT(*) FILTER (WHERE resolution IS NOT NULL
-                                     AND final_prob IS NOT NULL)         AS resolved_count,
+                                     AND (excluded_from_metrics IS NOT TRUE)) AS resolved_count,
 
                   COALESCE(SUM(net_cost)
                     FILTER (WHERE closed_at IS NULL), 0)                 AS total_deployed,
@@ -232,15 +272,17 @@ class Database:
                     FILTER (WHERE closed_at IS NULL
                               AND edge_net IS NOT NULL), 0)              AS unrealized_pnl_est,
 
-                  -- Realized PnL: closed trades with a known resolution.
-                  -- close_pnl is computed at close time from actual resolution.
+                  -- Realized PnL: admin-excluded trades omitted (close_pnl=0 by convention
+                  -- but excluded explicitly so they don't skew the aggregate).
                   COALESCE(SUM(close_pnl)
                     FILTER (WHERE closed_at IS NOT NULL
-                              AND close_pnl IS NOT NULL), 0)             AS realized_pnl,
+                              AND close_pnl IS NOT NULL
+                              AND (excluded_from_metrics IS NOT TRUE)), 0) AS realized_pnl,
 
                   COUNT(*) FILTER (WHERE closed_at IS NOT NULL
                                      AND close_pnl IS NOT NULL
-                                     AND close_pnl > 0)                  AS wins_realized,
+                                     AND close_pnl > 0
+                                     AND (excluded_from_metrics IS NOT TRUE)) AS wins_realized,
 
                   -- Calibration (Brier score transformed to higher-is-better):
                   -- 1 − AVG((final_prob − resolution)²) on resolved trades.
@@ -248,12 +290,15 @@ class Database:
                   -- resolution is 1.0 (YES won) or 0.0 (NO won).
                   -- Perfect calibration → 1.0 | Random → ~0.75 | Worst → 0.0
                   -- Returns NULL if fewer than 10 resolved trades with final_prob.
+                  -- Admin-excluded trades omitted from both threshold and average.
                   CASE
                     WHEN COUNT(*) FILTER (WHERE resolution IS NOT NULL
-                                            AND final_prob IS NOT NULL) >= 10
+                                            AND final_prob IS NOT NULL
+                                            AND (excluded_from_metrics IS NOT TRUE)) >= 10
                     THEN 1.0 - AVG((final_prob - resolution) * (final_prob - resolution))
                                FILTER (WHERE resolution IS NOT NULL
-                                        AND final_prob IS NOT NULL)
+                                        AND final_prob IS NOT NULL
+                                        AND (excluded_from_metrics IS NOT TRUE))
                     ELSE NULL
                   END                                                     AS calibration_score
 
@@ -285,12 +330,42 @@ class Database:
             return result
 
     async def get_metrics_history(self, days: int = 42) -> list[dict]:
+        """Return the closing snapshot of each UTC day, newest day first.
+
+        metrics_daily receives one snapshot per trading cycle (~1/min), so a
+        plain LIMIT over raw rows would cover minutes, not days. DISTINCT ON
+        collapses each calendar day to its last snapshot, making `days` bound
+        actual days. history[0] remains the most recent snapshot overall.
+        """
         async with self._pool.acquire() as conn:
             rows = await conn.fetch(
-                "SELECT * FROM metrics_daily ORDER BY timestamp DESC LIMIT $1", days
+                """
+                SELECT DISTINCT ON (timestamp::date) *
+                FROM metrics_daily
+                ORDER BY timestamp::date DESC, timestamp DESC
+                LIMIT $1
+                """, days
             )
             return [dict(r) for r in rows]
 
+    async def get_daily_pnl_closes(self) -> list[float]:
+        """Return the closing total_pnl of every observed UTC day, oldest first.
+
+        One value per calendar day with at least one metrics_daily snapshot
+        (the day's last snapshot, same collapse rule as get_metrics_history).
+        This is the input series for the Sharpe ratio: len() = days observed,
+        consecutive deltas = daily PnL changes.
+        """
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch(
+                """
+                SELECT DISTINCT ON (timestamp::date) total_pnl
+                FROM metrics_daily
+                ORDER BY timestamp::date ASC, timestamp DESC
+                """
+            )
+            return [float(r["total_pnl"] or 0) for r in rows]
+
     async def backfill_feature_columns(self) -> int:
         """Back-populate feat_*_lo for trades created before Phase 6.
 
@@ -360,22 +435,27 @@ class Database:
                          feat_fg_lo AS fval,
                          edge_net, net_cost, fee_usdc, closed_at, close_pnl
                   FROM trades WHERE feat_fg_lo IS NOT NULL
+                    AND (excluded_from_metrics IS NOT TRUE)
                   UNION ALL
                   SELECT 'mom', 0.05, feat_mom_lo,
                          edge_net, net_cost, fee_usdc, closed_at, close_pnl
                   FROM trades WHERE feat_mom_lo IS NOT NULL
+                    AND (excluded_from_metrics IS NOT TRUE)
                   UNION ALL
                   SELECT 'news', 0.10, feat_news_lo,
                          edge_net, net_cost, fee_usdc, closed_at, close_pnl
                   FROM trades WHERE feat_news_lo IS NOT NULL
+                    AND (excluded_from_metrics IS NOT TRUE)
                   UNION ALL
                   SELECT 'mfld', 0.10, feat_mfld_lo,
                          edge_net, net_cost, fee_usdc, closed_at, close_pnl
                   FROM trades WHERE feat_mfld_lo IS NOT NULL
+                    AND (excluded_from_metrics IS NOT TRUE)
                   UNION ALL
                   SELECT 'btc_dom', 0.05, feat_btc_dom_lo,
                          edge_net, net_cost, fee_usdc, closed_at, close_pnl
                   FROM trades WHERE feat_btc_dom_lo IS NOT NULL
+                    AND (excluded_from_metrics IS NOT TRUE)
                 )
                 SELECT
                   feature,
@@ -459,6 +539,7 @@ class Database:
                     ) AS dominant
                   FROM trades
                   WHERE feat_fg_lo IS NOT NULL
+                    AND (excluded_from_metrics IS NOT TRUE)
                 )
                 SELECT
                   COALESCE(dominant, 'none')                                          AS dominant_feature,
@@ -493,6 +574,530 @@ class Database:
         return result
 
 
+    async def save_manifold_audit(
+        self,
+        audit_id: str,
+        poly_market_id: str,
+        poly_question: str,
+        search_query: str,
+        mfld_market_id: Optional[str],
+        mfld_market_title: Optional[str],
+        mfld_market_url: Optional[str],
+        prob_raw: Optional[float],
+        prob_final: Optional[float],
+        inverted: bool,
+        match_score: Optional[float],
+        match_reason: Optional[str],
+        match_status: str,
+        poly_outcome_type: Optional[str] = None,
+        mfld_outcome_type: Optional[str] = None,
+        matcher_version: Optional[str] = None,
+    ) -> None:
+        async with self._pool.acquire() as conn:
+            await conn.execute("""
+                INSERT INTO manifold_match_audit (
+                    id, poly_market_id, poly_question, search_query,
+                    mfld_market_id, mfld_market_title, mfld_market_url,
+                    prob_raw, prob_final, inverted,
+                    match_score, match_reason, match_status, used_in_trade,
+                    poly_outcome_type, mfld_outcome_type, matcher_version
+                ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,FALSE,$14,$15,$16)
+            """,
+                audit_id, poly_market_id, poly_question, search_query,
+                mfld_market_id, mfld_market_title, mfld_market_url,
+                prob_raw, prob_final, inverted,
+                match_score, match_reason, match_status,
+                poly_outcome_type, mfld_outcome_type, matcher_version,
+            )
+
+    async def get_manifold_cooldown(self, poly_market_id: str) -> Optional[dict]:
+        """Return the cooldown row for a market, or None if no cooldown is recorded.
+
+        The caller decides whether the cooldown is still active by comparing
+        now() against retry_after (see bayesian.evaluate()).
+        """
+        async with self._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT poly_market_id, last_evaluated_at, last_status, "
+                "retry_after, cooldown_reason FROM manifold_eval_cooldown "
+                "WHERE poly_market_id = $1",
+                poly_market_id,
+            )
+        return dict(row) if row else None
+
+    async def upsert_manifold_cooldown(
+        self,
+        poly_market_id: str,
+        last_status: str,
+        retry_after,
+        cooldown_reason: Optional[str] = None,
+    ) -> None:
+        """Insert or refresh the cooldown for a market after a real evaluation.
+
+        last_evaluated_at is stamped server-side with now(); retry_after is the
+        caller-computed earliest re-evaluation time.
+        """
+        async with self._pool.acquire() as conn:
+            await conn.execute("""
+                INSERT INTO manifold_eval_cooldown (
+                    poly_market_id, last_evaluated_at, last_status,
+                    retry_after, cooldown_reason
+                ) VALUES ($1, now(), $2, $3, $4)
+                ON CONFLICT (poly_market_id) DO UPDATE SET
+                    last_evaluated_at = now(),
+                    last_status       = EXCLUDED.last_status,
+                    retry_after       = EXCLUDED.retry_after,
+                    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(
+                "UPDATE manifold_match_audit SET used_in_trade = TRUE WHERE id = $1",
+                audit_id,
+            )
+
+    async def get_manifold_matches(self, limit: int = 50) -> dict:
+        """Manifold match audit, with summary split by matcher version.
+
+        The summary separates the current matcher (MANIFOLD_MATCHER_VERSION) from
+        all-time totals and from legacy pre-outcome-guard records, whose accepted
+        matches would now be rejected by the outcome-compatibility guard and so
+        must not be conflated with current-version stats.
+        """
+        async with self._pool.acquire() as conn:
+            current = await conn.fetchrow("""
+                SELECT
+                    COUNT(*) FILTER (WHERE match_status = 'accepted')   AS total_accepted,
+                    COUNT(*) FILTER (WHERE match_status = 'rejected')   AS total_rejected,
+                    COUNT(*) FILTER (WHERE match_status = 'no_results') AS total_no_results,
+                    AVG(match_score) FILTER (WHERE match_status = 'accepted') AS avg_match_score,
+                    COUNT(*) FILTER (WHERE used_in_trade = TRUE)        AS used_in_trade
+                FROM manifold_match_audit
+                WHERE matcher_version = $1
+            """, MANIFOLD_MATCHER_VERSION)
+            all_time = await conn.fetchrow("""
+                SELECT
+                    COUNT(*) FILTER (WHERE match_status = 'accepted')   AS total_accepted,
+                    COUNT(*) FILTER (WHERE match_status = 'rejected')   AS total_rejected,
+                    COUNT(*) FILTER (WHERE match_status = 'no_results') AS total_no_results
+                FROM manifold_match_audit
+            """)
+            legacy = await conn.fetchrow("""
+                SELECT COUNT(*) AS accepted_without_outcome_type
+                FROM manifold_match_audit
+                WHERE matcher_version = 'legacy_pre_outcome_guard'
+                  AND match_status = 'accepted'
+            """)
+            unique_markets = await conn.fetchrow("""
+                SELECT
+                    COUNT(DISTINCT poly_market_id) AS evaluated,
+                    COUNT(DISTINCT poly_market_id) FILTER (
+                        WHERE match_status = 'accepted'
+                          AND matcher_version = $1
+                    ) AS accepted
+                FROM manifold_match_audit
+            """, MANIFOLD_MATCHER_VERSION)
+            mfld_dominated = await conn.fetchrow("""
+                SELECT COUNT(*) AS cnt FROM trades
+                WHERE (excluded_from_metrics IS NOT TRUE)
+                  AND mfld_match_status = 'accepted'
+                  AND feat_mfld_lo IS NOT NULL
+                  AND ABS(feat_mfld_lo) > 0.0001
+                  AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_fg_lo, 0))
+                  AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_mom_lo, 0))
+                  AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_news_lo, 0))
+                  AND ABS(feat_mfld_lo) > ABS(COALESCE(feat_btc_dom_lo, 0))
+            """)
+            rows = await conn.fetch(
+                "SELECT * FROM manifold_match_audit ORDER BY timestamp DESC LIMIT $1",
+                limit,
+            )
+        return {
+            "summary": {
+                "current_version": {
+                    "version":          MANIFOLD_MATCHER_VERSION,
+                    "total_accepted":   int(current["total_accepted"] or 0),
+                    "total_rejected":   int(current["total_rejected"] or 0),
+                    "total_no_results": int(current["total_no_results"] or 0),
+                    "avg_match_score":  _f(current["avg_match_score"]),
+                    "used_in_trade":    int(current["used_in_trade"] or 0),
+                },
+                "all_time": {
+                    "total_accepted":   int(all_time["total_accepted"] or 0),
+                    "total_rejected":   int(all_time["total_rejected"] or 0),
+                    "total_no_results": int(all_time["total_no_results"] or 0),
+                },
+                "legacy": {
+                    "accepted_without_outcome_type":
+                        int(legacy["accepted_without_outcome_type"] or 0),
+                },
+                "trades_dominated_by_mfld": int(mfld_dominated["cnt"] or 0),
+                "unique_markets": {
+                    "evaluated": int(unique_markets["evaluated"] or 0),
+                    "accepted":  int(unique_markets["accepted"] or 0),
+                    "coverage_rate": (
+                        float(unique_markets["accepted"]) / float(unique_markets["evaluated"])
+                        if unique_markets["evaluated"] else None
+                    ),
+                },
+            },
+            "recent_matches": [dict(r) for r in rows],
+        }
+
+    async def get_manifold_coverage_by_category(self) -> dict:
+        """Manifold coverage by semantic market category, counted by UNIQUE market.
+
+        Base table is manifold_match_audit filtered to the current matcher
+        (v3_outcome_guard).  Each poly_market_id is collapsed to one row first, so
+        a market is counted once regardless of how many audit attempts or trades it
+        has — this measures coverage, not retry volume.
+
+        Category is inferred from the market's trade family_key when available, else
+        from poly_question (LEFT JOIN: audited markets that never produced a trade
+        are kept).  Buckets (accepted/rejected/no_results) are not mutually exclusive
+        at the market level — a market that was no_results then later rejected counts
+        in both — matching COUNT(DISTINCT CASE WHEN status=...) semantics.
+        """
+        async with self._pool.acquire() as conn:
+            rows = await conn.fetch("""
+                WITH audit AS (
+                    SELECT
+                        poly_market_id,
+                        MAX(poly_question)                   AS poly_question,
+                        bool_or(match_status = 'accepted')   AS has_accepted,
+                        bool_or(match_status = 'rejected')   AS has_rejected,
+                        bool_or(match_status = 'no_results') AS has_no_results
+                    FROM manifold_match_audit
+                    WHERE matcher_version = 'v3_outcome_guard'
+                    GROUP BY poly_market_id
+                ),
+                fam AS (
+                    SELECT market_id, MAX(family_key) AS family_key
+                    FROM trades
+                    GROUP BY market_id
+                ),
+                categorized AS (
+                    SELECT
+                        a.has_accepted, a.has_rejected, a.has_no_results,
+                        CASE
+                            WHEN f.family_key ILIKE '%gubernatorial%' THEN 'gubernatorial'
+                            WHEN f.family_key ILIKE '%mayoral%'       THEN 'mayoral'
+                            WHEN f.family_key ILIKE '%senate%'        THEN 'senate'
+                            WHEN f.family_key ILIKE '%republican%'    THEN 'primary-republican'
+                            WHEN f.family_key ILIKE '%democrat%'      THEN 'primary-democrat'
+                            WHEN f.family_key ILIKE '%openai%'
+                              OR f.family_key ILIKE '%nvidia%'
+                              OR f.family_key ILIKE '%anthropic%'     THEN 'big-tech'
+                            -- family_key NULL or unmatched → infer from question
+                            WHEN a.poly_question ILIKE '%governor%'
+                              OR a.poly_question ILIKE '%gubernatorial%' THEN 'gubernatorial'
+                            WHEN a.poly_question ILIKE '%mayor%'
+                              OR a.poly_question ILIKE '%mayoral%'       THEN 'mayoral'
+                            WHEN a.poly_question ILIKE '%senate%'        THEN 'senate'
+                            WHEN a.poly_question ILIKE '%republican primary%' THEN 'primary-republican'
+                            WHEN a.poly_question ILIKE '%democratic primary%'
+                              OR a.poly_question ILIKE '%democrat primary%'   THEN 'primary-democrat'
+                            WHEN a.poly_question ILIKE '%openai%'
+                              OR a.poly_question ILIKE '%nvidia%'
+                              OR a.poly_question ILIKE '%anthropic%'     THEN 'big-tech'
+                            WHEN a.poly_question ILIKE '%russia%'
+                              OR a.poly_question ILIKE '%ukraine%'
+                              OR a.poly_question ILIKE '%israel%'
+                              OR a.poly_question ILIKE '%ceasefire%'
+                              OR a.poly_question ILIKE '%military%'      THEN 'geopolitics'
+                            ELSE 'other'
+                        END AS category
+                    FROM audit a
+                    LEFT JOIN fam f ON f.market_id = a.poly_market_id
+                )
+                SELECT
+                    category,
+                    COUNT(*)                              AS unique_evaluated,
+                    COUNT(*) FILTER (WHERE has_accepted)   AS unique_accepted,
+                    COUNT(*) FILTER (WHERE has_rejected)   AS unique_rejected,
+                    COUNT(*) FILTER (WHERE has_no_results) AS unique_no_results
+                FROM categorized
+                GROUP BY category
+                ORDER BY unique_evaluated DESC
+            """)
+
+        coverage_by_category = []
+        total_evaluated = 0
+        total_accepted = 0
+        categories_with_coverage = 0
+        for r in rows:
+            evaluated = int(r["unique_evaluated"] or 0)
+            accepted = int(r["unique_accepted"] or 0)
+            total_evaluated += evaluated
+            total_accepted += accepted
+            if accepted > 0:
+                categories_with_coverage += 1
+            coverage_by_category.append({
+                "category":          r["category"],
+                "unique_evaluated":  evaluated,
+                "unique_accepted":   accepted,
+                "unique_rejected":   int(r["unique_rejected"] or 0),
+                "unique_no_results": int(r["unique_no_results"] or 0),
+                "coverage_rate":     (accepted / evaluated) if evaluated else None,
+            })
+
+        return {
+            "coverage_by_category": coverage_by_category,
+            "summary": {
+                "total_unique_evaluated":  total_evaluated,
+                "total_unique_accepted":   total_accepted,
+                "overall_coverage_rate":   (total_accepted / total_evaluated) if total_evaluated else None,
+                "categories_with_coverage": categories_with_coverage,
+            },
+        }
+
+
 def _f(v) -> Optional[float]:
     """None-safe float cast for asyncpg Decimal/None values."""
     return float(v) if v is not None else None

bot/data/manifold.py

@@ -2,34 +2,49 @@
 Manifold Markets client — cross-platform prediction market probability signals.
 
 For each Polymarket question, searches Manifold for a matching binary market
-by keyword overlap and returns its probability as a calibration signal.
+by keyword overlap and returns a ManifoldMatchResult with full audit metadata.
 
-Inversion guard: if the Manifold market's winning side (Republican / Democrat)
-is the complement of the Polymarket question's winning side, the probability is
-automatically inverted (1 - prob).  This prevents "Democrats win Ohio governor"
-from consuming the probability of a Manifold market titled "Republicans win Ohio
-governor" without adjustment.
+Match threshold: >= 0.40 Jaccard overlap (raised from 0.25 for stricter semantics).
 
-Rejection guard: if the match score falls below _MATCH_THRESHOLD the market is
-rejected, even if inversion would otherwise apply.  All decisions are logged at
-INFO so they can be audited per-cycle.
+Outcome compatibility guard (conservative):
+  - Conditional Manifold markets ("If X, will Y?" / "Conditional on..." / "Assuming..."
+    / "Given that..." / mid-sentence "...if X is nominated, will...") are rejected:
+    a premise-gated question is not equivalent to a direct outcome question even when
+    token overlap is high. reason='conditional_market'.
+  - Each side is classified into an outcome_type (nomination | primary_win |
+    general_win | conditional | other). Matches with differing outcome_type — or any
+    conditional side — are rejected. reason='outcome_mismatch: poly=... manifold=...'.
 
-Cache TTL: 30 minutes (Manifold markets move slowly vs our 60 s cycle).
-Match threshold: >= 0.25 keyword overlap ratio between significant tokens.
+Inversion guard (conservative):
+  - If Polymarket question names a party (democrat/republican) AND the matched
+    Manifold market names the OPPOSITE party → invert probability (1 - prob).
+  - If Polymarket question names a party AND Manifold market has NO party keyword
+    → reject with reason='ambiguous_inversion' (can't determine if inversion applies).
+  - All other cases: no inversion, accept if score >= threshold.
+  - Ante duda, reject.
+
+Cache TTL: 30 minutes.
 """
 import logging
 import re
 import time
+from dataclasses import dataclass, field
 from typing import Optional
 
 import httpx
 
+# Version tag for every audit record this matcher produces. Persisted to
+# manifold_match_audit.matcher_version so metrics can isolate current-version
+# stats from legacy/pre-versioning records. Do NOT change this value once set;
+# bump to a new string only when matcher semantics change materially.
+MANIFOLD_MATCHER_VERSION = "v3_outcome_guard"
+
 MANIFOLD_API = "https://api.manifold.markets/v0"
 CACHE_TTL_SEC = 1800  # 30 minutes
 
 log = logging.getLogger(__name__)
 
-_MATCH_THRESHOLD = 0.25
+_MATCH_THRESHOLD = 0.40  # raised from 0.25
 
 _STOP_WORDS = frozenset([
     "will", "the", "a", "an", "is", "are", "was", "were", "be", "been",
@@ -43,11 +58,26 @@ _STOP_WORDS = frozenset([
     "before", "during", "until", "against", "between", "through",
 ])
 
-# Mutually exclusive political parties used for complement detection
 _REPUBLICAN_WORDS = frozenset(["republican", "republicans", "gop"])
 _DEMOCRAT_WORDS   = frozenset(["democrat", "democrats", "democratic"])
 
 
+@dataclass
+class ManifoldMatchResult:
+    status: str                        # 'accepted' | 'rejected' | 'no_results'
+    prob_final: Optional[float] = None
+    prob_raw:   Optional[float] = None
+    market_id:    Optional[str] = None  # Manifold internal market ID
+    market_title: Optional[str] = None
+    market_url:   Optional[str] = None
+    match_score:  Optional[float] = None  # 0-1 Jaccard
+    match_reason: Optional[str] = None   # human-readable explanation
+    inverted: bool = False
+    search_query: str = ""
+    poly_outcome_type: Optional[str] = None  # nomination|primary_win|general_win|conditional|other
+    mfld_outcome_type: Optional[str] = None
+
+
 def _significant_words(text: str) -> set[str]:
     words = re.findall(r"[a-zA-Z]+", text.lower())
     return {w for w in words if w not in _STOP_WORDS and len(w) >= 3}
@@ -69,27 +99,53 @@ def _detect_party(text: str) -> Optional[str]:
     return None
 
 
-def _best_match_with_audit(
-    poly_question: str,
-    results: list[dict],
-) -> tuple[Optional[dict], float, bool]:
-    """
-    Find the best-matching open binary Manifold market.
+# ── Conditional-market detection (Task 1) ──────────────────────────────────────
+# A market is "conditional" when its resolution is gated on a premise rather than
+# asking the outcome directly (e.g. "If X is the nominee, will he win?"). Such a
+# market is NOT equivalent to a direct outcome question even with high token overlap.
+_CONDITIONAL_PREFIXES = ("if ", "conditional on", "assuming ", "given that")
+# " if <clause>," — a mid-sentence conditional clause closed by a comma.
+_CONDITIONAL_CLAUSE_RE = re.compile(r"\sif\s[^,]*,")
 
-    Returns (match, score, needs_inversion):
-      match            — best result dict, or None if below threshold
-      score            — keyword overlap score of best candidate (even if rejected)
-      needs_inversion  — True when Manifold market favours the OPPOSITE party/side
-                         to the Polymarket question (probability should be 1 - prob)
+
+def _is_conditional(text: str) -> bool:
+    """True if the question is phrased conditionally (premise-gated)."""
+    t = (text or "").strip().lower()
+    if t.startswith(_CONDITIONAL_PREFIXES):
+        return True
+    return bool(_CONDITIONAL_CLAUSE_RE.search(t))
+
+
+def _classify_outcome(text: str) -> str:
     """
+    Coarse classification of what a question is *asking about*, used to reject
+    matches whose outcomes are not equivalent even when tokens overlap.
+
+    Returns one of: nomination | primary_win | general_win | conditional | other.
+    Order matters: conditional is checked first (premise-gated), then nomination
+    (which subsumes "primary nominee"), then primary, then general election.
+    """
+    t = (text or "").strip().lower()
+    if t.startswith(_CONDITIONAL_PREFIXES):
+        return "conditional"
+    if any(k in t for k in ("nominee", "nominated", "nomination")):
+        return "nomination"
+    if any(k in t for k in ("primary", "win the primary", "first round")):
+        return "primary_win"
+    if any(k in t for k in ("win the election", "win the race",
+                            "win the seat", "general election")):
+        return "general_win"
+    return "other"
+
+
+def _find_best_candidate(poly_question: str, results: list[dict]) -> tuple[Optional[dict], float]:
+    """Find the highest-scoring open binary Manifold market by Jaccard overlap."""
     poly_words = _significant_words(poly_question)
-    poly_party = _detect_party(poly_question)
     if not poly_words:
-        return None, 0.0, False
+        return None, 0.0
 
     best_score = 0.0
     best: Optional[dict] = None
-    best_needs_inv = False
 
     for result in results:
         if result.get("outcomeType") != "BINARY":
@@ -106,18 +162,14 @@ def _best_match_with_audit(
         if score > best_score:
             best_score = score
             best = result
-            manifold_party = _detect_party(title)
-            # Inversion is warranted only when both sides are unambiguously detected
-            # and they are confirmed opposites (republican ≠ democrat).
-            best_needs_inv = (
-                poly_party is not None
-                and manifold_party is not None
-                and poly_party != manifold_party
-            )
 
-    if best_score >= _MATCH_THRESHOLD and best is not None:
-        return best, best_score, best_needs_inv
-    return None, best_score, False
+    return best, best_score
+
+
+def _market_url(match: dict) -> Optional[str]:
+    slug    = match.get("slug", "")
+    creator = match.get("creatorUsername", "")
+    return f"https://manifold.markets/{creator}/{slug}" if slug else None
 
 
 class ManifoldClient:
@@ -125,27 +177,32 @@ class ManifoldClient:
 
     def __init__(self) -> None:
         self._client = httpx.AsyncClient(timeout=15)
-        # question → (fetched_at_monotonic, probability_or_None)
-        self._cache: dict[str, tuple[float, Optional[float]]] = {}
+        # question → (fetched_at_monotonic, ManifoldMatchResult)
+        self._cache: dict[str, tuple[float, ManifoldMatchResult]] = {}
 
-    async def get_probability(self, question: str) -> Optional[float]:
+    async def get_match(self, question: str) -> ManifoldMatchResult:
         """
-        Return Manifold probability for a matching market, or None.
+        Return a ManifoldMatchResult for the given Polymarket question.
 
-        Probability is already adjusted for party-direction inversion when
-        the matched Manifold market is the complement of our question.
-
-        Full audit log is emitted at INFO for every resolved query.
+        status='accepted'   → prob_final is set and ready to use as signal
+        status='rejected'   → match found but failed quality/inversion check
+        status='no_results' → API returned no results or call failed
         """
         now = time.monotonic()
         cached = self._cache.get(question)
         if cached and (now - cached[0]) < CACHE_TTL_SEC:
             return cached[1]
 
+        poly_outcome = _classify_outcome(question)
+
         query = _build_search_query(question)
         if not query:
-            self._cache[question] = (now, None)
-            return None
+            result = ManifoldMatchResult(
+                status="no_results", search_query="",
+                poly_outcome_type=poly_outcome,
+            )
+            self._cache[question] = (now, result)
+            return result
 
         try:
             resp = await self._client.get(
@@ -154,45 +211,140 @@ class ManifoldClient:
             )
             resp.raise_for_status()
             results = resp.json()
-        except Exception as e:
-            log.warning("Manifold API error for %r: %s", question[:40], e)
-            self._cache[question] = (now, None)
-            return None
+        except Exception as exc:
+            log.warning("Manifold API error for %r: %s", question[:40], exc)
+            result = ManifoldMatchResult(
+                status="no_results", search_query=query,
+                poly_outcome_type=poly_outcome,
+            )
+            self._cache[question] = (now, result)
+            return result
 
-        match, score, needs_inv = _best_match_with_audit(question, results)
+        if not results:
+            result = ManifoldMatchResult(
+                status="no_results", search_query=query,
+                poly_outcome_type=poly_outcome,
+            )
+            self._cache[question] = (now, result)
+            return result
 
-        if match is None:
+        best, score = _find_best_candidate(question, results)
+
+        # ── Score threshold ───────────────────────────────────────────────────
+        if best is None or score < _MATCH_THRESHOLD:
+            reason = f"jaccard={score:.2f}<{_MATCH_THRESHOLD:.2f}"
             log.info(
-                "Manifold no_match: %-50s | best_score=%.2f < %.2f | query=%r",
+                "Manifold REJECTED  %-50s | score=%.2f < threshold=%.2f | query=%r",
                 question[:50], score, _MATCH_THRESHOLD, query,
             )
-            self._cache[question] = (now, None)
-            return None
+            result = ManifoldMatchResult(
+                status="rejected",
+                market_title=best.get("question") if best else None,
+                match_score=score if best else None,
+                match_reason=reason,
+                search_query=query,
+                poly_outcome_type=poly_outcome,
+                mfld_outcome_type=_classify_outcome(best.get("question", "")) if best else None,
+            )
+            self._cache[question] = (now, result)
+            return result
 
-        prob_raw = float(match["probability"])
-        prob_final = (1.0 - prob_raw) if needs_inv else prob_raw
+        # ── Outcome compatibility + inversion analysis (conservative) ─────────
+        mfld_title     = best.get("question", "")
+        mfld_outcome   = _classify_outcome(mfld_title)
+        poly_party     = _detect_party(question)
+        manifold_party = _detect_party(mfld_title)
 
-        # Build market URL from slug (best-effort; may be missing)
-        slug = match.get("slug", "")
-        creator = match.get("creatorUsername", "")
-        url = f"https://manifold.markets/{creator}/{slug}" if slug else "n/a"
+        poly_words    = _significant_words(question)
+        mfld_words    = _significant_words(mfld_title)
+        matched_tokens = sorted(poly_words & mfld_words)[:6]
+
+        inverted = False
+        rejection_reason: Optional[str] = None
+
+        # Task 1 — conditional Manifold market is never equivalent to a direct
+        # outcome question, regardless of token overlap.
+        if _is_conditional(mfld_title):
+            rejection_reason = "conditional_market: manifold question is conditional"
+        # Task 2 — outcome types must match; any conditional side is rejected.
+        elif (poly_outcome == "conditional" or mfld_outcome == "conditional"
+              or poly_outcome != mfld_outcome):
+            rejection_reason = (
+                f"outcome_mismatch: poly={poly_outcome} manifold={mfld_outcome}"
+            )
+        elif poly_party is not None:
+            if manifold_party is None:
+                # Poly specifies a party; Manifold does not → can't verify inversion safety
+                rejection_reason = (
+                    f"ambiguous_inversion: poly_party={poly_party}, mfld_party=none"
+                )
+            elif manifold_party != poly_party:
+                # Clear opposite parties — apply inversion
+                inverted = True
+            # manifold_party == poly_party → same party, no inversion needed
+
+        if rejection_reason is not None:
+            url = _market_url(best)
+            log.info(
+                "Manifold REJECTED  %-50s | score=%.2f | reason=%s\n"
+                "  mfld_title: %s",
+                question[:50], score, rejection_reason, best.get("question", "")[:70],
+            )
+            result = ManifoldMatchResult(
+                status="rejected",
+                market_id=str(best.get("id", "")) or None,
+                market_title=best.get("question"),
+                market_url=url,
+                match_score=score,
+                match_reason=(
+                    f"jaccard={score:.2f}, tokens={matched_tokens}, {rejection_reason}"
+                ),
+                search_query=query,
+                poly_outcome_type=poly_outcome,
+                mfld_outcome_type=mfld_outcome,
+            )
+            self._cache[question] = (now, result)
+            return result
+
+        # ── Accepted ──────────────────────────────────────────────────────────
+        prob_raw   = float(best["probability"])
+        prob_final = (1.0 - prob_raw) if inverted else prob_raw
+        url        = _market_url(best)
+
+        match_reason = f"jaccard={score:.2f}, tokens={matched_tokens}"
+        if inverted:
+            match_reason += f", inverted=party({poly_party}≠{manifold_party})"
 
         log.info(
-            "Manifold %s: %-50s\n"
-            "  poly_question: %s\n"
-            "  manifold_title: %s\n"
-            "  manifold_url:   %s\n"
-            "  match_score: %.2f | prob_raw=%.3f | inverted=%s | prob_final=%.3f",
-            "MATCH_INVERTED" if needs_inv else "MATCH",
+            "Manifold %s %-50s\n"
+            "  poly: %s\n"
+            "  mfld: %s\n"
+            "  url:  %s\n"
+            "  score=%.2f | raw=%.3f | inverted=%s | final=%.3f",
+            "ACCEPTED_INVERTED" if inverted else "ACCEPTED  ",
             question[:50],
             question,
-            match.get("question", ""),
-            url,
-            score, prob_raw, needs_inv, prob_final,
+            best.get("question", ""),
+            url or "n/a",
+            score, prob_raw, inverted, prob_final,
         )
 
-        self._cache[question] = (now, prob_final)
-        return prob_final
+        result = ManifoldMatchResult(
+            status="accepted",
+            prob_final=prob_final,
+            prob_raw=prob_raw,
+            market_id=str(best.get("id", "")) or None,
+            market_title=best.get("question"),
+            market_url=url,
+            match_score=score,
+            match_reason=match_reason,
+            inverted=inverted,
+            search_query=query,
+            poly_outcome_type=poly_outcome,
+            mfld_outcome_type=mfld_outcome,
+        )
+        self._cache[question] = (now, result)
+        return result
 
     async def close(self) -> None:
         await self._client.aclose()

bot/data/news.py

@@ -51,7 +51,11 @@ _DATE_RE = re.compile(
     r"|\bQ[1-4]\b",
     flags=re.IGNORECASE,
 )
-_PUNCT_RE = re.compile(r"[?!\"'.,;:()\[\]{}]")
+# Hyphens/dashes are GNews query operators (a leading '-' means "exclude the
+# next term"), so a token like "El-Sayed" makes the API return HTTP 400.  Strip
+# them to spaces along with the rest of the punctuation so the query stays a
+# plain keyword list.  – = en dash, — = em dash.
+_PUNCT_RE = re.compile(r"[?!\"'.,;:()\[\]{}\-–—]")
 
 
 class NewsClient:
@@ -79,6 +83,18 @@ class NewsClient:
     # Public API
     # ------------------------------------------------------------------
 
+    @property
+    def enabled(self) -> bool:
+        """True only when a GNews API key is configured.
+
+        When False, get_sentiment() is a no-op that returns 0.0 without any
+        network call, so callers must skip GNews entirely — including the
+        per-cycle query budget accounting — instead of "spending" a query that
+        never reaches the API (which inflated gnews_queries_used to a phantom
+        5/5 while the key was missing).
+        """
+        return bool(self._api_key)
+
     async def get_sentiment(self, question: str) -> float:
         """
         Return a sentiment score ∈ [-1.0, +1.0] for the market question.

bot/data/polymarket.py

@@ -211,6 +211,32 @@ class Market:
     category: str = ""
 
 
+@dataclass
+class MarketResolution:
+    """Resolution state of a market, from Gamma API.
+
+    resolution is the final YES outcome price: 1.0 = YES won, 0.0 = NO won.
+    resolved is True only when the outcome is definitive — a market that is
+    closed but still in UMA dispute/proposal reports resolved=False.
+    """
+    resolved: bool
+    resolution: Optional[float] = None
+    resolved_at: Optional[datetime] = None
+
+
+def _parse_resolution_timestamp(raw: Optional[str]) -> Optional[datetime]:
+    """Parse Gamma timestamps: '2026-06-11 13:15:01+00' or '2026-06-11T13:15:01Z'."""
+    if not raw:
+        return None
+    try:
+        dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=timezone.utc)
+        return dt
+    except (ValueError, TypeError):
+        return None
+
+
 @dataclass
 class OrderBook:
     market_id: str
@@ -447,6 +473,74 @@ class PolymarketClient:
         )
         return markets
 
+    async def get_market_resolution(self, market_id: str) -> Optional[MarketResolution]:
+        """Fetch resolution state for a market by Gamma market id.
+
+        Observed Gamma API behaviour (GET /markets/{id}):
+          open market     → closed=false, umaResolutionStatus absent
+          resolved market → closed=true, umaResolutionStatus="resolved",
+                            outcomePrices='["0", "1"]' (final YES price = outcome)
+          unknown id      → HTTP 404
+
+        Returns None on API errors (caller retries next check). A closed market
+        whose outcome prices are not degenerate (0/1) or whose UMA status is not
+        "resolved" yet (proposed/disputed) reports resolved=False — we never
+        settle a position on an ambiguous outcome.
+        """
+        try:
+            resp = await self._client.get(f"{GAMMA_API}/markets/{market_id}")
+            if resp.status_code == 404:
+                log.warning("get_market_resolution: market %s not found (404)", market_id)
+                return None
+            resp.raise_for_status()
+            m = resp.json()
+        except httpx.HTTPError as e:
+            log.warning("get_market_resolution: API error for %s: %s", market_id, e)
+            return None
+
+        if not m.get("closed"):
+            return MarketResolution(resolved=False)
+
+        uma_status = (m.get("umaResolutionStatus") or "").lower()
+        if uma_status and uma_status != "resolved":
+            # Closed but UMA outcome still proposed/disputed — wait for finality
+            return MarketResolution(resolved=False)
+
+        raw_prices = m.get("outcomePrices", [])
+        if isinstance(raw_prices, str):
+            import json as _json
+            try:
+                raw_prices = _json.loads(raw_prices)
+            except ValueError:
+                raw_prices = []
+        try:
+            yes_final = float(raw_prices[0])
+        except (IndexError, TypeError, ValueError):
+            log.warning(
+                "get_market_resolution: market %s closed but outcomePrices "
+                "unparseable: %r", market_id, m.get("outcomePrices"),
+            )
+            return MarketResolution(resolved=False)
+
+        if yes_final >= 0.99:
+            resolution = 1.0
+        elif yes_final <= 0.01:
+            resolution = 0.0
+        else:
+            # Closed but prices not settled at 0/1 (partial / ambiguous outcome)
+            log.warning(
+                "get_market_resolution: market %s closed with non-binary final "
+                "price %.3f — not settling", market_id, yes_final,
+            )
+            return MarketResolution(resolved=False)
+
+        resolved_at = (
+            _parse_resolution_timestamp(m.get("closedTime"))
+            or _parse_resolution_timestamp(m.get("umaEndDate"))
+            or _parse_resolution_timestamp(m.get("endDate"))
+        )
+        return MarketResolution(resolved=True, resolution=resolution, resolved_at=resolved_at)
+
     async def get_order_book(self, token_id: str) -> Optional[OrderBook]:
         """Get order book for a specific token."""
         try:

bot/data/schema.sql

@@ -113,9 +113,10 @@ CREATE INDEX IF NOT EXISTS idx_trades_closed ON trades(closed_at) WHERE closed_a
 -- Fix 3: market resolution and realized P&L per trade
 --
 -- resolution: 1.0 if YES resolved, 0.0 if NO resolved, NULL if not yet settled.
--- close_pnl:  realized P&L in USDC at close time.
---   BUY_YES: (resolution - entry_price) * shares
---   BUY_NO:  ((1 - resolution) - entry_price) * shares
+-- close_pnl:  realized P&L in USDC at close time — NET of fee (payout − net_cost),
+--   the same definition PaperExecutor.close_position() reports in logs/Telegram.
+--   BUY_YES: resolution * shares - net_cost
+--   BUY_NO:  (1 - resolution) * shares - net_cost
 --   NULL if closed without a known resolution (legacy closes, inversion fixes).
 -- ─────────────────────────────────────────────────────────────────────────────
 ALTER TABLE trades ADD COLUMN IF NOT EXISTS close_pnl  DOUBLE PRECISION;
@@ -168,6 +169,103 @@ ALTER TABLE trades ADD COLUMN IF NOT EXISTS feat_btc_dom_lo DOUBLE PRECISION;
 CREATE INDEX IF NOT EXISTS idx_trades_feat_fg   ON trades(feat_fg_lo)  WHERE feat_fg_lo IS NOT NULL;
 CREATE INDEX IF NOT EXISTS idx_trades_feat_mfld ON trades(feat_mfld_lo) WHERE feat_mfld_lo IS NOT NULL;
 
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Manifold match audit — per-trade columns in trades
+--
+-- Persisted for every trade where Manifold was queried (status='accepted').
+-- mfld_match_status: 'accepted' | 'rejected' | 'no_results'
+-- mfld_inverted: TRUE when prob_final = 1 - prob_raw (party complement match)
+-- ─────────────────────────────────────────────────────────────────────────────
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_market_id    TEXT;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_market_title TEXT;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_market_url   TEXT;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_prob_raw     DOUBLE PRECISION;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_prob_final   DOUBLE PRECISION;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_inverted     BOOLEAN;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_match_score  DOUBLE PRECISION;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_match_reason TEXT;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS mfld_match_status TEXT;
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Manifold match audit table — records every Manifold query attempt
+--
+-- Populated for ALL queries: accepted, rejected, and no_results.
+-- used_in_trade=TRUE is set after executor confirms a trade was executed.
+-- poly_market_id: Market.id from the Polymarket Market dataclass (never NULL).
+-- ─────────────────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS manifold_match_audit (
+    id                TEXT PRIMARY KEY,
+    timestamp         TIMESTAMPTZ DEFAULT NOW(),
+    poly_market_id    TEXT NOT NULL,
+    poly_question     TEXT NOT NULL,
+    search_query      TEXT,
+    mfld_market_id    TEXT,
+    mfld_market_title TEXT,
+    mfld_market_url   TEXT,
+    prob_raw          DOUBLE PRECISION,
+    prob_final        DOUBLE PRECISION,
+    inverted          BOOLEAN DEFAULT FALSE,
+    match_score       DOUBLE PRECISION,
+    match_reason      TEXT,
+    match_status      TEXT NOT NULL,
+    used_in_trade     BOOLEAN DEFAULT FALSE,
+    poly_outcome_type TEXT,
+    mfld_outcome_type TEXT
+);
+
+CREATE INDEX IF NOT EXISTS idx_mfld_audit_timestamp ON manifold_match_audit(timestamp DESC);
+CREATE INDEX IF NOT EXISTS idx_mfld_audit_status    ON manifold_match_audit(match_status);
+CREATE INDEX IF NOT EXISTS idx_mfld_audit_poly_mkt  ON manifold_match_audit(poly_market_id);
+
+-- Backfill outcome-type columns on pre-existing tables (idempotent).
+ALTER TABLE manifold_match_audit ADD COLUMN IF NOT EXISTS poly_outcome_type TEXT;
+ALTER TABLE manifold_match_audit ADD COLUMN IF NOT EXISTS mfld_outcome_type TEXT;
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Matcher versioning — separate current-matcher metrics from legacy records
+--
+-- matcher_version tags each audit row with the matcher that produced it
+-- (MANIFOLD_MATCHER_VERSION in bot/data/manifold.py).  This lets the metrics
+-- endpoint isolate current_version stats from pre-versioning records, whose
+-- accepted matches would now be rejected by the outcome-compatibility guard.
+--
+-- Backfill is one-shot and idempotent (only touches NULL matcher_version rows):
+--   * rows with no outcome types  → 'legacy_pre_outcome_guard' (pre outcome-guard;
+--     accepted without any outcome-type validation)
+--   * rows with an outcome type   → 'v2_outcome_guard_no_version' (existed between
+--     the outcome-guard and this versioning; real version not persisted)
+-- We tag rather than infer the exact version that wasn't recorded.
+-- ─────────────────────────────────────────────────────────────────────────────
+ALTER TABLE manifold_match_audit ADD COLUMN IF NOT EXISTS matcher_version TEXT;
+
+UPDATE manifold_match_audit
+SET matcher_version = 'legacy_pre_outcome_guard'
+WHERE matcher_version IS NULL
+  AND poly_outcome_type IS NULL
+  AND mfld_outcome_type IS NULL;
+
+UPDATE manifold_match_audit
+SET matcher_version = 'v2_outcome_guard_no_version'
+WHERE matcher_version IS NULL
+  AND (poly_outcome_type IS NOT NULL OR mfld_outcome_type IS NOT NULL);
+
+CREATE INDEX IF NOT EXISTS idx_mfld_audit_version ON manifold_match_audit(matcher_version);
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Metric exclusion — administrative closure flag
+--
+-- excluded_from_metrics: TRUE for trades closed for non-signal reasons
+--   (bad matcher, data error, admin close).  These trades are excluded from
+--   win_rate, calibration_score, realized_pnl, and feature attribution.
+-- exclusion_reason: free-text label for the exclusion cause.
+--   e.g. 'invalid_manifold_match_legacy'
+-- ─────────────────────────────────────────────────────────────────────────────
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS excluded_from_metrics BOOLEAN DEFAULT FALSE;
+ALTER TABLE trades ADD COLUMN IF NOT EXISTS exclusion_reason TEXT;
+
+CREATE INDEX IF NOT EXISTS idx_trades_excluded ON trades(excluded_from_metrics)
+  WHERE excluded_from_metrics = TRUE;
+
 -- ─────────────────────────────────────────────────────────────────────────────
 -- Fix 3: extended metrics_daily columns for DB-computed metrics
 --
@@ -182,3 +280,175 @@ ALTER TABLE metrics_daily ADD COLUMN IF NOT EXISTS realized_pnl       DOUBLE PRE
 ALTER TABLE metrics_daily ADD COLUMN IF NOT EXISTS open_count         INTEGER;
 ALTER TABLE metrics_daily ADD COLUMN IF NOT EXISTS closed_count       INTEGER;
 ALTER TABLE metrics_daily ADD COLUMN IF NOT EXISTS resolved_count     INTEGER;
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Checkpoint alerts — one-shot and rate-limited Telegram observation alerts
+--
+-- fired_at:      timestamp of the first fire (immutable for one-shot checkpoints)
+-- last_fired_at: updated on every fire (used for rate-limiting repeatable alerts)
+-- ─────────────────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS checkpoint_alerts (
+    checkpoint_name TEXT PRIMARY KEY,
+    fired_at        TIMESTAMPTZ NOT NULL,
+    last_fired_at   TIMESTAMPTZ
+);
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- Manifold evaluation cooldown — per-market backoff for the Manifold matcher
+--
+-- The trading loop re-evaluates the same ~stable set of politics/tech markets
+-- every cycle (~60s).  Most resolve to a stable terminal verdict (no Manifold
+-- coverage, low-score, outcome mismatch, conditional market) that will not change
+-- on the next cycle.  Re-querying them every minute floods manifold_match_audit
+-- with redundant rows and makes the metrics uninterpretable.
+--
+-- This table records, per poly_market_id, when the market was last evaluated and
+-- the earliest time it should be evaluated again (retry_after).  evaluate() in
+-- bot/strategy/bayesian.py consults it BEFORE calling the matcher and skips the
+-- call (and the audit write) entirely while now() < retry_after.
+--
+-- last_status / cooldown_reason are stored for observability only.
+-- ─────────────────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS manifold_eval_cooldown (
+    poly_market_id    TEXT PRIMARY KEY,
+    last_evaluated_at TIMESTAMPTZ NOT NULL,
+    last_status       TEXT NOT NULL,
+    retry_after       TIMESTAMPTZ NOT NULL,
+    cooldown_reason   TEXT
+);
+
+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/executor/paper.py

@@ -22,6 +22,30 @@ log = logging.getLogger(__name__)
 # NOTE: this is a heuristic — see COMMISSION_RATE in bayesian.py for context.
 POLYMARKET_FEE = 0.02  # 2%
 
+# Strong references to in-flight notification tasks.  The event loop only
+# keeps a weak reference to tasks created via create_task(), so without this
+# set a pending Telegram notification could be garbage-collected before it
+# runs.  Tasks remove themselves from the set on completion.
+_background_tasks: set[asyncio.Task] = set()
+
+
+def _notify_in_background(coro) -> None:
+    """Fire-and-forget a Telegram notification, keeping the task referenced."""
+    task = asyncio.create_task(coro)
+    _background_tasks.add(task)
+    task.add_done_callback(_background_tasks.discard)
+
+
+def cash_available(bankroll: float, total_net_cost_open: float) -> float:
+    """Cash left after the net cost (fees included) of all open positions.
+
+    Single source of truth for the cash figure, shared by
+    PaperExecutor.initialize() and the /api/summary endpoint so both always
+    report the same number for the same DB state.
+    total_net_cost_open comes from Database.get_open_position_data().
+    """
+    return max(0.0, bankroll - total_net_cost_open)
+
 
 @dataclass
 class Trade:
@@ -57,6 +81,16 @@ class Trade:
     feat_news_lo:    float = 0.0
     feat_mfld_lo:    float = 0.0
     feat_btc_dom_lo: float = 0.0
+    # ── Manifold match audit ──────────────────────────────────────────────────
+    mfld_market_id:   Optional[str]   = None
+    mfld_market_title: Optional[str]  = None
+    mfld_market_url:  Optional[str]   = None
+    mfld_prob_raw:    Optional[float] = None
+    mfld_prob_final:  Optional[float] = None
+    mfld_inverted:    bool            = False
+    mfld_match_score: Optional[float] = None
+    mfld_match_reason: Optional[str]  = None
+    mfld_match_status: Optional[str]  = None
 
     def __str__(self) -> str:
         return (
@@ -98,7 +132,7 @@ class PaperExecutor:
 
         positions_value = sum(positions_size.values())
         self._portfolio.positions = positions_size
-        self._portfolio.cash = max(0.0, self._portfolio.cash - total_net_cost)
+        self._portfolio.cash = cash_available(self._portfolio.cash, total_net_cost)
 
         total_value = self._portfolio.cash + positions_value
         exposure_pct = positions_value / total_value if total_value > 0 else 0.0
@@ -176,6 +210,16 @@ class PaperExecutor:
             feat_news_lo=order.feat_news_lo,
             feat_mfld_lo=order.feat_mfld_lo,
             feat_btc_dom_lo=order.feat_btc_dom_lo,
+            # Manifold audit
+            mfld_market_id=order.mfld_market_id,
+            mfld_market_title=order.mfld_market_title,
+            mfld_market_url=order.mfld_market_url,
+            mfld_prob_raw=order.mfld_prob_raw,
+            mfld_prob_final=order.mfld_prob_final,
+            mfld_inverted=order.mfld_inverted,
+            mfld_match_score=order.mfld_match_score,
+            mfld_match_reason=order.mfld_match_reason,
+            mfld_match_status=order.mfld_match_status,
         )
 
         # Update paper portfolio
@@ -185,7 +229,7 @@ class PaperExecutor:
         # Persist to DB
         await self._db.save_trade(trade)
 
-        asyncio.create_task(
+        _notify_in_background(
             telegram.trade_opened(trade.question, trade.direction, trade.size_usdc, trade.edge_net)
         )
 
@@ -206,7 +250,7 @@ class PaperExecutor:
             "LEGACY_CLOSE market=%s | returned $%.2f to cash | %s",
             market_id, cost, reason[:80],
         )
-        asyncio.create_task(
+        _notify_in_background(
             telegram.trade_legacy_closed(question or market_id, cost, reason)
         )
         return cost
@@ -215,24 +259,53 @@ class PaperExecutor:
         """Close a paper position after market resolution.
 
         resolution: 1.0 if YES won, 0.0 if NO won.
-        Persists resolution and close_pnl to DB (computed via SQL from stored
-        entry_price and shares). Returns approximate P&L for logging.
+        Settlement payout per trade:
+          BUY_YES: shares * resolution
+          BUY_NO:  shares * (1 - resolution)
+        pnl = payout - net_cost.
+        Persists resolution and close_pnl to DB. Returns realized P&L for
+        logging, or None if no position is open.
         """
         if market_id not in self._portfolio.positions:
             return None
 
-        position_cost = self._portfolio.positions.pop(market_id)
-        self._portfolio.cash += position_cost * resolution  # pay out winnings
+        position_cost = self._portfolio.positions[market_id]
+        open_trades = await self._db.get_open_trades_for_market(market_id)
 
+        if open_trades:
+            payout = sum(
+                float(t["shares"])
+                * (resolution if t["direction"] == "BUY_YES" else 1.0 - resolution)
+                for t in open_trades
+            )
+            net_cost = sum(float(t["net_cost"]) for t in open_trades)
+            pnl = payout - net_cost
+        else:
+            # In-memory position with no open DB trades: direction/shares are
+            # unknown, so settle at break-even instead of guessing the payout.
+            log.warning(
+                "close_position: no open DB trades for market %s — "
+                "settling at break-even", market_id,
+            )
+            payout = position_cost
+            pnl = 0.0
+
+        # Persist first, mutate memory after: if the DB write fails, the
+        # in-memory portfolio must keep the position so the next resolution
+        # check can retry the close.
         await self._db.close_paper_position(
             market_id,
-            reason=f"market_resolved resolution={resolution:.1f}",
+            reason="resolved",
             resolution=resolution,
         )
-        approx_pnl = position_cost * resolution - position_cost
-        log.info("Closed position in %s, resolution=%.1f", market_id, resolution)
-        asyncio.create_task(
-            telegram.trade_closed(question or market_id, approx_pnl)
+
+        self._portfolio.positions.pop(market_id)
+        self._portfolio.cash += payout
+        log.info(
+            "Closed position in %s, resolution=%.1f payout=$%.2f pnl=%+.2f",
+            market_id, resolution, payout, pnl,
         )
-        # Approximate PnL: settlement value minus cost. Exact value is in close_pnl.
-        return approx_pnl
+        _notify_in_background(
+            telegram.trade_closed(question or market_id, pnl)
+        )
+        return pnl

bot/main.py

@@ -11,21 +11,89 @@ from bot.data.polymarket import PolymarketClient, Market, market_family_key
 from bot.data.external import ExternalDataClient
 from bot.data.news import NewsClient
 from bot.data.manifold import ManifoldClient
-from bot.strategy.bayesian import BayesianStrategy, gnews_priority, MAX_NEWS_QUERIES_PER_CYCLE
+from bot.strategy.bayesian import (
+    BayesianStrategy,
+    gnews_priority,
+    MAX_NEWS_QUERIES_PER_CYCLE,
+    MANIFOLD_SIGNAL_ENABLED,
+)
 from bot.risk.manager import RiskManager
 from bot.executor.paper import PaperExecutor
 from bot.metrics.tracker import MetricsTracker
 from bot.data.db import Database
+from bot.notify.checkpoints import CheckpointMonitor
 
 logging.basicConfig(
     level=logging.INFO,
     format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
 )
+# httpx logs every request URL at INFO, and the GNews URL carries the API key as
+# a `?token=` query param — that would leak GNEWS_API_KEY in plaintext into the
+# pod logs.  Raise httpx/httpcore to WARNING so request URLs never reach INFO.
+# The bot's own GNews log lines only print the sanitised query, not the token.
+logging.getLogger("httpx").setLevel(logging.WARNING)
+logging.getLogger("httpcore").setLevel(logging.WARNING)
 log = logging.getLogger("bot.main")
 
 PAPER_MODE = os.getenv("PAPER_MODE", "true").lower() == "true"
 PAPER_BANKROLL = float(os.getenv("PAPER_BANKROLL", "10000"))
 
+# Check open positions for market resolution every N trading cycles (~N minutes
+# at the 60s cycle cadence). Keeps Gamma API load at ~1 request per open
+# 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,
+    executor: PaperExecutor,
+    db: Database,
+) -> None:
+    """Detect resolved markets and settle their open paper positions.
+
+    For each open position, asks the Gamma API whether the market resolved.
+    On a definitive resolution, PaperExecutor.close_position() settles the
+    payout, persists close_reason='resolved' + resolution + close_pnl, and
+    sends the Telegram notification.
+    """
+    positions = await db.get_open_position_details()
+    checked = 0
+    resolved = 0
+    for pos in positions:
+        market_id = str(pos["market_id"])
+        try:
+            res = await poly.get_market_resolution(market_id)
+        except Exception as exc:
+            log.warning("Resolution check failed for market %s: %s", market_id, exc)
+            continue
+        checked += 1
+        if res is None or not res.resolved or res.resolution is None:
+            continue
+        try:
+            pnl = await executor.close_position(
+                market_id, res.resolution, question=pos.get("question") or "",
+            )
+        except Exception as exc:
+            log.error("Failed to close resolved market %s: %s", market_id, exc)
+            continue
+        resolved += 1
+        log.info(
+            "MARKET_RESOLVED market_id=%s resolution=%.1f pnl=%s | %s",
+            market_id,
+            res.resolution,
+            f"{pnl:+.2f}" if pnl is not None else "n/a",
+            (pos.get("question") or "")[:60],
+        )
+    log.info("Resolution check: %d positions checked, %d resolved", checked, resolved)
+
 
 async def run_trading_loop(
     poly: PolymarketClient,
@@ -38,9 +106,23 @@ async def run_trading_loop(
 ) -> None:
     """Main trading loop — runs every 60 seconds."""
     log.info("Trading loop started. PAPER_MODE=%s", PAPER_MODE)
+    checkpoint_monitor = CheckpointMonitor()
+    cycle_count = 0
 
     while True:
         try:
+            cycle_count += 1
+
+            # 0. Resolution detector — every RESOLUTION_CHECK_INTERVAL cycles,
+            #    settle paper positions whose market resolved on Polymarket.
+            #    Runs before evaluation so freed cash/families are usable this cycle.
+            if (
+                PAPER_MODE
+                and isinstance(executor, PaperExecutor)
+                and cycle_count % RESOLUTION_CHECK_INTERVAL == 0
+            ):
+                await check_resolutions(poly, executor, db)
+
             # 1. Fetch active markets (90-day window)
             markets = await poly.get_active_markets()
             log.info("Found %d active markets", len(markets))
@@ -48,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.
@@ -102,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(
@@ -109,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
@@ -136,11 +230,39 @@ async def run_trading_loop(
                 # 7. Execute (paper)
                 trade = await executor.execute(order)
                 if trade:
-                    await metrics.record_trade(trade)
                     log.info("Trade executed: %s", trade)
                     # 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:
+                            await db.mark_manifold_audit_used(signal.mfld_audit_id)
+                        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()
@@ -152,7 +274,17 @@ async def run_trading_loop(
                 if denom == 0:
                     return "0% (0/0)"
                 return f"{n * 100 // denom}% ({n}/{denom})"
-            gnews_cap = strategy._news_queries_this_cycle  # already updated by reset below
+
+            # The accepted/rejected counters only increment on the active-signal
+            # path, so with the signal disabled they always print 0/0 — say
+            # "disabled" instead of pretending the matcher found nothing.
+            if MANIFOLD_SIGNAL_ENABLED:
+                manifold_summary = (
+                    f"  manifold_matches_accepted:      {stats['manifold_matches_accepted']}\n"
+                    f"  manifold_matches_rejected:      {stats['manifold_matches_rejected']}"
+                )
+            else:
+                manifold_summary = "  manifold_signal:                disabled"
 
             log.info(
                 "[CYCLE SUMMARY]\n"
@@ -170,9 +302,7 @@ async def run_trading_loop(
                 "  gnews_queries_used:             %d/%d\n"
                 "  reentry_guard_blocked:          %d\n"
                 "  legacy_incomplete_seen:         %d\n"
-                "  family_conflicts_prevented:     %d\n"
-                "  manifold_matches_accepted:      %d\n"
-                "  manifold_matches_rejected:      %d",
+                "%s",
                 n_total,
                 n_uncertainty,
                 stats["max_edge_gross"],
@@ -187,14 +317,37 @@ async def run_trading_loop(
                 stats["gnews_queries_used"], MAX_NEWS_QUERIES_PER_CYCLE,
                 reentry_guard_count,
                 legacy_incomplete_count,
-                stats["skip_family"],
-                stats["manifold_matches_accepted"],
-                stats["manifold_matches_rejected"],
+                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()
 
+            # 10. Checkpoint alerts — one-shot / rate-limited Telegram notifications
+            current_portfolio = executor.get_portfolio()
+            try:
+                await checkpoint_monitor.check_all(
+                    db,
+                    exposure_pct=current_portfolio.exposure_pct,
+                    exposure_cap_pct=risk.max_exposure_pct,
+                )
+            except Exception as exc:
+                log.warning("checkpoint_monitor.check_all failed: %s", exc)
+
         except Exception as e:
             log.error("Error in trading loop: %s", e, exc_info=True)
 
@@ -204,14 +357,17 @@ async def run_trading_loop(
 async def run_legacy_scan(
     db: Database,
     markets: list,
-    manifold: ManifoldClient,
     executor: PaperExecutor,
     paper_mode: bool,
 ) -> None:
     """
     One-time startup scan: re-key all open DB positions with the current
-    market_family_key() logic, detect contradictions, re-validate Manifold
-    signals, and report KEEP / REVIEW / CLOSE_RECOMMENDED per position.
+    market_family_key() logic, detect family conflicts, and report
+    KEEP / REVIEW / CLOSE_RECOMMENDED per position.
+
+    Manifold is intentionally not consulted here: with
+    MANIFOLD_SIGNAL_ENABLED=false it is observational-only and must not
+    drive position closures.
 
     In paper_mode: auto-closes all CLOSE_RECOMMENDED positions after logging.
     """
@@ -250,8 +406,6 @@ async def run_legacy_scan(
             "family_key_old": old_fk,
             "family_key_new": new_fk,
             "fk_changed": new_fk != old_fk,
-            "manifold_prob_new": None,
-            "manifold_inverted": False,
             "recommendation": "legacy_incomplete" if is_legacy_incomplete else "OK",
             "rec_reason": "edge_net and live market unavailable" if is_legacy_incomplete else "no family conflict",
         })
@@ -289,31 +443,7 @@ async def run_legacy_scan(
                 p["market_id"], p["family_key_old"] or "none", p["family_key_new"],
             )
 
-    # Step 3: Manifold re-query for positions whose family key changed
-    for p in enriched:
-        if p["live_market"] and p["fk_changed"]:
-            prob = await manifold.get_probability(p["question"])
-            p["manifold_prob_new"] = prob
-            if prob is not None:
-                # Detect if original trade direction conflicts with corrected Manifold signal
-                if prob < 0.40 and p["direction"] == "BUY_YES":
-                    p["manifold_inverted"] = True
-                    note = f"Manifold:{prob:.3f} contradicts BUY_YES (inversion bug confirmed)"
-                    if p["recommendation"] in ("OK", "REVIEW"):
-                        p["recommendation"] = "CLOSE_RECOMMENDED"
-                        p["rec_reason"] = note
-                    else:
-                        p["rec_reason"] += f" | {note}"
-                elif prob > 0.60 and p["direction"] == "BUY_NO":
-                    p["manifold_inverted"] = True
-                    note = f"Manifold:{prob:.3f} contradicts BUY_NO (inversion bug confirmed)"
-                    if p["recommendation"] in ("OK", "REVIEW"):
-                        p["recommendation"] = "CLOSE_RECOMMENDED"
-                        p["rec_reason"] = note
-                    else:
-                        p["rec_reason"] += f" | {note}"
-
-    # Step 4: log the full scan report (before any closures)
+    # Step 3: log the full scan report (before any closures)
     n_close    = sum(1 for p in enriched if p["recommendation"] == "CLOSE_RECOMMENDED")
     n_keep     = sum(1 for p in enriched if p["recommendation"] == "KEEP")
     n_ok       = sum(1 for p in enriched if p["recommendation"] == "OK")
@@ -329,7 +459,6 @@ async def run_legacy_scan(
             "  [%-18s] market=%-8s | dir=%-8s | edge_net=%+.3f\n"
             "                        stored_family: %s\n"
             "                        new_family:    %s%s\n"
-            "                        manifold_new:  %s\n"
             "                        reason: %s",
             p["recommendation"],
             p["market_id"], p["direction"],
@@ -337,12 +466,11 @@ async def run_legacy_scan(
             p["family_key_old"] or "none",
             p["family_key_new"],
             " [CHANGED]" if p["fk_changed"] else "",
-            f"{p['manifold_prob_new']:.3f}" if p["manifold_prob_new"] is not None else "n/a",
             p["rec_reason"],
         )
     log.warning("━" * 70)
 
-    # Step 5: auto-close in paper mode
+    # Step 4: auto-close in paper mode
     if paper_mode and n_close > 0 and isinstance(executor, PaperExecutor):
         log.warning("PAPER MODE: auto-closing %d CLOSE_RECOMMENDED position(s)...", n_close)
         for p in enriched:
@@ -375,7 +503,7 @@ async def main() -> None:
     external = ExternalDataClient()
     news = NewsClient()
     manifold = ManifoldClient()
-    strategy = BayesianStrategy(news=news, manifold=manifold)
+    strategy = BayesianStrategy(news=news, manifold=manifold, db=db)
     risk = RiskManager(max_position_pct=0.05, max_exposure_pct=0.30)
     executor = PaperExecutor(db=db, bankroll=PAPER_BANKROLL) if PAPER_MODE else None
     metrics = MetricsTracker(db=db)
@@ -395,7 +523,7 @@ async def main() -> None:
     except Exception as e:
         log.warning("Could not fetch markets for legacy scan: %s — scan skipped", e)
         scan_markets = []
-    await run_legacy_scan(db, scan_markets, manifold, executor, PAPER_MODE)
+    await run_legacy_scan(db, scan_markets, executor, PAPER_MODE)
 
     try:
         await run_trading_loop(poly, external, strategy, risk, executor, metrics, db)

bot/metrics/sharpe.py

@@ -0,0 +1,79 @@
+"""
+Sharpe ratio from the paper portfolio's daily PnL curve, with a minimum-sample gate.
+
+The input series is the closing total_pnl of each observed UTC day
+(Database.get_daily_pnl_closes). Daily returns are PnL deltas normalized by
+the paper bankroll:
+
+    r_t = (pnl_t − pnl_{t−1}) / bankroll
+
+Sharpe = mean(r) / sample_std(r) × √365, annualized — prediction markets
+resolve every calendar day, so 365 is used instead of 252 trading days.
+Risk-free rate is taken as 0.
+
+Gate: with a tiny sample (e.g. 1 resolved trade over a flat curve plus one
++299 jump) any Sharpe value is statistically meaningless — artificially huge
+or tiny depending on where the jump lands. So no numeric Sharpe is exposed
+until BOTH minimums are met:
+
+    days observed     >= MIN_DAYS_OBSERVED   (30)
+    resolved trades   >= MIN_RESOLVED_TRADES (10)
+
+Below either minimum the value is None with status "insufficient_sample".
+A perfectly flat curve (zero variance) also yields None ("zero_variance"):
+Sharpe is undefined there, not infinite.
+"""
+from statistics import mean, stdev
+from typing import Optional
+
+MIN_DAYS_OBSERVED = 30
+MIN_RESOLVED_TRADES = 10
+ANNUALIZATION_DAYS = 365
+
+SHARPE_OK = "ok"
+SHARPE_INSUFFICIENT = "insufficient_sample"
+SHARPE_ZERO_VARIANCE = "zero_variance"
+
+
+def daily_returns(daily_pnl_closes: list[float], bankroll: float) -> list[float]:
+    """Bankroll-normalized day-over-day returns from a daily PnL-close series."""
+    return [
+        (curr - prev) / bankroll
+        for prev, curr in zip(daily_pnl_closes, daily_pnl_closes[1:])
+    ]
+
+
+def compute_sharpe(daily_pnl_closes: list[float], bankroll: float) -> Optional[float]:
+    """Annualized Sharpe of the daily PnL curve, or None if undefined.
+
+    None when there are fewer than 2 returns (need 3+ daily closes) or the
+    return series has zero variance. No sample-size gate here — see
+    sharpe_with_gate() for the exposed value.
+    """
+    returns = daily_returns(daily_pnl_closes, bankroll)
+    if len(returns) < 2:
+        return None
+    sd = stdev(returns)
+    if sd == 0:
+        return None
+    return mean(returns) / sd * ANNUALIZATION_DAYS ** 0.5
+
+
+def sharpe_with_gate(
+    daily_pnl_closes: list[float],
+    bankroll: float,
+    resolved_count: int,
+) -> tuple[Optional[float], str]:
+    """Return (sharpe, status) applying the minimum-sample gate.
+
+    status: "ok"                  — sharpe is a meaningful float
+            "insufficient_sample" — sample below minimums, sharpe is None
+            "zero_variance"       — sample OK but flat curve, sharpe is None
+    """
+    days_observed = len(daily_pnl_closes)
+    if days_observed < MIN_DAYS_OBSERVED or resolved_count < MIN_RESOLVED_TRADES:
+        return None, SHARPE_INSUFFICIENT
+    sharpe = compute_sharpe(daily_pnl_closes, bankroll)
+    if sharpe is None:
+        return None, SHARPE_ZERO_VARIANCE
+    return sharpe, SHARPE_OK

bot/metrics/tracker.py

@@ -15,13 +15,16 @@ win_rate            Fraction of resolved closed trades with close_pnl > 0.
                     NULL if fewer than 5 resolved trades.
 calibration_score   1 − AVG((final_prob − resolution)²) on resolved trades.
                     Brier score (higher = better calibration). NULL if < 10 resolved.
-sharpe_ratio        0.0 — requires a daily-return time series, not yet tracked.
+sharpe_ratio        Annualized Sharpe of the daily total_pnl curve (see
+                    bot/metrics/sharpe.py). NULL until the sample gate passes:
+                    >= 30 days observed AND >= 10 resolved trades.
 """
 import logging
+import os
 from datetime import datetime, UTC
 
 from bot.data.db import Database
-from bot.executor.paper import Trade
+from bot.metrics.sharpe import sharpe_with_gate
 
 log = logging.getLogger(__name__)
 
@@ -30,11 +33,6 @@ class MetricsTracker:
     def __init__(self, db: Database) -> None:
         self._db = db
 
-    async def record_trade(self, trade: Trade) -> None:
-        """Persist a trade to the DB. No in-memory accumulation."""
-        await self._db.save_trade(trade)
-        log.info("Trade recorded: %s", trade)
-
     async def update_daily_summary(self) -> None:
         """Compute metrics from DB and write a metrics_daily snapshot.
 
@@ -67,6 +65,12 @@ class MetricsTracker:
 
         avg_edge = total_pnl / total_deployed if total_deployed > 0 else 0.0
 
+        # Sharpe: real value from the daily PnL curve, NULL while the sample
+        # gate (>=30 days observed, >=10 resolved) is not met.
+        bankroll = float(os.getenv("PAPER_BANKROLL", "10000"))
+        daily_closes = await self._db.get_daily_pnl_closes()
+        sharpe, sharpe_status = sharpe_with_gate(daily_closes, bankroll, resolved)
+
         metrics = {
             "timestamp":          datetime.now(UTC),
             "total_trades":       int(raw["total_trades"]),
@@ -80,7 +84,7 @@ class MetricsTracker:
             "total_pnl":          total_pnl,
             "win_rate":           win_rate,
             "avg_edge":           avg_edge,
-            "sharpe_ratio":       0.0,   # requires daily-return series (not yet tracked)
+            "sharpe_ratio":       sharpe,   # NULL until sample gate passes
             "calibration_score":  calibration,
             "paper_mode":         True,
         }
@@ -89,9 +93,10 @@ class MetricsTracker:
         log.info(
             "Daily metrics | trades=%d (open=%d closed=%d resolved=%d) | "
             "unrealized=$%.2f realized=$%.2f total=$%.2f | "
-            "win_rate=%s calibration=%s",
+            "win_rate=%s calibration=%s sharpe=%s",
             metrics["total_trades"], open_count, closed_count, resolved,
             unrealized, realized, total_pnl,
             f"{win_rate:.1%}" if win_rate is not None else "n/a (<5)",
             f"{calibration:.3f}" if calibration is not None else "n/a (<10)",
+            f"{sharpe:.2f}" if sharpe is not None else f"n/a ({sharpe_status})",
         )

bot/notify/checkpoints.py

@@ -0,0 +1,184 @@
+"""One-shot and rate-limited Telegram checkpoint alerts.
+
+Called from the main trading loop at the end of each cycle.
+Errors are swallowed — checkpoint failures must never break the loop.
+"""
+import logging
+from datetime import datetime, timezone
+from typing import Optional
+
+from bot.notify import telegram
+
+log = logging.getLogger(__name__)
+
+_EXPOSURE_COOLDOWN_HOURS = 6
+
+
+class CheckpointMonitor:
+
+    async def check_all(
+        self,
+        db,
+        exposure_pct: float,
+        exposure_cap_pct: float,
+    ) -> None:
+        for name, coro in [
+            ("primer_match_accepted", self._check_primer_match_accepted(db)),
+            ("primer_trade_phase6",   self._check_primer_trade_phase6(db)),
+            ("primer_resolved",       self._check_primer_resolved(db)),
+            ("exposure_cerca_cap",    self._check_exposure_cerca_cap(db, exposure_pct, exposure_cap_pct)),
+        ]:
+            try:
+                await coro
+            except Exception as exc:
+                log.warning("checkpoint %s failed: %s", name, exc)
+
+    # ── helpers ──────────────────────────────────────────────────────────────
+
+    async def _one_shot_fired(self, db, name: str) -> bool:
+        async with db._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT 1 FROM checkpoint_alerts WHERE checkpoint_name = $1", name
+            )
+            return row is not None
+
+    async def _mark_one_shot(self, db, name: str) -> None:
+        async with db._pool.acquire() as conn:
+            await conn.execute(
+                "INSERT INTO checkpoint_alerts (checkpoint_name, fired_at) VALUES ($1, NOW())",
+                name,
+            )
+
+    async def _last_fired_at(self, db, name: str) -> Optional[datetime]:
+        async with db._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                "SELECT last_fired_at FROM checkpoint_alerts WHERE checkpoint_name = $1",
+                name,
+            )
+        if row is None:
+            return None
+        return row["last_fired_at"]
+
+    async def _upsert_repeatable(self, db, name: str) -> None:
+        async with db._pool.acquire() as conn:
+            await conn.execute(
+                """
+                INSERT INTO checkpoint_alerts (checkpoint_name, fired_at, last_fired_at)
+                VALUES ($1, NOW(), NOW())
+                ON CONFLICT (checkpoint_name) DO UPDATE SET last_fired_at = NOW()
+                """,
+                name,
+            )
+
+    # ── checkpoints ──────────────────────────────────────────────────────────
+
+    async def _check_primer_match_accepted(self, db) -> None:
+        if await self._one_shot_fired(db, "primer_match_accepted"):
+            return
+
+        async with db._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                """
+                SELECT match_score, poly_question, mfld_market_title
+                FROM manifold_match_audit
+                WHERE match_status = 'accepted'
+                ORDER BY timestamp ASC
+                LIMIT 1
+                """
+            )
+        if not row:
+            return
+
+        score  = float(row["match_score"] or 0.0)
+        poly_q = (row["poly_question"] or "")[:60]
+        mfld_t = (row["mfld_market_title"] or "")[:60]
+
+        await telegram._send(
+            f"✅ Primer match Manifold accepted — score={score:.2f} "
+            f"poly='{poly_q}' mfld='{mfld_t}'"
+        )
+        await self._mark_one_shot(db, "primer_match_accepted")
+        log.info("checkpoint primer_match_accepted fired")
+
+    async def _check_primer_trade_phase6(self, db) -> None:
+        if await self._one_shot_fired(db, "primer_trade_phase6"):
+            return
+
+        async with db._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                """
+                SELECT question, mfld_match_score, edge_net
+                FROM trades
+                WHERE mfld_match_score IS NOT NULL
+                  AND (excluded_from_metrics IS NOT TRUE)
+                ORDER BY timestamp ASC
+                LIMIT 1
+                """
+            )
+        if not row:
+            return
+
+        question = (row["question"] or "")[:70]
+        score    = float(row["mfld_match_score"] or 0.0)
+        edge     = float(row["edge_net"] or 0.0)
+
+        await telegram._send(
+            f"🎯 Primer trade Phase-6 limpio — {question} "
+            f"score={score:.2f} edge={edge:.3f}"
+        )
+        await self._mark_one_shot(db, "primer_trade_phase6")
+        log.info("checkpoint primer_trade_phase6 fired")
+
+    async def _check_primer_resolved(self, db) -> None:
+        if await self._one_shot_fired(db, "primer_resolved"):
+            return
+
+        async with db._pool.acquire() as conn:
+            row = await conn.fetchrow(
+                """
+                SELECT question, resolution, close_pnl
+                FROM trades
+                WHERE resolution IS NOT NULL
+                  AND (excluded_from_metrics IS NOT TRUE)
+                ORDER BY closed_at ASC
+                LIMIT 1
+                """
+            )
+        if not row:
+            return
+
+        question   = (row["question"] or "")[:70]
+        resolution = float(row["resolution"] or 0.0)
+        pnl        = float(row["close_pnl"] or 0.0)
+
+        await telegram._send(
+            f"🏁 Primer mercado resuelto — {question} "
+            f"result={resolution} pnl={pnl:.2f}"
+        )
+        await self._mark_one_shot(db, "primer_resolved")
+        log.info("checkpoint primer_resolved fired")
+
+    async def _check_exposure_cerca_cap(
+        self, db, exposure_pct: float, exposure_cap_pct: float
+    ) -> None:
+        if exposure_pct < 0.80 * exposure_cap_pct:
+            return
+
+        last = await self._last_fired_at(db, "exposure_cerca_cap")
+        if last is not None:
+            now_utc = datetime.now(timezone.utc)
+            if last.tzinfo is None:
+                last = last.replace(tzinfo=timezone.utc)
+            elapsed_hours = (now_utc - last).total_seconds() / 3600
+            if elapsed_hours < _EXPOSURE_COOLDOWN_HOURS:
+                return
+
+        await telegram._send(
+            f"⚠️ Exposure al 80% del cap — revisar posiciones "
+            f"({exposure_pct * 100:.1f}% / {exposure_cap_pct * 100:.1f}%)"
+        )
+        await self._upsert_repeatable(db, "exposure_cerca_cap")
+        log.info(
+            "checkpoint exposure_cerca_cap fired (%.1f%% / %.1f%%)",
+            exposure_pct * 100, exposure_cap_pct * 100,
+        )

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/risk/manager.py

@@ -62,6 +62,17 @@ class Order:
     feat_news_lo:    float = 0.0
     feat_mfld_lo:    float = 0.0
     feat_btc_dom_lo: float = 0.0
+    # Manifold audit fields (propagated from TradingSignal → Trade → DB)
+    mfld_audit_id:    Optional[str]   = None
+    mfld_market_id:   Optional[str]   = None
+    mfld_market_title: Optional[str]  = None
+    mfld_market_url:  Optional[str]   = None
+    mfld_prob_raw:    Optional[float] = None
+    mfld_prob_final:  Optional[float] = None
+    mfld_inverted:    bool            = False
+    mfld_match_score: Optional[float] = None
+    mfld_match_reason: Optional[str]  = None
+    mfld_match_status: Optional[str]  = None
 
 
 class RiskManager:
@@ -159,4 +170,15 @@ class RiskManager:
             feat_news_lo=signal.feat_news_lo,
             feat_mfld_lo=signal.feat_mfld_lo,
             feat_btc_dom_lo=signal.feat_btc_dom_lo,
+            # Manifold audit
+            mfld_audit_id=signal.mfld_audit_id,
+            mfld_market_id=signal.mfld_market_id,
+            mfld_market_title=signal.mfld_market_title,
+            mfld_market_url=signal.mfld_market_url,
+            mfld_prob_raw=signal.mfld_prob_raw,
+            mfld_prob_final=signal.mfld_prob_final,
+            mfld_inverted=signal.mfld_inverted,
+            mfld_match_score=signal.mfld_match_score,
+            mfld_match_reason=signal.mfld_match_reason,
+            mfld_match_status=signal.mfld_match_status,
         )

bot/strategy/bayesian.py

@@ -12,16 +12,21 @@ Polymarket might reflect in a slow-moving order book.
 """
 import logging
 import math
+import os
+import re
+import uuid
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import Optional, TYPE_CHECKING
 
 from bot.data.polymarket import Market, market_family_key
 from bot.data.external import ExternalSignals
+from bot.data.manifold import MANIFOLD_MATCHER_VERSION, ManifoldMatchResult
 
 if TYPE_CHECKING:
     from bot.data.news import NewsClient
     from bot.data.manifold import ManifoldClient
+    from bot.data.db import Database
 
 log = logging.getLogger(__name__)
 
@@ -58,11 +63,81 @@ NEWS_LOGODDS_WEIGHT = 1.5
 # Weaker than NEWS_LOGODDS_WEIGHT because Manifold can have illiquid/stale markets.
 MANIFOLD_LOGODDS_WEIGHT = 0.6
 
+
+def _env_bool(name: str, default: bool) -> bool:
+    return os.getenv(name, str(default)).strip().lower() in ("1", "true", "yes", "on")
+
+
+# ── Manifold activation flags ──────────────────────────────────────────────────
+# Manifold has been retired as an ACTIVE trading signal: a per-category coverage
+# audit (see /api/metrics/manifold-coverage) showed coverage_rate=0.0 across every
+# category in the bot's current universe, so any edge it produced was false edge.
+#
+# MANIFOLD_SIGNAL_ENABLED (default False): when False, Manifold is observational
+#   only — its probability never touches the edge model: no manifold_log_adj, no
+#   confidence bump, feat_mfld_lo stays 0.0 (so it can never be the dominant
+#   feature), and it never contributes to a trade.
+# MANIFOLD_AUDIT_ENABLED (default True): when True the matcher still runs and
+#   audit/coverage rows + cooldowns are written, preserving the trail so we can
+#   decide later whether to reactivate Manifold in a universe with real coverage.
+# The matcher is only called when at least one flag is on.
+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
 
 
+# ─────────────────────────────────────────────────────────────────────────────
+# Manifold evaluation cooldown
+#
+# Per-market backoff so the trading loop stops re-querying Manifold (and flooding
+# manifold_match_audit) for markets whose verdict is stable.  Computed from the
+# match result; longer for verdicts that essentially never change.
+#   no_results            → 24 h  (Manifold has no market on this topic)
+#   rejected/low_score    → 24 h  (best candidate below Jaccard threshold)
+#   rejected/outcome_mism. → 24 h  (outcome types differ)
+#   rejected/ambiguous    → 24 h  (party named but inversion unverifiable)
+#   rejected/conditional  →  7 d  (premise-gated market; structural, won't change)
+#   accepted              →  1 h  (signal is live; refresh probability hourly)
+# ─────────────────────────────────────────────────────────────────────────────
+def _cooldown_for(result: ManifoldMatchResult) -> tuple[timedelta, str]:
+    """Map a Manifold match result to (retry_delay, cooldown_reason)."""
+    if result.status == "accepted":
+        return timedelta(hours=1), "accepted"
+    if result.status == "no_results":
+        return timedelta(hours=24), "no_results"
+    # rejected — classify by the reason text the matcher produced
+    reason = result.match_reason or "rejected"
+    if "conditional_market" in reason:
+        return timedelta(days=7), reason
+    # outcome_mismatch, ambiguous_inversion, and low_score (jaccard<threshold)
+    # all settle in 24 h.
+    return timedelta(hours=24), reason
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # Phase 4 — Regime-based minimum edge (uses edge_NET, not edge_gross)
 # ─────────────────────────────────────────────────────────────────────────────
@@ -92,24 +167,82 @@ 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
 
 
+def has_token(text: str, token: str) -> bool:
+    """
+    True if `token` appears in `text` as a standalone word.
+
+    Short crypto tickers (eth, sol, ada, …) must NOT match inside ordinary
+    words — "Seth", "dissolved", "Canada" — but must still match the usual
+    market phrasings: "ETH", "$ETH", "ETH/USD", "SOL reach $200".  Boundaries
+    are any non-alphanumeric character (or start/end of string), so "$" and
+    "/" delimit correctly.
+    """
+    return re.search(
+        rf"(?<![A-Za-z0-9]){re.escape(token)}(?![A-Za-z0-9])", text, re.IGNORECASE
+    ) is not None
+
+
 # ─────────────────────────────────────────────────────────────────────────────
 # 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).
@@ -170,6 +303,19 @@ class TradingSignal:
     feat_news_lo:    float = 0.0
     feat_mfld_lo:    float = 0.0
     feat_btc_dom_lo: float = 0.0
+    # ── Manifold match audit (propagated → Order → Trade → DB) ───────────────
+    # mfld_audit_id: UUID of the manifold_match_audit row; used to mark
+    #   used_in_trade=TRUE after executor confirms the trade was executed.
+    mfld_audit_id:    Optional[str]   = None
+    mfld_market_id:   Optional[str]   = None
+    mfld_market_title: Optional[str]  = None
+    mfld_market_url:  Optional[str]   = None
+    mfld_prob_raw:    Optional[float] = None
+    mfld_prob_final:  Optional[float] = None
+    mfld_inverted:    bool            = False
+    mfld_match_score: Optional[float] = None
+    mfld_match_reason: Optional[str]  = None
+    mfld_match_status: Optional[str]  = None
 
 
 class BayesianStrategy:
@@ -201,10 +347,12 @@ class BayesianStrategy:
         self,
         news: Optional["NewsClient"] = None,
         manifold: Optional["ManifoldClient"] = None,
+        db: Optional["Database"] = None,
     ) -> None:
         self._signal_count = 0
         self._news = news
         self._manifold = manifold
+        self._db = db
         self._news_queries_this_cycle = 0
         # Per-cycle counters — reset by reset_cycle(), read by get_cycle_stats()
         self._skip_family: int = 0
@@ -216,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."""
@@ -227,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."""
@@ -246,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(
@@ -253,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
@@ -277,13 +495,18 @@ class BayesianStrategy:
             "below", "under", "less than", "lower", "drop",
         ])
 
-        is_btc = "btc" in question_lower or "bitcoin" in question_lower
-        is_eth = "eth" in question_lower or "ethereum" in question_lower
-        is_sol = "sol" in question_lower or "solana" in question_lower
-        is_xrp = "xrp" in question_lower or "ripple" in question_lower
-        is_doge = "doge" in question_lower or "dogecoin" in question_lower
+        # Short tickers need word boundaries: "Seth" contains "eth",
+        # "dissolved" contains "sol", "Canada" contains "ada".  Long
+        # unambiguous names (bitcoin, ethereum, …) stay as substrings.
+        is_btc = has_token(question_lower, "btc") or "bitcoin" in question_lower
+        is_eth = has_token(question_lower, "eth") or "ethereum" in question_lower
+        is_sol = has_token(question_lower, "sol") or "solana" in question_lower
+        is_xrp = has_token(question_lower, "xrp") or "ripple" in question_lower
+        is_doge = has_token(question_lower, "doge") or "dogecoin" in question_lower
         is_altcoin = is_sol or is_xrp or is_doge or any(
-            w in question_lower for w in ["ltc", "litecoin", "bnb", "ada", "cardano", "avax", "avalanche"]
+            has_token(question_lower, t) for t in ["ltc", "bnb", "ada", "avax"]
+        ) or any(
+            w in question_lower for w in ["litecoin", "cardano", "avalanche"]
         )
         is_general_crypto = any(
             w in question_lower for w in ["crypto", "market cap", "total market", "altcoin", "defi"]
@@ -306,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:
@@ -313,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 ────────────────────────────
@@ -324,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
@@ -331,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 ────────────────────────────────────
@@ -341,40 +568,47 @@ 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 ──────────────────────────────────
         sources: list[str] = [f"Prior=poly({prior:.3f})"]
         adjustments: list[float] = []
 
-        # Signal 1: price momentum (asset-specific or BTC as sentiment proxy)
+        # Momentum and Fear & Greed only make sense for price markets, where
+        # is_price_above gives the adjustment a meaningful sign.  For
+        # politics/tech/events there is no above/below notion — is_price_above
+        # defaults to False (or flips on accidental wording like "reach"), so
+        # applying these signals just injected sign noise.  Skip them entirely;
+        # their contributions stay 0.0 → feat_mom_lo / feat_fg_lo = 0.0.
+        is_non_price = is_politics or is_tech or is_events
+
+        # Signal 1: price momentum (asset-specific; price markets only)
+        _momentum_contribution = 0.0
+        if not is_non_price:
             if is_btc:
                 momentum = ext.btc_change_24h
                 asset_label = "BTC"
             elif is_eth:
                 momentum = ext.eth_change_24h
                 asset_label = "ETH"
-        elif is_politics or is_tech or is_events:
-            momentum = ext.btc_change_24h
-            asset_label = "BTC(sentiment)"
             else:
                 momentum = ext.total_market_cap_change
                 asset_label = "total mktcap"
 
-        _momentum_contribution = 0.0
             if abs(momentum) > 2:
                 momentum_adj = math.tanh(momentum / 20) * 0.15
-            if is_politics or is_tech or is_events:
-                momentum_adj *= 0.5
                 _momentum_contribution = momentum_adj if is_price_above else -momentum_adj
                 adjustments.append(_momentum_contribution)
                 sources.append(f"{asset_label} 24h: {momentum:+.1f}%")
 
-        # Signal 2: Fear & Greed
+        # Signal 2: Fear & Greed (price markets only)
+        _fg_contribution = 0.0
+        if not is_non_price:
             fg = ext.fear_greed_index
             if fg > 70:
                 fg_adj = 0.06
@@ -388,8 +622,13 @@ class BayesianStrategy:
             _fg_contribution = fg_adj if is_price_above else -fg_adj
             adjustments.append(_fg_contribution)
 
-        # Signal 3: BTC dominance — hurts altcoins when high
+        # Signal 3: BTC dominance — hurts altcoins when high (price markets only)
+        # Like momentum and Fear & Greed above: no demonstrated causality for
+        # politics/tech/events, even when they legitimately mention a ticker
+        # ("Will the ETH ETF be approved?").  For non-price markets the
+        # contribution stays 0.0 → feat_btc_dom_lo = 0.0.
         _btc_dom_contribution = 0.0
+        if not is_non_price:
             if (is_eth or is_altcoin or is_general_crypto) and ext.btc_dominance > 55:
                 _btc_dom_contribution = -0.03 if is_price_above else 0.03
                 adjustments.append(_btc_dom_contribution)
@@ -403,14 +642,24 @@ 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
-        if is_politics and self._news is not None:
+        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.
+        if is_politics and self._news is not None and self._news.enabled:
             if self._news_queries_this_cycle < MAX_NEWS_QUERIES_PER_CYCLE:
                 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,
@@ -419,18 +668,129 @@ class BayesianStrategy:
         # Signal 5: Manifold cross-market probability (politics + tech)
         # Applies a log-odds adjustment proportional to divergence from prior.
         # No query budget — 30 min cache means network cost is paid once per cycle.
+        # Now uses ManifoldMatchResult for stricter semantic validation and audit.
         manifold_log_adj = 0.0
         manifold_used = False
-        if (is_politics or is_tech) and self._manifold is not None:
-            manifold_prob = await self._manifold.get_probability(market.question)
-            if manifold_prob is not None:
+        manifold_result: Optional[ManifoldMatchResult] = None
+        audit_id: Optional[str] = None
+
+        if ((is_politics or is_tech) and self._manifold is not None
+                and (MANIFOLD_AUDIT_ENABLED or MANIFOLD_SIGNAL_ENABLED)):
+            # ── Cooldown gate ────────────────────────────────────────────────
+            # Skip markets whose Manifold verdict was recently settled to a
+            # stable value.  A skip is equivalent to a no-signal: the matcher is
+            # NOT called and NO manifold_match_audit row is written, so only real
+            # evaluations are recorded.  See _cooldown_for() and the
+            # manifold_eval_cooldown table.
+            in_cooldown = False
+            if self._db is not None and market.id:
+                try:
+                    cd = await self._db.get_manifold_cooldown(market.id)
+                except Exception as exc:
+                    log.warning("Failed to read manifold cooldown: %s", exc)
+                    cd = None
+                if cd is not None and datetime.now(timezone.utc) < cd["retry_after"]:
+                    in_cooldown = True
+                    log.info(
+                        "MANIFOLD_COOLDOWN skip market=%s | last_status=%s "
+                        "retry_after=%s | %s",
+                        market.id, cd["last_status"],
+                        cd["retry_after"].isoformat(), market.question[:50],
+                    )
+
+            if not in_cooldown:
+                manifold_result = await self._manifold.get_match(market.question)
+
+                # Persist audit record for ALL outcomes (accepted / rejected / no_results).
+                # Gated by MANIFOLD_AUDIT_ENABLED so the audit/coverage trail and
+                # cooldowns can be kept even while Manifold is observational-only.
+                if MANIFOLD_AUDIT_ENABLED and self._db is not None:
+                    if not market.id:
+                        log.error(
+                            "MANIFOLD_AUDIT: market.id is None/empty — skipping audit save | "
+                            "question=%r", market.question[:60],
+                        )
+                    else:
+                        audit_id = str(uuid.uuid4())
+                        try:
+                            await self._db.save_manifold_audit(
+                                audit_id=audit_id,
+                                poly_market_id=market.id,
+                                poly_question=market.question,
+                                search_query=manifold_result.search_query,
+                                mfld_market_id=manifold_result.market_id,
+                                mfld_market_title=manifold_result.market_title,
+                                mfld_market_url=manifold_result.market_url,
+                                prob_raw=manifold_result.prob_raw,
+                                prob_final=manifold_result.prob_final,
+                                inverted=manifold_result.inverted,
+                                match_score=manifold_result.match_score,
+                                match_reason=manifold_result.match_reason,
+                                match_status=manifold_result.status,
+                                poly_outcome_type=manifold_result.poly_outcome_type,
+                                mfld_outcome_type=manifold_result.mfld_outcome_type,
+                                matcher_version=MANIFOLD_MATCHER_VERSION,
+                            )
+                        except Exception as exc:
+                            log.warning("Failed to save manifold audit: %s", exc)
+                            audit_id = None
+
+                    # Record the cooldown so this market is not re-queried every
+                    # cycle.  Written even if the audit save above failed — we
+                    # still performed a real evaluation.
+                    if market.id:
+                        delay, cd_reason = _cooldown_for(manifold_result)
+                        try:
+                            await self._db.upsert_manifold_cooldown(
+                                poly_market_id=market.id,
+                                last_status=manifold_result.status,
+                                retry_after=datetime.now(timezone.utc) + delay,
+                                cooldown_reason=cd_reason,
+                            )
+                        except Exception as exc:
+                            log.warning("Failed to save manifold cooldown: %s", exc)
+
+                # Structured log — both forms for compatibility
+                log.info(
+                    "MANIFOLD_MATCH poly='%s' mfld='%s' score=%s raw=%s final=%s"
+                    " inverted=%s status=%s reason=%s",
+                    market.question, manifold_result.market_title,
+                    manifold_result.match_score, manifold_result.prob_raw,
+                    manifold_result.prob_final, manifold_result.inverted,
+                    manifold_result.status, manifold_result.match_reason,
+                )
+                log.info("MANIFOLD_MATCH", extra={
+                    "poly_question": market.question,
+                    "mfld_title":    manifold_result.market_title,
+                    "score":         manifold_result.match_score,
+                    "prob_raw":      manifold_result.prob_raw,
+                    "prob_final":    manifold_result.prob_final,
+                    "inverted":      manifold_result.inverted,
+                    "status":        manifold_result.status,
+                    "reason":        manifold_result.match_reason,
+                })
+
+                if (MANIFOLD_SIGNAL_ENABLED
+                        and manifold_result.status == "accepted"
+                        and manifold_result.prob_final is not None):
+                    # ACTIVE signal path — only when explicitly enabled.
                     manifold_used = True
                     self._manifold_fetched += 1
-                m_clamped = max(0.05, min(0.95, manifold_prob))
+                    m_clamped = max(0.05, min(0.95, manifold_result.prob_final))
                     m_log = math.log(m_clamped / (1 - m_clamped))
                     p_log = math.log(prior / (1 - prior))
                     manifold_log_adj = (m_log - p_log) * MANIFOLD_LOGODDS_WEIGHT
-                sources.append(f"Manifold:{manifold_prob:.2f}")
+                    sources.append(f"Manifold:{manifold_result.prob_final:.2f}")
+                elif not MANIFOLD_SIGNAL_ENABLED:
+                    # Observational-only: matched/audited but NEVER fed to the edge
+                    # model.  manifold_log_adj stays 0.0 → no confidence bump,
+                    # feat_mfld_lo=0.0 (cannot be dominant), no trade contribution.
+                    log.info(
+                        "Manifold: observational_only — signal disabled "
+                        "(MANIFOLD_SIGNAL_ENABLED=false) | market=%s status=%s",
+                        market.id, manifold_result.status,
+                    )
+                    sources.append("Manifold: observational_only")
 
         # Confidence cap: macro/politics/tech signals are weaker proxies
         confidence_cap = 0.65 if (is_macro or is_politics or is_tech or is_events) else 0.90
@@ -438,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
@@ -461,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} "
@@ -481,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:
@@ -509,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} | "
@@ -560,6 +1021,21 @@ class BayesianStrategy:
             feat_news_lo=feat_news_lo,
             feat_mfld_lo=feat_mfld_lo,
             feat_btc_dom_lo=feat_btc_dom_lo,
+            # Manifold match audit — propagated through Order → Trade → DB.
+            # mfld_audit_id is the hook main.py uses to flip the audit row's
+            # used_in_trade=TRUE; suppress it when observational so the trail
+            # truthfully shows Manifold drove no trades. The mfld_* fields below
+            # stay as observational record (feat_mfld_lo is already 0.0).
+            mfld_audit_id=(audit_id if MANIFOLD_SIGNAL_ENABLED else None),
+            mfld_market_id=manifold_result.market_id if manifold_result else None,
+            mfld_market_title=manifold_result.market_title if manifold_result else None,
+            mfld_market_url=manifold_result.market_url if manifold_result else None,
+            mfld_prob_raw=manifold_result.prob_raw if manifold_result else None,
+            mfld_prob_final=manifold_result.prob_final if manifold_result else None,
+            mfld_inverted=manifold_result.inverted if manifold_result else False,
+            mfld_match_score=manifold_result.match_score if manifold_result else None,
+            mfld_match_reason=manifold_result.match_reason if manifold_result else None,
+            mfld_match_status=manifold_result.status if manifold_result else None,
         )
 
 

dashboard/package.json

@@ -14,7 +14,7 @@
     "recharts": "^2.12.4"
   },
   "devDependencies": {
-    "@vitejs/plugin-react": "^6.0.0",
+    "@vitejs/plugin-react": "^4.2.1",
     "vite": "^5.2.0"
   }
 }

dashboard/src/App.jsx

@@ -200,8 +200,12 @@ export default function App() {
         <MetricCard
           title="Sharpe"
           value={fmt(summary.sharpe_ratio)}
-          subtitle="Objetivo ≥ 0.5"
-          progress={Math.min(1, summary.sharpe_ratio / 2)}
+          subtitle={
+            summary.sharpe_ratio == null
+              ? `Muestra insuficiente: ${summary.resolved_count}/${summary.min_resolved_required} resueltos, ${summary.days_observed}/${summary.min_days_required} días`
+              : 'Objetivo ≥ 0.5'
+          }
+          progress={summary.sharpe_ratio == null ? 0 : Math.min(1, summary.sharpe_ratio / 2)}
           progressColor={summary.sharpe_ratio >= 0.5 ? 'var(--green)' : 'var(--amber)'}
         />
         <MetricCard
@@ -216,6 +220,12 @@ export default function App() {
           value={fmtUSD(summary.total_deployed)}
           subtitle={`${summary.total_trades} trades`}
         />
+        <MetricCard
+          title="Cash Disponible"
+          value={fmtUSD(summary.cash_available)}
+          subtitle={`${fmtPct(summary.cash_available / summary.paper_bankroll)} del bankroll`}
+          progress={summary.cash_available / summary.paper_bankroll}
+        />
       </div>
 
       {/* Performance chart */}

docs/informe-final.md

@@ -0,0 +1,165 @@
+# Informe final de paper trading — polymarket-bot
+
+**Snapshot: 2026-07-06** · Generado desde la base de datos de producción con el bot
+aún operativo, como parte de la Fase 1 del plan de decomisión (bloqueo ISP de
+Polymarket en España, mayo 2026).
+
+---
+
+## ⚠️ Advertencia de tamaño muestral (léase primero)
+
+**12 trades ejecutados, de los cuales solo 2 llegaron a resolución, no constituyen
+evidencia estadística de nada.** Ni el P&L positivo demuestra que la estrategia
+funcione, ni su ausencia lo habría refutado. El propio bot lo sabía: sus criterios
+de promoción a dinero real exigían ≥10 mercados resueltos y ≥30 días de
+observación (`promotion_ready: false`, win rate y Sharpe deliberadamente
+reportados como `n/a (insufficient_sample)` en toda la instrumentación).
+
+El valor demostrable de este proyecto no está en el P&L: está en la
+infraestructura de evaluación, el pipeline de datos y la disciplina de gates que
+impidió que el bot tradease sin ventaja medible. Este informe presenta el P&L
+como lo que es — **una anécdota** — y dedica el grueso al sistema.
+
+---
+
+## 1. Resultados de trading (anecdóticos)
+
+Bankroll simulado: **$10.000** (USDC virtual) · Capital desplegado: **$2.447** ·
+Periodo de actividad: **14–22 de abril de 2026** (los 12 trades se abrieron en 9
+días; después, el endurecimiento de los gates de edge y confianza dejó el bot en
+0 trades — funcionando, evaluando ~1.350 ciclos/día, sin encontrar ventaja neta
+que superara sus propios umbrales).
+
+### P&L realizado: **+$247,78** (2 mercados resueltos)
+
+| Mercado | Dirección | Entrada | Resultado | P&L neto |
+|---|---|---|---|---|
+| Will Ken Paxton win the 2026 Texas Republican Primary? | BUY_YES | 0,618 | ✅ YES (2026-06-11) | **+$299,06** |
+| Will NVIDIA be the largest company in the world…? | BUY_NO | 0,156 | ❌ YES (2026-06-30) | **−$51,28** |
+
+P&L neto de comisiones (2% simulado por lado). 1 acierto / 1 fallo: n=2.
+
+### P&L no realizado (mark-to-market): **+$809,59** (5 posiciones abiertas)
+
+Marcado a los precios reales de Polymarket (Gamma API) el 2026-07-06. Estas
+posiciones **no están cerradas**: el número de abajo cambiaría cada día que los
+mercados sigan moviéndose, y varios no resuelven hasta noviembre de 2026.
+
+| Mercado | Dirección | Entrada | Precio actual (lado) | Coste neto | MTM P&L |
+|---|---|---|---|---|---|
+| Karen Bass gana la alcaldía de LA 2026 | BUY_YES | 0,268 | 0,595 | $510,00 | **+$600,07** |
+| OpenAI IPO antes del 31-dic-2026 | BUY_NO | 0,618 | 0,785 | $510,00 | **+$125,11** |
+| Demócratas ganan el Senado de Texas 2026 | BUY_YES | 0,418 | 0,435 | $509,49 | **+$10,32** |
+| Demócratas ganan gobernación de Ohio 2026 | BUY_NO | 0,458 | 0,520 | $407,74 | **+$46,12** |
+| Republicanos ganan gobernación de Nebraska 2026 | BUY_NO | 0,158 | 0,170 | $510,00 | **+$27,97** |
+
+> Nota metodológica: el tracker interno reportaba `unrealized_pnl_est = +$522,73`,
+> pero esa cifra es la **expectativa del propio modelo a fecha de entrada**
+> (edge_net × coste), no un marcado a mercado. La tabla de arriba usa precios
+> reales de mercado del día del snapshot, que es el criterio honesto. Ambas
+> cifras coinciden en el signo; ninguna de las dos es un resultado realizado.
+
+### Los otros 5 cierres (sin P&L): la historia real de abril
+
+De los 7 trades cerrados, solo 2 cerraron por resolución. Los otros 5 los cerró
+el propio bot al detectar defectos en su lógica — y son la parte más
+instructiva del histórico:
+
+- **3 por conflicto de familia**: dos posiciones sobre mercados hermanos del
+  mismo evento (p. ej. YES-demócratas y YES-republicanos en la misma
+  gobernación) — el agrupador de familias los detectó y cerró el lado débil.
+- **2 por el bug de inversión de Manifold**: la señal cruzada de Manifold se
+  aplicaba invertida en mercados espejo. Detectado por instrumentación, las
+  posiciones afectadas se cerraron y el matching se movió a modo
+  observacional-only con auditoría completa (165.538 filas en
+  `manifold_match_audit`).
+- **1 excluido de métricas** (`invalid_manifold_match_legacy`, P&L $0,00).
+
+Después de esa semana el bot no volvió a operar: los gates endurecidos
+(edge neto > mínimo por régimen, confianza ≥ 0,55, guardrail de noticias
+prior±0,25) filtraron el 100% de las ~223.000 evaluaciones posteriores.
+**Un sistema de paper trading cuyo resultado principal es "no tradees sin
+ventaja" es un resultado**, y está instrumentado hasta el último skip.
+
+---
+
+## 2. El sistema (el activo real)
+
+### Pipeline de señales
+
+| Métrica | Valor (snapshot 2026-07-06) |
+|---|---|
+| Ciclos de evaluación archivados | **5.356** (cadencia ~64 s) |
+| Evaluaciones mercado-ciclo archivadas | **222.738** |
+| Mercados distintos evaluados | 87 (events 114,9k · politics 96,0k · crypto/finance 7,7k · tech 4,2k evaluaciones) |
+| Archivo `signals` operativo desde | 2026-07-02 (replay R0) — crece ~55k filas/día |
+| Histórico de métricas diarias | **81 días** (2026-04-14 → 2026-07-06, `metrics_daily`) |
+| Snapshots de contexto externo | 5.356 (BTC, dominancia, Fear&Greed por ciclo) |
+
+Cada evaluación archiva el prior, la probabilidad estimada, el edge bruto/neto,
+la descomposición por feature (Fear&Greed, momentum, noticias, Manifold,
+dominancia BTC en log-odds), el gate que la bloqueó y el contexto de mercado —
+suficiente para reproducir la decisión completa offline.
+
+### Motor de replay (R0–R2, julio 2026)
+
+- Inyección de reloj + replay determinista de ciclos archivados:
+  **22.697/22.697 decisiones idénticas** al reproducir el histórico.
+- Joiner diario de resoluciones (CronJob 00:30 UTC) contra la Gamma API:
+  cobertura de outcomes 9/87 mercados al snapshot, creciendo sola.
+- Diseñado para calibrar sobre **todas** las evaluaciones (no solo trades),
+  esquivando el problema del n=12.
+
+### Estrategia (congelada desde 2026-07-03)
+
+Prior desde el precio de Polymarket → ajuste bayesiano en log-odds con señales
+externas (GNews con presupuesto de 5 consultas/día y guardrail prior±0,25;
+Fear&Greed; momentum BTC; Manifold observacional) → gates en cascada: prior
+extremo, edge bruto por régimen de volatilidad, edge neto tras comisión 2% y
+spread, confianza mínima 0,55, conflictos de familia, guard de reentrada.
+Sizing por Kelly fraccionado con techo por posición.
+
+### Infraestructura
+
+3 imágenes (bot / API FastAPI / dashboard) construidas por Gitea Actions solo
+cuando cambian sus fuentes, desplegadas por ArgoCD (prune + selfHeal) en k3s;
+PostgreSQL 16 en StatefulSet; secrets vía Infisical operator; smoke test
+PostSync contra la API; backups nocturnos con sync offsite (rclone → MEGA);
+observabilidad en Grafana + uptime-kuma + notificaciones Telegram (deploys,
+checkpoints del bot, fallos de sync). Retención de métricas y join de outcomes
+como CronJobs. **88 días de uptime del StatefulSet al snapshot.**
+
+---
+
+## 3. Estado de la base de datos al snapshot
+
+| Tabla | Filas | Contenido |
+|---|---|---|
+| `signals` | 222.549* | Archivo de evaluaciones por ciclo (desde 2026-07-02) |
+| `manifold_match_audit` | 165.538 | Auditoría completa del matching Polymarket↔Manifold |
+| `replay_decisions` | 22.697 | Decisiones reproducidas por el motor de replay |
+| `ext_snapshots` | 5.353* | Contexto externo por ciclo |
+| `metrics_daily` | 516 | Cierres diarios (81 días) + snapshots intradía de hoy |
+| `markets` | 87 | Universo evaluado |
+| `trades` | 12 | Ledger completo de paper trades |
+| `market_outcomes` | 9 | Resoluciones UMA-final joineadas |
+| `replay_runs` / `manifold_eval_cooldown` / `checkpoint_alerts` | 1 / 76 / 3 | Metadatos |
+
+\* La BD sigue viva; las tablas por ciclo crecen ~55k y ~1,3k filas/día
+respectivamente. Backup verificado del 2026-07-06 en
+`/data/backups/backups/polymarket-decommission/` (pg_dump -Fc + CSV.gz +
+checksums; restore de prueba con recuentos idénticos 11/11 tablas). El dump
+final se hará en la Fase 3, justo antes del apagado.
+
+---
+
+## 4. Conclusión
+
+El proyecto se archiva por una causa externa (bloqueo regulatorio del origen de
+datos), no por fracaso técnico. Lo que queda: un motor de evaluación bayesiano
+determinista y auditable, un pipeline de datos que archivó cada decisión con su
+contexto completo, y la evidencia — anecdótica en P&L, sólida en ingeniería —
+de que el sistema prefería no operar antes que operar sin ventaja. Todos los
+módulos centrales (edge, familias, replay, observabilidad) son agnósticos de la
+fuente y quedan listos para un eventual pivote a Manifold
+(`docs/pivot-manifold.md`, Fase 5).

tests/conftest.py

@@ -0,0 +1,10 @@
+"""Test environment shims.
+
+The bot runs on python:3.11-slim in production; local dev machines may have
+3.10, which lacks datetime.UTC (added in 3.11). Alias it so modules using
+`from datetime import UTC` import cleanly under 3.10.
+"""
+import datetime
+
+if not hasattr(datetime, "UTC"):
+    datetime.UTC = datetime.timezone.utc

tests/test_api_summary_consistency.py

@@ -0,0 +1,106 @@
+"""
+Tests for bug #7 — /api/summary must agree with the executor's cash model.
+
+Regression: /api/summary computed total_trades as len() over a LIMIT-500
+query (capped once history grows) and reimplemented cash as
+bankroll - sum(net_cost of open trades) from that same capped query.
+
+Fix: counts come from COUNT(*) (compute_metrics_from_db) and cash comes from
+cash_available() — the same helper PaperExecutor.initialize() uses — fed by
+the same source (get_open_position_data). This test runs both consumers
+against one fake DB state and asserts they report identical cash.
+"""
+import asyncio
+
+import pytest
+
+import api.main as api_main
+from bot.executor.paper import PaperExecutor, cash_available
+
+
+BANKROLL = 10_000.0  # PAPER_BANKROLL default used by both bot and API
+
+
+class FakeDB:
+    """One DB state served to both the API endpoint and the executor."""
+
+    def __init__(self, positions: dict[str, float], total_net_cost: float,
+                 total_trades: int, open_count: int):
+        self._positions = positions
+        self._total_net_cost = total_net_cost
+        self._total = total_trades
+        self._open = open_count
+
+    # Shared source: executor.initialize() and /api/summary both call this.
+    async def get_open_position_data(self):
+        return dict(self._positions), self._total_net_cost
+
+    # /api/summary only:
+    async def get_metrics_history(self, days=1):
+        return []
+
+    async def compute_metrics_from_db(self):
+        return {
+            "total_trades": self._total,
+            "open_count": self._open,
+            "closed_count": self._total - self._open,
+            "resolved_count": 0,
+        }
+
+    async def get_recently_closed_inverted(self, hours=24):
+        return set()
+
+    async def get_legacy_incomplete_count(self):
+        return 0
+
+    async def get_daily_pnl_closes(self):
+        return []
+
+
+def _run(db: FakeDB, monkeypatch) -> tuple[dict, PaperExecutor]:
+    monkeypatch.setattr(api_main, "db", db)
+    monkeypatch.delenv("PAPER_BANKROLL", raising=False)
+
+    async def run():
+        summary = await api_main.get_summary()
+        ex = PaperExecutor(db=db, bankroll=BANKROLL)
+        await ex.initialize()
+        return summary, ex
+
+    return asyncio.run(run())
+
+
+def test_api_and_executor_report_same_cash(monkeypatch):
+    db = FakeDB(
+        positions={"m1": 100.0, "m2": 80.0},
+        total_net_cost=183.60,   # 180 + fees
+        total_trades=12,
+        open_count=2,
+    )
+    summary, ex = _run(db, monkeypatch)
+    assert summary["cash_available"] == pytest.approx(ex.get_portfolio().cash)
+    assert summary["cash_available"] == pytest.approx(
+        cash_available(BANKROLL, 183.60)
+    )
+    assert summary["total_deployed"] == pytest.approx(183.60)
+
+
+def test_total_trades_not_capped_by_query_limit(monkeypatch):
+    """700 trades in DB: the old len(LIMIT 500) reported 500."""
+    db = FakeDB(
+        positions={"m1": 100.0},
+        total_net_cost=102.0,
+        total_trades=700,
+        open_count=1,
+    )
+    summary, _ = _run(db, monkeypatch)
+    assert summary["total_trades"] == 700
+    assert summary["open_trades_count"] == 1
+    assert summary["closed_trades_count"] == 699
+
+
+def test_cash_consistency_with_no_open_positions(monkeypatch):
+    db = FakeDB(positions={}, total_net_cost=0.0, total_trades=0, open_count=0)
+    summary, ex = _run(db, monkeypatch)
+    assert summary["cash_available"] == pytest.approx(BANKROLL)
+    assert ex.get_portfolio().cash == pytest.approx(BANKROLL)

tests/test_bayesian_asset_detection.py

@@ -0,0 +1,159 @@
+"""
+Tests for FASE 4 — crypto ticker detection must use word boundaries.
+
+Regression: short tickers were detected with substring matching over
+question_lower, so non-crypto markets triggered crypto flags:
+
+  "Israeli parliament dissolved"  contains "sol" → is_sol / is_altcoin
+  "Will Canada win Group B"       contains "ada" → is_altcoin
+  "Will Seth Moulton be the nominee" contains "eth" → is_eth
+
+Those flags armed the BTC-dominance signal (btc_dom_lo=+0.06 observed in
+production on politics markets).  The fix routes short tickers (btc, eth,
+sol, xrp, doge, ltc, bnb, ada, avax) through has_token(), which requires
+non-alphanumeric boundaries; long unambiguous names (bitcoin, ethereum,
+solana, cardano, …) remain substrings.
+
+The is_* flags are internal to evaluate(), so the integration tests assert
+on btc_dom_lo parsed from the structured audit log (same technique as
+test_bayesian_macro_signals.py), with btc_dominance=60 so the signal fires
+whenever an ETH/altcoin flag is set.
+"""
+import asyncio
+import logging
+import re
+
+import pytest
+
+from bot.data.external import ExternalSignals
+from bot.data.polymarket import Market
+from bot.strategy.bayesian import BayesianStrategy, has_token
+
+BTC_DOM_RE = re.compile(r"btc_dom_lo=([+-]\d+\.\d+)")
+
+
+def _make_market(question: str, category: str) -> Market:
+    return Market(
+        id="mkt-test-1",
+        condition_id="cond-test-1",
+        question=question,
+        yes_token_id="yes-tok",
+        no_token_id="no-tok",
+        yes_price=0.50,
+        no_price=0.50,
+        volume_24h=50_000.0,
+        end_date="2026-07-15T00:00:00Z",
+        active=True,
+        category=category,
+    )
+
+
+def _make_signals() -> ExternalSignals:
+    # btc_dominance=60 (>55) arms the BTC-dominance signal for any market
+    # flagged as ETH / altcoin / general-crypto.
+    return ExternalSignals(
+        btc_price=100_000.0,
+        btc_change_24h=10.0,
+        eth_price=4_000.0,
+        eth_change_24h=8.0,
+        btc_dominance=60.0,
+        fear_greed_index=80,
+        fear_greed_label="greed",
+        total_market_cap_change=5.0,
+        valid=True,
+    )
+
+
+def _evaluate_and_parse_btc_dom(question: str, category: str, caplog) -> float:
+    """Run BayesianStrategy.evaluate and return btc_dom_lo from the audit log."""
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    market = _make_market(question, category)
+    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 = BTC_DOM_RE.search(record.getMessage())
+        if m:
+            return float(m.group(1))
+    pytest.fail(
+        "No SKIP_EDGE_NET/TRADE log line with btc_dom_lo found; "
+        f"got: {[r.getMessage() for r in caplog.records]}"
+    )
+
+
+# ── has_token unit tests ─────────────────────────────────────────────────────
+
+def test_has_token_rejects_substrings_inside_words():
+    assert has_token("israeli parliament dissolved by june 30?", "sol") is False
+    assert has_token("will canada win group b?", "ada") is False
+    assert has_token("will seth moulton be the nominee?", "eth") is False
+
+def test_has_token_matches_common_market_formats():
+    assert has_token("will eth hit $5000?", "eth") is True
+    assert has_token("$eth above $5000?", "eth") is True
+    assert has_token("eth/usd above 5000?", "eth") is True
+    assert has_token("will sol reach $200?", "sol") is True
+    assert has_token("will ada reach $1?", "ada") is True
+    assert has_token("BTC to $150k?", "btc") is True  # case-insensitive
+
+
+# ── Regression: false positives must not arm the BTC-dominance signal ───────
+
+def test_israeli_parliament_market_is_not_sol(caplog):
+    """'dissolved' contains 'sol' — must NOT flag is_sol/is_altcoin."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "Israeli parliament dissolved by June 30?", "politics", caplog
+    )
+    assert btc_dom_lo == 0.0
+
+def test_canada_market_is_not_ada(caplog):
+    """'Canada' contains 'ada' — must NOT flag is_altcoin."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "Will Canada win Group B?", "events", caplog
+    )
+    assert btc_dom_lo == 0.0
+
+def test_seth_moulton_market_is_not_eth(caplog):
+    """'Seth' contains 'eth' — must NOT flag is_eth."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "Will Seth Moulton be the nominee?", "politics", caplog
+    )
+    assert btc_dom_lo == 0.0
+
+
+# ── Real ticker mentions must keep working ───────────────────────────────────
+
+def test_eth_market_detected(caplog):
+    """Standalone 'ETH' still flags is_eth: BTC-dom fires and momentum uses ETH."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "Will ETH hit $5000?", "crypto/finance", caplog
+    )
+    assert btc_dom_lo != 0.0
+    # Momentum picks the ETH branch only when is_eth is True.
+    full_log = "\n".join(r.getMessage() for r in caplog.records)
+    assert "ETH 24h: +8.0%" in full_log
+
+def test_dollar_eth_market_detected(caplog):
+    """'$ETH' format still flags is_eth."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "$ETH above $5000?", "crypto/finance", caplog
+    )
+    assert btc_dom_lo != 0.0
+    full_log = "\n".join(r.getMessage() for r in caplog.records)
+    assert "ETH 24h: +8.0%" in full_log
+
+def test_sol_market_detected(caplog):
+    """'SOL reach $200' still flags is_sol → is_altcoin → BTC-dom signal."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "Will SOL reach $200?", "crypto/finance", caplog
+    )
+    # 'reach' → is_price_above, dominance 60 → -0.03 contribution → -0.06 log-odds
+    assert btc_dom_lo == pytest.approx(-0.06, abs=1e-4)
+
+def test_ada_market_detected(caplog):
+    """'ADA reach $1' still flags is_altcoin."""
+    btc_dom_lo = _evaluate_and_parse_btc_dom(
+        "Will ADA reach $1?", "crypto/finance", caplog
+    )
+    assert btc_dom_lo == pytest.approx(-0.06, abs=1e-4)

tests/test_bayesian_btc_dom_gate.py

@@ -0,0 +1,130 @@
+"""
+Tests for FASE 5 — BTC-dominance signal must not apply to non-price markets.
+
+FASE 3 gated momentum and Fear & Greed behind is_non_price (politics / tech /
+events); FASE 4 fixed ticker detection so non-crypto questions no longer flag
+crypto assets by accident.  But a non-price market that LEGITIMATELY mentions
+a ticker ("Will the ETH ETF be approved?") still armed the BTC-dominance
+signal, which has no demonstrated causality for non-price outcomes.  FASE 5
+applies the same is_non_price gate to that signal.
+
+Note: the dominance signal only fires for is_eth / is_altcoin /
+is_general_crypto markets — a pure-BTC question never receives it, so the
+pro-Bitcoin test below is a regression guard rather than a gate exercise;
+the ETH-ETF test is the one that fails without the gate.
+
+Same caplog technique as test_bayesian_asset_detection.py: btc_dom_lo is
+parsed from the structured audit log, with btc_dominance=65 (>55) so the
+signal fires whenever it is allowed to.
+"""
+import asyncio
+import logging
+import re
+
+import pytest
+
+from bot.data.external import ExternalSignals
+from bot.data.polymarket import Market
+from bot.strategy.bayesian import BayesianStrategy
+
+BTC_DOM_RE = re.compile(r"btc_dom_lo=([+-]\d+\.\d+)")
+
+
+def _make_market(question: str, category: str) -> Market:
+    return Market(
+        id="mkt-test-1",
+        condition_id="cond-test-1",
+        question=question,
+        yes_token_id="yes-tok",
+        no_token_id="no-tok",
+        yes_price=0.50,
+        no_price=0.50,
+        volume_24h=50_000.0,
+        end_date="2026-07-15T00:00:00Z",
+        active=True,
+        category=category,
+    )
+
+
+def _make_signals() -> ExternalSignals:
+    # btc_dominance=65 (>55) arms the dominance signal wherever it is allowed.
+    # Momentum kept below the 2% threshold so price-market tests isolate the
+    # dominance contribution.
+    return ExternalSignals(
+        btc_price=100_000.0,
+        btc_change_24h=1.0,
+        eth_price=4_000.0,
+        eth_change_24h=1.0,
+        btc_dominance=65.0,
+        fear_greed_index=50,
+        fear_greed_label="neutral",
+        total_market_cap_change=1.0,
+        valid=True,
+    )
+
+
+def _evaluate(question: str, category: str, caplog) -> tuple[float, str]:
+    """Run evaluate() and return (btc_dom_lo, full_log) from the audit log."""
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    market = _make_market(question, category)
+    with caplog.at_level(logging.INFO, logger="bot.strategy.bayesian"):
+        asyncio.run(
+            strategy.evaluate(market, _make_signals(), occupied_families=set())
+        )
+    full_log = "\n".join(r.getMessage() for r in caplog.records)
+    for record in caplog.records:
+        m = BTC_DOM_RE.search(record.getMessage())
+        if m:
+            return float(m.group(1)), full_log
+    pytest.fail(
+        "No SKIP_EDGE_NET/TRADE log line with btc_dom_lo found; "
+        f"got: {[r.getMessage() for r in caplog.records]}"
+    )
+
+
+# ── Non-price markets: gate must zero the signal ─────────────────────────────
+
+def test_politics_market_mentioning_eth_gets_no_btc_dom(caplog):
+    """Legitimate ETH mention in a politics market → btc_dom_lo == 0.0."""
+    btc_dom_lo, full_log = _evaluate(
+        "Will the ETH ETF be approved?", "politics", caplog
+    )
+    assert btc_dom_lo == 0.0
+    assert "BTC dom" not in full_log
+
+def test_politics_market_mentioning_bitcoin_gets_no_btc_dom(caplog):
+    """Legitimate Bitcoin mention in a politics market → btc_dom_lo == 0.0."""
+    btc_dom_lo, full_log = _evaluate(
+        "Will a pro-Bitcoin candidate win the election?", "politics", caplog
+    )
+    assert btc_dom_lo == 0.0
+    assert "BTC dom" not in full_log
+
+def test_tech_and_events_markets_get_no_btc_dom(caplog):
+    for category in ("tech", "events"):
+        caplog.clear()
+        btc_dom_lo, full_log = _evaluate(
+            "Will the ETH foundation launch the product?", category, caplog
+        )
+        assert btc_dom_lo == 0.0, f"BTC dominance applied to {category} market"
+        assert "BTC dom" not in full_log
+
+
+# ── Price markets: current behavior preserved ───────────────────────────────
+
+def test_eth_price_market_keeps_btc_dom(caplog):
+    """ETH price market with dominance 65 → signal fires as before."""
+    btc_dom_lo, full_log = _evaluate(
+        "Will ETH be above $5000?", "crypto/finance", caplog
+    )
+    # 'above' → is_price_above, dominance 65 > 55 → -0.03 → -0.06 log-odds
+    assert btc_dom_lo == pytest.approx(-0.06, abs=1e-4)
+    assert "BTC dom: 65.0% (high → alt pressure)" in full_log
+
+def test_altcoin_price_market_keeps_btc_dom(caplog):
+    """SOL price market with dominance 65 → signal fires as before."""
+    btc_dom_lo, full_log = _evaluate(
+        "Will SOL reach $200?", "crypto/finance", caplog
+    )
+    assert btc_dom_lo == pytest.approx(-0.06, abs=1e-4)
+    assert "BTC dom: 65.0% (high → alt pressure)" in full_log

tests/test_bayesian_macro_signals.py

@@ -0,0 +1,123 @@
+"""
+Tests for FASE 3 — macro signals (momentum, Fear & Greed) must not apply to
+non-price markets (politics / tech / events).
+
+Regression: for "Will X win the election?"-style questions, is_price_above is
+False, so positive BTC momentum and high Fear & Greed were sign-flipped into
+evidence AGAINST the YES outcome. The fix skips both signals entirely for
+politics/tech/events, leaving their contributions (and feat_mom_lo /
+feat_fg_lo) at 0.0.
+
+evaluate_market only returns a TradingSignal on the TRADE path; on skips it
+returns None but always emits a structured log line containing the per-feature
+log-odds (fg_lo=… mom_lo=…). The tests parse that line via caplog.
+"""
+import asyncio
+import logging
+import math
+import re
+
+import pytest
+
+from bot.data.external import ExternalSignals
+from bot.data.polymarket import Market
+from bot.strategy.bayesian import BayesianStrategy
+
+FEAT_RE = re.compile(r"fg_lo=([+-]\d+\.\d+) mom_lo=([+-]\d+\.\d+)")
+
+
+def _make_market(question: str, category: str) -> Market:
+    return Market(
+        id="mkt-test-1",
+        condition_id="cond-test-1",
+        question=question,
+        yes_token_id="yes-tok",
+        no_token_id="no-tok",
+        yes_price=0.50,
+        no_price=0.50,
+        volume_24h=50_000.0,
+        end_date="2026-07-15T00:00:00Z",
+        active=True,
+        category=category,
+    )
+
+
+def _make_signals() -> ExternalSignals:
+    # Strong bullish macro environment: BTC +10%, extreme greed.
+    return ExternalSignals(
+        btc_price=100_000.0,
+        btc_change_24h=10.0,
+        eth_price=4_000.0,
+        eth_change_24h=8.0,
+        btc_dominance=50.0,
+        fear_greed_index=80,
+        fear_greed_label="greed",
+        total_market_cap_change=5.0,
+        valid=True,
+    )
+
+
+def _evaluate_and_parse_feats(question: str, category: str, caplog) -> tuple[float, float]:
+    """Run BayesianStrategy.evaluate and return (feat_fg_lo, feat_mom_lo) from the audit log."""
+    strategy = BayesianStrategy(news=None, manifold=None, db=None)
+    market = _make_market(question, category)
+    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 = FEAT_RE.search(record.getMessage())
+        if m:
+            return float(m.group(1)), float(m.group(2))
+    pytest.fail(
+        "No SKIP_EDGE_NET/TRADE log line with feature contributions found; "
+        f"got: {[r.getMessage() for r in caplog.records]}"
+    )
+
+
+def test_politics_market_ignores_momentum_and_fear_greed(caplog):
+    """Political market with BTC +10% and F&G=80 → both contributions 0.0."""
+    feat_fg_lo, feat_mom_lo = _evaluate_and_parse_feats(
+        "Will John Smith win the election?", "politics", caplog
+    )
+    assert feat_mom_lo == 0.0
+    assert feat_fg_lo == 0.0
+    # The signal sources must not mention momentum or Fear & Greed either.
+    full_log = "\n".join(r.getMessage() for r in caplog.records)
+    assert "Fear&Greed" not in full_log
+    assert "24h" not in full_log
+
+
+def test_tech_and_events_markets_ignore_macro_signals(caplog):
+    for category in ("tech", "events"):
+        caplog.clear()
+        feat_fg_lo, feat_mom_lo = _evaluate_and_parse_feats(
+            "Will the product launch happen this quarter?", category, caplog
+        )
+        assert feat_mom_lo == 0.0, f"momentum applied to {category} market"
+        assert feat_fg_lo == 0.0, f"Fear&Greed applied to {category} market"
+
+
+def test_btc_market_keeps_momentum_and_fear_greed(caplog):
+    """BTC price market with BTC +10% and F&G=80 → current behavior preserved."""
+    feat_fg_lo, feat_mom_lo = _evaluate_and_parse_feats(
+        "Will Bitcoin be above $150,000 on July 1?", "crypto/finance", caplog
+    )
+    assert feat_mom_lo > 0
+    assert feat_fg_lo > 0
+    # Exact values: is_price_above=True ("above"), so contributions are positive.
+    # momentum: tanh(10/20) * 0.15, ×2 → log-odds.  F&G>70: +0.06, ×2 → log-odds.
+    assert feat_mom_lo == pytest.approx(math.tanh(10 / 20) * 0.15 * 2, abs=1e-4)
+    assert feat_fg_lo == pytest.approx(0.06 * 2, abs=1e-4)
+    full_log = "\n".join(r.getMessage() for r in caplog.records)
+    assert "Fear&Greed: 80 (greed)" in full_log
+    assert "BTC 24h: +10.0%" in full_log
+
+
+def test_btc_below_market_sign_flip_preserved(caplog):
+    """'below' market: bullish macro lowers YES probability (sign flip intact)."""
+    feat_fg_lo, feat_mom_lo = _evaluate_and_parse_feats(
+        "Will Bitcoin drop below $50,000 by August?", "crypto/finance", caplog
+    )
+    assert feat_mom_lo < 0
+    assert feat_fg_lo < 0

tests/test_manifold_outcome.py

@@ -0,0 +1,152 @@
+"""
+Tests for the Manifold outcome-compatibility guard.
+
+Regression: a Polymarket *nomination* question must not match a Manifold
+*conditional* question ("If X is the nominee, will he win?") even at Jaccard=1.0.
+"""
+import asyncio
+
+import pytest
+
+from bot.data.manifold import (
+    ManifoldClient,
+    _classify_outcome,
+    _is_conditional,
+)
+
+
+# ── _is_conditional ────────────────────────────────────────────────────────────
+def test_is_conditional_prefixes():
+    assert _is_conditional("If Graham Platner is the nominee, will he win?")
+    assert _is_conditional("Conditional on a recession, will rates fall?")
+    assert _is_conditional("Assuming Trump runs, will he win?")
+    assert _is_conditional("Given that X happens, will Y?")
+
+
+def test_is_conditional_midsentence_clause():
+    assert _is_conditional("Will Biden, if he is nominated, win the election?")
+
+
+def test_is_not_conditional():
+    assert not _is_conditional("Will Graham Platner be the Democratic nominee?")
+    assert not _is_conditional("Will the GOP win the Senate?")
+    # "if" without a closing comma clause is not flagged
+    assert not _is_conditional("What happens if everything goes right")
+
+
+# ── _classify_outcome ───────────────────────────────────────────────────────────
+def test_classify_nomination():
+    assert _classify_outcome("Will X be the Democratic nominee for Senate?") == "nomination"
+    assert _classify_outcome("Will X be nominated?") == "nomination"
+    # "primary nominee" → nomination (checked before primary)
+    assert _classify_outcome("Will X be the primary nominee?") == "nomination"
+
+
+def test_classify_primary_win():
+    assert _classify_outcome("Will X win the primary?") == "primary_win"
+    assert _classify_outcome("Will X advance in the first round?") == "primary_win"
+
+
+def test_classify_general_win():
+    assert _classify_outcome("Will X win the election?") == "general_win"
+    assert _classify_outcome("Will X win the seat?") == "general_win"
+    assert _classify_outcome("Will X win the general election?") == "general_win"
+
+
+def test_classify_conditional():
+    assert _classify_outcome("If X is the nominee, will he win?") == "conditional"
+    assert _classify_outcome("Assuming a runoff, who wins?") == "conditional"
+
+
+def test_classify_other():
+    assert _classify_outcome("Will it rain tomorrow?") == "other"
+
+
+# ── End-to-end get_match with a stubbed Manifold API ────────────────────────────
+class _StubResponse:
+    def __init__(self, payload):
+        self._payload = payload
+
+    def raise_for_status(self):
+        pass
+
+    def json(self):
+        return self._payload
+
+
+class _StubHTTP:
+    def __init__(self, payload):
+        self._payload = payload
+
+    async def get(self, *args, **kwargs):
+        return _StubResponse(self._payload)
+
+    async def aclose(self):
+        pass
+
+
+async def _match(poly, mfld_market):
+    client = ManifoldClient()
+    client._client = _StubHTTP([mfld_market])
+    try:
+        return await client.get_match(poly)
+    finally:
+        await client.close()
+
+
+def test_graham_platner_conditional_rejected():
+    """Poly nomination vs Manifold conditional → rejected (Task 4.1)."""
+    poly = ("Will Graham Platner be the Democratic nominee for Senate "
+            "in Maine in 2026?")
+    mfld_market = {
+        "outcomeType": "BINARY",
+        "probability": 0.55,
+        "question": ("If Graham Platner is the Democratic nominee for Senate "
+                     "in Maine, will he win the general election?"),
+        "id": "abc123",
+        "slug": "graham-platner-win",
+        "creatorUsername": "someone",
+    }
+    result = asyncio.run(_match(poly, mfld_market))
+
+    assert result.status == "rejected"
+    assert result.match_reason is not None
+    assert ("conditional" in result.match_reason
+            or "outcome_mismatch" in result.match_reason)
+    # outcome types are classified and available for persistence
+    assert result.poly_outcome_type == "nomination"
+    assert result.mfld_outcome_type == "conditional"
+
+
+def test_outcome_mismatch_nomination_vs_general_rejected():
+    """Poly nomination vs Manifold general_win (non-conditional) → rejected."""
+    poly = "Will Jane Doe be the Republican nominee for Governor?"
+    mfld_market = {
+        "outcomeType": "BINARY",
+        "probability": 0.4,
+        "question": "Will Jane Doe win the election for Governor?",
+        "id": "x", "slug": "jane-doe", "creatorUsername": "u",
+    }
+    result = asyncio.run(_match(poly, mfld_market))
+
+    assert result.status == "rejected"
+    assert "outcome_mismatch" in result.match_reason
+    assert result.poly_outcome_type == "nomination"
+    assert result.mfld_outcome_type == "general_win"
+
+
+def test_matching_nomination_accepted():
+    """Poly nomination vs Manifold nomination (same outcome) → accepted."""
+    poly = "Will Graham Platner be the Democratic nominee for Senate in Maine?"
+    mfld_market = {
+        "outcomeType": "BINARY",
+        "probability": 0.62,
+        "question": "Will Graham Platner be the Democratic Senate nominee in Maine?",
+        "id": "ok", "slug": "platner-nominee", "creatorUsername": "u",
+    }
+    result = asyncio.run(_match(poly, mfld_market))
+
+    assert result.status == "accepted"
+    assert result.poly_outcome_type == "nomination"
+    assert result.mfld_outcome_type == "nomination"
+    assert result.prob_final == pytest.approx(0.62)

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_news_query.py

@@ -0,0 +1,77 @@
+"""Tests for the GNews layer minor fixes.
+
+Two faults found during the GNews capture/prioritisation diagnostic:
+
+1. Hyphens/dashes in a market question reached the GNews query verbatim and,
+   because '-' is GNews's exclusion operator, produced HTTP 400
+   (e.g. "Abdul El-Sayed Michigan Democratic Primary").
+
+2. The per-cycle GNews budget counter incremented in evaluate() *before*
+   get_sentiment() checked the API key, so with no key configured the
+   [CYCLE SUMMARY] reported a phantom "gnews_queries_used: 5/5" even though
+   zero real requests left the process.
+"""
+import asyncio
+
+from bot.data.news import NewsClient
+from bot.data.external import ExternalSignals
+from bot.data.polymarket import Market
+from bot.strategy.bayesian import BayesianStrategy
+
+
+# ── Fix 1: query sanitisation ────────────────────────────────────────────────
+
+def test_build_query_strips_hyphen_that_breaks_gnews():
+    q = NewsClient._build_query(
+        "Will Abdul El-Sayed win the 2026 Michigan Democratic Primary?"
+    )
+    assert "-" not in q              # the exclusion operator must be gone
+    assert "El-Sayed" not in q
+    assert "Sayed" in q             # the meaningful token survives as its own word
+
+
+def test_build_query_strips_unicode_dashes():
+    q = NewsClient._build_query("Trump–Putin summit — final outcome")
+    assert "–" not in q and "—" not in q
+    assert "Trump" in q and "Putin" in q
+
+
+# ── Fix 2: enabled property + budget accounting ──────────────────────────────
+
+def test_enabled_reflects_api_key(monkeypatch):
+    monkeypatch.delenv("GNEWS_API_KEY", raising=False)
+    assert NewsClient().enabled is False
+    monkeypatch.setenv("GNEWS_API_KEY", "deadbeefdeadbeefdeadbeefdeadbeef")
+    assert NewsClient().enabled is True
+
+
+def _politics_market() -> Market:
+    return Market(
+        id="m1", condition_id="c1",
+        question="Will candidate X win the 2026 governor election?",
+        yes_token_id="y", no_token_id="n",
+        yes_price=0.50, no_price=0.50, volume_24h=10_000.0,
+        end_date="2026-07-15T00:00:00Z", active=True, category="politics",
+    )
+
+
+def _signals() -> ExternalSignals:
+    return ExternalSignals(
+        btc_price=1.0, btc_change_24h=0.0, eth_price=1.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 test_disabled_news_consumes_no_gnews_budget(monkeypatch):
+    """Regression: no API key → gnews_queries_used stays 0 (was a phantom 1+)."""
+    monkeypatch.delenv("GNEWS_API_KEY", raising=False)
+    news = NewsClient()
+    assert news.enabled is False
+
+    strategy = BayesianStrategy(news=news, manifold=None, db=None)
+    strategy.reset_cycle()
+    asyncio.run(
+        strategy.evaluate(_politics_market(), _signals(), occupied_families=set())
+    )
+    assert strategy.get_cycle_stats()["gnews_queries_used"] == 0

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_paper_close.py

@@ -0,0 +1,130 @@
+"""
+Tests for PaperExecutor.close_position() settlement payout.
+
+Regression: the old code computed cash += position_cost * resolution, which
+ignores direction — a winning BUY_NO (resolution = 0.0) paid out $0.
+
+Correct settlement:
+  BUY_YES: payout = shares * resolution
+  BUY_NO:  payout = shares * (1 - resolution)
+  pnl = payout - net_cost
+"""
+import asyncio
+
+import pytest
+
+from bot.executor import paper
+from bot.executor.paper import PaperExecutor
+
+
+class FakeDB:
+    """Minimal Database stub for close_position()."""
+
+    def __init__(self, trades_by_market: dict[str, list[dict]]):
+        self._trades = trades_by_market
+        self.closed: list[tuple] = []
+
+    async def get_open_trades_for_market(self, market_id: str) -> list[dict]:
+        return self._trades.get(market_id, [])
+
+    async def close_paper_position(self, market_id, reason="", resolution=None):
+        self.closed.append((market_id, reason, resolution))
+
+
+def _close(direction: str, resolution: float):
+    """Open one paper trade (size $100 @ 0.5 → 200 shares, net_cost $102)
+    and settle it at `resolution`. Returns (pnl, executor, notifications)."""
+    notifications: list[tuple] = []
+
+    async def fake_trade_closed(question, pnl):
+        notifications.append((question, pnl))
+
+    async def run():
+        db = FakeDB({
+            "mkt1": [{"direction": direction, "shares": 200.0, "net_cost": 102.0}],
+        })
+        ex = PaperExecutor(db=db, bankroll=1000.0)
+        ex._portfolio.cash = 898.0           # 1000 - net_cost spent at entry
+        ex._portfolio.positions["mkt1"] = 100.0  # size_usdc, as execute() stores it
+
+        original = paper.telegram.trade_closed
+        paper.telegram.trade_closed = fake_trade_closed
+        try:
+            pnl = await ex.close_position("mkt1", resolution, question="Test market?")
+            await asyncio.sleep(0)  # let the notification task run
+        finally:
+            paper.telegram.trade_closed = original
+        return pnl, ex, db
+
+    pnl, ex, db = asyncio.run(run())
+    return pnl, ex, db, notifications
+
+
+def test_buy_yes_wins():
+    pnl, ex, db, notif = _close("BUY_YES", resolution=1.0)
+    assert pnl == pytest.approx(200.0 - 102.0)   # payout = 200 * 1.0
+    assert pnl > 0
+    assert ex._portfolio.cash == pytest.approx(898.0 + 200.0)
+    assert notif[0][1] > 0                        # Telegram reports a win
+
+
+def test_buy_yes_loses():
+    pnl, ex, db, notif = _close("BUY_YES", resolution=0.0)
+    assert pnl == pytest.approx(-102.0)           # payout = 0
+    assert pnl < 0
+    assert ex._portfolio.cash == pytest.approx(898.0)
+    assert notif[0][1] < 0                        # Telegram reports a loss
+
+
+def test_buy_no_wins():
+    pnl, ex, db, notif = _close("BUY_NO", resolution=0.0)
+    assert pnl == pytest.approx(200.0 - 102.0)   # payout = 200 * (1 - 0.0)
+    assert pnl > 0
+    assert ex._portfolio.cash == pytest.approx(898.0 + 200.0)
+    assert notif[0][1] > 0                        # win despite resolution = 0.0
+
+
+def test_buy_no_loses():
+    pnl, ex, db, notif = _close("BUY_NO", resolution=1.0)
+    assert pnl == pytest.approx(-102.0)           # payout = 200 * (1 - 1.0) = 0
+    assert pnl < 0
+    assert ex._portfolio.cash == pytest.approx(898.0)
+    assert notif[0][1] < 0                        # loss despite resolution = 1.0
+
+
+def test_position_is_removed_and_persisted():
+    pnl, ex, db, notif = _close("BUY_YES", resolution=1.0)
+    assert "mkt1" not in ex._portfolio.positions
+    assert db.closed == [("mkt1", "resolved", 1.0)]
+
+
+def test_unknown_market_returns_none():
+    async def run():
+        ex = PaperExecutor(db=FakeDB({}), bankroll=1000.0)
+        return await ex.close_position("nope", 1.0)
+    assert asyncio.run(run()) is None
+
+
+def test_db_failure_keeps_position_for_retry():
+    """Regression: a DB error during close must not mutate the in-memory
+    portfolio — otherwise the next resolution check skips the market
+    (not in positions) and the DB row stays open forever."""
+
+    class FailingDB(FakeDB):
+        async def close_paper_position(self, market_id, reason="", resolution=None):
+            raise RuntimeError("db down")
+
+    async def run():
+        db = FailingDB({
+            "mkt1": [{"direction": "BUY_YES", "shares": 200.0, "net_cost": 102.0}],
+        })
+        ex = PaperExecutor(db=db, bankroll=1000.0)
+        ex._portfolio.cash = 898.0
+        ex._portfolio.positions["mkt1"] = 100.0
+        with pytest.raises(RuntimeError):
+            await ex.close_position("mkt1", 1.0)
+        return ex
+
+    ex = asyncio.run(run())
+    assert ex._portfolio.positions == {"mkt1": 100.0}   # still open in memory
+    assert ex._portfolio.cash == pytest.approx(898.0)   # payout not credited

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_resolution_detector.py

@@ -0,0 +1,219 @@
+"""
+Tests for the automatic market-resolution detector (Phase 2).
+
+Covers:
+  - PolymarketClient.get_market_resolution() parsing of real Gamma API shapes
+    (resolved YES/NO, still open, UMA-disputed, ambiguous prices, 404, errors).
+  - check_resolutions() in bot/main.py: a resolved market settles the open
+    paper position via PaperExecutor.close_position() and persists
+    close_reason='resolved' with the resolution value.
+"""
+import asyncio
+import json
+
+import httpx
+import pytest
+
+from bot.data.polymarket import PolymarketClient, MarketResolution
+from bot.executor import paper
+from bot.executor.paper import PaperExecutor
+from bot.main import check_resolutions
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# get_market_resolution() — Gamma API response parsing
+# ─────────────────────────────────────────────────────────────────────────────
+
+class FakeResponse:
+    def __init__(self, status_code: int, payload: dict | None = None):
+        self.status_code = status_code
+        self._payload = payload or {}
+
+    def json(self):
+        return self._payload
+
+    def raise_for_status(self):
+        if self.status_code >= 400:
+            raise httpx.HTTPStatusError(
+                f"HTTP {self.status_code}", request=None, response=None
+            )
+
+
+class FakeHTTPClient:
+    def __init__(self, response):
+        self._response = response
+        self.requested_urls: list[str] = []
+
+    async def get(self, url, **kwargs):
+        self.requested_urls.append(url)
+        if isinstance(self._response, Exception):
+            raise self._response
+        return self._response
+
+
+def _resolution_for(response) -> MarketResolution | None:
+    client = PolymarketClient()
+    client._client = FakeHTTPClient(response)
+    return asyncio.run(client.get_market_resolution("12345"))
+
+
+def _gamma_market(closed: bool, yes_price: str, no_price: str,
+                  uma_status: str | None = "resolved") -> dict:
+    """Mirror the real Gamma /markets/{id} payload shape (observed 2026-06-11)."""
+    m = {
+        "id": "12345",
+        "question": "Test market?",
+        "closed": closed,
+        "active": True,
+        "outcomePrices": json.dumps([yes_price, no_price]),
+        "closedTime": "2026-06-11 13:15:01+00" if closed else None,
+        "umaEndDate": "2026-06-11T13:15:01Z" if closed else None,
+        "endDate": "2026-06-11T13:00:00Z",
+    }
+    if uma_status is not None:
+        m["umaResolutionStatus"] = uma_status
+    return m
+
+
+def test_resolution_no_won():
+    res = _resolution_for(FakeResponse(200, _gamma_market(True, "0", "1")))
+    assert res.resolved is True
+    assert res.resolution == 0.0
+    assert res.resolved_at is not None
+
+
+def test_resolution_yes_won():
+    res = _resolution_for(FakeResponse(200, _gamma_market(True, "1", "0")))
+    assert res.resolved is True
+    assert res.resolution == 1.0
+
+
+def test_open_market_not_resolved():
+    res = _resolution_for(FakeResponse(
+        200, _gamma_market(False, "0.51", "0.49", uma_status=None)
+    ))
+    assert res.resolved is False
+    assert res.resolution is None
+
+
+def test_closed_but_uma_disputed_not_settled():
+    res = _resolution_for(FakeResponse(
+        200, _gamma_market(True, "0", "1", uma_status="disputed")
+    ))
+    assert res.resolved is False
+
+
+def test_closed_with_ambiguous_prices_not_settled():
+    res = _resolution_for(FakeResponse(200, _gamma_market(True, "0.6", "0.4")))
+    assert res.resolved is False
+
+
+def test_market_not_found_returns_none():
+    assert _resolution_for(FakeResponse(404)) is None
+
+
+def test_api_error_returns_none():
+    assert _resolution_for(httpx.ConnectError("boom")) is None
+
+
+# ─────────────────────────────────────────────────────────────────────────────
+# check_resolutions() — detector loop settles paper positions
+# ─────────────────────────────────────────────────────────────────────────────
+
+class FakeDB:
+    """Database stub: one open BUY_NO paper position."""
+
+    def __init__(self, trades_by_market: dict[str, list[dict]]):
+        self._trades = trades_by_market
+        self.closed: list[tuple] = []
+
+    async def get_open_position_details(self) -> list[dict]:
+        return [
+            {"market_id": mid, "question": t[0].get("question", ""),
+             "direction": t[0]["direction"]}
+            for mid, t in self._trades.items()
+        ]
+
+    async def get_open_trades_for_market(self, market_id: str) -> list[dict]:
+        return self._trades.get(market_id, [])
+
+    async def close_paper_position(self, market_id, reason="", resolution=None):
+        self.closed.append((market_id, reason, resolution))
+
+
+class FakePoly:
+    def __init__(self, resolutions: dict[str, MarketResolution | None]):
+        self._resolutions = resolutions
+        self.checked: list[str] = []
+
+    async def get_market_resolution(self, market_id: str):
+        self.checked.append(market_id)
+        return self._resolutions.get(market_id)
+
+
+def _run_check(resolutions: dict, trades: dict):
+    notifications: list[tuple] = []
+
+    async def fake_trade_closed(question, pnl):
+        notifications.append((question, pnl))
+
+    async def run():
+        db = FakeDB(trades)
+        ex = PaperExecutor(db=db, bankroll=1000.0)
+        for mid, t in trades.items():
+            ex._portfolio.positions[mid] = sum(x["net_cost"] for x in t) - 2.0
+        ex._portfolio.cash = 898.0
+        poly = FakePoly(resolutions)
+
+        original = paper.telegram.trade_closed
+        paper.telegram.trade_closed = fake_trade_closed
+        try:
+            await check_resolutions(poly, ex, db)
+            await asyncio.sleep(0)  # let notification task run
+        finally:
+            paper.telegram.trade_closed = original
+        return db, ex, poly
+
+    db, ex, poly = asyncio.run(run())
+    return db, ex, poly, notifications
+
+
+BUY_NO_TRADE = {
+    "mkt1": [{
+        "direction": "BUY_NO", "shares": 200.0, "net_cost": 102.0,
+        "question": "Will X happen?",
+    }],
+}
+
+
+def test_resolved_buy_no_position_is_closed():
+    """BUY_NO position + market resolved NO (resolution=0.0) → winning close."""
+    db, ex, poly, notif = _run_check(
+        {"mkt1": MarketResolution(resolved=True, resolution=0.0)},
+        BUY_NO_TRADE,
+    )
+    assert poly.checked == ["mkt1"]
+    # close_paper_position called with close_reason='resolved' and the resolution
+    assert db.closed == [("mkt1", "resolved", 0.0)]
+    # Position removed and payout credited: 200 shares * (1 - 0.0) = $200
+    assert "mkt1" not in ex._portfolio.positions
+    assert ex._portfolio.cash == pytest.approx(898.0 + 200.0)
+    # Telegram notified with positive pnl (200 - 102)
+    assert notif == [("Will X happen?", pytest.approx(98.0))]
+
+
+def test_unresolved_position_stays_open():
+    db, ex, poly, notif = _run_check(
+        {"mkt1": MarketResolution(resolved=False)},
+        BUY_NO_TRADE,
+    )
+    assert poly.checked == ["mkt1"]
+    assert db.closed == []
+    assert "mkt1" in ex._portfolio.positions
+    assert notif == []
+
+
+def test_api_failure_leaves_position_open():
+    db, ex, poly, notif = _run_check({"mkt1": None}, BUY_NO_TRADE)
+    assert db.closed == []
+    assert "mkt1" in ex._portfolio.positions

tests/test_sharpe_gate.py

@@ -0,0 +1,242 @@
+"""
+Tests for the real Sharpe ratio with minimum-sample gate.
+
+Regression: sharpe_ratio was hardcoded to 0.0 in MetricsTracker and exposed
+as `latest.get("sharpe_ratio") or 0` in /api/summary, and promotion_ready
+could in principle flip on a statistically meaningless sample (e.g. 1
+resolved trade over ~40 days of flat PnL plus a single +299 jump).
+
+Fix: bot/metrics/sharpe.py computes an annualized Sharpe from the daily
+total_pnl close series, gated to None ("insufficient_sample") below 30 days
+observed / 10 resolved trades. /api/summary exposes the value plus an
+explanation (sharpe_status, days_observed, min_* fields), and
+promotion_ready additionally requires the sample minimums and non-null
+metrics.
+"""
+import asyncio
+from statistics import mean, stdev
+
+import pytest
+
+import api.main as api_main
+from bot.metrics.sharpe import (
+    MIN_DAYS_OBSERVED,
+    MIN_RESOLVED_TRADES,
+    SHARPE_INSUFFICIENT,
+    SHARPE_OK,
+    SHARPE_ZERO_VARIANCE,
+    compute_sharpe,
+    daily_returns,
+    sharpe_with_gate,
+)
+from bot.metrics.tracker import MetricsTracker
+
+
+BANKROLL = 10_000.0
+
+
+def _closes_from_deltas(deltas: list[float], start: float = 0.0) -> list[float]:
+    closes = [start]
+    for d in deltas:
+        closes.append(closes[-1] + d)
+    return closes
+
+
+# ── Pure computation ─────────────────────────────────────────────────────────
+
+def test_daily_returns_are_bankroll_normalized_deltas():
+    closes = [0.0, 100.0, 50.0, 50.0]
+    assert daily_returns(closes, BANKROLL) == pytest.approx([0.01, -0.005, 0.0])
+
+
+def test_compute_sharpe_matches_manual_formula():
+    deltas = [10.0, 14.0, 8.0, 12.0, 6.0, 13.0, 9.0]
+    closes = _closes_from_deltas(deltas)
+    rets = [d / BANKROLL for d in deltas]
+    expected = mean(rets) / stdev(rets) * 365 ** 0.5
+    assert compute_sharpe(closes, BANKROLL) == pytest.approx(expected)
+    assert compute_sharpe(closes, BANKROLL) > 0
+
+
+def test_compute_sharpe_undefined_cases_return_none():
+    assert compute_sharpe([], BANKROLL) is None
+    assert compute_sharpe([0.0], BANKROLL) is None
+    assert compute_sharpe([0.0, 50.0], BANKROLL) is None      # only 1 return
+    assert compute_sharpe([0.0] * 40, BANKROLL) is None        # zero variance
+
+
+# ── Minimum-sample gate ───────────────────────────────────────────────────────
+
+def test_gate_blocks_current_situation_one_resolved_trade():
+    """~40 flat days plus a single +299 jump, 1 resolved trade → no Sharpe."""
+    closes = [0.0] * 35 + [299.06] * 5
+    sharpe, status = sharpe_with_gate(closes, BANKROLL, resolved_count=1)
+    assert sharpe is None
+    assert status == SHARPE_INSUFFICIENT
+    # The raw (ungated) value would exist and be wildly misleading:
+    assert compute_sharpe(closes, BANKROLL) is not None
+
+
+def test_gate_blocks_too_few_days_even_with_enough_resolved():
+    closes = _closes_from_deltas([10.0, -5.0] * 10)  # 21 days < 30
+    sharpe, status = sharpe_with_gate(closes, BANKROLL, resolved_count=15)
+    assert sharpe is None
+    assert status == SHARPE_INSUFFICIENT
+
+
+def test_gate_passes_with_sufficient_sample():
+    deltas = [10.0, 14.0, 8.0, 12.0, 6.0] * 8  # 40 returns → 41 days
+    closes = _closes_from_deltas(deltas)
+    sharpe, status = sharpe_with_gate(closes, BANKROLL, resolved_count=MIN_RESOLVED_TRADES)
+    assert status == SHARPE_OK
+    assert sharpe == pytest.approx(compute_sharpe(closes, BANKROLL))
+
+
+def test_gate_flat_curve_with_sufficient_sample_is_zero_variance():
+    sharpe, status = sharpe_with_gate([0.0] * 40, BANKROLL, resolved_count=12)
+    assert sharpe is None
+    assert status == SHARPE_ZERO_VARIANCE
+
+
+# ── /api/summary ─────────────────────────────────────────────────────────────
+
+class FakeDB:
+    def __init__(self, daily_closes, resolved_count, total_trades=60,
+                 win_rate=0.6, calibration=0.8):
+        self._closes = daily_closes
+        self._resolved = resolved_count
+        self._total = total_trades
+        self._win_rate = win_rate
+        self._calibration = calibration
+
+    async def get_metrics_history(self, days=1):
+        return [{
+            "win_rate": self._win_rate,
+            "calibration_score": self._calibration,
+            "unrealized_pnl_est": 0.0,
+            "realized_pnl": 299.06,
+            "total_pnl": 299.06,
+        }]
+
+    async def compute_metrics_from_db(self):
+        return {
+            "total_trades": self._total,
+            "open_count": self._total - self._resolved,
+            "closed_count": self._resolved,
+            "resolved_count": self._resolved,
+        }
+
+    async def get_open_position_data(self):
+        return {}, 0.0
+
+    async def get_recently_closed_inverted(self, hours=24):
+        return set()
+
+    async def get_legacy_incomplete_count(self):
+        return 0
+
+    async def get_daily_pnl_closes(self):
+        return list(self._closes)
+
+
+def _summary(db, monkeypatch) -> dict:
+    monkeypatch.setattr(api_main, "db", db)
+    monkeypatch.delenv("PAPER_BANKROLL", raising=False)
+    return asyncio.run(api_main.get_summary())
+
+
+def test_api_insufficient_sample_returns_null_with_explanation(monkeypatch):
+    """Current prod situation: 1 resolved, ~40 days → null Sharpe, not ready."""
+    db = FakeDB(daily_closes=[0.0] * 35 + [299.06] * 5, resolved_count=1)
+    s = _summary(db, monkeypatch)
+    assert s["sharpe_ratio"] is None
+    assert s["sharpe_status"] == SHARPE_INSUFFICIENT
+    assert s["resolved_count"] == 1
+    assert s["min_resolved_required"] == MIN_RESOLVED_TRADES == 10
+    assert s["days_observed"] == 40
+    assert s["min_days_required"] == MIN_DAYS_OBSERVED == 30
+    # One lucky resolved trade must never promote, even with perfect
+    # win_rate/calibration and 50+ trades.
+    assert s["promotion_ready"] is False
+
+
+def test_api_sharpe_appears_with_sufficient_sample(monkeypatch):
+    deltas = [10.0, 14.0, 8.0, 12.0, 6.0] * 8
+    db = FakeDB(daily_closes=_closes_from_deltas(deltas), resolved_count=12)
+    s = _summary(db, monkeypatch)
+    assert s["sharpe_status"] == SHARPE_OK
+    assert s["sharpe_ratio"] == pytest.approx(
+        compute_sharpe(_closes_from_deltas(deltas), BANKROLL)
+    )
+    assert s["sharpe_ratio"] >= 0.5
+    assert s["promotion_ready"] is True
+
+
+def test_api_not_ready_when_sharpe_below_threshold(monkeypatch):
+    # Zero-drift curve: mean return ~0 → Sharpe ≈ 0 < 0.5
+    deltas = [50.0, -50.0] * 20
+    db = FakeDB(daily_closes=_closes_from_deltas(deltas), resolved_count=12)
+    s = _summary(db, monkeypatch)
+    assert s["sharpe_status"] == SHARPE_OK
+    assert s["sharpe_ratio"] < 0.5
+    assert s["promotion_ready"] is False
+
+
+def test_api_not_ready_when_metrics_null(monkeypatch):
+    db = FakeDB(
+        daily_closes=_closes_from_deltas([10.0, 14.0, 8.0, 12.0, 6.0] * 8),
+        resolved_count=12,
+        win_rate=None,
+        calibration=None,
+    )
+    s = _summary(db, monkeypatch)
+    assert s["sharpe_status"] == SHARPE_OK
+    assert s["promotion_ready"] is False
+
+
+# ── MetricsTracker: no hardcoded 0.0 in the snapshot ─────────────────────────
+
+class FakeTrackerDB:
+    def __init__(self, daily_closes, resolved_count):
+        self._closes = daily_closes
+        self._resolved = resolved_count
+        self.saved = None
+
+    async def compute_metrics_from_db(self):
+        return {
+            "total_trades": 60,
+            "open_count": 40,
+            "closed_count": 20,
+            "resolved_count": self._resolved,
+            "wins_realized": self._resolved,
+            "unrealized_pnl_est": 0.0,
+            "realized_pnl": 100.0,
+            "total_deployed": 1000.0,
+            "total_fees": 20.0,
+            "calibration_score": 0.8,
+        }
+
+    async def get_daily_pnl_closes(self):
+        return list(self._closes)
+
+    async def save_daily_metrics(self, metrics):
+        self.saved = metrics
+
+
+def test_tracker_stores_null_sharpe_below_gate(monkeypatch):
+    monkeypatch.delenv("PAPER_BANKROLL", raising=False)
+    db = FakeTrackerDB(daily_closes=[0.0] * 35 + [299.06] * 5, resolved_count=1)
+    asyncio.run(MetricsTracker(db).update_daily_summary())
+    assert db.saved is not None
+    assert db.saved["sharpe_ratio"] is None
+
+
+def test_tracker_stores_real_sharpe_above_gate(monkeypatch):
+    monkeypatch.delenv("PAPER_BANKROLL", raising=False)
+    closes = _closes_from_deltas([10.0, 14.0, 8.0, 12.0, 6.0] * 8)
+    db = FakeTrackerDB(daily_closes=closes, resolved_count=12)
+    asyncio.run(MetricsTracker(db).update_daily_summary())
+    assert db.saved["sharpe_ratio"] == pytest.approx(
+        compute_sharpe(closes, BANKROLL)
+    )
+    assert db.saved["sharpe_ratio"] != 0.0

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",
+    ]