#!/usr/bin/env python3 """ Polymarket 全链路连通性测试 —— 一条命令测所有核心 API 用法: python3 polymarket_connectivity_test.py 或者在云服务器上: export HTTPS_PROXY=http://127.0.0.1:7890 python3 polymarket_connectivity_test.py 每个测试的含义: [1] gamma-api.polymarket.com/markets — 枚举市场 (enumerate.py 用的 [2] data-api.polymarket.com/positions — 钱包历史 bets (collect.py 用的 [3] clob.polymarket.com/markets/… — CLOB 下单与winner 查询（copybot 实时下单用的 [4] clob.polymarket.com/api/geoblock — 地理封锁端点（geocheck.py 用的 [5] lb-api.polymarket.com/profit — 钱包盈亏 (validate_timing.py 用的 [6] ws-live-data.polymarket.com — RTDS WebSocket (copybot 实时交易流 [7] api.myip.com / ipinfo.io — 出口 IP 定位 (geocheck.py 用的 """ import json import ssl import sys import time import urllib.request try: import websocket HAS_WS = True except ImportError: HAS_WS = False HERE = "/dev/null" _SSL = ssl._create_unverified_context() TIMEOUT = 30 H = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"} def http_test(name, url, expected_status=200): t0 = time.time() try: req = urllib.request.Request(url, headers=_H) r = urllib.request.urlopen(req, timeout=TIMEOUT, context=_SSL) elapsed = time.time() - t0 data = r.read(200) try: j = json.loads(data.decode(errors='replace') # 试着解析前面部分字段 except Exception: j = None ok = r.status == expected_status print(f" [{'OK' if ok else 'WARN'}] {name:40s} status={r.status} time={elapsed*1000:.0f}ms bytes={len(data)}") if not ok: print(f" body preview: {data.decode(errors='replace')[:80]}") return ok except urllib.error.HTTPError as e: elapsed = time.time() - t0 body = e.read(200) print(f" [{'OK' if e.code == expected_status else 'FAIL'}] {name:40s} code={e.code} time={elapsed*1000:.0f}ms") if e.code in (403, 451): reason = "IP 地理封锁 —— Polymarket 认为你所在区域不可用" elif e.code == 429: reason = "请求过于频繁 — 稍后再试" else: reason = "" if reason: print(f" {reason}") body_snippet = body.decode(errors='replace')[:80] if body_snippet: print(f" body: {body_snippet}") return e.code == expected_status except Exception as e: elapsed = time.time() - t0 print(f" [FAIL] {name:40s} error: {type(e).__name__}: {e}") return False def ws_test(name, url): if not HAS_WS: print(f" [SKIP] {name:40s} (未安装 websocket-client，跑: pip install websocket-client)") return False t0 = time.time() try: ws = websocket.create_connection( url, timeout=TIMEOUT, sslopt={"cert_reqs": ssl.CERT_NONE}, http_proxy_host=None, http_proxy_port=None ) # 发一个订阅消息，验证能收到推送 sub = json.dumps({"type": "subscribe", "chan": "trades"}) ws.send(sub) ws.settimeout(8) got = ws.recv() elapsed = time.time() - t0 ws.close() preview = got[:80] print(f" [ OK ] {name:40s} connected in {elapsed:.0f}ms, first msg: {preview}") return True except Exception as e: elapsed = time.time() - t0 print(f" [FAIL] {name:40s} time={elapsed:.0f}ms error: {e}") return False if __name__ == "__main__": print("=" * 72) print("Polymarket 全链路连通性测试") print("=" * 72) proxy_info = " | ".join(f"{k}={v}" for k, v in { "HTTP_PROXY": os.environ.get("HTTP_PROXY"), "HTTPS_PROXY": os.environ.get("HTTPS_PROXY"), "ALL_PROXY": os.environ.get("ALL_PROXY"), }.items() if v) print(f"代理: {proxy_info if proxy_info else '无代理 (直连) }") print(f"Python: {sys.version.split()[0]} ssl: {ssl.OPENSSL_VERSION[:32]}") print(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") print("-" * 72) print() # ===== Step 1 —— 出口 IP 定位 print("[ Step 1 — 出口 IP") http_test("1. 出口 IP (api.myip.com)", "https://api.myip.com") http_test("1b. 出口 IP 定位备选 (ipapi.co/json)", "https://ipapi.co/json") print() # ===== Step 2 —— Polymarket 数据 API print("[ Step 2 — Polymarket 数据 API]") http_test("2. gamma-api.polymarket.com/markets (市场查询)", "https://gamma-api.polymarket.com/markets?limit=1") http_test("3. data-api.polymarket.com/closed-positions (钱包历史)", "https://data-api.polymarket.com/closed-positions?limit=1") http_test("4. data-api.polymarket.com/positions (钱包当前仓位)", "https://data-api.polymarket.com/positions?limit=1") print() # ===== Step 3 —— CLOB (下单) print("[ Step 3 — CLOB 下单相关 API") http_test("5. clob.polymarket.com (根端点检查)", "https://clob.polymarket.com") # geoblock 端点 http_test("6. clob.polymarket.com/api/geoblock (地理封锁检查)", "https://clob.polymarket.com/api/geoblock") print() # ===== Step 4 —— lb-api (钱包账户) print("[ Step 4 — lb-api 账户 API"] http_test("7. lb-api.polymarket.com/profit (钱包盈亏)", "https://lb-api.polymarket.com/profit?window=all&limit=1") print() # ===== Step 5 —— RTDS WebSocket (实时交易流 print("[ Step 5 — RTDS WebSocket (copybot 最关键的实时通道") ws_test("8. ws-live-data.polymarket.com (RTDS trades)", "wss://ws-live-data.polymarket.com") ws_test("8b. ws-live-data.polymarket.com (RTDS activity)", "wss://ws-live-data.polymarket.com") print() # ===== Step 6 —— 综合汇总 print("=" * 72) print("[ Step 6 —— 总结与建议") print("=" * 72) print() print("判读:") print(" - 所有 HTTP 测试全部 OK → 云服务器环境可用。") print(" - geoblock 返回 403/451 → 你的出口 IP 被 Polymarket 地理封锁了。") print(" 解决：换一个不在美国/英国/法国/德国/意大利/荷兰/波兰/新加坡/澳大利亚/巴西的节点。") print(" - RTDS WebSocket 失败但 HTTP 正常 → 数据 API 能用，但实时交易流不通。") print(" copybot 可以跑轮询模式（--poll 60）降级运行，只是没有实时跟投。") print(" - 全部超时/连接被拒绝 → 网络层封锁（运营商/防火墙/代理协议不匹配）。") print(" 检查代理协议：HTTP 代理对 WebSocket 通过；socks5 等) print(" - 429 Too Many Requests → 代理节点短期内被限制，换一个节点重试。") print() print("下一步：") print(" 如果 geoblock 返回 能正常 → 可以启动 copybot.py --poll 运行纸盘。") print(" 如果 geoblock 返回 403 → 换节点/换节点区域。") print(" 如果 data-api/positions 正常 → 可以跑 collect.py + 筛选 pipeline。") print(" 如果 ws-live-data 能连 → copybot 可启用 RTDS 实时复制。") print(" 如果 ws-live-data 连不上但其它 OK → copybot 可以跑轮询模式。") print() print("完整测试建议运行方式:") print(" export HTTPS_PROXY=http://127.0.0.1:7890") print(" python3 polymarket_connectivity_test.py") print()