""" Bayesian Market Making Strategy. Core idea: 1. Compute a prior probability for a market outcome using external data 2. Compare with Polymarket's current price 3. If divergence > threshold + confidence is high enough → generate signal For crypto markets: if BTC is up 5% and fear/greed is 75 (greed), a market asking "Will BTC be above $X?" should be priced higher than Polymarket might reflect in a slow-moving order book. """ import logging import math from dataclasses import dataclass from typing import Optional from bot.data.polymarket import Market from bot.data.external import ExternalSignals from bot.data.news import NewsClient log = logging.getLogger(__name__) # Minimum edge required to place a trade. # With an informed prior (poly price), 10% means our signals strongly disagree # with the market — much higher bar than before, but necessary to avoid noise. MIN_EDGE = 0.10 # 10% edge minimum MIN_CONFIDENCE = 0.55 # Minimum confidence in our estimate # Log-odds weight applied to the GNews sentiment score (range ±1.0). # A weight of 1.5 means a fully negative/positive signal shifts log-odds by ±1.5, # which moves a 50% prior to ~18%/82% — strong but not overwhelming. NEWS_LOGODDS_WEIGHT = 1.5 # 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 @dataclass class TradingSignal: market_id: str question: str polymarket_price: float # Current market price for YES (0-1) estimated_prob: float # Our Bayesian estimate (0-1) edge: float # estimated_prob - polymarket_price confidence: float # How confident we are (0-1) direction: str # "BUY_YES" | "BUY_NO" reasoning: str # Human-readable explanation for logging sources: list[str] # Data sources used class BayesianStrategy: """ Estimates true probability using external signals and Bayesian updating. Prior: Polymarket's current YES price (market consensus — not 0.5) Likelihood updates from: - BTC/ETH price momentum - Fear & Greed index - Market cap trend / BTC dominance We only bet when our signals move the estimate far enough from the prior to justify the fee + slippage cost (MIN_EDGE). """ def __init__(self, news: Optional[NewsClient] = None) -> None: self._signal_count = 0 self._news = news # Optional; degrades gracefully when None or key missing self._news_queries_this_cycle = 0 def reset_cycle(self) -> None: """Call once at the start of each trading cycle to reset the per-cycle GNews counter.""" self._news_queries_this_cycle = 0 async def evaluate( self, market: Market, ext: ExternalSignals, ) -> Optional[TradingSignal]: """ Evaluate a market and return a signal if edge exists. Returns None if no actionable opportunity. """ question_lower = market.question.lower() category = market.category # set by PolymarketClient # Classify what kind of market this is is_price_above = any(w in question_lower for w in ["above", "over", "exceed", "higher", "atleast", "reach"]) is_price_below = any(w in question_lower for w in ["below", "under", "less than", "lower", "drop"]) is_btc = "btc" in question_lower or "bitcoin" in question_lower is_eth = "eth" in question_lower or "ethereum" in question_lower is_sol = "sol" in question_lower or "solana" in question_lower is_xrp = "xrp" in question_lower or "ripple" in question_lower is_doge = "doge" in question_lower or "dogecoin" in question_lower is_altcoin = is_sol or is_xrp or is_doge or any( w in question_lower for w in ["ltc", "litecoin", "bnb", "ada", "cardano", "avax", "avalanche"] ) is_general_crypto = any( w in question_lower for w in ["crypto", "market cap", "total market", "altcoin", "defi"] ) is_macro = any( w in question_lower for w in ["nasdaq", "s&p", "sp500", "inflation", "fed rate", "interest rate", "tariff"] ) is_politics = category == "politics" is_tech = category == "tech" is_events = category == "events" is_any_supported = ( is_btc or is_eth or is_altcoin or is_general_crypto or is_macro or is_politics or is_tech or is_events ) if not is_any_supported: log.info( "SKIP %-50s | reason=unsupported category=%r", market.question[:50], category, ) return None if not ext.valid: log.info( "SKIP %-50s | reason=no external signals", market.question[:50], ) return None # Can't reason without external data # --- Bayesian probability estimation --- # Prior = Polymarket consensus price, clamped away from extremes. # The market already aggregates information from many traders; # our signals update from that informed baseline, not from 0.5. prior = max(0.05, min(0.95, market.yes_price)) sources: list[str] = [f"Prior=poly({prior:.3f})"] adjustments: list[float] = [] # Signal 1: Price momentum (asset-specific or total market cap as proxy) # For politics/tech/events use BTC as a broad sentiment proxy. if is_btc: momentum = ext.btc_change_24h asset_label = "BTC" elif is_eth: momentum = ext.eth_change_24h asset_label = "ETH" elif is_politics or is_tech or is_events: # BTC as risk-sentiment proxy for non-crypto categories momentum = ext.btc_change_24h asset_label = "BTC(sentiment)" else: # Altcoins and general crypto: use total market cap change as proxy momentum = ext.total_market_cap_change asset_label = "total mktcap" if abs(momentum) > 2: momentum_adj = math.tanh(momentum / 20) * 0.15 # Max ±15% # For non-directional markets (politics/events/tech), momentum is weaker signal if is_politics or is_tech or is_events: momentum_adj *= 0.5 adjustments.append(momentum_adj if is_price_above else -momentum_adj) sources.append(f"{asset_label} 24h: {momentum:+.1f}%") # Signal 2: Fear & Greed fg = ext.fear_greed_index if fg > 70: fg_adj = 0.06 sources.append(f"Fear&Greed: {fg} (greed)") elif fg < 30: fg_adj = -0.06 sources.append(f"Fear&Greed: {fg} (fear)") else: fg_adj = (fg - 50) / 50 * 0.04 sources.append(f"Fear&Greed: {fg} (neutral)") adjustments.append(fg_adj if is_price_above else -fg_adj) # Signal 3: BTC dominance — hurts altcoins when high if (is_eth or is_altcoin or is_general_crypto) and ext.btc_dominance > 55: dom_adj = -0.03 if is_price_above else 0.03 adjustments.append(dom_adj) sources.append(f"BTC dom: {ext.btc_dominance:.1f}% (high → alt pressure)") elif (is_eth or is_altcoin or is_general_crypto) and ext.btc_dominance < 45: dom_adj = 0.03 if is_price_above else -0.03 adjustments.append(dom_adj) sources.append(f"BTC dom: {ext.btc_dominance:.1f}% (low → alt season)") # Signal 4: GNews sentiment — politics markets only. # BTC/F&G already cover crypto and macro; GNews budget is too tight to # waste on tech/events. Cap at MAX_NEWS_QUERIES_PER_CYCLE per cycle so # we prioritise the soonest-resolving markets (caller sorts by end_date). # Applied as a direct log-odds shift — stronger signal than macro proxies. # Weight NEWS_LOGODDS_WEIGHT=1.5 means a ±1.0 sentiment score shifts # log-odds by ±1.5 (e.g. 50% prior → ~82% / ~18%). news_log_adj = 0.0 if is_politics and self._news is not None: if self._news_queries_this_cycle < MAX_NEWS_QUERIES_PER_CYCLE: self._news_queries_this_cycle += 1 sentiment = await self._news.get_sentiment(market.question) if abs(sentiment) > 0.05: news_log_adj = sentiment * NEWS_LOGODDS_WEIGHT sources.append(f"GNews: {sentiment:+.2f}") else: log.debug( "GNews cycle limit (%d) reached — skipping news for %r", MAX_NEWS_QUERIES_PER_CYCLE, market.question[:50], ) # Macro/politics/tech/events: cap confidence lower to reflect weaker signal quality if is_macro or is_politics or is_tech or is_events: confidence_cap = 0.65 else: confidence_cap = 0.90 # Compute posterior using log-odds updating. # total_adj (BTC/F&G/dominance) is amplified ×2 because those are weak proxies. # news_log_adj is applied at face value — it IS a direct log-odds signal. log_odds_prior = math.log(prior / (1 - prior)) total_adj = sum(adjustments) estimated_prob = _sigmoid(log_odds_prior + total_adj * 2 + news_log_adj) estimated_prob = max(0.05, min(0.95, estimated_prob)) # Compute edge edge = estimated_prob - market.yes_price direction = "BUY_YES" if edge > 0 else "BUY_NO" abs_edge = abs(edge) # Confidence based on signal agreement agreement = sum(1 for a in adjustments if (a > 0) == (total_adj > 0)) confidence = min(confidence_cap, 0.4 + (agreement / max(len(adjustments), 1)) * 0.5) # News signal available → boost confidence by 0.10 (news corroborates macro signals) if news_log_adj != 0.0: confidence = min(confidence_cap, confidence + 0.10) # Log evaluation result for every market action = "TRADE" if (abs_edge >= MIN_EDGE and confidence >= MIN_CONFIDENCE) else "SKIP" skip_reason = "" if action == "SKIP": reasons = [] if abs_edge < MIN_EDGE: reasons.append(f"edge={abs_edge:.3f}<{MIN_EDGE}") if confidence < MIN_CONFIDENCE: reasons.append(f"conf={confidence:.2f}<{MIN_CONFIDENCE}") skip_reason = " | reason=" + ",".join(reasons) log.info( "%-5s %-50s | cat=%-12s | prior=%.3f | est=%.3f | edge=%+.3f | conf=%.2f | dir=%-8s | signals=%s%s", action, market.question[:50], category, prior, estimated_prob, edge, confidence, direction, ", ".join(sources[1:]) or "none", skip_reason, ) # Filter: only trade if edge and confidence thresholds met if abs_edge < MIN_EDGE or confidence < MIN_CONFIDENCE: return None reasoning = ( f"Prior=poly({prior:.3f}) → estimate={estimated_prob:.3f} | " f"Poly price={market.yes_price:.3f} | " f"Edge={edge:+.3f} | " f"Direction={direction} | " f"Signals: {', '.join(sources[1:])}" # skip the prior label already shown ) self._signal_count += 1 return TradingSignal( market_id=market.id, question=market.question, polymarket_price=market.yes_price, estimated_prob=estimated_prob, edge=abs_edge, confidence=confidence, direction=direction, reasoning=reasoning, sources=sources, ) def _sigmoid(x: float) -> float: return 1 / (1 + math.exp(-x))