87 lines
2.8 KiB
Python
87 lines
2.8 KiB
Python
"""Trending Polymarket bets, via their public Gamma REST API
|
|
(gamma-api.polymarket.com) — unauthenticated, no key/wallet required, the
|
|
same endpoint their own frontend calls (the site itself renders market
|
|
data client-side, so there's nothing to scrape from its HTML).
|
|
|
|
Positions/volumes are high-churn like flights — kept as an in-memory
|
|
snapshot rather than persisted history, same pattern as flights.py.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
log = logging.getLogger("newsatlas.polymarket")
|
|
|
|
EVENTS_URL = "https://gamma-api.polymarket.com/events"
|
|
_HEADERS = {"User-Agent": "Mozilla/5.0 (NewsAtlas; self-hosted news dashboard)"}
|
|
|
|
_latest_markets: list[dict] = []
|
|
|
|
|
|
def _leading_outcome(event: dict) -> str:
|
|
"""Best-effort one-line summary of where the odds currently sit."""
|
|
markets = event.get("markets") or []
|
|
try:
|
|
if len(markets) == 1:
|
|
outcomes = json.loads(markets[0].get("outcomes") or "[]")
|
|
prices = json.loads(markets[0].get("outcomePrices") or "[]")
|
|
pairs = dict(zip(outcomes, prices))
|
|
if "Yes" in pairs:
|
|
return f"Yes {float(pairs['Yes']) * 100:.0f}%"
|
|
elif len(markets) > 1:
|
|
best_label, best_price = None, -1.0
|
|
for m in markets:
|
|
outcomes = json.loads(m.get("outcomes") or "[]")
|
|
prices = json.loads(m.get("outcomePrices") or "[]")
|
|
pairs = dict(zip(outcomes, prices))
|
|
price = float(pairs.get("Yes", 0) or 0)
|
|
if price > best_price:
|
|
best_label = m.get("groupItemTitle") or m.get("question") or ""
|
|
best_price = price
|
|
if best_label is not None:
|
|
return f"{best_label} {best_price * 100:.0f}%"
|
|
except (ValueError, TypeError, KeyError):
|
|
pass
|
|
return ""
|
|
|
|
|
|
def poll_polymarket() -> int:
|
|
global _latest_markets
|
|
|
|
params = {
|
|
"order": "volume24hr",
|
|
"ascending": "false",
|
|
"active": "true",
|
|
"closed": "false",
|
|
"limit": "10",
|
|
}
|
|
try:
|
|
resp = httpx.get(EVENTS_URL, params=params, headers=_HEADERS, timeout=15)
|
|
resp.raise_for_status()
|
|
events = resp.json()
|
|
except Exception:
|
|
log.exception("Failed to fetch Polymarket events")
|
|
return 0
|
|
|
|
markets = []
|
|
for e in events:
|
|
markets.append(
|
|
{
|
|
"title": e.get("title", ""),
|
|
"slug": e.get("slug", ""),
|
|
"url": f"https://polymarket.com/event/{e.get('slug', '')}",
|
|
"volume_24h": e.get("volume24hr"),
|
|
"leading_outcome": _leading_outcome(e),
|
|
}
|
|
)
|
|
|
|
_latest_markets = markets
|
|
log.info("Polymarket poll complete: %d trending markets", len(markets))
|
|
return len(markets)
|
|
|
|
|
|
def get_latest_polymarket() -> list[dict]:
|
|
return _latest_markets
|