Market SDK

pip install requests cryptography

import os
import time
import base64
import requests
from cryptography.hazmat.primitives.asymmetric import ed25519

class PolymarketMarketSDK:
    """SDK for Polymarket US Market API"""

    def __init__(self, api_key_id=None, private_key_base64=None):
        """
        Initialize the SDK

        Note: Market data endpoints are public and don't require authentication,
        but auth can be provided for rate limit increases.

        Args:
            api_key_id: Optional API key ID from developer portal
            private_key_base64: Optional base64-encoded Ed25519 private key
        """
        self.api_key_id = api_key_id
        self.base_url = "https://api.polymarket.us"
        self.private_key = None

        if private_key_base64:
            private_key_bytes = base64.b64decode(private_key_base64)
            self.private_key = ed25519.Ed25519PrivateKey.from_private_bytes(
                private_key_bytes[:32]
            )

    def _sign_request(self, method, path):
        """Generate Ed25519 signature for request (optional for market data)"""
        if not self.private_key or not self.api_key_id:
            return {}

        timestamp = str(int(time.time() * 1000))
        message = f"{timestamp}{method}{path}"
        signature = self.private_key.sign(message.encode('utf-8'))
        signature_base64 = base64.b64encode(signature).decode('utf-8')

        return {
            "X-PM-Access-Key": self.api_key_id,
            "X-PM-Timestamp": timestamp,
            "X-PM-Signature": signature_base64
        }

    def get_markets(self, limit=50, offset=0, active=True, closed=False,
                   archived=False, order_by=None, order_direction="desc",
                   **filters):
        """
        Get list of markets with optional filtering

        Args:
            limit: Number of markets to return
            offset: Pagination offset
            active: Filter for active markets
            closed: Filter for closed markets
            archived: Filter for archived markets
            order_by: List of fields to order by (e.g., ["volume", "liquidity"])
            order_direction: "asc" or "desc"
            **filters: Additional filters:
                - liquidityNumMin/Max: Liquidity range
                - volumeNumMin/Max: Volume range
                - startDateMin/Max: Start date range
                - endDateMin/Max: End date range
                - categories: List of categories
                - tagId: Tag ID filter

        Returns:
            dict: List of markets with metadata
        """
        path = "/v1/markets"
        headers = self._sign_request("GET", path)

        params = {
            "limit": limit,
            "offset": offset,
            "active": active,
            "closed": closed,
            "archived": archived,
            "orderDirection": order_direction
        }

        if order_by:
            params["orderBy"] = order_by

        # Add any additional filters
        params.update(filters)

        response = requests.get(
            f"{self.base_url}{path}",
            headers=headers,
            params=params
        )
        response.raise_for_status()
        return response.json()

    def get_market_by_slug(self, slug):
        """
        Get a specific market by its slug

        Args:
            slug: Market slug

        Returns:
            dict: Market details
        """
        markets = self.get_markets(slug=[slug], limit=1)
        if markets.get("data"):
            return markets["data"][0]
        return None

    def get_market_by_id(self, market_id):
        """
        Get a specific market by its ID

        Args:
            market_id: Market ID

        Returns:
            dict: Market details
        """
        markets = self.get_markets(id=[market_id], limit=1)
        if markets.get("data"):
            return markets["data"][0]
        return None

    def get_order_book(self, market_slug):
        """
        Get order book for a specific market

        Args:
            market_slug: Market slug

        Returns:
            dict: Order book with bids and asks
        """
        path = f"/v1/markets/{market_slug}/orderbook"
        headers = self._sign_request("GET", path)

        response = requests.get(
            f"{self.base_url}{path}",
            headers=headers
        )
        response.raise_for_status()
        return response.json()

    def get_best_prices(self, market_slug):
        """
        Get best bid and ask prices for a market

        Args:
            market_slug: Market slug

        Returns:
            dict: Best bid and ask prices
        """
        order_book = self.get_order_book(market_slug)

        bids = order_book.get("bids", [])
        asks = order_book.get("asks", [])

        best_bid = float(bids[0]["price"]["value"]) if bids else None
        best_ask = float(asks[0]["price"]["value"]) if asks else None

        spread = (best_ask - best_bid) if (best_bid and best_ask) else None

        return {
            "bestBid": best_bid,
            "bestAsk": best_ask,
            "spread": spread,
            "midPrice": (best_bid + best_ask) / 2 if (best_bid and best_ask) else None
        }

    def get_active_markets(self, limit=50):
        """
        Get all active markets

        Args:
            limit: Number of markets to return

        Returns:
            list: Active markets
        """
        response = self.get_markets(limit=limit, active=True, closed=False)
        return response.get("data", [])

    def search_markets(self, categories=None, min_liquidity=None,
                      min_volume=None, limit=50):
        """
        Search markets with filters

        Args:
            categories: List of category names
            min_liquidity: Minimum liquidity filter
            min_volume: Minimum volume filter
            limit: Number of results

        Returns:
            list: Matching markets
        """
        filters = {"limit": limit}

        if categories:
            filters["categories"] = categories
        if min_liquidity:
            filters["liquidityNumMin"] = min_liquidity
        if min_volume:
            filters["volumeNumMin"] = min_volume

        response = self.get_markets(**filters)
        return response.get("data", [])

    def get_market_stats(self, market_slug):
        """
        Get statistics for a specific market

        Args:
            market_slug: Market slug

        Returns:
            dict: Market statistics
        """
        market = self.get_market_by_slug(market_slug)
        if not market:
            return None

        return {
            "slug": market.get("slug"),
            "title": market.get("question"),
            "volume": float(market.get("volume", 0)),
            "liquidity": float(market.get("liquidity", 0)),
            "active": market.get("active"),
            "closed": market.get("closed"),
            "endDate": market.get("endDate"),
            "outcomePrices": market.get("outcomePrices", [])
        }


