Orders SDK

pip install requests cryptography

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

class PolymarketOrdersSDK:
    """SDK for Polymarket US Orders API"""

    # Order Intent constants (string enum values)
    ORDER_INTENT_BUY_LONG = "ORDER_INTENT_BUY_LONG"      # Buy YES
    ORDER_INTENT_SELL_LONG = "ORDER_INTENT_SELL_LONG"    # Sell YES
    ORDER_INTENT_BUY_SHORT = "ORDER_INTENT_BUY_SHORT"    # Buy NO
    ORDER_INTENT_SELL_SHORT = "ORDER_INTENT_SELL_SHORT"  # Sell NO

    # Order Type constants
    ORDER_TYPE_LIMIT = "ORDER_TYPE_LIMIT"
    ORDER_TYPE_MARKET = "ORDER_TYPE_MARKET"

    # Time in Force constants
    TIF_GOOD_TILL_CANCEL = "TIME_IN_FORCE_GOOD_TILL_CANCEL"
    TIF_GOOD_TILL_DATE = "TIME_IN_FORCE_GOOD_TILL_DATE"
    TIF_IMMEDIATE_OR_CANCEL = "TIME_IN_FORCE_IMMEDIATE_OR_CANCEL"
    TIF_FILL_OR_KILL = "TIME_IN_FORCE_FILL_OR_KILL"

    # Manual Order Indicator constants
    MANUAL_ORDER = "MANUAL_ORDER_INDICATOR_MANUAL"
    AUTOMATIC_ORDER = "MANUAL_ORDER_INDICATOR_AUTOMATIC"

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

        Args:
            api_key_id: Your API key ID from developer portal
            private_key_base64: Your base64-encoded Ed25519 private key (64 bytes)
        """
        self.api_key_id = api_key_id
        self.base_url = "https://api.polymarket.us"

        # Parse Ed25519 private key
        private_key_bytes = base64.b64decode(private_key_base64)
        self.private_key = ed25519.Ed25519PrivateKey.from_private_bytes(
            private_key_bytes[:32]  # First 32 bytes are the seed
        )

    def _sign_request(self, method, path):
        """Generate Ed25519 signature for request"""
        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 create_order(self, market_slug, intent, order_type=None, price=None,
                    quantity=None, tif=None, synchronous=False,
                    participate_dont_initiate=False, manual_order_indicator=None):
        """
        Create a new order

        Args:
            market_slug: Market slug to trade in
            intent: Order intent string (use class constants like ORDER_INTENT_BUY_LONG)
            order_type: Order type string (ORDER_TYPE_LIMIT or ORDER_TYPE_MARKET)
            price: Order price as dict {"value": "0.55", "currency": "USD"}
            quantity: Quantity in shares
            tif: Time in force string (use class constants like TIF_GOOD_TILL_CANCEL)
            synchronous: Wait for execution before returning
            participate_dont_initiate: If True, order must rest on book (maker only)
            manual_order_indicator: MANUAL_ORDER or AUTOMATIC_ORDER

        Returns:
            dict: Order response with order ID
        """
        path = "/v1/orders"
        headers = self._sign_request("POST", path)
        headers["Content-Type"] = "application/json"

        payload = {
            "marketSlug": market_slug,
            "intent": intent,
            "synchronousExecution": synchronous
        }

        if order_type:
            payload["type"] = order_type
        if price:
            payload["price"] = price
        if quantity:
            payload["quantity"] = quantity
        if tif:
            payload["tif"] = tif
        if participate_dont_initiate:
            payload["participateDontInitiate"] = True
        if manual_order_indicator:
            payload["manualOrderIndicator"] = manual_order_indicator

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

    def get_open_orders(self, market_slugs=None):
        """
        Get all open orders

        Args:
            market_slugs: Optional list of market slugs to filter by

        Returns:
            dict: List of open orders
        """
        path = "/v1/orders/open"
        headers = self._sign_request("GET", path)

        params = {}
        if market_slugs:
            params["slugs"] = market_slugs

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

    def get_order(self, order_id):
        """
        Get details for a specific order

        Args:
            order_id: Exchange-assigned order ID

        Returns:
            dict: Order details
        """
        path = f"/v1/order/{order_id}"
        headers = self._sign_request("GET", path)

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

    def modify_order(self, order_id, market_slug, price=None, quantity=None, tif=None):
        """
        Modify an existing order

        Args:
            order_id: Exchange-assigned order ID
            market_slug: Market slug of the order
            price: New order price as dict {"value": "0.55", "currency": "USD"}
            quantity: New quantity in shares
            tif: New time in force string

        Returns:
            dict: Empty response on success
        """
        path = f"/v1/order/{order_id}/modify"
        headers = self._sign_request("POST", path)
        headers["Content-Type"] = "application/json"

        payload = {"marketSlug": market_slug}
        if price:
            payload["price"] = price
        if quantity:
            payload["quantity"] = quantity
        if tif:
            payload["tif"] = tif

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

    def cancel_order(self, order_id, market_slug):
        """
        Cancel a specific order

        Args:
            order_id: Exchange-assigned order ID
            market_slug: Market slug of the order

        Returns:
            dict: Empty response on success
        """
        path = f"/v1/order/{order_id}/cancel"
        headers = self._sign_request("POST", path)
        headers["Content-Type"] = "application/json"

        response = requests.post(
            f"{self.base_url}{path}",
            headers=headers,
            json={"marketSlug": market_slug}
        )
        response.raise_for_status()
        return response.json()

    def cancel_all_orders(self, market_slugs=None):
        """
        Cancel all open orders

        Args:
            market_slugs: Optional list of market slugs to filter cancellation

        Returns:
            dict: List of canceled order IDs
        """
        path = "/v1/orders/open/cancel"
        headers = self._sign_request("POST", path)
        headers["Content-Type"] = "application/json"

        payload = {}
        if market_slugs:
            payload["slugs"] = market_slugs

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

    def close_position(self, market_slug, synchronous=False, manual_order_indicator=None,
                       slippage_bips=None, slippage_ticks=None, current_price=None):
        """
        Close an existing position

        Args:
            market_slug: Market slug of the position to close
            synchronous: Wait for execution before returning
            manual_order_indicator: MANUAL_ORDER or AUTOMATIC_ORDER
            slippage_bips: Slippage tolerance in basis points
            slippage_ticks: Slippage tolerance in ticks (takes priority over bips)
            current_price: Current price for slippage calculation {"value": "0.50", "currency": "USD"}

        Returns:
            dict: Close position order response
        """
        path = "/v1/order/close-position"
        headers = self._sign_request("POST", path)
        headers["Content-Type"] = "application/json"

        payload = {
            "marketSlug": market_slug,
            "synchronousExecution": synchronous
        }

        if manual_order_indicator:
            payload["manualOrderIndicator"] = manual_order_indicator

        if slippage_bips or slippage_ticks or current_price:
            slippage = {}
            if current_price:
                slippage["currentPrice"] = current_price
            if slippage_bips:
                slippage["bips"] = slippage_bips
            if slippage_ticks:
                slippage["ticks"] = slippage_ticks
            payload["slippageTolerance"] = slippage

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

    def preview_order(self, market_slug, intent, order_type=None, price=None, quantity=None):
        """
        Preview an order before submission

        Args:
            market_slug: Market slug to trade in
            intent: Order intent string (use class constants like ORDER_INTENT_BUY_LONG)
            order_type: Order type string (ORDER_TYPE_LIMIT or ORDER_TYPE_MARKET)
            price: Order price as dict {"value": "0.55", "currency": "USD"}
            quantity: Quantity in shares

        Returns:
            dict: Previewed order with calculated values
        """
        path = "/v1/order/preview"
        headers = self._sign_request("POST", path)
        headers["Content-Type"] = "application/json"

        request = {
            "marketSlug": market_slug,
            "intent": intent
        }

        if order_type:
            request["type"] = order_type
        if price:
            request["price"] = price
        if quantity:
            request["quantity"] = quantity

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


# Example usage
if __name__ == "__main__":
    # Load credentials from environment
    api_key_id = os.environ.get("POLYMARKET_API_KEY")
    private_key = os.environ.get("POLYMARKET_PRIVATE_KEY")

    # Initialize SDK
    sdk = PolymarketOrdersSDK(api_key_id, private_key)

    # Example 1: Preview a limit order to buy YES at $0.65
    preview = sdk.preview_order(
        market_slug="super-bowl-lix-chiefs-vs-eagles",
        intent=sdk.ORDER_INTENT_BUY_LONG,  # Buy YES
        order_type=sdk.ORDER_TYPE_LIMIT,
        price={"value": "0.65", "currency": "USD"},
        quantity=100
    )
    print("Order preview:", preview)

    # Example 2: Create a limit order to buy YES
    order = sdk.create_order(
        market_slug="super-bowl-lix-chiefs-vs-eagles",
        intent=sdk.ORDER_INTENT_BUY_LONG,  # Buy YES
        order_type=sdk.ORDER_TYPE_LIMIT,
        price={"value": "0.65", "currency": "USD"},
        quantity=100,
        tif=sdk.TIF_GOOD_TILL_CANCEL,
        manual_order_indicator=sdk.MANUAL_ORDER
    )
    print("Order created:", order["id"])

    # Example 3: Create a limit order to buy NO at $0.45
    no_order = sdk.create_order(
        market_slug="super-bowl-lix-chiefs-vs-eagles",
        intent=sdk.ORDER_INTENT_BUY_SHORT,  # Buy NO
        order_type=sdk.ORDER_TYPE_LIMIT,
        price={"value": "0.45", "currency": "USD"},
        quantity=50,
        tif=sdk.TIF_GOOD_TILL_CANCEL
    )
    print("NO order created:", no_order["id"])

    # Example 4: Get all open orders
    open_orders = sdk.get_open_orders()
    print(f"Open orders: {len(open_orders['orders'])}")

    # Example 5: Get specific order details
    if open_orders["orders"]:
        order_id = open_orders["orders"][0]["id"]
        order_details = sdk.get_order(order_id)
        print("Order details:", order_details)

    # Example 6: Modify an order's price
    if order:
        sdk.modify_order(
            order_id=order["id"],
            market_slug="super-bowl-lix-chiefs-vs-eagles",
            price={"value": "0.60", "currency": "USD"}
        )
        print("Order modified")

    # Example 7: Cancel a specific order
    if order:
        canceled = sdk.cancel_order(
            order_id=order["id"],
            market_slug="super-bowl-lix-chiefs-vs-eagles"
        )
        print("Order canceled")

    # Example 8: Close an existing position
    close_result = sdk.close_position(
        market_slug="super-bowl-lix-chiefs-vs-eagles",
        synchronous=True,
        slippage_ticks=5,
        current_price={"value": "0.50", "currency": "USD"}
    )
    print("Position closed:", close_result)

    # Example 9: Cancel all open orders for a market
    result = sdk.cancel_all_orders(market_slugs=["super-bowl-lix-chiefs-vs-eagles"])
    print(f"Canceled {len(result['canceledOrderIds'])} orders")

{
  "marketSlug": "your-market-slug",
  "type": "ORDER_TYPE_LIMIT",
  "price": { "value": "0.45", "currency": "USD" },
  "quantity": 10,
  "tif": "TIME_IN_FORCE_GOOD_TILL_CANCEL",
  "intent": "ORDER_INTENT_BUY_SHORT"
}

try:
    order = sdk.create_order(
        market_slug="super-bowl-lix-chiefs-vs-eagles",
        intent=sdk.ORDER_INTENT_BUY_LONG,
        order_type=sdk.ORDER_TYPE_LIMIT,
        price={"value": "0.65", "currency": "USD"},
        quantity=100
    )
except requests.exceptions.HTTPError as e:
    if e.response.status_code == 400:
        print("Invalid order parameters")
    elif e.response.status_code == 401:
        print("Authentication failed")
    else:
        print(f"Error: {e.response.text}")