PolyEdge 官方 SDK 使用指南

PolyEdge 为 TypeScript / Node.js、Python、Rust 和 Go 提供了官方客户端开发库，用于超低时延出块前内存池Polymarket交易推流及调用链上 Polymarket 交易分析 API。

内置连接稳定性保障

• 45 秒保活看门狗（Watchdog）：自动检测心跳中断与静默 TCP 半开连接，触发平滑重连。
• Last-Event-ID 增量恢复：重连时携带最新事件 ID，服务端自动补发缓冲期内的出块前内存池Polymarket交易。

---

TypeScript / Node.js

npm install polyedge

import { PolyEdgeClient } from "polyedge";

const client = new PolyEdgeClient({
  apiKey: process.env.POLYEDGE_API_KEY || "pe_live_your_api_key",
});

async function main() {
  // 1. 获取顶级交易员排行榜
  const leaderboard = await client.analytics.getLeaderboard({ limit: 5 });
  console.log(`全网追踪交易员总数: ${leaderboard.total_count}`);
  leaderboard.traders?.forEach((t, i) => {
    console.log(
      `#${i + 1} ${t.profile?.address} (${t.profile?.name || "匿名交易员"}) | 收益: $${t.total_pnl} | 胜率: ${t.win_rate}%`
    );
  });

  // 2. 创建按 Tag 和 Series 过滤的交易流
  const stream = await client.streams.create({
    name: "Crypto & Solana 5M Stream",
    tags: ["crypto"],
    series: ["sol-up-or-down-5m"],
  });
  console.log(`流创建成功: ${stream.id}`);

  // 3. 连接实时内存池 SSE 数据流
  const conn = client.streams.connect(stream.id, { heartbeatTimeoutMs: 45_000 });

  // 方式 A: 现代异步迭代器 (for await)
  for await (const tx of conn) {
    console.log(`⚡ [Tx] ${tx.tx_hash.slice(0, 18)}... | ${tx.market?.question}`);
    console.log(`   吃单: ${tx.taker?.outcome} (${tx.taker?.side}) | 份额: ${tx.taker?.shares} | USDC: ${tx.taker?.usdc}`);
  }

  // 方式 B: 事件监听器
  // conn.on("tx", (tx, id) => {
  //   console.log(`⚡ [Tx #${id}] ${tx.tx_hash.slice(0, 18)}... | ${tx.market?.question}`);
  // });
  // conn.on("error", (err) => console.error("数据流异常:", err));
}

main().catch(console.error);

---

Python

pip install polyedge

import os
import asyncio
from polyedge import PolyEdgeClient

client = PolyEdgeClient(api_key=os.getenv("POLYEDGE_API_KEY", "pe_live_your_api_key"))

async def main():
    # 1. 获取顶级交易员排行榜
    leaderboard = client.analytics.get_leaderboard(limit=5)
    print(f"全网追踪交易员总数: {leaderboard.get('total_count')}")
    for i, t in enumerate(leaderboard.get("traders", [])):
        profile = t.get("profile", {})
        print(f"#{i + 1} {profile.get('address')} ({profile.get('name') or '匿名交易员'}) | 收益: ${t.get('total_pnl')} | 胜率: {t.get('win_rate')}%")

    # 2. 创建按 Tag 和 Series 过滤的交易流
    stream = client.streams.create({
        "name": "Crypto & Solana 5M Stream",
        "tags": ["crypto"],
        "series": ["sol-up-or-down-5m"],
    })
    print(f"流创建成功: {stream['id']}")

    # 3. 连接实时内存池 SSE 数据流 (默认 45 秒看门狗保活)
    conn = client.streams.connect(stream["id"], heartbeat_timeout=45.0)
    try:
        async for tx in conn.subscribe():
            taker = tx.get("taker", {})
            print(f"⚡ [Tx] {tx.get('tx_hash', '')[:18]}... | {tx.get('market', {}).get('question')}")
            print(f"   吃单: {taker.get('outcome')} ({taker.get('side')}) | 份额: {taker.get('shares')} | USDC: {taker.get('usdc')}")
    finally:
        conn.close()

if __name__ == "__main__":
    asyncio.run(main())

---

Rust

cargo add polyedge tokio --features full futures-util

