Skip to main content

​
Requirements

  • Python 3.7 or higher
  • pip

​
Installation

1

Install dependencies

pip install requests urllib3
2

Copy the SDK files

Download polyedge.py, parser_compact_tx.py, and sseclient.py from the PolyEdge examples repository and copy them into your project directory.
sseclient.py is a lightweight, MIT-licensed Server-Sent Events client bundled with the SDK — no additional SSE library is required.

​
Instantiation

Import PolyEdgeClient and pass your API key. The base URL defaults to https://api.polyedge.dev/v1. 
from polyedge import PolyEdgeClient

client = PolyEdgeClient('your_api_key_here')
You can also pass a custom base URL as the second argument: 
client = PolyEdgeClient('your_api_key_here', base_url='https://api.polyedge.dev/v1')
Your API key is sent on every request via the X-PolyEdge-Key header.

​
Methods

Retrieve the trader leaderboard, optionally filtered and sorted.
data = client.get_leaderboard(params)
Parameters — params is an optional dict:
KeyTypeDescription
limitintNumber of results to return. Default: 50, max: 100.
offsetintPagination offset. Default: 0.
time_framestr'1H', '4H', '1D', '3D', '7D', '30D'. Default: '30D'.
sort_bystr'pnl', 'roi', 'win_rate', 'market_count'. Default: 'pnl'.
tagstrFilter by market tag, e.g. 'crypto', 'sports'.
min_win_ratefloatMinimum win rate (0–100).
max_win_ratefloatMaximum win rate (0–100).
min_roifloatMinimum ROI percentage.
max_roifloatMaximum ROI percentage.
last_active_hoursintOnly include traders active in the last N hours.
min_market_countintMinimum number of markets traded.
max_market_countintMaximum number of markets traded.
Returns dict.
Get a trader’s detailed profile including tags, stats, and metadata.
data = client.get_trader_profile('0xabc…')
ParameterTypeDescription
addressstrThe EVM wallet address of the trader.
Returns dict.
Retrieve a trader’s trading history aggregated by market.
data = client.get_trader_markets('0xabc…', params)
ParameterTypeDescription
addressstrThe EVM wallet address of the trader.
limitintNumber of markets to return. Default: 20, max: 100.
offsetintPagination offset. Default: 0.
is_activeboolFilter by active (unresolved) markets only.
Returns dict.
Retrieve a specific trader’s order history within a specific market.
data = client.get_trader_market_orders('0xabc…', 12345)
ParameterTypeDescription
addressstrThe EVM wallet address of the trader.
market_idstr | intThe specific market ID.
Returns dict.
Retrieve detailed information about a specific market.
data = client.get_market_detail(12345)
ParameterTypeDescription
market_idstr | intThe unique numeric ID of the market.
Returns dict.
Subscribe to live mempool orders via SSE. Returns a generator that yields parsed order dicts.
for order in client.stream_orders(params):
    print(order)
Break out of the loop at any time to stop consuming the stream.Parameters — params is an optional dict:
KeyTypeDescription
tradersstrComma-separated list of trader addresses to follow.
tagsstrComma-separated list of market tags to filter by.
seriesstrComma-separated list of series slugs.
min_usdcfloatMinimum USDC trade size.
max_usdcfloatMaximum USDC trade size.
min_sharesfloatMinimum shares amount.
max_sharesfloatMaximum shares amount.
min_pricefloatMinimum price (0–100).
max_pricefloatMaximum price (0–100).
formatstrSet to 'json' for structured JSON. Omit for compact array format (recommended).
Yields dict.
When format is omitted, each yielded value is the output of parse_compact_tx — a fully structured Python dict. See Compact format parsing below.

​
Compact format parsing

