gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement Polymarket API client for market and price data using py-clob-client with REST API fallbacks.

Browse Source
This commit is contained in:
2569718930@qq.com
2026-02-07 01:56:47 +08:00
parent 499046424c
commit f9c425b984
2 changed files with 196 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data_collection/polymarket_api.py
+85 -33
View File
@@ -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]:
"""
获取中点价格 (官方接口)
test_price.py
+111
View File
@@ -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()