""" Polymarket API 客户端 用于与Polymarket CLOB API交互，支持： - 搜索BTC相关市场 - 获取订单簿和价格 - 下单交易 """ import os import httpx from typing import Optional, Dict, List, Any from dataclasses import dataclass from dotenv import load_dotenv # 加载环境变量 load_dotenv() @dataclass class MarketInfo: """市场信息""" condition_id: str question: str outcomes: List[str] outcome_prices: List[float] tokens: List[Dict] volume: float liquidity: float end_date: str class PolymarketClient: """Polymarket API客户端""" GAMMA_API = "https://gamma-api.polymarket.com" CLOB_API = "https://clob.polymarket.com" def __init__(self, private_key: Optional[str] = None): """ 初始化客户端 Args: private_key: 钱包私钥(用于交易)，如不提供则只能读取数据 """ self.private_key = private_key or os.getenv("POLYMARKET_PRIVATE_KEY") # 配置代理 proxy = os.getenv("SOCKS5_PROXY") if os.getenv("USE_PROXY", "false").lower() == "true" else None # HTTP客户端 - 禁用压缩避免brotli问题 self.http = httpx.Client( timeout=30.0, proxy=proxy, headers={'Accept-Encoding': 'identity'} # 禁用压缩 ) # 如果有私钥，初始化CLOB客户端 self.clob_client = None if self.private_key: self._init_clob_client() def _init_clob_client(self): """初始化CLOB交易客户端""" try: from py_clob_client.client import ClobClient from py_clob_client.clob_types import ApiCreds # Polygon链ID chain_id = 137 # 初始化客户端 self.clob_client = ClobClient( host=self.CLOB_API, key=self.private_key, chain_id=chain_id, ) # 获取或创建API凭证 self.clob_client.set_api_creds(self.clob_client.create_or_derive_api_creds()) print("✓ CLOB客户端初始化成功") except Exception as e: print(f"⚠️ CLOB客户端初始化失败: {e}") self.clob_client = None def search_markets(self, query: str = "BTC", limit: int = 20) -> List[Dict]: """ 搜索市场 Args: query: 搜索关键词 limit: 返回数量限制 Returns: 市场列表 """ try: response = self.http.get( f"{self.GAMMA_API}/markets", params={ "active": "true", "closed": "false", "limit": limit, } ) response.raise_for_status() markets = response.json() # 过滤包含关键词的市场 filtered = [m for m in markets if query.lower() in m.get("question", "").lower()] return filtered except Exception as e: print(f"搜索市场失败: {e}") return [] def search_btc_1h_markets(self) -> List[Dict]: """ 搜索BTC 1小时价格预测市场 Returns: BTC 1H相关市场列表 """ try: response = self.http.get( f"{self.GAMMA_API}/markets", params={ "active": "true", "closed": "false", "limit": 100, } ) response.raise_for_status() markets = response.json() # 过滤BTC 1小时相关市场 btc_1h_markets = [] keywords = ["btc", "bitcoin"] time_keywords = ["1h", "1 hour", "hourly", "hour"] for m in markets: question = m.get("question", "").lower() # 检查是否包含BTC关键词 has_btc = any(kw in question for kw in keywords) # 检查是否包含1小时关键词 has_1h = any(kw in question for kw in time_keywords) if has_btc and has_1h: btc_1h_markets.append(m) return btc_1h_markets except Exception as e: print(f"搜索BTC 1H市场失败: {e}") return [] def get_market_info(self, condition_id: str) -> Optional[MarketInfo]: """ 获取市场详细信息 Args: condition_id: 市场条件ID Returns: 市场信息对象 """ try: response = self.http.get(f"{self.GAMMA_API}/markets/{condition_id}") response.raise_for_status() data = response.json() return MarketInfo( condition_id=data.get("conditionId", ""), question=data.get("question", ""), outcomes=data.get("outcomes", []), outcome_prices=data.get("outcomePrices", []), tokens=data.get("tokens", []), volume=float(data.get("volume", 0)), liquidity=float(data.get("liquidity", 0)), end_date=data.get("endDate", ""), ) except Exception as e: print(f"获取市场信息失败: {e}") return None def get_orderbook(self, token_id: str) -> Dict: """ 获取订单簿 Args: token_id: 代币ID Returns: 订单簿数据 """ try: response = self.http.get( f"{self.CLOB_API}/book", params={"token_id": token_id} ) response.raise_for_status() return response.json() except Exception as e: print(f"获取订单簿失败: {e}") return {} def get_price(self, token_id: str) -> Optional[float]: """ 获取代币当前价格 Args: token_id: 代币ID Returns: 当前价格(0-1之间) """ try: response = self.http.get( f"{self.CLOB_API}/price", params={"token_id": token_id, "side": "buy"} ) response.raise_for_status() data = response.json() return float(data.get("price", 0)) except Exception as e: print(f"获取价格失败: {e}") return None def place_market_order( self, token_id: str, side: str, # "BUY" or "SELL" amount: float, # USDC金额 ) -> Optional[Dict]: """ 下市价单 Args: token_id: 代币ID side: 买卖方向 "BUY" 或 "SELL" amount: 下注金额(USDC) Returns: 订单响应 """ if not self.clob_client: print("❌ CLOB客户端未初始化，无法下单") return None try: from py_clob_client.order_builder.constants import BUY, SELL order_side = BUY if side.upper() == "BUY" else SELL # 创建市价单 order = self.clob_client.create_market_order( token_id=token_id, amount=amount, side=order_side, ) # 提交订单 response = self.clob_client.post_order(order) print(f"✓ 订单已提交: {response}") return response except Exception as e: print(f"❌ 下单失败: {e}") return None def place_limit_order( self, token_id: str, side: str, # "BUY" or "SELL" price: float, # 价格(0-1) size: float, # 数量 ) -> Optional[Dict]: """ 下限价单 Args: token_id: 代币ID side: 买卖方向 "BUY" 或 "SELL" price: 价格(0-1之间) size: 购买数量 Returns: 订单响应 """ if not self.clob_client: print("❌ CLOB客户端未初始化，无法下单") return None try: from py_clob_client.order_builder.constants import BUY, SELL order_side = BUY if side.upper() == "BUY" else SELL # 创建限价单 order = self.clob_client.create_limit_order( token_id=token_id, price=price, size=size, side=order_side, ) # 提交订单 response = self.clob_client.post_order(order) print(f"✓ 限价单已提交: {response}") return response except Exception as e: print(f"❌ 下单失败: {e}") return None def get_positions(self) -> List[Dict]: """ 获取当前仓位 Returns: 仓位列表 """ if not self.clob_client: print("❌ CLOB客户端未初始化") return [] try: positions = self.clob_client.get_positions() return positions except Exception as e: print(f"获取仓位失败: {e}") return [] def cancel_order(self, order_id: str) -> bool: """ 取消订单 Args: order_id: 订单ID Returns: 是否成功 """ if not self.clob_client: print("❌ CLOB客户端未初始化") return False try: self.clob_client.cancel(order_id) print(f"✓ 订单已取消: {order_id}") return True except Exception as e: print(f"❌ 取消订单失败: {e}") return False def close(self): """关闭客户端""" self.http.close() # 测试函数 def test_search_markets(): """测试搜索市场功能""" client = PolymarketClient() print("搜索BTC相关市场...") markets = client.search_markets("BTC", limit=10) if markets: print(f"\n找到 {len(markets)} 个BTC相关市场:\n") for m in markets[:5]: print(f" 问题: {m.get('question', 'N/A')}") print(f" 条件ID: {m.get('conditionId', 'N/A')}") print(f" 结果: {m.get('outcomes', [])}") print(f" 价格: {m.get('outcomePrices', [])}") print("-" * 50) else: print("未找到BTC相关市场") print("\n搜索BTC 1H市场...") btc_1h = client.search_btc_1h_markets() if btc_1h: print(f"找到 {len(btc_1h)} 个BTC 1H市场") for m in btc_1h: print(f" {m.get('question', 'N/A')}") else: print("未找到BTC 1H市场") client.close() if __name__ == "__main__": test_search_markets()