Python SDK

Requirements

Python 3.7 or higher
pip

Installation

Install dependencies

pip install requests urllib3

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

get_leaderboard(params)

Retrieve the trader leaderboard, optionally filtered and sorted.

data = client.get_leaderboard(params)

Parameters — params is an optional dict:

Key	Type	Description
`limit`	int	Number of results to return. Default: `50`, max: `100`.
`offset`	int	Pagination offset. Default: `0`.
`time_frame`	str	`'1H'`, `'4H'`, `'1D'`, `'3D'`, `'7D'`, `'30D'`. Default: `'30D'`.
`sort_by`	str	`'pnl'`, `'roi'`, `'win_rate'`, `'market_count'`. Default: `'pnl'`.
`tag`	str	Filter by market tag, e.g. `'crypto'`, `'sports'`.
`min_win_rate`	float	Minimum win rate (0–100).
`max_win_rate`	float	Maximum win rate (0–100).
`min_roi`	float	Minimum ROI percentage.
`max_roi`	float	Maximum ROI percentage.
`last_active_hours`	int	Only include traders active in the last N hours.
`min_market_count`	int	Minimum number of markets traded.
`max_market_count`	int	Maximum number of markets traded.
`min_pnl`	int	Minimum PnL (USDC).

Returns dict.

get_trader_profile(address)

Get a trader’s detailed profile including tags, stats, and metadata.

data = client.get_trader_profile('0xabc…')

Parameter	Type	Description
`address`	str	The EVM wallet address of the trader.

Returns dict.

get_trader_markets(address, params)

Retrieve a trader’s trading history aggregated by market.

data = client.get_trader_markets('0xabc…', params)

Parameter	Type	Description
`address`	str	The EVM wallet address of the trader.
`limit`	int	Number of markets to return. Default: `20`, max: `100`.
`offset`	int	Pagination offset. Default: `0`.
`is_active`	bool	Filter by active (unresolved) markets only.
`include_orders`	bool	Include the trader’s individual orders for each market.

Returns dict.

get_trader_market_orders(address, market_id)

Retrieve a specific trader’s order history within a specific market.

data = client.get_trader_market_orders('0xabc…', 12345)

Parameter	Type	Description
`address`	str	The EVM wallet address of the trader.
`market_id`	str \| int	The specific market ID.

Returns dict.

get_trader_hourly_stats(address, params)

Get hourly aggregated performance data for a specific trader.

data = client.get_trader_hourly_stats('0xabc…', params)

Parameter	Type	Description
`address`	str	The EVM wallet address of the trader.
`tag`	str	(Optional) Filter by specific market tag (e.g., `'crypto'`, `'politics'`). Default: `'all'`.

Returns dict.

get_market_detail(market_id)

Retrieve detailed information about a specific market.

data = client.get_market_detail(12345)

Parameter	Type	Description
`market_id`	str \| int	The unique numeric ID of the market.

Returns dict.

stream_orders(params) — generator

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:

Key	Type	Description
`traders`	str	Comma-separated list of trader addresses to follow.
`tags`	str	Comma-separated list of market tags to filter by.
`series`	str	Comma-separated list of series slugs.
`min_usdc`	float	Minimum USDC trade size.
`max_usdc`	float	Maximum USDC trade size.
`min_shares`	float	Minimum shares amount.
`max_shares`	float	Maximum shares amount.
`min_price`	float	Minimum price (0–100).
`max_price`	float	Maximum price (0–100).
`format`	str	Set 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 trader hourly stats
hourly_response = client.get_trader_hourly_stats(trader_address)
for s in hourly_response.get('stats', [])[:3]:
    print(f"Hour: {s.get('hour')} | PNL: ${format_amount(s.get('pnl', '0'))} | Volume: ${format_amount(s.get('volume', '0'))}")

# 6. 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'))}")

# 7. 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

Get Started

Guides

SDKs

Requirements

Installation

Instantiation

Methods

Compact format parsing

Complete example

​Requirements

​Installation

​Instantiation

​Methods

​Compact format parsing

​Complete example

Requirements

Installation

Instantiation

Methods

Compact format parsing

Complete example