polymarket-bot/bot/strategy/bayesian.py

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

        # 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
        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)

        # 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))