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