polymarket-bot/bot/data/manifold.py

"""
Manifold Markets client — cross-platform prediction market probability signals.

For each Polymarket question, searches Manifold for a matching binary market
by keyword overlap and returns its probability as a calibration signal.

Inversion guard: if the Manifold market's winning side (Republican / Democrat)
is the complement of the Polymarket question's winning side, the probability is
automatically inverted (1 - prob).  This prevents "Democrats win Ohio governor"
from consuming the probability of a Manifold market titled "Republicans win Ohio
governor" without adjustment.

Rejection guard: if the match score falls below _MATCH_THRESHOLD the market is
rejected, even if inversion would otherwise apply.  All decisions are logged at
INFO so they can be audited per-cycle.

Cache TTL: 30 minutes (Manifold markets move slowly vs our 60 s cycle).
Match threshold: >= 0.25 keyword overlap ratio between significant tokens.
"""
import logging
import re
import time
from typing import Optional

import httpx

MANIFOLD_API = "https://api.manifold.markets/v0"
CACHE_TTL_SEC = 1800  # 30 minutes

log = logging.getLogger(__name__)

_MATCH_THRESHOLD = 0.25

_STOP_WORDS = frozenset([
    "will", "the", "a", "an", "is", "are", "was", "were", "be", "been",
    "by", "in", "on", "at", "to", "for", "of", "and", "or", "not",
    "this", "that", "with", "from", "have", "has", "had", "do", "does",
    "did", "can", "could", "would", "should", "may", "might", "shall",
    "win", "lose", "get", "become", "make", "take", "give", "see",
    "any", "who", "what", "when", "where", "which", "how", "over", "under",
    "than", "more", "most", "least", "its", "their", "they",
    "him", "her", "his", "she", "been", "being", "into", "after",
    "before", "during", "until", "against", "between", "through",
])

# Mutually exclusive political parties used for complement detection
_REPUBLICAN_WORDS = frozenset(["republican", "republicans", "gop"])
_DEMOCRAT_WORDS = frozenset(["democrat", "democrats", "democratic"])


def _significant_words(text: str) -> set[str]:
    words = re.findall(r"[a-zA-Z]+", text.lower())
    return {w for w in words if w not in _STOP_WORDS and len(w) >= 3}


def _build_search_query(question: str, max_words: int = 6) -> str:
    words = re.findall(r"[a-zA-Z0-9]+", question)
    sig = [w for w in words if w.lower() not in _STOP_WORDS and len(w) >= 3]
    return " ".join(sig[:max_words])


def _detect_party(text: str) -> Optional[str]:
    """Return 'republican', 'democrat', or None if no party detected."""
    words = set(re.findall(r"[a-zA-Z]+", text.lower()))
    if words & _REPUBLICAN_WORDS:
        return "republican"
    if words & _DEMOCRAT_WORDS:
        return "democrat"
    return None


def _best_match_with_audit(
    poly_question: str,
    results: list[dict],
) -> tuple[Optional[dict], float, bool]:
    """
    Find the best-matching open binary Manifold market.

    Returns (match, score, needs_inversion):
      match            — best result dict, or None if below threshold
      score            — keyword overlap score of best candidate (even if rejected)
      needs_inversion  — True when Manifold market favours the OPPOSITE party/side
                         to the Polymarket question (probability should be 1 - prob)
    """
    poly_words = _significant_words(poly_question)
    poly_party = _detect_party(poly_question)
    if not poly_words:
        return None, 0.0, False

    best_score = 0.0
    best: Optional[dict] = None
    best_needs_inv = False

    for result in results:
        if result.get("outcomeType") != "BINARY":
            continue
        prob = result.get("probability")
        if prob is None or not (0.02 < float(prob) < 0.98):
            continue
        title = result.get("question", "")
        m_words = _significant_words(title)
        if not m_words:
            continue
        overlap = len(poly_words & m_words)
        score = overlap / min(len(poly_words), len(m_words))
        if score > best_score:
            best_score = score
            best = result
            manifold_party = _detect_party(title)
            # Inversion is warranted only when both sides are unambiguously detected
            # and they are confirmed opposites (republican ≠ democrat).
            best_needs_inv = (
                poly_party is not None
                and manifold_party is not None
                and poly_party != manifold_party
            )

    if best_score >= _MATCH_THRESHOLD and best is not None:
        return best, best_score, best_needs_inv
    return None, best_score, False


class ManifoldClient:
    """Async Manifold Markets client for cross-platform probability signals."""

    def __init__(self) -> None:
        self._client = httpx.AsyncClient(timeout=15)
        # question → (fetched_at_monotonic, probability_or_None)
        self._cache: dict[str, tuple[float, Optional[float]]] = {}

    async def get_probability(self, question: str) -> Optional[float]:
        """
        Return Manifold probability for a matching market, or None.

        Probability is already adjusted for party-direction inversion when
        the matched Manifold market is the complement of our question.

        Full audit log is emitted at INFO for every resolved query.
        """
        now = time.monotonic()
        cached = self._cache.get(question)
        if cached and (now - cached[0]) < CACHE_TTL_SEC:
            return cached[1]

        query = _build_search_query(question)
        if not query:
            self._cache[question] = (now, None)
            return None

        try:
            resp = await self._client.get(
                f"{MANIFOLD_API}/search-markets",
                params={"term": query, "limit": 5, "filter": "open"},
            )
            resp.raise_for_status()
            results = resp.json()
        except Exception as e:
            log.warning("Manifold API error for %r: %s", question[:40], e)
            self._cache[question] = (now, None)
            return None

        match, score, needs_inv = _best_match_with_audit(question, results)

        if match is None:
            log.info(
                "Manifold no_match: %-50s | best_score=%.2f < %.2f | query=%r",
                question[:50], score, _MATCH_THRESHOLD, query,
            )
            self._cache[question] = (now, None)
            return None

        prob_raw = float(match["probability"])
        prob_final = (1.0 - prob_raw) if needs_inv else prob_raw

        # Build market URL from slug (best-effort; may be missing)
        slug = match.get("slug", "")
        creator = match.get("creatorUsername", "")
        url = f"https://manifold.markets/{creator}/{slug}" if slug else "n/a"

        log.info(
            "Manifold %s: %-50s\n"
            "  poly_question: %s\n"
            "  manifold_title: %s\n"
            "  manifold_url:   %s\n"
            "  match_score: %.2f | prob_raw=%.3f | inverted=%s | prob_final=%.3f",
            "MATCH_INVERTED" if needs_inv else "MATCH",
            question[:50],
            question,
            match.get("question", ""),
            url,
            score, prob_raw, needs_inv, prob_final,
        )

        self._cache[question] = (now, prob_final)
        return prob_final

    async def close(self) -> None:
        await self._client.aclose()