提交代码

polymarket 优化抓取.py

@@ -0,0 +1,144 @@
+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-1766912400", decimals=0, top_n=10))