""" 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 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 @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) -> None: self._signal_count = 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() # 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_any_supported = is_btc or is_eth or is_altcoin or is_general_crypto or is_macro if not is_any_supported: log.debug("Skipping unsupported market: %s", market.question[:60]) return None if not ext.valid: 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) if is_btc: momentum = ext.btc_change_24h asset_label = "BTC" elif is_eth: momentum = ext.eth_change_24h asset_label = "ETH" 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% 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)") # Macro markets: lower weight, rely only on Fear&Greed signal already added # Cap confidence below for macro to reflect weaker signal quality confidence_cap = 0.70 if is_macro else 0.90 # Compute posterior using log-odds updating log_odds_prior = math.log(prior / (1 - prior)) total_adj = sum(adjustments) estimated_prob = _sigmoid(log_odds_prior + total_adj * 2) 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) # Filter: only trade if edge and confidence thresholds met if abs_edge < MIN_EDGE or confidence < MIN_CONFIDENCE: log.debug( "No signal: edge=%.3f confidence=%.2f market=%s", abs_edge, confidence, market.question[:40] ) 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))