Compare commits

..
Author SHA1 Message Date
Renovate Bot 391e86c7a6 chore(deps): update python docker tag to v3.14 2026-05-27 12:01:27 +00:00
34 changed files with 275 additions and 5563 deletions
+10 -90
View File
@@ -19,64 +19,15 @@ 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"]
@@ -99,7 +50,6 @@ 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 \
@@ -111,7 +61,6 @@ 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 \
@@ -123,7 +72,6 @@ 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 \
@@ -136,7 +84,6 @@ jobs:
dashboard
- name: Verify images in registry
if: steps.changes.outputs.build_any == 'true'
run: |
TAG=${{ steps.tag.outputs.TAG }}
check_image() {
@@ -151,18 +98,11 @@ 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
check_image polymarket-bot
check_image polymarket-bot-api
check_image polymarket-bot-dashboard
- name: Update k8s manifests
if: steps.changes.outputs.build_any == 'true'
run: |
pip3 install pyyaml -q
@@ -174,21 +114,12 @@ 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/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|image: .*polymarket-bot[^-].*|image: git.chemavx.xyz/chemavx/polymarket-bot:${TAG}|g" \
polymarket-bot/deployment-bot.yaml
sed -i "s|image: .*polymarket-bot-api.*|image: git.chemavx.xyz/chemavx/polymarket-bot-api:${TAG}|g" \
polymarket-bot/deployment-api.yaml
sed -i "s|image: .*polymarket-bot-dashboard.*|image: git.chemavx.xyz/chemavx/polymarket-bot-dashboard:${TAG}|g" \
polymarket-bot/deployment-dashboard.yaml
sed -i "s|imagePullPolicy: Never|imagePullPolicy: Always|g" \
polymarket-bot/deployment-bot.yaml \
polymarket-bot/deployment-api.yaml \
@@ -223,21 +154,10 @@ 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
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
MSG="✅ Deploy polymarket-bot:${TAG} completado"
else
MSG="❌ Deploy polymarket-bot:${TAG} fallido (status: ${JOB_STATUS})"
fi
+1 -1
View File
@@ -1,4 +1,4 @@
FROM python:3.11-slim
FROM python:3.14-slim
WORKDIR /app
+1 -1
View File
@@ -1,4 +1,4 @@
FROM python:3.11-slim
FROM python:3.14-slim
WORKDIR /app
-35
View File
@@ -1,35 +0,0 @@
# polymarket-bot
Bot de paper-trading para Polymarket con estrategia bayesiana, API FastAPI y
dashboard React. Corre en k3s vía GitOps (Gitea Actions → registry → ArgoCD).
## 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: https://polymarket.chemavx.xyz
## CI/CD
`.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
```
+24 -121
View File
@@ -11,12 +11,6 @@ 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"
@@ -215,66 +209,6 @@ 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.
@@ -285,49 +219,28 @@ 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, counts, position_data, inverted, legacy_count, daily_closes = (
await asyncio.gather(
db.get_metrics_history(days=1),
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_metrics, open_trades, all_trades, inverted, legacy_count = await asyncio.gather(
db.get_metrics_history(days=1),
db.get_recent_trades(limit=500, status="open"),
db.get_recent_trades(limit=500),
db.get_recently_closed_inverted(hours=24),
db.get_legacy_incomplete_count(),
)
latest = latest_metrics[0] if latest_metrics else {}
paper_bankroll = float(os.getenv("PAPER_BANKROLL", "10000"))
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")
total_deployed = sum(t.get("net_cost", 0) for t in open_trades)
return {
# ── Portfolio state (live from DB) ──────────────────────────────────
"paper_mode": os.getenv("PAPER_MODE", "true") == "true",
"paper_bankroll": paper_bankroll,
"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_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_deployed": total_deployed, # exact, from DB
"cash_available": cash_available(paper_bankroll, total_net_cost_open),
"cash_available": max(0.0, paper_bankroll - total_deployed), # exact
"legacy_incomplete_count": legacy_count, # exact, from DB
"reentry_guard_blocks_24h": len(inverted), # exact, from DB
@@ -335,41 +248,31 @@ 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 — payout net_cost per trade (net of fee), matches logs/Telegram.
# Exact — computed from (resolution entry_price) × shares.
# 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 ──────────────────────────────────────────────
# ── Performance metrics (from latest metrics_daily snapshot) ─────────
# win_rate: fraction of resolved closed trades where close_pnl > 0.
# null if fewer than 5 resolved trades. Source: closed+resolved trades.
# 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.
# sharpe_ratio: 0.0 — requires daily-return time series (not yet tracked).
# calibration_score: 1 Brier score on resolved trades (higher = better).
# null if fewer than 10 resolved trades. Source: closed+resolved trades.
"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
"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
# ── Counters (live from DB) ──────────────────────────────────────────
"resolved_count": resolved_count,
# ── Counters from snapshot ───────────────────────────────────────────
"resolved_count": latest.get("resolved_count") or 0,
# ── Promotion gate ───────────────────────────────────────────────────
# 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.
# All thresholds must pass; null metrics count as not-ready.
"promotion_ready": (
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
(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
),
}
+16 -621
View File
@@ -4,8 +4,6 @@ import os
from typing import Optional
import asyncpg
from bot.data.manifold import MANIFOLD_MATCHER_VERSION
log = logging.getLogger(__name__)
@@ -38,15 +36,11 @@ 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,
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
feat_fg_lo, feat_mom_lo, feat_news_lo, feat_mfld_lo, feat_btc_dom_lo
) 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,$31,$32,$33,$34
$21,$22,$23,$24,$25
)
ON CONFLICT (id) DO NOTHING
""",
@@ -59,10 +53,6 @@ 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:
@@ -152,20 +142,6 @@ 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:
@@ -173,29 +149,19 @@ 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 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.
is computed in SQL so it matches the stored entry_price and shares exactly.
"""
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::double precision,
resolution = $3,
close_pnl = CASE
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
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
ELSE NULL
END
WHERE market_id = $1 AND closed_at IS NULL
@@ -252,14 +218,8 @@ 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 (excluded_from_metrics IS NOT TRUE)) AS resolved_count,
AND final_prob IS NOT NULL) AS resolved_count,
COALESCE(SUM(net_cost)
FILTER (WHERE closed_at IS NULL), 0) AS total_deployed,
@@ -272,17 +232,15 @@ class Database:
FILTER (WHERE closed_at IS NULL
AND edge_net IS NOT NULL), 0) AS unrealized_pnl_est,
-- Realized PnL: admin-excluded trades omitted (close_pnl=0 by convention
-- but excluded explicitly so they don't skew the aggregate).
-- Realized PnL: closed trades with a known resolution.
-- close_pnl is computed at close time from actual resolution.
COALESCE(SUM(close_pnl)
FILTER (WHERE closed_at IS NOT NULL
AND close_pnl IS NOT NULL
AND (excluded_from_metrics IS NOT TRUE)), 0) AS realized_pnl,
AND close_pnl IS NOT NULL), 0) AS realized_pnl,
COUNT(*) FILTER (WHERE closed_at IS NOT NULL
AND close_pnl IS NOT NULL
AND close_pnl > 0
AND (excluded_from_metrics IS NOT TRUE)) AS wins_realized,
AND close_pnl > 0) AS wins_realized,
-- Calibration (Brier score transformed to higher-is-better):
-- 1 AVG((final_prob resolution)²) on resolved trades.
@@ -290,15 +248,12 @@ 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
AND (excluded_from_metrics IS NOT TRUE)) >= 10
AND final_prob IS NOT NULL) >= 10
THEN 1.0 - AVG((final_prob - resolution) * (final_prob - resolution))
FILTER (WHERE resolution IS NOT NULL
AND final_prob IS NOT NULL
AND (excluded_from_metrics IS NOT TRUE))
AND final_prob IS NOT NULL)
ELSE NULL
END AS calibration_score
@@ -330,42 +285,12 @@ 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 DISTINCT ON (timestamp::date) *
FROM metrics_daily
ORDER BY timestamp::date DESC, timestamp DESC
LIMIT $1
""", days
"SELECT * FROM metrics_daily ORDER BY 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.
@@ -435,27 +360,22 @@ 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,
@@ -539,7 +459,6 @@ 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,
@@ -574,530 +493,6 @@ 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
+74 -226
View File
@@ -2,49 +2,34 @@
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 a ManifoldMatchResult with full audit metadata.
by keyword overlap and returns its probability as a calibration signal.
Match threshold: >= 0.40 Jaccard overlap (raised from 0.25 for stricter semantics).
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.
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=...'.
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.
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.
Cache TTL: 30 minutes (Manifold markets move slowly vs our 60 s cycle).
Match threshold: >= 0.25 keyword overlap ratio between significant tokens.
"""
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.40 # raised from 0.25
_MATCH_THRESHOLD = 0.25
_STOP_WORDS = frozenset([
"will", "the", "a", "an", "is", "are", "was", "were", "be", "been",
@@ -58,24 +43,9 @@ _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
_DEMOCRAT_WORDS = frozenset(["democrat", "democrats", "democratic"])
def _significant_words(text: str) -> set[str]:
@@ -99,53 +69,27 @@ def _detect_party(text: str) -> Optional[str]:
return None
# ── 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[^,]*,")
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:
def _best_match_with_audit(
poly_question: str,
results: list[dict],
) -> tuple[Optional[dict], float, bool]:
"""
Coarse classification of what a question is *asking about*, used to reject
matches whose outcomes are not equivalent even when tokens overlap.
Find the best-matching open binary Manifold market.
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.
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)
"""
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
return None, 0.0, False
best_score = 0.0
best: Optional[dict] = None
best_needs_inv = False
for result in results:
if result.get("outcomeType") != "BINARY":
@@ -162,14 +106,18 @@ def _find_best_candidate(poly_question: str, results: list[dict]) -> tuple[Optio
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
)
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
if best_score >= _MATCH_THRESHOLD and best is not None:
return best, best_score, best_needs_inv
return None, best_score, False
class ManifoldClient:
@@ -177,32 +125,27 @@ class ManifoldClient:
def __init__(self) -> None:
self._client = httpx.AsyncClient(timeout=15)
# question → (fetched_at_monotonic, ManifoldMatchResult)
self._cache: dict[str, tuple[float, ManifoldMatchResult]] = {}
# question → (fetched_at_monotonic, probability_or_None)
self._cache: dict[str, tuple[float, Optional[float]]] = {}
async def get_match(self, question: str) -> ManifoldMatchResult:
async def get_probability(self, question: str) -> Optional[float]:
"""
Return a ManifoldMatchResult for the given Polymarket question.
Return Manifold probability for a matching market, or None.
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
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.
"""
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:
result = ManifoldMatchResult(
status="no_results", search_query="",
poly_outcome_type=poly_outcome,
)
self._cache[question] = (now, result)
return result
self._cache[question] = (now, None)
return None
try:
resp = await self._client.get(
@@ -211,140 +154,45 @@ class ManifoldClient:
)
resp.raise_for_status()
results = resp.json()
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
except Exception as e:
log.warning("Manifold API error for %r: %s", question[:40], e)
self._cache[question] = (now, None)
return None
if not results:
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)
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}"
if match is None:
log.info(
"Manifold REJECTED %-50s | score=%.2f < threshold=%.2f | query=%r",
"Manifold no_match: %-50s | best_score=%.2f < %.2f | query=%r",
question[:50], score, _MATCH_THRESHOLD, query,
)
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
self._cache[question] = (now, None)
return None
# ── 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)
prob_raw = float(match["probability"])
prob_final = (1.0 - prob_raw) if needs_inv else prob_raw
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})"
# 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"
log.info(
"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 ",
"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",
question[:50],
question,
best.get("question", ""),
url or "n/a",
score, prob_raw, inverted, prob_final,
match.get("question", ""),
url,
score, prob_raw, needs_inv, 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
self._cache[question] = (now, prob_final)
return prob_final
async def close(self) -> None:
await self._client.aclose()
+1 -17
View File
@@ -51,11 +51,7 @@ _DATE_RE = re.compile(
r"|\bQ[1-4]\b",
flags=re.IGNORECASE,
)
# 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"[?!\"'.,;:()\[\]{}\-–—]")
_PUNCT_RE = re.compile(r"[?!\"'.,;:()\[\]{}]")
class NewsClient:
@@ -83,18 +79,6 @@ 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.
-94
View File
@@ -211,32 +211,6 @@ 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
@@ -473,74 +447,6 @@ 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:
+3 -273
View File
@@ -113,10 +113,9 @@ 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 — 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
-- close_pnl: realized P&L in USDC at close time.
-- BUY_YES: (resolution - entry_price) * shares
-- BUY_NO: ((1 - resolution) - entry_price) * shares
-- NULL if closed without a known resolution (legacy closes, inversion fixes).
-- ─────────────────────────────────────────────────────────────────────────────
ALTER TABLE trades ADD COLUMN IF NOT EXISTS close_pnl DOUBLE PRECISION;
@@ -169,103 +168,6 @@ 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
--
@@ -280,175 +182,3 @@ 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()
);
+14 -87
View File
@@ -22,30 +22,6 @@ 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:
@@ -81,16 +57,6 @@ 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 (
@@ -132,7 +98,7 @@ class PaperExecutor:
positions_value = sum(positions_size.values())
self._portfolio.positions = positions_size
self._portfolio.cash = cash_available(self._portfolio.cash, total_net_cost)
self._portfolio.cash = max(0.0, 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
@@ -210,16 +176,6 @@ 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
@@ -229,7 +185,7 @@ class PaperExecutor:
# Persist to DB
await self._db.save_trade(trade)
_notify_in_background(
asyncio.create_task(
telegram.trade_opened(trade.question, trade.direction, trade.size_usdc, trade.edge_net)
)
@@ -250,7 +206,7 @@ class PaperExecutor:
"LEGACY_CLOSE market=%s | returned $%.2f to cash | %s",
market_id, cost, reason[:80],
)
_notify_in_background(
asyncio.create_task(
telegram.trade_legacy_closed(question or market_id, cost, reason)
)
return cost
@@ -259,53 +215,24 @@ class PaperExecutor:
"""Close a paper position after market resolution.
resolution: 1.0 if YES won, 0.0 if NO won.
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.
Persists resolution and close_pnl to DB (computed via SQL from stored
entry_price and shares). Returns approximate P&L for logging.
"""
if market_id not in self._portfolio.positions:
return None
position_cost = self._portfolio.positions[market_id]
open_trades = await self._db.get_open_trades_for_market(market_id)
position_cost = self._portfolio.positions.pop(market_id)
self._portfolio.cash += position_cost * resolution # pay out winnings
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="resolved",
reason=f"market_resolved resolution={resolution:.1f}",
resolution=resolution,
)
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,
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)
)
_notify_in_background(
telegram.trade_closed(question or market_id, pnl)
)
return pnl
# Approximate PnL: settlement value minus cost. Exact value is in close_pnl.
return approx_pnl
+44 -172
View File
@@ -11,89 +11,21 @@ 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,
MANIFOLD_SIGNAL_ENABLED,
)
from bot.strategy.bayesian import BayesianStrategy, gnews_priority, MAX_NEWS_QUERIES_PER_CYCLE
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,
@@ -106,23 +38,9 @@ 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))
@@ -130,16 +48,6 @@ 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.
@@ -194,7 +102,6 @@ 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(
@@ -202,7 +109,6 @@ 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
@@ -230,39 +136,11 @@ 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()
@@ -274,17 +152,7 @@ async def run_trading_loop(
if denom == 0:
return "0% (0/0)"
return f"{n * 100 // denom}% ({n}/{denom})"
# 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"
gnews_cap = strategy._news_queries_this_cycle # already updated by reset below
log.info(
"[CYCLE SUMMARY]\n"
@@ -302,7 +170,9 @@ async def run_trading_loop(
" gnews_queries_used: %d/%d\n"
" reentry_guard_blocked: %d\n"
" legacy_incomplete_seen: %d\n"
"%s",
" family_conflicts_prevented: %d\n"
" manifold_matches_accepted: %d\n"
" manifold_matches_rejected: %d",
n_total,
n_uncertainty,
stats["max_edge_gross"],
@@ -317,37 +187,14 @@ async def run_trading_loop(
stats["gnews_queries_used"], MAX_NEWS_QUERIES_PER_CYCLE,
reentry_guard_count,
legacy_incomplete_count,
manifold_summary,
stats["skip_family"],
stats["manifold_matches_accepted"],
stats["manifold_matches_rejected"],
)
# 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)
@@ -357,17 +204,14 @@ 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 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.
market_family_key() logic, detect contradictions, re-validate Manifold
signals, and report KEEP / REVIEW / CLOSE_RECOMMENDED per position.
In paper_mode: auto-closes all CLOSE_RECOMMENDED positions after logging.
"""
@@ -406,6 +250,8 @@ 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",
})
@@ -443,7 +289,31 @@ async def run_legacy_scan(
p["market_id"], p["family_key_old"] or "none", p["family_key_new"],
)
# Step 3: log the full scan report (before any closures)
# 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)
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")
@@ -459,6 +329,7 @@ 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"],
@@ -466,11 +337,12 @@ 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 4: auto-close in paper mode
# Step 5: 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:
@@ -503,7 +375,7 @@ async def main() -> None:
external = ExternalDataClient()
news = NewsClient()
manifold = ManifoldClient()
strategy = BayesianStrategy(news=news, manifold=manifold, db=db)
strategy = BayesianStrategy(news=news, manifold=manifold)
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)
@@ -523,7 +395,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, executor, PAPER_MODE)
await run_legacy_scan(db, scan_markets, manifold, executor, PAPER_MODE)
try:
await run_trading_loop(poly, external, strategy, risk, executor, metrics, db)
-79
View File
@@ -1,79 +0,0 @@
"""
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_{t1}) / 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
+9 -14
View File
@@ -15,16 +15,13 @@ 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 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.
sharpe_ratio 0.0 requires a daily-return time series, not yet tracked.
"""
import logging
import os
from datetime import datetime, UTC
from bot.data.db import Database
from bot.metrics.sharpe import sharpe_with_gate
from bot.executor.paper import Trade
log = logging.getLogger(__name__)
@@ -33,6 +30,11 @@ 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.
@@ -65,12 +67,6 @@ 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"]),
@@ -84,7 +80,7 @@ class MetricsTracker:
"total_pnl": total_pnl,
"win_rate": win_rate,
"avg_edge": avg_edge,
"sharpe_ratio": sharpe, # NULL until sample gate passes
"sharpe_ratio": 0.0, # requires daily-return series (not yet tracked)
"calibration_score": calibration,
"paper_mode": True,
}
@@ -93,10 +89,9 @@ 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 sharpe=%s",
"win_rate=%s calibration=%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})",
)
-184
View File
@@ -1,184 +0,0 @@
"""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,
)
-208
View File
@@ -1,208 +0,0 @@
"""
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()
-394
View File
@@ -1,394 +0,0 @@
"""
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()
-22
View File
@@ -62,17 +62,6 @@ 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:
@@ -170,15 +159,4 @@ 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,
)
+76 -552
View File
@@ -12,21 +12,16 @@ 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, timedelta, timezone
from datetime import datetime, 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__)
@@ -63,81 +58,11 @@ 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)
# ─────────────────────────────────────────────────────────────────────────────
@@ -167,82 +92,24 @@ 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, 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.
"""
def _days_to_resolution(end_date: str) -> int:
"""Return calendar days until market resolution, or 30 if unknown."""
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)
now = as_of if as_of is not None else datetime.now(timezone.utc)
days = (dt - now).days
days = (dt - datetime.now(timezone.utc)).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).
@@ -303,19 +170,6 @@ 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:
@@ -347,12 +201,10 @@ 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
@@ -364,13 +216,6 @@ 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."""
@@ -382,54 +227,6 @@ 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."""
@@ -449,14 +246,6 @@ 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(
@@ -464,17 +253,10 @@ 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
@@ -495,18 +277,13 @@ class BayesianStrategy:
"below", "under", "less than", "lower", "drop",
])
# 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_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
is_altcoin = is_sol or is_xrp or is_doge or any(
has_token(question_lower, t) for t in ["ltc", "bnb", "ada", "avax"]
) or any(
w in question_lower for w in ["litecoin", "cardano", "avalanche"]
w in question_lower for w in ["ltc", "litecoin", "bnb", "ada", "cardano", "avax", "avalanche"]
)
is_general_crypto = any(
w in question_lower for w in ["crypto", "market cap", "total market", "altcoin", "defi"]
@@ -529,7 +306,6 @@ class BayesianStrategy:
"SKIP_UNSUPPORTED %-50s | cat=%r",
market.question[:50], category,
)
self._record(market, skip_reason="unsupported")
return None
if not ext.valid:
@@ -537,7 +313,6 @@ 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 ────────────────────────────
@@ -549,7 +324,6 @@ 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
@@ -557,7 +331,6 @@ 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 ────────────────────────────────────
@@ -568,98 +341,76 @@ 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, as_of)
days = _days_to_resolution(market.end_date)
regime_min = _regime_min_edge(category, days)
# ── Bayesian probability estimation ──────────────────────────────────
sources: list[str] = [f"Prior=poly({prior:.3f})"]
adjustments: list[float] = []
# 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 or BTC as sentiment proxy)
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"
# 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"
else:
momentum = ext.total_market_cap_change
asset_label = "total mktcap"
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}%")
if abs(momentum) > 2:
momentum_adj = math.tanh(momentum / 20) * 0.15
_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
fg = ext.fear_greed_index
if fg > 70:
fg_adj = 0.06
sources.append(f"Fear&Greed: {fg} (greed)")
elif fg < 30:
fg_adj = -0.06
sources.append(f"Fear&Greed: {fg} (fear)")
else:
fg_adj = (fg - 50) / 50 * 0.04
sources.append(f"Fear&Greed: {fg} (neutral)")
_fg_contribution = fg_adj if is_price_above else -fg_adj
adjustments.append(_fg_contribution)
# 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
sources.append(f"Fear&Greed: {fg} (greed)")
elif fg < 30:
fg_adj = -0.06
sources.append(f"Fear&Greed: {fg} (fear)")
else:
fg_adj = (fg - 50) / 50 * 0.04
sources.append(f"Fear&Greed: {fg} (neutral)")
_fg_contribution = fg_adj if is_price_above else -fg_adj
adjustments.append(_fg_contribution)
# 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.
# Signal 3: BTC dominance — hurts altcoins when high
_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)
sources.append(f"BTC dom: {ext.btc_dominance:.1f}% (high → alt pressure)")
elif (is_eth or is_altcoin or is_general_crypto) and ext.btc_dominance < 45:
_btc_dom_contribution = 0.03 if is_price_above else -0.03
adjustments.append(_btc_dom_contribution)
sources.append(f"BTC dom: {ext.btc_dominance:.1f}% (low → alt season)")
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)
sources.append(f"BTC dom: {ext.btc_dominance:.1f}% (high → alt pressure)")
elif (is_eth or is_altcoin or is_general_crypto) and ext.btc_dominance < 45:
_btc_dom_contribution = 0.03 if is_price_above else -0.03
adjustments.append(_btc_dom_contribution)
sources.append(f"BTC dom: {ext.btc_dominance:.1f}% (low → alt season)")
# Signal 4: GNews sentiment (politics only, budget-gated)
# Phase 3: caller has pre-sorted markets by gnews_priority() so the
# highest-value markets reach this block first.
news_log_adj = 0.0
news_sentiment = 0.0
# Replay R0: True when GNews was never consulted for this market this
# cycle (budget exhausted) — a replay must not read feat_news_lo=0.0 as
# "there was no news".
news_budget_skipped = False
# self._news.enabled gates the whole block: with no GNews API key the
# client is a no-op, so we must not consume (or report) query budget for
# it — see NewsClient.enabled.
if is_politics and self._news is not None and self._news.enabled:
if is_politics and self._news is not None:
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,
@@ -668,129 +419,18 @@ 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
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_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_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")
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_used = True
self._manifold_fetched += 1
m_clamped = max(0.05, min(0.95, manifold_prob))
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}")
# 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
@@ -798,31 +438,8 @@ class BayesianStrategy:
# Posterior via log-odds updating
log_odds_prior = math.log(prior / (1 - prior))
total_adj = sum(adjustments)
# 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),
)
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))
# ── Phase 1: edge_gross and edge_net ─────────────────────────────────
raw_edge = estimated_prob - market.yes_price
@@ -844,6 +461,15 @@ 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} "
@@ -855,80 +481,6 @@ 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:
@@ -957,21 +509,8 @@ 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 = (
prob_part +
f"Prior=poly({prior:.3f}) → estimate={estimated_prob:.3f} | "
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} | "
@@ -1021,21 +560,6 @@ 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,
)
+2 -12
View File
@@ -200,12 +200,8 @@ export default function App() {
<MetricCard
title="Sharpe"
value={fmt(summary.sharpe_ratio)}
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)}
subtitle="Objetivo ≥ 0.5"
progress={Math.min(1, summary.sharpe_ratio / 2)}
progressColor={summary.sharpe_ratio >= 0.5 ? 'var(--green)' : 'var(--amber)'}
/>
<MetricCard
@@ -220,12 +216,6 @@ 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 */}
-10
View File
@@ -1,10 +0,0 @@
"""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
-106
View File
@@ -1,106 +0,0 @@
"""
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)
-159
View File
@@ -1,159 +0,0 @@
"""
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)
-130
View File
@@ -1,130 +0,0 @@
"""
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
-123
View File
@@ -1,123 +0,0 @@
"""
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
-152
View File
@@ -1,152 +0,0 @@
"""
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)
-247
View File
@@ -1,247 +0,0 @@
"""
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"
-77
View File
@@ -1,77 +0,0 @@
"""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("TrumpPutin 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
-174
View File
@@ -1,174 +0,0 @@
"""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
-130
View File
@@ -1,130 +0,0 @@
"""
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
-367
View File
@@ -1,367 +0,0 @@
"""
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)
-219
View File
@@ -1,219 +0,0 @@
"""
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
-242
View File
@@ -1,242 +0,0 @@
"""
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
-224
View File
@@ -1,224 +0,0 @@
"""
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",
]