Official PolyEdge SDKs

PolyEdge provides official client libraries for TypeScript / Node.js, Python, Rust, and Go to stream Polymarket mempool trades with ultra-low latency and query the on-chain Polymarket trade analytics API.

Built-in Streaming Reliability

• 45s Keep-Alive Watchdog: Automatically detects silent TCP half-open connection drops and reconnects.
• Auto-Resume with Last-Event-ID: Replays buffered trades upon reconnection to prevent missed events.

---

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. Query Top Traders Leaderboard
  const leaderboard = await client.analytics.getLeaderboard({ limit: 5 });
  console.log(`Tracked Traders: ${leaderboard.total_count}`);
  leaderboard.traders?.forEach((t, i) => {
    console.log(
      `#${i + 1} ${t.profile?.address} (${t.profile?.name || "Anonymous"}) | PnL: $${t.total_pnl} | WinRate: ${t.win_rate}%`
    );
  });

  // 2. Create Stream Filtered by Tag and Series
  const stream = await client.streams.create({
    name: "Crypto & Solana 5M Stream",
    tags: ["crypto"],
    series: ["sol-up-or-down-5m"],
  });
  console.log(`Stream created: ${stream.id}`);

  // 3. Connect to Real-Time Mempool SSE Feed
  const conn = client.streams.connect(stream.id, { heartbeatTimeoutMs: 45_000 });

  // Option A: Modern Async Iterator (for await)
  for await (const tx of conn) {
    console.log(`⚡ [Tx] ${tx.tx_hash.slice(0, 18)}... | ${tx.market?.question}`);
    console.log(`   Taker: ${tx.taker?.outcome} (${tx.taker?.side}) | Shares: ${tx.taker?.shares} | USDC: ${tx.taker?.usdc}`);
  }

  // Option B: Event Listener
  // conn.on("tx", (tx, id) => {
  //   console.log(`⚡ [Tx #${id}] ${tx.tx_hash.slice(0, 18)}... | ${tx.market?.question}`);
  // });
  // conn.on("error", (err) => console.error("Stream 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. Query Top Traders Leaderboard
    leaderboard = client.analytics.get_leaderboard(limit=5)
    print(f"Tracked Traders: {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 'Anonymous'}) | PnL: ${t.get('total_pnl')} | WinRate: {t.get('win_rate')}%")

    # 2. Create Stream Filtered by Tag and Series
    stream = client.streams.create({
        "name": "Crypto & Solana 5M Stream",
        "tags": ["crypto"],
        "series": ["sol-up-or-down-5m"],
    })
    print(f"Stream created: {stream['id']}")

    # 3. Connect to Real-Time Mempool SSE Feed (default 45s watchdog)
    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: {taker.get('outcome')} ({taker.get('side')}) | Shares: {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. Query Top Traders Leaderboard
    let leaderboard = client.analytics().get_leaderboard(Some(5), None).await?;
    println!("Tracked Traders: {:?}", 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("Anonymous");
        println!("#{i + 1} {addr} ({name}) | PnL: ${:?} | WinRate: {:?}%", t.total_pnl, t.win_rate);
    }

    // 2. Create Stream Filtered by Tag and 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 created: {}", stream_id);

    // 3. Connect to Real-Time Mempool SSE Feed
    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!("   Taker: {} ({:?}) | Shares: {}", 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. Query Top Traders Leaderboard
	leaderboard, err := client.Analytics.GetLeaderboard(ctx, 5, 0)
	if err != nil {
		panic(err)
	}
	fmt.Printf("Tracked Traders: %d\n", leaderboard.TotalCount)
	for i, t := range leaderboard.Traders {
		name := t.Profile.Name
		if name == "" {
			name = "Anonymous"
		}
		fmt.Printf("#%d %s (%s) | PnL: $%d | WinRate: %.2f%%\n",
			i+1, t.Profile.Address, name, t.TotalPnl, t.WinRate)
	}

	// 2. Create Stream Filtered by Tag and 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("Stream created: %s\n", stream.ID)

	// 3. Connect to Real-Time Mempool SSE Feed
	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("   Taker: %s (%s) | Shares: %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("Stream error: %v\n", err)
			}

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