polymarket-bot/tests/test_news_guardrail.py

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