PolyEdge Python SDK

Official Python SDK for PolyEdge — Ultra-Low-Latency Polymarket Mempool Trade Streaming & Real-Time On-Chain Analytics API.

Engineered for Polymarket copy-trading bots, professional traders, and prediction market builders. Capture pending trades pre-block and track smart money with institutional-grade on-chain analytics.

Key Features

• 4 First-Class Namespaces: Aligned with OpenAPI 3.1:
  • client.streams: Real-time SSE trade streaming, sessions & filter management.
  • client.analytics: Polymarket on-chain trade attribution, trader leaderboards, market settlement audit.
  • client.account: User profile, API keys, request telemetry, and ledger.
  • client.subscription: Pricing tier catalog, upgrade quotes, and promo codes.
• Asynchronous SSE Stream Engine:
  • 45s Keep-Alive Watchdog: Native httpx read-timeout detection automatically cleans up dead socket half-open connections.
  • Automatic Resume with Last-Event-ID: Seamlessly retrieves buffered trades on reconnection.
  • Fail-Fast Policy: Immediate non-retry termination on fatal HTTP 401/403/404 errors.
  • Pythonic Async Iterator: Streamlined async for tx in stream.subscribe(): interface.
• Pre-Block Polymarket Trade Streaming: Peered with 100+ Polygon nodes to capture pending Polymarket trades pre-block. The World's Fastest — verify yourself.

Installation

pip install polyedge

Quick Start

1. Initialize Client

import os
from polyedge import PolyEdgeClient

client = PolyEdgeClient(api_key=os.environ.get("POLYEDGE_API_KEY", "your_polyedge_api_key"))

2. Real-Time Mempool Trade Stream (SSE)

import asyncio
from polyedge import PolyEdgeClient

async def main():
    client = PolyEdgeClient(api_key="your_polyedge_api_key")

    stream = client.streams.connect("your_stream_id", heartbeat_timeout=45.0)

    # Optional heartbeat listener
    stream.on_heartbeat(lambda: print("💓 Server heartbeat received"))

    async for tx in stream.subscribe():
        tx_hash = tx.get("tx_hash", "")[:18]
        question = tx.get("market", {}).get("question", "")
        taker = tx.get("taker", {})
        outcome = taker.get("outcome")
        side = taker.get("side")
        shares = taker.get("shares")
        makers_count = len(tx.get("makers", []))

        print(f"⚡ [Live Trade] Tx: {tx_hash}...")
        print(f"   Market: \"{question}\"")
        print(f"   Taker: {outcome} ({side}) | Shares: {shares}")
        print(f"   Matched Makers: {makers_count} orders")

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

3. Analytics: Leaderboard & Trader Dossier

# Leaderboard
leaderboard = client.analytics.get_leaderboard(limit=5)
print(f"Total active traders tracked: {leaderboard.get('total_count')}")

for i, t in enumerate(leaderboard.get("traders", [])):
    addr = t.get("profile", {}).get("address")
    name = t.get("profile", {}).get("name") or "Anonymous"
    pnl_usd = (t.get("total_pnl", 0)) / 1e6
    print(f"#{i + 1} {addr} ({name}) - Realized PnL: ${pnl_usd:,.2f}")

# Deep Trader Profile
profile = client.analytics.get_trader("0x51698a47f840a242abc2ca0351371c7ffac41842")
print(f"Taker Rebate Tier: {profile.get('taker_tier_name')}")

4. Account Profile & Telemetry

# Account info
profile = client.account.get_profile()
print(f"Email: {profile.get('email')}, Tier: {profile.get('tier')}")

# Daily telemetry
telemetry = client.account.get_telemetry()
print(f"Total API calls today: {telemetry.get('total_requests')}")

Complete API Service Reference (28 Methods)

1. Streams Service (client.streams)

2. Analytics Service (client.analytics)

3. Account Service (client.account)

4. Subscription Service (client.subscription)

License

MIT © 2026 PolyEdge Labs