145 lines
5.1 KiB
Python
145 lines
5.1 KiB
Python
import requests
|
||
import json
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
GAMMA = "https://gamma-api.polymarket.com"
|
||
CLOB = "https://clob.polymarket.com"
|
||
|
||
def parse_jsonish_list(v):
|
||
"""兼容 list 或 JSON 数组字符串。"""
|
||
if v is None:
|
||
return []
|
||
if isinstance(v, list):
|
||
return v
|
||
if isinstance(v, str):
|
||
s = v.strip()
|
||
if s.startswith("[") and s.endswith("]"):
|
||
return json.loads(s)
|
||
return [x.strip() for x in s.split(",") if x.strip()]
|
||
return []
|
||
|
||
def get_market_by_slug(session: requests.Session, slug: str) -> Dict[str, Any]:
|
||
r = session.get(f"{GAMMA}/markets/slug/{slug}", timeout=20)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
def get_price(session: requests.Session, token_id: str, side: str) -> Optional[float]:
|
||
"""
|
||
side='buy' -> best bid
|
||
side='sell' -> best ask
|
||
"""
|
||
r = session.get(f"{CLOB}/price", params={"token_id": token_id, "side": side}, timeout=20)
|
||
r.raise_for_status()
|
||
p = r.json().get("price")
|
||
return float(p) if p is not None else None
|
||
|
||
def get_book(session: requests.Session, token_id: str) -> Dict[str, Any]:
|
||
"""包括 bids/asks 的订单簿快照。"""
|
||
r = session.get(f"{CLOB}/book", params={"token_id": token_id}, timeout=20)
|
||
r.raise_for_status()
|
||
return r.json()
|
||
|
||
def compute_mid_from_price(bid: Optional[float], ask: Optional[float]) -> Tuple[Optional[float], Optional[float]]:
|
||
if bid is None and ask is None:
|
||
return None, None
|
||
if bid is None:
|
||
return ask, None
|
||
if ask is None:
|
||
return bid, None
|
||
return (bid + ask) / 2.0, (ask - bid)
|
||
|
||
def trim_book(book: Dict[str, Any], top_n: int = 10) -> Dict[str, Any]:
|
||
bids = book.get("bids") or []
|
||
asks = book.get("asks") or []
|
||
# 原始返回一般是按价优排序；这里做一下保险排序
|
||
bids_sorted = sorted(bids, key=lambda x: float(x["price"]), reverse=True)[:top_n]
|
||
asks_sorted = sorted(asks, key=lambda x: float(x["price"]))[:top_n]
|
||
return {"bids": bids_sorted, "asks": asks_sorted}
|
||
|
||
def fetch_token_bundle(session: requests.Session, token_id: str, top_n: int = 10) -> Dict[str, Any]:
|
||
"""
|
||
对某个 token 并发抓取：
|
||
- best bid (/price?side=buy)
|
||
- best ask (/price?side=sell)
|
||
- order book (/book)
|
||
"""
|
||
with ThreadPoolExecutor(max_workers=3) as ex:
|
||
futures = {
|
||
ex.submit(get_price, session, token_id, "buy"): "bid",
|
||
ex.submit(get_price, session, token_id, "sell"): "ask",
|
||
ex.submit(get_book, session, token_id): "book",
|
||
}
|
||
out: Dict[str, Any] = {"token_id": token_id}
|
||
for fut in as_completed(futures):
|
||
key = futures[fut]
|
||
out[key] = fut.result()
|
||
|
||
bid = out.get("bid")
|
||
ask = out.get("ask")
|
||
mid, spread = compute_mid_from_price(bid, ask)
|
||
|
||
out["mid"] = mid
|
||
out["spread"] = spread
|
||
out["book"] = trim_book(out.get("book") or {}, top_n=top_n)
|
||
return out
|
||
|
||
def polymarket_updown(slug: str, decimals: int = 0, top_n: int = 10) -> Dict[str, Any]:
|
||
"""
|
||
返回：
|
||
- UP/DOWN 百分比（尽量贴网页）
|
||
- UP/DOWN 的 token_id、bid/ask/mid/spread
|
||
- UP/DOWN 订单簿 topN
|
||
"""
|
||
session = requests.Session()
|
||
session.headers.update({"User-Agent": "Mozilla/5.0"})
|
||
|
||
m = get_market_by_slug(session, slug)
|
||
outcomes = [str(x) for x in parse_jsonish_list(m.get("outcomes"))]
|
||
token_ids = [str(x) for x in parse_jsonish_list(m.get("clobTokenIds"))]
|
||
|
||
if len(token_ids) < 2:
|
||
raise RuntimeError(f"clobTokenIds missing/too short: {token_ids}")
|
||
|
||
token_map = dict(zip(outcomes, token_ids))
|
||
up_id = token_map.get("Up") or token_map.get("Yes") or token_ids[0]
|
||
dn_id = token_map.get("Down") or token_map.get("No") or token_ids[1]
|
||
|
||
# 并发抓 UP / DOWN 两个 token 的 bundle
|
||
with ThreadPoolExecutor(max_workers=2) as ex:
|
||
fut_up = ex.submit(fetch_token_bundle, session, up_id, top_n)
|
||
fut_dn = ex.submit(fetch_token_bundle, session, dn_id, top_n)
|
||
up = fut_up.result()
|
||
dn = fut_dn.result()
|
||
|
||
if up["mid"] is None or dn["mid"] is None:
|
||
return {
|
||
"error": "missing mid price (bid/ask both None?)",
|
||
"slug": slug,
|
||
"market_id": m.get("id"),
|
||
"question": m.get("question"),
|
||
"up": up,
|
||
"down": dn,
|
||
}
|
||
|
||
# 归一化到 100（防止 mid 不严格互补导致 up+down != 1）
|
||
s = float(up["mid"]) + float(dn["mid"])
|
||
up_pct = round(float(up["mid"]) / s * 100, decimals)
|
||
dn_pct = round(float(dn["mid"]) / s * 100, decimals)
|
||
|
||
return {
|
||
"question": m.get("question"),
|
||
"slug": m.get("slug"),
|
||
"market_id": m.get("id"),
|
||
"outcomes": outcomes,
|
||
"token_map_rough": token_map,
|
||
"up_pct": up_pct,
|
||
"down_pct": dn_pct,
|
||
"up": up,
|
||
"down": dn,
|
||
"note": "UP/DOWN% 使用 /price 的 bid/ask 计算 mid 并归一化；book 为订单簿 topN。",
|
||
}
|
||
|
||
if __name__ == "__main__":
|
||
print(polymarket_updown("eth-updown-15m-1767069900", decimals=0, top_n=10))
|