Jeffrey Sperling
4fde52459d
Add events_status=active and keep_closed_markets=0 to Gamma API search params, filtering out resolved/closed markets that clutter results. These params are confirmed in the Polymarket OpenAPI spec.
560 lines
20 KiB
Python
560 lines
20 KiB
Python
"""Polymarket prediction market search via Gamma API (free, no auth required).
|
|
|
|
Uses gamma-api.polymarket.com for event/market discovery.
|
|
No API key needed - public read-only API with generous rate limits (15K req/10s).
|
|
"""
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import quote_plus, urlencode
|
|
|
|
from . import http
|
|
|
|
GAMMA_SEARCH_URL = "https://gamma-api.polymarket.com/public-search"
|
|
|
|
# Pages to fetch per query (API returns 5 events per page, limit param is a no-op)
|
|
DEPTH_CONFIG = {
|
|
"quick": 1,
|
|
"default": 3,
|
|
"deep": 4,
|
|
}
|
|
|
|
# Max events to return after merge + dedup + re-ranking
|
|
RESULT_CAP = {
|
|
"quick": 5,
|
|
"default": 15,
|
|
"deep": 25,
|
|
}
|
|
|
|
|
|
def _log(msg: str):
|
|
"""Log to stderr (only in TTY mode to avoid cluttering Claude Code output)."""
|
|
if sys.stderr.isatty():
|
|
sys.stderr.write(f"[PM] {msg}\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def _extract_core_subject(topic: str) -> str:
|
|
"""Extract core subject from topic string.
|
|
|
|
Strips common prefixes like 'last 7 days', 'what are people saying about', etc.
|
|
"""
|
|
topic = topic.strip()
|
|
# Remove common leading phrases
|
|
prefixes = [
|
|
r"^last \d+ days?\s+",
|
|
r"^what(?:'s| is| are) (?:people saying about|happening with|going on with)\s+",
|
|
r"^how (?:is|are)\s+",
|
|
r"^tell me about\s+",
|
|
r"^research\s+",
|
|
]
|
|
for pattern in prefixes:
|
|
topic = re.sub(pattern, "", topic, flags=re.IGNORECASE)
|
|
return topic.strip()
|
|
|
|
|
|
def _expand_queries(topic: str) -> List[str]:
|
|
"""Generate search queries to cast a wider net.
|
|
|
|
Strategy:
|
|
- Always include the core subject
|
|
- Add ALL individual words as standalone searches (not just first)
|
|
- Include the full topic if different from core
|
|
- Cap at 6 queries, dedupe
|
|
"""
|
|
core = _extract_core_subject(topic)
|
|
queries = [core]
|
|
|
|
# Add ALL individual words as separate queries
|
|
words = core.split()
|
|
if len(words) >= 2:
|
|
for word in words:
|
|
if len(word) > 1: # skip single-char words
|
|
queries.append(word)
|
|
|
|
# Add the full topic if different from core
|
|
if topic.lower().strip() != core.lower():
|
|
queries.append(topic.strip())
|
|
|
|
# Dedupe while preserving order, cap at 6
|
|
seen = set()
|
|
unique = []
|
|
for q in queries:
|
|
q_lower = q.lower().strip()
|
|
if q_lower and q_lower not in seen:
|
|
seen.add(q_lower)
|
|
unique.append(q.strip())
|
|
return unique[:6]
|
|
|
|
|
|
_GENERIC_TAGS = frozenset({"sports", "politics", "crypto", "science", "culture", "pop culture"})
|
|
|
|
|
|
def _extract_domain_queries(topic: str, events: List[Dict]) -> List[str]:
|
|
"""Extract domain-indicator search terms from first-pass event tags.
|
|
|
|
Uses structured tag metadata from Gamma API events to discover broader
|
|
domain categories (e.g., 'NCAA CBB' from a Big 12 basketball event).
|
|
Falls back to frequent title bigrams if no useful tags exist.
|
|
"""
|
|
query_words = set(_extract_core_subject(topic).lower().split())
|
|
|
|
# Collect tag labels from all first-pass events, count occurrences
|
|
tag_counts: Dict[str, int] = {}
|
|
for event in events:
|
|
tags = event.get("tags") or []
|
|
for tag in tags:
|
|
label = tag.get("label", "") if isinstance(tag, dict) else str(tag)
|
|
if not label:
|
|
continue
|
|
label_lower = label.lower()
|
|
# Skip generic category tags and tags matching existing queries
|
|
if label_lower in _GENERIC_TAGS:
|
|
continue
|
|
if label_lower in query_words:
|
|
continue
|
|
tag_counts[label] = tag_counts.get(label, 0) + 1
|
|
|
|
# Sort by frequency, take top 2 that appear in 2+ events
|
|
domain_queries = [
|
|
label for label, count in sorted(tag_counts.items(), key=lambda x: -x[1])
|
|
if count >= 2
|
|
][:2]
|
|
|
|
return domain_queries
|
|
|
|
|
|
def _search_single_query(query: str, page: int = 1) -> Dict[str, Any]:
|
|
"""Run a single search query against Gamma API."""
|
|
params = {
|
|
"q": query,
|
|
"page": str(page),
|
|
"events_status": "active",
|
|
"keep_closed_markets": "0",
|
|
}
|
|
url = f"{GAMMA_SEARCH_URL}?{urlencode(params)}"
|
|
|
|
try:
|
|
response = http.request("GET", url, timeout=15, retries=2)
|
|
return response
|
|
except http.HTTPError as e:
|
|
_log(f"Search failed for '{query}' page {page}: {e}")
|
|
return {"events": [], "error": str(e)}
|
|
except Exception as e:
|
|
_log(f"Search failed for '{query}' page {page}: {e}")
|
|
return {"events": [], "error": str(e)}
|
|
|
|
|
|
def _run_queries_parallel(
|
|
queries: List[str], pages: int, all_events: Dict, errors: List, start_idx: int = 0,
|
|
) -> None:
|
|
"""Run (query, page) combinations in parallel, merging into all_events."""
|
|
with ThreadPoolExecutor(max_workers=min(8, len(queries) * pages)) as executor:
|
|
futures = {}
|
|
for i, q in enumerate(queries, start=start_idx):
|
|
for p in range(1, pages + 1):
|
|
future = executor.submit(_search_single_query, q, p)
|
|
futures[future] = i
|
|
|
|
for future in as_completed(futures):
|
|
query_idx = futures[future]
|
|
try:
|
|
response = future.result(timeout=15)
|
|
if response.get("error"):
|
|
errors.append(response["error"])
|
|
|
|
events = response.get("events", [])
|
|
for event in events:
|
|
event_id = event.get("id", "")
|
|
if not event_id:
|
|
continue
|
|
if event_id not in all_events:
|
|
all_events[event_id] = (event, query_idx)
|
|
elif query_idx < all_events[event_id][1]:
|
|
all_events[event_id] = (event, query_idx)
|
|
except Exception as e:
|
|
errors.append(str(e))
|
|
|
|
|
|
def search_polymarket(
|
|
topic: str,
|
|
from_date: str,
|
|
to_date: str,
|
|
depth: str = "default",
|
|
) -> Dict[str, Any]:
|
|
"""Search Polymarket via Gamma API with two-pass query expansion.
|
|
|
|
Pass 1: Run expanded queries in parallel, merge and dedupe by event ID.
|
|
Pass 2: Extract domain-indicator terms from first-pass titles, search those.
|
|
|
|
Args:
|
|
topic: Search topic
|
|
from_date: Start date (YYYY-MM-DD) - used for activity filtering
|
|
to_date: End date (YYYY-MM-DD)
|
|
depth: 'quick', 'default', or 'deep'
|
|
|
|
Returns:
|
|
Dict with 'events' list and optional 'error'.
|
|
"""
|
|
pages = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
|
|
cap = RESULT_CAP.get(depth, RESULT_CAP["default"])
|
|
queries = _expand_queries(topic)
|
|
|
|
_log(f"Searching for '{topic}' with queries: {queries} (pages={pages})")
|
|
|
|
# Pass 1: run expanded queries in parallel
|
|
all_events: Dict[str, tuple] = {}
|
|
errors: List[str] = []
|
|
_run_queries_parallel(queries, pages, all_events, errors)
|
|
|
|
# Pass 2: extract domain-indicator terms from first-pass titles and search
|
|
first_pass_events = [ev for ev, _ in all_events.values()]
|
|
domain_queries = _extract_domain_queries(topic, first_pass_events)
|
|
# Filter out queries we already ran
|
|
seen_queries = {q.lower() for q in queries}
|
|
domain_queries = [dq for dq in domain_queries if dq.lower() not in seen_queries]
|
|
|
|
if domain_queries:
|
|
_log(f"Domain expansion queries: {domain_queries}")
|
|
_run_queries_parallel(domain_queries, 1, all_events, errors, start_idx=len(queries))
|
|
|
|
merged_events = [ev for ev, _ in sorted(all_events.values(), key=lambda x: x[1])]
|
|
total_queries = len(queries) + len(domain_queries)
|
|
_log(f"Found {len(merged_events)} unique events across {total_queries} queries")
|
|
|
|
result = {"events": merged_events, "_cap": cap}
|
|
if errors and not merged_events:
|
|
result["error"] = "; ".join(errors[:2])
|
|
return result
|
|
|
|
|
|
def _format_price_movement(market: Dict[str, Any]) -> Optional[str]:
|
|
"""Pick the most significant price change and format it.
|
|
|
|
Returns string like 'down 11.7% this month' or None if no significant change.
|
|
"""
|
|
changes = [
|
|
(abs(market.get("oneDayPriceChange") or 0), market.get("oneDayPriceChange"), "today"),
|
|
(abs(market.get("oneWeekPriceChange") or 0), market.get("oneWeekPriceChange"), "this week"),
|
|
(abs(market.get("oneMonthPriceChange") or 0), market.get("oneMonthPriceChange"), "this month"),
|
|
]
|
|
|
|
# Pick the largest absolute change
|
|
changes.sort(key=lambda x: x[0], reverse=True)
|
|
abs_change, raw_change, period = changes[0]
|
|
|
|
# Skip if change is less than 1% (noise)
|
|
if abs_change < 0.01:
|
|
return None
|
|
|
|
direction = "up" if raw_change > 0 else "down"
|
|
pct = abs_change * 100
|
|
return f"{direction} {pct:.1f}% {period}"
|
|
|
|
|
|
def _parse_outcome_prices(market: Dict[str, Any]) -> List[tuple]:
|
|
"""Parse outcomePrices JSON string into list of (outcome_name, price) tuples."""
|
|
outcomes_raw = market.get("outcomes") or []
|
|
prices_raw = market.get("outcomePrices")
|
|
|
|
if not prices_raw:
|
|
return []
|
|
|
|
# Both outcomes and outcomePrices can be JSON-encoded strings
|
|
try:
|
|
if isinstance(outcomes_raw, str):
|
|
outcomes = json.loads(outcomes_raw)
|
|
else:
|
|
outcomes = outcomes_raw
|
|
except (json.JSONDecodeError, TypeError):
|
|
outcomes = []
|
|
|
|
try:
|
|
if isinstance(prices_raw, str):
|
|
prices = json.loads(prices_raw)
|
|
else:
|
|
prices = prices_raw
|
|
except (json.JSONDecodeError, TypeError):
|
|
return []
|
|
|
|
result = []
|
|
for i, price in enumerate(prices):
|
|
try:
|
|
p = float(price)
|
|
except (ValueError, TypeError):
|
|
continue
|
|
name = outcomes[i] if i < len(outcomes) else f"Outcome {i+1}"
|
|
result.append((name, p))
|
|
|
|
return result
|
|
|
|
|
|
def _shorten_question(question: str) -> str:
|
|
"""Extract a short display name from a market question.
|
|
|
|
'Will Arizona win the 2026 NCAA Tournament?' -> 'Arizona'
|
|
'Will Duke be a number 1 seed in the 2026 NCAA...' -> 'Duke'
|
|
"""
|
|
q = question.strip().rstrip("?")
|
|
# Common patterns: "Will X win/be/...", "X wins/loses..."
|
|
m = re.match(r"^Will\s+(.+?)\s+(?:win|be|make|reach|have|lose|qualify|advance|strike|agree|pass|sign|get|become|remain|stay|leave|survive|next)\b", q, re.IGNORECASE)
|
|
if m:
|
|
return m.group(1).strip()
|
|
m = re.match(r"^Will\s+(.+?)\s+", q, re.IGNORECASE)
|
|
if m and len(m.group(1).split()) <= 4:
|
|
return m.group(1).strip()
|
|
# Fallback: truncate
|
|
return question[:40] if len(question) > 40 else question
|
|
|
|
|
|
def _compute_text_similarity(topic: str, title: str, outcomes: List[str] = None) -> float:
|
|
"""Score how well the event title (or outcome names) match the search topic.
|
|
|
|
Returns 0.0-1.0. Title substring match gets 1.0, outcome match gets 0.85/0.7,
|
|
title token overlap gets proportional score.
|
|
"""
|
|
core = _extract_core_subject(topic).lower()
|
|
title_lower = title.lower()
|
|
if not core:
|
|
return 0.5
|
|
|
|
# Full substring match in title
|
|
if core in title_lower:
|
|
return 1.0
|
|
|
|
# Check if topic appears in any outcome name (bidirectional)
|
|
if outcomes:
|
|
core_tokens = set(core.split())
|
|
best_outcome_score = 0.0
|
|
for outcome_name in outcomes:
|
|
outcome_lower = outcome_name.lower()
|
|
# Bidirectional: "arizona" in "arizona basketball" OR "arizona basketball" contains "arizona"
|
|
if core in outcome_lower or outcome_lower in core:
|
|
best_outcome_score = max(best_outcome_score, 0.85)
|
|
elif core_tokens & set(outcome_lower.split()):
|
|
best_outcome_score = max(best_outcome_score, 0.7)
|
|
if best_outcome_score > 0:
|
|
return best_outcome_score
|
|
|
|
# Token overlap fallback against title
|
|
topic_tokens = set(core.split())
|
|
title_tokens = set(title_lower.split())
|
|
if not topic_tokens:
|
|
return 0.5
|
|
overlap = len(topic_tokens & title_tokens)
|
|
return overlap / len(topic_tokens)
|
|
|
|
|
|
def _safe_float(val, default=0.0) -> float:
|
|
"""Safely convert a value to float."""
|
|
try:
|
|
return float(val or default)
|
|
except (ValueError, TypeError):
|
|
return default
|
|
|
|
|
|
def parse_polymarket_response(response: Dict[str, Any], topic: str = "") -> List[Dict[str, Any]]:
|
|
"""Parse Gamma API response into normalized item dicts.
|
|
|
|
Each event becomes one item showing its title and top markets.
|
|
|
|
Args:
|
|
response: Raw Gamma API response
|
|
topic: Original search topic (for relevance scoring)
|
|
|
|
Returns:
|
|
List of item dicts ready for normalization.
|
|
"""
|
|
events = response.get("events", [])
|
|
items = []
|
|
|
|
for i, event in enumerate(events):
|
|
event_id = event.get("id", "")
|
|
title = event.get("title", "")
|
|
slug = event.get("slug", "")
|
|
|
|
# Filter: skip closed/resolved events
|
|
if event.get("closed", False):
|
|
continue
|
|
if not event.get("active", True):
|
|
continue
|
|
|
|
# Get markets for this event
|
|
markets = event.get("markets", [])
|
|
if not markets:
|
|
continue
|
|
|
|
# Filter to active, open markets with liquidity (excludes resolved markets)
|
|
active_markets = []
|
|
for m in markets:
|
|
if m.get("closed", False):
|
|
continue
|
|
if not m.get("active", True):
|
|
continue
|
|
# Must have liquidity (resolved markets have 0 or None)
|
|
try:
|
|
liq = float(m.get("liquidity", 0) or 0)
|
|
except (ValueError, TypeError):
|
|
liq = 0
|
|
if liq > 0:
|
|
active_markets.append(m)
|
|
|
|
if not active_markets:
|
|
continue
|
|
|
|
# Sort markets by volume (most liquid first)
|
|
def market_volume(m):
|
|
try:
|
|
return float(m.get("volume", 0) or 0)
|
|
except (ValueError, TypeError):
|
|
return 0
|
|
active_markets.sort(key=market_volume, reverse=True)
|
|
|
|
# Take top market for the event
|
|
top_market = active_markets[0]
|
|
|
|
# Collect outcome names from ALL active markets (not just top) for similarity scoring
|
|
# Filter to outcomes with price > 1% to avoid noise
|
|
# Also extract subjects from market questions for neg-risk events (outcomes are Yes/No)
|
|
all_outcome_names = []
|
|
for m in active_markets:
|
|
for name, price in _parse_outcome_prices(m):
|
|
if price > 0.01 and name not in all_outcome_names:
|
|
all_outcome_names.append(name)
|
|
# For neg-risk binary markets (Yes/No outcomes), the team/entity name
|
|
# lives in the question, e.g., "Will Arizona win the NCAA Tournament?"
|
|
question = m.get("question", "")
|
|
if question and question != title:
|
|
all_outcome_names.append(question)
|
|
|
|
# Parse outcome prices - for multi-market events with Yes/No binary
|
|
# sub-markets, synthesize from market questions to show actual
|
|
# team/entity probabilities instead of a single market's Yes/No
|
|
outcome_prices = _parse_outcome_prices(top_market)
|
|
top_outcomes_are_binary = (
|
|
len(outcome_prices) == 2
|
|
and {n.lower() for n, _ in outcome_prices} == {"yes", "no"}
|
|
)
|
|
if top_outcomes_are_binary and len(active_markets) > 1:
|
|
synth_outcomes = []
|
|
for m in active_markets:
|
|
q = m.get("question", "")
|
|
if not q:
|
|
continue
|
|
pairs = _parse_outcome_prices(m)
|
|
yes_price = next((p for name, p in pairs if name.lower() == "yes"), None)
|
|
if yes_price is not None and yes_price > 0.005:
|
|
synth_outcomes.append((q, yes_price))
|
|
if synth_outcomes:
|
|
synth_outcomes.sort(key=lambda x: x[1], reverse=True)
|
|
outcome_prices = [(_shorten_question(q), p) for q, p in synth_outcomes]
|
|
|
|
# Format price movement
|
|
price_movement = _format_price_movement(top_market)
|
|
|
|
# Volume and liquidity - prefer event-level (more stable), fall back to market-level
|
|
event_volume1mo = _safe_float(event.get("volume1mo"))
|
|
event_volume1wk = _safe_float(event.get("volume1wk"))
|
|
event_liquidity = _safe_float(event.get("liquidity"))
|
|
event_competitive = _safe_float(event.get("competitive"))
|
|
volume24hr = _safe_float(event.get("volume24hr")) or _safe_float(top_market.get("volume24hr"))
|
|
liquidity = event_liquidity or _safe_float(top_market.get("liquidity"))
|
|
|
|
# Event URL
|
|
url = f"https://polymarket.com/event/{slug}" if slug else f"https://polymarket.com/event/{event_id}"
|
|
|
|
# Date: use updatedAt from event
|
|
updated_at = event.get("updatedAt", "")
|
|
date_str = None
|
|
if updated_at:
|
|
try:
|
|
date_str = updated_at[:10] # YYYY-MM-DD
|
|
except (IndexError, TypeError):
|
|
pass
|
|
|
|
# End date for the market
|
|
end_date = top_market.get("endDate")
|
|
if end_date:
|
|
try:
|
|
end_date = end_date[:10]
|
|
except (IndexError, TypeError):
|
|
end_date = None
|
|
|
|
# Quality-signal relevance (replaces position-based decay)
|
|
text_score = _compute_text_similarity(topic, title, all_outcome_names) if topic else 0.5
|
|
|
|
# Volume signal: log-scaled monthly volume (most stable signal)
|
|
vol_raw = event_volume1mo or event_volume1wk or volume24hr
|
|
vol_score = min(1.0, math.log1p(vol_raw) / 16) # ~$9M = 1.0
|
|
|
|
# Liquidity signal
|
|
liq_score = min(1.0, math.log1p(liquidity) / 14) # ~$1.2M = 1.0
|
|
|
|
# Price movement: daily weighted more than monthly
|
|
day_change = abs(top_market.get("oneDayPriceChange") or 0) * 3
|
|
week_change = abs(top_market.get("oneWeekPriceChange") or 0) * 2
|
|
month_change = abs(top_market.get("oneMonthPriceChange") or 0)
|
|
max_change = max(day_change, week_change, month_change)
|
|
movement_score = min(1.0, max_change * 5) # 20% change = 1.0
|
|
|
|
# Competitive bonus: markets near 50/50 are more interesting
|
|
competitive_score = event_competitive
|
|
|
|
relevance = min(1.0, (
|
|
0.30 * text_score +
|
|
0.30 * vol_score +
|
|
0.15 * liq_score +
|
|
0.15 * movement_score +
|
|
0.10 * competitive_score
|
|
))
|
|
|
|
# Surface the topic-matching outcome to the front before truncating
|
|
if topic and outcome_prices:
|
|
core = _extract_core_subject(topic).lower()
|
|
core_tokens = set(core.split())
|
|
reordered = []
|
|
rest = []
|
|
for pair in outcome_prices:
|
|
name_lower = pair[0].lower()
|
|
# Match if full core is substring, or name is substring of core,
|
|
# or any core token appears in the name (handles long question strings)
|
|
if (core in name_lower or name_lower in core
|
|
or any(tok in name_lower for tok in core_tokens if len(tok) > 2)):
|
|
reordered.append(pair)
|
|
else:
|
|
rest.append(pair)
|
|
if reordered:
|
|
outcome_prices = reordered + rest
|
|
|
|
# Top 3 outcomes for multi-outcome markets
|
|
top_outcomes = outcome_prices[:3]
|
|
remaining = len(outcome_prices) - 3
|
|
if remaining < 0:
|
|
remaining = 0
|
|
|
|
items.append({
|
|
"event_id": event_id,
|
|
"title": title,
|
|
"question": top_market.get("question", title),
|
|
"url": url,
|
|
"outcome_prices": top_outcomes,
|
|
"outcomes_remaining": remaining,
|
|
"price_movement": price_movement,
|
|
"volume24hr": volume24hr,
|
|
"volume1mo": event_volume1mo,
|
|
"liquidity": liquidity,
|
|
"date": date_str,
|
|
"end_date": end_date,
|
|
"relevance": round(relevance, 2),
|
|
"why_relevant": f"Prediction market: {title[:60]}",
|
|
})
|
|
|
|
# Sort by relevance (quality-signal ranked) and apply cap
|
|
items.sort(key=lambda x: x["relevance"], reverse=True)
|
|
cap = response.get("_cap", len(items))
|
|
return items[:cap]
|