Compare commits
13
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0816e19740 | ||
|
|
2b326ad54f | ||
|
|
124b6d8558 | ||
|
|
6c544e46e2 | ||
|
|
0ac48ba7f8 | ||
|
|
eb4f67414a | ||
|
|
919fe1617a | ||
|
|
117d2b33b2 | ||
|
|
7f84bc3ec7 | ||
|
|
9e21ecac21 | ||
|
|
a3ec69d2be | ||
|
|
af0d1fbc59 | ||
|
|
54fc8fa11a |
@@ -178,7 +178,8 @@ jobs:
|
||||
# keep their current (still existing) tag in the registry.
|
||||
if [ "${{ steps.changes.outputs.build_bot }}" = "true" ]; then
|
||||
sed -i "s|image: .*polymarket-bot[^-].*|image: git.chemavx.xyz/chemavx/polymarket-bot:${TAG}|g" \
|
||||
polymarket-bot/deployment-bot.yaml
|
||||
polymarket-bot/deployment-bot.yaml \
|
||||
polymarket-bot/cronjob-outcomes.yaml
|
||||
fi
|
||||
if [ "${{ steps.changes.outputs.build_api }}" = "true" ]; then
|
||||
sed -i "s|image: .*polymarket-bot-api.*|image: git.chemavx.xyz/chemavx/polymarket-bot-api:${TAG}|g" \
|
||||
|
||||
+245
@@ -650,6 +650,251 @@ class Database:
|
||||
cooldown_reason = EXCLUDED.cooldown_reason
|
||||
""", poly_market_id, last_status, retry_after, cooldown_reason)
|
||||
|
||||
# ── Replay R0: snapshot recorder ─────────────────────────────────────────
|
||||
|
||||
async def save_ext_snapshot(self, cycle_ts, ext) -> None:
|
||||
"""Persist the ExternalSignals snapshot for one cycle (Replay R0)."""
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute("""
|
||||
INSERT INTO ext_snapshots (
|
||||
cycle_ts, btc_price, btc_change_24h, eth_price, eth_change_24h,
|
||||
btc_dominance, fear_greed_index, fear_greed_label,
|
||||
total_market_cap_change, valid
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10)
|
||||
ON CONFLICT (cycle_ts) DO NOTHING
|
||||
""",
|
||||
cycle_ts, ext.btc_price, ext.btc_change_24h,
|
||||
ext.eth_price, ext.eth_change_24h, ext.btc_dominance,
|
||||
ext.fear_greed_index, ext.fear_greed_label,
|
||||
ext.total_market_cap_change, ext.valid,
|
||||
)
|
||||
|
||||
async def upsert_markets(self, markets: list) -> None:
|
||||
"""Refresh market metadata (Replay R0) — replay rebuilds Market from here."""
|
||||
rows = [
|
||||
(m.id, m.condition_id, m.question, m.category, m.end_date, m.active)
|
||||
for m in markets
|
||||
]
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.executemany("""
|
||||
INSERT INTO markets (id, condition_id, question, category, end_date, active, last_seen)
|
||||
VALUES ($1,$2,$3,$4,$5,$6, now())
|
||||
ON CONFLICT (id) DO UPDATE SET
|
||||
condition_id = EXCLUDED.condition_id,
|
||||
question = EXCLUDED.question,
|
||||
category = EXCLUDED.category,
|
||||
end_date = EXCLUDED.end_date,
|
||||
active = EXCLUDED.active,
|
||||
last_seen = now()
|
||||
""", rows)
|
||||
|
||||
async def save_signal_records(self, cycle_ts, records: list[dict]) -> None:
|
||||
"""Batch-insert one cycle's decision records into signals (Replay R0)."""
|
||||
if not records:
|
||||
return
|
||||
rows = [
|
||||
(
|
||||
r["market_id"], cycle_ts, cycle_ts,
|
||||
r["polymarket_price"], r["category"], r["volume_24h"],
|
||||
r["skip_reason"], r["family_key"],
|
||||
r["prior_prob"], r["estimated_prob"], r["raw_final_prob"],
|
||||
r["edge_gross"], r["edge_net"], r["regime_min_edge"],
|
||||
r["days_to_resolution"], r["confidence"], r["direction"],
|
||||
r["passed_gross"], r["passed_net"],
|
||||
r["news_sentiment"], r["news_budget_skipped"],
|
||||
r["guardrail_applied"], r["guardrail_changed_decision"],
|
||||
r["feat_fg_lo"], r["feat_mom_lo"], r["feat_news_lo"],
|
||||
r["feat_mfld_lo"], r["feat_btc_dom_lo"],
|
||||
r["edge_gross"], # legacy `edge` column mirrors edge_gross
|
||||
r["acted_on"],
|
||||
)
|
||||
for r in records
|
||||
]
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.executemany("""
|
||||
INSERT INTO signals (
|
||||
market_id, timestamp, cycle_ts,
|
||||
polymarket_price, category, volume_24h,
|
||||
skip_reason, family_key,
|
||||
prior_prob, estimated_prob, raw_final_prob,
|
||||
edge_gross, edge_net, regime_min_edge,
|
||||
days_to_resolution, confidence, direction,
|
||||
passed_gross, passed_net,
|
||||
news_sentiment, news_budget_skipped,
|
||||
guardrail_applied, guardrail_changed_decision,
|
||||
feat_fg_lo, feat_mom_lo, feat_news_lo,
|
||||
feat_mfld_lo, feat_btc_dom_lo,
|
||||
edge, acted_on
|
||||
) VALUES (
|
||||
$1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,
|
||||
$16,$17,$18,$19,$20,$21,$22,$23,$24,$25,$26,$27,$28,$29,$30
|
||||
)
|
||||
""", rows)
|
||||
|
||||
async def prune_signal_records(self, retention_days: int) -> int:
|
||||
"""Delete archive rows older than retention_days; returns rows deleted."""
|
||||
async with self._pool.acquire() as conn:
|
||||
result = await conn.execute(
|
||||
"DELETE FROM signals WHERE timestamp < now() - ($1 || ' days')::interval",
|
||||
str(retention_days),
|
||||
)
|
||||
await conn.execute(
|
||||
"DELETE FROM ext_snapshots WHERE cycle_ts < now() - ($1 || ' days')::interval",
|
||||
str(retention_days),
|
||||
)
|
||||
try:
|
||||
return int(result.split()[-1])
|
||||
except (ValueError, IndexError):
|
||||
return 0
|
||||
|
||||
# ── Replay R1: replay core ───────────────────────────────────────────────
|
||||
|
||||
async def get_replay_cycles(self, from_ts, to_ts) -> list:
|
||||
"""Return the cycle_ts values with archived decisions in [from_ts, to_ts)."""
|
||||
async with self._pool.acquire() as conn:
|
||||
rows = await conn.fetch("""
|
||||
SELECT DISTINCT cycle_ts FROM signals
|
||||
WHERE cycle_ts >= $1 AND cycle_ts < $2
|
||||
ORDER BY cycle_ts
|
||||
""", from_ts, to_ts)
|
||||
return [r["cycle_ts"] for r in rows]
|
||||
|
||||
async def get_ext_snapshot(self, cycle_ts) -> Optional[dict]:
|
||||
"""Return one cycle's ExternalSignals snapshot, or None if missing."""
|
||||
async with self._pool.acquire() as conn:
|
||||
row = await conn.fetchrow(
|
||||
"SELECT * FROM ext_snapshots WHERE cycle_ts = $1", cycle_ts
|
||||
)
|
||||
return dict(row) if row else None
|
||||
|
||||
async def get_cycle_signal_rows(self, cycle_ts) -> list[dict]:
|
||||
"""Return one cycle's archived decision rows in original evaluation
|
||||
order (id = insertion order = the order main.py evaluated them)."""
|
||||
async with self._pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
"SELECT * FROM signals WHERE cycle_ts = $1 ORDER BY id", cycle_ts
|
||||
)
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
async def get_markets_by_ids(self, market_ids: list[str]) -> dict[str, dict]:
|
||||
"""Return market metadata rows keyed by id (for Market reconstruction)."""
|
||||
if not market_ids:
|
||||
return {}
|
||||
async with self._pool.acquire() as conn:
|
||||
rows = await conn.fetch(
|
||||
"SELECT * FROM markets WHERE id = ANY($1::text[])", market_ids
|
||||
)
|
||||
return {r["id"]: dict(r) for r in rows}
|
||||
|
||||
async def save_replay_run(self, run: dict) -> None:
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute("""
|
||||
INSERT INTO replay_runs (
|
||||
run_id, git_sha, config_hash, config_json,
|
||||
from_ts, to_ts, cycles, decisions, matched, mismatched, note
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11)
|
||||
""",
|
||||
run["run_id"], run["git_sha"], run["config_hash"],
|
||||
run["config_json"], run["from_ts"], run["to_ts"],
|
||||
run["cycles"], run["decisions"], run["matched"],
|
||||
run["mismatched"], run["note"],
|
||||
)
|
||||
|
||||
async def save_replay_decisions(self, run_id: str, decisions: list[dict]) -> None:
|
||||
if not decisions:
|
||||
return
|
||||
rows = [
|
||||
(
|
||||
run_id, d["cycle_ts"], d["market_id"],
|
||||
d["skip_reason"], d["prior_prob"], d["estimated_prob"],
|
||||
d["raw_final_prob"], d["edge_gross"], d["edge_net"],
|
||||
d["regime_min_edge"], d["days_to_resolution"],
|
||||
d["confidence"], d["direction"], d["would_trade"],
|
||||
d["recorded_skip_reason"], d["matched"], d["mismatch_field"],
|
||||
)
|
||||
for d in decisions
|
||||
]
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.executemany("""
|
||||
INSERT INTO replay_decisions (
|
||||
run_id, cycle_ts, market_id,
|
||||
skip_reason, prior_prob, estimated_prob,
|
||||
raw_final_prob, edge_gross, edge_net,
|
||||
regime_min_edge, days_to_resolution,
|
||||
confidence, direction, would_trade,
|
||||
recorded_skip_reason, matched, mismatch_field
|
||||
) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)
|
||||
""", rows)
|
||||
|
||||
# ── Replay R2: outcomes + calibration metrics ────────────────────────────
|
||||
|
||||
async def get_unresolved_archived_market_ids(self) -> list[str]:
|
||||
"""Archived markets (present in signals) with no stored outcome yet."""
|
||||
async with self._pool.acquire() as conn:
|
||||
rows = await conn.fetch("""
|
||||
SELECT DISTINCT s.market_id FROM signals s
|
||||
LEFT JOIN market_outcomes o ON o.market_id = s.market_id
|
||||
WHERE o.market_id IS NULL
|
||||
ORDER BY s.market_id
|
||||
""")
|
||||
return [r["market_id"] for r in rows]
|
||||
|
||||
async def upsert_market_outcome(
|
||||
self, market_id: str, outcome: float, resolved_at
|
||||
) -> None:
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute("""
|
||||
INSERT INTO market_outcomes (market_id, outcome, resolved_at)
|
||||
VALUES ($1, $2, $3)
|
||||
ON CONFLICT (market_id) DO UPDATE
|
||||
SET outcome = EXCLUDED.outcome,
|
||||
resolved_at = EXCLUDED.resolved_at,
|
||||
fetched_at = NOW()
|
||||
""", market_id, outcome, resolved_at)
|
||||
|
||||
async def get_outcome_coverage(self) -> dict:
|
||||
"""How much of the archive is scorable: resolved vs archived markets."""
|
||||
async with self._pool.acquire() as conn:
|
||||
row = await conn.fetchrow("""
|
||||
SELECT
|
||||
(SELECT COUNT(DISTINCT market_id) FROM signals) AS archived,
|
||||
(SELECT COUNT(*) FROM market_outcomes
|
||||
WHERE market_id IN (SELECT DISTINCT market_id FROM signals)
|
||||
) AS resolved
|
||||
""")
|
||||
return dict(row)
|
||||
|
||||
async def get_calibration_rows(self, run_id: Optional[str] = None) -> list[dict]:
|
||||
"""Every archived evaluation with a full estimate AND a known outcome.
|
||||
|
||||
run_id None scores the R0 archive (signals); a run_id scores that
|
||||
replay run's re-estimates instead (counterfactual calibration).
|
||||
Rows without estimated_prob (skipped before estimation: prior_extreme,
|
||||
unsupported, family, no_signals) carry no model prediction to score.
|
||||
"""
|
||||
async with self._pool.acquire() as conn:
|
||||
if run_id is None:
|
||||
rows = await conn.fetch("""
|
||||
SELECT s.market_id, s.category,
|
||||
s.estimated_prob, s.prior_prob, o.outcome
|
||||
FROM signals s
|
||||
JOIN market_outcomes o ON o.market_id = s.market_id
|
||||
WHERE s.estimated_prob IS NOT NULL
|
||||
AND s.prior_prob IS NOT NULL
|
||||
""")
|
||||
else:
|
||||
rows = await conn.fetch("""
|
||||
SELECT d.market_id, m.category,
|
||||
d.estimated_prob, d.prior_prob, o.outcome
|
||||
FROM replay_decisions d
|
||||
JOIN market_outcomes o ON o.market_id = d.market_id
|
||||
LEFT JOIN markets m ON m.id = d.market_id
|
||||
WHERE d.run_id = $1
|
||||
AND d.estimated_prob IS NOT NULL
|
||||
AND d.prior_prob IS NOT NULL
|
||||
""", run_id)
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
async def mark_manifold_audit_used(self, audit_id: str) -> None:
|
||||
async with self._pool.acquire() as conn:
|
||||
await conn.execute(
|
||||
|
||||
+17
-1
@@ -51,7 +51,11 @@ _DATE_RE = re.compile(
|
||||
r"|\bQ[1-4]\b",
|
||||
flags=re.IGNORECASE,
|
||||
)
|
||||
_PUNCT_RE = re.compile(r"[?!\"'.,;:()\[\]{}]")
|
||||
# Hyphens/dashes are GNews query operators (a leading '-' means "exclude the
|
||||
# next term"), so a token like "El-Sayed" makes the API return HTTP 400. Strip
|
||||
# them to spaces along with the rest of the punctuation so the query stays a
|
||||
# plain keyword list. – = en dash, — = em dash.
|
||||
_PUNCT_RE = re.compile(r"[?!\"'.,;:()\[\]{}\-–—]")
|
||||
|
||||
|
||||
class NewsClient:
|
||||
@@ -79,6 +83,18 @@ class NewsClient:
|
||||
# Public API
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
@property
|
||||
def enabled(self) -> bool:
|
||||
"""True only when a GNews API key is configured.
|
||||
|
||||
When False, get_sentiment() is a no-op that returns 0.0 without any
|
||||
network call, so callers must skip GNews entirely — including the
|
||||
per-cycle query budget accounting — instead of "spending" a query that
|
||||
never reaches the API (which inflated gnews_queries_used to a phantom
|
||||
5/5 while the key was missing).
|
||||
"""
|
||||
return bool(self._api_key)
|
||||
|
||||
async def get_sentiment(self, question: str) -> float:
|
||||
"""
|
||||
Return a sentiment score ∈ [-1.0, +1.0] for the market question.
|
||||
|
||||
@@ -318,3 +318,137 @@ CREATE TABLE IF NOT EXISTS manifold_eval_cooldown (
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_mfld_cooldown_retry ON manifold_eval_cooldown(retry_after);
|
||||
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
-- Replay R0: snapshot recorder — the archive the replay engine reads from
|
||||
--
|
||||
-- The signals table (Phase 2/5 schema) never had a writer; R0 makes it the
|
||||
-- per-(market, cycle) decision archive. One row per evaluated market per
|
||||
-- cycle, carrying both the INPUTS the strategy saw (external signals, news
|
||||
-- sentiment, per-feature log-odds) and the OUTPUTS it produced (probs, edges,
|
||||
-- gates, skip_reason). A replay run rebuilds Market/ExternalSignals from
|
||||
-- these rows plus ext_snapshots and re-executes evaluate() deterministically.
|
||||
--
|
||||
-- cycle_ts groups all rows of one trading cycle and joins them to their
|
||||
-- ext_snapshots row (same timestamp; no FK to keep writes independent).
|
||||
-- days_to_resolution is persisted so replay does not depend on wall-clock.
|
||||
-- news_budget_skipped distinguishes "GNews had nothing" from "GNews was not
|
||||
-- asked this cycle" (5-query budget) — without it politics replay would treat
|
||||
-- budget starvation as absence of news.
|
||||
-- Retention: rows older than SIGNALS_RETENTION_DAYS (default 90) are pruned.
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS cycle_ts TIMESTAMPTZ;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS category TEXT;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS prior_prob DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS raw_final_prob DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS days_to_resolution INTEGER;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS volume_24h DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS news_sentiment DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS news_budget_skipped BOOLEAN;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS guardrail_applied BOOLEAN;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS guardrail_changed_decision BOOLEAN;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_fg_lo DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_mom_lo DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_news_lo DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_mfld_lo DOUBLE PRECISION;
|
||||
ALTER TABLE signals ADD COLUMN IF NOT EXISTS feat_btc_dom_lo DOUBLE PRECISION;
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_signals_cycle ON signals(cycle_ts);
|
||||
|
||||
-- One row per trading cycle: the ExternalSignals snapshot every market in
|
||||
-- that cycle was evaluated against. Written once per cycle before the
|
||||
-- evaluation loop; signals rows join on cycle_ts.
|
||||
CREATE TABLE IF NOT EXISTS ext_snapshots (
|
||||
cycle_ts TIMESTAMPTZ PRIMARY KEY,
|
||||
btc_price DOUBLE PRECISION,
|
||||
btc_change_24h DOUBLE PRECISION,
|
||||
eth_price DOUBLE PRECISION,
|
||||
eth_change_24h DOUBLE PRECISION,
|
||||
btc_dominance DOUBLE PRECISION,
|
||||
fear_greed_index INTEGER,
|
||||
fear_greed_label TEXT,
|
||||
total_market_cap_change DOUBLE PRECISION,
|
||||
valid BOOLEAN
|
||||
);
|
||||
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
-- Replay R1: replay core — re-execute evaluate() over the R0 archive
|
||||
--
|
||||
-- A replay run reads cycles from signals + ext_snapshots + markets, rebuilds
|
||||
-- the exact inputs (including archived news_sentiment — GNews is never called),
|
||||
-- re-runs BayesianStrategy.evaluate() with the archived cycle_ts as clock, and
|
||||
-- writes one replay_decisions row per (cycle, market).
|
||||
--
|
||||
-- replay_runs tags every run with the code (git_sha) and strategy constants
|
||||
-- (config_hash) that produced it: two runs over the same window with different
|
||||
-- config_hash values are a counterfactual comparison; same config_hash against
|
||||
-- the recorded rows is a determinism check (mismatches should be 0, modulo
|
||||
-- day-boundary crossings between cycle_ts and the original wall-clock).
|
||||
--
|
||||
-- matched: replayed decision equals the recorded one (skip_reason, probs,
|
||||
-- confidence, direction). NULL when not comparable — e.g. reentry_guard
|
||||
-- rows, recorded outside evaluate() with no decision fields to compare;
|
||||
-- the replay still re-evaluates them, which is extra calibration data.
|
||||
-- mismatch_field: first field that differed, for triage.
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
CREATE TABLE IF NOT EXISTS replay_runs (
|
||||
run_id TEXT PRIMARY KEY,
|
||||
created_at TIMESTAMPTZ DEFAULT NOW(),
|
||||
git_sha TEXT,
|
||||
config_hash TEXT,
|
||||
config_json TEXT,
|
||||
from_ts TIMESTAMPTZ,
|
||||
to_ts TIMESTAMPTZ,
|
||||
cycles INTEGER,
|
||||
decisions INTEGER,
|
||||
matched INTEGER,
|
||||
mismatched INTEGER,
|
||||
note TEXT
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS replay_decisions (
|
||||
id SERIAL PRIMARY KEY,
|
||||
run_id TEXT NOT NULL,
|
||||
cycle_ts TIMESTAMPTZ NOT NULL,
|
||||
market_id TEXT NOT NULL,
|
||||
-- replayed outputs (same semantics as the signals columns)
|
||||
skip_reason TEXT,
|
||||
prior_prob DOUBLE PRECISION,
|
||||
estimated_prob DOUBLE PRECISION,
|
||||
raw_final_prob DOUBLE PRECISION,
|
||||
edge_gross DOUBLE PRECISION,
|
||||
edge_net DOUBLE PRECISION,
|
||||
regime_min_edge DOUBLE PRECISION,
|
||||
days_to_resolution INTEGER,
|
||||
confidence DOUBLE PRECISION,
|
||||
direction TEXT,
|
||||
would_trade BOOLEAN,
|
||||
-- fidelity vs the recorded signals row
|
||||
recorded_skip_reason TEXT,
|
||||
matched BOOLEAN,
|
||||
mismatch_field TEXT
|
||||
);
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_replay_decisions_run ON replay_decisions(run_id);
|
||||
CREATE INDEX IF NOT EXISTS idx_replay_decisions_mkt ON replay_decisions(market_id);
|
||||
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
-- Replay R2: outcomes + calibration metrics
|
||||
--
|
||||
-- One row per resolved market, fetched from the Gamma API via
|
||||
-- get_market_resolution() (UMA-final only: a market closed but still in
|
||||
-- proposal/dispute is not stored). outcome is the final YES price:
|
||||
-- 1.0 = YES won, 0.0 = NO won.
|
||||
--
|
||||
-- Joining signals (or replay_decisions) to market_outcomes scores every
|
||||
-- archived estimate against reality — Brier / log-loss of estimated_prob
|
||||
-- benchmarked against the market price (prior_prob) on the same rows,
|
||||
-- answering "does the model add value over the market?" across ALL
|
||||
-- evaluations, not just executed trades.
|
||||
-- ─────────────────────────────────────────────────────────────────────────────
|
||||
CREATE TABLE IF NOT EXISTS market_outcomes (
|
||||
market_id TEXT PRIMARY KEY,
|
||||
outcome DOUBLE PRECISION NOT NULL,
|
||||
resolved_at TIMESTAMPTZ,
|
||||
fetched_at TIMESTAMPTZ DEFAULT NOW()
|
||||
);
|
||||
|
||||
+63
@@ -27,6 +27,12 @@ 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"
|
||||
@@ -37,6 +43,14 @@ PAPER_BANKROLL = float(os.getenv("PAPER_BANKROLL", "10000"))
|
||||
# position per 10 minutes.
|
||||
RESOLUTION_CHECK_INTERVAL = 10
|
||||
|
||||
# Replay R0: persist per-(market, cycle) decision records + the ExternalSignals
|
||||
# snapshot each cycle, so the replay engine can re-run past decisions. The
|
||||
# recorder must never break trading — every write is wrapped in try/except.
|
||||
SIGNAL_RECORDER_ENABLED = os.getenv("SIGNAL_RECORDER_ENABLED", "true").lower() == "true"
|
||||
SIGNALS_RETENTION_DAYS = int(os.getenv("SIGNALS_RETENTION_DAYS", "90"))
|
||||
# Prune the archive roughly once a day at the 60s cycle cadence.
|
||||
SIGNALS_PRUNE_INTERVAL_CYCLES = 1440
|
||||
|
||||
|
||||
async def check_resolutions(
|
||||
poly: PolymarketClient,
|
||||
@@ -116,6 +130,16 @@ async def run_trading_loop(
|
||||
# 2. Get external signals
|
||||
ext_data = await external.get_all_signals()
|
||||
|
||||
# 2b. Replay R0: archive this cycle's inputs (ext snapshot + market
|
||||
# metadata). cycle_ts groups all signals rows of this cycle.
|
||||
cycle_ts = datetime.now(timezone.utc)
|
||||
if SIGNAL_RECORDER_ENABLED:
|
||||
try:
|
||||
await db.save_ext_snapshot(cycle_ts, ext_data)
|
||||
await db.upsert_markets(markets)
|
||||
except Exception as exc:
|
||||
log.warning("Signal recorder (inputs) failed: %s", exc)
|
||||
|
||||
# 3. Build occupied_families from the current open portfolio positions.
|
||||
# This prevents re-entering a family where we already hold a position.
|
||||
# We also pull from DB to survive pod restarts.
|
||||
@@ -170,6 +194,7 @@ async def run_trading_loop(
|
||||
|
||||
reentry_guard_count = 0
|
||||
cycle_trades = 0
|
||||
traded_market_ids: set[str] = set()
|
||||
for market in markets:
|
||||
if market.id in inverted_guard:
|
||||
log.info(
|
||||
@@ -177,6 +202,7 @@ async def run_trading_loop(
|
||||
market.id, market.question[:60],
|
||||
)
|
||||
reentry_guard_count += 1
|
||||
strategy.record_skip(market, "reentry_guard")
|
||||
continue
|
||||
|
||||
# evaluate() returns None for all skips — reasons are logged internally
|
||||
@@ -208,6 +234,7 @@ async def run_trading_loop(
|
||||
# Block this family for the rest of the cycle (Phase 2)
|
||||
occupied_families.add(signal.family_key)
|
||||
cycle_trades += 1
|
||||
traded_market_ids.add(market.id)
|
||||
# Mark manifold audit record as used in this trade
|
||||
if signal.mfld_audit_id:
|
||||
try:
|
||||
@@ -215,6 +242,28 @@ async def run_trading_loop(
|
||||
except Exception as exc:
|
||||
log.warning("Failed to mark manifold audit used: %s", exc)
|
||||
|
||||
# 7b. Replay R0: flush this cycle's decision records to the archive.
|
||||
# acted_on marks records whose signal actually became a trade
|
||||
# (evaluate() can emit a signal that risk sizing later rejects).
|
||||
records = strategy.drain_cycle_records()
|
||||
if SIGNAL_RECORDER_ENABLED and records:
|
||||
for rec in records:
|
||||
if rec["market_id"] in traded_market_ids:
|
||||
rec["acted_on"] = True
|
||||
try:
|
||||
await db.save_signal_records(cycle_ts, records)
|
||||
except Exception as exc:
|
||||
log.warning("Signal recorder (records) failed: %s", exc)
|
||||
if cycle_count % SIGNALS_PRUNE_INTERVAL_CYCLES == 1:
|
||||
try:
|
||||
pruned = await db.prune_signal_records(SIGNALS_RETENTION_DAYS)
|
||||
log.info(
|
||||
"Signal archive pruned: %d rows older than %d days removed",
|
||||
pruned, SIGNALS_RETENTION_DAYS,
|
||||
)
|
||||
except Exception as exc:
|
||||
log.warning("Signal archive prune failed: %s", exc)
|
||||
|
||||
# 8. [CYCLE SUMMARY] — one block per cycle, stable format for grep/compare
|
||||
stats = strategy.get_cycle_stats()
|
||||
legacy_incomplete_count = await db.get_legacy_incomplete_count()
|
||||
@@ -271,6 +320,20 @@ async def run_trading_loop(
|
||||
manifold_summary,
|
||||
)
|
||||
|
||||
# NEWS SUMMARY — one compact line, only on cycles where at least
|
||||
# one market had a material GNews contribution (never an empty
|
||||
# section on news-less cycles).
|
||||
if stats["news_with_material"] > 0:
|
||||
log.info(
|
||||
"NEWS SUMMARY | with_news=%d | avg_shift=%+.2f | "
|
||||
"max_shift=%+.2f | guardrail_applied=%d | changed_decisions=%d",
|
||||
stats["news_with_material"],
|
||||
stats["news_avg_shift"],
|
||||
stats["news_max_shift"],
|
||||
stats["news_guardrail_applied"],
|
||||
stats["news_changed_decisions"],
|
||||
)
|
||||
|
||||
# 9. Update daily metrics
|
||||
await metrics.update_daily_summary()
|
||||
|
||||
|
||||
+208
@@ -0,0 +1,208 @@
|
||||
"""
|
||||
Replay R2 — outcomes + calibration metrics.
|
||||
|
||||
Two phases, one CLI:
|
||||
|
||||
1. Fetch: for every archived market (present in `signals`) without a stored
|
||||
outcome, ask the Gamma API via PolymarketClient.get_market_resolution()
|
||||
— the same UMA-finality gate the trading loop uses to settle positions.
|
||||
Definitive resolutions are upserted into `market_outcomes`; open, disputed
|
||||
or ambiguous markets are simply retried on the next invocation. There is
|
||||
no data-loss urgency here (unlike the R0 recorder): Gamma reports past
|
||||
resolutions at any time, so running this lazily loses nothing.
|
||||
|
||||
2. Score: join archived estimates to outcomes and compute Brier / log-loss of
|
||||
estimated_prob, benchmarked against the market price (prior_prob) on the
|
||||
same rows. This scores ALL evaluations with a full estimate — the sample
|
||||
multiplier the phase plan calls for — not just executed trades. With
|
||||
--run-id it scores a replay run's re-estimates instead (counterfactual
|
||||
calibration: did config X predict better than the market?).
|
||||
|
||||
Reading the numbers: lower is better for both metrics; model < prior means
|
||||
the model added information over the market price. Micro averages weight
|
||||
every evaluation equally, so long-lived markets (~1 evaluation/min while in
|
||||
the universe) dominate; macro averages score each market once (mean of its
|
||||
evaluations) and answer the same question per market. Evaluations of one
|
||||
market minutes apart are highly autocorrelated — n_evaluations overstates
|
||||
the effective sample size, n_markets is the honest one.
|
||||
|
||||
CLI:
|
||||
python -m bot.outcomes # fetch new outcomes, then score archive
|
||||
python -m bot.outcomes --fetch-only
|
||||
python -m bot.outcomes --metrics-only
|
||||
python -m bot.outcomes --run-id UUID # score a replay run (implies no fetch)
|
||||
"""
|
||||
import argparse
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
from collections import defaultdict
|
||||
from typing import Optional
|
||||
|
||||
from bot.data.db import Database
|
||||
from bot.data.polymarket import PolymarketClient
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Clip probabilities before log() so a (theoretical) hard 0/1 estimate on a
|
||||
# wrong outcome scores ~20.7 nats instead of infinity poisoning the mean.
|
||||
LOGLOSS_EPS = 1e-9
|
||||
|
||||
|
||||
async def fetch_outcomes(poly, market_ids: list[str]) -> list[dict]:
|
||||
"""Resolve archived markets against Gamma; returns only definitive ones.
|
||||
|
||||
Sequential on purpose: ~50 markets per invocation, and the Gamma API has
|
||||
no bulk endpoint. get_market_resolution() already returns None on API
|
||||
errors and resolved=False on open/disputed/ambiguous markets.
|
||||
"""
|
||||
resolved = []
|
||||
for market_id in market_ids:
|
||||
res = await poly.get_market_resolution(market_id)
|
||||
if res is None or not res.resolved or res.resolution is None:
|
||||
continue
|
||||
resolved.append({
|
||||
"market_id": market_id,
|
||||
"outcome": res.resolution,
|
||||
"resolved_at": res.resolved_at,
|
||||
})
|
||||
return resolved
|
||||
|
||||
|
||||
def _logloss(p: float, outcome: float) -> float:
|
||||
p = min(max(p, LOGLOSS_EPS), 1.0 - LOGLOSS_EPS)
|
||||
return -math.log(p) if outcome == 1.0 else -math.log(1.0 - p)
|
||||
|
||||
|
||||
def compute_calibration(rows: list[dict]) -> Optional[dict]:
|
||||
"""Score estimated_prob vs prior_prob against outcomes; None if no rows.
|
||||
|
||||
rows: dicts with market_id, category, estimated_prob, prior_prob, outcome.
|
||||
Pure function — the CLI feeds it DB rows, tests feed it literals.
|
||||
"""
|
||||
if not rows:
|
||||
return None
|
||||
|
||||
n = len(rows)
|
||||
brier_model = sum((r["estimated_prob"] - r["outcome"]) ** 2 for r in rows) / n
|
||||
brier_prior = sum((r["prior_prob"] - r["outcome"]) ** 2 for r in rows) / n
|
||||
logloss_model = sum(_logloss(r["estimated_prob"], r["outcome"]) for r in rows) / n
|
||||
logloss_prior = sum(_logloss(r["prior_prob"], r["outcome"]) for r in rows) / n
|
||||
|
||||
by_market: dict[str, list[dict]] = defaultdict(list)
|
||||
for r in rows:
|
||||
by_market[r["market_id"]].append(r)
|
||||
market_briers = [
|
||||
(
|
||||
sum((r["estimated_prob"] - r["outcome"]) ** 2 for r in mrows) / len(mrows),
|
||||
sum((r["prior_prob"] - r["outcome"]) ** 2 for r in mrows) / len(mrows),
|
||||
)
|
||||
for mrows in by_market.values()
|
||||
]
|
||||
brier_model_macro = sum(b[0] for b in market_briers) / len(market_briers)
|
||||
brier_prior_macro = sum(b[1] for b in market_briers) / len(market_briers)
|
||||
|
||||
by_category: dict[str, list[dict]] = defaultdict(list)
|
||||
for r in rows:
|
||||
by_category[r["category"] or "unknown"].append(r)
|
||||
per_category = {
|
||||
cat: {
|
||||
"n": len(crows),
|
||||
"markets": len({r["market_id"] for r in crows}),
|
||||
"brier_model": sum((r["estimated_prob"] - r["outcome"]) ** 2
|
||||
for r in crows) / len(crows),
|
||||
"brier_prior": sum((r["prior_prob"] - r["outcome"]) ** 2
|
||||
for r in crows) / len(crows),
|
||||
}
|
||||
for cat, crows in sorted(by_category.items())
|
||||
}
|
||||
|
||||
return {
|
||||
"n_evaluations": n,
|
||||
"n_markets": len(by_market),
|
||||
"brier_model": brier_model,
|
||||
"brier_prior": brier_prior,
|
||||
"brier_model_macro": brier_model_macro,
|
||||
"brier_prior_macro": brier_prior_macro,
|
||||
"logloss_model": logloss_model,
|
||||
"logloss_prior": logloss_prior,
|
||||
"per_category": per_category,
|
||||
}
|
||||
|
||||
|
||||
def print_report(metrics: Optional[dict], source: str) -> None:
|
||||
if metrics is None:
|
||||
print(f"calibration : no scorable rows yet for {source} "
|
||||
"(no archived estimate has a resolved outcome)")
|
||||
return
|
||||
print(f"calibration : {source} — {metrics['n_evaluations']} evaluations, "
|
||||
f"{metrics['n_markets']} markets")
|
||||
print(f"{'':14s}{'model':>10s}{'market':>10s}{'delta':>10s}")
|
||||
for label, m_key, p_key in (
|
||||
("Brier micro", "brier_model", "brier_prior"),
|
||||
("Brier macro", "brier_model_macro", "brier_prior_macro"),
|
||||
("logloss micro", "logloss_model", "logloss_prior"),
|
||||
):
|
||||
m, p = metrics[m_key], metrics[p_key]
|
||||
print(f" {label:12s}{m:>10.4f}{p:>10.4f}{m - p:>+10.4f}")
|
||||
print(" (delta < 0 = model beats the market price)")
|
||||
for cat, c in metrics["per_category"].items():
|
||||
print(f" {cat:12s}n={c['n']:<6d} markets={c['markets']:<3d} "
|
||||
f"brier model {c['brier_model']:.4f} vs market {c['brier_prior']:.4f}")
|
||||
|
||||
|
||||
async def _amain(args: argparse.Namespace) -> None:
|
||||
db = Database()
|
||||
await db.connect()
|
||||
try:
|
||||
if not args.metrics_only and args.run_id is None:
|
||||
pending = await db.get_unresolved_archived_market_ids()
|
||||
poly = PolymarketClient()
|
||||
try:
|
||||
resolved = await fetch_outcomes(poly, pending)
|
||||
finally:
|
||||
await poly.close()
|
||||
for out in resolved:
|
||||
await db.upsert_market_outcome(
|
||||
out["market_id"], out["outcome"], out["resolved_at"]
|
||||
)
|
||||
print(f"outcomes : {len(resolved)} newly resolved "
|
||||
f"(of {len(pending)} pending markets checked)")
|
||||
|
||||
coverage = await db.get_outcome_coverage()
|
||||
print(f"coverage : {coverage['resolved']}/{coverage['archived']} "
|
||||
"archived markets resolved")
|
||||
|
||||
if args.fetch_only:
|
||||
return
|
||||
rows = await db.get_calibration_rows(run_id=args.run_id)
|
||||
source = f"replay run {args.run_id}" if args.run_id else "R0 archive"
|
||||
print_report(compute_calibration(rows), source)
|
||||
finally:
|
||||
await db.disconnect()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="python -m bot.outcomes",
|
||||
description="Fetch market resolutions and score archived estimates.",
|
||||
)
|
||||
parser.add_argument("--fetch-only", action="store_true",
|
||||
help="only fetch/store outcomes, skip metrics")
|
||||
parser.add_argument("--metrics-only", action="store_true",
|
||||
help="skip the Gamma fetch, score what is stored")
|
||||
parser.add_argument("--run-id", default=None,
|
||||
help="score a replay run's re-estimates instead of "
|
||||
"the R0 archive (implies --metrics-only)")
|
||||
args = parser.parse_args()
|
||||
if args.fetch_only and args.metrics_only:
|
||||
parser.error("--fetch-only and --metrics-only are mutually exclusive")
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
)
|
||||
asyncio.run(_amain(args))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+394
@@ -0,0 +1,394 @@
|
||||
"""
|
||||
Replay R1 — replay core.
|
||||
|
||||
Re-executes BayesianStrategy.evaluate() over the R0 archive (signals +
|
||||
ext_snapshots + markets) and stores the outcome in replay_runs /
|
||||
replay_decisions.
|
||||
|
||||
Determinism contract: evaluate() is a pure function of
|
||||
(market, ext, occupied_families, as_of) plus the news client, so a replay
|
||||
rebuilds exactly those four inputs from the archive:
|
||||
|
||||
market — metadata from `markets`, per-cycle price/volume from `signals`
|
||||
ext — the cycle's `ext_snapshots` row
|
||||
families — a family-skipped row replays with its own family_key occupied;
|
||||
every other row replays with no occupancy (the recorded
|
||||
skip_reason already reflects the original portfolio state)
|
||||
as_of — the archived cycle_ts (clock injection, Replay R1)
|
||||
|
||||
GNews is never called: ReplayNews feeds back the archived news_sentiment.
|
||||
The per-cycle query budget is bypassed (reset before every market) because
|
||||
the archived sentiment already encodes the budget's effect — a
|
||||
budget-skipped market was recorded with sentiment 0.0.
|
||||
|
||||
Manifold and the DB are not wired into the replayed strategy (manifold=None,
|
||||
db=None): the signal is observational-only in production (feat_mfld_lo is
|
||||
always 0.0 in the archive), so the replay reproduces decisions without
|
||||
touching cooldowns or audit tables. If MANIFOLD_SIGNAL_ENABLED is ever
|
||||
turned on, replayed decisions will diverge from recorded ones and the
|
||||
matched/mismatch_field columns will say so.
|
||||
|
||||
Run tagging: every run stores the git sha and a hash of the strategy
|
||||
constants. Same config_hash vs the archive = determinism check (expect 0
|
||||
mismatches, modulo UTC-day-boundary crossings between cycle_ts and the
|
||||
original wall-clock). Different config_hash = counterfactual run.
|
||||
|
||||
CLI:
|
||||
python -m bot.replay --from 2026-07-02T00:00:00Z --to 2026-07-03 --note "..."
|
||||
"""
|
||||
import argparse
|
||||
import asyncio
|
||||
import hashlib
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import uuid
|
||||
from collections import Counter
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from typing import Optional
|
||||
|
||||
import bot.strategy.bayesian as bayesian
|
||||
from bot.data.db import Database
|
||||
from bot.data.external import ExternalSignals
|
||||
from bot.data.polymarket import Market
|
||||
from bot.strategy.bayesian import BayesianStrategy
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
# Absolute float tolerance for recorded-vs-replayed comparison. Archived
|
||||
# values are float8 (exact IEEE-754 round-trip of Python floats), so any real
|
||||
# divergence is far larger than this.
|
||||
FLOAT_TOL = 1e-9
|
||||
|
||||
# Strategy constants that define a replay configuration. Hashed into
|
||||
# replay_runs.config_hash; read from the module at call time so a
|
||||
# counterfactual run can monkeypatch them and be tagged distinctly.
|
||||
CONFIG_KEYS = (
|
||||
"SPREAD_ESTIMATE",
|
||||
"COMMISSION_RATE",
|
||||
"MIN_CONFIDENCE",
|
||||
"NEWS_LOGODDS_WEIGHT",
|
||||
"MANIFOLD_LOGODDS_WEIGHT",
|
||||
"MANIFOLD_SIGNAL_ENABLED",
|
||||
"NEWS_GUARDRAIL_ENABLED",
|
||||
"MAX_NEWS_ONLY_PROB_SHIFT",
|
||||
"NEWS_MATERIAL_LOGODDS_THRESHOLD",
|
||||
"MAX_NEWS_QUERIES_PER_CYCLE",
|
||||
)
|
||||
|
||||
# Rows recorded outside evaluate() (via record_skip) carry no decision fields;
|
||||
# the replay still re-evaluates them for calibration but cannot compare.
|
||||
NON_COMPARABLE_SKIPS = {"reentry_guard"}
|
||||
|
||||
|
||||
def strategy_config() -> dict:
|
||||
return {k: getattr(bayesian, k) for k in CONFIG_KEYS}
|
||||
|
||||
|
||||
def strategy_config_hash() -> str:
|
||||
blob = json.dumps(strategy_config(), sort_keys=True)
|
||||
return hashlib.sha256(blob.encode()).hexdigest()[:12]
|
||||
|
||||
|
||||
def _git_sha() -> str:
|
||||
sha = os.getenv("GIT_SHA", "")
|
||||
if sha:
|
||||
return sha
|
||||
try:
|
||||
return subprocess.run(
|
||||
["git", "rev-parse", "--short", "HEAD"],
|
||||
capture_output=True, text=True, timeout=5,
|
||||
).stdout.strip() or "unknown"
|
||||
except (OSError, subprocess.SubprocessError):
|
||||
return "unknown"
|
||||
|
||||
|
||||
class ReplayNews:
|
||||
"""NewsClient stand-in that feeds archived sentiment back into evaluate().
|
||||
|
||||
No HTTP, no cache: the engine sets `sentiment` to the archived value
|
||||
before each evaluate() call. Values below evaluate()'s 0.05 materiality
|
||||
threshold were archived as 0.0, so the round-trip is exact.
|
||||
"""
|
||||
enabled = True
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.sentiment: float = 0.0
|
||||
|
||||
async def get_sentiment(self, question: str) -> float:
|
||||
return self.sentiment
|
||||
|
||||
def get_freshness(self, question: str) -> float:
|
||||
return 1.0 # only used by gnews_priority(), which replay never calls
|
||||
|
||||
|
||||
def build_ext(snapshot: dict) -> ExternalSignals:
|
||||
"""Rebuild the ExternalSignals a cycle was evaluated against."""
|
||||
return ExternalSignals(
|
||||
btc_price=snapshot["btc_price"],
|
||||
btc_change_24h=snapshot["btc_change_24h"],
|
||||
eth_price=snapshot["eth_price"],
|
||||
eth_change_24h=snapshot["eth_change_24h"],
|
||||
btc_dominance=snapshot["btc_dominance"],
|
||||
fear_greed_index=snapshot["fear_greed_index"],
|
||||
fear_greed_label=snapshot["fear_greed_label"],
|
||||
total_market_cap_change=snapshot["total_market_cap_change"],
|
||||
valid=snapshot["valid"],
|
||||
)
|
||||
|
||||
|
||||
def build_market(market_row: dict, signal_row: dict) -> Market:
|
||||
"""Rebuild a Market: metadata from `markets`, per-cycle state from `signals`.
|
||||
|
||||
Token ids are irrelevant to evaluate() and left empty; no_price is the
|
||||
YES complement (evaluate() never reads it either).
|
||||
"""
|
||||
yes_price = signal_row["polymarket_price"]
|
||||
return Market(
|
||||
id=market_row["id"],
|
||||
condition_id=market_row["condition_id"] or "",
|
||||
question=market_row["question"],
|
||||
yes_token_id="",
|
||||
no_token_id="",
|
||||
yes_price=yes_price,
|
||||
no_price=1.0 - yes_price,
|
||||
volume_24h=signal_row["volume_24h"] or 0.0,
|
||||
end_date=market_row["end_date"] or "",
|
||||
active=True,
|
||||
category=signal_row["category"] or (market_row["category"] or ""),
|
||||
)
|
||||
|
||||
|
||||
def _compare(recorded: dict, replayed: dict) -> Optional[str]:
|
||||
"""Return the first field where replayed diverges from recorded, or None."""
|
||||
if recorded["skip_reason"] != replayed["skip_reason"]:
|
||||
return "skip_reason"
|
||||
for field in ("prior_prob", "estimated_prob", "raw_final_prob",
|
||||
"edge_net", "confidence"):
|
||||
a, b = recorded[field], replayed[field]
|
||||
if a is None and b is None:
|
||||
continue
|
||||
if a is None or b is None or abs(a - b) > FLOAT_TOL:
|
||||
return field
|
||||
if recorded["direction"] != replayed["direction"]:
|
||||
return "direction"
|
||||
return None
|
||||
|
||||
|
||||
async def replay_cycle(
|
||||
cycle_ts: datetime,
|
||||
snapshot: dict,
|
||||
signal_rows: list[dict],
|
||||
market_rows: dict[str, dict],
|
||||
) -> list[dict]:
|
||||
"""Re-evaluate one archived cycle; returns one decision dict per row.
|
||||
|
||||
Pure with respect to the DB — everything it needs is passed in, so tests
|
||||
can drive it with synthetic rows.
|
||||
"""
|
||||
news = ReplayNews()
|
||||
strategy = BayesianStrategy(news=news, manifold=None, db=None)
|
||||
ext = build_ext(snapshot)
|
||||
decisions: list[dict] = []
|
||||
|
||||
for row in signal_rows:
|
||||
recorded_skip = row["skip_reason"]
|
||||
decision = {
|
||||
"cycle_ts": cycle_ts,
|
||||
"market_id": row["market_id"],
|
||||
"skip_reason": None,
|
||||
"prior_prob": None,
|
||||
"estimated_prob": None,
|
||||
"raw_final_prob": None,
|
||||
"edge_gross": None,
|
||||
"edge_net": None,
|
||||
"regime_min_edge": None,
|
||||
"days_to_resolution": None,
|
||||
"confidence": None,
|
||||
"direction": None,
|
||||
"would_trade": None,
|
||||
"recorded_skip_reason": recorded_skip,
|
||||
"matched": None,
|
||||
"mismatch_field": None,
|
||||
}
|
||||
|
||||
market_row = market_rows.get(row["market_id"])
|
||||
if market_row is None:
|
||||
# Should not happen (R0 upserts markets every cycle) — record the
|
||||
# gap instead of crashing the run.
|
||||
decision["matched"] = False
|
||||
decision["mismatch_field"] = "market_missing"
|
||||
decisions.append(decision)
|
||||
continue
|
||||
|
||||
market = build_market(market_row, row)
|
||||
# A family-skipped row replays against its own occupied family; all
|
||||
# other rows replay unoccupied — their recorded skip_reason already
|
||||
# reflects whatever portfolio state existed, and evaluate() checks
|
||||
# the family gate before anything portfolio-dependent.
|
||||
families = (
|
||||
{row["family_key"]}
|
||||
if recorded_skip == "family" and row["family_key"]
|
||||
else set()
|
||||
)
|
||||
news.sentiment = row["news_sentiment"] or 0.0
|
||||
# Bypass the per-cycle GNews budget: archived sentiment already
|
||||
# encodes it (budget-skipped markets were recorded with 0.0).
|
||||
strategy._news_queries_this_cycle = 0
|
||||
|
||||
signal = await strategy.evaluate(market, ext, families, as_of=cycle_ts)
|
||||
rec = strategy.drain_cycle_records()[-1]
|
||||
|
||||
decision.update(
|
||||
skip_reason=rec["skip_reason"],
|
||||
prior_prob=rec["prior_prob"],
|
||||
estimated_prob=rec["estimated_prob"],
|
||||
raw_final_prob=rec["raw_final_prob"],
|
||||
edge_gross=rec["edge_gross"],
|
||||
edge_net=rec["edge_net"],
|
||||
regime_min_edge=rec["regime_min_edge"],
|
||||
days_to_resolution=rec["days_to_resolution"],
|
||||
confidence=rec["confidence"],
|
||||
direction=rec["direction"],
|
||||
would_trade=signal is not None,
|
||||
)
|
||||
if recorded_skip in NON_COMPARABLE_SKIPS:
|
||||
decision["matched"] = None # re-evaluated for calibration only
|
||||
else:
|
||||
mismatch = _compare(row, rec)
|
||||
decision["matched"] = mismatch is None
|
||||
decision["mismatch_field"] = mismatch
|
||||
decisions.append(decision)
|
||||
|
||||
return decisions
|
||||
|
||||
|
||||
async def run_replay(
|
||||
db: Database,
|
||||
from_ts: datetime,
|
||||
to_ts: datetime,
|
||||
note: str = "",
|
||||
limit_cycles: Optional[int] = None,
|
||||
) -> dict:
|
||||
"""Replay every archived cycle in [from_ts, to_ts) and persist the run.
|
||||
|
||||
Returns the replay_runs row (plus a mismatch_fields Counter) for reporting.
|
||||
"""
|
||||
run_id = str(uuid.uuid4())
|
||||
cycles = await db.get_replay_cycles(from_ts, to_ts)
|
||||
if limit_cycles:
|
||||
cycles = cycles[:limit_cycles]
|
||||
|
||||
decisions_total = 0
|
||||
matched = 0
|
||||
mismatched = 0
|
||||
mismatch_fields: Counter = Counter()
|
||||
skipped_cycles = 0
|
||||
|
||||
for cycle_ts in cycles:
|
||||
snapshot = await db.get_ext_snapshot(cycle_ts)
|
||||
if snapshot is None:
|
||||
skipped_cycles += 1
|
||||
log.warning("Replay: no ext_snapshot for cycle %s — skipped", cycle_ts)
|
||||
continue
|
||||
signal_rows = await db.get_cycle_signal_rows(cycle_ts)
|
||||
market_rows = await db.get_markets_by_ids(
|
||||
[r["market_id"] for r in signal_rows]
|
||||
)
|
||||
decisions = await replay_cycle(cycle_ts, snapshot, signal_rows, market_rows)
|
||||
await db.save_replay_decisions(run_id, decisions)
|
||||
|
||||
decisions_total += len(decisions)
|
||||
for d in decisions:
|
||||
if d["matched"] is True:
|
||||
matched += 1
|
||||
elif d["matched"] is False:
|
||||
mismatched += 1
|
||||
mismatch_fields[d["mismatch_field"]] += 1
|
||||
|
||||
run = {
|
||||
"run_id": run_id,
|
||||
"git_sha": _git_sha(),
|
||||
"config_hash": strategy_config_hash(),
|
||||
"config_json": json.dumps(strategy_config(), sort_keys=True),
|
||||
"from_ts": from_ts,
|
||||
"to_ts": to_ts,
|
||||
"cycles": len(cycles) - skipped_cycles,
|
||||
"decisions": decisions_total,
|
||||
"matched": matched,
|
||||
"mismatched": mismatched,
|
||||
"note": note,
|
||||
}
|
||||
await db.save_replay_run(run)
|
||||
run["mismatch_fields"] = dict(mismatch_fields)
|
||||
run["skipped_cycles"] = skipped_cycles
|
||||
return run
|
||||
|
||||
|
||||
def _parse_ts(value: str) -> datetime:
|
||||
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt
|
||||
|
||||
|
||||
async def _amain(args: argparse.Namespace) -> None:
|
||||
db = Database()
|
||||
await db.connect()
|
||||
try:
|
||||
run = await run_replay(
|
||||
db,
|
||||
from_ts=args.from_ts,
|
||||
to_ts=args.to_ts,
|
||||
note=args.note,
|
||||
limit_cycles=args.limit_cycles,
|
||||
)
|
||||
finally:
|
||||
await db.disconnect()
|
||||
|
||||
comparable = run["matched"] + run["mismatched"]
|
||||
print(f"run_id : {run['run_id']}")
|
||||
print(f"git_sha : {run['git_sha']} config_hash: {run['config_hash']}")
|
||||
print(f"window : {run['from_ts'].isoformat()} → {run['to_ts'].isoformat()}")
|
||||
print(f"cycles : {run['cycles']} (skipped: {run['skipped_cycles']})")
|
||||
print(f"decisions : {run['decisions']} ({comparable} comparable)")
|
||||
print(f"matched : {run['matched']}")
|
||||
print(f"mismatched : {run['mismatched']}")
|
||||
if run["mismatch_fields"]:
|
||||
for field, count in sorted(run["mismatch_fields"].items(), key=lambda x: -x[1]):
|
||||
print(f" {field}: {count}")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="python -m bot.replay",
|
||||
description="Replay archived trading cycles through the current strategy.",
|
||||
)
|
||||
now = datetime.now(timezone.utc)
|
||||
parser.add_argument(
|
||||
"--from", dest="from_ts", type=_parse_ts,
|
||||
default=now - timedelta(hours=24),
|
||||
help="window start, ISO-8601 (default: 24h ago)",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--to", dest="to_ts", type=_parse_ts, default=now,
|
||||
help="window end, ISO-8601, exclusive (default: now)",
|
||||
)
|
||||
parser.add_argument("--note", default="", help="free-text tag for replay_runs")
|
||||
parser.add_argument(
|
||||
"--limit-cycles", type=int, default=None,
|
||||
help="replay at most N cycles (smoke runs)",
|
||||
)
|
||||
args = parser.parse_args()
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
||||
)
|
||||
# evaluate() logs one INFO line per market — thousands per replay window.
|
||||
logging.getLogger("bot.strategy.bayesian").setLevel(logging.WARNING)
|
||||
asyncio.run(_amain(args))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
+267
-17
@@ -84,6 +84,27 @@ def _env_bool(name: str, default: bool) -> bool:
|
||||
MANIFOLD_SIGNAL_ENABLED = _env_bool("MANIFOLD_SIGNAL_ENABLED", False)
|
||||
MANIFOLD_AUDIT_ENABLED = _env_bool("MANIFOLD_AUDIT_ENABLED", True)
|
||||
|
||||
# ── GNews guardrail (catastrophic fuse) ────────────────────────────────────────
|
||||
# Post-mortem NVIDIA 631181: a single strong signal (legacy Manifold 0.13 at
|
||||
# weight 0.6) flipped a 0.845 market to 0.431 and lost. With Manifold now
|
||||
# observational-only and macro signals gated behind is_non_price, GNews
|
||||
# (weight 1.5) is the only live signal that can move politics markets 20-30 pp
|
||||
# against the order-book consensus. This is NOT a fine calibration — it is a
|
||||
# fuse against the extreme case: one uncorroborated signal violently inverting
|
||||
# the market.
|
||||
#
|
||||
# NEWS_GUARDRAIL_ENABLED: master switch for the fuse.
|
||||
# MAX_NEWS_ONLY_PROB_SHIFT: when GNews is the ONLY material signal, the final
|
||||
# probability is clamped to prior ± this value. 0.25 still allows a 25 pp
|
||||
# move (edge_net 0.21 after costs) — trades still happen, sizing is bounded.
|
||||
# NEWS_MATERIAL_LOGODDS_THRESHOLD: a signal counts as *material* iff its
|
||||
# |log-odds contribution| >= this value. Below it, a signal is noise and
|
||||
# does NOT count as corroboration. If ANY other signal (fg, momentum,
|
||||
# btc_dom, manifold) is material, the fuse does not apply.
|
||||
NEWS_GUARDRAIL_ENABLED = _env_bool("NEWS_GUARDRAIL_ENABLED", True)
|
||||
MAX_NEWS_ONLY_PROB_SHIFT = float(os.getenv("MAX_NEWS_ONLY_PROB_SHIFT", "0.25"))
|
||||
NEWS_MATERIAL_LOGODDS_THRESHOLD = float(os.getenv("NEWS_MATERIAL_LOGODDS_THRESHOLD", "0.10"))
|
||||
|
||||
# GNews free tier: 100 req/day. We limit to 5 queries per trading cycle
|
||||
# (politics markets only) and rely on 6 h cache to stay within budget.
|
||||
MAX_NEWS_QUERIES_PER_CYCLE = 5
|
||||
@@ -146,15 +167,22 @@ def _regime_min_edge(category: str, days_to_resolution: int) -> float:
|
||||
return 0.10 # tech, crypto/finance, events, default
|
||||
|
||||
|
||||
def _days_to_resolution(end_date: str) -> int:
|
||||
"""Return calendar days until market resolution, or 30 if unknown."""
|
||||
def _days_to_resolution(end_date: str, as_of: Optional[datetime] = None) -> int:
|
||||
"""Return calendar days until market resolution, or 30 if unknown.
|
||||
|
||||
as_of (Replay R1): reference clock for the computation. None (production)
|
||||
means wall-clock now; a replay run passes the archived cycle_ts so
|
||||
days-to-resolution — and therefore the regime edge threshold — is computed
|
||||
against the moment the decision was originally made.
|
||||
"""
|
||||
if not end_date:
|
||||
return 30 # conservative: treat as medium-term
|
||||
try:
|
||||
dt = datetime.fromisoformat(end_date.replace("Z", "+00:00"))
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
days = (dt - datetime.now(timezone.utc)).days
|
||||
now = as_of if as_of is not None else datetime.now(timezone.utc)
|
||||
days = (dt - now).days
|
||||
return max(0, days)
|
||||
except (ValueError, TypeError):
|
||||
return 30
|
||||
@@ -179,6 +207,42 @@ def has_token(text: str, token: str) -> bool:
|
||||
# Phase 3 — GNews priority scoring
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def apply_news_guardrail(
|
||||
prior: float,
|
||||
raw_final_prob: float,
|
||||
feat_news_lo: float,
|
||||
other_feats_lo: tuple[float, ...],
|
||||
) -> tuple[float, bool]:
|
||||
"""
|
||||
GNews guardrail (catastrophic fuse).
|
||||
|
||||
Clamp raw_final_prob to prior ± MAX_NEWS_ONLY_PROB_SHIFT when ALL hold:
|
||||
1. NEWS_GUARDRAIL_ENABLED
|
||||
2. |feat_news_lo| >= NEWS_MATERIAL_LOGODDS_THRESHOLD (news is material)
|
||||
3. every other signal's |log-odds contribution| is below the threshold
|
||||
(GNews is the ONLY material signal — no corroboration)
|
||||
|
||||
Returns (final_prob, guardrail_applied). guardrail_applied is True only
|
||||
when the clamp actually changed the value; a raw_final_prob already inside
|
||||
the band passes through untouched with applied=False.
|
||||
|
||||
Module globals are read at call time so tests can monkeypatch them.
|
||||
"""
|
||||
if not NEWS_GUARDRAIL_ENABLED:
|
||||
return raw_final_prob, False
|
||||
if abs(feat_news_lo) < NEWS_MATERIAL_LOGODDS_THRESHOLD:
|
||||
return raw_final_prob, False
|
||||
if any(abs(v) >= NEWS_MATERIAL_LOGODDS_THRESHOLD for v in other_feats_lo):
|
||||
return raw_final_prob, False # corroborated — fuse does not apply
|
||||
clamped = min(
|
||||
max(raw_final_prob, prior - MAX_NEWS_ONLY_PROB_SHIFT),
|
||||
prior + MAX_NEWS_ONLY_PROB_SHIFT,
|
||||
)
|
||||
if clamped == raw_final_prob:
|
||||
return raw_final_prob, False
|
||||
return clamped, True
|
||||
|
||||
|
||||
def gnews_priority(market: Market, news: "NewsClient") -> float:
|
||||
"""
|
||||
Score a market for GNews query priority (higher = more valuable to query).
|
||||
@@ -300,6 +364,13 @@ class BayesianStrategy:
|
||||
# (edge_gross, edge_net, regime_min) for every market that reached the
|
||||
# edge computation stage (passed prior-extreme, family, unsupported filters)
|
||||
self._evaluated_edges: list[tuple[float, float, float]] = []
|
||||
# GNews guardrail observability — only markets with material news
|
||||
self._news_shifts: list[float] = [] # final_prob - prior, signed
|
||||
self._news_guardrail_applied: int = 0
|
||||
self._news_changed_decisions: int = 0
|
||||
# Replay R0: per-(market, cycle) decision records, drained by main.py
|
||||
# into the signals table after each evaluation loop.
|
||||
self._cycle_records: list[dict] = []
|
||||
|
||||
def reset_cycle(self) -> None:
|
||||
"""Call once at the start of each trading cycle to reset per-cycle counters."""
|
||||
@@ -311,6 +382,54 @@ class BayesianStrategy:
|
||||
self._manifold_fetched = 0
|
||||
self._manifold_on_trade = 0
|
||||
self._evaluated_edges = []
|
||||
self._news_shifts = []
|
||||
self._news_guardrail_applied = 0
|
||||
self._news_changed_decisions = 0
|
||||
self._cycle_records = []
|
||||
|
||||
def record_skip(self, market: Market, skip_reason: str) -> None:
|
||||
"""Record a skip decided OUTSIDE evaluate() (e.g. reentry_guard in main)."""
|
||||
self._record(market, skip_reason=skip_reason)
|
||||
|
||||
def drain_cycle_records(self) -> list[dict]:
|
||||
"""Return and clear this cycle's decision records (Replay R0)."""
|
||||
records, self._cycle_records = self._cycle_records, []
|
||||
return records
|
||||
|
||||
def _record(self, market: Market, skip_reason: Optional[str], **fields) -> None:
|
||||
"""Append one decision record. Early skips leave most fields None —
|
||||
the archive still shows the market existed and why it went no further."""
|
||||
rec = {
|
||||
"market_id": market.id,
|
||||
"polymarket_price": market.yes_price,
|
||||
"category": market.category,
|
||||
"volume_24h": market.volume_24h,
|
||||
"skip_reason": skip_reason,
|
||||
"family_key": None,
|
||||
"prior_prob": None,
|
||||
"estimated_prob": None,
|
||||
"raw_final_prob": None,
|
||||
"edge_gross": None,
|
||||
"edge_net": None,
|
||||
"regime_min_edge": None,
|
||||
"days_to_resolution": None,
|
||||
"confidence": None,
|
||||
"direction": None,
|
||||
"passed_gross": None,
|
||||
"passed_net": None,
|
||||
"news_sentiment": None,
|
||||
"news_budget_skipped": None,
|
||||
"guardrail_applied": None,
|
||||
"guardrail_changed_decision": None,
|
||||
"feat_fg_lo": None,
|
||||
"feat_mom_lo": None,
|
||||
"feat_news_lo": None,
|
||||
"feat_mfld_lo": None,
|
||||
"feat_btc_dom_lo": None,
|
||||
"acted_on": False,
|
||||
}
|
||||
rec.update(fields)
|
||||
self._cycle_records.append(rec)
|
||||
|
||||
def get_cycle_stats(self) -> dict:
|
||||
"""Return per-cycle counters for the [CYCLE SUMMARY] log block."""
|
||||
@@ -330,6 +449,14 @@ class BayesianStrategy:
|
||||
"gross_gt_004": sum(1 for g in all_gross if g > 0.04),
|
||||
"manifold_matches_accepted": self._manifold_on_trade,
|
||||
"manifold_matches_rejected": self._manifold_fetched - self._manifold_on_trade,
|
||||
# GNews guardrail — markets with |news_lo| >= NEWS_MATERIAL_LOGODDS_THRESHOLD
|
||||
"news_with_material": len(self._news_shifts),
|
||||
"news_avg_shift": (sum(self._news_shifts) / len(self._news_shifts))
|
||||
if self._news_shifts else 0.0,
|
||||
"news_max_shift": max(self._news_shifts, key=abs)
|
||||
if self._news_shifts else 0.0,
|
||||
"news_guardrail_applied": self._news_guardrail_applied,
|
||||
"news_changed_decisions": self._news_changed_decisions,
|
||||
}
|
||||
|
||||
async def evaluate(
|
||||
@@ -337,10 +464,17 @@ class BayesianStrategy:
|
||||
market: Market,
|
||||
ext: ExternalSignals,
|
||||
occupied_families: set[str],
|
||||
as_of: Optional[datetime] = None,
|
||||
) -> Optional[TradingSignal]:
|
||||
"""
|
||||
Evaluate a market and return a TradingSignal if actionable.
|
||||
|
||||
as_of (Replay R1): clock injection — None in production (wall-clock
|
||||
now); a replay passes the archived cycle_ts so the regime threshold
|
||||
matches the original decision moment. Only days-to-resolution
|
||||
depends on the clock; everything else is a pure function of
|
||||
(market, ext, occupied_families) and the news/manifold clients.
|
||||
|
||||
Returns None with a structured log line in all skip cases.
|
||||
Skip reasons (Phase 5 observability):
|
||||
SKIP_UNSUPPORTED — category not supported
|
||||
@@ -395,6 +529,7 @@ class BayesianStrategy:
|
||||
"SKIP_UNSUPPORTED %-50s | cat=%r",
|
||||
market.question[:50], category,
|
||||
)
|
||||
self._record(market, skip_reason="unsupported")
|
||||
return None
|
||||
|
||||
if not ext.valid:
|
||||
@@ -402,6 +537,7 @@ class BayesianStrategy:
|
||||
"SKIP_NO_SIGNALS %-50s | reason=external data unavailable",
|
||||
market.question[:50],
|
||||
)
|
||||
self._record(market, skip_reason="no_signals")
|
||||
return None
|
||||
|
||||
# ── Phase 1: prior + prior-extreme filter ────────────────────────────
|
||||
@@ -413,6 +549,7 @@ class BayesianStrategy:
|
||||
"SKIP_PRIOR_EXTREME %-50s | cat=%-12s | prior=%.3f | reason=prior<0.08",
|
||||
market.question[:50], category, market.yes_price,
|
||||
)
|
||||
self._record(market, skip_reason="prior_extreme", prior_prob=prior)
|
||||
return None
|
||||
if market.yes_price > 0.92:
|
||||
self._skip_prior_extreme += 1
|
||||
@@ -420,6 +557,7 @@ class BayesianStrategy:
|
||||
"SKIP_PRIOR_EXTREME %-50s | cat=%-12s | prior=%.3f | reason=prior>0.92",
|
||||
market.question[:50], category, market.yes_price,
|
||||
)
|
||||
self._record(market, skip_reason="prior_extreme", prior_prob=prior)
|
||||
return None
|
||||
|
||||
# ── Phase 2: family deduplication ────────────────────────────────────
|
||||
@@ -430,10 +568,11 @@ class BayesianStrategy:
|
||||
"SKIP_FAMILY %-50s | cat=%-12s | family=%s",
|
||||
market.question[:50], category, family,
|
||||
)
|
||||
self._record(market, skip_reason="family", prior_prob=prior, family_key=family)
|
||||
return None
|
||||
|
||||
# ── Phase 4: regime min-edge ─────────────────────────────────────────
|
||||
days = _days_to_resolution(market.end_date)
|
||||
days = _days_to_resolution(market.end_date, as_of)
|
||||
regime_min = _regime_min_edge(category, days)
|
||||
|
||||
# ── Bayesian probability estimation ──────────────────────────────────
|
||||
@@ -503,14 +642,24 @@ class BayesianStrategy:
|
||||
# Phase 3: caller has pre-sorted markets by gnews_priority() so the
|
||||
# highest-value markets reach this block first.
|
||||
news_log_adj = 0.0
|
||||
if is_politics and self._news is not None:
|
||||
news_sentiment = 0.0
|
||||
# Replay R0: True when GNews was never consulted for this market this
|
||||
# cycle (budget exhausted) — a replay must not read feat_news_lo=0.0 as
|
||||
# "there was no news".
|
||||
news_budget_skipped = False
|
||||
# self._news.enabled gates the whole block: with no GNews API key the
|
||||
# client is a no-op, so we must not consume (or report) query budget for
|
||||
# it — see NewsClient.enabled.
|
||||
if is_politics and self._news is not None and self._news.enabled:
|
||||
if self._news_queries_this_cycle < MAX_NEWS_QUERIES_PER_CYCLE:
|
||||
self._news_queries_this_cycle += 1
|
||||
sentiment = await self._news.get_sentiment(market.question)
|
||||
if abs(sentiment) > 0.05:
|
||||
news_sentiment = sentiment
|
||||
news_log_adj = sentiment * NEWS_LOGODDS_WEIGHT
|
||||
sources.append(f"GNews: {sentiment:+.2f}")
|
||||
else:
|
||||
news_budget_skipped = True
|
||||
log.info(
|
||||
"SKIP_GNEWS_PRIORITY %-50s | reason=cycle budget %d reached",
|
||||
market.question[:50], MAX_NEWS_QUERIES_PER_CYCLE,
|
||||
@@ -649,8 +798,31 @@ class BayesianStrategy:
|
||||
# Posterior via log-odds updating
|
||||
log_odds_prior = math.log(prior / (1 - prior))
|
||||
total_adj = sum(adjustments)
|
||||
estimated_prob = _sigmoid(log_odds_prior + total_adj * 2 + news_log_adj + manifold_log_adj)
|
||||
estimated_prob = max(0.05, min(0.95, estimated_prob))
|
||||
# raw_final_prob: posterior BEFORE the news guardrail.
|
||||
raw_final_prob = _sigmoid(log_odds_prior + total_adj * 2 + news_log_adj + manifold_log_adj)
|
||||
raw_final_prob = max(0.05, min(0.95, raw_final_prob))
|
||||
|
||||
# Per-feature log-odds contributions (Phase 6) — computed here (not
|
||||
# after the edge gate) because the guardrail below needs them to decide
|
||||
# signal materiality.
|
||||
# fg / mom / btc_dom: probability-delta × 2 → log-odds.
|
||||
# news / mfld: already log-odds (LOGODDS_WEIGHT already applied).
|
||||
feat_fg_lo = _fg_contribution * 2
|
||||
feat_mom_lo = _momentum_contribution * 2
|
||||
feat_news_lo = news_log_adj
|
||||
feat_mfld_lo = manifold_log_adj
|
||||
feat_btc_dom_lo = _btc_dom_contribution * 2
|
||||
|
||||
# ── GNews guardrail (catastrophic fuse) ──────────────────────────────
|
||||
# When GNews is the ONLY material signal, clamp the posterior to
|
||||
# prior ± MAX_NEWS_ONLY_PROB_SHIFT. estimated_prob (post-guardrail) is
|
||||
# what edge/trading uses; raw_final_prob is kept for observability.
|
||||
estimated_prob, news_guardrail_applied = apply_news_guardrail(
|
||||
prior,
|
||||
raw_final_prob,
|
||||
feat_news_lo,
|
||||
(feat_fg_lo, feat_mom_lo, feat_btc_dom_lo, feat_mfld_lo),
|
||||
)
|
||||
|
||||
# ── Phase 1: edge_gross and edge_net ─────────────────────────────────
|
||||
raw_edge = estimated_prob - market.yes_price
|
||||
@@ -672,15 +844,6 @@ class BayesianStrategy:
|
||||
if manifold_log_adj != 0.0:
|
||||
confidence = min(confidence_cap, confidence + 0.08)
|
||||
|
||||
# Per-feature log-odds contributions (Phase 6).
|
||||
# fg / mom / btc_dom: probability-delta × 2 → log-odds.
|
||||
# news / mfld: already log-odds (LOGODDS_WEIGHT already applied).
|
||||
feat_fg_lo = _fg_contribution * 2
|
||||
feat_mom_lo = _momentum_contribution * 2
|
||||
feat_news_lo = news_log_adj
|
||||
feat_mfld_lo = manifold_log_adj
|
||||
feat_btc_dom_lo = _btc_dom_contribution * 2
|
||||
|
||||
feat_str = (
|
||||
f"fg_lo={feat_fg_lo:+.4f} mom_lo={feat_mom_lo:+.4f} "
|
||||
f"news_lo={feat_news_lo:+.4f} mfld_lo={feat_mfld_lo:+.4f} "
|
||||
@@ -692,6 +855,80 @@ class BayesianStrategy:
|
||||
passed_net = edge_net >= regime_min
|
||||
can_trade = passed_net and confidence >= MIN_CONFIDENCE
|
||||
|
||||
# ── Guardrail decision impact ────────────────────────────────────────
|
||||
# True when the un-clamped posterior's edge crossed the regime gate but
|
||||
# the clamped one no longer does — i.e. the fuse PREVENTED a trade.
|
||||
# Confidence is invariant under the clamp (it depends only on signal
|
||||
# agreement), so the edge gate is the only component that can flip.
|
||||
guardrail_changed_trade_decision = False
|
||||
if news_guardrail_applied:
|
||||
raw_edge_net = abs(raw_final_prob - market.yes_price) - TOTAL_COST_RATE
|
||||
guardrail_changed_trade_decision = (
|
||||
raw_edge_net >= regime_min and edge_net < regime_min
|
||||
)
|
||||
|
||||
# ── Guardrail observability — ONLY markets with material news ───────
|
||||
# Gated on materiality so the ~145 markets/cycle without news don't
|
||||
# flood the logs. posterior_before_news = everything except GNews.
|
||||
news_is_material = abs(feat_news_lo) >= NEWS_MATERIAL_LOGODDS_THRESHOLD
|
||||
if news_is_material:
|
||||
posterior_before_news = max(0.05, min(0.95, _sigmoid(
|
||||
log_odds_prior + total_adj * 2 + manifold_log_adj
|
||||
)))
|
||||
self._news_shifts.append(estimated_prob - prior)
|
||||
if news_guardrail_applied:
|
||||
self._news_guardrail_applied += 1
|
||||
if guardrail_changed_trade_decision:
|
||||
self._news_changed_decisions += 1
|
||||
log.info(
|
||||
"NEWS_MATERIAL %-50s | cat=%-12s | family=%-28s | "
|
||||
"prior=%.3f | before_news=%.3f | raw=%.3f | final=%.3f | "
|
||||
"sent=%+.2f | news_lo=%+.4f | "
|
||||
"edge_before_news=%.3f | edge_after_raw=%.3f | edge_after_guardrail=%.3f | "
|
||||
"guardrail=%s | changed_decision=%s | max_shift=%.2f",
|
||||
market.question[:50], category, family,
|
||||
prior, posterior_before_news, raw_final_prob, estimated_prob,
|
||||
news_sentiment, feat_news_lo,
|
||||
abs(posterior_before_news - market.yes_price),
|
||||
abs(raw_final_prob - market.yes_price),
|
||||
edge_gross,
|
||||
"applied" if news_guardrail_applied else "none",
|
||||
str(guardrail_changed_trade_decision).lower(),
|
||||
MAX_NEWS_ONLY_PROB_SHIFT,
|
||||
)
|
||||
|
||||
# Replay R0: full decision record — same fields for skip and trade paths.
|
||||
# skip_reason granularity: "edge_net" when the edge gate failed,
|
||||
# "confidence" when only the confidence gate blocked the trade.
|
||||
self._record(
|
||||
market,
|
||||
skip_reason=(
|
||||
None if can_trade
|
||||
else ("edge_net" if not passed_net else "confidence")
|
||||
),
|
||||
family_key=family,
|
||||
prior_prob=prior,
|
||||
estimated_prob=estimated_prob,
|
||||
raw_final_prob=raw_final_prob,
|
||||
edge_gross=edge_gross,
|
||||
edge_net=edge_net,
|
||||
regime_min_edge=regime_min,
|
||||
days_to_resolution=days,
|
||||
confidence=confidence,
|
||||
direction=direction,
|
||||
passed_gross=passed_gross,
|
||||
passed_net=passed_net,
|
||||
news_sentiment=news_sentiment,
|
||||
news_budget_skipped=news_budget_skipped,
|
||||
guardrail_applied=news_guardrail_applied,
|
||||
guardrail_changed_decision=guardrail_changed_trade_decision,
|
||||
feat_fg_lo=feat_fg_lo,
|
||||
feat_mom_lo=feat_mom_lo,
|
||||
feat_news_lo=feat_news_lo,
|
||||
feat_mfld_lo=feat_mfld_lo,
|
||||
feat_btc_dom_lo=feat_btc_dom_lo,
|
||||
)
|
||||
|
||||
if not can_trade:
|
||||
# Increment the appropriate edge-net counter
|
||||
if edge_net <= 0:
|
||||
@@ -720,8 +957,21 @@ class BayesianStrategy:
|
||||
)
|
||||
return None
|
||||
|
||||
# When GNews participated, expose raw vs final and the guardrail verdict
|
||||
# (Task 4 of the guardrail spec); otherwise keep the legacy format.
|
||||
if news_log_adj != 0.0:
|
||||
prob_part = (
|
||||
f"Prior=poly({prior:.3f}) → raw={raw_final_prob:.3f} "
|
||||
f"→ final={estimated_prob:.3f} | "
|
||||
f"GNews sent={news_sentiment:+.2f} | "
|
||||
f"guardrail={'applied' if news_guardrail_applied else 'none'} | "
|
||||
f"changed_decision={str(guardrail_changed_trade_decision).lower()} | "
|
||||
f"max_shift={MAX_NEWS_ONLY_PROB_SHIFT:.2f} | "
|
||||
)
|
||||
else:
|
||||
prob_part = f"Prior=poly({prior:.3f}) → estimate={estimated_prob:.3f} | "
|
||||
reasoning = (
|
||||
f"Prior=poly({prior:.3f}) → estimate={estimated_prob:.3f} | "
|
||||
prob_part +
|
||||
f"Poly price={market.yes_price:.3f} | "
|
||||
f"edge_gross={edge_gross:+.3f} | edge_net={edge_net:+.3f} | "
|
||||
f"regime_min={regime_min:.2f} | days={days} | "
|
||||
|
||||
+1
-1
@@ -1,7 +1,7 @@
|
||||
# Core
|
||||
asyncpg==0.29.0
|
||||
httpx==0.27.0
|
||||
fastapi==0.137.2
|
||||
fastapi==0.111.0
|
||||
uvicorn[standard]==0.29.0
|
||||
pydantic==2.7.0
|
||||
|
||||
|
||||
@@ -0,0 +1,247 @@
|
||||
"""
|
||||
Tests for the GNews guardrail (catastrophic fuse).
|
||||
|
||||
Post-mortem NVIDIA 631181: one uncorroborated signal at high weight flipped a
|
||||
0.845 market to 0.431. With Manifold observational-only and macro signals
|
||||
gated behind is_non_price, GNews is the only live signal able to move politics
|
||||
markets 20-30 pp against the order-book consensus. The fuse clamps the
|
||||
posterior to prior ± MAX_NEWS_ONLY_PROB_SHIFT when GNews is the ONLY material
|
||||
signal (|log-odds| >= NEWS_MATERIAL_LOGODDS_THRESHOLD); any other material
|
||||
signal counts as corroboration and disables the clamp.
|
||||
|
||||
Politics markets have no macro adjustments, so full-path tests exercise the
|
||||
"GNews only" branch naturally; the corroboration branch is tested through the
|
||||
pure helper apply_news_guardrail().
|
||||
|
||||
evaluate() emits a NEWS_MATERIAL log line for every market whose news
|
||||
contribution is material (trade or skip); tests parse it via caplog.
|
||||
"""
|
||||
import asyncio
|
||||
import logging
|
||||
import math
|
||||
import re
|
||||
|
||||
import pytest
|
||||
|
||||
import bot.strategy.bayesian as bayesian
|
||||
from bot.data.external import ExternalSignals
|
||||
from bot.data.polymarket import Market
|
||||
from bot.strategy.bayesian import (
|
||||
NEWS_LOGODDS_WEIGHT,
|
||||
BayesianStrategy,
|
||||
apply_news_guardrail,
|
||||
)
|
||||
|
||||
NEWS_MATERIAL_RE = re.compile(
|
||||
r"NEWS_MATERIAL.*raw=(\d+\.\d+) \| final=(\d+\.\d+).*"
|
||||
r"guardrail=(applied|none) \| changed_decision=(true|false)"
|
||||
)
|
||||
|
||||
|
||||
def _logodds(p: float) -> float:
|
||||
return math.log(p / (1 - p))
|
||||
|
||||
|
||||
def _sentiment_for(prior: float, target_raw: float) -> float:
|
||||
"""Sentiment that moves `prior` to exactly `target_raw` via GNews alone."""
|
||||
return (_logodds(target_raw) - _logodds(prior)) / NEWS_LOGODDS_WEIGHT
|
||||
|
||||
|
||||
class FakeNews:
|
||||
"""Deterministic NewsClient stub returning a fixed sentiment."""
|
||||
|
||||
enabled = True
|
||||
|
||||
def __init__(self, sentiment: float) -> None:
|
||||
self._sentiment = sentiment
|
||||
|
||||
async def get_sentiment(self, question: str) -> float:
|
||||
return self._sentiment
|
||||
|
||||
def get_freshness(self, question: str) -> float:
|
||||
return 1.0
|
||||
|
||||
|
||||
def _make_market(yes_price: float) -> Market:
|
||||
return Market(
|
||||
id="mkt-guardrail-1",
|
||||
condition_id="cond-guardrail-1",
|
||||
question="Will John Smith win the election?",
|
||||
yes_token_id="yes-tok",
|
||||
no_token_id="no-tok",
|
||||
yes_price=yes_price,
|
||||
no_price=1.0 - yes_price,
|
||||
volume_24h=50_000.0,
|
||||
end_date="2026-07-15T00:00:00Z", # politics <30 d → regime_min 0.08
|
||||
active=True,
|
||||
category="politics",
|
||||
)
|
||||
|
||||
|
||||
def _make_signals() -> ExternalSignals:
|
||||
# Neutral macro environment; irrelevant for politics (gated) but explicit.
|
||||
return ExternalSignals(
|
||||
btc_price=100_000.0,
|
||||
btc_change_24h=0.0,
|
||||
eth_price=4_000.0,
|
||||
eth_change_24h=0.0,
|
||||
btc_dominance=50.0,
|
||||
fear_greed_index=50,
|
||||
fear_greed_label="neutral",
|
||||
total_market_cap_change=0.0,
|
||||
valid=True,
|
||||
)
|
||||
|
||||
|
||||
def _evaluate(yes_price: float, sentiment: float, caplog) -> tuple[
|
||||
BayesianStrategy, tuple[float, float, str, str]
|
||||
]:
|
||||
"""Run evaluate() on a politics market and parse the NEWS_MATERIAL line."""
|
||||
strategy = BayesianStrategy(news=FakeNews(sentiment), manifold=None, db=None)
|
||||
market = _make_market(yes_price)
|
||||
with caplog.at_level(logging.INFO, logger="bot.strategy.bayesian"):
|
||||
asyncio.run(strategy.evaluate(market, _make_signals(), occupied_families=set()))
|
||||
for record in caplog.records:
|
||||
m = NEWS_MATERIAL_RE.search(record.getMessage())
|
||||
if m:
|
||||
return strategy, (
|
||||
float(m.group(1)), float(m.group(2)), m.group(3), m.group(4)
|
||||
)
|
||||
pytest.fail(
|
||||
"No NEWS_MATERIAL log line found; got: "
|
||||
f"{[r.getMessage() for r in caplog.records]}"
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Test 1 — extreme uncorroborated shift: clamp to prior - MAX_NEWS_ONLY_PROB_SHIFT
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_extreme_news_only_shift_is_clamped(caplog):
|
||||
"""prior=0.845, raw 0.431 (NVIDIA signature) → final clamped to 0.595."""
|
||||
strategy, (raw, final, guardrail, _) = _evaluate(
|
||||
yes_price=0.845, sentiment=_sentiment_for(0.845, 0.431), caplog=caplog
|
||||
)
|
||||
assert raw == pytest.approx(0.431, abs=1e-3)
|
||||
assert guardrail == "applied"
|
||||
assert final >= 0.595
|
||||
assert final == pytest.approx(0.845 - bayesian.MAX_NEWS_ONLY_PROB_SHIFT, abs=1e-3)
|
||||
assert strategy.get_cycle_stats()["news_guardrail_applied"] == 1
|
||||
assert strategy.get_cycle_stats()["news_with_material"] == 1
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Test 2 — moderate shift inside the band: passes through untouched
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_moderate_news_shift_inside_band_not_clamped(caplog):
|
||||
"""prior=0.50, raw 0.62 → within ±0.25 band → final=0.62, no clamp."""
|
||||
strategy, (raw, final, guardrail, _) = _evaluate(
|
||||
yes_price=0.50, sentiment=_sentiment_for(0.50, 0.62), caplog=caplog
|
||||
)
|
||||
assert raw == pytest.approx(0.62, abs=1e-3)
|
||||
assert final == pytest.approx(0.62, abs=1e-3)
|
||||
assert guardrail == "none"
|
||||
assert strategy.get_cycle_stats()["news_guardrail_applied"] == 0
|
||||
# Still counted as a material-news market for the NEWS SUMMARY.
|
||||
assert strategy.get_cycle_stats()["news_with_material"] == 1
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Test 3 — corroboration: any other material signal disables the fuse
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_corroborated_news_not_clamped():
|
||||
"""GNews material + another signal >= threshold → raw passes without clamp."""
|
||||
news_lo = _logodds(0.20) - _logodds(0.50) # ≈ -1.386, clearly material
|
||||
final, applied = apply_news_guardrail(
|
||||
prior=0.50,
|
||||
raw_final_prob=0.20,
|
||||
feat_news_lo=news_lo,
|
||||
other_feats_lo=(0.0, 0.15, 0.0, 0.0), # one corroborating signal
|
||||
)
|
||||
assert final == 0.20
|
||||
assert applied is False
|
||||
|
||||
|
||||
def test_corroboration_threshold_is_inclusive():
|
||||
"""|other| == threshold exactly counts as corroboration (>=, not >)."""
|
||||
final, applied = apply_news_guardrail(
|
||||
prior=0.50,
|
||||
raw_final_prob=0.20,
|
||||
feat_news_lo=-1.386,
|
||||
other_feats_lo=(bayesian.NEWS_MATERIAL_LOGODDS_THRESHOLD, 0.0, 0.0, 0.0),
|
||||
)
|
||||
assert final == 0.20
|
||||
assert applied is False
|
||||
|
||||
|
||||
def test_uncorroborated_helper_clamps():
|
||||
"""Same shift with only noise elsewhere → clamped to prior - 0.25."""
|
||||
final, applied = apply_news_guardrail(
|
||||
prior=0.50,
|
||||
raw_final_prob=0.20,
|
||||
feat_news_lo=-1.386,
|
||||
other_feats_lo=(0.05, -0.09, 0.0, 0.0), # all below threshold → noise
|
||||
)
|
||||
assert final == pytest.approx(0.25)
|
||||
assert applied is True
|
||||
|
||||
|
||||
def test_sub_material_news_never_clamped():
|
||||
"""|news_lo| below threshold → fuse not armed, whatever the shift."""
|
||||
final, applied = apply_news_guardrail(
|
||||
prior=0.50,
|
||||
raw_final_prob=0.10,
|
||||
feat_news_lo=0.09,
|
||||
other_feats_lo=(0.0, 0.0, 0.0, 0.0),
|
||||
)
|
||||
assert final == 0.10
|
||||
assert applied is False
|
||||
|
||||
|
||||
def test_guardrail_disabled_passthrough(monkeypatch):
|
||||
monkeypatch.setattr(bayesian, "NEWS_GUARDRAIL_ENABLED", False)
|
||||
final, applied = apply_news_guardrail(
|
||||
prior=0.845,
|
||||
raw_final_prob=0.431,
|
||||
feat_news_lo=-1.974,
|
||||
other_feats_lo=(0.0, 0.0, 0.0, 0.0),
|
||||
)
|
||||
assert final == 0.431
|
||||
assert applied is False
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Test 4 — changed_decision: the clamp moves the edge from tradeable to not
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_guardrail_changed_trade_decision(monkeypatch, caplog):
|
||||
"""
|
||||
With max_shift=0.10 the clamped edge (0.10 gross, 0.06 net) falls below the
|
||||
politics <30 d regime gate (0.08) while the raw edge (0.414 gross, 0.374
|
||||
net) crossed it → the fuse prevented the trade → changed_decision=true.
|
||||
|
||||
(With the default 0.25 the clamped edge_net is 0.21, above every regime
|
||||
minimum, so the flag can only fire with a tighter configured band.)
|
||||
"""
|
||||
monkeypatch.setattr(bayesian, "MAX_NEWS_ONLY_PROB_SHIFT", 0.10)
|
||||
strategy, (raw, final, guardrail, changed) = _evaluate(
|
||||
yes_price=0.845, sentiment=_sentiment_for(0.845, 0.431), caplog=caplog
|
||||
)
|
||||
assert raw == pytest.approx(0.431, abs=1e-3)
|
||||
assert final == pytest.approx(0.745, abs=1e-3)
|
||||
assert guardrail == "applied"
|
||||
assert changed == "true"
|
||||
stats = strategy.get_cycle_stats()
|
||||
assert stats["news_changed_decisions"] == 1
|
||||
assert stats["news_guardrail_applied"] == 1
|
||||
|
||||
|
||||
def test_default_band_does_not_change_decision(caplog):
|
||||
"""Default 0.25 band: clamp binds but edge_net 0.21 still crosses the gate."""
|
||||
_, (_, _, guardrail, changed) = _evaluate(
|
||||
yes_price=0.845, sentiment=_sentiment_for(0.845, 0.431), caplog=caplog
|
||||
)
|
||||
assert guardrail == "applied"
|
||||
assert changed == "false"
|
||||
@@ -0,0 +1,77 @@
|
||||
"""Tests for the GNews layer minor fixes.
|
||||
|
||||
Two faults found during the GNews capture/prioritisation diagnostic:
|
||||
|
||||
1. Hyphens/dashes in a market question reached the GNews query verbatim and,
|
||||
because '-' is GNews's exclusion operator, produced HTTP 400
|
||||
(e.g. "Abdul El-Sayed Michigan Democratic Primary").
|
||||
|
||||
2. The per-cycle GNews budget counter incremented in evaluate() *before*
|
||||
get_sentiment() checked the API key, so with no key configured the
|
||||
[CYCLE SUMMARY] reported a phantom "gnews_queries_used: 5/5" even though
|
||||
zero real requests left the process.
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
from bot.data.news import NewsClient
|
||||
from bot.data.external import ExternalSignals
|
||||
from bot.data.polymarket import Market
|
||||
from bot.strategy.bayesian import BayesianStrategy
|
||||
|
||||
|
||||
# ── Fix 1: query sanitisation ────────────────────────────────────────────────
|
||||
|
||||
def test_build_query_strips_hyphen_that_breaks_gnews():
|
||||
q = NewsClient._build_query(
|
||||
"Will Abdul El-Sayed win the 2026 Michigan Democratic Primary?"
|
||||
)
|
||||
assert "-" not in q # the exclusion operator must be gone
|
||||
assert "El-Sayed" not in q
|
||||
assert "Sayed" in q # the meaningful token survives as its own word
|
||||
|
||||
|
||||
def test_build_query_strips_unicode_dashes():
|
||||
q = NewsClient._build_query("Trump–Putin summit — final outcome")
|
||||
assert "–" not in q and "—" not in q
|
||||
assert "Trump" in q and "Putin" in q
|
||||
|
||||
|
||||
# ── Fix 2: enabled property + budget accounting ──────────────────────────────
|
||||
|
||||
def test_enabled_reflects_api_key(monkeypatch):
|
||||
monkeypatch.delenv("GNEWS_API_KEY", raising=False)
|
||||
assert NewsClient().enabled is False
|
||||
monkeypatch.setenv("GNEWS_API_KEY", "deadbeefdeadbeefdeadbeefdeadbeef")
|
||||
assert NewsClient().enabled is True
|
||||
|
||||
|
||||
def _politics_market() -> Market:
|
||||
return Market(
|
||||
id="m1", condition_id="c1",
|
||||
question="Will candidate X win the 2026 governor election?",
|
||||
yes_token_id="y", no_token_id="n",
|
||||
yes_price=0.50, no_price=0.50, volume_24h=10_000.0,
|
||||
end_date="2026-07-15T00:00:00Z", active=True, category="politics",
|
||||
)
|
||||
|
||||
|
||||
def _signals() -> ExternalSignals:
|
||||
return ExternalSignals(
|
||||
btc_price=1.0, btc_change_24h=0.0, eth_price=1.0, eth_change_24h=0.0,
|
||||
btc_dominance=50.0, fear_greed_index=50, fear_greed_label="neutral",
|
||||
total_market_cap_change=0.0, valid=True,
|
||||
)
|
||||
|
||||
|
||||
def test_disabled_news_consumes_no_gnews_budget(monkeypatch):
|
||||
"""Regression: no API key → gnews_queries_used stays 0 (was a phantom 1+)."""
|
||||
monkeypatch.delenv("GNEWS_API_KEY", raising=False)
|
||||
news = NewsClient()
|
||||
assert news.enabled is False
|
||||
|
||||
strategy = BayesianStrategy(news=news, manifold=None, db=None)
|
||||
strategy.reset_cycle()
|
||||
asyncio.run(
|
||||
strategy.evaluate(_politics_market(), _signals(), occupied_families=set())
|
||||
)
|
||||
assert strategy.get_cycle_stats()["gnews_queries_used"] == 0
|
||||
@@ -0,0 +1,174 @@
|
||||
"""Replay R2 tests — outcome fetching and calibration scoring."""
|
||||
import asyncio
|
||||
import math
|
||||
|
||||
import pytest
|
||||
|
||||
from bot.data.polymarket import MarketResolution
|
||||
from bot.outcomes import (
|
||||
LOGLOSS_EPS,
|
||||
compute_calibration,
|
||||
fetch_outcomes,
|
||||
print_report,
|
||||
)
|
||||
from datetime import datetime, timezone
|
||||
|
||||
|
||||
class FakePoly:
|
||||
"""get_market_resolution stand-in driven by a dict of canned responses."""
|
||||
|
||||
def __init__(self, responses: dict):
|
||||
self.responses = responses
|
||||
self.calls: list[str] = []
|
||||
|
||||
async def get_market_resolution(self, market_id: str):
|
||||
self.calls.append(market_id)
|
||||
return self.responses.get(market_id)
|
||||
|
||||
|
||||
RESOLVED_AT = datetime(2026, 7, 1, 12, 0, tzinfo=timezone.utc)
|
||||
|
||||
|
||||
def _row(market_id="m1", category="politics", est=0.6, prior=0.5, outcome=1.0):
|
||||
return {
|
||||
"market_id": market_id,
|
||||
"category": category,
|
||||
"estimated_prob": est,
|
||||
"prior_prob": prior,
|
||||
"outcome": outcome,
|
||||
}
|
||||
|
||||
|
||||
# ── fetch_outcomes ───────────────────────────────────────────────────────────
|
||||
|
||||
def test_fetch_keeps_only_definitive_resolutions():
|
||||
poly = FakePoly({
|
||||
"yes": MarketResolution(resolved=True, resolution=1.0,
|
||||
resolved_at=RESOLVED_AT),
|
||||
"no": MarketResolution(resolved=True, resolution=0.0,
|
||||
resolved_at=None),
|
||||
"open": MarketResolution(resolved=False),
|
||||
"disputed": MarketResolution(resolved=False),
|
||||
"apierror": None, # get_market_resolution returns None on HTTP errors
|
||||
})
|
||||
out = asyncio.run(
|
||||
fetch_outcomes(poly, ["yes", "no", "open", "disputed", "apierror"])
|
||||
)
|
||||
assert poly.calls == ["yes", "no", "open", "disputed", "apierror"]
|
||||
assert out == [
|
||||
{"market_id": "yes", "outcome": 1.0, "resolved_at": RESOLVED_AT},
|
||||
{"market_id": "no", "outcome": 0.0, "resolved_at": None},
|
||||
]
|
||||
|
||||
|
||||
def test_fetch_empty_list_is_noop():
|
||||
poly = FakePoly({})
|
||||
assert asyncio.run(fetch_outcomes(poly, [])) == []
|
||||
assert poly.calls == []
|
||||
|
||||
|
||||
# ── compute_calibration ──────────────────────────────────────────────────────
|
||||
|
||||
def test_no_rows_returns_none():
|
||||
assert compute_calibration([]) is None
|
||||
|
||||
|
||||
def test_single_row_known_values():
|
||||
m = compute_calibration([_row(est=0.8, prior=0.6, outcome=1.0)])
|
||||
assert m["n_evaluations"] == 1
|
||||
assert m["n_markets"] == 1
|
||||
assert m["brier_model"] == pytest.approx((0.8 - 1.0) ** 2)
|
||||
assert m["brier_prior"] == pytest.approx((0.6 - 1.0) ** 2)
|
||||
assert m["logloss_model"] == pytest.approx(-math.log(0.8))
|
||||
assert m["logloss_prior"] == pytest.approx(-math.log(0.6))
|
||||
# one market: macro == micro
|
||||
assert m["brier_model_macro"] == pytest.approx(m["brier_model"])
|
||||
assert m["brier_prior_macro"] == pytest.approx(m["brier_prior"])
|
||||
|
||||
|
||||
def test_logloss_no_outcome_branch():
|
||||
m = compute_calibration([_row(est=0.2, prior=0.7, outcome=0.0)])
|
||||
assert m["logloss_model"] == pytest.approx(-math.log(0.8))
|
||||
assert m["logloss_prior"] == pytest.approx(-math.log(0.3))
|
||||
|
||||
|
||||
def test_logloss_clipping_keeps_hard_miss_finite():
|
||||
# A hard 1.0 estimate on a NO outcome must not produce inf.
|
||||
m = compute_calibration([_row(est=1.0, prior=0.5, outcome=0.0)])
|
||||
assert math.isfinite(m["logloss_model"])
|
||||
assert m["logloss_model"] == pytest.approx(-math.log(LOGLOSS_EPS))
|
||||
|
||||
|
||||
def test_micro_weights_evaluations_macro_weights_markets():
|
||||
# Market a: 3 evaluations, model error 0.1; market b: 1 evaluation, error 0.5.
|
||||
rows = [
|
||||
_row(market_id="a", est=0.9, prior=0.8, outcome=1.0),
|
||||
_row(market_id="a", est=0.9, prior=0.8, outcome=1.0),
|
||||
_row(market_id="a", est=0.9, prior=0.8, outcome=1.0),
|
||||
_row(market_id="b", est=0.5, prior=0.6, outcome=1.0),
|
||||
]
|
||||
m = compute_calibration(rows)
|
||||
assert m["n_evaluations"] == 4
|
||||
assert m["n_markets"] == 2
|
||||
# micro: (3*0.01 + 0.25) / 4 ; macro: (0.01 + 0.25) / 2
|
||||
assert m["brier_model"] == pytest.approx((3 * 0.01 + 0.25) / 4)
|
||||
assert m["brier_model_macro"] == pytest.approx((0.01 + 0.25) / 2)
|
||||
assert m["brier_prior"] == pytest.approx((3 * 0.04 + 0.16) / 4)
|
||||
assert m["brier_prior_macro"] == pytest.approx((0.04 + 0.16) / 2)
|
||||
|
||||
|
||||
def test_model_beating_market_gives_negative_delta():
|
||||
# est closer to the outcome than the price on every row
|
||||
rows = [
|
||||
_row(market_id="a", est=0.8, prior=0.6, outcome=1.0),
|
||||
_row(market_id="b", est=0.3, prior=0.45, outcome=0.0),
|
||||
]
|
||||
m = compute_calibration(rows)
|
||||
assert m["brier_model"] < m["brier_prior"]
|
||||
assert m["logloss_model"] < m["logloss_prior"]
|
||||
|
||||
|
||||
def test_per_category_grouping_and_unknown():
|
||||
rows = [
|
||||
_row(market_id="a", category="politics", est=0.8, prior=0.6, outcome=1.0),
|
||||
_row(market_id="b", category="politics", est=0.7, prior=0.6, outcome=1.0),
|
||||
_row(market_id="c", category=None, est=0.4, prior=0.5, outcome=0.0),
|
||||
]
|
||||
m = compute_calibration(rows)
|
||||
assert set(m["per_category"]) == {"politics", "unknown"}
|
||||
pol = m["per_category"]["politics"]
|
||||
assert pol["n"] == 2 and pol["markets"] == 2
|
||||
assert pol["brier_model"] == pytest.approx((0.04 + 0.09) / 2)
|
||||
unk = m["per_category"]["unknown"]
|
||||
assert unk["n"] == 1 and unk["markets"] == 1
|
||||
assert unk["brier_model"] == pytest.approx(0.16)
|
||||
|
||||
|
||||
def test_repeated_market_counts_once_in_markets():
|
||||
rows = [
|
||||
_row(market_id="a", est=0.8, prior=0.6, outcome=1.0),
|
||||
_row(market_id="a", est=0.7, prior=0.55, outcome=1.0),
|
||||
]
|
||||
m = compute_calibration(rows)
|
||||
assert m["n_markets"] == 1
|
||||
assert m["per_category"]["politics"]["markets"] == 1
|
||||
|
||||
|
||||
# ── print_report ─────────────────────────────────────────────────────────────
|
||||
|
||||
def test_report_handles_no_metrics(capsys):
|
||||
print_report(None, "R0 archive")
|
||||
assert "no scorable rows yet" in capsys.readouterr().out
|
||||
|
||||
|
||||
def test_report_prints_all_metric_lines(capsys):
|
||||
m = compute_calibration([
|
||||
_row(market_id="a", est=0.8, prior=0.6, outcome=1.0),
|
||||
_row(market_id="b", category=None, est=0.4, prior=0.5, outcome=0.0),
|
||||
])
|
||||
print_report(m, "R0 archive")
|
||||
out = capsys.readouterr().out
|
||||
assert "2 evaluations, 2 markets" in out
|
||||
for label in ("Brier micro", "Brier macro", "logloss micro",
|
||||
"politics", "unknown"):
|
||||
assert label in out
|
||||
@@ -0,0 +1,367 @@
|
||||
"""
|
||||
Tests for the Replay R1 replay core (bot/replay.py) and the as_of clock
|
||||
injection in BayesianStrategy.evaluate().
|
||||
|
||||
The central contract is round-trip fidelity: a decision recorded by R0 and
|
||||
replayed through replay_cycle() with the same strategy constants must match
|
||||
field-for-field (matched=True, mismatch_field=None). Each round-trip test
|
||||
produces the "archive" by running the real evaluate() with FakeNews, then
|
||||
replays the drained record as if it had been read back from the signals table.
|
||||
"""
|
||||
import asyncio
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
import bot.strategy.bayesian as bayesian
|
||||
from bot.data.polymarket import Market, market_family_key
|
||||
from bot.strategy.bayesian import BayesianStrategy, _days_to_resolution
|
||||
from bot.replay import (
|
||||
ReplayNews,
|
||||
build_ext,
|
||||
build_market,
|
||||
replay_cycle,
|
||||
strategy_config_hash,
|
||||
)
|
||||
|
||||
from tests.test_news_guardrail import FakeNews, _sentiment_for
|
||||
|
||||
|
||||
def _end_date(days_ahead: int = 20) -> str:
|
||||
dt = datetime.now(timezone.utc) + timedelta(days=days_ahead)
|
||||
return dt.strftime("%Y-%m-%dT00:00:00Z")
|
||||
|
||||
|
||||
def _make_market(
|
||||
yes_price: float,
|
||||
question: str = "Will John Smith win the election?",
|
||||
category: str = "politics",
|
||||
market_id: str = "mkt-replay-1",
|
||||
end_date: str = None,
|
||||
) -> Market:
|
||||
return Market(
|
||||
id=market_id,
|
||||
condition_id="cond-replay-1",
|
||||
question=question,
|
||||
yes_token_id="yes-tok",
|
||||
no_token_id="no-tok",
|
||||
yes_price=yes_price,
|
||||
no_price=1.0 - yes_price,
|
||||
volume_24h=50_000.0,
|
||||
end_date=end_date if end_date is not None else _end_date(),
|
||||
active=True,
|
||||
category=category,
|
||||
)
|
||||
|
||||
|
||||
def _snapshot(valid: bool = True) -> dict:
|
||||
"""An ext_snapshots row as read back from the DB."""
|
||||
return {
|
||||
"btc_price": 100_000.0,
|
||||
"btc_change_24h": 0.0,
|
||||
"eth_price": 4_000.0,
|
||||
"eth_change_24h": 0.0,
|
||||
"btc_dominance": 50.0,
|
||||
"fear_greed_index": 50,
|
||||
"fear_greed_label": "neutral",
|
||||
"total_market_cap_change": 0.0,
|
||||
"valid": valid,
|
||||
}
|
||||
|
||||
|
||||
def _market_row(market: Market) -> dict:
|
||||
"""A markets-table row for the given Market."""
|
||||
return {
|
||||
"id": market.id,
|
||||
"condition_id": market.condition_id,
|
||||
"question": market.question,
|
||||
"category": market.category,
|
||||
"end_date": market.end_date,
|
||||
}
|
||||
|
||||
|
||||
def _record_with_live_evaluate(
|
||||
market: Market,
|
||||
news=None,
|
||||
families: set = frozenset(),
|
||||
) -> dict:
|
||||
"""Run the real evaluate() and return the R0 record it produced —
|
||||
the same dict save_signal_records() would have archived."""
|
||||
strategy = BayesianStrategy(news=news, manifold=None, db=None)
|
||||
asyncio.run(strategy.evaluate(market, build_ext(_snapshot()), set(families)))
|
||||
return strategy.drain_cycle_records()[0]
|
||||
|
||||
|
||||
def _replay_one(record: dict, market: Market, snapshot: dict = None) -> dict:
|
||||
cycle_ts = datetime.now(timezone.utc)
|
||||
decisions = asyncio.run(replay_cycle(
|
||||
cycle_ts,
|
||||
snapshot or _snapshot(),
|
||||
[record],
|
||||
{market.id: _market_row(market)},
|
||||
))
|
||||
assert len(decisions) == 1
|
||||
return decisions[0]
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Clock injection
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_days_to_resolution_uses_injected_clock():
|
||||
end = "2026-08-01T00:00:00Z"
|
||||
as_of = datetime(2026, 7, 2, 12, 0, tzinfo=timezone.utc)
|
||||
assert _days_to_resolution(end, as_of) == 29
|
||||
assert _days_to_resolution(end, as_of - timedelta(days=60)) == 89
|
||||
|
||||
|
||||
def test_default_clock_is_wall_clock():
|
||||
end = _end_date(days_ahead=40)
|
||||
assert _days_to_resolution(end) == _days_to_resolution(
|
||||
end, datetime.now(timezone.utc)
|
||||
)
|
||||
|
||||
|
||||
def test_as_of_changes_regime_threshold():
|
||||
"""Same politics market: <30 d out → regime 0.08; replayed from 60 d
|
||||
earlier → regime 0.12. The clock, not the wall time, must decide."""
|
||||
market = _make_market(0.470)
|
||||
sentiment = _sentiment_for(0.470, 0.601)
|
||||
|
||||
def _regime(as_of):
|
||||
strategy = BayesianStrategy(news=FakeNews(sentiment), manifold=None, db=None)
|
||||
asyncio.run(strategy.evaluate(
|
||||
market, build_ext(_snapshot()), set(), as_of=as_of,
|
||||
))
|
||||
return strategy.drain_cycle_records()[0]["regime_min_edge"]
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
assert _regime(now) == pytest.approx(0.08)
|
||||
assert _regime(now - timedelta(days=60)) == pytest.approx(0.12)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Round-trip fidelity: record with live evaluate(), replay, expect match
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_roundtrip_confidence_skip():
|
||||
"""Georgia signature: edge passes, confidence blocks — full-field match."""
|
||||
sentiment = _sentiment_for(0.470, 0.601)
|
||||
market = _make_market(0.470)
|
||||
record = _record_with_live_evaluate(market, news=FakeNews(sentiment))
|
||||
assert record["skip_reason"] == "confidence"
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["mismatch_field"] is None
|
||||
assert decision["skip_reason"] == "confidence"
|
||||
assert decision["estimated_prob"] == pytest.approx(record["estimated_prob"])
|
||||
assert decision["edge_net"] == pytest.approx(record["edge_net"])
|
||||
assert decision["confidence"] == pytest.approx(record["confidence"])
|
||||
assert decision["direction"] == record["direction"]
|
||||
assert decision["would_trade"] is False
|
||||
|
||||
|
||||
def test_roundtrip_edge_net_skip():
|
||||
market = _make_market(0.50)
|
||||
record = _record_with_live_evaluate(market)
|
||||
assert record["skip_reason"] == "edge_net"
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["would_trade"] is False
|
||||
|
||||
|
||||
def test_roundtrip_guardrail_clamp():
|
||||
"""Clamped posterior must reproduce exactly (raw != final in archive)."""
|
||||
market = _make_market(0.845)
|
||||
record = _record_with_live_evaluate(
|
||||
market, news=FakeNews(_sentiment_for(0.845, 0.431))
|
||||
)
|
||||
assert record["guardrail_applied"] is True
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["raw_final_prob"] == pytest.approx(record["raw_final_prob"])
|
||||
assert decision["estimated_prob"] == pytest.approx(record["estimated_prob"])
|
||||
|
||||
|
||||
def test_roundtrip_prior_extreme():
|
||||
market = _make_market(0.03)
|
||||
record = _record_with_live_evaluate(market)
|
||||
assert record["skip_reason"] == "prior_extreme"
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["skip_reason"] == "prior_extreme"
|
||||
|
||||
|
||||
def test_roundtrip_family_skip():
|
||||
"""Family-skipped rows replay with their own family injected as occupied."""
|
||||
market = _make_market(0.50)
|
||||
record = _record_with_live_evaluate(
|
||||
market, families={market_family_key(market)}
|
||||
)
|
||||
assert record["skip_reason"] == "family"
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["skip_reason"] == "family"
|
||||
|
||||
|
||||
def test_roundtrip_unsupported():
|
||||
market = _make_market(0.50, question="Will it rain tomorrow?", category="")
|
||||
record = _record_with_live_evaluate(market)
|
||||
assert record["skip_reason"] == "unsupported"
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
|
||||
|
||||
def test_roundtrip_no_signals():
|
||||
"""ext.valid=False archived → replay rebuilds the invalid snapshot."""
|
||||
market = _make_market(0.50)
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
asyncio.run(strategy.evaluate(market, build_ext(_snapshot(valid=False)), set()))
|
||||
record = strategy.drain_cycle_records()[0]
|
||||
assert record["skip_reason"] == "no_signals"
|
||||
|
||||
decision = _replay_one(record, market, snapshot=_snapshot(valid=False))
|
||||
assert decision["matched"] is True
|
||||
|
||||
|
||||
def test_roundtrip_trade_path(monkeypatch):
|
||||
"""skip_reason=None (tradeable) round-trips with would_trade=True.
|
||||
Politics can't clear MIN_CONFIDENCE=0.55 (the known ceiling), so the
|
||||
gate is lowered for this test only — both record and replay see the
|
||||
same constant, which is exactly the config_hash contract."""
|
||||
monkeypatch.setattr(bayesian, "MIN_CONFIDENCE", 0.45)
|
||||
sentiment = _sentiment_for(0.470, 0.601)
|
||||
market = _make_market(0.470)
|
||||
record = _record_with_live_evaluate(market, news=FakeNews(sentiment))
|
||||
assert record["skip_reason"] is None
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["skip_reason"] is None
|
||||
assert decision["would_trade"] is True
|
||||
assert decision["direction"] == "BUY_YES"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Replay-specific semantics
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_budget_skipped_row_replays_without_news():
|
||||
"""A budget-skipped archive row (sentiment 0.0) must replay to the same
|
||||
no-news decision — and never consume a replay-side budget."""
|
||||
market = _make_market(0.50)
|
||||
strategy = BayesianStrategy(news=FakeNews(0.9), manifold=None, db=None)
|
||||
strategy._news_queries_this_cycle = bayesian.MAX_NEWS_QUERIES_PER_CYCLE
|
||||
asyncio.run(strategy.evaluate(market, build_ext(_snapshot()), set()))
|
||||
record = strategy.drain_cycle_records()[0]
|
||||
assert record["news_budget_skipped"] is True
|
||||
assert record["news_sentiment"] == 0.0
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is True
|
||||
assert decision["estimated_prob"] == pytest.approx(record["estimated_prob"])
|
||||
|
||||
|
||||
def test_reentry_guard_row_is_recalibrated_not_compared():
|
||||
"""record_skip() rows carry no decision fields; the replay re-evaluates
|
||||
them (calibration data) but marks them non-comparable."""
|
||||
market = _make_market(0.50)
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
strategy.record_skip(market, "reentry_guard")
|
||||
record = strategy.drain_cycle_records()[0]
|
||||
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is None
|
||||
assert decision["recorded_skip_reason"] == "reentry_guard"
|
||||
# Re-evaluated on its merits: a full decision despite the recorded skip
|
||||
assert decision["estimated_prob"] is not None
|
||||
assert decision["skip_reason"] == "edge_net"
|
||||
|
||||
|
||||
def test_missing_market_row_flagged_not_crashed():
|
||||
market = _make_market(0.50)
|
||||
record = _record_with_live_evaluate(market)
|
||||
|
||||
decisions = asyncio.run(replay_cycle(
|
||||
datetime.now(timezone.utc), _snapshot(), [record], {},
|
||||
))
|
||||
assert decisions[0]["matched"] is False
|
||||
assert decisions[0]["mismatch_field"] == "market_missing"
|
||||
|
||||
|
||||
def test_mismatch_detected_when_config_differs(monkeypatch):
|
||||
"""Counterfactual sanity: replaying under a different guardrail band
|
||||
must produce matched=False with the diverging field named."""
|
||||
market = _make_market(0.845)
|
||||
record = _record_with_live_evaluate(
|
||||
market, news=FakeNews(_sentiment_for(0.845, 0.431))
|
||||
)
|
||||
assert record["guardrail_applied"] is True
|
||||
|
||||
monkeypatch.setattr(bayesian, "MAX_NEWS_ONLY_PROB_SHIFT", 0.10)
|
||||
decision = _replay_one(record, market)
|
||||
assert decision["matched"] is False
|
||||
# Tighter clamp (prior 0.845 ± 0.10 → est 0.745): edge_net drops from
|
||||
# 0.21 to 0.06 < regime 0.08, so the skip flips confidence → edge_net
|
||||
# and skip_reason is the first field _compare() sees diverge.
|
||||
assert decision["mismatch_field"] == "skip_reason"
|
||||
assert decision["skip_reason"] == "edge_net"
|
||||
|
||||
|
||||
def test_multi_row_cycle_preserves_order_and_isolation():
|
||||
"""Rows replay independently within a cycle: a family skip and a full
|
||||
evaluation with different sentiments don't bleed into each other."""
|
||||
m1 = _make_market(0.470, market_id="m1")
|
||||
m2 = _make_market(
|
||||
0.50, market_id="m2",
|
||||
question="Will Jane Doe win the Georgia Senate race?",
|
||||
)
|
||||
r1 = _record_with_live_evaluate(m1, news=FakeNews(_sentiment_for(0.470, 0.601)))
|
||||
r2 = _record_with_live_evaluate(m2) # no news → edge_net skip
|
||||
|
||||
decisions = asyncio.run(replay_cycle(
|
||||
datetime.now(timezone.utc),
|
||||
_snapshot(),
|
||||
[r1, r2],
|
||||
{"m1": _market_row(m1), "m2": _market_row(m2)},
|
||||
))
|
||||
assert [d["market_id"] for d in decisions] == ["m1", "m2"]
|
||||
assert all(d["matched"] is True for d in decisions)
|
||||
assert decisions[0]["skip_reason"] == "confidence"
|
||||
assert decisions[1]["skip_reason"] == "edge_net"
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Run tagging
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_config_hash_stable_and_sensitive(monkeypatch):
|
||||
h1 = strategy_config_hash()
|
||||
assert strategy_config_hash() == h1
|
||||
monkeypatch.setattr(bayesian, "MAX_NEWS_ONLY_PROB_SHIFT", 0.10)
|
||||
assert strategy_config_hash() != h1
|
||||
|
||||
|
||||
def test_replay_news_returns_current_sentiment():
|
||||
news = ReplayNews()
|
||||
assert asyncio.run(news.get_sentiment("q")) == 0.0
|
||||
news.sentiment = -0.42
|
||||
assert asyncio.run(news.get_sentiment("q")) == -0.42
|
||||
|
||||
|
||||
def test_build_market_reconstruction():
|
||||
market = _make_market(0.37)
|
||||
record = _record_with_live_evaluate(market)
|
||||
rebuilt = build_market(_market_row(market), record)
|
||||
assert rebuilt.id == market.id
|
||||
assert rebuilt.yes_price == pytest.approx(0.37)
|
||||
assert rebuilt.volume_24h == pytest.approx(market.volume_24h)
|
||||
assert rebuilt.end_date == market.end_date
|
||||
assert rebuilt.category == "politics"
|
||||
assert market_family_key(rebuilt) == market_family_key(market)
|
||||
@@ -0,0 +1,224 @@
|
||||
"""
|
||||
Tests for the Replay R0 snapshot recorder (strategy-side record accumulation).
|
||||
|
||||
Every evaluate() call must leave exactly one record in _cycle_records, whatever
|
||||
exit path it takes, so the signals archive is a complete account of each cycle.
|
||||
DB persistence itself (save_signal_records) is exercised in prod; these tests
|
||||
cover the record-building contract the replay engine will rely on:
|
||||
|
||||
- one record per market per evaluate() call, drained per cycle
|
||||
- skip_reason granularity (prior_extreme / family / edge_net / confidence /
|
||||
unsupported / reentry_guard via record_skip)
|
||||
- full input/output fields on records that reached edge computation
|
||||
- news_budget_skipped distinguishes "not asked" from "no news"
|
||||
"""
|
||||
import asyncio
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
|
||||
import bot.strategy.bayesian as bayesian
|
||||
from bot.data.external import ExternalSignals
|
||||
from bot.data.polymarket import Market
|
||||
from bot.strategy.bayesian import (
|
||||
MAX_NEWS_QUERIES_PER_CYCLE,
|
||||
BayesianStrategy,
|
||||
)
|
||||
|
||||
from tests.test_news_guardrail import FakeNews, _sentiment_for
|
||||
|
||||
|
||||
def _end_date(days_ahead: int = 20) -> str:
|
||||
dt = datetime.now(timezone.utc) + timedelta(days=days_ahead)
|
||||
return dt.strftime("%Y-%m-%dT00:00:00Z")
|
||||
|
||||
|
||||
def _make_market(
|
||||
yes_price: float,
|
||||
question: str = "Will John Smith win the election?",
|
||||
category: str = "politics",
|
||||
market_id: str = "mkt-recorder-1",
|
||||
) -> Market:
|
||||
return Market(
|
||||
id=market_id,
|
||||
condition_id="cond-recorder-1",
|
||||
question=question,
|
||||
yes_token_id="yes-tok",
|
||||
no_token_id="no-tok",
|
||||
yes_price=yes_price,
|
||||
no_price=1.0 - yes_price,
|
||||
volume_24h=50_000.0,
|
||||
end_date=_end_date(), # ~20 d → politics regime_min 0.08
|
||||
active=True,
|
||||
category=category,
|
||||
)
|
||||
|
||||
|
||||
def _make_signals() -> ExternalSignals:
|
||||
return ExternalSignals(
|
||||
btc_price=100_000.0,
|
||||
btc_change_24h=0.0,
|
||||
eth_price=4_000.0,
|
||||
eth_change_24h=0.0,
|
||||
btc_dominance=50.0,
|
||||
fear_greed_index=50,
|
||||
fear_greed_label="neutral",
|
||||
total_market_cap_change=0.0,
|
||||
valid=True,
|
||||
)
|
||||
|
||||
|
||||
def _evaluate(strategy: BayesianStrategy, market: Market, families=None) -> None:
|
||||
asyncio.run(strategy.evaluate(market, _make_signals(), families or set()))
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Full-evaluation records: every input/output field the replay needs
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_confidence_skip_record_has_full_fields():
|
||||
"""Politics market whose edge passes but confidence blocks (the known
|
||||
politics ceiling): record must carry the complete decision context."""
|
||||
sentiment = _sentiment_for(0.470, 0.601) # Georgia signature: edge_net 0.091
|
||||
strategy = BayesianStrategy(news=FakeNews(sentiment), manifold=None, db=None)
|
||||
market = _make_market(0.470)
|
||||
_evaluate(strategy, market)
|
||||
|
||||
records = strategy.drain_cycle_records()
|
||||
assert len(records) == 1
|
||||
rec = records[0]
|
||||
assert rec["market_id"] == "mkt-recorder-1"
|
||||
assert rec["skip_reason"] == "confidence"
|
||||
assert rec["category"] == "politics"
|
||||
assert rec["polymarket_price"] == pytest.approx(0.470)
|
||||
assert rec["prior_prob"] == pytest.approx(0.470)
|
||||
assert rec["estimated_prob"] == pytest.approx(0.601, abs=1e-3)
|
||||
assert rec["raw_final_prob"] == pytest.approx(0.601, abs=1e-3)
|
||||
assert rec["edge_net"] == pytest.approx(0.091, abs=1e-3)
|
||||
assert rec["regime_min_edge"] == pytest.approx(0.08)
|
||||
assert rec["passed_net"] is True
|
||||
assert rec["confidence"] == pytest.approx(0.50)
|
||||
assert rec["direction"] == "BUY_YES"
|
||||
assert rec["news_sentiment"] == pytest.approx(sentiment, abs=1e-6)
|
||||
assert rec["feat_news_lo"] != 0.0
|
||||
assert rec["news_budget_skipped"] is False
|
||||
assert rec["guardrail_applied"] is False
|
||||
assert rec["guardrail_changed_decision"] is False
|
||||
assert rec["days_to_resolution"] is not None
|
||||
assert rec["acted_on"] is False
|
||||
|
||||
|
||||
def test_edge_net_skip_record():
|
||||
"""No news, no edge → skip_reason=edge_net with passed_net False."""
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
market = _make_market(0.50)
|
||||
_evaluate(strategy, market)
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["skip_reason"] == "edge_net"
|
||||
assert rec["passed_net"] is False
|
||||
assert rec["estimated_prob"] == pytest.approx(0.50, abs=1e-3)
|
||||
assert rec["feat_news_lo"] == 0.0
|
||||
|
||||
|
||||
def test_guardrail_fields_recorded_when_clamped():
|
||||
"""Guardrail clamp shows up in the record (applied=True, raw != final)."""
|
||||
strategy = BayesianStrategy(
|
||||
news=FakeNews(_sentiment_for(0.845, 0.431)), manifold=None, db=None
|
||||
)
|
||||
market = _make_market(0.845)
|
||||
_evaluate(strategy, market)
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["guardrail_applied"] is True
|
||||
assert rec["raw_final_prob"] == pytest.approx(0.431, abs=1e-3)
|
||||
assert rec["estimated_prob"] == pytest.approx(
|
||||
0.845 - bayesian.MAX_NEWS_ONLY_PROB_SHIFT, abs=1e-3
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Early-skip records: minimal but present
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_prior_extreme_record():
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
_evaluate(strategy, _make_market(0.03))
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["skip_reason"] == "prior_extreme"
|
||||
assert rec["polymarket_price"] == pytest.approx(0.03)
|
||||
assert rec["prior_prob"] == pytest.approx(0.05) # clamped prior
|
||||
assert rec["estimated_prob"] is None
|
||||
assert rec["edge_net"] is None
|
||||
|
||||
|
||||
def test_family_skip_record():
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
market = _make_market(0.50)
|
||||
from bot.data.polymarket import market_family_key
|
||||
_evaluate(strategy, market, families={market_family_key(market)})
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["skip_reason"] == "family"
|
||||
assert rec["family_key"] is not None
|
||||
|
||||
|
||||
def test_unsupported_record():
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
market = _make_market(0.50, question="Will it rain tomorrow?", category="")
|
||||
_evaluate(strategy, market)
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["skip_reason"] == "unsupported"
|
||||
|
||||
|
||||
def test_record_skip_external_reason():
|
||||
"""main.py records reentry-guard skips through record_skip()."""
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
strategy.record_skip(_make_market(0.50), "reentry_guard")
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["skip_reason"] == "reentry_guard"
|
||||
assert rec["estimated_prob"] is None
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
# Budget flag + cycle lifecycle
|
||||
# ─────────────────────────────────────────────────────────────────────────────
|
||||
|
||||
def test_news_budget_skipped_flag():
|
||||
"""With the cycle budget exhausted, the record must say GNews was never
|
||||
asked — feat_news_lo=0.0 alone would be indistinguishable from no-news."""
|
||||
strategy = BayesianStrategy(news=FakeNews(0.9), manifold=None, db=None)
|
||||
strategy._news_queries_this_cycle = MAX_NEWS_QUERIES_PER_CYCLE
|
||||
_evaluate(strategy, _make_market(0.50))
|
||||
|
||||
rec = strategy.drain_cycle_records()[0]
|
||||
assert rec["news_budget_skipped"] is True
|
||||
assert rec["news_sentiment"] == 0.0
|
||||
assert rec["feat_news_lo"] == 0.0
|
||||
|
||||
|
||||
def test_drain_empties_and_reset_clears():
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
_evaluate(strategy, _make_market(0.50))
|
||||
assert len(strategy.drain_cycle_records()) == 1
|
||||
assert strategy.drain_cycle_records() == []
|
||||
|
||||
_evaluate(strategy, _make_market(0.50))
|
||||
strategy.reset_cycle()
|
||||
assert strategy.drain_cycle_records() == []
|
||||
|
||||
|
||||
def test_one_record_per_market_accumulates_in_order():
|
||||
strategy = BayesianStrategy(news=None, manifold=None, db=None)
|
||||
_evaluate(strategy, _make_market(0.03, market_id="m1")) # prior_extreme
|
||||
_evaluate(strategy, _make_market(0.50, market_id="m2")) # edge_net
|
||||
_evaluate(strategy, _make_market(0.97, market_id="m3")) # prior_extreme
|
||||
|
||||
records = strategy.drain_cycle_records()
|
||||
assert [r["market_id"] for r in records] == ["m1", "m2", "m3"]
|
||||
assert [r["skip_reason"] for r in records] == [
|
||||
"prior_extreme", "edge_net", "prior_extreme",
|
||||
]
|
||||
Reference in New Issue
Block a user