diff --git a/src/data_collection/polymarket_api.py b/src/data_collection/polymarket_api.py index 136a2527..f4ff209a 100644 --- a/src/data_collection/polymarket_api.py +++ b/src/data_collection/polymarket_api.py @@ -105,28 +105,52 @@ class PolymarketClient: def get_price(self, token_id: str, side: str = "ask") -> Optional[float]: """ - 获取 Token 的实时盘口价格 (纯官方库实现) + 获取 Token 的实时盘口价格 + 优先使用官方库，失败时使用直接 REST API """ + # 方法1: 尝试官方库 try: - # 用户侧语义：BUY 代表我要买 (Ask)，SELL 代表我要卖 (Bid) sdk_side = "BUY" if side == "ask" else "SELL" price_str = self.clob_client.get_price(token_id=token_id, side=sdk_side) if price_str: return float(price_str) except Exception as e: logger.debug(f"官方库 get_price 失败 ({token_id}): {e}") + + # 方法2: 直接调用 REST API (与 test_price.py 一致) + try: + resp = self.session.get(f"{self.base_url}/price", params={ + "token_id": token_id, + "side": side.upper() + }, timeout=10) + data = resp.json() + return float(data.get("price", 0)) + except Exception as e: + logger.debug(f"REST API get_price 失败 ({token_id}): {e}") return None def get_orderbook(self, token_id: str) -> Optional[Dict]: """ - 获取订单簿深度 (纯官方库实现) + 获取订单簿深度 + 优先使用官方库，失败时使用直接 REST API """ + # 方法1: 尝试官方库 try: return self.clob_client.get_orderbook(token_id=token_id) except Exception as e: logger.debug(f"官方库 get_orderbook 失败 ({token_id}): {e}") + + # 方法2: 直接调用 REST API (与 test_price.py 一致) + try: + resp = self.session.get(f"{self.base_url}/book", params={ + "token_id": token_id + }, timeout=10) + return resp.json() + except Exception as e: + logger.debug(f"REST API get_orderbook 失败 ({token_id}): {e}") return None + def get_buy_prices(self, yes_token_id: str, no_token_id: str) -> Optional[Dict]: """ 获取买入价格 (Buy Yes 和 Buy No) @@ -171,7 +195,8 @@ class PolymarketClient: def get_multiple_prices(self, token_requests: List[Dict]) -> Dict[str, float]: """ - 批量获取多个 token 的价格 (官方接口 + 线程池并行实现) + 批量获取多个 token 的价格 + 优先使用官方库，失败时使用直接 REST API """ if not token_requests: return {} @@ -188,43 +213,70 @@ class PolymarketClient: chunks = [token_requests[i : i + batch_size] for i in range(0, len(token_requests), batch_size)] - def fetch_chunk(chunk): - for attempt in range(3): + def fetch_chunk_sdk(chunk): + """使用官方 SDK 批量获取""" + try: + batch_req = [] + for r in chunk: + sdk_side = "BUY" if r.get("side") == "ask" else "SELL" + batch_req.append(BookParams(token_id=r["token_id"], side=sdk_side)) + + results = self.clob_client.get_prices(batch_req) + + chunk_prices = {} + if isinstance(results, list): + for item in results: + tid = item.get("token_id") + price_raw = item.get("price") + res_side = item.get("side") + if tid and price_raw: + val = robust_float(price_raw) + key_side = "ask" if res_side == "BUY" else "bid" + chunk_prices[f"{tid}:{key_side}"] = val + return chunk_prices + except Exception as e: + logger.debug(f"SDK batch fetch failed: {e}") + return None + + def fetch_chunk_rest(chunk): + """使用 REST API 逐个获取 (备用方案)""" + chunk_prices = {} + for r in chunk: try: - batch_req = [] - for r in chunk: - sdk_side = "BUY" if r.get("side") == "ask" else "SELL" - batch_req.append(BookParams(token_id=r["token_id"], side=sdk_side)) - - results = self.clob_client.get_prices(batch_req) - - chunk_prices = {} - if isinstance(results, list): - for item in results: - tid = item.get("token_id") - price_raw = item.get("price") - res_side = item.get("side") - if tid and price_raw: - val = robust_float(price_raw) - key_side = "ask" if res_side == "BUY" else "bid" - chunk_prices[f"{tid}:{key_side}"] = val - return chunk_prices + resp = self.session.get(f"{self.base_url}/price", params={ + "token_id": r["token_id"], + "side": r.get("side", "ask").upper() + }, timeout=10) + data = resp.json() + price = robust_float(data.get("price", 0)) + if price > 0: + chunk_prices[f"{r['token_id']}:{r.get('side', 'ask')}"] = price except Exception as e: - if attempt < 2: - time.sleep(0.5 * (attempt + 1)) - continue - logger.warning(f"Batch fetch failed after 3 attempts: {e}") - return {} + logger.debug(f"REST API price fetch failed for {r['token_id'][:16]}...: {e}") + return chunk_prices - # 使用更保守的线程池并发抓取 + # 使用线程池并发抓取 with ThreadPoolExecutor(max_workers=3) as executor: - future_results = list(executor.map(fetch_chunk, chunks)) - + future_results = list(executor.map(fetch_chunk_sdk, chunks)) + + # 检查结果，如果 SDK 全部失败，使用 REST API 备用 + sdk_success = False for chunk_result in future_results: - all_prices.update(chunk_result) + if chunk_result: + all_prices.update(chunk_result) + sdk_success = True + + # 如果 SDK 完全失败，使用 REST API + if not sdk_success and token_requests: + logger.info("SDK 批量获取失败，使用 REST API 逐个获取...") + with ThreadPoolExecutor(max_workers=5) as executor: + future_results = list(executor.map(fetch_chunk_rest, chunks)) + for chunk_result in future_results: + all_prices.update(chunk_result) return all_prices + def get_midpoint(self, token_id: str) -> Optional[float]: """ 获取中点价格 (官方接口) diff --git a/test_price.py b/test_price.py new file mode 100644 index 00000000..ddd6f9a7 --- /dev/null +++ b/test_price.py @@ -0,0 +1,111 @@ +""" +Polymarket 天气市场价格查询 +直接获取 YES/NO 的真实买入价格（Ask） +""" +import requests + +# ============ 配置 ============ +# 替换成你要查的 token_id（从 Gamma API 获取） +YES_TOKEN_ID = "你的YES_token_id" +NO_TOKEN_ID = "你的NO_token_id" + +CLOB_BASE = "https://clob.polymarket.com" +GAMMA_BASE = "https://gamma-api.polymarket.com" +# ============ 方法1: 快速查价格 ============ +def get_price(token_id, side="buy"): +"""获取单个 token 的买入/卖出价格""" +resp = requests.get(f"{CLOB_BASE}/price", params={ +"token_id": token_id, +"side": side.upper() +}) +data = resp.json() +return float(data.get("price", 0)) + +# ============ 方法2: 查完整盘口 ============ +def get_orderbook(token_id): +"""获取完整 orderbook，含深度""" +resp = requests.get(f"{CLOB_BASE}/book", params={ +"token_id": token_id +}) +return resp.json() + +# ============ 方法3: 从 Gamma 发现天气市场 ============ +def discover_weather_markets(city="New York"): +"""自动发现天气市场，获取 token_id""" +resp = requests.get(f"{GAMMA_BASE}/markets", params={ +"tag": "weather", +"closed": "false", +"limit": 50 +}) +markets = resp.json() +results = [] +for m in markets: +if city.lower() in m.get("question", "").lower(): +tokens = m.get("tokens", []) +if len(tokens) >= 2: + yes_token = None + no_token = None + for t in tokens: + if t.get("outcome") == "Yes": + yes_token = t["token_id"] + else: + no_token = t["token_id"] + if yes_token and no_token: + results.append({ + "question": m["question"], + "yes_token": yes_token, + "no_token": no_token, + "slug": m.get("market_slug", "") + }) +return results + + +# ============ 主流程 ============ +def main(): +print("📡 正在发现 NYC 天气市场...\n") +markets = discover_weather_markets("New York") +if not markets: +print("❌ 未找到天气市场，检查 Gamma API") +return + +print(f"✅ 找到 {len(markets)} 个市场\n") +print("=" * 55) + +for m in markets[:10]: +q = m["question"] +yes_id = m["yes_token"] +no_id = m["no_token"] + +# 获取真实 Ask 价格 +yes_ask = get_price(yes_id, "buy") +no_ask = get_price(no_id, "buy") + +# 获取 Bid +yes_bid = get_price(yes_id, "sell") + +spread = yes_ask - yes_bid if yes_ask and yes_bid else None + +# 获取盘口深度 +book = get_orderbook(yes_id) +ask_depth = sum(float(o.get("size", 0)) for o in book.get("asks", [])[:3]) +bid_depth = sum(float(o.get("size", 0)) for o in book.get("bids", [])[:3]) + +# 流动性判断 +if ask_depth < 50: +liq = "🔴枯竭" +elif ask_depth < 500: +liq = "🟡正常" +else: +liq = "🟢充裕" + +print(f"\n📊 {q}") +print(f" YES Ask: {yes_ask*100:.1f}¢ | NO Ask: {no_ask*100:.1f}¢") +print(f" YES Bid: {yes_bid*100:.1f}¢ | Spread: {spread*100:.1f}¢" if spread else " YES Bid: --") +print(f" 深度: Ask ${ask_depth:.0f} | Bid ${bid_depth:.0f} | {liq}") +print(f" 🔗 https://polymarket.com/event/{m['slug']}") + +print("\n" + "=" * 55) +print("💡 价格单位: Ask = 你买入要付的价格") + +if __name__ == "__main__": +main() \ No newline at end of file