By default, stream_orders receives events in a compact array format optimised for low bandwidth. The parser_compact_tx.py utility converts each raw list into a structured dict before yielding it. The parsed dict has this shape: 
{
    "txHash": str,
    "timestamp": str,
    "market": {
        "id": str,
        "question": str,
        "slug": str,
        "start_date": str,
        "end_date": str,
        "icon": str,
        "neg_risk": bool,
        "event_slug": str,
        "series_slug": str,
        "tags": list,
        "outcome_1": str,
        "outcome_2": str,
        "group_item_title": str,
        "event_title": str,
        "event_icon": str
    },
    "takerOrder": {
        "side": str,        # 'BUY' or 'SELL'
        "trader": {
            "address": str,
            "name": str,
            "x_username": str,
            "profile_image": str,
            "profile_created_at": str
        },
        "outcome": str,
        "tokenId": str,
        "usdc": str,        # raw 1e6 format
        "shares": str,      # raw 1e6 format
        "fee": str,
        "feeRateBps": str
    },
    "makerOrders": [ # same shape as takerOrder ]
}
USDC and shares values are returned as raw strings in 1e6 format. Divide by 1e6 to convert to human-readable amounts: float(order['takerOrder']['usdc']) / 1e6.
If you prefer fully structured JSON from the API rather than parsing compact arrays yourself, pass format='json' in the params dict to stream_orders — the raw response is yielded directly without calling the parser.

​
Complete example

The following example exercises every method in the SDK, then opens the SSE stream and prints the first three events. 
from polyedge import PolyEdgeClient

API_KEY = 'your_api_key_here'
client = PolyEdgeClient(API_KEY)

def format_amount(raw_str):
    if not raw_str:
        return "0.00"
    num = float(raw_str) / 1e6
    return f"{num:,.2f}"

# 1. Get leaderboard — top 3 traders by PNL over the last 7 days
leaderboard = client.get_leaderboard({'limit': 3, 'sort_by': 'pnl', 'time_frame': '7D'})
trader_address = leaderboard['traders'][0]['profile']['address']
for t in leaderboard['traders']:
    name = t['profile'].get('name') or 'Unknown'
    pnl = format_amount(t.get('total_pnl', '0'))
    print(f"{t['rank']}. {name} ({t['profile']['address']})")
    print(f"   PNL: ${pnl} | ROI: {t.get('roi', 0)}% | Win Rate: {t.get('win_rate', 0)}%")

# 2. Get trader profile
profile_response = client.get_trader_profile(trader_address)
p = profile_response.get('profile', {})
print(f"Name: {p.get('name') or 'Unknown'} | Address: {p.get('address')}")

# 3. Get trader markets
markets_response = client.get_trader_markets(trader_address, {'limit': 1})
hm = markets_response['history'][0]
market_id = hm['market']['id']
print(f"Market: {hm['market']['question']} (ID: {market_id})")

# 4. Get trader market orders
orders_response = client.get_trader_market_orders(trader_address, market_id)
for i, o in enumerate(orders_response.get('orders', [])[:3]):
    price = f"{(float(o.get('price', '0')) / 1e6):.3f}"
    print(f"[Order {i + 1}] {o.get('side')} {o.get('outcome')} @ ${price}")

# 5. Get market detail
market_detail = client.get_market_detail(market_id)
md = market_detail.get('market', {})
print(f"Question: {md.get('question')}")
print(f"Total Volume: ${format_amount(market_detail.get('total_volume', '0'))}")

# 6. Stream live orders — exit after 3 events
event_count = 0
for order in client.stream_orders():
    if not order:
        continue
    event_count += 1
    taker = order.get('takerOrder', {})
    shares = float(taker.get('shares', '0'))
    price_str = f"{float(taker.get('usdc', '0')) / shares:.3f}" if shares else "0.000"

    market = order.get('market', {})
    trader_info = taker.get('trader') or {}

    print(f"[Event #{event_count}] {order.get('timestamp')}")
    print(f"  Market: {market.get('question')} ({market.get('id')})")
    print(f"  Trader: {trader_info.get('address', 'Unknown')}")
    print(f"  Action: {taker.get('side')} {taker.get('outcome')} @ ~${price_str}")
    print(f"  Size: {format_amount(taker.get('shares', '0'))} shares (${format_amount(taker.get('usdc', '0'))})")

    if event_count >= 3:
        break