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