# Example usage
if __name__ == "__main__":
    # Market data is public - no auth required
    # But you can provide auth for higher rate limits
    api_key_id = os.environ.get("POLYMARKET_API_KEY")
    private_key = os.environ.get("POLYMARKET_PRIVATE_KEY")

    # Initialize SDK (with or without auth)
    sdk = PolymarketMarketSDK(api_key_id, private_key)

    # Example 1: Get active markets
    markets = sdk.get_active_markets(limit=10)
    print(f"Active markets: {len(markets)}")

    for market in markets[:3]:
        print(f"\nMarket: {market.get('question')}")
        print(f"  Slug: {market.get('slug')}")
        print(f"  Volume: \${market.get('volume', 0):,.0f}")
        print(f"  Liquidity: \${market.get('liquidity', 0):,.0f}")

    # Example 2: Get specific market by slug
    market = sdk.get_market_by_slug("will-trump-win-2024")
    if market:
        print(f"\nMarket: {market.get('question')}")
        print(f"  Active: {market.get('active')}")
        print(f"  End Date: {market.get('endDate')}")

    # Example 3: Get order book
    order_book = sdk.get_order_book("will-trump-win-2024")
    print(f"\nOrder Book:")
    print(f"  Bids: {len(order_book.get('bids', []))}")
    print(f"  Asks: {len(order_book.get('asks', []))}")

    # Show top 5 bids and asks
    for bid in order_book.get("bids", [])[:5]:
        print(f"  BID: \${bid['price']['value']} x {bid['quantity']}")

    for ask in order_book.get("asks", [])[:5]:
        print(f"  ASK: \${ask['price']['value']} x {ask['quantity']}")

    # Example 4: Get best prices
    prices = sdk.get_best_prices("will-trump-win-2024")
    print(f"\nBest Prices:")
    print(f"  Bid: \${prices['bestBid']}")
    print(f"  Ask: \${prices['bestAsk']}")
    print(f"  Spread: \${prices['spread']:.4f}")
    print(f"  Mid: \${prices['midPrice']:.4f}")

    # Example 5: Search markets by category
    politics_markets = sdk.search_markets(
        categories=["Politics"],
        min_volume=10000,
        limit=20
    )
    print(f"\nPolitics markets with volume > \$10k: {len(politics_markets)}")

    # Example 6: Get market statistics
    stats = sdk.get_market_stats("will-trump-win-2024")
    if stats:
        print(f"\nMarket Statistics:")
        print(f"  Title: {stats['title']}")
        print(f"  Volume: \${stats['volume']:,.0f}")
        print(f"  Liquidity: \${stats['liquidity']:,.0f}")
        print(f"  Active: {stats['active']}")

    # Example 7: Filter by liquidity and volume
    liquid_markets = sdk.get_markets(
        liquidityNumMin=50000,
        volumeNumMin=100000,
        active=True,
        limit=10,
        order_by=["volume"],
        order_direction="desc"
    )
    print(f"\nHigh liquidity markets: {len(liquid_markets.get('data', []))}")

{
  "bids": [
    {
      "price": {"value": "0.65", "currency": "USD"},
      "quantity": 100
    }
  ],
  "asks": [
    {
      "price": {"value": "0.67", "currency": "USD"},
      "quantity": 50
    }
  ]
}

def calculate_market_metrics(sdk, market_slug):
    """Calculate various market metrics"""
    market = sdk.get_market_by_slug(market_slug)
    prices = sdk.get_best_prices(market_slug)

    volume = float(market.get("volume", 0))
    liquidity = float(market.get("liquidity", 0))

    return {
        "volume": volume,
        "liquidity": liquidity,
        "spread": prices["spread"],
        "spreadBps": (prices["spread"] / prices["midPrice"] * 10000) if prices["midPrice"] else None,
        "volumeToLiquidity": (volume / liquidity) if liquidity > 0 else None
    }

metrics = calculate_market_metrics(sdk, "will-trump-win-2024")
print(f"Spread: {metrics['spread']:.4f} (\${metrics['spreadBps']:.1f} bps)")
print(f"Volume/Liquidity: {metrics['volumeToLiquidity']:.2f}x")

try:
    markets = sdk.get_markets(limit=50)
except requests.exceptions.HTTPError as e:
    print(f"Error: {e.response.status_code} - {e.response.text}")