winning-wallet-finder/polymarket_connectivity_test.py

#!/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()