use futures_util::StreamExt;
use polyedge::client::UserCreateStreamRequest;
use polyedge::PolyEdgeClient;
use std::env;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let api_key = env::var("POLYEDGE_API_KEY").unwrap_or_else(|_| "pe_live_your_api_key".to_string());
    let client = PolyEdgeClient::new(&api_key);

    // 1. 获取顶级交易员排行榜
    let leaderboard = client.analytics().get_leaderboard(Some(5), None).await?;
    println!("全网追踪交易员总数: {:?}", leaderboard.total_count);
    for (i, t) in leaderboard.traders.unwrap_or_default().iter().enumerate() {
        let addr = t.profile.as_ref().and_then(|p| p.address.as_deref()).unwrap_or("unknown");
        let name = t.profile.as_ref().and_then(|p| p.name.as_deref()).unwrap_or("匿名交易员");
        println!("#{i + 1} {addr} ({name}) | 收益: ${:?} | 胜率: {:?}%", t.total_pnl, t.win_rate);
    }

    // 2. 创建按 Tag 和 Series 过滤的交易流
    let req = UserCreateStreamRequest {
        name: Some("Crypto & Solana 5M Stream".to_string()),
        tags: Some(vec!["crypto".to_string()]),
        series: Some(vec!["sol-up-or-down-5m".to_string()]),
        addresses: None,
    };
    let stream = client.streams().create(&req).await?;
    let stream_id = stream.id.unwrap_or_default();
    println!("流创建成功: {}", stream_id);

    // 3. 连接实时内存池 SSE 数据流
    let conn = client.streams().connect(&stream_id);
    let mut stream_iter = Box::pin(conn.subscribe());

    while let Some(item) = stream_iter.next().await {
        if let Ok(tx) = item {
            let hash = tx.tx_hash.unwrap_or_default();
            let short_hash = if hash.len() > 18 { &hash[..18] } else { &hash };
            let question = tx.market.and_then(|m| m.question).unwrap_or_default();
            let taker = tx.taker;
            let outcome = taker.as_ref().and_then(|t| t.outcome.as_deref()).unwrap_or_default();
            let side = taker.as_ref().and_then(|t| t.side.as_ref());
            let shares = taker.as_ref().and_then(|t| t.shares.as_deref()).unwrap_or_default();

            println!("⚡ [Tx] {}... | {}", short_hash, question);
            println!("   吃单: {} ({:?}) | 份额: {}", outcome, side, shares);
        }
    }

    Ok(())
}

---

Go

go get github.com/PolyEdgeDev/polyedge-go

package main

import (
	"context"
	"fmt"
	"os"
	"time"

	polyedge "github.com/PolyEdgeDev/polyedge-go"
)

func main() {
	apiKey := os.Getenv("POLYEDGE_API_KEY")
	if apiKey == "" {
		apiKey = "pe_live_your_api_key"
	}

	client, err := polyedge.NewClient(apiKey)
	if err != nil {
		panic(err)
	}

	ctx := context.Background()

	// 1. 获取顶级交易员排行榜
	leaderboard, err := client.Analytics.GetLeaderboard(ctx, 5, 0)
	if err != nil {
		panic(err)
	}
	fmt.Printf("全网追踪交易员总数: %d\n", leaderboard.TotalCount)
	for i, t := range leaderboard.Traders {
		name := t.Profile.Name
		if name == "" {
			name = "匿名交易员"
		}
		fmt.Printf("#%d %s (%s) | 收益: $%d | 胜率: %.2f%%\n",
			i+1, t.Profile.Address, name, t.TotalPnl, t.WinRate)
	}

	// 2. 创建按 Tag 和 Series 过滤的交易流
	stream, err := client.Streams.Create(ctx, &polyedge.UserCreateStreamRequest{
		Name:   "Crypto & Solana 5M Stream",
		Tags:   []string{"crypto"},
		Series: []string{"sol-up-or-down-5m"},
	})
	if err != nil {
		panic(err)
	}
	fmt.Printf("流创建成功: %s\n", stream.ID)

	// 3. 连接实时内存池 SSE 数据流
	conn, err := client.Streams.Connect(stream.ID, polyedge.StreamOptions{
		HeartbeatTimeout: 45 * time.Second,
	})
	if err != nil {
		panic(err)
	}

	txCh, errCh := conn.Subscribe(ctx)
	for {
		select {
		case tx, ok := <-txCh:
			if !ok {
				return
			}
			hash := tx.TxHash
			if len(hash) > 18 {
				hash = hash[:18]
			}
			fmt.Printf("⚡ [Tx] %s... | %s\n", hash, tx.Market.Question)
			fmt.Printf("   吃单: %s (%s) | 份额: %s | USDC: %s\n",
				tx.Taker.Outcome, tx.Taker.Side, tx.Taker.Shares, tx.Taker.USDC)

		case err, ok := <-errCh:
			if ok && err != nil {
				fmt.Printf("数据流异常: %v\n", err)
			}

		case <-ctx.Done():
			return
		}
	}
}